diff --git a/docs/adr/ADR-111-ceg-outcome-replay.md b/docs/adr/ADR-111-ceg-outcome-replay.md new file mode 100644 index 0000000..3390ab3 --- /dev/null +++ b/docs/adr/ADR-111-ceg-outcome-replay.md @@ -0,0 +1,17 @@ +# ADR-111: CEG deterministic outcome replay + +## Status +Accepted (TASK-033) + +## Context +TASK-057 produces Odoo replay inputs (`l9.odoo.outcome_replay_input.v1`). CEG needs an offline consumer that yields identical outcome hashes across runs without Gate or Neo4j. + +## Decision +- Accept Odoo schema `l9.odoo.outcome_replay_input.v1` only +- Reject `gate_mutation=true` +- Emit `l9.ceg.outcome_replay.v1` with per-event `outcome_hash` and aggregate `outcome_set_hash` +- Pure function / CLI: `gate_calls=0`, `network=false` +- No handler registration changes; observational tooling only + +## Consequences +TASK-058 can chain Odoo input hash → CEG `outcome_set_hash`. Live stack is out of scope for this ADR. diff --git a/docs/runbooks/CEG_OUTCOME_REPLAY.md b/docs/runbooks/CEG_OUTCOME_REPLAY.md new file mode 100644 index 0000000..bbbc10a --- /dev/null +++ b/docs/runbooks/CEG_OUTCOME_REPLAY.md @@ -0,0 +1,19 @@ +# Runbook: CEG outcome replay (TASK-033) + +## Input +Odoo artifact from TASK-057 (`schema=l9.odoo.outcome_replay_input.v1`). + +## Command +```bash +env PYTHONPATH=. python3.12 tools/outcome_replay.py \ + --input /path/to/odoo-replay-input.json \ + --output /tmp/ceg-outcome-replay.json +``` + +## Guarantees +- Deterministic: same input → same `outcome_set_hash` +- No Gate / Neo4j / HTTP +- Compatible field set: `tenant`, `action`, `packet_id` (+ optional payload) + +## Recovery +Discard generated JSON; revert `engine/replay/` and `tools/outcome_replay.py` if needed. diff --git a/engine/replay/__init__.py b/engine/replay/__init__.py new file mode 100644 index 0000000..789eb41 --- /dev/null +++ b/engine/replay/__init__.py @@ -0,0 +1,17 @@ +"""Deterministic CEG outcome replay from Odoo/fixture inputs (TASK-033).""" + +from engine.replay.outcome import ( + ODOO_INPUT_SCHEMA, + REPLAY_OUTCOME_SCHEMA, + ReplayError, + load_odoo_replay_input, + replay_outcomes, +) + +__all__ = [ + "ODOO_INPUT_SCHEMA", + "REPLAY_OUTCOME_SCHEMA", + "ReplayError", + "load_odoo_replay_input", + "replay_outcomes", +] diff --git a/engine/replay/outcome.py b/engine/replay/outcome.py new file mode 100644 index 0000000..deed830 --- /dev/null +++ b/engine/replay/outcome.py @@ -0,0 +1,113 @@ +""" +--- L9_META --- +l9_schema: 1 +origin: engine-specific +engine: graph +layer: [replay] +tags: [replay, outcome, offline, task-033] +owner: engine-team +status: active +--- /L9_META --- + +Offline outcome replay compatible with Odoo TASK-057 input schema. +""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +ODOO_INPUT_SCHEMA = "l9.odoo.outcome_replay_input.v1" +REPLAY_OUTCOME_SCHEMA = "l9.ceg.outcome_replay.v1" +REQUIRED_EVENT_FIELDS = ("tenant", "action", "packet_id") + + +class ReplayError(ValueError): + """Invalid replay input or forbidden live-path usage.""" + + +def _canonical(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + + +def _digest(value: Any) -> str: + return "sha256:" + hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest() + + +def load_odoo_replay_input(document: dict[str, Any]) -> dict[str, Any]: + """Validate Odoo replay input; never mutates Gate or opens network.""" + if not isinstance(document, dict): + raise ReplayError("replay input must be an object") + if document.get("schema") != ODOO_INPUT_SCHEMA: + raise ReplayError(f"unsupported schema: {document.get('schema')!r}") + if document.get("gate_mutation") is True: + raise ReplayError("gate_mutation=true is forbidden in replay mode") + events = document.get("events") + if not isinstance(events, list): + raise ReplayError("events must be a list") + normalized: list[dict[str, Any]] = [] + for raw in events: + if not isinstance(raw, dict): + raise ReplayError("each event must be an object") + missing = [f for f in REQUIRED_EVENT_FIELDS if not str(raw.get(f) or "").strip()] + if missing: + raise ReplayError(f"event missing required fields: {missing}") + event = { + "action": str(raw["action"]).strip(), + "packet_id": str(raw["packet_id"]).strip(), + "tenant": str(raw["tenant"]).strip(), + } + for optional in ("correlation_id", "source_model", "source_id", "observed_at"): + if raw.get(optional) is not None: + event[optional] = str(raw[optional]) + if isinstance(raw.get("payload"), dict): + event["payload"] = raw["payload"] + normalized.append(event) + normalized.sort(key=lambda e: (e["packet_id"], e["action"], e["tenant"])) + return { + "schema": ODOO_INPUT_SCHEMA, + "schema_version": str(document.get("schema_version") or ""), + "producer": document.get("producer"), + "producer_task": document.get("producer_task"), + "gate_mutation": False, + "event_count": len(normalized), + "events": normalized, + "content_hash": document.get("content_hash"), + } + + +def replay_outcomes(document: dict[str, Any]) -> dict[str, Any]: + """Derive deterministic outcome hashes from recorded events (no I/O).""" + loaded = load_odoo_replay_input(document) + outcomes: list[dict[str, Any]] = [] + for event in loaded["events"]: + body = { + "action": event["action"], + "packet_id": event["packet_id"], + "payload": event.get("payload", {}), + "tenant": event["tenant"], + } + for optional in ("correlation_id", "source_model", "source_id", "observed_at"): + if optional in event: + body[optional] = event[optional] + outcomes.append( + { + "action": event["action"], + "outcome_hash": _digest(body), + "packet_id": event["packet_id"], + "tenant": event["tenant"], + } + ) + result: dict[str, Any] = { + "event_count": len(outcomes), + "gate_calls": 0, + "input_content_hash": loaded.get("content_hash"), + "input_schema": ODOO_INPUT_SCHEMA, + "network": False, + "outcomes": outcomes, + "replay_mode": True, + "schema": REPLAY_OUTCOME_SCHEMA, + } + result["outcome_set_hash"] = _digest({"outcomes": outcomes, "schema": REPLAY_OUTCOME_SCHEMA}) + return result diff --git a/engine/shadow/compare.py b/engine/shadow/compare.py index 30b6be0..7bf1d9a 100644 --- a/engine/shadow/compare.py +++ b/engine/shadow/compare.py @@ -1,4 +1,16 @@ -"""Deterministic primary-vs-shadow ranking comparison (observational only).""" +""" +--- L9_META --- +l9_schema: 1 +origin: engine-specific +engine: graph +layer: [shadow] +tags: [shadow, comparison, observational, task-055] +owner: engine-team +status: active +--- /L9_META --- + +Deterministic primary-vs-shadow ranking comparison (observational only). +""" from __future__ import annotations diff --git a/tests/contracts/test_contracts.py b/tests/contracts/test_contracts.py index 207fef5..cef0c13 100644 --- a/tests/contracts/test_contracts.py +++ b/tests/contracts/test_contracts.py @@ -579,6 +579,8 @@ def test_no_unexpected_top_level_dirs_in_engine(self): "diagnostics", "hoprag", "models", + "shadow", + "replay", } for item in ENGINE_DIR.iterdir(): if item.is_dir() and not item.name.startswith("."): diff --git a/tests/unit/test_outcome_replay.py b/tests/unit/test_outcome_replay.py new file mode 100644 index 0000000..1c861ab --- /dev/null +++ b/tests/unit/test_outcome_replay.py @@ -0,0 +1,100 @@ +"""Unit tests for deterministic CEG outcome replay (TASK-033).""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from engine.replay.outcome import ( + ODOO_INPUT_SCHEMA, + ReplayError, + replay_outcomes, +) + +ROOT = Path(__file__).resolve().parents[2] + + +def _fixture() -> dict: + return { + "schema": ODOO_INPUT_SCHEMA, + "schema_version": "1.0.0", + "producer": "odoo.outcome_replay_export", + "producer_task": "TASK-057", + "gate_mutation": False, + "event_count": 2, + "events": [ + { + "tenant": "plasticos", + "action": "match", + "packet_id": "pkt-b", + "payload": {"score": 0.2}, + }, + { + "tenant": "plasticos", + "action": "match", + "packet_id": "pkt-a", + "payload": {"score": 0.9}, + }, + ], + "content_hash": "sha256:fixture", + } + + +@pytest.mark.unit +def test_replay_deterministic_across_two_runs() -> None: + a = replay_outcomes(_fixture()) + b = replay_outcomes(_fixture()) + assert a == b + assert a["network"] is False + assert a["gate_calls"] == 0 + assert a["outcome_set_hash"].startswith("sha256:") + assert len(a["outcomes"]) == 2 + + +@pytest.mark.unit +def test_event_order_does_not_change_hash() -> None: + doc = _fixture() + doc["events"] = list(reversed(doc["events"])) + assert replay_outcomes(doc)["outcome_set_hash"] == replay_outcomes(_fixture())["outcome_set_hash"] + + +@pytest.mark.unit +def test_rejects_gate_mutation_and_wrong_schema() -> None: + bad = _fixture() + bad["gate_mutation"] = True + with pytest.raises(ReplayError): + replay_outcomes(bad) + bad2 = _fixture() + bad2["schema"] = "other" + with pytest.raises(ReplayError): + replay_outcomes(bad2) + + +@pytest.mark.unit +def test_cli_two_runs_identical(tmp_path: Path) -> None: + import importlib.util + + spec = importlib.util.spec_from_file_location("ceg_outcome_replay_cli", ROOT / "tools" / "outcome_replay.py") + assert spec is not None + assert spec.loader is not None + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + + inp = tmp_path / "in.json" + out1 = tmp_path / "o1.json" + out2 = tmp_path / "o2.json" + inp.write_text(json.dumps(_fixture()), encoding="utf-8") + for dest in (out1, out2): + assert mod.main(["--input", str(inp), "--output", str(dest)]) == 0 + assert out1.read_text(encoding="utf-8") == out2.read_text(encoding="utf-8") + + +@pytest.mark.unit +def test_source_has_no_network_imports() -> None: + src = (ROOT / "engine" / "replay" / "outcome.py").read_text(encoding="utf-8") + assert "neo4j" not in src.lower() + assert "httpx" not in src + assert "GateClient" not in src + assert "requests" not in src diff --git a/tools/outcome_replay.py b/tools/outcome_replay.py new file mode 100644 index 0000000..62718da --- /dev/null +++ b/tools/outcome_replay.py @@ -0,0 +1,50 @@ +#!/usr/bin/env python3 +"""CLI: deterministic CEG outcome replay from Odoo fixtures (TASK-033). + +No Gate / Neo4j / network calls. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from engine.replay.outcome import ReplayError, replay_outcomes + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="CEG offline outcome replay") + parser.add_argument("--input", required=True, help="Odoo replay input JSON") + parser.add_argument("--output", required=True, help="Write replay outcome JSON") + args = parser.parse_args(argv) + try: + document = json.loads(Path(args.input).read_text(encoding="utf-8")) + result = replay_outcomes(document) + except (OSError, json.JSONDecodeError, ReplayError) as exc: + print(str(exc), file=sys.stderr) + return 1 + out = Path(args.output) + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text( + json.dumps(result, sort_keys=True, indent=2, ensure_ascii=True) + "\n", + encoding="utf-8", + ) + print( + json.dumps( + { + "output": str(out), + "outcome_set_hash": result["outcome_set_hash"], + "event_count": result["event_count"], + } + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())