mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-10-14 12:15:25 +02:00
fix(langgraph): stop astream_events dropping control/interrupts on v1/v2 (#9219)
### Problem `Pregel.astream_events` declared `interrupt_before`, `interrupt_after` and `control` as keyword-only parameters, but forwarded them only on the `version="v3"` branch. On async `version="v1"`/`"v2"` the values were bound to named parameters and never entered `**kwargs`, so they were silently dropped on the way to `astream` — no error, no effect: - `request_drain()` on a caller-supplied `RunControl` did nothing; the graph created its own control and ran to completion. - Static interrupts passed as `astream_events(..., interrupt_before=[...])` never fired. This is a regression introduced by #7677 (first released in `1.2.0a3`). Before that, `Pregel` did not override `astream_events`, and these keywords reached `astream` through `**kwargs` — langchain-core's v1 and v2 event implementations both forward caller kwargs to `astream` (`tracers/log_stream.py`, `tracers/event_stream.py`). Downstream impact: langgraph-api passes run-level `interrupt_before` / `interrupt_after` into `graph.astream_events(..., version="v2", **kwargs)` for runs with `stream_mode="events"` (`models/run.py`, `stream.py`), so server runs using that stream mode have silently dropped static interrupts since langgraph 1.2.0a3. Sync `stream_events(version="v1"/"v2")` is unaffected: langchain-core raises `NotImplementedError` there, so there was no working sync path to break. ### Fix Remove the three parameters from the public runtime implementations of `stream_events` / `astream_events` (and from the explicit passing in their v3 dispatch calls), so the values travel through `**kwargs` again: - **v1/v2** recover the pre-#7677 passthrough. `astream` receives exactly what the caller passed — a supplied value, an explicit `None`, or nothing for an omitted argument. - **v3** is unchanged. The values bind on `_pregel_stream_v3` / `_apregel_stream_v3`, which keep their named parameters and since-inception (#7519, `1.2.0a1`) explicit-`None` defaults: an omitted argument arrives at `astream` as `None`, exactly as before. What `astream` receives after this change: | caller supplies | async v1/v2 | v3 (sync and async) | |---|---|---| | nothing | argument absent | explicit `None` | | explicit `None` | `None` | `None` | | a value | the value | the value | The typed `version="v3"` `@overload` stubs are untouched, so v3 call sites keep their static types (invalid values still fail `ty` there, same as on `main`). `RemoteGraph` and the v3-only `transformers` parameter are untouched. Complexity: net −24 lines in `main.py` with no branches, sentinels or wrappers added. The public capture-and-drop pattern that caused the bug is gone; the two private helpers each have a single call path and forward unconditionally, so they cannot drop anything. Docstrings now describe the three parameters as honored on every async version (type-checked only on the v3 overloads), and no longer promise synchronous v1/v2 event streaming, which langchain-core does not implement. ### Compatibility notes - The three parameters were keyword-only, so no caller can break; keyword callers bind through `**kwargs` identically. - The names leave the runtime signatures (`inspect.signature`), while the v3 overloads keep the typing. - v3 is unchanged: `_pregel_stream_v3`/`_apregel_stream_v3` still default all three to `None` and pass them explicitly, so an override's non-`None` default never applied on v3 and still doesn't. The only behavior change beyond restoring the pre-#7677 passthrough is on v1/v2 with an explicit `None`: since 1.2.0a3 the named-parameter capture dropped it, so an override's non-`None` default was silently applied instead; now the explicit `None` reaches `astream`, as it did before 1.2.0 and as it does on v3. ### Testing New tests in `tests/test_stream_events_v3_kwarg_forwarding.py`: - `interrupt_before` / `interrupt_after` / pre-drained `control` parametrized over v1/v2/v3 (12 of them fail on unpatched `main`). - Sync v3 static interrupts. - Mid-run drain: a node calls `request_drain()`; asserts `GraphDrained` propagates, the caller's own `RunControl` was used, and the checkpoint keeps the pending step. - Recording tests pin, for all three names and on both the async and sync v3 paths, exactly what `(a)stream` receives: supplied values (with `control` identity), explicit `None`, and absence for omitted arguments. Verified by mutation: replacing the implementation with each rejected alternative (unpatched parent; injecting `None` for omitted kwargs on v1/v2; dropping explicit `None` on v1/v2; kwargs-only private helpers) makes the corresponding tests fail. Suites: 246 passed on the four adjacent streaming test files; 906 passed / 1 skipped across every test file calling `(a)stream_events` plus `test_runtime.py`. `ruff check` / `format --check` clean; `ty check langgraph` clean. Not verifiable locally (CI please): Python 3.10 (the new tests carry no version skip), minimum supported langchain-core, and a live langgraph-api server run. ### Release note > Fixed a regression since 1.2.0a3: `Pregel.astream_events` silently dropped > `interrupt_before`, `interrupt_after` and `control` for > `version="v1"`/`"v2"`. Static interrupts and `RunControl` drains now work on > every async version. Server runs using `stream_mode="events"` were affected > and get their static interrupts back.
This commit is contained in:
Igor Soarez
@@ -3714,16 +3714,15 @@ class Pregel(
|
||||
config: RunnableConfig | None = None,
|
||||
*,
|
||||
version: Literal["v1", "v2", "v3"] = "v2",
|
||||
interrupt_before: All | Sequence[str] | None = None,
|
||||
interrupt_after: All | Sequence[str] | None = None,
|
||||
control: RunControl | None = None,
|
||||
transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
"""Stream events from this graph.
|
||||
|
||||
For `version="v1"` / `"v2"`, yields `StreamEvent` dicts (see
|
||||
`Runnable.stream_events`). For `version="v3"`, returns a
|
||||
For `version="v1"` / `"v2"`, delegates to
|
||||
`Runnable.(a)stream_events`; synchronous v1/v2 event streaming is
|
||||
not implemented in langchain-core, so use `astream_events` for
|
||||
those versions. For `version="v3"`, returns a
|
||||
`GraphRunStream` whose typed projections the caller drives by
|
||||
iterating — no background thread.
|
||||
|
||||
@@ -3753,20 +3752,27 @@ class Pregel(
|
||||
config: Optional runnable config.
|
||||
version: Streaming-event schema version. `"v3"` selects the
|
||||
content-block-centric streaming protocol.
|
||||
interrupt_before: Nodes to interrupt before, if any. Only
|
||||
used for `version="v3"`.
|
||||
interrupt_after: Nodes to interrupt after, if any. Only
|
||||
used for `version="v3"`.
|
||||
interrupt_before: Nodes to interrupt before, if any.
|
||||
Honored on every version that can run; type-checked
|
||||
only on the `version="v3"` overloads.
|
||||
interrupt_after: Nodes to interrupt after, if any. Honored
|
||||
on every version that can run; type-checked only on
|
||||
the `version="v3"` overloads.
|
||||
control: Optional run control used to request cooperative
|
||||
drain. Only used for `version="v3"`.
|
||||
drain. Honored on every version that can run;
|
||||
type-checked only on the `version="v3"` overloads.
|
||||
transformers: Extra transformer classes or configured
|
||||
factories appended after compile-time
|
||||
`stream_transformers`. Factories are called as
|
||||
`factory(scope)` so they can propagate to subgraph
|
||||
scopes. Only used for `version="v3"`.
|
||||
**kwargs: For `version="v1"`/`"v2"`, forwarded to
|
||||
`Runnable.stream_events`. For `version="v3"`, forwarded
|
||||
to the underlying `stream(...)` call (e.g. `context`,
|
||||
**kwargs: For `version="v1"`/`"v2"` on `astream_events`,
|
||||
forwarded to `Runnable.astream_events`, which passes
|
||||
them through to `astream` — so execution kwargs such as
|
||||
`context`, `durability`, `interrupt_before`,
|
||||
`interrupt_after` and `control` are honored on every
|
||||
version that can run. For `version="v3"`, forwarded to the
|
||||
underlying `stream(...)` call (e.g. `context`,
|
||||
`durability`, `output_keys`, `print_mode`, `debug`).
|
||||
`stream_mode` and `subgraphs` are not accepted under
|
||||
`version="v3"` and raise `TypeError` if supplied; v3
|
||||
@@ -3774,16 +3780,15 @@ class Pregel(
|
||||
|
||||
Returns:
|
||||
For `version="v3"`, a `GraphRunStream` the caller iterates
|
||||
to drive the run. Otherwise an `Iterator[StreamEvent]`.
|
||||
to drive the run. For `version="v1"`/`"v2"`,
|
||||
`astream_events` yields `StreamEvent` dicts; the synchronous
|
||||
v1/v2 path is not implemented in langchain-core.
|
||||
"""
|
||||
if version == "v3":
|
||||
_reject_v3_invariant_kwargs(kwargs)
|
||||
return self._pregel_stream_v3(
|
||||
input,
|
||||
config,
|
||||
interrupt_before=interrupt_before,
|
||||
interrupt_after=interrupt_after,
|
||||
control=control,
|
||||
transformers=transformers,
|
||||
**kwargs,
|
||||
)
|
||||
@@ -3819,9 +3824,6 @@ class Pregel(
|
||||
config: RunnableConfig | None = None,
|
||||
*,
|
||||
version: Literal["v1", "v2", "v3"] = "v2",
|
||||
interrupt_before: All | Sequence[str] | None = None,
|
||||
interrupt_after: All | Sequence[str] | None = None,
|
||||
control: RunControl | None = None,
|
||||
transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> AsyncIterator[StreamEvent] | Awaitable[Any]:
|
||||
@@ -3844,9 +3846,6 @@ class Pregel(
|
||||
return self._apregel_stream_v3(
|
||||
input,
|
||||
config,
|
||||
interrupt_before=interrupt_before,
|
||||
interrupt_after=interrupt_after,
|
||||
control=control,
|
||||
transformers=transformers,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
@@ -8,20 +8,33 @@ caller kwargs to the inner ``(a)stream`` call but rejects ``stream_mode`` and
|
||||
``subgraphs`` since v3 owns them (``stream_mode`` is built from the
|
||||
transformer mux; ``subgraphs`` is forced True so nested namespaces flow
|
||||
through scoped muxes).
|
||||
|
||||
A second regression is pinned here: #7677 (first released in 1.2.0a3)
|
||||
declared `interrupt_before` / `interrupt_after` / `control` as named
|
||||
parameters on the `Pregel.stream_events` / `astream_events` dispatchers
|
||||
but forwarded them only on the v3 branch, silently dropping them on v1/v2
|
||||
(where they had reached `(a)stream` through `**kwargs` before).
|
||||
`TestAstreamKwargsForwardedOnEveryVersion` and friends pin that they
|
||||
reach `(a)stream` on every version, with the v1/v2 passthrough semantics
|
||||
restored (exactly what the caller passed, including an explicit `None`)
|
||||
and v3's explicit-default semantics preserved.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from collections.abc import AsyncIterator, Callable
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
from typing_extensions import TypedDict
|
||||
|
||||
from langgraph.constants import END, START
|
||||
from langgraph.errors import GraphDrained
|
||||
from langgraph.graph import StateGraph
|
||||
from langgraph.runtime import Runtime
|
||||
from langgraph.runtime import RunControl, Runtime
|
||||
|
||||
NEEDS_CONTEXTVARS = pytest.mark.skipif(
|
||||
sys.version_info < (3, 11),
|
||||
@@ -106,3 +119,274 @@ class TestKwargForwardingAsync:
|
||||
version="v3",
|
||||
subgraphs=False,
|
||||
)
|
||||
|
||||
|
||||
_KWARG_NAMES = ("control", "interrupt_before", "interrupt_after")
|
||||
|
||||
|
||||
def _build_two_step_graph(
|
||||
first: Callable[[_State], dict[str, Any]] | None = None,
|
||||
) -> Any:
|
||||
"""A `first -> second` graph with a checkpointer, for interrupt/drain tests."""
|
||||
|
||||
def default_first(state: _State) -> dict[str, Any]:
|
||||
return {"message": state["message"] + " first"}
|
||||
|
||||
def second(state: _State) -> dict[str, Any]:
|
||||
return {"message": state["message"] + " second"}
|
||||
|
||||
builder = StateGraph(_State)
|
||||
builder.add_node("first", first or default_first)
|
||||
builder.add_node("second", second)
|
||||
builder.add_edge(START, "first")
|
||||
builder.add_edge("first", "second")
|
||||
builder.add_edge("second", END)
|
||||
return builder.compile(checkpointer=InMemorySaver())
|
||||
|
||||
|
||||
async def _drive_astream_events(
|
||||
graph: Any, config: dict[str, Any], version: str, **kwargs: Any
|
||||
) -> None:
|
||||
"""Consume an astream_events run for `version` to completion."""
|
||||
if version == "v3":
|
||||
run = await graph.astream_events(
|
||||
{"message": "hi"}, config, version="v3", **kwargs
|
||||
)
|
||||
await run.output()
|
||||
else:
|
||||
async for _ in graph.astream_events(
|
||||
{"message": "hi"}, config, version=version, **kwargs
|
||||
):
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.filterwarnings("ignore:astream_events version='v1' is deprecated")
|
||||
@pytest.mark.parametrize("version", ["v1", "v2", "v3"])
|
||||
class TestAstreamKwargsForwardedOnEveryVersion:
|
||||
"""`interrupt_before`/`interrupt_after`/`control` reach `astream` on every
|
||||
version.
|
||||
|
||||
Regression test for #7677 (first released in 1.2.0a3): the dispatchers
|
||||
captured these parameters as named arguments but forwarded them only on
|
||||
the v3 branch, silently dropping them on v1/v2.
|
||||
"""
|
||||
|
||||
async def test_interrupt_before(self, version: str) -> None:
|
||||
graph = _build_two_step_graph()
|
||||
config = {"configurable": {"thread_id": "ib"}}
|
||||
await _drive_astream_events(graph, config, version, interrupt_before=["second"])
|
||||
state = await graph.aget_state(config)
|
||||
assert state.next == ("second",)
|
||||
assert state.values == {"message": "hi first"}
|
||||
|
||||
async def test_interrupt_after(self, version: str) -> None:
|
||||
graph = _build_two_step_graph()
|
||||
config = {"configurable": {"thread_id": "ia"}}
|
||||
await _drive_astream_events(graph, config, version, interrupt_after=["first"])
|
||||
state = await graph.aget_state(config)
|
||||
assert state.next == ("second",)
|
||||
assert state.values == {"message": "hi first"}
|
||||
|
||||
async def test_pre_drained_control(self, version: str) -> None:
|
||||
graph = _build_two_step_graph()
|
||||
config = {"configurable": {"thread_id": "drain"}}
|
||||
control = RunControl()
|
||||
control.request_drain("sigterm")
|
||||
with pytest.raises(GraphDrained, match="sigterm"):
|
||||
await _drive_astream_events(graph, config, version, control=control)
|
||||
|
||||
|
||||
class TestStreamEventsV3SyncInterrupts:
|
||||
"""Sync v3 static interrupts reach `stream()` after the kwargs rewire."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("kwarg", "node"),
|
||||
[("interrupt_before", "second"), ("interrupt_after", "first")],
|
||||
)
|
||||
def test_static_interrupt(self, kwarg: str, node: str) -> None:
|
||||
graph = _build_two_step_graph()
|
||||
config = {"configurable": {"thread_id": "sync"}}
|
||||
run = graph.stream_events(
|
||||
{"message": "hi"}, config, version="v3", **{kwarg: [node]}
|
||||
)
|
||||
list(run.values)
|
||||
state = graph.get_state(config)
|
||||
assert state.next == ("second",)
|
||||
assert state.values == {"message": "hi first"}
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.filterwarnings("ignore:astream_events version='v1' is deprecated")
|
||||
@pytest.mark.parametrize("version", ["v1", "v2", "v3"])
|
||||
class TestAstreamMidRunDrain:
|
||||
"""A drain requested from inside a node propagates out of `astream_events`.
|
||||
|
||||
This is the graceful-shutdown scenario: `request_drain()` called while
|
||||
the run is in flight (e.g. from a signal handler), with v1/v2 running
|
||||
inside core's event-stream task. The caller's own `RunControl` is used
|
||||
(the drain reason proves identity), `GraphDrained` propagates to the
|
||||
consumer, and the checkpoint keeps the pending step.
|
||||
"""
|
||||
|
||||
async def test_drain_requested_inside_first_node(self, version: str) -> None:
|
||||
control = RunControl()
|
||||
|
||||
def first(state: _State) -> dict[str, Any]:
|
||||
control.request_drain("sigterm-mid")
|
||||
return {"message": state["message"] + " first"}
|
||||
|
||||
graph = _build_two_step_graph(first)
|
||||
config = {"configurable": {"thread_id": "midrun"}}
|
||||
|
||||
with pytest.raises(GraphDrained, match="sigterm-mid"):
|
||||
await _drive_astream_events(graph, config, version, control=control)
|
||||
state = await graph.aget_state(config)
|
||||
assert state.next == ("second",)
|
||||
assert state.values == {"message": "hi first"}
|
||||
|
||||
|
||||
def _record_astream_kwargs(graph: Any) -> list[dict[str, Any]]:
|
||||
"""Patch `graph.astream` to record the kwargs each call receives."""
|
||||
received: list[dict[str, Any]] = []
|
||||
original = graph.astream
|
||||
|
||||
async def recording_astream(
|
||||
input: Any, config: Any = None, **kwargs: Any
|
||||
) -> AsyncIterator[Any]:
|
||||
received.append(kwargs)
|
||||
async for chunk in original(input, config, **kwargs):
|
||||
yield chunk
|
||||
|
||||
graph.astream = recording_astream # type: ignore[method-assign]
|
||||
return received
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.filterwarnings("ignore:astream_events version='v1' is deprecated")
|
||||
@pytest.mark.parametrize("version", ["v1", "v2"])
|
||||
class TestAstreamV1V2KwargsPassthrough:
|
||||
"""v1/v2 forward to `astream` exactly what the caller passed.
|
||||
|
||||
Pre-#7677 semantics: an explicit `None` is forwarded as `None`, and an
|
||||
omitted argument is not forwarded at all (so an override's own default
|
||||
would apply).
|
||||
"""
|
||||
|
||||
@pytest.mark.parametrize("name", _KWARG_NAMES)
|
||||
async def test_passed_values_reach_astream(self, version: str, name: str) -> None:
|
||||
graph = _build_two_step_graph()
|
||||
received = _record_astream_kwargs(graph)
|
||||
value: Any = RunControl() if name == "control" else ["second"]
|
||||
await _drive_astream_events(
|
||||
graph, {"configurable": {"thread_id": "rec"}}, version, **{name: value}
|
||||
)
|
||||
assert len(received) == 1
|
||||
assert received[0][name] == value
|
||||
if name == "control":
|
||||
assert received[0][name] is value
|
||||
|
||||
@pytest.mark.parametrize("name", _KWARG_NAMES)
|
||||
async def test_explicit_none_is_forwarded(self, version: str, name: str) -> None:
|
||||
graph = _build_two_step_graph()
|
||||
received = _record_astream_kwargs(graph)
|
||||
await _drive_astream_events(
|
||||
graph, {"configurable": {"thread_id": "rec-none"}}, version, **{name: None}
|
||||
)
|
||||
assert len(received) == 1
|
||||
assert received[0][name] is None
|
||||
|
||||
@pytest.mark.parametrize("name", _KWARG_NAMES)
|
||||
async def test_omitted_values_are_absent(self, version: str, name: str) -> None:
|
||||
graph = _build_two_step_graph()
|
||||
received = _record_astream_kwargs(graph)
|
||||
await _drive_astream_events(
|
||||
graph, {"configurable": {"thread_id": "rec-omit"}}, version
|
||||
)
|
||||
assert len(received) == 1
|
||||
assert name not in received[0]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
class TestAstreamV3KwargsDefaults:
|
||||
"""v3 keeps its since-inception explicit-default semantics (#7519).
|
||||
|
||||
Omitted `interrupt_before`/`interrupt_after`/`control` are supplied to
|
||||
`astream` as `None`; passed values are forwarded as-is.
|
||||
"""
|
||||
|
||||
async def test_omitted_values_arrive_as_none(self) -> None:
|
||||
graph = _build_two_step_graph()
|
||||
received = _record_astream_kwargs(graph)
|
||||
await _drive_astream_events(
|
||||
graph, {"configurable": {"thread_id": "v3-rec"}}, "v3"
|
||||
)
|
||||
assert len(received) == 1
|
||||
assert received[0]["control"] is None
|
||||
assert received[0]["interrupt_before"] is None
|
||||
assert received[0]["interrupt_after"] is None
|
||||
|
||||
@pytest.mark.parametrize("name", _KWARG_NAMES)
|
||||
async def test_passed_values_reach_astream(self, name: str) -> None:
|
||||
graph = _build_two_step_graph()
|
||||
received = _record_astream_kwargs(graph)
|
||||
value: Any = RunControl() if name == "control" else ["second"]
|
||||
await _drive_astream_events(
|
||||
graph, {"configurable": {"thread_id": "v3-rec-2"}}, "v3", **{name: value}
|
||||
)
|
||||
assert len(received) == 1
|
||||
assert received[0][name] == value
|
||||
if name == "control":
|
||||
assert received[0][name] is value
|
||||
|
||||
|
||||
def _record_stream_kwargs(graph: Any) -> list[dict[str, Any]]:
|
||||
"""Patch `graph.stream` to record the kwargs each call receives."""
|
||||
received: list[dict[str, Any]] = []
|
||||
original = graph.stream
|
||||
|
||||
def recording_stream(input: Any, config: Any = None, **kwargs: Any) -> Any:
|
||||
received.append(kwargs)
|
||||
yield from original(input, config, **kwargs)
|
||||
|
||||
graph.stream = recording_stream # type: ignore[method-assign]
|
||||
return received
|
||||
|
||||
|
||||
def _drive_stream_events_v3(graph: Any, config: dict[str, Any], **kwargs: Any) -> None:
|
||||
"""Consume a sync v3 stream_events run to completion."""
|
||||
run = graph.stream_events({"message": "hi"}, config, version="v3", **kwargs)
|
||||
list(run.values)
|
||||
|
||||
|
||||
class TestStreamEventsV3SyncKwargsDefaults:
|
||||
"""Sync v3 keeps its since-inception explicit-default semantics (#7519).
|
||||
|
||||
Mirror of `TestAstreamV3KwargsDefaults`: omitted
|
||||
`interrupt_before`/`interrupt_after`/`control` are supplied to `stream`
|
||||
as `None`; passed values are forwarded as-is. Pins the sync helper
|
||||
against a kwargs-only "simplification" that would change subclass
|
||||
default handling.
|
||||
"""
|
||||
|
||||
def test_omitted_values_arrive_as_none(self) -> None:
|
||||
graph = _build_two_step_graph()
|
||||
received = _record_stream_kwargs(graph)
|
||||
_drive_stream_events_v3(graph, {"configurable": {"thread_id": "s-rec"}})
|
||||
assert len(received) == 1
|
||||
assert received[0]["control"] is None
|
||||
assert received[0]["interrupt_before"] is None
|
||||
assert received[0]["interrupt_after"] is None
|
||||
|
||||
@pytest.mark.parametrize("name", _KWARG_NAMES)
|
||||
def test_passed_values_reach_stream(self, name: str) -> None:
|
||||
graph = _build_two_step_graph()
|
||||
received = _record_stream_kwargs(graph)
|
||||
value: Any = RunControl() if name == "control" else ["second"]
|
||||
_drive_stream_events_v3(
|
||||
graph, {"configurable": {"thread_id": "s-rec-2"}}, **{name: value}
|
||||
)
|
||||
assert len(received) == 1
|
||||
assert received[0][name] == value
|
||||
if name == "control":
|
||||
assert received[0][name] is value
|
||||
|
||||
Reference in New Issue
Block a user