diff --git a/api/app.py b/api/app.py index 2bc82bb..5e41082 100644 --- a/api/app.py +++ b/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]: diff --git a/events/logger.py b/events/logger.py index 5e36411..615087b 100644 --- a/events/logger.py +++ b/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" diff --git a/forgetting/contradiction.py b/forgetting/contradiction.py index 0cd5283..fcf2559 100644 --- a/forgetting/contradiction.py +++ b/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 diff --git a/forgetting/service.py b/forgetting/service.py index dde94d4..e0b0593 100644 --- a/forgetting/service.py +++ b/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 diff --git a/runtime.py b/runtime.py index a78001c..ffbc7fb 100644 --- a/runtime.py +++ b/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", diff --git a/stores/procedural_store.py b/stores/procedural_store.py index 657a36a..3f49189 100644 --- a/stores/procedural_store.py +++ b/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, diff --git a/stores/semantic_store.py b/stores/semantic_store.py index f925f35..c3a5734 100644 --- a/stores/semantic_store.py +++ b/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,7 +188,12 @@ 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") @@ -164,7 +201,10 @@ def _all_records(self, *, include_embeddings: bool) -> list[SemanticMemory]: 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", "") diff --git a/tests/test_api_runtime_edges.py b/tests/test_api_runtime_edges.py index 5e4e08e..29a584d 100644 --- a/tests/test_api_runtime_edges.py +++ b/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 diff --git a/tests/test_forgetting_integration.py b/tests/test_forgetting_integration.py index 9468781..6992905 100644 --- a/tests/test_forgetting_integration.py +++ b/tests/test_forgetting_integration.py @@ -1,9 +1,11 @@ """Verify forgetting endpoints, contradiction lookup, and full cycle integration.""" +import asyncio import os import shutil import sys import tempfile +import threading import httpx import pytest @@ -39,6 +41,7 @@ async def test_semantic_store_returns_potential_contradictions(): assert r1.status_code == 200 body1 = r1.json() assert "potential_contradictions" in body1 + assert body1["contradiction_check"]["status"] == "completed" assert isinstance(body1["potential_contradictions"], list) r2 = await client.post( @@ -48,6 +51,7 @@ async def test_semantic_store_returns_potential_contradictions(): assert r2.status_code == 200 body2 = r2.json() assert "potential_contradictions" in body2 + assert body2["contradiction_check"]["status"] == "completed" @pytest.mark.anyio @@ -60,6 +64,7 @@ async def test_forgetting_preview_returns_stable_report(): assert response.status_code == 200 report = response.json() + assert report["status"] == "completed" assert report["dry_run"] is True assert report["scanned"] == 2 assert isinstance(report["decisions"], list) @@ -94,6 +99,7 @@ async def test_forgetting_run_executes_cycle(): assert response.status_code == 200 report = response.json() + assert report["status"] == "completed" assert report["dry_run"] is False assert report["scanned"] >= 1 assert isinstance(report["decisions"], list) @@ -184,3 +190,67 @@ async def test_full_cycle_supersede_then_prune(): overview = await client.get("/api/overview") assert overview.json()["semantic_count"] == 1 + + +@pytest.mark.anyio +async def test_retrieval_hides_superseded_semantic_records_by_default(): + async with make_client() as client: + old_response = await client.post( + "/api/memories/semantic", + json={"content": "Deployment region is us-east-1"}, + ) + old_id = old_response.json()["record"]["id"] + + new_response = await client.post( + "/api/memories/semantic", + json={"content": "Deployment region is eu-west-1"}, + ) + new_id = new_response.json()["record"]["id"] + + resolve = await client.post( + "/api/forgetting/resolve", + json={"keep_id": new_id, "supersede_id": old_id}, + ) + assert resolve.status_code == 200 + + query = await client.post( + "/api/retrieval/query", + json={"query": "deployment region", "top_k": 5, "memory_types": ["semantic"]}, + ) + + assert query.status_code == 200 + ids = {item["record"]["id"] for item in query.json()["results"]} + assert new_id in ids + assert old_id not in ids + + +@pytest.mark.anyio +async def test_overlapping_forgetting_requests_return_already_running(): + async with make_client() as client: + await client.post("/api/memories/semantic", json={"content": "Overlap witness"}) + await client.get("/api/overview") + + entered = threading.Event() + release = threading.Event() + original = client.app.state.service.forgetting_service.run_cycle + + def blocking_run_cycle(dry_run: bool = False): + entered.set() + release.wait() + return original(dry_run=dry_run) + + client.app.state.service.forgetting_service.run_cycle = blocking_run_cycle + + first_task = asyncio.create_task(client.post("/api/forgetting/run")) + await asyncio.to_thread(entered.wait) + second_response = await client.post("/api/forgetting/run") + release.set() + first_response = await first_task + + assert first_response.status_code == 200 + assert first_response.json()["status"] == "completed" + assert second_response.status_code == 200 + assert second_response.json() == { + "status": "already_running", + "dry_run": False, + } diff --git a/tests/test_forgetting_service.py b/tests/test_forgetting_service.py index 8dc682d..1106844 100644 --- a/tests/test_forgetting_service.py +++ b/tests/test_forgetting_service.py @@ -42,8 +42,15 @@ def store(self, record): def get_by_id(self, record_id): return self._records.get(record_id) - def get_all_records(self, include_embeddings: bool = False): + def get_all_records( + self, + include_embeddings: bool = False, + *, + include_superseded: bool = False, + ): records = list(self._records.values()) + if not include_superseded: + records = [record for record in records if not getattr(record, "superseded_by", None)] if include_embeddings: return records return [replace(record, embedding=None) for record in records] @@ -66,6 +73,24 @@ def replace(self, record) -> None: self._records[record.id] = record +class TrackingSemanticStore(FakeStore): + def __init__(self, records=None): + super().__init__(records) + self.last_include_superseded = None + + def get_all_records( + self, + include_embeddings: bool = False, + *, + include_superseded: bool = False, + ): + self.last_include_superseded = include_superseded + return super().get_all_records( + include_embeddings=include_embeddings, + include_superseded=include_superseded, + ) + + class EventRecorder: def __init__(self, bus: EventBus, *event_types: str): self.events = [] @@ -73,6 +98,11 @@ def __init__(self, bus: EventBus, *event_types: str): bus.subscribe(event_type, self.events.append) +class _NoopDuplicateDetector: + def find_likely_duplicates_batch(self, threshold=0.95): + return [] + + def _media_store(): root = tempfile.mkdtemp(prefix="forgetting_media_") return MediaStore(root), root @@ -93,7 +123,7 @@ def _make_service( procedural_store=FakeStore(procedural_records), media_store=media_store, event_bus=event_bus, - contradiction_detector=FakeDuplicateDetector(duplicate_pairs), + contradiction_detector=FakeDuplicateDetector(duplicate_pairs) if duplicate_pairs is not None else _NoopDuplicateDetector(), ) return service, media_root @@ -102,6 +132,31 @@ def _ids_for(report): return {decision.record_id: decision for decision in report.decisions} +def test_scan_records_includes_superseded_semantic_rows(): + superseded = SemanticMemory( + id="semantic-old", + content="Old fact", + superseded_by="semantic-new", + ) + active = SemanticMemory( + id="semantic-new", + content="New fact", + ) + semantic_store = TrackingSemanticStore([superseded, active]) + service = ForgettingService( + semantic_store=semantic_store, + episodic_store=FakeStore(), + procedural_store=FakeStore(), + media_store=MediaStore(tempfile.mkdtemp(prefix="forgetting_media_scan_")), + contradiction_detector=_NoopDuplicateDetector(), + ) + + records = service._scan_records() + + assert semantic_store.last_include_superseded is True + assert {"semantic-old", "semantic-new"} <= set(records) + + def test_dry_run_resolves_duplicate_clusters_component_wise(monkeypatch): now = datetime(2026, 3, 1, 12, 0, tzinfo=timezone.utc) first = SemanticMemory( diff --git a/tests/test_logger.py b/tests/test_logger.py index 77dd1fe..530c333 100644 --- a/tests/test_logger.py +++ b/tests/test_logger.py @@ -42,6 +42,14 @@ def test_console_logger_formats_events(): ], }, ) + bus.emit( + "memory.updated", + { + "record_id": "abc-123", + "memory_type": "procedural", + "update_type": "procedural_outcome", + }, + ) bus.emit( "memory.accessed", { @@ -61,7 +69,10 @@ def test_console_logger_formats_events(): assert output[2] == ( "[EVENT] memory.ranked | query='Who created Python?' | top_score=0.8500 | results=1" ) - assert output[3] == "[EVENT] memory.accessed | record_id=abc-123 | count=1" + assert output[3] == ( + "[EVENT] memory.updated | record_id=abc-123 | type=procedural | update=procedural_outcome" + ) + assert output[4] == "[EVENT] memory.accessed | record_id=abc-123 | count=1" print(" PASS ConsoleLogger formats event lines consistently") diff --git a/tests/test_procedural_store.py b/tests/test_procedural_store.py index cf1515d..09807c9 100644 --- a/tests/test_procedural_store.py +++ b/tests/test_procedural_store.py @@ -227,6 +227,30 @@ def test_record_outcome_rewrites_metadata_without_reembedding(): print(" PASS record_outcome rewrites metadata only and preserves embedding/document data") +def test_record_outcome_emits_memory_updated_event(): + bus = EventBus() + recorder = EventRecorder(bus, "memory.updated") + store, _ = fresh_setup(event_bus=bus) + record = ProceduralMemory( + content="Deploy to Lambda", + steps=["Package", "Deploy"], + ) + store.store(record) + + store.record_outcome(record.id, True) + + assert len(recorder.events) == 1 + event = recorder.events[0] + assert event.data["record_id"] == record.id + assert event.data["memory_type"] == "procedural" + assert event.data["update_type"] == "procedural_outcome" + assert event.data["success"] is True + assert event.data["success_count"] == 1 + assert event.data["failure_count"] == 0 + assert event.data["total_outcomes"] == 1 + print(" PASS record_outcome emits memory.updated with outcome counters") + + def test_record_outcome_missing_record_is_a_no_op(): store, _ = fresh_setup() @@ -358,6 +382,7 @@ def test_get_best_procedures_returns_empty_for_empty_store(): test_store_emits_memory_stored_with_procedural_fields() test_media_backed_procedure_round_trip_uses_combined_embedding_path() test_record_outcome_rewrites_metadata_without_reembedding() + test_record_outcome_emits_memory_updated_event() test_record_outcome_missing_record_is_a_no_op() test_get_all_records_returns_records_without_embeddings_by_default() test_procedural_delete_is_idempotent_and_replace_preserves_embedding() diff --git a/tests/test_semantic_store.py b/tests/test_semantic_store.py index eb59715..ded7d91 100644 --- a/tests/test_semantic_store.py +++ b/tests/test_semantic_store.py @@ -223,6 +223,44 @@ def fail_add(*args, **kwargs): print(" PASS failed semantic writes clean up owned media") +def test_retrieve_filters_superseded_records_by_default(): + embedder = DeterministicMultimodalEmbedder(dimensions=4) + store, _ = fresh_setup(embedder=embedder) + active = SemanticMemory(content="Current deployment region is eu-west-1") + superseded = SemanticMemory( + content="Old deployment region is us-east-1", + superseded_by=active.id, + ) + store.store(active) + store.store(superseded) + + records = store.get_all_records() + results = store.retrieve("deployment region", top_k=5) + + assert {record.id for record in records} == {active.id} + assert {record.id for record, _ in results} == {active.id} + print(" PASS semantic recall hides superseded records by default") + + +def test_retrieve_can_include_superseded_records_for_admin_flows(): + embedder = DeterministicMultimodalEmbedder(dimensions=4) + store, _ = fresh_setup(embedder=embedder) + active = SemanticMemory(content="Current deployment region is eu-west-1") + superseded = SemanticMemory( + content="Old deployment region is us-east-1", + superseded_by=active.id, + ) + store.store(active) + store.store(superseded) + + records = store.get_all_records(include_superseded=True) + results = store.retrieve("deployment region", top_k=5, include_superseded=True) + + assert {record.id for record in records} == {active.id, superseded.id} + assert {record.id for record, _ in results} == {active.id, superseded.id} + print(" PASS semantic admin recall can include superseded records") + + def test_get_all_records_returns_full_collection_without_embeddings_by_default(): store, _ = fresh_setup() first = SemanticMemory(content="First fact", importance=0.2)