Human-in-the-Loop & Time Travel

This content was generated with the assistance of AI and should be verified against the official documentation of each tool and framework before being relied on in production.

This section’s bibliography lists the reference material consulted while preparing these pages.

This page was written against langchain 1.4.3, langchain-core 1.6.5, langgraph 1.2.12 and langgraph-checkpoint-sqlite on Python 3.12, plus langgraph-cli 0.4.32, langgraph-api 0.15.1 and langgraph-sdk 0.4.5 for the double-texting section, as verified on 2026-09-29. The running example is DocsAssistant, which answers questions over this site’s documentation with Ollama’s llama3.1:8b chat model over the pgvector store built in Integrating with LangChain; its one write action is opening a GitHub issue about a documentation gap. Ollama was not available while the snippets were checked, so every agent run below used a scripted fake chat model (a GenericFakeChatModel subclass that replays prepared messages, including the tool calls); the graphs, the middleware, the checkpointers and the Agent Server were the real packages. No output comes from a live model.

A person is the most reliable guard an agent can have and the most expensive one, so the design question is where to ask, not whether. The design side of that question — which actions need approval, escalation thresholds, the autonomy slider — is in Human-in-the-Loop and the interface side in Agent UX. This page is the LangChain and LangGraph mechanism underneath: how a run stops and later continues, how a reviewer approves, edits or rejects a tool call, how the saved state can be edited, how a thread can be replayed or forked from any past checkpoint, and what happens when a user sends a second message while the first is still running. All of it rests on the checkpointers described in Short-Term Memory & Checkpointers and on the graph model in LangGraph Fundamentals.

interrupt() and Command(resume=…)

interrupt() pauses a graph from inside a node. LangGraph saves the state with the checkpointer, hands the payload passed to interrupt() to the caller, and waits, for as long as it takes, until the same thread is invoked again with Command(resume=value). That value becomes the return value of the interrupt() call. Three things are needed: a checkpointer, a thread_id in the config, and a JSON-serialisable payload. The graph below drafts a GitHub issue and stops for approval before filing it:

An approval node that pauses with interrupt()

from typing import Literal, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class IssueState(TypedDict):
    question: str
    draft: str
    status: str


def draft_issue(state: IssueState):
    print("draft_issue runs")
    return {"draft": f"Docs gap: {state['question']}"}


def approval(state: IssueState) -> Command[Literal["file", "discard"]]:
    print("approval starts")                       # runs again on every resume
    decision = interrupt({"question": "File this issue?", "draft": state["draft"]})
    return Command(goto="file" if decision else "discard")


def file(state: IssueState):
    return {"status": "filed"}


def discard(state: IssueState):
    return {"status": "discarded"}


builder = StateGraph(IssueState)
builder.add_node("draft_issue", draft_issue)
builder.add_node("approval", approval)
builder.add_node("file", file)
builder.add_node("discard", discard)
builder.add_edge(START, "draft_issue")
builder.add_edge("draft_issue", "approval")
builder.add_edge("file", END)
builder.add_edge("discard", END)
graph = builder.compile(checkpointer=InMemorySaver())

cfg = {"configurable": {"thread_id": "issue-1"}}
result = graph.invoke({"question": "ef_search is undocumented"}, cfg)
print("__interrupt__:", result["__interrupt__"])
print("next:", graph.get_state(cfg).next)
final = graph.invoke(Command(resume=True), cfg)
print("final:", final["status"])

Output

draft_issue runs
approval starts
__interrupt__: [Interrupt(value={'question': 'File this issue?', 'draft': 'Docs gap: ef_search is undocumented'}, id='781cb8a7f8bc95450be8f74e9c4c5122')]
next: ('approval',)
approval starts
final: filed

The first invoke returned early with the payload under the interrupt key and the graph stopped with approval still pending (next). The resume used the same thread_id, and True came back from interrupt(), routing to file. Notice that draft_issue did not run a second time, because its result was already checkpointed, but approval starts printed twice: on resume the runtime restarts the whole node from its first line and only then lets interrupt() return the resume value. Anything before the interrupt() call must therefore be cheap and idempotent. See LangGraph — Interrupts.

Passing version="v2" to invoke returns a GraphOutput whose .interrupts attribute holds the pending Interrupt objects and whose .value holds the state, which is what the agent snippets below use. Newer code can also drive a run with graph.stream_events(…, version="v3") and read stream.interrupted and stream.interrupts after draining stream.output; the same run through it printed interrupted: True [\{'question': 'File this issue?', 'draft': 'Docs gap: x'}], but langgraph 1.2.12 emits a LangChainBetaWarning saying the v3 streaming protocol on Pregel is experimental, so this page sticks to invoke. The streaming side is in Streaming. See LangGraph — Event streaming.

