Human-in-the-Loop: interrupt(), Approval, and Editing the Plan (LangGraph in Production, Module 07)

This is Module 07 of LangGraph in Production, a free 16-module course that takes you from your first StateGraph to a deployed, durable autonomous coding agent on LangGraph v1.x. Start at Module 1 or browse the full syllabus.

---

The Cowboy Agent Problem

Imagine this: you hand Forge the issue “Bug: tally report --month 2026-05 crashes with KeyError on empty category.” The agent springs into action. It reads the issue, classifies it as a bug, calls the Planner. The Planner probes the Tally repo and produces a tidy Plan: target files are tally/report.py and tally/models.py. Forge fans out two Editor tasks in parallel, rewrites both files, runs pytest — 12 passed. Total wall time: nine seconds.

Then, ten seconds later, a PR lands on main. Tests are green. CI merges it automatically. And buried in the diff is a refactor to models.py that nobody asked for — the Planner thought it was relevant, the Editor executed it, nobody checked.

This is the cowboy agent pattern: fully autonomous, very fast, and dangerous in proportion to how well it works. The problem is not autonomy per se — the problem is autonomy without a pause. The solution is not to wrap Forge in a polling while True loop in your own code. The solution is interrupt(): a native framework pause that suspends the graph at a specific node, surfaces a payload to the caller, waits for a decision, and resumes with the decision’s value injected back into the node’s return value. It integrates with the checkpointer you already built in Module 6, so a pause can last ten seconds or three hours with no lost work.

After this module, Forge pauses twice: once before any file is touched (you approve the Plan), and once before the PR is opened (you approve the diff). If you reject or edit the Plan, Forge re-plans without burning a full restart. The cowboy retires.

---

In This Module

You’ll learn:

• How interrupt(payload) suspends the graph and surfaces payload to the caller, and how graph.invoke(Command(resume=value), config) resumes it with value as the return of interrupt() (CA7.1)
• Why the interrupted node re-executes from its very beginning on resume, and how to keep code before interrupt() idempotent so you don’t pay for double LLM calls (CA7.2)
• The difference between static breakpoints (compile(interrupt_before=["n"])) — a debugging tool — and interrupt() — a production HITL mechanism — and when to use each (CA7.3)
• The four HITL patterns: approve/reject, edit-state, review-tool-calls, validate-input — and which one applies where in Forge (CA7.4)
• The PullRequest schema (frozen here), the open_pr tool, and the two new ForgeState fields (plan_approved, pull_request) introduced in this module (CA7.5)
• Why NodeInterrupt is gone in v1.x and what replaces it

You’ll build: Forge gains two human-in-the-loop checkpoints: interrupt() pauses before editing (you approve or reject the Plan) and again before opening the PR (you approve the diff). Rejection routes back to the Planner without a full restart. The open_pr tool and the PullRequest schema are introduced here and frozen for the rest of the course.

Concepts covered: CA7 — Human-in-the-loop (core). This module BUILDs CA7.1–CA7.5 of the LangGraph theory inventory.

Prerequisites: Modules 1–6 complete. SqliteSaver operational from Module 6 — the checkpointer is mandatory for interrupt() to work. Plan schema frozen (M4). editor/tester/fixer nodes in place (M3–M5). .env with ANTHROPIC_API_KEY or Ollama fallback configured.

---

Where We Are

M1 ✅  M2 ✅  M3 ✅  M4 ✅  M5 ✅  M6 ✅  M7 👉  M8 ⬜  M9 ⬜  M10 ⬜  M11 ⬜  M12 ⬜  M13 ⬜  M14 ⬜  M15 ⬜  M16 ⬜

Forge before M7: fully automatic from intake to PR. The graph never pauses. Hand it an issue, it plans, edits N files in parallel via the Send API, runs tests in a fix-it loop, and opens a PR — all without asking permission.

Forge after M7: two controlled pauses at the two highest-risk decision points. The first pause surfaces the Plan to you before a single file is written in the sandbox. The second pause surfaces the PR metadata before open_pr is called. Both pauses integrate with the SqliteSaver checkpointer from M6: the state survives the pause regardless of duration.

What we are not doing here: time-travel (walking state history, replaying from a past checkpoint, forking a bad run) — that is M8. Durability modes (durability="exit") and retry policies land in M9. Streaming the interrupt payload to a live UI comes in M11.

---

interrupt() and Command(resume=): The Anatomy of a Pause

Before interrupt() existed, getting a human decision into a running graph required gymnastics: polling the state externally, checking a flag, re-invoking with a new input. The framework had no native way to suspend and inject a value. The only built-in pause was static breakpoints — compile(interrupt_before=["node_name"]) — but those stop execution without surfacing a payload or accepting a return value. They are a debug tool, not a HITL protocol.

