fix(langgraph): let other tasks finish when a resume value is invalid (#9265)

Found while testing per-call human-in-the-loop
(langchain-ai/langchain#41062). The code depends only on #9232, which is
already on `main`.

When one answer in a multi-answer resume fails `response_schema`
validation, the runner cancelled the other tasks in the step, so in
async runs a task cut off after its side effect kept its saved answer
and ran again on the next resume. Invalid resume values (already tagged
by `is_invalid_resume` from #9232) no longer stop the other tasks: the
error is raised once they finish, so their results are saved.

Hit by per-call human-in-the-loop: approving one tool call while giving
another an invalid edit, in one resume, sent the approved call's email
twice once the second answer was fixed.

## How to verify

- `test_invalid_resume_lets_other_answered_tasks_finish` fails on `main`
(the approved task is still pending after its side effect) and passes
with this change. Like the other async interrupt tests, it is skipped
before Python 3.11, where async nodes can't call `interrupt()`.
- The per-call HITL reproduction against this branch: the approved async
tool runs once (twice on `main` whenever it awaits after its side
effect).
- `make lint` passes. With `NO_DOCKER=true`, the full `libs/langgraph`
suite has the same failures as `main` locally (`test_prebuilt_tool_chat`
and `test_stream_mode_messages_command`, sync and async), and nothing
new.

libs/langgraph/langgraph/pregel/_runner.py

@@ -41,7 +41,7 @@ from langgraph._internal._future import chain_future, run_coroutine_threadsafe
 from langgraph._internal._scratchpad import PregelScratchpad
 from langgraph._internal._typing import MISSING
 from langgraph.constants import TAG_HIDDEN
-from langgraph.errors import GraphBubbleUp, GraphInterrupt
+from langgraph.errors import GraphBubbleUp, GraphInterrupt, is_invalid_resume
 from langgraph.pregel._algo import Call
 from langgraph.pregel._executor import Submit
 from langgraph.pregel._retry import arun_with_retry, run_with_retry
@@ -621,7 +621,11 @@ def _should_stop_others(
     handled_exception_ids: set[int] | None = None,
 ) -> bool:
     """Check if any task failed, if so, cancel all other tasks.
-    GraphInterrupts are not considered failures."""
+    GraphInterrupts are not considered failures.
+
+    Neither is an invalid resume value: it's raised once the other tasks finish, so a
+    task answered in the same resume isn't cut off after its side effects and then run
+    again with its saved answer."""
     for fut in done:
         if fut.cancelled():
             continue
@@ -629,6 +633,7 @@ def _should_stop_others(
             if (
                 id(exc) not in (handled_exception_ids or set())
                 and not isinstance(exc, GraphBubbleUp)
+                and not is_invalid_resume(exc)
                 and fut not in SKIP_RERAISE_SET
             ):
                 return True

libs/langgraph/tests/test_interruption.py

@@ -1,5 +1,8 @@
+import asyncio
+import operator
+import sys
 from dataclasses import dataclass
-from typing import Any
+from typing import Annotated, Any
 
 import pytest
 from langgraph.checkpoint.base import BaseCheckpointSaver
@@ -251,3 +254,61 @@ def test_interrupt_response_schema_invalid_resume_after_earlier_interrupt(
     assert graph.invoke(resume({"approved": True}), config) == {
         "answer": ["ok", Decision(approved=True)]
     }
+
+
+@pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="Python 3.11+ is required for async contextvars support",
+)
+async def test_invalid_resume_lets_other_answered_tasks_finish(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """An invalid answer fails the resume without cutting off the other answered tasks.
+
+    A task cancelled after its side effect keeps its answer and runs again on the next
+    resume, so the side effect would happen twice.
+    """
+    sent: list[str] = []
+
+    class State(TypedDict):
+        log: Annotated[list[str], operator.add]
+
+    async def send(state: State) -> State:
+        interrupt("send?", response_schema=Decision)
+        sent.append("email")
+        await asyncio.sleep(0.05)  # waiting on the reply to a request already sent
+        return {"log": ["sent"]}
+
+    async def check(state: State) -> State:
+        interrupt("check?", response_schema=Decision)
+        return {"log": ["checked"]}
+
+    graph = (
+        StateGraph(State)
+        .add_node("send", send)
+        .add_node("check", check)
+        .add_edge(START, "send")
+        .add_edge(START, "check")
+        .compile(checkpointer=async_checkpointer)
+    )
+    config = {"configurable": {"thread_id": "1"}}
+    await graph.ainvoke({"log": []}, config)
+    ids = {i.value: i.id for i in (await graph.aget_state(config)).interrupts}
+
+    with pytest.raises(ValidationError, match="approved"):
+        await graph.ainvoke(
+            Command(
+                resume={
+                    ids["send?"]: {"approved": True},
+                    ids["check?"]: {"approved": "nope"},
+                }
+            ),
+            config,
+        )
+    pending = [i.value for i in (await graph.aget_state(config)).interrupts]
+    assert (pending, sent) == (["check?"], ["email"])
+
+    result = await graph.ainvoke(
+        Command(resume={ids["check?"]: {"approved": True}}), config
+    )
+    assert (sorted(result["log"]), sent) == (["checked", "sent"], ["email"])