Coverage for src/ai_jury/orchestrator.py: 99%
507 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-17 21:31 +0000
1"""Jury orchestration: review -> debate -> synthesis.
3The orchestrator owns the round structure and prompt assembly; adapters only run
4their CLI. Rounds run agents concurrently (thread pool) because each call is an
5independent, IO-bound subprocess.
6"""
8from __future__ import annotations
10import random
11import string
12import time
13from concurrent.futures import ThreadPoolExecutor
14from dataclasses import dataclass, field, replace
16from . import convergence, injection, largediff, prompts
17from .adapters import RETRYABLE_ERROR_CODES, Adapter, AgentResult, make_adapter
18from .config import JuryConfig
19from .consensus import FindingGroup, demote_local_only_groups, group_findings
20from .findings import (
21 Finding,
22 Verdict,
23 emitted_findings_block,
24 parse_findings,
25 parse_verdicts,
26)
27from .policy import ReviewPolicy, render_policy_section
28from .privilege import audit_privilege
29from .redaction import redact
32class RunBudget:
33 """Wall-clock budget for one jury run (issue #30).
35 Tracks the elapsed time since construction and derives the timeout to pass a
36 single agent call from the optional total-run and per-phase budgets. ``None``
37 for either budget means uncapped; when both are unset ``call_timeout``
38 returns ``None`` so adapters fall back to their own per-agent timeout and
39 behaviour is identical to having no budget at all.
40 """
42 def __init__(self, total_timeout: int | None, phase_timeout: int | None):
43 self.total = total_timeout
44 self.phase = phase_timeout
45 self._start = time.monotonic()
47 def elapsed(self) -> float:
48 return time.monotonic() - self._start
50 def remaining(self) -> float | None:
51 if self.total is None:
52 return None
53 return max(0.0, self.total - self.elapsed())
55 def expired(self) -> bool:
56 return self.total is not None and self.elapsed() >= self.total
58 def call_timeout(self) -> int | None:
59 """Per-call timeout: the min of the phase budget and remaining total.
61 The agent's own per-agent timeout is applied by the adapter (it takes the
62 min with this value), so it is not needed here. Returns ``None`` when
63 neither budget caps the call, leaving the adapter to use its configured
64 per-agent timeout.
65 """
66 caps: list[float] = []
67 if self.phase is not None:
68 caps.append(float(self.phase))
69 remaining = self.remaining()
70 if remaining is not None:
71 caps.append(remaining)
72 if not caps:
73 return None
74 return max(1, int(min(caps)))
77def _run_with_retry(
78 adapter: Adapter,
79 prompt: str,
80 phase: str,
81 budget: RunBudget,
82 retries: int,
83 log,
84) -> AgentResult:
85 """Run one agent for one phase, retrying transient failures (issue #30).
87 Retries only failures whose typed error code is in
88 ``RETRYABLE_ERROR_CODES`` (timeout/rate-limit/spawn), up to ``retries`` extra
89 attempts. A deterministic failure (auth, missing CLI, empty output, generic
90 nonzero exit) is returned immediately. The returned result's ``attempts``
91 records how many tries were made. Retrying stops early when the run budget is
92 exhausted so a retry never overruns the total timeout.
93 """
94 max_attempts = max(1, retries + 1)
95 result = adapter.run(prompt, phase=phase, timeout=budget.call_timeout())
96 attempts = 1
97 while (
98 not result.ok
99 and result.error_code in RETRYABLE_ERROR_CODES
100 and attempts < max_attempts
101 and not budget.expired()
102 ):
103 log(f"{adapter.name}: {phase} attempt {attempts} failed ({result.error_code}); retrying")
104 result = adapter.run(prompt, phase=phase, timeout=budget.call_timeout())
105 attempts += 1
106 result.attempts = attempts
107 return result
110def _order_by_agents(results: list[AgentResult], order: list[str]) -> list[AgentResult]:
111 """Reorder phase results into the configured/enabled agent order.
113 Round phases run agents concurrently (ThreadPoolExecutor.map), so the order
114 in which results arrive is not guaranteed across runs. The report and all
115 downstream consumers must NOT depend on thread-completion order, so we sort
116 every phase's results by each agent's index in ``order`` (the stable
117 enabled-agent list). Agents not present in ``order`` (should not happen)
118 sort to the end, preserving their relative arrival order as a stable
119 tiebreak so the sort is total and deterministic.
120 """
121 index = {name: i for i, name in enumerate(order)}
122 fallback = len(order)
123 return sorted(results, key=lambda r: index.get(r.agent, fallback))
126@dataclass
127class JuryOutcome:
128 reviews: list[AgentResult]
129 debate: list[AgentResult]
130 synthesis: AgentResult | None
131 chair: str
132 findings: list[Finding] = field(default_factory=list)
133 warnings: list[str] = field(default_factory=list)
134 groups: list[FindingGroup] = field(default_factory=list)
135 verify: AgentResult | None = None
136 verdicts: list[Verdict] = field(default_factory=list)
137 context_mode: str = "diff-only"
138 redact_secrets: bool = True
139 redaction_count: int = 0
140 injection_hits: list = field(default_factory=list)
141 # Execution/partial-result signals (issue #30): agents skipped because their
142 # CLI was unavailable (name, reason), and whether the run budget was
143 # exhausted before all phases completed.
144 skipped: list = field(default_factory=list)
145 budget_exhausted: bool = False
146 # Adaptive-rounds signals (issue #40): rounds actually executed and a short
147 # human-readable reason for why the debate ran / stopped.
148 rounds_executed: int = 1
149 stop_reason: str = ""
150 # Set when this outcome was served from the local result cache (issue #33),
151 # so the report/metadata can mark it as cached rather than freshly computed.
152 from_cache: bool = False
155def _run_phase(
156 adapters: list[Adapter],
157 prompt_for: dict[str, str],
158 phase: str,
159 parallel: bool,
160 *,
161 budget: RunBudget,
162 retries: int,
163 log,
164) -> list[AgentResult]:
165 def task(a: Adapter) -> AgentResult:
166 return _run_with_retry(a, prompt_for[a.name], phase, budget, retries, log)
168 if parallel and len(adapters) > 1:
169 with ThreadPoolExecutor(max_workers=len(adapters)) as pool:
170 return list(pool.map(task, adapters))
171 return [task(a) for a in adapters]
174def _others(reviews: list[AgentResult], me: str) -> str:
175 """Identity-labeled peer reviews (legacy path; ``anonymize_debate = false``).
177 Renders each *other* reviewer's round-1 output with its real agent/vendor
178 identity in the stable enabled-agent order. This is the pre-#37 behaviour and
179 leaks both identity and position; the anonymizing path below is the default.
180 """
181 chunks = [
182 f"### {r.agent} ({r.vendor})\n{r.output}"
183 for r in reviews
184 if r.agent != me and r.ok and r.output
185 ]
186 return "\n\n".join(chunks) if chunks else "_(no other reviews available)_"
189def _anon_label(i: int) -> str:
190 """Stable anonymous reviewer label: 0->'A', 1->'B', ... 26->'AA'."""
191 letters = string.ascii_uppercase
192 label = ""
193 i += 1
194 while i > 0:
195 i, rem = divmod(i - 1, 26)
196 label = letters[rem] + label
197 return label
200def _anonymize_peers(
201 reviews: list[AgentResult], me: str, rng: random.Random
202) -> tuple[str, dict[str, str]]:
203 """Chatham House peer view for a debater (#37).
205 Returns ``(prompt_text, label_to_agent)`` where the prompt text renders each
206 *other* successful reviewer's round-1 output under an anonymous
207 ``### Reviewer A`` / ``### Reviewer B`` heading — NO vendor or agent name.
208 The debater's OWN review is excluded (it is passed separately as
209 ``own_review``). Presentation order is shuffled DETERMINISTICALLY using the
210 shared run RNG so neither identity nor position is a stable signal; the same
211 seed yields the same order, different seeds may differ.
213 ``label_to_agent`` keeps the anonymous-label -> real-agent mapping internal so
214 callers can still recover authorship (the report attributes by real name).
215 """
216 peers = [r for r in reviews if r.agent != me and r.ok and r.output]
217 if not peers:
218 return "_(no other reviews available)_", {}
219 # Deterministic per-debater shuffle from the shared run RNG. We shuffle a
220 # copy so the caller's review list (used elsewhere) is untouched.
221 order = list(peers)
222 rng.shuffle(order)
223 chunks: list[str] = []
224 label_to_agent: dict[str, str] = {}
225 for i, r in enumerate(order):
226 label = f"Reviewer {_anon_label(i)}"
227 label_to_agent[label] = r.agent
228 chunks.append(f"### {label}\n{r.output}")
229 return "\n\n".join(chunks), label_to_agent
232def _debate_round(
233 debaters: list[Adapter],
234 reviews: list[AgentResult],
235 diff: str,
236 config: JuryConfig,
237 run_rng: random.Random,
238 agent_order: list[str],
239 prior: list[AgentResult],
240 budget: RunBudget,
241 retries: int,
242 log,
243 round_no: int,
244 template: str = prompts.DEBATE,
245) -> list[AgentResult]:
246 """Run one debate round and return its results in stable agent order.
248 ``prior`` holds the previous round's debate outputs (empty for the first
249 debate round); when present they are appended to each debater's prompt as a
250 "prior debate" addendum so later rounds in an adaptive run (issue #40) build
251 on, rather than repeat, earlier cross-examination. The peer-review anonymizing
252 path (#37) is preserved unchanged.
253 """
254 log(f"round {round_no}: {len(debaters)} agents cross-examining")
255 own = {r.agent: r.output for r in reviews if r.ok}
256 # Prior-round debate output quotes attacker-controlled diff text, so it is
257 # untrusted: neutralize sentinels (issue #316/L-1) before it is fenced and
258 # appended below, matching every other peer-output slot.
259 prior_txt = prompts.neutralize_sentinels(
260 "\n\n".join(f"### {r.agent}\n{r.output}" for r in prior if r.ok and r.output)
261 )
262 debate_prompt: dict[str, str] = {}
263 for a in debaters:
264 if config.anonymize_debate:
265 # Per-debater deterministic shuffle: derive a child RNG from the
266 # shared run RNG so each debater gets an independent but reproducible
267 # peer ordering (same seed -> same order).
268 peer_rng = random.Random(run_rng.random())
269 other_reviews, _label_map = _anonymize_peers(reviews, a.name, peer_rng)
270 else:
271 other_reviews = _others(reviews, a.name)
272 text = template.format(
273 name=a.name,
274 diff=prompts.neutralize_sentinels(diff),
275 own_review=prompts.neutralize_sentinels(
276 own.get(a.name, "_(your review was unavailable)_")
277 ),
278 other_reviews=prompts.neutralize_sentinels(other_reviews),
279 notice=prompts._UNTRUSTED_NOTICE,
280 )
281 if prior_txt:
282 text += (
283 "\n\n=== PRIOR DEBATE (earlier round) ===\n"
284 "Build on this; do not just repeat it. Only keep a DISPUTE or "
285 "MISSED item if it is still unresolved.\n\n"
286 "<<<UNTRUSTED_REVIEW\n" + prior_txt + "\nUNTRUSTED_REVIEW>>>\n"
287 )
288 debate_prompt[a.name] = text
289 results = _run_phase(
290 debaters,
291 debate_prompt,
292 "debate",
293 config.parallel,
294 budget=budget,
295 retries=retries,
296 log=log,
297 )
298 # Same stable-ordering guarantee as round 1: independent of thread-pool
299 # completion order.
300 return _order_by_agents(results, agent_order)
303def run_jury(
304 config: JuryConfig,
305 diff: str,
306 *,
307 context: str = "",
308 mock: bool = False,
309 strict: bool = False,
310 seed: int | None = None,
311 policy: ReviewPolicy | None = None,
312 log=lambda _msg: None,
313 budget: RunBudget | None = None,
314 on_event=None,
315 mode: str = "code",
316) -> JuryOutcome:
317 # Jury mode (issue #221): "code" (default) reviews a diff with the code-review
318 # rubric; "issue" reviews a GitHub issue's prose for completeness/clarity.
319 # Only the prompt TEMPLATES differ — the round structure, consensus, voting,
320 # verification, ordering, and determinism are identical. ``tmpl`` selects the
321 # four phase templates; each is threaded into the phase that uses it so the
322 # call sites are otherwise unchanged.
323 tmpl = prompts.for_mode(mode)
324 # Live play-by-play hook (issue #210): an optional callback fired after each
325 # phase result is produced — ``on_event(kind, result, round_no=None)`` with
326 # kind in {"review", "debate", "verify", "synthesis"}. It lets a caller stream
327 # the deliberation as it happens (CLI ``--live``) without the orchestrator
328 # doing any I/O itself. Fired in stable per-phase order (not thread-completion
329 # order) so the event sequence is deterministic. Defaults to a no-op.
330 emit = on_event or (lambda *_a, **_k: None)
331 # Repository review policy (optional, #8): maintainer-authored, TRUSTED
332 # content rendered into each REVIEW prompt in a clearly separated section.
333 # When ``policy`` is None a sentinel placeholder is used, so the prompt is
334 # unchanged except for that section. The policy is distinct from the
335 # agent-runtime ``config`` and never enters the untrusted diff/context fences.
336 policy_section = render_policy_section(policy)
337 # Run reproducibility: a single shared RNG seeds every randomized
338 # orchestration decision (future: anonymized-rebuttal order, rotating
339 # chair, tie-breaks). The seed comes from the explicit ``seed`` argument if
340 # given, else from ``config.seed``. We construct a dedicated
341 # ``random.Random`` instance rather than touching the global ``random``
342 # module so seeding a jury run never perturbs unrelated global state.
343 # When the seed is None the RNG is unseeded (still deterministic
344 # orchestration; randomness, if any, is just not reproducible run-to-run).
345 # LLM output itself is never made deterministic by this — only the
346 # orchestration around it. ``run_rng`` is the shared run RNG: pass it to
347 # any feature that needs reproducible randomness instead of using ``random``.
348 run_seed = seed if seed is not None else config.seed
349 run_rng = random.Random(run_seed) # shared run RNG (see docstring)
351 # Run budget (issue #30): a single wall-clock budget threaded through every
352 # phase. Defaults (both None) leave behaviour identical to no budget, with
353 # each agent bounded only by its own per-agent timeout. ``retries`` is the
354 # number of extra attempts for transient (retryable) failures. A caller may
355 # pass a SHARED budget so ``total_timeout`` spans a whole chunked review
356 # rather than resetting per chunk (issue #31 / review finding).
357 if budget is None:
358 budget = RunBudget(config.total_timeout, config.phase_timeout)
359 retries = config.retries
361 # Context policy: diff-only sends only the diff; expanded includes context.
362 ctx_cfg = getattr(config, "context", None)
363 context_mode = getattr(ctx_cfg, "mode", "diff-only") if ctx_cfg else "diff-only"
364 redact_on = getattr(ctx_cfg, "redact_secrets", True) if ctx_cfg else True
365 if context_mode == "diff-only":
366 context = ""
367 redaction_count = 0
368 if redact_on:
369 diff, _n1 = redact(diff)
370 context, _n2 = redact(context)
371 redaction_count = _n1 + _n2
372 if redaction_count:
373 log(f"redacted {redaction_count} secret(s) before sending to agents")
375 # Prompt-injection heuristic (OWASP LLM01): scan untrusted diff/context for
376 # patterns that try to override instructions, then SURFACE them as a synthetic
377 # finding/warning. We never act on them; the CI gate is derived from
378 # structured consensus (see ci.evaluate_ci), so an injected "APPROVE"
379 # cannot flip the verdict.
380 injection_hits = injection.scan_inputs(diff, context)
381 injection_findings: list[Finding] = []
382 if injection_hits:
383 log(f"prompt-injection heuristic: {len(injection_hits)} suspicious pattern(s) flagged")
384 syn = injection.hits_to_finding(injection_hits)
385 if syn is not None: 385 ↛ 390line 385 didn't jump to line 390 because the condition on line 385 was always true
386 injection_findings.append(syn)
388 # Least-privilege audit: warn when a configured agent could perform
389 # write/tool actions while reviewing attacker-controlled content.
390 privilege_warnings = audit_privilege(config.enabled_agents)
391 for w in privilege_warnings:
392 log(f"least-privilege warning: {w}")
393 if strict and privilege_warnings:
394 raise RuntimeError(
395 "least-privilege check failed (--strict): " + "; ".join(privilege_warnings)
396 )
398 specs = config.enabled_agents
399 adapters = [make_adapter(s, mock=mock) for s in specs]
401 # Filter to available agents (unless strict, where a missing CLI is fatal).
402 # Skipped agents are recorded (name, reason) so the report can state exactly
403 # which agents never ran — part of the partial-result policy (issue #30).
404 usable: list[Adapter] = []
405 skipped: list[tuple[str, str]] = []
406 for a in adapters:
407 if a.available():
408 usable.append(a)
409 elif strict:
410 raise RuntimeError(f"agent '{a.name}' CLI not available: {a.spec.command}")
411 else:
412 reason = f"CLI not found ({a.spec.command})"
413 log(f"skipping '{a.name}': {reason}")
414 skipped.append((a.name, reason))
415 if not usable:
416 raise RuntimeError("no usable agents — install at least one agent CLI or use --mock")
418 usable_names = [a.name for a in usable]
420 # Round 1: independent reviews.
421 log(f"round 1: {len(usable)} agents reviewing")
422 review_prompt = {
423 a.name: tmpl["review"].format(
424 name=a.name,
425 context=prompts.neutralize_sentinels(context or "_(none)_"),
426 diff=prompts.neutralize_sentinels(diff),
427 policy=policy_section,
428 notice=prompts._UNTRUSTED_NOTICE,
429 )
430 for a in usable
431 }
432 reviews = _run_phase(
433 usable,
434 review_prompt,
435 "review",
436 config.parallel,
437 budget=budget,
438 retries=retries,
439 log=log,
440 )
441 # Stable ordering: the thread pool can return results in any completion
442 # order. Reorder to the enabled-agent order so the report (and every
443 # downstream consumer) is independent of which thread finished first.
444 agent_order = [a.name for a in usable]
445 reviews = _order_by_agents(reviews, agent_order)
447 # Parse structured findings from each successful review and aggregate them.
448 # Seed with the synthetic injection finding/warnings so they surface in the
449 # report and outcome.warnings without ever influencing agent behaviour.
450 all_findings: list[Finding] = list(injection_findings)
451 all_warnings: list[str] = injection.hits_to_warnings(injection_hits)
452 all_warnings.extend(privilege_warnings)
453 for r in reviews:
454 if not r.ok:
455 continue
456 found, warns = parse_findings(r.output, r.agent)
457 r.findings = found
458 r.warnings = warns
459 r.structured = emitted_findings_block(r.output)
460 all_findings.extend(found)
461 all_warnings.extend(warns)
463 # Stream round-1 reviews as they're now finalized (stable order).
464 for r in reviews:
465 emit("review", r)
467 # Deterministic consensus grouping across reviewers.
468 groups = group_findings(all_findings, len(reviews))
470 # Names of agents whose round-1 review succeeded — the chair resolver uses
471 # this to (optionally) prefer a non-reviewer chair (#38).
472 reviewer_names = [r.agent for r in reviews if r.ok]
474 # Resolve the chair ONCE for the whole run so verify and synthesis use the
475 # SAME chair. ``chair = "rotate"`` and prefer-non-reviewer both consume the
476 # shared run RNG / reviewer info, so resolving once (rather than recomputing
477 # per phase) is what keeps a rotating chair stable within a run (#38).
478 chair_name = resolve_chair(config, usable_names, reviewer_names, run_rng)
480 # Round 2+: debate. Only agents whose round-1 review succeeded participate.
481 # Two modes (issue #40):
482 # - fixed (early_stop = false): honour ``rounds`` exactly — run one debate
483 # round iff rounds >= 2. Reproducible fixed-N behaviour for benchmarking.
484 # - adaptive (early_stop = true): skip the debate when round-1 reviewers
485 # already agree, otherwise run debate up to ``max_rounds`` rounds and stop
486 # as soon as a round resolves all disputes.
487 debate: list[AgentResult] = []
488 rounds_executed = 1
489 stop_reason = ""
490 budget_exhausted = False
491 debaters = [a for a in usable if any(r.agent == a.name and r.ok for r in reviews)]
492 can_debate = len(debaters) >= 2
494 if config.early_stop:
495 max_rounds = config.effective_max_rounds
496 if not can_debate:
497 stop_reason = "stopped after round 1: need >=2 successful reviews to debate"
498 log(stop_reason)
499 elif max_rounds < 2:
500 stop_reason = "stopped after round 1: max_rounds < 2"
501 log(stop_reason)
502 else:
503 converged, why = convergence.review_convergence(groups, len(reviews))
504 if converged:
505 stop_reason = f"early stop after round 1: {why}"
506 log(stop_reason)
507 else:
508 log(f"early stop active: {why}; running debate up to {max_rounds} round(s)")
509 prior: list[AgentResult] = []
510 round_no = 1
511 while round_no < max_rounds:
512 if budget.expired():
513 budget_exhausted = True
514 stop_reason = f"stopped at round {rounds_executed}: run budget exhausted"
515 log(stop_reason)
516 break
517 round_no += 1
518 debate = _debate_round(
519 debaters,
520 reviews,
521 diff,
522 config,
523 run_rng,
524 agent_order,
525 prior,
526 budget,
527 retries,
528 log,
529 round_no,
530 template=tmpl["debate"],
531 )
532 rounds_executed = round_no
533 for r in debate:
534 emit("debate", r, round_no)
535 dconv, dwhy = convergence.debate_convergence(debate)
536 if dconv:
537 stop_reason = f"converged after round {round_no}: {dwhy}"
538 log(stop_reason)
539 break
540 prior = debate
541 stop_reason = f"ran {round_no} rounds: {dwhy}"
542 else:
543 stop_reason = stop_reason or (
544 f"reached max_rounds ({max_rounds}) with disagreement remaining"
545 )
546 else:
547 # Fixed-N: exactly the historical behaviour.
548 if config.rounds >= 2 and can_debate:
549 if budget.expired():
550 budget_exhausted = True
551 stop_reason = "round 2 skipped: run budget exhausted"
552 log(stop_reason)
553 else:
554 debate = _debate_round(
555 debaters,
556 reviews,
557 diff,
558 config,
559 run_rng,
560 agent_order,
561 [],
562 budget,
563 retries,
564 log,
565 2,
566 template=tmpl["debate"],
567 )
568 rounds_executed = 2
569 for r in debate:
570 emit("debate", r, 2)
571 elif config.rounds >= 2:
572 stop_reason = "round 2 skipped: need >=2 successful reviews to debate"
573 log(stop_reason)
574 else:
575 stop_reason = "single round (rounds = 1)"
577 # Verification: the chair judges candidate findings to reduce false
578 # positives. Skipped when the run budget is exhausted (issue #30) so a
579 # partial run still returns what completed instead of overrunning.
580 verify_result: AgentResult | None = None
581 verdicts: list[Verdict] = []
582 if config.verify:
583 if budget.expired():
584 budget_exhausted = True
585 msg = "verification skipped: run budget exhausted"
586 log(msg)
587 all_warnings.append(msg)
588 else:
589 verify_result, verdicts, verify_warnings = _verify(
590 chair_name,
591 usable,
592 all_findings,
593 diff,
594 context,
595 budget,
596 retries,
597 log,
598 template=tmpl["verify"],
599 )
600 all_warnings.extend(verify_warnings)
601 _apply_verdicts(groups, verdicts)
602 if verify_result is not None:
603 emit("verify", verify_result)
605 # Local-only demotion (issue #442) runs AFTER verification, never before:
606 # _reject_targets' member-tier guard (orchestrator._reject_targets) assumes
607 # group.severity == max(member severities) to decide whether a rejecting
608 # verdict may suppress the whole group. Demoting group.severity earlier
609 # would desync it from that invariant and could let a verdict aimed at a
610 # minor local-only duplicate collateral-reject a genuinely critical,
611 # never-verified co-located finding merged into the same group.
612 if config.demote_local_only:
613 vendor_by_reviewer = {a.name: a.vendor for a in config.agents}
614 demote_local_only_groups(groups, vendor_by_reviewer)
616 # Synthesis: the chair consolidates. When the resolved chair is ALSO a
617 # round-1 reviewer, feed it an anonymized view of the reviews (#38 guardrail)
618 # so it cannot preferentially weight its own findings; the report still
619 # attributes by real name because it renders the real outcome data, not this
620 # synthesis prompt.
621 synthesis: AgentResult | None = None
622 if budget.expired():
623 budget_exhausted = True
624 msg = "synthesis skipped: run budget exhausted"
625 log(msg)
626 if msg not in all_warnings: 626 ↛ 648line 626 didn't jump to line 648 because the condition on line 626 was always true
627 all_warnings.append(msg)
628 else:
629 chair_is_reviewer = chair_name in reviewer_names
630 anonymize_synthesis = config.anonymize_debate and chair_is_reviewer
631 synthesis = _synthesize(
632 chair_name,
633 usable,
634 reviews,
635 debate,
636 diff,
637 budget,
638 retries,
639 log,
640 verdicts=verdicts,
641 anonymize_reviews=anonymize_synthesis,
642 rng=run_rng,
643 template=tmpl["synthesis"],
644 )
645 if synthesis is not None:
646 emit("synthesis", synthesis)
648 return JuryOutcome(
649 reviews=reviews,
650 debate=debate,
651 synthesis=synthesis,
652 chair=chair_name,
653 findings=all_findings,
654 warnings=all_warnings,
655 groups=groups,
656 verify=verify_result,
657 verdicts=verdicts,
658 context_mode=context_mode,
659 redact_secrets=redact_on,
660 redaction_count=redaction_count,
661 injection_hits=injection_hits,
662 skipped=skipped,
663 budget_exhausted=budget_exhausted,
664 rounds_executed=rounds_executed,
665 stop_reason=stop_reason,
666 )
669def resolve_chair(
670 config: JuryConfig,
671 usable: list[str],
672 reviewers: list[str],
673 rng: random.Random,
674) -> str:
675 """Resolve the chair for a run as a PURE function of its inputs (#38).
677 Precedence:
678 1. ``chair = "rotate"`` — pick deterministically from the usable agents
679 using the shared run ``rng``. Same seed -> same chair; different seeds
680 may differ. Falls back to the first usable agent when none are usable.
681 2. An explicit ``config.chair`` that names a usable agent — honoured as-is
682 (an operator-chosen chair always wins).
683 3. ``prefer_non_reviewer_chair`` — when set and a usable agent that was NOT
684 a successful round-1 reviewer exists, prefer the first such agent
685 (neutral chair). This only applies when the configured chair is not
686 itself a usable agent.
687 4. Fallback to the first usable agent (legacy behaviour).
689 Keeping this pure (no Adapter objects, no I/O) makes it directly
690 unit-testable and guarantees ``_verify`` and ``_synthesize`` agree because
691 the caller resolves it ONCE and threads the result through both.
692 """
693 if not usable:
694 return config.chair
695 names = set(usable)
697 if config.chair == "rotate":
698 # Deterministic rotation: sort for a stable candidate order independent
699 # of dict/thread ordering, then index with the shared run RNG. Sorting
700 # the candidate list (not iterating the set) makes the pick a pure
701 # function of (seed, usable-name set): same seed + same agents -> same
702 # chair, regardless of RNG-consumption order elsewhere.
703 candidates = sorted(names)
704 return candidates[rng.randrange(len(candidates))]
706 if config.chair in names:
707 return config.chair
709 if config.prefer_non_reviewer_chair:
710 reviewer_set = set(reviewers)
711 non_reviewers = [n for n in usable if n not in reviewer_set]
712 if non_reviewers:
713 return non_reviewers[0]
715 return usable[0]
718def _format_findings_for_verify(findings: list[Finding]) -> str:
719 """Render candidate findings for the chair's verification prompt.
721 Reviewer identity is omitted (#250) so the chair can't favour its own
722 findings while judging them — parity with the #37/#38 anonymization.
723 Verdicts match back by file/line/claim, so dropping it is safe.
724 """
725 if not findings:
726 return "_(no candidate findings)_"
727 lines = []
728 for f in findings:
729 loc = f.file or "?"
730 if f.line is not None:
731 loc = f"{loc}:{f.line}"
732 lines.append(f"- [{f.severity}] {loc} — {f.claim}")
733 return "\n".join(lines)
736def _format_verdicts(verdicts: list[Verdict]) -> str:
737 if not verdicts:
738 return "_(no verification verdicts)_"
739 lines = []
740 for v in verdicts:
741 loc = v.file or "?"
742 if v.line is not None:
743 loc = f"{loc}:{v.line}"
744 lines.append(f"- [{v.status}] {loc} — {v.claim}: {v.reasoning}")
745 return "\n".join(lines)
748def _verify(
749 chair_name,
750 usable,
751 findings,
752 diff,
753 context,
754 budget,
755 retries,
756 log,
757 template=prompts.VERIFY,
758) -> tuple[AgentResult | None, list[Verdict], list[str]]:
759 chair = next((a for a in usable if a.name == chair_name), None)
760 if chair is None:
761 return None, [], []
762 log(f"verification: chair '{chair_name}' judging {len(findings)} candidate findings")
763 prompt = template.format(
764 diff=prompts.neutralize_sentinels(diff),
765 findings=prompts.neutralize_sentinels(_format_findings_for_verify(findings)),
766 context=prompts.neutralize_sentinels(context or "_(none)_"),
767 notice=prompts._UNTRUSTED_NOTICE,
768 )
769 result = _run_with_retry(chair, prompt, "verify", budget, retries, log)
770 if not result.ok:
771 return result, [], [f"verification failed: {result.error}"]
772 verdicts, warnings = parse_verdicts(result.output, chair_name)
773 return result, verdicts, warnings
776def _verdict_matches_group(verdict: Verdict, group: FindingGroup) -> bool:
777 from .consensus import _normalize_claim, _normalize_path
779 rep = group.representative
780 # Case-EXACT path match (fold_case=False): on a case-sensitive filesystem
781 # ``Config.py`` != ``config.py``, so a verdict must not reject a finding it
782 # only case-collapses onto (audit 2026-06-13 r6/M).
783 if _normalize_path(verdict.file, fold_case=False) != _normalize_path(rep.file, fold_case=False):
784 return False
785 if verdict.line is not None and rep.line is not None and abs(verdict.line - rep.line) > 3:
786 return False
787 v_claim = _normalize_claim(verdict.claim)
788 r_claim = _normalize_claim(rep.claim)
789 if not v_claim:
790 # An empty verdict claim is allowed to match the finding *at this
791 # location* (the verifier may omit the claim and refer to it by
792 # position). But a verdict with NEITHER a claim NOR a line has no
793 # location precision at all: it would otherwise match — and, when
794 # ``unsupported``, REJECT — every finding group in the file, including
795 # unrelated criticals, flipping the CI gate from FAIL to PASS. Such a
796 # claim-less, line-less verdict is a file-wide wildcard and must not
797 # match (security audit 2026-06-13 r6/M). Require a concrete line that
798 # actually pins the finding before honoring an empty-claim match.
799 return verdict.line is not None and rep.line is not None
800 if v_claim == r_claim:
801 return True
802 v_tokens, r_tokens = set(v_claim.split()), set(r_claim.split())
803 if not v_tokens or not r_tokens:
804 return False
805 inter = len(v_tokens & r_tokens)
806 union = len(v_tokens) + len(r_tokens) - inter
807 return (inter / union if union else 0.0) >= 0.5
810def _claim_sim(a_claim: str, b_claim: str) -> float:
811 """Token-set similarity between two claims: 1.0 exact, else Jaccard, 0.0 if
812 either side is empty."""
813 from .consensus import _normalize_claim
815 a = _normalize_claim(a_claim)
816 b = _normalize_claim(b_claim)
817 if not a or not b:
818 return 0.0
819 if a == b:
820 return 1.0
821 at, bt = set(a.split()), set(b.split())
822 inter = len(at & bt)
823 union = len(at) + len(bt) - inter
824 return (inter / union) if union else 0.0
827# Verdict statuses that move a finding into a non-blocking bucket (suppress it).
828_REJECTING_STATUSES = frozenset({"unsupported", "needs_human_decision"})
829# Apply most-blocking statuses first so a contradictory verdict pair on one
830# finding is fail-closed: a `verified` (blocking) judgement is recorded before
831# any `unsupported`/`needs_human_decision` and cannot then be flipped to
832# non-blocking by verdict array ordering (audit 2026-06-13 r8/M).
833_STATUS_PRIORITY = {"verified": 0, "needs_human_decision": 1, "unsupported": 2}
834# Minimum claim similarity for a verdict to be considered "about" a finding at
835# all. The PRIMARY defence against a verdict dismissing a co-located *distinct*
836# finding is that a rejection attaches to AT MOST the single best-matching group
837# (`_best_reject_target`), so a verdict whose claim copies a benign neighbour
838# routes to that neighbour, not to the co-located critical (audit 2026-06-13
839# r8/M). The threshold stays moderate so the verifier's legitimate paraphrased
840# rejections (it drops the reviewer-name prefix etc.) still apply.
841_REJECT_CLAIM_THRESHOLD = 0.5
844def _reject_targets(verdict: Verdict, groups: list[FindingGroup]) -> list[FindingGroup]:
845 """Un-statused groups a rejecting verdict may suppress (fail-closed, r7/r8).
847 Defences (each closes a distinct collateral-rejection vector found across
848 audit rounds 6-9):
850 0. **Line required.** A rejecting verdict must pin a concrete line. A
851 line-less verdict is too imprecise to safely suppress a finding and would
852 act as a file-wide-by-claim wildcard (audit r9/M, the claim-ful
853 counterpart of the round-6 line-less-wildcard fix).
854 1. **Member-tier guard.** A group may merge findings of different severities
855 (consensus keeps the max). A verdict is "about" the member whose claim it
856 best matches; if that member is *less severe* than the group's max, the
857 verdict is dismissing a lesser co-located finding and must NOT suppress
858 the (e.g. critical) group.
859 2. **Best-tier only.** Across candidate groups, suppress only those at the
860 highest match similarity — a verdict copying a benign neighbour rejects
861 that neighbour (and its duplicate phrasings, which tie) but not a
862 separate, less-similar critical group.
863 3. **Least-severe within a tie.** If the best-similarity tier still spans
864 severities (an exact `_claim_sim` tie between a critical and a benign
865 decoy), suppress only the *least*-severe groups — a tie must never drag a
866 critical down alongside a decoy (audit r9/M).
867 """
868 from .findings import SEVERITY_ORDER
870 if verdict.line is None:
871 return []
872 scored: list[tuple[float, FindingGroup]] = []
873 for group in groups:
874 if group.status:
875 continue
876 if not _verdict_matches_group(verdict, group):
877 continue
878 members = getattr(group, "members", None) or [group.representative]
879 best_sim, best_member = max(
880 ((_claim_sim(verdict.claim, m.claim), m) for m in members),
881 key=lambda t: t[0],
882 )
883 if best_sim < _REJECT_CLAIM_THRESHOLD:
884 continue
885 # Member-tier guard: refuse if the verdict best-names a member less
886 # severe than the group's max severity (lower rank = more severe).
887 if SEVERITY_ORDER.get(best_member.severity, 99) > SEVERITY_ORDER.get(group.severity, 99):
888 continue
889 scored.append((best_sim, group))
890 if not scored:
891 return []
892 best = max(sim for sim, _ in scored)
893 tier = [(sim, group) for sim, group in scored if sim >= best]
894 # Within the top-similarity tier, keep only the least-severe groups.
895 least_rank = max(SEVERITY_ORDER.get(g.severity, 99) for _, g in tier)
896 return [g for _, g in tier if SEVERITY_ORDER.get(g.severity, 99) == least_rank]
899def _apply_verdicts(groups: list[FindingGroup], verdicts: list[Verdict]) -> None:
900 """Attach verification statuses to consensus groups.
902 unsupported -> bucket 'rejected'; needs_human_decision -> bucket 'disputed';
903 verified -> status recorded, bucket unchanged.
904 """
905 # Stable-sort by blocking priority so contradictions resolve fail-closed.
906 for verdict in sorted(verdicts, key=lambda v: _STATUS_PRIORITY.get(v.status, 3)):
907 if verdict.status in _REJECTING_STATUSES:
908 # Suppress only the best-similarity tier this verdict names — never
909 # collaterally a co-located, less-similar distinct finding.
910 bucket = "rejected" if verdict.status == "unsupported" else "disputed"
911 for target in _reject_targets(verdict, groups):
912 target.status = verdict.status
913 target.status_reasoning = verdict.reasoning
914 target.bucket = bucket
915 continue
916 # A verifying (non-suppressing) verdict may attach to every matching
917 # group: when reviewers phrase the same issue differently it can land in
918 # more than one group, and all should carry the judgement.
919 for group in groups:
920 if group.status:
921 continue
922 if _verdict_matches_group(verdict, group):
923 group.status = verdict.status
924 group.status_reasoning = verdict.reasoning
927def _synthesize(
928 chair_name,
929 usable,
930 reviews,
931 debate,
932 diff,
933 budget,
934 retries,
935 log,
936 verdicts=None,
937 anonymize_reviews=False,
938 rng=None,
939 template=prompts.SYNTHESIS,
940) -> AgentResult | None:
941 chair = next((a for a in usable if a.name == chair_name), None)
942 if chair is None:
943 return None
944 log(f"synthesis: chair '{chair_name}' consolidating verdict")
945 if anonymize_reviews:
946 # Chair self-preference guardrail (#38): present round-1 reviews to the
947 # chair under anonymous labels (no agent/vendor identity, no stable
948 # order) so it cannot tell which review is "its own". Uses the shared run
949 # RNG for deterministic-but-unstable ordering. ``me=None`` keeps ALL
950 # reviews (we are not excluding a debater here, only stripping identity).
951 peer_rng = random.Random(rng.random()) if rng is not None else random.Random()
952 reviews_txt, _label_map = _anonymize_peers(reviews, None, peer_rng)
953 else:
954 reviews_txt = (
955 "\n\n".join(
956 f"### {r.agent} ({r.vendor})\n{r.output}" for r in reviews if r.ok and r.output
957 )
958 or "_(no reviews)_"
959 )
960 debate_txt = (
961 "\n\n".join(f"### {r.agent}\n{r.output}" for r in debate if r.ok and r.output)
962 or "_(no debate round)_"
963 )
964 prompt = template.format(
965 diff=prompts.neutralize_sentinels(diff),
966 reviews=prompts.neutralize_sentinels(reviews_txt),
967 debate=prompts.neutralize_sentinels(debate_txt),
968 notice=prompts._UNTRUSTED_NOTICE,
969 )
970 if verdicts:
971 # The verdicts quote candidate findings, which transitively quote
972 # untrusted diff text (issue v1.5.0/M-1: this addendum was the one slot
973 # the #316/L-1 fix missed). Fence + neutralize it like every other
974 # peer-output slot so an embedded closing token can't break out.
975 prompt += (
976 "\n\n=== VERIFICATION VERDICTS (may quote UNTRUSTED text) ===\n"
977 "<<<UNTRUSTED_FINDINGS\n"
978 + prompts.neutralize_sentinels(_format_verdicts(verdicts))
979 + "\nUNTRUSTED_FINDINGS>>>\n"
980 )
981 return _run_with_retry(chair, prompt, "synthesis", budget, retries, log)
984def _merge_results_by_agent(phase_lists: list[list[AgentResult]]) -> list[AgentResult]:
985 """Merge per-chunk results for the same agent into one result (issue #31).
987 Outputs are concatenated under per-chunk headers, durations summed, ``ok`` is
988 true if the agent succeeded on any chunk, and ``attempts`` keeps the max so a
989 retried chunk is still visible. Agent order follows first appearance.
990 """
991 order: list[str] = []
992 by_agent: dict[str, list[AgentResult]] = {}
993 for lst in phase_lists:
994 for r in lst:
995 if r.agent not in by_agent:
996 by_agent[r.agent] = []
997 order.append(r.agent)
998 by_agent[r.agent].append(r)
1000 merged: list[AgentResult] = []
1001 for name in order:
1002 parts = by_agent[name]
1004 # bolt: Consolidate multiple metrics (ok, body, total_duration, max_attempts)
1005 # into a single-pass O(N) explicit loop to bypass multiple generator instantiations
1006 ok = False
1007 body_parts = []
1008 first_err = None
1009 total_duration = 0.0
1010 max_attempts = 0
1012 for i, p in enumerate(parts, 1):
1013 if p.ok:
1014 ok = True
1015 if p.output:
1016 body_parts.append(f"#### chunk {i}\n{p.output}")
1017 elif first_err is None:
1018 first_err = p
1020 total_duration += p.duration_s
1021 if p.attempts > max_attempts:
1022 max_attempts = p.attempts
1024 body = "\n\n".join(body_parts)
1026 merged.append(
1027 AgentResult(
1028 name,
1029 parts[0].vendor,
1030 ok,
1031 body,
1032 round(total_duration, 3),
1033 error=None if ok else (first_err.error if first_err else None),
1034 error_code=None if ok else (first_err.error_code if first_err else None),
1035 attempts=max_attempts,
1036 )
1037 )
1038 return merged
1041def _combine_chair_results(results: list[AgentResult], chair: str) -> AgentResult | None:
1042 """Combine per-chunk chair results (verify/synthesis) into one labelled result."""
1043 ok_parts = [r for r in results if r.ok and r.output]
1044 if not ok_parts:
1045 return results[0] if results else None
1046 vendor = ok_parts[0].vendor
1048 # bolt: Consolidate body text concatenation and duration sum into a single-pass O(N) loop
1049 body_parts = []
1050 total_duration = 0.0
1051 for i, r in enumerate(ok_parts, 1):
1052 body_parts.append(f"### chunk {i}\n{r.output}")
1053 total_duration += r.duration_s
1055 body = "\n\n".join(body_parts)
1056 return AgentResult(chair, vendor, True, body, round(total_duration, 3))
1059def _merge_chunk_outcomes(outcomes: list[JuryOutcome], config: JuryConfig) -> JuryOutcome:
1060 """Fold per-chunk outcomes (issue #31) into one renderable JuryOutcome.
1062 Findings are unioned and re-grouped across all chunks so the consensus view
1063 is global; verdicts are re-applied to the merged groups. Review/debate/chair
1064 outputs are merged per agent with chunk labels so the report stays coherent.
1065 """
1066 if len(outcomes) == 1:
1067 return outcomes[0]
1068 base = outcomes[0]
1070 reviews = _merge_results_by_agent([o.reviews for o in outcomes])
1071 debate = (
1072 _merge_results_by_agent([o.debate for o in outcomes])
1073 if any(o.debate for o in outcomes)
1074 else []
1075 )
1076 findings = [f for o in outcomes for f in o.findings]
1077 groups = group_findings(findings, len(reviews))
1078 verdicts = [v for o in outcomes for v in o.verdicts]
1079 # Scope each chunk's verdicts to that chunk's own findings. A verdict is
1080 # produced while verifying ONE chunk (whose prompt held only that chunk's
1081 # findings, but whose attacker-controlled diff text could steer it); after
1082 # the global merge an unscoped verdict could reject a *different* chunk's
1083 # structured critical and flip the CI gate (audit 2026-06-13 r7/M). Chunks
1084 # are file-disjoint, so apply each chunk's verdicts only to groups whose
1085 # location is one of that chunk's files.
1086 from .consensus import _normalize_path
1088 for o in outcomes:
1089 chunk_files = {_normalize_path(f.file, fold_case=False) for f in o.findings if f.file}
1090 chunk_groups = [
1091 g
1092 for g in groups
1093 if _normalize_path(g.representative.file, fold_case=False) in chunk_files
1094 ]
1095 _apply_verdicts(chunk_groups, o.verdicts)
1097 # Runs AFTER verdicts are applied — see the matching comment in run_jury for
1098 # why (the _reject_targets member-tier guard assumes group.severity is the
1099 # true max member severity; demoting earlier would desync that invariant).
1100 if config.demote_local_only:
1101 vendor_by_reviewer = {a.name: a.vendor for a in config.agents}
1102 demote_local_only_groups(groups, vendor_by_reviewer)
1104 warnings = [w for o in outcomes for w in o.warnings]
1106 synthesis = _combine_chair_results([o.synthesis for o in outcomes if o.synthesis], base.chair)
1107 verify = _combine_chair_results([o.verify for o in outcomes if o.verify], base.chair)
1109 # bolt: Consolidate collection aggregations (sum, extend, max, any) into a single-pass O(N) explicit loop
1110 redaction_count = 0
1111 injection_hits = []
1112 budget_exhausted = False
1113 rounds_executed = 0
1115 for o in outcomes:
1116 redaction_count += o.redaction_count
1117 injection_hits.extend(o.injection_hits)
1118 if o.budget_exhausted:
1119 budget_exhausted = True
1120 if o.rounds_executed > rounds_executed:
1121 rounds_executed = o.rounds_executed
1123 return JuryOutcome(
1124 reviews=reviews,
1125 debate=debate,
1126 synthesis=synthesis,
1127 chair=base.chair,
1128 findings=findings,
1129 warnings=warnings,
1130 groups=groups,
1131 verify=verify,
1132 verdicts=verdicts,
1133 context_mode=base.context_mode,
1134 redact_secrets=base.redact_secrets,
1135 redaction_count=redaction_count,
1136 injection_hits=injection_hits,
1137 skipped=base.skipped,
1138 budget_exhausted=budget_exhausted,
1139 rounds_executed=rounds_executed,
1140 stop_reason=f"chunked review across {len(outcomes)} part(s)",
1141 )
1144def review_diff(
1145 config: JuryConfig,
1146 diff: str,
1147 *,
1148 context: str = "",
1149 mock: bool = False,
1150 strict: bool = False,
1151 seed: int | None = None,
1152 policy: ReviewPolicy | None = None,
1153 log=lambda _msg: None,
1154 on_event=None,
1155) -> tuple[JuryOutcome, largediff.DiffPlan]:
1156 """Plan a diff (filter + size + mode) then run the jury (issue #31).
1158 The single entry point the CLI uses: it measures and filters the diff,
1159 reports the size and the selected handling mode, and dispatches:
1161 - ``full`` — review the filtered diff in one ``run_jury`` pass;
1162 - ``chunked`` — review each chunk and merge the outcomes;
1163 - ``too_large`` — raise ``RuntimeError`` with an actionable message.
1165 Returns ``(outcome, plan)`` so the caller can surface the plan. Existing
1166 callers of :func:`run_jury` are unaffected.
1167 """
1168 dc = config.diff
1169 plan = largediff.plan_diff(
1170 diff,
1171 max_bytes=dc.max_bytes,
1172 chunk=dc.chunk,
1173 chunk_max_bytes=dc.chunk_max_bytes,
1174 exclude_generated=dc.exclude_generated,
1175 exclude=dc.exclude,
1176 include=dc.include,
1177 )
1178 log(
1179 f"diff size: {plan.total_bytes} B total, {plan.kept_bytes} B after filters "
1180 f"({len(plan.kept)} file(s) kept, {len(plan.excluded)} excluded); "
1181 f"mode: {plan.mode}"
1182 )
1183 if plan.excluded:
1184 log("excluded: " + ", ".join(f"{p} [{why}]" for p, why in plan.excluded))
1185 log(plan.reason)
1187 if plan.mode == largediff.MODE_TOO_LARGE:
1188 raise RuntimeError(f"diff too large to review: {plan.reason}")
1189 if not plan.chunks:
1190 raise RuntimeError(
1191 "nothing to review after filters — all files were excluded "
1192 "(check [jury.diff] include/exclude patterns)"
1193 )
1195 # One shared budget across all chunks so ``total_timeout`` bounds the WHOLE
1196 # review, not each chunk independently (review finding). ``phase_timeout`` and
1197 # per-agent timeouts still apply per call via the same budget.
1198 shared_budget = RunBudget(config.total_timeout, config.phase_timeout)
1200 # Redact the shared context ONCE here, before fan-out (#249). The same context
1201 # is reviewed against every chunk; letting each per-chunk ``run_jury`` redact
1202 # it would count its secrets once per chunk and ``_merge_chunk_outcomes`` would
1203 # sum them, inflating ``redaction_count`` (e.g. a 1-secret context over 8
1204 # chunks reported 8). Pre-redacting makes each chunk's re-redaction a no-op —
1205 # the ``[REDACTED:…]`` placeholders no longer match — so we add the one-time
1206 # context count back at the end. Diff/chunk redactions are still counted
1207 # per chunk and summed, which is correct (each chunk's diff is distinct).
1208 # `config.context` is the right path: `_from_dict` flattens the `[jury]`
1209 # table onto JuryConfig, so the `[jury.context]` sub-table is `config.context`
1210 # (a ContextConfig), NOT `config.jury.context` — there is no `config.jury`.
1211 # This mirrors how run_jury() reads it.
1212 ctx_cfg = getattr(config, "context", None)
1213 ctx_mode = getattr(ctx_cfg, "mode", "diff-only") if ctx_cfg else "diff-only"
1214 redact_on = getattr(ctx_cfg, "redact_secrets", True) if ctx_cfg else True
1215 context_redactions = 0
1216 if redact_on and ctx_mode != "diff-only" and context:
1217 context, context_redactions = redact(context)
1219 def _run(chunk: str) -> JuryOutcome:
1220 return run_jury(
1221 config,
1222 chunk,
1223 context=context,
1224 mock=mock,
1225 strict=strict,
1226 seed=seed,
1227 policy=policy,
1228 log=log,
1229 budget=shared_budget,
1230 on_event=on_event,
1231 )
1233 def _finalize(outcome: JuryOutcome) -> JuryOutcome:
1234 # Add the one-time context redaction count (per-chunk runs saw an already-
1235 # redacted context and counted 0 for it).
1236 if not context_redactions:
1237 return outcome
1238 return replace(outcome, redaction_count=outcome.redaction_count + context_redactions)
1240 if plan.mode == largediff.MODE_FULL:
1241 return _finalize(_run(plan.chunks[0])), plan
1243 outcomes = []
1244 for i, chunk in enumerate(plan.chunks, 1):
1245 log(f"reviewing chunk {i}/{len(plan.chunks)}")
1246 outcomes.append(_run(chunk))
1247 return _finalize(_merge_chunk_outcomes(outcomes, config)), plan