Merge branch 'main' into nh/namespace-keyed-subagents

Comment follow up

libs/checkpoint-conformance/README.md

@@ -63,6 +63,7 @@ The suite tests **base** capabilities (required) and **extended** capabilities (
 | `delete_for_runs` | no | `adelete_for_runs` |
 | `copy_thread` | no | `acopy_thread` |
 | `prune` | no | `aprune` |
+| `delta_channel_history` | no | `aget_delta_channel_history` |
 
 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

@@ -23,6 +23,7 @@ class Capability(str, Enum):
     DELETE_FOR_RUNS = "delete_for_runs"
     COPY_THREAD = "copy_thread"
     PRUNE = "prune"
+    DELTA_CHANNEL_HISTORY = "delta_channel_history"
 
 
 # Capabilities that every checkpointer must support.
@@ -42,6 +43,7 @@ EXTENDED_CAPABILITIES = frozenset(
         Capability.DELETE_FOR_RUNS,
         Capability.COPY_THREAD,
         Capability.PRUNE,
+        Capability.DELTA_CHANNEL_HISTORY,
     }
 )
 
@@ -57,6 +59,7 @@ _CAPABILITY_METHOD_MAP: dict[Capability, str] = {
     Capability.DELETE_FOR_RUNS: "adelete_for_runs",
     Capability.COPY_THREAD: "acopy_thread",
     Capability.PRUNE: "aprune",
+    Capability.DELTA_CHANNEL_HISTORY: "aget_delta_channel_history",
 }
 
 

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

