TASK-055: emit CEG shadow comparison outputs

docs/adr/ADR-110-ceg-shadow-comparison-outputs.md

@@ -0,0 +1,17 @@
+# ADR-110: CEG shadow comparison outputs
+
+## Status
+Accepted (TASK-055)
+
+## Context
+Wave-8 needs observational comparison between primary Gate match rankings and a shadow scorer without changing production match authority.
+
+## Decision
+- Emit structured comparison artifacts via `engine/shadow/` and `tools/shadow_comparison.py`.
+- Artifact is observational (`replaces_primary=false`).
+- Mismatch classes: `rank`, `score`, `missing`, `extra`.
+- Serialization is deterministic for identical inputs (stable checksum).
+- Primary `handle_match` remains the authority path; shadow does not replace its response.
+
+## Consequences
+Downstream dual-write/shadow validation (TASK-056) consumes these artifacts. No production cutover in this ADR.

docs/runbooks/CEG_SHADOW_COMPARISON.md

@@ -0,0 +1,28 @@
+# Runbook: CEG shadow comparison outputs (TASK-055)
+
+## Purpose
+Produce observational primary-vs-shadow ranking JSON for a Gate `packet_id`.
+
+## Emit offline
+```bash
+env PYTHONPATH=. python3.12 tools/shadow_comparison.py \
+  --input /tmp/shadow-input.json \
+  --output /tmp/shadow-comparison.json
+```
+
+Input shape:
+```json
+{
+  "packet_id": "...",
+  "primary": [{"candidate_id": "…", "score": 0.9, "rank": 1}],
+  "shadow": [{"candidate_id": "…", "score": 0.8, "rank": 1}]
+}
+```
+
+## Safety
+- Does not modify `handle_match` responses.
+- Does not write to Neo4j.
+- Treat output as evidence only until TASK-056 integration.
+
+## Rollback
+Delete or ignore comparison artifacts; primary match path unchanged.

engine/shadow/__init__.py

@@ -0,0 +1,12 @@
+"""Observational shadow comparison outputs (TASK-055).
+
+Does not replace primary match authority.
+"""
+
+from engine.shadow.compare import RankedCandidate, ShadowComparison, emit_shadow_comparison
+
+__all__ = [
+    "RankedCandidate",
+    "ShadowComparison",
+    "emit_shadow_comparison",
+]

engine/shadow/compare.py

