langchain-ai/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-07-15 12:35:43 -04:00
Code Issues 314 Packages Projects Releases 17 Wiki Activity

Interrupt resume values misrouted between tools when using a ToolNode #1079

New Issue
Closed
opened 2026-02-20 17:42:59 -05:00 by yindo · 2 comments
yindo commented 2026-02-20 17:42:59 -05:00
Owner
Copy Link
Copy Source

Originally created by @d4n-a on GitHub (Dec 3, 2025).

Checked other resources

  • This is a bug, not a usage question. For questions, please use the LangChain Forum (https://forum.langchain.com/).
  • I added a clear and detailed title that summarizes the issue.
  • I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example).
  • I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue.

Example Code

import asyncio
import logging
import math
import random
from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.types import Command, interrupt

logger = logging.getLogger(__name__)


@tool
async def tool_a():
    """The tool raises Interrupt A and expects that exact value in order to continue"""
    expected = "Interrupt A"
    await delay()
    result = interrupt(expected)
    assert expected == result, f"Expected {expected}, got {result}"
    return result


@tool
async def tool_b():
    """The tool raises Interrupt B and expects that exact value in order to continue"""
    expected = "Interrupt B"
    await delay()
    result = interrupt(expected)
    assert expected == result, f"Expected {expected}, got {result}"
    return result


async def parallel_interrupts():
    """Resubmitting interrupt messages back to confirm they are channeled correctly"""
    graph = make_graph()
    config = {"configurable": {"thread_id": "test-thread-1"}}

    # First invocation to trigger both tools in parallel
    first_result = await graph.ainvoke(
        {"messages": [HumanMessage(content="Tool calls")]}, config=config, stream_mode="debug"
    )

    first_interrupt = first_result[-1]["payload"]["interrupts"][0]
    first_interrupt_id = first_interrupt["id"]
    first_interrupt_value = first_interrupt["value"]
    logger.info(f"Resuming first interrupt with id {first_interrupt_id} and value {first_interrupt_value}")

    second_result = await graph.ainvoke(
        Command(resume={first_interrupt_id: first_interrupt_value}), config=config, stream_mode="debug"
    )

    # Resume second interrupt
    second_interrupt = second_result[-1]["payload"]["interrupts"][0]
    second_interrupt_id = second_interrupt["id"]
    second_interrupt_value = second_interrupt["value"]
    logger.info(f"Resuming second interrupt with id {second_interrupt_id} and value {second_interrupt_value}")

    _ = await graph.ainvoke(
        Command(resume={second_interrupt["id"]: second_interrupt["value"]}), config=config, stream_mode="debug"
    )

    assert first_interrupt_id == second_interrupt_id  # never fails


async def test_main():
    num_iterations = 100  # repeat multiple times to trigger race condition
    for i in range(num_iterations):
        await parallel_interrupts()


async def delay():
    """
    Simulate async work with random delays and CPU-bound tasks to force the event loop switch coroutines.
    """
    await asyncio.sleep(random.uniform(0.0, 0.05))

    for _ in range(random.randint(3, 10)):
        x = 0
        for i in range(1000):
            x += math.sin(i + i**2)
        await asyncio.sleep(0)

    await asyncio.sleep(random.uniform(0.0, 0.05))


def agent_node(*args, **kwargs):
    return {
        "messages": [
            AIMessage(
                content="Calling both tools in parallel",
                tool_calls=[
                    {"name": "tool_a", "args": {}, "id": "call_a", "type": "tool_call"},
                    {"name": "tool_b", "args": {}, "id": "call_b", "type": "tool_call"},
                ],
            )
        ]
    }


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


def should_continue(state: State):
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END


def make_graph():
    graph = StateGraph(State)
    graph.add_node("agent", agent_node)
    graph.add_node("tools", ToolNode([tool_a, tool_b], handle_tool_errors=False))
    graph.add_edge(START, "agent")
    graph.add_conditional_edges("agent", should_continue, ["tools", END])
    graph.add_edge("tools", END)
    memory = MemorySaver()  # for interrupt support
    app = graph.compile(checkpointer=memory)
    return app


if __name__ == "__main__":
    asyncio.run(test_main())

Error Message and Stack Trace (if applicable)

Description

When using ToolNode from langgraph-prebuilt with multiple tools that each call interrupt(), the resume values are sometimes routed to the wrong tool.