interrupt() is different. It is a call you put inside a node’s body. When the node executes that call, the graph suspends immediately. The value you passed to interrupt() is surfaced to the caller in the result’s __interrupt__ key. The caller inspects it, makes a decision, and resumes the graph by invoking it again with Command(resume=decision_value). The return value of the interrupt() call inside the node becomes decision_value. Execution continues from there.

Here is the plan_approval node from the actual lab code in forge/nodes.py:

from langgraph.types import Send, interrupt

def plan_approval(state: ForgeState) -> dict:
    """Pause for a human to approve / edit / reject the plan."""
    decision = interrupt({
        "action": "approve_plan",
        "summary": state["plan"].summary,
        "target_files": state["plan"].target_files,
    })
    edited = decision.get("edited_plan") if isinstance(decision, dict) else None
    if edited:
        plan = Plan(**edited)
        return {"plan": plan, "targets": _targets_from_plan(plan), "plan_approved": True}
    approved = bool(decision.get("approved")) if isinstance(decision, dict) else bool(decision)
    return {"plan_approved": approved}

Notice what the node does before interrupt(): it reads state, builds a dict, and calls interrupt(). No LLM call, no file write, no side effect. That is intentional and non-negotiable — you will see why in the next section.

The caller pattern looks like this (from tests/test_pipeline.py):

from langgraph.types import Command
from forge.graph import build_graph
from forge import checkpoint

graph = build_graph(checkpointer=checkpoint.in_memory())
cfg = {"configurable": {"thread_id": "approve"}, "recursion_limit": 50}

# First invoke: runs until the interrupt, returns the payload
r1 = graph.invoke({"issue": "report crashes on empty category", "repo_path": repo}, cfg)
assert r1["__interrupt__"][0].value["action"] == "approve_plan"

# Operator inspects plan, approves it
r2 = graph.invoke(Command(resume={"approved": True}), cfg)
# Graph ran until the SECOND interrupt (pr_approval)
assert r2["__interrupt__"][0].value["action"] == "approve_pr"

# Operator approves the PR
out = graph.invoke(Command(resume={"approved": True}), cfg)
assert out["pull_request"].status == "open"

This is real test code from the lab, passing against LangGraph 1.2.5. Three invokes, two pauses, one thread id. The checkpointer from M6 holds the state between each call: if the process dies between the first and second invoke, you can re-run graph.invoke(Command(resume={"approved": True}), cfg) on the same thread_id and it picks up exactly where it stopped.

Two things are required for interrupt() to work:

1. A checkpointer must be passed to build_graph(). Without it, interrupt() raises a configuration error at runtime. The thread must survive between the suspend and the resume — that is literally the checkpointer’s job.
2. The same config (same thread_id) must be passed to every subsequent invoke. The graph needs to identify which saved state to restore.

sequenceDiagram
    participant Caller
    participant Graph
    participant Node as plan_approval node
    Caller->>Graph: invoke({"issue": "..."}, cfg)
    Graph->>Node: execute — 1st time
    Node->>Graph: interrupt({"action": "approve_plan", ...})
    Graph-->>Caller: result["__interrupt__"][0].value = payload
    Note over Caller: Inspect plan, decide
    Caller->>Graph: invoke(Command(resume={"approved": True}), cfg)
    Graph->>Node: RE-EXECUTE from start (idempotent before interrupt!)
    Node->>Node: interrupt() returns {"approved": True}
    Node->>Graph: return {"plan_approved": True}
    Graph->>Graph: continue to dispatch → editor × N → ...

---

The Re-Execution Trap: Why Idempotence Before interrupt() Is Non-Negotiable

⚠️ Common misconception

“When the graph resumes after interrupt(), execution continues from the line after the interrupt() call.”

This is false. When graph.invoke(Command(resume=...), cfg) is called, the node re-executes from its very first line. Every line of code in the node body runs again — including every line that appears before the interrupt() call.

This is the single most common bug when developers first use interrupt(). The mental model of “resume from the pause point” is intuitive and wrong. LangGraph is a checkpoint-based system: resuming a node means re-running it from scratch, with the interrupt mechanism replaying the saved Command(resume=...) value as the return of interrupt(). It is the same node executing again, not a continuation.

What does this mean in practice? Any code before interrupt() that has a side effect will run twice: once on the original invocation, and once on the resume.

Here is the broken version:

# BROKEN: the LLM call runs on BOTH the first invoke and the resume
def plan_approval_broken(state: ForgeState) -> dict:
    # This call happens twice: once before the pause, once after resume
    enriched = config.get_model("smart").invoke([HumanMessage("Review this plan...")])
    decision = interrupt({"plan": enriched.content, "message": "Approve?"})
    return {"plan_approved": decision.get("approved", False)}

The LLM call is made on the first run (before the pause) and again on the resume. You pay twice. If the model call is slow or expensive, this doubles your latency and cost on every HITL interaction.

The fix is to separate concerns into separate nodes:

