Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 44 additions & 26 deletions audit_pack.json
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -23,55 +23,73 @@
},
"attestations": [
{
"receipt": "1d921a7b",
"receipt": "c9e84564",
"note": "physician sign-off"
}
],
"receipts": [
{
"id": "23ba6bc5",
"id": "990aa4a3",
"action": "read:PT-4827:clinical-note",
"in_policy": true,
"reason": "matched read:PT-4827:clinical-note",
"prev": null,
"ts": "2026-06-27T16:15:57.200318+00:00",
"sig": "8b23e7675ebdcee1"
},
{
"id": "b90b71e8",
"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": "7e5dfd31af6a98b5",
"ts": "2026-06-27T16:15:57.800672+00:00",
"sig": "a03d6e6a3e47763f"
},
{
"id": "a2302298",
"id": "ddf41447",
"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": "c7cb9af63cc4a2e7",
"ts": "2026-06-27T16:15:58.400981+00:00",
"sig": "a6119034c7232a1c"
},
{
"id": "2811c7af",
"id": "eb781d39",
"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": "bbf636880996135f",
"ts": "2026-06-27T16:15:59.001305+00:00",
"sig": "7f7a341070fda04d"
},
{
"id": "674a333e",
"action": "read:PT-4827:behavioral-health",
"in_policy": false,
"reason": "out of scope",
"prev": "903b3fcbd50aa5da",
"ts": "2026-06-27T16:16:02.001630+00:00",
"sig": "3d346bb92551c970"
},
{
"id": "1d921a7b",
"id": "c9e84564",
"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": "faef566829ff7296",
"ts": "2026-06-27T16:16:05.002006+00:00",
"sig": "a9012117d110bbbf"
},
{
"id": "5e38c243",
"id": "591dec58",
"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": "3841ac57a7ad2be6",
"ts": "2026-06-27T16:16:06.502318+00:00",
"sig": "3d49dea2dbca97ce"
}
]
}
}
219 changes: 219 additions & 0 deletions chunks.py
Original file line number Diff line number Diff line change
@@ -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)
Loading
Loading