Specifically, the resume value intended for the first tool's interrupt can be delivered to the second tool instead, and vice versa.

Take a look at the attached MRE to reproduce the issue. It's expected that each tool receives the interrupt value it initially raised back. Actual behavior causes assertions to fail when interrupt values are misrouted.

The issue appears to be a race condition and typically fails within the first few iterations. Both interrupts are assigned the same ID, which is used to pass the resume value back to the interrupt.

Is there a workaround to use ToolNode with interrupts?

System Info

System Information

OS: Darwin
OS Version: Darwin Kernel Version 24.6.0: Wed Oct 15 21:12:05 PDT 2025; root:xnu-11417.140.69.703.14~1/RELEASE_ARM64_T6030
Python Version: 3.11.12 (main, Apr 8 2025, 14:15:29) [Clang 16.0.0 (clang-1600.0.26.6)]

Package Information

langchain_core: 1.1.0
langchain: 1.1.0
langsmith: 0.4.12
langchain_google_vertexai: 2.1.2
langchain_text_splitters: 0.3.9

Other Dependencies

langgraph: 1.0.4
langgraph-prebuilt: 1.0.5
pydantic: 2.11.7
pytest: 8.3.3
typing-extensions: 4.14.1

Originally created by @d4n-a on GitHub (Dec 3, 2025). ### Checked other resources - [x] This is a bug, not a usage question. For questions, please use the LangChain Forum (https://forum.langchain.com/). - [x] I added a clear and detailed title that summarizes the issue. - [x] I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example). - [x] I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue. ### Example Code ```python import asyncio import logging import math import random from typing import Annotated, TypedDict from langchain_core.messages import AIMessage, HumanMessage from langchain_core.tools import tool from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import END, START, StateGraph from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode from langgraph.types import Command, interrupt logger = logging.getLogger(__name__) @tool async def tool_a(): """The tool raises Interrupt A and expects that exact value in order to continue""" expected = "Interrupt A" await delay() result = interrupt(expected) assert expected == result, f"Expected {expected}, got {result}" return result @tool async def tool_b(): """The tool raises Interrupt B and expects that exact value in order to continue""" expected = "Interrupt B" await delay() result = interrupt(expected) assert expected == result, f"Expected {expected}, got {result}" return result async def parallel_interrupts(): """Resubmitting interrupt messages back to confirm they are channeled correctly""" graph = make_graph() config = {"configurable": {"thread_id": "test-thread-1"}} # First invocation to trigger both tools in parallel first_result = await graph.ainvoke( {"messages": [HumanMessage(content="Tool calls")]}, config=config, stream_mode="debug" ) first_interrupt = first_result[-1]["payload"]["interrupts"][0] first_interrupt_id = first_interrupt["id"] first_interrupt_value = first_interrupt["value"] logger.info(f"Resuming first interrupt with id {first_interrupt_id} and value {first_interrupt_value}") second_result = await graph.ainvoke( Command(resume={first_interrupt_id: first_interrupt_value}), config=config, stream_mode="debug" ) # Resume second interrupt second_interrupt = second_result[-1]["payload"]["interrupts"][0] second_interrupt_id = second_interrupt["id"] second_interrupt_value = second_interrupt["value"] logger.info(f"Resuming second interrupt with id {second_interrupt_id} and value {second_interrupt_value}") _ = await graph.ainvoke( Command(resume={second_interrupt["id"]: second_interrupt["value"]}), config=config, stream_mode="debug" ) assert first_interrupt_id == second_interrupt_id # never fails async def test_main(): num_iterations = 100 # repeat multiple times to trigger race condition for i in range(num_iterations): await parallel_interrupts() async def delay(): """ Simulate async work with random delays and CPU-bound tasks to force the event loop switch coroutines. """ await asyncio.sleep(random.uniform(0.0, 0.05)) for _ in range(random.randint(3, 10)): x = 0 for i in range(1000): x += math.sin(i + i**2) await asyncio.sleep(0) await asyncio.sleep(random.uniform(0.0, 0.05)) def agent_node(*args, **kwargs): return { "messages": [ AIMessage( content="Calling both tools in parallel", tool_calls=[ {"name": "tool_a", "args": {}, "id": "call_a", "type": "tool_call"}, {"name": "tool_b", "args": {}, "id": "call_b", "type": "tool_call"}, ], ) ] } class State(TypedDict): messages: Annotated[list, add_messages] def should_continue(state: State): last_message = state["messages"][-1] if hasattr(last_message, "tool_calls") and last_message.tool_calls: return "tools" return END def make_graph(): graph = StateGraph(State) graph.add_node("agent", agent_node) graph.add_node("tools", ToolNode([tool_a, tool_b], handle_tool_errors=False)) graph.add_edge(START, "agent") graph.add_conditional_edges("agent", should_continue, ["tools", END]) graph.add_edge("tools", END) memory = MemorySaver() # for interrupt support app = graph.compile(checkpointer=memory) return app if __name__ == "__main__": asyncio.run(test_main()) ``` ### Error Message and Stack Trace (if applicable) ```shell ``` ### Description When using `ToolNode` from [langgraph-prebuilt](https://github.com/langchain-ai/langgraph/blob/7f08a6dafda133f0b4db9d169b0445a3dee7b466/libs/prebuilt/langgraph/prebuilt/tool_node.py#L610) with multiple tools that each call `interrupt()`, the resume values are sometimes routed to the wrong tool. Specifically, the resume value intended for the first tool's interrupt can be delivered to the second tool instead, and vice versa. Take a look at the attached MRE to reproduce the issue. It's expected that each tool receives the interrupt value it initially raised back. Actual behavior causes assertions to fail when interrupt values are misrouted. The issue appears to be a race condition and typically fails within the first few iterations. Both interrupts are assigned the same ID, which is used to pass the resume value back to the interrupt. Is there a workaround to use ToolNode with interrupts? ### System Info System Information ------------------ > OS: Darwin > OS Version: Darwin Kernel Version 24.6.0: Wed Oct 15 21:12:05 PDT 2025; root:xnu-11417.140.69.703.14~1/RELEASE_ARM64_T6030 > Python Version: 3.11.12 (main, Apr 8 2025, 14:15:29) [Clang 16.0.0 (clang-1600.0.26.6)] Package Information ------------------- > langchain_core: 1.1.0 > langchain: 1.1.0 > langsmith: 0.4.12 > langchain_google_vertexai: 2.1.2 > langchain_text_splitters: 0.3.9 Other Dependencies ------------------ > langgraph: 1.0.4 > langgraph-prebuilt: 1.0.5 > pydantic: 2.11.7 > pytest: 8.3.3 > typing-extensions: 4.14.1
yindo added the bugpending labels 2026-02-20 17:42:59 -05:00
yindo closed this issue 2026-02-20 17:42:59 -05:00
yindo commented 2026-02-20 17:42:59 -05:00
Author
Owner
Copy Link
Copy Source

