Build approve, edit, reject and timeout paths with LangGraph interrupts¶

Draft outline: not validated

Drafted with AI from official sources on 16 Oct 2026; updated on 22 Oct 2026 for task AGT-05-T01. Do not use on a live system until it is written, dry-run by someone other than the author and signed off (planned 2026-11-24). Author: Sanjay Sarvesh · dry-run: Yasar · sign-off: James. Work items: AGT-05-T01, AGT-05-T02, AGT-05-T03, AGT-05-T04.

1. Summary¶

This how-to shows how to build a human-in-the-loop (HITL) approval pattern using LangGraph interrupts and checkpointers. It provides four distinct resolution paths—approve, edit, reject, and timeout—with state isolation, schema re-validation, and an immutable audit trail. Implementing these controls ensures sensitive agent actions never execute without valid authorization or produce duplicate side effects upon resumption.

Risk: High (side effects before the interrupt can run twice). Typical duration: 1-2 days.

2. When to use and when not to¶

Use it when:

An agent proposes a high-risk or irreversible action (such as financial transactions, resource deletion, external communications, or permission grants).
Approvers need the ability to approve without changes, edit arguments before execution, reject the request, or handle approval timeouts.
Execution must pause durably across process boundaries, server restarts, or worker scaling while waiting for human review.

Do not use it when:

Actions are low-risk or read-only (such as information retrieval, web browsing, or text formatting); human gates add unnecessary latency to these tasks.
The system runs in an unattended batch processing pipeline where no human is assigned to review decisions (use deterministic policy checks or guardrails instead).
Approvals must be delivered interactively through chat channels (Teams, Slack) or email with async webhook callbacks: use PB-AGT-05-02.
You are triaging an operational backlog of queued or expired approvals during an incident: use PB-AGT-05-03.
You are recovering or replaying paused threads after a breaking graph schema deployment: use PB-AGT-05-04.

3. What you must know first¶

interrupt(value) pauses node execution dynamically and surfaces a JSON-serialisable payload to the caller (LangGraph: Interrupts).
Graphs using interrupt() require a persistent checkpointer; compiling without a checkpointer causes runtime failures on interrupt (LangGraph: Interrupts).
Command(resume=...) resumes an interrupted thread when supplied with the identical thread_id (LangGraph: Interrupts).
On resume, LangGraph re-runs the node from its first line; any code placed before interrupt() will execute again (LangGraph: Interrupts).
Never wrap interrupt() in a broad try/except Exception: block, as this catches the internal GraphInterrupt exception and prevents graph pausing (LangGraph: Interrupts).
LangGraph does not include built-in wall-clock timers inside interrupt(); timeouts must be monitored and triggered by an external scheduler or queue monitor.
LangChain HumanInTheLoopMiddleware provides high-level tool approval with approve, edit, reject, and respond choices (LangChain: Human-in-the-loop middleware).

4. Prerequisites and access¶

Python 3.10 or higher with langgraph at the template lock-file version, and no lower than 0.2.47 (first release with interrupt()), and langchain-core>=0.3.0.
Template repository cloned.
Durable checkpointer configured for multi-process environments (such as PostgresSaver via langgraph-checkpoint-postgres or Azure Cosmos DB checkpointer). InMemorySaver is suitable only for sandbox testing.
Defined catalog of sensitive actions with risk classifications and parameter schemas agreed with the risk team.
Append-only audit store (such as PostgreSQL audit table, Azure Cosmos DB, or Azure Monitor Log Analytics) accessible via service credentials.
Test runner and access to the sandbox environment.

5. Steps¶

Follow these steps to build and verify the four approval paths:

