Coverage for src/keel/ledger.py: 100%
207 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 12:05 +0000
1"""Structured run ledger helpers for keel workflows."""
3from __future__ import annotations
5import json
6from pathlib import Path
7from typing import Any
9from . import capture, redaction, workspace
10from . import config as cfg
12LEDGER_SCHEMA_VERSION = "keel.run-ledger.v1"
13CAPTURE_HEALTH_SCHEMA_VERSION = "keel.capture-health.v1"
14DEFAULT_LEDGER_PATH = ".keel/state/run-ledger.jsonl"
15RECORD_TYPE_SHIP_RUN = "ship_run"
18class LedgerError(ValueError):
19 """Raised when a ledger file cannot be decoded as the stable schema."""
22def ledger_contract_as_dict(config: cfg.ProjectConfig) -> dict[str, Any]:
23 """Return the project-neutral ledger storage and schema contract."""
24 path, source = configured_ledger_path(config)
25 return {
26 "schema_version": LEDGER_SCHEMA_VERSION,
27 "format": "jsonl",
28 "path": path,
29 "path_source": source,
30 "missing_handling": "treat-as-empty",
31 "append_owner": ["ship"],
32 "readers": ["morning", "wrap", "overnight", "capture-verification", "ledger"],
33 "consumer_neutral": True,
34 "capture_redaction": redaction.contract_as_dict(config),
35 "capture_contract": capture.contract_as_dict(config),
36 "capture_health": capture_health_contract_as_dict(),
37 "record_types": [RECORD_TYPE_SHIP_RUN],
38 }
41def capture_health_contract_as_dict() -> dict[str, Any]:
42 """Return the ledger-derived capture-health summary contract."""
43 return {
44 "schema_version": CAPTURE_HEALTH_SCHEMA_VERSION,
45 "source": "run-ledger ship_run records",
46 "readers": ["morning", "wrap", "status", "ledger"],
47 "consumer_neutral": True,
48 "missing_ledger_handling": "clean-empty-history",
49 "dry_run": {
50 "no_mutations": True,
51 "safe_reconcile_actions_only": True,
52 },
53 "states": ["clean", "needs-reconcile"],
54 "item_statuses": ["applied", "deferred", "skipped", "missing-marker"],
55 }
58def configured_ledger_path(config: cfg.ProjectConfig) -> tuple[str, str]:
59 """Return the configured ledger path and the config source that supplied it."""
60 pack = config.policy_pack or {}
61 reports = pack.get("reports") if isinstance(pack.get("reports"), dict) else {}
62 value = reports.get("run_ledger")
63 if isinstance(value, str) and value.strip():
64 return value, "policy_pack.reports.run_ledger"
65 return DEFAULT_LEDGER_PATH, "default"
68def resolve_path(root: str | Path, config: cfg.ProjectConfig) -> Path:
69 """Resolve the configured ledger path under ``root`` and reject escapes."""
70 raw, _ = configured_ledger_path(config)
71 path = Path(raw)
72 if path.is_absolute():
73 raise LedgerError("run ledger path must be relative to the project root")
74 root_path = Path(root).resolve()
75 resolved = (root_path / path).resolve()
76 try:
77 resolved.relative_to(root_path)
78 except ValueError as exc:
79 raise LedgerError("run ledger path escapes the project root") from exc
80 return resolved
83def build_ship_run_record(
84 *,
85 command: str,
86 base_branch: str,
87 changed_files: list[str] | None,
88 declared_files: list[str] | None = None,
89 outcomes: list[Any],
90 verdict: Any,
91 assessment: Any,
92 issue_intake: dict[str, Any] | None = None,
93 target: str | None = None,
94 run_id: str | None = None,
95 issue_number: int | None = None,
96 pr_number: int | None = None,
97 branch: str | None = None,
98 head_sha: str | None = None,
99 capture_status: str | None = None,
100 capture_reason: str | None = None,
101 capture_artifact: str | None = None,
102 issue_title: str | None = None,
103 issue_labels: list[str] | tuple[str, ...] = (),
104 existing_records: list[dict[str, Any]] | None = None,
105 config: cfg.ProjectConfig | None = None,
106 implementer: str | None = None,
107 reviewer_agents: list[str] | None = None,
108 tester: str | None = None,
109 host_agent: str | None = None,
110 transport: str | None = None,
111 profile: str | None = None,
112 jury_mode: str | None = None,
113 consent_status: str | None = None,
114 consent_scopes: list[str] | tuple[str, ...] | None = None,
115 run_controls: dict[str, Any] | None = None,
116) -> dict[str, Any]:
117 """Build one deterministic consumer-neutral ship ledger record."""
118 return {
119 "schema_version": LEDGER_SCHEMA_VERSION,
120 "record_type": RECORD_TYPE_SHIP_RUN,
121 "command": command,
122 "run_id": run_id,
123 "target": target,
124 "issue": {"number": issue_number},
125 "pull_request": {"number": pr_number},
126 "git": {
127 "base_branch": base_branch,
128 "branch": branch,
129 "head_sha": head_sha,
130 },
131 # `None` means git could not be read, and stays distinct from `[]` all the way
132 # into the record: a consumer of the ledger (or of the closure comment rendered
133 # from it) must not read "we could not see the diff" as "the diff was empty" —
134 # that is the conflation this whole family of fixes exists to remove, and a
135 # record claiming TIER-3 with zero files is self-contradictory besides.
136 "changes": {
137 "file_count": None if changed_files is None else len(changed_files),
138 "files": None if changed_files is None else list(changed_files),
139 "unreadable": changed_files is None,
140 },
141 "declared": _declared_block(declared_files),
142 "gates": [
143 {
144 "gate": outcome.gate,
145 "ok": outcome.ok,
146 "skipped": outcome.skipped,
147 "timed_out": outcome.timed_out,
148 # getattr: outcomes reach here from adapters and fixtures that predate
149 # these fields. Defaulting not_run to False keeps an older producer's
150 # record readable; on_fail defaults to the strict value so a record
151 # that *does* carry not_run without a severity fails closed.
152 "not_run": getattr(outcome, "not_run", False),
153 "on_fail": getattr(outcome, "on_fail", "block"),
154 "error": outcome.error,
155 "finding_count": len(outcome.findings),
156 }
157 for outcome in outcomes
158 ],
159 "verdict": {
160 "blocked": verdict.blocked,
161 "counts": dict(verdict.counts),
162 },
163 "assessment": {
164 "tier": assessment.tier,
165 "reviewers": assessment.reviewers,
166 "window_open": assessment.window_open,
167 "ci_ok": assessment.ci_ok,
168 "merge": {
169 "action": assessment.merge.action,
170 "reason": assessment.merge.reason,
171 },
172 "halted": assessment.halted,
173 "bypassed_window": assessment.bypassed_window,
174 },
175 "actors": {
176 "implementer": implementer,
177 "reviewers": list(reviewer_agents or ()),
178 "tester": tester,
179 },
180 "run_context": _run_context(
181 host_agent=host_agent,
182 transport=transport,
183 profile=profile,
184 jury_mode=jury_mode,
185 consent_status=consent_status,
186 consent_scopes=consent_scopes,
187 ),
188 "run_controls": run_controls,
189 "issue_intake": issue_intake,
190 "capture": capture.record_marker(
191 pr_number=pr_number,
192 status=capture_status,
193 reason=capture_reason,
194 artifact=capture_artifact,
195 title=issue_title,
196 labels=issue_labels,
197 changed_files=changed_files,
198 existing_records=existing_records or [],
199 config=config,
200 ),
201 }
204def _declared_block(declared_files: list[str] | None) -> dict[str, Any] | None:
205 """Build the implementer's declared-scope block, or ``None`` when unset.
207 ``declared_files`` is the implementer's contract of which files the change is
208 *supposed* to touch (distinct from the observed ``changes`` diff). When the
209 implementer does not declare a scope, the block is omitted so readers can
210 degrade to advisory back-compat behavior.
211 """
212 if declared_files is None:
213 return None
214 files = [str(path) for path in declared_files]
215 return {"file_count": len(files), "files": files}
218def declared_files_for_record(record: dict[str, Any]) -> list[str] | None:
219 """Return the implementer's declared file list from a ship_run ``record``.
221 Returns ``None`` when no declared scope was recorded (a missing or malformed
222 ``declared`` block), letting ``scope-verify`` degrade to an advisory pass.
223 """
224 declared = record.get("declared")
225 if not isinstance(declared, dict):
226 return None
227 files = declared.get("files")
228 if not isinstance(files, list):
229 return None
230 return [str(path) for path in files]
233def _run_context(
234 *,
235 host_agent: str | None,
236 transport: str | None,
237 profile: str | None,
238 jury_mode: str | None,
239 consent_status: str | None,
240 consent_scopes: list[str] | tuple[str, ...] | None,
241) -> dict[str, Any]:
242 """Build the deterministic consumer-neutral preflight run-context block.
244 Every field is optional; a missing scalar degrades to ``None`` so the
245 closure renderer can present ``unknown``/``none`` without a schema change.
246 Consent is a small summary: a status and the approved mutation scopes,
247 reusing the operator/approve-scope inputs already resolved by the caller.
248 """
249 scopes = [str(scope) for scope in (consent_scopes or ()) if str(scope).strip()]
250 return {
251 "host_agent": host_agent if _nonblank(host_agent) else None,
252 "transport": transport if _nonblank(transport) else None,
253 "profile": profile if _nonblank(profile) else None,
254 "jury_mode": jury_mode if _nonblank(jury_mode) else None,
255 "consent": {
256 "status": consent_status if _nonblank(consent_status) else None,
257 "scopes": scopes,
258 },
259 }
262def _nonblank(value: Any) -> bool:
263 return isinstance(value, str) and bool(value.strip())
266def encode_record(record: dict[str, Any]) -> str:
267 """Encode one ledger record as stable JSONL."""
268 _validate_record(record)
269 return json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n"
272def parse_records(text: str) -> list[dict[str, Any]]:
273 """Parse ledger JSONL text into validated records."""
274 records: list[dict[str, Any]] = []
275 for line_number, raw in enumerate(text.splitlines(), start=1):
276 if not raw.strip():
277 continue
278 try:
279 record = json.loads(raw)
280 except json.JSONDecodeError as exc:
281 raise LedgerError(f"line {line_number}: invalid JSON") from exc
282 _validate_record(record, line_number=line_number)
283 records.append(record)
284 return records
287def read_records(path: str | Path) -> list[dict[str, Any]]:
288 """Read a ledger file; a missing ledger is a valid empty history."""
289 ledger_path = Path(path)
290 if not ledger_path.exists():
291 return []
292 return parse_records(ledger_path.read_text(encoding="utf-8"))
295def latest_ship_run_for_pr(
296 records: list[dict[str, Any]],
297 pr_number: int,
298) -> dict[str, Any] | None:
299 """Return the last ship_run record whose pull_request matches ``pr_number``.
301 Records are appended in chronological order, so the last match is the most
302 recent ship run for that PR. Returns ``None`` when no record matches.
303 """
304 match: dict[str, Any] | None = None
305 for record in records:
306 if record.get("record_type") != RECORD_TYPE_SHIP_RUN:
307 continue
308 pull_request = record.get("pull_request")
309 number = pull_request.get("number") if isinstance(pull_request, dict) else None
310 if number == pr_number:
311 match = record
312 return match
315def record_gates_passed(record: dict[str, Any]) -> bool:
316 """Return whether a ship_run record's gates count as a clean pass.
318 A pass requires that the run was not blocked by findings and that every
319 recorded gate either ran clean (``ok``) or was deliberately skipped, with no
320 gate reporting an error. A missing or malformed ``gates``/``verdict`` block
321 degrades to "not a pass" so a corrupt record can never authorize a merge.
323 A gate marked ``not_run`` was never executed by the runner that wrote the
324 record — an ``agentic`` gate reaching the command-only runner. For a
325 ``block``-severity gate that is **not** a pass: certifying "gates passed" for a
326 blocking review nobody ran is exactly the fail-open this check exists to stop.
327 Soft (``warn``/``suggest``) not-run gates are tolerated, matching ``skipped``.
328 Records written before this field existed have no ``not_run`` key and are
329 unaffected.
330 """
331 verdict = record.get("verdict")
332 if not isinstance(verdict, dict) or verdict.get("blocked") is not False:
333 return False
334 gates = record.get("gates")
335 if not isinstance(gates, list) or not gates:
336 return False
337 for gate in gates:
338 if not isinstance(gate, dict):
339 return False
340 if gate.get("error"):
341 return False
342 if not (gate.get("ok") is True or gate.get("skipped") is True):
343 return False
344 # Missing `on_fail` defaults to the strict value *here*, at read time. Defaulting
345 # only at write time protects records keel wrote and no others: any producer that
346 # learns `not_run` without its sibling key would otherwise fail open in exactly
347 # the certification path this check exists to close.
348 # Fail closed on anything not explicitly soft: a missing key, a JSON-round-tripped
349 # `None`, or a severity name keel does not know all mean "we cannot tell this was
350 # optional", and this is the certification path.
351 if gate.get("not_run") is True and gate.get("on_fail") not in ("warn", "suggest"):
352 return False
353 return True
356def gates_pass_for_head(
357 records: list[dict[str, Any]],
358 pr_number: int,
359 head_sha: str,
360) -> tuple[bool, dict[str, Any] | None]:
361 """Find a passing gates run recorded against ``head_sha`` for ``pr_number``.
363 Returns ``(matched, record)``. ``matched`` is ``True`` only when the **latest**
364 ship_run record for the PR carrying the exact current ``head_sha`` passed its
365 gates (see :func:`record_gates_passed`); the record is returned on a match and
366 ``None`` otherwise. A blank ``head_sha`` never matches — an unknown head must
367 not be authorized by a stale green run. This is a pure function: it reads only
368 its arguments and performs no I/O.
370 Latest-wins, not any-pass. Re-gating the same head is ordinary — a flaky suite
371 settles, a fix-loop re-runs, a dependency lands — so the same ``head_sha`` can
372 carry a green record followed by a red one. Scanning for *any* green would let
373 the superseded pass authorize the merge and the later red never be consulted:
374 a fail-open in exactly the gate that exists to hold the merge closed. Only the
375 most recent verdict for the head counts.
376 """
377 if not isinstance(head_sha, str) or not head_sha.strip():
378 return False, None
379 latest: dict[str, Any] | None = None
380 for record in records:
381 if record.get("record_type") != RECORD_TYPE_SHIP_RUN:
382 continue
383 pull_request = record.get("pull_request")
384 number = pull_request.get("number") if isinstance(pull_request, dict) else None
385 if number != pr_number:
386 continue
387 git = record.get("git")
388 record_sha = git.get("head_sha") if isinstance(git, dict) else None
389 if record_sha != head_sha:
390 continue
391 latest = record
392 if latest is None or not record_gates_passed(latest):
393 return False, None
394 return True, latest
397def capture_health_summary(records: list[dict[str, Any]]) -> dict[str, Any]:
398 """Summarize capture visibility for morning, wrap, status, and ledger readers."""
399 items = [_capture_health_item(record) for record in records]
400 counts = {
401 "applied": 0,
402 "marker_only": 0,
403 "create_learning": 0,
404 "duplicate_learning": 0,
405 "deferred": 0,
406 "skipped": 0,
407 "missing_marker": 0,
408 "needs_reconcile": 0,
409 }
410 skipped_by_reason: dict[str, int] = {}
411 for item in items:
412 status = item["status"]
413 learning = item["learning_decision"]
414 if status == "applied":
415 counts["applied"] += 1
416 elif status == "deferred":
417 counts["deferred"] += 1
418 elif status == "skipped":
419 counts["skipped"] += 1
420 skipped_by_reason[item["reason"] or "unspecified"] = (
421 skipped_by_reason.get(item["reason"] or "unspecified", 0) + 1
422 )
423 else:
424 counts["missing_marker"] += 1
425 if learning == "marker-only":
426 counts["marker_only"] += 1
427 elif learning == "create-learning":
428 counts["create_learning"] += 1
429 elif learning == "duplicate":
430 counts["duplicate_learning"] += 1
431 if item["needs_reconcile"]:
432 counts["needs_reconcile"] += 1
433 return {
434 "schema_version": CAPTURE_HEALTH_SCHEMA_VERSION,
435 "status": "needs-reconcile" if counts["needs_reconcile"] else "clean",
436 "record_count": len(records),
437 "counts": counts,
438 "skipped_by_reason": dict(sorted(skipped_by_reason.items())),
439 "items": items,
440 "reconcile_actions": [
441 action for item in items for action in item["reconcile_actions"]
442 ],
443 "dry_run": {
444 "no_mutations": True,
445 "description": "Morning and wrap surface these actions; they do not mutate "
446 "ledger, GitHub, or capture destinations.",
447 },
448 }
451def _capture_marker(record: dict[str, Any]) -> str | None:
452 capture = record.get("capture")
453 marker = capture.get("marker") if isinstance(capture, dict) else None
454 return marker if isinstance(marker, str) and marker.strip() else None
457def existing_capture_marker(
458 records: list[dict[str, Any]], record: dict[str, Any]
459) -> dict[str, Any] | None:
460 """The already-recorded capture marker ``record`` would duplicate, if any.
462 Exactly one capture marker per merged PR is an invariant that was only ever
463 *detected*, never prevented: :func:`keel.capture.verify_session` refuses the whole
464 session on a second one ("multiple capture markers found for merged PR"),
465 ``capture-reconcile`` returns ``blocked`` with no actions to offer, and nothing in
466 this module can remove a line — so the only exit is editing the ledger by hand.
468 Re-running the same append is the most natural thing to do after a crash mid-s11,
469 which made the obvious recovery the very action that bricks the run. Checking here
470 costs one pass over records the caller already holds. Returns the conflicting
471 record so the caller can name it; ``None`` when the append is new.
472 """
473 if _capture_marker(record) is None:
474 return None
475 pull_request = record.get("pull_request")
476 pr = pull_request.get("number") if isinstance(pull_request, dict) else None
477 if not isinstance(pr, int):
478 return None
479 for existing in records:
480 if existing.get("record_type") != RECORD_TYPE_SHIP_RUN:
481 continue
482 other = existing.get("pull_request")
483 if (other.get("number") if isinstance(other, dict) else None) != pr:
484 continue
485 if _capture_marker(existing) is not None:
486 return existing
487 return None
490def append_record(path: str | Path, record: dict[str, Any]) -> None:
491 """Append one validated JSONL record, creating parent directories as needed."""
492 ledger_path = Path(path)
493 ledger_path.parent.mkdir(parents=True, exist_ok=True)
494 workspace.ensure_runtime_gitignore_for(ledger_path)
495 with ledger_path.open("a", encoding="utf-8") as handle:
496 handle.write(encode_record(record))
499def sanitize_record(
500 record: dict[str, Any],
501 config: cfg.ProjectConfig | None = None,
502) -> dict[str, Any]:
503 """Apply capture redaction before a ledger record becomes durable."""
504 result = redaction.sanitize(record, redaction.policy_from_config(config))
505 sanitized = dict(result.value)
506 sanitized["redaction"] = result.audit
507 return sanitized
510def _validate_record(record: Any, *, line_number: int | None = None) -> None:
511 prefix = f"line {line_number}: " if line_number is not None else ""
512 if not isinstance(record, dict):
513 raise LedgerError(f"{prefix}record must be an object")
514 if record.get("schema_version") != LEDGER_SCHEMA_VERSION:
515 raise LedgerError(f"{prefix}unsupported schema_version")
516 if record.get("record_type") != RECORD_TYPE_SHIP_RUN:
517 raise LedgerError(f"{prefix}unsupported record_type")
520def _capture_health_item(record: dict[str, Any]) -> dict[str, Any]:
521 block = record.get("capture") if isinstance(record.get("capture"), dict) else {}
522 status = block.get("status")
523 marker = block.get("marker")
524 reason = block.get("marker_reason") or block.get("reason")
525 learning = block.get("learning") if isinstance(block.get("learning"), dict) else {}
526 item_status = _capture_health_status(status, marker)
527 needs_reconcile = item_status in {"missing-marker", "deferred"}
528 pr_number = (record.get("pull_request") or {}).get("number")
529 item = {
530 "run_id": record.get("run_id"),
531 "issue": (record.get("issue") or {}).get("number"),
532 "pull_request": pr_number,
533 "status": item_status,
534 "capture_status": status,
535 "marker": marker if isinstance(marker, str) and marker.strip() else None,
536 "reason": reason if isinstance(reason, str) and reason.strip() else None,
537 "learning_decision": learning.get("decision"),
538 "learning_reason": learning.get("reason"),
539 "needs_reconcile": needs_reconcile,
540 "reconcile_actions": [],
541 }
542 if needs_reconcile:
543 item["reconcile_actions"].append(_capture_reconcile_action(pr_number, item_status))
544 return item
547def _capture_health_status(status: Any, marker: Any) -> str:
548 if not isinstance(marker, str) or not marker.strip():
549 return "missing-marker"
550 if status == "deferred":
551 return "deferred"
552 if status == "skipped":
553 return "skipped"
554 return "applied"
557def _capture_reconcile_action(pr_number: Any, status: str) -> dict[str, Any]:
558 return {
559 "type": "capture-reconcile",
560 "reason": status,
561 "pr": pr_number if isinstance(pr_number, int) else None,
562 "command": (
563 f"keel capture-reconcile .keel/project.yaml --root . --merged-pr {pr_number}"
564 if isinstance(pr_number, int)
565 else "keel capture-reconcile .keel/project.yaml --root ."
566 ),
567 "dry_run": True,
568 }