Coverage for src/keel/checkpoint.py: 100%
225 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 12:05 +0000
1"""Resumable checkpoint helpers for keel work runs."""
3from __future__ import annotations
5import json
6from pathlib import Path
7from typing import Any
9from . import config as cfg
10from . import model, workspace
12CHECKPOINT_SCHEMA_VERSION = "keel.checkpoint.v1"
13DEFAULT_CHECKPOINT_PATH = ".keel/state/checkpoint.json"
14RECORD_TYPE_RUN_CHECKPOINT = "run_checkpoint"
16COMMANDS = ("ship", "work-block", "overnight")
17STEP_IDS = tuple(step.id for step in model.BACKBONE)
18_STEP_IDS_SET = frozenset(STEP_IDS)
19MERGE_STATES = ("not-started", "pending", "merged", "failed", "skipped")
20CAPTURE_STATES = ("not-started", "applied", "deferred", "skipped", "failed")
21CLOSE_STATES = ("not-started", "closed", "failed")
22LIVE_PR_STATES = ("unknown", "missing", "open", "merged", "closed")
23LIVE_WORKTREE_STATES = ("unknown", "present", "missing")
25_IDEMPOTENT_STEPS = {
26 "s0": "load-config",
27 "s1": "select-or-reconcile-issue",
28 "s2": "ensure-branch-and-worktree",
29 "s3": "rerun-guards",
30 "s4": "continue-implementation",
31 "s5": "reclassify-diff",
32 "s6": "recheck-ci",
33 "s7": "rerun-review",
34 "s8": "rerun-test",
35 "s9": "continue-fixloop",
36 "s10": "revalidate-merge-state",
37 "s11": "run-or-verify-capture",
38 "s12": "run-or-verify-close",
39}
42class CheckpointError(ValueError):
43 """Raised when a checkpoint cannot be decoded as the stable schema."""
46def checkpoint_contract_as_dict(config: cfg.ProjectConfig) -> dict[str, Any]:
47 """Return the project-neutral checkpoint storage and resume contract."""
48 path, source = configured_checkpoint_path(config)
49 return {
50 "schema_version": CHECKPOINT_SCHEMA_VERSION,
51 "format": "json",
52 "path": path,
53 "path_source": source,
54 "missing_handling": "no-checkpoint",
55 "write_owner": ["ship", "work-block", "overnight"],
56 "resume_command": "resume",
57 "checkpoint_command": "checkpoint",
58 "steps": [
59 {
60 "step_id": step.id,
61 "step_name": step.name,
62 "safe_boundary": step.id in _IDEMPOTENT_STEPS,
63 "resume_action": _IDEMPOTENT_STEPS.get(step.id),
64 }
65 for step in model.BACKBONE
66 ],
67 "consumer_neutral": True,
68 }
71def configured_checkpoint_path(config: cfg.ProjectConfig) -> tuple[str, str]:
72 """Return the configured checkpoint path and source."""
73 pack = config.policy_pack or {}
74 reports = pack.get("reports") if isinstance(pack.get("reports"), dict) else {}
75 value = reports.get("checkpoint")
76 if isinstance(value, str) and value.strip():
77 return value, "policy_pack.reports.checkpoint"
78 return DEFAULT_CHECKPOINT_PATH, "default"
81def resolve_path(root: str | Path, config: cfg.ProjectConfig) -> Path:
82 """Resolve the checkpoint path under ``root`` and reject escapes."""
83 raw, _ = configured_checkpoint_path(config)
84 path = Path(raw)
85 if path.is_absolute():
86 raise CheckpointError("checkpoint path must be relative to the project root")
87 root_path = Path(root).resolve()
88 resolved = (root_path / path).resolve()
89 try:
90 resolved.relative_to(root_path)
91 except ValueError as exc:
92 raise CheckpointError("checkpoint path escapes the project root") from exc
93 return resolved
96def build_checkpoint_record(
97 *,
98 run_id: str,
99 command: str,
100 current_step: str,
101 base_branch: str,
102 target: str | None = None,
103 issue_queue: list[int] | None = None,
104 active_issue: int | None = None,
105 branch: str | None = None,
106 worktree: str | None = None,
107 pull_request: int | None = None,
108 head_sha: str | None = None,
109 completed_steps: list[str] | None = None,
110 last_gate: str | None = None,
111 last_review: str | None = None,
112 last_check: str | None = None,
113 jury_mode: str | None = None,
114 merge_state: str = "not-started",
115 capture_state: str = "not-started",
116 close_state: str = "not-started",
117 stop_reason: str | None = None,
118) -> dict[str, Any]:
119 """Build one deterministic checkpoint record."""
120 record = {
121 "schema_version": CHECKPOINT_SCHEMA_VERSION,
122 "record_type": RECORD_TYPE_RUN_CHECKPOINT,
123 "run_id": run_id,
124 "command": command,
125 "target": target,
126 "queue": {
127 "issues": list(issue_queue or ()),
128 "active_issue": active_issue,
129 },
130 "position": {
131 "current_step": current_step,
132 "completed_steps": list(completed_steps or ()),
133 },
134 "identifiers": {
135 "base_branch": base_branch,
136 "branch": branch,
137 "worktree": worktree,
138 "pull_request": pull_request,
139 "head_sha": head_sha,
140 },
141 "state": {
142 "last_gate": last_gate,
143 "last_review": last_review,
144 "last_check": last_check,
145 "jury_mode": jury_mode,
146 "merge": merge_state,
147 "capture": capture_state,
148 "close": close_state,
149 "stop_reason": stop_reason,
150 },
151 "resume": {
152 "safe_boundary": current_step in _IDEMPOTENT_STEPS,
153 "action": _IDEMPOTENT_STEPS.get(current_step),
154 "must_reconcile_live_state": True,
155 "repeat_policy": {
156 "merge": "never-repeat-if-live-pr-merged",
157 "comments": "idempotent-anchor-or-skip",
158 "worktree": "refuse-unrelated-existing-path",
159 },
160 },
161 }
162 validate_checkpoint(record)
163 return record
166def encode_checkpoint(record: dict[str, Any]) -> str:
167 """Encode one checkpoint as stable JSON."""
168 validate_checkpoint(record)
169 return json.dumps(record, indent=2, sort_keys=True) + "\n"
172def parse_checkpoint(text: str) -> dict[str, Any]:
173 """Parse and validate one checkpoint record."""
174 try:
175 record = json.loads(text)
176 except json.JSONDecodeError as exc:
177 raise CheckpointError("invalid JSON") from exc
178 validate_checkpoint(record)
179 return record
182def read_checkpoint(path: str | Path) -> dict[str, Any] | None:
183 """Read a checkpoint; a missing checkpoint means there is no resumable run."""
184 checkpoint_path = Path(path)
185 if not checkpoint_path.exists():
186 return None
187 return parse_checkpoint(checkpoint_path.read_text(encoding="utf-8"))
190def write_checkpoint(path: str | Path, record: dict[str, Any]) -> None:
191 """Write one validated checkpoint, replacing the previous resume point."""
192 checkpoint_path = Path(path)
193 checkpoint_path.parent.mkdir(parents=True, exist_ok=True)
194 workspace.ensure_runtime_gitignore_for(checkpoint_path)
195 checkpoint_path.write_text(encode_checkpoint(record), encoding="utf-8")
198def resume_plan_as_dict(
199 record: dict[str, Any] | None,
200 *,
201 live_pr_state: str = "unknown",
202 live_worktree_state: str = "unknown",
203 live_head_sha: str | None = None,
204) -> dict[str, Any]:
205 """Return a deterministic dry-run resume plan after live-state reconciliation.
207 ``live_head_sha`` is the branch's actual head. The checkpoint records the head it
208 was written at, and nothing compared the two (#635), so a branch that moved since
209 resumed with stale context and no signal. A mismatch warns rather than blocks: the
210 common cause is a legitimate push that crashed before the next checkpoint, and s10
211 still fails closed on its own (``ledger.gates_pass_for_head`` pins the merge to the
212 live head).
213 """
214 if live_pr_state not in LIVE_PR_STATES:
215 raise CheckpointError("unsupported live_pr_state")
216 if live_worktree_state not in LIVE_WORKTREE_STATES:
217 raise CheckpointError("unsupported live_worktree_state")
218 if record is None:
219 return {
220 "status": "no-checkpoint",
221 "can_resume": False,
222 "reason": "checkpoint file is missing",
223 "next_step": None,
224 "resume_action": None,
225 "reconcile": _live_state(live_pr_state, live_worktree_state),
226 "warnings": [],
227 }
228 validate_checkpoint(record)
229 warnings: list[str] = []
230 state = record["state"]
231 identifiers = record["identifiers"]
232 current_step = record["position"]["current_step"]
233 status = "ready"
234 can_resume = True
235 reason = "resume from the recorded safe boundary"
236 next_step = current_step
237 action = record["resume"]["action"]
239 # A checkpoint claiming `merge: merged` used to win over *any* live state, so a
240 # checkpoint written optimistically before the merge landed sent every later resume
241 # straight to capture and close — closing the issue and flipping status:done on a PR
242 # that never merged. The post-hoc auditor cannot catch it either: `closeorder`
243 # attests the merge *decision*, not the merge. Live evidence that the PR is not
244 # merged is the stronger signal and makes this ambiguous.
245 recorded_head = identifiers.get("head_sha")
246 if live_head_sha and recorded_head and live_head_sha != recorded_head:
247 warnings.append(
248 f"branch head moved since the checkpoint was written "
249 f"({recorded_head[:12]} -> {live_head_sha[:12]}); the recorded context may be "
250 "stale — re-read the diff before resuming"
251 )
253 claimed_merged = state["merge"] == "merged"
254 contradicted = (
255 claimed_merged
256 and live_pr_state in {"open", "closed", "missing"}
257 and identifiers.get("pull_request")
258 )
259 if contradicted:
260 status = "ambiguous"
261 can_resume = False
262 reason = (
263 "checkpoint records the merge as complete but live state reports the "
264 f"pull request {live_pr_state}"
265 )
266 warnings.append(
267 "confirm whether the merge actually landed before resuming; resuming here "
268 "would close the issue for a merge that may never have happened"
269 )
270 elif live_pr_state == "merged" or claimed_merged:
271 if state["capture"] in {"not-started", "failed"}:
272 status = "needs-capture"
273 next_step = "s11"
274 action = _IDEMPOTENT_STEPS["s11"]
275 reason = "merge is complete; resume at capture without repeating merge"
276 elif state["close"] in {"not-started", "failed"}:
277 status = "needs-close"
278 next_step = "s12"
279 action = _IDEMPOTENT_STEPS["s12"]
280 reason = "capture is complete or skipped; resume at close"
281 else:
282 status = "complete"
283 can_resume = False
284 next_step = None
285 action = None
286 reason = "merge, capture, and close are already complete"
287 elif state["merge"] == "pending" and live_pr_state == "unknown" \
288 and identifiers.get("pull_request"):
289 # A crash *during* the merge is the genuinely ambiguous case, and it used to
290 # resume as a plain `pr-open / next: s10` with no warning at all (#635). Only
291 # live evidence resolves it: `merged` and `open` are handled above/below and
292 # both answer the question. `unknown` does not, so refuse to guess.
293 status = "ambiguous"
294 can_resume = False
295 reason = (
296 "checkpoint records the merge as in-flight and live state is unknown; "
297 "whether the merge landed cannot be determined from the checkpoint alone"
298 )
299 warnings.append(
300 "check whether the pull request merged before resuming; resuming here "
301 "would re-attempt a merge that may already have landed"
302 )
303 elif live_pr_state == "closed" and identifiers.get("pull_request"):
304 status = "ambiguous"
305 can_resume = False
306 reason = "checkpoint references a pull request that live state reports closed"
307 warnings.append("reopen the PR or reconcile the checkpoint before resuming")
308 elif live_worktree_state == "missing" and identifiers.get("worktree"):
309 status = "ambiguous"
310 can_resume = False
311 reason = "checkpoint references a worktree that live state reports missing"
312 warnings.append("recreate or reconcile the recorded worktree before resuming")
313 elif live_pr_state == "missing" and identifiers.get("pull_request"):
314 status = "ambiguous"
315 can_resume = False
316 reason = "checkpoint references a pull request that live state reports missing"
317 warnings.append("verify whether the PR was deleted or the checkpoint is stale")
318 elif current_step == "s6":
319 status = "waiting-on-ci"
320 reason = "resume by rechecking CI before review, test, or merge"
321 elif identifiers.get("pull_request"):
322 status = "pr-open"
323 reason = "resume against the recorded pull request after refreshing live state"
325 return {
326 "status": status,
327 "can_resume": can_resume,
328 "reason": reason,
329 "next_step": next_step,
330 "resume_action": action,
331 "checkpoint": record,
332 "reconcile": _live_state(live_pr_state, live_worktree_state),
333 "warnings": warnings,
334 }
337COVERAGE_STATES = ("covered", "missing", "stale-step")
340def covering_checkpoint(
341 record: dict[str, Any] | None,
342 run_id: str,
343 expected_step: str,
344) -> dict[str, Any]:
345 """Decide whether a checkpoint covers ``run_id`` at ``expected_step``.
347 Pure and deterministic — no I/O. ``record`` is the current checkpoint
348 (or ``None`` when no checkpoint file exists); ``expected_step`` must be a
349 backbone step id. A run is *covered* when its checkpoint is for the same
350 ``run_id`` and the run actually progressed to ``expected_step`` (the
351 recorded ``current_step`` is at or past it, or the step is in
352 ``completed_steps``). The result distinguishes:
354 * ``covered`` — a current checkpoint for this run reached ``expected_step``;
355 * ``missing`` — no checkpoint, or a checkpoint for a different run;
356 * ``stale-step`` — a checkpoint for this run that has not reached the step.
357 """
358 if expected_step not in STEP_IDS:
359 raise CheckpointError("expected_step must be a backbone step id")
360 expected_index = STEP_IDS.index(expected_step)
361 if record is None:
362 return {
363 "status": "missing",
364 "covered": False,
365 "run_id": run_id,
366 "expected_step": expected_step,
367 "checkpoint_run_id": None,
368 "checkpoint_step": None,
369 "reason": f"no current checkpoint for run {run_id} at step {expected_step}",
370 }
371 validate_checkpoint(record)
372 checkpoint_run_id = record.get("run_id")
373 position = record["position"]
374 current_step = position["current_step"]
375 completed = position.get("completed_steps", [])
376 base = {
377 "run_id": run_id,
378 "expected_step": expected_step,
379 "checkpoint_run_id": checkpoint_run_id,
380 "checkpoint_step": current_step,
381 }
382 if checkpoint_run_id != run_id:
383 return {
384 **base,
385 "status": "missing",
386 "covered": False,
387 "reason": (
388 f"no current checkpoint for run {run_id} at step {expected_step} "
389 f"(checkpoint is for run {checkpoint_run_id})"
390 ),
391 }
392 reached = expected_step in completed or STEP_IDS.index(current_step) >= expected_index
393 if not reached:
394 return {
395 **base,
396 "status": "stale-step",
397 "covered": False,
398 "reason": (
399 f"no current checkpoint for run {run_id} at step {expected_step} "
400 f"(run is at {current_step})"
401 ),
402 }
403 return {
404 **base,
405 "status": "covered",
406 "covered": True,
407 "reason": f"checkpoint for run {run_id} reached step {expected_step}",
408 }
411def find_orphans(
412 *,
413 live_branches: list[str] | None = None,
414 live_pull_requests: list[int] | None = None,
415 checkpoint_record: dict[str, Any] | None = None,
416 ledger_records: list[dict[str, Any]] | None = None,
417) -> dict[str, Any]:
418 """Return live branches/PRs with no covering checkpoint or ledger record.
420 Pure and deterministic — no I/O. An *orphan* is a branch or pull request
421 that exists on the git/transport side but is referenced by neither the
422 current checkpoint nor any ledger record, i.e. keel has no covering state
423 for it (the GAP-13 hazard from the git side). Advisory only.
424 """
425 known_branches, known_prs = _known_references(checkpoint_record, ledger_records)
426 orphan_branches = [
427 branch for branch in (live_branches or []) if branch not in known_branches
428 ]
429 orphan_prs = [pr for pr in (live_pull_requests or []) if pr not in known_prs]
430 return {
431 "branches": orphan_branches,
432 "pull_requests": orphan_prs,
433 "orphan_count": len(orphan_branches) + len(orphan_prs),
434 "known_branches": sorted(known_branches),
435 "known_pull_requests": sorted(known_prs),
436 }
439def _known_references(
440 checkpoint_record: dict[str, Any] | None,
441 ledger_records: list[dict[str, Any]] | None,
442) -> tuple[set[str], set[int]]:
443 branches: set[str] = set()
444 prs: set[int] = set()
445 if checkpoint_record is not None:
446 identifiers = checkpoint_record.get("identifiers", {})
447 branch = identifiers.get("branch")
448 if isinstance(branch, str) and branch:
449 branches.add(branch)
450 pull_request = identifiers.get("pull_request")
451 if isinstance(pull_request, int):
452 prs.add(pull_request)
453 for record in ledger_records or []:
454 git = record.get("git") if isinstance(record.get("git"), dict) else {}
455 branch = git.get("branch")
456 if isinstance(branch, str) and branch:
457 branches.add(branch)
458 pr_entry = record.get("pull_request")
459 if isinstance(pr_entry, dict):
460 number = pr_entry.get("number")
461 if isinstance(number, int):
462 prs.add(number)
463 return branches, prs
466def validate_checkpoint(record: Any) -> None:
467 """Validate the stable checkpoint shape."""
468 if not isinstance(record, dict):
469 raise CheckpointError("checkpoint must be an object")
470 if record.get("schema_version") != CHECKPOINT_SCHEMA_VERSION:
471 raise CheckpointError("unsupported schema_version")
472 if record.get("record_type") != RECORD_TYPE_RUN_CHECKPOINT:
473 raise CheckpointError("unsupported record_type")
474 if record.get("command") not in COMMANDS:
475 raise CheckpointError("unsupported command")
476 queue = record.get("queue")
477 if not isinstance(queue, dict) or not isinstance(queue.get("issues"), list):
478 raise CheckpointError("queue must be an object with issues")
479 position = record.get("position")
480 if not isinstance(position, dict) or position.get("current_step") not in _IDEMPOTENT_STEPS:
481 raise CheckpointError("unsupported current_step")
482 completed = position.get("completed_steps")
483 if not isinstance(completed, list):
484 raise CheckpointError("unsupported completed_steps")
485 try:
486 if not _STEP_IDS_SET.issuperset(completed):
487 raise CheckpointError("unsupported completed_steps")
488 except TypeError as err:
489 raise CheckpointError("unsupported completed_steps") from err
490 identifiers = record.get("identifiers")
491 if not isinstance(identifiers, dict) or "base_branch" not in identifiers:
492 raise CheckpointError("identifiers must include base_branch")
493 state = record.get("state")
494 if not isinstance(state, dict):
495 raise CheckpointError("state must be an object")
496 if state.get("merge") not in MERGE_STATES:
497 raise CheckpointError("unsupported merge state")
498 if state.get("capture") not in CAPTURE_STATES:
499 raise CheckpointError("unsupported capture state")
500 if state.get("close") not in CLOSE_STATES:
501 raise CheckpointError("unsupported close state")
502 resume = record.get("resume")
503 if not isinstance(resume, dict):
504 raise CheckpointError("resume must be an object")
505 if resume.get("action") != _IDEMPOTENT_STEPS[position["current_step"]]:
506 raise CheckpointError("resume action does not match current_step")
507 if not isinstance(resume.get("repeat_policy"), dict):
508 raise CheckpointError("resume repeat_policy must be an object")
511def _live_state(live_pr_state: str, live_worktree_state: str) -> dict[str, str]:
512 return {
513 "pull_request": live_pr_state,
514 "worktree": live_worktree_state,
515 "source": "live-git-github-state-or-adapter-supplied-dry-run-state",
516 }