State Update After Interrupt Not Reaching Subgraph Node on Resume

[wicusverhoef]

LangGraph Nested Graph State Update Clarification Q

This example shows that when resuming a nested LangGraph after a NodeInterrupt, state updates made via update_state do not propagate into the subgraph where the resume occurs. As a result, the resumed node does not see the updated state, raising questions about intended behaviour and best practices for state injection in nested workflows.
Or I am just being very daft.

Context

This minimal example explores LangGraph's checkpointing and state update mechanism for nested graphs with human-in-the-loop-like interruptions.

Scenario

• We use LangGraph with nested subgraphs and HIL nodes: nodes that interrupt and require human input.
• The system is stateless between invocations (e.g., GCP Cloud Run), so the graph and its state must be persisted and reconstituted.
• The goal is to allow a human or external process to update the graph's state after an interrupt, then resume execution at the exact point of interruption, with the updated state visible to the resumed node - even if deep inside a subgraph.

What This Example Shows

• The graph is three layers deep: outer graph → mid subgraph → deep subgraph. The latter has a HIL node.
• On first run, the HIL node interrupts as expected.
• We then call update_state on the graph, setting a flag (resume_flag=True) and a message, intending for these to be visible to the HIL node on resume. Printing the state at this stage shows the update.
• On resume, the correct HIL node is invoked again, but the updated state does not appear to be visible; the node sees only the original state, not the update.

Request for Clarification

• According to LangGraph documentation https://langchain-ai.github.io/langgraph/concepts/persistence/ and the human-in-the-loop time-travel example, it seems possible to update the state after an interrupt and have the update propagate to the resumed node, even in a nested subgraph.
• In this example here, the state update appears to only affect the outer graph's checkpoint, not the subgraph's context where the resume actually happens.
• Is this the intended behaviour? If so, what is the recommended way to inject new state into a subgraph after an interrupt?

Key Questions for the Community

• Is there a supported way to update the state so that the resumed node sees the update?
• Are there workarounds or best practices for propagating state updates into subgraphs after an interrupt?

Minimal Example

• This file is a minimal, self-contained reproduction of the issue.
• To reproduce:
  1. Run the script. The HIL node will interrupt.
  2. The script updates the state and resumes.
  3. Observe whether the HIL node sees the updated state on resume.

---

Any insight or clarification would be greatly appreciated!

• I am using langgraph==0.4.5

from langgraph.graph import StateGraph, END
from typing import TypedDict, Optional
import datetime
from langgraph.errors import NodeInterrupt
from langgraph.checkpoint.memory import MemorySaver

# --- Minimal Node ---
class PrintNode:
    def __init__(self, node_id: str, message: str):
        self.node_id = node_id
        self.message = message
    def run(self, state: dict) -> dict:
        print(f"[Node {self.node_id}] State in: {state}")
        return {"history": state.get("history", []) + [self.message]}

class HITLNode:
    def __init__(self, node_id: str, message: str = "HIL required"):
        self.node_id = node_id
        self.message = message
    def run(self, state: dict) -> dict:
        print(f"[HITLNode {self.node_id}] State in: {state}")
        if not state.get("resume_flag"):
            print(f"[HITLNode {self.node_id}] Raising NodeInterrupt!")
            raise NodeInterrupt(self.message)
        print(f"[HITLNode {self.node_id}] Resumed successfully!")
        return {"history": state.get("history", []) + [f"{self.node_id} resumed"]}

# --- State Type ---
class State(TypedDict):
    history: list[str]
    resume_flag: Optional[bool]
    printable_message: Optional[str]
# --- Build Fractal Graph ---
def build_fractal_graph():
    # Deepest subgraph
    deep_node1 = PrintNode("deep1", "deep1 done")
    deep_node2 = HITLNode("deep2")
    deep_graph = StateGraph(State)
    deep_graph.set_entry_point("deep1")
    deep_graph.add_node("deep1", deep_node1.run)
    deep_graph.add_node("deep2", deep_node2.run)
    deep_graph.add_edge("deep1", "deep2")
    deep_graph.add_edge("deep2", END)
    compiled_deep = deep_graph.compile(checkpointer=True) # Have found that without this checkpointer=True in the subgraphs, the resume does not happen at the nested node level

    # Middle subgraph
    mid_node = PrintNode("mid", "mid done")
    mid_graph = StateGraph(State)
    mid_graph.set_entry_point("mid")
    mid_graph.add_node("mid", mid_node.run)
    mid_graph.add_node("deep_subgraph", compiled_deep)
    mid_graph.add_edge("mid", "deep_subgraph")
    mid_graph.add_edge("deep_subgraph", END)
    compiled_mid = mid_graph.compile(checkpointer=True) 

    # Outer graph
    outer_node = PrintNode("outer", "outer done")
    outer_graph = StateGraph(State)
    outer_graph.set_entry_point("outer")
    outer_graph.add_node("outer", outer_node.run)
    outer_graph.add_node("mid_subgraph", compiled_mid)
    outer_graph.add_edge("outer", "mid_subgraph")
    outer_graph.add_edge("mid_subgraph", END)
    memory = MemorySaver()
    compiled_outer = outer_graph.compile(checkpointer=memory)

    return compiled_outer, memory

