Harden retrieval and forgetting runtime paths

api/app.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
+import asyncio
 import json
+import logging
 import mimetypes
 import os
 import tempfile
 from collections.abc import Mapping
+from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
 from typing import Any
@@ -32,6 +35,7 @@
 DEFAULT_MEDIA_DIR = Path(os.getenv("MEMORY_MEDIA_DIR", config.MEDIA_STORAGE_PATH))
 _SUPPORTED_FILE_MODALITIES = {"audio", "image", "video", "multimodal"}
 _ALLOWED_MEDIA_TYPES = {"image", "audio", "video", "pdf"}
+logger = logging.getLogger(__name__)
 
 
 def _normalise_origins(origins: list[str] | None = None) -> list[str]:
@@ -323,14 +327,27 @@ def overview(self) -> dict[str, Any]:
         }
 
 
+@dataclass(frozen=True, slots=True)
+class ContradictionCheckResult:
+    status: str
+    candidates: list[ContradictionCandidate]
+    detail: str | None = None
+
+
 def _safe_contradiction_lookup(
     active_service: MemoryAPIService,
     record: SemanticMemory,
-) -> list[ContradictionCandidate]:
+) -> ContradictionCheckResult:
     try:
-        return active_service.contradiction_detector.find_potential_contradictions(record)
-    except ValueError:
-        return []
+        return ContradictionCheckResult(
+            status="completed",
+            candidates=active_service.contradiction_detector.find_potential_contradictions(record),
+        )
+    except ValueError as exc:
+        return ContradictionCheckResult(status="skipped", candidates=[], detail=str(exc))
+    except Exception as exc:
+        logger.exception("Contradiction lookup failed for semantic record %s", record.id)
+        return ContradictionCheckResult(status="error", candidates=[], detail=str(exc))
 
 
 def _serialise_contradiction_candidate(candidate: ContradictionCandidate) -> dict[str, Any]:
