From f567a28cbd2a1500710166299924b236467da35a Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Sat, 22 Aug 2026 16:41:51 -0400 Subject: [PATCH] feat: add scaffold-verifier + explain operator commands, outcome epilogues, and run social card Two read-first operator helpers close the 'what do I do next' gap after a run ends without a receipt: - flow scaffold-verifier : drafts (never approves) an effect-oracle contract (effect_contract.yaml) from the demonstration's retained write-shaped evidence -- observed system-of-record deltas split identity from payload exactly as compiler.effect_mining does; every observed value carries a TODO; a deployment effects: skeleton matches docs/EFFECT_KIT.md. Refuses demonstrations with no consequential step. - flow explain : pure read-only plain-language outcome summary (what happened / which check fired for a HALT / the exact next command). - outcome-aware 3-line epilogues on non-VERIFIED replay endings, failing lint, and non-VERIFIED tutorial runs: presentation only, fail-closed semantics and exit codes untouched. - scripts/social_card.py: deterministic PHI-free PNG share card from a run's report.json (+receipt.json/bench.json when present); no network, PIL default fonts only. Existing subcommands are unchanged (parser choices diff: only the two additions). --- openadapt_flow/__main__.py | 163 +++++++- openadapt_flow/scaffold_verifier.py | 564 ++++++++++++++++++++++++++++ openadapt_flow/tutorial.py | 58 +++ scripts/social_card.py | 178 +++++++++ tests/test_outcome_epilogues.py | 179 +++++++++ tests/test_scaffold_verifier.py | 342 +++++++++++++++++ 6 files changed, 1483 insertions(+), 1 deletion(-) create mode 100644 openadapt_flow/scaffold_verifier.py create mode 100644 scripts/social_card.py create mode 100644 tests/test_outcome_epilogues.py create mode 100644 tests/test_scaffold_verifier.py diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index 6f91cc94..d0901f53 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -37,6 +37,12 @@ - ``certify`` — enforce a safety policy on a bundle (refuse it if it fails). - ``qualify`` — create and edit the versioned qualification project, import customer-controlled case results, explain refusals, and persist certification. +- ``scaffold-verifier`` — draft (never approve) an effect-oracle contract + (``effect_contract.yaml``) from a recording or bundle's write-shaped steps; + refuses demonstrations with no consequential step. +- ``explain`` — plain-language, read-only read of a completed run directory: + what happened, why the outcome is the safe one, and the next suggested + command. - ``console`` — serve the localhost-only operator console (a read-first web UI over bundles / runs / skill libraries; requires the ``console`` extra). - ``emit-skill`` — emit an Agent Skills folder for a bundle. @@ -774,6 +780,74 @@ def _build_and_run_replayer( ) +def _replay_outcome_epilogue( + report, + outcome: str, + run_dir: Path, +) -> Optional[str]: + """Three-line epilogue after a non-VERIFIED replay (presentation only). + + What happened / why this is the safe behavior / the exact next command. + Fail-closed semantics are untouched: this only explains an ending that + already happened. + """ + from openadapt_flow.tutorial import outcome_epilogue_lines + + halt = getattr(report, "halt", None) + halted_step = next( + ( + result.step_id + for result in report.results + if not result.skipped + and ( + result.effect_verified is False + or result.safety_halt + or ( + result.identity is not None and result.identity.status == "mismatch" + ) + ) + ), + None, + ) + if outcome == "HALTED": + where = f" at step `{halted_step}`" if halted_step else "" + what = f"the run stopped{where} and ended {outcome}" + ( + f" ({halt.reason})" if halt is not None and halt.reason else "" + ) + why_safe = ( + "the engine halts instead of acting on unproven state; nothing " + "further was executed once the check failed" + ) + return "\n".join( + outcome_epilogue_lines( + what=what, + why_safe=why_safe, + next_command=f"openadapt-flow explain {run_dir}", + ) + ) + if outcome == "COMPLETED_UNVERIFIED": + return "\n".join( + outcome_epilogue_lines( + what=( + "every executed step finished on screen, but nothing " + f"independently proved the writes landed ({outcome})" + ), + why_safe=("screen-only completion can never claim success under Flow"), + next_command=( + "openadapt-flow scaffold-verifier " + "# draft an oracle, wire effects:, re-run" + ), + ) + ) + return "\n".join( + outcome_epilogue_lines( + what=f"the run ended {outcome} and reported the failure loudly", + why_safe="a failure is never guessed into a success", + next_command=f"openadapt-flow explain {run_dir}", + ) + ) + + def _finish_replay( run_dir: Path, report, @@ -794,6 +868,10 @@ def _finish_replay( "NOTE: a model-grounding component was wired for this run — " "screenshots could have left the box (see REPORT.md)." ) + if outcome not in {"VERIFIED", "success"}: + epilogue = _replay_outcome_epilogue(report, outcome, run_dir) + if epilogue: + print(f"\n{epilogue}") _maybe_report_break(run_dir, report) _maybe_report_run(run_dir, report, args, backend_kind=backend_kind) _maybe_attest_run(run_dir, report, args) @@ -1216,6 +1294,7 @@ def _cmd_tutorial(args: argparse.Namespace) -> int: TutorialError, _next_steps_block, run_tutorial, + tutorial_epilogue, ) out = ( @@ -1274,6 +1353,10 @@ def _cmd_tutorial(args: argparse.Namespace) -> int: "watch the engine halt:\n openadapt-flow tutorial --break-it" ) print(f"\n{_next_steps_block()}") + else: + # Presentation-only epilogue for the non-VERIFIED endings: what + # happened, why this is the safe behavior, the exact next command. + print("\n" + "\n".join(tutorial_epilogue(result))) if result.execution_outcome != "VERIFIED": return 1 return 0 @@ -2498,7 +2581,47 @@ def _cmd_lint(args: argparse.Namespace) -> int: # (an unarmed or vacuous IRREVERSIBLE step). `--strict` also fails on warn. threshold = "warn" if args.strict else "error" fail = SEVERITY_ORDER[report.max_severity] >= SEVERITY_ORDER[threshold] - return 1 if (report.findings and fail) else 0 + if report.findings and fail: + _print_lint_epilogue(args.bundle, threshold) + return 1 + return 0 + + +def _print_lint_epilogue(bundle: str, threshold: str) -> None: + """Three-line epilogue after a failing lint (presentation only).""" + from openadapt_flow.tutorial import outcome_epilogue_lines + + lines = outcome_epilogue_lines( + what=( + f"lint found coverage gaps at or above the '{threshold}' severity " + f"in {bundle}" + ), + why_safe=( + "gaps are reported instead of silently running unguarded or " + "unverifiable steps" + ), + next_command=f"openadapt-flow certify {bundle} --policy ", + ) + print("\n" + "\n".join(lines)) + + +def _cmd_scaffold_verifier(args: argparse.Namespace) -> int: + """Draft (never approve) an effect-oracle contract from retained evidence.""" + from openadapt_flow.scaffold_verifier import NEXT_COMMANDS_TEMPLATE, write_draft + + out, count = write_draft( + Path(args.source), Path(args.out) if getattr(args, "out", None) else None + ) + print(NEXT_COMMANDS_TEMPLATE.format(out=out, count=count)) + return 0 + + +def _cmd_explain(args: argparse.Namespace) -> int: + """Print a plain-language, read-only summary of one completed run.""" + from openadapt_flow.scaffold_verifier import explain_run + + print(explain_run(Path(args.run_dir))) + return 0 def _cmd_visualize(args: argparse.Namespace) -> int: @@ -5519,6 +5642,44 @@ def build_parser() -> argparse.ArgumentParser: ) p.set_defaults(func=_cmd_lint) + p = sub.add_parser( + "scaffold-verifier", + help=( + "Draft an effect-oracle contract (effect_contract.yaml) from a " + "recording or bundle's write-shaped steps. The output is a DRAFT " + "requiring human edit; refuses demonstrations with no " + "consequential (write) step" + ), + ) + p.add_argument( + "source", + help="Recording directory OR workflow bundle directory", + ) + p.add_argument( + "-o", + "--out", + default=None, + metavar="DIR", + help=( + "Directory for the drafted effect_contract.yaml (default: beside the input)" + ), + ) + p.set_defaults(func=_cmd_scaffold_verifier) + + p = sub.add_parser( + "explain", + help=( + "Plain-language read of a completed run dir: what happened, why " + "the outcome is the safe one, and the next suggested command " + "(read-only)" + ), + ) + p.add_argument( + "run_dir", + help="Completed run directory (holds report.json)", + ) + p.set_defaults(func=_cmd_explain) + p = sub.add_parser( "visualize", help=( diff --git a/openadapt_flow/scaffold_verifier.py b/openadapt_flow/scaffold_verifier.py new file mode 100644 index 00000000..9bbb015a --- /dev/null +++ b/openadapt_flow/scaffold_verifier.py @@ -0,0 +1,564 @@ +"""Operator tooling: draft effect-oracle scaffolds + plain-language run reads. + +Two read-first operator helpers live here. + +``flow scaffold-verifier `` + Reads the retained action/effect evidence in a recording directory or a + compiled bundle and emits a DRAFT effect-oracle contract + (``effect_contract.yaml``) pre-filled from the demonstration's write-shaped + steps: a proposed system-of-record read, expected postcondition fields with + explicit ``TODO`` markers, and a deployment ``effects:`` section skeleton + matching ``docs/EFFECT_KIT.md`` and the shipped + :class:`~openadapt_flow.runtime.effects.RestRecordVerifier` conventions. + + The output is explicitly a DRAFT requiring human edit. Nothing is + auto-approved: the file carries a loud header, every observed value that + needs operator confirmation is marked ``TODO``, and the command prints the + review + qualification commands next. A demonstration with no consequential + (write-shaped) step is REFUSED rather than scaffolded. + +``flow explain `` + Pure read-only plain-language outcome summary over a completed run's + artifacts (``report.json``, ``REPORT.md``, ``receipt.json``): what + happened, why the outcome is the safe one (which check fired for a HALT), + and the exact suggested next command (re-run, pair an oracle via + ``scaffold-verifier``, or qualify). + +Neither helper executes a workflow, contacts a network, or weakens any gate. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional + +BUNDLE_MARKERS = ("workflow.json", "workflow.json.enc") +RECORDING_MARKER = "events.jsonl" + +#: The draft file this module writes. Deliberately YAML: it is the shape an +#: operator copies into ``deployment.yaml`` (``effects:``) and reviews by hand. +CONTRACT_FILENAME = "effect_contract.yaml" + + +class ScaffoldRefused(SystemExit): + """The input has nothing to scaffold from; refuse with a clear message.""" + + +# --------------------------------------------------------------------------- +# Target inspection (recording dir vs bundle dir) +# --------------------------------------------------------------------------- + + +def classify_target(path: Path) -> str: + """Return ``"bundle"`` or ``"recording"`` for an existing artifact path.""" + if not path.exists(): + raise ScaffoldRefused(f"scaffold-verifier: path not found: {path}") + if any((path / marker).is_file() for marker in BUNDLE_MARKERS): + return "bundle" + if (path / RECORDING_MARKER).is_file(): + return "recording" + raise ScaffoldRefused( + f"scaffold-verifier: {path} is neither a workflow bundle " + f"(workflow.json) nor a recording directory ({RECORDING_MARKER})" + ) + + +# --------------------------------------------------------------------------- +# Write-step candidates from retained evidence +# --------------------------------------------------------------------------- + + +@dataclass +class WriteCandidate: + """One write-shaped step observed in the retained evidence.""" + + step_id: str + action: str + #: Human-readable target text the write shape was inferred from. + label: str + #: Why this step is consequential (mirrors risk.py's honest explanations). + basis: str + #: Identity selector for the intended record (observed or compiled). + match: dict[str, str] = field(default_factory=dict) + #: Expected post-write field values (observed payload or bound params). + payload: dict[str, str] = field(default_factory=dict) + #: Observed idempotency key (field, value) when the record carried one. + idempotency: Optional[tuple[str, str]] = None + + @property + def observed_delta(self) -> bool: + return bool(self.match or self.payload) + + +def _event_text(event: dict[str, Any]) -> str: + """Every piece of retained human-readable target text on one event.""" + parts = [str(event.get("text") or "")] + structural = event.get("structural") + if isinstance(structural, dict): + parts.append(str(structural.get("name") or "")) + identity = event.get("structured_identity") + if isinstance(identity, str): + parts.append(identity) + parts.append(str(event.get("field_label") or "")) + return " ".join(part for part in parts if part) + + +def _sor_delta(event: dict[str, Any]) -> tuple[Optional[list], Optional[list]]: + """The captured before/after system-of-record snapshots, when present.""" + from openadapt_flow.compiler.effect_mining import ( + SOR_AFTER_KEY, + SOR_BEFORE_KEY, + _as_records, + ) + + return ( + _as_records(event.get(SOR_BEFORE_KEY)), + _as_records(event.get(SOR_AFTER_KEY)), + ) + + +def _demonstrated_param_values(recording_dir: Path) -> tuple[str, ...]: + """The demonstration's parameter values (from ``meta.json``), if any.""" + import json + + try: + meta = json.loads((recording_dir / "meta.json").read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return () + params = meta.get("params") + if not isinstance(params, dict): + return () + return tuple(str(value) for value in params.values() if value) + + +def _candidate_from_record( + step_id: str, + action: str, + label: str, + record: dict[str, Any], + *, + demonstrated_values: tuple[str, ...], +) -> WriteCandidate: + """Split one observed new record into identity vs payload fields. + + Uses the compiler miner's own helpers so the draft splits identity from + payload exactly as a real mined contract would (same surrogate-id + exclusion, same selector rules, same demonstrated-value payload split). + """ + from openadapt_flow.compiler.effect_mining import ( + IDEMPOTENCY_KEY_FIELD, + _match_selector, + _param_value_fields, + ) + + payload = _param_value_fields(record, demonstrated_values) + selector = _match_selector(record, set(payload)) + idempotency = None + raw_key = record.get(IDEMPOTENCY_KEY_FIELD) + if raw_key not in (None, ""): + idempotency = (IDEMPOTENCY_KEY_FIELD, str(raw_key)) + # The at-most-once key is surfaced separately (it must be bound to a + # per-run param, never the frozen demo literal), so it never doubles + # as an identity selector. + selector.pop(IDEMPOTENCY_KEY_FIELD, None) + return WriteCandidate( + step_id=step_id, + action=action, + label=label, + basis="observed system-of-record delta (one new record)", + match=selector, + payload=payload, + idempotency=idempotency, + ) + + +def candidates_from_recording(recording_dir: Path) -> list[WriteCandidate]: + """Write-shaped candidates straight from ``events.jsonl`` evidence.""" + import json + + from openadapt_flow.risk import is_write_shaped + + events_path = recording_dir / RECORDING_MARKER + try: + lines = events_path.read_text(encoding="utf-8").splitlines() + except OSError as exc: + raise ScaffoldRefused( + f"scaffold-verifier: could not read {events_path}: {exc}" + ) from exc + + # Demonstrated parameter values separate a record's PAYLOAD fields (the + # typed note -> postcondition read-backs) from its identity fields (the + # match selector), exactly as the compiler's miner does. + demonstrated_values = _demonstrated_param_values(recording_dir) + + candidates: list[WriteCandidate] = [] + for index, line in enumerate(lines): + if not line.strip(): + continue + try: + event = json.loads(line) + except json.JSONDecodeError as exc: + raise ScaffoldRefused( + f"scaffold-verifier: {events_path} line {index + 1} is not JSON: {exc}" + ) from exc + kind = str(event.get("kind") or "") + step_id = f"step_{index:03d}" # positional, matching the compiler + label = _event_text(event) + + # Strongest evidence first: a captured system-of-record snapshot whose + # after-state holds exactly one NEW record -- the demonstration itself + # observed the write land. + before, after = _sor_delta(event) + if before is not None and after is not None: + from openadapt_flow.compiler.effect_mining import _new_records + + new_records = _new_records(before, after) + if len(new_records) == 1: + candidates.append( + _candidate_from_record( + step_id, + kind, + label, + new_records[0], + demonstrated_values=demonstrated_values, + ) + ) + continue + + pointer_write = kind in {"click", "double_click", "right_click"} and ( + is_write_shaped(label) + ) + submit_key = kind == "key" and (event.get("key") or "").lower() in { + "enter", + "return", + } + if pointer_write or submit_key: + basis = ( + "write-shaped control label" + if pointer_write + else "submission key with no system-of-record snapshot" + ) + candidates.append( + WriteCandidate(step_id=step_id, action=kind, label=label, basis=basis) + ) + return candidates + + +def candidates_from_bundle(bundle_dir: Path) -> list[WriteCandidate]: + """Write-shaped candidates from a compiled bundle's typed steps/effects.""" + from openadapt_flow.ir import ActionKind, Workflow + + workflow = Workflow.load(bundle_dir) + candidates: list[WriteCandidate] = [] + for step in workflow.steps: + real_effects = [ + effect for effect in step.effects if not effect.needs_operator_confirmation + ] + if not real_effects and step.risk != "irreversible": + continue + candidate = WriteCandidate( + step_id=step.id, + action=step.action.value, + label=step.intent, + basis=( + "compiled effect contract" + if real_effects + else "compile-time risk classification (consequential write)" + ), + ) + for effect in real_effects: + if effect.kind.value == "record_written": + for key, expr in effect.match.items(): + candidate.match[key] = ( + f"{{param: {expr.param}}}" if expr.param else str(expr.literal) + ) + elif effect.kind.value == "field_equals" and effect.value is not None: + key = effect.field or "TODO-field" + candidate.payload[key] = ( + f"{{param: {effect.value.param}}}" + if effect.value.param + else str(effect.value.literal) + ) + if step.action is ActionKind.TYPE and step.param: + name = step.field_label or step.param + candidate.payload[name] = f"{{param: {step.param}}}" + candidates.append(candidate) + return candidates + + +# --------------------------------------------------------------------------- +# Draft rendering +# --------------------------------------------------------------------------- + +_DRAFT_HEADER = """\ +# OPENADAPT EFFECT-ORACLE DRAFT -- REQUIRES HUMAN EDIT BEFORE USE +# +# Generated by `openadapt-flow scaffold-verifier` from the demonstration +# evidence below. This file is a STARTING POINT, not an approved contract: +# * resolve EVERY `TODO` against the application's REAL system of record +# (a person who knows the app must do this), +# * copy the `deployment.effects` section into your deployment.yaml, +# * re-run lint/certify and qualify the bundle before any governed run. +# Nothing here is auto-approved, and a draft that is never edited, wired, and +# qualified verifies nothing. +""" + + +def _yaml_scalar(value: str) -> str: + """Quote a scalar only when YAML would otherwise misread it.""" + special = ":{}[],&*#?|-<>=!%@`\"'\n" + if value == "" or any(character in value for character in special): + return repr(value) + return value + + +def render_draft_yaml( + *, + source: Path, + source_kind: str, + workflow_name: str, + candidates: list[WriteCandidate], +) -> str: + """Render the draft contract: deterministic, commented, TODO-marked.""" + lines: list[str] = [_DRAFT_HEADER.rstrip()] + lines += [ + "", + f"workflow: {_yaml_scalar(workflow_name)}", + f"source: {_yaml_scalar(str(source))}", + f"source_kind: {source_kind}", + "draft_status: requires-human-edit", + "", + "# The write-shaped steps retained by the demonstration.", + "steps:", + ] + for candidate in candidates: + lines.append(f" - step_id: {candidate.step_id}") + lines.append(f" action: {candidate.action}") + lines.append(f" evidence_basis: {_yaml_scalar(candidate.basis)}") + if candidate.label: + lines.append(f" label: {_yaml_scalar(candidate.label[:120])}") + lines += [ + "", + "# Proposed per-step effect contracts (what the oracle must prove).", + "effects:", + ] + for candidate in candidates: + lines += [ + f" - step_id: {candidate.step_id}", + " kind: record_written", + " expected_count: 1 # at-most-once for this write", + ] + lines.extend(_effect_block_lines(candidate)) + lines += [ + "", + "# Deployment wiring: copy into your deployment.yaml (docs/EFFECT_KIT.md).", + "deployment:", + " effects:", + " kind: rest", + " base_url: TODO-system-of-record-base-url", + " records_path: TODO-path-that-returns-the-records-document", + " records_key: records # TODO key holding the records list", + " # auth names ENV VARS, never credential literals:", + " # auth:", + " # bearer_env: SOR_BEARER_TOKEN", + "", + ] + return "\n".join(lines) + + +def _effect_block_lines(candidate: WriteCandidate) -> list[str]: + """The per-step proposed contract block for one write candidate.""" + lines: list[str] = [] + if not candidate.observed_delta: + return lines + [ + f" # Evidence basis: {candidate.basis}; no system-of-record", + " # snapshot was captured for this step, so a human must bind it.", + " match: {} # TODO bind the intended record in the system of record", + " postconditions: {} # TODO expected post-write field values", + ] + lines.append(" match:") + if candidate.match: + for key, value in candidate.match.items(): + lines.append( + f" {_yaml_scalar(key)}: {_yaml_scalar(value)} # TODO confirm" + ) + else: + lines.append( + " TODO: stable-identifier # fields that identify the intended" + " record across runs" + ) + lines.append(" postconditions:") + if candidate.payload: + for key, value in candidate.payload.items(): + shown = value if value.startswith("{param:") else _yaml_scalar(value) + lines.append(f" {_yaml_scalar(key)}: {shown} # TODO confirm") + else: + lines.append(" TODO-field: TODO-value") + if candidate.idempotency: + key_field, observed = candidate.idempotency + lines.append( + f" idempotency_key_field: {key_field} # observed at-most-once key" + ) + lines.append( + " # bind its value to a run param (--param), never the frozen" + f" demo literal {observed!r}" + ) + return lines + + +NEXT_COMMANDS_TEMPLATE = """\ +Draft written to {out} ({count} write-shaped step(s)) + +This is a DRAFT oracle, not an approved contract: it verifies nothing until a +person resolves every TODO, wires the deployment section, and qualifies the +bundle. + +Next commands: + 1. edit {out} + 2. openadapt-flow lint + 3. openadapt-flow certify --policy + 4. openadapt-flow qualify init --target ... +""" + + +def write_draft(source: Path, out_dir: Optional[Path] = None) -> tuple[Path, int]: + """Build and write the draft contract; return ``(path, candidate_count)``.""" + source_kind = classify_target(source) + if source_kind == "bundle": + from openadapt_flow.ir import Workflow + + workflow_name = Workflow.load(source).name + candidates = candidates_from_bundle(source) + else: + import json + + try: + meta = json.loads((source / "meta.json").read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + meta = {} + workflow_name = str(meta.get("app_url") or source.name) + candidates = candidates_from_recording(source) + + if not candidates: + raise ScaffoldRefused( + "scaffold-verifier REFUSED: the demonstration has no consequential " + "(write-shaped) step to verify. An effect oracle asserts that a " + "WRITE landed in a system of record; this input shows only " + "read/navigation actions, so there is nothing to scaffold." + ) + + text = render_draft_yaml( + source=source, + source_kind=source_kind, + workflow_name=workflow_name, + candidates=candidates, + ) + destination = (out_dir or source) / CONTRACT_FILENAME + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(text, encoding="utf-8") + return destination, len(candidates) + + +# --------------------------------------------------------------------------- +# flow explain: read-only plain-language run summary +# --------------------------------------------------------------------------- + + +def _halt_check_line(report: Any) -> str: + """Name the exact check that fired for a halt, from retained evidence.""" + for result in report.results: + if result.effect_verified is False: + return ( + f"the independent effect check on step `{result.step_id}` " + "REFUTED the declared write against the system of record" + ) + for result in report.results: + identity = result.identity + if identity is not None and identity.status == "mismatch": + return ( + f"the identity gate on step `{result.step_id}` detected a target " + "DIFFERENT from the demonstrated one" + ) + for result in report.results: + if result.postconditions_ok is False and not result.skipped: + return ( + f"the screen postcondition on step `{result.step_id}` did not " + "hold after the action" + ) + halt = getattr(report, "halt", None) + if halt is not None and halt.reason: + return f"the engine halted: {halt.reason}" + return "a governed check refused to let an unproven step claim success" + + +def explain_run(run_dir: Path) -> str: + """Read-only plain-language summary of one completed run directory.""" + from openadapt_flow.ir import RunReport + + report_path = run_dir / "report.json" + if not report_path.is_file(): + raise SystemExit( + f"explain: {run_dir} holds no report.json -- nothing to explain " + "(is this a completed run directory?)" + ) + try: + report = RunReport.model_validate_json(report_path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise SystemExit( + f"explain: {report_path} could not be read as a run report: {exc}" + ) from exc + outcome = report.execution_outcome or ("success" if report.success else "FAILED") + executed = sum(1 for result in report.results if not result.skipped) + ok_steps = sum(1 for result in report.results if result.ok) + + lines = [ + f"What happened: run '{report.workflow_name}' finished {outcome}: " + f"{ok_steps}/{len(report.results)} steps ok ({executed} executed), " + f"{report.heal_count} heal(s), model calls {report.model_calls}." + ] + receipt = run_dir / "receipt.json" + if outcome == "VERIFIED": + lines.append( + "Why this is safe: every declared effect was independently confirmed " + "in a system of record before the run claimed success." + ) + if receipt.is_file(): + lines.append(f"Shareable receipt: {receipt}") + lines.append( + "Next: qualify this workflow for your environment -- " + "openadapt-flow qualify init --target ..." + ) + elif outcome == "HALTED": + lines.append(f"Why this is safe: {_halt_check_line(report)}.") + lines.append( + "The engine stopped instead of acting on unproven state -- that is " + "the fail-closed contract working, not a defect." + ) + lines.append( + f"Next: read {run_dir / 'REPORT.md'} for the halt evidence, fix the " + "cause, then re-run the same command." + ) + elif outcome == "COMPLETED_UNVERIFIED": + lines.append( + "Why this is safe: the steps completed on screen but nothing " + "independently proved the writes reached a system of record, so " + "this outcome must never be reported as success." + ) + lines.append( + "Next: pair an oracle -- openadapt-flow scaffold-verifier " + " drafts one; wire deployment.yaml effects:, " + "then re-run under the standard profile." + ) + else: + lines.append( + "Why this is safe: the run failed loudly and reported failure " + "instead of guessing." + ) + lines.append( + f"Next: read {run_dir / 'REPORT.md'}, then re-run the same command " + "once the cause is fixed." + ) + if (run_dir / "REPORT.md").is_file(): + lines.append(f"Plain-language evidence: {run_dir / 'REPORT.md'}") + return "\n".join(lines) diff --git a/openadapt_flow/tutorial.py b/openadapt_flow/tutorial.py index 467054ae..5ce08dd2 100644 --- a/openadapt_flow/tutorial.py +++ b/openadapt_flow/tutorial.py @@ -166,6 +166,64 @@ def _next_steps_block() -> str: ) +def outcome_epilogue_lines( + *, + what: str, + why_safe: str, + next_command: str, +) -> list[str]: + """The three-line outcome epilogue: what / why-safe / exact next command. + + Presentation only -- every caller keeps its own exit code and fail-closed + semantics unchanged. Shared by the replay finisher, the lint failure path, + and the tutorial so every non-VERIFIED ending speaks with one voice. + """ + + return [ + f"What happened: {what}.", + f"Why this is safe: {why_safe}.", + f"Next command: {next_command}", + ] + + +def tutorial_epilogue(result: "TutorialResult") -> list[str]: + """The epilogue for a NON-VERIFIED tutorial run (never printed on success). + + A halt or unverified completion is a correct result but earns no receipt; + these lines say what happened, why that is the safe behavior, and give the + exact next command instead of leaving the operator at a dead end. + """ + + outcome = result.execution_outcome + if outcome == "COMPLETED_UNVERIFIED": + what = ( + f"the run completed on screen but ended {outcome} -- no independent " + "system-of-record proof, so no receipt was issued" + ) + why_safe = ( + "a demo-profile completion can never claim success under Flow; " + "only independently confirmed writes earn VERIFIED" + ) + next_command = ( + "openadapt-flow scaffold-verifier " + f"{result.recording_dir} # draft an oracle, wire deployment.yaml " + "effects:, re-run under the standard profile" + ) + else: # HALTED / FAILED / ROLLED_BACK + what = ( + f"the run stopped at a governed check and ended {outcome} -- no " + "receipt was issued for an unproven run" + ) + why_safe = ( + "the engine reports only what independent evidence proves; halting " + "on a failed check is the fail-closed contract working" + ) + next_command = f"openadapt-flow explain {result.run_dir}" + return outcome_epilogue_lines( + what=what, why_safe=why_safe, next_command=next_command + ) + + def _http_json(url: str, *, method: str = "GET", body: Any = None) -> Any: data = None if body is None else json.dumps(body).encode("utf-8") request = Request( # noqa: S310 - loopback only, built from serve()'s URL diff --git a/scripts/social_card.py b/scripts/social_card.py new file mode 100644 index 00000000..70249fa6 --- /dev/null +++ b/scripts/social_card.py @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 +"""Render a shareable PNG card from one completed run's local artifacts. + +Usage: + python scripts/social_card.py -o card.png + +Reads ``/report.json`` (a typed :class:`openadapt_flow.ir.RunReport`) +plus, when present, ``receipt.json`` and ``bench.json``. Deterministic layout, +no network access, no fonts beyond Pillow's default: the same run directory +always renders byte-identical pixels. + +The card states only closed, PHI-free facts the run retained about itself: +the outcome badge, workflow name, duration median, executed trials, the +model-call count (0 on a healthy governed run), short artifact SHA-256 hashes, +and the openadapt.ai URL. No screenshot, parameter, URL, or free-form halt +text is drawn. +""" + +from __future__ import annotations + +import argparse +import json +import statistics +import sys +from pathlib import Path + +_BG = (16, 18, 22) +_FG = (233, 236, 241) +_MUTED = (140, 148, 160) +_OK = (86, 204, 132) +_WARN = (232, 176, 84) + +_WIDTH = 720 +_URL = "openadapt.ai" + + +def _load_report(run_dir: Path): + from openadapt_flow.ir import RunReport + + report_path = run_dir / "report.json" + if not report_path.is_file(): + raise SystemExit(f"social_card: {run_dir} holds no report.json") + return RunReport.model_validate_json(report_path.read_text(encoding="utf-8")) + + +def _card_stats(run_dir: Path, report) -> dict[str, object]: + """Closed stats for the card, from the typed report + optional artifacts.""" + executed_ms = [result.elapsed_ms for result in report.results if not result.skipped] + duration_median = statistics.median(executed_ms) if executed_ms else 0.0 + + trials = len([result for result in report.results if not result.skipped]) + bench_median = None + bench_path = run_dir / "bench.json" + if bench_path.is_file(): + try: + bench = json.loads(bench_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + bench = {} + n = int(bench.get("n") or 0) + if n > 0: + trials = n + bench_median = float(bench.get("total_ms_p50") or 0.0) + if bench_median is not None: + duration_median = bench_median + + receipt_digest = "" + receipt_path = run_dir / "receipt.json" + if receipt_path.is_file(): + try: + receipt_digest = str( + json.loads(receipt_path.read_text(encoding="utf-8")).get( + "receipt_digest", "" + ) + ) + except (OSError, json.JSONDecodeError): + receipt_digest = "" + + return { + "outcome": str( + getattr(report, "execution_outcome", None) + or ("success" if report.success else "FAILED") + ), + "name": str(report.workflow_name), + "duration_median": duration_median, + "trials": trials, + "model_calls": int(report.model_calls), + "bundle_short": str(report.bundle_content_digest or "")[:12], + "receipt_short": receipt_digest[:12], + } + + +def render_card(stats: dict[str, object]): + """Draw the card deterministically; returns a PIL Image.""" + from PIL import Image, ImageDraw, ImageFont + + def font(size: int): + try: + return ImageFont.load_default(size=size) + except TypeError: # pragma: no cover - very old Pillow + return ImageFont.load_default() + + outcome = str(stats["outcome"]) + accent = _OK if outcome == "VERIFIED" else _WARN + title_font = font(30) + big_font = font(20) + label_font = font(15) + foot_font = font(13) + + pad = 28 + row_h = 26 + duration_median = float(stats["duration_median"]) # type: ignore[arg-type] + rows = [ + ("duration median", f"{duration_median:.0f} ms"), + ("trials", str(stats["trials"])), + ( + "model calls", + f"{stats['model_calls']}" + + ( + " (healthy governed runs make none)" if not stats["model_calls"] else "" + ), + ), + ("bundle sha256", str(stats["bundle_short"]) or "unbound"), + ("receipt sha256", str(stats["receipt_short"]) or "none issued"), + ] + height = pad + 46 + pad + 34 + len(rows) * row_h + pad + 24 + + image = Image.new("RGB", (_WIDTH, height), _BG) + draw = ImageDraw.Draw(image) + draw.rectangle([(0, 0), (6, height)], fill=accent) + + y = pad + badge_w = max(120, len(outcome) * 14 + 32) + draw.rounded_rectangle([(pad, y), (pad + badge_w, y + 38)], radius=8, fill=accent) + draw.text((pad + 16, y + 7), outcome, font=big_font, fill=_BG) + y += 52 + + name = str(stats["name"]) + if draw.textlength(name, font=title_font) > _WIDTH - 2 * pad: + while name and draw.textlength(name + "...", font=title_font) > ( + _WIDTH - 2 * pad + ): + name = name[:-1] + name += "..." + draw.text((pad, y), name, font=title_font, fill=_FG) + y += 40 + + for key, value in rows: + draw.text((pad, y), key, font=label_font, fill=_MUTED) + draw.text((pad + 200, y), value, font=label_font, fill=_FG) + y += row_h + + draw.text( + (pad, height - pad - 12), + f"{_URL} | deterministic local execution - no screenshots, no " + "parameters drawn", + font=foot_font, + fill=_MUTED, + ) + return image + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("run_dir", help="Completed run directory (report.json)") + parser.add_argument("-o", "--out", default="card.png", help="Output PNG path") + args = parser.parse_args(argv) + + run_dir = Path(args.run_dir) + stats = _card_stats(run_dir, _load_report(run_dir)) + out_path = Path(args.out) + out_path.parent.mkdir(parents=True, exist_ok=True) + render_card(stats).save(out_path, format="PNG") + print(f"Social card written to {out_path}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_outcome_epilogues.py b/tests/test_outcome_epilogues.py new file mode 100644 index 00000000..1cc969b2 --- /dev/null +++ b/tests/test_outcome_epilogues.py @@ -0,0 +1,179 @@ +"""Outcome-aware epilogues (presentation only) on the non-VERIFIED endings. + +* tutorial non-VERIFIED runs print the 3-line epilogue (what / why-safe / + next command); VERIFIED runs keep printing the success-rail block only; +* a failing `lint` exits nonzero AND prints the epilogue; +* `_finish_replay` appends the epilogue for HALTED / COMPLETED_UNVERIFIED + runs and leaves exit codes untouched (fail-closed semantics identical). +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from openadapt_flow import tutorial as tutorial_module +from openadapt_flow.__main__ import _finish_replay, main +from openadapt_flow.ir import RunReport, StepResult +from openadapt_flow.tutorial import TutorialResult, outcome_epilogue_lines + + +def _tutorial_result(root: Path, execution_outcome: str) -> TutorialResult: + verified = execution_outcome == "VERIFIED" + return TutorialResult( + recording_dir=root / "recording", + bundle_dir=root / "bundle", + run_dir=root / "run", + execution_outcome=execution_outcome, + transaction_outcome=execution_outcome, + execution_profile="standard", + transaction_billable=verified, + model_calls=0, + effects_required=2, + effects_confirmed=2 if verified else 0, + effect_tier=1 if verified else None, + bundle_digest="d" * 64, + system_of_record_records=1 if verified else 0, + ) + + +def _wire(monkeypatch: pytest.MonkeyPatch, result_for) -> None: + def fake_run_tutorial(work_dir: Path, **kwargs): + return result_for(Path(work_dir), kwargs) + + monkeypatch.setattr(tutorial_module, "run_tutorial", fake_run_tutorial) + + +class TestTutorialEpilogue: + def test_halted_tutorial_prints_explain_command( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys + ) -> None: + _wire( + monkeypatch, + lambda root, kwargs: _tutorial_result(root, "HALTED"), + ) + assert main(["tutorial", "--out", str(tmp_path / "t")]) == 1 + out = capsys.readouterr().out + assert "What happened:" in out + assert "Why this is safe:" in out + assert f"openadapt-flow explain {tmp_path}/t/run" in out.replace("//", "/") + + def test_unverified_tutorial_points_at_scaffold_verifier( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys + ) -> None: + _wire( + monkeypatch, + lambda root, kwargs: _tutorial_result(root, "COMPLETED_UNVERIFIED"), + ) + assert main(["tutorial", "--out", str(tmp_path / "t")]) == 1 + out = capsys.readouterr().out + assert "scaffold-verifier" in out + + def test_verified_tutorial_prints_no_epilogue( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys + ) -> None: + _wire( + monkeypatch, + lambda root, kwargs: _tutorial_result(root, "VERIFIED"), + ) + assert main(["tutorial", "--out", str(tmp_path / "t")]) == 0 + out = capsys.readouterr().out + assert "What happened:" not in out + + +class TestLintEpilogue: + @pytest.fixture() + def gap_bundle(self, tmp_path: Path) -> Path: + """A bundle whose unarmed irreversible click fails lint at 'error'.""" + from openadapt_flow.ir import ActionKind, Step, Workflow + + bundle = tmp_path / "gap-bundle" + Workflow( + name="gappy", + steps=[ + Step( + id="step_000", + intent="click 'Delete record'", + action=ActionKind.CLICK, + risk="irreversible", + ) + ], + ).save(bundle) + return bundle + + def test_failing_lint_prints_three_line_epilogue( + self, gap_bundle: Path, capsys + ) -> None: + assert main(["lint", str(gap_bundle)]) == 1 + out = capsys.readouterr().out + assert "What happened: lint found coverage gaps" in out + assert "Why this is safe:" in out + assert f"Next command: openadapt-flow certify {gap_bundle}" in out + + def test_clean_lint_prints_no_epilogue(self, tmp_path: Path, capsys) -> None: + from openadapt_flow.ir import ActionKind, Step, Workflow + + bundle = tmp_path / "clean" + Workflow( + name="clean", + steps=[Step(id="s", intent="click 'Open'", action=ActionKind.CLICK)], + ).save(bundle) + assert main(["lint", str(bundle)]) == 0 + assert "What happened:" not in capsys.readouterr().out + + +class TestReplayFinisherEpilogue: + """_finish_replay gains lines only; exit codes are unchanged.""" + + @staticmethod + def _run_and_finish(tmp_path: Path, outcome: str, success: bool) -> tuple[int, str]: + run_dir = tmp_path / f"run-{outcome}" + report = RunReport( + workflow_name="wf", + started_at="2026-08-20T12:00:00+00:00", + execution_outcome=outcome, + results=[ + StepResult( + step_id="step_004", + intent="click 'Save encounter'", + ok=success, + safety_halt=outcome == "HALTED", + effect_verified=False if outcome == "HALTED" else None, + postconditions_ok=False if outcome == "HALTED" else None, + elapsed_ms=100.0, + ) + ], + success=success, + ) + # render_run_report needs report.json on disk first. + report.save(run_dir) + code = _finish_replay(run_dir, report) + return code, run_dir + + def test_halted_replay_appends_epilogue_exit_code_kept( + self, tmp_path: Path, capsys + ) -> None: + code, _ = self._run_and_finish(tmp_path, "HALTED", False) + out = capsys.readouterr().out + assert "What happened: the run stopped at step `step_004`" in out + assert "Why this is safe:" in out + assert "Next command: openadapt-flow explain" in out + assert code == 1 + + def test_completed_unverified_names_scaffold_verifier( + self, tmp_path: Path, capsys + ) -> None: + code, _ = self._run_and_finish(tmp_path, "COMPLETED_UNVERIFIED", True) + out = capsys.readouterr().out + assert "scaffold-verifier" in out + assert "can never claim success under Flow" in out + assert code == 0 + + +def test_outcome_epilogue_lines_shape() -> None: + lines = outcome_epilogue_lines(what="x", why_safe="y", next_command="z") + assert len(lines) == 3 + assert lines[0].startswith("What happened:") + assert lines[1].startswith("Why this is safe:") + assert lines[2].startswith("Next command:") diff --git a/tests/test_scaffold_verifier.py b/tests/test_scaffold_verifier.py new file mode 100644 index 00000000..cff40787 --- /dev/null +++ b/tests/test_scaffold_verifier.py @@ -0,0 +1,342 @@ +"""scaffold-verifier + explain operator tooling (scaffold_verifier.py). + +Fixture run dirs / recordings are built in-test, mirroring +tests/test_report.py's synthetic builders. The load-bearing contracts: + +* scaffold-verifier drafts, and never approves: every draft carries TODO + markers plus the loud requires-human-edit header; +* a demonstration with no consequential step is REFUSED (nonzero), never + scaffolded anyway; +* explain is pure read-only and names the check that fired for a HALT. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openadapt_flow.__main__ import build_parser, main +from openadapt_flow.ir import ( + ActionKind, + Effect, + HaltObservation, + RunReport, + Step, + StepResult, + Workflow, +) +from openadapt_flow.scaffold_verifier import ( + CONTRACT_FILENAME, + candidates_from_bundle, + candidates_from_recording, + classify_target, + write_draft, +) + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +def _recording_dir(tmp_path: Path, *, consequential: bool = True) -> Path: + """A minimal recording dir; one save click carrying a SoR delta.""" + rec = tmp_path / "recording" + rec.mkdir() + events: list[dict] = [ + { + "i": 0, + "kind": "click", + "x": 100, + "y": 200, + "t": 0.1, + "structural": {"role": "button", "name": "New encounter"}, + } + ] + if consequential: + events.append( + { + "i": 1, + "kind": "click", + "x": 300, + "y": 400, + "t": 0.4, + "structural": {"role": "button", "name": "Save encounter"}, + "sor_before": [{"id": 1, "patient_id": "p1", "type": "Intake"}], + "sor_after": [ + {"id": 1, "patient_id": "p1", "type": "Intake"}, + { + "id": 2, + "patient_id": "p2", + "type": "Triage", + "note": "Synthetic follow-up in two weeks", + "key": "demo-key-1", + }, + ], + } + ) + (rec / "events.jsonl").write_text( + "".join(json.dumps(event) + "\n" for event in events), encoding="utf-8" + ) + (rec / "meta.json").write_text( + json.dumps({"params": {"note": "Synthetic follow-up in two weeks"}}), + encoding="utf-8", + ) + return rec + + +def _bundle_dir(tmp_path: Path) -> Path: + """A compiled bundle with a typed record_written effect on its last step.""" + from openadapt_flow.runtime.effects.effect import ValueExpr + + steps = [ + Step(id="step_000", intent="click 'Open'", action=ActionKind.CLICK), + Step( + id="step_001", + intent="type note", + action=ActionKind.TYPE, + param="note", + field_label="Note", + text="ignored-at-draft-time", + ), + Step( + id="step_002", + intent="click 'Save encounter'", + action=ActionKind.CLICK, + risk="irreversible", + effects=[ + Effect( + kind="record_written", + match={ + "patient_id": ValueExpr(literal="p2"), + "type": ValueExpr(literal="Triage"), + }, + ) + ], + ), + ] + bundle = tmp_path / "bundle" + Workflow(name="triage demo", steps=steps).save(bundle) + return bundle + + +def _run_report(outcome: str, *, halted: bool = False) -> RunReport: + results = [ + StepResult(step_id="step_000", intent="open", ok=True, elapsed_ms=120.0), + StepResult( + step_id="step_004", + intent="click 'Save encounter'", + ok=not halted, + safety_halt=halted, + effect_verified=False if halted else None, + postconditions_ok=False if halted else None, + elapsed_ms=480.0, + ), + ] + return RunReport( + workflow_name="local-quickstart", + started_at="2026-08-20T12:00:00+00:00", + execution_outcome=outcome, + execution_profile="standard" if outcome != "COMPLETED_UNVERIFIED" else "demo", + results=results, + success=outcome == "VERIFIED", + halt=( + HaltObservation( + state_id="step_004", + intent="click 'Save encounter'", + reason="record_written refuted against the system of record", + ) + if halted + else None + ), + model_calls=0, + total_ms=600.0, + ) + + +# --------------------------------------------------------------------------- +# scaffold-verifier +# --------------------------------------------------------------------------- + + +class TestScaffoldVerifier: + def test_classify_bundle_and_recording(self, tmp_path: Path) -> None: + assert classify_target(_bundle_dir(tmp_path)) == "bundle" + assert classify_target(_recording_dir(tmp_path)) == "recording" + with pytest.raises(SystemExit, match="path not found"): + classify_target(tmp_path / "missing") + empty = tmp_path / "empty-dir" + empty.mkdir() + with pytest.raises(SystemExit, match="neither a workflow bundle"): + classify_target(empty) + + def test_recording_candidates_split_identity_and_payload( + self, tmp_path: Path + ) -> None: + rec = _recording_dir(tmp_path) + candidates = candidates_from_recording(rec) + assert [candidate.step_id for candidate in candidates] == ["step_001"] + candidate = candidates[0] + assert candidate.observed_delta + assert candidate.match == {"patient_id": "p2", "type": "Triage"} + assert candidate.payload == {"note": "Synthetic follow-up in two weeks"} + assert candidate.idempotency == ("key", "demo-key-1") + + def test_write_draft_from_recording_is_todo_marked(self, tmp_path: Path) -> None: + rec = _recording_dir(tmp_path) + out_dir = tmp_path / "drafted" + out, count = write_draft(rec, out_dir) + assert out == out_dir / CONTRACT_FILENAME and count == 1 + text = out.read_text(encoding="utf-8") + assert "REQUIRES HUMAN EDIT BEFORE USE" in text + assert "requires-human-edit" in text + assert "TODO" in text + assert "patient_id: p2" in text + assert "kind: rest" in text + # Valid YAML despite all the comments. + import yaml + + parsed = yaml.safe_load(text) + assert parsed["source_kind"] == "recording" + assert parsed["effects"][0]["kind"] == "record_written" + + def test_write_draft_from_bundle(self, tmp_path: Path) -> None: + bundle = _bundle_dir(tmp_path) + candidates = candidates_from_bundle(bundle) + assert [candidate.step_id for candidate in candidates] == ["step_002"] + assert candidates[0].match == {"patient_id": "p2", "type": "Triage"} + out, count = write_draft(bundle) + assert count == 1 + text = out.read_text(encoding="utf-8") + assert "compiled effect contract" in text + assert "TODO" in text + + def test_no_consequential_step_refused(self, tmp_path: Path) -> None: + rec = _recording_dir(tmp_path, consequential=False) + with pytest.raises(SystemExit, match="REFUSED.*no consequential"): + write_draft(rec) + + def test_cli_scaffold_verifier(self, tmp_path: Path, capsys) -> None: + rec = _recording_dir(tmp_path) + out_dir = tmp_path / "cli-out" + code = main(["scaffold-verifier", str(rec), "-o", str(out_dir)]) + assert code == 0 + out = capsys.readouterr().out + assert "DRAFT oracle" in out + assert "Next commands:" in out + assert (out_dir / CONTRACT_FILENAME).is_file() + + def test_parser_registers_both_new_commands(self) -> None: + parser = build_parser() + helptext = parser.format_help() + assert "scaffold-verifier" in helptext + assert "explain" in helptext + + +# --------------------------------------------------------------------------- +# flow explain +# --------------------------------------------------------------------------- + + +class TestExplainRun: + def _run_dir(self, tmp_path: Path, outcome: str) -> Path: + run_dir = tmp_path / "run" + _run_report(outcome, halted=outcome == "HALTED").save(run_dir) + return run_dir + + def test_explains_a_halt_and_names_the_fired_check( + self, tmp_path: Path, capsys + ) -> None: + run_dir = self._run_dir(tmp_path, "HALTED") + assert main(["explain", str(run_dir)]) == 0 + out = capsys.readouterr().out + assert "finished HALTED" in out + assert "independent effect check on step `step_004`" in out + assert "Why this is safe" in out + assert f"read {run_dir}/REPORT.md" in out + + def test_unverified_points_at_scaffold_verifier( + self, tmp_path: Path, capsys + ) -> None: + run_dir = self._run_dir(tmp_path, "COMPLETED_UNVERIFIED") + assert main(["explain", str(run_dir)]) == 0 + out = capsys.readouterr().out + assert "must never be reported as success" in out + assert "scaffold-verifier" in out + + def test_missing_report_refuses(self, tmp_path: Path) -> None: + with pytest.raises(SystemExit, match="holds no report.json"): + main(["explain", str(tmp_path)]) + + def test_explain_is_read_only(self, tmp_path: Path) -> None: + import hashlib + + run_dir = self._run_dir(tmp_path, "HALTED") + + def snapshot(root: Path) -> dict[str, bytes]: + return { + str(path.relative_to(root)): hashlib.sha256( + path.read_bytes() + ).hexdigest() + for path in sorted(root.rglob("*")) + if path.is_file() + } + + before = snapshot(run_dir) + from openadapt_flow.scaffold_verifier import explain_run + + assert "HALTED" in explain_run(run_dir) + assert snapshot(run_dir) == before + + def test_halt_line_falls_back_to_halt_reason(self, tmp_path: Path) -> None: + from openadapt_flow.scaffold_verifier import _halt_check_line + + report = RunReport( + workflow_name="w", + started_at="2026-08-20T12:00:00+00:00", + execution_outcome="HALTED", + halt=HaltObservation(state_id="s", reason="unhandled screen state"), + ) + assert "unhandled screen state" in _halt_check_line(report) + + +# --------------------------------------------------------------------------- +# social_card.py smoke + determinism +# --------------------------------------------------------------------------- + + +class TestSocialCard: + def _stats(self, tmp_path: Path) -> dict[str, object]: + run_dir = tmp_path / "run" + _run_report("VERIFIED").save(run_dir) + from scripts.social_card import _card_stats + + return _card_stats( + run_dir, + RunReport.model_validate_json((run_dir / "report.json").read_text()), + ) + + def test_renders_nonempty_png_into_tmp(self, tmp_path: Path) -> None: + from scripts.social_card import render_card + + stats = self._stats(tmp_path) + out = tmp_path / "card.png" + image = render_card(stats) + image.save(out, format="PNG") + assert out.is_file() and out.stat().st_size > 0 + + @pytest.mark.parametrize("outcome", ["VERIFIED", "HALTED", "COMPLETED_UNVERIFIED"]) + def test_smoke_all_outcomes_render(self, tmp_path: Path, outcome: str) -> None: + from scripts.social_card import render_card + + stats = self._stats(tmp_path) | {"outcome": outcome} + assert render_card(stats).size[0] > 0 + + def test_deterministic_pixels(self, tmp_path: Path) -> None: + from scripts.social_card import render_card + + stats = self._stats(tmp_path) + first = render_card(stats).tobytes() + second = render_card(stats).tobytes() + assert first == second