feat: support list of hooks for pre_model_hook and post_model_hook

Allow `pre_model_hook` and `post_model_hook` in `create_react_agent`
to accept a list of `RunnableLike` callables in addition to a single
hook. When a list is provided the hooks are composed in order: each
hook receives the graph state merged with all prior hooks' updates,
and the final merged update is returned to the graph.

This mirrors the middleware-stack pattern familiar from HTTP frameworks
(Express, FastAPI, Starlette) and allows hook logic to be written as
small, reusable units that can be composed without manually threading
state between them.

libs/checkpoint-conformance/README.md

@@ -64,8 +64,6 @@ The suite tests **base** capabilities (required) and **extended** capabilities (
 | `copy_thread` | no | `acopy_thread` |
 | `prune` | no | `aprune` |
 | `delta_channel_history` | no | `aget_delta_channel_history` |
-| `delta_channel_keepset` | no | `aget_delta_channel_keepset` |
-| `delta_channel_reconstruction` | no | `aput` |
 
 Extended capabilities are detected by checking whether the method is overridden from `BaseCheckpointSaver`. If not overridden, those tests are skipped.
 

libs/checkpoint-conformance/langgraph/checkpoint/conformance/capabilities.py

@@ -24,8 +24,6 @@ class Capability(str, Enum):
     COPY_THREAD = "copy_thread"
     PRUNE = "prune"
     DELTA_CHANNEL_HISTORY = "delta_channel_history"
-    DELTA_CHANNEL_KEEPSET = "delta_channel_keepset"
-    DELTA_CHANNEL_RECONSTRUCTION = "delta_channel_reconstruction"
 
 
 # Capabilities that every checkpointer must support.
@@ -46,8 +44,6 @@ EXTENDED_CAPABILITIES = frozenset(
         Capability.COPY_THREAD,
         Capability.PRUNE,
         Capability.DELTA_CHANNEL_HISTORY,
-        Capability.DELTA_CHANNEL_KEEPSET,
-        Capability.DELTA_CHANNEL_RECONSTRUCTION,
     }
 )
 
@@ -64,8 +60,6 @@ _CAPABILITY_METHOD_MAP: dict[Capability, str] = {
     Capability.COPY_THREAD: "acopy_thread",
     Capability.PRUNE: "aprune",
     Capability.DELTA_CHANNEL_HISTORY: "aget_delta_channel_history",
-    Capability.DELTA_CHANNEL_KEEPSET: "aget_tuple",
-    Capability.DELTA_CHANNEL_RECONSTRUCTION: "aput",
 }
 
 

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/__init__.py

