From b480be5e6e9ed1dcdb0363eafcf2114a73393dca Mon Sep 17 00:00:00 2001 From: byhow <25713361+byhow@users.noreply.github.com> Date: Wed, 7 Oct 2026 12:08:39 -0700 Subject: [PATCH] fix(langgraph,prebuilt): re-raise invalid resume values from interrupts inside tools A tool that runs a graph calling `interrupt(response_schema=...)` let a bad resume value's `ValidationError` out of its body, and `ToolNode` reported it to the model as invalid tool arguments: the run carried on and the interrupt was lost. `interrupt()` now marks that error and `ToolNode` re-raises marked errors, so the run fails without committing and the interrupt can be answered again. --- libs/langgraph/langgraph/types.py | 24 +++++- libs/langgraph/tests/test_interruption.py | 4 +- libs/prebuilt/langgraph/prebuilt/tool_node.py | 10 +++ libs/prebuilt/tests/test_tool_node.py | 86 ++++++++++++++++++- 4 files changed, 118 insertions(+), 6 deletions(-) diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 54510c8fc..9013371c1 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -20,7 +20,7 @@ from warnings import warn from langchain_core.messages import AnyMessage from langchain_core.runnables import Runnable, RunnableConfig from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata -from pydantic import TypeAdapter +from pydantic import TypeAdapter, ValidationError from typing_extensions import ( NotRequired, TypeAliasType, @@ -882,6 +882,22 @@ class Command(Generic[N], ToolOutputMixin): PARENT: ClassVar[Literal["__parent__"]] = "__parent__" +_INVALID_RESUME_ATTR = "langgraph_invalid_resume" +"""Set on the `ValidationError` for a resume value that doesn't match `response_schema`. + +`ToolNode` reads it by name, so a graph run by a tool re-raises the error instead of +reporting it as invalid tool arguments. +""" + + +def _validate_resume(adapter: TypeAdapter[Any], value: Any) -> Any: + try: + return adapter.validate_python(value) + except ValidationError as exc: + setattr(exc, _INVALID_RESUME_ATTR, True) + raise + + @overload def interrupt(value: Any, *, response_schema: type[ResponseT]) -> ResponseT: ... @@ -989,6 +1005,8 @@ def interrupt( Raises: GraphInterrupt: On the first invocation within the node, halts execution and surfaces the provided value to the client. pydantic.ValidationError: When a resume value does not match a Pydantic model, `TypedDict`, or dataclass `response_schema`. + Inside a graph run by a tool, `ToolNode` re-raises it rather than reporting it + as invalid tool arguments, so the interrupt can be answered again. """ from langgraph._internal._constants import ( CONFIG_KEY_CHECKPOINT_NS, @@ -1012,14 +1030,14 @@ def interrupt( if scratchpad.resume: if idx < len(scratchpad.resume): v = scratchpad.resume[idx] - validated = adapter.validate_python(v) if adapter else v + validated = _validate_resume(adapter, v) if adapter else v conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume[: idx + 1])]) return validated # find current resume value v = scratchpad.get_null_resume(True) if v is not None: assert len(scratchpad.resume) == idx, (scratchpad.resume, idx) - validated = adapter.validate_python(v) if adapter else v + validated = _validate_resume(adapter, v) if adapter else v scratchpad.resume.append(v) conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)]) return validated diff --git a/libs/langgraph/tests/test_interruption.py b/libs/langgraph/tests/test_interruption.py index f4d401e94..6140f19fd 100644 --- a/libs/langgraph/tests/test_interruption.py +++ b/libs/langgraph/tests/test_interruption.py @@ -202,8 +202,10 @@ def test_interrupt_response_schema_rejects_invalid_resume( def resume(value: dict[str, Any]) -> Command: return Command(resume=value if resume_style == "null" else {pending.id: value}) - with pytest.raises(ValidationError, match="approved"): + with pytest.raises(ValidationError, match="approved") as exc_info: graph.invoke(resume({"approved": "nope"}), config) + # `ToolNode` reads this by name to re-raise the error from a graph run by a tool. + assert getattr(exc_info.value, "langgraph_invalid_resume", False) assert graph.invoke(resume({"approved": False}), config) == { "answer": Decision(approved=False) diff --git a/libs/prebuilt/langgraph/prebuilt/tool_node.py b/libs/prebuilt/langgraph/prebuilt/tool_node.py index 95e161b90..f43cd19b8 100644 --- a/libs/prebuilt/langgraph/prebuilt/tool_node.py +++ b/libs/prebuilt/langgraph/prebuilt/tool_node.py @@ -957,6 +957,11 @@ class ToolNode(RunnableCallable): try: response = tool.invoke(call_args, config) except ValidationError as exc: + if getattr(exc, "langgraph_invalid_resume", False): + # Set by `interrupt()`: a graph this tool ran got a resume value + # that doesn't match its `response_schema`. The tool's own + # arguments were fine, so fail the run and keep the interrupt. + raise # Filter out errors for injected arguments injected = self._injected_args.get(call["name"]) filtered_errors = _filter_validation_errors(exc, injected) @@ -1104,6 +1109,11 @@ class ToolNode(RunnableCallable): try: response = await tool.ainvoke(call_args, config) except ValidationError as exc: + if getattr(exc, "langgraph_invalid_resume", False): + # Set by `interrupt()`: a graph this tool ran got a resume value + # that doesn't match its `response_schema`. The tool's own + # arguments were fine, so fail the run and keep the interrupt. + raise # Filter out errors for injected arguments injected = self._injected_args.get(call["name"]) filtered_errors = _filter_validation_errors(exc, injected) diff --git a/libs/prebuilt/tests/test_tool_node.py b/libs/prebuilt/tests/test_tool_node.py index 47ebdcae0..0d3200af2 100644 --- a/libs/prebuilt/tests/test_tool_node.py +++ b/libs/prebuilt/tests/test_tool_node.py @@ -25,6 +25,7 @@ from langchain_core.runnables.config import RunnableConfig from langchain_core.tools import BaseTool, InjectedToolArg, ToolException from langchain_core.tools import tool as dec_tool from langchain_core.tools.base import InjectedToolCallId +from langgraph.checkpoint.base import BaseCheckpointSaver from langgraph.config import get_stream_writer from langgraph.errors import GraphBubbleUp, GraphInterrupt from langgraph.graph import START, MessagesState, StateGraph @@ -32,8 +33,8 @@ from langgraph.graph.message import REMOVE_ALL_MESSAGES, add_messages from langgraph.runtime import ExecutionInfo, ServerInfo from langgraph.store.base import BaseStore from langgraph.store.memory import InMemoryStore -from langgraph.types import Command, Send -from pydantic import BaseModel +from langgraph.types import Command, Send, interrupt +from pydantic import BaseModel, ValidationError from pydantic.v1 import BaseModel as BaseModelV1 from typing_extensions import TypedDict @@ -626,6 +627,87 @@ def test_tool_node_node_interrupt() -> None: assert exc_info.value == "foo" +class _Approval(BaseModel): + approved: bool + + +class _AskState(TypedDict, total=False): + answer: str + + +def _approval_graph(): + """A graph that asks a human for approval with a typed interrupt.""" + + def ask(state: _AskState) -> _AskState: + approval = interrupt("Approve?", response_schema=_Approval) + return {"answer": f"approved={approval.approved}"} + + return StateGraph(_AskState).add_node("ask", ask).add_edge(START, "ask").compile() + + +def _ask_human_call() -> dict[str, list[AnyMessage]]: + call = ToolCall(name="ask_human", args={}, id="call_1") + return {"messages": [AIMessage("", tool_calls=[call])]} + + +def test_tool_node_reraises_invalid_resume_from_nested_interrupt( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + asker = _approval_graph() + + @dec_tool + def ask_human() -> str: + """Ask a human for approval.""" + return asker.invoke({})["answer"] + + graph = ( + StateGraph(MessagesState) + .add_node("tools", ToolNode([ask_human])) + .add_edge(START, "tools") + .compile(checkpointer=sync_checkpointer) + ) + config: RunnableConfig = {"configurable": {"thread_id": "1"}} + [pending] = graph.invoke(_ask_human_call(), config)["__interrupt__"] + + # A bad answer isn't a bad tool argument: the run fails without committing, + # so the same interrupt can be answered again. + with pytest.raises(ValidationError, match="approved"): + graph.invoke(Command(resume={pending.id: {"approved": "maybe"}}), config) + assert [i.id for i in graph.get_state(config).interrupts] == [pending.id] + + result = graph.invoke(Command(resume={pending.id: {"approved": True}}), config) + assert result["messages"][-1].content == "approved=True" + + +async def test_tool_node_reraises_invalid_resume_from_nested_interrupt_async( + async_checkpointer: BaseCheckpointSaver, +) -> None: + asker = _approval_graph() + + @dec_tool + async def ask_human() -> str: + """Ask a human for approval.""" + return (await asker.ainvoke({}))["answer"] + + graph = ( + StateGraph(MessagesState) + .add_node("tools", ToolNode([ask_human])) + .add_edge(START, "tools") + .compile(checkpointer=async_checkpointer) + ) + config: RunnableConfig = {"configurable": {"thread_id": "1"}} + [pending] = (await graph.ainvoke(_ask_human_call(), config))["__interrupt__"] + + with pytest.raises(ValidationError, match="approved"): + await graph.ainvoke(Command(resume={pending.id: {"approved": "maybe"}}), config) + state = await graph.aget_state(config) + assert [i.id for i in state.interrupts] == [pending.id] + + resume = Command(resume={pending.id: {"approved": True}}) + result = await graph.ainvoke(resume, config) + assert result["messages"][-1].content == "approved=True" + + @pytest.mark.parametrize("input_type", ["dict", "tool_calls"]) async def test_tool_node_command(input_type: str) -> None: