fix(memory): track MCP retrieval access (#2065)

## Description

Track successful native MCP `memory_search` retrievals in persistent
memory metadata. Returned memories now increment `access_count` and
update `last_accessed`, so MCP usage contributes to memory budget and
retention signals.

Closes #2061

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Add an atomic, deduplicated `MemoryStore.record_access` operation.
- Expose access recording through `HierarchicalMemory` and
`LocalBackend`, invalidating stale cache entries.
- Record only the final active memories actually returned by MCP search.
- Fail open if usage metadata cannot be written.
- Add SQLite and MCP regression coverage.

## Testing

- [x] Unit tests pass (`pytest`)
- [ ] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
pytest tests/test_memory --ignore=tests/test_memory/test_learn_flag.py -q
368 passed, 142 skipped, 158 warnings in 3.28s

pytest tests/test_memory/test_hierarchical.py tests/test_memory/test_mcp_server.py tests/test_memory/test_factory.py -q
40 passed, 53 skipped, 158 warnings in 0.75s
```

## Real Behavior Proof

- Environment: macOS, Python 3.13, SQLite memory store.
- Exact command / steps: save two memories; call `record_access` with
duplicate IDs plus a missing ID; read both rows; call it again for one
row.
- Observed result: each existing memory increments once per call,
duplicates do not double-count, missing IDs are ignored, and
`last_accessed` advances to the supplied timestamp.
- Not tested: the full repository suite and
`tests/test_memory/test_learn_flag.py`; the source checkout does not
include the compiled `headroom._core` Rust extension. Ruff and mypy were
not available in the local development environment; CI remains
authoritative for those checks.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Additional Notes

Documentation and changelog changes are not included because this is an
internal retrieval-metadata correction with no user-facing configuration
change. Access tracking is intentionally fail-open so a metadata write
failure cannot suppress a valid memory search result.

---------

Co-authored-by: xuyidiao <xuyidiao@bytedance.com>
Co-authored-by: JerrettDavis <mxjerrett@gmail.com>

headroom/memory/adapters/sqlite.py

@@ -342,6 +342,31 @@ class SQLiteMemoryStore:
 
             return [self._row_to_memory(row) for row in cursor]
 
+    async def record_access(
+        self,
+        memory_ids: list[str],
+        accessed_at: datetime | None = None,
+    ) -> int:
+        """Atomically record one retrieval for each distinct memory ID."""
+        unique_ids = list(dict.fromkeys(memory_ids))
+        if not unique_ids:
+            return 0
+
+        timestamp = accessed_at or datetime.utcnow()
+        placeholders = ", ".join("?" for _ in unique_ids)
+        with self._get_conn() as conn:
+            cursor = conn.execute(
+                f"""
+                UPDATE memories
+                SET access_count = access_count + 1,
+                    last_accessed = ?
+                WHERE id IN ({placeholders})
+                """,  # nosec B608
+                [timestamp.isoformat(), *unique_ids],
+            )
+            conn.commit()
+            return cursor.rowcount
+
     async def delete(self, memory_id: str) -> bool:
         """Delete a memory by ID.
 

headroom/memory/backends/local.py

@@ -516,6 +516,12 @@ class LocalBackend:
         results.sort(key=lambda x: x.score, reverse=True)
         return results[:top_k]
 
+    async def record_access(self, memory_ids: list[str]) -> int:
+        """Record retrieval metadata for memories returned to a caller."""
+        await self._ensure_initialized()
+        assert self._hierarchical_memory is not None
+        return await self._hierarchical_memory.record_access(memory_ids)
+
     async def update_memory(
         self,
         memory_id: str,

headroom/memory/core.py

@@ -304,6 +304,21 @@ class HierarchicalMemory:
 
         return memory
 
+    async def record_access(
+        self,
+        memory_ids: list[str],
+        accessed_at: datetime | None = None,
+    ) -> int:
+        """Record retrieval metadata for memories returned to a caller."""
+        unique_ids = list(dict.fromkeys(memory_ids))
+        if not unique_ids:
+            return 0
+
+        updated = await self._store.record_access(unique_ids, accessed_at)
+        if self._cache is not None:
+            await self._cache.invalidate_batch(unique_ids)
+        return updated
+
     async def query(self, filter: MemoryFilter) -> list[Memory]:
         """Query memories with filtering.
 

headroom/memory/mcp_server.py

@@ -245,6 +245,12 @@ async def _handle_search(
         # Trim to requested top_k
         active_results = active_results[:top_k]
 
+        try:
+            await backend.record_access([r.memory.id for r in active_results])
+        except Exception as e:
+            # Usage metadata must never make a successful retrieval fail.
+            logger.warning(f"Memory MCP: failed to record access: {e}")
+
         lines = []
         for i, r in enumerate(active_results, 1):
             score = f"{r.score:.2f}" if hasattr(r, "score") else "?"

headroom/memory/ports.py

@@ -311,6 +311,22 @@ class MemoryStore(Protocol):
         """
         ...
 
+    async def record_access(
+        self,
+        memory_ids: list[str],
+        accessed_at: datetime | None = None,
+    ) -> int:
+        """Record one retrieval for each distinct memory ID.
+
+        Args:
+            memory_ids: IDs of memories actually returned to a caller.
+            accessed_at: Retrieval time (defaults to now).
+
+        Returns:
+            Number of existing memories updated.
+        """
+        ...
+
     async def delete(self, memory_id: str) -> bool:
         """
         Delete a memory by ID.

tests/test_memory/test_hierarchical.py

@@ -182,6 +182,36 @@ class TestSQLiteMemoryStore:
             assert retrieved is not None
             assert retrieved.content == memory.content
 
+    @pytest.mark.asyncio
+    async def test_record_access_is_atomic_and_deduplicates_ids(self, store):
+        memories = [Memory(content=f"Memory {i}", user_id="alice") for i in range(2)]
+        await store.save_batch(memories)
+
+        first_access = datetime(2026, 7, 12, 9, 30)
+        updated = await store.record_access(
+            [memories[0].id, memories[0].id, memories[1].id, "missing"],
+            first_access,
+        )
+
+        assert updated == 2
+        first = await store.get(memories[0].id)
+        second = await store.get(memories[1].id)
+        assert first is not None
+        assert second is not None
+        assert first.access_count == 1
+        assert second.access_count == 1
+        assert first.last_accessed == first_access
+        assert second.last_accessed == first_access
+
+        second_access = datetime(2026, 7, 12, 9, 31)
+        assert await store.record_access([memories[0].id], second_access) == 1
+        first = await store.get(memories[0].id)
+        assert first is not None
+        assert first.access_count == 2
+        assert first.last_accessed == second_access
+
+        assert await store.record_access([]) == 0
+
     @pytest.mark.asyncio
     async def test_delete(self, store, sample_memory):
         """Test deleting a memory."""

tests/test_memory/test_mcp_server.py

@@ -163,3 +163,68 @@ def test_main_logs_memory_mcp_startup_context(monkeypatch, tmp_path, caplog) ->
         and "resolution=dynamic-cwd" in record.message
         for record in caplog.records
     )
+
+
+def test_search_records_access_only_for_returned_memories() -> None:
+    active = Memory(content="Active preference", user_id="alice")
+    extra = Memory(content="Lower-ranked preference", user_id="alice")
+    backend = SimpleNamespace(
+        search_memories=AsyncMock(
+            return_value=[
+                SimpleNamespace(memory=active, score=0.9, related_entities=[]),
+                SimpleNamespace(memory=extra, score=0.8, related_entities=[]),
+            ]
+        ),
+        get_memory=AsyncMock(
+            side_effect=lambda memory_id: {
+                active.id: active,
+                extra.id: extra,
+            }[memory_id]
+        ),
+        record_access=AsyncMock(return_value=1),
+    )
+
+    result = asyncio.run(
+        mcp_server_mod._handle_search(
+            backend,
+            {"query": "preference", "top_k": 1},
+            "alice",
+        )
+    )
+
+    backend.record_access.assert_awaited_once_with([active.id])
+    assert "Active preference" in result[0].kwargs["text"]
+    assert "Lower-ranked preference" not in result[0].kwargs["text"]
+
+
+def test_search_does_not_record_superseded_memories() -> None:
+    superseded = Memory(content="Old preference", user_id="alice")
+    replacement = Memory(content="Current preference", user_id="alice")
+    superseded.superseded_by = replacement.id
+    backend = SimpleNamespace(
+        search_memories=AsyncMock(
+            return_value=[SimpleNamespace(memory=superseded, score=0.9, related_entities=[])]
+        ),
+        get_memory=AsyncMock(return_value=superseded),
+        record_access=AsyncMock(),
+    )
+
+    result = asyncio.run(mcp_server_mod._handle_search(backend, {"query": "preference"}, "alice"))
+
+    backend.record_access.assert_not_awaited()
+    assert result[0].kwargs["text"] == "No memories found."
+
+
+def test_search_fails_open_when_access_tracking_fails() -> None:
+    memory = Memory(content="Useful preference", user_id="alice")
+    backend = SimpleNamespace(
+        search_memories=AsyncMock(
+            return_value=[SimpleNamespace(memory=memory, score=0.9, related_entities=[])]
+        ),
+        get_memory=AsyncMock(return_value=memory),
+        record_access=AsyncMock(side_effect=RuntimeError("write failed")),
+    )
+
+    result = asyncio.run(mcp_server_mod._handle_search(backend, {"query": "preference"}, "alice"))
+
+    assert "Useful preference" in result[0].kwargs["text"]