fix(lint): remove redundant cast flagged by mypy

chore: apply ruff format/lint fixes
fix(message): full add_messages parity for _messages_delta_reducer
2026-09-25 11:05:12 +02:00 · 2026-05-06 15:08:37 -04:00 · 2026-05-06 15:03:13 -04:00 · 2026-05-06 14:59:12 -04:00 · 2026-05-06 14:42:36 -04:00
40 changed files with 814 additions and 2364 deletions
@@ -63,7 +63,6 @@ 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.

@@ -23,7 +23,6 @@ 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.
@@ -43,7 +42,6 @@ EXTENDED_CAPABILITIES = frozenset(
        Capability.DELETE_FOR_RUNS,
        Capability.COPY_THREAD,
        Capability.PRUNE,
-        Capability.DELTA_CHANNEL_HISTORY,
    }
 )

@@ -59,7 +57,6 @@ _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",
 }


@@ -9,9 +9,6 @@ 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
@@ -27,5 +24,4 @@ __all__ = [
    "run_delete_for_runs_tests",
    "run_copy_thread_tests",
    "run_prune_tests",
-    "run_delta_channel_history_tests",
 ]
@@ -1,99 +0,0 @@
-"""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
@@ -1,247 +0,0 @@
-"""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
@@ -19,9 +19,6 @@ 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
@@ -38,7 +35,6 @@ _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,
 }


@@ -43,11 +43,7 @@ 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"

