feat(decay): native usage-aware memory decay engine + run_decay_scan MCP tool

alembic/versions/20260704_add_memory_usage_tracking.py

@@ -0,0 +1,59 @@
+"""add usage tracking columns to memories (decay engine)
+
+Revision ID: 20260704_usage_tracking
+Revises: 20260408_provenance_all
+Create Date: 2026-07-04
+
+Adds two columns to the `memories` table for the forgetful-hulkito decay
+engine (plan `native_memory_decay_engine`):
+
+- `access_count` (INTEGER NOT NULL DEFAULT 0) — incremented each time a
+  memory is returned by a real read path (query_memory final results,
+  get_memory). Mutated only by the internal `record_memory_access`
+  repository method so `updated_at` is not distorted.
+- `last_accessed_at` (DATETIME WITH TIME ZONE, nullable) — last time the
+  memory was returned by a read path. Nullable because existing rows have
+  never been accessed through the new path.
+
+Also adds an index `ix_memories_last_accessed_at` to support the
+`get_decay_candidates` query.
+
+Both columns are backwards-compatible: existing rows get `access_count=0`
+and `last_accessed_at=NULL`, and the decay formula treats NULL
+`last_accessed_at` as `updated_at ?? created_at`.
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "20260704_usage_tracking"
+down_revision: str | Sequence[str] | None = "20260408_provenance_all"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Add access_count + last_accessed_at columns and index."""
+    op.add_column(
+        "memories",
+        sa.Column("access_count", sa.Integer(), nullable=False, server_default="0"),
+    )
+    op.add_column(
+        "memories",
+        sa.Column("last_accessed_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_index(
+        "ix_memories_last_accessed_at",
+        "memories",
+        ["last_accessed_at"],
+    )
+
+
+def downgrade() -> None:
+    """Remove access_count + last_accessed_at columns and index."""
+    op.drop_index("ix_memories_last_accessed_at", table_name="memories")
+    op.drop_column("memories", "last_accessed_at")
+    op.drop_column("memories", "access_count")

app/models/memory_models.py

@@ -286,6 +286,15 @@ class Memory(MemoryCreate):
     superseded_by: int | None = Field(default=None, description="ID of memory that supersedes this one")
     obsoleted_at: datetime | None = Field(default=None, description="When this memory was marked obsolete")
 
+    # Usage tracking (read-only from the API surface; mutated only by the
+    # internal `record_memory_access` repository method so `updated_at` is not
+    # distorted and external callers cannot fake read counters).
+    access_count: int = Field(default=0, description="Number of times this memory has been returned by a read path")
+    last_accessed_at: datetime | None = Field(
+        default=None,
+        description="Last time this memory was returned by a read path (null = never accessed)",
+    )
+
     model_config = ConfigDict(from_attributes=True)
 
 class MemorySummary(BaseModel):
@@ -404,4 +413,70 @@ def validate_related_ids(cls, v, info):
         return v
 
 
+# ---------------------------------------------------------------------------
+# Memory decay engine (forgetful-hulkito fork)
+#
+# Usage tracking columns `access_count` / `last_accessed_at` are populated on
+# real read paths only (query_memory final results, get_memory). The
+# `run_decay_scan` MCP tool uses them together with age to propose confidence
+# decay and obsolete candidates. See plan `native_memory_decay_engine`.
+# ---------------------------------------------------------------------------
+
+
+class DecayScanRequest(BaseModel):
+    """Scope parameters for a decay scan.
+
+    Args:
+        memory_ids: Explicit ids to scan (already user-scoped server-side).
+        project_id: Restrict to a single project owned by the caller.
+        dry_run: When True, only report proposed changes; do not write.
+    """
+
+    memory_ids: list[int] | None = Field(
+        default=None,
+        description="Explicit memory ids to scan; leave null to use project_id or global user scope",
+    )
+    project_id: int | None = Field(
+        default=None,
+        description="Restrict the scan to memories of a single project owned by the caller",
+    )
+    dry_run: bool = Field(
+        default=True,
+        description="Preview-only mode (no writes). Set to False to apply confidence decay and obsolete mutations.",
+    )
+
+
+class DecayCandidateAction(BaseModel):
+    """A single proposed mutation for one memory.
+
+    `kind` is one of:
+    - "decay": lower `confidence` by `delta` (clamped at 0.0).
+    - "obsolete": mark the memory obsolete with `reason`.
+    - "skip": report the candidate without mutating it (used when
+      `confidence is None` or another guard fires); `reason` explains why.
+    """
+
+    kind: str
+    memory_id: int
+    title: str | None = None
+    current_confidence: float | None = None
+    proposed_confidence: float | None = None
+    delta: float | None = None
+    reason: str | None = None
+    age_days: int | None = None
+    days_since_access: int | None = None
+    access_count: int | None = None
+
+
+class DecayScanResponse(BaseModel):
+    """Result of a decay scan (dry-run or live)."""
+
+    dry_run: bool
+    scanned: int
+    actions: list[DecayCandidateAction] = Field(default_factory=list)
+    applied: int = Field(0, description="Number of actions actually applied (0 in dry-run)")
+    failed: list[dict] = Field(default_factory=list, description="Per-memory failures: {memory_id, reason}")
+
+
+
 

app/protocols/memory_protocol.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Any, Protocol
 from uuid import UUID
 
@@ -265,4 +266,40 @@ async def validate_search_works(self) -> bool:
         """Run a smoke-test semantic search"""
         ...
 
+    # ------------------------------------------------------------------
+    # Usage tracking + decay engine (forgetful-hulkito fork)
+    # ------------------------------------------------------------------
+
+    async def record_memory_access(
+        self,
+        user_id: UUID,
+        memory_ids: list[int],
+        accessed_at: datetime | None = None,
+    ) -> int:
+        """Increment `access_count` and update `last_accessed_at` for the given
+        memory ids, scoped to `user_id`.
+
+        MUST NOT mutate `updated_at` — usage tracking is an internal read-side
+        concern and must not distort normal memory recency. Returns the number
+        of rows actually updated (owned + non-obsolete).
+        """
+        ...
+
+    async def get_decay_candidates(
+        self,
+        user_id: UUID,
+        memory_ids: list[int] | None = None,
+        project_id: int | None = None,
+        max_age_days: int | None = None,
+    ) -> list[Memory]:
+        """Return non-obsolete memories owned by `user_id` that are eligible
+        for decay review, ordered oldest first.
+
+        Filters: explicit `memory_ids` (already user-scoped) and/or a single
+        `project_id`. `max_age_days` optionally limits to memories whose
+        effective access date (`last_accessed_at ?? updated_at ?? created_at`)
+        is older than the given threshold.
+        """
+        ...
+
 

app/repositories/postgres/memory_repository.py

@@ -1,6 +1,6 @@
 """Memory repository for postgres data access operations
 """
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from typing import Any
 from uuid import UUID
 
@@ -1759,3 +1759,92 @@ async def get_subgraph_nodes(
             })
 
             return nodes, truncated
