Coverage for src/keel/ship.py: 100%
88 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-16 18:07 +0000
1"""The deterministic ship decisions — keel's value-add as pure functions.
3The agentic steps and the git/gh plumbing live in the adapter + I/O layer; the
4*decisions* (how many reviewers, whether to merge / defer / block, whether to keep
5fixing) are pure and live here, so they are reproducible and fully unit-tested.
6"""
8from __future__ import annotations
10from dataclasses import dataclass
11from typing import Any
13from . import classify
14from .findings import Verdict
15from .window import is_merge_open
17#: Hard cap on review→fix rounds (matches ship's budget).
18MAX_FIX_ROUNDS = 3
20#: GitHub check-rollup conclusions that count as "not failing".
21CI_OK_STATES = frozenset({"SUCCESS", "NEUTRAL", "SKIPPED"})
23POSTING_MODES = frozenset({"inline", "summary"})
25REVIEW_FOCUS_A = (
26 "logic correctness",
27 "null safety",
28 "language interop",
29)
30REVIEW_FOCUS_B = (
31 "platform compatibility",
32 "lifecycle safety",
33 "API compatibility",
34 "threading",
35)
36REVIEW_FOCUS_C = (
37 "test coverage",
38 "docs gate",
39 "scope creep",
40 "CI prediction",
41 "security",
42)
45def reviewer_count(tier: int) -> int:
46 """Reviewers for a risk tier: TIER-3→3, TIER-2→2, TIER-1→1 (default 2)."""
47 return {3: 3, 2: 2, 1: 1}.get(tier, 2)
50def reviewer_focuses(count: int) -> tuple[dict[str, Any], ...]:
51 """Focus coverage for each reviewer slot. Lower counts merge focus; none are dropped."""
52 if count <= 1:
53 return ({
54 "slot": "A",
55 "focus": list(REVIEW_FOCUS_A + REVIEW_FOCUS_B + REVIEW_FOCUS_C),
56 "merged_from": ["A", "B", "C"],
57 },)
58 if count == 2:
59 return (
60 {
61 "slot": "A",
62 "focus": list(REVIEW_FOCUS_A + REVIEW_FOCUS_B),
63 "merged_from": ["A", "B"],
64 },
65 {
66 "slot": "C",
67 "focus": list(REVIEW_FOCUS_C),
68 "merged_from": ["C"],
69 },
70 )
71 return (
72 {"slot": "A", "focus": list(REVIEW_FOCUS_A), "merged_from": ["A"]},
73 {"slot": "B", "focus": list(REVIEW_FOCUS_B), "merged_from": ["B"]},
74 {"slot": "C", "focus": list(REVIEW_FOCUS_C), "merged_from": ["C"]},
75 )
78def resolve_jury(
79 *,
80 tier: int | None,
81 gates: tuple[str, ...] = (),
82 jury: bool = False,
83 no_jury: bool = False,
84 jury_advisory: bool = False,
85) -> dict[str, Any]:
86 """Resolve the cross-vendor jury mode using ship flag precedence."""
87 if no_jury:
88 enabled = False
89 reason = "--no-jury"
90 elif jury:
91 enabled = True
92 reason = "--jury"
93 elif tier == 3:
94 enabled = True
95 reason = "tier-3 auto"
96 else:
97 enabled = False
98 reason = "default"
99 mode = "off" if not enabled else ("advisory" if jury_advisory else "gating")
100 return {
101 "enabled": enabled,
102 "mode": mode,
103 "reason": reason,
104 "configured_gate": "jury" in gates,
105 "fail_soft": True,
106 "minimum_vendors": 2,
107 "verified_consensus_gates": enabled and mode == "gating",
108 "severity_policy": {
109 "critical": "block",
110 "major": "block",
111 "minor": "gated-suggestion",
112 "nit": "advisory",
113 },
114 }
117def resolve_review_contract(
118 *,
119 tier: int | None,
120 reviewer_override: int | None = None,
121 review_comments: str = "inline",
122 gates: tuple[str, ...] = (),
123 policy_pack: dict[str, Any] | None = None,
124 jury: bool = False,
125 no_jury: bool = False,
126 jury_advisory: bool = False,
127 require_distinct_vendors: bool = False,
128) -> dict[str, Any]:
129 """Machine-readable review, jury, test, and merge-gate plan for ship-like flows."""
130 if reviewer_override is not None and reviewer_override not in {1, 2, 3}:
131 raise ValueError("reviewer_override must be one of 1, 2, or 3")
132 if review_comments not in POSTING_MODES:
133 raise ValueError("review_comments must be 'inline' or 'summary'")
134 count = reviewer_override if reviewer_override is not None else reviewer_count(tier or 2)
135 source = (
136 "override" if reviewer_override is not None
137 else ("risk-tier" if tier is not None else "unresolved")
138 )
139 pack = policy_pack or {}
140 review_policy = pack.get("review", {}) if isinstance(pack.get("review", {}), dict) else {}
141 return {
142 "reviewers": {
143 "count": count,
144 "source": source,
145 "tier": tier,
146 "independent": True,
147 "self_review_counts_toward_lgtm": False,
148 "minimum_lgtm": count,
149 "require_distinct_vendors": bool(require_distinct_vendors),
150 "orchestrator_owns_writes": True,
151 "focuses": list(reviewer_focuses(count)),
152 "project_additions": list(review_policy.get("additions", [])),
153 "required_sections": list(review_policy.get("required_sections", [])),
154 },
155 "posting": {
156 "mode": review_comments,
157 "inline_default": True,
158 "per_reviewer_inline_fallback": "summary",
159 "summary_mode": review_comments == "summary",
160 },
161 "jury": resolve_jury(
162 tier=tier,
163 gates=gates,
164 jury=jury,
165 no_jury=no_jury,
166 jury_advisory=jury_advisory,
167 ),
168 "finding_policy": {
169 "critical": "block",
170 "major": "block",
171 "minor": "gated-suggestion",
172 "nit": "advisory",
173 "suggestions_require_fix_or_explicit_deferral": True,
174 "parser_source": "reviewer-returned-findings",
175 },
176 "fixloop": {
177 "max_rounds": MAX_FIX_ROUNDS,
178 "blocker_rerun": "full-review",
179 "suggestion_only_rerun": "narrowed-originating-focus",
180 },
181 "ci": {
182 "failure_before_pending": True,
183 "empty_check_set_allowed_for_docs_only": True,
184 "retry_budget": 3,
185 },
186 "test_gates": {
187 "configured_gates": list(gates),
188 "no_jury_preserves_review_and_test_gates": True,
189 },
190 "merge_gate": {
191 "merge_window_applies_to": "literal-merge-only",
192 "merge_lock_scope": "literal-merge-only",
193 "final_mergeability_recheck_inside_lock": True,
194 "hotfix_bypasses_window_only": True,
195 "hotfix_never_bypasses_findings_or_ci": True,
196 "pr_merged_state_authoritative": True,
197 },
198 "closeout": {
199 "comment_targets": ["issue", "pull_request"],
200 "capture_marker_required": True,
201 "status_done_after_merge_only": True,
202 },
203 }
206@dataclass(frozen=True)
207class MergeDecision:
208 action: str # "merge" | "defer" | "block"
209 reason: str
212def decide_merge(verdict: Verdict, *, window_open: bool, is_blocker: bool = False) -> MergeDecision:
213 """Decide what to do with a green-or-not PR given the window.
215 * blocking findings ⇒ **block** (never merges);
216 * outside the merge window and not a blocker ⇒ **defer** to the morning queue;
217 * otherwise ⇒ **merge**. A blocker bypasses the window (but never the findings).
218 """
219 if verdict.blocked:
220 return MergeDecision("block", "blocking findings present")
221 if not window_open and not is_blocker:
222 return MergeDecision("defer", "outside merge window (night no-merge)")
223 reason = "blocker bypass" if (is_blocker and not window_open) else "clear to merge"
224 return MergeDecision("merge", reason)
227def should_run_fixloop(verdict: Verdict, *, current_round: int, cap: int = MAX_FIX_ROUNDS) -> bool:
228 """True if there are blocking findings and the fix budget is not exhausted."""
229 return verdict.blocked and current_round < cap
232def ci_passing(ci_conclusion: str | None) -> bool | None:
233 """Interpret a check-rollup string (e.g. ``"SUCCESS,FAILURE"``). ``None`` == unknown."""
234 if ci_conclusion is None:
235 return None
236 parts = [p.strip().upper() for p in ci_conclusion.split(",") if p.strip()]
237 if not parts:
238 return None
239 return all(p in CI_OK_STATES for p in parts)
242def is_hotfix(labels: list[str] | tuple[str, ...], *, hotfix_label: str = "hotfix") -> bool:
243 """True if the issue/PR carries the hotfix label (case-insensitive)."""
244 return any(label.strip().lower() == hotfix_label.lower() for label in labels)
247@dataclass(frozen=True)
248class ShipAssessment:
249 tier: int
250 reviewers: int
251 window_open: bool
252 ci_ok: bool | None
253 merge: MergeDecision
254 halted: bool = False # pause mode + outside window ⇒ pipeline halted
255 bypassed_window: bool = False # hotfix merged outside the window (audited)
256 review_contract: dict[str, Any] | None = None
259def assess(
260 *,
261 changed_files: list[str],
262 gate_verdict: Verdict,
263 tier3_globs: tuple[str, ...] = (),
264 docs_globs: tuple[str, ...] = (),
265 timezone: str | None = None,
266 merge_window: str | None = None,
267 merge_window_mode: str = "freeze",
268 ci_conclusion: str | None = None,
269 now=None,
270 is_blocker: bool = False,
271 reviewer_override: int | None = None,
272 review_comments: str = "inline",
273 gates: tuple[str, ...] = (),
274 policy_pack: dict[str, Any] | None = None,
275 jury: bool = False,
276 no_jury: bool = False,
277 jury_advisory: bool = False,
278) -> ShipAssessment:
279 """The whole deterministic ship decision in one place: tier → reviewers, window,
280 CI, and the final merge action. Pure — identical inputs give identical output.
282 ``merge_window_mode`` 'pause' halts the pipeline outside the window; 'freeze'
283 (default) only blocks the merge. ``is_blocker`` (a hotfix) bypasses the window —
284 but never the findings or a failing CI."""
285 tier = classify.tier_for_files(changed_files, tier3_globs=tier3_globs, docs_globs=docs_globs)
286 reviewers = reviewer_override if reviewer_override is not None else reviewer_count(tier)
287 window_open = (
288 is_merge_open(timezone, merge_window, now=now) if (timezone and merge_window) else True
289 )
290 halted = (merge_window_mode == "pause") and not window_open and not is_blocker
291 ci_ok = ci_passing(ci_conclusion)
292 if ci_ok is False:
293 merge = MergeDecision("block", "CI failing")
294 else:
295 merge = decide_merge(gate_verdict, window_open=window_open, is_blocker=is_blocker)
296 bypassed = is_blocker and not window_open and merge.action == "merge"
297 review_contract = resolve_review_contract(
298 tier=tier,
299 reviewer_override=reviewer_override,
300 review_comments=review_comments,
301 gates=gates,
302 policy_pack=policy_pack,
303 jury=jury,
304 no_jury=no_jury,
305 jury_advisory=jury_advisory,
306 )
307 return ShipAssessment(
308 tier, reviewers, window_open, ci_ok, merge, halted, bypassed, review_contract
309 )