feat(runner+cli): run-attempt receipts + madp report (issue #5 proposals 6+7)

docs/technical-reference.md

@@ -48,6 +48,16 @@ madp owner-decide DIR --decision DECISION.md
 turn; it is not a required transition. Automatic multi-turn loops are
 deliberately out of scope; every launch is one explicit turn.
 
+Two auxiliary read-only/local commands sit outside the production
+path: **`madp report DIR`** derives the evidence index on demand (one
+row per accepted turn: commit SHA, recorded vs recomputed digests,
+provider/model/session, probed CLI version) — never committed, never
+read by acceptance or validation logic, and a tampered artifact or
+evidence file is flagged here AND by `validate` independently;
+**`madp canary --adapter A --dialogue DIR`** runs one turn through the
+real acceptance path in a fresh local dialogue and validates it with
+the production gate (always local-only).
+
 ```bash
 python3 -m unittest discover -s tests   # full suite (PYTHONPATH=src, or run scripts/verify.py)
 python3 scripts/verify.py               # compile + tests + schemas + secret scan + git hygiene
@@ -87,6 +97,17 @@ manually controlled recovery.
   dialogue** — claim, release, and complete all refuse it; recovery
   from `BLOCKED` is a human decision outside the protocol.
 
+**Run-attempt receipts**: every `run --launch` writes a
+`run-attempt-receipt` JSON under `work/run-attempts/` — before the
+process starts (`outcome: in_flight`, argv digest only, never the
+plaintext argv; runner pid; start time) and again when the attempt
+finalizes (`completed`/`failed`, exit status, claim-cleanup state).
+The directory is Git-ignored: receipts are crash forensics for the
+recovery path, **never** ledger content, and nothing in acceptance or
+validation reads them. A receipt stuck at `in_flight` names the
+attempt that never reported an outcome — the ledger stays the
+authority on what actually completed.
+
 ## Completion provenance (`completed_via`)
 
 Every completed turn records how it entered the dialogue, with exactly

examples/fakes/bin/fake-worker

@@ -7,13 +7,18 @@ is never identity proof. Evidence comes from the separately configured
 external verifier (see fake-verifier).
 
 Usage: fake-worker --task T --turn-output O --round R --actor A
+
+Extra knob: FAKE_SLEEP_SECONDS — sleep that long after the spawn mark,
+before writing the turn (crash/kill-forensics fixtures).
 """
 
 from __future__ import annotations
 
 import argparse
 import json
+import os
 import sys
+import time
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
@@ -33,6 +38,10 @@ if __name__ == "__main__":
     _fakelib.mark_spawn("worker")
     _fakelib.forced_exit()
 
+    sleep = float(os.environ.get("FAKE_SLEEP_SECONDS", "0") or 0)
+    if sleep > 0:
+        time.sleep(sleep)
+
     seed_value = _fakelib.seed(args.actor, args.round)
     turn_text = (
         f"# {args.round} by {args.actor}\n\n"

src/multi_agent_dialogue/cli.py

@@ -142,6 +142,12 @@ def cmd_owner_decide(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_report(args: argparse.Namespace) -> int:
+    report = engine.Dialogue(args.dialogue).build_report()
+    _emit(report)
+    return 0 if report["ok"] else 1
+
+
 def cmd_canary(args: argparse.Namespace) -> int:
     report = canary.run_canary(
         args.dialogue,
@@ -215,6 +221,14 @@ def build_parser() -> argparse.ArgumentParser:
     p.add_argument("--decision", required=True, type=Path)
     p.set_defaults(func=cmd_owner_decide)
 
+    p = sub.add_parser(
+        "report",
+        help="derive the evidence index on demand (read-only; never "
+             "committed, never read by acceptance logic)",
+    )
+    p.add_argument("dialogue", type=Path)
+    p.set_defaults(func=cmd_report)
+
     p = sub.add_parser(
         "canary",
         help="run one turn through the real acceptance path in a fresh local "

src/multi_agent_dialogue/engine.py

@@ -789,6 +789,119 @@ def validate(
             "errors": errors,
         }
 
+    def build_report(self) -> dict:
+        """Derived evidence index over the accepted ledger.
+
+        On-demand and read-only: the report is never committed and never
+        read by acceptance or validation logic. Digests are re-derived
+        from the raw published files, so a tampered byte is flagged here
+        AND by ``validate`` independently.
+        """
+        state = self.state()
+        definition = self.definition()
+        provenance = self._git_provenance(True, state)
+        commits = {
+            entry.get("round_id"): entry
+            for entry in provenance.get("turn_commits") or []
+        }
+        rows: list[dict] = []
+        mismatches: list[str] = []
+        for record in state.get("completed_turns", []):
+            round_id = record.get("round_id")
+            row: dict = {
+                "round_id": round_id,
+                "actor_id": record.get("actor_id"),
+                "completed_via": record.get("completed_via"),
+                "commit": (commits.get(round_id) or {}).get("commit"),
+                "artifact_file": record.get("artifact_file"),
+                "artifact_sha256": record.get("artifact_sha256"),
+                "evidence_file": record.get("evidence_file"),
+                "evidence_sha256": record.get("evidence_sha256"),
+                "session_id": record.get("session_id"),
+                "completed_at": record.get("completed_at"),
+            }
+            for file_key, sha_key, label in (
+                ("artifact_file", "artifact_sha256", "artifact"),
+                ("evidence_file", "evidence_sha256", "evidence"),
+            ):
+                recorded = record.get(sha_key)
+                rel_path = record.get(file_key)
+                if not isinstance(rel_path, str) or not rel_path:
+                    mismatches.append(
+                        f"turn {round_id}: state record has no usable "
+                        f"{file_key}"
+                    )
+                    row[f"{label}_digest_ok"] = False
+                    continue
+                try:
+                    actual = artifacts.sha256_file(self.directory / rel_path)
+                except Exception as exc:  # read-only report must not crash
+                    mismatches.append(
+                        f"turn {round_id}: {label} unreadable: {exc}"
+                    )
+                    row[f"{label}_digest_ok"] = False
+                    continue
+                row[f"{label}_digest_ok"] = actual == recorded
+                if actual != recorded:
+                    mismatches.append(
+                        f"turn {round_id}: {label} digest mismatch — "
+                        f"recorded {recorded!r}, file now hashes {actual!r}"
+                    )
+            # Index fields re-read from the raw evidence bytes.
+            evidence_rel = record.get("evidence_file")
+            if not isinstance(evidence_rel, str) or not evidence_rel:
+                row["evidence_index"] = None
+                rows.append(row)
+                continue
+            try:
+                turn_evidence = evidence.load_evidence(
+                    self.directory / evidence_rel
+                )
+            except evidence.EvidenceError as exc:
+                mismatches.append(f"turn {round_id}: evidence unreadable: {exc}")
+                row["evidence_index"] = None
+            else:
+                cli_version = turn_evidence.get("cli_version")
+                if cli_version is not None and not isinstance(cli_version, dict):
+                    mismatches.append(
+                        f"turn {round_id}: cli_version is not an object "
+                        f"({cli_version!r:.80})"
+                    )
+                    cli_version = None
+                row["evidence_index"] = {
+                    "evidence_version": turn_evidence.get("evidence_version"),
+                    "adapter": turn_evidence.get("adapter"),
+                    "provider": turn_evidence.get("provider"),
+                    "model": turn_evidence.get("model"),
+                    "outcome": turn_evidence.get("outcome"),
+                    "cli_version": (
+                        None
+                        if cli_version is None
+                        else {
+                            "output": cli_version.get("output"),
+                            "output_sha256": cli_version.get("output_sha256"),
+                        }
+                    ),
+                }
+            rows.append(row)
+        provenance_errors = list(provenance.get("errors") or [])
+        return {
+            "ok": not mismatches and not provenance_errors,
+            "protocol_id": state.get("protocol_id"),
+            "status": state.get("status"),
+            "definition_digest": definition.digest(),
+            "turn_count": len(rows),
+            "turns": rows,
+            "mismatches": mismatches,
+            "provenance_errors": provenance_errors,
+            "derived": True,
+            "note": (
+                "derived on demand from the accepted commits and raw "
+                "files; never committed, never read by acceptance or "
+                "validation logic"
+            ),
+        }
+
     def _git_provenance(self, require_git: bool, state: dict) -> dict:
         import subprocess
 

src/multi_agent_dialogue/runner.py

@@ -20,12 +20,107 @@
 from __future__ import annotations
 
 import json
+import os
+import secrets
+import time
+from datetime import datetime, timezone
 from pathlib import Path
 
 from . import adapters, artifacts, config, engine
 
 WORK_DIR = "work"
 
+# Run-attempt receipts live under work/ — the gitignored transient
+# namespace — so they NEVER enter the ledger: they are crash forensics
+# for the recovery path, not acceptance input. Nothing in the acceptance
+# or validation logic reads them.
+RUN_ATTEMPTS_DIR = "run-attempts"
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _receipt_path(context: adapters.PrepareContext, token: str) -> Path:
+    return (
+        context.work_dir.parent
+        / RUN_ATTEMPTS_DIR
+        / f"{context.turn.round_id}-{context.actor.actor_id}-{token}.json"
+    )
+
+
+def _write_receipt(path: Path, receipt: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    engine.atomic_write_json(path, receipt)
+
+
+def _start_receipt(
+    context: adapters.PrepareContext, packet: adapters.CommandPacket
+) -> tuple[Path, dict]:
+    """Write the in-flight receipt BEFORE any process starts.
+
+    If the orchestrator is killed mid-turn, this receipt is what names
+    the attempt afterwards: round, actor, argv digest (never the
+    plaintext argv — it embeds the briefing), runner pid, start time.
+    A receipt that still says ``in_flight`` means the attempt never
+    reported an outcome.
+    """
+    started = time.time()
+    receipt = {
+        "kind": "run-attempt-receipt",
+        "round_id": context.turn.round_id,
+        "actor_id": context.actor.actor_id,
+        "adapter": context.actor.transport,
+        "argv_sha256": artifacts.sha256_bytes(
+            config.canonical_json(list(packet.argv)).encode("utf-8")
+        ),
+        "runner_pid": os.getpid(),
+        "started_at": _utc_now(),
+        "finalized_at": None,
+        "outcome": "in_flight",
+        "exit_status": None,
+        "cleanup": "pending",
+        "detail": None,
+    }
+    path = _receipt_path(
+        context,
+        f"{int(started * 1000):x}-{os.getpid():x}-{secrets.token_hex(2)}",
+    )
+    _write_receipt(path, receipt)
+    return path, receipt
+
+
+def _finalize_receipt(
+    path: Path | None,
+    receipt: dict | None,
+    *,
+    outcome: str,
+    cleanup: str,
+    exit_status: int | None = None,
+    detail: str | None = None,
+) -> None:
+    """Best-effort outcome stamp; the ledger stays the authority.
+
+    A failed finalize leaves the receipt ``in_flight`` — itself a
+    truthful "no outcome was recorded" signal — and must never mask the
+    real outcome of the turn.
+    """
+    if path is None or receipt is None:
+        return
+    receipt.update(
+        {
+            "finalized_at": _utc_now(),
+            "outcome": outcome,
+            "exit_status": exit_status,
+            "cleanup": cleanup,
+            "detail": detail,
+        }
+    )
+    try:
+        _write_receipt(path, receipt)
+    except Exception:
+        pass
+
 
 def _context_for(
     dialogue: engine.Dialogue, actor_id: str
@@ -184,6 +279,14 @@ def launch(dialogue: engine.Dialogue, actor_id: str, timeout: int | None = None)
         raise engine.ProtocolError(str(exc)) from exc
 
     dialogue.claim(actor_id)
+    # Forensics are best-effort: a receipt that cannot be written must
+    # never hold a claim hostage or block a turn the ledger can prove.
+    try:
+        receipt_path: Path | None
+        receipt: dict | None
+        receipt_path, receipt = _start_receipt(context, packet)
+    except Exception:
+        receipt_path, receipt = None, None
     try:
         context.work_dir.mkdir(parents=True, exist_ok=True)
         briefing = build_task_briefing(
@@ -216,20 +319,39 @@ def launch(dialogue: engine.Dialogue, actor_id: str, timeout: int | None = None)
             context.evidence_file,
             completed_via=engine.COMPLETED_VIA_RUNNER_LAUNCH,
         )
+        _finalize_receipt(
+            receipt_path, receipt,
+            outcome="completed",
+            cleanup="claim-consumed-by-completion",
+            exit_status=record.get("exit_status"),
+        )
     except adapters.CleanupUnprovenError as exc:
         # The external worker lane may still be alive. Releasing the claim
         # would let a retry start a duplicate worker beside it, so the
         # claim and its lock stay in place and the dialogue locks BLOCKED
         # (release() refuses BLOCKED dialogues) until a human resolves it.
+        _finalize_receipt(
+            receipt_path, receipt,
+            outcome="failed",
+            cleanup="claim-retained-blocked",
+            detail=str(exc),
+        )
         dialogue.block(f"unproven worker-lane cleanup: {exc}")
         raise engine.ProtocolError(str(exc)) from exc
-    except BaseException:
+    except BaseException as exc:
         # Fail closed but leave the turn claimable again: the adapter has
         # already proven that no worker lane survived this failure.
+        cleanup = "claim-released"
         try:
             dialogue.release(actor_id)
         except engine.ProtocolError:
-            pass
+            cleanup = "release-failed"
+        _finalize_receipt(
+            receipt_path, receipt,
+            outcome="failed",
+            cleanup=cleanup,
+            detail=str(exc),
+        )
         raise
     return {
         "dry_run": False,