@@ -62,9 +58,6 @@ 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/"
@@ -279,8 +279,8 @@ def _build_delta_stage2_sql(
        )
    for _ in channels_with_seed:
        branches.append(
-            "SELECT 'b'::text AS _kind, NULL::text AS checkpoint_id, channel, "
-            "type, blob, NULL::text AS task_id, NULL::int AS idx, version "
+            "SELECT 'b'::text, NULL, channel, "
+            "type, blob, NULL, NULL, version "
            "FROM checkpoint_blobs "
            "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
            "AND version = %s"
@@ -1,35 +0,0 @@
-"""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}")
@@ -63,11 +63,6 @@ class CheckpointMetadata(TypedDict, total=False):
    delta_updates_since_snapshot: dict[str, int]
    """Per-channel update count since the last `_DeltaSnapshot` was written.

-    !!! warning "Beta"
-
-        This metadata field backs `DeltaChannel` (beta). The key name and
-        contents may change while the delta-channel design stabilizes.
-
    Maps channel name → number of supersteps that wrote to this channel
    since its last snapshot blob. Used by `pregel.create_checkpoint` to
    decide when to write the next snapshot (when the count reaches the
@@ -140,11 +135,6 @@ class CheckpointTuple(NamedTuple):
 class DeltaChannelHistory(TypedDict):
    """Per-channel result entry from `BaseCheckpointSaver.get_delta_channel_history`.

-    !!! warning "Beta"
-
-        Part of the `DeltaChannel` support surface; in beta. Field names and
-        semantics may change.
-
    Storage-level view of what one channel contributed across the ancestor
    chain of a target checkpoint:

@@ -327,14 +317,6 @@ 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

@@ -348,17 +330,6 @@ 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

@@ -374,34 +345,6 @@ 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

@@ -518,13 +461,6 @@ 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

@@ -538,13 +474,6 @@ 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

@@ -560,13 +489,6 @@ 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

@@ -575,14 +497,6 @@ class BaseCheckpointSaver(Generic[V]):
    ) -> Mapping[str, DeltaChannelHistory]:
        """Walk the parent chain returning per-channel writes + seed.

-        !!! warning "Beta"
-
-            This method is part of the `DeltaChannel` support surface and is
-            in beta. The signature, return shape (`DeltaChannelHistory`), and
-            interaction with `_DeltaSnapshot` blobs may change. Override at
-            your own risk; the default implementation will continue to work
-            against the public `BaseCheckpointSaver` contract.
-
        For each requested channel, walks ancestors of the checkpoint
        identified by `config` (following `parent_config`) and accumulates
        `pending_writes` for that channel. The walk terminates per-channel
@@ -642,13 +556,7 @@ class BaseCheckpointSaver(Generic[V]):
    async def aget_delta_channel_history(
        self, *, config: RunnableConfig, channels: Sequence[str]
    ) -> Mapping[str, DeltaChannelHistory]:
-        """Async version of `get_delta_channel_history`.
-
-        !!! warning "Beta"
-
-            This method is part of the `DeltaChannel` support surface and is
-            in beta. See `get_delta_channel_history` for caveats.
-        """
+        """Async version of `get_delta_channel_history`."""
        if not channels:
            return {}
        collected_by_ch: dict[str, list[PendingWrite]] = {c: [] for c in channels}
@@ -1,33 +0,0 @@
-"""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}")
@@ -1 +1 @@
-__version__ = "0.4.25"
+__version__ = "0.4.24"
@@ -26,15 +26,6 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
    """Reducer channel that stores only a sentinel in checkpoint blobs and
    reconstructs state by replaying ancestor writes through the reducer.

-    !!! warning "Beta"
-
-        `DeltaChannel` is in beta. The API and on-disk representation may
-        change in future releases. Threads written with `DeltaChannel` today
-        are expected to remain readable, but the surrounding contract
-        (`BaseCheckpointSaver.get_delta_channel_history`, the
-        `_DeltaSnapshot` blob shape, the `delta_updates_since_snapshot`
-        metadata field) is not yet stable.
-
    The reducer receives the current accumulated value and a batch of writes
    in one call: `reducer(state, [write1, write2, ...]) -> new_state`.

@@ -249,17 +249,15 @@ def _messages_delta_reducer(
 ) -> list[AnyMessage]:
    """**Experimental.** Batch reducer for use with `DeltaChannel`.

-    Processes all writes in one pass — dedup by ID, `RemoveMessage`
-    tombstoning — without calling `add_messages`.
+    Provides full `add_messages` parity: dedup by ID, `RemoveMessage`
+    tombstoning, `REMOVE_ALL_MESSAGES` reset, `BaseMessageChunk` coercion,
+    and UUID assignment for ID-less messages — all in a single batched pass.

    This reducer is batching-invariant, as required by `DeltaChannel`:
    `reducer(reducer(state, xs), ys) == reducer(state, xs + ys)`.

    Raw dict / string / tuple inputs are coerced to typed `BaseMessage`
-    objects so that HTTP-driven graphs work without a separate coercion
-    step. This is not full `add_messages` parity — `REMOVE_ALL_MESSAGES`,
-    unknown-id `RemoveMessage` errors, missing-id UUID assignment, and
-    `BaseMessageChunk` conversion are not handled here.
+    objects so that HTTP-driven graphs work without a separate coercion step.

    Example::

@@ -280,24 +278,51 @@ def _messages_delta_reducer(
            flat.extend(w)
        else:
            flat.append(w)
-    # Steady state: the reducer's own output is already typed, so skip
-    # `convert_to_messages` on state when the first element is a BaseMessage.
+    # Steady state: the reducer's own output is already typed BaseMessages
+    # (never chunks), so skip convert_to_messages on the fast path.
    # Only raw input (initial dicts, deserialized blobs) hits the slow path.
    if state and isinstance(state[0], BaseMessage):
        state_msgs = state
    else:
-        state_msgs = cast("list[AnyMessage]", convert_to_messages(state))
-    msgs = cast("list[AnyMessage]", convert_to_messages(flat))
+        state_msgs = cast(
+            "list[AnyMessage]",
+            [
+                message_chunk_to_message(cast(BaseMessageChunk, m))
+                for m in convert_to_messages(state)
+            ],
+        )
+    # Coerce chunks to full messages — streaming nodes can emit BaseMessageChunk.
+    msgs = cast(
+        "list[AnyMessage]",
+        [
+            message_chunk_to_message(cast(BaseMessageChunk, m))
+            for m in convert_to_messages(flat)
+        ],
+    )

-    index: dict[str, int] = {
-        m.id: i for i, m in enumerate(state_msgs) if m.id is not None
-    }
+    # REMOVE_ALL_MESSAGES resets everything; find the last sentinel and
+    # discard all state plus all writes before it.
+    remove_all_idx = None
+    for idx, m in enumerate(msgs):
+        if isinstance(m, RemoveMessage) and m.id == REMOVE_ALL_MESSAGES:
+            remove_all_idx = idx
+    if remove_all_idx is not None:
+        state_msgs = []
+        msgs = msgs[remove_all_idx + 1 :]
+
+    # Build index and assign missing IDs in one pass (parity with add_messages
+    # so that eviction and RemoveMessage tombstoning work on ID-less messages).
+    index: dict[str, int] = {}
+    for i, m in enumerate(state_msgs):
+        if m.id is None:
+            m.id = str(uuid.uuid4())
+        index[m.id] = i
    result: list[AnyMessage | None] = list(state_msgs)
    for msg in msgs:
+        if msg.id is None:
+            msg.id = str(uuid.uuid4())
        mid = msg.id
-        if mid is None:
-            result.append(msg)
-        elif isinstance(msg, RemoveMessage):
+        if isinstance(msg, RemoveMessage):
            if mid in index:
                result[index[mid]] = None
                del index[mid]
@@ -34,23 +34,28 @@ def empty_checkpoint() -> Checkpoint:
    )


-def delta_channels_to_snapshot(
-    channels: Mapping[str, BaseChannel],
-    counts: Mapping[str, int],
-) -> set[str]:
-    """Return the set of DeltaChannel names that should snapshot now.
+def _should_snapshot_delta(
+    name: str,
+    ch: DeltaChannel,
+    updates_since_snapshot: Mapping[str, int],
+    *,
+    force: bool,
+) -> bool:
+    """Decide whether `ch` should write a `_DeltaSnapshot` this step.

-    A channel snapshots when its accumulated update count (since the last
-    snapshot) reaches or exceeds `snapshot_frequency`. This is a pure
-    predicate — no mutation.
+    Triggers:
+      * `force` — always snapshot (used by `durability="exit"`).
+      * Update-count: this channel has accumulated at least
+        `snapshot_frequency` updates since its last snapshot. The count
+        is supplied by the caller via `updates_since_snapshot[name]` and
+        is reset to `0` whenever a snapshot fires.
+
+    Version-format-independent: works for `int`, `float`, and `str`
+    versioning schemes alike.
    """
-    return {
-        name
-        for name, ch in channels.items()
-        if isinstance(ch, DeltaChannel)
-        and ch.is_available()
-        and counts.get(name, 0) >= ch.snapshot_frequency
-    }
+    if force:
+        return True
+    return updates_since_snapshot.get(name, 0) >= ch.snapshot_frequency


 def create_checkpoint(
@@ -61,19 +66,34 @@ def create_checkpoint(
    id: str | None = None,
    updated_channels: set[str] | None = None,
    get_next_version: GetNextVersion | None = None,
-    channels_to_snapshot: set[str] | None = None,
+    force_delta_snapshot: bool = False,
+    updates_since_snapshot: Mapping[str, int] | None = None,
+    new_updates_since_snapshot: dict[str, int] | None = None,
 ) -> Checkpoint:
-    """Build a new Checkpoint from the previous one and live channel state.
+    """Create a checkpoint for the given channels.

-    For each name in `channels_to_snapshot`, a `_DeltaSnapshot(value)` blob
-    is written into `channel_values[k]`. Other delta channels are omitted
-    from `channel_values` — the ancestor walk reconstructs their state
-    from `checkpoint_writes`. Callers compute the set via
-    `delta_channels_to_snapshot(channels, counts)`; defaults to empty
-    (no snapshots) when not provided.
+    For each `DeltaChannel`, a `_DeltaSnapshot(value)` blob is written into
+    `channel_values[k]` when this channel has accumulated at least
+    `snapshot_frequency` updates since its last snapshot (counter supplied
+    via `updates_since_snapshot`). Otherwise the channel is omitted from
+    `channel_values`; its `channel_versions` entry still bumps so that the
+    saver tracks the channel and the ancestor walk can replay writes.
+
+    Snapshots are eager: even if the channel had no write this step, a
+    version bump is forced (via `get_next_version`) so `put()` includes
+    the channel in `new_versions` and stores the blob.
+
+    `force_delta_snapshot` ignores the cadence and always snapshots —
+    used by `durability="exit"` where intermediate writes are not stored
+    as ancestor `checkpoint_writes`.
+
+    If `new_updates_since_snapshot` is provided, the function resets the
+    counter to `0` for any channel that snapshotted this step. Counters
+    for channels that did not snapshot are left untouched (the caller is
+    responsible for incrementing them based on `updated_channels`).
    """
    ts = datetime.now(timezone.utc).isoformat()
-    channels_to_snapshot = channels_to_snapshot or set()
+    counts = updates_since_snapshot or {}
    if channels is None:
        values = checkpoint["channel_values"]
        channel_versions = checkpoint["channel_versions"]
@@ -84,23 +104,25 @@ def create_checkpoint(
            if k not in channel_versions:
                continue
            ch = channels[k]
-            if k in channels_to_snapshot:
-                # In exit mode, the snapshot decision is deferred to exit
-                # time (intermediate steps have do_checkpoint=False). The
-                # channel's count may have reached snapshot_frequency over
-                # several supersteps, but the LAST superstep may not have
-                # written to this channel. In that case apply_writes()
-                # (in _algo.py) didn't bump this channel's version, so
-                # saver.put() wouldn't include it in new_versions and
-                # the snapshot blob would be silently dropped. The manual
-                # bump below closes the gap. In sync/async durability this
-                # branch is effectively dead code (the step that pushes
-                # the count to freq always writes the channel).
+            if (
+                isinstance(ch, DeltaChannel)
+                and ch.is_available()
+                and _should_snapshot_delta(
+                    k,
+                    ch,
+                    counts,
+                    force=force_delta_snapshot,
+                )
+            ):
+                # Eager snapshot: bump version if not already written this step
+                # so put() includes this channel in new_versions and stores blob.
                if get_next_version is not None and (
                    updated_channels is None or k not in updated_channels
                ):
                    channel_versions[k] = get_next_version(channel_versions[k], None)
                values[k] = _DeltaSnapshot(ch.get())
+                if new_updates_since_snapshot is not None:
+                    new_updates_since_snapshot[k] = 0
            else:
                v = ch.checkpoint()
                if v is not MISSING:
@@ -100,7 +100,6 @@ from langgraph.pregel._checkpoint import (
    channels_from_checkpoint,
    copy_checkpoint,
    create_checkpoint,
-    delta_channels_to_snapshot,
    empty_checkpoint,
 )
 from langgraph.pregel._executor import (
@@ -195,40 +194,8 @@ class PregelLoop:
    _migrate_checkpoint: Callable[[Checkpoint], None] | None
    submit: Submit
    channels: Mapping[str, BaseChannel]
-    # Futures from `checkpointer.put_writes` calls that produced delta-channel
-    # writes. `_checkpointer_put_after_previous` drains this list (swap to a
-    # local `futs` then reset to `[]` and wait/gather) before putting the
-    # next checkpoint, so a checkpoint never becomes durable before the
-    # writes that produced it. Initialised to `[]` in both sync and async
-    # `__enter__`; stays `None` only when no checkpointer.
+    # Only set on AsyncPregelLoop; sync loops keep this as None.
    _delta_write_futs: list[Any] | None = None
-
-    # Exit-mode accumulator: every delta-channel write produced during this
-    # run (input writes from `_first` + per-superstep writes captured in
-    # `after_tick`). At exit, `_put_exit_delta_writes` filters out channels
-    # that will snapshot, then persists the rest under an anchor parent.
-    # `None` when not in exit mode (so the capture sites are no-ops).
-    # Each tuple is `(step, task_id, channel, value)` — `step` drives the
-    # synthetic step-prefixed task_id used to preserve chronological order
-    # under the saver's `ORDER BY task_id, idx` sorting.
-    _exit_delta_writes: list[tuple[int, str, str, Any]] | None = None
-
-    # The checkpoint_config that points at the parent loaded at `__enter__`
-    # (or the synthetic-empty checkpoint, on first run). We capture it
-    # eagerly because every `_put_checkpoint` advances `self.checkpoint_config`
-    # to the newly-saved checkpoint's id — by exit time the original parent
-    # config would otherwise be lost. `_put_exit_delta_writes` uses this:
-    # on resumed runs as the anchor for exit delta writes; on first runs
-    # to derive the lazy stub's config (its `checkpoint_id` is the
-    # synthetic-empty id we want the stub persisted under).
-    _initial_checkpoint_config: RunnableConfig
-
-    # True iff the saver actually returned a tuple at `__enter__`. False
-    # on the first-ever run for a thread (no parent persisted yet).
-    # `_put_exit_delta_writes` uses this to decide between anchoring on
-    # the existing parent (True) or creating a lazy stub (False).
-    _has_persisted_parent: bool = False
-
    managed: ManagedValueMapping
    checkpoint: Checkpoint
    checkpoint_id_saved: str
@@ -670,11 +637,6 @@ class PregelLoop:
            self._emit(
                "values", map_output_values, self.output_keys, writes, self.channels
            )
-        # capture delta-channel writes for exit-mode accumulator before clearing
-        if self._exit_delta_writes is not None:
-            for tid, ch, v in self.checkpoint_pending_writes:
-                if isinstance(self.specs.get(ch), DeltaChannel):
-                    self._exit_delta_writes.append((self.step, tid, ch, v))
        # clear pending writes
        self.checkpoint_pending_writes.clear()
        # only replay (re-execute) done tasks on the first tick
@@ -892,27 +854,6 @@ class PregelLoop:
                self.checkpointer_get_next_version,
                self.trigger_to_nodes,
            )
-            # Input writes go through `apply_writes` directly (above) — they
-            # never enter `checkpoint_pending_writes`, so the after_tick
-            # capture site does not see them. In exit mode, capture them
-            # here so `_exit_delta_writes` includes the input's delta writes
-            # alongside per-superstep writes; otherwise the input would be
-            # lost on read (it's not in final_checkpoint.channel_values for
-            # sub-freq channels, and walks ignore target.pending_writes).
-            if self._exit_delta_writes is not None:
-                for c, v in input_writes:
-                    if isinstance(self.specs.get(c), DeltaChannel):
-                        self._exit_delta_writes.append((self.step, NULL_TASK_ID, c, v))
-            # Persist delta-channel input writes so sub-freq inputs are
-            # recoverable via ancestor walk (mirrors the Command input path).
-            if self.durability != "exit":
-                delta_input = [
-                    (c, v)
-                    for c, v in input_writes
-                    if isinstance(self.specs.get(c), DeltaChannel)
-                ]
-                if delta_input:
-                    self.put_writes(NULL_TASK_ID, delta_input)
            # save input checkpoint
            self.updated_channels = updated_channels
            self._put_checkpoint({"source": "input"})
@@ -964,60 +905,36 @@ class PregelLoop:
        return updated_channels

    def _put_checkpoint(self, metadata: CheckpointMetadata) -> None:
-        # `is` (object identity) — not `==`. Three of four call sites pass a
-        # fresh dict ({"source":"input"|"loop"|"fork"}); only
-        # `_suppress_interrupt`(will rename to _on_loop_exit soon)
-        # at exit reuses the existing `self.checkpoint_metadata` instance. So
-        # `metadata is self.checkpoint_metadata` is True only on the exit call,
-        # which is what we use to gate exit-only behaviour (skip count-bump,
-        # don't replace metadata). Could be replaced by an explicit
-        # `exiting: bool = False` parameter; left as-is to match the existing
-        # idiom in this file.
-        # TODO: replace with an explicit `exiting: bool = False` parameter.
+        # assign step and parents
        exiting = metadata is self.checkpoint_metadata
        if exiting and self.checkpoint["id"] == self.checkpoint_id_saved:
            # checkpoint already saved
            return
-        # Per-delta-channel update bookkeeping.
-        #
-        # `_put_checkpoint` is called once per superstep with a fresh
-        # metadata dict (source="input"|"loop"|"fork") — those are the
-        # intermediate calls that bump the count by +1 for each delta
-        # channel touched that step. In exit mode,
-        # `_suppress_interrupt`(will rename to _on_loop_exit soon)
-        # additionally calls `_put_checkpoint(self.checkpoint_metadata)` AT
-        # EXIT to commit the final checkpoint — this runs *after* the last
-        # intermediate call already counted the last superstep. So the
-        # exit call must NOT bump again or it would double-count the last
-        # superstep. (Sync/async durability does not call `_put_checkpoint`
-        # at exit, so the issue only surfaces in exit mode. force_delta_snapshot
-        # used to mask this latent bug by resetting every count to 0.)
+        # Carry per-delta-channel update bookkeeping forward across
+        # supersteps. Capture from the OLD metadata before potentially
+        # replacing it with a fresh dict that wouldn't contain it. Then
+        # increment for any delta channel updated this step (so the count
+        # reflects "supersteps that wrote to this channel since last
+        # snapshot"). create_checkpoint will reset entries to 0 for any
+        # channel that fires a snapshot this step.
+        prev_counts = dict(
+            self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
+        )
+        new_counts = dict(prev_counts)
+        if self.updated_channels:
+            for ch_name in self.updated_channels:
+                ch_obj = self.channels.get(ch_name)
+                if isinstance(ch_obj, DeltaChannel):
+                    new_counts[ch_name] = new_counts.get(ch_name, 0) + 1
        if not exiting:
-            prev_counts = dict(
-                self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
-            )
-            new_counts = dict(prev_counts)
-            if self.updated_channels:
-                for ch_name in self.updated_channels:
-                    if isinstance(self.channels.get(ch_name), DeltaChannel):
-                        new_counts[ch_name] = new_counts.get(ch_name, 0) + 1
            metadata["step"] = self.step
            metadata["parents"] = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})
            self.checkpoint_metadata = metadata
-        else:
-            new_counts = dict(
-                self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
-            )
        # do checkpoint?
        do_checkpoint = self._checkpointer_put_after_previous is not None and (
            exiting or self.durability != "exit"
        )
        # create new checkpoint
-        channels_to_snapshot = (
-            delta_channels_to_snapshot(self.channels, new_counts)
-            if do_checkpoint
-            else set()
-        )
        self.checkpoint = create_checkpoint(
            self.checkpoint,
            self.channels if do_checkpoint else None,
@@ -1027,10 +944,10 @@ class PregelLoop:
            get_next_version=self.checkpointer_get_next_version
            if do_checkpoint
            else None,
-            channels_to_snapshot=channels_to_snapshot,
+            force_delta_snapshot=exiting and self.durability == "exit",
+            updates_since_snapshot=new_counts,
+            new_updates_since_snapshot=new_counts,
        )
-        for k in channels_to_snapshot:
-            new_counts[k] = 0
        if new_counts:
            self.checkpoint_metadata["delta_updates_since_snapshot"] = new_counts
        elif "delta_updates_since_snapshot" in self.checkpoint_metadata:
@@ -1093,97 +1010,6 @@ class PregelLoop:
            # increment step
            self.step += 1

-    def _put_exit_delta_writes(self) -> None:
-        """Stage stub + accumulated delta writes so final_checkpoint's put
-        waits on them (visibility invariant: both must be durable before
-        final_checkpoint becomes visible to readers).
-
-        Stub is created lazily — only when no persisted parent exists AND at
-        least one delta channel has writes that won't be snapshotted.
-        """
-        if (
-            not self._exit_delta_writes
-            or self.checkpointer is None
-            or self._checkpointer_put_after_previous is None
-            or self.checkpointer_put_writes is None
-        ):
-            return
-
-        counts = self.checkpoint_metadata.get("delta_updates_since_snapshot", {}) or {}
-        channels_to_snapshot = delta_channels_to_snapshot(self.channels, counts)
-
-        pending = [
-            (step, tid, ch, v)
-            for (step, tid, ch, v) in self._exit_delta_writes
-            if ch not in channels_to_snapshot
-        ]
-        if not pending:
-            return
-
-        if self._has_persisted_parent:
-            # _initial_checkpoint_config's checkpoint_id is the saved parent's
-            # id (saver returned a real tuple at __enter__).
-            anchor_config = self._initial_checkpoint_config
-        else:
-            stub_cp = empty_checkpoint()
-            stub_cp["id"] = self.checkpoint_id_saved
-            stub_cp["ts"] = datetime.now(timezone.utc).isoformat()
-            # Stub has no parent (checkpoint_id=None in config).
-            stub_put_config = patch_configurable(
-                self._initial_checkpoint_config,
-                {CONFIG_KEY_CHECKPOINT_ID: None},
-            )
-            # Anchor config for put_writes: checkpoint_id = stub's id.
-            anchor_config = patch_configurable(
-                self._initial_checkpoint_config,
-                {CONFIG_KEY_CHECKPOINT_ID: stub_cp["id"]},
-            )
-            self._put_checkpoint_fut = self.submit(
-                self._checkpointer_put_after_previous,
-                getattr(self, "_put_checkpoint_fut", None),
-                stub_put_config,
-                stub_cp,
-                {"step": -2},
-                {},
-            )
-            # Set checkpoint_config so final_checkpoint's _put_checkpoint
-            # sees the stub as its parent.
-            self.checkpoint_config = anchor_config
-
-        # Step-prefixed synthetic task_id preserves chronological superstep
-        # order under the saver's ORDER BY task_id, idx sorting.
-        grouped: dict[tuple[int, str], list[tuple[str, Any]]] = {}
-        for step, tid, ch, v in pending:
-            grouped.setdefault((step, tid), []).append((ch, v))
-        anchor_write_config = patch_configurable(
-            anchor_config,
-            {
-                CONFIG_KEY_CHECKPOINT_NS: self.config[CONF].get(
-                    CONFIG_KEY_CHECKPOINT_NS, ""
-                ),
-                CONFIG_KEY_CHECKPOINT_ID: anchor_config[CONF][CONFIG_KEY_CHECKPOINT_ID],
-            },
-        )
-        for (step, tid), entries in grouped.items():
-            synth_tid = f"{step:08d}-{tid}"
-            if self.checkpointer_put_writes_accepts_task_path:
-                fut = self.submit(
-                    self.checkpointer_put_writes,
-                    anchor_write_config,
-                    entries,
-                    synth_tid,
-                    "",
-                )
-            else:
-                fut = self.submit(
-                    self.checkpointer_put_writes,
-                    anchor_write_config,
-                    entries,
-                    synth_tid,
-                )
-            if self._delta_write_futs is not None:
-                self._delta_write_futs.append(fut)
-
    def _suppress_interrupt(
        self,
        exc_type: type[BaseException] | None,
@@ -1199,7 +1025,6 @@ class PregelLoop:
            # or a nested graph with checkpointer=True
            or all(NS_END not in part for part in self.checkpoint_ns)
        ):
-            self._put_exit_delta_writes()
            self._put_checkpoint(self.checkpoint_metadata)
            self._put_pending_writes()
        # suppress interrupt
@@ -1405,9 +1230,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
-        if self._delta_write_futs:
-            futs, self._delta_write_futs = self._delta_write_futs, []
-            concurrent.futures.wait(futs)
        try:
            if prev is not None:
                prev.result()
@@ -1525,10 +1347,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
            # graph/thread. Returns None on first invocation.
            saved = self.checkpointer.get_tuple(self.checkpoint_config)

-        # Capture before the synthetic-empty fallback below overwrites `saved`.
-        # `_put_exit_delta_writes` uses this on first run (no persisted parent)
-        # to lazy-create a stub instead of anchoring delta writes on a parent.
-        self._has_persisted_parent = saved is not None
        if saved is None:
            saved = CheckpointTuple(
                self.checkpoint_config, empty_checkpoint(), {"step": -2}, None, []
@@ -1544,7 +1362,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
                **saved.config.get(CONF, {}),
            },
        }
-        self._initial_checkpoint_config = self.checkpoint_config
        self.prev_checkpoint_config = saved.parent_config
        self.checkpoint_id_saved = saved.checkpoint["id"]
        self.checkpoint = saved.checkpoint
@@ -1554,10 +1371,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
            if saved.pending_writes is not None
            else []
        )
-        self._delta_write_futs = []
-        self._exit_delta_writes = (
-            [] if self.durability == "exit" and self.checkpointer is not None else None
-        )
        self.submit = self.stack.enter_context(BackgroundExecutor(self.config))
        self.channels, self.managed = channels_from_checkpoint(
            self.specs,
@@ -1783,10 +1596,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
            # graph/thread. Returns None on first invocation.
            saved = await self.checkpointer.aget_tuple(self.checkpoint_config)

-        # Capture before the synthetic-empty fallback below overwrites `saved`.
-        # `_put_exit_delta_writes` uses this on first run (no persisted parent)
-        # to lazy-create a stub instead of anchoring delta writes on a parent.
-        self._has_persisted_parent = saved is not None
        if saved is None:
            saved = CheckpointTuple(
                self.checkpoint_config, empty_checkpoint(), {"step": -2}, None, []
@@ -1802,7 +1611,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
                **saved.config.get(CONF, {}),
            },
        }
-        self._initial_checkpoint_config = self.checkpoint_config
        self.prev_checkpoint_config = saved.parent_config
        self.checkpoint_id_saved = saved.checkpoint["id"]
        self.checkpoint = saved.checkpoint
@@ -1813,9 +1621,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
            else []
        )
        self._delta_write_futs = []
-        self._exit_delta_writes = (
-            [] if self.durability == "exit" and self.checkpointer is not None else None
-        )
        self.submit = await self.stack.enter_async_context(
            AsyncBackgroundExecutor(self.config)
        )
@@ -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",
@@ -12,7 +12,7 @@ from langgraph.stream._mux import StreamMux
 from langgraph.stream._types import ProtocolEvent

 if TYPE_CHECKING:
-    from langgraph.stream.transformers import LifecycleEvent
+    from langgraph.stream.transformers import SubgraphStatus


 def _drive_until_done(pump: Callable[[], bool]) -> None:
@@ -523,8 +523,8 @@ class _SubgraphRunStreamMixin:

    path: tuple[str, ...]
    graph_name: str | None
-    parent_task_id: str | None
-    status: LifecycleEvent
+    trigger_call_id: str | None
+    status: SubgraphStatus
    error: str | None
    _seen_terminal: bool

@@ -538,7 +538,7 @@ class SubgraphRunStream(GraphRunStream, _SubgraphRunStreamMixin):
        *,
        path: tuple[str, ...],
        graph_name: str | None = None,
-        parent_task_id: str | None = None,
+        trigger_call_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.parent_task_id = parent_task_id
+        self.trigger_call_id = trigger_call_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,
-        parent_task_id: str | None = None,
+        trigger_call_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.parent_task_id = parent_task_id
+        self.trigger_call_id = trigger_call_id
        self.status = "started"
        self.error = None
        self._seen_terminal = False
@@ -327,31 +327,11 @@ class MessagesTransformer(StreamTransformer):
        self._by_run.clear()


-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.
-"""
+SubgraphStatus = Literal["started", "completed", "failed", "interrupted", "drained"]


 def _parse_ns_segment(segment: str) -> tuple[str, str | None]:
-    """Split a namespace segment into `(graph_name, parent_task_id)`.
+    """Split a namespace segment into `(graph_name, trigger_call_id)`.

    Segments are formatted `node_name:task_id` by `prepare_next_tasks`.
    Returns `(segment, None)` if no `:` is present.
@@ -360,36 +340,6 @@ 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.

@@ -399,54 +349,11 @@ class LifecyclePayload(TypedDict, total=False):
    `run.lifecycle`.
    """

-    event: LifecycleEvent
-    """State transition. See `LifecycleEvent` for per-value semantics."""
+    event: SubgraphStatus
    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]
-    """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`.
-    """
+    trigger_call_id: NotRequired[str]
    error: NotRequired[str]
-    """Error string. Set on `failed` events; absent otherwise."""


 class _TasksLifecycleBase(StreamTransformer):
@@ -464,13 +371,13 @@ class _TasksLifecycleBase(StreamTransformer):

    - `_should_track(ns)` — scope filter (e.g. multi-depth vs
      direct-children-only).
-    - `_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.
+    - `_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.

    Tasks events are suppressed from the main event log (`process`
    returns False) — they're folded into whichever projection the
@@ -483,15 +390,9 @@ 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 dispatching task whose
+        # Maps tracked namespace -> task_id of the parent 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) ---

@@ -503,33 +404,19 @@ class _TasksLifecycleBase(StreamTransformer):
        self,
        ns: tuple[str, ...],
        graph_name: str | None,
-        parent_task_id: str | None,
-        tool_call_id: str | None = None,
+        trigger_call_id: str | None,
    ) -> None:
-        """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.
-        """
+        """Fired once per discovered namespace (first observed task event)."""
        raise NotImplementedError

    def _on_terminal(
        self,
        ns: tuple[str, ...],
-        status: LifecycleEvent,
+        status: SubgraphStatus,
        error: str | None,
-        parent_task_id: str,
    ) -> None:
-        """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`.
+        """Fired once per tracked namespace when its parent's result arrives,
+        or via finalize/fail safety-net sweeps.
        """
        raise NotImplementedError

@@ -543,96 +430,56 @@ class _TasksLifecycleBase(StreamTransformer):
        if "result" in data:
            self._handle_task_result(ns, data)
        else:
-            self._handle_task_start(ns, data)
+            self._handle_task_start(ns)
        # 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, ...], 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)
+    def _handle_task_start(self, ns: tuple[str, ...]) -> None:
        if not self._should_track(ns) or ns in self._seen:
            return
        self._seen.add(ns)
-        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
+        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

    def _pop_terminal_transitions(
        self, ns: tuple[str, ...], data: dict[str, Any]
-    ) -> 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`.
-        """
+    ) -> list[tuple[tuple[str, ...], SubgraphStatus, str | None]]:
+        """Return and remove tracked children closed by this task result."""
        result_id = data.get("id")
        if not result_id:
            return []
-        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:
+        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:
                continue
            status, error = _terminal_from_result(data)
-            transitions.append((child_ns, status, error, dispatching_task_id))
+            transitions.append((child_ns, status, error))
            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, parent_task_id in self._pop_terminal_transitions(
-            ns, data
-        ):
-            self._on_terminal(child_ns, status, error, parent_task_id)
+        for child_ns, status, error in self._pop_terminal_transitions(ns, data):
+            self._on_terminal(child_ns, status, error)

    def finalize(self) -> None:
        """Emit `completed` for any tracked namespace still open at run end."""
-        for ns, parent_task_id in list(self._open.items()):
-            self._on_terminal(ns, "completed", None, parent_task_id)
+        for ns in list(self._open):
+            self._on_terminal(ns, "completed", None)
        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, parent_task_id in list(self._open.items()):
-            self._on_terminal(ns, status, error_str, parent_task_id)
+        for ns in list(self._open):
+            self._on_terminal(ns, status, error_str)
        self._open.clear()


-def _status_from_exception(err: BaseException) -> tuple[LifecycleEvent, str | None]:
+def _status_from_exception(err: BaseException) -> tuple[SubgraphStatus, str | None]:
    """Map a run exception to a subgraph terminal status and error string."""
    if isinstance(err, GraphDrained):
        return "drained", None
@@ -643,7 +490,7 @@ def _status_from_exception(err: BaseException) -> tuple[LifecycleEvent, str | No

 def _terminal_from_result(
    payload: dict[str, Any],
-) -> tuple[LifecycleEvent, str | None]:
+) -> tuple[SubgraphStatus, str | None]:
    """Map a `TaskResultPayload` to a `(status, error)` pair.

    Order matters: a result with both `error` and `interrupts` prefers
@@ -692,37 +539,26 @@ class LifecycleTransformer(_TasksLifecycleBase):
        self,
        ns: tuple[str, ...],
        graph_name: str | None,
-        parent_task_id: str | None,
-        tool_call_id: str | None = None,
+        trigger_call_id: str | None,
    ) -> None:
-        if parent_task_id is None:
-            # Without a task id we can't correlate a dispatching-task-result
+        if trigger_call_id is None:
+            # Without a task id we can't correlate a parent-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),
-            "parent_task_id": parent_task_id,
-        }
+        payload: LifecyclePayload = {"event": "started", "namespace": list(ns)}
        if graph_name:
            payload["graph_name"] = graph_name
-        if tool_call_id is not None:
-            payload["metadata"] = {"type": "tool_call", "tool_call_id": tool_call_id}
+        payload["trigger_call_id"] = trigger_call_id
        self._channel.push(payload)

    def _on_terminal(
        self,
        ns: tuple[str, ...],
-        status: LifecycleEvent,
+        status: SubgraphStatus,
        error: str | None,
-        parent_task_id: str,
    ) -> None:
-        payload: LifecyclePayload = {
-            "event": status,
-            "namespace": list(ns),
-            "parent_task_id": parent_task_id,
-        }
+        payload: LifecyclePayload = {"event": status, "namespace": list(ns)}
        if error is not None:
            payload["error"] = error
        self._channel.push(payload)
@@ -775,8 +611,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
        self,
        ns: tuple[str, ...],
        graph_name: str | None,
-        parent_task_id: str | None,
-        tool_call_id: str | None = None,  # noqa: ARG002
+        trigger_call_id: str | None,
    ) -> None:
        if self._mux is None:
            return
@@ -789,7 +624,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
            mux=child_mux,
            path=ns,
            graph_name=graph_name,
-            parent_task_id=parent_task_id,
+            trigger_call_id=trigger_call_id,
        )
        self._handles[ns] = handle
        self._log.push(handle)
@@ -797,9 +632,8 @@ class SubgraphTransformer(_TasksLifecycleBase):
    def _on_terminal(
        self,
        ns: tuple[str, ...],
-        status: LifecycleEvent,
+        status: SubgraphStatus,
        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):
@@ -809,9 +643,8 @@ class SubgraphTransformer(_TasksLifecycleBase):
    async def _aon_terminal(
        self,
        ns: tuple[str, ...],
-        status: LifecycleEvent,
+        status: SubgraphStatus,
        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):
@@ -821,7 +654,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
    def _mark_terminal(
        self,
        handle: SubgraphRunStream | AsyncSubgraphRunStream,
-        status: LifecycleEvent,
+        status: SubgraphStatus,
        error: str | None,
    ) -> bool:
        """Mark a handle terminal once. Returns True on first transition."""
@@ -836,7 +669,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
    def _close_or_fail_handle(
        self,
        handle: SubgraphRunStream | AsyncSubgraphRunStream,
-        status: LifecycleEvent,
+        status: SubgraphStatus,
        error: str | None,
    ) -> None:
        if handle._mux is None or handle._mux._events._closed:
@@ -849,7 +682,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
    async def _aclose_or_fail_handle(
        self,
        handle: SubgraphRunStream | AsyncSubgraphRunStream,
-        status: LifecycleEvent,
+        status: SubgraphStatus,
        error: str | None,
    ) -> None:
        if handle._mux is None or handle._mux._events._closed:
@@ -888,15 +721,10 @@ class SubgraphTransformer(_TasksLifecycleBase):
            ns = tuple(event["params"]["namespace"])
            data = event["params"]["data"]
            if "result" in data:
-                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)
+                for child_ns, status, error in self._pop_terminal_transitions(ns, data):
+                    await self._aon_terminal(child_ns, status, error)
            else:
-                self._handle_task_start(ns, data)
+                self._handle_task_start(ns)
            keep = False
        else:
            keep = True
@@ -908,9 +736,9 @@ class SubgraphTransformer(_TasksLifecycleBase):

    def _complete_open_handles(self) -> BaseException | None:
        first_error: BaseException | None = None
-        for ns, parent_task_id in list(self._open.items()):
+        for ns in list(self._open):
            try:
-                self._on_terminal(ns, "completed", None, parent_task_id)
+                self._on_terminal(ns, "completed", None)
            except BaseException as e:
                if first_error is None:
                    first_error = e
@@ -926,9 +754,9 @@ class SubgraphTransformer(_TasksLifecycleBase):

    async def _acomplete_open_handles(self) -> BaseException | None:
        first_error: BaseException | None = None
-        for ns, parent_task_id in list(self._open.items()):
+        for ns in list(self._open):
            try:
-                await self._aon_terminal(ns, "completed", None, parent_task_id)
+                await self._aon_terminal(ns, "completed", None)
            except BaseException as e:
                if first_error is None:
                    first_error = e
@@ -3,7 +3,12 @@ from collections.abc import Sequence
 from typing import Annotated

 import pytest
-from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
+from langchain_core.messages import (
+    AIMessage,
+    AIMessageChunk,
+    HumanMessage,
+    RemoveMessage,
+)
 from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import NotRequired, TypedDict
@@ -16,7 +21,7 @@ from langgraph.channels.topic import Topic
 from langgraph.channels.untracked_value import UntrackedValue
 from langgraph.errors import EmptyChannelError, InvalidUpdateError
 from langgraph.graph import START, StateGraph
-from langgraph.graph.message import _messages_delta_reducer
+from langgraph.graph.message import REMOVE_ALL_MESSAGES, _messages_delta_reducer
 from langgraph.graph.state import _get_channel
 from langgraph.types import Overwrite

@@ -284,6 +289,61 @@ def test_messages_delta_reducer_tuple_write_is_one_message() -> None:
    assert result[0].content == "hi"


+def test_messages_delta_reducer_assigns_uuid_to_id_less_messages() -> None:
+    """Messages without IDs get UUIDs assigned, matching add_messages behavior.
+
+    Without UUID assignment, RemoveMessage tombstoning fails on messages that
+    were created without explicit IDs.
+    """
+    m1 = HumanMessage(content="hi")
+    m2 = AIMessage(content="hello")
+    assert m1.id is None
+    assert m2.id is None
+
+    result = _messages_delta_reducer([], [[m1, m2]])
+    assert len(result) == 2
+    assert result[0].id is not None
+    assert result[1].id is not None
+
+    # RemoveMessage tombstoning must work on the now-assigned IDs.
+    result2 = _messages_delta_reducer(result, [RemoveMessage(id=result[1].id)])
+    assert len(result2) == 1
+    assert result2[0].content == "hi"
+
+
+def test_messages_delta_reducer_remove_all_messages() -> None:
+    """REMOVE_ALL_MESSAGES sentinel clears all state and preceding writes."""
+    state = [HumanMessage(content="old", id="h1"), AIMessage(content="prior", id="a1")]
+
+    # Sentinel mid-batch: everything before it (including state) is discarded.
+    result = _messages_delta_reducer(
+        state,
+        [
+            [
+                RemoveMessage(id=REMOVE_ALL_MESSAGES),
+                HumanMessage(content="fresh", id="h2"),
+            ]
+        ],
+    )
+    assert len(result) == 1
+    assert result[0].content == "fresh"
+
+    # Batching-invariant: split across two calls must equal one combined call.
+    step1 = _messages_delta_reducer(state, [[RemoveMessage(id=REMOVE_ALL_MESSAGES)]])
+    step2 = _messages_delta_reducer(step1, [[HumanMessage(content="fresh", id="h2")]])
+    assert step2 == result
+
+
+def test_messages_delta_reducer_coerces_message_chunks() -> None:
+    """BaseMessageChunk writes are coerced to full messages."""
+    chunk = AIMessageChunk(content="hello", id="a1")
+    result = _messages_delta_reducer([], [[chunk]])
+    assert len(result) == 1
+    assert not isinstance(result[0], AIMessageChunk)
+    assert result[0].content == "hello"
+    assert result[0].id == "a1"
+
+
 def test_delta_channel_checkpoint_returns_missing() -> None:
    """checkpoint() always returns MISSING regardless of state.

@@ -1,365 +0,0 @@
-"""Tests for exit-mode delta channel persistence redesign.
-
-Validates that `durability="exit"` correctly persists delta-channel writes
-using count-based snapshot decisions (rather than force-snapshotting every
-channel), lazy stub creation when no parent exists, and proper read-path
-reconstruction via ancestor walks.
-"""
-
-from typing import Annotated, Any
-
-import pytest
-from langchain_core.messages import AIMessage, HumanMessage
-from langgraph.checkpoint.memory import InMemorySaver
-from langgraph.checkpoint.serde.types import _DeltaSnapshot
-from typing_extensions import TypedDict
-
-from langgraph.channels.delta import DeltaChannel
-from langgraph.graph import START, StateGraph
-from langgraph.graph.message import _messages_delta_reducer
-
-pytestmark = pytest.mark.anyio
-
-
-def _build_graph(
-    checkpointer: InMemorySaver,
-    *,
-    freq: int = 1000,
-) -> Any:
-    channel = DeltaChannel(_messages_delta_reducer, snapshot_frequency=freq)
-    # Functional TypedDict form: class form can't reference `channel` (a
-    # local variable) inside Annotated due to forward-ref evaluation rules.
-    State = TypedDict("State", {"messages": Annotated[list, channel]})  # type: ignore[call-overload]  # noqa: UP013
-
-    def respond(state: dict) -> dict:
-        i = len(state["messages"])
-        return {"messages": [AIMessage(content=f"reply-{i}", id=f"ai{i}")]}
-
-    builder = StateGraph(State)
-    builder.add_node("respond", respond)
-    builder.add_edge(START, "respond")
-    return builder.compile(checkpointer=checkpointer)
-
-
-# ---------------------------------------------------------------------------
-# 8a. Write-path / structural tests
-# ---------------------------------------------------------------------------
-
-
-async def test_exit_first_run_no_delta_writes() -> None:
-    """Graph with delta channel invoked with input that doesn't touch it.
-    Only one checkpoint row, no stub."""
-    State = TypedDict(  # noqa: UP013
-        "State",
-        {
-            "messages": Annotated[list, DeltaChannel(_messages_delta_reducer)],
-            "value": str,
-        },
-    )  # type: ignore[call-overload]
-
-    def noop(state: dict) -> dict:
-        return {"value": "done"}
-
-    saver = InMemorySaver()
-    builder = StateGraph(State)
-    builder.add_node("noop", noop)
-    builder.add_edge(START, "noop")
-    graph = builder.compile(checkpointer=saver)
-    config = {"configurable": {"thread_id": "no-delta-writes"}}
-
-    graph.invoke({"value": "start"}, config, durability="exit")
-
-    checkpoints = list(saver.list(config))
-    assert len(checkpoints) == 1
-    stubs = [t for t in checkpoints if t.metadata.get("step") == -2]
-    assert len(stubs) == 0
-
-
-async def test_exit_first_run_all_snapshot() -> None:
-    """snapshot_frequency=1 forces every channel to snapshot.
-    No stub needed; final_checkpoint has _DeltaSnapshot."""
-    saver = InMemorySaver()
-    graph = _build_graph(saver, freq=1)
-    config = {"configurable": {"thread_id": "all-snapshot"}}
-
-    result = graph.invoke(
-        {"messages": [HumanMessage(content="hi", id="h1")]},
-        config,
-        durability="exit",
-    )
-    assert len(result["messages"]) == 2
-
-    checkpoints = list(saver.list(config))
-    stubs = [t for t in checkpoints if t.metadata.get("step") == -2]
-    assert len(stubs) == 0
-
-    head = saver.get_tuple(config)
-    assert head is not None
-    assert isinstance(head.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
-
-    state = graph.get_state(config)
-    assert [m.content for m in state.values["messages"]] == ["hi", "reply-1"]
-
-
-async def test_exit_first_run_sub_freq_with_writes() -> None:
-    """First run with default snapshot_frequency (1000), writes below threshold.
-    A stub is created; writes are anchored under it; get_state reconstructs."""
-    saver = InMemorySaver()
-    graph = _build_graph(saver)
-    config = {"configurable": {"thread_id": "sub-freq-first"}}
-
-    result = graph.invoke(
-        {"messages": [HumanMessage(content="hello", id="h1")]},
-        config,
-        durability="exit",
-    )
-    assert [m.content for m in result["messages"]] == ["hello", "reply-1"]
-
-    checkpoints = list(saver.list(config))
-    stubs = [t for t in checkpoints if t.metadata.get("step") == -2]
-    assert len(stubs) == 1, f"Expected 1 stub, got {len(stubs)}"
-
-    head = saver.get_tuple(config)
-    assert head is not None
-    assert "messages" not in head.checkpoint["channel_values"]
-    assert "messages" in head.checkpoint["channel_versions"]
-
-    state = graph.get_state(config)
-    assert [m.content for m in state.values["messages"]] == ["hello", "reply-1"]
-
-
-async def test_exit_resumed_run_sub_freq() -> None:
-    """Two consecutive exit runs. Second run anchors on the first's
-    final_checkpoint (no new stub). Ordering preserved."""
-    saver = InMemorySaver()
-    graph = _build_graph(saver)
-    config = {"configurable": {"thread_id": "resumed-sub-freq"}}
-
-    graph.invoke(
-        {"messages": [HumanMessage(content="msg1", id="h1")]},
-        config,
-        durability="exit",
-    )
-
-    graph.invoke(
-        {"messages": [HumanMessage(content="msg2", id="h2")]},
-        config,
-        durability="exit",
-    )
-
-    checkpoints = list(saver.list(config))
-    stubs = [t for t in checkpoints if t.metadata.get("step") == -2]
-    assert len(stubs) == 1
-
-    state = graph.get_state(config)
-    contents = [m.content for m in state.values["messages"]]
-    assert len(contents) == 4
-    assert contents[0] == "msg1"
-    assert contents[2] == "msg2"
-    assert contents[0:4:2] == ["msg1", "msg2"]
-
-
-async def test_exit_count_parity_sync_vs_exit() -> None:
-    """Sync and exit durability produce the same delta_updates_since_snapshot
-    after an equivalent run."""
-    for durability in ("sync", "exit"):
-        saver = InMemorySaver()
-        graph = _build_graph(saver)
-        config = {"configurable": {"thread_id": f"parity-{durability}"}}
-
-        graph.invoke(
-            {"messages": [HumanMessage(content="hi", id="h1")]},
-            config,
-            durability=durability,
-        )
-
-        head = saver.get_tuple(config)
-        assert head is not None
-        counts = head.metadata.get("delta_updates_since_snapshot", {})
-        assert counts.get("messages") == 2, (
-            f"durability={durability}: expected count=2, got {counts}"
-        )
-
-
-async def test_exit_snapshot_fires_at_frequency() -> None:
-    """With snapshot_frequency=3, after 3 exit runs (each incrementing count
-    by 2: input + superstep), the 2nd run hits count=4>=3, triggering snapshot.
-    After that run, count resets to 0 and channel_values has _DeltaSnapshot."""
-    saver = InMemorySaver()
-    graph = _build_graph(saver, freq=3)
-    config = {"configurable": {"thread_id": "snapshot-at-freq"}}
-
-    graph.invoke(
-        {"messages": [HumanMessage(content="m1", id="h1")]},
-        config,
-        durability="exit",
-    )
-    head = saver.get_tuple(config)
-    assert head is not None
-    count1 = head.metadata.get("delta_updates_since_snapshot", {}).get("messages", 0)
-    assert count1 == 2
-
-    graph.invoke(
-        {"messages": [HumanMessage(content="m2", id="h2")]},
-        config,
-        durability="exit",
-    )
-    head = saver.get_tuple(config)
-    assert head is not None
-    count2 = head.metadata.get("delta_updates_since_snapshot", {}).get("messages", 0)
-    assert count2 == 0, f"Expected reset to 0 after snapshot, got {count2}"
-    assert isinstance(head.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
-
-
-async def test_exit_mixed_snapshot_and_non_snapshot() -> None:
-    """One delta channel at freq=1 (always snapshot) and one at freq=1000
-    (never snapshot within this test). Verify correct behavior for both."""
-
-    fast_ch = DeltaChannel(_messages_delta_reducer, snapshot_frequency=1)
-    slow_ch = DeltaChannel(_messages_delta_reducer, snapshot_frequency=1000)
-    State = TypedDict(  # noqa: UP013
-        "State",
-        {"fast": Annotated[list, fast_ch], "slow": Annotated[list, slow_ch]},
-    )  # type: ignore[call-overload]
-
-    def respond(state: dict) -> dict:
-        return {
-            "fast": [AIMessage(content="fast-reply", id="f1")],
-            "slow": [AIMessage(content="slow-reply", id="s1")],
-        }
-
-    saver = InMemorySaver()
-    builder = StateGraph(State)
-    builder.add_node("respond", respond)
-    builder.add_edge(START, "respond")
-    graph = builder.compile(checkpointer=saver)
-    config = {"configurable": {"thread_id": "mixed-freq"}}
-
-    graph.invoke(
-        {
-            "fast": [HumanMessage(content="fast-in", id="fi")],
-            "slow": [HumanMessage(content="slow-in", id="si")],
-        },
-        config,
-        durability="exit",
-    )
-
-    head = saver.get_tuple(config)
-    assert head is not None
-    assert isinstance(head.checkpoint["channel_values"].get("fast"), _DeltaSnapshot)
-    assert "slow" not in head.checkpoint["channel_values"]
-
-    state = graph.get_state(config)
-    assert [m.content for m in state.values["fast"]] == ["fast-in", "fast-reply"]
-    assert [m.content for m in state.values["slow"]] == ["slow-in", "slow-reply"]
-
-
-# ---------------------------------------------------------------------------
-# 8b. Read-path tests
-# ---------------------------------------------------------------------------
-
-
-async def test_exit_multi_run_replay_chain() -> None:
-    """K=4 consecutive exit runs, each adding a message. After each run,
-    get_state returns all messages in chronological order."""
-    saver = InMemorySaver()
-    graph = _build_graph(saver)
-    config = {"configurable": {"thread_id": "replay-chain"}}
-
-    for i in range(4):
-        graph.invoke(
-            {"messages": [HumanMessage(content=f"user-{i}", id=f"h{i}")]},
-            config,
-            durability="exit",
-        )
-
-        state = graph.get_state(config)
-        contents = [m.content for m in state.values["messages"]]
-        user_msgs = [c for c in contents if c.startswith("user-")]
-        assert user_msgs == [f"user-{j}" for j in range(i + 1)], (
-            f"After run {i}: user messages out of order: {user_msgs}"
-        )
-        assert len(contents) == (i + 1) * 2
-
-
-async def test_exit_metadata_round_trip() -> None:
-    """K=5 consecutive exit runs with snapshot_frequency=5. Verify metadata
-    delta_updates_since_snapshot increments correctly across runs."""
-    freq = 5
-    saver = InMemorySaver()
-    graph = _build_graph(saver, freq=freq)
-    config = {"configurable": {"thread_id": "metadata-rt"}}
-
-    for i in range(1, 6):
-        graph.invoke(
-            {"messages": [HumanMessage(content=f"m{i}", id=f"h{i}")]},
-            config,
-            durability="exit",
-        )
-        head = saver.get_tuple(config)
-        assert head is not None
-        count = head.metadata.get("delta_updates_since_snapshot", {}).get("messages", 0)
-        cumulative = i * 2
-        if cumulative >= freq:
-            assert count == 0 or count == cumulative % freq or count < freq, (
-                f"After run {i}: count={count} should have reset or be partial"
-            )
-        else:
-            assert count == cumulative, (
-                f"After run {i}: expected {cumulative}, got {count}"
-            )
-
-
-async def test_exit_mixed_durability_round_trip() -> None:
-    """Alternate sync and exit durability; verify counts stay monotonic
-    and state accumulates correctly."""
-    saver = InMemorySaver()
-    graph = _build_graph(saver)
-    config = {"configurable": {"thread_id": "mixed-durability"}}
-
-    for i, dur in enumerate(["sync", "exit", "sync", "exit"]):
-        graph.invoke(
-            {"messages": [HumanMessage(content=f"msg-{i}", id=f"h{i}")]},
-            config,
-            durability=dur,
-        )
-
-        state = graph.get_state(config)
-        contents = [m.content for m in state.values["messages"]]
-        user_msgs = [c for c in contents if c.startswith("msg-")]
-        assert user_msgs == [f"msg-{j}" for j in range(i + 1)], (
-            f"After run {i} (durability={dur}): {user_msgs}"
-        )
-        assert len(contents) == (i + 1) * 2
-
-
-async def test_exit_snapshot_then_tail_deltas() -> None:
-    """Run 1 forces snapshot (freq=1). Run 2 at freq=1000 adds more writes
-    that don't snapshot. Reading after run 2 must combine the snapshot seed
-    with the tail deltas."""
-    saver = InMemorySaver()
-
-    graph1 = _build_graph(saver, freq=1)
-    config = {"configurable": {"thread_id": "snapshot-then-tail"}}
-    graph1.invoke(
-        {"messages": [HumanMessage(content="seed-msg", id="h1")]},
-        config,
-        durability="exit",
-    )
-
-    head = saver.get_tuple(config)
-    assert head is not None
-    assert isinstance(head.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
-
-    graph2 = _build_graph(saver, freq=1000)
-    graph2.invoke(
-        {"messages": [HumanMessage(content="tail-msg", id="h2")]},
-        config,
-        durability="exit",
-    )
-
-    state = graph2.get_state(config)
-    contents = [m.content for m in state.values["messages"]]
-    assert "seed-msg" in contents
-    assert "tail-msg" in contents
-    assert contents.index("seed-msg") < contents.index("tail-msg")
@@ -1674,28 +1674,15 @@ 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(idle_timeout_s), name="heartbeat"
-    )
+    task = _make_task(HeartbeatProc(), timeout=_idle_timeout(0.2), name="heartbeat")
    task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = events.append
    assert await arun_with_retry(task, retry_policy=None) == "ok"

@@ -1704,13 +1691,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.25s; with the task running for
-    # ~400ms we expect 1–2 progress events (well below the 9 heartbeats).
+    # 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).
    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 == idle_timeout_s
+        assert ev.context.idle_timeout_secs == 0.2
        assert isinstance(ev.progress_at, datetime)


@@ -35,16 +35,8 @@ def _tasks_start(
    *,
    task_id: str,
    name: str,
-    input: Any = None,
 ) -> dict[str, Any]:
-    """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`.
-    """
+    """Build a `tasks` ProtocolEvent carrying a TaskPayload (start)."""
    return {
        "type": "event",
        "method": "tasks",
@@ -54,7 +46,7 @@ def _tasks_start(
            "data": {
                "id": task_id,
                "name": name,
-                "input": input,
+                "input": None,
                "triggers": [],
            },
        },
@@ -131,220 +123,7 @@ 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["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
+    assert payload["trigger_call_id"] == "abc123"


 def test_started_dedup_on_repeat_namespace() -> None:
@@ -409,12 +188,8 @@ 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"))

-    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)
+    events = [p["event"] for p in _drain_lifecycle(mux)]
+    assert events == ["started", "completed"]


 def test_failed_on_parent_task_result_with_error() -> None:
@@ -490,54 +265,6 @@ 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()
@@ -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.parent_task_id == "abc"
+    assert handle.trigger_call_id == "abc"
    assert handle.status == "started"
    _child_mux(handle)  # mini-mux backed

@@ -1849,14 +1849,14 @@ dev = [
    { name = "pytest-watch" },
    { name = "ruff", specifier = "==0.15.12" },
    { name = "starlette" },
-    { name = "ty", specifier = "==0.0.33" },
+    { name = "ty", specifier = "==0.0.23" },
 ]
 lint = [
    { name = "codespell" },
    { name = "mypy", specifier = "==1.20.2" },
    { name = "ruff", specifier = "==0.15.12" },
    { name = "starlette" },
-    { name = "ty", specifier = "==0.0.33" },
+    { name = "ty", specifier = "==0.0.23" },
 ]
 test = [
    { name = "pytest" },
@@ -618,14 +618,14 @@ dev = [
    { name = "pytest-watch" },
    { name = "ruff", specifier = "==0.15.12" },
    { name = "starlette" },
-    { name = "ty", specifier = "==0.0.33" },
+    { name = "ty", specifier = "==0.0.23" },
 ]
 lint = [
    { name = "codespell" },
    { name = "mypy", specifier = "==1.20.2" },
    { name = "ruff", specifier = "==0.15.12" },
    { name = "starlette" },
-    { name = "ty", specifier = "==0.0.33" },
+    { name = "ty", specifier = "==0.0.23" },
 ]
 test = [
    { name = "pytest" },
@@ -110,7 +110,7 @@ def get_client(
    if url is None:
        url = "http://api"
        if os.environ.get("__LANGGRAPH_DEFER_LOOPBACK_TRANSPORT") == "true":
-            transport = get_asgi_transport()(app=None, root_path="/noauth")  # ty: ignore[invalid-argument-type]
+            transport = get_asgi_transport()(app=None, root_path="/noauth")  # type: ignore[invalid-argument-type]
            _registered_transports.append(transport)
        else:
            try:
@@ -122,7 +122,7 @@ def get_client(
                    "Failed to connect to in-process LangGraph server. Deferring configuration.",
                    exc_info=True,
                )
-                transport = get_asgi_transport()(app=None, root_path="/noauth")  # ty: ignore[invalid-argument-type]
+                transport = get_asgi_transport()(app=None, root_path="/noauth")  # type: ignore[invalid-argument-type]
                _registered_transports.append(transport)

    if transport is None:
@@ -131,7 +131,7 @@ def get_client(
        base_url=url,
        transport=transport,
        timeout=(
-            httpx.Timeout(timeout)  # ty: ignore[invalid-argument-type]
+            httpx.Timeout(timeout)  # type: ignore[arg-type]
            if timeout is not None
            else httpx.Timeout(connect=5, read=300, write=300, pool=5)
        ),
@@ -49,7 +49,7 @@ async def _wrap_stream_v2(
    async for part in raw:
        v2 = _sse_to_v2_dict(part.event, part.data)
        if v2 is not None:
-            yield v2  # ty: ignore[invalid-yield]
+            yield v2


 class RunsClient:
@@ -144,9 +144,8 @@ def _resolve_timezone(tz: str | tzinfo | ZoneInfo | None) -> str | None:
        return tz
    if isinstance(tz, tzinfo):
        # ZoneInfo objects have a .key attribute with the IANA name
-        key = getattr(tz, "key", None)
-        if isinstance(key, str):
-            return key
+        if hasattr(tz, "key"):
+            return tz.key  # type: ignore[union-attr]
        # Fall back to tzname for fixed-offset timezones like datetime.timezone.utc
        name = tz.tzname(None)
        if name is not None:
@@ -210,7 +209,7 @@ def configure_loopback_transports(app: Any) -> None:
@functools.lru_cache(maxsize=1)
 def get_asgi_transport() -> type[httpx.ASGITransport]:
    try:
-        from langgraph_api import asgi_transport  # ty: ignore[unresolved-import]
+        from langgraph_api import asgi_transport  # type: ignore[unresolved-import]

        return asgi_transport.ASGITransport
    except ImportError:
@@ -77,7 +77,7 @@ def get_sync_client(
        base_url=url,
        transport=transport,
        timeout=(
-            httpx.Timeout(timeout)  # ty: ignore[invalid-argument-type]
+            httpx.Timeout(timeout)  # type: ignore[arg-type]
            if timeout is not None
            else httpx.Timeout(connect=5, read=300, write=300, pool=5)
        ),
@@ -49,7 +49,7 @@ def _wrap_stream_v2_sync(
    for part in raw:
        v2 = _sse_to_v2_dict(part.event, part.data)
        if v2 is not None:
-            yield v2  # ty: ignore[invalid-yield]
+            yield v2


 class SyncRunsClient:
@@ -16,10 +16,10 @@ T = TypeVar("T")
 CacheStatus = Literal["miss", "fresh", "stale", "expired"]

 try:
-    from langgraph_api.cache import (  # ty: ignore[unresolved-import]
+    from langgraph_api.cache import (  # type: ignore[unresolved-import]
        cache_get as _cache_get,
    )
-    from langgraph_api.cache import (  # ty: ignore[unresolved-import]
+    from langgraph_api.cache import (  # type: ignore[unresolved-import]
        cache_set as _cache_set,
    )
 except ImportError:
@@ -28,8 +28,8 @@ except ImportError:


 try:
-    from langgraph_api.cache import SWRResult  # ty: ignore[unresolved-import]
-    from langgraph_api.cache import swr as _api_swr  # ty: ignore[unresolved-import]
+    from langgraph_api.cache import SWRResult  # type: ignore[unresolved-import]
+    from langgraph_api.cache import swr as _api_swr  # type: ignore[unresolved-import]

 except ImportError:
    _api_swr = None
@@ -40,10 +40,7 @@ except ImportError:
        value: T
        status: CacheStatus

-        async def mutate(
-            self,
-            value: T = ...,  # ty: ignore[invalid-parameter-default]
-        ) -> T:  # ty: ignore[empty-body]
+        async def mutate(self, value: T = ...) -> T:  # type: ignore[assignment]
            """Update or revalidate the cached value."""
            ...

@@ -37,7 +37,7 @@ class APIError(httpx.HTTPStatusError, LangGraphError):
            req = response_or_request
            response = None

-        httpx.HTTPStatusError.__init__(self, message, request=req, response=response)  # ty: ignore[invalid-argument-type]
+        httpx.HTTPStatusError.__init__(self, message, request=req, response=response)  # type: ignore[arg-type]
        LangGraphError.__init__(self, message)

        self.request = req
@@ -156,7 +156,7 @@ class _ExecutionRuntime(_ServerRuntimeBase[ContextT], Generic[ContextT]):
        This API is in beta and may change in future releases.
    """

-    context: ContextT = field(default=None)  # ty: ignore[invalid-assignment]
+    context: ContextT = field(default=None)  # type: ignore[assignment]
    """The graph run context, typed by the graph's `context_schema`.

    Only available during `threads.create_run`.
@@ -55,7 +55,7 @@ class BytesLineDecoder:
            # Include any existing buffer in the first portion of the
            # splitlines result.
            self.buffer.extend(lines[0])
-            lines = [self.buffer, *lines[1:]]
+            lines = cast(list[BytesLike], [self.buffer, *lines[1:]])
            self.buffer = bytearray()

        if not trailing_newline:
@@ -69,7 +69,7 @@ class BytesLineDecoder:
        if not self.buffer and not self.trailing_cr:
            return []

-        lines: list[BytesLike] = [self.buffer]
+        lines = [self.buffer]
        self.buffer = bytearray()
        self.trailing_cr = False
        return lines
@@ -102,7 +102,7 @@ class SSEDecoder:

            sse = StreamPart(
                event=self._event,
-                data=orjson.loads(self._data) if self._data else None,  # ty: ignore[invalid-argument-type]
+                data=orjson.loads(self._data) if self._data else None,  # type: ignore[invalid-argument-type]
                id=self.last_event_id,
            )

@@ -33,7 +33,7 @@ lint = [
    "ruff==0.15.12",
    "codespell",
    "mypy==1.20.2",
-    "ty==0.0.33",
+    "ty==0.0.23",
    "starlette",
 ]
 dev = [
@@ -388,7 +388,7 @@ async def test_async_stream_v2_client_side_conversion() -> None:
            event="values", data={"messages": [{"role": "user", "content": "hi"}]}
        )
        yield StreamPart(event="updates|sub:abc", data={"node": {"out": 1}})
-        yield StreamPart(event="end", data=None)  # ty: ignore[invalid-argument-type]
+        yield StreamPart(event="end", data=None)  # type: ignore[arg-type]

    parts: list[StreamPartV2] = [part async for part in _wrap_stream_v2(mock_stream())]
    assert len(parts) == 3
@@ -420,7 +420,7 @@ def test_sync_stream_v2_client_side_conversion() -> None:
    def mock_stream() -> Any:
        yield StreamPart(event="metadata", data={"run_id": "r1"})
        yield StreamPart(event="values", data={"state": "full"})
-        yield StreamPart(event="end", data=None)  # ty: ignore[invalid-argument-type]
+        yield StreamPart(event="end", data=None)  # type: ignore[arg-type]

    parts: list[StreamPartV2] = list(_wrap_stream_v2_sync(mock_stream()))
    assert len(parts) == 2
@@ -67,6 +67,6 @@ class TestHandlerValidation:

        with pytest.raises(TypeError, match="must accept exactly 2 parameters"):

-            @encryption.encrypt.blob  # ty: ignore[invalid-argument-type]
+            @encryption.encrypt.blob  # type: ignore[arg-type]
            async def wrong_params(ctx):
                return ctx
@@ -533,14 +533,14 @@ dev = [
    { name = "pytest-watch" },
    { name = "ruff", specifier = "==0.15.12" },
    { name = "starlette" },
-    { name = "ty", specifier = "==0.0.33" },
+    { name = "ty", specifier = "==0.0.23" },
 ]
 lint = [
    { name = "codespell" },
    { name = "mypy", specifier = "==1.20.2" },
    { name = "ruff", specifier = "==0.15.12" },
    { name = "starlette" },
-    { name = "ty", specifier = "==0.0.33" },
+    { name = "ty", specifier = "==0.0.23" },
 ]
 test = [
    { name = "pytest" },
@@ -1275,26 +1275,26 @@ wheels = [

 [[package]]
 name = "ty"
-version = "0.0.33"
+version = "0.0.23"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/84/44/9478c50c266826c1bf30d1692e589755bffa8f1c0a3eb7af8a346c255991/ty-0.0.33.tar.gz", hash = "sha256:46d63bda07403322cb6c28ccfdd5536be916e13df725c29f7ccd0a21f06bd9e8", size = 5559373, upload-time = "2026-04-28T10:45:13.18Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/75/ba/d3c998ff4cf6b5d75b39356db55fe1b7caceecc522b9586174e6a5dee6f7/ty-0.0.23.tar.gz", hash = "sha256:5fb05db58f202af366f80ef70f806e48f5237807fe424ec787c9f289e3f3a4ef", size = 5341461, upload-time = "2026-03-13T12:34:23.125Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/e9/24/e287388c63a19191be26b32ff4dbd06029834068150ebe2532939bc4c851/ty-0.0.33-py3-none-linux_armv6l.whl", hash = "sha256:94d0a9d2234261a8911396d59e506b5923fe0971dbda43b9dcea287936887fcc", size = 11021308, upload-time = "2026-04-28T10:45:43.34Z" },
-    { url = "https://files.pythonhosted.org/packages/00/ca/ba1eed819895bd239fba8ee35dfcd5fcb266c203b0914a17a59579096bb5/ty-0.0.33-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:e4a2b5ba078f90de342f56b5f7979bb77c9b9b1d8625a041352ffc6ee93c4073", size = 10777272, upload-time = "2026-04-28T10:45:32.905Z" },
-    { url = "https://files.pythonhosted.org/packages/25/a8/c3131d37b44b3fea1d6654a1c929a0cd0873822f77a90482b8ec28f6fbbd/ty-0.0.33-py3-none-macosx_11_0_arm64.whl", hash = "sha256:84ff5707825e9af9668d2bcf66975f93e520a63b524ab494e3a8265735be2563", size = 10201078, upload-time = "2026-04-28T10:45:23.374Z" },
-    { url = "https://files.pythonhosted.org/packages/7b/db/d8e37ff0045810cc65e1ff36aa0da0a2253c05659787ac987df8a16c7897/ty-0.0.33-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:e375285736f57886868e7af0b11c7b0ec5b6543fa15e7ad2a714fed9f077d4e0", size = 10732347, upload-time = "2026-04-28T10:45:21.444Z" },
-    { url = "https://files.pythonhosted.org/packages/e0/1a/20e83a412506a918e4684fc67b567cf7cc13b105470b3428cb23c3d5aa13/ty-0.0.33-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:5680f6350c3b4e46b8bff6d7bb132366ea239463d6cad4892725d06046e65464", size = 10808238, upload-time = "2026-04-28T10:45:38.565Z" },
-    { url = "https://files.pythonhosted.org/packages/5d/4b/d0a39f4464dc6cb4cc2c159473ce216bd1846bfb684c0323a3cb36dce5c6/ty-0.0.33-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:c5535538bad8d0f7e62bcdff02197cdb30e41451d80b35d27e17d128f2e1dc5d", size = 11288348, upload-time = "2026-04-28T10:45:08.419Z" },
-    { url = "https://files.pythonhosted.org/packages/35/7e/f1745e0f9583363d7a83d9a4990fc244f76ecc30840ddad83dc16a33c52d/ty-0.0.33-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:da196c42bbbc069e1e21e3e52107c061aa9660352dae57a41930690b56e2c02d", size = 11789907, upload-time = "2026-04-28T10:45:19.064Z" },
-    { url = "https://files.pythonhosted.org/packages/a5/71/25f39f46a12d662859d45bc648555d0661044eb43db6b5648c9947487da9/ty-0.0.33-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:9281672921ef6d4460e03146b5e6c18cb1a3e3a3b8a1a88f6f33226d05a469b7", size = 11500774, upload-time = "2026-04-28T10:45:48.012Z" },
-    { url = "https://files.pythonhosted.org/packages/94/ec/136959ecbb7c71cb90537f5aea441c73f4ab24612868a6ecdc9d7444d32d/ty-0.0.33-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:82c1b8f303f82da64e878108e764be3ecbcd7c9903ac0a7f7031614ed00b97ab", size = 11360314, upload-time = "2026-04-28T10:45:05.402Z" },
-    { url = "https://files.pythonhosted.org/packages/cf/95/32809575c222f00beed498cb728e9290a0f5009f930025381bb7253b2206/ty-0.0.33-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:efe3af412c9ff67bce5fa37d0a2b0d8555c24072b145a5bac6c79637f1c83abe", size = 10707785, upload-time = "2026-04-28T10:45:10.836Z" },
-    { url = "https://files.pythonhosted.org/packages/13/89/c8e9531f7aa4a093359e15fa32c8e1277fbbe90d16894d7c6032d29f4b34/ty-0.0.33-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:aeec29c91ea768601747da546c3efc20b72c2fb1bd52bcc786a5c6eeff51d27b", size = 10834987, upload-time = "2026-04-28T10:45:40.738Z" },
-    { url = "https://files.pythonhosted.org/packages/31/16/9835fbcf5338af1a1917bd28fdb8a7193c210b83f243aa286fa9f79cb3ad/ty-0.0.33-py3-none-musllinux_1_2_i686.whl", hash = "sha256:a535977c52bbb5f7e96b8b70a6ad375ad077f4a9ff2492508ea3816a2b403819", size = 10968968, upload-time = "2026-04-28T10:45:30.26Z" },
-    { url = "https://files.pythonhosted.org/packages/36/69/64c76aabc1bc70c7f24b686cd93c3407f8ea430905e395f59bf9603ef571/ty-0.0.33-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:1d732facf39fcb221ba279d469c5040d37883e964f123b1563888efd34818180", size = 11458077, upload-time = "2026-04-28T10:45:45.971Z" },
-    { url = "https://files.pythonhosted.org/packages/91/84/fae27b0c4718776a298690d31ca4cc1995f2e3e1c63a7b59e84c41498e9a/ty-0.0.33-py3-none-win32.whl", hash = "sha256:d90960b574428dc252f85e8598ec5fcb7f619794196b2fc95a90da075ed4681c", size = 10345364, upload-time = "2026-04-28T10:45:16.836Z" },
-    { url = "https://files.pythonhosted.org/packages/3c/a0/a2938b23ae3e1a09a2d7c189e2ac5f7113676bae4e0e23948b568e18e5f8/ty-0.0.33-py3-none-win_amd64.whl", hash = "sha256:c1c3aec62c44de610c6e95f0a4e97ac3dbc07934bfdbf1fd90d758c9ff72f48e", size = 11342470, upload-time = "2026-04-28T10:45:26.455Z" },
-    { url = "https://files.pythonhosted.org/packages/ab/62/7fb948aace38d2f6329261bb33c035a8484549c74f1db28649c7a4c6fed9/ty-0.0.33-py3-none-win_arm64.whl", hash = "sha256:0d44f99ba1b441e55e2aa301b2ac0a21112784931b46a5f66f4ea9efe5620d97", size = 10742673, upload-time = "2026-04-28T10:45:35.555Z" },
+    { url = "https://files.pythonhosted.org/packages/f4/21/aab32603dfdfacd4819e52fa8c6074e7bd578218a5142729452fc6a62db6/ty-0.0.23-py3-none-linux_armv6l.whl", hash = "sha256:e810eef1a5f1cfc0731a58af8d2f334906a96835829767aed00026f1334a8dd7", size = 10329096, upload-time = "2026-03-13T12:34:09.432Z" },
+    { url = "https://files.pythonhosted.org/packages/9f/a9/dd3287a82dce3df546ec560296208d4905dcf06346b6e18c2f3c63523bd1/ty-0.0.23-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:e43d36bd89a151ddcad01acaeff7dcc507cb73ff164c1878d2d11549d39a061c", size = 10156631, upload-time = "2026-03-13T12:34:53.122Z" },
+    { url = "https://files.pythonhosted.org/packages/0f/01/3f25909b02fac29bb0a62b2251f8d62e65d697781ffa4cf6b47a4c075c85/ty-0.0.23-py3-none-macosx_11_0_arm64.whl", hash = "sha256:bd6a340969577b4645f231572c4e46012acba2d10d4c0c6570fe1ab74e76ae00", size = 9653211, upload-time = "2026-03-13T12:34:15.049Z" },
+    { url = "https://files.pythonhosted.org/packages/d5/60/bfc0479572a6f4b90501c869635faf8d84c8c68ffc5dd87d04f049affabc/ty-0.0.23-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:341441783e626eeb7b1ec2160432956aed5734932ab2d1c26f94d0c98b229937", size = 10156143, upload-time = "2026-03-13T12:34:34.468Z" },
+    { url = "https://files.pythonhosted.org/packages/3a/81/8a93e923535a340f54bea20ff196f6b2787782b2f2f399bd191c4bc132d6/ty-0.0.23-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:8ce1dc66c26d4167e2c78d12fa870ef5a7ec9cc344d2baaa6243297cfa88bd52", size = 10136632, upload-time = "2026-03-13T12:34:28.832Z" },
+    { url = "https://files.pythonhosted.org/packages/da/cb/2ac81c850c58acc9f976814404d28389c9c1c939676e32287b9cff61381e/ty-0.0.23-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:bae1e7a294bf8528836f7617dc5c360ea2dddb63789fc9471ae6753534adca05", size = 10655025, upload-time = "2026-03-13T12:34:37.105Z" },
+    { url = "https://files.pythonhosted.org/packages/b5/9b/bac771774c198c318ae699fc013d8cd99ed9caf993f661fba11238759244/ty-0.0.23-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:d2b162768764d9dc177c83fb497a51532bb67cbebe57b8fa0f2668436bf53f3c", size = 11230107, upload-time = "2026-03-13T12:34:20.751Z" },
+    { url = "https://files.pythonhosted.org/packages/14/09/7644fb0e297265e18243f878aca343593323b9bb19ed5278dcbc63781be0/ty-0.0.23-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:d28384e48ca03b34e4e2beee0e230c39bbfb68994bb44927fec61ef3642900da", size = 10934177, upload-time = "2026-03-13T12:34:17.904Z" },
+    { url = "https://files.pythonhosted.org/packages/18/14/69a25a0cad493fb6a947302471b579a03516a3b00e7bece77fdc6b4afb9b/ty-0.0.23-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:559d9a299df793cb7a7902caed5eda8a720ff69164c31c979673e928f02251ee", size = 10752487, upload-time = "2026-03-13T12:34:31.785Z" },
+    { url = "https://files.pythonhosted.org/packages/9d/2a/42fc3cbccf95af0a62308ebed67e084798ab7a85ef073c9986ef18032743/ty-0.0.23-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:32a7b8a14a98e1d20a9d8d2af23637ed7efdb297ac1fa2450b8e465d05b94482", size = 10133007, upload-time = "2026-03-13T12:34:42.838Z" },
+    { url = "https://files.pythonhosted.org/packages/e1/69/307833f1b52fa3670e0a1d496e43ef7df556ecde838192d3fcb9b35e360d/ty-0.0.23-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:6f803b9b9cca87af793467973b9abdd4b83e6b96d9b5e749d662cff7ead70b6d", size = 10169698, upload-time = "2026-03-13T12:34:12.351Z" },
+    { url = "https://files.pythonhosted.org/packages/89/ae/5dd379ec22d0b1cba410d7af31c366fcedff191d5b867145913a64889f66/ty-0.0.23-py3-none-musllinux_1_2_i686.whl", hash = "sha256:4a0bf086ec8e2197b7ea7ebfcf4be36cb6a52b235f8be61647ef1b2d99d6ffd3", size = 10346080, upload-time = "2026-03-13T12:34:40.012Z" },
+    { url = "https://files.pythonhosted.org/packages/98/c7/dfc83203d37998620bba9c4873a080c8850a784a8a46f56f8163c5b4e320/ty-0.0.23-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:252539c3fcd7aeb9b8d5c14e2040682c3e1d7ff640906d63fd2c4ce35865a4ba", size = 10848162, upload-time = "2026-03-13T12:34:45.421Z" },
+    { url = "https://files.pythonhosted.org/packages/89/08/05481511cfbcc1fd834b6c67aaae090cb609a079189ddf2032139ccfc490/ty-0.0.23-py3-none-win32.whl", hash = "sha256:51b591d19eef23bbc3807aef77d38fa1f003c354e1da908aa80ea2dca0993f77", size = 9748283, upload-time = "2026-03-13T12:34:50.607Z" },
+    { url = "https://files.pythonhosted.org/packages/31/2e/eaed4ff5c85e857a02415084c394e02c30476b65e158eec1938fdaa9a205/ty-0.0.23-py3-none-win_amd64.whl", hash = "sha256:1e137e955f05c501cfbb81dd2190c8fb7d01ec037c7e287024129c722a83c9ad", size = 10698355, upload-time = "2026-03-13T12:34:26.134Z" },
+    { url = "https://files.pythonhosted.org/packages/91/29/b32cb7b4c7d56b9ed50117f8ad6e45834aec293e4cb14749daab4e9236d5/ty-0.0.23-py3-none-win_arm64.whl", hash = "sha256:a0399bd13fd2cd6683fd0a2d59b9355155d46546d8203e152c556ddbdeb20842", size = 10155890, upload-time = "2026-03-13T12:34:48.082Z" },
 ]

 [[package]]
Author	SHA1	Message	Date
Sydney Runkle	2239aae856	fix(lint): remove redundant cast flagged by mypy	2026-05-06 15:08:37 -04:00
Sydney Runkle	e8c6d9cc31	chore: apply ruff format/lint fixes	2026-05-06 15:03:13 -04:00
Sydney Runkle	6c43a254ef	fix(message): full add_messages parity for _messages_delta_reducer Add REMOVE_ALL_MESSAGES sentinel handling and BaseMessageChunk coercion, completing parity with add_messages. REMOVE_ALL_MESSAGES resets state and discards all preceding writes; chunks are coerced to full messages via message_chunk_to_message. Both properties are batching-invariant.	2026-05-06 14:59:12 -04:00
Sydney Runkle	eddfb40703	fix(message): assign UUIDs to ID-less messages in _messages_delta_reducer Without UUID assignment, message eviction and RemoveMessage tombstoning fail for messages created without explicit IDs. Matches the behavior of add_messages. IDs are now assigned inline within the existing iterations over state_msgs and msgs to avoid an extra pass.	2026-05-06 14:42:36 -04:00