Allows us to put some more information for tracing purposes (e.g.,
ls_integration) without dumping it into the streaming APIs (good for
performance)
Move some other metadata into tracing only since it's not needed in
streaming APIs
---------
Co-authored-by: Eugene Yurtsev <eyurtsev@gmail.com>
## Description
The `_build_server_info` function was reading `assistant_id` and
`graph_id` from `config["metadata"]`, but the server puts these values
in `config["configurable"]`. This updates the source to read from
`configurable` consistently.
## Test Plan
- [ ] Verify `server_info.assistant_id` and `server_info.graph_id` are
correctly populated from `config["configurable"]`
_Opened collaboratively by Sydney Runkle and open-swe._
Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>
Switches test_sync_streaming_with_functional_api to time.monotonic() for
both emitted task timestamps and observed arrival times so the assertion
is based on a monotonic clock instead of wall time. This makes the
streaming timing check less flaky on systems where time.time() can jump
or lack sufficient precision.
Created with [Deep Agents
CLI](https://docs.langchain.com/oss/python/deepagents/cli/overview)
using gpt-5.4 (provider: openai).
## Summary
This change adds first-class graph lifecycle callbacks to LangGraph so
interrupt and resume transitions can be observed without overloading the
existing LangChain custom event system. It introduces a dedicated graph
callback manager and wires lifecycle emission through Pregel execution
in both sync and async paths.
## Changes
- **`libs/langgraph/langgraph/callbacks.py`**: Adds
`GraphCallbackHandler` and `GraphCallbackManager` (built on LangChain
base callback classes), plus config plumbing via `graph_callbacks` and
`get_graph_callback_manager_for_config`.
- **`libs/langgraph/langgraph/pregel/_loop.py`**: Introduces
`GraphLifecycleEvent` and records lifecycle transitions (`resume`,
`interrupt`) into an internal FIFO queue with
`shift_graph_lifecycle_event()`.
- **`libs/langgraph/langgraph/pregel/main.py`**: Resolves graph callback
manager from config and drains lifecycle events while loop execution
progresses, dispatching `on_resume` and `on_interrupt` consistently in
sync and async runtimes.
- **`libs/langgraph/tests/test_graph_callbacks.py`**: Adds sync and
async coverage verifying lifecycle callbacks fire correctly and remain
distinct from LangChain `on_custom_event` handlers.
---------
Co-authored-by: Eugene Yurtsev <eyurtsev@gmail.com>
## Summary
- Fixes a bug where `runtime.execution_info` is `None` in the
distributed runtime (LangGraph Platform) because tasks are prepared by
the server and deserialized in the executor, bypassing the OSS
`_algo.py` code that normally creates `ExecutionInfo`
- Adds `_ensure_execution_info()` in `_retry.py` that lazily constructs
`ExecutionInfo` from the task config when it's missing, called in both
`run_with_retry` and `arun_with_retry`
## Why
PR #7363 introduced `ExecutionInfo` on `Runtime`, populated during task
preparation in `_algo.py`. However, on LangGraph Platform (distributed
runtime), tasks are serialized by the server and deserialized in a
separate executor process — `_algo.py` task prep never runs in that
context, so `execution_info` remains `None`. Any user code or tooling
relying on `runtime.execution_info` (e.g. for tracing, logging, or auth)
would fail with `None` fields.
The fix reconstructs `ExecutionInfo` from config keys (`checkpoint_id`,
`checkpoint_ns`, `task_id`, `thread_id`, `run_id`) that are already
present in the deserialized task config, so no server-side changes are
needed.
## Test plan
- [x] `_ensure_execution_info` is a no-op when `execution_info` already
exists
- [x] `_ensure_execution_info` creates `ExecutionInfo` from config when
missing
- [x] Falls back to `task.id` when `CONFIG_KEY_TASK_ID` is absent from
config
- [x] End-to-end `run_with_retry` test simulating the distributed
runtime scenario (runtime with `execution_info=None`)
## Summary
Enhances `ExecutionInfo` and `Runtime` to surface richer execution
context and introduces `ServerInfo` for LangGraph Server metadata.
### `ExecutionInfo` expansion
Converted from `NamedTuple` to a frozen `dataclass`. Added identity
fields populated during task preparation in `_algo.py`:
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `checkpoint_id` | `str` | required | Checkpoint ID for the current
execution |
| `checkpoint_ns` | `str` | required | Checkpoint namespace for the
current execution |
| `task_id` | `str` | required | Task ID for the current execution |
| `thread_id` | `str \| None` | `None` | Thread ID (None without a
checkpointer) |
| `run_id` | `str \| None` | `None` | Run ID (None when not provided in
config) |
| `node_attempt` | `int` | `1` | Current node execution attempt number
(1-indexed) |
| `node_first_attempt_time` | `float \| None` | `None` | Unix timestamp
for when the first attempt started |
`checkpoint_id`, `checkpoint_ns`, and `task_id` are required (no
defaults) — they are always populated during task preparation in
`_algo.py`. `Runtime.execution_info` is `None` until that point.
### New `ServerInfo` type
Frozen dataclass with `assistant_id: str`, `graph_id: str`, and optional
`user: BaseUser | None`. Populated from config metadata (`assistant_id`,
`graph_id`) and `configurable["langgraph_auth_user"]` via
`_build_server_info()` in `pregel/main.py`.
User detection uses `isinstance(BaseUser)` with a `hasattr("identity")`
fallback — needed because the server's `ProxyUser` provides
`permissions` via `__getattr__`, which Python's `runtime_checkable`
Protocol check doesn't see.
### `Runtime` changes
- `execution_info` is now `ExecutionInfo | None` (default `None`), set
during task prep
- Added `server_info: ServerInfo | None` field, wired through `merge()`
and `override()`
### `ToolNode` / `ToolRuntime` forwarding
`execution_info` and `server_info` are forwarded from `Runtime` to
`ToolRuntime` so tools can access execution and server context.
### New public API surface
```python
from langgraph.runtime import BaseUser, ExecutionInfo, Runtime, ServerInfo, get_runtime
```
## Test plan
- [x] `ExecutionInfo` defaults, patch, and frozen behavior
- [x] Integration tests verifying identity fields are populated in sync
and async execution
- [x] Retry tests confirming identity fields persist and `node_attempt`
increments
- [x] `ServerInfo` construction, frozen behavior, and `Runtime.merge`
precedence
- [x] `server_info` populated from config metadata and
`langgraph_auth_user` (including starlette-style proxy user)
- [x] `server_info` is `None` when no server metadata present
- [x] `ToolRuntime` forwarding of `execution_info` and `server_info`
🤖 Generated with [Claude Code](https://claude.com/claude-code)
- [x] **Add tests and docs**: If you're adding a new integration, you
must include:
1. A test for the integration, preferably unit tests that do not rely on
network access,
2. An example notebook showing its use. It lives in
`docs/docs/integrations` directory.
- [x] **Lint and test**: Run `make format`, `make lint` and `make test`
from the root of the package(s) you've modified. We will not consider a
PR unless these three are passing in CI. See [contribution
guidelines](https://docs.langchain.com/oss/python/contributing/overview)
for more.
Additional guidelines:
- Make sure optional dependencies are imported within a function.
- Please do not add dependencies to `pyproject.toml` files (even
optional ones) unless they are **required** for unit tests.
- Most PRs should not touch more than one package.
- Changes should be backwards compatible.
## Summary
Fixes two bugs when time-traveling to a subgraph checkpoint
(`graph.invoke(None, sub_config)`):
1. **Wrong checkpoint loaded:** The old `__enter__` checked
`replay_state` before explicit `checkpoint_id` for nested graphs. When
time-traveling, `checkpoint_map` resolves a specific checkpoint_id for
the target subgraph, but `replay_state.get_checkpoint()` would find an
*earlier* checkpoint (via `list(before=parent_ckpt_id)`), causing the
subgraph to re-execute from the beginning (step_a, ask_1, etc. all
re-ran). Fix: check for explicit `checkpoint_id` first.
2. **Stale RESUME writes kept:** The parent sets `RESUMING=True` on
subgraph configs — it can't distinguish time-travel from normal resume.
The subgraph sees `RESUMING=True` and preserves old RESUME writes, so
`interrupt()` returns cached answers instead of re-firing. Fix: check
whether the subgraph's own namespace appears in `checkpoint_map` (it
only does during time-travel — normally the map only has ancestor
entries). When present, force-strip RESUME writes.
## Test plan
Tests cover time travel (replay + fork) at different interrupt points
for both single-nested and double-nested subgraphs, including cases
where the middle subgraph itself has interrupts. All tests have sync and
async variants.
### Single nested (2 levels: parent → executor subgraph)
- `test_subgraph_time_travel_to_first_interrupt` — time travel to 1st
interrupt checkpoint; verifies step_a doesn't re-run, ask_1 re-fires
(replay + fork)
- `test_subgraph_time_travel_to_second_interrupt` — time travel to 2nd
interrupt checkpoint; verifies step_a and ask_1 don't re-run, ask_2
re-fires (replay + fork)
- `test_subgraph_time_travel_after_completion` — replay from final
parent checkpoint after full completion; verifies no nodes re-run and
all state values preserved
### Double nested (3 levels: parent → outer → inner subgraph)
- `test_3_levels_deep_time_travel_to_first_interrupt` — time travel to
innermost checkpoint at 1st interrupt (replay + fork)
- `test_3_levels_deep_time_travel_to_second_interrupt` — time travel to
innermost checkpoint at 2nd interrupt (replay + fork)
- `test_3_levels_deep_time_travel_to_middle_subgraph` — time travel to
middle-level subgraph checkpoint (replay + fork)
- `test_3_levels_deep_middle_has_interrupts` — middle subgraph has its
own `interrupt()` calls (in `pre` node) plus an inner subgraph with
interrupts; time travel to middle checkpoint at each interrupt point
(replay + fork at both the middle's own interrupt and the inner's
interrupt)
## Summary
Fix time travel (replay and fork) for graphs with interrupts and
subgraphs.
## Problem
Two issues with replaying/forking from earlier checkpoints:
1. **Stale interrupt values during replay** — Replays incorrectly reused
cached `RESUME` values from prior `interrupt()` calls, so interrupts
silently returned stale answers instead of re-firing.
2. **Wrong subgraph state during time travel** — Subgraphs always loaded
their **latest** checkpoint instead of the one corresponding to the
parent's historical state. This caused subgraphs to skip execution or
produce incorrect results during replay/fork.
## Changes
Code changes span `libs/langgraph/langgraph/pregel/_loop.py`,
`libs/langgraph/langgraph/_internal/_constants.py`, and a new
`libs/langgraph/langgraph/_internal/_replay.py` module:
- **Strip stale `RESUME` writes on replay** — During replays, cached
`RESUME` writes are filtered out so `interrupt()` re-fires instead of
returning old values. Genuine resumes (`Command(resume=...)`) preserve
these writes.
- **Rename `skip_done_tasks` → `is_replaying`** — Clearer naming for the
flag that tracks whether the current run is replaying from a specific
checkpoint.
- **New `ReplayState` class (`_replay.py`)** — Encapsulates subgraph
checkpoint loading during time-travel. Tracks a parent checkpoint ID
upper bound and which subgraph namespaces have already loaded their
pre-replay checkpoint. On the first visit to a subgraph namespace, it
loads the latest checkpoint created *before* the replay point (via
`checkpointer.list(..., before=...)` with `limit=1`). On subsequent
visits (e.g. the same subgraph in a later loop iteration), it falls back
to normal latest-checkpoint loading. The task-id suffix is stripped from
namespaces so the same logical subgraph is recognized across loop
iterations.
- **New `CONFIG_KEY_REPLAY_STATE` config key** — The parent graph
creates a `ReplayState` instance and passes it to subgraphs via config.
For forks (`source=update`), the replay state uses the fork's parent
checkpoint ID since the fork was created after the subgraph's original
checkpoints. The single `ReplayState` instance is shared by reference
across all derived configs within one parent execution.
- **Subgraph checkpoint loading in `__enter__`/`__aenter__`** — When a
subgraph detects a `ReplayState` in its config, it delegates checkpoint
loading to `ReplayState.get_checkpoint`/`aget_checkpoint` instead of
using the default `get_tuple`. It also clears `CONFIG_KEY_RESUMING` so
`_first` re-applies input and recreates ephemeral routing channels.
## Tests
New test files `test_time_travel.py` (~2500 lines) and
`test_time_travel_async.py` (~2200 lines) covering:
- Replay and fork with interrupts (single and multiple)
- Replay and fork for graphs with and without subgraphs
- Correct subgraph checkpoint restoration during parent time travel
- `get_state` with subgraph state during replay
Fixes incorrect `Command.PARENT` bubbling when checkpoint namespaces
include numeric task-disambiguation segments like `|1`. In some
nested-invoke/fanout scenarios, the runtime inserts a purely-numeric
namespace segment between `name:task_id` segments (e.g.
`parent_first:<id>|1|node:<id>`). The previous ParentCommand rewrite
logic only handled numeric segments at the end of the namespace, which
could produce a malformed parent graph identifier (e.g.
`parent_first:<id>|1`) and prevent the command from routing to the
intended parent node.
This change normalizes checkpoint namespaces by dropping numeric
segments before computing the parent namespace in both sync and async
retry paths. Added a minimal regression test that exercises the
nested-invoke case and asserts that `Command(graph=Command.PARENT,
goto=...)` reliably routes to the parent graph, regardless of whether
the jump comes from the first or second nested invocation.
In the original proposed fix
https://github.com/langchain-ai/langgraph/pull/6863, we were guarding
this specific case by only checking for null resume in the scratchpad.
But I noticed the tests passed on main for the sync case, indicating an
inconsistency between the two paths.
Seems like we were making a redundant put_writes call when replaying the
async tasks
---------
Co-authored-by: William FH <13333726+hinthornw@users.noreply.github.com>
Problem: `StreamMessagesHandler.on_chain_start` only populated the seen
set for dict inputs, so Pydantic/dataclass state, objects were skipped —
causing duplicate messages after interrupt/resume
Solution: Extract a shared `_state_values()` helper that handles dict,
BaseModel, and dataclass inputs uniformly, and use it in both
on_chain_start and on_chain_end
Also replaces the fragile dir()/getattr fallback in on_chain_end with
the same helper
Tests: Added test_stream_messages_dedupe_pydantic_subgraph_interrupt —
confirms no duplicate message IDs across interrupt/resume with Pydantic
subgraph state (fails on main, passes with fix)
`aiosqlite` changed it's Connection type to no longer subclass
`threading.Thread`. This removed the is_alive method, which is called
proactively in setup().
This PR handles this in a backwards compat way.
Changed "BaseMessge" to "BaseMessage" in test comments.
This critical 2-character fix prevents mass confusion among developers
who might have spent milliseconds wondering what a "Messge" is.
The world is now a safer place.
Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
Description: Catch invalid checkpointer objects early by validating any
checkpointer argument before compilation/execution, raising a clear
TypeError that instructs users to pass a proper BaseCheckpointSaver
(e.g., AsyncPostgresSaver) instead of stores like AsyncPostgresStore.
Includes shared validation logic and a regression test so we don’t see
AttributeError: 'AsyncPostgresStore' object has no attribute
'get_next_version' again.
Issue: Fixes#6585
Dependencies: None
Twitter handle: none
- **Description:** Provide the id of the event for routes that use SSE
streams. This will allow for more custom retry logic when streams
disconnect if needed.
- **Issue:** N/A
- **Dependencies:** N/A
- **Twitter handle:** N/A
---------
Co-authored-by: William FH <13333726+hinthornw@users.noreply.github.com>
Otherwise, you cannot use `context` with stateful runs, because the
server throws if you provide both configurable and context in a single
call (due to ambiguous parameters)
This PR improves the consistency of interrupt streaming.
- when streaming with stream_mode values, the stream chunk now contains
the entire state alongside the interrupt:
```python
class State(TypedDict):
robot_input: str
# at this point in time robot_input is already set to "beep boop i am a robot"
app.stream(..., stream_mode="values")
# before
{"__interrupt__": (Interrupt(value="interrupt",))}}
# after
{"robot_input": "beep boop i am a robot", "__interrupt__": (Interrupt(value="interrupt"))}
```
- when streaming with stream_mode=["values", "updates"], interrupts are
surfaced in both an update stream chunk and the value stream chunk, when
previously we keep interrupt in values only if we request values mode
only
```python
class State(TypedDict):
robot_input: str
# at this point in time robot_input is already set to "beep boop i am a robot"
app.stream(..., stream_mode=["values", "updates"])
# before (interrupt would only emit on update chunk, there would be no values chunk)
("updates", {"__interrupt__": (Interrupt(value="interrupt",))}})
# after
("updates", {"__interrupt__": (Interrupt(value="interrupt",))}})
("values", {"robot_input": "beep boop i am a robot", "__interrupt__": (Interrupt(value="interrupt"))})
```
For housekeeping: this PR improves on this revert:
https://github.com/langchain-ai/langgraph/pull/6141
PR #6195 fixed `bulk_update_state` to populate `task.result` by calling
`prepare_next_tasks` to discover task IDs. Before #6195,
prepare_next_tasks was gated by the condition `CONFIG_KEY_CHECKPOINT_ID
not in config[CONF]` - so it only ran if we were resuming from an empty
checkpoint. This check was removed in order to properly populate task
results. However, the removal of this check inadvertently applied
pending writes during manual state updates which caused issues when
forking:
- When you fork from a checkpoint by calling `update_state(config,
new_values, as_node="mynode")`, pending writes from the original
execution were being applied
- This caused stale data to leak into forked threads (eg. old tool call
results appearing in forked execution)
Changes
Removed pending writes application from `bulk_update_state` and
`abulk_update_state`:
- Still call `prepare_next_tasks` to discover task IDs, but skip the
code that applies null writes and regular pending writes
Tests
- Added `test_fork_does_not_apply_pending_writes` for sync and async
which verifies forking doesn't include stale pending writes from
original execution
bumping core dependency for `langgraph-prebuilt` to `>1.0.0` so that we
can take advantage of internal utils that allow `ToolRuntime` injection.
We were previously bumping the version in lock step with prebuilt (prev
version was 0.3.67), so this pattern is in line with that.
Also updating snapshots accordingly:
* New mermaid syntax for a few graphs
* Removal of `examples` from `AIMessage`
UntrackedValue is a special channel type where the values in it are not
persisted to memory. Our v1 create_agent middleware used UntrackedValue
in middleware (e.g. ShellToolMiddleware) for some cool features like
temp files.
If a user has elected to use a checkpointer, we normally enforce that
the values they write to channels are serializable. However, this
doesn't make sense to enforce for UntrackedValues because the contract
is they're never written to checkpoint - so the user should not be
forced to make the contents of the channel serializable
However when using a checkpointer and durability sync/async, we found
that writes would still be persisted that contained UntrackedValue
contents in two forms:
a) UntrackedValue channel objects
b) Send objects - in the state passed to another node
Patched this in put_writes by a) skipping persisting writes to
UntrackedValue channels altogether and b) popping all UntrackedValue kv
pairs nested within Send packets. We also need to sanitize in
_put_checkpoint which is called when durability=="exit".
Added a basic test for UntrackedValue in test_channel.py and added more
comprehensive tests using Send under some different scenarios in
test_pregel.py
See https://github.com/langchain-ai/langgraph/pull/6277
Adds langgraph.types.Overwrite, a deterministic way to bypass a reducer.
When encountering a value wrapped with Overwrite,
BinaryOperatorAggregate overwrites the channel value.
<img width="227" height="329" alt="image"
src="https://github.com/user-attachments/assets/f2136117-9aa3-4246-863d-d5df0e7d1df1"
/>
If either node_b or node_c overwrite (but not both), then at END the
channel is equal to the value node_b or node_c wrote. Order of execution
doesn't matter because once an Overwrite value is encountered, regular
values are ignored (self.operator is not called for the rest of the
update)
If multiple nodes overwrite in the same superstep then
InvalidUpdateError is thrown
Usage
```python
from langgraph.types import Overwrite
def node_b(state:State):
return {"messages": Overwrite(["b"])}
```
or
``` python
def node_b(state:State):
return {"messages": {"__overwrite__": ["b"]}}
```
* catching error thrown by asyncio
* using 2nd check for annotations given Pydantic 2.12 changes
* skipping tests for remote graph bc langgraph-api is dependent on
`jsonschema-rs`
* skipping tests w/ pydantic v1 models
```bash
hint: This usually indicates a problem with the package or the build environment.
help: `jsonschema-rs` (v0.29.1) was included because `langgraph:dev` (v1.0.0rc1) depends on `langgraph-cli[inmem]` which
depends on `langgraph-api` (v0.4.29) which depends on `jsonschema-rs`
```
not yet testing for free threaded python, that'll be much more involved!
ended up separating lint / testing deps during this process bc I was
getting a ton of not required deps while testing that were complicating
things :/
some of these changes were obvious, and some were less obvious. In a few
spots, it felt like a judgement call if we should be saying LangSmith
Deployment of LangGraph Server. But hopefully either works.
### Description
Prevents interrupt tasks from executing when the resume value has not
yet been specified.
Implemented for sync and async Pregel loop
If a task execution is skipped, the skipped interrupt is still included
in the graph result for consistency:
``` python
result = graph.invoke(...)
interrupts = result.get("__interrupt__", []) # [interrupt_1, interrupt_2]
partial_result = graph.invoke(Command(resume=interrupt_1_resume_map), ...)
remaining_interrupts = partial_result.get("__interrupt__", []) # [interrupt_2]
```
### Tests
- `test_interrupt_with_send_payloads`: test for a single resume map that
resumes all interrupts at once
- `test_interrupt_with_send_payloads_sequential_resume`: test for two
resume maps delivered in sequence
- `test_node_with_multiple_interrupts_requires_full_resume` test
optimization for multiple interrupts within a single node
Solves https://github.com/langchain-ai/langgraph/issues/6208
---------
Co-authored-by: Eugene Yurtsev <eyurtsev@gmail.com>
### Description
Fix `bulk_update_state` and `abulk_update_state` so history populates
`tasks[*].result` when creating state via supersteps.
There was a branch in these functions that I'm guessing was meant to be
triggered when a `StateUpdate.as_node` was the name of a real node (not
`"__input__"` or `"__copy__"`), but was never being triggered because of
a condition `CONFIG_KEY_CHECKPOINT_ID not in config[CONF]`:
```python
# apply pending writes, if not on specific checkpoint
if (
CONFIG_KEY_CHECKPOINT_ID not in config[CONF]
and saved is not None
and saved.pending_writes
):
next_tasks = prepare_next_tasks(...)
```
From what I can tell, in the bulk-update flow every superstep carries a
`checkpoint_id`, so the condition was always false. That skipped
`prepare_next_tasks(...)` and prevented us from discovering the task IDs
that we would need to attach the task result. So, I removed this check.
I also replaced the `pending_writes` check with a more lenient one (just
check it is not None to satisfy type checkers). I found that
`saved.pending_writes` was sometimes just `[]`, and in this case we
would skip `prepare_next_tasks(...)` and never attach the task result.
Now for each task discovered in `prepare_next_tasks(...)`, I collect the
task IDs and reuse them when running all writers of the chosen node
(applying the updates).
### Tests
- `test_supersteps_populate_task_results` for `PregelLoop` and
`AsyncPregelLoop`
These tests build a single node graph and compare history from two
threads: one uses `.invoke` and the other is build from supersteps. Both
tests fail on main and pass with this PR.
### Issue
Solves https://github.com/langchain-ai/langgraph/issues/6206
This PR ensures that even if a type has multiple annotations, we can
still detect the `BaseChannel` subclasses attached.
```py
class State(TypedDict):
# recognized as EphemeralValue(int)
foo: Annotated[int, EphemeralValue]
# now recognized as EphemeralValue(int)
bar: Annotated[int, EphemeralValue, OtherMetadata]
# now recognized as EphemeralValue(int)
baz: Annotated[int, SomeMetadata, EphemeralValue, OtherMetadata]
```
### Summary
This PR fixes an issue where `AsyncPregelLoop` could leave behind an
orphaned `stream.wait()` task, resulting in warnings like:
```
Task was destroyed but it is pending!
```
### Related Discussion
This PR is in response to:
[langchain-ai/langgraph#6163](https://github.com/langchain-ai/langgraph/discussions/6163)
### Problem
* In the async path, `get_waiter()` was creating a new `asyncio.Task`
via
```python
aioloop.create_task(stream.wait())
```
but never tracked or cleaned it up.
* On cancellation or shutdown, these tasks remained pending and produced
warnings.
### Solution
* Changed `get_waiter()` to:
* Maintain a **single waiter task** (similar to the sync path).
* Auto-clear the reference when the task finishes.
* Added `_cleanup_waiter()`:
* On exit, attempt to wake the waiter (`stream._count.release()` if
available).
* Otherwise, cancel and `await` the pending task to ensure proper
cleanup.
* Wrapped the `while loop.tick():` block in a `try/finally` to guarantee
`_cleanup_waiter()` runs on exit.
* Added missing `import contextlib`.
### Impact
* Prevents orphaned `stream.wait()` tasks.
* Removes noisy `"Task was destroyed but it is pending!"` warnings.
* Behavior of async streaming remains unchanged, only lifecycle
management improved.
### Test Plan
* Reproduced the issue by running async streaming with cancellation.
* Verified warnings no longer appear after the fix.
* Ran existing test suite (all passing).
### Notes
* Sync and Async implementations now follow the same principle: *only
one waiter at a time, always cleaned up on exit*.
* Backwards-compatible; no API changes.
### Repro & Verification
To confirm the issue and the fix I used the following minimal repro
snippet:
```python
# lg_repro.py
import asyncio
import os
# Enable asyncio debug logs to surface pending task warnings
os.environ.setdefault("PYTHONASYNCIODEBUG", "1")
from langgraph.graph import START, END, StateGraph
State = dict
# Slow async node: processes once, then sleeps to keep the waiter alive
async def slow_node(state: State) -> State:
await asyncio.sleep(0.2) # simulate work
state["count"] = state.get("count", 0) + 1
await asyncio.sleep(1.0) # keep stream.wait() waiter active
return state
# Build simple graph: START -> slow_node -> END
builder = StateGraph(State)
builder.add_node("slow", slow_node)
builder.add_edge(START, "slow")
builder.add_edge("slow", END)
graph = builder.compile()
async def run_and_cancel():
# astream with messages mode triggers internal stream.wait() waiter
async def consumer():
async for _ in graph.astream({"msg": "hi"}, stream_mode="messages"):
await asyncio.sleep(0.05)
t = asyncio.create_task(consumer(), name="astream-consumer")
# Allow the stream to start, then cancel the consumer
await asyncio.sleep(0.1)
t.cancel()
try:
await t
except asyncio.CancelledError:
pass
# Let loop settle to show pending waiter task if not cleaned
await asyncio.sleep(0.05)
def main():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.set_debug(True)
try:
loop.run_until_complete(run_and_cancel())
finally:
# If the internal waiter is not cleaned, closing the loop will warn
loop.close()
if __name__ == "__main__":
main()
````
**How to run**
```powershell
# Before (main branch)
git checkout main
pip install -e libs/langgraph
$env:PYTHONASYNCIODEBUG=1; python lg_repro.py
# After (patched branch)
git checkout async-waiter-cleanup
pip install -e libs/langgraph
$env:PYTHONASYNCIODEBUG=1; python lg_repro.py
```
**Observed results**
* **main branch (before fix):**
Shows warnings like:
```
Task was destroyed but it is pending!
... coro=<AsyncQueue.wait() ...>
created at langgraph/pregel/main.py:2927
```
* **patched branch (after fix):**
No warnings. The single waiter is properly cleaned up on exit via
`_cleanup_waiter()` (release semaphore if available, then cancel/await).
---
This confirms that the patch removes the orphaned `stream.wait()` task
and prevents
`"Task was destroyed but it is pending!"` warnings during
cancellation/shutdown.
---------
Co-authored-by: Caspar Broekhuizen <caspar@langchain.dev>
