fix(memory): keep vector metadata in sync (#2295)

## Description

Fixes #2296.

Metadata-only memory updates can leave the primary store, vector-index
metadata, and cache inconsistent. TrafficLearner also performs an atomic
SQLite evidence increment that bypasses normal secondary-index refresh.

## 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

- Refresh vector metadata for HierarchicalMemory metadata-only,
importance, and entity-reference updates.
- Add a LocalBackend path that reloads a memory from the primary store
and refreshes vector metadata plus cache state.
- Preserve the atomic TrafficLearner SQL evidence increment, then
refresh secondary state only when a row was updated.
- Keep refresh failures fail-open and distinguish them from
primary-store increment failures in logs.
- Add backend-neutral contract tests instead of inspecting a specific
vector adapter private field.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`) — verified locally: mypy
1.20.2, no issues in 504 source files
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
141 passed, 1 skipped
ruff check: passed
ruff format --check: passed
```

The first CI run exposed one backend-specific test assertion against
HNSW private state while CI used SQLiteVectorIndex. Commit a1aff399
removes that assertion and keeps the backend-neutral mock contract test.

## Real Behavior Proof

- Environment: macOS, Python 3.13, current Headroom main.
- Exact command / steps: update a Memory with metadata only through
HierarchicalMemory, assert the vector index receives the updated Memory,
then perform a TrafficLearner evidence bump and assert LocalBackend
refresh is called only for an existing row.
- Observed result: metadata-only update refreshes vector metadata
without re-embedding content; evidence bump remains atomic and refreshes
vector/cache state; an unknown memory ID triggers no refresh.
- Not tested: remote memory backends, full repository suite, or live
multi-process writers against one SQLite database.

## 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 code where the behavior is not self-explanatory
- [x] I have made corresponding documentation changes (N/A — internal
bug fix, no user-facing docs/changelog impact)
- [x] My changes generate no new warnings
- [x] I have added tests that prove the fix is effective
- [x] New and existing focused unit tests pass locally
- [x] I have updated CHANGELOG.md if applicable (N/A — internal bug fix,
no user-facing docs/changelog impact)

## Screenshots (if applicable)

Not applicable.

## Additional Notes

Draft for storage-owner feedback on the refresh API and write overhead.
The refresh reuses the existing embedding and does not invoke the
embedder.

headroom/memory/backends/local.py

@@ -528,6 +528,12 @@ class LocalBackend:
         assert self._hierarchical_memory is not None
         return await self._hierarchical_memory.record_access(memory_ids)
 
+    async def refresh_memory_indexes(self, memory_id: str) -> Memory | None:
+        """Refresh vector metadata and cache state from the primary store."""
+        await self._ensure_initialized()
+        assert self._hierarchical_memory is not None
+        return await self._hierarchical_memory.refresh_memory_indexes(memory_id)
+
     async def update_memory(
         self,
         memory_id: str,

headroom/memory/core.py

@@ -494,10 +494,14 @@ class HierarchicalMemory:
         # Save updates
         await self._store.save(memory)
 
-        # Update indexes
+        # Vector indexes also carry filterable metadata. Refresh them for
+        # metadata-only updates, not just when the embedding content changes.
+        vector_metadata_changed = any(
+            value is not None for value in (importance, entity_refs, metadata)
+        )
+        if (content_changed or vector_metadata_changed) and memory.embedding is not None:
+            await self._vector_index.index(memory)
         if content_changed:
-            if memory.embedding is not None:
-                await self._vector_index.index(memory)
             await self._index_for_text_search(memory)
 
         # Invalidate and re-cache
@@ -507,6 +511,21 @@ class HierarchicalMemory:
 
         return memory
 
+    async def refresh_memory_indexes(self, memory_id: str) -> Memory | None:
+        """Refresh secondary index metadata from the primary memory store."""
+        memory = await self._store.get(memory_id)
+        if memory is None:
+            return None
+
+        if memory.embedding is not None:
+            await self._vector_index.index(memory)
+
+        if self._cache is not None:
+            await self._cache.invalidate(memory_id)
+            await self._cache.put(memory)
+
+        return memory
+
     async def supersede(
         self,
         old_memory_id: str,

headroom/memory/traffic_learner.py

@@ -1389,10 +1389,10 @@ class TrafficLearner:
 
         now_iso = datetime.now(timezone.utc).isoformat()
 
-        def _bump() -> None:
+        def _bump() -> bool:
             conn = sqlite3.connect(str(db_path))
             try:
-                conn.execute(
+                cursor = conn.execute(
                     "UPDATE memories SET metadata = json_set("
                     "metadata, '$.evidence_count', "
                     "COALESCE(json_extract(metadata, '$.evidence_count'), 0) + 1, "
@@ -1401,13 +1401,26 @@ class TrafficLearner:
                     (now_iso, memory_id),
                 )
                 conn.commit()
+                return cursor.rowcount > 0
             finally:
                 conn.close()
 
         try:
-            await asyncio.to_thread(_bump)
+            updated = await asyncio.to_thread(_bump)
         except Exception as e:
             logger.debug("Traffic learner evidence bump failed for %s: %s", memory_id, e)
+            return
+
+        refresh = getattr(self._backend, "refresh_memory_indexes", None)
+        if updated and refresh is not None:
+            try:
+                await refresh(memory_id)
+            except Exception as e:
+                logger.debug(
+                    "Traffic learner evidence index refresh failed for %s: %s",
+                    memory_id,
+                    e,
+                )
 
     # =========================================================================
     # Convenience: Extract from Anthropic messages format

tests/test_memory/test_core_metadata_index_sync.py

@@ -0,0 +1,59 @@
+"""Regression tests for primary-store and vector-metadata consistency."""
+
+from unittest.mock import AsyncMock
+
+import numpy as np
+import pytest
+
+from headroom.memory.core import HierarchicalMemory
+from headroom.memory.models import Memory
+
+
+def _memory_system(memory: Memory, *, cache=None):
+    store = AsyncMock()
+    store.get.return_value = memory
+    vector_index = AsyncMock()
+    text_index = AsyncMock()
+    embedder = AsyncMock()
+    system = HierarchicalMemory(store, vector_index, text_index, embedder, cache=cache)
+    return system, store, vector_index, text_index
+
+
+@pytest.mark.asyncio
+async def test_metadata_only_update_refreshes_vector_metadata() -> None:
+    memory = Memory(
+        id="memory-1",
+        content="stable content",
+        user_id="default",
+        embedding=np.array([0.1, 0.2], dtype=np.float32),
+        metadata={"evidence_count": 5},
+    )
+    system, store, vector_index, text_index = _memory_system(memory)
+
+    updated = await system.update(memory.id, metadata={"evidence_count": 7})
+
+    assert updated is memory
+    store.save.assert_awaited_once_with(memory)
+    vector_index.index.assert_awaited_once_with(memory)
+    text_index.index.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_refresh_memory_indexes_reloads_store_and_cache() -> None:
+    memory = Memory(
+        id="memory-1",
+        content="stable content",
+        user_id="default",
+        embedding=np.array([0.1, 0.2], dtype=np.float32),
+        metadata={"evidence_count": 7},
+    )
+    cache = AsyncMock()
+    system, store, vector_index, _ = _memory_system(memory, cache=cache)
+
+    refreshed = await system.refresh_memory_indexes(memory.id)
+
+    assert refreshed is memory
+    store.get.assert_awaited_once_with(memory.id)
+    vector_index.index.assert_awaited_once_with(memory)
+    cache.invalidate.assert_awaited_once_with(memory.id)
+    cache.put.assert_awaited_once_with(memory)

tests/test_memory/test_traffic_learner.py

@@ -904,6 +904,10 @@ class _FakeBackend:
 
         self._config = _types.SimpleNamespace(db_path=str(db_path))
         self._db_path = str(db_path)
+        self.refreshed_ids = []
+
+    async def refresh_memory_indexes(self, memory_id: str):
+        self.refreshed_ids.append(memory_id)
 
     async def save_memory(
         self,
@@ -1465,6 +1469,7 @@ class TestBumpEdgeCases:
         learner = TrafficLearner(backend=backend, min_evidence=1)
         await learner._bump_persisted_evidence("no-such-id")
         assert _read_traffic_rows(db) == []
+        assert backend.refreshed_ids == []
 
 
 # =============================================================================
@@ -2092,6 +2097,7 @@ class TestBumpPersistsLastSeenAt:
         # Should be parseable back.
         parsed = _parse_iso_timestamp(meta["last_seen_at"])
         assert parsed is not None
+        assert backend.refreshed_ids == ["row-1"]
 
 
 class TestHydrateLegacyRow: