Coverage for src/ai_jury/replay.py: 98%
83 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-17 21:31 +0000
1"""Replay a saved jury outcome in the deliberation theater (issue #449).
3``jury replay <outcome.json>`` re-drives the presentation layer — the theater
4scene or the plain ``--live`` transcript stream — from a serialized
5:class:`~ai_jury.orchestrator.JuryOutcome`. No orchestration, no network, no
6agents: this module only re-issues the same ``on_event`` sequence the
7orchestrator emits during a live run (reviews in panel order → the recorded
8debate round → verify → synthesis), plus the ``set_vote``/``close`` finale the
9CLI performs on the theater.
11Accepted input shapes (sniffed by top-level keys):
13* a bare outcome dict — the exact shape :func:`ai_jury.cache.outcome_to_dict`
14 produces (top-level ``reviews`` / ``debate`` / ``chair`` ...);
15* a result-cache entry — the on-disk ``*.json`` the cache writes, which wraps
16 that same dict under an ``"outcome"`` key.
18A ``--format json`` report (``schema_version`` + ``metadata`` at top level) is
19recognised and rejected with a clear message: it carries findings and the
20verdict but NOT the per-agent deliberation stream, so there is nothing to
21replay from it.
23Security posture: the file is untrusted input. It is read with a hard byte cap
24(mirroring ``config._read_toml_bounded`` / ``cache._MAX_CACHE_BYTES``), parsed
25with plain :func:`json.loads` (never eval), and every parse failure surfaces as
26a :class:`ReplayError` rather than a traceback. Output hardening (control-byte
27scrubbing, bidi/zero-width stripping) is the theater's and report renderer's
28existing responsibility — replay adds no new output channel.
29"""
31from __future__ import annotations
33import json
34from collections.abc import Iterator
35from pathlib import Path
37from .adapters import AgentResult
38from .cache import outcome_from_dict
39from .orchestrator import JuryOutcome
40from .redaction import redact
42# Upper bound on a replay-file read (issue #449). Matches the result cache's
43# ceiling (``cache._MAX_CACHE_BYTES``): a serialized outcome is a few KB, so a
44# multi-MB file is either corrupt or hostile — reject it without pulling it
45# fully into memory.
46_MAX_REPLAY_BYTES = 8 * 1024 * 1024
49class ReplayError(ValueError):
50 """A replay input problem the user can act on (bad path/shape/JSON)."""
53def load_outcome(path: Path | str) -> JuryOutcome:
54 """Load and validate a serialized outcome from ``path``.
56 Accepts a bare ``outcome_to_dict`` dict or a cache entry wrapping one under
57 ``"outcome"``. Raises :class:`ReplayError` — never a raw traceback — for a
58 missing/unreadable file, an oversized file, invalid JSON, an unrecognized
59 shape, or a malformed outcome.
60 """
61 path = Path(path)
62 try:
63 # Cap the READ itself (not stat-then-read, which is a TOCTOU): read at
64 # most the ceiling + 1 so an oversized file is detected without ever
65 # being held fully in memory.
66 with path.open("r", encoding="utf-8") as fh:
67 raw = fh.read(_MAX_REPLAY_BYTES + 1)
68 except OSError as exc:
69 raise ReplayError(f"cannot read '{path}': {redact(str(exc))[0]}") from None
70 except UnicodeDecodeError as exc:
71 raise ReplayError(f"'{path}' is not UTF-8 text: {redact(str(exc))[0]}") from None
72 if len(raw) > _MAX_REPLAY_BYTES:
73 raise ReplayError(f"'{path}' exceeds the {_MAX_REPLAY_BYTES}-byte replay limit")
74 try:
75 data = json.loads(raw)
76 except (ValueError, RecursionError) as exc:
77 # RecursionError on deeply nested JSON is not a ValueError; catch it so
78 # a hostile file cannot crash the loader (mirrors cache.py).
79 raise ReplayError(f"'{path}' is not valid JSON: {redact(str(exc))[0]}") from None
81 if not isinstance(data, dict):
82 raise ReplayError(f"'{path}' is not a JSON object")
84 if isinstance(data.get("outcome"), dict):
85 # Result-cache entry ({"cache_schema": ..., "outcome": {...}, "mac": ...}).
86 # The MAC is deliberately NOT verified here: it authenticates entries
87 # for the cache-hit fast path; replay is presentation-only and the user
88 # chose this file explicitly.
89 inner = data["outcome"]
90 elif "reviews" in data:
91 # Bare outcome_to_dict shape.
92 inner = data
93 elif "schema_version" in data and "metadata" in data:
94 raise ReplayError(
95 f"'{path}' looks like a `jury --format json` report, which does not "
96 "contain the per-agent deliberation stream (reviews/debate), so it "
97 "cannot be replayed. Pass a serialized outcome instead: a result-"
98 "cache entry (see `jury --cache`) or an `outcome_to_dict` dump."
99 )
100 else:
101 raise ReplayError(
102 f"'{path}' is not a recognized outcome shape (expected a serialized "
103 "outcome with a top-level 'reviews' key, or a cache entry with a "
104 "top-level 'outcome' key)"
105 )
107 _coerce_agent_results(inner)
108 try:
109 outcome = outcome_from_dict(inner)
110 except (KeyError, TypeError, AttributeError, ValueError) as exc:
111 raise ReplayError(f"'{path}' holds a malformed outcome: {redact(str(exc))[0]}") from None
112 if not outcome.reviews:
113 raise ReplayError(f"'{path}' contains no reviews — nothing to replay")
114 return outcome
117def _coerce_agent_results(inner: dict) -> None:
118 """Coerce type-invalid AgentResult fields in place (untrusted file).
120 ``outcome_from_dict``/``cache._agent_result`` copy values without type
121 validation, so a hand-edited file with ``"output": null`` or a string
122 ``duration_s`` would pass loading and crash far later inside the render
123 loop with a raw traceback (review finding). Coerce the fields the render
124 path consumes: ``output`` to str, ``duration_s`` to float, ``ok`` to bool,
125 ``agent``/``vendor`` to str.
126 """
127 for key in ("reviews", "debate"):
128 items = inner.get(key)
129 if not isinstance(items, list):
130 continue
131 for item in items:
132 if isinstance(item, dict):
133 _coerce_one(item)
134 for key in ("synthesis", "verify"):
135 item = inner.get(key)
136 if isinstance(item, dict):
137 _coerce_one(item)
140def _coerce_one(item: dict) -> None:
141 item["agent"] = str(item.get("agent") or "")
142 item["vendor"] = str(item.get("vendor") or "")
143 item["ok"] = bool(item.get("ok"))
144 out = item.get("output")
145 item["output"] = out if isinstance(out, str) else ("" if out is None else str(out))
146 try:
147 item["duration_s"] = float(item.get("duration_s") or 0.0)
148 except (TypeError, ValueError):
149 item["duration_s"] = 0.0
152def replay_events(
153 outcome: JuryOutcome,
154) -> Iterator[tuple[str, AgentResult, int | None]]:
155 """Yield the ``(kind, result, round_no)`` sequence a live run would emit.
157 Mirrors the orchestrator's ``on_event`` stream: each review in panel order,
158 then the recorded debate round, then verify, then synthesis — phases absent
159 from the outcome are simply skipped, exactly as a live run that skipped
160 them. The outcome stores only the FINAL debate round (earlier rounds are
161 superseded, not serialized), numbered from ``rounds_executed`` — debate
162 rounds start at 2, review being round 1.
163 """
164 for r in outcome.reviews:
165 yield ("review", r, None)
166 if outcome.debate:
167 round_no = outcome.rounds_executed if outcome.rounds_executed >= 2 else 2
168 for r in outcome.debate:
169 yield ("debate", r, round_no)
170 if outcome.verify is not None:
171 yield ("verify", outcome.verify, None)
172 if outcome.synthesis is not None:
173 yield ("synthesis", outcome.synthesis, None)
176def replay_into(court, outcome: JuryOutcome, vote=None) -> None:
177 """Drive ``court`` (a theater ``Courtroom``-like object) from ``outcome``.
179 Uses exactly the API the live CLI path uses: ``open()``, ``step(kind,
180 result, round_no)`` per event, ``set_vote(vote)`` when a panel vote is
181 supplied, and ``close()`` (always — the terminal must be restored even if a
182 step raises).
183 """
184 court.open()
185 try:
186 for kind, result, round_no in replay_events(outcome):
187 court.step(kind, result, round_no)
188 if vote is not None:
189 court.set_vote(vote)
190 finally:
191 court.close()