Coverage for src/ai_jury/replay.py: 98%
82 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 23:18 +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
41# Upper bound on a replay-file read (issue #449). Matches the result cache's
42# ceiling (``cache._MAX_CACHE_BYTES``): a serialized outcome is a few KB, so a
43# multi-MB file is either corrupt or hostile — reject it without pulling it
44# fully into memory.
45_MAX_REPLAY_BYTES = 8 * 1024 * 1024
48class ReplayError(ValueError):
49 """A replay input problem the user can act on (bad path/shape/JSON)."""
52def load_outcome(path: Path | str) -> JuryOutcome:
53 """Load and validate a serialized outcome from ``path``.
55 Accepts a bare ``outcome_to_dict`` dict or a cache entry wrapping one under
56 ``"outcome"``. Raises :class:`ReplayError` — never a raw traceback — for a
57 missing/unreadable file, an oversized file, invalid JSON, an unrecognized
58 shape, or a malformed outcome.
59 """
60 path = Path(path)
61 try:
62 # Cap the READ itself (not stat-then-read, which is a TOCTOU): read at
63 # most the ceiling + 1 so an oversized file is detected without ever
64 # being held fully in memory.
65 with path.open("r", encoding="utf-8") as fh:
66 raw = fh.read(_MAX_REPLAY_BYTES + 1)
67 except OSError as exc:
68 raise ReplayError(f"cannot read '{path}': {exc}") from exc
69 except UnicodeDecodeError as exc:
70 raise ReplayError(f"'{path}' is not UTF-8 text: {exc}") from exc
71 if len(raw) > _MAX_REPLAY_BYTES:
72 raise ReplayError(
73 f"'{path}' exceeds the {_MAX_REPLAY_BYTES}-byte replay limit"
74 )
75 try:
76 data = json.loads(raw)
77 except (ValueError, RecursionError) as exc:
78 # RecursionError on deeply nested JSON is not a ValueError; catch it so
79 # a hostile file cannot crash the loader (mirrors cache.py).
80 raise ReplayError(f"'{path}' is not valid JSON: {exc}") from exc
82 if not isinstance(data, dict):
83 raise ReplayError(f"'{path}' is not a JSON object")
85 if isinstance(data.get("outcome"), dict):
86 # Result-cache entry ({"cache_schema": ..., "outcome": {...}, "mac": ...}).
87 # The MAC is deliberately NOT verified here: it authenticates entries
88 # for the cache-hit fast path; replay is presentation-only and the user
89 # chose this file explicitly.
90 inner = data["outcome"]
91 elif "reviews" in data:
92 # Bare outcome_to_dict shape.
93 inner = data
94 elif "schema_version" in data and "metadata" in data:
95 raise ReplayError(
96 f"'{path}' looks like a `jury --format json` report, which does not "
97 "contain the per-agent deliberation stream (reviews/debate), so it "
98 "cannot be replayed. Pass a serialized outcome instead: a result-"
99 "cache entry (see `jury --cache`) or an `outcome_to_dict` dump."
100 )
101 else:
102 raise ReplayError(
103 f"'{path}' is not a recognized outcome shape (expected a serialized "
104 "outcome with a top-level 'reviews' key, or a cache entry with a "
105 "top-level 'outcome' key)"
106 )
108 _coerce_agent_results(inner)
109 try:
110 outcome = outcome_from_dict(inner)
111 except (KeyError, TypeError, AttributeError, ValueError) as exc:
112 raise ReplayError(f"'{path}' holds a malformed outcome: {exc!r}") from exc
113 if not outcome.reviews:
114 raise ReplayError(f"'{path}' contains no reviews — nothing to replay")
115 return outcome
118def _coerce_agent_results(inner: dict) -> None:
119 """Coerce type-invalid AgentResult fields in place (untrusted file).
121 ``outcome_from_dict``/``cache._agent_result`` copy values without type
122 validation, so a hand-edited file with ``"output": null`` or a string
123 ``duration_s`` would pass loading and crash far later inside the render
124 loop with a raw traceback (review finding). Coerce the fields the render
125 path consumes: ``output`` to str, ``duration_s`` to float, ``ok`` to bool,
126 ``agent``/``vendor`` to str.
127 """
128 for key in ("reviews", "debate"):
129 items = inner.get(key)
130 if not isinstance(items, list):
131 continue
132 for item in items:
133 if isinstance(item, dict):
134 _coerce_one(item)
135 for key in ("synthesis", "verify"):
136 item = inner.get(key)
137 if isinstance(item, dict):
138 _coerce_one(item)
141def _coerce_one(item: dict) -> None:
142 item["agent"] = str(item.get("agent") or "")
143 item["vendor"] = str(item.get("vendor") or "")
144 item["ok"] = bool(item.get("ok"))
145 out = item.get("output")
146 item["output"] = out if isinstance(out, str) else ("" if out is None else str(out))
147 try:
148 item["duration_s"] = float(item.get("duration_s") or 0.0)
149 except (TypeError, ValueError):
150 item["duration_s"] = 0.0
153def replay_events(
154 outcome: JuryOutcome,
155) -> Iterator[tuple[str, AgentResult, int | None]]:
156 """Yield the ``(kind, result, round_no)`` sequence a live run would emit.
158 Mirrors the orchestrator's ``on_event`` stream: each review in panel order,
159 then the recorded debate round, then verify, then synthesis — phases absent
160 from the outcome are simply skipped, exactly as a live run that skipped
161 them. The outcome stores only the FINAL debate round (earlier rounds are
162 superseded, not serialized), numbered from ``rounds_executed`` — debate
163 rounds start at 2, review being round 1.
164 """
165 for r in outcome.reviews:
166 yield ("review", r, None)
167 if outcome.debate:
168 round_no = outcome.rounds_executed if outcome.rounds_executed >= 2 else 2
169 for r in outcome.debate:
170 yield ("debate", r, round_no)
171 if outcome.verify is not None:
172 yield ("verify", outcome.verify, None)
173 if outcome.synthesis is not None:
174 yield ("synthesis", outcome.synthesis, None)
177def replay_into(court, outcome: JuryOutcome, vote=None) -> None:
178 """Drive ``court`` (a theater ``Courtroom``-like object) from ``outcome``.
180 Uses exactly the API the live CLI path uses: ``open()``, ``step(kind,
181 result, round_no)`` per event, ``set_vote(vote)`` when a panel vote is
182 supplied, and ``close()`` (always — the terminal must be restored even if a
183 step raises).
184 """
185 court.open()
186 try:
187 for kind, result, round_no in replay_events(outcome):
188 court.step(kind, result, round_no)
189 if vote is not None:
190 court.set_vote(vote)
191 finally:
192 court.close()