@@ -12,12 +12,6 @@ from langgraph.checkpoint.conformance.spec.test_delete_thread import (
 from langgraph.checkpoint.conformance.spec.test_delta_channel_history import (
     run_delta_channel_history_tests,
 )
-from langgraph.checkpoint.conformance.spec.test_delta_channel_keepset import (
-    run_delta_channel_keepset_tests,
-)
-from langgraph.checkpoint.conformance.spec.test_delta_channel_reconstruction import (
-    run_delta_channel_reconstruction_tests,
-)
 from langgraph.checkpoint.conformance.spec.test_get_tuple import run_get_tuple_tests
 from langgraph.checkpoint.conformance.spec.test_list import run_list_tests
 from langgraph.checkpoint.conformance.spec.test_prune import run_prune_tests
@@ -34,6 +28,4 @@ __all__ = [
     "run_copy_thread_tests",
     "run_prune_tests",
     "run_delta_channel_history_tests",
-    "run_delta_channel_keepset_tests",
-    "run_delta_channel_reconstruction_tests",
 ]

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_keepset.py

@@ -1,172 +0,0 @@
-"""DELTA_CHANNEL_KEEPSET capability tests — aget_delta_channel_keepset contract."""
-
-from __future__ import annotations
-
-import traceback
-from collections.abc import Callable
-from uuid import uuid4
-
-from langgraph.checkpoint.base import BaseCheckpointSaver
-
-from langgraph.checkpoint.conformance.spec._delta_fixtures import build_delta_chain
-
-
-async def test_keepset_empty_channels_returns_target_only(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Empty channels → keep-set is just {target_id}."""
-    tid = str(uuid4())
-    configs = await build_delta_chain(
-        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=4
-    )
-    head = configs[-1]
-    keep = await saver.aget_delta_channel_keepset(config=head, channels=[])
-    head_id = head["configurable"]["checkpoint_id"]
-    assert keep == {head_id}, f"Expected only target, got {keep}"
-
-
-async def test_keepset_snapshot_at_target(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """When target itself has a snapshot, keep-set is just {target_id}."""
-    tid = str(uuid4())
-    configs = await build_delta_chain(
-        saver,
-        thread_id=tid,
-        channel="ch",
-        snapshots_at_steps=[0, 3],
-        total_steps=4,
-    )
-    head = configs[3]
-    keep = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
-    head_id = head["configurable"]["checkpoint_id"]
-    assert keep == {head_id}, f"Snapshot at target should yield only target, got {keep}"
-
-
-async def test_keepset_snapshot_n_back(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Snapshot N steps back → target + intermediates + snapshot ancestor."""
-    tid = str(uuid4())
-    configs = await build_delta_chain(
-        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=5
-    )
-    head = configs[-1]
-    keep = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
-    expected_ids = {c["configurable"]["checkpoint_id"] for c in configs}
-    assert keep == expected_ids, f"Expected all ancestors, got {keep}"
-
-
-async def test_keepset_multi_channel_union(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Multi-channel keep-set is the union (max chain per channel)."""
-    tid = str(uuid4())
-    from langgraph.checkpoint.base import Checkpoint
-    from langgraph.checkpoint.base.id import uuid6
-    from langgraph.checkpoint.serde.types import _DeltaSnapshot
-
-    from langgraph.checkpoint.conformance.test_utils import generate_metadata
-
-    configs: list = []
-    parent_cfg = None
-    for step in range(6):
-        config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
-        if parent_cfg:
-            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
-                "checkpoint_id"
-            ]
-        cv: dict = {}
-        cvs: dict = {}
-        # Channel "a" has snapshot at step 3 (recent)
-        if step == 3:
-            cv["a"] = _DeltaSnapshot("snap_a")
-            cvs["a"] = step + 1
-        # Channel "b" has snapshot at step 1 (further back)
-        if step == 1:
-            cv["b"] = _DeltaSnapshot("snap_b")
-            cvs["b"] = step + 1
-        cp = Checkpoint(
-            v=1,
-            id=str(uuid6(clock_seq=-1)),
-            ts="",
-            channel_values=cv,
-            channel_versions=cvs,
-            versions_seen={},
-            updated_channels=None,
-        )
-        parent_cfg = await saver.aput(config, cp, generate_metadata(step=step), cvs)
-        configs.append(parent_cfg)
-
-    head = configs[-1]
-    keep = await saver.aget_delta_channel_keepset(config=head, channels=["a", "b"])
-    # Union: b needs back to step 1, so steps 1..5 (all except step 0) plus head
-    expected_ids = {c["configurable"]["checkpoint_id"] for c in configs[1:]}
-    expected_ids.add(head["configurable"]["checkpoint_id"])
-    assert keep == expected_ids, f"Expected union, got {keep} vs {expected_ids}"
-
-
-async def test_keepset_walk_to_root(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """No snapshot anywhere → entire chain to root is in keep-set."""
-    tid = str(uuid4())
-    configs = await build_delta_chain(
-        saver, thread_id=tid, channel="ch", snapshots_at_steps=[], total_steps=4
-    )
-    head = configs[-1]
-    keep = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
-    all_ids = {c["configurable"]["checkpoint_id"] for c in configs}
-    assert keep == all_ids, f"Expected full chain, got {keep}"
-
-
-async def test_keepset_deterministic(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Same inputs return identical sets."""
-    tid = str(uuid4())
-    configs = await build_delta_chain(
-        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=5
-    )
-    head = configs[-1]
-    keep1 = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
-    keep2 = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
-    assert keep1 == keep2
-
-
-ALL_DELTA_CHANNEL_KEEPSET_TESTS = [
-    test_keepset_empty_channels_returns_target_only,
-    test_keepset_snapshot_at_target,
-    test_keepset_snapshot_n_back,
-    test_keepset_multi_channel_union,
-    test_keepset_walk_to_root,
-    test_keepset_deterministic,
-]
-
-
-async def run_delta_channel_keepset_tests(
-    saver: BaseCheckpointSaver,
-    on_test_result: Callable[[str, str, bool, str | None], None] | None = None,
-) -> tuple[int, int, list[str]]:
-    """Run all delta_channel_keepset tests. Returns (passed, failed, failure_names)."""
-    passed = 0
-    failed = 0
-    failures: list[str] = []
-    for test_fn in ALL_DELTA_CHANNEL_KEEPSET_TESTS:
-        try:
-            await test_fn(saver)
-            passed += 1
-            if on_test_result:
-                on_test_result("delta_channel_keepset", test_fn.__name__, True, None)
-        except Exception:
-            failed += 1
-            msg = f"{test_fn.__name__}: {traceback.format_exc()}"
-            failures.append(msg)
-            if on_test_result:
-                on_test_result(
-                    "delta_channel_keepset",
-                    test_fn.__name__,
-                    False,
-                    traceback.format_exc(),
-                )
-    return passed, failed, failures

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_reconstruction.py

@@ -1,164 +0,0 @@
-"""DELTA_CHANNEL_RECONSTRUCTION capability tests — end-to-end round-trip.
-
-Exercises: aput + aput_writes + aget_delta_channel_history + reconstruction.
-This catches the most common silent-corruption mode: failing to round-trip
-`_DeltaSnapshot` blobs through serialization.
-
-NOTE: This test does NOT import from `langgraph` (which is not a dependency
-of checkpoint-conformance). Instead it inlines a minimal reconstruction
-equivalent: seed + fold writes through a simple list-append reducer.
-"""
-
-from __future__ import annotations
-
-import traceback
-from collections.abc import Callable
-from typing import Any
-from uuid import uuid4
-
-from langgraph.checkpoint.base import BaseCheckpointSaver
-
-from langgraph.checkpoint.conformance.spec._delta_fixtures import build_delta_chain
-
-
-def _reconstruct(seed: Any, writes: list) -> list:
-    """Minimal DeltaChannel reconstruction: list-append reducer.
-
-    Mirrors DeltaChannel.from_checkpoint(seed) + replay_writes(writes).
-    """
-    from langgraph.checkpoint.serde.types import _DeltaSnapshot
-
-    if seed is None:
-        base: list = []
-    elif isinstance(seed, _DeltaSnapshot):
-        base = list(seed.value)
-    else:
-        base = list(seed)
-    for _task_id, _ch, value in writes:
-        base = base + value
-    return base
-
-
-async def test_reconstruction_basic(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Reconstruct delta channel value from history matches expected."""
-    tid = str(uuid4())
-    # 5 steps: snapshot at 0 (value=[0]), writes at 1,2,3,4.
-    # Head = step 4. Walk from parent (step 3) collects writes 1,2,3.
-    configs = await build_delta_chain(
-        saver,
-        thread_id=tid,
-        channel="msgs",
-        snapshots_at_steps=[0],
-        total_steps=5,
-        write_value_fn=lambda step: [step],
-    )
-    head = configs[-1]
-    result = await saver.aget_delta_channel_history(config=head, channels=["msgs"])
-    history = result["msgs"]
-
-    seed = history.get("seed")
-    reconstructed = _reconstruct(seed, history["writes"])
-
-    # seed=[0] from step 0, writes from steps 1,2,3
-    expected = [0] + [1] + [2] + [3]
-    assert reconstructed == expected, (
-        f"Reconstructed {reconstructed} != expected {expected}"
-    )
-
-
-async def test_reconstruction_mid_chain_snapshot(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Reconstruction works when snapshot is mid-chain."""
-    tid = str(uuid4())
-    # 6 steps: snapshots at 0 and 3, writes at 1,2,4,5.
-    # Head = step 5. Walk from step 4 stops at step 3 (snapshot).
-    # Collects writes from step 4.
-    configs = await build_delta_chain(
-        saver,
-        thread_id=tid,
-        channel="msgs",
-        snapshots_at_steps=[0, 3],
-        total_steps=6,
-        write_value_fn=lambda step: [step],
-    )
-    head = configs[-1]
-    result = await saver.aget_delta_channel_history(config=head, channels=["msgs"])
-    history = result["msgs"]
-
-    seed = history.get("seed")
-    reconstructed = _reconstruct(seed, history["writes"])
-
-    # Snapshot at step 3 = [3], write from step 4
-    expected = [3] + [4]
-    assert reconstructed == expected, (
-        f"Reconstructed {reconstructed} != expected {expected}"
-    )
-
-
-async def test_reconstruction_no_snapshot(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Reconstruction from root (no snapshot) gives all writes accumulated."""
-    tid = str(uuid4())
-    # 4 steps: no snapshot, writes at 0,1,2,3.
-    # Head = step 3. Walk from step 2 collects writes 0,1,2.
-    configs = await build_delta_chain(
-        saver,
-        thread_id=tid,
-        channel="msgs",
-        snapshots_at_steps=[],
-        total_steps=4,
-        write_value_fn=lambda step: [step],
-    )
-    head = configs[-1]
-    result = await saver.aget_delta_channel_history(config=head, channels=["msgs"])
-    history = result["msgs"]
-
-    seed = history.get("seed")
-    reconstructed = _reconstruct(seed, history["writes"])
-
-    # No seed → start empty, writes from steps 0,1,2
-    expected = [0] + [1] + [2]
-    assert reconstructed == expected, (
-        f"Reconstructed {reconstructed} != expected {expected}"
-    )
-
-
-ALL_DELTA_CHANNEL_RECONSTRUCTION_TESTS = [
-    test_reconstruction_basic,
-    test_reconstruction_mid_chain_snapshot,
-    test_reconstruction_no_snapshot,
-]
-
-
-async def run_delta_channel_reconstruction_tests(
-    saver: BaseCheckpointSaver,
-    on_test_result: Callable[[str, str, bool, str | None], None] | None = None,
-) -> tuple[int, int, list[str]]:
-    """Run all reconstruction tests. Returns (passed, failed, failure_names)."""
-    passed = 0
-    failed = 0
-    failures: list[str] = []
-    for test_fn in ALL_DELTA_CHANNEL_RECONSTRUCTION_TESTS:
-        try:
-            await test_fn(saver)
-            passed += 1
-            if on_test_result:
-                on_test_result(
-                    "delta_channel_reconstruction", test_fn.__name__, True, None
-                )
-        except Exception:
-            failed += 1
-            msg = f"{test_fn.__name__}: {traceback.format_exc()}"
-            failures.append(msg)
-            if on_test_result:
-                on_test_result(
-                    "delta_channel_reconstruction",
-                    test_fn.__name__,
-                    False,
-                    traceback.format_exc(),
-                )
-    return passed, failed, failures

libs/checkpoint-conformance/langgraph/checkpoint/conformance/validate.py

@@ -22,12 +22,6 @@ from langgraph.checkpoint.conformance.spec.test_delete_thread import (
 from langgraph.checkpoint.conformance.spec.test_delta_channel_history import (
     run_delta_channel_history_tests,
 )
-from langgraph.checkpoint.conformance.spec.test_delta_channel_keepset import (
-    run_delta_channel_keepset_tests,
-)
-from langgraph.checkpoint.conformance.spec.test_delta_channel_reconstruction import (
-    run_delta_channel_reconstruction_tests,
-)
 from langgraph.checkpoint.conformance.spec.test_get_tuple import run_get_tuple_tests
 from langgraph.checkpoint.conformance.spec.test_list import run_list_tests
 from langgraph.checkpoint.conformance.spec.test_prune import run_prune_tests
@@ -45,8 +39,6 @@ _RUNNERS = {
     Capability.COPY_THREAD: run_copy_thread_tests,
     Capability.PRUNE: run_prune_tests,
     Capability.DELTA_CHANNEL_HISTORY: run_delta_channel_history_tests,
-    Capability.DELTA_CHANNEL_KEEPSET: run_delta_channel_keepset_tests,
-    Capability.DELTA_CHANNEL_RECONSTRUCTION: run_delta_channel_reconstruction_tests,
 }
 
 

libs/checkpoint-conformance/pyproject.toml

@@ -50,7 +50,6 @@ unresolved-attribute = "ignore"
 unresolved-import = "ignore"
 invalid-argument-type = "ignore"
 invalid-return-type = "ignore"
-missing-typed-dict-key = "ignore"
 
 [tool.ruff]
 lint.select = [

libs/checkpoint-sqlite/tests/test_conformance_delta.py

@@ -27,8 +27,6 @@ async def test_delta_channel_conformance():
         sqlite_saver,
         capabilities={
             "delta_channel_history",
-            "delta_channel_keepset",
-            "delta_channel_reconstruction",
         },
     )
     for cap, result in report.results.items():

libs/checkpoint/langgraph/checkpoint/base/__init__.py

@@ -680,123 +680,6 @@ class BaseCheckpointSaver(Generic[V]):
             result[ch] = entry
         return result
 
-    def get_delta_channel_keepset(
-        self,
-        *,
-        config: RunnableConfig,
-        channels: Sequence[str],
-    ) -> set[str]:
-        """Return ancestor checkpoint_ids that must survive deletion.
-
-        !!! warning "Beta"
-
-            This method is part of the `DeltaChannel` support surface and is
-            in beta. The signature may change while the delta-channel design
-            stabilizes.
-
-        Walks the parent chain from `config` backward, collecting visited
-        checkpoint_ids (inclusive of the target), and terminates per-channel
-        when that channel has a populated `channel_values[ch]` (a
-        `_DeltaSnapshot` blob or a pre-migration plain value). The returned
-        set is the minimum keep-set: every checkpoint_id whose removal would
-        break reconstruction of the listed channels at `config`.
-
-        Pass `channels=[]` to return just `{config.checkpoint_id}` — useful
-        for graphs that don't use `DeltaChannel`.
-
-        Compose this into custom `prune` / `delete_for_runs` / `copy_thread`::
-
-            keep = saver.get_delta_channel_keepset(
-                config=head_config, channels=delta_channels,
-            )
-            delete_rows_not_in(keep)
-
-        Note:
-            The default implementation here uses repeated `get_tuple` calls
-            to walk the parent chain. This is a basic reference implementation
-            suitable for low-frequency maintenance operations (prune, etc.).
-            Custom checkpointer backends may override with a more efficient
-            version tailored to their data model (e.g. a single SQL query
-            with a recursive CTE), but it is not required — the default works
-            correctly for any saver that implements `get_tuple`.
-
-        Args:
-            config: Configuration identifying the target checkpoint.
-            channels: Channel names whose delta history must be preserved.
-                Empty sequence means only the target checkpoint_id is kept.
-
-        Returns:
-            Set of checkpoint_ids that must not be deleted.
-        """
-        target_tuple = self.get_tuple(config)
-        if target_tuple is None:
-            return set()
-        target_id = target_tuple.config["configurable"]["checkpoint_id"]
-        keep: set[str] = {target_id}
-        if not channels:
-            return keep
-        remaining: set[str] = set(channels)
-        for ch in list(remaining):
-            if ch in target_tuple.checkpoint["channel_values"]:
-                remaining.discard(ch)
-        if not remaining:
-            return keep
-        cursor_config: RunnableConfig | None = target_tuple.parent_config
-        while cursor_config is not None and remaining:
-            tup = self.get_tuple(cursor_config)
-            if tup is None:
-                break
-            cid = tup.config["configurable"]["checkpoint_id"]
-            keep.add(cid)
-            for ch in list(remaining):
-                if ch in tup.checkpoint["channel_values"]:
-                    remaining.discard(ch)
-            if not remaining:
-                break
-            cursor_config = tup.parent_config
-        return keep
-
-    async def aget_delta_channel_keepset(
-        self,
-        *,
-        config: RunnableConfig,
-        channels: Sequence[str],
-    ) -> set[str]:
-        """Async version of `get_delta_channel_keepset`.
-
-        !!! warning "Beta"
-
-            This method is part of the `DeltaChannel` support surface and is
-            in beta. See `get_delta_channel_keepset` for full documentation.
-        """
-        target_tuple = await self.aget_tuple(config)
-        if target_tuple is None:
-            return set()
-        target_id = target_tuple.config["configurable"]["checkpoint_id"]
-        keep: set[str] = {target_id}
-        if not channels:
-            return keep
-        remaining: set[str] = set(channels)
-        for ch in list(remaining):
-            if ch in target_tuple.checkpoint["channel_values"]:
-                remaining.discard(ch)
-        if not remaining:
-            return keep
-        cursor_config: RunnableConfig | None = target_tuple.parent_config
-        while cursor_config is not None and remaining:
-            tup = await self.aget_tuple(cursor_config)
-            if tup is None:
-                break
-            cid = tup.config["configurable"]["checkpoint_id"]
-            keep.add(cid)
-            for ch in list(remaining):
-                if ch in tup.checkpoint["channel_values"]:
-                    remaining.discard(ch)
-            if not remaining:
-                break
-            cursor_config = tup.parent_config
-        return keep
-
     def get_next_version(self, current: V | None, channel: None) -> V:
         """Generate the next version ID for a channel.
 

libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py

@@ -44,7 +44,7 @@ if TYPE_CHECKING:
         AllowedMsgpackModules,
     )
 
-LC_REVIVER = Reviver()
+LC_REVIVER = Reviver(allowed_objects="core")
 EMPTY_BYTES = b""
 logger = logging.getLogger(__name__)
 

libs/checkpoint/tests/test_conformance_delta.py

@@ -25,8 +25,6 @@ async def test_delta_channel_conformance():
         mem_saver,
         capabilities={
             "delta_channel_history",
-            "delta_channel_keepset",
-            "delta_channel_reconstruction",
         },
     )
     for cap, result in report.results.items():

libs/langgraph/langgraph/pregel/_loop.py

@@ -776,17 +776,6 @@ class PregelLoop:
                 and configurable.get(CONFIG_KEY_CHECKPOINT_NS, "")
                 in configurable.get(CONFIG_KEY_CHECKPOINT_MAP, {})
             )
-            # Outer graph: time-travel-resume — explicit non-head
-            # checkpoint paired with a Command(resume=...). Without this,
-            # `Command(resume=...) + checkpoint=<non-head>` would match the
-            # plain-resume exclusion below and skip the cleanup needed to
-            # treat this as a fork.
-            or (
-                not self.is_nested
-                and getattr(self, "_loaded_explicit_non_head", False)
-                and input_is_command
-                and cast(Command, self.input).resume is not None
-            )
             or not (
                 # Outer graph: resume arrives as Command(resume=...)
                 (input_is_command and cast(Command, self.input).resume is not None)
@@ -1511,13 +1500,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
 
     def __enter__(self) -> Self:
         self._graph_lifecycle_events = deque()
-        # Set by the explicit-id branch when the loaded checkpoint is not
-        # the latest for the (thread, ns). Lets _first distinguish a
-        # plain resume (`Command(resume=...)` against head, which carries
-        # an explicit checkpoint_id only because clients like LangGraph
-        # Studio echo it back) from time-travel-resume (`Command(resume=
-        # ...)` against an explicit non-head checkpoint).
-        self._loaded_explicit_non_head = False
         if not self.checkpointer:
             saved = None
         elif self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
@@ -1525,14 +1507,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             # 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 not self.is_nested:
-                head = self.checkpointer.get_tuple(
-                    patch_configurable(
-                        self.checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
-                    )
-                )
-                if head is not None and head.checkpoint["id"] != saved.checkpoint["id"]:
-                    self._loaded_explicit_non_head = True
         elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
             # Subgraph replay: the parent is replaying and passed us a
             # replay_state with its checkpoint_id. Look up our checkpoint
@@ -1784,7 +1758,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
 
     async def __aenter__(self) -> Self:
         self._graph_lifecycle_events = deque()
-        self._loaded_explicit_non_head = False
         if not self.checkpointer:
             saved = None
         elif self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
@@ -1792,14 +1765,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             # 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 not self.is_nested:
-                head = await self.checkpointer.aget_tuple(
-                    patch_configurable(
-                        self.checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
-                    )
-                )
-                if head is not None and head.checkpoint["id"] != saved.checkpoint["id"]:
-                    self._loaded_explicit_non_head = True
         elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
             # Subgraph replay: the parent is replaying and passed us a
             # replay_state with its checkpoint_id. Look up our checkpoint

libs/langgraph/tests/test_time_travel.py

@@ -3964,96 +3964,3 @@ def test_subgraph_called_in_loop_loads_state_on_replay(
 
     assert len(observed) == 1
     assert observed[0] == ("sub_step", {"sub_trail": ["s", "s", "s"]})
-
-
-def test_subgraph_time_travel_resume_with_command_in_one_call(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Repro for time-travel-resume issued as a SINGLE call:
-    `invoke(Command(resume=...), config={..., checkpoint_id: <non-head>})`.
-
-    Equivalent to the existing two-call pattern
-    (`test_subgraph_time_travel_resume_from_first_interrupt`) which does
-    `invoke(None, sub_config)` then `invoke(Command(resume=...))`. Clients
-    like LangGraph API server / Studio can submit this as one request.
-    """
-    called: list[str] = []
-
-    def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    executor = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("executor", executor)
-        .add_edge(START, "executor")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run to completion through both interrupts.
-    graph.invoke({"value": []}, config)
-    parent_state_at_first = graph.get_state(config, subgraphs=True)
-    sub_config_at_first = parent_state_at_first.tasks[0].state.config
-    graph.invoke(Command(resume="answer_1"), config)
-    graph.invoke(Command(resume="answer_2"), config)
-
-    final_after_original = graph.get_state(config).values
-    assert final_after_original["value"] == [
-        "step_a_done",
-        "ask_1:answer_1",
-        "ask_2:answer_2",
-    ]
-
-    # ONE-CALL time-travel-resume: pair Command(resume=...) with a
-    # non-head parent checkpoint_id in the same invocation.
-    called.clear()
-    result = graph.invoke(Command(resume="new_answer_1"), sub_config_at_first)
-
-    # `step_a` ran before the time-travel target — must not re-execute.
-    assert "step_a" not in called, f"step_a re-executed; called={called}"
-    # The new answer must be applied (not the cached `answer_1`).
-    assert result.get("__interrupt__"), (
-        f"expected to pause at the next interrupt; got result={result}"
-    )
-    assert result["__interrupt__"][0].value == "Question 2?"
-    sub_state_now = graph.get_state(config, subgraphs=True).tasks[0].state
-    assert sub_state_now.values["value"] == ["step_a_done", "ask_1:new_answer_1"], (
-        f"subgraph state at int2 should reflect new_answer_1; "
-        f"got {sub_state_now.values}"
-    )
-
-    # Resume the second interrupt with another new answer; ensure the
-    # whole branch concludes consistently.
-    called.clear()
-    final = graph.invoke(Command(resume="new_answer_2"), config)
-    assert "step_a" not in called
-    assert "ask_1" not in called
-    assert final["value"] == [
-        "step_a_done",
-        "ask_1:new_answer_1",
-        "ask_2:new_answer_2",
-    ]

libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py

@@ -6,6 +6,7 @@ from typing import (
     Any,
     Literal,
     TypeVar,
+    Union,
     cast,
     get_type_hints,
 )
@@ -125,6 +126,9 @@ Prompt = (
     | Runnable[StateSchema, LanguageModelInput]
 )
 
+# A single hook or a list of hooks to be composed in order.
+HookLike = Union[RunnableLike, Sequence[RunnableLike]]
+
 
 def _get_state_value(state: StateSchema, key: str, default: Any = None) -> Any:
     return (
@@ -134,6 +138,131 @@ def _get_state_value(state: StateSchema, key: str, default: Any = None) -> Any:
     )
 
 
+def _set_state_value(state: StateSchema, key: str, value: Any) -> None:
+    """Set a value in the state, supporting both dict and Pydantic model states."""
+    if isinstance(state, dict):
+        state[key] = value
+    else:
+        setattr(state, key, value)
+
+
+def _merge_state_update(state: StateSchema, update: dict) -> StateSchema:
+    """Return a shallow copy of *state* with *update* applied.
+
+    This is used when chaining multiple hooks: each hook receives the state
+    as it would look after all previous hooks have run, so that hooks later
+    in the chain can observe updates made by earlier ones.
+
+    Note: only simple key-level merging is performed here (no reducer logic).
+    The full reducer logic is applied by the graph engine when the final
+    combined update dict is written back to the state.
+    """
+    if isinstance(state, dict):
+        return {**state, **update}  # type: ignore[return-value]
+    else:
+        # Pydantic / dataclass – make a shallow copy and patch fields
+        try:
+            merged = state.model_copy()  # pydantic v2
+        except AttributeError:
+            merged = state.copy()  # pydantic v1 / dataclass fallback
+        for k, v in update.items():
+            setattr(merged, k, v)
+        return merged  # type: ignore[return-value]
+
+
+def _coerce_to_runnable(hook: RunnableLike) -> RunnableCallable:
+    """Wrap a plain callable into a RunnableCallable if necessary."""
+    if isinstance(hook, RunnableCallable):
+        return hook
+    if isinstance(hook, Runnable):
+        # Already a Runnable – wrap so we get a uniform interface
+        sync_fn = hook.invoke
+        async_fn = hook.ainvoke
+        return RunnableCallable(sync_fn, async_fn)
+    if inspect.iscoroutinefunction(hook):
+        return RunnableCallable(None, hook)
+    if callable(hook):
+        return RunnableCallable(hook)
+    raise TypeError(f"Expected a callable or Runnable, got {type(hook)!r}")
+
+
+def _chain_hooks(hooks: Sequence[RunnableLike]) -> RunnableCallable:
+    """Compose multiple hook callables into a single hook.
+
+    Each hook is called in order.  After each hook the returned update dict is
+    merged into a running copy of the graph state so that subsequent hooks can
+    observe the changes made by earlier ones.  The accumulated update dict
+    (union of all individual update dicts, with later hooks winning on key
+    conflicts) is returned as the final state update.
+
+    Args:
+        hooks: A sequence of :data:`RunnableLike` objects.  Each must accept
+            the graph state as its first positional argument and return a
+            ``dict`` of state updates.
+
+    Returns:
+        A :class:`~langgraph._internal._runnable.RunnableCallable` that behaves
+        like a single hook but applies all of *hooks* in sequence.
+    """
+    if not hooks:
+        raise ValueError("_chain_hooks requires at least one hook")
+    if len(hooks) == 1:
+        return _coerce_to_runnable(hooks[0])
+
+    runnables = [_coerce_to_runnable(h) for h in hooks]
+
+    def _sync_chained(state: Any, **kwargs: Any) -> dict:
+        accumulated: dict = {}
+        current_state = state
+        for runnable in runnables:
+            # Pass extra kwargs (e.g. config, store) through if the hook
+            # accepts them; RunnableCallable handles introspection.
+            update = runnable.invoke(current_state, **kwargs)
+            if update:
+                accumulated.update(update)
+                current_state = _merge_state_update(current_state, update)
+        return accumulated
+
+    async def _async_chained(state: Any, **kwargs: Any) -> dict:
+        accumulated: dict = {}
+        current_state = state
+        for runnable in runnables:
+            update = await runnable.ainvoke(current_state, **kwargs)
+            if update:
+                accumulated.update(update)
+                current_state = _merge_state_update(current_state, update)
+        return accumulated
+
+    return RunnableCallable(_sync_chained, _async_chained, name="chained_hooks")
+
+
+def _resolve_hook(hook: HookLike | None) -> RunnableLike | None:
+    """Normalise *hook* to a single ``RunnableLike`` (or ``None``).
+
+    * If *hook* is ``None`` → return ``None``.
+    * If *hook* is already a ``RunnableLike`` → return it unchanged.
+    * If *hook* is a non-empty :class:`~collections.abc.Sequence` of
+      ``RunnableLike`` → chain them with :func:`_chain_hooks`.
+    """
+    if hook is None:
+        return None
+    # A Sequence[RunnableLike] but NOT a single Runnable/callable
+    if (
+        isinstance(hook, Sequence)
+        and not isinstance(hook, str)
+        and not isinstance(hook, Runnable)
+        and not callable(hook)
+    ):
+        hooks_list: list[RunnableLike] = list(hook)
+        if not hooks_list:
+            return None
+        if len(hooks_list) == 1:
+            return hooks_list[0]
+        return _chain_hooks(hooks_list)
+    # Single hook – return as-is
+    return hook  # type: ignore[return-value]
+
+
 def _get_prompt_runnable(prompt: Prompt | None) -> Runnable:
     prompt_runnable: Runnable
     if prompt is None:
@@ -293,8 +422,8 @@ def create_react_agent(
     response_format: StructuredResponseSchema
     | tuple[str, StructuredResponseSchema]
     | None = None,
-    pre_model_hook: RunnableLike | None = None,
-    post_model_hook: RunnableLike | None = None,
+    pre_model_hook: HookLike | None = None,
+    post_model_hook: HookLike | None = None,
     state_schema: StateSchemaType | None = None,
     context_schema: type[Any] | None = None,
     checkpointer: Checkpointer | None = None,
@@ -393,10 +522,21 @@ def create_react_agent(
                 The graph will make a separate call to the LLM to generate the structured response after the agent loop is finished.
                 This is not the only strategy to get structured responses, see more options in [this guide](https://langchain-ai.github.io/langgraph/how-tos/react-agent-structured-output/).
 
-        pre_model_hook: An optional node to add before the `agent` node (i.e., the node that calls the LLM).
-            Useful for managing long message histories (e.g., message trimming, summarization, etc.).
-            Pre-model hook must be a callable or a runnable that takes in current graph state and returns a state update in the form of
-                ```python
+        pre_model_hook: An optional node (or list of nodes) to add before the
+            ``agent`` node (i.e., the node that calls the LLM). Useful for
+            managing long message histories (e.g., message trimming,
+            summarization, etc.) or for composing multiple pre-processing
+            steps.
+
+            A single hook **or a list of hooks** may be provided. When a list
+            is given the hooks are executed in order: each hook receives the
+            graph state as updated by all preceding hooks, and the union of
+            all their return dicts is applied to the graph state before the
+            agent node runs.
+
+            Each hook must be a callable or a runnable that takes the current
+            graph state and returns a state update::
+
                 # At least one of `messages` or `llm_input_messages` MUST be provided
                 {
                     # If provided, will UPDATE the `messages` in the state
@@ -407,27 +547,63 @@ def create_react_agent(
                     # Any other state keys that need to be propagated
                     ...
                 }
-                ```
 
             !!! Important
-                At least one of `messages` or `llm_input_messages` MUST be provided and will be used as an input to the `agent` node.
-                The rest of the keys will be added to the graph state.
+                At least one of `messages` or `llm_input_messages` MUST be
+                provided (by at least one hook in the chain) and will be used
+                as an input to the ``agent`` node.  The rest of the keys will
+                be added to the graph state.
 
             !!! Warning
-                If you are returning `messages` in the pre-model hook, you should OVERWRITE the `messages` key by doing the following:
+                If you are returning `messages` in the pre-model hook, you
+                should OVERWRITE the `messages` key::
+
+                    {
+                        "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *new_messages]
+                        ...
+                    }
+
+            !!! Example "Composing multiple pre-model hooks"
 
                 ```python
-                {
-                    "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *new_messages]
-                    ...
-                }
+                from langchain_core.messages import RemoveMessage
+                from langgraph.graph.message import REMOVE_ALL_MESSAGES
+
+                def trim_messages(state):
+                    # Keep only the last 10 messages
+                    return {
+                        "messages": [
+                            RemoveMessage(id=REMOVE_ALL_MESSAGES),
+                            *state["messages"][-10:],
+                        ]
+                    }
+
+                def inject_system_prompt(state):
+                    return {
+                        "llm_input_messages": [
+                            SystemMessage("You are a helpful assistant."),
+                            *state["messages"],
+                        ]
+                    }
+
+                agent = create_react_agent(
+                    model,
+                    tools,
+                    pre_model_hook=[trim_messages, inject_system_prompt],
+                )
                 ```
-        post_model_hook: An optional node to add after the `agent` node (i.e., the node that calls the LLM).
-            Useful for implementing human-in-the-loop, guardrails, validation, or other post-processing.
-            Post-model hook must be a callable or a runnable that takes in current graph state and returns a state update.
+
+        post_model_hook: An optional node (or list of nodes) to add after the
+            ``agent`` node (i.e., the node that calls the LLM). Useful for
+            implementing human-in-the-loop, guardrails, validation, or other
+            post-processing steps.
+
+            Accepts the same single-hook-or-list-of-hooks form as
+            ``pre_model_hook``.
 
             !!! Note
-                Only available with `version="v2"`.
+                Only available with ``version="v2"``.
+
         state_schema: An optional state schema that defines graph state.
             Must have `messages` and `remaining_steps` keys.
             Defaults to `AgentState` that defines those two keys.
@@ -551,6 +727,10 @@ def create_react_agent(
             else AgentState
         )
 
+    # Normalise hook arguments: a list of hooks is composed into a single hook.
+    resolved_pre_model_hook: RunnableLike | None = _resolve_hook(pre_model_hook)
+    resolved_post_model_hook: RunnableLike | None = _resolve_hook(post_model_hook)
+
     llm_builtin_tools: list[dict] = []
     if isinstance(tools, ToolNode):
         tool_classes = list(tools.tools_by_name.values())
@@ -634,7 +814,7 @@ def create_react_agent(
         return False
 
     def _get_model_input_state(state: StateSchema) -> StateSchema:
-        if pre_model_hook is not None:
+        if resolved_pre_model_hook is not None:
             messages = (
                 _get_state_value(state, "llm_input_messages")
             ) or _get_state_value(state, "messages")
@@ -721,7 +901,7 @@ def create_react_agent(
         return {"messages": [response]}
 
     input_schema: StateSchemaType
-    if pre_model_hook is not None:
+    if resolved_pre_model_hook is not None:
         # Dynamically create a schema that inherits from state_schema and adds 'llm_input_messages'
         if isinstance(state_schema, type) and issubclass(state_schema, BaseModel):
             # For Pydantic schemas
@@ -792,8 +972,8 @@ def create_react_agent(
             RunnableCallable(call_model, acall_model),
             input_schema=input_schema,
         )
-        if pre_model_hook is not None:
-            workflow.add_node("pre_model_hook", pre_model_hook)  # type: ignore[arg-type]
+        if resolved_pre_model_hook is not None:
+            workflow.add_node("pre_model_hook", resolved_pre_model_hook)  # type: ignore[arg-type]
             workflow.add_edge("pre_model_hook", "agent")
             entrypoint = "pre_model_hook"
         else:
@@ -801,8 +981,8 @@ def create_react_agent(
 
         workflow.set_entry_point(entrypoint)
 
-        if post_model_hook is not None:
-            workflow.add_node("post_model_hook", post_model_hook)  # type: ignore[arg-type]
+        if resolved_post_model_hook is not None:
+            workflow.add_node("post_model_hook", resolved_post_model_hook)  # type: ignore[arg-type]
             workflow.add_edge("agent", "post_model_hook")
 
         if response_format is not None:
@@ -813,7 +993,7 @@ def create_react_agent(
                     agenerate_structured_response,
                 ),
             )
-            if post_model_hook is not None:
+            if resolved_post_model_hook is not None:
                 workflow.add_edge("post_model_hook", "generate_structured_response")
             else:
                 workflow.add_edge("agent", "generate_structured_response")
@@ -833,7 +1013,7 @@ def create_react_agent(
         last_message = messages[-1]
         # If there is no function call, then we finish
         if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
-            if post_model_hook is not None:
+            if resolved_post_model_hook is not None:
                 return "post_model_hook"
             elif response_format is not None:
                 return "generate_structured_response"
@@ -844,7 +1024,7 @@ def create_react_agent(
             if version == "v1":
                 return "tools"
             elif version == "v2":
-                if post_model_hook is not None:
+                if resolved_post_model_hook is not None:
                     return "post_model_hook"
                 return [
                     Send(
@@ -873,8 +1053,8 @@ def create_react_agent(
 
     # Optionally add a pre-model hook node that will be called
     # every time before the "agent" (LLM-calling node)
-    if pre_model_hook is not None:
-        workflow.add_node("pre_model_hook", pre_model_hook)  # type: ignore[arg-type]
+    if resolved_pre_model_hook is not None:
+        workflow.add_node("pre_model_hook", resolved_pre_model_hook)  # type: ignore[arg-type]
         workflow.add_edge("pre_model_hook", "agent")
         entrypoint = "pre_model_hook"
     else:
@@ -888,8 +1068,8 @@ def create_react_agent(
     post_model_hook_paths = [entrypoint, "tools"]
 
     # Add a post model hook node if post_model_hook is provided
-    if post_model_hook is not None:
-        workflow.add_node("post_model_hook", post_model_hook)  # type: ignore[arg-type]
+    if resolved_post_model_hook is not None:
+        workflow.add_node("post_model_hook", resolved_post_model_hook)  # type: ignore[arg-type]
         agent_paths.append("post_model_hook")
         workflow.add_edge("agent", "post_model_hook")
     else:
@@ -904,17 +1084,17 @@ def create_react_agent(
                 agenerate_structured_response,
             ),
         )
-        if post_model_hook is not None:
+        if resolved_post_model_hook is not None:
             post_model_hook_paths.append("generate_structured_response")
         else:
             agent_paths.append("generate_structured_response")
     else:
-        if post_model_hook is not None:
+        if resolved_post_model_hook is not None:
             post_model_hook_paths.append(END)
         else:
             agent_paths.append(END)
 
-    if post_model_hook is not None:
+    if resolved_post_model_hook is not None:
 
         def post_model_hook_router(state: StateSchema) -> str | list[Send]:
             """Route to the next node after post_model_hook.
@@ -1012,4 +1192,5 @@ __all__ = [
     "AgentStatePydantic",
     "AgentStateWithStructuredResponse",
     "AgentStateWithStructuredResponsePydantic",
+    "HookLike",
 ]

libs/sdk-py/langgraph_sdk/_async/cron.py

@@ -18,6 +18,7 @@ from langgraph_sdk.schema import (
     CronSortBy,
     Durability,
     Input,
+    Json,
     OnCompletionBehavior,
     QueryParamTypes,
     Run,
@@ -413,6 +414,7 @@ class CronClient:
         assistant_id: str | None = None,
         thread_id: str | None = None,
         enabled: bool | None = None,
+        metadata: Json = None,
         limit: int = 10,
         offset: int = 0,
         sort_by: CronSortBy | None = None,
@@ -427,6 +429,8 @@ class CronClient:
             assistant_id: The assistant ID or graph name to search for.
             thread_id: the thread ID to search for.
             enabled: The enabled status to search for.
+            metadata: Metadata to filter by. Exact match filter for each KV pair.
+                !!! version-added "Added in Agent Server version 0.9.0"
             limit: The maximum number of results to return.
             offset: The number of results to skip.
             headers: Optional custom headers to include with the request.
@@ -481,6 +485,8 @@ class CronClient:
             "limit": limit,
             "offset": offset,
         }
+        if metadata:
+            payload["metadata"] = metadata
         if sort_by:
             payload["sort_by"] = sort_by
         if sort_order:
@@ -497,6 +503,7 @@ class CronClient:
         *,
         assistant_id: str | None = None,
         thread_id: str | None = None,
+        metadata: Json = None,
         headers: Mapping[str, str] | None = None,
         params: QueryParamTypes | None = None,
     ) -> int:
@@ -505,6 +512,8 @@ class CronClient:
         Args:
             assistant_id: Assistant ID to filter by.
             thread_id: Thread ID to filter by.
+            metadata: Metadata to filter by. Exact match filter for each KV pair.
+                !!! version-added "Added in Agent Server version 0.9.0"
             headers: Optional custom headers to include with the request.
             params: Optional query parameters to include with the request.
 
@@ -516,6 +525,8 @@ class CronClient:
             payload["assistant_id"] = assistant_id
         if thread_id:
             payload["thread_id"] = thread_id
+        if metadata:
+            payload["metadata"] = metadata
         return await self.http.post(
             "/runs/crons/count", json=payload, headers=headers, params=params
         )

libs/sdk-py/langgraph_sdk/_sync/cron.py

@@ -18,6 +18,7 @@ from langgraph_sdk.schema import (
     CronSortBy,
     Durability,
     Input,
+    Json,
     OnCompletionBehavior,
     QueryParamTypes,
     Run,
@@ -402,6 +403,7 @@ class SyncCronClient:
         assistant_id: str | None = None,
         thread_id: str | None = None,
         enabled: bool | None = None,
+        metadata: Json = None,
         limit: int = 10,
         offset: int = 0,
         sort_by: CronSortBy | None = None,
@@ -416,6 +418,8 @@ class SyncCronClient:
             assistant_id: The assistant ID or graph name to search for.
             thread_id: the thread ID to search for.
             enabled: Whether the cron job is enabled.
+            metadata: Metadata to filter by. Exact match filter for each KV pair.
+                !!! version-added "Added in Agent Server version 0.9.0"
             limit: The maximum number of results to return.
             offset: The number of results to skip.
             headers: Optional custom headers to include with the request.
@@ -468,6 +472,8 @@ class SyncCronClient:
             "limit": limit,
             "offset": offset,
         }
+        if metadata:
+            payload["metadata"] = metadata
         if sort_by:
             payload["sort_by"] = sort_by
         if sort_order:
@@ -484,6 +490,7 @@ class SyncCronClient:
         *,
         assistant_id: str | None = None,
         thread_id: str | None = None,
+        metadata: Json = None,
         headers: Mapping[str, str] | None = None,
         params: QueryParamTypes | None = None,
     ) -> int:
@@ -492,6 +499,8 @@ class SyncCronClient:
         Args:
             assistant_id: Assistant ID to filter by.
             thread_id: Thread ID to filter by.
+            metadata: Metadata to filter by. Exact match filter for each KV pair.
+                !!! version-added "Added in Agent Server version 0.9.0"
             headers: Optional custom headers to include with the request.
             params: Optional query parameters to include with the request.
 
@@ -503,6 +512,8 @@ class SyncCronClient:
             payload["assistant_id"] = assistant_id
         if thread_id:
             payload["thread_id"] = thread_id
+        if metadata:
+            payload["metadata"] = metadata
         return self.http.post(
             "/runs/crons/count", json=payload, headers=headers, params=params
         )

libs/sdk-py/tests/test_crons_client.py

@@ -485,3 +485,165 @@ def test_sync_update_with_enabled_parameter(enabled_value):
         )
 
     assert result == cron
+
+
+@pytest.mark.asyncio
+async def test_async_search_with_metadata():
+    """Test that CronClient.search forwards metadata in the request body."""
+    cron = _cron_response()
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/runs/crons/search"
+
+        body = json.loads(request.content)
+        assert body["metadata"] == {"owner": "alice"}
+        assert body["limit"] == 10
+        assert body["offset"] == 0
+
+        return httpx.Response(200, json=[cron])
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        http_client = HttpClient(client)
+        cron_client = CronClient(http_client)
+        result = await cron_client.search(metadata={"owner": "alice"})
+
+    assert result == [cron]
+
+
+@pytest.mark.asyncio
+async def test_async_search_omits_empty_metadata():
+    """Test that CronClient.search does not send metadata when not provided."""
+    cron = _cron_response()
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        assert "metadata" not in body
+        return httpx.Response(200, json=[cron])
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        http_client = HttpClient(client)
+        cron_client = CronClient(http_client)
+        await cron_client.search()
+
+
+@pytest.mark.asyncio
+async def test_async_count_with_metadata():
+    """Test that CronClient.count forwards metadata in the request body."""
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/runs/crons/count"
+
+        body = json.loads(request.content)
+        assert body["metadata"] == {"team": "infra"}
+
+        return httpx.Response(200, json=2)
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        http_client = HttpClient(client)
+        cron_client = CronClient(http_client)
+        result = await cron_client.count(metadata={"team": "infra"})
+
+    assert result == 2
+
+
+@pytest.mark.asyncio
+async def test_async_count_omits_empty_metadata():
+    """Test that CronClient.count does not send metadata when not provided."""
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        assert "metadata" not in body
+        return httpx.Response(200, json=0)
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://example.com"
+    ) as client:
+        http_client = HttpClient(client)
+        cron_client = CronClient(http_client)
+        await cron_client.count()
+
+
+def test_sync_search_with_metadata():
+    """Test that SyncCronClient.search forwards metadata in the request body."""
+    cron = _cron_response()
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/runs/crons/search"
+
+        body = json.loads(request.content)
+        assert body["metadata"] == {"owner": "alice"}
+
+        return httpx.Response(200, json=[cron])
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        http_client = SyncHttpClient(client)
+        cron_client = SyncCronClient(http_client)
+        result = cron_client.search(metadata={"owner": "alice"})
+
+    assert result == [cron]
+
+
+def test_sync_search_omits_empty_metadata():
+    """Test that SyncCronClient.search does not send metadata when not provided."""
+    cron = _cron_response()
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        assert "metadata" not in body
+        return httpx.Response(200, json=[cron])
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        http_client = SyncHttpClient(client)
+        cron_client = SyncCronClient(http_client)
+        cron_client.search()
+
+
+def test_sync_count_with_metadata():
+    """Test that SyncCronClient.count forwards metadata in the request body."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/runs/crons/count"
+
+        body = json.loads(request.content)
+        assert body["metadata"] == {"team": "infra"}
+
+        return httpx.Response(200, json=2)
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        http_client = SyncHttpClient(client)
+        cron_client = SyncCronClient(http_client)
+        result = cron_client.count(metadata={"team": "infra"})
+
+    assert result == 2
+
+
+def test_sync_count_omits_empty_metadata():
+    """Test that SyncCronClient.count does not send metadata when not provided."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        assert "metadata" not in body
+        return httpx.Response(200, json=0)
+
+    transport = httpx.MockTransport(handler)
+    with httpx.Client(transport=transport, base_url="https://example.com") as client:
+        http_client = SyncHttpClient(client)
+        cron_client = SyncCronClient(http_client)
+        cron_client.count()