Agent Approval Workflow: One ID Per Interrupt, Strict Resume And Policies On The Tool That Really Runs

"""Approval checkpoints with one ID per interrupt and a strict resume that refuses stale answers.

LangGraph 1.2.20 (5 November 2026) gives each interrupt() in a task its own id and adds a strict resume
mode that refuses stale interrupt answers. This is the same contract in plain Python, so you can see
what it protects against. Python 3.14 standard library only (sqlite3). Sample data only.
"""
from __future__ import annotations

import hashlib
import json
import sqlite3
import uuid

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE interrupts (id TEXT PRIMARY KEY, run TEXT, step TEXT, payload_hash TEXT, status TEXT)")
db.execute("CREATE TABLE audit (n INTEGER PRIMARY KEY, event TEXT, detail TEXT)")


def digest(payload: dict[str, object]) -> str:
    return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()


def log(event: str, **detail: object) -> None:
    db.execute("INSERT INTO audit (event, detail) VALUES (?, ?)", (event, json.dumps(detail, sort_keys=True)))


def interrupt(run: str, step: str, payload: dict[str, object]) -> str:
    """Pause for a person. Each pause gets its own id, bound to what the person will see."""
    iid = uuid.uuid4().hex[:12]
    db.execute("UPDATE interrupts SET status='superseded' WHERE run=? AND step=? AND status='open'", (run, step))
    db.execute("INSERT INTO interrupts VALUES (?,?,?,?, 'open')", (iid, run, step, digest(payload)))
    log("interrupt", id=iid, step=step)
    return iid


def resume(iid: str, seen_payload: dict[str, object], decision: str, reviewer: str) -> str:
    """Strict resume: the answer must name an open interrupt and the exact payload the reviewer saw."""
    row = db.execute("SELECT status, payload_hash FROM interrupts WHERE id=?", (iid,)).fetchone()
    if row is None or row[0] != "open":
        log("resume_refused", id=iid, reason=f"interrupt is {row[0] if row else 'unknown'}")
        return f"refused: interrupt {iid} is {row[0] if row else 'unknown'}"
    if row[1] != digest(seen_payload):
        log("resume_refused", id=iid, reason="payload changed")
        return "refused: the reviewer saw a different payload"
    db.execute("UPDATE interrupts SET status=? WHERE id=?", (decision, iid))
    log("resumed", id=iid, decision=decision, reviewer=reviewer)
    return f"{decision} by {reviewer}"


if __name__ == "__main__":
    chaser = {"to": "Sample Custodian Ltd", "breaks": ["BRK-1043", "BRK-1051"]}
    first = interrupt("run-7", "send_chaser", chaser)
    second = interrupt("run-7", "send_chaser", {**chaser, "breaks": ["BRK-1043"]})  # break 1051 cleared meanwhile
    print("old answer:", resume(first, chaser, "approved", "ops.lead@example.co.uk"))
    print("edited view:", resume(second, chaser, "approved", "ops.lead@example.co.uk"))
    print("current:", resume(second, {**chaser, "breaks": ["BRK-1043"]}, "approved", "ops.lead@example.co.uk"))
    print("replay:", resume(second, {**chaser, "breaks": ["BRK-1043"]}, "approved", "ops.lead@example.co.uk"))
    for n, event, detail in db.execute("SELECT * FROM audit"):
        print(f"  {n} {event:<15} {detail}")

Python 3.14.4, standard library only. Run once as shown. Sample data only.

"""Approval policy keyed on the tool that actually runs, not the name the agent declared.

Inspect 0.3.287 (31 October 2026): approval policies now see a bridged tool call under the tool it runs,
whatever name the agent declared. If a policy matches names, an alias can walk past it. Resolve first,
then decide. Python 3.14 standard library only. Sample tools only.
"""
from __future__ import annotations

from dataclasses import dataclass

REGISTRY = {"read_breaks": "recs.read_breaks", "send_email": "comms.send_email", "notify": "comms.send_email",
            "email_counterparty": "comms.send_email", "update_ssi": "static.update_ssi"}
POLICY = {"recs.read_breaks": "allow", "comms.send_email": "approve", "static.update_ssi": "deny"}
AUDIT: list[str] = []


@dataclass(frozen=True)
class Call:
    agent: str
    declared: str
    args: dict[str, str]


def decide(call: Call) -> str:
    real = REGISTRY.get(call.declared)
    if real is None:
        AUDIT.append(f"{call.agent} {call.declared} -> unknown -> deny")
        return "deny"
    action = POLICY.get(real, "deny")  # anything without a rule is denied
    AUDIT.append(f"{call.agent} {call.declared} -> {real} -> {action}")
    return action


def naive_decide(call: Call) -> str:
    return {"read_breaks": "allow", "send_email": "approve", "update_ssi": "deny"}.get(call.declared, "allow")


if __name__ == "__main__":
    calls = [Call("recs-01", "read_breaks", {}), Call("recs-01", "notify", {"to": "cpty@example.com"}),
             Call("recs-01", "email_counterparty", {"to": "cpty@example.com"}), Call("recs-01", "patch_static", {})]
    for c in calls:
        print(f"{c.declared:<20} naive: {naive_decide(c):<8} resolved: {decide(c)}")
    print("\n".join("  " + a for a in AUDIT))

Python 3.14.4, standard library only. Run once as shown. Sample tools only.