TASK-033: support deterministic CEG outcome replay

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.

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.

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",
+]

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

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
 

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("."):

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

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())