A pause is durable only if the checkpointer is. The next run wrote its checkpoints to a SQLite file (SqliteSaver, from langgraph-checkpoint-sqlite), paused in one Python process, and was resumed by a second process started afterwards, which is how an approval that arrives hours later, or after a deployment, works:

Pausing in one process and resuming in another

import sqlite3, sys
from typing import TypedDict

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class S(TypedDict, total=False):
    draft: str
    status: str


def draft(state: S):
    return {"draft": "Docs gap: ef_search"}


def review(state: S):
    return {"status": "filed" if interrupt({"draft": state["draft"]}) else "discarded"}


saver = SqliteSaver(sqlite3.connect("approvals.sqlite", check_same_thread=False))
graph = (StateGraph(S).add_node("draft", draft).add_node("review", review)
         .add_edge(START, "draft").add_edge("draft", "review").add_edge("review", END)
         .compile(checkpointer=saver))

cfg = {"configurable": {"thread_id": "issue-42"}}
if sys.argv[1] == "ask":
    print("paused:", graph.invoke({}, cfg)["__interrupt__"][0].value)
else:
    print("resumed:", graph.invoke(Command(resume=True), cfg)["status"])

Output (python approve.py ask, then, in a new process, python approve.py approve)

paused: {'draft': 'Docs gap: ef_search'}
resumed: filed

Production deployments use a Postgres checkpointer for the same reason; the options are compared in Short-Term Memory & Checkpointers. See LangGraph — Persistence.

How a run stops and continues

sequenceDiagram participant U as Reviewer UI participant A as Graph run participant C as Checkpointer U->>A: invoke(input, thread_id) A->>A: nodes run, interrupt() raises GraphInterrupt A->>C: save checkpoint and pending writes A-->>U: __interrupt__ payload Note over U,C: minutes, hours or a deploy later U->>A: invoke(Command(resume=value), same thread_id) C-->>A: load last checkpoint A->>A: restart the interrupted node, interrupt() returns value A-->>U: final state

Figure 1. An interrupt saves state and returns; the resume restarts the node

The diagram shows why the same thread_id is essential: it is the only key the checkpointer has for finding the paused run.

Static breakpoints and the older interrupt_before

Before interrupt(), graphs were paused with static breakpoints: interrupt_before=[…] or interrupt_after=[…] given at compile() or at invoke time stops the run around a named node. They still work and are useful for debugging one node at a time, because the pause needs no code in the node. Resuming is invoke(None, cfg):

A static breakpoint before the file node

g = (StateGraph(S)
     .add_node("draft", lambda s: {"draft": "Docs gap: ef_search"})
     .add_node("file", lambda s: {"status": "filed"})
     .add_edge(START, "draft").add_edge("draft", "file").add_edge("file", END)
     .compile(checkpointer=InMemorySaver(), interrupt_before=["file"]))
cfg = {"configurable": {"thread_id": "bp"}}
g.invoke({}, cfg)
print("paused before:", g.get_state(cfg).next, "| draft:", g.get_state(cfg).values["draft"])
print("resumed:", g.invoke(None, cfg))          # None input continues from the breakpoint

Output

paused before: ('file',) | draft: Docs gap: ef_search
resumed: {'draft': 'Docs gap: ef_search', 'status': 'filed'}

The differences that matter for application code: a breakpoint cannot carry a payload to the reviewer, cannot receive a value from them, and cannot be conditional on the state, whereas interrupt() can do all three. Guides written for LangGraph 0.x approve tool calls with interrupt_before=["tools"]; in 1.x the maintained way is interrupt() or HumanInTheLoopMiddleware. See LangGraph — Interrupts.

The rules of interrupts

Because the node restarts on resume and the interrupt is implemented by raising an exception, a few rules keep it predictable. The LangGraph interrupts page states them and the snippets below confirm two of them.

Do not wrap interrupt() in a bare try/except. The pause is a GraphInterrupt exception; a broad except Exception swallows it and the run continues as if nobody had been asked.
Keep interrupts in the same order on every execution. When a node calls interrupt() several times, resume values are matched by index, so a conditional or reordered call shifts the answers onto the wrong questions.
Call interrupt() once per node execution when validating input. To re-ask after an invalid answer, store the new question in state and route back to the node with a conditional edge, rather than a while True loop that replays every earlier iteration on each resume.
Make code before interrupt() idempotent, since it runs again on resume, and put side effects after it or in a later node.

