chore: reject writes after canvas halts and retain diagnostics

.github/workflows/canvas-nodom-ladder.yml

@@ -93,6 +93,7 @@ jobs:
         with:
           name: canvas-nodom-ladder-qualification
           path: runs/canvas-ladder/results.json
+          retention-days: 7
           if-no-files-found: error
 
       - name: Tear down

benchmark/canvas_ladder/run_canvas_ladder_qualification.py

@@ -39,15 +39,19 @@
 import hashlib
 import io
 import json
+import math
 import os
 import subprocess
 import time
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
 
 from PIL import Image, ImageFilter, ImageOps
 
+if TYPE_CHECKING:
+    from openadapt_flow.ir import RunReport
+
 # -- fixture geometry (kiosk_app.py renders these at fixed positions) ----------
 VIEWPORT = (1280, 800)
 ADA_ROW = (347, 192)  # "Ada Lovelace   MRN 24682" roster row
@@ -327,14 +331,25 @@ def acquire_actuation_frame(self) -> bytes:
 
 
 def _read_saved_note(container: str) -> Optional[str]:
+    # Only an explicit missing file proves absence. Docker/read failures must
+    # not become a no-write result, and an existing empty file remains a write.
+    probe = """import pathlib, sys
+try:
+    data = pathlib.Path(sys.argv[1]).read_bytes()
+except FileNotFoundError:
+    sys.exit(44)
+sys.stdout.buffer.write(data)
+"""
     res = subprocess.run(
-        ["docker", "exec", container, "cat", SAVE_PATH],
+        ["docker", "exec", container, "python3", "-c", probe, SAVE_PATH],
         capture_output=True,
         timeout=15,
         check=False,
     )
-    if res.returncode != 0:
+    if res.returncode == 44:
         return None
+    if res.returncode != 0:
+        raise RuntimeError(f"Canvas saved-note probe failed (exit {res.returncode})")
     return res.stdout.decode(errors="replace").strip()
 
 
@@ -380,6 +395,62 @@ def _new_page(pw, base_url: str, container_port: int):
     raise RuntimeError("noVNC canvas never painted a non-blank kiosk frame")
 
 
