### 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 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`.
### 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>
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**: 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>
### 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
* 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:** 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
This PR introduces the `--build-command` and `--install-command`
arguments to `langgraph build`.
`--install-command` is a custom install command. If passed, it will be
run from wherever the `langgraph build` call was made, i.e. NOT where
the langgraph.json file lives (except if these are the same place). This
will override the detected install command that we previously used.
`--build-command` is a custom build command. This will run from wherever
the langgraph.json file lives, and will be done after the install has
been run.
You don't need to provide both. Just providing one will make the install
(detected or supplied) run in the directory from where `langgraph build
was called` and then have the build command (if one exists) run in the
directory where langgraph.json exists.
I think we should probably allow configuring the directories from which
these commands get run, but I don't think this needs to be part of the
MVP.
---------
Co-authored-by: William FH <13333726+hinthornw@users.noreply.github.com>
Thank you for contributing to LangGraph! Follow these steps to mark your
pull request as ready for review. **If any of these steps are not
completed, your PR will not be considered for review.**
- [ ] **PR title**: Follows the format: {TYPE}({SCOPE}): {DESCRIPTION}
- Examples:
- feat(core): add multi-tenant support
- fix(cli): resolve flag parsing error
- docs(openai): update API usage examples
- Allowed `{TYPE}` values:
- feat, fix, docs, style, refactor, perf, test, build, ci, chore,
revert, release
- Allowed `{SCOPE}` values (optional):
- langgraph, docs, cli, checkpoint, checkpoint-postgres,
checkpoint-sqlite, prebuilt, scheduler-kafka, sdk-py
- Once you've written the title, please delete this checklist item; do
not include it in the PR.
- [ ] **PR message**: ***Delete this entire checklist*** and replace
with
- **Description:** a description of the change. Include a [closing
keyword](https://docs.github.com/en/issues/tracking-your-work-with-issues/using-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword)
if applicable.
- **Issue:** the issue # it fixes, if applicable
- **Dependencies:** any dependencies required for this change
- **Twitter handle:** if your PR gets announced, and you'd like a
mention, we'll gladly shout you out!
- [ ] **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.
- [ ] **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://github.com/langchain-ai/langgraph/blob/main/CONTRIBUTING.md)
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.
### Description
Adds Redis as a supported cache backend for LangGraph node-level
caching, enabling distributed caching across multiple processes/servers.
This implementation follows the same patterns as existing InMemoryCache
and SqliteCache.
### Key changes
- New RedisCache class implementing the BaseCache interface
- Support for TTL-based expiration and batch operations
- Worker-specific cache prefixes for parallel test isolation
### Dependencies
- redis package (already included in dev dependencies)
### Test Plan
- Unit tests: Added Redis cache tests covering basic operations, TTL,
batch operations, and error handling
- Integration tests: Redis cache integrated into existing LangGraph test
suite, tested with all checkpointer combinations
William Fu-Hinthorn