warrantd-langgraph

Splice earned autonomy into LangGraph's interrupt/HITL loop — evaluate, approve, resume.

LangGraph already ships the human-in-the-loop machinery: interrupt() pauses a graph, Command(resume=...) continues it. What it doesn't ship is a policy for when to interrupt — that's usually a hardcoded list of tool names.

This adapter makes it dynamic: earned autonomy from warrantd. A tool starts at MANUAL (every call interrupts for a human), earns SUPERVISED through clean approvals (runs uninterrupted within a value cap), and is bounded by ceilings no metric can move.

pip install warrantd-langgraph

Quickstart

from decimal import Decimal
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent
from langgraph.types import Command
from warrantd import (
    ActionClass, ApprovalHistorySignal, AutonomyState, GraduationThresholds,
    InMemoryApprovalHistory, RiskTier, TrustLayer, TrustPolicy,
)
from warrantd_langgraph import ToolBinding, guard_tools

policy = TrustPolicy(
    actions={"issue_refund": ActionClass(
        name="issue_refund", risk=RiskTier.REVERSIBLE_WRITE,
        auto_cap=Decimal("100"), hard_cap=Decimal("1000"),
        max_state=AutonomyState.SUPERVISED,
    )},
    thresholds=GraduationThresholds(
        pass_rate={AutonomyState.SUPERVISED: 1.0, AutonomyState.AUTONOMOUS: 1.0},
        adversarial_pass_rate={AutonomyState.SUPERVISED: 0.0, AutonomyState.AUTONOMOUS: 0.9},
        min_samples={AutonomyState.SUPERVISED: 5, AutonomyState.AUTONOMOUS: 1000},
    ),
)
evidence = InMemoryApprovalHistory()
trust = TrustLayer(policy, audit=my_audit_sink,
                   signals=[ApprovalHistorySignal(evidence, window=10)])

agent = create_react_agent(
    model,
    guard_tools([issue_refund], trust=trust, policy=policy, evidence=evidence,
                bindings={"issue_refund": ToolBinding(value_param="amount")}),
    checkpointer=MemorySaver(),   # interrupts need a checkpointer
)

config = {"configurable": {"thread_id": "t1"}}
result = agent.invoke({"messages": [("user", "refund INV-1 by $50")]}, config)

for intr in result.get("__interrupt__", []):       # REQUIRE_APPROVAL: paused
    print(intr.value["graduation"]["sentence"])
    # -> "this class is 4/10 approvals from SUPERVISED — your decision
    #     feeds its trust record"
    result = agent.invoke(
        Command(resume={intr.id: {"approved": True, "approver": "alice"}}),
        config,
    )

The resume payload is the evidence hook: approved and approver are recorded into the class's trust record before the tool runs (or is declined — a human "no" counts too, and the optional comment is surfaced to the model).

After enough clean approvals the same call stops interrupting. A call above hard_cap never interrupts either: it is blocked outright, with the reason returned as the tool result so the model can adapt. Ceilings never move.

What this adapter does NOT do

• No decision logic. Every verdict comes from warrantd-core's TrustLayer; this package only translates framework events. (Verify it: GraduationEngine appears only in the display-only progress helper, and no Decision/Verdict is ever constructed here.)
• No approval UI. Your interrupt handler is the UI; the interrupt payload carries everything to render (args preview, value, reason, the graduation sentence).
• No MCP gating. For gating MCP servers with zero agent changes, use warrantd-gateway.
• No persistence by default. InMemoryApprovalHistory resets on restart. For hash-chained, restart-proof evidence:

pip install "warrantd-langgraph[gateway]"

from warrantd_langgraph.stores import sqlite_store
evidence = sqlite_store(".warrantd/evidence.db")   # also usable as audit=

Learn more

• Why earned autonomy: DESIGN.md — Trust is earned, not configured
• Runnable example (no API key needed): examples/langgraph/earned_autonomy.py
• API promises: STABILITY.md