Coverage for src/keel/branchscope.py: 100%
47 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-16 18:07 +0000
1"""Pure branch-contract verification — base ancestry + worktree isolation.
3keel's s2 branch contract has three rules: cut the work branch off an
4*up-to-date* ``origin/<base_branch>``; keep the work in one repo-nested,
5gitignored linked worktree per issue; and never mutate the operator's primary
6checkout. A branch cut from a *stale* local base produces phantom diffs and
7wrong tier classification, and edits landing in the primary checkout contaminate
8the operator's tree. Until now that contract was unenforced (audit GAP-5).
10This module makes the contract a pure function of gathered git facts. The CLI
11(:func:`keel.cli._cmd_verify_branch`) collects the live facts via the thin
12``git``/``gh`` wrappers; this module compares them and returns a structured
13verdict. There is no I/O here, so the verdict is a deterministic function of its
14arguments alone and is fully unit-testable offline.
16Two independent checks compose into one verdict:
18* **Base ancestry** — the PR head's merge-base with ``origin/<base_branch>``
19 must equal the current base tip (strict) *or* be a recent ancestor within a
20 bounded tolerance (``base_distance`` commits behind the tip). Beyond the
21 tolerance the branch was cut from a stale base → ``stale``. The
22 ``--allow-stale-base`` operator escape (consent scope ``git``) downgrades a
23 stale verdict to an advisory pass, recorded on the report.
24* **Worktree isolation** — when run locally with a linked worktree nested under
25 the repo root, that is the clean topology. A working branch living in the
26 *primary* checkout (or a worktree outside the repo root) is ``contaminated``.
27 In CI / PR-only mode there is no local worktree to inspect; the check is N/A
28 and skipped gracefully.
29"""
31from __future__ import annotations
33from typing import Any
35SCHEMA_VERSION = "keel.verify-branch.v1"
37#: Default ancestry tolerance: how many commits the merge-base may sit behind the
38#: current base tip before the branch counts as stale. ``0`` is strict (merge-base
39#: must equal the base tip). A small positive default tolerates a base that moved
40#: forward by a few commits after the branch was cut, which is normal and benign.
41DEFAULT_BASE_DISTANCE = 5
44def verify(
45 *,
46 base_branch: str,
47 head_sha: str | None,
48 merge_base_sha: str | None,
49 base_tip_sha: str | None,
50 base_distance: int | None,
51 worktree_path: str | None = None,
52 repo_root: str | None = None,
53 is_linked_worktree: bool | None = None,
54 tolerance: int = DEFAULT_BASE_DISTANCE,
55 allow_stale_base: bool = False,
56) -> dict[str, Any]:
57 """Compare gathered git facts against the s2 branch contract.
59 Parameters describe *facts*, not I/O:
61 * ``head_sha`` / ``merge_base_sha`` / ``base_tip_sha`` — the PR head, its
62 merge-base with ``origin/<base_branch>``, and the current base tip.
63 * ``base_distance`` — commits the merge-base sits behind the base tip
64 (``git rev-list --count merge_base..base_tip``), or ``None`` when it could
65 not be computed.
66 * ``worktree_path`` / ``repo_root`` / ``is_linked_worktree`` — the local
67 working-tree facts; all ``None`` in CI / PR-only mode, where the isolation
68 check is skipped.
69 * ``tolerance`` — max allowed ``base_distance`` before stale (pure knob).
70 * ``allow_stale_base`` — operator escape that downgrades stale to advisory.
72 Returns a JSON-compatible report with a ``status`` of ``pass``/``fail`` and a
73 ``verdict`` of ``ok``/``stale``/``contaminated``, plus per-check detail and an
74 advisory note recording any escape that was applied.
75 """
76 ancestry = _check_ancestry(
77 head_sha=head_sha,
78 merge_base_sha=merge_base_sha,
79 base_tip_sha=base_tip_sha,
80 base_distance=base_distance,
81 tolerance=tolerance,
82 allow_stale_base=allow_stale_base,
83 )
84 isolation = _check_isolation(
85 worktree_path=worktree_path,
86 repo_root=repo_root,
87 is_linked_worktree=is_linked_worktree,
88 )
90 if isolation["verdict"] == "contaminated":
91 verdict = "contaminated"
92 elif ancestry["verdict"] == "stale":
93 verdict = "stale"
94 else:
95 verdict = "ok"
97 blocking = (isolation["status"] == "fail") or (ancestry["status"] == "fail")
98 # Surface the most actionable note: a contamination failure first, then a
99 # stale/advisory ancestry note, then any informational skip.
100 if isolation["verdict"] == "contaminated":
101 note = isolation["note"]
102 elif ancestry["verdict"] == "stale":
103 note = ancestry["note"]
104 else:
105 note = ancestry["note"] or isolation["note"]
106 return {
107 "schema_version": SCHEMA_VERSION,
108 "status": "fail" if blocking else "pass",
109 "verdict": verdict,
110 "base_branch": base_branch,
111 "allow_stale_base": allow_stale_base,
112 "tolerance": tolerance,
113 "note": note,
114 "ancestry": ancestry,
115 "isolation": isolation,
116 }
119def _check_ancestry(
120 *,
121 head_sha: str | None,
122 merge_base_sha: str | None,
123 base_tip_sha: str | None,
124 base_distance: int | None,
125 tolerance: int,
126 allow_stale_base: bool,
127) -> dict[str, Any]:
128 """Verify the PR head was cut from an up-to-date base.
130 The merge-base of the head with the base tip must be the base tip itself
131 (strict) or sit within ``tolerance`` commits behind it. Missing facts (a base
132 or merge-base that could not be resolved) are an advisory skip — fail-soft, so
133 a transient ``gh``/``git`` gap never hard-blocks. ``allow_stale_base``
134 downgrades an otherwise-blocking stale verdict to an advisory pass.
135 """
136 if merge_base_sha is None or base_tip_sha is None or base_distance is None:
137 return {
138 "verdict": "unknown",
139 "status": "pass",
140 "advisory": True,
141 "note": "base ancestry not resolved; skipping",
142 "head_sha": head_sha,
143 "merge_base_sha": merge_base_sha,
144 "base_tip_sha": base_tip_sha,
145 "base_distance": base_distance,
146 "tolerance": tolerance,
147 }
148 up_to_date = merge_base_sha == base_tip_sha
149 within_tolerance = base_distance <= tolerance
150 fresh = up_to_date or within_tolerance
151 if fresh:
152 return {
153 "verdict": "ok",
154 "status": "pass",
155 "advisory": False,
156 "note": None,
157 "head_sha": head_sha,
158 "merge_base_sha": merge_base_sha,
159 "base_tip_sha": base_tip_sha,
160 "base_distance": base_distance,
161 "tolerance": tolerance,
162 }
163 reason = (
164 f"base is stale: head was cut {base_distance} commit(s) behind the current "
165 f"base tip (tolerance {tolerance})"
166 )
167 return {
168 "verdict": "stale",
169 "status": "pass" if allow_stale_base else "fail",
170 "advisory": allow_stale_base,
171 "note": (
172 f"{reason}; downgraded to advisory by --allow-stale-base"
173 if allow_stale_base
174 else reason
175 ),
176 "head_sha": head_sha,
177 "merge_base_sha": merge_base_sha,
178 "base_tip_sha": base_tip_sha,
179 "base_distance": base_distance,
180 "tolerance": tolerance,
181 }
184def _check_isolation(
185 *,
186 worktree_path: str | None,
187 repo_root: str | None,
188 is_linked_worktree: bool | None,
189) -> dict[str, Any]:
190 """Verify the working branch lives in a repo-nested linked worktree.
192 With no local worktree facts (CI / PR-only mode) the check is N/A and skipped
193 gracefully. Locally, a primary-checkout edit (``is_linked_worktree`` false) or
194 a worktree outside the repo root is contamination → fail.
195 """
196 if worktree_path is None or repo_root is None or is_linked_worktree is None:
197 return {
198 "verdict": "n/a",
199 "status": "pass",
200 "advisory": True,
201 "note": "no local worktree (CI/PR-only); isolation check skipped",
202 "worktree_path": worktree_path,
203 "repo_root": repo_root,
204 "is_linked_worktree": is_linked_worktree,
205 "nested": None,
206 }
207 nested = _is_nested(worktree_path, repo_root)
208 if not is_linked_worktree:
209 return {
210 "verdict": "contaminated",
211 "status": "fail",
212 "advisory": False,
213 "note": "working branch is the primary checkout, not a linked worktree",
214 "worktree_path": worktree_path,
215 "repo_root": repo_root,
216 "is_linked_worktree": is_linked_worktree,
217 "nested": nested,
218 }
219 if not nested:
220 return {
221 "verdict": "contaminated",
222 "status": "fail",
223 "advisory": False,
224 "note": "linked worktree is not nested under the repo root",
225 "worktree_path": worktree_path,
226 "repo_root": repo_root,
227 "is_linked_worktree": is_linked_worktree,
228 "nested": nested,
229 }
230 return {
231 "verdict": "ok",
232 "status": "pass",
233 "advisory": False,
234 "note": None,
235 "worktree_path": worktree_path,
236 "repo_root": repo_root,
237 "is_linked_worktree": is_linked_worktree,
238 "nested": nested,
239 }
242def _is_nested(worktree_path: str, repo_root: str) -> bool:
243 """True when ``worktree_path`` is strictly under ``repo_root`` (pure path math).
245 Uses normalized POSIX-style segment comparison so it is deterministic and
246 platform-independent — no filesystem access. A path equal to the root is not
247 "nested" (the primary checkout sits at the root).
248 """
249 root_parts = _segments(repo_root)
250 wt_parts = _segments(worktree_path)
251 if len(wt_parts) <= len(root_parts):
252 return False
253 return wt_parts[: len(root_parts)] == root_parts
256def _segments(path: str) -> list[str]:
257 """Split a path into non-empty, ``.``-free segments (normalized, no I/O)."""
258 normalized = path.replace("\\", "/")
259 return [part for part in normalized.split("/") if part not in ("", ".")]