The first rule is easy to violate by accident, for example with a retry decorator or a logging wrapper around the node body:

The bare except that swallows the pause

def swallow(state):
    try:
        interrupt("never reaches the caller")
    except Exception as e:                       # catches GraphInterrupt
        print("swallowed:", type(e).__name__)
    return {"vals": ["done"]}

Output

swallowed: GraphInterrupt
interrupted: False ['done']

When parallel branches interrupt in the same super-step, one resume call has to answer all of them. Command(resume=…) then takes a dictionary from each Interrupt.id to its answer, and every pending interrupt is resumed at once:

Answering two parallel interrupts by ID

import operator
from typing import Annotated, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class State(TypedDict):
    vals: Annotated[list[str], operator.add]


def ask_title(state):
    return {"vals": [f"title:{interrupt('Issue title?')}"]}


def ask_label(state):
    return {"vals": [f"label:{interrupt('Issue label?')}"]}


g = (StateGraph(State)
     .add_node("ask_title", ask_title).add_node("ask_label", ask_label)
     .add_edge(START, "ask_title").add_edge(START, "ask_label")
     .add_edge("ask_title", END).add_edge("ask_label", END)
     .compile(checkpointer=InMemorySaver()))
cfg = {"configurable": {"thread_id": "p"}}
out = g.invoke({"vals": []}, cfg)
print([i.value for i in out["__interrupt__"]])
answers = {"Issue title?": "Explain ef_search", "Issue label?": "docs"}
final = g.invoke(Command(resume={i.id: answers[i.value] for i in out["__interrupt__"]}), cfg)
print(sorted(final["vals"]))

Output

['Issue label?', 'Issue title?']
['label:docs', 'title:Explain ef_search']

The two questions arrived in no fixed order, which is why the mapping is keyed by ID rather than by position. See LangGraph — Interrupts.

HumanInTheLoopMiddleware

Writing an interrupt() node for every dangerous tool would repeat the same bookkeeping, so LangChain ships it as HumanInTheLoopMiddleware (an after_model hook, described with the other built-ins in Middleware). It inspects the tool calls in each model reply, and for the ones its policy names it interrupts with an action request before any tool runs. The policy is the interrupt_on dictionary: a tool name maps to True (interrupt, all decisions allowed), False (never interrupt) or a config with allowed_decisions. The four decisions are approve, edit, reject and respond. Below is DocsAssistant’s open_github_issue tool, with the three decisions that matter for a side-effecting tool, each in a fresh thread. The credential is read from an environment variable at call time, is scoped to issue writes on one repository, and was a placeholder value in these runs:

The three decisions on open_github_issue

import os

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langchain_core.messages import AIMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

opened = []


@tool
def open_github_issue(title: str, body: str) -> str:
    """Open an issue in the documentation repository. Destructive: creates public content."""
    token = os.environ["DOCS_ISSUES_TOKEN"]      # fine-grained token: Issues write on one repository
    opened.append((title, body))                 # ... POST to the GitHub API with the token ...
    return f"Opened issue: {title}"


def build(model):
    return create_agent(
        model, tools=[open_github_issue], checkpointer=InMemorySaver(),
        middleware=[HumanInTheLoopMiddleware(
            interrupt_on={"open_github_issue": {"allowed_decisions": ["approve", "edit", "reject"]}},
            description_prefix="DocsAssistant wants to open an issue")])


ask = {"messages": [{"role": "user", "content": "File a docs issue about ef_search"}]}
for name, decision in [
    ("approve", {"type": "approve"}),
    ("edit", {"type": "edit", "edited_action": {"name": "open_github_issue",
              "args": {"title": "Doc gap: hnsw ef_search", "body": "Explain ef_search and its default."}}}),
    ("reject", {"type": "reject", "message": "Duplicate of an open issue. Do not retry."}),
]:
    opened.clear()
    agent = build(scripted_model)                # replays: one open_github_issue call, then "Done."
    cfg = {"configurable": {"thread_id": name}}
    agent.invoke(ask, cfg, version="v2")         # pauses before the tool runs
    out = agent.invoke(Command(resume={"decisions": [decision]}), cfg, version="v2")
    print(f"--- {name}: opened={opened}")
    for m in out.value["messages"][1:]:
        print("  ", type(m).__name__, "|", getattr(m, "tool_calls", None) or m.content)

Output

--- approve: opened=[('Doc gap: ef_search', 'Explain ef_search.')]
   AIMessage | [{'name': 'open_github_issue', 'args': {'title': 'Doc gap: ef_search', 'body': 'Explain ef_search.'}, 'id': 'c1', 'type': 'tool_call'}]
   ToolMessage | Opened issue: Doc gap: ef_search
   AIMessage | Done.
--- edit: opened=[('Doc gap: hnsw ef_search', 'Explain ef_search and its default.')]
   AIMessage | [{'name': 'open_github_issue', 'args': {'title': 'Doc gap: ef_search', 'body': 'Explain ef_search.'}, 'id': 'c1', 'type': 'tool_call'}]
   ToolMessage | Note: a human reviewer replaced this tool call before it ran. The call recorded in your message is the one you produced, not the one that executed. This was intentional and authorized. Do not re-issue your original call. Executed instead: open_github_issue with arguments {"title": "Doc gap: hnsw ef_search", "body": "Explain ef_search and its default."}.

Tool response:
Opened issue: Doc gap: hnsw ef_search
   AIMessage | Done.
--- reject: opened=[]
   AIMessage | [{'name': 'open_github_issue', 'args': {'title': 'Doc gap: ef_search', 'body': 'Explain ef_search.'}, 'id': 'c1', 'type': 'tool_call'}]
   ToolMessage | User rejected the tool call for `open_github_issue` with reason: Duplicate of an open issue. Do not retry.
   AIMessage | Done.

Each decision changes something different:

approve runs the call exactly as the model proposed it.
edit runs the tool with the reviewer’s edited_action (tool name and arguments). The model’s own message still shows the original call, so the middleware appends a note to the tool result saying a human replaced it; the text is the edit_notice constructor argument (set it to None to drop it), and it stops the model from re-issuing the call it believes failed to run. Edit conservatively: large changes can make the model re-plan and call the tool again.
reject never runs the tool and returns the reviewer’s message to the model as the tool result. Without a message the middleware sends a default text saying the tool was not executed and should not be retried. For a write action, say in the message whether the agent should give up, ask a follow-up or try something safer.
respond also skips the tool, but returns the message as a successful result. It is for "ask the user" tools, whose real implementation is a person’s answer, and must not be used to refuse an action, because the model will read it as the action having succeeded.

The decisions list holds one entry per pending action, in the order of the interrupt’s action_requests. A policy can also depend on the arguments: an InterruptOnConfig accepts a when predicate (a function of the ToolCallRequest that returns True to interrupt), so only some calls of a tool reach a person. The next run gives open_github_issue a predicate that skips drafts, adds an ask_user tool that only allows respond, and lets the model call both in one turn:

A when predicate and a respond decision

from langchain.agents.middleware import ToolCallRequest


@tool
def ask_user(question: str) -> str:
    """Ask the user a clarifying question."""
    raise RuntimeError("never executed: a human answers through the 'respond' decision")


def is_not_a_draft(request: ToolCallRequest) -> bool:
    return not request.tool_call["args"].get("title", "").startswith("[draft]")


agent = create_agent(
    scripted_model,    # one turn calling open_github_issue("[draft] typo") and ask_user("Which page?"), then "Thanks."
    tools=[open_github_issue, ask_user], checkpointer=InMemorySaver(),
    middleware=[HumanInTheLoopMiddleware(interrupt_on={
        "open_github_issue": {"allowed_decisions": ["approve", "edit", "reject"], "when": is_not_a_draft},
        "ask_user": {"allowed_decisions": ["respond"]},
    })])
cfg = {"configurable": {"thread_id": "t"}}
r = agent.invoke({"messages": [{"role": "user", "content": "go"}]}, cfg, version="v2")
req = r.interrupts[0].value
print([a["name"] for a in req["action_requests"]], [c["allowed_decisions"] for c in req["review_configs"]])
out = agent.invoke(Command(resume={"decisions": [{"type": "respond", "message": "vector-rag/hnsw.adoc"}]}), cfg, version="v2")
for m in out.value["messages"][2:]:
    print(type(m).__name__, "|", m.content)

Output

['ask_user'] [['respond']]
ToolMessage | vector-rag/hnsw.adoc
ToolMessage | Opened issue: [draft] typo
AIMessage | Thanks.

Only ask_user reached the reviewer: the draft issue failed the predicate, so it ran without asking and was never added to the interrupt batch, and there is exactly one decision to supply. The when predicate needs langchain 1.3.3 or later, and the installed 1.4.3 accepts it. See LangChain — Human-in-the-loop and LangChain — Built-in middleware.

Editing state while paused

