"""Issue E — /invalidate + /supersede for active entities and memories.""" import pytest from fastapi.testclient import TestClient from atocore.engineering.service import ( create_entity, get_entity, get_relationships, init_engineering_schema, invalidate_active_entity, supersede_entity, ) from atocore.main import app from atocore.memory.service import create_memory, get_memories def _get_memory(memory_id): for status in ("active", "candidate", "invalid", "superseded"): for m in get_memories(status=status, active_only=False, limit=5000): if m.id == memory_id: return m return None from atocore.models.database import init_db @pytest.fixture def env(tmp_data_dir, tmp_path, monkeypatch): registry_path = tmp_path / "test-registry.json" registry_path.write_text('{"projects": []}', encoding="utf-8") monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path)) from atocore import config config.settings = config.Settings() init_db() init_engineering_schema() yield tmp_data_dir def test_invalidate_active_entity_transitions_to_invalid(env): e = create_entity(entity_type="component", name="tower-to-kill") ok, code = invalidate_active_entity(e.id, reason="duplicate") assert ok is True assert code == "invalidated" assert get_entity(e.id).status == "invalid" def test_invalidate_on_candidate_is_409(env): e = create_entity(entity_type="component", name="still-candidate", status="candidate") ok, code = invalidate_active_entity(e.id) assert ok is False assert code == "not_active" def test_invalidate_is_idempotent_on_invalid(env): e = create_entity(entity_type="component", name="already-gone") invalidate_active_entity(e.id) ok, code = invalidate_active_entity(e.id) assert ok is True assert code == "already_invalid" def test_supersede_creates_relationship(env): old = create_entity(entity_type="component", name="old-tower") new = create_entity(entity_type="component", name="new-tower") ok = supersede_entity(old.id, superseded_by=new.id, note="replaced") assert ok is True assert get_entity(old.id).status == "superseded" rels = get_relationships(new.id, direction="outgoing") assert any( r.relationship_type == "supersedes" and r.target_entity_id == old.id for r in rels ), "supersedes relationship must be auto-created" def test_supersede_rejects_self(): # no db needed — validation is pre-write. Using a fresh env anyway. pass # covered below via API test def test_api_invalidate_entity(env): e = create_entity(entity_type="component", name="api-kill") client = TestClient(app) r = client.post( f"/entities/{e.id}/invalidate", json={"reason": "test cleanup"}, ) assert r.status_code == 200 assert r.json()["status"] == "invalidated" def test_api_invalidate_entity_idempotent(env): e = create_entity(entity_type="component", name="api-kill-2") client = TestClient(app) client.post(f"/entities/{e.id}/invalidate", json={"reason": "first"}) r = client.post(f"/entities/{e.id}/invalidate", json={"reason": "second"}) assert r.status_code == 200 assert r.json()["status"] == "already_invalid" def test_api_invalidate_unknown_entity_is_404(env): client = TestClient(app) r = client.post( "/entities/nonexistent-id/invalidate", json={"reason": "missing"}, ) assert r.status_code == 404 def test_api_invalidate_candidate_entity_is_409(env): e = create_entity(entity_type="component", name="cand", status="candidate") client = TestClient(app) r = client.post(f"/entities/{e.id}/invalidate", json={"reason": "x"}) assert r.status_code == 409 def test_api_supersede_entity(env): old = create_entity(entity_type="component", name="api-old-tower") new = create_entity(entity_type="component", name="api-new-tower") client = TestClient(app) r = client.post( f"/entities/{old.id}/supersede", json={"superseded_by": new.id, "reason": "dedup"}, ) assert r.status_code == 200 body = r.json() assert body["status"] == "superseded" assert body["superseded_by"] == new.id assert get_entity(old.id).status == "superseded" def test_api_supersede_self_is_400(env): e = create_entity(entity_type="component", name="self-sup") client = TestClient(app) r = client.post( f"/entities/{e.id}/supersede", json={"superseded_by": e.id, "reason": "oops"}, ) assert r.status_code == 400 def test_api_supersede_missing_replacement_is_400(env): old = create_entity(entity_type="component", name="orphan-old") client = TestClient(app) r = client.post( f"/entities/{old.id}/supersede", json={"superseded_by": "does-not-exist", "reason": "missing"}, ) assert r.status_code == 400 def test_api_invalidate_memory(env): m = create_memory( memory_type="project", content="memory to retract", project="p05", ) client = TestClient(app) r = client.post( f"/memory/{m.id}/invalidate", json={"reason": "outdated"}, ) assert r.status_code == 200 assert r.json()["status"] == "invalidated" assert _get_memory(m.id).status == "invalid" def test_api_supersede_memory(env): m = create_memory( memory_type="project", content="memory to supersede", project="p05", ) client = TestClient(app) r = client.post( f"/memory/{m.id}/supersede", json={"reason": "replaced by newer fact"}, ) assert r.status_code == 200 assert r.json()["status"] == "superseded" assert _get_memory(m.id).status == "superseded" def test_v1_aliases_present(env): client = TestClient(app) spec = client.get("/openapi.json").json() paths = spec["paths"] for p in ( "/v1/entities/{entity_id}/invalidate", "/v1/entities/{entity_id}/supersede", "/v1/memory/{memory_id}/invalidate", "/v1/memory/{memory_id}/supersede", ): assert p in paths, f"{p} missing" # --------------------------------------------------------------------------- # Wave 1 (2026-04-29) — invalidation route used to do # `_get_memories(status='active', limit=1)` and look for the target id # inside that single highest-confidence row, so any active memory # outside slot 0 fell through as 404. Direct id lookup fixes it. # --------------------------------------------------------------------------- def test_api_invalidate_finds_active_memory_outside_top_one(env): """An active memory not at the top of the confidence sort must still be invalidatable via POST /memory/{id}/invalidate.""" high = create_memory( memory_type="knowledge", content="high-confidence top row", confidence=0.99, ) low = create_memory( memory_type="knowledge", content="lower-confidence target", confidence=0.55, ) client = TestClient(app) r = client.post(f"/memory/{low.id}/invalidate", json={"reason": "wave1 regression"}) assert r.status_code == 200, r.text assert r.json()["status"] == "invalidated" # And confirm the high-confidence row is untouched assert _get_memory(high.id).status == "active" assert _get_memory(low.id).status == "invalid" def test_api_invalidate_already_invalid_is_idempotent(env): m = create_memory(memory_type="knowledge", content="already invalid") client = TestClient(app) r1 = client.post(f"/memory/{m.id}/invalidate", json={"reason": "first"}) assert r1.status_code == 200 r2 = client.post(f"/memory/{m.id}/invalidate", json={"reason": "again"}) assert r2.status_code == 200 assert r2.json()["status"] == "already_invalid" def test_api_invalidate_candidate_returns_409(env): m = create_memory( memory_type="knowledge", content="candidate route", status="candidate" ) client = TestClient(app) r = client.post(f"/memory/{m.id}/invalidate", json={"reason": "wrong route"}) assert r.status_code == 409 def test_api_invalidate_unknown_id_is_404(env): client = TestClient(app) r = client.post("/memory/no-such-id/invalidate", json={"reason": "ghost"}) assert r.status_code == 404 def test_api_supersede_candidate_returns_409(env): """Mirror of the invalidate guard: candidates must not silently flip to superseded via the active-only supersede route.""" m = create_memory( memory_type="knowledge", content="candidate target", status="candidate" ) client = TestClient(app) r = client.post(f"/memory/{m.id}/supersede", json={"reason": "wrong route"}) assert r.status_code == 409 # Row should still be a candidate assert _get_memory(m.id).status == "candidate" def test_api_supersede_already_superseded_is_idempotent(env): m = create_memory(memory_type="knowledge", content="will be superseded") client = TestClient(app) r1 = client.post(f"/memory/{m.id}/supersede", json={"reason": "first"}) assert r1.status_code == 200 r2 = client.post(f"/memory/{m.id}/supersede", json={"reason": "again"}) assert r2.status_code == 200 assert r2.json()["status"] == "already_superseded" def test_api_supersede_unknown_id_is_404(env): client = TestClient(app) r = client.post("/memory/no-such-id/supersede", json={"reason": "ghost"}) assert r.status_code == 404 def test_admin_dashboard_active_count_matches_full_table(env): """/admin/dashboard memories.active must match the SQL aggregate even when there are more active memories than the legacy sample limit (500). This guards the Codex finding that the dashboard was deriving counts from a confidence-sorted limit=500 fetch, hiding rows past the cap. We don't need 500 rows in the test — a small corpus that exercises the SQL-aggregate path is enough; the integrity-vs-dashboard equality is the invariant being asserted. """ # Mix of statuses to exercise the by_status aggregate create_memory(memory_type="knowledge", content="a") create_memory(memory_type="knowledge", content="b", project="p06-polisher") create_memory(memory_type="project", content="c-cand", status="candidate") cand = create_memory(memory_type="project", content="d-cand", status="candidate") # Invalidate one to seed an "invalid" bucket from atocore.memory.service import invalidate_memory target_id = cand.id # Promote it first via direct DB so invalidate does flip a candidate # to invalid via the service path (mirrors actual API trajectory). invalidate_memory(target_id) client = TestClient(app) dash = client.get("/admin/dashboard").json() assert dash["memories"]["active"] == 2 assert dash["memories"]["candidates"] == 1 assert dash["memories"]["by_status"]["invalid"] == 1 assert dash["memories"]["total"] == 4 assert dash["memories"]["by_project"].get("p06-polisher") == 1 # "(none)" bucket is the COALESCE label for empty/null project assert "(none)" in dash["memories"]["by_project"]