@@ -9,6 +9,9 @@ from langgraph.checkpoint.conformance.spec.test_delete_for_runs import (
 from langgraph.checkpoint.conformance.spec.test_delete_thread import (
     run_delete_thread_tests,
 )
+from langgraph.checkpoint.conformance.spec.test_delta_channel_history import (
+    run_delta_channel_history_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
@@ -24,4 +27,5 @@ __all__ = [
     "run_delete_for_runs_tests",
     "run_copy_thread_tests",
     "run_prune_tests",
+    "run_delta_channel_history_tests",
 ]

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

@@ -0,0 +1,99 @@
+"""Shared fixtures for delta-channel conformance tests.
+
+Builds a parent chain with `_DeltaSnapshot` blobs at known positions via
+direct `aput` / `aput_writes` calls. No langgraph or Pregel dependency.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+from uuid import uuid4
+
+from langchain_core.runnables import RunnableConfig
+from langgraph.checkpoint.base import BaseCheckpointSaver, Checkpoint
+from langgraph.checkpoint.base.id import uuid6
+
+from langgraph.checkpoint.conformance.test_utils import generate_metadata
+
+
+async def build_delta_chain(
+    saver: BaseCheckpointSaver,
+    *,
+    thread_id: str | None = None,
+    checkpoint_ns: str = "",
+    channel: str = "messages",
+    snapshots_at_steps: Sequence[int] = (0,),
+    total_steps: int = 6,
+    write_value_fn: Any | None = None,
+) -> list[RunnableConfig]:
+    """Build a parent chain with `_DeltaSnapshot` at known positions.
+
+    Args:
+        saver: Checkpointer instance.
+        thread_id: Defaults to a random UUID.
+        checkpoint_ns: Namespace (default root).
+        channel: Channel name used for snapshots and writes.
+        snapshots_at_steps: Steps at which a `_DeltaSnapshot` blob is stored
+            in `channel_values[channel]`. Step 0 is the oldest checkpoint.
+        total_steps: Number of checkpoints in the chain.
+        write_value_fn: Callable(step) -> write value. Defaults to step index.
+
+    Returns:
+        List of stored configs (oldest first), one per step.
+    """
+    if write_value_fn is None:
+
+        def write_value_fn(step: int) -> Any:
+            return step
+
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+
+    thread_id = thread_id or str(uuid4())
+    snapshot_set = set(snapshots_at_steps)
+    stored: list[RunnableConfig] = []
+    parent_cfg: RunnableConfig | None = None
+
+    for step in range(total_steps):
+        config: RunnableConfig = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": checkpoint_ns,
+            }
+        }
+        if parent_cfg:
+            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
+                "checkpoint_id"
+            ]
+
+        channel_values: dict[str, Any] = {}
+        channel_versions: dict[str, int] = {}
+        if step in snapshot_set:
+            channel_values[channel] = _DeltaSnapshot(
+                write_value_fn(step),
+            )
+            channel_versions[channel] = step + 1
+
+        cp = Checkpoint(
+            v=1,
+            id=str(uuid6(clock_seq=-1)),
+            ts="",
+            channel_values=channel_values,
+            channel_versions=channel_versions,
+            versions_seen={},
+            updated_channels=None,
+        )
+        new_versions = dict(channel_versions)
+        parent_cfg = await saver.aput(
+            config, cp, generate_metadata(step=step), new_versions
+        )
+        stored.append(parent_cfg)
+
+        # Write a pending write for non-snapshot steps so the walk has
+        # something to collect.
+        if step not in snapshot_set:
+            await saver.aput_writes(
+                parent_cfg, [(channel, write_value_fn(step))], str(uuid4())
+            )
+
+    return stored

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

@@ -0,0 +1,247 @@
+"""DELTA_CHANNEL_HISTORY capability tests — aget_delta_channel_history 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_history_returns_writes_oldest_first(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Writes are returned oldest-to-newest."""
+    tid = str(uuid4())
+    # 5 steps: snapshot at 0, writes at 1,2,3,4.
+    # Head is step 4. Walk starts at step 3 (parent of head).
+    # Collects writes from steps 1,2,3 (between snapshot at 0 and head's parent).
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=5
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert values == [1, 2, 3], f"Expected [1,2,3], got {values}"
+
+
+async def test_history_seed_is_nearest_snapshot(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Seed is the value from the nearest ancestor with channel_values populated."""
+    tid = str(uuid4())
+    # 6 steps: snapshots at 0 and 3, writes at 1,2,4,5.
+    # Head is step 5. Walk from step 4 backward stops at step 3 (snapshot).
+    # Collects writes from step 4 only (between step 3 and head's parent step 4).
+    configs = await build_delta_chain(
+        saver,
+        thread_id=tid,
+        channel="ch",
+        snapshots_at_steps=[0, 3],
+        total_steps=6,
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    assert "seed" in result["ch"], "Expected seed from snapshot at step 3"
+    seed = result["ch"]["seed"]
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+
+    actual_value = seed.value if isinstance(seed, _DeltaSnapshot) else seed
+    assert actual_value == 3, f"Expected seed value 3 (step 3), got {actual_value}"
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert values == [4], f"Expected [4], got {values}"
+
+
+async def test_history_excludes_target_pending_writes(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Target's own pending_writes are NOT included in the history."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=3
+    )
+    head = configs[-1]
+    # Add writes directly to the head checkpoint
+    await saver.aput_writes(head, [("ch", "extra")], str(uuid4()))
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert "extra" not in values, f"Target's writes should be excluded, got {values}"
+
+
+async def test_history_multi_channel(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Multiple channels have independent walk termination."""
+    tid = str(uuid4())
+    configs: list = []
+    parent_cfg = None
+    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
+
+    for step in range(5):
+        config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
+        if parent_cfg:
+            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
+                "checkpoint_id"
+            ]
+        cv: dict = {}
+        cvs: dict = {}
+        if step == 1:
+            cv["a"] = _DeltaSnapshot("snap_a")
+            cvs["a"] = step + 1
+        if step == 3:
+            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)
+        await saver.aput_writes(parent_cfg, [("a", step), ("b", step)], str(uuid4()))
+
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["a", "b"])
+    a_writes = [w[2] for w in result["a"]["writes"]]
+    b_writes = [w[2] for w in result["b"]["writes"]]
+    assert a_writes == [1, 2, 3], f"Expected a writes [1,2,3], got {a_writes}"
+    assert b_writes == [3], f"Expected b writes [3], got {b_writes}"
+
+
+async def test_history_empty_channels_returns_empty(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Empty channels list returns empty mapping."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=3
+    )
+    result = await saver.aget_delta_channel_history(config=configs[-1], channels=[])
+    assert result == {}
+
+
+async def test_history_walk_to_root_no_seed(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Walk reaches root without finding seed — no 'seed' key in result."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver,
+        thread_id=tid,
+        channel="ch",
+        snapshots_at_steps=[],
+        total_steps=4,
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    assert "seed" not in result["ch"], f"Expected no seed, got {result['ch']}"
+
+
+async def test_history_migration_plain_value_as_seed(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Pre-delta plain value in channel_values acts as seed (migration case).
+
+    When a thread was originally using a regular channel (BinaryOperatorAggregate)
+    and later switches to DeltaChannel, the old checkpoint has a plain value in
+    channel_values[ch] (not a _DeltaSnapshot). The walk should treat it as the
+    seed and terminate there.
+    """
+    from langgraph.checkpoint.base import Checkpoint
+    from langgraph.checkpoint.base.id import uuid6
+
+    from langgraph.checkpoint.conformance.test_utils import generate_metadata
+
+    tid = str(uuid4())
+    configs: list = []
+    parent_cfg = None
+
+    for step in range(4):
+        config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
+        if parent_cfg:
+            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
+                "checkpoint_id"
+            ]
+        cv: dict = {}
+        cvs: dict = {}
+        # Step 1: plain value (migration case — old checkpoint before delta)
+        if step == 1:
+            cv["ch"] = [10, 20, 30]
+            cvs["ch"] = 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)
+        if step != 1:
+            await saver.aput_writes(parent_cfg, [("ch", step)], str(uuid4()))
+
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    # Seed should be the plain value from step 1
+    assert "seed" in result["ch"], "Expected seed from migration plain value at step 1"
+    seed = result["ch"]["seed"]
+    assert seed == [10, 20, 30], f"Expected plain value [10,20,30], got {seed}"
+    # Writes should be from step 2 only (between seed at step 1 and head's parent step 2)
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert values == [2], f"Expected [2], got {values}"
+
+
+ALL_DELTA_CHANNEL_HISTORY_TESTS = [
+    test_history_returns_writes_oldest_first,
+    test_history_seed_is_nearest_snapshot,
+    test_history_excludes_target_pending_writes,
+    test_history_multi_channel,
+    test_history_empty_channels_returns_empty,
+    test_history_walk_to_root_no_seed,
+    test_history_migration_plain_value_as_seed,
+]
+
+
+async def run_delta_channel_history_tests(
+    saver: BaseCheckpointSaver,
+    on_test_result: Callable[[str, str, bool, str | None], None] | None = None,
+) -> tuple[int, int, list[str]]:
+    """Run all delta_channel_history tests. Returns (passed, failed, failure_names)."""
+    passed = 0
+    failed = 0
+    failures: list[str] = []
+    for test_fn in ALL_DELTA_CHANNEL_HISTORY_TESTS:
+        try:
+            await test_fn(saver)
+            passed += 1
+            if on_test_result:
+                on_test_result("delta_channel_history", 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_history",
+                    test_fn.__name__,
+                    False,
+                    traceback.format_exc(),
+                )
+    return passed, failed, failures

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

@@ -19,6 +19,9 @@ from langgraph.checkpoint.conformance.spec.test_delete_for_runs import (
 from langgraph.checkpoint.conformance.spec.test_delete_thread import (
     run_delete_thread_tests,
 )
+from langgraph.checkpoint.conformance.spec.test_delta_channel_history import (
+    run_delta_channel_history_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
@@ -35,6 +38,7 @@ _RUNNERS = {
     Capability.DELETE_FOR_RUNS: run_delete_for_runs_tests,
     Capability.COPY_THREAD: run_copy_thread_tests,
     Capability.PRUNE: run_prune_tests,
+    Capability.DELTA_CHANNEL_HISTORY: run_delta_channel_history_tests,
 }
 
 

libs/checkpoint-conformance/pyproject.toml

@@ -43,7 +43,11 @@ asyncio_mode = "auto"
 # The extended methods (acopy_thread, adelete_for_runs, aprune) are checked
 # at runtime via capability detection and may not exist on the installed
 # base class. Dict literal inference is also overly strict for RunnableConfig.
+# Delta-channel tests import from `langgraph` (not a declared dep of this
+# package — at test time it is installed alongside); private `_DeltaSnapshot`
+# imports are intentional (beta surface).
 unresolved-attribute = "ignore"
+unresolved-import = "ignore"
 invalid-argument-type = "ignore"
 invalid-return-type = "ignore"
 
@@ -58,6 +62,9 @@ lint.select = [
 lint.ignore = ["E501", "B008"]
 target-version = "py310"
 
+[tool.uv.sources]
+langgraph-checkpoint = {path = "../checkpoint", editable = true}
+
 [[tool.uv.index]]
 name = "testpypi"
 url = "https://test.pypi.org/simple/"

libs/checkpoint-sqlite/tests/test_conformance_delta.py

@@ -0,0 +1,35 @@
+"""Run delta-channel conformance capabilities against AsyncSqliteSaver."""
+
+from __future__ import annotations
+
+import pytest
+
+pytest.importorskip(
+    "langgraph.checkpoint.conformance",
+    reason="langgraph-checkpoint-conformance not installed",
+)
+pytest.importorskip("aiosqlite", reason="aiosqlite not installed")
+
+
+@pytest.mark.asyncio
+async def test_delta_channel_conformance():
+    from langgraph.checkpoint.conformance import validate
+    from langgraph.checkpoint.conformance.initializer import checkpointer_test
+
+    from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+
+    @checkpointer_test(name="AsyncSqliteSaver")
+    async def sqlite_saver():
+        async with AsyncSqliteSaver.from_conn_string(":memory:") as saver:
+            yield saver
+
+    report = await validate(
+        sqlite_saver,
+        capabilities={
+            "delta_channel_history",
+        },
+    )
+    for cap, result in report.results.items():
+        if result.passed is False:
+            details = "\n".join(result.failures or [])
+            pytest.fail(f"Capability {cap} failed:\n{details}")

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

@@ -327,6 +327,14 @@ class BaseCheckpointSaver(Generic[V]):
 
         Args:
             run_ids: The run IDs whose checkpoints should be deleted.
+
+        !!! warning "DeltaChannel"
+
+            Deleting a run that produced ancestor `checkpoint_writes` — or
+            the only `_DeltaSnapshot` blob — for a still-live thread will
+            break reconstruction of any `DeltaChannel` whose history
+            depended on those rows. See the `DeltaChannel` note on `prune`
+            for safe-recovery strategies.
         """
         raise NotImplementedError
 
@@ -340,6 +348,17 @@ class BaseCheckpointSaver(Generic[V]):
         Args:
             source_thread_id: The thread ID to copy from.
             target_thread_id: The thread ID to copy to.
+
+        !!! warning "DeltaChannel"
+
+            Implementations must copy the **complete** parent chain (all
+            ancestor checkpoints and their `checkpoint_writes`) — copying
+            only the head checkpoint will leave the target thread with
+            `DeltaChannel` state that cannot be reconstructed (no path back
+            to a `_DeltaSnapshot` ancestor). Equivalently, the copy must
+            include enough ancestors that every `DeltaChannel`-backed key
+            has either a `_DeltaSnapshot` in `channel_values` somewhere in
+            the chain, or a complete write history back to the chain root.
         """
         raise NotImplementedError
 
@@ -355,6 +374,34 @@ class BaseCheckpointSaver(Generic[V]):
             thread_ids: The thread IDs to prune.
             strategy: The pruning strategy. `"keep_latest"` retains only the most
                 recent checkpoint per namespace. `"delete"` removes all checkpoints.
+
+        !!! warning "DeltaChannel"
+
+            Custom implementations must be `DeltaChannel`-aware. `DeltaChannel`
+            stores only a sentinel in `channel_values` for non-snapshot steps;
+            reconstruction walks the parent chain via
+            `get_delta_channel_history`, accumulating rows from
+            `checkpoint_writes` until it reaches an ancestor whose
+            `channel_values` contains a `_DeltaSnapshot` blob (written every
+            `snapshot_frequency` updates).
+
+            A naive `"keep_latest"` that drops intermediate checkpoints and
+            their writes can sever that chain: the surviving "latest"
+            checkpoint is rarely a snapshot point itself, so its delta
+            channels would silently reconstruct as empty (no error raised —
+            `get_delta_channel_history` simply returns no `seed`). Safe
+            options when the graph uses `DeltaChannel`:
+
+            * Walk back from each kept checkpoint and preserve every
+              ancestor (plus its `checkpoint_writes`) up to the nearest one
+              whose `channel_values` already contains a `_DeltaSnapshot` for
+              every `DeltaChannel`-backed key.
+            * Force a fresh snapshot on the kept checkpoint before deleting
+              ancestors — rewrite `channel_values[k] = _DeltaSnapshot(value)`
+              for each delta channel `k` (resolving `value` via the existing
+              ancestor walk first), then prune.
+            * Skip pruning threads whose graph uses `DeltaChannel` until one
+              of the above is implemented.
         """
         raise NotImplementedError
 
@@ -471,6 +518,13 @@ class BaseCheckpointSaver(Generic[V]):
 
         Args:
             run_ids: The run IDs whose checkpoints should be deleted.
+
+        !!! warning "DeltaChannel"
+
+            See `delete_for_runs` — deleting rows a still-live thread's
+            `DeltaChannel` reconstruction depends on (writes between the
+            head and its nearest `_DeltaSnapshot` ancestor) will silently
+            corrupt that channel's state.
         """
         raise NotImplementedError
 
@@ -484,6 +538,13 @@ class BaseCheckpointSaver(Generic[V]):
         Args:
             source_thread_id: The thread ID to copy from.
             target_thread_id: The thread ID to copy to.
+
+        !!! warning "DeltaChannel"
+
+            See `copy_thread` — the copy must carry the complete parent
+            chain (or at least back to a `_DeltaSnapshot` ancestor for every
+            `DeltaChannel`) so the target thread can reconstruct delta
+            state.
         """
         raise NotImplementedError
 
@@ -499,6 +560,13 @@ class BaseCheckpointSaver(Generic[V]):
             thread_ids: The thread IDs to prune.
             strategy: The pruning strategy. `"keep_latest"` retains only the most
                 recent checkpoint per namespace. `"delete"` removes all checkpoints.
+
+        !!! warning "DeltaChannel"
+
+            See `prune` for the full `DeltaChannel` caveat. In short:
+            `"keep_latest"` must not drop ancestor checkpoints / writes that
+            sit between the kept checkpoint and the nearest `_DeltaSnapshot`
+            ancestor, or delta channels will silently reconstruct as empty.
         """
         raise NotImplementedError
 

libs/checkpoint/tests/test_conformance_delta.py

@@ -0,0 +1,33 @@
+"""Run delta-channel conformance capabilities against InMemorySaver."""
+
+from __future__ import annotations
+
+import pytest
+
+conformance = pytest.importorskip(
+    "langgraph.checkpoint.conformance",
+    reason="langgraph-checkpoint-conformance not installed",
+)
+
+
+@pytest.mark.asyncio
+async def test_delta_channel_conformance():
+    from langgraph.checkpoint.conformance import validate
+    from langgraph.checkpoint.conformance.initializer import checkpointer_test
+
+    from langgraph.checkpoint.memory import InMemorySaver
+
+    @checkpointer_test(name="InMemorySaver")
+    async def mem_saver():
+        yield InMemorySaver()
+
+    report = await validate(
+        mem_saver,
+        capabilities={
+            "delta_channel_history",
+        },
+    )
+    for cap, result in report.results.items():
+        if result.passed is False:
+            details = "\n".join(result.failures or [])
+            pytest.fail(f"Capability {cap} failed:\n{details}")

libs/langgraph/langgraph/stream/__init__.py

@@ -17,9 +17,9 @@ from langgraph.stream.transformers import (
     CheckpointsTransformer,
     CustomTransformer,
     DebugTransformer,
+    LifecycleEvent,
     LifecyclePayload,
     LifecycleTransformer,
-    SubgraphStatus,
     SubgraphTransformer,
     TasksTransformer,
     UpdatesTransformer,
@@ -32,13 +32,13 @@ __all__ = [
     "CustomTransformer",
     "DebugTransformer",
     "GraphRunStream",
+    "LifecycleEvent",
     "LifecyclePayload",
     "LifecycleTransformer",
     "ProtocolEvent",
     "StreamChannel",
     "StreamTransformer",
     "SubgraphRunStream",
-    "SubgraphStatus",
     "SubgraphTransformer",
     "TasksTransformer",
     "UpdatesTransformer",

libs/langgraph/langgraph/stream/run_stream.py

@@ -12,7 +12,7 @@ from langgraph.stream._mux import StreamMux
 from langgraph.stream._types import ProtocolEvent
 
 if TYPE_CHECKING:
-    from langgraph.stream.transformers import SubgraphStatus
+    from langgraph.stream.transformers import LifecycleEvent
 
 
 def _drive_until_done(pump: Callable[[], bool]) -> None:
@@ -523,8 +523,8 @@ class _SubgraphRunStreamMixin:
 
     path: tuple[str, ...]
     graph_name: str | None
-    trigger_call_id: str | None
-    status: SubgraphStatus
+    parent_task_id: str | None
+    status: LifecycleEvent
     error: str | None
     _seen_terminal: bool
 
@@ -538,7 +538,7 @@ class SubgraphRunStream(GraphRunStream, _SubgraphRunStreamMixin):
         *,
         path: tuple[str, ...],
         graph_name: str | None = None,
-        trigger_call_id: str | None = None,
+        parent_task_id: str | None = None,
     ) -> None:
         # Capture the parent-inherited pump before super().__init__
         # touches anything; we delegate to it from `_pump_next`.
@@ -550,7 +550,7 @@ class SubgraphRunStream(GraphRunStream, _SubgraphRunStreamMixin):
         )
         self.path = path
         self.graph_name = graph_name
-        self.trigger_call_id = trigger_call_id
+        self.parent_task_id = parent_task_id
         self.status = "started"
         self.error = None
         self._seen_terminal = False
@@ -581,7 +581,7 @@ class AsyncSubgraphRunStream(AsyncGraphRunStream, _SubgraphRunStreamMixin):
         *,
         path: tuple[str, ...],
         graph_name: str | None = None,
-        trigger_call_id: str | None = None,
+        parent_task_id: str | None = None,
     ) -> None:
         self._parent_apump_fn: Callable[[], Awaitable[bool]] | None = mux._apump_fn
         super().__init__(
@@ -591,7 +591,7 @@ class AsyncSubgraphRunStream(AsyncGraphRunStream, _SubgraphRunStreamMixin):
         )
         self.path = path
         self.graph_name = graph_name
-        self.trigger_call_id = trigger_call_id
+        self.parent_task_id = parent_task_id
         self.status = "started"
         self.error = None
         self._seen_terminal = False

libs/langgraph/langgraph/stream/transformers.py

@@ -327,11 +327,31 @@ class MessagesTransformer(StreamTransformer):
         self._by_run.clear()
 
 
-SubgraphStatus = Literal["started", "completed", "failed", "interrupted", "drained"]
+LifecycleEvent = Literal["started", "completed", "failed", "interrupted", "drained"]
+"""State transition surfaced on the `lifecycle` channel for a tracked subgraph.
+
+Each value:
+
+- `started` — first `tasks` event observed at the tracked namespace. Carries
+  `graph_name` / `parent_task_id` / optional `metadata` describing what spawned
+  the subgraph.
+- `completed` — the dispatching task's `TaskResultPayload` arrived with neither
+  error nor interrupts. Also emitted by `finalize` for any tracked namespace
+  still open at run end.
+- `failed` — the dispatching task's result carried an `error`, OR the run
+  failed at top level with a non-interrupt / non-drain exception. Carries
+  `error` (string).
+- `interrupted` — the dispatching task's result carried `interrupts` (takes
+  precedence over `error` if both present), OR the run failed with
+  `GraphInterrupt`.
+- `drained` — the run was cooperatively stopped at a superstep boundary via
+  `RunControl.request_drain()` (e.g. SIGTERM). The checkpoint is saved and
+  the run is resumable.
+"""
 
 
 def _parse_ns_segment(segment: str) -> tuple[str, str | None]:
-    """Split a namespace segment into `(graph_name, trigger_call_id)`.
+    """Split a namespace segment into `(graph_name, parent_task_id)`.
 
     Segments are formatted `node_name:task_id` by `prepare_next_tasks`.
     Returns `(segment, None)` if no `:` is present.
@@ -340,6 +360,36 @@ def _parse_ns_segment(segment: str) -> tuple[str, str | None]:
     return name, task_id if sep else None
 
 
+def _extract_dispatching_tool_call_id(payload: Any) -> str | None:
+    """Return the model-side `tool_call_id` from a per-call dispatched task's
+    `input`, or `None` if the payload doesn't match a recognised shape.
+
+    Two shapes are recognised; both are duck-typed so any tool runner
+    that mimics the layout participates without naming any specific
+    dispatcher's types:
+
+    1. Single-element list of tool-call dicts:
+       `[{"id": ..., "name": ..., "args": {...}}]`. The current public
+       shape — `langchain.agents.create_agent` Send-fans this out per
+       pending tool call.
+    2. Dict envelope wrapping a tool call:
+       `{"tool_call": {"id": ..., "args": {...}, ...}, ...}`. Older
+       prebuilt agent paths Send-fan-out this shape.
+    """
+    if isinstance(payload, dict):
+        tool_call = payload.get("tool_call")
+        if not isinstance(tool_call, dict):
+            return None
+    elif (
+        isinstance(payload, list) and len(payload) == 1 and isinstance(payload[0], dict)
+    ):
+        tool_call = payload[0]
+    else:
+        return None
+    raw_id = tool_call.get("id")
+    return raw_id if isinstance(raw_id, str) else None
+
+
 class LifecyclePayload(TypedDict, total=False):
     """Payload of a lifecycle event surfaced on the `lifecycle` channel.
 
@@ -349,11 +399,54 @@ class LifecyclePayload(TypedDict, total=False):
     `run.lifecycle`.
     """
 
-    event: SubgraphStatus
+    event: LifecycleEvent
+    """State transition. See `LifecycleEvent` for per-value semantics."""
     namespace: list[str]
+    """Checkpoint namespace of the subgraph the event is about. Always present.
+
+    A list of `node_name:task_id` segments, one per nesting level (root has
+    `[]`, a direct child of root has `["agent:abc123"]`, a grandchild has
+    `["agent:abc123", "tool:def456"]`, etc.). Stable identity across the
+    `started → terminal` pair for the same subgraph instance.
+    """
     graph_name: NotRequired[str]
-    trigger_call_id: NotRequired[str]
+    """Name of the parent-scope node that dispatched this subgraph
+    (`add_node` name, surrounding tool's name for in-tool invokes,
+    `Send` target name, etc.) — parsed from the namespace tail
+    segment. Absent when the segment has no `:` separator.
+    """
+    parent_task_id: str
+    """Pregel task id of the dispatching task — the task whose execution
+    spawned this subgraph.
+
+    Always present on every event for the same subgraph instance. This is
+    the join key for correlating `started` ↔ terminal events and for
+    matching a `started` back to its `tasks` parent. Each Send produces
+    its own pregel task with its own id, so the join is 1:1 even when a
+    model dispatches multiple parallel tool calls in one turn.
+    """
+    metadata: NotRequired[dict[str, Any]]
+    """Optional generic descriptor of *what triggered* this subgraph.
+    Forwarded by protocol layers as the wire `lifecycle.started.metadata` field.
+
+    Shape:
+
+    - `{"type": "tool_call", "tool_call_id": "<id>"}` — set when the subgraph
+      was triggered by a per-call tool dispatch (a model tool call routed
+      through whatever tool node the agent uses). `tool_call_id` is the
+      model-side id of the originating tool call, exposed so UI consumers
+      can anchor the lifecycle event back to the AI message that dispatched
+      it. The langgraph layer deliberately doesn't mine `args` — those live
+      on the AIMessage's `tool_calls[i].args` already, and consumers that
+      want descriptive intent (subagent type, prompt text, etc.) look it
+      up there to keep one source of truth.
+
+    Absent for structurally-triggered subgraphs (parallel branches via
+    `Send` with non-tool-call payloads, nested `graph.invoke()`, etc.) and
+    for tool dispatches whose envelope carried no `id`.
+    """
     error: NotRequired[str]
+    """Error string. Set on `failed` events; absent otherwise."""
 
 
 class _TasksLifecycleBase(StreamTransformer):
@@ -371,13 +464,13 @@ class _TasksLifecycleBase(StreamTransformer):
 
     - `_should_track(ns)` — scope filter (e.g. multi-depth vs
       direct-children-only).
-    - `_on_started(ns, graph_name, trigger_call_id)` — first sighting
-      action (push payload / build handle / etc.). Called once per
-      discovered namespace.
-    - `_on_terminal(ns, status, error)` — terminal action (push
-      terminal payload / mark handle status). Called once per
-      tracked namespace at result time, or via `finalize` / `fail`
-      sweeps if no parent result arrived.
+    - `_on_started(ns, graph_name, parent_task_id, tool_call_id)` —
+      first sighting action (push payload / build handle / etc.).
+      Called once per discovered namespace.
+    - `_on_terminal(ns, status, error, parent_task_id)` — terminal
+      action (push terminal payload / mark handle status). Called
+      once per tracked namespace at result time, or via `finalize` /
+      `fail` sweeps if no parent result arrived.
 
     Tasks events are suppressed from the main event log (`process`
     returns False) — they're folded into whichever projection the
@@ -390,9 +483,15 @@ class _TasksLifecycleBase(StreamTransformer):
     def __init__(self, scope: tuple[str, ...] = ()) -> None:
         super().__init__(scope)
         self._seen: set[tuple[str, ...]] = set()
-        # Maps tracked namespace -> task_id of the parent task whose
+        # Maps tracked namespace -> task_id of the dispatching task whose
         # `TaskResultPayload` will close it.
         self._open: dict[tuple[str, ...], str] = {}
+        # Maps task_id -> model-side `tool_call_id` for tasks whose `input`
+        # matched a recognized per-call tool-dispatch shape. The lifecycle
+        # hook joins on this when a child subgraph fires its first task
+        # event so it can anchor the lifecycle.started to the originating
+        # AI message tool call.
+        self._dispatching_tool_call_id: dict[str, str] = {}
 
     # --- Template-method hooks (subclass overrides) ---
 
@@ -404,19 +503,33 @@ class _TasksLifecycleBase(StreamTransformer):
         self,
         ns: tuple[str, ...],
         graph_name: str | None,
-        trigger_call_id: str | None,
+        parent_task_id: str | None,
+        tool_call_id: str | None = None,
     ) -> None:
-        """Fired once per discovered namespace (first observed task event)."""
+        """Fired once per discovered namespace (first observed task event).
+
+        `tool_call_id` is the model-side id of the originating tool call
+        (from the per-call dispatched task's `input`). `None` for
+        structurally-triggered subgraphs or per-call envelopes that omitted
+        an `id`. Consumers join on `parent_task_id` (the pregel task id)
+        for identity; `tool_call_id` is purely an anchor back to the AI
+        message that dispatched the subgraph.
+        """
         raise NotImplementedError
 
     def _on_terminal(
         self,
         ns: tuple[str, ...],
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
+        parent_task_id: str,
     ) -> None:
-        """Fired once per tracked namespace when its parent's result arrives,
-        or via finalize/fail safety-net sweeps.
+        """Fired once per tracked namespace when its dispatching task's
+        result arrives, or via finalize/fail safety-net sweeps.
+
+        `parent_task_id` is the same id paired with the namespace at
+        `_on_started` time, so subscribers can correlate the terminal
+        event back to its `started`.
         """
         raise NotImplementedError
 
@@ -430,56 +543,96 @@ class _TasksLifecycleBase(StreamTransformer):
         if "result" in data:
             self._handle_task_result(ns, data)
         else:
-            self._handle_task_start(ns)
+            self._handle_task_start(ns, data)
         # Tasks events are folded into the synthesized projections;
         # suppress from the main event log so iterators don't double-see
         # the same information in two shapes.
         return False
 
-    def _handle_task_start(self, ns: tuple[str, ...]) -> None:
+    def _handle_task_start(self, ns: tuple[str, ...], data: dict[str, Any]) -> None:
+        # Mine input shape on every tasks event (not just tracked ones)
+        # so we capture dispatching tasks that themselves live outside the
+        # tracked region but whose `id` will appear as `parent_task_id`
+        # for a child subgraph.
+        self._record_dispatching_tool_call_id(data)
         if not self._should_track(ns) or ns in self._seen:
             return
         self._seen.add(ns)
-        graph_name, trigger_call_id = _parse_ns_segment(ns[-1])
-        self._on_started(ns, graph_name or None, trigger_call_id)
-        if trigger_call_id is not None:
-            self._open[ns] = trigger_call_id
+        graph_name, parent_task_id = _parse_ns_segment(ns[-1])
+        tool_call_id = (
+            self._dispatching_tool_call_id.pop(parent_task_id, None)
+            if parent_task_id is not None
+            else None
+        )
+        self._on_started(
+            ns,
+            graph_name or None,
+            parent_task_id,
+            tool_call_id,
+        )
+        if parent_task_id is not None:
+            self._open[ns] = parent_task_id
+
+    def _record_dispatching_tool_call_id(self, data: dict[str, Any]) -> None:
+        """Remember `task_id -> tool_call_id` if the task input matches
+        a recognized per-call tool-dispatch shape.
+
+        Shape detection and id extraction both live in
+        `_extract_dispatching_tool_call_id`; this method just records the
+        mapping under the dispatching task's own `id` so the lifecycle hook
+        can anchor a child subgraph back to the originating AI message
+        tool call when that subgraph's first task event arrives.
+        """
+        task_id = data.get("id")
+        if not isinstance(task_id, str):
+            return
+        tool_call_id = _extract_dispatching_tool_call_id(data.get("input"))
+        if tool_call_id is None:
+            return
+        self._dispatching_tool_call_id[task_id] = tool_call_id
 
     def _pop_terminal_transitions(
         self, ns: tuple[str, ...], data: dict[str, Any]
-    ) -> list[tuple[tuple[str, ...], SubgraphStatus, str | None]]:
-        """Return and remove tracked children closed by this task result."""
+    ) -> list[tuple[tuple[str, ...], LifecycleEvent, str | None, str]]:
+        """Return and remove tracked children closed by this task result.
+
+        Each tuple is `(child_ns, status, error, parent_task_id)`.
+        `parent_task_id` is the dispatching task's id — the same id
+        we'd already paired with the namespace at `_on_started`.
+        """
         result_id = data.get("id")
         if not result_id:
             return []
-        transitions: list[tuple[tuple[str, ...], SubgraphStatus, str | None]] = []
-        for child_ns, parent_task_id in list(self._open.items()):
-            if child_ns[:-1] != ns or parent_task_id != result_id:
+        transitions: list[tuple[tuple[str, ...], LifecycleEvent, str | None, str]] = []
+        for child_ns, dispatching_task_id in list(self._open.items()):
+            if child_ns[:-1] != ns or dispatching_task_id != result_id:
                 continue
             status, error = _terminal_from_result(data)
-            transitions.append((child_ns, status, error))
+            transitions.append((child_ns, status, error, dispatching_task_id))
             del self._open[child_ns]
         return transitions
 
     def _handle_task_result(self, ns: tuple[str, ...], data: dict[str, Any]) -> None:
-        for child_ns, status, error in self._pop_terminal_transitions(ns, data):
-            self._on_terminal(child_ns, status, error)
+        for child_ns, status, error, parent_task_id in self._pop_terminal_transitions(
+            ns, data
+        ):
+            self._on_terminal(child_ns, status, error, parent_task_id)
 
     def finalize(self) -> None:
         """Emit `completed` for any tracked namespace still open at run end."""
-        for ns in list(self._open):
-            self._on_terminal(ns, "completed", None)
+        for ns, parent_task_id in list(self._open.items()):
+            self._on_terminal(ns, "completed", None, parent_task_id)
         self._open.clear()
 
     def fail(self, err: BaseException) -> None:
         """Emit terminal status for any tracked namespace still open."""
         status, error_str = _status_from_exception(err)
-        for ns in list(self._open):
-            self._on_terminal(ns, status, error_str)
+        for ns, parent_task_id in list(self._open.items()):
+            self._on_terminal(ns, status, error_str, parent_task_id)
         self._open.clear()
 
 
-def _status_from_exception(err: BaseException) -> tuple[SubgraphStatus, str | None]:
+def _status_from_exception(err: BaseException) -> tuple[LifecycleEvent, str | None]:
     """Map a run exception to a subgraph terminal status and error string."""
     if isinstance(err, GraphDrained):
         return "drained", None
@@ -490,7 +643,7 @@ def _status_from_exception(err: BaseException) -> tuple[SubgraphStatus, str | No
 
 def _terminal_from_result(
     payload: dict[str, Any],
-) -> tuple[SubgraphStatus, str | None]:
+) -> tuple[LifecycleEvent, str | None]:
     """Map a `TaskResultPayload` to a `(status, error)` pair.
 
     Order matters: a result with both `error` and `interrupts` prefers
@@ -539,26 +692,37 @@ class LifecycleTransformer(_TasksLifecycleBase):
         self,
         ns: tuple[str, ...],
         graph_name: str | None,
-        trigger_call_id: str | None,
+        parent_task_id: str | None,
+        tool_call_id: str | None = None,
     ) -> None:
-        if trigger_call_id is None:
-            # Without a task id we can't correlate a parent-result
+        if parent_task_id is None:
+            # Without a task id we can't correlate a dispatching-task-result
             # event back to this namespace — skip the started payload
             # and rely on finalize/fail to close.
             return
-        payload: LifecyclePayload = {"event": "started", "namespace": list(ns)}
+        payload: LifecyclePayload = {
+            "event": "started",
+            "namespace": list(ns),
+            "parent_task_id": parent_task_id,
+        }
         if graph_name:
             payload["graph_name"] = graph_name
-        payload["trigger_call_id"] = trigger_call_id
+        if tool_call_id is not None:
+            payload["metadata"] = {"type": "tool_call", "tool_call_id": tool_call_id}
         self._channel.push(payload)
 
     def _on_terminal(
         self,
         ns: tuple[str, ...],
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
+        parent_task_id: str,
     ) -> None:
-        payload: LifecyclePayload = {"event": status, "namespace": list(ns)}
+        payload: LifecyclePayload = {
+            "event": status,
+            "namespace": list(ns),
+            "parent_task_id": parent_task_id,
+        }
         if error is not None:
             payload["error"] = error
         self._channel.push(payload)
@@ -611,7 +775,8 @@ class SubgraphTransformer(_TasksLifecycleBase):
         self,
         ns: tuple[str, ...],
         graph_name: str | None,
-        trigger_call_id: str | None,
+        parent_task_id: str | None,
+        tool_call_id: str | None = None,  # noqa: ARG002
     ) -> None:
         if self._mux is None:
             return
@@ -624,7 +789,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
             mux=child_mux,
             path=ns,
             graph_name=graph_name,
-            trigger_call_id=trigger_call_id,
+            parent_task_id=parent_task_id,
         )
         self._handles[ns] = handle
         self._log.push(handle)
@@ -632,8 +797,9 @@ class SubgraphTransformer(_TasksLifecycleBase):
     def _on_terminal(
         self,
         ns: tuple[str, ...],
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
+        parent_task_id: str,  # noqa: ARG002
     ) -> None:
         handle = self._handles.get(ns)
         if handle is None or not self._mark_terminal(handle, status, error):
@@ -643,8 +809,9 @@ class SubgraphTransformer(_TasksLifecycleBase):
     async def _aon_terminal(
         self,
         ns: tuple[str, ...],
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
+        parent_task_id: str,  # noqa: ARG002
     ) -> None:
         handle = self._handles.get(ns)
         if handle is None or not self._mark_terminal(handle, status, error):
@@ -654,7 +821,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
     def _mark_terminal(
         self,
         handle: SubgraphRunStream | AsyncSubgraphRunStream,
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
     ) -> bool:
         """Mark a handle terminal once. Returns True on first transition."""
@@ -669,7 +836,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
     def _close_or_fail_handle(
         self,
         handle: SubgraphRunStream | AsyncSubgraphRunStream,
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
     ) -> None:
         if handle._mux is None or handle._mux._events._closed:
@@ -682,7 +849,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
     async def _aclose_or_fail_handle(
         self,
         handle: SubgraphRunStream | AsyncSubgraphRunStream,
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
     ) -> None:
         if handle._mux is None or handle._mux._events._closed:
@@ -721,10 +888,15 @@ class SubgraphTransformer(_TasksLifecycleBase):
             ns = tuple(event["params"]["namespace"])
             data = event["params"]["data"]
             if "result" in data:
-                for child_ns, status, error in self._pop_terminal_transitions(ns, data):
-                    await self._aon_terminal(child_ns, status, error)
+                for (
+                    child_ns,
+                    status,
+                    error,
+                    parent_task_id,
+                ) in self._pop_terminal_transitions(ns, data):
+                    await self._aon_terminal(child_ns, status, error, parent_task_id)
             else:
-                self._handle_task_start(ns)
+                self._handle_task_start(ns, data)
             keep = False
         else:
             keep = True
@@ -736,9 +908,9 @@ class SubgraphTransformer(_TasksLifecycleBase):
 
     def _complete_open_handles(self) -> BaseException | None:
         first_error: BaseException | None = None
-        for ns in list(self._open):
+        for ns, parent_task_id in list(self._open.items()):
             try:
-                self._on_terminal(ns, "completed", None)
+                self._on_terminal(ns, "completed", None, parent_task_id)
             except BaseException as e:
                 if first_error is None:
                     first_error = e
@@ -754,9 +926,9 @@ class SubgraphTransformer(_TasksLifecycleBase):
 
     async def _acomplete_open_handles(self) -> BaseException | None:
         first_error: BaseException | None = None
-        for ns in list(self._open):
+        for ns, parent_task_id in list(self._open.items()):
             try:
-                await self._aon_terminal(ns, "completed", None)
+                await self._aon_terminal(ns, "completed", None, parent_task_id)
             except BaseException as e:
                 if first_error is None:
                     first_error = e

libs/langgraph/tests/test_retry.py

@@ -1674,15 +1674,28 @@ async def test_arun_with_retry_timeout_observer_tracks_attempts():
 async def test_arun_with_retry_timeout_observer_emits_progress_on_heartbeat():
     events: list = []
 
+    # `_TimedAttemptScope.__init__` sets `_last_progress` to `time.monotonic()`,
+    # but the watchdog itself doesn't start running until after `wrap_config`
+    # and task scheduling — under CI load that gap can be large enough to eat
+    # the entire idle window before the task body's first await even runs. We
+    # defend against that by:
+    #   1. Using a generous idle_timeout so scheduling slack stays well within it.
+    #   2. Calling `runtime.heartbeat()` BEFORE the first sleep, which resets
+    #      `_last_progress` to "now" the moment the task body actually starts.
+    idle_timeout_s = 1.0
+
     class HeartbeatProc:
         async def ainvoke(self, input, config):
             runtime = config[CONF][CONFIG_KEY_RUNTIME]
+            runtime.heartbeat()  # reset the idle clock at task-body entry
             for _ in range(8):
                 await asyncio.sleep(0.05)
                 runtime.heartbeat()
             return "ok"
 
-    task = _make_task(HeartbeatProc(), timeout=_idle_timeout(0.2), name="heartbeat")
+    task = _make_task(
+        HeartbeatProc(), timeout=_idle_timeout(idle_timeout_s), name="heartbeat"
+    )
     task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = events.append
     assert await arun_with_retry(task, retry_policy=None) == "ok"
 
@@ -1691,13 +1704,13 @@ async def test_arun_with_retry_timeout_observer_emits_progress_on_heartbeat():
     assert by_event[-1] == "finish"
     progress = [ev for ev in events if ev.event == "progress"]
     assert progress, "expected at least one progress event from heartbeat"
-    # Rate limit is `idle_timeout / 4` = 0.05s; with 8 heartbeats spaced ~0.05s
-    # we should see at most ~one progress event per heartbeat (well below 8).
+    # Rate limit is `idle_timeout / 4` = 0.25s; with the task running for
+    # ~400ms we expect 1–2 progress events (well below the 9 heartbeats).
     assert len(progress) <= len(by_event)
     for ev in progress:
         assert ev.context.task_name == "heartbeat"
         assert ev.context.attempt == 1
-        assert ev.context.idle_timeout_secs == 0.2
+        assert ev.context.idle_timeout_secs == idle_timeout_s
         assert isinstance(ev.progress_at, datetime)
 
 

libs/langgraph/tests/test_stream_lifecycle_transformer.py

@@ -35,8 +35,16 @@ def _tasks_start(
     *,
     task_id: str,
     name: str,
+    input: Any = None,
 ) -> dict[str, Any]:
-    """Build a `tasks` ProtocolEvent carrying a TaskPayload (start)."""
+    """Build a `tasks` ProtocolEvent carrying a TaskPayload (start).
+
+    Pass `input=[{"id": ..., "name": ..., "args": {...}}]` (the per-call
+    list shape `langchain.agents.create_agent` Send-fans out) or
+    `input={"tool_call": {"id": ..., ...}, ...}` (the dict envelope older
+    prebuilt agent paths emit) to exercise the lifecycle transformer's
+    `tool_call_id` mining for `lifecycle.started.metadata`.
+    """
     return {
         "type": "event",
         "method": "tasks",
@@ -46,7 +54,7 @@ def _tasks_start(
             "data": {
                 "id": task_id,
                 "name": name,
-                "input": None,
+                "input": input,
                 "triggers": [],
             },
         },
@@ -123,7 +131,220 @@ def test_started_emitted_on_first_direct_child_task() -> None:
     assert payload["event"] == "started"
     assert payload["namespace"] == ["agent:abc123"]
     assert payload["graph_name"] == "agent"
-    assert payload["trigger_call_id"] == "abc123"
+    assert payload["parent_task_id"] == "abc123"
+
+
+def test_started_carries_metadata_for_dict_envelope_input() -> None:
+    """When the dispatching task's `input` is a dict envelope with a
+    `tool_call` field (the shape older prebuilt agent paths Send-fan
+    out per call), the transformer mines `tool_call_id` from
+    `tool_call.id` and remembers it keyed by the dispatching task id.
+    When that task triggers a subgraph (the child's namespace ends in
+    `name:<dispatching_task_id>`), `lifecycle.started.metadata` carries
+    `{"type": "tool_call", "tool_call_id": ...}`. Identity correlation
+    still uses `parent_task_id`; `tool_call_id` is exposed so UI
+    consumers can anchor the lifecycle event back to the originating
+    AI message tool call. Args are deliberately NOT mined — they live
+    on the AIMessage and have a single source of truth there.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="abc123",
+            name="tools",
+            input={
+                "tool_call": {
+                    "id": "call_xyz",
+                    "name": "task",
+                    "args": {"subagent_type": "researcher"},
+                }
+            },
+        )
+    )
+    mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="model"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert payload["event"] == "started"
+    assert payload["parent_task_id"] == "abc123"
+    assert payload["metadata"] == {
+        "type": "tool_call",
+        "tool_call_id": "call_xyz",
+    }
+
+
+def test_started_carries_metadata_for_list_shape_per_call_input() -> None:
+    """`langchain.agents.create_agent` Send-fans out a per-call task
+    whose `input` is a single-element list of tool-call dicts:
+    `[{"id": ..., "name": ..., "args": {...}}]`. The transformer mines
+    `tool_call_id` exactly as for the dict envelope shape, so
+    `lifecycle.started.metadata` fires regardless of which agent factory
+    drove the dispatch.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="abc123",
+            name="tools",
+            input=[
+                {
+                    "id": "tc-1",
+                    "name": "task",
+                    "args": {"subagent_type": "researcher"},
+                }
+            ],
+        )
+    )
+    mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="model"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert payload["event"] == "started"
+    assert payload["parent_task_id"] == "abc123"
+    assert payload["metadata"] == {"type": "tool_call", "tool_call_id": "tc-1"}
+
+
+def test_started_carries_metadata_when_args_absent() -> None:
+    """`tool_call_id` is the only field metadata needs; the dispatching
+    envelope can omit `args` entirely (or have non-dict args) and we
+    still produce a metadata as long as `id` is a string."""
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="abc123",
+            name="tools",
+            input=[{"id": "tc-1", "name": "some_tool"}],
+        )
+    )
+    mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="model"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert payload["metadata"] == {"type": "tool_call", "tool_call_id": "tc-1"}
+
+
+def test_list_shape_ignored_when_not_single_element() -> None:
+    """Only single-element lists are recognized as the per-call shape;
+    a 0- or 2+-element list is some other batched/multi-call payload
+    and must not be mined."""
+    # Two-element list — not the per-call shape.
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="abc123",
+            name="tools",
+            input=[
+                {"id": "tc-1", "name": "task"},
+                {"id": "tc-2", "name": "task"},
+            ],
+        )
+    )
+    mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="model"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert "metadata" not in payload
+
+    # Empty list.
+    mux2 = _build_lifecycle_mux()
+    mux2.push(_tasks_start([], task_id="def456", name="tools", input=[]))
+    mux2.push(_tasks_start(["agent:def456"], task_id="t1", name="model"))
+    [payload2] = _drain_lifecycle(mux2)
+    assert "metadata" not in payload2
+
+
+def test_list_shape_robust_to_non_dict_or_missing_id() -> None:
+    """Duck-typing safety: a single-element list whose element isn't a
+    dict, or whose dict has no string `id`, must not raise — it just
+    leaves `metadata` absent."""
+    # Element is not a dict.
+    mux = _build_lifecycle_mux()
+    mux.push(_tasks_start([], task_id="t-a", name="tools", input=["not-a-dict"]))
+    mux.push(_tasks_start(["agent:t-a"], task_id="t1", name="model"))
+    [payload] = _drain_lifecycle(mux)
+    assert "metadata" not in payload
+
+    # Element has no `id`.
+    mux2 = _build_lifecycle_mux()
+    mux2.push(_tasks_start([], task_id="t-b", name="tools", input=[{"name": "task"}]))
+    mux2.push(_tasks_start(["agent:t-b"], task_id="t1", name="model"))
+    [payload2] = _drain_lifecycle(mux2)
+    assert "metadata" not in payload2
+
+    # Element's `id` is not a string.
+    mux3 = _build_lifecycle_mux()
+    mux3.push(_tasks_start([], task_id="t-c", name="tools", input=[{"id": 123}]))
+    mux3.push(_tasks_start(["agent:t-c"], task_id="t1", name="model"))
+    [payload3] = _drain_lifecycle(mux3)
+    assert "metadata" not in payload3
+
+
+def test_parallel_dispatches_attributed_to_correct_parent() -> None:
+    """Two dispatching task envelopes in the same model turn each fan
+    out to their own child subgraph; each child's `metadata.tool_call_id`
+    must reflect its own dispatching envelope, not the other.
+
+    Defends the `parent_task_id` (pregel task id) join: that id is
+    parsed from the child namespace segment and is unique per Send,
+    so it disambiguates parallel dispatches 1:1. Both children share
+    the same `subagent_type` (in args, not on metadata) — only the
+    pregel task id can tell them apart, so the `tool_call_id` must
+    follow the pregel id, not anything from `args`.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="parent_A",
+            name="tools",
+            input={
+                "tool_call": {
+                    "id": "call_1",
+                    "name": "task",
+                    "args": {"subagent_type": "researcher"},
+                }
+            },
+        )
+    )
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="parent_B",
+            name="tools",
+            input={
+                "tool_call": {
+                    "id": "call_2",
+                    "name": "task",
+                    "args": {"subagent_type": "researcher"},
+                }
+            },
+        )
+    )
+    mux.push(_tasks_start(["agent:parent_A"], task_id="t1", name="model"))
+    mux.push(_tasks_start(["agent:parent_B"], task_id="t2", name="model"))
+
+    payloads = _drain_lifecycle(mux)
+    by_ns = {tuple(p["namespace"]): p for p in payloads}
+    assert by_ns[("agent:parent_A",)]["metadata"] == {
+        "type": "tool_call",
+        "tool_call_id": "call_1",
+    }
+    assert by_ns[("agent:parent_B",)]["metadata"] == {
+        "type": "tool_call",
+        "tool_call_id": "call_2",
+    }
+
+
+def test_started_omits_metadata_for_structurally_triggered_subgraph() -> None:
+    """Subgraphs triggered without a recognizable tool-call envelope on
+    the parent's input (Send with custom payloads, plain nested
+    `graph.invoke`, etc.) don't get a `metadata` field on
+    `lifecycle.started`."""
+    mux = _build_lifecycle_mux()
+    mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="tool"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert "metadata" not in payload
 
 
 def test_started_dedup_on_repeat_namespace() -> None:
@@ -188,8 +409,12 @@ def test_completed_on_parent_task_result() -> None:
     mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
     mux.push(_tasks_result([], task_id="abc", name="agent"))
 
-    events = [p["event"] for p in _drain_lifecycle(mux)]
-    assert events == ["started", "completed"]
+    payloads = _drain_lifecycle(mux)
+    assert [p["event"] for p in payloads] == ["started", "completed"]
+    # `parent_task_id` is required on every event for the same subgraph
+    # so consumers can correlate `started` ↔ terminal without joining
+    # via `namespace`.
+    assert all(p["parent_task_id"] == "abc" for p in payloads)
 
 
 def test_failed_on_parent_task_result_with_error() -> None:
@@ -265,6 +490,54 @@ def test_fail_emits_failed_for_other_exceptions() -> None:
     assert payloads[1]["error"] == "boom"
 
 
+def test_parent_task_id_present_on_every_terminal_path() -> None:
+    """Every exit path that emits a terminal event (parent-result with
+    error / interrupts, finalize sweep, fail sweep) must carry
+    `parent_task_id` so consumers can correlate the terminal event
+    back to its `started` without falling back to namespace joins."""
+    # Path 1: parent-result with error.
+    mux = _build_lifecycle_mux()
+    mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
+    mux.push(_tasks_result([], task_id="abc", name="agent", error="boom"))
+    [_, terminal] = _drain_lifecycle(mux)
+    assert terminal["event"] == "failed"
+    assert terminal["parent_task_id"] == "abc"
+
+    # Path 2: parent-result with interrupts.
+    mux2 = _build_lifecycle_mux()
+    mux2.push(_tasks_start(["agent:def"], task_id="t1", name="tool"))
+    mux2.push(
+        _tasks_result([], task_id="def", name="agent", interrupts=[{"value": "pause"}])
+    )
+    [_, terminal2] = _drain_lifecycle(mux2)
+    assert terminal2["event"] == "interrupted"
+    assert terminal2["parent_task_id"] == "def"
+
+    # Path 3: finalize sweep (no parent result arrived).
+    mux3 = _build_lifecycle_mux()
+    mux3.push(_tasks_start(["agent:ghi"], task_id="t1", name="tool"))
+    mux3.close()
+    [_, terminal3] = _drain_lifecycle(mux3)
+    assert terminal3["event"] == "completed"
+    assert terminal3["parent_task_id"] == "ghi"
+
+    # Path 4: fail sweep with GraphInterrupt.
+    mux4 = _build_lifecycle_mux()
+    mux4.push(_tasks_start(["agent:jkl"], task_id="t1", name="tool"))
+    mux4.fail(GraphInterrupt())
+    [_, terminal4] = _drain_lifecycle(mux4)
+    assert terminal4["event"] == "interrupted"
+    assert terminal4["parent_task_id"] == "jkl"
+
+    # Path 5: fail sweep with generic exception.
+    mux5 = _build_lifecycle_mux()
+    mux5.push(_tasks_start(["agent:mno"], task_id="t1", name="tool"))
+    mux5.fail(RuntimeError("kaboom"))
+    [_, terminal5] = _drain_lifecycle(mux5)
+    assert terminal5["event"] == "failed"
+    assert terminal5["parent_task_id"] == "mno"
+
+
 def test_unrelated_methods_pass_through() -> None:
     """Non-`tasks` events are not consumed and don't emit lifecycle."""
     mux = _build_lifecycle_mux()

libs/langgraph/tests/test_stream_subgraph_transformer.py

@@ -195,7 +195,7 @@ def test_handle_created_on_first_direct_child_task() -> None:
     [handle] = _drain_subgraphs(mux)
     assert handle.path == ("agent:abc",)
     assert handle.graph_name == "agent"
-    assert handle.trigger_call_id == "abc"
+    assert handle.parent_task_id == "abc"
     assert handle.status == "started"
     _child_mux(handle)  # mini-mux backed