diff --git a/libs/langgraph/langgraph/pregel/_runner.py b/libs/langgraph/langgraph/pregel/_runner.py index 5e283716f..6799d86bd 100644 --- a/libs/langgraph/langgraph/pregel/_runner.py +++ b/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 diff --git a/libs/langgraph/tests/test_interruption.py b/libs/langgraph/tests/test_interruption.py index 50b8ed6d0..b404523b9 100644 --- a/libs/langgraph/tests/test_interruption.py +++ b/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"])