A person can also change the state of a paused run, not just answer its question. get_state(config) returns a StateSnapshot whose values are the current state, whose next names the nodes waiting to run and whose interrupts hold the pending payloads. update_state(config, values) writes a new checkpoint, applying the values through the graph’s reducers as if a node had returned them. This is how a reviewer corrects a draft before resuming, or how an application edits the conversation from outside the graph:

Editing the draft while the run is paused at review

calls = {"draft": 0}


class S(TypedDict, total=False):
    question: str
    draft: str
    status: str


def draft(state: S):
    calls["draft"] += 1
    return {"draft": f"Docs gap: {state['question']} (draft #{calls['draft']})"}


def review(state: S):
    ok = interrupt({"draft": state["draft"]})
    return {"status": "filed" if ok else "discarded"}

# ... graph: START -> draft -> review -> END, compiled with InMemorySaver, as in the first section ...

cfg = {"configurable": {"thread_id": "edit"}}
g.invoke({"question": "ef_search"}, cfg)
snap = g.get_state(cfg)
print("next:", snap.next, "| interrupts:", [i.value for i in snap.interrupts])
print("draft before:", snap.values["draft"])

g.update_state(cfg, {"draft": "Docs gap: explain ef_search and its default (edited by a human)"})
print("draft after :", g.get_state(cfg).values["draft"])
print("next after update:", g.get_state(cfg).next)
out = g.invoke(Command(resume=True), cfg)
print("final:", out["status"], "| draft():", calls["draft"], "call(s)")
for s in g.get_state_history(cfg):
    print(s.metadata["step"], s.metadata["source"], s.next, s.values.get("draft"))

Output