@@ -0,0 +1,136 @@
+"""Deterministic primary-vs-shadow ranking comparison (observational only)."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import asdict, dataclass, field
+from typing import Any, Literal
+
+MismatchClass = Literal["rank", "score", "missing", "extra"]
+
+
+@dataclass(frozen=True)
+class RankedCandidate:
+    candidate_id: str
+    score: float
+    rank: int
+
+
+@dataclass
+class Mismatch:
+    mismatch_class: MismatchClass
+    candidate_id: str
+    primary_rank: int | None = None
+    shadow_rank: int | None = None
+    primary_score: float | None = None
+    shadow_score: float | None = None
+    detail: str = ""
+
+
+@dataclass
+class ShadowComparison:
+    schema: str = "l9.ceg.shadow_comparison.v1"
+    packet_id: str = ""
+    observational: bool = True
+    replaces_primary: bool = False
+    primary: list[RankedCandidate] = field(default_factory=list)
+    shadow: list[RankedCandidate] = field(default_factory=list)
+    mismatches: list[Mismatch] = field(default_factory=list)
+    checksum: str = ""
+
+    def to_dict(self) -> dict[str, Any]:
+        body = {
+            "schema": self.schema,
+            "packet_id": self.packet_id,
+            "observational": self.observational,
+            "replaces_primary": self.replaces_primary,
+            "primary": [asdict(x) for x in self.primary],
+            "shadow": [asdict(x) for x in self.shadow],
+            "mismatches": [asdict(x) for x in self.mismatches],
+        }
+        blob = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+        self.checksum = "sha256:" + hashlib.sha256(blob.encode()).hexdigest()
+        body["checksum"] = self.checksum
+        return body
+
+
+def _normalize(rows: list[RankedCandidate]) -> list[RankedCandidate]:
+    ordered = sorted(rows, key=lambda r: (r.rank, r.candidate_id))
+    return [RankedCandidate(candidate_id=r.candidate_id, score=float(r.score), rank=int(r.rank)) for r in ordered]
+
+
+def emit_shadow_comparison(
+    *,
+    packet_id: str,
+    primary: list[RankedCandidate],
+    shadow: list[RankedCandidate],
+    score_epsilon: float = 1e-9,
+) -> ShadowComparison:
+    """Compare primary match ranking to a shadow scorer ranking.
+
+    Pure function: never mutates match handler outputs; observational artifact only.
+    """
+    primary_n = _normalize(primary)
+    shadow_n = _normalize(shadow)
+    primary_by_id = {r.candidate_id: r for r in primary_n}
+    shadow_by_id = {r.candidate_id: r for r in shadow_n}
+    mismatches: list[Mismatch] = []
+
+    for cid, prow in primary_by_id.items():
+        srow = shadow_by_id.get(cid)
+        if srow is None:
+            mismatches.append(
+                Mismatch(
+                    mismatch_class="missing",
+                    candidate_id=cid,
+                    primary_rank=prow.rank,
+                    primary_score=prow.score,
+                    detail="present in primary, absent in shadow",
+                )
+            )
+            continue
+        if prow.rank != srow.rank:
+            mismatches.append(
+                Mismatch(
+                    mismatch_class="rank",
+                    candidate_id=cid,
+                    primary_rank=prow.rank,
+                    shadow_rank=srow.rank,
+                    primary_score=prow.score,
+                    shadow_score=srow.score,
+                    detail="rank differs",
+                )
+            )
+        if abs(prow.score - srow.score) > score_epsilon:
+            mismatches.append(
+                Mismatch(
+                    mismatch_class="score",
+                    candidate_id=cid,
+                    primary_rank=prow.rank,
+                    shadow_rank=srow.rank,
+                    primary_score=prow.score,
+                    shadow_score=srow.score,
+                    detail="score differs beyond epsilon",
+                )
+            )
+
+    for cid, srow in shadow_by_id.items():
+        if cid not in primary_by_id:
+            mismatches.append(
+                Mismatch(
+                    mismatch_class="extra",
+                    candidate_id=cid,
+                    shadow_rank=srow.rank,
+                    shadow_score=srow.score,
+                    detail="present in shadow, absent in primary",
+                )
+            )
+
+    mismatches.sort(key=lambda m: (m.mismatch_class, m.candidate_id))
+    return ShadowComparison(
+        packet_id=packet_id,
+        primary=primary_n,
+        shadow=shadow_n,
+        mismatches=mismatches,
+    )

tests/unit/test_shadow_comparison.py

@@ -0,0 +1,69 @@
+"""Unit tests for observational shadow comparison (TASK-055)."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from engine.shadow.compare import RankedCandidate, emit_shadow_comparison
+
+
+@pytest.mark.unit
+def test_emit_identical_rankings_no_mismatches() -> None:
+    rows = [
+        RankedCandidate("a", 0.9, 1),
+        RankedCandidate("b", 0.5, 2),
+    ]
+    result = emit_shadow_comparison(packet_id="pkt-1", primary=rows, shadow=list(rows))
+    assert result.observational is True
+    assert result.replaces_primary is False
+    assert result.mismatches == []
+    d1 = result.to_dict()
+    d2 = emit_shadow_comparison(packet_id="pkt-1", primary=rows, shadow=list(rows)).to_dict()
+    assert d1["checksum"] == d2["checksum"]
+
+
+@pytest.mark.unit
+def test_rank_score_missing_extra_classes() -> None:
+    primary = [
+        RankedCandidate("a", 0.9, 1),
+        RankedCandidate("b", 0.5, 2),
+        RankedCandidate("c", 0.1, 3),
+    ]
+    shadow = [
+        RankedCandidate("b", 0.8, 1),  # rank+score vs primary
+        RankedCandidate("a", 0.9, 2),  # rank vs primary
+        RankedCandidate("d", 0.05, 3),  # extra
+    ]
+    result = emit_shadow_comparison(packet_id="pkt-2", primary=primary, shadow=shadow)
+    classes = {m.mismatch_class for m in result.mismatches}
+    assert "rank" in classes
+    assert "score" in classes
+    assert "missing" in classes  # c
+    assert "extra" in classes  # d
+    assert result.replaces_primary is False
+
+
+@pytest.mark.unit
+def test_cli_writes_deterministic_artifact(tmp_path: Path) -> None:
+    from tools.shadow_comparison import main
+
+    inp = tmp_path / "in.json"
+    out = tmp_path / "out.json"
+    inp.write_text(
+        json.dumps(
+            {
+                "packet_id": "pkt-cli",
+                "primary": [{"candidate_id": "x", "score": 1.0, "rank": 1}],
+                "shadow": [{"candidate_id": "x", "score": 1.0, "rank": 1}],
+            }
+        )
+    )
+    rc = main(["--input", str(inp), "--output", str(out)])
+    assert rc == 0
+    data = json.loads(out.read_text())
+    assert data["observational"] is True
+    assert data["replaces_primary"] is False
+    assert data["checksum"].startswith("sha256:")

tools/shadow_comparison.py

@@ -0,0 +1,57 @@
+#!/usr/bin/env python3
+"""CLI: emit observational CEG shadow comparison JSON (TASK-055).
+
+Does not call Neo4j or alter primary match responses.
+"""
+
+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.shadow.compare import RankedCandidate, emit_shadow_comparison
+
+
+def _load_ranked(items: list[dict]) -> list[RankedCandidate]:
+    out: list[RankedCandidate] = []
+    for item in items:
+        out.append(
+            RankedCandidate(
+                candidate_id=str(item["candidate_id"]),
+                score=float(item["score"]),
+                rank=int(item["rank"]),
+            )
+        )
+    return out
+
+
+def main(argv: list[str] | None = None) -> int:
+    p = argparse.ArgumentParser(description="Emit CEG shadow comparison artifact")
+    p.add_argument("--input", required=True, help="JSON with packet_id, primary[], shadow[]")
+    p.add_argument("--output", required=True, help="Write comparison JSON here")
+    args = p.parse_args(argv)
+    data = json.loads(Path(args.input).read_text())
+    comparison = emit_shadow_comparison(
+        packet_id=str(data["packet_id"]),
+        primary=_load_ranked(data.get("primary") or []),
+        shadow=_load_ranked(data.get("shadow") or []),
+    )
+    out = Path(args.output)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    payload = comparison.to_dict()
+    out.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    print(
+        json.dumps(
+            {"ok": True, "output": str(out), "checksum": payload["checksum"], "mismatches": len(payload["mismatches"])}
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())