Invalidate profile cache after memory ingest

Author: strongkeep-debug

src/api/routes/memory.py

@@ -193,6 +193,7 @@ async def _run_ingest_payload(
             image_url=payload.get("image_url", ""),
             effort_level=payload.get("effort_level", "low"),
         )
+    _invalidate_profile_cache(user_id)
     data = IngestResponse(
         model=_model_name(pipeline.model),
         classification=_safe_classifications(result),
@@ -764,6 +765,13 @@ def _safe_classifications(result: Dict[str, Any]) -> list:
     return []
 
 
+def _invalidate_profile_cache(user_id: str) -> None:
+    try:
+        get_retrieval_pipeline().invalidate_profile_cache(user_id)
+    except Exception as exc:
+        logger.warning("Failed to invalidate profile cache for user=%s: %s", user_id, exc)
+
+
 async def _read_user_job(job_id: str, user_id: str) -> Dict[str, Any] | None:
     job = await asyncio.to_thread(get_default_job_store().get, job_id)
     if not job:

src/pipelines/retrieval.py

@@ -697,6 +697,14 @@ async def _get_profile_catalog(self, user_id: str):
         self._trim_cache(self._profile_catalog_cache, _PROFILE_CATALOG_CACHE_LIMIT)
         return catalog, results
 
+    def invalidate_profile_cache(self, user_id: str) -> None:
+        """Clear cached profile records after a user's memories are ingested."""
+
+        self._profile_catalog_cache.pop(user_id, None)
+        for key in list(self._retrieval_plan_cache):
+            if key[0] == user_id:
+                self._retrieval_plan_cache.pop(key, None)
+
     def _fetch_profile_catalog(self, user_id: str):
         """Fetch all profile entries for a user.
 

tests/api/test_memory_search_routes.py

@@ -18,6 +18,7 @@ class FakeSearchPipeline:
 
     def __init__(self) -> None:
         self.answer_calls = 0
+        self.invalidated_users: list[str] = []
         self.search_calls: list[dict[str, object]] = []
         self.latencies: dict[str, list[float]] = {}
 
@@ -69,14 +70,24 @@ def get_latency_snapshot(self):
             for mode, samples in self.latencies.items()
         }
 
+    def invalidate_profile_cache(self, user_id: str) -> None:
+        self.invalidated_users.append(user_id)
+
+
+class FakeIngestPipeline:
+    model = SimpleNamespace(model="fake-ingest")
+
+    async def run(self, **kwargs):
+        return {"classification_result": SimpleNamespace(classifications=[])}
+
 
 @pytest.fixture
 def memory_search_app(monkeypatch):
     pipeline = FakeSearchPipeline()
     monkeypatch.setattr(deps.settings, "api_keys", ["test-static-key"], raising=False)
     deps._init_error = None
     deps._pipelines_ready.set()
-    deps.set_pipelines(SimpleNamespace(), pipeline)
+    deps.set_pipelines(FakeIngestPipeline(), pipeline)
 
     app = FastAPI()
     app.add_middleware(RequestContextMiddleware)
@@ -152,3 +163,19 @@ def test_memory_search_route_accepts_code_domain(memory_search_app):
     assert payload["data"]["results"][0]["domain"] == "code"
     assert payload["data"]["results"][0]["metadata"]["target_file"] == "src/retry.py"
     assert pipeline.search_calls[0]["domains"] == ["code"]
+
+
+def test_memory_ingest_invalidates_retrieval_profile_cache(memory_search_app):
+    app, pipeline = memory_search_app
+    response = TestClient(app).post(
+        "/v1/memory/ingest",
+        headers={"Authorization": "Bearer test-static-key"},
+        json={
+            "user_query": "Remember that I work at XMem",
+            "agent_response": "Acknowledged.",
+            "user_id": "ignored-by-auth",
+        },
+    )
+
+    assert response.status_code == 200
+    assert pipeline.invalidated_users == ["Static Key User"]

tests/integration/test_retrieval_pipeline.py

@@ -332,6 +332,33 @@ def test_retrieval_plan_cache_evicts_oldest_entry(vector_store, neo4j_client):
     assert pipeline._get_cached_retrieval_plan(first_key) is None
 
 
+def test_invalidate_profile_cache_clears_user_profile_and_plan_entries(
+    vector_store, neo4j_client
+):
+    model = FakeChatModel()
+    pipeline = RetrievalPipeline(
+        model=model, vector_store=vector_store, neo4j_client=neo4j_client
+    )
+
+    pipeline._profile_catalog_cache["alice"] = (999999999.0, [], [])
+    pipeline._profile_catalog_cache["bob"] = (999999999.0, [], [])
+    pipeline._cache_retrieval_plan(
+        ("alice", "where do I work?", 5, "catalog-a"),
+        FakeLLMResponse("alice-plan"),
+    )
+    pipeline._cache_retrieval_plan(
+        ("bob", "where do I work?", 5, "catalog-b"),
+        FakeLLMResponse("bob-plan"),
+    )
+
+    pipeline.invalidate_profile_cache("alice")
+
+    assert "alice" not in pipeline._profile_catalog_cache
+    assert "bob" in pipeline._profile_catalog_cache
+    assert not any(key[0] == "alice" for key in pipeline._retrieval_plan_cache)
+    assert any(key[0] == "bob" for key in pipeline._retrieval_plan_cache)
+
+
 @pytest.mark.asyncio
 async def test_answer_from_sources_skips_tool_selection(vector_store, neo4j_client):
     model = FakeChatModel(responses=["Alice works at XMem."])