# --- Main ---
if __name__ == "__main__":
    graph, memory = build_fractal_graph()
    thread_id = "thread_" + datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S")
    config = {"configurable": {"thread_id": thread_id}}
    start_state = {"history": []}

    print("\n=== Initial run (expect interrupt) ===")
    result = graph.invoke(start_state, config)
    print("\nFinal state:", result)

    # Check for interrupts in the latest checkpoint
    snapshot = graph.get_state(config)
    interrupts = getattr(snapshot, 'interrupts', None)
    if interrupts:
        print("\n=== INTERRUPT DETECTED in checkpoint ===")
        print(interrupts)
    else:
        print("\nNo interrupt detected in checkpoint.")

    print("\n=== State history after interrupt ===")
    for snap in graph.get_state_history(config):
        print(snap)

    # Simulate human input/state update and resume
    print("\n=== Update state (set resume_flag=True + printable_message) and resume ===")
    graph.update_state(config, {"resume_flag": True, "printable_message": "If this is visible, the state update propagated into the subgraph."})
    snap = graph.get_state(config)
    print("After update_state, state is:", snap.values)
    # Resume from updated checkpoint
    result = graph.invoke(None, config)
    print("\nFinal state after resume:", result)
    print("\n=== State history after resume ===")
    for snap in graph.get_state_history(config):
        print(snap)

[pablgave]

Hey @wicusverhoef! Just wanted to know if your doubt was ever solved, cause I just arrived to a similar situation using Langgraph: After an interrupt in a subgraph node, when sending the command, the state only has the original attributes of the parent graph ;(, and "erases" the extra ones it got from the subgraph.

[az-ihsan]

I am currently also stuck in this similar problem, can anyone help me too?

[rahul-langchain]

Hi @wicusverhoef

The unexpected behavior in your example is because you are using the update_state function to bypass the interrupt. That function is part of our time travel feature for debugging graphs, and has broader effects on the graph run that make it unsuitable for a human-in-the-loop workflow. I recommend you instead use the interrupt() function, and pass the graph a Command when resuming it, as documented here.

Here is a rewritten version of your code that works as expected:

import datetime
from typing import TypedDict, Optional

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


# --- Minimal Node ---
class PrintNode:
    def __init__(self, node_id: str, message: str):
        self.node_id = node_id
        self.message = message
    def run(self, state: dict) -> dict:
        print(f"[Node {self.node_id}] State in: {state}")
        return {"history": state.get("history", []) + [self.message]}

class HITLNode:
    def __init__(self, node_id: str, message: str = "HIL required"):
        self.node_id = node_id
        self.message = message
    def run(self, state: dict) -> dict:
        print(f"[HITLNode {self.node_id}] State in: {state}")
        print(f"[HITLNode {self.node_id}] Raising interrupt!")
        # Use interrupt() function here
        _ = interrupt(self.message)
        print(f"[HITLNode {self.node_id}] Resumed successfully!")
        return {"history": state.get("history", []) + [f"{self.node_id} resumed"]}

# --- State Type ---
class State(TypedDict):
    history: list[str]
    printable_message: Optional[str]
# --- Build Fractal Graph ---
def build_fractal_graph():
    # Deepest subgraph
    deep_node1 = PrintNode("deep1", "deep1 done")
    deep_node2 = HITLNode("deep2")
    deep_graph = StateGraph(State)
    deep_graph.set_entry_point("deep1")
    deep_graph.add_node("deep1", deep_node1.run)
    deep_graph.add_node("deep2", deep_node2.run)
    deep_graph.add_edge("deep1", "deep2")
    deep_graph.add_edge("deep2", END)
    compiled_deep = deep_graph.compile(checkpointer=True) # Have found that without this checkpointer=True in the subgraphs, the resume does not happen at the nested node level

    # Middle subgraph
    mid_node = PrintNode("mid", "mid done")
    mid_graph = StateGraph(State)
    mid_graph.set_entry_point("mid")
    mid_graph.add_node("mid", mid_node.run)
    mid_graph.add_node("deep_subgraph", compiled_deep)
    mid_graph.add_edge("mid", "deep_subgraph")
    mid_graph.add_edge("deep_subgraph", END)
    compiled_mid = mid_graph.compile(checkpointer=True)

    # Outer graph
    outer_node = PrintNode("outer", "outer done")
    outer_graph = StateGraph(State)
    outer_graph.set_entry_point("outer")
    outer_graph.add_node("outer", outer_node.run)
    outer_graph.add_node("mid_subgraph", compiled_mid)
    outer_graph.add_edge("outer", "mid_subgraph")
    outer_graph.add_edge("mid_subgraph", END)
    memory = MemorySaver()
    compiled_outer = outer_graph.compile(checkpointer=memory)

    return compiled_outer, memory

# --- Main ---
if __name__ == "__main__":
    graph, memory = build_fractal_graph()
    thread_id = "thread_" + datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S")
    config = {"configurable": {"thread_id": thread_id}}
    start_state = {"history": []}

    print("\n=== Initial run (expect interrupt) ===")
    result = graph.invoke(start_state, config)
    print("\nFinal state:", result)

    # Check for interrupts in the latest checkpoint
    snapshot = graph.get_state(config)
    interrupts = getattr(snapshot, 'interrupts', None)
    if interrupts:
        print("\n=== INTERRUPT DETECTED in checkpoint ===")
        print(interrupts)
    else:
        print("\nNo interrupt detected in checkpoint.")

    print("\n=== State history after interrupt ===")
    for snap in graph.get_state_history(config):
        print(snap)

    # Resume from updated checkpoint by passing Command object
    result = graph.invoke(Command(resume=True), config)
    print("\nFinal state after resume:", result)
    print("\n=== State history after resume ===")
    for snap in graph.get_state_history(config):
        print(snap)