feat(api): invalidate + supersede for active entities and memories (Issue E)

Public retraction path so mistakes can be corrected without SQL. Unblocks
the correction workflows that the live AKC p05 session exposed.

- engineering/service.py: invalidate_active_entity returns (ok, code)
  with codes invalidated/already_invalid/not_active/not_found for clean
  HTTP mapping. supersede_entity gains superseded_by + auto-creates the
  supersedes relationship (new SUPERSEDES old), rejects self-supersede
- memory/service.py: invalidate_memory/supersede_memory accept reason
  string that lands in audit note
- api/routes.py: POST /entities/{id}/invalidate, /supersede;
  POST /memory/{id}/invalidate, /supersede (all 4 behind /v1 aliases)
- tests/test_invalidate_supersede.py: 15 tests (idempotency, 404/409,
  supersede relationship auto-creation, self-supersede rejection,
  missing-replacement rejection, v1 alias presence)
- DEV-LEDGER.md: session log + test_count 494 -> 509

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-21 21:56:24 -04:00
parent 069d155585
commit 081c058f77
6 changed files with 427 additions and 9 deletions

View File

@@ -462,9 +462,76 @@ def reject_entity_candidate(entity_id: str, actor: str = "api", note: str = "")
return _set_entity_status(entity_id, "invalid", actor=actor, note=note)
def supersede_entity(entity_id: str, actor: str = "api", note: str = "") -> bool:
"""Mark an active entity as superseded by a newer one."""
return _set_entity_status(entity_id, "superseded", actor=actor, note=note)
def supersede_entity(
entity_id: str,
actor: str = "api",
note: str = "",
superseded_by: str | None = None,
) -> bool:
"""Mark an active entity as superseded.
When ``superseded_by`` names a real entity, also create a
``supersedes`` relationship from the new entity to the old one
(semantics: ``new SUPERSEDES old``). This keeps the graph
navigable without the caller remembering to make that edge.
"""
if superseded_by:
new_entity = get_entity(superseded_by)
if new_entity is None:
raise ValueError(
f"superseded_by entity not found: {superseded_by}"
)
if new_entity.id == entity_id:
raise ValueError("entity cannot supersede itself")
ok = _set_entity_status(entity_id, "superseded", actor=actor, note=note)
if not ok:
return False
if superseded_by:
try:
create_relationship(
source_entity_id=superseded_by,
target_entity_id=entity_id,
relationship_type="supersedes",
source_refs=[f"supersede-api:{actor}"],
)
except Exception as e:
log.warning(
"supersede_relationship_create_failed",
entity_id=entity_id,
superseded_by=superseded_by,
error=str(e),
)
return True
def invalidate_active_entity(
entity_id: str,
actor: str = "api",
reason: str = "",
) -> tuple[bool, str]:
"""Mark an active entity as invalid (Issue E — retraction path).
Returns (success, status_code) where status_code is one of:
- "invalidated" — happy path
- "not_found" — no such entity
- "already_invalid" — already invalid (idempotent)
- "not_active" — entity is candidate/superseded; use the
appropriate other endpoint
This is the public retraction API distinct from
``reject_entity_candidate`` (which only handles candidate→invalid).
"""
entity = get_entity(entity_id)
if entity is None:
return False, "not_found"
if entity.status == "invalid":
return True, "already_invalid"
if entity.status != "active":
return False, "not_active"
ok = _set_entity_status(entity_id, "invalid", actor=actor, note=reason)
return ok, "invalidated" if ok else "not_active"
def get_entity_audit(entity_id: str, limit: int = 100) -> list[dict]: