Script 'mail_helper' called by obssrc Hello community, here is the log from the commit of package python-langgraph for openSUSE:Factory checked in at 2026-10-17 21:26:57 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ Comparing /work/SRC/openSUSE:Factory/python-langgraph (Old) and /work/SRC/openSUSE:Factory/.python-langgraph.new.3698736 (New) ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Package is "python-langgraph" Sat Oct 17 21:26:57 2026 rev:11 rq:1385353 version:1.2.16 Changes: -------- --- /work/SRC/openSUSE:Factory/python-langgraph/python-langgraph.changes 2026-10-15 12:25:40.513964789 +0200 +++ /work/SRC/openSUSE:Factory/.python-langgraph.new.3698736/python-langgraph.changes 2026-10-17 21:27:00.553902447 +0200 @@ -1,0 +2,10 @@ +Sat Oct 17 06:28:17 UTC 2026 - Martin Pluskal <[email protected]> + +- Update to 1.2.16: + * BinOpChannel restores threads written while the field + was a DeltaChannel + * Pregel tracks tasks written during the run for exit mode + * An invalid resume value no longer cuts off a task + answered in the same resume + +------------------------------------------------------------------- Old: ---- langgraph-1.2.15.tar.gz New: ---- langgraph-1.2.16.tar.gz ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ Other differences: ------------------ ++++++ python-langgraph.spec ++++++ --- /var/tmp/diff_new_pack.GHuZjU/_old 2026-10-17 21:27:01.230930695 +0200 +++ /var/tmp/diff_new_pack.GHuZjU/_new 2026-10-17 21:27:01.231930736 +0200 @@ -17,7 +17,7 @@ Name: python-langgraph -Version: 1.2.15 +Version: 1.2.16 Release: 0 Summary: Library for building stateful, multi-actor applications with LLMs License: MIT ++++++ langgraph-1.2.15.tar.gz -> langgraph-1.2.16.tar.gz ++++++ diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/PKG-INFO new/langgraph-1.2.16/PKG-INFO --- old/langgraph-1.2.15/PKG-INFO 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/PKG-INFO 2020-02-02 01:00:00.000000000 +0100 @@ -1,6 +1,6 @@ Metadata-Version: 2.5 Name: langgraph -Version: 1.2.15 +Version: 1.2.16 Summary: Building stateful, multi-actor applications with LLMs Project-URL: Homepage, https://docs.langchain.com/oss/python/langgraph/overview Project-URL: Documentation, https://reference.langchain.com/python/langgraph/ diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/langgraph/channels/binop.py new/langgraph-1.2.16/langgraph/channels/binop.py --- old/langgraph-1.2.15/langgraph/channels/binop.py 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/langgraph/channels/binop.py 2020-02-02 01:00:00.000000000 +0100 @@ -2,6 +2,7 @@ from collections.abc import Callable, Sequence from typing import Any, Generic +from langgraph.checkpoint.serde.types import _DeltaSnapshot from typing_extensions import NotRequired, Required, Self from langgraph._internal._constants import OVERWRITE @@ -116,6 +117,9 @@ def from_checkpoint(self, checkpoint: Value) -> Self: empty = self.__class__(self.typ, self.operator) empty.key = self.key + # A thread written while this field was a DeltaChannel stores snapshots. + if isinstance(checkpoint, _DeltaSnapshot): + checkpoint = checkpoint.value if checkpoint is not MISSING: empty.value = checkpoint return empty diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/langgraph/pregel/_checkpoint.py new/langgraph-1.2.16/langgraph/pregel/_checkpoint.py --- old/langgraph-1.2.15/langgraph/pregel/_checkpoint.py 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/langgraph/pregel/_checkpoint.py 2020-02-02 01:00:00.000000000 +0100 @@ -4,7 +4,7 @@ from collections.abc import Callable, Iterable, Mapping, Sequence from datetime import datetime, timezone from inspect import signature -from typing import Any, Literal, cast +from typing import Any, Literal from langchain_core.runnables import RunnableConfig from langgraph.checkpoint.base import ( @@ -12,6 +12,7 @@ ChannelVersions, Checkpoint, CheckpointTuple, + DeltaChannelHistory, PendingWrite, ) from langgraph.checkpoint.base.id import uuid6 @@ -32,7 +33,7 @@ ) from langgraph._internal._typing import MISSING from langgraph.channels.base import BaseChannel -from langgraph.channels.binop import _get_overwrite +from langgraph.channels.binop import BinaryOperatorAggregate, _get_overwrite from langgraph.channels.delta import DeltaChannel from langgraph.managed.base import ManagedValueMapping, ManagedValueSpec @@ -409,23 +410,39 @@ def _delta_channels_to_replay( specs: Mapping[str, BaseChannel], checkpoint: Checkpoint ) -> list[str]: - """DeltaChannels whose value at `checkpoint` the ancestor walk rebuilds. + """Channels whose value at `checkpoint` the ancestor walk rebuilds. A `_DeltaSnapshot` blob or a plain value (migration) resolves directly via `from_checkpoint`, so only a channel with nothing stored here needs the walk. A channel with no version was never written, so it is empty without one; a walk for it would find no snapshot to stop at and read every ancestor, every time the thread is loaded. + + A `BinaryOperatorAggregate` stores its value at every version, so one with + nothing stored was written as a `DeltaChannel` before its annotation + changed, and is rebuilt the same way. """ return [ k for k, spec in specs.items() - if isinstance(spec, DeltaChannel) + if isinstance(spec, (DeltaChannel, BinaryOperatorAggregate)) and k in checkpoint["channel_versions"] and checkpoint["channel_values"].get(k, MISSING) is MISSING ] +def _replay(spec: BaseChannel, history: DeltaChannelHistory) -> BaseChannel: + ch = spec.from_checkpoint(history.get("seed", MISSING)) + if isinstance(ch, DeltaChannel): + ch.replay_writes(history["writes"]) + else: + # One update per write: the writes span many supersteps, and a single + # call would reject a second Overwrite and drop the writes after one. + for _, _, value in history["writes"]: + ch.update([value]) + return ch + + def _require_saver_for_history( delta_channels: list[str], saver: BaseCheckpointSaver | None, @@ -465,7 +482,7 @@ delta_channels = _delta_channels_to_replay(channel_specs, checkpoint) _require_saver_for_history(delta_channels, saver, config) - histories: Mapping[str, Any] = {} + histories: Mapping[str, DeltaChannelHistory] = {} if delta_channels and saver is not None and config is not None: histories = saver.get_delta_channel_history( config=config, channels=delta_channels @@ -475,11 +492,7 @@ for k, spec in channel_specs.items(): ch: BaseChannel if k in histories: - delta_spec = cast(DeltaChannel, spec) - history = histories[k] - replay_ch = delta_spec.from_checkpoint(history.get("seed", MISSING)) - replay_ch.replay_writes(history["writes"]) - ch = replay_ch + ch = _replay(spec, histories[k]) else: ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING)) channels[k] = ch @@ -504,7 +517,7 @@ delta_channels = _delta_channels_to_replay(channel_specs, checkpoint) _require_saver_for_history(delta_channels, saver, config) - histories: Mapping[str, Any] = {} + histories: Mapping[str, DeltaChannelHistory] = {} if delta_channels and saver is not None and config is not None: histories = await saver.aget_delta_channel_history( config=config, channels=delta_channels @@ -514,11 +527,7 @@ for k, spec in channel_specs.items(): ch: BaseChannel if k in histories: - delta_spec = cast(DeltaChannel, spec) - history = histories[k] - replay_ch = delta_spec.from_checkpoint(history.get("seed", MISSING)) - replay_ch.replay_writes(history["writes"]) - ch = replay_ch + ch = _replay(spec, histories[k]) else: ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING)) channels[k] = ch diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/langgraph/pregel/_loop.py new/langgraph-1.2.16/langgraph/pregel/_loop.py --- old/langgraph-1.2.15/langgraph/pregel/_loop.py 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/langgraph/pregel/_loop.py 2020-02-02 01:00:00.000000000 +0100 @@ -257,8 +257,10 @@ _resume_loaded_writes: Sequence[PendingWrite] = () # Tasks `_reapply_writes_to_succeeded_nodes` handed loaded writes back to. _reapplied_task_ids: set[str] - # An update or fork checkpoint addressed by `checkpoint_id` that the thread - # has moved past, so it may already have children. + # Tasks that wrote during this run, the only ones exit mode saves at exit. + _written_task_ids: set[str] + # A checkpoint addressed by `checkpoint_id` that the thread has moved past, + # so it may already have children. _addressed_checkpoint_superseded: bool = False # The checkpoint_config that points at the parent loaded at `__enter__` @@ -349,7 +351,6 @@ self.interrupt_before = interrupt_before self.manager = manager self.is_nested = CONFIG_KEY_TASK_ID in self.config.get(CONF, {}) - self.is_replaying = CONFIG_KEY_CHECKPOINT_ID in config[CONF] self._migrate_checkpoint = migrate_checkpoint self.trigger_to_nodes = trigger_to_nodes self.retry_policy = retry_policy @@ -407,6 +408,14 @@ else () ) self.prev_checkpoint_config = None + # Whether the run points this loop at a checkpoint: a truthy + # `checkpoint_id`, this graph's `checkpoint_map` entry, or a replaying + # parent. Every task config carries `checkpoint_id: None`, so a key + # check would count every subgraph. + self._addressed = bool( + self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID) + or self.config[CONF].get(CONFIG_KEY_REPLAY_STATE) + ) runtime = self.config[CONF].get(CONFIG_KEY_RUNTIME) self.control = runtime.control if isinstance(runtime, Runtime) else None @@ -455,6 +464,7 @@ """Put writes for a task, to be read by the next tick.""" if not writes: return + self._written_task_ids.add(task_id) # deduplicate writes to special channels, last write wins if all(w[0] in WRITES_IDX_MAP for w in writes): writes = list({w[0]: w for w in writes}.values()) @@ -468,6 +478,11 @@ for w in self.checkpoint_pending_writes if w[0] != task_id or w[1] not in WRITES_IDX_MAP ] + # Savers key a regular write by its position in the list, and the + # special ones dropped above won't be there next time, so they go + # last: a resume saved ahead of an update would move it to a new + # slot once the list is saved again, and it would apply twice. + writes = sorted(writes, key=lambda w: w[0] in WRITES_IDX_MAP) writes_to_save: WritesT = [ w[1:] for w in self.checkpoint_pending_writes if w[0] == task_id ] + list(writes) @@ -556,10 +571,12 @@ CONFIG_KEY_CHECKPOINT_ID: self.checkpoint["id"], }, ) - # group by task id + # Writes loaded with the checkpoint are stored on it already. Saving them + # again can file them under new indexes, and then they apply twice. by_task = defaultdict(list) for task_id, channel, value in self.checkpoint_pending_writes: - by_task[task_id].append((channel, value)) + if task_id in self._written_task_ids: + by_task[task_id].append((channel, value)) # submit writes to checkpointer for task_id, writes in by_task.items(): if self.checkpointer_put_writes_accepts_task_path and hasattr( @@ -904,6 +921,7 @@ if tid != NULL_TASK_ID and isinstance(self.specs.get(ch), DeltaChannel) } self._exit_command_writes = [] + self._written_task_ids = set() # Resuming from a previous checkpoint requires two things: # 1. A prior checkpoint exists (channel_versions is non-empty) # 2. The input signals continuation (not a fresh run with new input) @@ -931,7 +949,7 @@ # writes so that interrupt() calls re-fire instead of returning # stale values. But if we're actively resuming, keep them — # multi-interrupt scenarios need previously resolved values preserved. - is_time_traveling = self.is_replaying and ( + is_time_traveling = self._addressed and ( # Time-travel to a subgraph checkpoint: the parent sets # RESUMING=True (it can't distinguish time-travel from resume), # so we check if this subgraph's own ns is in checkpoint_map. @@ -950,6 +968,10 @@ or configurable.get(CONFIG_KEY_RESUMING, False) ) ) + # Finished tasks rerun only on time travel, or on a resume at a + # checkpoint the thread has moved past. A resume addressed at the + # thread's latest checkpoint hands their writes back, as without an id. + self.is_replaying = is_time_traveling or self._addressed_checkpoint_superseded if is_time_traveling: self.checkpoint_pending_writes = [ w for w in self.checkpoint_pending_writes if w[1] != RESUME @@ -1165,11 +1187,11 @@ # checkpoint ID since the fork was created after the subgraph's # checkpoints from the original execution. # - # Only gate on is_time_traveling (not is_replaying). When the + # Only gate on is_time_traveling (not _addressed). When the # client resumes with an explicit checkpoint_id that happens to # point at the current head (e.g. LangGraph Studio sending # `checkpoint: {checkpoint_id}` alongside Command(resume=...)), - # is_replaying is True but is_time_traveling is False. In that + # the run is addressed but not time traveling. In that # case subgraphs should load their latest checkpoint normally, # not go through ReplayState's before-bound lookup which would # miss subgraph checkpoints created during processing of the @@ -1789,7 +1811,7 @@ # This covers both normal replay and subgraphs resolved via # checkpoint_map during time-travel. saved = self.checkpointer.get_tuple(self.checkpoint_config) - if saved is not None and saved.metadata.get("source") in ("update", "fork"): + if saved is not None: self._addressed_checkpoint_superseded = checkpoint_superseded( self.checkpointer, self.checkpoint_config, saved ) @@ -2045,7 +2067,7 @@ # This covers both normal replay and subgraphs resolved via # checkpoint_map during time-travel. saved = await self.checkpointer.aget_tuple(self.checkpoint_config) - if saved is not None and saved.metadata.get("source") in ("update", "fork"): + if saved is not None: self._addressed_checkpoint_superseded = await acheckpoint_superseded( self.checkpointer, self.checkpoint_config, saved ) diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/langgraph/pregel/_runner.py new/langgraph-1.2.16/langgraph/pregel/_runner.py --- old/langgraph-1.2.15/langgraph/pregel/_runner.py 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/langgraph/pregel/_runner.py 2020-02-02 01:00:00.000000000 +0100 @@ -41,7 +41,7 @@ from langgraph._internal._scratchpad import PregelScratchpad from langgraph._internal._typing import MISSING from langgraph.constants import TAG_HIDDEN -from langgraph.errors import GraphBubbleUp, GraphInterrupt +from langgraph.errors import GraphBubbleUp, GraphInterrupt, is_invalid_resume from langgraph.pregel._algo import Call from langgraph.pregel._executor import Submit from langgraph.pregel._retry import arun_with_retry, run_with_retry @@ -621,7 +621,11 @@ handled_exception_ids: set[int] | None = None, ) -> bool: """Check if any task failed, if so, cancel all other tasks. - GraphInterrupts are not considered failures.""" + GraphInterrupts are not considered failures. + + Neither is an invalid resume value: it's raised once the other tasks finish, so a + task answered in the same resume isn't cut off after its side effects and then run + again with its saved answer.""" for fut in done: if fut.cancelled(): continue @@ -629,6 +633,7 @@ if ( id(exc) not in (handled_exception_ids or set()) and not isinstance(exc, GraphBubbleUp) + and not is_invalid_resume(exc) and fut not in SKIP_RERAISE_SET ): return True diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/pyproject.toml new/langgraph-1.2.16/pyproject.toml --- old/langgraph-1.2.15/pyproject.toml 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/pyproject.toml 2020-02-02 01:00:00.000000000 +0100 @@ -4,7 +4,7 @@ [project] name = "langgraph" -version = "1.2.15" +version = "1.2.16" description = "Building stateful, multi-actor applications with LLMs" authors = [] requires-python = ">=3.10" diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/tests/test_delta_channel_exit_mode.py new/langgraph-1.2.16/tests/test_delta_channel_exit_mode.py --- old/langgraph-1.2.15/tests/test_delta_channel_exit_mode.py 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/tests/test_delta_channel_exit_mode.py 2020-02-02 01:00:00.000000000 +0100 @@ -521,10 +521,11 @@ graph = builder.compile(checkpointer=sync_checkpointer) config = {"configurable": {"thread_id": "t"}} graph.invoke(_both("in"), config, durability=durability) - head = graph.get_state(config).config + paused = graph.get_state(config).config + graph.update_state(config, None) live = graph.invoke( - Command(resume="yes", update={"flag": True}), head, durability=durability + Command(resume="yes", update={"flag": True}), paused, durability=durability ) state = graph.get_state(config) diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/tests/test_delta_channel_to_plain.py new/langgraph-1.2.16/tests/test_delta_channel_to_plain.py --- old/langgraph-1.2.15/tests/test_delta_channel_to_plain.py 1970-01-01 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/tests/test_delta_channel_to_plain.py 2020-02-02 01:00:00.000000000 +0100 @@ -0,0 +1,148 @@ +import operator +from typing import Annotated, Any +from unittest.mock import patch + +import pytest +from langchain_core.messages import AIMessage, HumanMessage +from langchain_core.runnables import RunnableConfig +from langgraph.checkpoint.base import BaseCheckpointSaver +from langgraph.checkpoint.memory import InMemorySaver +from langgraph.checkpoint.serde.types import _DeltaSnapshot +from typing_extensions import TypedDict + +from langgraph.channels.binop import BinaryOperatorAggregate +from langgraph.channels.delta import DeltaChannel +from langgraph.graph import END, START, StateGraph +from langgraph.graph.message import _messages_delta_reducer, add_messages +from langgraph.pregel._checkpoint import channels_from_checkpoint, empty_checkpoint +from langgraph.types import Overwrite + +pytestmark = pytest.mark.anyio + + +class DeltaState(TypedDict): + messages: Annotated[ + list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=3) + ] + + +class PlainState(TypedDict): + messages: Annotated[list, add_messages] + + +def _app(schema: type, checkpointer: BaseCheckpointSaver) -> Any: + builder = StateGraph(schema) + builder.add_node( + "chat", + lambda state: {"messages": [AIMessage(f"reply {len(state['messages'])}")]}, + ) + builder.add_edge(START, "chat") + builder.add_edge("chat", END) + return builder.compile(checkpointer=checkpointer) + + +def _delta_thread(checkpointer: BaseCheckpointSaver) -> RunnableConfig: + config: RunnableConfig = {"configurable": {"thread_id": "1"}} + delta = _app(DeltaState, checkpointer) + for i in range(4): + delta.invoke({"messages": [HumanMessage(f"q{i}")]}, config) + assert any( + isinstance(t.checkpoint["channel_values"].get("messages"), _DeltaSnapshot) + for t in checkpointer.list(config) + ), "the thread should have a snapshot to read through" + return config + + +def test_plain_reducer_reads_every_delta_written_checkpoint( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + config = _delta_thread(sync_checkpointer) + delta = _app(DeltaState, sync_checkpointer) + plain = _app(PlainState, sync_checkpointer) + + assert [s.values for s in plain.get_state_history(config)] == [ + s.values for s in delta.get_state_history(config) + ] + + +def test_plain_reducer_continues_delta_written_thread( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + config = _delta_thread(sync_checkpointer) + delta = _app(DeltaState, sync_checkpointer) + before = delta.get_state(config).values["messages"] + + after = _app(PlainState, sync_checkpointer).invoke( + {"messages": [HumanMessage("q4")]}, config + )["messages"] + + assert after[:-2] == before + assert [m.content for m in after[-2:]] == ["q4", f"reply {len(before) + 1}"] + head = sync_checkpointer.get_tuple(config) + assert head is not None + assert head.checkpoint["channel_values"]["messages"] == after + assert delta.get_state(config).values["messages"] == after + + +async def test_plain_reducer_reads_and_continues_delta_written_thread_async( + async_checkpointer: BaseCheckpointSaver, +) -> None: + config: RunnableConfig = {"configurable": {"thread_id": "1"}} + delta = _app(DeltaState, async_checkpointer) + for i in range(4): + await delta.ainvoke({"messages": [HumanMessage(f"q{i}")]}, config) + plain = _app(PlainState, async_checkpointer) + + expected = [s.values async for s in delta.aget_state_history(config)] + assert [s.values async for s in plain.aget_state_history(config)] == expected + + after = (await plain.ainvoke({"messages": [HumanMessage("q4")]}, config))[ + "messages" + ] + assert after[:-2] == expected[0]["messages"] + + [email protected]( + ("seed", "writes", "expected"), + [ + (_DeltaSnapshot(["a"]), [["b"], ["c"]], ["a", "b", "c"]), + (["a"], [["b"], Overwrite(["c"]), ["d"]], ["c", "d"]), + (None, [["b"], Overwrite(["c"]), Overwrite(["d"]), ["e"]], ["d", "e"]), + ], +) +def test_plain_reducer_replays_delta_history_one_write_at_a_time( + seed: Any, writes: list[Any], expected: list[str] +) -> None: + history: dict[str, Any] = {"writes": [("t", "items", w) for w in writes]} + if seed is not None: + history["seed"] = seed + checkpoint = empty_checkpoint() + checkpoint["channel_versions"]["items"] = 1 + saver = InMemorySaver() + + with patch.object( + saver, "get_delta_channel_history", return_value={"items": history} + ): + channels, _ = channels_from_checkpoint( + {"items": BinaryOperatorAggregate(list, operator.add)}, + checkpoint, + saver=saver, + config={"configurable": {"thread_id": "1"}}, + ) + + assert channels["items"].get() == expected + + +def test_plain_thread_reads_never_walk_history() -> None: + saver = InMemorySaver() + config: RunnableConfig = {"configurable": {"thread_id": "1"}} + plain = _app(PlainState, saver) + + with patch.object(saver, "get_delta_channel_history") as walk: + plain.invoke({"messages": [HumanMessage("q0")]}, config) + plain.update_state(config, {"messages": [HumanMessage("note")]}) + plain.invoke({"messages": [HumanMessage("q1")]}, config) + plain.get_state(config) + list(plain.get_state_history(config)) + + walk.assert_not_called() diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/tests/test_interrupt_state.py new/langgraph-1.2.16/tests/test_interrupt_state.py --- old/langgraph-1.2.15/tests/test_interrupt_state.py 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/tests/test_interrupt_state.py 2020-02-02 01:00:00.000000000 +0100 @@ -152,6 +152,83 @@ assert asked == [["A1", "B"]] +def test_finished_task_output_applies_once( + sync_checkpointer: BaseCheckpointSaver, durability: Durability +) -> None: + builder = StateGraph(State) + builder.add_node("a", lambda s: {"log": [f"a:{interrupt('A')}"]}) + builder.add_node("b", lambda s: {"log": [f"b:{interrupt('B')}"]}) + builder.add_node("c", lambda s: {"log": [f"c:{interrupt('C')}"]}) + builder.add_edge(START, "a") + builder.add_edge(START, "b") + builder.add_edge(START, "c") + graph = builder.compile(checkpointer=sync_checkpointer) + config = _config() + + result = graph.invoke({"log": []}, config, durability=durability) + while pending := result.get("__interrupt__"): + result = graph.invoke( + Command(resume={pending[0].id: "yes"}), config, durability=durability + ) + + assert sorted(result["log"]) == ["a:yes", "b:yes", "c:yes"] + + +def test_resume_with_update_applies_the_update_once( + sync_checkpointer: BaseCheckpointSaver, durability: Durability +) -> None: + def ask_three_times(state: State) -> State: + answers = [interrupt(str(question)) for question in range(1, 4)] + return {"log": ["a:" + "".join(answers)]} + + graph = ( + StateGraph(State) + .add_node("a", ask_three_times) + .add_edge(START, "a") + .compile(checkpointer=sync_checkpointer) + ) + config = _config() + + result = graph.invoke({"log": []}, config, durability=durability) + for i in range(1, 4): + result = graph.invoke( + Command(resume=str(i), update={"log": [f"u{i}"]}), + config, + durability=durability, + ) + + assert sorted(result["log"]) == ["a:123", "u1", "u2", "u3"] + assert sorted(graph.get_state(config).values["log"]) == sorted(result["log"]) + + [email protected]("checkpoint_id", ["none", "head"]) +def test_resume_at_the_latest_checkpoint_does_not_rerun_finished_task( + sync_checkpointer: BaseCheckpointSaver, checkpoint_id: str +) -> None: + calls: Counter[str] = Counter() + graph = _build_parallel(sync_checkpointer, calls) + config = _config() + graph.invoke({"log": []}, config) + + def resume(question: str, answer: str) -> Any: + snapshot = graph.get_state(config) + target = ( + snapshot.config + if checkpoint_id == "head" + else {"configurable": {**config["configurable"], "checkpoint_id": None}} + ) + return graph.invoke( + Command(resume={_interrupt_by_value(snapshot, question).id: answer}), + target, + ) + + resume("A1", "yes") + result = resume("B", "ok") + + assert sorted(result["log"]) == ["a:yes", "b:ok"] + assert calls == {"a": 2, "b": 3}, "resuming B ran the finished A again" + + @NEEDS_CONTEXTVARS async def test_finished_task_does_not_report_answered_interrupt_async( async_checkpointer: BaseCheckpointSaver, @@ -474,6 +551,64 @@ assert calls == {"a": 2, "b": 3} +def test_subgraph_finished_task_is_not_rerun( + sync_checkpointer: BaseCheckpointSaver, durability: Durability +) -> None: + calls: Counter[str] = Counter() + builder = StateGraph(State) + builder.add_node("child", _build_parallel(None, calls)) + builder.add_edge(START, "child") + graph = builder.compile(checkpointer=sync_checkpointer) + config = _config() + + graph.invoke({"log": []}, config, durability=durability) + snapshot = graph.get_state(config) + graph.invoke( + Command(resume={_interrupt_by_value(snapshot, "A1").id: "yes"}), + config, + durability=durability, + ) + snapshot = graph.get_state(config) + result = graph.invoke( + Command(resume={_interrupt_by_value(snapshot, "B").id: "ok"}), + config, + durability=durability, + ) + + assert sorted(result["log"]) == ["a:yes", "b:ok"] + assert calls == {"a": 2, "b": 3}, "resuming B ran the finished A again" + + +def test_nested_entrypoint_finished_task_is_not_rerun( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + calls: Counter[str] = Counter() + + @task + def step_1(query: str) -> str: + calls["step_1"] += 1 + return f"{query} bar" + + @task + def human_feedback(query: str) -> str: + calls["human_feedback"] += 1 + return f"{query} {interrupt('feedback?')}" + + @entrypoint() + def child(query: str) -> str: + return human_feedback(step_1(query).result()).result() + + @entrypoint(checkpointer=sync_checkpointer) + def parent(query: str) -> str: + return child.invoke(query) + + config = _config() + parent.invoke("foo", config) + + assert parent.invoke(Command(resume="baz"), config) == "foo bar baz" + assert calls == {"step_1": 1, "human_feedback": 2} + + # --- Reading task status from recorded writes --- diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/tests/test_interruption.py new/langgraph-1.2.16/tests/test_interruption.py --- old/langgraph-1.2.15/tests/test_interruption.py 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/tests/test_interruption.py 2020-02-02 01:00:00.000000000 +0100 @@ -1,5 +1,8 @@ +import asyncio +import operator +import sys from dataclasses import dataclass -from typing import Any +from typing import Annotated, Any import pytest from langgraph.checkpoint.base import BaseCheckpointSaver @@ -251,3 +254,61 @@ assert graph.invoke(resume({"approved": True}), config) == { "answer": ["ok", Decision(approved=True)] } + + [email protected]( + sys.version_info < (3, 11), + reason="Python 3.11+ is required for async contextvars support", +) +async def test_invalid_resume_lets_other_answered_tasks_finish( + async_checkpointer: BaseCheckpointSaver, +) -> None: + """An invalid answer fails the resume without cutting off the other answered tasks. + + A task cancelled after its side effect keeps its answer and runs again on the next + resume, so the side effect would happen twice. + """ + sent: list[str] = [] + + class State(TypedDict): + log: Annotated[list[str], operator.add] + + async def send(state: State) -> State: + interrupt("send?", response_schema=Decision) + sent.append("email") + await asyncio.sleep(0.05) # waiting on the reply to a request already sent + return {"log": ["sent"]} + + async def check(state: State) -> State: + interrupt("check?", response_schema=Decision) + return {"log": ["checked"]} + + graph = ( + StateGraph(State) + .add_node("send", send) + .add_node("check", check) + .add_edge(START, "send") + .add_edge(START, "check") + .compile(checkpointer=async_checkpointer) + ) + config = {"configurable": {"thread_id": "1"}} + await graph.ainvoke({"log": []}, config) + ids = {i.value: i.id for i in (await graph.aget_state(config)).interrupts} + + with pytest.raises(ValidationError, match="approved"): + await graph.ainvoke( + Command( + resume={ + ids["send?"]: {"approved": True}, + ids["check?"]: {"approved": "nope"}, + } + ), + config, + ) + pending = [i.value for i in (await graph.aget_state(config)).interrupts] + assert (pending, sent) == (["check?"], ["email"]) + + result = await graph.ainvoke( + Command(resume={ids["check?"]: {"approved": True}}), config + ) + assert (sorted(result["log"]), sent) == (["checked", "sent"], ["email"]) diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/tests/test_pregel.py new/langgraph-1.2.16/tests/test_pregel.py --- old/langgraph-1.2.15/tests/test_pregel.py 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/tests/test_pregel.py 2020-02-02 01:00:00.000000000 +0100 @@ -7737,6 +7737,8 @@ def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> None: # --- CHILD GRAPH --- + answered: Counter[str] = Counter() + class ChildState(BaseModel): prompt: str = Field(..., description="What is going to be asked to the user?") human_input: str | None = Field(None, description="What the human said") @@ -7746,6 +7748,7 @@ def get_human_input(state: ChildState): human_input = interrupt(state.prompt) + answered["human"] += 1 return dict( human_inputs=[human_input], # update parent state @@ -7753,6 +7756,7 @@ def get_dolphin_input(state: ChildState): human_input = interrupt(state.prompt) + answered["dolphin"] += 1 return dict( human_inputs=[human_input], # update parent state @@ -7846,6 +7850,7 @@ assert invokes == 5 assert len(events) == 5 + assert answered == {"human": 2, "dolphin": 2} def test_pregel_loop_refcount(): diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/tests/test_prune_postgres.py new/langgraph-1.2.16/tests/test_prune_postgres.py --- old/langgraph-1.2.15/tests/test_prune_postgres.py 1970-01-01 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/tests/test_prune_postgres.py 2020-02-02 01:00:00.000000000 +0100 @@ -0,0 +1,467 @@ +import operator +import sys +from collections.abc import Callable +from contextlib import contextmanager +from typing import Annotated, Any + +import pytest +from langchain_core.messages import AIMessage, AnyMessage, HumanMessage +from langchain_core.runnables import RunnableConfig +from langgraph.checkpoint.postgres import PostgresSaver +from langgraph.checkpoint.postgres.base import PRUNE_DELTA_THREADS_SQL, PRUNE_HEADS_SQL +from typing_extensions import TypedDict + +from langgraph.channels.delta import DeltaChannel +from langgraph.graph import START, StateGraph +from langgraph.graph.message import _messages_delta_reducer, add_messages +from langgraph.types import Command, StateSnapshot, interrupt +from tests.conftest import NO_DOCKER +from tests.conftest_checkpointer import ( + DEFAULT_POSTGRES_URI, + _checkpointer_postgres, + _checkpointer_postgres_aio, + _checkpointer_postgres_aio_pipe, + _checkpointer_postgres_aio_pool, + _checkpointer_postgres_pipe, + _checkpointer_postgres_pool, +) + +pytestmark = [ + pytest.mark.anyio, + pytest.mark.skipif(NO_DOCKER, reason="needs Postgres"), +] + +NEEDS_CONTEXTVARS = pytest.mark.skipif( + sys.version_info < (3, 11), + reason="Python 3.11+ is required for async contextvars support", +) + +SYNC = [ + _checkpointer_postgres, + _checkpointer_postgres_pipe, + _checkpointer_postgres_pool, +] +ASYNC = [ + _checkpointer_postgres_aio, + _checkpointer_postgres_aio_pipe, + _checkpointer_postgres_aio_pool, +] + + +class State(TypedDict): + messages: Annotated[list[AnyMessage], add_messages] + notes: Annotated[list[str], operator.add] + + +class DeltaState(TypedDict): + messages: Annotated[list[AnyMessage], DeltaChannel(_messages_delta_reducer)] + + +def _reply(ask: bool): + def reply(state: dict) -> dict: + n = len(state["messages"]) + answer = interrupt("approve?") if ask else "ok" + return {"messages": [AIMessage(f"{answer} {n}", id=f"reply-{n}")]} + + return reply + + +def _graph( + checkpointer: Any, + *, + ask: bool = False, + fail: set[str] | None = None, + calls: dict[str, int] | None = None, +) -> Any: + def note(name: str): + def node(state: State) -> dict: + if calls is not None: + calls[name] = calls.get(name, 0) + 1 + if fail is not None and name in fail: + fail.discard(name) + raise ValueError(f"{name} failed") + return {"notes": [name]} + + return node + + builder = StateGraph(State) + builder.add_node("left", note("left")) + builder.add_node("right", note("right")) + builder.add_node("reply", _reply(ask)) + builder.add_edge(START, "left") + builder.add_edge(START, "right") + builder.add_edge(["left", "right"], "reply") + return builder.compile(checkpointer=checkpointer) + + +def _counting_subgraph(name: str, *, checkpointer: bool | None) -> Any: + class Calls(TypedDict): + messages: Annotated[list[AnyMessage], add_messages] + calls: Annotated[list[str], operator.add] + + def count(state: Calls) -> dict: + n = len(state["messages"]) + reply = AIMessage(f"{name} call {len(state['calls']) + 1}", id=f"{name}-{n}") + return {"calls": [name], "messages": [reply]} + + builder = StateGraph(Calls) + builder.add_node("count", count) + builder.add_edge(START, "count") + return builder.compile(checkpointer=checkpointer) + + +def _delta_graph(checkpointer: Any) -> Any: + builder = StateGraph(DeltaState) + builder.add_node("reply", _reply(ask=False)) + builder.add_edge(START, "reply") + return builder.compile(checkpointer=checkpointer) + + +def _config(thread_id: str) -> RunnableConfig: + return {"configurable": {"thread_id": thread_id}} + + +def _say(i: int) -> dict: + return {"messages": [HumanMessage(f"hi {i}", id=f"hi-{i}")]} + + +def _checkpoint_ids(checkpointer: Any, thread_id: str) -> list[str]: + return [c.checkpoint["id"] for c in checkpointer.list(_config(thread_id))] + + +def _namespaces(checkpointer: Any, thread_id: str) -> list[str]: + return sorted( + c.config["configurable"]["checkpoint_ns"] + for c in checkpointer.list(_config(thread_id)) + ) + + +async def _acheckpoint_ids(checkpointer: Any, thread_id: str) -> list[str]: + return [c.checkpoint["id"] async for c in checkpointer.alist(_config(thread_id))] + + +def _parentless(snapshot: StateSnapshot) -> StateSnapshot: + return snapshot._replace( + parent_config=None, + tasks=tuple( + t._replace(state=_parentless(t.state)) + if isinstance(t.state, StateSnapshot) + else t + for t in snapshot.tasks + ), + ) + + +def _run_once_after( + saver: PostgresSaver, sql: str, action: Callable[[], Any], *, nth: int = 1 +) -> None: + seen = 0 + cursor = saver._cursor + + class Hooked: + def __init__(self, cur: Any) -> None: + self.cur = cur + + def __getattr__(self, name: str) -> Any: + return getattr(self.cur, name) + + def execute(self, query: Any, params: Any = None, **kwargs: Any) -> None: + nonlocal seen + self.cur.execute(query, params, **kwargs) + if query is sql: + seen += 1 + if seen == nth: + action() + + @contextmanager + def hooked(**kwargs: Any): + with cursor(**kwargs) as cur: + yield Hooked(cur) + + saver._cursor = hooked + + [email protected]("durability", ["sync", "async", "exit"]) [email protected]("make_checkpointer", SYNC) +def test_keep_latest_keeps_the_state_and_the_thread_runs_on_like_its_twin( + make_checkpointer: Any, durability: str +) -> None: + with make_checkpointer() as checkpointer: + graph = _graph(checkpointer) + for i in range(3): + for thread_id in ("pruned", "twin"): + graph.invoke(_say(i), _config(thread_id), durability=durability) + before = graph.get_state(_config("pruned")) + + checkpointer.prune(["pruned"]) + + assert _checkpoint_ids(checkpointer, "pruned") == [ + before.config["configurable"]["checkpoint_id"] + ] + assert graph.get_state(_config("pruned")) == _parentless(before) + for thread_id in ("pruned", "twin"): + graph.invoke(_say(3), _config(thread_id), durability=durability) + assert ( + graph.get_state(_config("pruned")).values + == graph.get_state(_config("twin")).values + ) + + [email protected]("make_checkpointer", SYNC) +def test_keep_latest_keeps_a_pending_interrupt_resumable( + make_checkpointer: Any, +) -> None: + with make_checkpointer() as checkpointer: + graph = _graph(checkpointer, ask=True) + for thread_id in ("pruned", "twin"): + for i in range(2): + graph.invoke(_say(i), _config(thread_id)) + graph.invoke(Command(resume=f"yes {i}"), _config(thread_id)) + graph.invoke(_say(2), _config(thread_id)) + before = graph.get_state(_config("pruned")) + assert before.interrupts + + checkpointer.prune(["pruned"]) + + assert graph.get_state(_config("pruned")) == _parentless(before) + for thread_id in ("pruned", "twin"): + graph.invoke(Command(resume="yes 2"), _config(thread_id)) + assert ( + graph.get_state(_config("pruned")).values + == graph.get_state(_config("twin")).values + ) + + [email protected]("make_checkpointer", SYNC) +def test_keep_latest_keeps_writes_of_tasks_that_finished_before_a_failure( + make_checkpointer: Any, +) -> None: + with make_checkpointer() as checkpointer: + calls: dict[str, int] = {} + fail: set[str] = set() + graph = _graph(checkpointer, fail=fail, calls=calls) + for thread_id in ("pruned", "twin"): + for i in range(2): + graph.invoke(_say(i), _config(thread_id)) + fail.add("right") + with pytest.raises(ValueError, match="right failed"): + graph.invoke(_say(2), _config(thread_id)) + before = graph.get_state(_config("pruned")) + + checkpointer.prune(["pruned"]) + + assert graph.get_state(_config("pruned")) == _parentless(before) + left_runs = calls["left"] + graph.invoke(None, _config("pruned")) + assert calls["left"] == left_runs, "a task that had finished ran again" + graph.invoke(None, _config("twin")) + assert ( + graph.get_state(_config("pruned")).values + == graph.get_state(_config("twin")).values + ) + + [email protected]("make_checkpointer", SYNC) +def test_keep_latest_keeps_an_interrupted_subgraph_resumable( + make_checkpointer: Any, +) -> None: + with make_checkpointer() as checkpointer: + sub = StateGraph(State) + sub.add_node("reply", _reply(ask=True)) + sub.add_edge(START, "reply") + parent = StateGraph(State) + parent.add_node("sub", sub.compile()) + parent.add_edge(START, "sub") + graph = parent.compile(checkpointer=checkpointer) + for thread_id in ("pruned", "twin"): + for i in range(2): + graph.invoke(_say(i), _config(thread_id)) + graph.invoke(Command(resume=f"yes {i}"), _config(thread_id)) + graph.invoke(_say(2), _config(thread_id)) + before = graph.get_state(_config("pruned"), subgraphs=True) + interrupted = before.tasks[0].state.config["configurable"]["checkpoint_ns"] + + checkpointer.prune(["pruned"]) + + assert _namespaces(checkpointer, "pruned") == ["", interrupted] + assert graph.get_state(_config("pruned"), subgraphs=True) == _parentless(before) + for thread_id in ("pruned", "twin"): + graph.invoke(Command(resume="yes 2"), _config(thread_id)) + assert ( + graph.get_state(_config("pruned")).values + == graph.get_state(_config("twin")).values + ) + + [email protected]("durability", ["sync", "async", "exit"]) [email protected]("make_checkpointer", SYNC) +def test_keep_latest_keeps_a_checkpointer_true_subgraph_but_not_finished_calls( + make_checkpointer: Any, durability: str +) -> None: + with make_checkpointer() as checkpointer: + parent = StateGraph(State) + parent.add_node("memory", _counting_subgraph("memory", checkpointer=True)) + parent.add_node("call", _counting_subgraph("call", checkpointer=None)) + parent.add_edge(START, "memory") + parent.add_edge("memory", "call") + graph = parent.compile(checkpointer=checkpointer) + for i in range(3): + for thread_id in ("pruned", "twin"): + graph.invoke(_say(i), _config(thread_id), durability=durability) + + checkpointer.prune(["pruned"]) + + assert _namespaces(checkpointer, "pruned") == ["", "memory"] + for thread_id in ("pruned", "twin"): + graph.invoke(_say(3), _config(thread_id), durability=durability) + assert ( + graph.get_state(_config("pruned")).values + == graph.get_state(_config("twin")).values + ) + + +def test_keep_latest_keeps_a_turn_saved_after_it_read_the_heads() -> None: + with ( + _checkpointer_postgres() as checkpointer, + PostgresSaver.from_conn_string( + DEFAULT_POSTGRES_URI + checkpointer.conn.info.dbname + ) as pruner, + ): + graph = _graph(checkpointer) + for i in range(3): + for thread_id in ("pruned", "twin"): + graph.invoke(_say(i), _config(thread_id)) + _run_once_after( + pruner, PRUNE_HEADS_SQL, lambda: graph.invoke(_say(3), _config("pruned")) + ) + + pruner.prune(["pruned"]) + + graph.invoke(_say(3), _config("twin")) + assert ( + graph.get_state(_config("pruned")).values + == graph.get_state(_config("twin")).values + ) + + +def test_keep_latest_refuses_a_thread_that_switches_to_a_delta_channel_mid_batch() -> ( + None +): + with ( + _checkpointer_postgres() as checkpointer, + PostgresSaver.from_conn_string( + DEFAULT_POSTGRES_URI + checkpointer.conn.info.dbname + ) as pruner, + ): + graph, delta_graph = _graph(checkpointer), _delta_graph(checkpointer) + for i in range(3): + for thread_id in ("switched", "twin"): + graph.invoke(_say(i), _config(thread_id)) + _run_once_after( + pruner, + PRUNE_DELTA_THREADS_SQL, + lambda: delta_graph.invoke(_say(3), _config("switched")), + ) + + with pytest.raises(NotImplementedError, match="switched"): + pruner.prune(["switched"]) + + delta_graph.invoke(_say(3), _config("twin")) + assert ( + delta_graph.get_state(_config("switched")).values + == delta_graph.get_state(_config("twin")).values + ) + + +def test_keep_latest_keeps_a_delta_channel_turn_saved_after_the_thread_check() -> None: + with ( + _checkpointer_postgres() as checkpointer, + PostgresSaver.from_conn_string( + DEFAULT_POSTGRES_URI + checkpointer.conn.info.dbname + ) as pruner, + ): + graph, delta_graph = _graph(checkpointer), _delta_graph(checkpointer) + for i in range(3): + for thread_id in ("switched", "twin"): + graph.invoke(_say(i), _config(thread_id)) + head = _checkpoint_ids(checkpointer, "switched")[0] + _run_once_after( + pruner, + PRUNE_DELTA_THREADS_SQL, + lambda: delta_graph.invoke(_say(3), _config("switched")), + nth=2, + ) + + pruner.prune(["switched"]) + + assert _checkpoint_ids(checkpointer, "switched")[-1] == head + delta_graph.invoke(_say(3), _config("twin")) + assert ( + delta_graph.get_state(_config("switched")).values + == delta_graph.get_state(_config("twin")).values + ) + + [email protected]("make_checkpointer", SYNC) +def test_keep_latest_refuses_a_batch_with_a_delta_channel_thread( + make_checkpointer: Any, +) -> None: + with make_checkpointer() as checkpointer: + graph, delta_graph = _graph(checkpointer), _delta_graph(checkpointer) + for i in range(3): + graph.invoke(_say(i), _config("ordinary")) + delta_graph.invoke(_say(i), _config("delta")) + before = {t: _checkpoint_ids(checkpointer, t) for t in ("ordinary", "delta")} + delta_before = delta_graph.get_state(_config("delta")) + + with pytest.raises(NotImplementedError, match="delta"): + checkpointer.prune(["ordinary", "delta"]) + + assert {t: _checkpoint_ids(checkpointer, t) for t in before} == before + assert delta_graph.get_state(_config("delta")) == delta_before + + +@NEEDS_CONTEXTVARS [email protected]("make_checkpointer", ASYNC) +async def test_akeep_latest_keeps_a_pending_interrupt_resumable( + make_checkpointer: Any, +) -> None: + async with make_checkpointer() as checkpointer: + graph = _graph(checkpointer, ask=True) + for thread_id in ("pruned", "twin"): + for i in range(2): + await graph.ainvoke(_say(i), _config(thread_id)) + await graph.ainvoke(Command(resume=f"yes {i}"), _config(thread_id)) + await graph.ainvoke(_say(2), _config(thread_id)) + before = await graph.aget_state(_config("pruned")) + + await checkpointer.aprune(["pruned"]) + + assert await _acheckpoint_ids(checkpointer, "pruned") == [ + before.config["configurable"]["checkpoint_id"] + ] + assert await graph.aget_state(_config("pruned")) == _parentless(before) + for thread_id in ("pruned", "twin"): + await graph.ainvoke(Command(resume="yes 2"), _config(thread_id)) + assert (await graph.aget_state(_config("pruned"))).values == ( + await graph.aget_state(_config("twin")) + ).values + + [email protected]("make_checkpointer", ASYNC) +async def test_akeep_latest_refuses_a_batch_with_a_delta_channel_thread( + make_checkpointer: Any, +) -> None: + async with make_checkpointer() as checkpointer: + graph, delta_graph = _graph(checkpointer), _delta_graph(checkpointer) + for i in range(3): + await graph.ainvoke(_say(i), _config("ordinary")) + await delta_graph.ainvoke(_say(i), _config("delta")) + before = { + t: await _acheckpoint_ids(checkpointer, t) for t in ("ordinary", "delta") + } + + with pytest.raises(NotImplementedError, match="delta"): + await checkpointer.aprune(["ordinary", "delta"]) + + assert {t: await _acheckpoint_ids(checkpointer, t) for t in before} == before diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/tests/test_time_travel.py new/langgraph-1.2.16/tests/test_time_travel.py --- old/langgraph-1.2.15/tests/test_time_travel.py 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/tests/test_time_travel.py 2020-02-02 01:00:00.000000000 +0100 @@ -1177,6 +1177,43 @@ assert result["value"] == ["sub_a", "human:answer", "sub_b"] +def test_subgraph_replay_reasks_answered_parallel_interrupt( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + """Replaying a subgraph checkpoint asks its step's questions again, including + one already answered. An update checkpoint replays in place instead of + forking, so its tasks keep their ids and its saved answers could be reused.""" + + subgraph = ( + StateGraph(State) + .add_node("a", lambda s: {"value": [f"a:{interrupt('a')}"]}) + .add_node("b", lambda s: {"value": [f"b:{interrupt('b')}"]}) + .add_edge(START, "a") + .add_edge(START, "b") + .compile(checkpointer=True) + ) + graph = ( + StateGraph(State) + .add_node("subgraph_node", subgraph) + .add_edge(START, "subgraph_node") + .compile(checkpointer=sync_checkpointer) + ) + config = {"configurable": {"thread_id": "1"}} + + def subgraph_head() -> RunnableConfig: + return graph.get_state(config, subgraphs=True).tasks[0].state.config + + graph.invoke({"value": []}, config) + graph.update_state(subgraph_head(), {"value": ["u"]}, as_node="__start__") + result = graph.invoke(None, config) + question_a = next(i for i in result["__interrupt__"] if i.value == "a") + graph.invoke(Command(resume={question_a.id: "yes"}), config) + + result = graph.invoke(None, subgraph_head()) + + assert sorted(i.value for i in result["__interrupt__"]) == ["a", "b"] + + def test_subgraph_replay_loads_accumulated_state_then_resume( sync_checkpointer: BaseCheckpointSaver, ) -> None: diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn' '--exclude=.svnignore' old/langgraph-1.2.15/uv.lock new/langgraph-1.2.16/uv.lock --- old/langgraph-1.2.15/uv.lock 2020-02-02 01:00:00.000000000 +0100 +++ new/langgraph-1.2.16/uv.lock 2020-02-02 01:00:00.000000000 +0100 @@ -1451,7 +1451,7 @@ [[package]] name = "langgraph" -version = "1.2.15" +version = "1.2.16" source = { editable = "." } dependencies = [ { name = "langchain-core" }, @@ -1685,7 +1685,7 @@ [[package]] name = "langgraph-checkpoint-postgres" -version = "3.1.2" +version = "3.2.0" source = { editable = "../checkpoint-postgres" } dependencies = [ { name = "langgraph-checkpoint" }, @@ -1734,7 +1734,7 @@ [[package]] name = "langgraph-checkpoint-sqlite" -version = "3.1.1" +version = "3.1.2" source = { editable = "../checkpoint-sqlite" } dependencies = [ { name = "aiosqlite" }, @@ -1835,7 +1835,7 @@ [[package]] name = "langgraph-prebuilt" -version = "1.1.0" +version = "1.1.1" source = { editable = "../prebuilt" } dependencies = [ { name = "langchain-core" },
