Coverage for src/keel/stepverifier.py: 100%
163 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-02 20:26 +0000
1"""Deterministic backbone step completion verification.
3Agentic ship steps may be performed by different runtimes, but advancing the
4backbone must not depend on private prose. This module defines the shared
5"done" contract for each step, the structured handoff shape between steps, and
6the fail-closed transition check adapters can run before moving forward.
8:data:`HANDOFF_FIELDS` is the single declaration of the handoff schema. The producer
9(:func:`build_handoff`), the published contract (:func:`contract_as_dict`) and the
10verifier (:func:`_check_handoff_schema`) all read it, so the verifier cannot fall
11behind the shape the renderer emits — which is exactly how it came to check two of
12ten fields while the renderer emitted all ten (#1101).
13"""
15from __future__ import annotations
17from dataclasses import dataclass
18from typing import Any
20from . import artifacts, evidence, model, provenance
22SCHEMA_VERSION = "keel.step-verification.v1"
23HANDOFF_SCHEMA_VERSION = "keel.step-handoff.v1"
24HANDOFF_MARKER = artifacts.STEP_HANDOFF_MARKER
25COMPLETE_STATUS = "complete"
27#: JSON type names the handoff schema uses, and the Python type each one is once the
28#: document has been read back out of JSON.
29_JSON_TYPES: dict[str, type] = {"string": str, "array": list, "object": dict}
32@dataclass(frozen=True)
33class HandoffField:
34 """One field of the published ``keel.step-handoff.v1`` handoff object."""
36 name: str
37 json_type: str
38 #: ``None`` is a value the canonical renderer legitimately emits for this field.
39 nullable: bool = False
40 #: An empty string or an empty array is legitimate for this field.
41 may_be_blank: bool = False
43 def as_dict(self) -> dict[str, Any]:
44 return {"name": self.name, "type": self.json_type, "nullable": self.nullable}
46 def problem(self, handoff: dict[str, Any]) -> str | None:
47 """Name what is wrong with this field in ``handoff``, or return ``None``.
49 Every reason names the field, because "something is missing" is not a
50 message an agent can act on and not one an operator can audit.
51 """
52 if self.name not in handoff:
53 return f"handoff field missing: {self.name}"
54 value = handoff[self.name]
55 if value is None:
56 return None if self.nullable else f"handoff field null: {self.name}"
57 if not isinstance(value, _JSON_TYPES[self.json_type]):
58 return f"handoff field wrong type: {self.name}"
59 if not self.may_be_blank and not _has_content(value):
60 return f"handoff field empty: {self.name}"
61 return None
64#: The published handoff schema — the *one* declaration shared with the producer.
65#: A field added here is emitted by :func:`build_handoff`, published by
66#: :func:`contract_as_dict` and required by the verifier in the same commit. A second,
67#: hand-kept list is how the verifier fell eight fields behind the renderer (#1101).
68HANDOFF_FIELDS: tuple[HandoffField, ...] = (
69 HandoffField("schema_version", "string"),
70 HandoffField("step_id", "string"),
71 HandoffField("step_name", "string"),
72 HandoffField("status", "string"),
73 HandoffField("summary", "string"),
74 HandoffField("evidence_ids", "array", may_be_blank=True),
75 HandoffField("next_step", "string", nullable=True, may_be_blank=True),
76 HandoffField("producer", "string", nullable=True, may_be_blank=True),
77 HandoffField("provenance", "object"),
78 HandoffField("rendered", "string"),
79)
82@dataclass(frozen=True)
83class StepRequirement:
84 """Required evidence for one backbone step."""
86 step_id: str
87 step_name: str
88 required_evidence: tuple[str, ...] = ()
89 verifier: str = "keel.stepverifier.verify_step_completion"
91 def as_dict(self) -> dict[str, Any]:
92 return {
93 "step_id": self.step_id,
94 "step_name": self.step_name,
95 "required_evidence": list(self.required_evidence),
96 "verifier": self.verifier,
97 }
100def contract_as_dict(
101 review_contract: dict[str, Any],
102 *,
103 dry_run: bool = False,
104 enforced: bool = True,
105) -> dict[str, Any]:
106 """Return the deterministic step-completion contract for ship-like flows."""
107 del dry_run # The contract describes the required done-state even for dry-run output.
108 requirements = step_requirements(review_contract, dry_run=False, enforced=enforced)
109 return {
110 "schema_version": SCHEMA_VERSION,
111 "consumer_neutral": True,
112 "deterministic": True,
113 "fail_closed": True,
114 "dry_run_disables_runtime_gating": True,
115 "source": "backbone_plan + evidence",
116 "no_premature_termination": True,
117 "handoff_schema": {
118 "schema_version": HANDOFF_SCHEMA_VERSION,
119 "required_fields": [field.name for field in HANDOFF_FIELDS],
120 "fields": [field.as_dict() for field in HANDOFF_FIELDS],
121 "renderer": "keel.artifacts.render_step_handoff",
122 "marker": HANDOFF_MARKER,
123 "rendered_body_matches_fields": True,
124 "completed_step_claims_required_evidence": True,
125 },
126 "completion_rule": (
127 "A step may transition as success only when its structured handoff has "
128 "status=complete and every required evidence id for that step is ok."
129 ),
130 "steps": [requirement.as_dict() for requirement in requirements],
131 }
134def step_requirements(
135 review_contract: dict[str, Any],
136 *,
137 dry_run: bool = False,
138 enforced: bool = True,
139) -> tuple[StepRequirement, ...]:
140 """Map the public evidence contract onto the fixed backbone steps."""
141 evidence_ids = [
142 item.id
143 for item in evidence.required_items(
144 review_contract,
145 dry_run=dry_run,
146 enforced=enforced,
147 )
148 ]
149 by_step = {
150 "s7": tuple(item for item in evidence_ids if item.startswith("review-verdict-")),
151 "s8": tuple(item for item in evidence_ids if item == "jury-verdict"),
152 "s12": tuple(item for item in evidence_ids if item.startswith("closure-comment-")),
153 }
154 return tuple(
155 StepRequirement(
156 step_id=step.id,
157 step_name=step.name,
158 required_evidence=by_step.get(step.id, ()),
159 )
160 for step in model.BACKBONE
161 )
164def build_handoff(
165 *,
166 step_id: str,
167 status: str = COMPLETE_STATUS,
168 summary: str | None = None,
169 evidence_ids: tuple[str, ...] | list[str] = (),
170 next_step: str | None = None,
171 producer: str | None = None,
172 vendor: str | None = None,
173 model_name: str | None = None,
174 allowed_capabilities: tuple[str, ...] | list[str] = (),
175) -> dict[str, Any]:
176 """Build a structured handoff object rendered through canonical artifacts."""
177 step = model.get_step(step_id)
178 clean_evidence = tuple(
179 item.strip() for item in evidence_ids if isinstance(item, str) and item.strip()
180 )
181 rendered = artifacts.render_step_handoff(
182 step_id=step.id,
183 step_name=step.name,
184 status=status,
185 summary=summary,
186 next_step=next_step,
187 evidence_ids=clean_evidence,
188 )
189 values = {
190 "schema_version": HANDOFF_SCHEMA_VERSION,
191 "step_id": step.id,
192 "step_name": step.name,
193 "status": status,
194 # `.strip()` before the default: a whitespace-only summary is truthy, so the
195 # bare `or` stored it as written and the verifier — which refuses a blank
196 # field — rejected a document this very function had produced. A producer
197 # that can emit what its own verifier refuses is the synchronisation this
198 # change is about, pointing the other way.
199 "summary": (summary or "").strip() or "No summary recorded.",
200 "evidence_ids": list(clean_evidence),
201 "next_step": next_step,
202 "producer": producer,
203 "provenance": provenance.source_tag(
204 source_agent=producer,
205 step_id=step.id,
206 vendor=vendor,
207 model=model_name,
208 allowed_capabilities=allowed_capabilities,
209 ),
210 "rendered": rendered,
211 }
212 # Emit the schema, not a dict that happens to resemble it: a field declared in
213 # HANDOFF_FIELDS and not built above raises here rather than shipping a handoff
214 # the verifier will refuse, and one built above and not declared cannot leak into
215 # the document at all.
216 return {field.name: values[field.name] for field in HANDOFF_FIELDS}
219def verify_step_completion(
220 *,
221 step_id: str,
222 handoff: dict[str, Any] | None,
223 evidence_report: dict[str, Any] | None,
224 review_contract: dict[str, Any],
225 dry_run: bool = False,
226 enforced: bool = True,
227) -> dict[str, Any]:
228 """Verify that one step can be marked complete without trusting prose."""
229 requirement = _requirement_for(
230 step_id,
231 review_contract,
232 dry_run=dry_run,
233 enforced=enforced,
234 )
235 checks = [
236 _check_handoff_schema(step_id, handoff),
237 _check_handoff_status(handoff),
238 _check_handoff_marker(handoff),
239 _check_handoff_provenance(step_id, handoff),
240 _check_handoff_evidence(requirement, handoff),
241 _check_required_evidence(requirement, evidence_report),
242 ]
243 missing = [reason for check in checks if not check["ok"] for reason in check["missing"]]
244 return {
245 "schema_version": SCHEMA_VERSION,
246 "step_id": step_id,
247 "status": "pass" if not missing else "fail",
248 "no_premature_termination": True,
249 "required_evidence": list(requirement.required_evidence),
250 "missing": missing,
251 "checks": checks,
252 }
255def _requirement_for(
256 step_id: str,
257 review_contract: dict[str, Any],
258 *,
259 dry_run: bool,
260 enforced: bool,
261) -> StepRequirement:
262 requirements = {
263 requirement.step_id: requirement
264 for requirement in step_requirements(
265 review_contract,
266 dry_run=dry_run,
267 enforced=enforced,
268 )
269 }
270 if step_id not in requirements:
271 raise KeyError(f"unknown backbone step: {step_id}")
272 return requirements[step_id]
275def _check_handoff_schema(step_id: str, handoff: dict[str, Any] | None) -> dict[str, Any]:
276 """Check the whole published schema, naming every field that is absent or wrong."""
277 if not isinstance(handoff, dict):
278 return _check("handoff_schema", False, "handoff missing")
279 problems = [field.problem(handoff) for field in HANDOFF_FIELDS]
280 named = [problem for problem in problems if problem]
281 if named:
282 return _check("handoff_schema", False, *named)
283 if handoff["schema_version"] != HANDOFF_SCHEMA_VERSION:
284 return _check("handoff_schema", False, "handoff schema mismatch")
285 if handoff["step_id"] != step_id:
286 return _check("handoff_schema", False, "handoff step mismatch")
287 # The backbone names its own steps, so a handoff that names one wrong was not
288 # written from the step it claims to be reporting.
289 if handoff["step_name"] != model.get_step(step_id).name:
290 return _check("handoff_schema", False, "handoff step name mismatch")
291 return _check("handoff_schema", True)
294def _check_handoff_status(handoff: dict[str, Any] | None) -> dict[str, Any]:
295 if not isinstance(handoff, dict):
296 return _check("handoff_status", False, "handoff missing")
297 if handoff.get("status") != COMPLETE_STATUS:
298 return _check("handoff_status", False, "handoff not complete")
299 return _check("handoff_status", True)
302def _check_handoff_marker(handoff: dict[str, Any] | None) -> dict[str, Any]:
303 """Check the rendered body against the renderer, not against a substring.
305 A marker can be typed; the canonical rendering of *these* fields cannot be typed
306 into disagreement with them. Re-rendering the handoff's own structured fields and
307 comparing catches a body pasted from another step, or one whose prose says
308 something the fields do not.
309 """
310 if not isinstance(handoff, dict):
311 return _check("handoff_renderer", False, "handoff missing")
312 rendered = handoff.get("rendered")
313 if not isinstance(rendered, str) or HANDOFF_MARKER not in rendered:
314 return _check("handoff_renderer", False, "canonical handoff renderer missing")
315 if rendered != _render_from(handoff):
316 return _check("handoff_renderer", False, "handoff body does not match its own fields")
317 return _check("handoff_renderer", True)
320def _render_from(handoff: dict[str, Any]) -> str:
321 evidence_ids = handoff.get("evidence_ids")
322 return artifacts.render_step_handoff(
323 step_id=handoff.get("step_id"),
324 step_name=handoff.get("step_name"),
325 status=handoff.get("status"),
326 summary=handoff.get("summary"),
327 next_step=handoff.get("next_step"),
328 evidence_ids=evidence_ids if isinstance(evidence_ids, list) else [],
329 )
332def _canonical_provenance_keys() -> tuple[frozenset[str], frozenset[str], frozenset[str]]:
333 """The key sets ``provenance.source_tag`` emits: tag, ``source``, ``capability_scope``.
335 Derived from the producer rather than retyped, for the same reason
336 :data:`HANDOFF_FIELDS` is: a field added to the tag must not be able to go
337 unchecked here. The *keys* are required, not their values — ``vendor`` and
338 ``model`` are legitimately ``None`` when the caller knows neither.
339 """
340 tag = provenance.source_tag(source_agent="a", step_id="s")
341 return (
342 frozenset(tag),
343 frozenset(tag["source"]),
344 frozenset(tag["capability_scope"]),
345 )
348def _check_handoff_provenance(step_id: str, handoff: dict[str, Any] | None) -> dict[str, Any]:
349 """Check that the provenance tag is a canonical one, bound to this step.
351 The evidence chain downstream reads this tag. A handoff carrying an arbitrary
352 object under ``provenance`` has answered "who says so?" with nothing.
353 """
354 if not isinstance(handoff, dict):
355 return _check("handoff_provenance", False, "handoff missing")
356 tag = handoff.get("provenance")
357 if not isinstance(tag, dict):
358 return _check("handoff_provenance", False, "handoff field wrong type: provenance")
359 if tag.get("schema_version") != provenance.SCHEMA_VERSION:
360 return _check("handoff_provenance", False, "handoff provenance schema mismatch")
361 if (
362 tag.get("role") != provenance.UNTRUSTED_ROLE
363 or tag.get("trusted_as_instructions") is not False
364 ):
365 return _check("handoff_provenance", False, "handoff provenance is not untrusted output")
366 scope = tag.get("capability_scope")
367 if not isinstance(scope, dict) or scope.get("can_expand_capabilities") is not False:
368 return _check("handoff_provenance", False, "handoff provenance expands capabilities")
369 source = tag.get("source")
370 if not isinstance(source, dict):
371 return _check("handoff_provenance", False, "handoff provenance names no source")
372 if source.get("step_id") != step_id:
373 return _check("handoff_provenance", False, "handoff provenance step mismatch")
374 # Required unconditionally, not only when ``producer`` names someone. ``source_tag``
375 # always writes this key — ``unknown-agent`` when the caller passes no agent — so a
376 # tag that omits it was not built by the renderer, and the one field answering "who
377 # says so?" is exactly the field a fabricated tag has no reason to fill in. Gating it
378 # on a named ``producer`` let a handoff with ``producer: null`` clear the check by
379 # naming nobody, which is the shape of defect this whole verifier exists to refuse.
380 agent_id = source.get("agent_id")
381 if not (isinstance(agent_id, str) and agent_id.strip()):
382 return _check("handoff_provenance", False, "handoff provenance names no agent")
383 producer = handoff.get("producer")
384 named = producer.strip() if isinstance(producer, str) else ""
385 if named and agent_id != named:
386 return _check("handoff_provenance", False, "handoff provenance producer mismatch")
387 # Last, so a specific violation is still reported specifically. "Canonical" has to
388 # mean the shape the renderer emits, or the word is doing no work: a tag carrying
389 # only the members the checks above happen to read is not the tag `source_tag`
390 # builds, and accepting it claims a check that was never made.
391 tag_keys, source_keys, scope_keys = _canonical_provenance_keys()
392 for present, required, where in (
393 (tag, tag_keys, "provenance"),
394 (source, source_keys, "provenance.source"),
395 (scope, scope_keys, "provenance.capability_scope"),
396 ):
397 missing = sorted(required - set(present))
398 if missing:
399 return _check(
400 "handoff_provenance",
401 False,
402 f"handoff provenance is not canonical, {where} missing: {', '.join(missing)}",
403 )
404 # Types, not values. `vendor` and `model` are free-form strings the producer
405 # copies from the run, so there is nothing here to compare them against; what can
406 # be said is that a list is a list. Extra keys are deliberately allowed — a tag
407 # carrying a member this version has not heard of is a newer producer, not a
408 # forgery, and refusing it would make the verifier the reason a handoff cannot
409 # cross a version boundary.
410 for key in ("vendor", "model"):
411 if source[key] is not None and not isinstance(source[key], str):
412 return _check(
413 "handoff_provenance", False, f"handoff provenance source.{key} is not a string"
414 )
415 for key in ("allowed_capabilities", "unknown_capabilities"):
416 if not isinstance(scope[key], list):
417 return _check(
418 "handoff_provenance",
419 False,
420 f"handoff provenance capability_scope.{key} is not a list",
421 )
422 return _check("handoff_provenance", True)
425def _check_handoff_evidence(
426 requirement: StepRequirement,
427 handoff: dict[str, Any] | None,
428) -> dict[str, Any]:
429 """Check that a completed handoff claims the evidence its step owes.
431 The required-evidence check below reads a *separately supplied* report. This one
432 binds the handoff to it: a step that reports completed work without naming the
433 evidence for it is the fabrication this verifier exists to refuse.
434 """
435 if not isinstance(handoff, dict):
436 return _check("handoff_evidence", False, "handoff missing")
437 claimed = handoff.get("evidence_ids")
438 if not isinstance(claimed, list):
439 return _check("handoff_evidence", False, "handoff field wrong type: evidence_ids")
440 if any(not (isinstance(item, str) and item.strip()) for item in claimed):
441 return _check("handoff_evidence", False, "handoff evidence ids are not all named")
442 if handoff.get("status") != COMPLETE_STATUS:
443 return _check("handoff_evidence", True)
444 ids = {item.strip() for item in claimed}
445 unclaimed = [item for item in requirement.required_evidence if item not in ids]
446 return _check(
447 "handoff_evidence",
448 not unclaimed,
449 *(f"handoff claims no evidence id: {item}" for item in unclaimed),
450 )
453def _check_required_evidence(
454 requirement: StepRequirement,
455 evidence_report: dict[str, Any] | None,
456) -> dict[str, Any]:
457 if not requirement.required_evidence:
458 return _check("required_evidence", True)
459 ok_ids = {
460 result["id"]
461 for result in _evidence_results(evidence_report)
462 if result.get("ok") is True and isinstance(result.get("id"), str)
463 }
464 missing = [
465 evidence_id for evidence_id in requirement.required_evidence if evidence_id not in ok_ids
466 ]
467 return _check("required_evidence", not missing, *missing)
470def _evidence_results(evidence_report: dict[str, Any] | None) -> tuple[dict[str, Any], ...]:
471 if not isinstance(evidence_report, dict):
472 return ()
473 results = evidence_report.get("results")
474 if not isinstance(results, list):
475 return ()
476 return tuple(item for item in results if isinstance(item, dict))
479def _has_content(value: Any) -> bool:
480 return bool(value.strip()) if isinstance(value, str) else bool(value)
483def _check(name: str, ok: bool, *missing: str) -> dict[str, Any]:
484 return {
485 "name": name,
486 "ok": ok,
487 "missing": list(missing),
488 }