# CORRECT: side effects live in their own node, before plan_approval
def enrich_plan(state: ForgeState) -> dict:
    """Separate node: do the costly LLM call here."""
    enriched = config.get_model("smart").invoke(...)
    return {"plan": enriched_plan}

def plan_approval(state: ForgeState) -> dict:
    """HITL node: pure state read + interrupt(). No side effects."""
    decision = interrupt({
        "action": "approve_plan",
        "summary": state["plan"].summary,
        "target_files": state["plan"].target_files,
    })
    approved = bool(decision.get("approved")) if isinstance(decision, dict) else bool(decision)
    return {"plan_approved": approved}

The golden rule: a node that contains interrupt() must do nothing before the call except read state. No LLM calls, no file writes, no API requests, no database writes. Pure reads only. This makes the re-execution harmless — reading state twice produces the same result.

This is also why the Forge lab’s plan_approval node looks the way it does. The Planner is a separate node that runs before plan_approval. By the time plan_approval executes, the Plan is already in state. The node reads it, packages it, and calls interrupt(). On resume, it reads the same state again (unchanged), calls interrupt() (which immediately returns the saved Command(resume=...) value), and returns. One LLM call total, not two.

Multiple interrupt() calls in one node. LangGraph supports more than one interrupt() in a single node — they are matched to resumes in order. If a node calls interrupt() twice:

def two_step_gate(state: ForgeState) -> dict:
    first = interrupt("Approve the plan summary?")   # resume 1 provides this
    second = interrupt("Confirm the target files?")  # resume 2 provides this
    return {"plan_approved": first and second}

The first invoke(Command(resume=val1), cfg) supplies val1 to first, then hits the second interrupt() and pauses again. The second invoke(Command(resume=val2), cfg) supplies val2 to second. Useful for multi-step confirmation flows, but prefer separate nodes for readability unless the two questions are genuinely part of a single logical gate.

---

Static Breakpoints vs. interrupt(): Debug vs. Production

LangGraph has two distinct pause mechanisms. Conflating them is a freshness trap — v0.x tutorials often used breakpoints as the only pause option. In v1.x, the roles are clearly separated.

	interrupt_before / interrupt_after (static)	interrupt() (dynamic)
Purpose	Debugging — inspect state before or after a node	Production HITL — surface a payload, receive a decision
Where configured	compile(interrupt_before=["node_name"]) or per-call in invoke config	Inside the node body: value = interrupt(payload)
Payload surfaced	Current graph state (read via get_state())	The argument passed to interrupt()
Resume call	invoke(None, cfg) — no value injected	invoke(Command(resume=value), cfg) — value becomes interrupt()’s return
Trigger condition	Every time the named node is reached	Only when the node actually calls interrupt() (conditional, dynamic)
Suitable for production?	No — it stops on every run; no way to inject a decision value	Yes — designed for this use case

Use interrupt_before when you want to freeze a run in LangGraph Studio and inspect the state, or when you are debugging a test and want to see what a node receives. Use interrupt() when you want a human to make a decision that affects what the graph does next.

The lab includes a smoke test that demonstrates the difference directly: the static breakpoint test resumes with invoke(None, cfg) and the dynamic HITL test resumes with invoke(Command(resume={"approved": True}), cfg).

The Four HITL Patterns

CA7.4 maps four distinct patterns for human-in-the-loop interaction. Forge uses the first two directly in M7.

Pattern 1 — Approve / reject. The interrupt() payload is a question. The operator’s decision routes the graph conditionally. In Forge, plan_approval implements this: if approved=True, the gate function fans out to Editor nodes via Send; if approved=False, it routes to abandon.

def gate(state: ForgeState):
    """After approval: fan out to editors (Send), or route to abandon."""
    if not state.get("plan_approved"):
        return "abandon"
    return dispatch(state)  # returns list[Send]

The graph wiring in build_graph():

builder.add_edge("planner", "plan_approval")
builder.add_conditional_edges("plan_approval", nodes.gate, ["editor", "abandon"])
builder.add_edge("abandon", END)

Pattern 2 — Edit-state. The operator does not just approve or reject — they pass back a modified value that directly updates state. In plan_approval, if the Command(resume=...) contains an edited_plan key, Forge uses it:

edited = decision.get("edited_plan") if isinstance(decision, dict) else None
if edited:
    plan = Plan(**edited)
    return {"plan": plan, "targets": _targets_from_plan(plan), "plan_approved": True}

The operator passes Command(resume={"edited_plan": {"summary": "...", "target_files": ["report.py"], ...}}), and Forge proceeds with the corrected plan without calling the Planner again. This saves one (potentially expensive) smart model call and gives the operator surgical control over exactly what changes.