+
+    # ------------------------------------------------------------------
+    # Usage tracking + decay engine (forgetful-hulkito fork)
+    # ------------------------------------------------------------------
+
+    async def record_memory_access(
+        self,
+        user_id: UUID,
+        memory_ids: list[int],
+        accessed_at: datetime | None = None,
+    ) -> int:
+        """Increment `access_count` and update `last_accessed_at` for the given
+        memory ids, scoped to `user_id`. Does NOT mutate `updated_at`.
+
+        Returns the number of rows actually updated (owned + non-obsolete).
+        """
+        if not memory_ids:
+            return 0
+        ts = accessed_at or datetime.now(UTC)
+        async with self.db_adapter.system_session() as session:
+            stmt = (
+                update(MemoryTable)
+                .where(
+                    MemoryTable.user_id == user_id,
+                    MemoryTable.is_obsolete.is_(False),
+                    MemoryTable.id.in_(memory_ids),
+                )
+                .values(
+                    access_count=MemoryTable.access_count + 1,
+                    last_accessed_at=ts,
+                )
+            )
+            result = await session.execute(stmt)
+            return result.rowcount or 0
+
+    async def get_decay_candidates(
+        self,
+        user_id: UUID,
+        memory_ids: list[int] | None = None,
+        project_id: int | None = None,
+        max_age_days: int | None = None,
+    ) -> list[Memory]:
+        """Return non-obsolete memories owned by `user_id` eligible for decay
+        review, ordered oldest (by effective access date) first.
+
+        Effective access date = `last_accessed_at ?? updated_at ?? created_at`.
+        The optional `max_age_days` filter is applied in Python after reading
+        to keep the COALESCE-on-nullable-datetime logic backend-agnostic.
+        """
+        conditions = [
+            MemoryTable.user_id == user_id,
+            MemoryTable.is_obsolete.is_(False),
+        ]
+        if memory_ids:
+            conditions.append(MemoryTable.id.in_(memory_ids))
+
+        stmt = (
+            select(MemoryTable)
+            .options(
+                selectinload(MemoryTable.projects),
+                selectinload(MemoryTable.linked_memories),
+                selectinload(MemoryTable.linking_memories),
+                selectinload(MemoryTable.code_artifacts),
+                selectinload(MemoryTable.documents),
+                selectinload(MemoryTable.files),
+                selectinload(MemoryTable.skills),
+            )
+            .where(*conditions)
+        )
+        if project_id is not None:
+            stmt = stmt.join(MemoryTable.projects).where(
+                ProjectsTable.id == project_id,
+                ProjectsTable.user_id == user_id,
+            )
+
+        async with self.db_adapter.system_session() as session:
+            result = await session.execute(stmt)
+            memories_orm = result.scalars().unique().all()
+            memories = [Memory.model_validate(m) for m in memories_orm]
+
+        def effective_access(mem: Memory) -> datetime:
+            return mem.last_accessed_at or mem.updated_at or mem.created_at
+
+        if max_age_days is not None:
+            cutoff = datetime.now(UTC) - timedelta(days=max_age_days)
+            memories = [m for m in memories if effective_access(m) <= cutoff]
+
+        memories.sort(key=lambda m: (effective_access(m), m.id))
+        return memories

app/repositories/postgres/postgres_tables.py

@@ -212,6 +212,11 @@ class MemoryTable(Base):
     superseded_by: Mapped[int] = mapped_column(Integer, ForeignKey("memories.id", ondelete="SET NULL"), nullable=True)
     obsoleted_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=True)
 
+    # Usage tracking (decay engine). Mutated only by the internal
+    # `record_memory_access` repository method so `updated_at` is not distorted.
+    access_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    last_accessed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=True)
+
     # Timestamps
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
@@ -401,6 +406,7 @@ def entity_ids(self) -> list[int]:
         Index("ix_memories_is_obsolete", "is_obsolete"),
         Index("ix_memories_superseded_by", "superseded_by"),
         Index("ix_memories_confidence", "confidence"),
+        Index("ix_memories_last_accessed_at", "last_accessed_at"),
     )
 
 class MemoryLinkTable(Base):