@@ -342,6 +359,7 @@ def _serialise_contradiction_candidate(candidate: ContradictionCandidate) -> dic
 
 def _serialise_forgetting_report(report: ForgettingReport) -> dict[str, Any]:
     return {
+        "status": "completed",
         "dry_run": report.dry_run,
         "scanned": report.scanned,
         "kept": report.kept,
@@ -388,6 +406,7 @@ def create_app(
         allow_headers=["*"],
     )
     app.state.service = None
+    app.state.forgetting_lock = asyncio.Lock()
     app.state.service_config = {
         "chroma_path": chroma_path,
         "media_root": Path(media_root) if media_root else None,
@@ -400,6 +419,21 @@ def service() -> MemoryAPIService:
             app.state.service = MemoryAPIService(**app.state.service_config)
         return app.state.service
 
+    async def run_forgetting_cycle(*, dry_run: bool) -> dict[str, Any]:
+        lock = app.state.forgetting_lock
+        # This is intentionally single-flight per app process. Multi-worker
+        # deployments still need an external coordinator if cross-process
+        # forgetting exclusivity becomes a requirement.
+        if lock.locked():
+            return {
+                "status": "already_running",
+                "dry_run": dry_run,
+            }
+
+        async with lock:
+            report = await asyncio.to_thread(service().forgetting_service.run_cycle, dry_run)
+            return _serialise_forgetting_report(report)
+
     @app.get("/health")
     async def health() -> dict[str, str]:
         return {"status": "ok"}
@@ -467,12 +501,21 @@ async def create_semantic_memory(payload: dict[str, Any]) -> dict[str, Any]:
         except Exception:
             _cleanup_owned_media(active_service, record.media_ref)
             raise
-        candidates = _safe_contradiction_lookup(active_service, record)
+        contradiction_check = _safe_contradiction_lookup(active_service, record)
         return {
             "record": _serialise_record(record),
             "potential_contradictions": [
-                _serialise_contradiction_candidate(c) for c in candidates
+                _serialise_contradiction_candidate(c)
+                for c in contradiction_check.candidates
             ],
+            "contradiction_check": {
+                "status": contradiction_check.status,
+                **(
+                    {"detail": contradiction_check.detail}
+                    if contradiction_check.detail is not None
+                    else {}
+                ),
+            },
         }
 
     @app.post("/api/memories/episodic/text")
@@ -793,13 +836,11 @@ async def by_time_range(start: datetime, end: datetime) -> dict[str, Any]:
 
     @app.post("/api/forgetting/preview")
     async def forgetting_preview() -> dict[str, Any]:
-        report = service().forgetting_service.run_cycle(dry_run=True)
-        return _serialise_forgetting_report(report)
+        return await run_forgetting_cycle(dry_run=True)
 
     @app.post("/api/forgetting/run")
     async def forgetting_run() -> dict[str, Any]:
-        report = service().forgetting_service.run_cycle(dry_run=False)
-        return _serialise_forgetting_report(report)
+        return await run_forgetting_cycle(dry_run=False)
 
     @app.post("/api/forgetting/resolve")
     async def forgetting_resolve(payload: dict[str, Any]) -> dict[str, Any]:

events/logger.py

@@ -14,6 +14,7 @@ def __init__(self, stream: TextIO | None = None):
 
     def register(self, bus: EventBus) -> None:
         bus.subscribe("memory.stored", self.on_memory_stored)
+        bus.subscribe("memory.updated", self.on_memory_updated)
         bus.subscribe("memory.retrieved", self.on_memory_retrieved)
         bus.subscribe("memory.ranked", self.on_memory_ranked)
         bus.subscribe("memory.accessed", self.on_memory_accessed)
@@ -35,6 +36,13 @@ def on_memory_retrieved(self, event: MemoryEvent) -> None:
             file=self._stream,
         )
 
+    def on_memory_updated(self, event: MemoryEvent) -> None:
+        print(
+            f"[EVENT] {event.event_type} | record_id={event.data.get('record_id')} "
+            f"| type={event.data.get('memory_type')} | update={event.data.get('update_type')}",
+            file=self._stream,
+        )
+
     def on_memory_ranked(self, event: MemoryEvent) -> None:
         results = event.data.get("results", ())
         top_score = "n/a"

forgetting/contradiction.py

@@ -98,11 +98,18 @@ def find_likely_duplicates_batch(
         pairs: list[tuple[str, str, float]] = []
         seen_pairs: set[tuple[str, str]] = set()
 
-        for record in self._store.get_all_records(include_embeddings=True):
+        for record in self._store.get_all_records(
+            include_embeddings=True,
+            include_superseded=True,
+        ):
             if record.embedding is None:
                 continue
 
-            matches = self._store.retrieve_by_vector(record.embedding, top_k=2)
+            matches = self._store.retrieve_by_vector(
+                record.embedding,
+                top_k=2,
+                include_superseded=True,
+            )
             best_other = next((match for match in matches if match[0].id != record.id), None)
             if best_other is None:
                 continue

forgetting/service.py

@@ -133,8 +133,8 @@ def __init__(
         episodic_store: BaseStore,
         procedural_store: BaseStore,
         media_store: MediaStore,
+        contradiction_detector: ContradictionDetector,
         event_bus: EventBus | None = None,
-        contradiction_detector: ContradictionDetector | None = None,
     ):
         self._stores = {
             "semantic": semantic_store,
@@ -143,7 +143,7 @@ def __init__(
         }
         self._media_store = media_store
         self._event_bus = event_bus
-        self._duplicate_detector = contradiction_detector or ContradictionDetector(semantic_store)
+        self._duplicate_detector = contradiction_detector
 
     def run_cycle(self, dry_run: bool = False) -> ForgettingReport:
         cycle_now = datetime.now(timezone.utc)
@@ -195,8 +195,15 @@ def _plan_cycle(self, cycle_now: datetime) -> list[_PlannedDecision]:
 
     def _scan_records(self) -> dict[str, MemoryRecord]:
         records: dict[str, MemoryRecord] = {}
-        for store in self._stores.values():
-            for record in store.get_all_records(include_embeddings=True):
+        for memory_type, store in self._stores.items():
+            if memory_type == "semantic":
+                scanned = store.get_all_records(  # type: ignore[call-arg]
+                    include_embeddings=True,
+                    include_superseded=True,
+                )
+            else:
+                scanned = store.get_all_records(include_embeddings=True)
+            for record in scanned:
                 records[record.id] = record
         return records
 

runtime.py

@@ -35,6 +35,7 @@ def __init__(self, bus: EventBus, *, max_events: int = 200):
         self._events: deque[dict[str, Any]] = deque(maxlen=max_events)
         for event_type in (
             "memory.stored",
+            "memory.updated",
             "memory.retrieved",
             "memory.ranked",
             "memory.accessed",

stores/procedural_store.py

@@ -145,6 +145,19 @@ def record_outcome(self, record_id: str, success: bool) -> None:
             return
         record.record_outcome(success)
         self.replace(record)
+        self._emit_event(
+            "memory.updated",
+            {
+                "record_id": record.id,
+                "memory_type": record.memory_type,
+                "update_type": "procedural_outcome",
+                "success": success,
+                "success_count": record.success_count,
+                "failure_count": record.failure_count,
+                "total_outcomes": record.total_outcomes,
+                "wilson_score": record.wilson_score,
+            },
+        )
 
     def get_best_procedures(
         self,

stores/semantic_store.py

@@ -78,21 +78,49 @@ def get_by_id(self, record_id: str) -> SemanticMemory | None:
             return None
         return self._from_result(result, 0)
 
-    def get_all_records(self, include_embeddings: bool = False) -> list[SemanticMemory]:
-        return self._all_records(include_embeddings=include_embeddings)
+    def get_all_records(
+        self,
+        include_embeddings: bool = False,
+        *,
+        include_superseded: bool = False,
+    ) -> list[SemanticMemory]:
+        return self._all_records(
+            include_embeddings=include_embeddings,
+            include_superseded=include_superseded,
+        )
 
-    def retrieve(self, query: str, top_k: int = 5) -> list[tuple[SemanticMemory, float]]:
+    def retrieve(
+        self,
+        query: str,
+        top_k: int = 5,
+        *,
+        include_superseded: bool = False,
+    ) -> list[tuple[SemanticMemory, float]]:
         query_embedding = self._embedder.embed_query(query)
-        return self.retrieve_by_vector(query_embedding, top_k=top_k)
+        return self.retrieve_by_vector(
+            query_embedding,
+            top_k=top_k,
+            include_superseded=include_superseded,
+        )
 
     def retrieve_by_vector(
         self,
         vector: list[float],
         top_k: int = 5,
+        *,
+        include_superseded: bool = False,
     ) -> list[tuple[SemanticMemory, float]]:
+        n_results = max(1, top_k)
+        if not include_superseded:
+            # We need enough raw candidates to still return `top_k` active
+            # records after filtering superseded ones, and Chroma cannot apply
+            # that predicate natively for us. This makes default semantic recall
+            # O(N) in collection size, which is acceptable for the current
+            # scale but should be revisited if the store grows substantially.
+            n_results = max(1, self._collection.count())
         result = self._collection.query(
             query_embeddings=[vector],
-            n_results=top_k,
+            n_results=n_results,
             include=["embeddings", "documents", "metadatas", "distances"],
         )
         if not result["ids"] or not result["ids"][0]:
@@ -101,9 +129,13 @@ def retrieve_by_vector(
         pairs = []
         for i in range(len(result["ids"][0])):
             record = self._from_query_result(result, i)
+            if not include_superseded and record.superseded_by:
+                continue
             # ChromaDB cosine distance = 1 - similarity
             similarity = 1.0 - result["distances"][0][i]
             pairs.append((record, similarity))
+            if len(pairs) >= top_k:
+                break
         return pairs
 
     # ── serialisation helpers ──────────────────────────────────────────────
@@ -156,15 +188,23 @@ def _to_metadata(self, record: SemanticMemory) -> dict:
             "text_description": record.text_description or "",
         }
 
-    def _all_records(self, *, include_embeddings: bool) -> list[SemanticMemory]:
+    def _all_records(
+        self,
+        *,
+        include_embeddings: bool,
+        include_superseded: bool,
+    ) -> list[SemanticMemory]:
         include = ["documents", "metadatas"]
         if include_embeddings:
             include.insert(0, "embeddings")
 
         result = self._collection.get(include=include)
         if not result["ids"]:
             return []
-        return [self._from_result(result, index) for index in range(len(result["ids"]))]
+        records = [self._from_result(result, index) for index in range(len(result["ids"]))]
+        if include_superseded:
+            return records
+        return [record for record in records if not record.superseded_by]
 
     def _build_record(self, doc: str, id: str, embedding, meta: dict) -> SemanticMemory:
         last_accessed = meta.get("last_accessed_at", "")

tests/test_api_runtime_edges.py

@@ -1,4 +1,4 @@
-"""Edge-case API tests around the shared runtime refactor."""
+"""Edge-case API tests around the runtime hardening path."""
 
 import os
 import sys
@@ -45,10 +45,14 @@ async def test_semantic_write_swallows_valueerror_from_contradiction_lookup():
 
     assert response.status_code == 200
     assert response.json()["potential_contradictions"] == []
+    assert response.json()["contradiction_check"] == {
+        "status": "skipped",
+        "detail": "synthetic contradiction miss",
+    }
 
 
 @pytest.mark.anyio
-async def test_semantic_write_surfaces_programming_error_from_contradiction_lookup():
+async def test_semantic_write_reports_programming_error_from_contradiction_lookup():
     async with make_client() as client:
         await client.get("/api/overview")
         client.app.state.service.contradiction_detector.find_potential_contradictions = (
@@ -60,7 +64,12 @@ async def test_semantic_write_surfaces_programming_error_from_contradiction_look
             json={"content": "TypeError contradiction lookup should not be suppressed"},
         )
 
-    assert response.status_code == 500
+    assert response.status_code == 200
+    assert response.json()["potential_contradictions"] == []
+    assert response.json()["contradiction_check"] == {
+        "status": "error",
+        "detail": "synthetic programming error",
+    }
 
 
 @pytest.mark.anyio