@sydney-runkle commented on GitHub (Jan 12, 2026):

Hiya!

Thanks for your question. If you'd like to use tools w/ interrupts, we recommend using Send to trigger the tool node w/ each of your tool calls.

Here's a minimalistic solution. This is roughly we do inside of create_agent, and why create_react_agent introduced a version arg

"""
Solution: Use Send to dispatch each tool call as a separate task.

When each tool call is dispatched via Send, it becomes its own PUSH task
with a unique checkpoint namespace, resulting in unique interrupt IDs.
"""

import asyncio
from typing import Annotated
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph, add_messages
from langgraph.prebuilt import ToolNode
from langgraph.prebuilt.tool_node import ToolCallWithContext
from langgraph.types import Command, Send, interrupt
from typing_extensions import TypedDict


@tool
def tool_a(x: str) -> str:
    """Tool A."""
    response = interrupt({"tool": "A", "x": x})
    return f"Tool A completed with: {response}"


@tool
def tool_b(x: str) -> str:
    """Tool B."""
    response = interrupt({"tool": "B", "x": x})
    return f"Tool B completed with: {response}"


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


def agent(state: State):
    """Returns 2 parallel tool calls."""
    return {
        "messages": [
            AIMessage(
                content="Calling tools",
                tool_calls=[
                    {"id": "call_a", "name": "tool_a", "args": {"x": "1"}},
                    {"id": "call_b", "name": "tool_b", "args": {"x": "2"}},
                ],
            )
        ]
    }