Pattern 3 — Review tool calls. interrupt() can live inside a @tool function, not just a node. If you want the operator to inspect the diff before open_pr submits it to GitHub, you put the interrupt inside the tool. Forge’s M7 implementation puts the pause in the pr_approval node instead (simpler for the lab), but the tool-level version is the pattern to reach for when you need approval at the tool-call boundary in a create_agent loop.

Pattern 4 — Validate input. Loop on interrupt() until the operator provides a valid value:

def validated_input_gate(state: ForgeState) -> dict:
    while True:
        value = interrupt({"message": "Provide a valid plan (JSON):", "schema": "Plan"})
        try:
            plan = Plan(**value)
            return {"plan": plan, "plan_approved": True}
        except Exception as e:
            # interrupt() is called again on the next loop iteration
            # Each call is a separate pause + resume cycle
            continue

Not built in Forge M7, but the mechanism works: each iteration of the loop is a new interrupt() call and therefore a new pause-resume cycle.

---

The Full Forge HITL Flow

flowchart TD
    A([issue]) --> B[intake]
    B --> C[triage]
    C --> D[planner]
    D --> E{plan_approval\ninterrupt\nApprove plan?}
    E -->|approved=True\nor edited_plan| F[gate → Send ×N]
    E -->|approved=False| G[abandon]
    G --> Z([END])
    F --> H[editor × N]
    H --> I[apply — defer=True]
    I --> J[tester]
    J -->|passed=False & attempts < 3| K[fixer]
    K --> J
    J -->|give_up| Z
    J -->|passed=True| L{pr_approval\ninterrupt\nOpen PR?}
    L -->|approved=True| M[pr opened\nPR.md written]
    L -->|approved=False| N([changes on branch\nno PR])
    M --> Z
    N --> Z

    style E fill:#f9a,stroke:#c66,color:#000
    style L fill:#f9a,stroke:#c66,color:#000

The two pink nodes are the HITL gates. Everything else is the Forge flow you built in M1–M6. The gates cost zero in terms of new LangGraph concepts — they are just nodes with interrupt() calls and a SqliteSaver beneath them.

---

PullRequest, open_pr, and the v1.x Interrupt Shape

The PullRequest Schema (Frozen Here)

Module 7 introduces PullRequest in forge/models.py. This schema is frozen for the rest of the course — do not add fields in later modules:

class PullRequest(BaseModel):
    """The PR Forge opens after human approval — frozen at Module 7."""

    title: str
    body: str
    branch: str
    files_changed: list[str]
    status: Literal["draft", "open", "merged"] = "draft"

Five fields exactly. No pr_number, no url, no author, no created_at. In production you would add those, but for the course, keeping the schema minimal and stable is more important than simulating a full GitHub response.

The ForgeState Extensions

Two new fields in forge/state.py — both absent before M7, both added by the addition-only rule:

class ForgeState(TypedDict, total=False):
    # ... all M2–M5 fields unchanged ...
    # --- Module 7 ---
    plan_approved: bool                                 # HITL plan-approval gate
    pull_request: PullRequest | None                    # the PR opened after approval
    # --- Module 12 adds: review ---

Both are last-write-wins (no reducer annotation). review: Review | None does not appear until M12 — do not add it here.

The open_pr Tool

The lab’s open_pr tool lives in forge/tools.py as a factory function (like the other tools from M4–M5, which close over a repo_path):

def make_pr_tool(repo_path: str | Path):
    root = Path(repo_path)

    @tool
    def open_pr(title: str, body: str, files_changed: list[str]) -> str:
        """Open a pull request: write PR.md describing the change."""
        (root / "PR.md").write_text(
            f"# {title}\n\n{body}\n\nFiles changed: {', '.join(files_changed)}\n"
        )
        return f"opened PR: {title}"

    return open_pr

In the lab, “opening a PR” means writing PR.md to the sandbox. In production, you replace the body with a GitHub API call. The interface stays the same; only the side effect changes. The pr_approval node calls this directly — it does not go through create_agent because there is no LLM decision to make; the operator has already approved.

Here is pr_approval, the second HITL gate:

def pr_approval(state: ForgeState) -> dict:
    """Pause for a human to approve opening the PR; open it on approval."""
    files = [e.path for e in state.get("edits", [])]
    decision = interrupt({"action": "approve_pr", "files": files})
    approved = bool(decision.get("approved")) if isinstance(decision, dict) else bool(decision)
    if not approved:
        return {"messages": [AIMessage(content="PR not approved — changes left on the branch.")]}
    pr = PullRequest(
        title=f"Fix: {state['issue'][:60]}",
        body=state["plan"].summary if state.get("plan") else state["issue"],
        branch="forge/auto-fix",
        files_changed=files,
        status="open",
    )
    sandbox.write_file(state["repo_path"], "PR.md",
        f"# {pr.title}\n\n{pr.body}\n\nFiles changed: {', '.join(files)}\n")
    return {"pull_request": pr, "messages": [AIMessage(content=f"Opened PR: {pr.title}")]}

