feat: Per-user ingestion coordination to prevent cross-source race conditions

debug_test.py

@@ -0,0 +1,36 @@
+import asyncio
+from fastapi.testclient import TestClient
+from unittest.mock import AsyncMock, patch
+
+from src.api.app import create_app
+
+app = create_app()
+client = TestClient(app)
+
+with patch("src.api.routes.memory.require_api_key", return_value={"username": "test_user"}):
+    from src.api.dependencies import require_api_key, enforce_rate_limit, require_ready
+    app.dependency_overrides[require_api_key] = lambda: {"username": "test_user"}
+    app.dependency_overrides[enforce_rate_limit] = lambda: True
+    app.dependency_overrides[require_ready] = lambda: True
+
+    payload = {
+        "items": [
+            {
+                "user_query": "Hello world",
+                "agent_response": "Hi there",
+                "user_id": "test_user_1",
+            }
+        ]
+    }
+
+    try:
+        response = client.post(
+            "/v1/memory/batch-ingest",
+            json=payload,
+            headers={"Authorization": "Bearer test-key"}
+        )
+        print("Status code:", response.status_code)
+        import json
+        print(json.dumps(response.json(), indent=2))
+    except Exception as e:
+        print("Exception:", e)

server.py

@@ -50,6 +50,7 @@
 
 from src.pipelines.ingest import IngestPipeline
 from src.pipelines.retrieval import RetrievalPipeline
+from src.api.ingestion_coordinator import UserIngestionCoordinator
 
 
 # ═══════════════════════════════════════════════════════════════════
@@ -82,6 +83,7 @@ def emit(self, record: logging.LogRecord) -> None:
 _pipelines_ready = asyncio.Event()
 _init_error: str | None = None
 SKIP_PIPELINES = os.getenv("XMEM_SKIP_PIPELINES", "").lower() in {"1", "true", "yes"}
+_user_coordinator = UserIngestionCoordinator()
 
 
 def _init_pipelines_sync() -> None:
@@ -315,14 +317,15 @@ async def v1_ingest_memory(req: IngestRequest):
         )
 
     try:
-        result = await ingest_pipeline.run(
-            user_query=req.user_query,
-            agent_response=req.agent_response or "Acknowledged.",
-            user_id=req.user_id,
-            session_datetime=req.session_datetime,
-            image_url=req.image_url,
-            effort_level=req.effort_level,
-        )
+        async with _user_coordinator.acquire(req.user_id):
+            result = await ingest_pipeline.run(
+                user_query=req.user_query,
+                agent_response=req.agent_response or "Acknowledged.",
+                user_id=req.user_id,
+                session_datetime=req.session_datetime,
+                image_url=req.image_url,
+                effort_level=req.effort_level,
+            )
 
         data = {
             "model": _get_model_name(ingest_pipeline.model),
@@ -368,11 +371,12 @@ async def api_ingest(req: IngestRequest):
         lg.addHandler(capture)
 
     try:
-        result = await ingest_pipeline.run(
-            user_query=req.user_query,
-            agent_response=req.agent_response or "Acknowledged.",
-            user_id=req.user_id,
-        )
+        async with _user_coordinator.acquire(req.user_id):
+            result = await ingest_pipeline.run(
+                user_query=req.user_query,
+                agent_response=req.agent_response or "Acknowledged.",
+                user_id=req.user_id,
+            )
 
         # Build structured response
         response: Dict[str, Any] = {

src/agents/judge.py

@@ -19,6 +19,7 @@
 
 import asyncio
 import json
+from difflib import SequenceMatcher
 from typing import Any, Callable, Dict, List, Optional
 
 from langchain_core.language_models import BaseChatModel
@@ -162,7 +163,9 @@ def __init__(
     # Public entry point
     # ------------------------------------------------------------------
 
-    async def arun(self, state: Dict[str, Any]) -> JudgeResult:
+    async def arun(
+        self, state: Dict[str, Any], pending_ops: Optional[List[Operation]] = None
+    ) -> JudgeResult:
         domain_str = state.get("domain", "")
         try:
             domain = JudgeDomain(domain_str)
@@ -186,6 +189,7 @@ async def arun(self, state: Dict[str, Any]) -> JudgeResult:
             new_items=new_items,
             user_id=user_id,
             domain=domain,
+            pending_ops=pending_ops,
         )
 
         if domain == JudgeDomain.SUMMARY and not _has_summary_judge_candidates(matches_per_item):
@@ -218,7 +222,9 @@ async def arun(self, state: Dict[str, Any]) -> JudgeResult:
 
         return result
 
-    async def arun_deterministic(self, state: Dict[str, Any]) -> JudgeResult:
+    async def arun_deterministic(
+        self, state: Dict[str, Any], pending_ops: Optional[List[Operation]] = None
+    ) -> JudgeResult:
         """Build operations without an LLM for structured domains.
 
         Profile and temporal extraction already returns normalized structured
@@ -238,15 +244,15 @@ async def arun_deterministic(self, state: Dict[str, Any]) -> JudgeResult:
             return JudgeResult()
 
         if domain == JudgeDomain.PROFILE:
-            result = await self._deterministic_profile(new_items, user_id)
+            result = await self._deterministic_profile(new_items, user_id, pending_ops=pending_ops)
         elif domain == JudgeDomain.TEMPORAL:
-            result = await self._deterministic_temporal(new_items, user_id)
+            result = await self._deterministic_temporal(new_items, user_id, pending_ops=pending_ops)
         else:
             self.logger.warning(
                 "Deterministic judge unsupported for %s; falling back to LLM judge.",
                 domain.value,
             )
-            return await self.arun(state)
+            return await self.arun(state, pending_ops=pending_ops)
 
         self._log_result(domain, result)
         return result
@@ -275,11 +281,16 @@ async def _fetch_similar(
         new_items: list,
         user_id: str,
         domain: JudgeDomain,
+        pending_ops: Optional[List[Operation]] = None,
     ) -> Dict[str, List[SearchResult]]:
         if domain == JudgeDomain.TEMPORAL:
-            return await self._fetch_similar_temporal(items_strings, new_items, user_id)
+            return await self._fetch_similar_temporal(
+                items_strings, new_items, user_id, pending_ops=pending_ops
+            )
         else:
-            return await self._fetch_similar_vector(items_strings, new_items, user_id, domain)
+            return await self._fetch_similar_vector(
+                items_strings, new_items, user_id, domain, pending_ops=pending_ops
+            )
 
     # -- Profile / Summary: Pinecone (vector store) -----------------------
 
@@ -289,6 +300,7 @@ async def _fetch_similar_vector(
         new_items: list,
         user_id: str,
         domain: JudgeDomain,
+        pending_ops: Optional[List[Operation]] = None,
     ) -> Dict[str, List[SearchResult]]:
         if not self.vector_store:
             self.logger.debug("No vector store attached — skipping similarity search.")
@@ -297,7 +309,7 @@ async def _fetch_similar_vector(
         # Profile domain: use deterministic metadata lookup
         if domain == JudgeDomain.PROFILE:
             return await self._fetch_similar_profile_metadata(
-                items_strings, new_items, user_id,
+                items_strings, new_items, user_id, pending_ops=pending_ops
             )
 
         # Summary / other: parallel semantic search across all items
@@ -320,7 +332,43 @@ async def _search_one(item_str: str) -> tuple[str, List[SearchResult]]:
                 return item_str, []
 
         pairs = await asyncio.gather(*(_search_one(s) for s in items_strings))
-        return dict(pairs)
+        matches_per_item = dict(pairs)
+
+        if pending_ops:
+            for item_str in items_strings:
+                matches = matches_per_item.get(item_str, [])
+
+                # Apply deletes first
+                deletes = {op.embedding_id for op in pending_ops if op.type == OperationType.DELETE and op.embedding_id}
+                if deletes:
+                    matches = [m for m in matches if m.id not in deletes]
+
+                # Apply adds/updates
+                for op in pending_ops:
+                    if op.type in (OperationType.ADD, OperationType.UPDATE):
+                        ratio = SequenceMatcher(None, _norm_text(item_str), _norm_text(op.content)).ratio()
+                        if ratio > 0.5:
+                            simulated = SearchResult(
+                                id=op.embedding_id or f"pending_{domain.value}_{hash(op.content)}",
+                                content=op.content,
+                                score=ratio,
+                                metadata={
+                                    "domain": domain.value,
+                                    "user_id": user_id,
+                                }
+                            )
+                            # Replace if same ID already exists in matches, otherwise prepend
+                            existing_idx = next((i for i, m in enumerate(matches) if m.id == simulated.id), None)
+                            if existing_idx is not None:
+                                matches[existing_idx] = simulated
+                            else:
+                                matches.insert(0, simulated)
+
+                # Sort matches by score descending
+                matches = sorted(matches, key=lambda x: x.score or 0.0, reverse=True)[:self.top_k]
+                matches_per_item[item_str] = matches
+
+        return matches_per_item
 
     # -- Profile: deterministic metadata lookup ----------------------------
 
@@ -329,6 +377,7 @@ async def _fetch_similar_profile_metadata(
         items_strings: List[str],
         new_items: list,
         user_id: str,
+        pending_ops: Optional[List[Operation]] = None,
     ) -> Dict[str, List[SearchResult]]:
         """Fetch existing profile records by exact topic_subtopic match (parallel).
 
@@ -377,7 +426,36 @@ async def _lookup_one(idx: int, item_str: str) -> tuple[str, List[SearchResult]]
         pairs = await asyncio.gather(
             *(_lookup_one(i, s) for i, s in enumerate(items_strings))
         )
-        return dict(pairs)
+        matches_per_item = dict(pairs)
+
+        if pending_ops:
+            for idx, item_str in enumerate(items_strings):
+                item = new_items[idx] if idx < len(new_items) else {}
+                meta_key = _build_profile_metadata_key(item)
+                if not meta_key:
+                    continue
+
+                for op in pending_ops:
+                    op_meta_key = _profile_meta_key_from_content(op.content)
+                    if op_meta_key == meta_key:
+                        if op.type in (OperationType.ADD, OperationType.UPDATE):
+                            _, _, memo = _parse_profile_content(op.content)
+                            simulated = SearchResult(
+                                id=op.embedding_id or f"pending_profile_{meta_key}",
+                                content=op.content,
+                                score=1.0,
+                                metadata={
+                                    "main_content": meta_key,
+                                    "subcontent": memo,
+                                    "domain": "profile",
+                                    "user_id": user_id,
+                                }
+                            )
+                            matches_per_item[item_str] = [simulated]
+                        elif op.type == OperationType.DELETE:
+                            matches_per_item[item_str] = []
+
+        return matches_per_item
 
     # -- Semantic search fallback (summary domain) -------------------------
 
@@ -406,6 +484,7 @@ async def _fetch_similar_temporal(
         items_strings: List[str],
         new_items: list,
         user_id: str,
+        pending_ops: Optional[List[Operation]] = None,
     ) -> Dict[str, List[SearchResult]]:
         if not self.graph_event_search:
             self.logger.debug("No graph_event_search provided — skipping Neo4j lookup.")
@@ -434,7 +513,36 @@ async def _lookup_one(idx: int, item_str: str) -> tuple[str, List[SearchResult]]
         pairs = await asyncio.gather(
             *(_lookup_one(i, s) for i, s in enumerate(items_strings))
         )
-        return dict(pairs)
+        matches_per_item = dict(pairs)
+
+        if pending_ops:
+            for idx, item_str in enumerate(items_strings):
+                event = new_items[idx] if idx < len(new_items) else {}
+                event_name = event.get("event_name", "") if isinstance(event, dict) else ""
+                if not event_name:
+                    continue
+                norm_event_name = _norm_text(event_name)
+
+                for op in pending_ops:
+                    fields = _temporal_fields_from_content(op.content)
+                    op_event_name = fields.get("event_name", "")
+                    if _norm_text(op_event_name) == norm_event_name:
+                        if op.type in (OperationType.ADD, OperationType.UPDATE):
+                            simulated = SearchResult(
+                                id=op.embedding_id or f"pending_temporal_{norm_event_name}",
+                                content=op.content,
+                                score=1.0,
+                                metadata={
+                                    **fields,
+                                    "domain": "temporal",
+                                    "user_id": user_id,
+                                }
+                            )
+                            matches_per_item[item_str] = [simulated]
+                        elif op.type == OperationType.DELETE:
+                            matches_per_item[item_str] = []
+
+        return matches_per_item
 
     # -- Deterministic operation builders ---------------------------------
 
@@ -687,3 +795,20 @@ def _temporal_fields_from_match(match: SearchResult) -> Dict[str, str]:
 def _same_temporal_event(incoming: Dict[str, str], existing: Dict[str, str]) -> bool:
     keys = ["date", "event_name", "desc", "year", "time", "date_expression"]
     return all(_norm_text(incoming.get(key)) == _norm_text(existing.get(key)) for key in keys)
+
+
+def _parse_profile_content(content: str) -> tuple[str, str, str]:
+    if " = " not in content:
+        return "", "", content
+    left, memo = content.split(" = ", 1)
+    if " / " not in left:
+        return left.strip(), "", memo.strip()
+    topic, sub_topic = left.split(" / ", 1)
+    return topic.strip(), sub_topic.strip(), memo.strip()
+
+
+def _profile_meta_key_from_content(content: str) -> str:
+    topic, sub_topic, _ = _parse_profile_content(content)
+    if not topic or not sub_topic:
+        return ""
+    return f"{topic}_{sub_topic}".replace(" ", "_").lower()