Define typed state and approval schemas. Create strict data contracts for the graph state, the approval request payload, and the resume decision payload (approve, edit, reject, timeout). Check: Verify schemas reject invalid decision types and missing fields with Pydantic validation errors.
Isolate the review gate in a dedicated node without pre-interrupt side effects. Create an approval node that only constructs the interrupt payload and calls interrupt(). Do not call external APIs, send emails, or write database state prior to interrupt(). Check: Confirm the review node has no network or database calls prior to interrupt().
Emit the structured approval request payload via interrupt(). Pass an immutable payload containing the action name, proposed parameters, requester context, risk tier, and an expiry timestamp to interrupt(). Check: Confirm graph.get_state(config).tasks[0].interrupts returns the expected dictionary payload when paused.
Implement the four resume decision paths in the review node. Inspect the resumed value returned by interrupt():
approve: Update status to approved and transition to the execution node.
edit: Validate modified parameters against the tool schema before updating state, then route to execution.
reject: Update status to rejected and terminate the flow at END.
timeout: Update status to timed_out and route to END or a remediation node. Check: Unit test each decision branch with sample payloads and verify the returned Command targets the correct destination.
Execute sensitive actions in a downstream node using idempotency keys. !!! warning "Do not execute actions in the review node" Executing actions inside the review node causes duplicate side effects if the node re-runs or is resumed twice. Always execute side effects in a dedicated downstream node with an idempotency key. Generate a deterministic idempotency key ${THREAD_ID}:${ACTION_NAME} and pass it to the target service. Check: Confirm the execution node executes exactly once when status == "approved" or status == "edited".
Log immutable audit records for the request and decision. Write the request record in a separate node that runs before the review node (for example record_request), or make the write idempotent on thread_id plus the interrupt ID so a re-run does not create a second row. Write the decision record after interrupt() returns. Check: Resume a thread twice in test and confirm exactly one request record and one decision record per approval.
Implement the external timeout evaluator. Configure an external scheduler or queue consumer that periodically queries paused threads past their expiry SLA and resumes them with Command(resume={"type": "timeout"}). Check: Trigger the timeout worker against a test thread and confirm the graph transitions to timed_out.
Verify recovery across process restarts. Run the graph up to the interrupt point, terminate the host Python process, re-initialize the graph against the persistent checkpointer, and resume execution. Check: Confirm the graph successfully resumes from the persisted state without losing thread history.

Draft snippet: Approve, edit, reject and timeout graph¶

from typing import Literal, TypedDict
from pydantic import BaseModel, Field
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


# Schemas for parameter validation
class PaymentParams(BaseModel):
    payee: str
    amount: float = Field(gt=0, le=10000)


class PayState(TypedDict):
    payee: str
    amount: float
    status: str
    execution_result: str


def record_request(state: PayState) -> dict:
    # Log approval request before entering review; prevents duplicate audit rows on resume
    return {}


def human_review(state: PayState) -> Command[Literal["execute_action", "__end__"]]:
    # Rule: Zero side effects before interrupt().
    # On resume, this function restarts from line 1.
    payload = {
        "action": "execute_payment",
        "payee": state["payee"],
        "amount": state["amount"],
        "description": "Approve outgoing vendor payment",
    }

    # Pause execution and return payload to client
    decision = interrupt(payload)

    # Write the decision audit record after interrupt() returns
    decision_type = decision.get("type")

    if decision_type == "approve":
        return Command(
            goto="execute_action",
            update={"status": "approved"}
        )

    if decision_type == "edit":
        # Re-validate edited values against the schema before use
        edited_amount = decision.get("amount")
        validated = PaymentParams(payee=state["payee"], amount=float(edited_amount))
        return Command(
            goto="execute_action",
            update={
                "amount": validated.amount,
                "status": "edited",
            }
        )

    if decision_type == "timeout":
        return Command(
            goto=END,
            update={"status": "timed_out"}
        )

    # Default to reject for unknown or explicit rejection types
    return Command(
        goto=END,
        update={"status": "rejected"}
    )


def execute_action(state: PayState) -> dict:
    # Downstream execution node with idempotency
    idempotency_key = f"{state['payee']}-{state['amount']}"
    return {
        "status": "completed",
        "execution_result": f"Paid {state['amount']} to {state['payee']} (Key: {idempotency_key})",
    }


