Coverage for src/keel/evidence.py: 100%
352 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 12:05 +0000
1"""Deterministic pre-merge evidence verification.
3The ship adapter is agentic, but the artifacts it must leave behind are not:
4reviewer verdict comments/reviews, the optional jury verdict, and the stable
5closure comment marker. This module keeps the check pure so CI can enforce it
6without trusting prose in an agent prompt.
7"""
9from __future__ import annotations
11import hashlib
12import re
13from collections.abc import Sequence
14from dataclasses import dataclass
15from typing import Any
17from . import agents, closure
19SCHEMA_VERSION = "keel.evidence.v1"
20AGENT_LABEL_PREFIX = "agent:"
21REVIEW_VERDICT_MARKER = "keel.review-verdict.v1"
22JURY_VERDICT_MARKER = "keel.jury-verdict.v1"
23SHIP_ASSESSMENT_HEADING = "### \U0001f6a2 keel ship"
24DEFAULT_WAIVER_LABEL = "keel:evidence-waived"
25TRUSTED_AUTHOR_ASSOCIATIONS = frozenset({"OWNER", "MEMBER", "COLLABORATOR"})
26TRUSTED_SHIP_ASSESSMENT_BOTS = frozenset({"github-actions", "github-actions[bot]"})
28_FIELD_RE = re.compile(
29 r"^\s*(?P<key>reviewer|head|vendor|model|vendors)\s*:\s*(?P<value>\S+)\s*$",
30 re.IGNORECASE | re.MULTILINE,
31)
32_SHIP_BRANCH_RE = re.compile(r"^(feature|fix|chore|docs|test)/issue-\d+(?:-|$)")
34# Evidence phases. An artifact is required in the phase that produces it, mirroring
35# the step mapping stepverifier already applies (review -> s7, jury -> s8, closure ->
36# s12). The merge gate at s10 asks for PHASE_PRE_MERGE, because the closure comment
37# is a post-merge record and requiring it at s10 makes the backbone unsatisfiable.
38PHASE_PRE_MERGE = "pre-merge"
39PHASE_POST_MERGE = "post-merge"
40PHASE_ALL = "all"
41PHASES = (PHASE_PRE_MERGE, PHASE_POST_MERGE, PHASE_ALL)
44@dataclass(frozen=True)
45class EvidenceItem:
46 id: str
47 kind: str
48 required: bool
49 description: str
50 phase: str = PHASE_PRE_MERGE
52 def as_dict(self) -> dict[str, Any]:
53 return {
54 "id": self.id,
55 "kind": self.kind,
56 "required": self.required,
57 "description": self.description,
58 "phase": self.phase,
59 }
62def gate_active(labels: Sequence[str] | None, gate_label: str) -> bool:
63 """Return whether ``gate_label`` is present in ``labels`` (None/empty -> False).
65 An empty ``gate_label`` is never active, so a misconfigured (blank) label can
66 never silently match — the schema also forbids an empty ``evidence_gate_label``.
67 """
68 if not gate_label:
69 return False
70 return gate_label in set(labels or ())
73def gate_decision(
74 labels: Sequence[str] | None,
75 gate_label: str,
76 *,
77 waiver_label: str = DEFAULT_WAIVER_LABEL,
78 head_ref: str | None = None,
79 pr_comments: list[dict[str, Any]] | None = None,
80 pr_reviews: list[dict[str, Any]] | None = None,
81 ledger_records: Sequence[object] | None = None,
82) -> dict[str, Any]:
83 """Return the fail-closed evidence-gate arming decision.
85 Ship provenance arms the gate by default. The only disarm path is an explicit
86 waiver label applied by an operator; the legacy gate label remains an
87 additional arming signal for already-installed workflows.
88 """
89 label_set = set(labels or ())
90 if waiver_label and waiver_label in label_set:
91 return _gate_decision(False, "operator-waiver-label", waiver_label, waived=True)
92 if gate_active(labels, gate_label):
93 return _gate_decision(True, "gate-label", gate_label)
94 if head_ref and _SHIP_BRANCH_RE.search(head_ref):
95 return _gate_decision(True, "ship-branch", head_ref)
96 if _has_trusted_ship_assessment(pr_comments or []):
97 return _gate_decision(True, "ship-assessment-comment", SHIP_ASSESSMENT_HEADING)
98 if _has_trusted_review_marker([*(pr_comments or []), *(pr_reviews or [])]):
99 return _gate_decision(True, "review-verdict-marker", REVIEW_VERDICT_MARKER)
100 if ledger_records:
101 return _gate_decision(True, "ship-run-ledger", "ship_run")
102 return _gate_decision(False, "no-ship-provenance", None)
105def _gate_decision(
106 enforced: bool,
107 reason: str,
108 source: str | None,
109 *,
110 waived: bool = False,
111) -> dict[str, Any]:
112 return {
113 "schema_version": SCHEMA_VERSION,
114 "enforced": enforced,
115 "waived": waived,
116 "reason": reason,
117 "source": source,
118 }
121def _has_trusted_ship_assessment(items: list[dict[str, Any]]) -> bool:
122 return any(
123 _is_ship_assessment_source(item) and _is_ship_assessment(_body(item))
124 for item in items
125 )
128def _is_ship_assessment_source(item: dict[str, Any]) -> bool:
129 if _is_trusted_source(item, enforced=True):
130 return True
131 user = item.get("user") if isinstance(item.get("user"), dict) else {}
132 login = user.get("login") if isinstance(user.get("login"), str) else None
133 return bool(login and login.lower() in TRUSTED_SHIP_ASSESSMENT_BOTS)
136def contract_as_dict(
137 review_contract: dict[str, Any],
138 *,
139 dry_run: bool = False,
140 enforced: bool = True,
141 deferrals: tuple[str, ...] = (),
142) -> dict[str, Any]:
143 """Return the required evidence set derived from review/jury flags."""
144 return {
145 "schema_version": SCHEMA_VERSION,
146 "enforced": enforced,
147 "source": "review_merge_contract + closure_comment",
148 "dry_run_disables_gating": True,
149 "fail_closed": True,
150 "require_distinct_vendors": _require_distinct_vendors(review_contract),
151 "accepted_sources": {
152 "closure": (
153 "trusted issue/PR comments carrying keel.closure-comment.v1"
154 ),
155 "review": (
156 "trusted PR review/comment carrying keel.review-verdict.v1 and current head"
157 ),
158 "jury": "trusted PR comment carrying keel.jury-verdict.v1 and current head",
159 },
160 "not_accepted": [
161 "pull_request_body",
162 "chat_summary",
163 "untrusted_public_comment",
164 "keel_ship_assessment_comment",
165 ],
166 "deferrals": list(deferrals),
167 "required": [
168 item.as_dict()
169 for item in required_items(review_contract, dry_run=False, enforced=enforced)
170 ],
171 "active_required": [
172 item.as_dict()
173 for item in required_items(review_contract, dry_run=dry_run, enforced=enforced)
174 ],
175 }
178def required_items(
179 review_contract: dict[str, Any],
180 *,
181 dry_run: bool = False,
182 enforced: bool = True,
183 phase: str = PHASE_ALL,
184) -> tuple[EvidenceItem, ...]:
185 """Return the tier/flag-derived evidence requirements for ``phase``.
187 ``phase`` selects which artifacts are in scope: ``pre-merge`` covers the
188 review verdicts and a gating jury verdict (everything that must exist before
189 s10 authorizes a merge), ``post-merge`` covers the closure comments s11
190 posts, and ``all`` — the default, so existing callers are unchanged — covers
191 both. An unknown phase raises, so a typo cannot silently drop requirements.
192 """
193 if phase not in PHASES:
194 raise ValueError(f"unknown evidence phase {phase!r}; expected one of {', '.join(PHASES)}")
195 if dry_run or not enforced:
196 return ()
197 reviewers = review_contract.get("reviewers")
198 reviewer_count = reviewers.get("count") if isinstance(reviewers, dict) else 0
199 reviewer_count = reviewer_count if isinstance(reviewer_count, int) and reviewer_count > 0 else 0
200 jury = review_contract.get("jury")
201 jury_required = (
202 isinstance(jury, dict)
203 and bool(jury.get("enabled"))
204 and jury.get("mode") == "gating"
205 )
206 items: list[EvidenceItem] = [
207 EvidenceItem(
208 "closure-comment-pr",
209 "closure",
210 True,
211 "PR conversation comment with keel.closure-comment.v1 marker",
212 PHASE_POST_MERGE,
213 ),
214 EvidenceItem(
215 "closure-comment-issue",
216 "closure",
217 True,
218 "Linked issue comment with keel.closure-comment.v1 marker",
219 PHASE_POST_MERGE,
220 ),
221 ]
222 for index in range(1, reviewer_count + 1):
223 items.append(EvidenceItem(
224 f"review-verdict-{index}",
225 "review",
226 True,
227 "Distinct posted s7 reviewer verdict for the current PR",
228 PHASE_PRE_MERGE,
229 ))
230 if jury_required:
231 items.append(EvidenceItem(
232 "jury-verdict",
233 "jury",
234 True,
235 "Posted gating jury verdict comment for the current PR",
236 PHASE_PRE_MERGE,
237 ))
238 if phase == PHASE_ALL:
239 return tuple(items)
240 return tuple(item for item in items if item.phase == phase)
243def verify(
244 review_contract: dict[str, Any],
245 *,
246 pr_comments: list[dict[str, Any]] | None = None,
247 issue_comments: list[dict[str, Any]] | None = None,
248 pr_reviews: list[dict[str, Any]] | None = None,
249 pr_body: str | None = None,
250 pr_labels: Sequence[str] | None = None,
251 head_sha: str | None = None,
252 ledger_record: dict[str, Any] | None = None,
253 dry_run: bool = False,
254 enforced: bool = True,
255 deferrals: tuple[str, ...] = (),
256 phase: str = PHASE_ALL,
257 require_armed: bool = False,
258 waived: bool = False,
259) -> dict[str, Any]:
260 """Verify required evidence artifacts and return a deterministic report.
262 ``phase`` narrows the requirement set to the artifacts that phase produces;
263 see :func:`required_items`. The s10 merge gate passes ``pre-merge`` so it
264 does not demand the closure comments s11 has not written yet.
266 ``require_armed`` closes the vacuous-pass hole: with the gate unarmed there
267 are no requirements, so the report would otherwise pass without having
268 checked anything, and a green result could not be told apart from "could not
269 tell whether this was a ship run". When set, an unarmed gate is a blocking
270 finding instead. A deliberately non-ship PR still goes green through the
271 operator waiver label, which disarms explicitly rather than by accident.
273 When ``ledger_record`` is the ship_run record for this PR, a closure comment
274 only counts when its content matches the canonical render of that record
275 (closure-comment fidelity). Without a record the marker-only behavior holds.
277 When the gate is active, ``pr_labels`` are additionally checked for the
278 mandatory ``agent:<vendor>`` attribution label (and cross-checked against the
279 ledger implementer vendor when a record is present); see
280 :func:`attribution_check`.
281 """
282 del pr_body # Explicitly not accepted as evidence.
283 items = required_items(review_contract, dry_run=dry_run, enforced=enforced, phase=phase)
284 deferred = set(deferrals)
285 counts = _evidence_counts(
286 pr_comments=pr_comments or [],
287 issue_comments=issue_comments or [],
288 pr_reviews=pr_reviews or [],
289 head_sha=head_sha,
290 enforced=enforced,
291 ledger_record=ledger_record,
292 )
293 findings = _run_context_findings(
294 pr_comments=pr_comments or [],
295 issue_comments=issue_comments or [],
296 enforced=enforced,
297 ledger_record=ledger_record,
298 )
299 mismatch = _closure_mismatch_scopes(
300 pr_comments=pr_comments or [],
301 issue_comments=issue_comments or [],
302 enforced=enforced,
303 ledger_record=ledger_record,
304 )
305 results = []
306 for item in items:
307 present = _is_present(item, counts)
308 is_deferred = item.id in deferred or item.kind in deferred or "all" in deferred
309 ok = present or is_deferred
310 results.append({
311 "id": item.id,
312 "kind": item.kind,
313 "required": item.required,
314 "present": present,
315 "deferred": is_deferred,
316 "ok": ok,
317 "reason": None if ok else _result_reason(item, mismatch),
318 })
319 missing = [result["id"] for result in results if not result["ok"]]
320 distinct = _distinct_vendor_finding(
321 review_contract,
322 items=items,
323 deferred=deferred,
324 pr_comments=pr_comments or [],
325 pr_reviews=pr_reviews or [],
326 head_sha=head_sha,
327 enforced=enforced,
328 )
329 if distinct is not None:
330 findings = [*findings, distinct]
331 attribution = _attribution_finding(
332 pr_labels=pr_labels,
333 enforced=enforced and not dry_run,
334 ledger_record=ledger_record,
335 )
336 if attribution is not None:
337 findings = [*findings, attribution]
338 unarmed = _unarmed_finding(
339 enforced=enforced,
340 dry_run=dry_run,
341 require_armed=require_armed,
342 waived=waived,
343 )
344 if unarmed is not None:
345 findings = [*findings, unarmed]
346 blocking_findings = [finding for finding in findings if finding["severity"] == "major"]
347 return {
348 "schema_version": SCHEMA_VERSION,
349 "status": "pass" if not missing and not blocking_findings else "fail",
350 "dry_run": dry_run,
351 "enforced": enforced,
352 "phase": phase,
353 "required_count": len(items),
354 "missing": missing,
355 "results": results,
356 "counts": counts,
357 "findings": findings,
358 }
361def _require_distinct_vendors(review_contract: dict[str, Any]) -> bool:
362 reviewers = review_contract.get("reviewers")
363 return bool(reviewers.get("require_distinct_vendors")) if isinstance(reviewers, dict) else False
366def _distinct_vendor_finding(
367 review_contract: dict[str, Any],
368 *,
369 items: tuple[EvidenceItem, ...],
370 deferred: set[str],
371 pr_comments: list[dict[str, Any]],
372 pr_reviews: list[dict[str, Any]],
373 head_sha: str | None,
374 enforced: bool,
375) -> dict[str, Any] | None:
376 """Return a blocking finding when the optional vendor-distinctness check fails.
378 Off by default: ``None`` unless ``reviewers.require_distinct_vendors`` is set
379 on the contract. Skipped when review evidence is deferred so the knob never
380 overrides an explicit deferral.
381 """
382 if not _require_distinct_vendors(review_contract):
383 return None
384 if "review" in deferred or "all" in deferred:
385 return None
386 required = sum(1 for item in items if item.kind == "review" and item.id not in deferred)
387 if required <= 0:
388 return None
389 provenance = _review_vendor_provenance(
390 [*pr_comments, *pr_reviews],
391 head_sha=head_sha,
392 enforced=enforced,
393 )
394 result = distinct_vendor_check(list(provenance.values()), required_count=required)
395 if result["ok"]:
396 return None
397 return {
398 "id": "review-vendor-distinctness",
399 "severity": "major",
400 "kind": "review",
401 "message": f"require_distinct_vendors: {result['reason']}.",
402 }
405_CLOSURE_MISMATCH_REASON = (
406 "closure comment does not match the ship_run ledger record"
407)
410def _result_reason(item: EvidenceItem, mismatch: set[str]) -> str:
411 if item.id == "closure-comment-pr" and "pr" in mismatch:
412 return _CLOSURE_MISMATCH_REASON
413 if item.id == "closure-comment-issue" and "issue" in mismatch:
414 return _CLOSURE_MISMATCH_REASON
415 return f"missing required evidence: {item.id}"
418def _closure_mismatch_scopes(
419 *,
420 pr_comments: list[dict[str, Any]],
421 issue_comments: list[dict[str, Any]],
422 enforced: bool,
423 ledger_record: dict[str, Any] | None,
424) -> set[str]:
425 """Return scopes ({"pr"}/{"issue"}) where a marker closure mismatched the ledger.
427 A scope is reported only when a trusted marker-bearing closure exists but none
428 of them match the record — so a stale comment alongside a correct re-post does
429 not produce a misleading mismatch reason.
430 """
431 if ledger_record is None:
432 return set()
433 scopes: set[str] = set()
434 for scope, comments in (("pr", pr_comments), ("issue", issue_comments)):
435 markered = [
436 comment for comment in comments
437 if _is_trusted_source(comment, enforced=enforced)
438 and _has_closure_marker(_body(comment))
439 ]
440 if markered and not any(
441 closure_body_matches_record(_body(comment), ledger_record)
442 for comment in markered
443 ):
444 scopes.add(scope)
445 return scopes
448def _evidence_counts(
449 *,
450 pr_comments: list[dict[str, Any]],
451 issue_comments: list[dict[str, Any]],
452 pr_reviews: list[dict[str, Any]],
453 head_sha: str | None = None,
454 enforced: bool = True,
455 ledger_record: dict[str, Any] | None = None,
456) -> dict[str, int]:
457 review_keys = _review_evidence_keys(
458 [*pr_comments, *pr_reviews],
459 head_sha=head_sha,
460 enforced=enforced,
461 )
462 return {
463 "closure_pr": sum(
464 _is_closure_comment(comment, enforced=enforced, record=ledger_record)
465 for comment in pr_comments
466 ),
467 "closure_issue": sum(
468 _is_closure_comment(comment, enforced=enforced, record=ledger_record)
469 for comment in issue_comments
470 ),
471 "review_verdict": len(review_keys),
472 "jury_verdict": sum(_is_jury_verdict(comment, head_sha=head_sha, enforced=enforced)
473 for comment in pr_comments),
474 }
477def _is_present(item: EvidenceItem, counts: dict[str, int]) -> bool:
478 if item.id == "closure-comment-pr":
479 return counts["closure_pr"] >= 1
480 if item.id == "closure-comment-issue":
481 return counts["closure_issue"] >= 1
482 if item.kind == "review":
483 index = int(item.id.rsplit("-", 1)[1])
484 return counts["review_verdict"] >= index
485 if item.id == "jury-verdict":
486 return counts["jury_verdict"] >= 1
487 return False
490def _body(item: dict[str, Any]) -> str:
491 body = item.get("body")
492 return body if isinstance(body, str) else ""
495def _has_closure_marker(body: str) -> bool:
496 return closure.COMMENT_MARKER in body
499#: The idempotency marker ``keel post-comment`` appends to a posted body so a re-post
500#: can find and edit its own comment. It is transport bookkeeping, not content, so it is
501#: stripped before a closure body is compared to its canonical render.
502#:
503#: Matched in the *exact* form the transport emits — a run id, then the close, then end
504#: of line. A permissive ``.*?`` would let a trusted author smuggle arbitrary text past
505#: the verbatim comparison: an HTML comment ends at its first ``-->``, so anything after
506#: that renders visibly on the page while the whole line still normalizes away.
507RUN_ID_MARKER_RE = re.compile(r"^\s*<!--\s*keel\.run-id:\s*[\w.:@/+-]+\s*-->\s*$")
510def _normalize_closure_body(body: str) -> str:
511 """Normalize a closure body for content comparison.
513 Robust to harmless formatting drift but sensitive to real content changes:
514 trailing whitespace is stripped per line, runs of blank lines collapse to a
515 single blank line, and leading/trailing blank lines are dropped.
517 The transport's ``keel.run-id`` marker line is dropped too. Without that, closure
518 fidelity and post-comment idempotency were mutually exclusive: the marker is what
519 lets a re-post edit its own comment instead of duplicating, and its presence made
520 the body differ from the canonical render.
521 """
522 lines = [line.rstrip() for line in body.splitlines()
523 if not RUN_ID_MARKER_RE.match(line)]
524 normalized: list[str] = []
525 for line in lines:
526 if not line and (not normalized or not normalized[-1]):
527 continue
528 normalized.append(line)
529 while normalized and not normalized[-1]:
530 normalized.pop()
531 return "\n".join(normalized)
534def closure_body_matches_record(body: str, record: dict[str, Any]) -> bool:
535 """Return whether ``body`` matches the canonical render of ``record``."""
536 expected = closure.render_closure_comment(record)
537 return _normalize_closure_body(body) == _normalize_closure_body(expected)
540def _is_closure_comment(
541 item: dict[str, Any],
542 *,
543 enforced: bool = True,
544 record: dict[str, Any] | None = None,
545) -> bool:
546 if not _is_trusted_source(item, enforced=enforced):
547 return False
548 if not _has_closure_marker(_body(item)):
549 return False
550 if record is None:
551 return True
552 return closure_body_matches_record(_body(item), record)
555def _run_context_findings(
556 *,
557 pr_comments: list[dict[str, Any]],
558 issue_comments: list[dict[str, Any]],
559 enforced: bool,
560 ledger_record: dict[str, Any] | None = None,
561) -> list[dict[str, Any]]:
562 comments = [*pr_comments, *issue_comments]
563 findings: list[dict[str, Any]] = []
564 for item in comments:
565 if not _is_closure_comment(item, enforced=enforced, record=ledger_record):
566 continue
567 body = _body(item)
568 if _has_empty_run_context(body):
569 findings.append({
570 "id": "run-context-empty",
571 "severity": "major" if enforced else "minor",
572 "kind": "closure",
573 "message": "Closure comment Run context is fully degraded.",
574 })
575 return findings
578def _has_empty_run_context(body: str) -> bool:
579 if "### Run context" not in body:
580 return False
581 fields = _run_context_fields(body)
582 return fields == {
583 "host agent": "unknown",
584 "transport": "unknown",
585 "profile": "unknown",
586 "jury": "off",
587 "consent": "unknown (scopes: none)",
588 }
591def _run_context_fields(body: str) -> dict[str, str]:
592 fields: dict[str, str] = {}
593 in_block = False
594 for line in body.splitlines():
595 if line.strip() == "### Run context":
596 in_block = True
597 continue
598 if in_block and line.startswith("### "):
599 break
600 if not in_block:
601 continue
602 match = re.match(r"^-\s+\*\*(?P<key>[^*]+):\*\*\s+(?P<value>.+?)\s*$", line)
603 if match:
604 fields[match.group("key").strip().lower()] = match.group("value").strip().lower()
605 return fields
608def _is_trusted_source(item: dict[str, Any], *, enforced: bool = True) -> bool:
609 """Return whether GitHub marks this evidence source as trusted.
611 Live GitHub comment/review payloads include ``author_association``. Enforced
612 evidence fails closed when that field is absent because offline fixtures are
613 agent-writable and must not manufacture trust. Untrusted explicit
614 associations fail closed even if the author type is ``Bot``.
615 """
616 association = item.get("author_association")
617 if association is None:
618 return not enforced
619 if isinstance(association, str) and association.upper() in TRUSTED_AUTHOR_ASSOCIATIONS:
620 return True
621 return False
624def _is_ship_assessment(body: str) -> bool:
625 return SHIP_ASSESSMENT_HEADING in body or "keel ship \u2014" in body
628def count_review_verdicts(
629 pr_comments: list[dict[str, Any]] | None = None,
630 pr_reviews: list[dict[str, Any]] | None = None,
631 *,
632 head_sha: str | None = None,
633 enforced: bool = True,
634) -> int:
635 """Count distinct trusted review-verdict reviewers for a PR.
637 This is the same evidence-side counting the verify report uses for the
638 ``review`` items: it collapses idempotent re-posts by the same reviewer to
639 one verdict and only counts trusted, head-bound verdicts. Reused by capture
640 reconcile to cross-check the ledger's recorded reviewer count.
641 """
642 keys = _review_evidence_keys(
643 [*(pr_comments or []), *(pr_reviews or [])],
644 head_sha=head_sha,
645 enforced=enforced,
646 )
647 return len(keys)
650def _review_evidence_keys(
651 items: list[dict[str, Any]],
652 *,
653 head_sha: str | None = None,
654 enforced: bool = True,
655) -> set[str]:
656 keys: set[str] = set()
657 for item in items:
658 if not _is_trusted_source(item, enforced=enforced):
659 continue
660 body = _body(item)
661 if not _is_review_verdict_body(body):
662 continue
663 if not _matches_head(item, body, head_sha):
664 continue
665 keys.add(_reviewer_key(item, body))
666 return keys
669def _review_vendor_provenance(
670 items: list[dict[str, Any]],
671 *,
672 head_sha: str | None = None,
673 enforced: bool = True,
674) -> dict[str, str | None]:
675 """Map each accepted review-verdict reviewer-key to its declared vendor.
677 The value is the lower-cased ``vendor:`` provenance for that verdict, or
678 ``None`` when the verdict carries no vendor field. Keys mirror
679 :func:`_review_evidence_keys`, so duplicate reviewer-keys collapse to one
680 entry (idempotent re-posts do not inflate the vendor set).
681 """
682 provenance: dict[str, str | None] = {}
683 for item in items:
684 if not _is_trusted_source(item, enforced=enforced):
685 continue
686 body = _body(item)
687 if not _is_review_verdict_body(body):
688 continue
689 if not _matches_head(item, body, head_sha):
690 continue
691 key = _reviewer_key(item, body)
692 if key in provenance:
693 continue
694 vendor = _fields(body).get("vendor")
695 provenance[key] = vendor.lower() if vendor else None
696 return provenance
699def distinct_vendor_check(
700 vendors: Sequence[str | None],
701 *,
702 required_count: int,
703) -> dict[str, Any]:
704 """Pure vendor-distinctness check over review-verdict provenance.
706 ``vendors`` is one entry per accepted review verdict: the declared vendor, or
707 ``None`` when the verdict carries no vendor provenance. The check passes only
708 when at least ``required_count`` verdicts each declare a vendor and those
709 vendors are all distinct. It fails when a required verdict is missing vendor
710 provenance, or when two required verdicts share a vendor.
712 Returns ``{ok, reason, duplicated, missing_provenance}``. No I/O — fully
713 unit-testable. A non-positive ``required_count`` always passes (nothing to
714 require).
715 """
716 if required_count <= 0:
717 return {"ok": True, "reason": None, "duplicated": [], "missing_provenance": 0}
718 present = [vendor for vendor in vendors if vendor]
719 missing = len(vendors) - len(present)
720 seen: set[str] = set()
721 duplicated: list[str] = []
722 for vendor in present:
723 if vendor in seen and vendor not in duplicated:
724 duplicated.append(vendor)
725 seen.add(vendor)
726 if len(present) < required_count:
727 return {
728 "ok": False,
729 "reason": "missing vendor provenance on required review verdict(s)",
730 "duplicated": duplicated,
731 "missing_provenance": missing,
732 }
733 if duplicated:
734 return {
735 "ok": False,
736 "reason": f"review verdicts share a vendor: {', '.join(sorted(duplicated))}",
737 "duplicated": sorted(duplicated),
738 "missing_provenance": missing,
739 }
740 return {"ok": True, "reason": None, "duplicated": [], "missing_provenance": missing}
743def agent_label_vendors(labels: Sequence[str] | None) -> list[str]:
744 """Return the lower-cased vendor slugs from every ``agent:<vendor>`` label.
746 A blank vendor (a bare ``agent:`` label) is ignored. Order is preserved and
747 duplicates are kept so callers can reason about the raw label set; this is a
748 pure helper with no I/O.
749 """
750 vendors: list[str] = []
751 for label in labels or ():
752 if not isinstance(label, str) or not label.startswith(AGENT_LABEL_PREFIX):
753 continue
754 vendor = label[len(AGENT_LABEL_PREFIX):].strip().lower()
755 if vendor:
756 vendors.append(vendor)
757 return vendors
760def ledger_implementer_vendor(ledger_record: dict[str, Any] | None) -> str | None:
761 """Return the implementer's vendor slug from a ship_run ``ledger_record``.
763 The ledger stores the effective implementer as a codename or ``vendor:model``
764 string under ``actors.implementer``; the vendor is the part before the first
765 ``:``. Returns ``None`` when no record, no implementer, or a blank implementer
766 is recorded so the cross-check can degrade to presence-only. Pure — no I/O.
767 """
768 if not isinstance(ledger_record, dict):
769 return None
770 actors = ledger_record.get("actors")
771 implementer = actors.get("implementer") if isinstance(actors, dict) else None
772 if not isinstance(implementer, str) or not implementer.strip():
773 return None
774 vendor, _ = agents.split_delegate(implementer.strip())
775 vendor = vendor.strip().lower()
776 return vendor or None
779def attribution_check(
780 labels: Sequence[str] | None,
781 *,
782 implementer_vendor: str | None = None,
783) -> dict[str, Any]:
784 """Pure attribution-label check over a PR's labels and the ledger implementer.
786 Two layers, both fail-closed only on a real contradiction:
788 * **Presence** — at least one non-blank ``agent:<vendor>`` label must exist.
789 Missing one is a ``missing-label`` finding.
790 * **Cross-check** — when ``implementer_vendor`` is known (a ship_run record
791 recorded an implementer), one of the PR's ``agent:*`` vendors must match it.
792 A mismatch is a ``vendor-mismatch`` finding. When ``implementer_vendor`` is
793 ``None`` (no record / no implementer) only the presence layer runs, so PRs
794 that predate attribution recording are not broken.
796 Returns ``{ok, reason, label_vendors, implementer_vendor}``. No I/O.
797 """
798 label_vendors = agent_label_vendors(labels)
799 implementer = implementer_vendor.strip().lower() if implementer_vendor else None
800 if not label_vendors:
801 return {
802 "ok": False,
803 "reason": "missing-label",
804 "label_vendors": label_vendors,
805 "implementer_vendor": implementer,
806 }
807 if implementer is not None and implementer not in label_vendors:
808 return {
809 "ok": False,
810 "reason": "vendor-mismatch",
811 "label_vendors": label_vendors,
812 "implementer_vendor": implementer,
813 }
814 return {
815 "ok": True,
816 "reason": None,
817 "label_vendors": label_vendors,
818 "implementer_vendor": implementer,
819 }
822def _unarmed_finding(
823 *,
824 enforced: bool,
825 dry_run: bool,
826 require_armed: bool,
827 waived: bool,
828) -> dict[str, Any] | None:
829 """Return a blocking finding when the gate was never armed, else ``None``.
831 Opt-in via ``require_armed`` so existing callers keep today's behavior. An
832 unarmed gate derives no requirements, so without this the report passes
833 having verified nothing — indistinguishable from a genuine pass. Skipped
834 under ``dry_run``, where producing no evidence is the expected outcome, and
835 when ``waived``: an operator disarming the gate on purpose is the sanctioned
836 way out, and the whole point is to separate that from arming by accident.
837 """
838 if not require_armed or dry_run or enforced or waived:
839 return None
840 return {
841 "id": "gate-unarmed",
842 "severity": "major",
843 "kind": "arming",
844 "message": (
845 "Evidence gate is not armed, so no requirements were checked. Arm it via ship "
846 "provenance (ship branch, posted review verdict, ship-run ledger, or the gate "
847 "label), or disarm deliberately with the operator waiver label."
848 ),
849 }
852def _attribution_finding(
853 *,
854 pr_labels: Sequence[str] | None,
855 enforced: bool,
856 ledger_record: dict[str, Any] | None,
857) -> dict[str, Any] | None:
858 """Return a blocking attribution finding when the gate is active, else ``None``.
860 Only runs when the evidence gate is active (``enforced``) *and* PR labels were
861 actually fetched (``pr_labels is not None``): the presence check is cheap and
862 default-on, while the vendor cross-check engages only when the ledger recorded
863 an implementer vendor. Degrades gracefully — labels not available skips the
864 check entirely, no record means presence-only, and a gate-inactive run skips
865 the check (back-compat with callers that never pass labels).
866 """
867 if not enforced or pr_labels is None:
868 return None
869 implementer_vendor = ledger_implementer_vendor(ledger_record)
870 result = attribution_check(pr_labels, implementer_vendor=implementer_vendor)
871 if result["ok"]:
872 return None
873 if result["reason"] == "missing-label":
874 message = "PR is missing a mandatory agent:<vendor> attribution label."
875 else:
876 message = (
877 "PR agent:<vendor> attribution "
878 f"({', '.join(result['label_vendors'])}) does not match the ship_run "
879 f"ledger implementer vendor ({result['implementer_vendor']})."
880 )
881 return {
882 "id": "attribution-label",
883 "severity": "major",
884 "kind": "attribution",
885 "message": message,
886 }
889def _reviewer_key(item: dict[str, Any], body: str) -> str:
890 fields = _fields(body)
891 reviewer = fields.get("reviewer")
892 if reviewer:
893 return f"reviewer:{reviewer.lower()}"
894 user = item.get("user")
895 if isinstance(user, dict) and isinstance(user.get("login"), str) and user["login"]:
896 return f"user:{user['login'].lower()}"
897 digest = hashlib.sha256(body.encode("utf-8")).hexdigest()
898 return f"body:{digest}"
901def _matches_head(item: dict[str, Any], body: str, head_sha: str | None) -> bool:
902 if not head_sha:
903 return True
904 fields = _fields(body)
905 recorded = fields.get("head")
906 if recorded:
907 return recorded == head_sha
908 commit_id = item.get("commit_id")
909 return isinstance(commit_id, str) and commit_id == head_sha
912def _fields(body: str) -> dict[str, str]:
913 return {match.group("key").lower(): match.group("value")
914 for match in _FIELD_RE.finditer(body)}
917def _is_review_verdict_body(body: str) -> bool:
918 if not body or _is_ship_assessment(body) or _has_closure_marker(body):
919 return False
920 if JURY_VERDICT_MARKER in body:
921 return False
922 return REVIEW_VERDICT_MARKER in body
925def _has_trusted_review_marker(items: list[dict[str, Any]]) -> bool:
926 return any(
927 _is_trusted_source(item, enforced=True) and REVIEW_VERDICT_MARKER in _body(item)
928 for item in items
929 )
932def jury_participating_vendors(
933 pr_comments: list[dict[str, Any]] | None = None,
934 pr_reviews: list[dict[str, Any]] | None = None,
935 *,
936 head_sha: str | None = None,
937 enforced: bool = True,
938) -> int | None:
939 """Return the panel size declared by a posted jury verdict, or ``None``.
941 Reads the ``vendors: <N>`` field from a trusted, head-bound
942 ``keel.jury-verdict.v1`` comment. This is how the participating-vendor count
943 reaches a CI evidence check: the run ledger and the jury artifact both live
944 under the gitignored ``.keel/state/``, so a hosted runner cannot read either,
945 but PR comments are always visible.
947 ``None`` means "not declared" — no jury verdict posted, or one that predates
948 the field — and leaves the jury mode untouched rather than assuming a short
949 panel. Only a verdict that actually states the count may relax the gate.
951 When several verdicts qualify, the largest declared count wins: a re-post
952 correcting an earlier partial run should not be capped by the stale one.
953 """
954 counts = [
955 parsed
956 for item in [*(pr_comments or []), *(pr_reviews or [])]
957 if _is_jury_verdict(item, head_sha=head_sha, enforced=enforced)
958 if (parsed := _parse_vendor_count(_fields(_body(item)).get("vendors"))) is not None
959 ]
960 return max(counts) if counts else None
963def _parse_vendor_count(raw: str | None) -> int | None:
964 """Parse a declared vendor count, rejecting anything not a plain non-negative int."""
965 if raw is None:
966 return None
967 try:
968 parsed = int(raw)
969 except ValueError:
970 return None
971 return parsed if parsed >= 0 else None
974def _is_jury_verdict(
975 item: dict[str, Any],
976 *,
977 head_sha: str | None = None,
978 enforced: bool = True,
979) -> bool:
980 if not _is_trusted_source(item, enforced=enforced):
981 return False
982 body = _body(item)
983 if not body or _is_ship_assessment(body) or _has_closure_marker(body):
984 return False
985 return JURY_VERDICT_MARKER in body and _matches_head(item, body, head_sha)