next: ('review',) | interrupts: [{'draft': 'Docs gap: ef_search (draft #1)'}]
draft before: Docs gap: ef_search (draft #1)
draft after : Docs gap: explain ef_search and its default (edited by a human)
next after update: ('review',)
final: filed | draft(): 1 call(s)
3 loop () Docs gap: explain ef_search and its default (edited by a human)
2 update ('review',) Docs gap: explain ef_search and its default (edited by a human)
1 loop ('review',) Docs gap: ef_search (draft #1)
0 loop ('draft',) None
-1 input ('__start__',) None

update_state did not disturb the pending interrupt (next is still review) and the draft node did not run again, so the edited text is what review showed and filed. The history is the point: the edit is a checkpoint of its own with source update at step 2, next to the original at step 1, so an audit can see what the model wrote and what the human changed it to. update_state takes an as_node argument that says which node’s writers apply the values; LangGraph infers it from the checkpoint, and it has to be given explicitly only for parallel branches that finished in the same step, for a fresh thread with no history, or to skip nodes. See LangGraph — Use time-travel and LangGraph — Persistence.

Time travel: replay and fork

Every checkpoint of a thread stays in the checkpointer, and any of them can be the starting point of a new run. get_state_history(config) lists them, newest first; each snapshot’s config includes a checkpoint_id. There are two operations, and both re-execute the nodes after the chosen checkpoint (nodes before it are not run again):

Replay invokes the graph with None as input and the past snapshot’s config. The same state runs forward again, which is useful for retrying a failed or unlucky step.
Fork first calls update_state on the past snapshot’s config, which creates a new checkpoint that branches from it with different values, and then invokes with None and the config that update_state returned.

Replaying and forking DocsAssistant’s issue flow

runs = {"search": 0, "draft": 0}
# search -> draft -> review (interrupt) -> END, compiled with InMemorySaver

cfg = {"configurable": {"thread_id": "tt"}}
g.invoke({"question": "ef_search"}, cfg)
g.invoke(Command(resume=True), cfg)
print("first run:", runs)

history = list(g.get_state_history(cfg))            # newest first
for s in history:
    print(s.metadata["step"], s.next)
before_draft = next(s for s in history if s.next == ("draft",))

r = g.invoke(None, before_draft.config)             # replay
print("after replay:", runs, "| interrupted:", "__interrupt__" in r)

fork = g.update_state(before_draft.config, {"sections": ["pgvector.adoc#hnsw"]})
r = g.invoke(None, fork)                            # fork
print("after fork  :", runs, "| draft:", r["__interrupt__"][0].value["draft"])
print("checkpoints in the thread:", len(history), "->", len(list(g.get_state_history(cfg))))
print("original final checkpoint still holds:", g.get_state(history[0].config).values["status"])

Output

first run: {'search': 1, 'draft': 1}
3 ()
2 ('review',)
1 ('draft',)
0 ('search',)
-1 ('__start__',)
after replay: {'search': 1, 'draft': 2} | interrupted: True
after fork  : {'search': 1, 'draft': 3} | draft: Docs gap: ef_search -- see pgvector.adoc#hnsw
checkpoints in the thread: 5 -> 9
original final checkpoint still holds: filed

The counters show the semantics. search ran once in total, because the checkpoint chosen was after it, while draft ran again for the replay and again for the fork. Both runs then stopped at review, because the interrupt re-fires whenever a node containing it re-executes: time travel never reuses an old human answer for a node that runs again. The thread grew from 5 to 9 checkpoints and the original final checkpoint still says filed, so a fork does not roll anything back; it adds a branch.

flowchart LR I["-1 input"] --> S["0 search"] S --> D["1 draft"] D --> R["2 review (paused)"] R --> F["3 filed"] D -. "update_state: fork" .-> D2["draft, sections changed"] D2 --> R2["review (paused again)"] D -. "replay" .-> D3["draft again"] D3 --> R3["review (paused again)"]

Figure 2. Replay and fork branch from the same checkpoint and leave the original run intact

Because replay re-executes nodes, it also re-executes their side effects. That matters most for an agent whose tool already ran with approval. The next snippet uses the approval-gated agent from the previous section, with a docs_search tool that never interrupts, and goes back to the moment after the model proposed the issue:

Replaying a run whose tool was already approved

# run() is the gate loop of the last section wrapped in a function: search, propose, approve, file
agent, cfg, res = run(lambda a: {"type": "approve"})
print("approve ->", opened, "|", res.value["messages"][-1].content)

hist = list(agent.get_state_history(cfg))
proposed = next(s for s in hist if s.next == ("HumanInTheLoopMiddleware.after_model",)
                and s.values["messages"][-1].tool_calls
                and s.values["messages"][-1].tool_calls[0]["name"] == "open_github_issue")
print("checkpoint:", proposed.metadata["step"], proposed.next, "| issues opened so far:", opened)
again = agent.invoke(None, proposed.config, version="v2")
print("replay pauses again:", [a["name"] for a in again.interrupts[0].value["action_requests"]])
reject = {"type": "reject", "message": "Changed my mind."}
out = agent.invoke(Command(resume={"decisions": [reject]}), cfg, version="v2")   # thread config: latest checkpoint
print("opened after the rejected replay:", opened)
print(out.value["messages"][-2].content)

Output

approve -> ['Doc gap: ef_search'] | Filed.
checkpoint: 4 ('HumanInTheLoopMiddleware.after_model',) | issues opened so far: ['Doc gap: ef_search']
replay pauses again: ['open_github_issue']
opened after the rejected replay: ['Doc gap: ef_search']
User rejected the tool call for `open_github_issue` with reason: Changed my mind.

The replay asked for a decision again, so the middleware gate protects a replay as well as the first run; and the rejection was honoured, leaving one issue opened, the one from the original run. One detail is easy to get wrong. The resume above goes through the plain thread config, which addresses the newest checkpoint, the one where the replay paused. In a check that differed only in resuming with proposed.config (the old checkpoint’s config) again, the tool executed a second time despite the reject decision and the issue was opened twice, on two separate runs. Use the checkpoint config only to start a replay or a fork, and the thread config to answer the interrupt it produces.

Time travel also works below the top graph. A subgraph compiled without its own checkpointer is one super-step to the parent, so it can only be replayed from before it. Compiling it with checkpointer=True gives it its own history, and a fork can start between its nodes; get_state(config, subgraphs=True) exposes the subgraph’s config:

Forking between two interrupts inside a subgraph

sub = (StateGraph(State).add_node("ask_page", ask_page).add_node("ask_title", ask_title)
       .add_edge(START, "ask_page").add_edge("ask_page", "ask_title")
       .compile(checkpointer=True))                     # the subgraph keeps its own checkpoint history
graph = (StateGraph(State).add_node("form", sub).add_edge(START, "form")
         .compile(checkpointer=InMemorySaver()))
cfg = {"configurable": {"thread_id": "1"}}
graph.invoke({"value": []}, cfg)
graph.invoke(Command(resume="hnsw.adoc"), cfg)          # now paused at ask_title
sub_cfg = graph.get_state(cfg, subgraphs=True).tasks[0].state.config
fork = graph.update_state(sub_cfg, {"value": ["forked"]})
print(graph.invoke(None, fork)["__interrupt__"][0].value)   # ask_page result kept, ask_title asks again

Output

Issue title?

The fork resumed inside the subgraph and asked the second question again without repeating the first. Subgraphs as a design tool are covered in Multi-Agent & Deep Agents. See LangGraph — Use time-travel and LangGraph — Subgraphs.

Double texting

A user who sends a second message while the first is still being processed is double texting. In a bare LangGraph process you decide what happens (a lock per thread, a queue, cancelling the first task). The Agent Server, the runtime behind langgraph dev and LangSmith Deployment (see Deployment), builds the decision into the run API as multitask_strategy, and the LangChain documentation says the feature belongs to LangSmith Deployment and is not part of the open-source library. The four values:

Table 1. The double-texting strategies
`multitask_strategy`	What the server does with the second run	Use when
`enqueue` (default)	Waits until the current run finishes, then runs the second one after it, on top of the first run’s state.	Every message matters and order must be kept; the safe default.
`reject`	Refuses the second run; the API answers with an HTTP error and the first run continues.	The client can be told "busy" and a duplicate should never run, such as a submit button pressed twice.
`interrupt`	Halts the current run, keeps the progress it made, and runs the new input from that state.	The newest message supersedes the old one, as in a chat that can be corrected mid-answer. Partly executed tool calls must be handled by the graph.
`rollback`	Halts the current run, deletes it including its input, and runs the new input as a fresh run.	The first message was a mistake and should leave no trace in the thread.

They were run against a local Agent Server started with langgraph dev, serving a one-node graph (the node sleeps three seconds to stand in for a slow model call). The client is langgraph_sdk, and the second run is created half a second after the first:

The graph served by langgraph dev (langgraph.json maps docs_assistant to it)

import asyncio
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages


class State(TypedDict):
    messages: Annotated[list, add_messages]


async def answer(state: State):
    await asyncio.sleep(3)                       # stands in for a slow model call
    last = state["messages"][-1].content
    return {"messages": [{"role": "assistant", "content": f"echo: {last}"}]}


graph = StateGraph(State).add_node("answer", answer).add_edge(START, "answer").add_edge("answer", END).compile()

The client that double texts

import asyncio

import httpx
from langgraph_sdk import get_client

client = get_client(url="http://127.0.0.1:2199")
ASSISTANT = "docs_assistant"


def user(text):
    return {"messages": [{"role": "user", "content": text}]}


async def try_strategy(strategy):
    thread = await client.threads.create()
    tid = thread["thread_id"]
    first = await client.runs.create(tid, ASSISTANT, input=user("What is ef_search?"))
    await asyncio.sleep(0.5)                     # the first run is still executing
    try:
        second = await client.runs.create(
            tid, ASSISTANT, input=user("Actually, what is ef_construction?"),
            multitask_strategy=strategy)
    except httpx.HTTPStatusError as e:
        print(f"{strategy:9} second run refused: HTTP {e.response.status_code}")
        await client.runs.join(tid, first["run_id"])
    else:
        await client.runs.join(tid, second["run_id"])
    await asyncio.sleep(0.3)
    runs = await client.runs.list(tid)
    state = await client.threads.get_state(tid)
    texts = [m["content"] for m in state["values"]["messages"]]
    print(f"{strategy:9} runs={sorted(r['status'] for r in runs)} messages={texts}")


async def main():
    for s in ["enqueue", "reject", "interrupt", "rollback"]:
        await try_strategy(s)

asyncio.run(main())

Output

enqueue   runs=['success', 'success'] messages=['What is ef_search?', 'echo: What is ef_search?', 'Actually, what is ef_construction?', 'echo: Actually, what is ef_construction?']
reject    second run refused: HTTP 409
reject    runs=['success'] messages=['What is ef_search?', 'echo: What is ef_search?']
interrupt runs=['interrupted', 'success'] messages=['Actually, what is ef_construction?', 'echo: Actually, what is ef_construction?']
rollback  runs=['success'] messages=['Actually, what is ef_construction?', 'echo: Actually, what is ef_construction?']

The four lines differ exactly as the table says. enqueue answered both questions in order; reject answered only the first and told the client with a 409; interrupt left the first run with status interrupted; rollback deleted it and kept a single successful run. In this small check the interrupted first run left no messages in the thread, because its only node had not finished when it was halted, so with interrupt the new input started from an essentially empty state; a graph with several steps keeps the completed ones. Choosing a strategy is a product decision: chat UIs usually pick enqueue or interrupt, and anything that triggers a side effect should pair interrupt or rollback with the idempotency rules above. See LangSmith — Double texting and its how-to guides for enqueue, reject, interrupt and rollback.

DocsAssistant’s approval before opening an issue

Everything above meets in one agent: DocsAssistant may search freely but must get a person’s approval before it writes to GitHub. The tool set is docs_search, which is read-only and listed as False in interrupt_on, and open_github_issue, which is gated. The reviewer is a function here; in a product it is a web form or a chat message, and the loop is the same: run, read interrupts, collect decisions, resume with the same thread_id:

The approval gate around open_github_issue

import os

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

REPO = "albertoirurueta/docs"          # the only repository this tool may write to


@tool
def docs_search(query: str) -> str:
    """Search the documentation and return matching sections."""
    return "no section explains ef_search"


@tool
def open_github_issue(title: str, body: str) -> str:
    """Open an issue in the documentation repository. Destructive: creates public content."""
    token = os.environ["DOCS_ISSUES_TOKEN"]     # fine-grained, Issues write on REPO only
    # ... POST https://api.github.com/repos/{REPO}/issues with the token ...
    return f"Opened issue in {REPO}: {title}"


def review(interrupt_payload, reviewer):
    """Turn each pending action into a decision; `reviewer` is the UI (here: a function)."""
    return {"decisions": [reviewer(a) for a in interrupt_payload["action_requests"]]}


agent = create_agent(
    model, tools=[docs_search, open_github_issue], checkpointer=InMemorySaver(),
    middleware=[HumanInTheLoopMiddleware(interrupt_on={
        "docs_search": False,
        "open_github_issue": {"allowed_decisions": ["approve", "edit", "reject"]}})])
cfg = {"configurable": {"thread_id": "gate"}}
res = agent.invoke({"messages": [{"role": "user", "content": "ef_search is undocumented"}]}, cfg, version="v2")
while res.interrupts:
    res = agent.invoke(Command(resume=review(res.interrupts[0].value, reviewer)), cfg, version="v2")

With a reviewer that approves everything, the scripted model searched, proposed the issue and filed it after the one pause (approve → ['Doc gap: ef_search'] | Filed. in the replay snippet above), and the reject and edit runs in the previous section show the other outcomes. A few properties make this safe enough to ship:

The gate is enforced in the runtime, not in the prompt: the model cannot skip it by phrasing.
The credential lives in the tool, read from an environment variable when the tool runs, so the pause happens before the token is even loaded, and it grants nothing but issue writes on one repository.
Rejections and edits are recorded in the thread, so a later audit can see who changed what; the reviewer’s identity belongs in the run metadata or your own audit log.
Replays and forks hit the same gate, and resuming a replay must use the thread config, as shown above.
The gate applies per tool call, so a tool that can do several kinds of damage should be split into narrow tools, as in Tools.

The surrounding agent is the one from Agents with create_agent; how reviewers are shown a pending action and how autonomy is loosened over time is in Human-in-the-Loop. The JavaScript package @langchain/langgraph exports the same primitives (interrupt, Command) in LangChain.js, covered on LangChain.js, and the front-end side is on the Streaming page. See LangChain — Human-in-the-loop.

Summary

Table 2. Which mechanism for which need
Need	Mechanism	Remember
Ask a person anything, anywhere in a node	`interrupt(payload)` and `Command(resume=value)`	Same `thread_id`; the node restarts, so code before `interrupt()` must be idempotent.
Gate dangerous tools in an agent	`HumanInTheLoopMiddleware` with `interrupt_on`	`approve`, `edit`, `reject` for actions, `respond` only for ask-the-user tools; one decision per action, in order.
Only ask for some calls	`when` predicate in the tool’s config	Non-matching calls run without asking.
Debug one node at a time	`interrupt_before` / `interrupt_after`	No payload, no resume value; not the way to approve tool calls in 1.x.
Fix the state before continuing	`get_state` and `update_state`	The edit is a new checkpoint; `as_node` only when LangGraph cannot infer it.
Retry from the past	`get_state_history`, then `invoke(None, snapshot.config)`	Nodes after the checkpoint run again, including interrupts and side effects.
Try an alternative	`update_state(snapshot.config, values)`, then `invoke(None, fork_config)`	A branch, not a rollback; answer new interrupts through the thread config.
A second message during a run	`multitask_strategy`: `enqueue`, `reject`, `interrupt`, `rollback`	Agent Server feature, not in open-source LangGraph.

Human-in-the-Loop & Time Travel

interrupt() and Command(resume=…​)

How a run stops and continues

Static breakpoints and the older interrupt_before

The rules of interrupts

HumanInTheLoopMiddleware

Editing state while paused

Time travel: replay and fork

Double texting

DocsAssistant’s approval before opening an issue

Summary

References

interrupt() and Command(resume=…)