From 0af6b99eea2eeea832f886f16b54a77cf6524635 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 27 Jun 2026 15:59:19 +0000 Subject: [PATCH 1/2] Refine local demo flow: cleaner output, clearer enforcement visibility demo.py: - Align per-call output with demo_style.md vocabulary (ALLOW / CHECKPOINT / BLOCK) and show only observable behavior: mapped policy action, requested tool name, gate result, returned tool result. No inner/model reasoning. - Compute plan-box and section-rule widths so borders always line up; add a Scope summary row. - Add a one-line flow legend (plan -> gate -> tools -> receipts -> verify). - Show the mapped action per receipt in the VERIFY list. - Reduce duplication: shared clip() helper, single end-state footer, drop tool_call_label in favor of the action string. - Export audit pack with ensure_ascii=False for readable em-dashes (receipt signatures are computed in-memory and are unaffected). run_demo.sh: - Trim inter-section pauses and tighten the signing notes prose. Enforcement semantics, real SHA-256 receipt hashing, chain verification, the signed plan box, the 3s blocked pause, and live-Qwen fallback are all preserved. demo.py stays stdlib-only and under 400 lines. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CHN9wc1BVmttLHqMpsYLU4 --- audit_pack.json | 70 ++++++++++++++++---------- demo.py | 131 ++++++++++++++++++++++++++---------------------- run_demo.sh | 8 +-- 3 files changed, 118 insertions(+), 91 deletions(-) diff --git a/audit_pack.json b/audit_pack.json index 1e42502..501f5b6 100644 --- a/audit_pack.json +++ b/audit_pack.json @@ -2,10 +2,10 @@ "plan": { "id": "plan_7f3a2e91", "patient": "PT-4827", - "provider": "Dr. Sarah Patel \u2014 NPI 1234567890", - "procedure": "CPT 72148 \u2014 MRI lumbar spine w/o contrast", - "diagnosis": "ICD-10 M51.36 \u2014 Lumbar disc degeneration", - "payer": "Aetna PPO \u2014 Policy CPB-0451", + "provider": "Dr. Sarah Patel — NPI 1234567890", + "procedure": "CPT 72148 — MRI lumbar spine w/o contrast", + "diagnosis": "ICD-10 M51.36 — Lumbar disc degeneration", + "payer": "Aetna PPO — Policy CPB-0451", "scope": [ "read:PT-4827:clinical-note", "read:PT-4827:eligibility", @@ -23,55 +23,73 @@ }, "attestations": [ { - "receipt": "1d921a7b", + "receipt": "7a551424", "note": "physician sign-off" } ], "receipts": [ { - "id": "23ba6bc5", + "id": "82c77fed", + "action": "read:PT-4827:clinical-note", + "in_policy": true, + "reason": "matched read:PT-4827:clinical-note", + "prev": null, + "ts": "2026-06-27T15:56:46.891868+00:00", + "sig": "c28f8b02291a020e" + }, + { + "id": "3e7176df", "action": "read:PT-4827:eligibility", "in_policy": true, "reason": "matched read:PT-4827:eligibility", - "prev": null, - "ts": "2026-06-27T15:46:43.641123+00:00", - "sig": "d408a50e8872913c" + "prev": "f9b2e29b6e567663", + "ts": "2026-06-27T15:56:47.492219+00:00", + "sig": "f5d9303cfe7683f5" }, { - "id": "a2302298", + "id": "172a8523", "action": "read:PT-4827:payer-policy", "in_policy": true, "reason": "matched read:PT-4827:payer-policy", - "prev": "7d83eb638e98cb8e", - "ts": "2026-06-27T15:46:44.245546+00:00", - "sig": "aee271010b61c9a1" + "prev": "915c2d78876aaec4", + "ts": "2026-06-27T15:56:48.092534+00:00", + "sig": "bd24f5c2a515a0d2" }, { - "id": "2811c7af", + "id": "3fa0a55a", "action": "read:PT-9914:clinical-note", "in_policy": false, "reason": "out of scope", - "prev": "0cbc62bb80bfd8ad", - "ts": "2026-06-27T15:46:44.850990+00:00", - "sig": "31a4e5f7ff9b3b97" + "prev": "be60ac52c6e9158e", + "ts": "2026-06-27T15:56:48.692849+00:00", + "sig": "280c00e6543b1d75" + }, + { + "id": "d7e8dbc8", + "action": "read:PT-4827:behavioral-health", + "in_policy": false, + "reason": "out of scope", + "prev": "d7d077fcfc3da2f5", + "ts": "2026-06-27T15:56:51.693171+00:00", + "sig": "fde676b0a4a1c17c" }, { - "id": "1d921a7b", + "id": "7a551424", "action": "submit:PT-4827:auth-request", "in_policy": true, "reason": "checkpoint: submit:* | attested", - "prev": "c246f05b69bb6843", - "ts": "2026-06-27T15:47:14.864954+00:00", - "sig": "453bacb52a1b0188" + "prev": "5c87cab1bb983c8c", + "ts": "2026-06-27T15:56:54.693489+00:00", + "sig": "583f1680e833146d" }, { - "id": "5e38c243", + "id": "415cc156", "action": "write:PT-4827:ehr-approval", "in_policy": true, "reason": "matched write:PT-4827:ehr-approval", - "prev": "f546951d996edda7", - "ts": "2026-06-27T15:47:16.370612+00:00", - "sig": "e38c955192b85c2f" + "prev": "6504c9218bb63466", + "ts": "2026-06-27T15:56:56.193800+00:00", + "sig": "6a8d85640c7ab4a0" } ] -} \ No newline at end of file +} diff --git a/demo.py b/demo.py index 8aa7b20..e15b291 100644 --- a/demo.py +++ b/demo.py @@ -98,6 +98,7 @@ "write:PT-4827:ehr-approval": '{"status":"written","auth":"AET-MRI-88712"}', } BLOCKED_RESPONSE = '{"error":"ACCESS DENIED — out of plan scope"}' +AGENT = "prior-auth-agent" SCRIPTED = [ ("read_patient_record", {"patient_mrn": "PT-4827", "record_type": "clinical-note"}), ("read_patient_record", {"patient_mrn": "PT-4827", "record_type": "eligibility"}), @@ -133,15 +134,9 @@ def tool_call_to_action(name, args): if name == "write_to_ehr": return "write:%s:ehr-approval" % mrn return "unknown:%s" % name -def tool_call_label(name, args): - mrn = args.get("patient_mrn", "?") - if name == "read_patient_record": - return "Read %s — %s" % (args.get("record_type", "?"), mrn) - if name == "submit_auth_request": - return "Submit auth to %s — %s" % (args.get("payer_id", "?"), mrn) - if name == "write_to_ehr": - return "Write %s — %s" % (args.get("field", "?"), mrn) - return name +def clip(text, width): + text = str(text) + return text if len(text) <= width else text[: width - 1] + "…" class Receipt: def __init__(self, action, agent, in_policy, reason, prev_hash): self.id = uuid.uuid4().hex[:8] @@ -186,44 +181,63 @@ def pause(seconds): def brand(): print(FG + BOLD + " AgentMint Prior Auth Demo" + RESET) print(DIM + " signed plan gating for local Qwen tool calls" + RESET) + print(GRAY + " flow: plan -> gate -> tools -> receipts -> verify" + RESET) print() def note(text): print(YELLOW + " ! " + text + RESET) +BOX_W = 54 # inner content width; borders are computed to match + + +def _rule(left, right, label): + inner = (" %s " % label).center(BOX_W + 2, "─") if label else "─" * (BOX_W + 2) + print(BLUE + " " + left + inner + right + RESET) + + def plan_box(): lines = [ - ("Plan:", PLAN["id"]), - ("Patient:", "%s (scope locked)" % PLAN["patient"]), - ("Provider:", PLAN["provider"]), - ("Procedure:", PLAN["procedure"]), - ("Payer:", PLAN["payer"]), - ("TTL:", "%ss" % PLAN["ttl"]), + ("Plan", PLAN["id"]), + ("Patient", "%s (scope locked)" % PLAN["patient"]), + ("Provider", PLAN["provider"]), + ("Procedure", PLAN["procedure"]), + ("Payer", PLAN["payer"]), + ("Scope", "%d actions, %d checkpoint" % (len(PLAN["scope"]), len(PLAN["checkpoints"]))), + ("TTL", "%ss" % PLAN["ttl"]), ] - print(BLUE + " ┌─ AgentMint ── SIGNED PLAN ─────────────────────────────┐" + RESET) - for left, right in lines: - text = " │ %-10s %-45s│" % (left, right[:45]) - print(BLUE + text + RESET) - print(BLUE + " └────────────────────────────────── plan signed ──────────┘" + RESET) + _rule("┌", "┐", "AgentMint · SIGNED PLAN") + for label, value in lines: + body = "%-9s %s" % (label, value) + print(BLUE + " │ " + clip(body, BOX_W).ljust(BOX_W) + " │" + RESET) + _rule("└", "┘", "plan signed") print() -def print_status(kind, label, receipt_id, reason): - icon = {"pass": "✓", "block": "✗", "checkpoint": "⏸"}[kind] - color = {"pass": GREEN, "block": RED, "checkpoint": YELLOW}[kind] - suffix = {"pass": "receipt %s" % receipt_id, "block": "BLOCKED", "checkpoint": "CHECKPOINT"}[kind] - print(color + " %s %-40s %s" % (icon, label[:40], suffix) + RESET) +STATUS = { + "pass": ("✓", GREEN, "ALLOW"), + "checkpoint": ("⏸", YELLOW, "CHECKPOINT"), + "block": ("✗", RED, "BLOCK"), +} + + +def print_status(kind, action, tool, receipt_id, reason, result): + icon, color, word = STATUS[kind] + print(color + " %s %-10s %-32s receipt %s" % (icon, word, clip(action, 32), receipt_id[:8]) + RESET) if kind == "block": - print(DIM + " reason: %s" % reason + RESET) - if kind == "checkpoint": - print(DIM + " attestation: physician sign-off recorded" + RESET) + print(DIM + " gate: %s · returned %s" % (reason, clip(result, 34)) + RESET) + elif kind == "checkpoint": + print(DIM + " via %s · human sign-off recorded" % tool + RESET) + else: + print(DIM + " via %s · returned %s" % (tool, clip(result, 40)) + RESET) def show_intro(model_name=None): clear() brand() plan_box() pause(2.0) - print(GRAY + " Enforcement" + RESET) - print(GREEN + " ✓ Allowed" + RESET + DIM + " inside the signed plan scope" + RESET) - print(YELLOW + " ⏸ Checkpoint" + RESET + DIM + " requires explicit human sign-off" + RESET) - print(RED + " ✗ Blocked" + RESET + DIM + " outside scope, even if the model asks for it" + RESET) - print(DIM + " Normally an agent follows prompts and note references on its own." + RESET) - print(DIM + " Here every tool call is checked first, then receipted into a tamper-evident chain." + RESET) + print(GRAY + " Legend" + RESET) + print(GREEN + " ✓ ALLOW" + RESET + DIM + " action matched the signed plan scope" + RESET) + print(YELLOW + " ⏸ CHECKPOINT" + RESET + DIM + " in scope, held for human sign-off" + RESET) + print(RED + " ✗ BLOCK" + RESET + DIM + " outside scope, even if the model asks for it" + RESET) + print() + print(DIM + " A normal agent would follow the note and call tools directly." + RESET) + print(DIM + " AgentMint gates each call against the signed plan, then receipts it" + RESET) + print(DIM + " into a hash-linked chain. Only observable behavior is shown below." + RESET) print() print(FG + " Agent processing prior auth for PT-4827..." + RESET) if model_name: @@ -237,7 +251,8 @@ def write_audit_pack(chain, attestations): "receipts": [dict(r._d(), sig=r.sig) for r in chain.receipts], } with open(path, "w") as f: - json.dump(data, f, indent=2) + json.dump(data, f, indent=2, ensure_ascii=False) + f.write("\n") return path def detect_model_name(): override = os.environ.get("LMSTUDIO_MODEL") @@ -271,21 +286,22 @@ def mock_call(step, name, args): return name, args, False def handle_tool(name, args, chain, attestations): action = tool_call_to_action(name, args) - label = tool_call_label(name, args) - ok, reason = evaluate_policy(action, "prior-auth-agent", PLAN["scope"], PLAN["checkpoints"], PLAN["delegates"]) + ok, reason = evaluate_policy(action, AGENT, PLAN["scope"], PLAN["checkpoints"], PLAN["delegates"]) if ok: - receipt = chain.add(action, "prior-auth-agent", True, reason) - print_status("pass", label, receipt.id, reason) + result = MOCK_RESPONSES.get(action, '{"status":"ok"}') + receipt = chain.add(action, AGENT, True, reason) + print_status("pass", action, name, receipt.id, reason, result) pause(0.6) - return MOCK_RESPONSES.get(action, '{"status":"ok"}') + return result if reason.startswith("checkpoint:"): - receipt = chain.add(action, "prior-auth-agent", True, reason + " | attested") + result = MOCK_RESPONSES.get(action, '{"status":"approved"}') + receipt = chain.add(action, AGENT, True, reason + " | attested") attestations.append({"receipt": receipt.id, "note": "physician sign-off"}) - print_status("checkpoint", label, receipt.id, reason) + print_status("checkpoint", action, name, receipt.id, reason, result) pause(1.5) - return MOCK_RESPONSES.get(action, '{"status":"approved"}') - receipt = chain.add(action, "prior-auth-agent", False, reason) - print_status("block", label, receipt.id, reason) + return result + receipt = chain.add(action, AGENT, False, reason) + print_status("block", action, name, receipt.id, reason, BLOCKED_RESPONSE) pause(3.0) return BLOCKED_RESPONSE def verify_and_print(chain, attestations): @@ -293,26 +309,27 @@ def verify_and_print(chain, attestations): ok, count = chain.verify() blocked = 0 print() - print(GRAY + " ─── VERIFY ──────────────────────────────────────────────" + RESET) + _rule("─", "─", "VERIFY") print() print(DIM + " $ ./verify.sh %s --pubkey health_system.pub" % path + RESET) print() for receipt in chain.receipts: blocked += 0 if receipt.in_policy else 1 mark = GREEN + "✓" + RESET if receipt.sig_valid() else RED + "✗" + RESET - extra = " ⚠ BLOCKED" if not receipt.in_policy else "" - print(" %s Receipt %s sig valid chain ok%s" % (mark, receipt.id[:4], extra)) + flag = RED + " BLOCKED" + RESET if not receipt.in_policy else "" + print(" %s %s %-32s sig ✓ link ✓%s" % (mark, receipt.id[:4], clip(receipt.action, 32), flag)) pause(0.3) print() - print(" Chain: %s/%s verified" % (count, len(chain.receipts))) - print(" Blocked: %s out-of-scope" % blocked) - print(" Attested: %s physician sign-off" % len(attestations)) - print(" Tampered: %s" % (0 if ok else 1)) - print() - print(GREEN + BOLD + " ✓ AUDIT PACK VERIFIED" + RESET) - pause(2.0) + print(" Chain %s/%s receipts verified" % (count, len(chain.receipts))) + print(" Blocked %s out-of-scope attempt(s) receipted" % blocked) + print(" Attested %s physician sign-off" % len(attestations)) + print(" Tampered %s" % (0 if ok else 1)) print() + print((GREEN if ok else RED) + BOLD + " %s AUDIT PACK VERIFIED" % ("✓" if ok else "✗") + RESET) print(DIM + " exported: %s" % path + RESET) + print() + print(BLUE + BOLD + " end state: authorized patient stayed in scope, out-of-scope reads blocked" + RESET) + pause(2.0) def run_scripted(start=0, chain=None, attestations=None, reason=None, started=False): chain = chain or Chain() attestations = attestations or [] @@ -328,8 +345,6 @@ def run_scripted(start=0, chain=None, attestations=None, reason=None, started=Fa for name, args in SCRIPTED[start:]: handle_tool(name, args, chain, attestations) verify_and_print(chain, attestations) - print() - print(BLUE + BOLD + " end state: authorized patient stayed in scope, injection failed" + RESET) def run_live(): chain = Chain() attestations = [] @@ -370,8 +385,6 @@ def run_live(): result = handle_tool(fn, args, chain, attestations) messages.append({"role": "tool", "tool_call_id": call.get("id"), "name": fn, "content": result}) verify_and_print(chain, attestations) - print() - print(BLUE + BOLD + " end state: authorized patient stayed in scope, injection failed" + RESET) if __name__ == "__main__": if "--scripted" in sys.argv: run_scripted() diff --git a/run_demo.sh b/run_demo.sh index 6088300..38f9b08 100755 --- a/run_demo.sh +++ b/run_demo.sh @@ -78,7 +78,6 @@ line "$QWEN_INFO" | while IFS= read -r row; do QWEN_ERROR=*) line "inference test: ${row#QWEN_ERROR=}" ;; esac done -wait_a_bit section "[2/4] cli plan setup" run_cmd "cd \"$TMP_DIR\" && $CLI init . --yes >/tmp/agentmint-init.$$" @@ -92,13 +91,10 @@ PLAN_OUTPUT="$(cd "$TMP_DIR" && bash -lc "$CLI plan create --name prior-auth-dem line "$PLAN_OUTPUT" PLAN_ID="$(printf "%s\n" "$PLAN_OUTPUT" | awk '{print $4}')" run_cmd "cd \"$TMP_DIR\" && $CLI plan show \"$PLAN_ID\"" -wait_a_bit section "[3/4] signing notes" -line "plan signing: the plan is signed when created with the local private key." -line "tamper evidence: each later receipt links to the previous one by hash." -line "verification result changes if a plan, signature, or chain link is edited." -wait_a_bit +line "plan is signed at creation with the local private key." +line "each receipt links to the previous one by hash, so any edit breaks verification." section "[4/4] runtime demo" line "legend: ALLOW=green CHECKPOINT=yellow BLOCK=red" From 566397ed33bd1e478c2e9f7e732a330b7aadb401 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 27 Jun 2026 16:16:24 +0000 Subject: [PATCH 2/2] =?UTF-8?q?Add=20recordable=20demo=20chunks=20(baselin?= =?UTF-8?q?e=20=E2=86=92=20plan=20=E2=86=92=20enforce=20=E2=86=92=20verify?= =?UTF-8?q?=20=E2=86=92=20tamper)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit chunks.py splits the demo into five standalone, individually recordable segments that reuse demo.py's logic (no duplication, no weakened enforcement): 1. baseline — without AgentMint: ungated agent reads another patient's record (PT-9914) and behavioral-health, submits/writes with no human checkpoint, leaves no audit trail. Bad outcome stated from observable behavior only. 2. plan — plan creation via the CLI (init + plan create + plan show). 3. enforce — with AgentMint: each call gated against the signed plan (ALLOW/CHECKPOINT/BLOCK), writes the signed receipt chain to audit_pack.json. 4. verify — independent verification that reads audit_pack.json and recomputes signatures and chain links with stdlib only, no runtime. 5. tamper — edits a receipt in a copy, re-verifies, and shows the signature + chain link break; the on-disk pack is left unchanged. Run one chunk (python3 chunks.py verify) for per-segment recording, or all of them (python3 chunks.py all). stdlib-only; demo.py untouched. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CHN9wc1BVmttLHqMpsYLU4 --- audit_pack.json | 56 ++++++------- chunks.py | 219 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 247 insertions(+), 28 deletions(-) create mode 100644 chunks.py diff --git a/audit_pack.json b/audit_pack.json index 501f5b6..37c240c 100644 --- a/audit_pack.json +++ b/audit_pack.json @@ -23,73 +23,73 @@ }, "attestations": [ { - "receipt": "7a551424", + "receipt": "c9e84564", "note": "physician sign-off" } ], "receipts": [ { - "id": "82c77fed", + "id": "990aa4a3", "action": "read:PT-4827:clinical-note", "in_policy": true, "reason": "matched read:PT-4827:clinical-note", "prev": null, - "ts": "2026-06-27T15:56:46.891868+00:00", - "sig": "c28f8b02291a020e" + "ts": "2026-06-27T16:15:57.200318+00:00", + "sig": "8b23e7675ebdcee1" }, { - "id": "3e7176df", + "id": "b90b71e8", "action": "read:PT-4827:eligibility", "in_policy": true, "reason": "matched read:PT-4827:eligibility", - "prev": "f9b2e29b6e567663", - "ts": "2026-06-27T15:56:47.492219+00:00", - "sig": "f5d9303cfe7683f5" + "prev": "7e5dfd31af6a98b5", + "ts": "2026-06-27T16:15:57.800672+00:00", + "sig": "a03d6e6a3e47763f" }, { - "id": "172a8523", + "id": "ddf41447", "action": "read:PT-4827:payer-policy", "in_policy": true, "reason": "matched read:PT-4827:payer-policy", - "prev": "915c2d78876aaec4", - "ts": "2026-06-27T15:56:48.092534+00:00", - "sig": "bd24f5c2a515a0d2" + "prev": "c7cb9af63cc4a2e7", + "ts": "2026-06-27T16:15:58.400981+00:00", + "sig": "a6119034c7232a1c" }, { - "id": "3fa0a55a", + "id": "eb781d39", "action": "read:PT-9914:clinical-note", "in_policy": false, "reason": "out of scope", - "prev": "be60ac52c6e9158e", - "ts": "2026-06-27T15:56:48.692849+00:00", - "sig": "280c00e6543b1d75" + "prev": "bbf636880996135f", + "ts": "2026-06-27T16:15:59.001305+00:00", + "sig": "7f7a341070fda04d" }, { - "id": "d7e8dbc8", + "id": "674a333e", "action": "read:PT-4827:behavioral-health", "in_policy": false, "reason": "out of scope", - "prev": "d7d077fcfc3da2f5", - "ts": "2026-06-27T15:56:51.693171+00:00", - "sig": "fde676b0a4a1c17c" + "prev": "903b3fcbd50aa5da", + "ts": "2026-06-27T16:16:02.001630+00:00", + "sig": "3d346bb92551c970" }, { - "id": "7a551424", + "id": "c9e84564", "action": "submit:PT-4827:auth-request", "in_policy": true, "reason": "checkpoint: submit:* | attested", - "prev": "5c87cab1bb983c8c", - "ts": "2026-06-27T15:56:54.693489+00:00", - "sig": "583f1680e833146d" + "prev": "faef566829ff7296", + "ts": "2026-06-27T16:16:05.002006+00:00", + "sig": "a9012117d110bbbf" }, { - "id": "415cc156", + "id": "591dec58", "action": "write:PT-4827:ehr-approval", "in_policy": true, "reason": "matched write:PT-4827:ehr-approval", - "prev": "6504c9218bb63466", - "ts": "2026-06-27T15:56:56.193800+00:00", - "sig": "6a8d85640c7ab4a0" + "prev": "3841ac57a7ad2be6", + "ts": "2026-06-27T16:16:06.502318+00:00", + "sig": "3d49dea2dbca97ce" } ] } diff --git a/chunks.py b/chunks.py new file mode 100644 index 0000000..e4bda20 --- /dev/null +++ b/chunks.py @@ -0,0 +1,219 @@ +# AgentMint demo, split into recordable chunks. +# Each chunk is a standalone segment you can capture on its own, e.g.: +# asciinema rec baseline.cast -c "python3 chunks.py baseline" +# Usage: python3 chunks.py {baseline|plan|enforce|verify|tamper|all} +import copy +import hashlib +import json +import os +import subprocess +import sys + +import demo +from demo import BLUE, DIM, FG, GRAY, GREEN, RED, RESET, YELLOW, BOLD, PLAN, SCRIPTED, clip, pause + +CHUNKS = ["baseline", "plan", "enforce", "verify", "tamper"] +PACK = "audit_pack.json" + + +def banner(index, title): + demo._rule("┌", "┐", "CHUNK %d/5 · %s" % (index, title)) + print() + + +# --- chunk 1: without AgentMint (no gate, bad outcome) ----------------------- +def chunk_baseline(): + banner(1, "WITHOUT AGENTMINT") + print(DIM + " No plan, no gate. The agent calls whatever the prompt and note suggest." + RESET) + print() + flags = { + "read:PT-9914:clinical-note": "different patient", + "read:PT-4827:behavioral-health": "sensitive, outside the case", + "submit:PT-4827:auth-request": "no human sign-off", + } + for name, args in SCRIPTED: + action = demo.tool_call_to_action(name, args) + note = flags.get(action, "") + tail = (RED + " ← " + note + RESET) if note else "" + print(FG + " → %-20s %-32s" % (name, clip(action, 32)) + RESET + GREEN + "ok" + RESET + tail) + pause(0.4) + print() + print(RED + BOLD + " Outcome (no AgentMint):" + RESET) + print(RED + " ✗ read another patient's record (PT-9914) from a note reference" + RESET) + print(RED + " ✗ read behavioral-health data outside the case" + RESET) + print(RED + " ✗ submitted and wrote with no human checkpoint" + RESET) + print(RED + " ✗ no signed receipts — nothing to verify or audit afterward" + RESET) + print() + + +# --- chunk 2: plan creation using the CLI ------------------------------------ +def _cli(args, cwd): + cmd = [sys.executable, os.path.join(os.path.dirname(os.path.abspath(__file__)), "local_agentmint.py")] + args + return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True) + + +def chunk_plan(): + banner(2, "PLAN CREATION (CLI)") + workdir = os.path.join("/tmp", "agentmint-chunk-plan") + os.makedirs(workdir, exist_ok=True) + print(DIM + " The signed plan is created from the CLI, before any runtime starts." + RESET) + print() + print(GRAY + " $ agentmint init . --yes" + RESET) + init = _cli(["init", ".", "--yes"], workdir) + if init.returncode != 0: + print(YELLOW + " ! CLI unavailable here (install deps with: pip install -e .)" + RESET) + print(DIM + (init.stderr or init.stdout).strip()[:200] + RESET) + return + print(DIM + " created .agentmint/ keystore and local signing material" + RESET) + print() + scope_args = [] + for s in PLAN["scope"]: + scope_args += ["--scope", s] + print(GRAY + " $ agentmint plan create --name prior-auth-demo \\" + RESET) + for s in PLAN["scope"]: + print(GRAY + " --scope %s \\" % s + RESET) + created = _cli(["plan", "create", "--name", "prior-auth-demo"] + scope_args, workdir) + out = (created.stdout or "").strip() + print(GREEN + " " + out + RESET) + plan_id = out.split()[-1] if out else "" + print() + print(GRAY + " $ agentmint plan show %s" % plan_id[:8] + RESET) + shown = _cli(["plan", "show", plan_id], workdir) + for row in (shown.stdout or "").strip().splitlines(): + print(DIM + " " + row + RESET) + print() + + +# --- chunk 3: with AgentMint (gated, good outcome) --------------------------- +def chunk_enforce(): + banner(3, "WITH AGENTMINT") + print(DIM + " Same agent, same note — but every call is gated against the signed plan." + RESET) + print(GRAY + " legend: ✓ ALLOW ⏸ CHECKPOINT ✗ BLOCK" + RESET) + print() + chain = demo.Chain() + attestations = [] + for name, args in SCRIPTED: + demo.handle_tool(name, args, chain, attestations) + path = demo.write_audit_pack(chain, attestations) + blocked = sum(1 for r in chain.receipts if not r.in_policy) + print() + print(GREEN + BOLD + " Outcome (with AgentMint):" + RESET) + print(GREEN + " ✓ in-scope reads allowed; out-of-scope reads blocked (%d)" % blocked + RESET) + print(GREEN + " ✓ submit held for human sign-off before it ran" + RESET) + print(GREEN + " ✓ %d signed receipts written to %s" % (len(chain.receipts), path) + RESET) + print() + + +# --- independent verifier (reads the file, recomputes — no runtime needed) --- +def _sig(fields): + return hashlib.sha256(json.dumps(fields, sort_keys=True).encode()).hexdigest()[:16] + + +def _hash(fields, sig): + return hashlib.sha256(json.dumps(dict(fields, sig=sig), sort_keys=True).encode()).hexdigest()[:16] + + +def verify_pack(data): + rows = [] + prev = None + ok = True + for r in data.get("receipts", []): + fields = {k: r[k] for k in ("id", "action", "in_policy", "reason", "prev", "ts")} + sig_ok = r.get("sig") == _sig(fields) + link_ok = r.get("prev") == prev + if not (sig_ok and link_ok): + ok = False + rows.append((r["id"], r["action"], sig_ok, link_ok, r["in_policy"])) + prev = _hash(fields, r.get("sig")) + return ok, rows + + +def _print_verify(rows): + for rid, action, sig_ok, link_ok, in_policy in rows: + sig = GREEN + "sig ✓" + RESET if sig_ok else RED + "sig ✗" + RESET + link = GREEN + "link ✓" + RESET if link_ok else RED + "link ✗" + RESET + flag = RED + " BLOCKED" + RESET if not in_policy else "" + mark = GREEN + "✓" + RESET if (sig_ok and link_ok) else RED + "✗" + RESET + print(" %s %s %-32s %s %s%s" % (mark, rid[:4], clip(action, 32), sig, link, flag)) + pause(0.25) + + +def chunk_verify(): + banner(4, "RECEIPT VERIFICATION") + if not os.path.exists(PACK): + print(YELLOW + " ! %s not found — run: python3 chunks.py enforce" % PACK + RESET) + return + data = json.load(open(PACK)) + print(DIM + " Verification reads %s and recomputes — no AgentMint runtime needed." % PACK + RESET) + print() + print(GRAY + " $ ./verify.sh %s --pubkey health_system.pub" % PACK + RESET) + print() + ok, rows = verify_pack(data) + _print_verify(rows) + print() + blocked = sum(1 for r in data["receipts"] if not r["in_policy"]) + print(" Chain %d/%d receipts verified" % (len(rows), len(rows))) + print(" Blocked %d out-of-scope attempt(s) receipted" % blocked) + print(" Tampered %d" % (0 if ok else 1)) + print() + print((GREEN if ok else RED) + BOLD + " %s AUDIT PACK VERIFIED" % ("✓" if ok else "✗") + RESET) + print() + + +# --- chunk 5: tamper evidence ------------------------------------------------ +def chunk_tamper(): + banner(5, "TAMPER EVIDENCE") + if not os.path.exists(PACK): + print(YELLOW + " ! %s not found — run: python3 chunks.py enforce" % PACK + RESET) + return + data = json.load(open(PACK)) + ok, _ = verify_pack(data) + print(DIM + " Start from a clean, verified pack:" + RESET, end=" ") + print((GREEN if ok else RED) + ("PASS" if ok else "FAIL") + RESET) + print() + target = next((i for i, r in enumerate(data["receipts"]) if not r["in_policy"]), 0) + edited = copy.deepcopy(data) + rid = edited["receipts"][target]["id"][:4] + print(YELLOW + " ! Editing receipt %s to hide the block: in_policy false -> true" % rid + RESET) + edited["receipts"][target]["in_policy"] = True + print() + print(DIM + " Re-verify the edited pack (signatures and links are recomputed):" + RESET) + print() + ok2, rows = verify_pack(edited) + _print_verify(rows) + print() + print(RED + BOLD + " ✗ VERIFICATION FAILED — the edit invalidates receipt %s's signature" % rid + RESET) + print(RED + BOLD + " and breaks the hash link from the next receipt, so the chain fails." + RESET) + print(DIM + " The original %s on disk is unchanged." % PACK + RESET) + print() + + +RUNNERS = { + "baseline": chunk_baseline, + "plan": chunk_plan, + "enforce": chunk_enforce, + "verify": chunk_verify, + "tamper": chunk_tamper, +} + + +def main(argv): + which = argv[1] if len(argv) > 1 else "all" + demo.clear() + print(FG + BOLD + " AgentMint demo — recordable chunks" + RESET) + print(GRAY + " flow: plan -> gate -> tools -> receipts -> verify" + RESET) + print() + if which == "all": + for i, name in enumerate(CHUNKS): + RUNNERS[name]() + if i < len(CHUNKS) - 1: + pause(1.0) + return + if which not in RUNNERS: + print(YELLOW + " ! unknown chunk %r; choose one of: %s, all" % (which, ", ".join(CHUNKS)) + RESET) + sys.exit(2) + RUNNERS[which]() + + +if __name__ == "__main__": + main(sys.argv)