From 5965d72ff7d1733342eea78cf565d93f7320b779 Mon Sep 17 00:00:00 2001 From: Baihao Wang <25713361+byhow@users.noreply.github.com> Date: Thu, 8 Oct 2026 12:59:25 -0700 Subject: [PATCH] fix(langgraph,prebuilt): stop ToolNode from swallowing invalid resume values (#9232) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **Problem.** `interrupt(response_schema=...)` checks a human's answer when the graph resumes. If the answer doesn't match the schema, the resume fails, nothing is saved, and the human can answer again (#8886). That works when the interrupt runs in a graph node, but not when it runs inside a tool: either the tool calls `interrupt()` itself, or it starts a graph that does, such as a subagent called as a tool. There, the resume returns normally and the interrupt disappears. **Why.** `interrupt()` raises a `pydantic.ValidationError` for an invalid answer. When one comes out of a tool, `ToolNode` assumes the model passed bad arguments, so it hands the error to the model ("Error invoking tool … Please fix the error and try again") and the run carries on. The human never sees the error, the model is blamed for it, and the interrupt is no longer pending. **Fix.** `interrupt()` still raises the same `ValidationError`, but marks it so a new helper, `langgraph.errors.is_invalid_resume(error)`, can recognize it. `ToolNode` re-raises these errors instead of handing them to the model, whatever `handle_tool_errors` is set to, so a bad answer fails the resume the same way inside a tool as outside one: nothing is saved and the interrupt can be answered again. Every other `ValidationError` from a tool is handled as before. **Also fixed, in the same error handling.** With `wrap_tool_call` set and a `handle_tool_errors` other than the default (for example `True`), `ToolNode` turned any exception from the wrapper into an error `ToolMessage`, including `GraphInterrupt`. So an interrupt in that setup never paused: the model got `Error: GraphInterrupt(...)` instead. Interrupts now propagate there too, as they already do without a wrapper. `create_agent` uses the default handler, so it was not affected. **Compatibility.** Not a breaking change: the error type is unchanged. Worth a careful look: a wrapped `ToolNode` with a non-default `handle_tool_errors` now pauses on interrupts instead of returning them to the model as errors. With an older `langgraph` that doesn't have `is_invalid_resume`, `ToolNode` keeps today's behavior. --- libs/langgraph/langgraph/errors.py | 19 +++ libs/langgraph/langgraph/types.py | 17 +- libs/langgraph/tests/test_interruption.py | 11 +- libs/prebuilt/langgraph/prebuilt/tool_node.py | 42 ++++- libs/prebuilt/tests/test_tool_node.py | 148 +++++++++++++++++- 5 files changed, 227 insertions(+), 10 deletions(-) diff --git a/libs/langgraph/langgraph/errors.py b/libs/langgraph/langgraph/errors.py index 65b57d61e..d8ac56200 100644 --- a/libs/langgraph/langgraph/errors.py +++ b/libs/langgraph/langgraph/errors.py @@ -28,6 +28,7 @@ __all__ = ( "ParentCommand", "EmptyInputError", "TaskNotFound", + "is_invalid_resume", ) @@ -239,3 +240,21 @@ class NodeTimeoutError(Exception): self.kind = kind self.idle_timeout = idle_timeout self.run_timeout = run_timeout + + +_INVALID_RESUME = "_langgraph_invalid_resume" + + +def _mark_invalid_resume(error: BaseException) -> None: + setattr(error, _INVALID_RESUME, True) + + +def is_invalid_resume(error: BaseException) -> bool: + """Whether `error` was raised because a resume value didn't match `response_schema`. + + `interrupt()` raises a `pydantic.ValidationError` in that case. `ToolNode` uses + this to tell it apart from invalid tool arguments when a tool calls `interrupt()` + or runs a graph that does, so the resume fails and the interrupt can be answered + again. + """ + return getattr(error, _INVALID_RESUME, False) is True diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 54510c8fc..8fba752a1 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,16 @@ class Command(Generic[N], ToolOutputMixin): PARENT: ClassVar[Literal["__parent__"]] = "__parent__" +def _validate_resume(adapter: TypeAdapter[Any], value: Any) -> Any: + from langgraph.errors import _mark_invalid_resume + + try: + return adapter.validate_python(value) + except ValidationError as exc: + _mark_invalid_resume(exc) + raise + + @overload def interrupt(value: Any, *, response_schema: type[ResponseT]) -> ResponseT: ... @@ -989,6 +999,7 @@ 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`. + Nothing is saved, so the interrupt can be answered again. `is_invalid_resume` identifies it. """ from langgraph._internal._constants import ( CONFIG_KEY_CHECKPOINT_NS, @@ -1012,14 +1023,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..50b8ed6d0 100644 --- a/libs/langgraph/tests/test_interruption.py +++ b/libs/langgraph/tests/test_interruption.py @@ -6,6 +6,7 @@ from langgraph.checkpoint.base import BaseCheckpointSaver from pydantic import BaseModel, ValidationError from typing_extensions import TypedDict +from langgraph.errors import is_invalid_resume from langgraph.graph import END, START, StateGraph from langgraph.types import Command, Durability, Interrupt, interrupt from tests.any_str import AnyStr @@ -202,14 +203,22 @@ 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) + assert is_invalid_resume(exc_info.value) assert graph.invoke(resume({"approved": False}), config) == { "answer": Decision(approved=False) } +def test_is_invalid_resume_ignores_other_errors() -> None: + with pytest.raises(ValidationError) as exc_info: + Decision.model_validate({"approved": "nope"}) + assert not is_invalid_resume(exc_info.value) + assert not is_invalid_resume(ValueError("nope")) + + @pytest.mark.parametrize("resume_style", ["null", "id_map"]) def test_interrupt_response_schema_invalid_resume_after_earlier_interrupt( sync_checkpointer: BaseCheckpointSaver, resume_style: str diff --git a/libs/prebuilt/langgraph/prebuilt/tool_node.py b/libs/prebuilt/langgraph/prebuilt/tool_node.py index 95e161b90..c0398f8f4 100644 --- a/libs/prebuilt/langgraph/prebuilt/tool_node.py +++ b/libs/prebuilt/langgraph/prebuilt/tool_node.py @@ -99,6 +99,14 @@ if TYPE_CHECKING: from langgraph.runtime import Runtime from pydantic_core import ErrorDetails +try: + from langgraph.errors import is_invalid_resume +except ImportError: # `langgraph` before `is_invalid_resume` never marks resume errors + + def is_invalid_resume(error: BaseException) -> bool: + return False + + # right now we use a dict as the default, can change this to AgentState, but depends # on if this lives in LangChain or LangGraph... ideally would have some typed # messages key @@ -957,6 +965,11 @@ class ToolNode(RunnableCallable): try: response = tool.invoke(call_args, config) except ValidationError as exc: + if is_invalid_resume(exc): + # An `interrupt()` in this tool, or in a graph it ran, got a resume + # value that doesn't match its `response_schema`. That's not a bad + # tool argument: fail the run so the interrupt can be answered again. + raise # Filter out errors for injected arguments injected = self._injected_args.get(call["name"]) filtered_errors = _filter_validation_errors(exc, injected) @@ -982,6 +995,10 @@ class ToolNode(RunnableCallable): except GraphBubbleUp: raise except Exception as e: + # The model can't fix a resume value that doesn't match an interrupt's + # `response_schema`, so no `handle_tool_errors` setting handles it. + if is_invalid_resume(e): + raise # Determine which exception types are handled handled_types: tuple[type[Exception], ...] if isinstance(self._handle_tool_errors, type) and issubclass( @@ -1053,9 +1070,13 @@ class ToolNode(RunnableCallable): # Call wrapper with request and execute callable try: return self._wrap_tool_call(tool_request, execute) + except GraphBubbleUp: + # Interrupts always propagate, as they do without a wrapper. + raise except Exception as e: - # Wrapper threw an exception - if not self._handle_tool_errors: + # Wrapper threw an exception. The model can't fix a resume value that + # doesn't match an interrupt's `response_schema`, so it's never handled. + if not self._handle_tool_errors or is_invalid_resume(e): raise # Convert to error message content = _handle_tool_error(e, flag=self._handle_tool_errors) @@ -1104,6 +1125,11 @@ class ToolNode(RunnableCallable): try: response = await tool.ainvoke(call_args, config) except ValidationError as exc: + if is_invalid_resume(exc): + # An `interrupt()` in this tool, or in a graph it ran, got a resume + # value that doesn't match its `response_schema`. That's not a bad + # tool argument: fail the run so the interrupt can be answered again. + raise # Filter out errors for injected arguments injected = self._injected_args.get(call["name"]) filtered_errors = _filter_validation_errors(exc, injected) @@ -1129,6 +1155,10 @@ class ToolNode(RunnableCallable): except GraphBubbleUp: raise except Exception as e: + # The model can't fix a resume value that doesn't match an interrupt's + # `response_schema`, so no `handle_tool_errors` setting handles it. + if is_invalid_resume(e): + raise # Determine which exception types are handled handled_types: tuple[type[Exception], ...] if isinstance(self._handle_tool_errors, type) and issubclass( @@ -1208,9 +1238,13 @@ class ToolNode(RunnableCallable): # None check was performed above already self._wrap_tool_call = cast("ToolCallWrapper", self._wrap_tool_call) return self._wrap_tool_call(tool_request, _sync_execute) + except GraphBubbleUp: + # Interrupts always propagate, as they do without a wrapper. + raise except Exception as e: - # Wrapper threw an exception - if not self._handle_tool_errors: + # Wrapper threw an exception. The model can't fix a resume value that + # doesn't match an interrupt's `response_schema`, so it's never handled. + if not self._handle_tool_errors or is_invalid_resume(e): raise # Convert to error message content = _handle_tool_error(e, flag=self._handle_tool_errors) diff --git a/libs/prebuilt/tests/test_tool_node.py b/libs/prebuilt/tests/test_tool_node.py index 47ebdcae0..546a41770 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,149 @@ 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 _handle_any(e): # no annotation: handles every error + return "handled" + + +# A bad answer to an interrupt must fail the run whatever `handle_tool_errors` is, +# including settings that cover `ValidationError` (a `ValueError`), with or without +# a wrapper. `create_agent` always runs tools through a wrapper (its middleware). +_TOOL_NODES = pytest.mark.parametrize( + ("wrapped", "handle_tool_errors"), + [ + (wrapped, handler) + for wrapped in (False, True) + for handler in (None, True, (ValueError,), _handle_any) + ], + ids=[ + f"{wrapped}-{handler}" + for wrapped in ("plain", "wrapped") + for handler in ("default", "handle_true", "handle_value_error", "untyped") + ], +) +# The interrupt either runs in a graph the tool starts (a subagent) or in the tool. +_SHAPE = pytest.mark.parametrize("nested", [True, False], ids=["nested", "direct"]) + + +@_TOOL_NODES +@_SHAPE +def test_tool_node_reraises_invalid_resume( + sync_checkpointer: BaseCheckpointSaver, + wrapped: bool, + nested: bool, + handle_tool_errors: Any, +) -> None: + asker = _approval_graph() + + @dec_tool + def ask_human() -> str: + """Ask a human for approval.""" + if nested: + return asker.invoke({})["answer"] + approval = interrupt("Approve?", response_schema=_Approval) + return f"approved={approval.approved}" + + def pass_through(request, handler): + return handler(request) + + errors = ( + {} if handle_tool_errors is None else {"handle_tool_errors": handle_tool_errors} + ) + graph = ( + StateGraph(MessagesState) + .add_node( + "tools", + ToolNode( + [ask_human], wrap_tool_call=pass_through if wrapped else None, **errors + ), + ) + .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 saving, 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" + + +@_TOOL_NODES +@_SHAPE +async def test_tool_node_reraises_invalid_resume_async( + async_checkpointer: BaseCheckpointSaver, + wrapped: bool, + nested: bool, + handle_tool_errors: Any, +) -> None: + asker = _approval_graph() + + @dec_tool + async def ask_human() -> str: + """Ask a human for approval.""" + if nested: + return (await asker.ainvoke({}))["answer"] + approval = interrupt("Approve?", response_schema=_Approval) + return f"approved={approval.approved}" + + async def pass_through(request, handler): + return await handler(request) + + errors = ( + {} if handle_tool_errors is None else {"handle_tool_errors": handle_tool_errors} + ) + graph = ( + StateGraph(MessagesState) + .add_node( + "tools", + ToolNode( + [ask_human], awrap_tool_call=pass_through if wrapped else None, **errors + ), + ) + .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: