feat(kernel): administrative terminal_superseded events — immutable-terminal corrections (#54)

loop/events.py

@@ -14,13 +14,14 @@
 from .emit import _ITERATION_OUTCOMES, _RECEIPT_OUTCOMES, _RECEIPT_ROLES
 
 EVENT_SCHEMA_ID = "loop-engineer/event@1"
-EVENT_TYPES = ("contract_opened", "iteration_appended", "receipt_appended", "terminal_written")
+EVENT_TYPES = ("contract_opened", "iteration_appended", "receipt_appended", "terminal_written", "terminal_superseded")
 
 _PAYLOAD_REQUIRED_FIELDS: dict[str, tuple[str, ...]] = {
     "contract_opened": ("workspace",),
     "iteration_appended": ("iteration_id", "outcome"),
     "receipt_appended": ("iteration_id", "role", "model", "outcome"),
     "terminal_written": ("state", "criteria_met", "evidence", "false_completion"),
+    "terminal_superseded": ("state", "criteria_met", "evidence", "false_completion", "justification", "authority"),
 }
 
 _CREATE_EVENTS_TABLE = """
@@ -115,6 +116,24 @@ def _payload_issues(event_type: str, payload: Any) -> list[str]:
             issues.append("terminal_written.evidence must be an array")
         if "false_completion" in payload and not isinstance(payload["false_completion"], bool):
             issues.append("terminal_written.false_completion must be a boolean")
+    elif event_type == "terminal_superseded":
+        if "state" in payload and not isinstance(payload["state"], str):
+            issues.append("terminal_superseded.state must be a string")
+        if "criteria_met" in payload and not isinstance(payload["criteria_met"], dict):
+            issues.append("terminal_superseded.criteria_met must be an object")
+        if "evidence" in payload and not isinstance(payload["evidence"], list):
+            issues.append("terminal_superseded.evidence must be an array")
+        if "false_completion" in payload and not isinstance(payload["false_completion"], bool):
+            issues.append("terminal_superseded.false_completion must be a boolean")
+        if "justification" in payload and (not isinstance(payload["justification"], str)
+                                            or not payload["justification"].strip()):
+            issues.append("terminal_superseded.justification must be a non-empty string")
+        if "authority" in payload:
+            authority = payload["authority"]
+            if (not isinstance(authority, dict)
+                    or not isinstance(authority.get("by"), str) or not authority.get("by", "").strip()
+                    or not isinstance(authority.get("at"), str) or not authority.get("at", "").strip()):
+                issues.append("terminal_superseded.authority must be an object with non-empty by/at strings")
     return issues
 
 

loop/reducer.py

@@ -16,28 +16,39 @@ class EventReplayError(ValueError):
 
 def _empty_projection(run_id: str | None) -> dict[str, Any]:
     return {"run_id": run_id, "state": None, "iteration_id": None, "active_task": None,
-            "terminal": None, "runlog_entries": [], "receipts": [], "event_count": 0,
+            "terminal": None, "runlog_entries": [], "receipts": [], "superseded_history": [], "event_count": 0,
             "last_sequence": None}
 
 
 def _validate_terminal_payload_semantics(payload: Mapping[str, Any]) -> None:
     state = payload.get("state")
     if state not in TERMINAL_STATES:
-        raise EventReplayError(f"terminal_written payload has non-canonical state {state!r}")
+        raise EventReplayError(f"terminal payload has non-canonical state {state!r}")
     if state != "Succeeded":
         return
     if payload.get("false_completion") is True:
-        raise EventReplayError("refusing Succeeded terminal_written with false_completion=True (G1)")
+        raise EventReplayError("refusing Succeeded terminal with false_completion=True (G1)")
     try:
         policy = normalize_completion_policy(payload.get("completion_policy"))
     except CompletionPolicyError as exc:
         raise EventReplayError(f"invalid completion_policy: {exc}") from exc
     criteria = payload.get("criteria_met")
     if not isinstance(criteria, dict) or not criteria_satisfy_completion(criteria, policy):
-        raise EventReplayError("Succeeded terminal_written does not satisfy the completion policy (G1)")
+        raise EventReplayError("Succeeded terminal does not satisfy the completion policy (G1)")
     evidence = payload.get("evidence")
     if not isinstance(evidence, list) or not evidence:
-        raise EventReplayError("Succeeded terminal_written has empty evidence (G1)")
+        raise EventReplayError("Succeeded terminal has empty evidence (G1)")
+
+
+def _validate_superseded_payload_semantics(payload: Mapping[str, Any]) -> None:
+    _validate_terminal_payload_semantics(payload)
+    if not isinstance(payload.get("justification"), str) or not payload["justification"].strip():
+        raise EventReplayError("terminal_superseded payload missing non-empty justification")
+    authority = payload.get("authority")
+    if (not isinstance(authority, dict)
+            or not isinstance(authority.get("by"), str) or not authority.get("by", "").strip()
+            or not isinstance(authority.get("at"), str) or not authority.get("at", "").strip()):
+        raise EventReplayError("terminal_superseded payload missing authority.by/authority.at")
 
 
 def _reduce_one(state: dict[str, Any], event: Mapping[str, Any]) -> dict[str, Any]:
@@ -52,9 +63,12 @@ def _reduce_one(state: dict[str, Any], event: Mapping[str, Any]) -> dict[str, An
     expected_sequence = 0 if state["last_sequence"] is None else state["last_sequence"] + 1
     if event["sequence"] != expected_sequence:
         raise EventReplayError(f"non-monotonic sequence: expected {expected_sequence}, got {event['sequence']!r}")
-    if state["terminal"] is not None:
-        raise EventReplayError("event appended after terminal — terminal is immutable")
     event_type = event["type"]
+    if state["terminal"] is not None:
+        if event_type != "terminal_superseded":
+            raise EventReplayError("event appended after terminal — terminal is immutable")
+    elif event_type == "terminal_superseded":
+        raise EventReplayError("terminal_superseded has nothing to supersede (no terminal record yet)")
     if event_type == "contract_opened" and state["last_sequence"] is not None:
         raise EventReplayError("contract_opened must be the first event in a run")
     if event_type != "contract_opened" and state["state"] is None:
@@ -86,12 +100,23 @@ def _reduce_one(state: dict[str, Any], event: Mapping[str, Any]) -> dict[str, An
         _validate_terminal_payload_semantics(payload)
         new_state["state"] = fsm.TERMINAL_MARKER
         new_state["terminal"] = dict(payload, event_id=event["event_id"], ts=event["ts"])
+    elif event_type == "terminal_superseded":
+        current_terminal = state["terminal"]
+        if event.get("causation_id") != current_terminal.get("event_id"):
+            raise EventReplayError(
+                "terminal_superseded.causation_id must reference the event_id "
+                "of the terminal record it corrects"
+            )
+        _validate_superseded_payload_semantics(payload)
+        history_entry = {**current_terminal, "superseded_by": event["event_id"], "superseded_at": event["ts"]}
+        new_state["superseded_history"] = state["superseded_history"] + [history_entry]
+        new_state["terminal"] = dict(payload, event_id=event["event_id"], ts=event["ts"])
     return new_state
 
 
 def reduce_events(events: Iterable[Mapping[str, Any]], *, initial: Mapping[str, Any] | None = None) -> dict[str, Any]:
     """Fold events without mutating the supplied stream or initial projection."""
-    state = ({**initial, "runlog_entries": list(initial["runlog_entries"]), "receipts": list(initial["receipts"])} if initial is not None else _empty_projection(None))
+    state = ({**initial, "runlog_entries": list(initial["runlog_entries"]), "receipts": list(initial["receipts"]), "superseded_history": list(initial["superseded_history"])} if initial is not None else _empty_projection(None))
     for event in events:
         state = _reduce_one(state, event)
     return state

reference/repo-os-contract.md

@@ -641,6 +641,30 @@ SHA-256 comparison is attempted.
 
 ---
 
+## 18. `terminal_superseded` — administrative terminal corrections
+
+`terminal_superseded` is an append-only administrative event that corrects the
+currently effective terminal record while preserving the original decision and
+every later correction in the reducer projection's oldest-first
+`superseded_history`. Each correction carries the corrected terminal fields,
+non-empty `justification`, and `{by, at}` `authority`, and its `causation_id`
+must identify the terminal event it corrects; chained corrections therefore
+remain auditable without replacing any record.
+
+**Scope boundary:** this is a fifth event type with no corresponding
+`loop.emit` writer operation — deliberately: unlike the other four,
+`terminal_superseded` is administrative and event-log-only; §16's
+“one-to-one with `loop.emit`'s four writer operations” describes the other four
+types and predates this addition. It is not file-based `terminal@1` replacement
+or an `emit`/`doctor` workflow.
+
+**Domain enforcement:** the EventStore validates envelope and payload shape
+only; the reducer alone admits this type after a terminal, verifies its
+causation anchor, and reuses G1 completion checks when a correction sets
+`state` to `Succeeded`. All other event types remain forbidden after a terminal.
+
+---
+
 Sources: "Designing a Loop Engineer Skill for Frontier Agent Workflows" (2026), synthesizing
 Anthropic guidance on long-running agent harnesses (anthropic.com, 2025), OpenAI Agents/Codex guidance, Google
 Conductor, and arXiv PreFlect (2602.07187), SWE-Marathon (2606.07682), Web Agents

schemas/event.schema.json

@@ -10,7 +10,7 @@
     "event_id": { "type": "string", "minLength": 1 },
     "run_id": { "type": "string", "minLength": 1 },
     "sequence": { "type": "integer", "minimum": 0 },
-    "type": { "enum": ["contract_opened", "iteration_appended", "receipt_appended", "terminal_written"] },
+    "type": { "enum": ["contract_opened", "iteration_appended", "receipt_appended", "terminal_written", "terminal_superseded"] },
     "actor": { "type": "string", "minLength": 1 },
     "causation_id": { "type": ["string", "null"] },
     "correlation_id": { "type": ["string", "null"] },

scripts/test_eventstore.py

@@ -117,3 +117,66 @@ def append(index: int) -> int:
     with ThreadPoolExecutor(max_workers=8) as pool:
         sequences = list(pool.map(append, range(8)))
     assert set(sequences) == set(range(1, 9))
+
+
+def terminal_superseded_event(**overrides: object) -> dict[str, object]:
+    result: dict[str, object] = {
+        "schema": EVENT_SCHEMA_ID, "event_id": "superseded-1", "run_id": "run-1", "sequence": 1,
+        "type": "terminal_superseded", "actor": "test", "causation_id": "terminal-1",
+        "correlation_id": None, "ts": "2026-01-01T00:00:01+00:00",
+        "payload": {"state": "FailedSafety", "criteria_met": {"done": True}, "evidence": ["proof"],
+                    "false_completion": False, "justification": "audit correction",
+                    "authority": {"by": "ops", "at": "2026-01-01T00:00:01+00:00"}},
+    }
+    result.update(overrides)
+    return result
+
+
+def test_event_types_include_terminal_superseded_and_match_schema_enum() -> None:
+    schema = __import__("json").load(open("schemas/event.schema.json", encoding="utf-8"))
+    from loop.events import EVENT_TYPES
+
+    assert "terminal_superseded" in EVENT_TYPES
+    assert "terminal_superseded" in schema["properties"]["type"]["enum"]
+    assert set(EVENT_TYPES) == set(schema["properties"]["type"]["enum"])
+
+
+@pytest.mark.parametrize("mode", ["basic", "release"])
+def test_terminal_superseded_validates_in_basic_and_release_modes(mode: str) -> None:
+    if mode == "release":
+        pytest.importorskip("jsonschema")
+    assert validate_event(terminal_superseded_event(), mode=mode)["ok"] is True
+
+
+@pytest.mark.parametrize("mode", ["basic", "release"])
+@pytest.mark.parametrize("payload", [
+    {"criteria_met": {}, "evidence": [], "false_completion": False, "justification": "j", "authority": {"by": "a", "at": "t"}},
+    {"state": 1, "criteria_met": {}, "evidence": [], "false_completion": False, "justification": "j", "authority": {"by": "a", "at": "t"}},
+    {"state": "FailedSafety", "evidence": [], "false_completion": False, "justification": "j", "authority": {"by": "a", "at": "t"}},
+    {"state": "FailedSafety", "criteria_met": [], "evidence": [], "false_completion": False, "justification": "j", "authority": {"by": "a", "at": "t"}},
+    {"state": "FailedSafety", "criteria_met": {}, "false_completion": False, "justification": "j", "authority": {"by": "a", "at": "t"}},
+    {"state": "FailedSafety", "criteria_met": {}, "evidence": "proof", "false_completion": False, "justification": "j", "authority": {"by": "a", "at": "t"}},
+    {"state": "FailedSafety", "criteria_met": {}, "evidence": [], "justification": "j", "authority": {"by": "a", "at": "t"}},
+    {"state": "FailedSafety", "criteria_met": {}, "evidence": [], "false_completion": 0, "justification": "j", "authority": {"by": "a", "at": "t"}},
+    {"state": "FailedSafety", "criteria_met": {}, "evidence": [], "false_completion": False, "authority": {"by": "a", "at": "t"}},
+    {"state": "FailedSafety", "criteria_met": {}, "evidence": [], "false_completion": False, "justification": " ", "authority": {"by": "a", "at": "t"}},
+    {"state": "FailedSafety", "criteria_met": {}, "evidence": [], "false_completion": False, "justification": "j"},
+    {"state": "FailedSafety", "criteria_met": {}, "evidence": [], "false_completion": False, "justification": "j", "authority": {"by": "", "at": "t"}},
+])
+def test_terminal_superseded_rejects_malformed_payload_fields(mode: str, payload: dict[str, object]) -> None:
+    if mode == "release":
+        pytest.importorskip("jsonschema")
+    assert validate_event(terminal_superseded_event(payload=payload), mode=mode)["ok"] is False
+    with pytest.raises(EventValidationError):
+        SQLiteEventStore(":memory:").append("run", "terminal_superseded", payload, actor="test")
+
+
+def test_store_appends_well_formed_terminal_superseded(tmp_path) -> None:
+    store = SQLiteEventStore(tmp_path / "events.db")
+    store.append("run", "contract_opened", {"workspace": "w"}, actor="test")
+    terminal = store.append("run", "terminal_written", {"state": "Succeeded", "criteria_met": {"done": True},
+                             "evidence": ["proof"], "false_completion": False}, actor="test")
+    correction = store.append("run", "terminal_superseded", terminal_superseded_event()["payload"], actor="test",
+                              causation_id=terminal["event_id"])
+    assert correction["type"] == "terminal_superseded"
+    assert correction["causation_id"] == terminal["event_id"]