# Assemble graph
builder = StateGraph(PayState)
builder.add_node("record_request", record_request)
builder.add_node("human_review", human_review)
builder.add_node("execute_action", execute_action)
builder.add_edge(START, "record_request")
builder.add_edge("record_request", "human_review")
builder.add_edge("execute_action", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# Execution test
cfg = {"configurable": {"thread_id": "payment-tx-101"}}
graph.invoke({"payee": "Contoso Ltd", "amount": 250.0, "status": "pending", "execution_result": ""}, cfg)

# Resume with an edit
resumed = graph.invoke(Command(resume={"type": "edit", "amount": 200.0}), cfg)
print("Resumed status:", resumed["status"])
print("Result:", resumed["execution_result"])

Snippets to add:

snippets/templates/hitl/approval_graph.py: Complete production graph with PostgreSQL checkpointer and Pydantic validation (python)
snippets/tests/test_approval_paths.py: Pytest suite exercising approve, edit, reject, and timeout paths plus process restart (python)

6. Verify it works¶

To verify the implementation, execute the test suite against the sandbox environment:

Test Approve Path: Invoke the graph with {"payee": "Contoso Ltd", "amount": 100.0}. Confirm execution pauses at human_review. Resume with Command(resume={"type": "approve"}). Confirm status is completed and execution_result is populated.
Test Edit Path: Invoke the graph. Resume with Command(resume={"type": "edit", "amount": 75.0}). Confirm amount is updated to 75.0 and action executes with edited parameters.
Test Edit Validation Failure: Resume with Command(resume={"type": "edit", "amount": -50.0}). Confirm Pydantic validation fails and does not execute the action.
Test Reject Path: Invoke the graph. Resume with Command(resume={"type": "reject"}). Confirm status is rejected and execute_action never runs.
Test Timeout Path: Invoke the graph. Resume with Command(resume={"type": "timeout"}). Confirm status is timed_out and no payment executes.
Test Process Restart Recovery: Invoke the graph with a durable checkpointer. Terminate the process. Start a new process and resume using the original thread_id. Confirm the thread completes normally.
Test Audit Trail Idempotency: Resume a thread twice in test and inspect the audit table. Confirm exactly one request record and one decision record exist per approval.

7. Known issues and fixes¶

Symptom	Likely cause	Fix
Side effect runs twice (e.g., duplicate email or payment)	Side effect placed before `interrupt()` re-executed when node resumed	Move all side effects into a downstream execution node or behind idempotency checks
Duplicate audit record created on resume	Request audit record written inside the review node before `interrupt()`	Write the request record in a separate pre-review node (such as `record_request`) or enforce idempotency on `thread_id` and interrupt ID
Graph never pauses on `interrupt()`	Broad `try/except Exception:` caught `GraphInterrupt`	Remove bare `except Exception:` around `interrupt()` or explicitly re-raise `GraphInterrupt`
Resume starts a new conversation instead of resuming	Different or omitted `thread_id` supplied in `configurable`	Persist `thread_id` with the approval request and reuse the exact same `thread_id` on resume
Edited payload bypasses business limits or safety constraints	Modified parameters merged into state without re-validation	Run edited values through Pydantic/JSON schema validation before returning `Command`
Graph raises runtime error on `interrupt()`	Graph compiled without a checkpointer	Pass a checkpointer instance (such as `PostgresSaver`) to `builder.compile(checkpointer=...)`
`TypeError: Object not JSON serializable` during interrupt	Non-primitive objects passed into `interrupt()` payload	Ensure payload contains only primitive dictionaries, strings, numbers, or serialized Pydantic models
Threads remain paused indefinitely in checkpointer	External timeout worker not running or missed thread SLA	Implement scheduled monitoring worker to scan paused threads and send timeout resume commands

8. Monitoring and alerts¶

Signal	Alert when	Action
Sensitive action executed without an approval record	Greater than 0 in any window	Security on-call investigates immediate bypass breach
Threads stuck in interrupted state past SLA	Any thread interrupted past its approval expiry	Trigger timeout worker or escalate to approver queue manager
Checkpoint serialisation or deserialisation failures	Greater than 0 in 15 minutes	Inspect payload for non-JSON serialisable types
Validation failure rate on edited arguments	Above the agreed baseline (threshold TO VERIFY)	Inspect approver UI validation logic for malformed inputs
Resumes attempted on non-existent or completed threads	Any in 15 minutes	Verify webhook callback idempotency and deduplicate resume events
Interrupt-related errors in logs	Any in 15 minutes	Check thread_id handling and resume payloads

9. AWS and open-source equivalents¶

OpenAI Agents SDK: Supports @function_tool(needs_approval=True). The agent pauses on sensitive tool calls and produces result.interruptions. Convert with state = result.to_state() (serialisable RunState), call state.approve(interruption) or state.reject(interruption) for each item, and resume with Runner.run(agent, state) (OpenAI Agents SDK: Human-in-the-loop).
Microsoft Agent Framework (preview): Workflows model steps as Executor classes. Pauses are initiated with await ctx.request_info(request_data=..., response_type=...) (a RequestPort in .NET). The workflow suspends into a durable checkpoint, emits a request_info event (RequestInfoEvent in .NET), and resumes when the response is delivered to the @response_handler (Microsoft Agent Framework: Workflow human-in-the-loop).
AWS Bedrock Agents & Step Functions: AWS Lambda action groups integrate with AWS Step Functions using task tokens (.waitForTaskToken). Execution pauses until an external system calls SendTaskSuccess or SendTaskFailure.

10. Validation record¶

Pending. Planned dry-run scenario: A second engineer builds the flow and runs the four paths plus a restart between interrupt and resume; pass when each path ends correctly and the action runs at most once.