test(langgraph): skip async @task RemoteGraph retry case below Python 3.11

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

libs/langgraph/langgraph/_internal/_constants.py

@@ -74,6 +74,8 @@ CONFIG_KEY_RUNTIME = sys.intern("__pregel_runtime")
 # holds a `Runtime` instance with context, store, stream writer, etc.
 CONFIG_KEY_RESUME_MAP = sys.intern("__pregel_resume_map")
 # holds a mapping of task ns -> resume value for resuming tasks
+CONFIG_KEY_RESUME_MAP_SENT = sys.intern("__pregel_resume_map_sent")
+# holds the set of resume map ids already forwarded to a remote graph in this run
 CONFIG_KEY_STREAM_MESSAGES_V2 = sys.intern("__pregel_stream_messages_v2")
 # when True, attach StreamMessagesHandlerV2 so content-block (v2) events
 # flow through stream_mode="messages"; set by StreamingHandler only.
@@ -134,6 +136,7 @@ RESERVED = {
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_TIMED_ATTEMPT_OBSERVER,
     CONFIG_KEY_RESUME_MAP,
+    CONFIG_KEY_RESUME_MAP_SENT,
     CONFIG_KEY_STREAM_MESSAGES_V2,
     # other constants
     PUSH,

libs/langgraph/langgraph/pregel/_loop.py

@@ -43,6 +43,7 @@ from langgraph._internal._constants import (
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_REPLAY_STATE,
     CONFIG_KEY_RESUME_MAP,
+    CONFIG_KEY_RESUME_MAP_SENT,
     CONFIG_KEY_RESUMING,
     CONFIG_KEY_RUNTIME,
     CONFIG_KEY_SCRATCHPAD,
@@ -988,6 +989,7 @@ class PregelLoop:
                     and all(is_xxh3_128_hexdigest(k) for k in resume)
                 ):
                     self.config[CONF][CONFIG_KEY_RESUME_MAP] = resume
+                    self.config[CONF][CONFIG_KEY_RESUME_MAP_SENT] = set()
                 else:
                     if len(self._pending_interrupts()) > 1:
                         raise RuntimeError(

libs/langgraph/langgraph/pregel/remote.py

@@ -29,10 +29,12 @@ from langgraph_sdk.client import (
     get_client,
     get_sync_client,
 )
+from langgraph_sdk.errors import NotFoundError
 from langgraph_sdk.schema import (
     Checkpoint,
     Context,
     QueryParamTypes,
+    Thread,
     ThreadState,
 )
 from langgraph_sdk.schema import (
@@ -49,6 +51,10 @@ from langgraph._internal._constants import (
     CONFIG_KEY_CHECKPOINT_ID,
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_CHECKPOINT_NS,
+    CONFIG_KEY_RESUME_MAP,
+    CONFIG_KEY_RESUME_MAP_SENT,
+    CONFIG_KEY_RESUMING,
+    CONFIG_KEY_SCRATCHPAD,
     CONFIG_KEY_STREAM,
     CONFIG_KEY_TASK_ID,
     INTERRUPT,
@@ -81,6 +87,9 @@ _CONF_DROPLIST = frozenset(
         CONFIG_KEY_CHECKPOINT_ID,
         CONFIG_KEY_CHECKPOINT_NS,
         CONFIG_KEY_TASK_ID,
+        CONFIG_KEY_RESUME_MAP,
+        CONFIG_KEY_RESUME_MAP_SENT,
+        CONFIG_KEY_RESUMING,
     ),
 )
 
@@ -433,6 +442,43 @@ class RemoteGraph(PregelProtocol):
 
         return sanitized
 
+    @staticmethod
+    def _is_resuming(config: RunnableConfig) -> bool:
+        """Whether a parent graph is re-running this node to resume or retry it."""
+        conf = config.get(CONF, {})
+        return CONFIG_KEY_SCRATCHPAD in conf and bool(conf.get(CONFIG_KEY_RESUMING))
+
+    @staticmethod
+    def _resume_command(
+        config: RunnableConfig, thread: Thread | None
+    ) -> CommandSDK | None:
+        """Build a command that resumes an interrupted remote thread."""
+        if thread is None or thread["status"] != "interrupted":
+            return None
+        pending = [
+            Interrupt(**i)
+            for ints in (thread.get("interrupts") or {}).values()
+            for i in ints
+        ]
+        if not pending:
+            return None
+        conf = config[CONF]
+        if resume_map := conf.get(CONFIG_KEY_RESUME_MAP):
+            sent: set[str] = conf.get(CONFIG_KEY_RESUME_MAP_SENT, set())
+            # Do not forward the whole map: it can hold answers for other remotes.
+            # Send each answer once per run: a later interrupt in the same remote
+            # node reuses the id, so a retry would apply the old answer to it.
+            if scoped := {
+                i.id: resume_map[i.id]
+                for i in pending
+                if i.id in resume_map and i.id not in sent
+            }:
+                sent.update(scoped)
+                return {"resume": scoped}
+        elif (resume := conf[CONFIG_KEY_SCRATCHPAD].get_null_resume(True)) is not None:
+            return {"resume": resume}
+        raise GraphInterrupt(pending)
+
     def get_state(
         self,
         config: RunnableConfig,
@@ -794,13 +840,21 @@ class RemoteGraph(PregelProtocol):
         stream_modes, requested, req_single, stream = self._get_stream_modes(
             stream_mode, config
         )
-        if isinstance(input, Command):
-            command: CommandSDK | None = cast(CommandSDK, asdict(input))
-            input = None
-        else:
-            command = None
         thread_id = sanitized_config.get("configurable", {}).pop("thread_id", None)
+        command: CommandSDK | None = None
+        if isinstance(input, Command):
+            command = cast(CommandSDK, asdict(input))
+        elif thread_id is not None and self._is_resuming(merged_config):
+            try:
+                thread = sync_client.threads.get(thread_id, headers=headers)
+            except NotFoundError:
+                thread = None
+            command = self._resume_command(merged_config, thread)
+        if command is not None:
+            input = None
 
+        interrupted = False
+        interrupts: dict[str, Interrupt] = {}
         for chunk in sync_client.runs.stream(
             thread_id=thread_id,
             assistant_id=self.assistant_id,
@@ -832,16 +886,24 @@ class RemoteGraph(PregelProtocol):
             if caller_ns := (config or {}).get(CONF, {}).get(CONFIG_KEY_CHECKPOINT_NS):
                 caller_ns = tuple(caller_ns.split(NS_SEP))
                 ns = caller_ns + ns
+            is_interrupt = (
+                chunk.event.startswith("updates")
+                and isinstance(chunk.data, dict)
+                and INTERRUPT in chunk.data
+            )
+            if interrupted and not is_interrupt and not chunk.event.startswith("error"):
+                continue
             # stream to parent stream
             if stream is not None and mode in stream.modes:
                 stream((ns, mode, chunk.data))
-            # raise interrupt or errors
-            if chunk.event.startswith("updates"):
-                if isinstance(chunk.data, dict) and INTERRUPT in chunk.data:
-                    if caller_ns:
-                        raise GraphInterrupt(
-                            [Interrupt(**i) for i in chunk.data[INTERRUPT]]
-                        )
+            # collect interrupts or raise errors
+            if is_interrupt:
+                if caller_ns:
+                    interrupted = True
+                    for i in chunk.data[INTERRUPT]:
+                        item = Interrupt(**i)
+                        interrupts.setdefault(item.id, item)
+                    continue
             elif chunk.event.startswith("error"):
                 raise RemoteException(chunk.data)
             # filter for what was actually requested
@@ -874,6 +936,8 @@ class RemoteGraph(PregelProtocol):
                 yield chunk.data
             else:
                 yield chunk
+        if interrupted:
+            raise GraphInterrupt(list(interrupts.values()))
 
     @overload
     def astream(
@@ -949,13 +1013,21 @@ class RemoteGraph(PregelProtocol):
         stream_modes, requested, req_single, stream = self._get_stream_modes(
             stream_mode, config
         )
-        if isinstance(input, Command):
-            command: CommandSDK | None = cast(CommandSDK, asdict(input))
-            input = None
-        else:
-            command = None
         thread_id = sanitized_config.get("configurable", {}).pop("thread_id", None)
+        command: CommandSDK | None = None
+        if isinstance(input, Command):
+            command = cast(CommandSDK, asdict(input))
+        elif thread_id is not None and self._is_resuming(merged_config):
+            try:
+                thread = await client.threads.get(thread_id, headers=headers)
+            except NotFoundError:
+                thread = None
+            command = self._resume_command(merged_config, thread)
+        if command is not None:
+            input = None
 
+        interrupted = False
+        interrupts: dict[str, Interrupt] = {}
         async for chunk in client.runs.stream(
             thread_id=thread_id,
             assistant_id=self.assistant_id,
@@ -987,16 +1059,24 @@ class RemoteGraph(PregelProtocol):
             if caller_ns := (config or {}).get(CONF, {}).get(CONFIG_KEY_CHECKPOINT_NS):
                 caller_ns = tuple(caller_ns.split(NS_SEP))
                 ns = caller_ns + ns
+            is_interrupt = (
+                chunk.event.startswith("updates")
+                and isinstance(chunk.data, dict)
+                and INTERRUPT in chunk.data
+            )
+            if interrupted and not is_interrupt and not chunk.event.startswith("error"):
+                continue
             # stream to parent stream
             if stream is not None and mode in stream.modes:
                 stream((ns, mode, chunk.data))
-            # raise interrupt or errors
-            if chunk.event.startswith("updates"):
-                if isinstance(chunk.data, dict) and INTERRUPT in chunk.data:
-                    if caller_ns:
-                        raise GraphInterrupt(
-                            [Interrupt(**i) for i in chunk.data[INTERRUPT]]
-                        )
+            # collect interrupts or raise errors
+            if is_interrupt:
+                if caller_ns:
+                    interrupted = True
+                    for i in chunk.data[INTERRUPT]:
+                        item = Interrupt(**i)
+                        interrupts.setdefault(item.id, item)
+                    continue
             elif chunk.event.startswith("error"):
                 raise RemoteException(chunk.data)
             # filter for what was actually requested
@@ -1029,6 +1109,8 @@ class RemoteGraph(PregelProtocol):
                 yield chunk.data
             else:
                 yield chunk
+        if interrupted:
+            raise GraphInterrupt(list(interrupts.values()))
 
     def stream_events(
         self,

libs/langgraph/tests/test_remote_graph.py

@@ -16,11 +16,13 @@ from langgraph_sdk.schema import StreamPart
 from pydantic import BaseModel
 from typing_extensions import TypedDict
 
+from langgraph.channels.last_value import LastValue
 from langgraph.errors import GraphInterrupt
+from langgraph.func import task
 from langgraph.graph import END, START, MessagesState, StateGraph, add_messages
-from langgraph.pregel import Pregel
-from langgraph.pregel.remote import RemoteGraph
-from langgraph.types import Interrupt, StateSnapshot
+from langgraph.pregel import NodeBuilder, Pregel
+from langgraph.pregel.remote import RemoteException, RemoteGraph
+from langgraph.types import Command, Interrupt, RetryPolicy, StateSnapshot
 from tests.any_str import AnyStr
 from tests.conftest import NO_DOCKER
 from tests.example_app.example_graph import app
@@ -884,6 +886,441 @@ def test_stream_sanitizes_thread_id():
     assert not passed_config["configurable"]
 
 
+class _ApprovalState(TypedDict, total=False):
+    request: str
+    approved: bool
+    result: str
+
+
+_REMOTE_INTERRUPT_ID = "0123456789abcdef0123456789abcdef"
+
+_REMOTE_INTERRUPT_RUN = [
+    StreamPart(
+        event="updates",
+        data={
+            "__interrupt__": [
+                {"value": {"question": "Approve X?"}, "id": _REMOTE_INTERRUPT_ID}
+            ]
+        },
+    )
+]
+
+_REMOTE_DONE_RUN = [
+    StreamPart(
+        event="values", data={"request": "X", "approved": True, "result": "done"}
+    )
+]
+
+
+_INTERRUPTED_THREAD = {
+    "status": "interrupted",
+    "interrupts": {
+        "task": [{"value": {"question": "Approve X?"}, "id": _REMOTE_INTERRUPT_ID}]
+    },
+}
+
+_IDLE_THREAD = {"status": "idle", "interrupts": {}}
+
+_RETRY = RetryPolicy(initial_interval=0, jitter=False, retry_on=ValueError)
+
+
+def _remote_interrupt_parent(remote_client: MagicMock, next_client: MagicMock):
+    return (
+        StateGraph(_ApprovalState)
+        .add_node(
+            "agent_b",
+            RemoteGraph("agent_b", sync_client=remote_client, client=remote_client),
+            retry_policy=_RETRY,
+        )
+        .add_node(
+            "agent_c",
+            RemoteGraph("agent_c", sync_client=next_client, client=next_client),
+            retry_policy=_RETRY,
+        )
+        .add_edge(START, "agent_b")
+        .add_edge("agent_b", "agent_c")
+        .compile(checkpointer=InMemorySaver())
+    )
+
+
+def _sync_clients(remote_runs: list, next_runs: list) -> tuple[MagicMock, MagicMock]:
+    remote_client = MagicMock()
+    remote_client.runs.stream.side_effect = remote_runs
+    remote_client.threads.get.return_value = _INTERRUPTED_THREAD
+    next_client = MagicMock()
+    next_client.runs.stream.side_effect = next_runs
+    next_client.threads.get.return_value = _IDLE_THREAD
+    return remote_client, next_client
+
+
+@pytest.mark.parametrize(
+    "resume",
+    [{"approved": True}, {_REMOTE_INTERRUPT_ID: {"approved": True}}],
+    ids=["value", "interrupt_id"],
+)
+def test_subgraph_resume_is_forwarded_to_remote_thread(resume):
+    remote_client, next_client = _sync_clients(
+        [_REMOTE_INTERRUPT_RUN, _REMOTE_DONE_RUN], [_REMOTE_DONE_RUN]
+    )
+    parent = _remote_interrupt_parent(remote_client, next_client)
+    config = {"configurable": {"thread_id": "parent_thread"}}
+
+    out = parent.invoke({"request": "X"}, config)
+    assert out["__interrupt__"] == [
+        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
+    ]
+
+    out = parent.invoke(Command(resume=resume), config)
+    assert "__interrupt__" not in out
+    assert out["result"] == "done"
+
+    first, second = remote_client.runs.stream.call_args_list
+    assert first.kwargs["input"] == {"request": "X"}
+    assert first.kwargs["command"] is None
+    assert second.kwargs["input"] is None
+    assert second.kwargs["command"] == {"resume": resume}
+    assert second.kwargs["thread_id"] == first.kwargs["thread_id"]
+
+    next_kwargs = next_client.runs.stream.call_args.kwargs
+    assert next_kwargs["command"] is None
+    assert next_kwargs["input"] == {"request": "X", "approved": True, "result": "done"}
+
+
+def test_subgraph_resume_for_other_interrupt_reraises_remote_interrupt():
+    remote_client, next_client = _sync_clients([_REMOTE_INTERRUPT_RUN], [])
+    parent = _remote_interrupt_parent(remote_client, next_client)
+    config = {"configurable": {"thread_id": "parent_thread"}}
+    parent.invoke({"request": "X"}, config)
+
+    other_id = "fedcba9876543210fedcba9876543210"
+    out = parent.invoke(Command(resume={other_id: {"approved": True}}), config)
+
+    assert out["__interrupt__"] == [
+        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
+    ]
+    assert remote_client.runs.stream.call_count == 1
+
+
+def test_subgraph_retry_after_resume_sends_input_to_idle_remote_thread():
+    remote_client, next_client = _sync_clients(
+        [_REMOTE_INTERRUPT_RUN, _REMOTE_DONE_RUN],
+        [ValueError("transient"), _REMOTE_DONE_RUN],
+    )
+    parent = _remote_interrupt_parent(remote_client, next_client)
+    config = {"configurable": {"thread_id": "parent_thread"}}
+    parent.invoke({"request": "X"}, config)
+
+    out = parent.invoke(
+        Command(resume={_REMOTE_INTERRUPT_ID: {"approved": True}}), config
+    )
+
+    assert out["result"] == "done"
+    retry = next_client.runs.stream.call_args_list[1]
+    assert retry.kwargs["command"] is None
+    assert retry.kwargs["input"] == {"request": "X", "approved": True, "result": "done"}
+
+
+def test_subgraph_retry_after_consumed_resume_reraises_remote_interrupt():
+    remote_client, next_client = _sync_clients(
+        [_REMOTE_INTERRUPT_RUN, ValueError("transient"), _REMOTE_DONE_RUN],
+        [_REMOTE_DONE_RUN],
+    )
+    parent = _remote_interrupt_parent(remote_client, next_client)
+    config = {"configurable": {"thread_id": "parent_thread"}}
+    parent.invoke({"request": "X"}, config)
+
+    out = parent.invoke(Command(resume={"approved": True}), config)
+    assert out["__interrupt__"] == [
+        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
+    ]
+    assert remote_client.runs.stream.call_count == 2
+
+    out = parent.invoke(Command(resume={"approved": True}), config)
+    assert out["result"] == "done"
+    for call in remote_client.runs.stream.call_args_list[1:]:
+        assert call.kwargs["input"] is None
+        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 _resumed_then_dropped():
+    # A remote node's next interrupt() reuses the id of the one just resumed.
+    yield from _REMOTE_INTERRUPT_RUN
+    raise ValueError("transient")
+
+
+def test_subgraph_retry_does_not_resend_resume_map():
+    remote_client, next_client = _sync_clients(
+        [_REMOTE_INTERRUPT_RUN, _resumed_then_dropped()], []
+    )
+    parent = _remote_interrupt_parent(remote_client, next_client)
+    config = {"configurable": {"thread_id": "parent_thread"}}
+    parent.invoke({"request": "X"}, config)
+
+    out = parent.invoke(
+        Command(resume={_REMOTE_INTERRUPT_ID: {"approved": True}}), config
+    )
+
+    assert out["__interrupt__"] == [
+        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
+    ]
+    assert remote_client.runs.stream.call_count == 2
+
+
+def _ancestor_retry_parent(remote_client: MagicMock, via: str = "node"):
+    # A low-level Pregel subgraph has no START step, so an ancestor retry's
+    # RESUMING flag reaches the RemoteGraph task with a fresh node_attempt.
+    remote = RemoteGraph("agent_b", sync_client=remote_client, client=remote_client)
+    inner = Pregel(
+        nodes={
+            "agent_b": NodeBuilder()
+            .subscribe_only("request")
+            .do(remote)
+            .write_to("result")
+        },
+        channels={"request": LastValue(str), "result": LastValue(object)},
+        input_channels=["request"],
+        output_channels=["result"],
+    )
+    if via == "node":
+        p = inner
+    elif via == "task":
+
+        @task
+        def child(request: str):
+            return inner.invoke({"request": request})["result"]
+
+        def p(state: _ApprovalState):
+            return {"result": child(state["request"]).result()}
+
+    else:
+
+        @task
+        async def child(request: str):
+            return (await inner.ainvoke({"request": request}))["result"]
+
+        async def p(state: _ApprovalState):
+            return {"result": await child(state["request"])}
+
+    return (
+        StateGraph(_ApprovalState)
+        .add_node("p", p, retry_policy=_RETRY)
+        .add_edge(START, "p")
+        .compile(checkpointer=InMemorySaver())
+    )
+
+
+@pytest.mark.parametrize("via", ["node", "task"])
+def test_subgraph_ancestor_retry_does_not_resend_resume_map(via):
+    remote_client, _ = _sync_clients(
+        [_REMOTE_INTERRUPT_RUN, _resumed_then_dropped(), _REMOTE_DONE_RUN], []
+    )
+    parent = _ancestor_retry_parent(remote_client, via)
+    config = {"configurable": {"thread_id": "parent_thread"}}
+    parent.invoke({"request": "X"}, config)
+
+    out = parent.invoke(Command(resume={_REMOTE_INTERRUPT_ID: "first"}), config)
+    assert out["__interrupt__"] == [
+        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
+    ]
+
+    parent.invoke(Command(resume={_REMOTE_INTERRUPT_ID: "second"}), config)
+    assert [c.kwargs["command"] for c in remote_client.runs.stream.call_args_list] == [
+        None,
+        {"resume": {_REMOTE_INTERRUPT_ID: "first"}},
+        {"resume": {_REMOTE_INTERRUPT_ID: "second"}},
+    ]
+
+
+def test_subgraph_empty_remote_interrupt_pauses_parent():
+    remote_client, next_client = _sync_clients(
+        [[StreamPart(event="updates", data={"__interrupt__": []})]], []
+    )
+    parent = _remote_interrupt_parent(remote_client, next_client)
+    config = {"configurable": {"thread_id": "parent_thread"}}
+
+    parent.invoke({"request": "X"}, config)
+
+    assert parent.get_state(config).next == ("agent_b",)
+    next_client.runs.stream.assert_not_called()
+
+
+def test_subgraph_remote_error_after_interrupt_is_raised():
+    remote_client, next_client = _sync_clients(
+        [
+            [
+                *_REMOTE_INTERRUPT_RUN,
+                StreamPart(event="error", data={"error": "boom", "message": "x"}),
+            ]
+        ],
+        [],
+    )
+    parent = _remote_interrupt_parent(remote_client, next_client)
+
+    with pytest.raises(RemoteException):
+        parent.invoke({"request": "X"}, {"configurable": {"thread_id": "t"}})
+
+
+def test_subgraph_surfaces_all_remote_interrupts():
+    second_id = "fedcba9876543210fedcba9876543210"
+    remote_client, next_client = _sync_clients(
+        [
+            [
+                *_REMOTE_INTERRUPT_RUN,
+                StreamPart(event="values", data={"request": "X"}),
+                StreamPart(
+                    event="updates",
+                    data={"__interrupt__": [{"value": "second", "id": second_id}]},
+                ),
+            ]
+        ],
+        [],
+    )
+    parent = _remote_interrupt_parent(remote_client, next_client)
+    config = {"configurable": {"thread_id": "parent_thread"}}
+
+    out = parent.invoke({"request": "X"}, config)
+
+    assert out["__interrupt__"] == [
+        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID),
+        Interrupt(value="second", id=second_id),
+    ]
+
+
+async def test_subgraph_resume_is_forwarded_to_remote_thread_async():
+    def async_runs(*runs: list) -> list[MagicMock]:
+        iters = []
+        for parts in runs:
+            async_iter = MagicMock()
+            async_iter.__aiter__.return_value = parts
+            iters.append(async_iter)
+        return iters
+
+    remote_client = MagicMock()
+    remote_client.runs.stream.side_effect = async_runs(
+        _REMOTE_INTERRUPT_RUN, _REMOTE_DONE_RUN
+    )
+    remote_client.threads.get = AsyncMock(return_value=_INTERRUPTED_THREAD)
+    next_client = MagicMock()
+    next_client.runs.stream.side_effect = async_runs(_REMOTE_DONE_RUN)
+    next_client.threads.get = AsyncMock(return_value=_IDLE_THREAD)
+    parent = _remote_interrupt_parent(remote_client, next_client)
+    config = {"configurable": {"thread_id": "parent_thread"}}
+
+    out = await parent.ainvoke({"request": "X"}, config)
+    assert "__interrupt__" in out
+
+    out = await parent.ainvoke(Command(resume={"approved": True}), config)
+    assert "__interrupt__" not in out
+    assert out["result"] == "done"
+
+    second = remote_client.runs.stream.call_args_list[1]
+    assert second.kwargs["input"] is None
+    assert second.kwargs["command"] == {"resume": {"approved": True}}
+    assert next_client.runs.stream.call_args.kwargs["command"] is None
+
+
+async def test_subgraph_retry_does_not_resend_resume_map_async():
+    async def run(*parts: StreamPart, error: Exception | None = None):
+        for part in parts:
+            yield part
+        if error:
+            raise error
+
+    remote_client = MagicMock()
+    remote_client.runs.stream.side_effect = [
+        run(*_REMOTE_INTERRUPT_RUN),
+        run(*_REMOTE_INTERRUPT_RUN, error=ValueError("transient")),
+    ]
+    remote_client.threads.get = AsyncMock(return_value=_INTERRUPTED_THREAD)
+    parent = _remote_interrupt_parent(remote_client, MagicMock())
+    config = {"configurable": {"thread_id": "parent_thread"}}
+    await parent.ainvoke({"request": "X"}, config)
+
+    resume = {_REMOTE_INTERRUPT_ID: {"approved": True}}
+    out = await parent.ainvoke(Command(resume=resume), config)
+
+    assert out["__interrupt__"] == [
+        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
+    ]
+    assert remote_client.runs.stream.call_count == 2
+    assert remote_client.runs.stream.call_args.kwargs["command"] == {"resume": resume}
+
+
+@pytest.mark.parametrize(
+    "via", ["node", pytest.param("atask", marks=NEEDS_CONTEXTVARS)]
+)
+async def test_subgraph_ancestor_retry_does_not_resend_resume_map_async(via):
+    async def run(*parts: StreamPart, error: Exception | None = None):
+        for part in parts:
+            yield part
+        if error:
+            raise error
+
+    remote_client = MagicMock()
+    remote_client.runs.stream.side_effect = [
+        run(*_REMOTE_INTERRUPT_RUN),
+        run(*_REMOTE_INTERRUPT_RUN, error=ValueError("transient")),
+        run(*_REMOTE_DONE_RUN),
+    ]
+    remote_client.threads.get = AsyncMock(return_value=_INTERRUPTED_THREAD)
+    parent = _ancestor_retry_parent(remote_client, via)
+    config = {"configurable": {"thread_id": "parent_thread"}}
+    await parent.ainvoke({"request": "X"}, config)
+
+    out = await parent.ainvoke(Command(resume={_REMOTE_INTERRUPT_ID: "first"}), config)
+    assert out["__interrupt__"] == [
+        Interrupt(value={"question": "Approve X?"}, id=_REMOTE_INTERRUPT_ID)
+    ]
+
+    await parent.ainvoke(Command(resume={_REMOTE_INTERRUPT_ID: "second"}), config)
+    assert [c.kwargs["command"] for c in remote_client.runs.stream.call_args_list] == [
+        None,
+        {"resume": {_REMOTE_INTERRUPT_ID: "first"}},
+        {"resume": {_REMOTE_INTERRUPT_ID: "second"}},
+    ]
+
+
 @pytest.mark.anyio
 async def test_ainvoke():
     # set up test