From 9a0b96693613e2e5bba4609b5d2c842a79839dbb Mon Sep 17 00:00:00 2001 From: Dariel Dato-on <293921406+darieldatoon@users.noreply.github.com> Date: Fri, 9 Oct 2026 11:06:38 -0500 Subject: [PATCH] fix(langgraph): send each RemoteGraph only its own resume map entries Co-Authored-By: Claude Opus 5.5 --- libs/langgraph/langgraph/pregel/remote.py | 7 ++-- libs/langgraph/tests/test_remote_graph.py | 41 +++++++++++++++++++++++ 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/remote.py b/libs/langgraph/langgraph/pregel/remote.py index 032577ce6..d596ebf42 100644 --- a/libs/langgraph/langgraph/pregel/remote.py +++ b/libs/langgraph/langgraph/pregel/remote.py @@ -462,8 +462,11 @@ class RemoteGraph(PregelProtocol): return None conf = config[CONF] if resume_map := conf.get(CONFIG_KEY_RESUME_MAP): - if any(i.id in resume_map for i in pending): - return {"resume": resume_map} + # Do not forward the whole map: it can hold answers for other remotes. + if scoped := { + i.id: resume_map[i.id] for i in pending if i.id in resume_map + }: + return {"resume": scoped} elif (resume := conf[CONFIG_KEY_SCRATCHPAD].get_null_resume(True)) is not None: return {"resume": resume} raise GraphInterrupt(pending) diff --git a/libs/langgraph/tests/test_remote_graph.py b/libs/langgraph/tests/test_remote_graph.py index 40b28ceba..42b8b93a6 100644 --- a/libs/langgraph/tests/test_remote_graph.py +++ b/libs/langgraph/tests/test_remote_graph.py @@ -1040,6 +1040,47 @@ def test_subgraph_retry_after_consumed_resume_reraises_remote_interrupt(): assert call.kwargs["command"] == {"resume": {"approved": True}} +def test_subgraph_resume_map_is_scoped_to_each_remote(): + id_a, id_b = _REMOTE_INTERRUPT_ID, "fedcba9876543210fedcba9876543210" + + def client(interrupt_id: str, done: dict) -> MagicMock: + pending = [{"value": "q", "id": interrupt_id}] + c = MagicMock() + c.runs.stream.side_effect = [ + [StreamPart(event="updates", data={"__interrupt__": pending})], + [StreamPart(event="values", data=done)], + ] + c.threads.get.return_value = { + "status": "interrupted", + "interrupts": {"task": pending}, + } + return c + + client_a = client(id_a, {"approved": True}) + client_b = client(id_b, {"result": "done"}) + parent = ( + StateGraph(_ApprovalState) + .add_node("a", RemoteGraph("a", sync_client=client_a, client=client_a)) + .add_node("b", RemoteGraph("b", sync_client=client_b, client=client_b)) + .add_edge(START, "a") + .add_edge(START, "b") + .compile(checkpointer=InMemorySaver()) + ) + config = {"configurable": {"thread_id": "parent_thread"}} + parent.invoke({"request": "X"}, config) + + out = parent.invoke(Command(resume={id_a: "answer a", id_b: "answer b"}), config) + + assert out["approved"] is True + assert out["result"] == "done" + assert client_a.runs.stream.call_args.kwargs["command"] == { + "resume": {id_a: "answer a"} + } + assert client_b.runs.stream.call_args.kwargs["command"] == { + "resume": {id_b: "answer b"} + } + + def test_subgraph_surfaces_all_remote_interrupts(): second_id = "fedcba9876543210fedcba9876543210" remote_client, next_client = _sync_clients(