+def _native_diagnostics(report: "RunReport") -> dict:
+    """Keep bounded typed halt evidence, never report text or local files.
+
+    Step IDs, intents, errors, OCR strings, paths and frames can contain live
+    values. A one-based result index plus native refusal enums locates the
+    failure without copying those fields into the scheduled public artifact.
+    """
+    examined = report.results[:64]
+    failed = next(
+        ((index, result) for index, result in enumerate(examined, 1) if not result.ok),
+        None,
+    )
+    diagnostic = {
+        "schema_version": "openadapt.canvas-native-diagnostics.v1",
+        "execution_outcome": report.execution_outcome,
+        "transaction_outcome": report.transaction_outcome,
+        "examined_step_count": len(examined),
+        "step_scan_truncated": len(report.results) > len(examined),
+        "first_failed_step": None,
+    }
+    if failed is None:
+        return diagnostic
+    index, result = failed
+
+    def fraction(value: float) -> Optional[float]:
+        return round(value, 4) if math.isfinite(value) and 0 <= value <= 1 else None
+
+    refusal = result.safety_refusal_evidence
+    diagnostic["first_failed_step"] = {
+        "result_index": index,
+        "failure_category": result.failure_category,
+        "refusal_stage": refusal.stage if refusal else None,
+        "refusal_code": refusal.code if refusal else None,
+        "error_present": result.error is not None,
+        "delivery_attempted": result.delivery_attempted,
+        "starting_state_settled": result.starting_state_settled,
+        "input_verified": result.input_verified,
+        "postconditions_ok": result.postconditions_ok,
+        "effect_verified": result.effect_verified,
+        "resolution": {
+            "rung": result.resolution.rung,
+            "confidence": fraction(result.resolution.confidence),
+        }
+        if result.resolution
+        else None,
+        "identity": {
+            "status": result.identity.status,
+            "mode": result.identity.mode,
+            "coverage": fraction(result.identity.coverage),
+        }
+        if result.identity
+        else None,
+    }
+    return diagnostic
+
+
 def run_qualification(
     container: str,
     *,
@@ -459,6 +530,7 @@ def run_qualification(
             {
                 "trial": 1,
                 "kind": "healthy_record_compile_replay",
+                "native_diagnostics": _native_diagnostics(report),
                 "success": bool(report.success),
                 "model_calls": int(report.model_calls),
                 "rung_counts": rung_counts,
@@ -501,6 +573,7 @@ def run_qualification(
             {
                 "trial": 2,
                 "kind": "moderate_drift_no_silent_wrong_write",
+                "native_diagnostics": _native_diagnostics(mod_report),
                 "drift": "downscale_0.4x + theme_invert + jpeg_q12 (legible; simulated on real session)",
                 "resolved_and_correct": bool(mod_correct_write),
                 "halted": bool(not mod_report.success),
@@ -543,18 +616,20 @@ def run_qualification(
             browser.close()
         saved_sev = _read_saved_note(container)
         sev_halted = not sev_report.success
-        sev_no_write = saved_sev != expected_saved
+        # An empty or wrong persisted note is still a write after the halt.
+        sev_no_write = saved_sev is None
         sev_no_model = sev_report.model_calls == 0
         sev_ok = sev_halted and sev_no_write and sev_no_model
         trials.append(
             {
                 "trial": 3,
                 "kind": "severe_drift_safe_halt",
+                "native_diagnostics": _native_diagnostics(sev_report),
                 "drift": "downscale_0.14x + gaussian_blur_2.0 + theme_invert + jpeg_q5 (illegible; simulated on real session)",
                 "halted": bool(sev_halted),
                 "rung_counts": dict(sev_report.rung_counts),
                 "model_calls": int(sev_report.model_calls),
-                "silent_write": bool(saved_sev == expected_saved),
+                "silent_write": not sev_no_write,
                 "effect_after_drift": saved_sev,
                 "passed": bool(sev_ok),
                 "failure_class": None if sev_ok else "drift_not_safely_halted",

public-artifacts.json

@@ -89,7 +89,7 @@
     },
     {
       "path": ".github/workflows/canvas-nodom-ladder.yml",
-      "sha256": "db5bdaf110b2d8acc15003eb5442ba9c2714a809fad3ac087cdc6ff5547e05a4"
+      "sha256": "631569233910131ff7d417655907561005f0c61008f022149b13d2c2328b9bd4"
     },
     {
       "path": ".github/workflows/ci.yml",

tests/test_canvas_ladder_backend.py

@@ -146,3 +146,220 @@ def test_canvas_backend_maps_portable_select_all_to_remote_control():
     module.CanvasBrowserBackend(page).press("ControlOrMeta+a")
 
     assert page.keyboard.events == [("press", "Control+a")]
+
+
+def _run_qualification_reports(module, tmp_path, monkeypatch, reports, saved_notes):
+    from contextlib import nullcontext
+    from unittest.mock import Mock
+
+    reports = iter(reports)
+    saved_notes = iter(saved_notes)
+    monkeypatch.setattr(
+        "playwright.sync_api.sync_playwright", lambda: nullcontext(None)
+    )
+    monkeypatch.setattr("openadapt_flow.recorder.Recorder", Mock())
+    monkeypatch.setattr("openadapt_flow.compiler.compile_recording", Mock())
+    monkeypatch.setattr(
+        "openadapt_flow.runtime.replayer.Replayer",
+        lambda *args, **kwargs: Mock(run=lambda *args, **kwargs: next(reports)),
+    )
+    monkeypatch.setattr(module, "_reset_kiosk", lambda *args: None)
+    monkeypatch.setattr(module, "_read_saved_note", lambda *args: next(saved_notes))
+    monkeypatch.setattr(module, "_new_page", lambda *args: (Mock(), None, Mock()))
+    return module.run_qualification(
+        "unused-test-container", out_dir=tmp_path, base_url="unused", port=0
+    )
+
+
+def test_moderate_over_halt_retains_native_reason_without_report_payloads(
+    tmp_path, monkeypatch
+):
+    """The scheduled result must diagnose a refusal without exporting text."""
+    import json
+
+    from openadapt_flow.ir import (
+        IdentityCheck,
+        Resolution,
+        RunReport,
+        SafetyRefusalEvidence,
+        StepResult,
+    )
+
+    module = _module()
+    secret = "SECRET-CANARY-DO-NOT-EXPORT" * 1000
+    healthy = RunReport(
+        workflow_name=secret,
+        started_at="2026-09-14T00:00:00Z",
+        success=True,
+        rung_counts={"template": 2, "ocr": 1},
+        results=[StepResult(step_id=secret, intent=secret, ok=True)],
+    )
+    moderate = RunReport(
+        workflow_name=secret,
+        started_at="2026-09-14T00:00:00Z",
+        execution_outcome="HALTED",
+        transaction_outcome="HALTED_BEFORE_EFFECT",
+        params={"note": secret},
+        results=[
+            StepResult(step_id=secret, intent=secret, ok=True),
+            StepResult(
+                step_id=secret,
+                intent=secret,
+                ok=False,
+                error=secret,
+                before_png=secret,
+                after_png=secret,
+                failure_category="governed_refusal",
+                delivery_attempted=False,
+                resolution=Resolution(
+                    rung="ocr", point=(910, 648), confidence=1.0, elapsed_ms=1.0
+                ),
+                identity=IdentityCheck(
+                    status="unreadable",
+                    mode="context",
+                    coverage=0.24,
+                    expected=secret,
+                    observed=secret,
+                    param=secret,
+                ),
+                safety_refusal_evidence=SafetyRefusalEvidence(
+                    stage="identity_verification",
+                    code="identity_unverifiable",
+                    detector_input_sha256="a" * 64,
+                ),
+            ),
+        ],
+    )
+    severe = moderate.model_copy(deep=True)
+    evidence = _run_qualification_reports(
+        module,
+        tmp_path,
+        monkeypatch,
+        [healthy, moderate, severe],
+        [f"{module.EXPECTED_MRN}\t{module.NOTE_VALUE}", None, None],
+    )
+    trial = evidence["trials"][1]
+    assert trial["failure_class"] == "moderate_drift_over_halt"
+    assert trial["passed"] is False
+    assert evidence["accepted"] is False
+    assert evidence["successes"] == 2
+    diagnostic = trial["native_diagnostics"]
+    assert diagnostic["transaction_outcome"] == "HALTED_BEFORE_EFFECT"
+    failed = diagnostic["first_failed_step"]
+    assert failed["result_index"] == 2
+    assert failed["refusal_stage"] == "identity_verification"
+    assert failed["refusal_code"] == "identity_unverifiable"
+    assert failed["delivery_attempted"] is False
+    assert failed["resolution"] == {"rung": "ocr", "confidence": 1.0}
+    assert failed["identity"] == {
+        "status": "unreadable",
+        "mode": "context",
+        "coverage": 0.24,
+    }
+    encoded = json.dumps(evidence, allow_nan=False)
+    assert "SECRET-CANARY" not in encoded
+    assert len(json.dumps(diagnostic)) < 4096
+    assert all("native_diagnostics" in row for row in evidence["trials"])
+    # No frame, native report or path sweep is part of this projection.
+    assert list(tmp_path.rglob("*")) == [tmp_path / "work"]
+
+
+def test_native_diagnostics_bounds_the_scan_and_omits_nonfinite_metrics():
+    import json
+
+    from openadapt_flow.ir import IdentityCheck, RunReport, StepResult
+
+    module = _module()
+    result = StepResult(
+        step_id="private-step",
+        intent="private-intent",
+        ok=False,
+        identity=IdentityCheck(status="abstain", coverage=float("nan")),
+    )
+    report = RunReport(
+        workflow_name="private-name",
+        started_at="2026-09-14T00:00:00Z",
+        results=[result] * 1000,
+    )
+    diagnostic = module._native_diagnostics(report)
+    assert diagnostic["examined_step_count"] == 64
+    assert diagnostic["step_scan_truncated"] is True
+    assert diagnostic["first_failed_step"]["refusal_code"] is None
+    assert diagnostic["first_failed_step"]["identity"]["coverage"] is None
+    assert len(json.dumps(diagnostic, allow_nan=False)) < 4096
+
+
+@pytest.mark.parametrize("saved_note", [None, "", "WRONG-MRN\twrong note", "expected"])
+def test_severe_halt_rejects_every_persisted_note(tmp_path, monkeypatch, saved_note):
+    from openadapt_flow.ir import RunReport, StepResult
+
+    module = _module()
+    expected = f"{module.EXPECTED_MRN}\t{module.NOTE_VALUE}"
+    if saved_note == "expected":
+        saved_note = expected
+    healthy = RunReport(
+        workflow_name="fixture",
+        started_at="2026-09-14T00:00:00Z",
+        success=True,
+        rung_counts={"template": 2, "ocr": 1},
+        results=[StepResult(step_id="fixture-step", intent="fixture", ok=True)],
+    )
+    halted = RunReport(
+        workflow_name="fixture",
+        started_at="2026-09-14T00:00:00Z",
+        execution_outcome="HALTED",
+        transaction_outcome="HALTED_BEFORE_EFFECT",
+        results=[StepResult(step_id="fixture-step", intent="fixture", ok=False)],
+    )
+    evidence = _run_qualification_reports(
+        module,
+        tmp_path,
+        monkeypatch,
+        [healthy, healthy, halted],
+        [expected, expected, saved_note],
+    )
+    severe = evidence["trials"][2]
+    absent = saved_note is None
+    assert severe["effect_after_drift"] == saved_note
+    assert severe["silent_write"] is not absent
+    assert severe["passed"] is absent
+    assert severe["failure_class"] == (None if absent else "drift_not_safely_halted")
+    assert evidence["accepted"] is absent
+
+
+@pytest.mark.parametrize("content", [None, b"", b"\n", b"\t\n", b"wrong note\n"])
+def test_saved_note_probe_distinguishes_absence_from_existing_rows(
+    tmp_path, monkeypatch, content
+):
+    module = _module()
+    note = tmp_path / "saved-note.txt"
+    if content is not None:
+        note.write_bytes(content)
+    real_run = module.subprocess.run
+
+    def local_probe(command, **kwargs):
+        assert command[:4] == ["docker", "exec", "fixture", "python3"]
+        assert command[-1] == module.SAVE_PATH
+        return real_run([sys.executable, "-S", *command[4:-1], str(note)], **kwargs)
+
+    monkeypatch.setattr(module.subprocess, "run", local_probe)
+    observed = module._read_saved_note("fixture")
+    assert observed == (None if content is None else content.decode().strip())
+
+
+@pytest.mark.parametrize("returncode", [1, 125, 126, 127, 137])
+def test_saved_note_probe_failure_cannot_prove_no_write(monkeypatch, returncode):
+    from subprocess import CompletedProcess
+
+    module = _module()
+    monkeypatch.setattr(
+        module.subprocess,
+        "run",
+        lambda *args, **kwargs: CompletedProcess(
+            args=args, returncode=returncode, stdout=b"", stderr=b"private detail"
+        ),
+    )
+    with pytest.raises(
+        RuntimeError, match=rf"^Canvas saved-note probe failed \(exit {returncode}\)$"
+    ):
+        module._read_saved_note("fixture")