def route(state: State) -> list[Send] | str:
    """Route using Send to dispatch each tool call as a separate task."""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        # Check if we already have tool messages for these calls
        tool_message_ids = {
            m.tool_call_id for m in state["messages"] if isinstance(m, ToolMessage)
        }
        pending_calls = [
            c for c in last_message.tool_calls if c["id"] not in tool_message_ids
        ]
        if pending_calls:
            # Dispatch each tool call as a SEPARATE task via Send
            return [
                Send(
                    "tools",
                    ToolCallWithContext(
                        __type="tool_call_with_context",
                        tool_call=call,
                        state=state,
                    ),
                )
                for call in pending_calls
            ]
    return END


async def main():
    graph = StateGraph(State)
    graph.add_node("agent", agent)
    graph.add_node("tools", ToolNode([tool_a, tool_b]))
    graph.add_edge(START, "agent")
    graph.add_conditional_edges("agent", route, ["tools", END])
    graph.add_edge("tools", "agent")

    app = graph.compile(checkpointer=MemorySaver())
    config = {"configurable": {"thread_id": "1"}}

    # First invocation triggers interrupts
    print("=" * 60)
    print("Solution: Using Send for parallel tool calls")
    print("=" * 60)

    result = await app.ainvoke({"messages": [HumanMessage("go")]}, config)

    # Check interrupts
    state = await app.aget_state(config)
    interrupts = [i for t in state.tasks for i in (t.interrupts or [])]

    print(f"\nInterrupts: {len(interrupts)}")
    for i, intr in enumerate(interrupts):
        print(f"  {i}: id={intr.id[:16]}... tool={intr.value.get('tool')}")

    unique_ids = {i.id for i in interrupts}
    print(f"\nUnique IDs: {len(unique_ids)} (expected: 2)")

    if len(unique_ids) == len(interrupts):
        print("\nSUCCESS: Each interrupt has a unique ID!")

        # Build resume dict - each tool gets its own response
        interrupt_by_tool = {i.value["tool"]: i for i in interrupts}
        resume = {
            interrupt_by_tool["A"].id: "approved_A",
            interrupt_by_tool["B"].id: "approved_B",
        }
        print(f"\nResume dict: {len(resume)} entries")

        # Resume all interrupts
        print("\n--- Resuming ---")
        result = await app.ainvoke(Command(resume=resume), config)

        # Check final state
        tool_messages = [m for m in result["messages"] if isinstance(m, ToolMessage)]
        print(f"\nTool messages: {len(tool_messages)}")
        for tm in tool_messages:
            print(f"  {tm.name}: {tm.content}")

        # Verify each tool got its correct response
        tool_a_msg = next((m for m in tool_messages if m.name == "tool_a"), None)
        tool_b_msg = next((m for m in tool_messages if m.name == "tool_b"), None)

        if tool_a_msg and "approved_A" in tool_a_msg.content:
            print("\nTool A got correct response!")
        if tool_b_msg and "approved_B" in tool_b_msg.content:
            print("Tool B got correct response!")
    else:
        print("\nFAILED: IDs are not unique")


if __name__ == "__main__":
    asyncio.run(main())
