From 2e4f6410675b97707f6b5b986bf3cecf6e4d75bf Mon Sep 17 00:00:00 2001 From: Vesper Date: Thu, 20 Aug 2026 06:01:58 +0000 Subject: [PATCH 1/2] feat(runner+cli): run-attempt receipts and madp report (issue #5 proposals 6+7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Run-attempt receipt (redesigned per review): every run --launch writes a receipt under the gitignored work/run-attempts/ recovery namespace — in-flight BEFORE any process starts (argv digest only, never the plaintext argv; runner pid; start time), finalized on outcome (completed/failed, exit status, claim-cleanup state). Never committed, never read by acceptance logic. Kill canary: a launch killed mid-turn leaves an in_flight receipt, an untouched ledger (init commit only), and an unadvanced schedule. madp report (redesigned per review): derives the evidence index on demand from the accepted commits and raw files — per-turn commit SHA, recorded vs recomputed digests, provider/model/session, probed CLI version. Never committed; acceptance/validation never read it. A tampered evidence file is flagged by report AND independently by validate. --- docs/technical-reference.md | 21 +++++ examples/fakes/bin/fake-worker | 9 ++ src/multi_agent_dialogue/cli.py | 14 +++ src/multi_agent_dialogue/engine.py | 96 +++++++++++++++++++++ src/multi_agent_dialogue/runner.py | 113 +++++++++++++++++++++++- tests/test_cli.py | 93 ++++++++++++++++++++ tests/test_provenance_recovery.py | 2 +- tests/test_real_contracts.py | 134 +++++++++++++++++++++++++++++ 8 files changed, 479 insertions(+), 3 deletions(-) diff --git a/docs/technical-reference.md b/docs/technical-reference.md index ed34a96..e0144f3 100644 --- a/docs/technical-reference.md +++ b/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 diff --git a/examples/fakes/bin/fake-worker b/examples/fakes/bin/fake-worker index 00b9b3b..31039b1 100755 --- a/examples/fakes/bin/fake-worker +++ b/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" diff --git a/src/multi_agent_dialogue/cli.py b/src/multi_agent_dialogue/cli.py index a6aed6a..d2f1bab 100644 --- a/src/multi_agent_dialogue/cli.py +++ b/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 " diff --git a/src/multi_agent_dialogue/engine.py b/src/multi_agent_dialogue/engine.py index 2ca0ab8..77759fc 100644 --- a/src/multi_agent_dialogue/engine.py +++ b/src/multi_agent_dialogue/engine.py @@ -789,6 +789,102 @@ 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) + try: + actual = artifacts.sha256_file( + self.directory / record[file_key] + ) + 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. + try: + turn_evidence = evidence.load_evidence( + self.directory / record["evidence_file"] + ) + 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") or 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 diff --git a/src/multi_agent_dialogue/runner.py b/src/multi_agent_dialogue/runner.py index 2a6f07b..d6da30e 100644 --- a/src/multi_agent_dialogue/runner.py +++ b/src/multi_agent_dialogue/runner.py @@ -20,12 +20,101 @@ from __future__ import annotations import json +import os +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}") + _write_receipt(path, receipt) + return path, receipt + + +def _finalize_receipt( + path: Path, + receipt: dict, + *, + 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. + """ + 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 +273,7 @@ def launch(dialogue: engine.Dialogue, actor_id: str, timeout: int | None = None) raise engine.ProtocolError(str(exc)) from exc dialogue.claim(actor_id) + receipt_path, receipt = _start_receipt(context, packet) try: context.work_dir.mkdir(parents=True, exist_ok=True) briefing = build_task_briefing( @@ -216,20 +306,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, diff --git a/tests/test_cli.py b/tests/test_cli.py index bff9cee..b292573 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -7,6 +7,7 @@ import subprocess import sys import tempfile +import time import unittest from pathlib import Path @@ -359,5 +360,97 @@ def test_real_binary_override_requires_explicit_identity(self) -> None: self.assertIn("never guesses identity", result.stderr) +class ReportCliTests(ValidateAndOwnerTests): + def test_report_command_indexes_the_ledger(self) -> None: + self.init() + self.run_all_turns() + result = run_cli("report", str(self.dialogue_dir)) + self.assertEqual(result.returncode, 0, result.stderr) + payload = json.loads(result.stdout) + self.assertTrue(payload["ok"], payload["mismatches"]) + self.assertEqual(payload["turn_count"], 4) + self.assertTrue(payload["derived"]) + for row in payload["turns"]: + self.assertTrue(row["artifact_digest_ok"]) + self.assertTrue(row["evidence_digest_ok"]) + + def test_report_flags_tamper_with_nonzero_exit(self) -> None: + self.init() + self.run_all_turns() + state = json.loads( + (self.dialogue_dir / "state.json").read_text(encoding="utf-8") + ) + target = self.dialogue_dir / state["completed_turns"][1]["artifact_file"] + with target.open("a", encoding="utf-8") as handle: + handle.write("\ntampered\n") + result = run_cli("report", str(self.dialogue_dir)) + self.assertEqual(result.returncode, 1) + payload = json.loads(result.stdout) + self.assertFalse(payload["ok"]) + self.assertTrue(payload["mismatches"]) + + +class KillReceiptCliTests(CliTestCase): + """Kill canary: a launch killed mid-turn leaves an in-flight receipt + in the recovery namespace and an untouched ledger.""" + + def setUp(self) -> None: + super().setUp() + for actor in self.raw["actors"]: + actor["settings"].setdefault("env", {})["FAKE_SLEEP_SECONDS"] = "30" + self.definition_path.write_text( + json.dumps(self.raw), encoding="utf-8" + ) + + def test_killed_mid_turn(self) -> None: + self.init() + env = dict(os.environ) + env["PYTHONPATH"] = ( + str(support.SRC) + os.pathsep + env.get("PYTHONPATH", "") + ) + proc = subprocess.Popen( + [sys.executable, "-m", "multi_agent_dialogue", "run", + str(self.dialogue_dir), "--actor", "worker-a", "--launch"], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + env=env, cwd=support.REPO_ROOT, + ) + receipt_dir = self.dialogue_dir / "work" / "run-attempts" + receipt_path = None + try: + deadline = time.time() + 20 + while time.time() < deadline: + found = ( + sorted(receipt_dir.glob("*.json")) + if receipt_dir.is_dir() + else [] + ) + if found: + receipt_path = found[0] + break + time.sleep(0.2) + self.assertIsNotNone( + receipt_path, "the in-flight receipt was never written" + ) + finally: + proc.kill() + proc.wait() + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + self.assertEqual(receipt["outcome"], "in_flight") + self.assertIsNone(receipt["finalized_at"]) + self.assertEqual(receipt["cleanup"], "pending") + # The ledger never saw the turn: no completion, no turn commit. + state = json.loads( + (self.dialogue_dir / "state.json").read_text(encoding="utf-8") + ) + self.assertEqual(state["completed_turns"], []) + self.assertEqual(state["turn_index"], 0) + commits = subprocess.run( + ["git", "-C", str(self.dialogue_dir), + "rev-list", "--count", "HEAD"], + capture_output=True, text=True, check=True, + ).stdout.strip() + self.assertEqual(commits, "1", "only the init commit exists") + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_provenance_recovery.py b/tests/test_provenance_recovery.py index 69010ec..62203b7 100644 --- a/tests/test_provenance_recovery.py +++ b/tests/test_provenance_recovery.py @@ -159,7 +159,7 @@ def test_top_level_help_has_exactly_the_public_commands(self) -> None: self.assertEqual( subcommands(result.stdout), {"init", "status", "next", "run", "validate", "owner-decide", - "canary"}, + "canary", "report"}, "top-level madp must expose exactly the public path " "(claim/prepare/complete are recovery-only)", ) diff --git a/tests/test_real_contracts.py b/tests/test_real_contracts.py index 9d287ef..0a375f3 100644 --- a/tests/test_real_contracts.py +++ b/tests/test_real_contracts.py @@ -1303,5 +1303,139 @@ def test_worker_without_version_flag_is_recorded_not_fatal(self) -> None: self.assertEqual(dialogue.state()["turn_index"], 1) +class RunAttemptReceiptTests(CommandFixtureBase): + """Run-attempt receipts: crash forensics in the gitignored work/ + namespace — never committed to the ledger, never read by acceptance.""" + + def _settings(self) -> dict: + return { + "argv": self.worker_argv(), + "identity_verifier_argv": self.verifier_argv(), + "env": { + "FAKE_PROVIDER": "fake-provider-a", + "FAKE_MODEL": "fake-model-a", + "FAKE_SPAWN_MARKER": str(self.marker), + }, + } + + def _receipts(self, dialogue: engine.Dialogue) -> list[Path]: + directory = dialogue.directory / "work" / "run-attempts" + if not directory.is_dir(): + return [] + return sorted(directory.glob("*.json")) + + def test_completed_turn_writes_finalized_receipt(self) -> None: + dialogue = self.make_dialogue(self._settings()) + runner.launch(dialogue, "worker-a") + receipts = self._receipts(dialogue) + self.assertEqual(len(receipts), 1) + receipt = json.loads(receipts[0].read_text(encoding="utf-8")) + self.assertEqual(receipt["kind"], "run-attempt-receipt") + self.assertEqual(receipt["round_id"], "R01") + self.assertEqual(receipt["actor_id"], "worker-a") + self.assertEqual(receipt["outcome"], "completed") + self.assertEqual(receipt["exit_status"], 0) + self.assertEqual(receipt["cleanup"], "claim-consumed-by-completion") + self.assertEqual(receipt["runner_pid"], os.getpid()) + self.assertIsNotNone(receipt["finalized_at"]) + self.assertEqual(len(receipt["argv_sha256"]), 64) + self.assertNotIn("argv", receipt, "redacted: digest only") + + def test_failed_turn_receipt_records_claim_release(self) -> None: + settings = self._settings() + settings["env"]["FAKE_EXIT"] = "1" + dialogue = self.make_dialogue(settings) + with self.assertRaises((engine.ProtocolError, adapters.AdapterError)): + runner.launch(dialogue, "worker-a") + receipts = self._receipts(dialogue) + self.assertEqual(len(receipts), 1) + receipt = json.loads(receipts[0].read_text(encoding="utf-8")) + self.assertEqual(receipt["outcome"], "failed") + self.assertEqual(receipt["cleanup"], "claim-released") + self.assertIn("exited 1", receipt["detail"]) + + def test_receipt_never_enters_the_ledger(self) -> None: + dialogue = self.make_dialogue(self._settings()) + runner.launch(dialogue, "worker-a") + tracked = support.git(dialogue.directory, "ls-files").stdout + self.assertNotIn("run-attempts", tracked) + report = dialogue.validate( + require_git=True, require_runner_completion=True + ) + self.assertTrue(report["ok"], report["errors"]) + + +class BuildReportTests(CommandFixtureBase): + """`madp report`: derived evidence index — read-only, never committed.""" + + def _settings(self) -> dict: + return { + "argv": self.worker_argv(), + "identity_verifier_argv": self.verifier_argv(), + "env": { + "FAKE_PROVIDER": "fake-provider-a", + "FAKE_MODEL": "fake-model-a", + "FAKE_SPAWN_MARKER": str(self.marker), + }, + } + + def test_report_indexes_accepted_turns(self) -> None: + dialogue = self.make_dialogue(self._settings()) + runner.launch(dialogue, "worker-a") + report = dialogue.build_report() + self.assertTrue(report["ok"], report["mismatches"]) + self.assertEqual(report["turn_count"], 1) + row = report["turns"][0] + self.assertEqual(row["round_id"], "R01") + self.assertEqual(row["completed_via"], "runner-launch") + self.assertTrue(row["artifact_digest_ok"]) + self.assertTrue(row["evidence_digest_ok"]) + self.assertEqual(len(row["commit"]), 40) + index = row["evidence_index"] + self.assertEqual(index["provider"], "fake-provider-a") + self.assertEqual(index["model"], "fake-model-a") + self.assertEqual(index["evidence_version"], 1) + self.assertIn("fake-worker", index["cli_version"]["output"]) + + def test_tampered_evidence_is_flagged_by_report_and_validate(self) -> None: + dialogue = self.make_dialogue(self._settings()) + runner.launch(dialogue, "worker-a") + evidence_file = ( + dialogue.directory + / dialogue.state()["completed_turns"][0]["evidence_file"] + ) + payload = json.loads(evidence_file.read_text(encoding="utf-8")) + payload["model"] = "tampered-model" + evidence_file.write_text( + json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + + "\n", + encoding="utf-8", + ) + report = dialogue.build_report() + self.assertFalse(report["ok"]) + self.assertTrue( + any("evidence digest mismatch" in m for m in report["mismatches"]) + ) + # validate re-checks the raw evidence independently of the report. + validation = dialogue.validate(require_git=True) + self.assertFalse(validation["ok"]) + + def test_report_is_read_only(self) -> None: + dialogue = self.make_dialogue(self._settings()) + runner.launch(dialogue, "worker-a") + before = support.git( + dialogue.directory, "rev-list", "--count", "HEAD" + ).stdout.strip() + dialogue.build_report() + after = support.git( + dialogue.directory, "rev-list", "--count", "HEAD" + ).stdout.strip() + self.assertEqual(before, after, "report never commits") + status = support.git( + dialogue.directory, "status", "--porcelain", "--", "dialogue" + ).stdout.strip() + self.assertEqual(status, "", "report leaves the worktree untouched") + + if __name__ == "__main__": unittest.main() From ad896a36d2756fa032a64b64b464350a061530e0 Mon Sep 17 00:00:00 2001 From: Vesper Date: Thu, 20 Aug 2026 06:12:10 +0000 Subject: [PATCH 2/2] fix(runner+engine): review hardening for receipts and report MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - receipt start is best-effort (never holds a claim hostage); filename token gains pid + random suffix so fast release-and-retry cannot overwrite a prior receipt; - build_report guards malformed state records (missing file keys) and non-dict cli_version instead of crashing — tampering is flagged, not fatal to the report; - tests: run_all_turns moved to the shared fixture base (no inherited re-run); the kill canary kills the process group and drains the pipes, so no sleeping fake-worker is orphaned. --- src/multi_agent_dialogue/engine.py | 27 ++++++++++++++++++++++----- src/multi_agent_dialogue/runner.py | 21 +++++++++++++++++---- tests/test_cli.py | 20 ++++++++++++-------- 3 files changed, 51 insertions(+), 17 deletions(-) diff --git a/src/multi_agent_dialogue/engine.py b/src/multi_agent_dialogue/engine.py index 77759fc..76f1f3b 100644 --- a/src/multi_agent_dialogue/engine.py +++ b/src/multi_agent_dialogue/engine.py @@ -825,10 +825,16 @@ def build_report(self) -> dict: ("evidence_file", "evidence_sha256", "evidence"), ): recorded = record.get(sha_key) - try: - actual = artifacts.sha256_file( - self.directory / record[file_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}" @@ -842,15 +848,26 @@ def build_report(self) -> dict: 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 / record["evidence_file"] + 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") or None + 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"), diff --git a/src/multi_agent_dialogue/runner.py b/src/multi_agent_dialogue/runner.py index d6da30e..f28378b 100644 --- a/src/multi_agent_dialogue/runner.py +++ b/src/multi_agent_dialogue/runner.py @@ -21,6 +21,7 @@ import json import os +import secrets import time from datetime import datetime, timezone from pathlib import Path @@ -81,14 +82,17 @@ def _start_receipt( "cleanup": "pending", "detail": None, } - path = _receipt_path(context, f"{int(started * 1000):x}") + 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, - receipt: dict, + path: Path | None, + receipt: dict | None, *, outcome: str, cleanup: str, @@ -101,6 +105,8 @@ def _finalize_receipt( 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(), @@ -273,7 +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) - receipt_path, receipt = _start_receipt(context, packet) + # 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( diff --git a/tests/test_cli.py b/tests/test_cli.py index b292573..b3f1c04 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -4,6 +4,7 @@ import json import os +import signal import subprocess import sys import tempfile @@ -67,6 +68,11 @@ def spawn_count(self) -> int: lines = self.marker.read_text(encoding="utf-8").splitlines() return sum(1 for line in lines if line == "worker") + def run_all_turns(self) -> None: + for actor in ("worker-a", "worker-b", "worker-a", "worker-b"): + result = run_cli("run", str(self.dialogue_dir), "--actor", actor, "--launch") + self.assertEqual(result.returncode, 0, result.stderr) + class InitStatusNextTests(CliTestCase): def test_init_status_next(self) -> None: @@ -215,11 +221,6 @@ def test_complete_with_fake_identity_fails(self) -> None: class ValidateAndOwnerTests(CliTestCase): - def run_all_turns(self) -> None: - for actor in ("worker-a", "worker-b", "worker-a", "worker-b"): - result = run_cli("run", str(self.dialogue_dir), "--actor", actor, "--launch") - self.assertEqual(result.returncode, 0, result.stderr) - def test_validate_clean_and_tampered(self) -> None: self.init() self.run_all_turns() @@ -360,7 +361,7 @@ def test_real_binary_override_requires_explicit_identity(self) -> None: self.assertIn("never guesses identity", result.stderr) -class ReportCliTests(ValidateAndOwnerTests): +class ReportCliTests(CliTestCase): def test_report_command_indexes_the_ledger(self) -> None: self.init() self.run_all_turns() @@ -413,6 +414,7 @@ def test_killed_mid_turn(self) -> None: str(self.dialogue_dir), "--actor", "worker-a", "--launch"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env, cwd=support.REPO_ROOT, + start_new_session=True, ) receipt_dir = self.dialogue_dir / "work" / "run-attempts" receipt_path = None @@ -432,8 +434,10 @@ def test_killed_mid_turn(self) -> None: receipt_path, "the in-flight receipt was never written" ) finally: - proc.kill() - proc.wait() + # Kill the whole process group: proc.kill() alone would + # orphan the sleeping fake-worker child for up to 30s. + os.killpg(proc.pid, signal.SIGKILL) + proc.communicate() receipt = json.loads(receipt_path.read_text(encoding="utf-8")) self.assertEqual(receipt["outcome"], "in_flight") self.assertIsNone(receipt["finalized_at"])