This PR updates the dependencies in all Python packages using `uv lock
--upgrade`.
This is an automated PR created by the UV Lock Upgrade workflow.
Co-authored-by: sydney-runkle <54324534+sydney-runkle@users.noreply.github.com>
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]
```
This adds `StateSnapshot` to the union type annotation of
`CheckpointTask.state`.
The annotation was previously incomplete: `map_debug_checkpoint()`
generates `CheckpointPayload` objects from `PregelTask` objects, and the
`state` field in `PregelTask` is of type `None | RunnableConfig |
StateSnapshot`.
We currently only support auth on the default routes; we'd like to be
able to support it on all (non-meta/liveness probe) routes by default.
This is the first step in that direction.
The `$contains` auth operator supports subset containment checks, but
this has previously been undocumented. This updates `FilterType` and its
associated docstring to reflect this support.
### 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>
The original implementation for `refresh_on_read=True` in `asearch` for
AsyncSqliteStore used a CTE with an UPDATE statement, which is not
well-supported by SQLite in that specific construction, leading to a
syntax error.
This commit changes the approach:
1. `_prepare_batch_search_queries` in `BaseSqliteStore` no longer
constructs a CTE-based UPDATE. Instead, it returns a flag indicating if
TTL refresh is needed for the searched items.
2. `_batch_search_ops` in both `AsyncSqliteStore` and `SqliteStore` now
check this flag. If true, they perform a separate UPDATE statement after
fetching the search results to refresh the TTL of those items.
Additionally, a new test case `test_async_asearch_refresh_ttl` was added
and existing test logic was refined to accurately verify this behavior.
---------
Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
Co-authored-by: William FH <13333726+hinthornw@users.noreply.github.com>
Co-authored-by: Caspar Broekhuizen <caspar@langchain.dev>
This adds a configuration option in `HttpConfig` that allows LangGraph
Platform users to apply custom authentication hooks before (other)
custom middleware. Currently, the order is fixed (custom middleware is
always evaluated before custom auth).
(Apologies for the noise in
[de187a9](https://github.com/langchain-ai/langgraph/pull/6179/commits/de187a989e807c5687c22db1fc065d24030fa6b7),
apparently from the forced application of new linter rules.)
This PR updates the dependencies in all Python packages using `uv lock
--upgrade`.
This is an automated PR created by the UV Lock Upgrade workflow.
Co-authored-by: sydney-runkle <54324534+sydney-runkle@users.noreply.github.com>
### Description
Added unit tests for util.py.
Authored by @oumizx. Had to copy #6113 into this separate PR because
langgraph/libs/cli was having issues with secrets.
**Description:**
Add test for before and limit parameters for the list in SqliteSaver
which was marked as TODO.
---------
Co-authored-by: Caspar Broekhuizen <caspar@langchain.dev>
### Description
https://github.com/langchain-ai/langgraph/issues/6137 and
https://github.com/langchain-ai/langgraph/issues/5677 reported issues
where older checkpoints read by AsyncPostgresSaver/PostgresSaver from
`langgraph-checkpoint-postgres==2.0.19` fail to read channel values,
throwing `NoneType object is not a mapping`. This was due to a bug in
how `channel_values` is assembled:
```python
"channel_values": {
**value["checkpoint"].get("channel_values"), # <--- if channel_values doesn't exist (old checkpoint), **None errors
**self._load_blobs(value["channel_values"]),
},
```
This bug was observed for checkpoints generated by
`langgraph-checkpoint-postgres<=2.0.19`.
Fixed by providing a fallback to
`value["checkpoint"].get("channel_values")`:
```python
**value["checkpoint"],
"channel_values": {
**(
value["checkpoint"].get("channel_values") or {}
), # 'or {}' needed for backwards compat with v3 checkpoints and below, as v4 introduced channel_values key
**self._load_blobs(value["channel_values"]),
},
```
### Tests
Added test for AsyncPostgresSaver and test for PostgresSaver, using
monkeypatch to remove `channel_values` before CheckpointTuple is
assembled in `_load_checkpoint_tuple`.
### Solves
https://github.com/langchain-ai/langgraph/issues/6137 and
https://github.com/langchain-ai/langgraph/issues/5677
---------
Co-authored-by: Shahrukh Shaik <144558473+shahrukh-shaik@users.noreply.github.com>
**Description**: fix#6050.
Root cause: In nested graphs, the first tick after resume often included
a checkpoint_id, which set skip_done_tasks=False. This skipped matching
pending writes and re-executed already-completed helper @task on
subsequent resumes.
Change: Initialize skip_done_tasks=True when resuming inside a nested
graph. Use original config[CONF] for checkpoint_id presence, and
self.config[CONF] for resuming (current loop state). Added a concise
comment clarifying the different config sources.
**Issue**: #6050
**Tests**:
Add regression test `test_nested_graph_resume_reuses_cached_task_writes`
---------
Signed-off-by: jitokim <pigberger70@gmail.com>
Co-authored-by: Caspar Broekhuizen <casparbroekhuizen@gmail.com>
## Summary
- add a public accessor for the last received SSE event id
- retry async and sync SSE streams using the Location reconnect path and
Last-Event-ID while skipping empty events
- add regression tests that simulate interrupted SSE streams for both
async and sync clients
## Testing
- make format
- make lint
- make test
------
https://chatgpt.com/codex/tasks/task_e_68ca8bfa26cc832d98bcb359884962ec
### Description
`test_embed_with_path` was failing on x86_64 architecture due to numeric
precision differences. `pytest.approx` was already used later on in this
test for float comparison, so this PR just updates a missed assertion.
Fixes https://github.com/langchain-ai/langgraph/issues/5845
## Summary
- ensure both async and sync HTTP clients flush the SSE decoder after
streaming
- add regression tests covering trailing SSE events without a
terminating blank line
## Testing
- make format
- make lint
- make test
------
https://chatgpt.com/codex/tasks/task_e_68c9727ca9f8832d9f207323c5e02a72
### Description
Revert change in #5201 that prevented the surfacing of interrupts when
`stream_mode="values"`. [Comment highlighting affected
lines](https://github.com/langchain-ai/langgraph/pull/5201#discussion_r2344884841)
Resolves#5409
### Test
Add test to verify interrupts are properly surfaced when
`stream_mode="values"` (`test_interrupt_stream_mode_values`)
This PR updates the dependencies in all Python packages using `uv lock
--upgrade`.
This is an automated PR created by the UV Lock Upgrade workflow.
To make tests pass:
* linting fixes
* whitespace fixes in snapshots
---------
Co-authored-by: sydney-runkle <54324534+sydney-runkle@users.noreply.github.com>
Co-authored-by: Sydney Runkle <sydneymarierunkle@gmail.com>
**Description:**
This PR adds missing documentation for the `merge` parameter in the
`push_ui_message` function. The parameter was present in the function
signature but lacked documentation in the docstring, which could confuse
API users.
Changes made:
- Added clear documentation for the `merge` parameter
- Explains the behavior difference between `merge=True`
(merges props) and `merge=False` (replaces props)
- Includes default value information
**Issue:**
N/A - Documentation improvement
**Dependencies:**
None
Description:
Corrects _parse_version to support Docker versions with SemVer build
metadata (e.g., 28.1.1+1), resolving #5965. Adds comprehensive unit
tests for version parsing, including normal, v-prefixed, prerelease,
build metadata, combined prerelease/build metadata, and edge cases with
missing components.
Issue:
Closes#5965
Dependencies:
None
---------
Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>
### Description
* Fix `get_graph()` generating an unexpected conditional edge to
`__end__` when the last step has a single (non-terminal) source and the
graph is cyclic.
### Issue
* There was a fallback path that was triggered in `draw_graph()` when,
for a Pregel instance; no termini exist and there is only a single step
source (the last one). In this case an edge was added: (last source) ->
`__end__`, even when another node already had a valid edge: (node) ->
`__end__`.
See this example:
<details>
<summary>code</summary>
```python
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel
class State(TypedDict):
messages: list[str]
def chatbot_node(state: State) -> State:
return {"messages": state["messages"] + ["chatbot"]}
def tools_node(state: State) -> State:
return {"messages": state["messages"] + ["tools"]}
def human_node(state: State) -> State:
return {"messages": state["messages"] + ["human"]}
def tools_condition(_: State) -> str:
return "tools"
def end_condition(_: State) -> str:
return "chatbot"
workflow = StateGraph(State)
workflow.add_node("chatbot", chatbot_node)
workflow.add_node("tools", tools_node)
workflow.add_node("human", human_node)
workflow.add_edge(START, "human")
workflow.add_edge("tools", "chatbot")
# graph_builder.add_edge("chatbot", "human") !!!
workflow.add_conditional_edges(
"chatbot", tools_condition, {"tools": "tools", "human": "human"}
)
workflow.add_conditional_edges(
"human", end_condition, {"chatbot": "chatbot", END: END}
)
app = workflow.compile()
mermaid = app.get_graph().draw_mermaid()
```
</details>
The code above, as-is, generates the graph on the left. There is an
unexpected conditional edge: chatbot -> `__end__`. If you uncomment the
commented line and introduce a static edge: chatbot -> human,
`get_graph()` returns the correct representation:
1 Without `graph_builder.add_edge("chatbot", "human")` | 2 With
`graph_builder.add_edge("chatbot", "human")`
:-------------------------:|:-------------------------:
|
* In case 1), the graph is cyclic so termini is empty, and the last
`step_sources` set during the static walk contains only the chatbot
node, so an edge is added: chatbot -> `__end__`.
* In case 2), the graph is cyclic so termini is empty, and the last
`step_sources` set during the static walk contains only the human node,
so an edge is added: human -> `__end__`, but `add_edge()` dedups (the
edge already exists) so the graph appears correct.
### Solution
* Check that no valid edges: (node) -> `__end__` exist before triggering
the fallback path and creating an edge.
Before | After
:-------------------------:|:-------------------------:
|
After: The graph is cyclic so termini is empty, and the last
`step_sources` contains the chatbot node, but an edge already exists:
human -> `__end__`, so no more edges are added.
### Tests
* `test_get_graph_nonterminal_last_step_source()` which asserts no
unexpected edge to `__end__` is produced from the last nonterminal step
source.
### Issue
Closes#4394
---------
Co-authored-by: Sydney Runkle <sydneymarierunkle@gmail.com>
Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>
The idea here is that we don't want to allow resuming a graph w/ an
arbitrary resume value if there are multiple interrupts in the queue,
because the order in which interrupts enter the queue is not
deterministic. We want to instead enforce that each resume value is
mapped to an interrupt id.
Instead, when multiple interrupts are present, a user should invoke w/ a
resume map, mapping interrupt id -> resume value.
The logic was more complex than expected because there are 2 copies of
an interrupt in `checkpoint_pending_writes` for the cases w/ the
functional API, because an interrupt in a task interrupts the task and
entrypoint.
This is technically breaking (users resuming multiple hanging interrupts
w/ multiple resume calls can no longer do this... but the behavior for
this case was non-deterministic in the first place so we can sell this
as a fix).
### Description
Export PoolConfig from langgraph.store.postgres.__init__ so the
documented import from langgraph.store.postgres import
AsyncPostgresStore, PoolConfig works as shown in the AsyncPostgresStore
examples. This resolves a docs vs. code inconsistency without changing
behavior.
### Issue
N/A
### Dependencies:
None
---------
Co-authored-by: William FH <13333726+hinthornw@users.noreply.github.com>
Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>
### Description
* Set `ensure_ascii=False` for all `json.dumps` calls in
`get_text_at_path`. Preserves non-ASCII text instead of embedding
`\uXXXX` escapes.
**Before**
```python
store.put(("user_123", "memories"), "1", {"text": "这是中文"})
# embeds {"text": "\\u8fd9\\u662f\\u4e2d\\u6587"}
```
**After**
```python
store.put(("user_123", "memories"), "1", {"text": "这是中文"})
# embeds {"text": "这是中文"}
```
### Tests & Docs
* Add unit test `test_non_ascii` that writes three records (Chinese,
Japanese, Korean) to an `InMemoryStore`, searches with the same strings,
and asserts the correct top hit with a score >= 0.15 for each.
### Issue
Fixes#5946
**Description:** Fix a typo where "runtie" was incorrectly used instead
of "runtime" in line 158 of the StateGraph class in state.py. This
resolves the example error caused by the misspelled variable name.
**Dependencies:** None
basically - for conditional edges, we use this to merge the updates from
state with the state object (before the actual update really occurs in
the tick.after)
otherwise - an emphemeral value will actually last through the logic in
the conditional edge of the node after
We previously errored when a user had prerelease dependencies, this PR
passes the `--prereleases=allow` flag to our `uv pip install` call.
This PR also adds a test to verify that said deployments will build and
run as expected.