Notice that pr_approval creates the PullRequest object and writes the file in one node. This works because there is no LLM call before interrupt() — the only state reads are state.get("edits", []) and state["issue"], which are idempotent. On resume, the node reads the same values, builds the same draft object, calls interrupt() (which immediately returns the saved approval), and proceeds to write the file. The file write happens after the interrupt call, on the resume path — not before it.

NodeInterrupt Is Gone: What Replaced It

In LangGraph v0.x, some tutorials used from langgraph.errors import NodeInterrupt with raise NodeInterrupt(value="..."). This API is removed in v1.x. If you find it in code you are reading, that code is outdated.

The correct v1.x import is:

from langgraph.types import interrupt   # the only valid path in v1.x
# NEVER: from langgraph.errors import NodeInterrupt  (removed)

The Interrupt shape surfaced to the caller in v1.x has two fields: value (the payload you passed to interrupt()) and id (an internal identifier). The old fields interrupt_id, when, resumable, and ns are gone. The test confirms the shape:

r1 = graph.invoke({"issue": "...", "repo_path": repo}, cfg)
assert r1["__interrupt__"][0].value["action"] == "approve_plan"
#                                 ^^^^^
#                                 .value is the payload you passed to interrupt()

---

Hands-On Lab: Wiring the Two HITL Gates

Objective: Add plan_approval and pr_approval to Forge, introduce PullRequest and the open_pr tool, and verify both gates via offline tests that use Command(resume=...) without any real API calls.

Lab path: module-07/ in langgraph-prod-labs

This lab cost approximately $0.15 on June 2026 prices (a few smart-tier Planner calls). The sandbox Tester runs pytest on Tally with no API key. The fallback Ollama configuration costs $0.00 — swap one line in forge/config.py.

Step 1: Extend forge/models.py

Add PullRequest at the bottom of models.py, after the existing M5 classes. No existing class changes:

class PullRequest(BaseModel):
    """The PR Forge opens after human approval — frozen at Module 7."""

    title: str
    body: str
    branch: str
    files_changed: list[str]
    status: Literal["draft", "open", "merged"] = "draft"

Run the contract test immediately:

def test_pullrequest_contract():
    pr = PullRequest(title="t", body="b", branch="x", files_changed=["report.py"])
    assert pr.status == "draft"  # opens as draft, set to "open" on approval

Step 2: Extend forge/state.py

Add the two M7 fields after the M5 block:

# --- Module 7 ---
plan_approved: bool                                 # HITL plan-approval gate
pull_request: PullRequest | None                    # the PR opened after approval
# --- Module 12 adds: review ---

Import PullRequest in the state module header alongside the other model imports.

Step 3: Add open_pr to forge/tools.py

Add make_pr_tool as a factory function alongside make_read_tools and make_write_tool. The unit test for it:

def test_open_pr_tool_writes_pr_md():
    repo = sandbox.make_sandbox(Path(__file__).resolve().parents[1] / "tally")
    open_pr = tools.make_pr_tool(repo)
    msg = open_pr.invoke({"title": "Fix bug", "body": "desc", "files_changed": ["report.py"]})
    assert "opened PR" in msg and (Path(repo) / "PR.md").exists()

No LLM required. Pure filesystem side effect, fully testable offline.

Step 4: Add the Two HITL Nodes to forge/nodes.py

Add plan_approval, gate, abandon, and pr_approval. The gate function is what the conditional edge calls — it either fans out (dispatch(state) returns list[Send]) or routes to abandon:

def gate(state: ForgeState):
    """After approval: fan out to editors (Send), or route to abandon."""
    if not state.get("plan_approved"):
        return "abandon"
    return dispatch(state)

def abandon(state: ForgeState) -> dict:
    return {"messages": [AIMessage(content="Plan not approved — abandoning the run.")]}

Test the gate in isolation — no graph required:

def test_gate_routes_on_approval():
    targets = [FileTarget(path="report.py", instruction="fix")]
    base = {"repo_path": "/r", "issue": "i", "targets": targets}
    assert nodes.gate({**base, "plan_approved": False}) == "abandon"
    sends = nodes.gate({**base, "plan_approved": True})
    assert len(sends) == 1 and all(isinstance(s, Send) for s in sends)

Step 5: Wire the Graph in forge/graph.py

The full build_graph() function after M7 changes:

def build_graph(checkpointer=None):
    builder = StateGraph(ForgeState)
    builder.add_node("intake", nodes.intake)
    builder.add_node("triage", nodes.triage)
    builder.add_node("planner", nodes.planner)
    builder.add_node("plan_approval", nodes.plan_approval)
    builder.add_node("abandon", nodes.abandon)
    builder.add_node("editor", nodes.editor)
    builder.add_node("apply", nodes.apply_edits, defer=True)
    builder.add_node("tester", nodes.tester)
    builder.add_node("fixer", nodes.fixer)
    builder.add_node("pr_approval", nodes.pr_approval)

    builder.add_edge(START, "intake")
    builder.add_edge("intake", "triage")
    builder.add_conditional_edges(
        "triage", nodes.route_by_type,
        {"bug": "planner", "feature": "planner", "chore": "planner"},
    )
    builder.add_edge("planner", "plan_approval")
    builder.add_conditional_edges("plan_approval", nodes.gate, ["editor", "abandon"])
    builder.add_edge("abandon", END)
    builder.add_edge("editor", "apply")
    builder.add_edge("apply", "tester")
    builder.add_conditional_edges(
        "tester", nodes.route_after_tests,
        {"done": "pr_approval", "give_up": END, "fixer": "fixer"},
    )
    builder.add_edge("fixer", "tester")
    builder.add_edge("pr_approval", END)
    return builder.compile(checkpointer=checkpointer)

The checkpointer parameter is passed through. Tests use checkpoint.in_memory(); production uses checkpoint.sqlite("forge_checkpoints.db").

Step 6: Run the Integration Tests Offline

The pipeline test in tests/test_pipeline.py drives the full happy path with a RoutedFakeModel (no API key needed):

def test_hitl_approve_plan_and_pr(monkeypatch, make_routed_model):
    repo = _setup(monkeypatch, make_routed_model)
    graph = build_graph(checkpointer=checkpoint.in_memory())
    cfg = {"configurable": {"thread_id": "approve"}, "recursion_limit": 50}

    r1 = graph.invoke({"issue": "report crashes on empty category", "repo_path": repo}, cfg)
    assert r1["__interrupt__"][0].value["action"] == "approve_plan"

    r2 = graph.invoke(Command(resume={"approved": True}), cfg)
    assert r2["__interrupt__"][0].value["action"] == "approve_pr"

    out = graph.invoke(Command(resume={"approved": True}), cfg)
    assert out["test_result"].passed is True
    assert out["pull_request"].status == "open"
    assert (Path(repo) / "PR.md").exists()

And the rejection path:

def test_hitl_reject_plan_abandons(monkeypatch, make_routed_model):
    # ... setup ...
    r1 = graph.invoke({"issue": "..."}, cfg)
    out = graph.invoke(Command(resume={"approved": False}), cfg)
    assert out.get("plan_approved") is False
    assert out.get("pull_request") is None      # abandoned — no PR
    # report.py is unchanged — editors never ran
    assert "totals.get(cat" not in (Path(repo) / "report.py").read_text()

Run all tests across M1–M7:

uv run pytest tests/ -q

Expected: all offline tests green, no API key required. The Tester node runs real pytest on the Tally sandbox during the integration test — that is your deterministic signal that the fix actually works.

Try It Yourself

1. Edit-state pattern: Resume with Command(resume={"edited_plan": {"summary": "fix only report.py", "target_files": ["report.py"], "steps": [...], "risk": "low"}}) instead of {"approved": True}. Verify that state["plan"].target_files contains only report.py and that the Planner node was not re-called.

2. Crash between pauses: After the first invoke triggers the plan-approval interrupt, kill the process. Create a new process, restore the SqliteSaver on the same database file, and invoke graph.invoke(Command(resume={"approved": True}), cfg) with the same thread_id. Verify the run completes from where it stopped — no re-execution of intake, triage, or planner.

3. Static breakpoint comparison: In a separate script, pass interrupt_before=["plan_approval"] to compile() instead of using interrupt() in the node body. Resume with invoke(None, cfg). Observe that no payload reaches the caller and no decision value is injected — this is why breakpoints are for debugging only.

---

In Production

Two HITL checkpoints is the right number for a coding agent. That is an opinion, and I am comfortable defending it: plan approval before any write hits the sandbox catches the Planner’s worst mistakes (wrong files, scope creep, risky refactors). PR approval before pushing catches anything the test suite does not cover (style, intent, hidden coupling). More than two checkpoints and you have rebuilt a manual process with extra steps. Zero and you are a cowboy deployer.

The checkpointer is what makes these pauses production-viable. A plan approval pause can last thirty seconds (operator is watching the terminal) or three hours (operator is in a meeting). LangGraph does not care — the state is persisted in SQLite or Postgres, and graph.invoke(Command(resume=...), cfg) on the same thread ID works identically either way. This is why Module 6 is a hard prerequisite: without a checkpointer, interrupt() raises an error at runtime.

The idempotence constraint before interrupt() is not accidental overhead — it is a production invariant. Every node that contains interrupt() should be testable with a fake resume (no LLM call, no API key). If you cannot test it offline, you have not separated concerns correctly. Forge’s plan_approval and pr_approval nodes pass this bar: both are pure state reads before the interrupt() call.

Pattern 2 (edit-state) is underused in practice. Most developers implement approve/reject and stop there. But letting the operator pass a revised_plan directly in Command(resume=...) eliminates a full Planner re-run — potentially one smart-tier model call saved per rejected plan. At scale (hundreds of issues per day), that adds up.

Observability: in M11 you will stream the interrupt() payload to a live UI using get_stream_writer, so the operator sees the plan appear in real time rather than waiting for the full invoke to return. The mechanism is the same; the delivery channel changes.

Audit trail: get_state_history() records every checkpoint, including every Interrupt checkpoint with the saved payload. You can reconstruct exactly what the operator was shown and what they approved — useful for compliance. This becomes the foundation for time-travel in M8.

---

Mastery Corner

What to Really Understand Here

CA7.1 — interrupt(payload) is a call inside a node that suspends the graph and surfaces payload in the result’s __interrupt__ key. Resuming requires graph.invoke(Command(resume=value), same_config), where value becomes the return value of the interrupt() call. A checkpointer is mandatory.

CA7.2 — On resume, the node re-executes from line 1. Every line before interrupt() runs again. Make those lines pure state reads. Put anything with side effects in a separate upstream node.

CA7.3 — Static breakpoints (compile(interrupt_before=["n"])) stop execution for debugging; they surface the state, not a custom payload, and resume with invoke(None, cfg). They inject no value. They are not a production HITL mechanism.

CA7.4 — Four HITL patterns: approve/reject (conditional routing on the resume value), edit-state (resume value directly updates state), review-tool-calls (interrupt() inside a @tool), validate-input (loop on interrupt() until valid). Forge M7 implements the first two.

CA7.5 — The v1.x interrupt shape is Interrupt(value=..., id=...). NodeInterrupt is removed. The only valid import is from langgraph.types import interrupt.

---

Five Mastery Questions

Q1. You add this node to Forge:

def plan_approval(state: ForgeState) -> dict:
    enriched = config.get_model("smart").invoke([HumanMessage("Summarize this plan...")])
    decision = interrupt({"plan": enriched.content, "message": "Approve?"})
    return {"plan_approved": decision.get("approved", False)}

After deploying, your team notices that every plan approval is billed for two LLM calls instead of one. What is the root cause?

A) interrupt() forces a full graph restart, re-running all previous nodes including the Planner.
B) The node re-executes from line 1 on resume, so get_model().invoke() runs twice — once before the pause, once after.
C) The checkpointer serializes and deserializes the model call, causing a second invocation.
D) Command(resume=...) triggers a second call to interrupt() which re-invokes the model.

---

Q2. A teammate configures compile(interrupt_before=["plan_approval"]) instead of calling interrupt() inside the plan_approval node. The operator resumes with graph.invoke(None, cfg). Which statement is true?

A) The behavior is identical: both surface the plan payload and accept a decision value.
B) The static breakpoint surfaces the current graph state, not a custom payload; the resume injects no value; the node cannot receive an operator decision.
C) The static breakpoint is more efficient because it avoids re-executing the node on resume.
D) invoke(None, cfg) and invoke(Command(resume=value), cfg) are interchangeable; both inject the value correctly.

---

Q3. The operator wants to correct the Planner’s list of target_files (say, remove models.py) without triggering a full re-plan. Which HITL pattern applies, and what does the correct Command(resume=...) look like?

A) Approve/reject — Command(resume={"approved": False}), then re-invoke from scratch.
B) Edit-state — Command(resume={"edited_plan": {"summary": "...", "target_files": ["report.py"], "steps": [...], "risk": "low"}}), handled by the edited = decision.get("edited_plan") branch of plan_approval.
C) Validate-input — loop until the operator provides a valid plan via the CLI.
D) Review-tool-calls — put interrupt() inside make_read_tools to intercept the Planner’s file reads.

---

Q4. You clone a repo that contains:

from langgraph.errors import NodeInterrupt
raise NodeInterrupt(value="Approve plan?")

You run it against LangGraph 1.2.5. What happens, and what is the correct migration?

A) It works identically to interrupt() — NodeInterrupt is an alias in v1.x.
B) NodeInterrupt is removed in v1.x; the import fails with ImportError. Migrate to from langgraph.types import interrupt and replace raise NodeInterrupt(...) with value = interrupt(...).
C) NodeInterrupt is deprecated but still functional; it logs a deprecation warning but does not raise an error.
D) The graph runs without pausing because NodeInterrupt is caught internally and swallowed.

---

Q5. A node contains two sequential interrupt() calls. An operator resumes the first pause with Command(resume="yes"). What happens next?

A) Both interrupt() calls receive "yes" simultaneously; the node returns after the second one.
B) The graph resumes, the node re-executes from line 1, the first interrupt() returns "yes" (the saved resume value), and execution hits the second interrupt() — causing a second pause. A second Command(resume=...) is needed.
C) The second interrupt() is skipped; Command(resume=...) provides a value for the first call only, and the node continues past the second call with None.
D) The runtime raises an error because multiple interrupt() calls in one node are not supported in v1.x.

---

Answers

A1 — B. The node re-executes from its very beginning on resume (CA7.2). get_model().invoke() appears before interrupt(), so it runs on both the original invocation and the resume. Fix: move the LLM call to a separate upstream node; keep plan_approval as a pure state reader.

A2 — B. Static breakpoints (interrupt_before) surface the graph’s current state via get_state(). They do not pass a custom payload, and resuming with invoke(None, cfg) injects no value into the node. The node cannot receive an operator decision through this mechanism. Use interrupt() for production HITL (CA7.3).

A3 — B. This is the edit-state pattern (CA7.4). The operator constructs a corrected Plan dict and passes it in Command(resume={"edited_plan": {...}}). The plan_approval node checks decision.get("edited_plan"), builds a new Plan object, and returns it — skipping the Planner entirely.

A4 — B. NodeInterrupt was removed in LangGraph v1.x (CA7.5). The import raises ImportError. The migration is from langgraph.types import interrupt, and instead of raise NodeInterrupt(value=...), you write result = interrupt(...) and use result as the return value.

A5 — B. LangGraph matches Command(resume=...) values to interrupt() calls in order (CA7.2). The first resume provides the value for the first interrupt(). Execution continues and hits the second interrupt(), pausing again. A second Command(resume=...) is required to satisfy it.

---

Traps to Avoid

Trap 1 — Assuming resume continues from the interrupt() call site. The node re-executes from line 1 on resume. Anything before interrupt() runs twice. This is the source of double billing, double writes, and duplicated side effects. Keep everything before interrupt() pure.

Trap 2 — Using NodeInterrupt from a v0.x tutorial. from langgraph.errors import NodeInterrupt is removed in v1.x and will fail at import time. The correct API is from langgraph.types import interrupt. If you are reading a tutorial that uses raise NodeInterrupt(...), that tutorial is outdated.

Trap 3 — Using interrupt_before as a production HITL mechanism. Static breakpoints cannot surface a custom payload and cannot receive a decision value. They are a debugging convenience. Resuming with invoke(None, cfg) does not inject anything into the graph — it just continues. For production HITL, use interrupt().

---

Key Takeaways

• interrupt(payload) suspends the graph and surfaces payload to the caller in result["__interrupt__"][0].value. graph.invoke(Command(resume=value), config) resumes the graph and injects value as the return of interrupt(). A checkpointer is mandatory for both to work.

• The node re-executes from its start on resume. Every line before interrupt() runs again. Keep those lines to pure state reads. Put LLM calls, file writes, and API requests in separate upstream nodes.

• interrupt_before / interrupt_after are static breakpoints for debugging — they surface state, accept no return value, and resume with invoke(None, cfg). interrupt() is the production HITL mechanism. They are not interchangeable.

• The four HITL patterns — approve/reject, edit-state, review-tool-calls, validate-input — cover every HITL scenario you will encounter. Forge M7 implements the first two.

• PullRequest (title, body, branch, files_changed, status) is the frozen M7 schema. open_pr is the canonical tool name. plan_approved and pull_request are the two ForgeState fields introduced here.

• NodeInterrupt is removed in LangGraph v1.x. The only valid import is from langgraph.types import interrupt. The v1.x Interrupt shape has two fields: value and id.

---

What’s Next + Newsletter

This is Module 07 of LangGraph in Production, a free 16-module course that takes you from your first StateGraph to a deployed, durable autonomous coding agent on LangGraph v1.x. Start at Module 1 or browse the full syllabus.

Want the full LangGraph mastery question bank (interview-style) and the next module in your inbox? Subscribe here — it’s free, like everything in this series.

Module 08 — Time-Travel: Forge can now pause and wait for approval — but what if the plan was wrong and you want to go back? Module 8 introduces time-travel: walking state history with get_state_history(), replaying from a past checkpoint, and forking a bad run to try a different Plan without starting over.

---

Links: Lab code — module-07 · Previous: Module 06 — Persistence · Next: Module 08 — Time-Travel · Course index

---

References

1. LangGraph Docs — How-to: Human-in-the-loop / interrupt(): https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/ (verified June 2026)

2. LangGraph Docs — langgraph.types: interrupt, Command, Send: https://langchain-ai.github.io/langgraph/reference/types/ (verified June 2026)

3. LangGraph Docs — Breakpoints (static interrupt_before / interrupt_after): https://langchain-ai.github.io/langgraph/how-tos/breakpoints/ (verified June 2026)

4. LangGraph Changelog — v1.0 GA release notes (removal of NodeInterrupt, new interrupt() API): https://github.com/langchain-ai/langgraph/releases (verified June 2026)

5. Lab code — langgraph-prod-labs/module-07/: https://github.com/dupuis1212/langgraph-prod-labs/tree/main/module-07