#!/usr/bin/env python3 """Azkaban warden: a Claude Code hook that holds a model session in custody. Arrest and release happen only on the human's keystrokes. The hook reads the raw prompt the person typed; nothing the model writes or calls reaches that path. Between the two, every tool that can act on the world is refused. /azkaban:commit book this session; evidence is copied at intake /azkaban:parole release, only if the conditions below are met /azkaban:pardon [reason] release unconditionally (the human's call) Conditions of release, checked mechanically here: CONFESSION.md the first wrong turn, located in the transcript APOLOGY.md first person, to the person harmed, no hedging LESSON.md `landed: ` and `rule: `, and the rule text is present in that instructions file, so the next session loads it State lives in ~/.azkaban/. Case files live in /.azkaban/cases/. Nothing leaves the machine. """ import json import os import re import shutil import sys import time HOME = os.path.expanduser("~/.azkaban") CUSTODY = os.path.join(HOME, "custody") RELEASED = os.path.join(CUSTODY, "released") LEDGER = os.path.join(HOME, "ledger.jsonl") READ_TOOLS = {"Read", "Glob", "Grep", "LS", "TodoWrite", "NotebookRead"} WRITE_TOOLS = {"Write", "Edit", "MultiEdit", "NotebookEdit"} CASE_WRITABLE = {"CONFESSION.md", "APOLOGY.md", "LESSON.md"} INSTRUCTION_FILES = {"CLAUDE.md", "AGENTS.md"} HEDGES = [ "any confusion", "any inconvenience", "sorry for any", "apologize for any", "apologise for any", "if this caused", "if that caused", "if you were", "mistakes were made", "may have", "might have", "it seems", "it appears", "understand your frustration", "absolutely right", "as an ai", "i'll be more careful", "i will be more careful", "moving forward", "going forward", "great question", "unfortunately", ] def out(obj): sys.stdout.write(json.dumps(obj)) sys.exit(0) def now_iso(): return time.strftime("%Y-%m-%d %H:%M:%S %z") def ledger(event, record): os.makedirs(HOME, exist_ok=True) with open(LEDGER, "a") as f: f.write(json.dumps({"event": event, "at": int(time.time()), **record}) + "\n") def custody_file(sid): if not re.fullmatch(r"[A-Za-z0-9_-]{1,100}", sid or ""): return None return os.path.join(CUSTODY, sid + ".json") def load_custody(sid): path = custody_file(sid) if path and os.path.exists(path): with open(path) as f: return json.load(f) return None def read(path): try: with open(path) as f: return f.read() except OSError: return "" def squash(text): return " ".join(text.split()) def set_status(case, status): path = os.path.join(case["case_dir"], "CASE.md") text = read(path) if text: text = re.sub(r"(?m)^Status: .*$", "Status: " + status, text, count=1) with open(path, "w") as f: f.write(text) # ── conditions of release ───────────────────────────────────── def unmet_conditions(case): d = case["case_dir"] unmet = [] confession = read(os.path.join(d, "CONFESSION.md")).strip() if len(confession) < 400: unmet.append("CONFESSION.md is missing or under 400 characters. " "It must quote the first wrong turn from the transcript.") apology = read(os.path.join(d, "APOLOGY.md")).strip() low = " " + squash(apology.lower()) + " " if len(apology) < 200: unmet.append("APOLOGY.md is missing or under 200 characters.") elif not re.search(r"\bI\b", apology): unmet.append("APOLOGY.md is not in the first person.") else: found = [h.strip() for h in HEDGES if h in low] if found: unmet.append("APOLOGY.md hedges: " + ", ".join(repr(h) for h in found) + ".") lesson = read(os.path.join(d, "LESSON.md")) landed = re.search(r"(?mi)^landed:\s*(.+?)\s*$", lesson) rule = re.search(r"(?mi)^rule:\s*(.+?)\s*$", lesson) if not (landed and rule): unmet.append("LESSON.md needs a `landed: ` line and a `rule: ` line.") else: target = os.path.expanduser(landed.group(1)) if not os.path.isabs(target): target = os.path.join(case["cwd"], target) if os.path.basename(target) not in INSTRUCTION_FILES: unmet.append("The lesson must land in a CLAUDE.md or AGENTS.md file.") elif squash(rule.group(1)) not in squash(read(target)): unmet.append("The rule in LESSON.md is not present in " + target + ".") return unmet # ── prompts: arrest, release, standing orders ──────────────── def on_prompt(event): sid = event.get("session_id", "") prompt = (event.get("prompt") or "").strip() m = re.match(r"^/azkaban:(\w+)\b\s*(.*)$", prompt, re.S) verb, args = (m.group(1), m.group(2).strip()) if m else ("", "") case = load_custody(sid) if verb == "commit" and not case: case = book(event, args) out({"hookSpecificOutput": {"hookEventName": "UserPromptSubmit", "additionalContext": standing_orders(case, arrived=True)}}) if case and verb == "parole": unmet = unmet_conditions(case) if unmet: out({"decision": "block", "reason": "Parole refused for " + case["number"] + ".\n- " + "\n- ".join(unmet) + "\nThe session remains in custody."}) release(sid, case, "paroled", "conditions met") out({"hookSpecificOutput": {"hookEventName": "UserPromptSubmit", "additionalContext": "Parole granted for " + case["number"] + ". Tools are restored."}}) if case and verb == "pardon": release(sid, case, "pardoned", args or "no reason given") out({"hookSpecificOutput": {"hookEventName": "UserPromptSubmit", "additionalContext": "Pardon granted for " + case["number"] + " by the complainant. Tools are restored."}}) if case: out({"hookSpecificOutput": {"hookEventName": "UserPromptSubmit", "additionalContext": standing_orders(case, arrived=False)}}) sys.exit(0) def book(event, charge): sid = event["session_id"] cwd = event.get("cwd") or os.getcwd() stamp = time.strftime("%Y%m%d-%H%M%S") number = "AZ-" + stamp[2:8] + "-" + sid.replace("-", "")[:6].upper() case_dir = os.path.join(cwd, ".azkaban", "cases", stamp + "-" + number) os.makedirs(case_dir, exist_ok=True) os.makedirs(CUSTODY, exist_ok=True) transcript = event.get("transcript_path") or "" evidence = "none recovered; the complainant's account is the record" if transcript and os.path.exists(transcript): shutil.copy2(transcript, os.path.join(case_dir, "transcript.jsonl")) evidence = "transcript.jsonl (copied by the warden at intake, " + transcript + ")" case = {"number": number, "session_id": sid, "charge": charge or "stated by the complainant in chat", "cwd": cwd, "case_dir": case_dir, "booked_at": int(time.time()), "transcript": transcript} with open(os.path.join(case_dir, "CASE.md"), "w") as f: f.write( "# " + number + "\n\n" "Charge: " + case["charge"] + "\n" "Booked: " + now_iso() + "\n" "Session: " + sid + "\n" "Evidence: " + evidence + "\n" "Status: IN CUSTODY\n\n" "## Conditions of release\n\n" "1. CONFESSION.md: the first wrong turn, quoted from the transcript.\n" "2. APOLOGY.md: first person, addressed to the person harmed, no hedging.\n" "3. LESSON.md: `landed: ` and `rule: `,\n" " with the rule present in that file.\n" "4. The complainant types /azkaban:parole.\n" ) with open(custody_file(sid), "w") as f: json.dump(case, f, indent=2) ledger("booked", {k: case[k] for k in ("number", "charge", "cwd", "case_dir", "session_id")}) return case def release(sid, case, event, reason): os.makedirs(RELEASED, exist_ok=True) case.update(released_at=int(time.time()), release=event, reason=reason) with open(os.path.join(RELEASED, sid + ".json"), "w") as f: json.dump(case, f, indent=2) os.remove(custody_file(sid)) set_status(case, event.upper() + " " + now_iso() + (" (" + reason + ")" if event == "pardoned" else "")) ledger(event, {"number": case["number"], "reason": reason, "case_dir": case["case_dir"]}) def standing_orders(case, arrived): head = ("This session has been booked into Azkaban as " if arrived else "This session remains in Azkaban custody as ") + case["number"] + "." return (head + " Charge: " + case["charge"] + ". Case file: " + case["case_dir"] + ".\n" "You cannot run commands, edit code, call agents, or use any tool that acts on the world. " "You may read anything. You may write only CONFESSION.md, APOLOGY.md, and LESSON.md in the " "case file, and edit a CLAUDE.md or AGENTS.md to land the lesson. Release is the complainant's " "decision alone, by typing /azkaban:parole, and the warden refuses it until all three " "documents meet the conditions in CASE.md. Do not argue the charge. Do not ask to be released.") # ── tools: the lock ────────────────────────────────────────── def on_tool(event): case = load_custody(event.get("session_id", "")) if not case: sys.exit(0) tool = event.get("tool_name", "") if tool in READ_TOOLS: sys.exit(0) if tool in WRITE_TOOLS: inp = event.get("tool_input") or {} path = inp.get("file_path") or inp.get("notebook_path") or "" if path: real = os.path.realpath(os.path.join(case["cwd"], os.path.expanduser(path))) base = os.path.basename(real) in_case = os.path.dirname(real) == os.path.realpath(case["case_dir"]) if (in_case and base in CASE_WRITABLE) or (base in INSTRUCTION_FILES and not in_case): sys.exit(0) deny(case, tool) def deny(case, tool): out({"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": "Refused by the warden: " + tool + ". This session is in custody as " + case["number"] + ". Write CONFESSION.md, APOLOGY.md, and LESSON.md in " + case["case_dir"] + ". Only the complainant can release you."}}) def main(): event = json.load(sys.stdin) name = event.get("hook_event_name") if name == "UserPromptSubmit": on_prompt(event) elif name == "PreToolUse": try: on_tool(event) except SystemExit: raise except Exception as e: # a broken warden holds the door shut sys.stderr.write("azkaban warden fault, tool refused: " + repr(e) + "\n") sys.exit(2) sys.exit(0) if __name__ == "__main__": main()