diff --git a/loop/events.py b/loop/events.py index 6b8b7c0..d5938b7 100644 --- a/loop/events.py +++ b/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 diff --git a/loop/reducer.py b/loop/reducer.py index 7d86728..329fcd8 100644 --- a/loop/reducer.py +++ b/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 diff --git a/reference/repo-os-contract.md b/reference/repo-os-contract.md index b53a06d..7eebd2c 100644 --- a/reference/repo-os-contract.md +++ b/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 diff --git a/schemas/event.schema.json b/schemas/event.schema.json index 04d6359..947bcca 100644 --- a/schemas/event.schema.json +++ b/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"] }, diff --git a/scripts/test_eventstore.py b/scripts/test_eventstore.py index c754cac..fa0f686 100644 --- a/scripts/test_eventstore.py +++ b/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"] diff --git a/scripts/test_reducer.py b/scripts/test_reducer.py index fce606c..1c1dea6 100644 --- a/scripts/test_reducer.py +++ b/scripts/test_reducer.py @@ -67,3 +67,96 @@ def test_reducer_rejects_event_after_terminal() -> None: events.append({**events[-1], "event_id": "later", "sequence": 5, "type": "receipt_appended", "payload": {"iteration_id": 2, "role": "read", "model": "m", "outcome": "ok"}}) with pytest.raises(EventReplayError, match="immutable"): reduce_events(events) + + +def superseded_event(*, event_id: str = "e5", sequence: int = 5, causation_id: str | None = "e4", + payload: dict[str, object] | None = None) -> dict[str, object]: + terminal = stream()[-1] + return {**terminal, "event_id": event_id, "sequence": sequence, "type": "terminal_superseded", + "ts": f"2026-01-01T00:00:{sequence:02d}+00:00", "causation_id": causation_id, + "payload": payload if payload is not None else { + "state": "FailedSafety", "criteria_met": {"done": True}, "evidence": ["proof"], + "false_completion": False, "justification": "audit correction", + "authority": {"by": "ops", "at": "2026-01-01T00:00:05+00:00"}, + }} + + +def test_reducer_admits_terminal_superseded_after_terminal() -> None: + result = reduce_events(stream() + [superseded_event()]) + assert result["terminal"]["event_id"] == "e5" + assert [entry["event_id"] for entry in result["superseded_history"]] == ["e4"] + + +@pytest.mark.parametrize("event_type, payload", [ + ("contract_opened", {"workspace": "w"}), + ("iteration_appended", {"iteration_id": 3, "outcome": "task_passed"}), + ("receipt_appended", {"iteration_id": 2, "role": "read", "model": "m", "outcome": "ok"}), + ("terminal_written", {"state": "FailedSafety", "criteria_met": {}, "evidence": [], "false_completion": False}), +]) +def test_reducer_rejects_every_other_event_type_after_terminal(event_type: str, payload: dict[str, object]) -> None: + terminal = stream()[-1] + events = stream() + [{**terminal, "event_id": "later", "sequence": 5, "type": event_type, "payload": payload}] + with pytest.raises(EventReplayError, match="event appended after terminal — terminal is immutable"): + reduce_events(events) + + +def test_reducer_rejects_terminal_superseded_before_any_terminal() -> None: + event = superseded_event(event_id="e0", sequence=0, causation_id=None) + with pytest.raises(EventReplayError, match="nothing to supersede"): + reduce_events([event]) + + +@pytest.mark.parametrize("causation_id", ["wrong", None]) +def test_reducer_rejects_terminal_superseded_with_mismatched_causation_id(causation_id: str | None) -> None: + with pytest.raises(EventReplayError, match="causation_id"): + reduce_events(stream() + [superseded_event(causation_id=causation_id)]) + + +def test_reducer_rejects_terminal_superseded_citing_a_stale_superseded_record() -> None: + events = stream() + [superseded_event(), superseded_event(event_id="e6", sequence=6, causation_id="e4")] + with pytest.raises(EventReplayError, match="causation_id"): + reduce_events(events) + + +def test_reducer_chains_multiple_terminal_supersessions_preserving_history_order() -> None: + result = reduce_events(stream() + [superseded_event(), superseded_event(event_id="e6", sequence=6, causation_id="e5")]) + assert [entry["event_id"] for entry in result["superseded_history"]] == ["e4", "e5"] + assert [entry["superseded_by"] for entry in result["superseded_history"]] == ["e5", "e6"] + assert result["terminal"]["event_id"] == "e6" + + +@pytest.mark.parametrize("payload, message", [ + ({"state": "succeeded", "criteria_met": {"done": True}, "evidence": ["proof"], "false_completion": False, "justification": "j", "authority": {"by": "a", "at": "t"}}, "non-canonical state"), + ({"state": "Succeeded", "criteria_met": {"done": True}, "evidence": ["proof"], "false_completion": True, "justification": "j", "authority": {"by": "a", "at": "t"}}, "false_completion"), + ({"state": "Succeeded", "criteria_met": {"done": False}, "evidence": ["proof"], "false_completion": False, "justification": "j", "authority": {"by": "a", "at": "t"}}, "completion policy"), + ({"state": "Succeeded", "criteria_met": {"done": True}, "evidence": [], "false_completion": False, "justification": "j", "authority": {"by": "a", "at": "t"}}, "empty evidence"), +]) +def test_reducer_terminal_superseded_enforces_g1_when_correcting_to_succeeded(payload: dict[str, object], message: str) -> None: + with pytest.raises(EventReplayError, match=message): + reduce_events(stream() + [superseded_event(payload=payload)]) + + +def test_reducer_terminal_superseded_allows_correcting_succeeded_to_a_failed_state() -> None: + events = stream() + [superseded_event(), superseded_event(event_id="e6", sequence=6, causation_id="e5", payload={ + "state": "FailedBlocked", "criteria_met": {}, "evidence": [], "false_completion": False, + "justification": "blocker confirmed", "authority": {"by": "ops", "at": "t"}, + })] + assert reduce_events(events)["terminal"]["state"] == "FailedBlocked" + + +@pytest.mark.parametrize("payload, message", [ + ({"state": "FailedSafety", "criteria_met": {}, "evidence": [], "false_completion": False, "authority": {"by": "a", "at": "t"}}, "justification"), + ({"state": "FailedSafety", "criteria_met": {}, "evidence": [], "false_completion": False, "justification": " " , "authority": {"by": "a", "at": "t"}}, "justification"), + ({"state": "FailedSafety", "criteria_met": {}, "evidence": [], "false_completion": False, "justification": "j"}, "authority"), + ({"state": "FailedSafety", "criteria_met": {}, "evidence": [], "false_completion": False, "justification": "j", "authority": {"by": "", "at": "t"}}, "authority"), +]) +def test_reducer_rejects_terminal_superseded_missing_justification_or_authority(payload: dict[str, object], message: str) -> None: + with pytest.raises(EventReplayError, match=message): + reduce_events(stream() + [superseded_event(payload=payload)]) + + +def test_reducer_terminal_superseded_is_deterministic_and_resumable() -> None: + events = stream() + [superseded_event(), superseded_event(event_id="e6", sequence=6, causation_id="e5")] + whole = reduce_events(events) + assert json.dumps(whole, sort_keys=True) == json.dumps(reduce_events(events), sort_keys=True) + assert reduce_events(events[3:], initial=reduce_events(events[:3])) == whole