Coverage for src/keel/review.py: 100%
114 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 12:05 +0000
1"""Pure orchestration for ``keel review`` — the evidence-bundle orchestrator.
3The host agent runs the actual reviewers and produces the review *content*. This
4module takes that supplied content and deterministically decides what to render
5and where to post it: one head-pinned reviewer verdict per supplied review, an
6optional closure comment posted to both the PR and the linked issue, and the
7run-id sub-keys that bind each post to a stable, idempotent comment.
9Everything here is pure — no network, no subprocess, no clock, no randomness.
10Rendering uses the already-pure artifact/closure renderers. The CLI handler owns
11the head-SHA fetch, the actual posting, and the optional re-verify.
12"""
14from __future__ import annotations
16from dataclasses import dataclass, field
17from typing import Any
19from . import artifacts, closure, evidence
21SCHEMA_VERSION = "keel.review.v1"
24class ReviewError(ValueError):
25 """Raised when the supplied review bundle is malformed or under-count."""
28@dataclass(frozen=True)
29class ReviewItem:
30 """One parsed reviewer verdict supplied by the host agent."""
32 reviewer: str
33 verdict: str
34 scope: str | None
35 findings: tuple[dict[str, Any], ...]
36 testing: str | None
37 vendor: str | None = None
38 model: str | None = None
41@dataclass(frozen=True)
42class PostTarget:
43 """A single planned post: a rendered body bound to a target and run-id sub-key."""
45 artifact: str
46 target_kind: str
47 target_number: int
48 run_id: str
49 marker: str
50 body: str
52 def as_dict(self) -> dict[str, Any]:
53 return {
54 "artifact": self.artifact,
55 "target": {"kind": self.target_kind, "number": self.target_number},
56 "run_id": self.run_id,
57 "marker": self.marker,
58 "body": self.body,
59 }
62@dataclass(frozen=True)
63class ReviewPlan:
64 """The deterministic plan: rendered verdicts, optional closure, post targets."""
66 pull_request: int
67 issue: int | None
68 head_sha: str | None
69 run_id: str
70 tier: int | None
71 required_count: int
72 supplied_count: int
73 posts: tuple[PostTarget, ...] = field(default_factory=tuple)
75 def as_dict(self) -> dict[str, Any]:
76 return {
77 "schema_version": SCHEMA_VERSION,
78 "pull_request": self.pull_request,
79 "issue": self.issue,
80 "head_sha": self.head_sha,
81 "run_id": self.run_id,
82 "tier": self.tier,
83 "required_count": self.required_count,
84 "supplied_count": self.supplied_count,
85 "posts": [post.as_dict() for post in self.posts],
86 }
89def parse_reviews(raw: object) -> tuple[ReviewItem, ...]:
90 """Parse and validate a ``--reviews`` JSON payload into ``ReviewItem`` records."""
91 if not isinstance(raw, list):
92 raise ReviewError("reviews file must contain a JSON array of review objects")
93 items: list[ReviewItem] = []
94 for index, entry in enumerate(raw):
95 items.append(_parse_review(entry, index))
96 return tuple(items)
99def _parse_review(entry: object, index: int) -> ReviewItem:
100 if not isinstance(entry, dict):
101 raise ReviewError(f"review #{index + 1} must be a JSON object")
102 reviewer = entry.get("reviewer")
103 if not isinstance(reviewer, str) or not reviewer.strip():
104 raise ReviewError(f"review #{index + 1} requires a non-empty 'reviewer' string")
105 verdict = entry.get("verdict")
106 if not isinstance(verdict, str) or not verdict.strip():
107 raise ReviewError(f"review #{index + 1} requires a non-empty 'verdict' string")
108 scope = entry.get("scope")
109 if scope is not None and not isinstance(scope, str):
110 raise ReviewError(f"review #{index + 1} 'scope' must be a string when present")
111 testing = entry.get("testing")
112 if testing is not None and not isinstance(testing, str):
113 raise ReviewError(f"review #{index + 1} 'testing' must be a string when present")
114 vendor = _parse_provenance_field(entry.get("vendor"), index, "vendor")
115 model = _parse_provenance_field(entry.get("model"), index, "model")
116 findings = _parse_findings(entry.get("findings"), index)
117 return ReviewItem(
118 reviewer=reviewer.strip(),
119 verdict=verdict.strip(),
120 scope=scope,
121 findings=findings,
122 testing=testing,
123 vendor=vendor,
124 model=model,
125 )
128def _parse_provenance_field(value: object, index: int, name: str) -> str | None:
129 """Parse an optional ``vendor``/``model`` provenance string from a review entry."""
130 if value is None:
131 return None
132 if not isinstance(value, str):
133 raise ReviewError(f"review #{index + 1} '{name}' must be a string when present")
134 cleaned = value.strip()
135 return cleaned or None
138def _parse_findings(raw: object, index: int) -> tuple[dict[str, Any], ...]:
139 if raw is None:
140 return ()
141 if not isinstance(raw, list):
142 raise ReviewError(f"review #{index + 1} 'findings' must be a list when present")
143 findings: list[dict[str, Any]] = []
144 for finding_index, finding in enumerate(raw):
145 if not isinstance(finding, dict):
146 raise ReviewError(
147 f"review #{index + 1} finding #{finding_index + 1} must be a JSON object"
148 )
149 findings.append(dict(finding))
150 return tuple(findings)
153def parse_cycle_reviewers(raw: object) -> tuple[dict[str, Any], ...]:
154 """Parse a review-cycle findings payload into reviewer records for rendering.
156 The payload is the host-supplied structured block each reviewer returns
157 (codename · focus · verdict · findings · clean areas). Validation is shallow
158 on purpose: the renderer applies per-field fallbacks, so this only enforces
159 the outer shape and surfaces a clear error when it is malformed.
160 """
161 if not isinstance(raw, list):
162 raise ReviewError("review-cycle findings must be a JSON array of reviewer objects")
163 reviewers: list[dict[str, Any]] = []
164 for index, entry in enumerate(raw):
165 if not isinstance(entry, dict):
166 raise ReviewError(f"reviewer #{index + 1} must be a JSON object")
167 reviewers.append(dict(entry))
168 return tuple(reviewers)
171def review_run_id(run_id: str, reviewer: str) -> str:
172 """Stable per-reviewer run-id sub-key, e.g. ``<run-id>:rv-<reviewer-slug>``."""
173 return f"{run_id}:rv-{artifacts.slug(reviewer)}"
176def closure_run_id(run_id: str) -> str:
177 """Stable run-id sub-key for the closure-comment artifact."""
178 return f"{run_id}:closure"
181def build_review_plan(
182 reviews: tuple[ReviewItem, ...],
183 *,
184 required_count: int,
185 head_sha: str | None,
186 pull_request: int,
187 issue: int | None,
188 run_id: str,
189 tier: int | None,
190 closure_record: dict[str, Any] | None = None,
191) -> ReviewPlan:
192 """Validate the bundle against the required count and build the post plan.
194 Fewer supplied reviews than required fails; exact or more is allowed. Each
195 review renders as a head-pinned verdict posted to the PR. A closure record,
196 when supplied, renders once and posts to both the PR and the linked issue.
197 """
198 supplied = len(reviews)
199 if supplied < required_count:
200 raise ReviewError(
201 f"supplied {supplied} review(s) but tier requires at least "
202 f"{required_count}; refusing to under-post evidence"
203 )
204 posts: list[PostTarget] = []
205 for item in reviews:
206 body = artifacts.render_review_verdict(
207 reviewer=item.reviewer,
208 head_sha=head_sha,
209 verdict=item.verdict,
210 scope=item.scope,
211 findings=list(item.findings),
212 testing=item.testing,
213 vendor=item.vendor,
214 model=item.model,
215 )
216 posts.append(
217 PostTarget(
218 artifact="review-verdict",
219 target_kind="pr",
220 target_number=pull_request,
221 run_id=review_run_id(run_id, item.reviewer),
222 marker=evidence.REVIEW_VERDICT_MARKER,
223 body=body,
224 )
225 )
226 if closure_record is not None:
227 posts.extend(
228 _closure_posts(
229 closure_record,
230 pull_request=pull_request,
231 issue=issue,
232 run_id=run_id,
233 )
234 )
235 return ReviewPlan(
236 pull_request=pull_request,
237 issue=issue,
238 head_sha=head_sha,
239 run_id=run_id,
240 tier=tier,
241 required_count=required_count,
242 supplied_count=supplied,
243 posts=tuple(posts),
244 )
247def _closure_posts(
248 closure_record: dict[str, Any],
249 *,
250 pull_request: int,
251 issue: int | None,
252 run_id: str,
253) -> list[PostTarget]:
254 if not isinstance(closure_record, dict):
255 raise ReviewError("closure file must contain a JSON object")
256 body = closure.render_closure_comment(closure_record)
257 sub_run_id = closure_run_id(run_id)
258 targets: list[PostTarget] = [
259 PostTarget(
260 artifact="closure-comment",
261 target_kind="pr",
262 target_number=pull_request,
263 run_id=sub_run_id,
264 marker=closure.COMMENT_MARKER,
265 body=body,
266 )
267 ]
268 if issue is not None:
269 targets.append(
270 PostTarget(
271 artifact="closure-comment",
272 target_kind="issue",
273 target_number=issue,
274 run_id=sub_run_id,
275 marker=closure.COMMENT_MARKER,
276 body=body,
277 )
278 )
279 return targets