@sydney-runkle commented on GitHub (Jan 12, 2026): Hiya! Thanks for your question. If you'd like to use tools w/ interrupts, we recommend using `Send` to trigger the tool node w/ each of your tool calls. Here's a minimalistic solution. This is roughly we do inside of `create_agent`, and why `create_react_agent` introduced a version arg ```py """ Solution: Use Send to dispatch each tool call as a separate task. When each tool call is dispatched via Send, it becomes its own PUSH task with a unique checkpoint namespace, resulting in unique interrupt IDs. """ import asyncio from typing import Annotated from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage from langchain_core.tools import tool from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import END, START, StateGraph, add_messages from langgraph.prebuilt import ToolNode from langgraph.prebuilt.tool_node import ToolCallWithContext from langgraph.types import Command, Send, interrupt from typing_extensions import TypedDict @tool def tool_a(x: str) -> str: """Tool A.""" response = interrupt({"tool": "A", "x": x}) return f"Tool A completed with: {response}" @tool def tool_b(x: str) -> str: """Tool B.""" response = interrupt({"tool": "B", "x": x}) return f"Tool B completed with: {response}" class State(TypedDict): messages: Annotated[list[AnyMessage], add_messages] def agent(state: State): """Returns 2 parallel tool calls.""" return { "messages": [ AIMessage( content="Calling tools", tool_calls=[ {"id": "call_a", "name": "tool_a", "args": {"x": "1"}}, {"id": "call_b", "name": "tool_b", "args": {"x": "2"}}, ], ) ] } def route(state: State) -> list[Send] | str: """Route using Send to dispatch each tool call as a separate task.""" last_message = state["messages"][-1] if hasattr(last_message, "tool_calls") and last_message.tool_calls: # Check if we already have tool messages for these calls tool_message_ids = { m.tool_call_id for m in state["messages"] if isinstance(m, ToolMessage) } pending_calls = [ c for c in last_message.tool_calls if c["id"] not in tool_message_ids ] if pending_calls: # Dispatch each tool call as a SEPARATE task via Send return [ Send( "tools", ToolCallWithContext( __type="tool_call_with_context", tool_call=call, state=state, ), ) for call in pending_calls ] return END async def main(): graph = StateGraph(State) graph.add_node("agent", agent) graph.add_node("tools", ToolNode([tool_a, tool_b])) graph.add_edge(START, "agent") graph.add_conditional_edges("agent", route, ["tools", END]) graph.add_edge("tools", "agent") app = graph.compile(checkpointer=MemorySaver()) config = {"configurable": {"thread_id": "1"}} # First invocation triggers interrupts print("=" * 60) print("Solution: Using Send for parallel tool calls") print("=" * 60) result = await app.ainvoke({"messages": [HumanMessage("go")]}, config) # Check interrupts state = await app.aget_state(config) interrupts = [i for t in state.tasks for i in (t.interrupts or [])] print(f"\nInterrupts: {len(interrupts)}") for i, intr in enumerate(interrupts): print(f" {i}: id={intr.id[:16]}... tool={intr.value.get('tool')}") unique_ids = {i.id for i in interrupts} print(f"\nUnique IDs: {len(unique_ids)} (expected: 2)") if len(unique_ids) == len(interrupts): print("\nSUCCESS: Each interrupt has a unique ID!") # Build resume dict - each tool gets its own response interrupt_by_tool = {i.value["tool"]: i for i in interrupts} resume = { interrupt_by_tool["A"].id: "approved_A", interrupt_by_tool["B"].id: "approved_B", } print(f"\nResume dict: {len(resume)} entries") # Resume all interrupts print("\n--- Resuming ---") result = await app.ainvoke(Command(resume=resume), config) # Check final state tool_messages = [m for m in result["messages"] if isinstance(m, ToolMessage)] print(f"\nTool messages: {len(tool_messages)}") for tm in tool_messages: print(f" {tm.name}: {tm.content}") # Verify each tool got its correct response tool_a_msg = next((m for m in tool_messages if m.name == "tool_a"), None) tool_b_msg = next((m for m in tool_messages if m.name == "tool_b"), None) if tool_a_msg and "approved_A" in tool_a_msg.content: print("\nTool A got correct response!") if tool_b_msg and "approved_B" in tool_b_msg.content: print("Tool B got correct response!") else: print("\nFAILED: IDs are not unique") if __name__ == "__main__": asyncio.run(main()) ```
yindo commented 2026-02-20 17:43:00 -05:00
Author
Owner
Copy Link
Copy Source

@sydney-runkle commented on GitHub (Jan 12, 2026):

A bit more history, if you're curious: https://github.com/langchain-ai/langgraph/commit/a37c4d6f4928a3e1d91f2061fc6af142b17e0408

@sydney-runkle commented on GitHub (Jan 12, 2026): A bit more history, if you're curious: https://github.com/langchain-ai/langgraph/commit/a37c4d6f4928a3e1d91f2061fc6af142b17e0408
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
agents
awaiting author revision
bug
change
chat models
checkpointer
ci
code quality
code quality
dependencies
documentation
drawing
duplicate
enhancement
good first issue
help wanted
invalid
maintainer
meta
missing information
not maintained by langchain
open-swe-auto
open-swe-max
open-swe-max-auto
pending
performance
pull-request
Mirrored from GitHub Pull Request
 question
requires design
sdk
server
stale
studio
tests
type checking
usage question
v0.5
v1.1
No Label bug pending
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraph#1079
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?