Add sync StreamMux, SubgraphRunStream, and chat model stream enhancements

Adds AsyncStreamMux sync counterpart, exports SubgraphRunStream, improves
EventLog future resolution safety, and expands run_stream with sync graph
run support. Includes comprehensive test updates across event log, mux,
reducers, and run stream modules.

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

@@ -339,37 +339,6 @@ async def test_list_metadata_custom_keys(
     assert results[0].metadata["run_id"] == "run-abc"
 
 
-async def test_list_global_search(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """alist(None, filter=...) searches across all threads."""
-    tid1, tid2 = str(uuid4()), str(uuid4())
-
-    # Use a unique marker so we don't collide with other tests' data
-    marker = str(uuid4())
-
-    cfg1 = generate_config(tid1)
-    cp1 = generate_checkpoint()
-    await saver.aput(cfg1, cp1, generate_metadata(source="input", marker=marker), {})
-
-    cfg2 = generate_config(tid2)
-    cp2 = generate_checkpoint()
-    await saver.aput(cfg2, cp2, generate_metadata(source="loop", marker=marker), {})
-
-    # Search across all threads with filter
-    results = []
-    async for tup in saver.alist(None, filter={"source": "input", "marker": marker}):
-        results.append(tup)
-    assert len(results) == 1
-    assert results[0].config["configurable"]["thread_id"] == tid1
-
-    # Search with marker only — should find both
-    results = []
-    async for tup in saver.alist(None, filter={"marker": marker}):
-        results.append(tup)
-    assert len(results) == 2
-
-
 ALL_LIST_TESTS = [
     test_list_all,
     test_list_by_thread,
@@ -380,7 +349,6 @@ ALL_LIST_TESTS = [
     test_list_metadata_filter_multiple_keys,
     test_list_metadata_filter_no_match,
     test_list_metadata_custom_keys,
-    test_list_global_search,
     test_list_before,
     test_list_limit,
     test_list_limit_plus_before,

libs/checkpoint-conformance/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-checkpoint-conformance"
-version = "0.0.1"
+version = "0.0.2"
 description = "Conformance test suite for LangGraph checkpointer implementations."
 authors = [{name = "William FH", email = "13333726+hinthornw@users.noreply.github.com"}]
 requires-python = ">=3.10"

libs/checkpoint-conformance/uv.lock

@@ -231,7 +231,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.2.22"
+version = "1.2.28"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -243,9 +243,9 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/b1/a3/c4cd6827a1df46c821e7214b7f7b7a28b189e6c9b84ef15c6d629c5e3179/langchain_core-1.2.22.tar.gz", hash = "sha256:8d8f726d03d3652d403da915126626bb6250747e8ba406537d849e68b9f5d058", size = 842487, upload-time = "2026-03-24T18:48:44.9Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/a4/317a1a3ac1df33a64adb3670bf88bbe3b3d5baa274db6863a979db472897/langchain_core-1.2.28.tar.gz", hash = "sha256:271a3d8bd618f795fdeba112b0753980457fc90537c46a0c11998516a74dc2cb", size = 846119, upload-time = "2026-04-08T18:19:34.867Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c7/a6/2ffacf0f1a3788f250e75d0b52a24896c413be11be3a6d42bcdf46fbea48/langchain_core-1.2.22-py3-none-any.whl", hash = "sha256:7e30d586b75918e828833b9ec1efc25465723566845dd652c277baf751e9c04b", size = 506829, upload-time = "2026-03-24T18:48:43.286Z" },
+    { url = "https://files.pythonhosted.org/packages/a8/92/32f785f077c7e898da97064f113c73fbd9ad55d1e2169cf3a391b183dedb/langchain_core-1.2.28-py3-none-any.whl", hash = "sha256:80764232581eaf8057bcefa71dbf8adc1f6a28d257ebd8b95ba9b8b452e8c6ac", size = 508727, upload-time = "2026-04-08T18:19:32.823Z" },
 ]
 
 [[package]]
@@ -263,7 +263,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint-conformance"
-version = "0.0.1"
+version = "0.0.2"
 source = { editable = "." }
 dependencies = [
     { name = "langgraph-checkpoint" },

libs/checkpoint-postgres/uv.lock

@@ -240,7 +240,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.2.22"
+version = "1.2.28"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -252,9 +252,9 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/b1/a3/c4cd6827a1df46c821e7214b7f7b7a28b189e6c9b84ef15c6d629c5e3179/langchain_core-1.2.22.tar.gz", hash = "sha256:8d8f726d03d3652d403da915126626bb6250747e8ba406537d849e68b9f5d058", size = 842487, upload-time = "2026-03-24T18:48:44.9Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/a4/317a1a3ac1df33a64adb3670bf88bbe3b3d5baa274db6863a979db472897/langchain_core-1.2.28.tar.gz", hash = "sha256:271a3d8bd618f795fdeba112b0753980457fc90537c46a0c11998516a74dc2cb", size = 846119, upload-time = "2026-04-08T18:19:34.867Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c7/a6/2ffacf0f1a3788f250e75d0b52a24896c413be11be3a6d42bcdf46fbea48/langchain_core-1.2.22-py3-none-any.whl", hash = "sha256:7e30d586b75918e828833b9ec1efc25465723566845dd652c277baf751e9c04b", size = 506829, upload-time = "2026-03-24T18:48:43.286Z" },
+    { url = "https://files.pythonhosted.org/packages/a8/92/32f785f077c7e898da97064f113c73fbd9ad55d1e2169cf3a391b183dedb/langchain_core-1.2.28-py3-none-any.whl", hash = "sha256:80764232581eaf8057bcefa71dbf8adc1f6a28d257ebd8b95ba9b8b452e8c6ac", size = 508727, upload-time = "2026-04-08T18:19:32.823Z" },
 ]
 
 [[package]]

libs/checkpoint/uv.lock

@@ -267,7 +267,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.2.23"
+version = "1.2.28"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -279,9 +279,9 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/1d/47/a5f21b651e9cbd7a26c3e5809336d10a0be94ef7bdf6bea47f2ad9fff1a8/langchain_core-1.2.23.tar.gz", hash = "sha256:fdec64f90cfea25317e88d9803c44684af1f4e30dec4e58320dd7393bb0f0785", size = 841684, upload-time = "2026-03-27T23:28:14.6Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/a4/317a1a3ac1df33a64adb3670bf88bbe3b3d5baa274db6863a979db472897/langchain_core-1.2.28.tar.gz", hash = "sha256:271a3d8bd618f795fdeba112b0753980457fc90537c46a0c11998516a74dc2cb", size = 846119, upload-time = "2026-04-08T18:19:34.867Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/9b/5a/6ff2d76618e4cac531ea51d4ef44c6add36575a84c3f0f8877aee68c951a/langchain_core-1.2.23-py3-none-any.whl", hash = "sha256:70866dfc5275b7840ce272ff70f0ff216c8666ab25dc1b41964a4ef58c02a3ff", size = 506709, upload-time = "2026-03-27T23:28:13.372Z" },
+    { url = "https://files.pythonhosted.org/packages/a8/92/32f785f077c7e898da97064f113c73fbd9ad55d1e2169cf3a391b183dedb/langchain_core-1.2.28-py3-none-any.whl", hash = "sha256:80764232581eaf8057bcefa71dbf8adc1f6a28d257ebd8b95ba9b8b452e8c6ac", size = 508727, upload-time = "2026-04-08T18:19:32.823Z" },
 ]
 
 [[package]]

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.19"
+__version__ = "0.4.21"

libs/cli/langgraph_cli/cli.py

@@ -817,6 +817,48 @@ def dev(
     )
 
 
+# ---------------------------------------------------------------------------
+# validate command
+# ---------------------------------------------------------------------------
+
+
+@OPT_CONFIG
+@cli.command(help="✅ Validate the LangGraph configuration file.")
+@log_command
+def validate(config: pathlib.Path):
+    import json
+
+    try:
+        with open(config) as f:
+            raw_config = json.load(f)
+    except json.JSONDecodeError as e:
+        raise click.UsageError(f"Invalid JSON in {config}: {e.args[0]}") from None
+
+    # Check for unknown keys before validation so they show alongside any error.
+    unknown_warnings = langgraph_cli.config.get_unknown_keys(raw_config)
+
+    try:
+        config_json = langgraph_cli.config.validate_config_file(config)
+    except (click.UsageError, ValueError) as e:
+        click.secho(f"Error: {e}", fg="red", err=True)
+        if unknown_warnings:
+            click.echo(err=True)
+            for warning in unknown_warnings:
+                click.secho(f"  warning: {warning}", fg="yellow", err=True)
+        raise SystemExit(1) from None
+
+    num_graphs = len(config_json.get("graphs", {}))
+    click.secho(
+        f"Configuration file {config} is valid. "
+        f"({num_graphs} graph{'s' if num_graphs != 1 else ''} found)",
+        fg="green",
+    )
+    if unknown_warnings:
+        click.echo()
+        for warning in unknown_warnings:
+            click.secho(f"  warning: {warning}", fg="yellow")
+
+
 # ---------------------------------------------------------------------------
 # new command
 # ---------------------------------------------------------------------------

libs/cli/langgraph_cli/config.py

@@ -182,7 +182,11 @@ def validate_config(config: Config) -> Config:
                     "Version must be major or major.minor or major.minor.patch."
                 )
         except TypeError:
-            raise click.UsageError(f"Invalid version format: {api_version}") from None
+            raise click.UsageError(
+                f"Invalid version format: {api_version}.\n\n"
+                "Pin to a minor version, e.g.:\n"
+                '  "api_version": "0.8"'
+            ) from None
 
     config = {
         "node_version": node_version,
@@ -220,45 +224,51 @@ def validate_config(config: Config) -> Config:
             if major < min_major:
                 raise click.UsageError(
                     f"Node.js version {node_version} is not supported. "
-                    f"Minimum required version is {MIN_NODE_VERSION}."
+                    f"Minimum required version is {MIN_NODE_VERSION}.\n\n"
+                    f"Set node_version to {MIN_NODE_VERSION} or higher:\n"
+                    f'  "node_version": "{MIN_NODE_VERSION}"'
                 )
         except ValueError as e:
             raise click.UsageError(str(e)) from None
 
     if pip_installer := config.get("pip_installer"):
-        if pip_installer == "uv_lock":
-            raise click.UsageError(
-                "pip_installer 'uv_lock' has been replaced. Use "
-                '`source: {"kind": "uv", "root": "..", '
-                '"package": "my-agent"}`.'
-            )
         if pip_installer not in ["auto", "pip", "uv"]:
             raise click.UsageError(
                 f"Invalid pip_installer: '{pip_installer}'. "
-                "Must be 'auto', 'pip', or 'uv'."
+                "Consider using uv-based source management instead:\n\n"
+                '  "source": {"kind": "uv", "root": ".."}'
             )
 
     source = config.get("source")
     source_kind = _get_source_kind(config)
     if source is not None and not isinstance(source, dict):
-        raise click.UsageError("`source` must be an object.")
+        raise click.UsageError(
+            "`source` must be an object, e.g.:\n"
+            '  "source": {"kind": "uv", "root": ".."}'
+        )
     if source is not None and source_kind != "uv":
-        raise click.UsageError("Invalid source.kind. Supported values: 'uv'.")
+        raise click.UsageError(
+            "Invalid source.kind. The only supported value is 'uv':\n"
+            '  "source": {"kind": "uv", "root": ".."}'
+        )
 
     if config.get("python_version"):
         pyversion = config["python_version"]
         if not pyversion.count(".") == 1 or not all(
             part.isdigit() for part in pyversion.split("-")[0].split(".")
         ):
+            parts = pyversion.split("-")[0].split(".")
+            fix = f"{parts[0]}.{parts[1]}" if len(parts) >= 2 else MIN_PYTHON_VERSION
             raise click.UsageError(
                 f"Invalid Python version format: {pyversion}. "
-                "Use 'major.minor' format (e.g., '3.11'). "
-                "Patch version cannot be specified."
+                "Use 'major.minor' format — patch version cannot be specified.\n\n"
+                f'  "python_version": "{fix}"'
             )
         if _parse_version(pyversion) < _parse_version(MIN_PYTHON_VERSION):
             raise click.UsageError(
                 f"Python version {pyversion} is not supported. "
-                f"Minimum required version is {MIN_PYTHON_VERSION}."
+                f"Minimum required version is {MIN_PYTHON_VERSION}.\n\n"
+                f'  "python_version": "{MIN_PYTHON_VERSION}"'
             )
         if "bullseye" in pyversion:
             raise click.UsageError(
@@ -269,12 +279,16 @@ def validate_config(config: Config) -> Config:
         if source_kind != "uv" and not config["dependencies"]:
             raise click.UsageError(
                 "No dependencies found in config. "
-                "Add at least one dependency to 'dependencies' list."
+                "Consider using uv-based source management:\n\n"
+                '  "source": {"kind": "uv", "root": ".."}'
             )
 
     if not config.get("graphs"):
         raise click.UsageError(
-            "No graphs found in config. Add at least one graph to 'graphs' dictionary."
+            "No graphs found in config. Add at least one graph, e.g.:\n"
+            '  "graphs": {\n'
+            '    "agent": "./my_agent/graph.py:graph"\n'
+            "  }"
         )
 
     # Validate image_distro config
@@ -287,7 +301,8 @@ def validate_config(config: Config) -> Config:
         if image_distro not in Distros.__args__:
             raise click.UsageError(
                 f"Invalid image_distro: '{image_distro}'. "
-                "Must be one of 'debian', 'wolfi', or 'bookworm'."
+                f"Must be one of: {', '.join(repr(d) for d in Distros.__args__)}.\n\n"
+                '  "image_distro": "wolfi"  (recommended)'
             )
 
     if source_kind == "uv":
@@ -369,6 +384,51 @@ def validate_config(config: Config) -> Config:
     return config
 
 
+# Keys recognized by validate_config (used to detect unknown fields).
+_KNOWN_CONFIG_KEYS = {
+    "python_version",
+    "node_version",
+    "api_version",
+    "base_image",
+    "image_distro",
+    "pip_config_file",
+    "pip_installer",
+    "source",
+    "dependencies",
+    "dockerfile_lines",
+    "graphs",
+    "env",
+    "store",
+    "auth",
+    "encryption",
+    "http",
+    "webhooks",
+    "checkpointer",
+    "ui",
+    "ui_config",
+    "keep_pkg_tools",
+    # Internal / legacy (still recognized, may error separately)
+    "_INTERNAL_docker_tag",
+    "project_root",
+    "package",
+}
+
+
+def get_unknown_keys(raw_config: dict) -> list[str]:
+    """Return warnings for unrecognized top-level keys (typos, etc.)."""
+    import difflib
+
+    unknown = set(raw_config) - _KNOWN_CONFIG_KEYS
+    warnings: list[str] = []
+    for key in sorted(unknown):
+        close = difflib.get_close_matches(key, _KNOWN_CONFIG_KEYS, n=1)
+        if close:
+            warnings.append(f"Unknown key '{key}' — did you mean '{close[0]}'?")
+        else:
+            warnings.append(f"Unknown key '{key}' is not a recognized config field.")
+    return warnings
+
+
 def validate_config_file(config_path: pathlib.Path) -> Config:
     """Load and validate a configuration file."""
     with open(config_path) as f:

libs/cli/tests/unit_tests/test_config.py

@@ -404,7 +404,7 @@ def test_validate_config_pip_installer():
             }
         )
     assert "Invalid pip_installer: 'conda'" in str(exc_info.value)
-    assert "Must be 'auto', 'pip', or 'uv'" in str(exc_info.value)
+    assert "uv-based source management" in str(exc_info.value)
 
     with pytest.raises(click.UsageError) as exc_info:
         validate_config(
@@ -417,7 +417,7 @@ def test_validate_config_pip_installer():
         )
     assert "Invalid pip_installer: 'invalid'" in str(exc_info.value)
 
-    with pytest.raises(click.UsageError, match="has been replaced"):
+    with pytest.raises(click.UsageError, match="Invalid pip_installer: 'uv_lock'"):
         validate_config(
             {
                 "python_version": "3.11",

libs/cli/uv-examples/monorepo/uv.lock

@@ -215,7 +215,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.2.27"
+version = "1.2.28"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -227,9 +227,9 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/13/5c/56d19a252bbb26247b7a7cd20821d48804d7ca03212fec709cd8db7c2516/langchain_core-1.2.27.tar.gz", hash = "sha256:c18372e4c4c1454d49bf23a2e484431e71bd39b64173a0f621f0fc283d7183a4", size = 844935, upload-time = "2026-04-07T14:56:32.364Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/a4/317a1a3ac1df33a64adb3670bf88bbe3b3d5baa274db6863a979db472897/langchain_core-1.2.28.tar.gz", hash = "sha256:271a3d8bd618f795fdeba112b0753980457fc90537c46a0c11998516a74dc2cb", size = 846119, upload-time = "2026-04-08T18:19:34.867Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c1/c3/6e0865bc130c448270eb9511b47863a3f9145cdb519b19f6e4758fa63d6f/langchain_core-1.2.27-py3-none-any.whl", hash = "sha256:9ecd6b0393b969fe88f6b9b309367134080ab095946d79e6937dd3911aa42bd5", size = 508315, upload-time = "2026-04-07T14:56:30.93Z" },
+    { url = "https://files.pythonhosted.org/packages/a8/92/32f785f077c7e898da97064f113c73fbd9ad55d1e2169cf3a391b183dedb/langchain_core-1.2.28-py3-none-any.whl", hash = "sha256:80764232581eaf8057bcefa71dbf8adc1f6a28d257ebd8b95ba9b8b452e8c6ac", size = 508727, upload-time = "2026-04-08T18:19:32.823Z" },
 ]
 
 [[package]]

libs/cli/uv-examples/simple/uv.lock

@@ -191,7 +191,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.2.27"
+version = "1.2.28"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -203,9 +203,9 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/13/5c/56d19a252bbb26247b7a7cd20821d48804d7ca03212fec709cd8db7c2516/langchain_core-1.2.27.tar.gz", hash = "sha256:c18372e4c4c1454d49bf23a2e484431e71bd39b64173a0f621f0fc283d7183a4", size = 844935, upload-time = "2026-04-07T14:56:32.364Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/a4/317a1a3ac1df33a64adb3670bf88bbe3b3d5baa274db6863a979db472897/langchain_core-1.2.28.tar.gz", hash = "sha256:271a3d8bd618f795fdeba112b0753980457fc90537c46a0c11998516a74dc2cb", size = 846119, upload-time = "2026-04-08T18:19:34.867Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c1/c3/6e0865bc130c448270eb9511b47863a3f9145cdb519b19f6e4758fa63d6f/langchain_core-1.2.27-py3-none-any.whl", hash = "sha256:9ecd6b0393b969fe88f6b9b309367134080ab095946d79e6937dd3911aa42bd5", size = 508315, upload-time = "2026-04-07T14:56:30.93Z" },
+    { url = "https://files.pythonhosted.org/packages/a8/92/32f785f077c7e898da97064f113c73fbd9ad55d1e2169cf3a391b183dedb/langchain_core-1.2.28-py3-none-any.whl", hash = "sha256:80764232581eaf8057bcefa71dbf8adc1f6a28d257ebd8b95ba9b8b452e8c6ac", size = 508727, upload-time = "2026-04-08T18:19:32.823Z" },
 ]
 
 [[package]]

libs/langgraph/langgraph/pregel/_messages_v2.py

@@ -0,0 +1,807 @@
+"""Protocol-native content-block message handler for StreamingHandler.
+
+Emits structured content-block lifecycle events (message-start,
+content-block-start/delta/finish, message-finish) instead of raw
+``(AIMessageChunk, metadata)`` tuples.  The existing
+:class:`~langgraph.pregel._messages.StreamMessagesHandler` is NOT
+modified — this handler is only activated when
+``__protocol_messages_stream`` is ``True`` in the run's configurable.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncIterator, Callable, Iterator, Sequence
+from dataclasses import dataclass, field
+from typing import Any, TypeVar, cast
+from uuid import UUID, uuid4
+
+from langchain_core.callbacks import BaseCallbackHandler
+from langchain_core.messages import AIMessageChunk, BaseMessage
+from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, LLMResult
+
+from langgraph._internal._constants import NS_SEP
+from langgraph.constants import TAG_HIDDEN, TAG_NOSTREAM
+from langgraph.pregel.protocol import StreamChunk
+from langgraph.stream._types import (
+    ContentBlockDeltaData,
+    ContentBlockFinishData,
+    ContentBlockStartData,
+    FinishReason,
+    InvalidToolCallBlock,
+    MessageErrorData,
+    MessageStartData,
+    ReasoningBlock,
+    TextBlock,
+    ToolCallBlock,
+    UsageInfo,
+)
+
+try:
+    from langchain_core.tracers._streaming import _StreamingCallbackHandler
+except ImportError:
+    _StreamingCallbackHandler = object  # type: ignore
+
+T = TypeVar("T")
+Meta = tuple[tuple[str, ...], dict[str, Any]]
+
+PROTOCOL_MESSAGES_STREAM_KEY = "__protocol_messages_stream"
+
+# ---------------------------------------------------------------------------
+# Content-block accumulation helpers
+# ---------------------------------------------------------------------------
+
+# A "compatible content block" is a dict matching one of the protocol block
+# TypedDicts (TextBlock, ReasoningBlock, ToolCallChunkBlock, etc.).
+CompatBlock = dict[str, Any]
+
+
+@dataclass
+class _ProtocolRunState:
+    """Per-run state for tracking the active message lifecycle."""
+
+    message_id: str | None = None
+    started: bool = False
+    blocks: dict[int, CompatBlock] = field(default_factory=dict)
+    usage: dict[str, Any] | None = None
+
+
+def _accumulate_block(accumulated: CompatBlock, delta: CompatBlock) -> CompatBlock:
+    """Merge *delta* into *accumulated*, returning the updated block."""
+    btype = accumulated.get("type", "text")
+    if btype == "text" and delta.get("type", "text") == "text":
+        accumulated["text"] = accumulated.get("text", "") + delta.get("text", "")
+    elif btype == "reasoning" and delta.get("type") == "reasoning":
+        accumulated["reasoning"] = accumulated.get("reasoning", "") + delta.get(
+            "reasoning", ""
+        )
+    elif btype == "tool_call_chunk" and delta.get("type") == "tool_call_chunk":
+        accumulated["args"] = accumulated.get("args", "") + delta.get("args", "")
+        if delta.get("id") is not None:
+            accumulated["id"] = delta["id"]
+        if delta.get("name") is not None:
+            accumulated["name"] = delta["name"]
+    return accumulated
+
+
+def _delta_block(previous: CompatBlock, current: CompatBlock) -> CompatBlock | None:
+    """Compute the delta between *previous* and *current*.
+
+    Returns ``None`` if there is nothing new to emit.
+    """
+    btype = current.get("type", "text")
+    if btype == "text":
+        prev_text = previous.get("text", "")
+        cur_text = current.get("text", "")
+        delta_text = cur_text[len(prev_text) :]
+        if not delta_text:
+            return None
+        return TextBlock(type="text", text=delta_text)
+    elif btype == "reasoning":
+        prev_r = previous.get("reasoning", "")
+        cur_r = current.get("reasoning", "")
+        delta_r = cur_r[len(prev_r) :]
+        if not delta_r:
+            return None
+        return ReasoningBlock(type="reasoning", reasoning=delta_r)
+    elif btype == "tool_call_chunk":
+        prev_args = previous.get("args", "")
+        cur_args = current.get("args", "")
+        delta_args = cur_args[len(prev_args) :]
+        has_meta = current.get("id") is not None or current.get("name") is not None
+        if not delta_args and not has_meta:
+            return None
+        result: CompatBlock = {"type": "tool_call_chunk", "args": delta_args}
+        if current.get("id") is not None and previous.get("id") is None:
+            result["id"] = current["id"]
+        if current.get("name") is not None and previous.get("name") is None:
+            result["name"] = current["name"]
+        return result
+    # Unrecognized block type — pass through unchanged
+    return current
+
+
+def _finalize_block(block: CompatBlock) -> CompatBlock:
+    """Convert a ``tool_call_chunk`` block to a finalized ``tool_call`` or
+    ``invalid_tool_call`` block.  Other block types pass through unchanged.
+    """
+    if block.get("type") != "tool_call_chunk":
+        return block
+    raw_args = block.get("args", "{}")
+    try:
+        parsed_args = json.loads(raw_args) if raw_args else {}
+        return ToolCallBlock(
+            type="tool_call",
+            id=block.get("id", ""),
+            name=block.get("name", ""),
+            args=parsed_args,
+        )
+    except (json.JSONDecodeError, TypeError):
+        return InvalidToolCallBlock(
+            type="invalid_tool_call",
+            id=block.get("id"),
+            name=block.get("name"),
+            args=raw_args,
+            error="Failed to parse tool call arguments as JSON",
+        )
+
+
+def _normalize_finish_reason(value: Any) -> FinishReason:
+    """Map provider-specific stop reasons to protocol finish reasons."""
+    if value == "length":
+        return "length"
+    if value == "content_filter":
+        return "content_filter"
+    if value in ("tool_use", "tool_calls"):
+        return "tool_use"
+    # "end_turn", "stop", None, and anything else → "stop"
+    return "stop"
+
+
+def _accumulate_usage(
+    current: dict[str, Any] | None, delta: Any
+) -> dict[str, Any] | None:
+    """Accumulate usage metadata from streamed chunks."""
+    if not isinstance(delta, dict):
+        return current
+    if current is None:
+        return dict(delta)
+    for key in ("input_tokens", "output_tokens", "total_tokens", "cached_tokens"):
+        if key in delta:
+            current[key] = current.get(key, 0) + delta[key]
+    # Merge detail dicts
+    for detail_key in ("input_token_details", "output_token_details"):
+        if detail_key in delta and isinstance(delta[detail_key], dict):
+            if detail_key not in current:
+                current[detail_key] = {}
+            current[detail_key].update(delta[detail_key])
+    return current
+
+
+def _to_protocol_usage(usage: dict[str, Any] | None) -> UsageInfo | None:
+    """Convert LangChain usage metadata to protocol ``UsageInfo``."""
+    if usage is None:
+        return None
+    result: dict[str, Any] = {}
+    if "input_tokens" in usage:
+        result["input_tokens"] = usage["input_tokens"]
+    if "output_tokens" in usage:
+        result["output_tokens"] = usage["output_tokens"]
+    if "total_tokens" in usage:
+        result["total_tokens"] = usage["total_tokens"]
+    if "cached_tokens" in usage:
+        result["cached_tokens"] = usage["cached_tokens"]
+    return UsageInfo(**result) if result else None
+
+
+# ---------------------------------------------------------------------------
+# Extracting content blocks from LangChain messages
+# ---------------------------------------------------------------------------
+
+
+def _extract_blocks_from_chunk(msg: AIMessageChunk) -> list[tuple[int, CompatBlock]]:
+    """Extract ``(index, block)`` pairs from an ``AIMessageChunk``.
+
+    LangChain stores content in several places:
+    - ``content: str`` — a single text block at index 0
+    - ``content: list[dict]`` — explicit content blocks with their own types
+    - ``tool_call_chunks`` — separate list for streamed tool call deltas
+    """
+    blocks: list[tuple[int, CompatBlock]] = []
+    content = msg.content
+    if isinstance(content, str) and content:
+        blocks.append((0, dict(TextBlock(type="text", text=content))))
+    elif isinstance(content, list):
+        for i, item in enumerate(content):
+            if not isinstance(item, dict):
+                continue
+            ctype = item.get("type", "")
+            if ctype == "text" and item.get("text"):
+                blocks.append(
+                    (
+                        item.get("index", i),
+                        dict(TextBlock(type="text", text=item["text"])),
+                    )
+                )
+            elif ctype in ("reasoning_content", "reasoning", "thinking"):
+                reasoning_text = (
+                    item.get("reasoning_content")
+                    or item.get("reasoning")
+                    or item.get("thinking", "")
+                )
+                if reasoning_text:
+                    blocks.append(
+                        (
+                            item.get("index", i),
+                            dict(
+                                ReasoningBlock(
+                                    type="reasoning", reasoning=reasoning_text
+                                )
+                            ),
+                        )
+                    )
+
+    # Tool call chunks live in a separate field
+    for tc in msg.tool_call_chunks or []:
+        idx = tc.get("index")
+        if idx is None:
+            # Assign indices after text content blocks
+            idx = len(blocks)
+        block: CompatBlock = {"type": "tool_call_chunk", "args": tc.get("args", "")}
+        if tc.get("id") is not None:
+            block["id"] = tc["id"]
+        if tc.get("name") is not None:
+            block["name"] = tc["name"]
+        blocks.append((idx, block))
+
+    return blocks
+
+
+# ---------------------------------------------------------------------------
+# The handler
+# ---------------------------------------------------------------------------
+
+
+class StreamProtocolMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
+    """Callback handler that emits content-block protocol events.
+
+    Activated when ``__protocol_messages_stream`` is ``True`` in the run's
+    configurable metadata.  Emits ``StreamChunk`` tuples of the form
+    ``(namespace, "messages", data)`` where *data* is one of the
+    ``MessagesData`` event types (``message-start``, ``content-block-start``,
+    etc.).
+    """
+
+    run_inline = True
+
+    def __init__(
+        self,
+        stream: Callable[[StreamChunk], None],
+        subgraphs: bool,
+        *,
+        parent_ns: tuple[str, ...] | None = None,
+    ) -> None:
+        self.stream = stream
+        self.subgraphs = subgraphs
+        self.parent_ns = parent_ns
+        # Per-run metadata: run_id → (namespace, metadata_dict)
+        self.metadata: dict[UUID, Meta] = {}
+        # Per-run protocol state for streamed messages
+        self.protocol_runs: dict[UUID, _ProtocolRunState] = {}
+        # Stable message ID mapping: run_id → message_id
+        self.stable_message_ids: dict[UUID, str] = {}
+        # Seen message IDs for deduplication of chain-emitted messages
+        self.seen: set[str | int] = set()
+
+    def _emit(self, meta: Meta, data: Any) -> None:
+        """Emit a protocol event as a StreamChunk.
+
+        The node name from *meta* is embedded at ``"__node__"`` so the
+        stream pump can lift it into ``params.node`` without changing the
+        ``StreamChunk`` tuple shape.
+        """
+        node = meta[1].get("langgraph_node")
+        if node and isinstance(data, dict):
+            data = {**data, "__node__": node}
+        self.stream((meta[0], "messages", data))
+
+    # -- Chat model callbacks -----------------------------------------------
+
+    def on_chat_model_start(
+        self,
+        serialized: dict[str, Any],
+        messages: list[list[BaseMessage]],
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        tags: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        if metadata and (not tags or (TAG_NOSTREAM not in tags)):
+            ns = tuple(cast(str, metadata["langgraph_checkpoint_ns"]).split(NS_SEP))[
+                :-1
+            ]
+            if not self.subgraphs and len(ns) > 0 and ns != self.parent_ns:
+                return
+            if tags:
+                if filtered := [t for t in tags if not t.startswith("seq:step")]:
+                    metadata["tags"] = filtered
+            self.metadata[run_id] = (ns, metadata)
+            self.protocol_runs[run_id] = _ProtocolRunState()
+
+    def on_llm_new_token(
+        self,
+        token: str,
+        *,
+        chunk: ChatGenerationChunk | None = None,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        tags: list[str] | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        if not isinstance(chunk, ChatGenerationChunk):
+            return
+        meta = self.metadata.get(run_id)
+        if meta is None:
+            return
+        state = self.protocol_runs.get(run_id)
+        if state is None:
+            return
+
+        msg = chunk.message
+        if not isinstance(msg, AIMessageChunk):
+            return
+
+        # Emit message-start on first token
+        if not state.started:
+            message_id = self._normalize_message_id(msg, run_id)
+            state.message_id = message_id
+            state.started = True
+            start_data = dict(
+                MessageStartData(
+                    event="message-start",
+                    role="ai",
+                )
+            )
+            if message_id:
+                start_data["message_id"] = message_id
+            self._emit(meta, start_data)
+
+        # Extract content blocks from this chunk
+        extracted = _extract_blocks_from_chunk(msg)
+        for idx, delta_block in extracted:
+            if idx not in state.blocks:
+                # New block — emit content-block-start
+                state.blocks[idx] = dict(delta_block)
+                # Start block has empty content placeholder
+                start_block = _make_start_block(delta_block)
+                self._emit(
+                    meta,
+                    ContentBlockStartData(
+                        event="content-block-start",
+                        index=idx,
+                        content_block=start_block,
+                    ),
+                )
+                # Then emit the first delta
+                first_delta = _delta_block(
+                    _make_start_block(delta_block), state.blocks[idx]
+                )
+                if first_delta is not None:
+                    self._emit(
+                        meta,
+                        ContentBlockDeltaData(
+                            event="content-block-delta",
+                            index=idx,
+                            content_block=first_delta,
+                        ),
+                    )
+            else:
+                # Existing block — compute delta, accumulate, emit
+                previous = dict(state.blocks[idx])
+                state.blocks[idx] = _accumulate_block(state.blocks[idx], delta_block)
+                delta = _delta_block(previous, state.blocks[idx])
+                if delta is not None:
+                    self._emit(
+                        meta,
+                        ContentBlockDeltaData(
+                            event="content-block-delta",
+                            index=idx,
+                            content_block=delta,
+                        ),
+                    )
+
+        # Accumulate usage from chunk
+        if msg.usage_metadata:
+            state.usage = _accumulate_usage(state.usage, msg.usage_metadata)
+
+    def on_llm_end(
+        self,
+        response: LLMResult,
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        meta = self.metadata.pop(run_id, None)
+        state = self.protocol_runs.pop(run_id, None)
+        if meta is None or state is None:
+            return
+
+        # Extract finish reason and usage from the final generation
+        finish_reason: FinishReason = "stop"
+        final_usage = state.usage
+
+        if response.generations and response.generations[0]:
+            gen = response.generations[0][0]
+            if isinstance(gen, ChatGeneration):
+                final_msg = gen.message
+                # Get finish reason from response_metadata
+                rm = getattr(final_msg, "response_metadata", {}) or {}
+                raw_reason = rm.get("finish_reason") or rm.get("stop_reason")
+                if raw_reason:
+                    finish_reason = _normalize_finish_reason(raw_reason)
+                # If we have tool calls in the final message, infer tool_use
+                if (
+                    finish_reason == "stop"
+                    and hasattr(final_msg, "tool_calls")
+                    and final_msg.tool_calls
+                ):
+                    finish_reason = "tool_use"
+                # Get usage from final message if not accumulated from chunks
+                if final_usage is None and hasattr(final_msg, "usage_metadata"):
+                    final_usage = (
+                        dict(final_msg.usage_metadata)
+                        if final_msg.usage_metadata
+                        else None
+                    )
+
+                # If we never got streaming tokens (non-streamed model call),
+                # emit the full message lifecycle now
+                if not state.started:
+                    self._emit_full_message(meta, final_msg, finish_reason, final_usage)
+                    return
+
+        # Close out any open content blocks
+        for idx in sorted(state.blocks):
+            finalized = _finalize_block(state.blocks[idx])
+            self._emit(
+                meta,
+                ContentBlockFinishData(
+                    event="content-block-finish",
+                    index=idx,
+                    content_block=finalized,
+                ),
+            )
+
+        # Emit message-finish
+        finish_data: dict[str, Any] = {
+            "event": "message-finish",
+            "reason": finish_reason,
+        }
+        usage_info = _to_protocol_usage(final_usage)
+        if usage_info is not None:
+            finish_data["usage"] = usage_info
+        self._emit(meta, finish_data)
+
+        # Track the message as seen for dedup
+        if state.message_id:
+            self.seen.add(state.message_id)
+
+    def on_llm_error(
+        self,
+        error: BaseException,
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        meta = self.metadata.pop(run_id, None)
+        state = self.protocol_runs.pop(run_id, None)
+        self.stable_message_ids.pop(run_id, None)
+        if meta is None or state is None:
+            return
+        if state.started:
+            self._emit(
+                meta,
+                MessageErrorData(
+                    event="error",
+                    message=str(error),
+                ),
+            )
+
+    # -- Chain callbacks (for node-level message dedup) ---------------------
+
+    def on_chain_start(
+        self,
+        serialized: dict[str, Any],
+        inputs: dict[str, Any],
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        tags: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        if (
+            metadata
+            and kwargs.get("name") == metadata.get("langgraph_node")
+            and (not tags or TAG_HIDDEN not in tags)
+        ):
+            ns = tuple(cast(str, metadata["langgraph_checkpoint_ns"]).split(NS_SEP))[
+                :-1
+            ]
+            if not self.subgraphs and len(ns) > 0:
+                return
+            self.metadata[run_id] = (ns, metadata)
+            # Record input message IDs for deduplication
+            self._record_seen_messages(inputs)
+
+    def on_chain_end(
+        self,
+        response: Any,
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        meta = self.metadata.pop(run_id, None)
+        if meta is None:
+            return
+        # Emit protocol events for any new messages in the node's output
+        self._emit_chain_messages(meta, response)
+
+    def on_chain_error(
+        self,
+        error: BaseException,
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        self.metadata.pop(run_id, None)
+
+    # -- Iterator taps (required by _StreamingCallbackHandler) ---------------
+
+    def tap_output_aiter(
+        self, run_id: UUID, output: AsyncIterator[T]
+    ) -> AsyncIterator[T]:
+        return output
+
+    def tap_output_iter(self, run_id: UUID, output: Iterator[T]) -> Iterator[T]:
+        return output
+
+    # -- Internal helpers ---------------------------------------------------
+
+    def _normalize_message_id(self, msg: BaseMessage, run_id: UUID) -> str | None:
+        """Return a stable message ID for this run, creating one if needed."""
+        msg_id = msg.id
+        if msg_id is None:
+            msg_id = self.stable_message_ids.get(run_id)
+        if msg_id is None:
+            msg_id = f"run-{run_id}"
+        self.stable_message_ids[run_id] = msg_id
+        # Mutate the message for consistency downstream
+        if msg.id != msg_id:
+            msg.id = msg_id
+        return msg_id
+
+    def _emit_full_message(
+        self,
+        meta: Meta,
+        msg: BaseMessage,
+        finish_reason: FinishReason,
+        usage: dict[str, Any] | None,
+        role: str = "ai",
+    ) -> None:
+        """Emit a complete message lifecycle for a non-streamed model call."""
+        message_id = msg.id or str(uuid4())
+        if message_id in self.seen:
+            return
+        self.seen.add(message_id)
+
+        # message-start
+        start_data = dict(
+            MessageStartData(
+                event="message-start",
+                role=role,
+            )
+        )
+        start_data["message_id"] = message_id
+        self._emit(meta, start_data)
+
+        # Extract all blocks from the final message
+        blocks = _extract_final_blocks(msg)
+        for idx, block in blocks:
+            # content-block-start with the full content
+            self._emit(
+                meta,
+                ContentBlockStartData(
+                    event="content-block-start",
+                    index=idx,
+                    content_block=_make_start_block(block),
+                ),
+            )
+            # content-block-delta with the full content
+            delta = _delta_block(_make_start_block(block), block)
+            if delta is not None:
+                self._emit(
+                    meta,
+                    ContentBlockDeltaData(
+                        event="content-block-delta",
+                        index=idx,
+                        content_block=delta,
+                    ),
+                )
+            # content-block-finish
+            finalized = _finalize_block(block)
+            self._emit(
+                meta,
+                ContentBlockFinishData(
+                    event="content-block-finish",
+                    index=idx,
+                    content_block=finalized,
+                ),
+            )
+
+        # message-finish
+        finish_data: dict[str, Any] = {
+            "event": "message-finish",
+            "reason": finish_reason,
+        }
+        usage_info = _to_protocol_usage(usage)
+        if usage_info is not None:
+            finish_data["usage"] = usage_info
+        self._emit(meta, finish_data)
+
+    def _record_seen_messages(self, obj: Any) -> None:
+        """Record message IDs from node inputs for deduplication."""
+        if isinstance(obj, BaseMessage):
+            if obj.id is not None:
+                self.seen.add(obj.id)
+        elif isinstance(obj, dict):
+            for value in obj.values():
+                self._record_seen_messages(value)
+        elif isinstance(obj, Sequence) and not isinstance(obj, (str, bytes)):
+            for item in obj:
+                self._record_seen_messages(item)
+
+    def _emit_chain_messages(self, meta: Meta, response: Any) -> None:
+        """Emit protocol events for messages found in chain output."""
+        from langgraph.types import Command
+
+        if isinstance(response, Command):
+            self._emit_chain_messages(meta, response.update)
+        elif isinstance(response, BaseMessage):
+            self._emit_message_from_chain(meta, response)
+        elif isinstance(response, Sequence) and not isinstance(response, (str, bytes)):
+            for item in response:
+                if isinstance(item, Command):
+                    self._emit_chain_messages(meta, item.update)
+                elif isinstance(item, BaseMessage):
+                    self._emit_message_from_chain(meta, item)
+        elif isinstance(response, dict):
+            for value in response.values():
+                if isinstance(value, BaseMessage):
+                    self._emit_message_from_chain(meta, value)
+                elif isinstance(value, Sequence) and not isinstance(
+                    value, (str, bytes)
+                ):
+                    for item in value:
+                        if isinstance(item, BaseMessage):
+                            self._emit_message_from_chain(meta, item)
+
+    def _emit_message_from_chain(self, meta: Meta, msg: BaseMessage) -> None:
+        """Emit a full message lifecycle for a message from a chain output,
+        deduplicating against previously-seen messages."""
+        if msg.id is not None and msg.id in self.seen:
+            return
+        if msg.id is None:
+            msg.id = str(uuid4())
+
+        # Determine role and finish reason
+        role = "ai"
+        if hasattr(msg, "type"):
+            if msg.type == "human":
+                role = "human"
+            elif msg.type == "system":
+                role = "system"
+
+        finish_reason: FinishReason = "stop"
+        rm = getattr(msg, "response_metadata", {}) or {}
+        raw_reason = rm.get("finish_reason") or rm.get("stop_reason")
+        if raw_reason:
+            finish_reason = _normalize_finish_reason(raw_reason)
+        if finish_reason == "stop" and hasattr(msg, "tool_calls") and msg.tool_calls:
+            finish_reason = "tool_use"
+
+        raw_usage = getattr(msg, "usage_metadata", None)
+        usage = dict(raw_usage) if raw_usage else None
+
+        self._emit_full_message(meta, msg, finish_reason, usage, role=role)
+
+
+# ---------------------------------------------------------------------------
+# Block extraction for finalized (non-streamed) messages
+# ---------------------------------------------------------------------------
+
+
+def _extract_final_blocks(msg: BaseMessage) -> list[tuple[int, CompatBlock]]:
+    """Extract ``(index, block)`` pairs from a finalized ``AIMessage``."""
+    blocks: list[tuple[int, CompatBlock]] = []
+    content = msg.content
+
+    if isinstance(content, str) and content:
+        blocks.append((0, dict(TextBlock(type="text", text=content))))
+    elif isinstance(content, list):
+        for i, item in enumerate(content):
+            if not isinstance(item, dict):
+                continue
+            ctype = item.get("type", "")
+            if ctype == "text" and item.get("text"):
+                blocks.append((i, dict(TextBlock(type="text", text=item["text"]))))
+            elif ctype in ("reasoning_content", "reasoning", "thinking"):
+                reasoning_text = (
+                    item.get("reasoning_content")
+                    or item.get("reasoning")
+                    or item.get("thinking", "")
+                )
+                if reasoning_text:
+                    blocks.append(
+                        (
+                            i,
+                            dict(
+                                ReasoningBlock(
+                                    type="reasoning", reasoning=reasoning_text
+                                )
+                            ),
+                        )
+                    )
+
+    # Finalized tool calls (already parsed, not chunks)
+    for tc in getattr(msg, "tool_calls", None) or []:
+        idx = len(blocks)
+        blocks.append(
+            (
+                idx,
+                dict(
+                    ToolCallBlock(
+                        type="tool_call",
+                        id=tc.get("id", ""),
+                        name=tc.get("name", ""),
+                        args=tc.get("args", {}),
+                    )
+                ),
+            )
+        )
+
+    return blocks
+
+
+def _make_start_block(block: CompatBlock) -> CompatBlock:
+    """Create an empty start placeholder for a content block."""
+    btype = block.get("type", "text")
+    if btype == "text":
+        return TextBlock(type="text", text="")
+    elif btype == "reasoning":
+        return ReasoningBlock(type="reasoning", reasoning="")
+    elif btype == "tool_call_chunk":
+        result: CompatBlock = {"type": "tool_call_chunk", "args": ""}
+        if "id" in block:
+            result["id"] = block["id"]
+        if "name" in block:
+            result["name"] = block["name"]
+        return result
+    elif btype == "tool_call":
+        # Already finalized — return as-is for start event
+        return ToolCallBlock(
+            type="tool_call",
+            id=block.get("id", ""),
+            name=block.get("name", ""),
+            args=block.get("args", {}),
+        )
+    return dict(block)
+
+
+__all__ = ["PROTOCOL_MESSAGES_STREAM_KEY", "StreamProtocolMessagesHandler"]

libs/langgraph/langgraph/pregel/main.py

@@ -128,6 +128,10 @@ from langgraph.pregel._loop import (
     SyncPregelLoop,
 )
 from langgraph.pregel._messages import StreamMessagesHandler
+from langgraph.pregel._messages_v2 import (
+    PROTOCOL_MESSAGES_STREAM_KEY,
+    StreamProtocolMessagesHandler,
+)
 from langgraph.pregel._read import DEFAULT_BOUND, PregelNode
 from langgraph.pregel._retry import RetryPolicy
 from langgraph.pregel._runner import PregelRunner
@@ -2616,8 +2620,15 @@ class Pregel(
             # set up messages stream mode
             if "messages" in stream_modes:
                 ns_ = cast(str | None, config[CONF].get(CONFIG_KEY_CHECKPOINT_NS))
+                _msg_cls = (
+                    StreamProtocolMessagesHandler
+                    if config.get("configurable", {}).get(
+                        PROTOCOL_MESSAGES_STREAM_KEY, False
+                    )
+                    else StreamMessagesHandler
+                )
                 run_manager.inheritable_handlers.append(
-                    StreamMessagesHandler(
+                    _msg_cls(
                         stream.put,
                         subgraphs,
                         parent_ns=tuple(ns_.split(NS_SEP)) if ns_ else None,
@@ -2935,7 +2946,10 @@ class Pregel(
                     True
                     for h in run_manager.handlers
                     if isinstance(h, _StreamingCallbackHandler)
-                    and not isinstance(h, StreamMessagesHandler)
+                    and not isinstance(
+                        h,
+                        (StreamMessagesHandler, StreamProtocolMessagesHandler),
+                    )
                 ),
                 False,
             )
@@ -2972,10 +2986,16 @@ class Pregel(
                 config[CONF][CONFIG_KEY_CHECKPOINT_NS] = recast_checkpoint_ns(ns)
             # set up messages stream mode
             if "messages" in stream_modes:
-                # namespace can be None in a root level graph?
                 ns_ = cast(str | None, config[CONF].get(CONFIG_KEY_CHECKPOINT_NS))
+                _msg_cls = (
+                    StreamProtocolMessagesHandler
+                    if config.get("configurable", {}).get(
+                        PROTOCOL_MESSAGES_STREAM_KEY, False
+                    )
+                    else StreamMessagesHandler
+                )
                 run_manager.inheritable_handlers.append(
-                    StreamMessagesHandler(
+                    _msg_cls(
                         stream_put,
                         subgraphs,
                         parent_ns=tuple(ns_.split(NS_SEP)) if ns_ else None,

libs/langgraph/langgraph/stream/__init__.py

@@ -0,0 +1,49 @@
+"""Stream protocol types and infrastructure for LangGraph."""
+
+from langgraph.stream._convert import STREAM_V2_MODES, convert_to_protocol_event
+from langgraph.stream._event_log import EventLog
+from langgraph.stream._mux import AsyncStreamMux, StreamMux
+from langgraph.stream._types import (
+    InterruptPayload,
+    ProtocolEvent,
+    StreamTransformer,
+)
+from langgraph.stream.chat_model_stream import AsyncChatModelStream, ChatModelStream
+from langgraph.stream.run_stream import (
+    AsyncGraphRunStream,
+    AsyncSubgraphRunStream,
+    GraphRunStream,
+    SubgraphRunStream,
+    create_async_graph_run_stream,
+    create_graph_run_stream,
+)
+from langgraph.stream.stream_channel import StreamChannel, is_stream_channel
+from langgraph.stream.streaming_handler import StreamingHandler
+from langgraph.stream.transformers import (
+    MessagesTransformer,
+    ValuesTransformer,
+)
+
+__all__ = [
+    "STREAM_V2_MODES",
+    "AsyncStreamMux",
+    "AsyncChatModelStream",
+    "AsyncGraphRunStream",
+    "AsyncSubgraphRunStream",
+    "ChatModelStream",
+    "EventLog",
+    "GraphRunStream",
+    "InterruptPayload",
+    "MessagesTransformer",
+    "ProtocolEvent",
+    "StreamChannel",
+    "StreamMux",
+    "SubgraphRunStream",
+    "StreamTransformer",
+    "StreamingHandler",
+    "ValuesTransformer",
+    "convert_to_protocol_event",
+    "create_async_graph_run_stream",
+    "create_graph_run_stream",
+    "is_stream_channel",
+]

libs/langgraph/langgraph/stream/_convert.py

@@ -0,0 +1,79 @@
+"""Convert raw ``StreamChunk`` tuples to ``ProtocolEvent`` envelopes.
+
+Each ``StreamMode`` is mapped to a ``ProtocolEvent`` whose ``method``
+field matches the mode name and whose ``params.data`` wraps the
+original payload.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any
+
+from langgraph.stream._types import ProtocolEvent, _ProtocolEventParams
+from langgraph.types import StreamMode
+
+#: All stream modes requested by ``StreamingHandler`` when calling the
+#: underlying ``stream()`` / ``astream()``.
+STREAM_V2_MODES: list[StreamMode] = [
+    "values",
+    "updates",
+    "messages",
+    "custom",
+    "checkpoints",
+    "tasks",
+    "debug",
+]
+
+_SUPPORTED_MODES: set[str] = set(STREAM_V2_MODES)
+
+
+def convert_to_protocol_event(
+    ns: tuple[str, ...],
+    mode: str,
+    payload: Any,
+    *,
+    node: str | None = None,
+) -> ProtocolEvent | None:
+    """Convert a ``StreamChunk`` to a ``ProtocolEvent``.
+
+    Returns ``None`` for unsupported or unknown modes.
+
+    The ``seq`` field is left as ``0`` here; the :class:`StreamMux` is
+    the sole seq assigner and overwrites it inside ``push()``.
+
+    Parameters
+    ----------
+    ns:
+        Namespace tuple from the ``StreamChunk``.
+    mode:
+        Stream mode string (``"values"``, ``"updates"``, etc.).
+    payload:
+        The raw payload from the stream.
+    node:
+        Optional node name for provenance.
+    """
+    if mode not in _SUPPORTED_MODES:
+        return None
+
+    params: _ProtocolEventParams = {
+        "namespace": list(ns),
+        "timestamp": _now_ms(),
+        "data": payload,
+    }
+    if node is not None:
+        params["node"] = node
+
+    return ProtocolEvent(
+        type="event",
+        method=mode,
+        params=params,
+    )
+
+
+def _now_ms() -> int:
+    """Current time in milliseconds since epoch."""
+    return int(time.time() * 1000)
+
+
+__all__ = ["STREAM_V2_MODES", "convert_to_protocol_event"]

libs/langgraph/langgraph/stream/_event_log.py

@@ -0,0 +1,136 @@
+"""Replayable append-only event buffer for StreamingHandler.
+
+``EventLog`` stores protocol events in an ordered list and supports
+multiple independent async iterators, each with their own cursor
+offset.  Subscribers that join mid-stream replay from a given offset
+without losing earlier events.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+def _resolve_future(fut: asyncio.Future[None]) -> None:
+    """Set a future's result if it hasn't already completed or been cancelled.
+
+    Runs on the event loop thread (scheduled via ``call_soon_threadsafe``)
+    so that the ``done()`` check and ``set_result`` are atomic with
+    respect to cancellation.
+    """
+    if not fut.done():
+        fut.set_result(None)
+
+
+class EventLog(Generic[T]):
+    """Append-only event buffer with cursor-based async iteration.
+
+    Multiple consumers can subscribe independently and each will see
+    every event from their starting offset onward.
+    """
+
+    __slots__ = ("_items", "_closed", "_error", "_waiters", "_lock")
+
+    def __init__(self) -> None:
+        self._items: list[T] = []
+        self._closed = False
+        self._error: BaseException | None = None
+        self._waiters: list[asyncio.Future[None]] = []
+        self._lock = threading.Lock()
+
+    # -- Producer API -------------------------------------------------------
+
+    def append(self, item: T) -> None:
+        """Append an event and wake all waiting consumers."""
+        with self._lock:
+            if self._closed:
+                raise RuntimeError("EventLog is closed")
+            self._items.append(item)
+            self._wake_all()
+
+    def close(self) -> None:
+        """Mark the log as complete.  Iterators will end gracefully."""
+        with self._lock:
+            self._closed = True
+            self._wake_all()
+
+    def fail(self, error: BaseException) -> None:
+        """Mark the log as failed.  Iterators will raise *error*."""
+        with self._lock:
+            self._error = error
+            self._closed = True
+            self._wake_all()
+
+    # -- Consumer API -------------------------------------------------------
+
+    def __aiter__(self) -> _Cursor[T]:
+        """Return a fresh cursor from the beginning of the log."""
+        return _Cursor(self)
+
+    # -- Inspection ---------------------------------------------------------
+
+    def __len__(self) -> int:
+        return len(self._items)
+
+    def __getitem__(self, index: int) -> T:
+        return self._items[index]
+
+    @property
+    def closed(self) -> bool:
+        return self._closed
+
+    # -- Internal -----------------------------------------------------------
+
+    def _wake_all(self) -> None:
+        for fut in self._waiters:
+            try:
+                fut.get_loop().call_soon_threadsafe(_resolve_future, fut)
+            except RuntimeError:
+                # Loop already closed — ignore.
+                pass
+        self._waiters.clear()
+
+
+class _Cursor(Generic[T]):
+    """An independent async iterator over an :class:`EventLog`."""
+
+    __slots__ = ("_log", "_offset")
+
+    def __init__(self, log: EventLog[T]) -> None:
+        self._log = log
+        self._offset = 0
+
+    def __aiter__(self) -> _Cursor[T]:
+        return self
+
+    async def __anext__(self) -> T:
+        while True:
+            with self._log._lock:
+                if self._offset < len(self._log._items):
+                    item = self._log._items[self._offset]
+                    self._offset += 1
+                    return item
+                if self._log._error is not None:
+                    raise self._log._error
+                if self._log._closed:
+                    raise StopAsyncIteration
+                # Nothing available yet — register a waiter
+                fut: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+                self._log._waiters.append(fut)
+            # Wait outside the lock
+            try:
+                await fut
+            except asyncio.CancelledError:
+                with self._log._lock:
+                    try:
+                        self._log._waiters.remove(fut)
+                    except ValueError:
+                        pass  # Already removed by _wake_all
+                raise
+
+
+__all__ = ["EventLog"]

libs/langgraph/langgraph/stream/_mux.py

@@ -0,0 +1,425 @@
+"""Central event dispatcher with transformer pipeline for StreamingHandler.
+
+``StreamMux`` is the sync-safe core: it holds the main
+:class:`EventLog`, tracks discovered namespaces for subgraph stream
+creation, and pipes every event through the registered
+:class:`StreamTransformer` pipeline before appending it to the log.
+
+``AsyncStreamMux`` extends the base with async subscription endpoints
+(output futures, namespace waiters, filtered event iteration).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from collections.abc import AsyncIterator
+from typing import Any
+
+from langgraph.stream._event_log import EventLog
+from langgraph.stream._types import InterruptPayload, ProtocolEvent, StreamTransformer
+from langgraph.stream.stream_channel import StreamChannel, is_stream_channel
+
+
+class StreamMux:
+    """Sync-safe event dispatcher for the StreamingHandler infrastructure.
+
+    The mux owns the main event log, applies the transformer pipeline to
+    every incoming event, and tracks namespace discovery and latest values.
+
+    For async subscription endpoints (output futures, namespace waiters,
+    filtered event iteration), use :class:`AsyncStreamMux`.
+    """
+
+    def __init__(self, transformers: list[StreamTransformer] | None = None) -> None:
+        self._event_log: EventLog[ProtocolEvent] = EventLog()
+        self._transformers: list[StreamTransformer] = list(transformers or [])
+        self._channels: list[StreamChannel[Any]] = []
+        self._current_namespace: list[str] = []
+        self._next_emit_seq: int = 0
+
+        # Namespace discovery: maps top-level ns segment → True
+        self._discovered_ns: dict[str, bool] = {}
+
+        # Latest values per namespace (list-of-strings key)
+        self._latest_values: dict[str, Any] = {}
+
+        # Interrupt tracking
+        self._interrupts: list[InterruptPayload] = []
+        self._interrupted = False
+
+        # Closed state
+        self._closed = False
+        self._error: BaseException | None = None
+
+    # -- Producer API -------------------------------------------------------
+
+    def push(self, event: ProtocolEvent) -> None:
+        """Push an event through the transformer pipeline and into the log.
+
+        Each registered transformer's ``process()`` is called in order.
+        If any transformer returns ``False``, the event is suppressed
+        (not appended to the main log).
+        """
+        if self._closed:
+            return
+
+        # Mux is the sole seq assigner — ensures all events in the log
+        # (including those from StreamChannel forwarders) share a single
+        # monotonically increasing counter.
+        event["seq"] = self._next_emit_seq
+        self._next_emit_seq += 1
+
+        # Track namespace
+        ns = event["params"].get("namespace", [])
+        if ns:
+            top_segment = ns[0]
+            if top_segment not in self._discovered_ns:
+                self._discovered_ns[top_segment] = True
+                self._on_ns_discovered(top_segment)
+
+        # Track values
+        if event["method"] == "values":
+            ns_key = _ns_key(ns)
+            self._latest_values[ns_key] = event["params"]["data"]
+
+        # Track interrupts from values events
+        if event["method"] == "values":
+            data = event["params"]["data"]
+            if isinstance(data, dict) and "__interrupt__" in data:
+                interrupt_info = data["__interrupt__"]
+                if isinstance(interrupt_info, (list, tuple)):
+                    for item in interrupt_info:
+                        iid = getattr(item, "id", None) or str(id(item))
+                        self._interrupts.append(
+                            InterruptPayload(
+                                interrupt_id=iid,
+                                payload=item,
+                            )
+                        )
+                    self._interrupted = True
+
+        # Run transformer pipeline
+        self._current_namespace = ns
+        keep = True
+        for transformer in self._transformers:
+            result = transformer.process(event)
+            if result is False:
+                keep = False
+        self._current_namespace = []
+
+        # Append to main log if not suppressed
+        if keep:
+            self._event_log.append(event)
+
+    def close(self, output: Any = None) -> None:
+        """Close the mux, finalizing transformers and the event log."""
+        if self._closed:
+            return
+        self._closed = True
+
+        # Finalize transformers (optional method)
+        for transformer in self._transformers:
+            if hasattr(transformer, "finalize"):
+                transformer.finalize()
+
+        # Close wired channels
+        for channel in self._channels:
+            channel._close()
+
+        # Close the event log
+        self._event_log.close()
+
+    def fail(self, error: BaseException) -> None:
+        """Fail the mux, propagating the error to transformers and channels."""
+        if self._closed:
+            return
+        self._closed = True
+        self._error = error
+
+        # Fail transformers (optional method)
+        for transformer in self._transformers:
+            if hasattr(transformer, "fail"):
+                transformer.fail(error)
+
+        # Fail wired channels
+        for channel in self._channels:
+            channel._fail(error)
+
+        # Fail the event log
+        self._event_log.fail(error)
+
+    # -- Inspection ---------------------------------------------------------
+
+    @property
+    def interrupted(self) -> bool:
+        return self._interrupted
+
+    @property
+    def interrupts(self) -> list[InterruptPayload]:
+        return list(self._interrupts)
+
+    @property
+    def event_log(self) -> EventLog[ProtocolEvent]:
+        return self._event_log
+
+    def get_latest_values(self, ns: list[str] | None = None) -> Any:
+        """Return the most recent values for a namespace."""
+        return self._latest_values.get(_ns_key(ns or []))
+
+    # -- Internal -----------------------------------------------------------
+
+    def _on_ns_discovered(self, segment: str) -> None:
+        """Hook called when a new top-level namespace segment is discovered.
+
+        The base implementation is a no-op.  :class:`AsyncStreamMux`
+        overrides this to wake namespace waiters.
+        """
+
+    def register_transformer(self, transformer: StreamTransformer) -> None:
+        """Register a new transformer and replay all buffered events through it.
+
+        This is the safe way to add a late-arriving transformer after the mux
+        has already started processing events.  The sequence is:
+
+        1. Snapshot the current log length (no await → no gap possible in
+           asyncio's cooperative threading model).
+        2. Append the transformer so future ``push()`` calls reach it.
+        3. Replay events ``[0, snapshot)`` through the transformer.
+        4. If the mux is already closed, call ``finalize()`` immediately so
+           the transformer's log/channel terminates cleanly.
+
+        ``process()`` is only called for events whose namespace starts with
+        any prefix — callers that need namespace filtering should do so inside
+        their ``process()`` implementation, or wrap this call with their own
+        filtering logic.
+        """
+        snapshot = len(self._event_log)
+        self._transformers.append(transformer)
+        for i in range(snapshot):
+            transformer.process(self._event_log[i])
+        if self._closed:
+            if hasattr(transformer, "finalize"):
+                transformer.finalize()
+
+    def wire_channels(self, projection: Any) -> None:
+        """Scan *projection* for :class:`StreamChannel` instances and wire them.
+
+        For each ``StreamChannel`` found, registers a push callback that
+        appends a :class:`ProtocolEvent` directly to the main event log
+        with ``method`` set to the channel's name.
+
+        Channel events bypass the transformer pipeline (matching the JS
+        implementation).  They are visible to raw event iteration and
+        remote SDK clients but not to other transformers' ``process()``.
+        """
+        if projection is None:
+            return
+        items: dict[str, Any] = {}
+        if isinstance(projection, dict):
+            items = projection
+        elif hasattr(projection, "__dict__"):
+            items = vars(projection)
+        for _key, value in items.items():
+            if is_stream_channel(value):
+                channel: StreamChannel[Any] = value
+                self._channels.append(channel)
+
+                def _make_forwarder(ch: StreamChannel[Any]) -> Any:
+                    def _forward(item: Any) -> None:
+                        if self._closed:
+                            return
+                        # Append directly to the event log, bypassing
+                        # the transformer pipeline.  This matches the JS
+                        # implementation and avoids re-entrancy bugs
+                        # (namespace clobbering, infinite recursion).
+                        self._event_log.append(
+                            ProtocolEvent(
+                                type="event",
+                                seq=self._next_emit_seq,
+                                method=ch.channel_name,
+                                params={
+                                    "namespace": list(self._current_namespace),
+                                    "timestamp": int(time.time() * 1000),
+                                    "data": item,
+                                },
+                            )
+                        )
+                        self._next_emit_seq += 1
+
+                    return _forward
+
+                channel._wire(_make_forwarder(channel))
+
+
+class AsyncStreamMux(StreamMux):
+    """Async extension of :class:`StreamMux`.
+
+    Adds output futures, namespace waiters, and async subscription
+    endpoints (``subscribe_events``, ``subscribe_subgraphs``,
+    ``get_output_future``).
+    """
+
+    def __init__(self, transformers: list[StreamTransformer] | None = None) -> None:
+        super().__init__(transformers)
+        # Waiters for new namespace discovery
+        self._ns_waiters: list[asyncio.Future[None]] = []
+        # Output promise tracking
+        self._output_futures: dict[str, asyncio.Future[Any]] = {}
+
+    # -- Producer API overrides ---------------------------------------------
+
+    def close(self, output: Any = None) -> None:
+        """Close the mux, resolving all output futures."""
+        if self._closed:
+            return
+
+        # Let the base class finalize transformers, channels, and event log
+        super().close(output)
+
+        # Resolve output futures
+        for ns_key, fut in self._output_futures.items():
+            if not fut.done():
+                value = self._latest_values.get(ns_key)
+                try:
+                    fut.get_loop().call_soon_threadsafe(fut.set_result, value)
+                except RuntimeError:
+                    pass
+
+        # Wake namespace waiters
+        self._wake_ns_waiters()
+
+    def fail(self, error: BaseException) -> None:
+        """Fail the mux, rejecting all output futures."""
+        if self._closed:
+            return
+
+        # Let the base class fail transformers, channels, and event log
+        super().fail(error)
+
+        # Reject output futures
+        for fut in self._output_futures.values():
+            if not fut.done():
+                try:
+                    fut.get_loop().call_soon_threadsafe(fut.set_exception, error)
+                except RuntimeError:
+                    pass
+
+        # Wake namespace waiters
+        self._wake_ns_waiters()
+
+    # -- Consumer API -------------------------------------------------------
+
+    def subscribe_events(
+        self, path: list[str] | None = None
+    ) -> AsyncIterator[ProtocolEvent]:
+        """Return an async iterator over events matching *path*.
+
+        If *path* is ``None`` or empty, all events are yielded.
+        Otherwise, only events whose namespace starts with *path*
+        are yielded.
+        """
+        cursor = aiter(self._event_log)
+        if not path:
+            return cursor
+        return _FilteredEventIterator(cursor, path)
+
+    async def subscribe_subgraphs(
+        self, path: list[str] | None = None, offset: int = 0
+    ) -> AsyncIterator[str]:
+        """Yield top-level namespace segments as they are discovered.
+
+        Each yielded value is the first namespace segment of a newly
+        discovered subgraph (e.g. ``"agent:0"``).
+        """
+        yielded: set[str] = set()
+        while True:
+            # Yield any newly discovered namespaces
+            for ns_segment in list(self._discovered_ns):
+                if ns_segment not in yielded:
+                    # Filter by path prefix if specified
+                    if path:
+                        if not ns_segment.startswith(path[0]):
+                            continue
+                    yielded.add(ns_segment)
+                    yield ns_segment
+
+            if self._closed:
+                return
+
+            # Wait for new namespaces
+            loop = asyncio.get_running_loop()
+            fut: asyncio.Future[None] = loop.create_future()
+            self._ns_waiters.append(fut)
+            await fut
+
+    def get_output_future(self, ns: list[str] | None = None) -> asyncio.Future[Any]:
+        """Get or create an output future for a namespace.
+
+        The future resolves to the latest ``values`` event data when
+        the mux is closed.
+        """
+        ns_key = _ns_key(ns or [])
+        if ns_key not in self._output_futures:
+            loop = asyncio.get_running_loop()
+            self._output_futures[ns_key] = loop.create_future()
+
+            # If already closed, resolve immediately
+            if self._closed:
+                value = self._latest_values.get(ns_key)
+                if self._error is not None:
+                    self._output_futures[ns_key].set_exception(self._error)
+                else:
+                    self._output_futures[ns_key].set_result(value)
+
+        return self._output_futures[ns_key]
+
+    # -- Internal -----------------------------------------------------------
+
+    def _on_ns_discovered(self, segment: str) -> None:
+        """Wake namespace waiters when a new namespace is discovered."""
+        self._wake_ns_waiters()
+
+    def _wake_ns_waiters(self) -> None:
+        for fut in self._ns_waiters:
+            if not fut.done():
+                try:
+                    fut.get_loop().call_soon_threadsafe(fut.set_result, None)
+                except RuntimeError:
+                    pass
+        self._ns_waiters.clear()
+
+
+class _FilteredEventIterator:
+    """Async iterator that filters events by namespace prefix."""
+
+    __slots__ = ("_cursor", "_path")
+
+    def __init__(self, cursor: AsyncIterator[ProtocolEvent], path: list[str]) -> None:
+        self._cursor = cursor
+        self._path = path
+
+    def __aiter__(self) -> _FilteredEventIterator:
+        return self
+
+    async def __anext__(self) -> ProtocolEvent:
+        while True:
+            event = await self._cursor.__anext__()
+            ns = event["params"].get("namespace", [])
+            if _ns_starts_with(ns, self._path):
+                return event
+
+
+def _ns_key(ns: list[str] | tuple[str, ...]) -> str:
+    """Convert a namespace list to a hashable key."""
+    return "|".join(ns)
+
+
+def _ns_starts_with(ns: list[str], prefix: list[str]) -> bool:
+    """Check if *ns* starts with *prefix*."""
+    if len(ns) < len(prefix):
+        return False
+    return ns[: len(prefix)] == prefix
+
+
+__all__ = ["AsyncStreamMux", "StreamMux"]

libs/langgraph/langgraph/stream/_types.py

@@ -0,0 +1,167 @@
+"""Protocol types for StreamingHandler.
+
+Re-exports CDDL-derived types from ``langchain-protocol`` and defines
+in-process-only types needed by the LangGraph streaming infrastructure.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Protocol, runtime_checkable
+
+# ---------------------------------------------------------------------------
+# Re-exports from langchain-protocol (CDDL-derived)
+# ---------------------------------------------------------------------------
+# Primitives
+# Content blocks
+# Messages data
+# Tools data
+from langchain_protocol import (
+    Annotation,
+    Citation,
+    ContentBlock,
+    ContentBlockDeltaData,
+    ContentBlockFinishData,
+    ContentBlockStartData,
+    FinalizedContentBlock,
+    FinishReason,
+    InvalidToolCallBlock,
+    MessageErrorData,
+    MessageFinishData,
+    MessageMetadata,
+    MessageRole,
+    MessagesData,
+    MessageStartData,
+    MetadataScalar,
+    Namespace,
+    ReasoningBlock,
+    TextBlock,
+    ToolCallBlock,
+    ToolCallChunkBlock,
+    ToolErrorData,
+    ToolFinishedData,
+    ToolOutputDeltaData,
+    ToolsData,
+    ToolStartedData,
+    UsageInfo,
+)
+from typing_extensions import NotRequired, TypedDict
+
+# ---------------------------------------------------------------------------
+# In-process types (not in the CDDL spec)
+# ---------------------------------------------------------------------------
+
+
+class _ProtocolEventParams(TypedDict):
+    """Payload envelope for a :class:`ProtocolEvent`."""
+
+    namespace: Namespace
+    timestamp: int
+    node: NotRequired[str]
+    data: Any
+
+
+class ProtocolEvent(TypedDict):
+    """A single protocol event emitted by the StreamingHandler infrastructure.
+
+    ``method`` corresponds to a
+    :pydata:`~langgraph.types.StreamMode` value (``"messages"``,
+    ``"updates"``, etc.).
+    """
+
+    type: str  # always "event"
+    seq: NotRequired[int]  # assigned by StreamMux.push(); absent before push()
+    method: str  # StreamMode value
+    params: _ProtocolEventParams
+
+
+@runtime_checkable
+class StreamTransformer(Protocol):
+    """Extension point for custom stream projections.
+
+    Implementations are registered with ``StreamingHandler`` and receive every
+    :class:`ProtocolEvent` before it is appended to the event log.
+
+    Any :class:`~langgraph.stream.stream_channel.StreamChannel` instances
+    returned by ``init()`` are automatically wired to the protocol event
+    stream by the mux.
+
+    """
+
+    def init(self) -> Any:
+        """Return the initial projection value.
+
+        Called once before the run.  Any
+        :class:`~langgraph.stream.stream_channel.StreamChannel` instances
+        in the return value are automatically wired by the mux.
+        """
+        ...
+
+    def process(self, event: ProtocolEvent) -> bool:
+        """Process an event.
+
+        Return ``True`` to keep the event in the log, ``False`` to suppress
+        it.
+        """
+        ...
+
+    def finalize(self) -> None:
+        """Called once when the run completes successfully.
+
+        Optional — the mux auto-closes any :class:`StreamChannel` instances,
+        so transformers that only use channels can omit this.
+        """
+        ...
+
+    def fail(self, err: BaseException) -> None:
+        """Called once when the run fails.
+
+        Optional — the mux auto-fails any :class:`StreamChannel` instances,
+        so transformers that only use channels can omit this.
+        """
+        ...
+
+
+class InterruptPayload(TypedDict):
+    """An interrupt produced during a StreamingHandler run."""
+
+    interrupt_id: str
+    payload: Any
+
+
+__all__ = [
+    # Primitives (re-exported)
+    "Namespace",
+    "MessageRole",
+    "MessageMetadata",
+    "MetadataScalar",
+    # Content blocks (re-exported)
+    "TextBlock",
+    "ReasoningBlock",
+    "ToolCallBlock",
+    "ToolCallChunkBlock",
+    "InvalidToolCallBlock",
+    "ContentBlock",
+    "FinalizedContentBlock",
+    "Annotation",
+    "Citation",
+    # Messages data (re-exported)
+    "MessagesData",
+    "MessageStartData",
+    "ContentBlockStartData",
+    "ContentBlockDeltaData",
+    "ContentBlockFinishData",
+    "MessageFinishData",
+    "MessageErrorData",
+    "FinishReason",
+    "UsageInfo",
+    # Tools data (re-exported)
+    "ToolsData",
+    "ToolStartedData",
+    "ToolOutputDeltaData",
+    "ToolFinishedData",
+    "ToolErrorData",
+    # In-process types
+    "ProtocolEvent",
+    "StreamTransformer",
+    "InterruptPayload",
+]

libs/langgraph/langgraph/stream/chat_model_stream.py

@@ -0,0 +1,402 @@
+"""Per-message streaming objects for StreamingHandler.
+
+``ChatModelStream`` is the synchronous variant returned by
+``GraphRunStream.messages``.  Properties (``.text``, ``.reasoning``,
+``.usage``) return final accumulated values.
+
+``AsyncChatModelStream`` is the asynchronous variant returned by
+``AsyncGraphRunStream.messages``.  Projections are dual
+async-iterable + awaitable (e.g. ``async for delta in msg.text``
+or ``full = await msg.text``).
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable, Generator, Iterator
+from typing import Any
+
+from langgraph.stream._types import UsageInfo
+
+
+# ---------------------------------------------------------------------------
+# Sync dual projection — iterable of deltas, str() for accumulated text
+# ---------------------------------------------------------------------------
+
+
+class _SyncDualProjection:
+    """Pump-driven sync iterable of string deltas.
+
+    Iterating yields incremental text fragments as the pump delivers
+    new ``content-block-delta`` events.  Calling ``str()`` drains the
+    pump and returns the full accumulated string.
+
+    This is the sync counterpart of :class:`_DualProjection` (the async
+    variant used by ``AsyncChatModelStream``).
+    """
+
+    __slots__ = ("_stream", "_attr", "_pump_one")
+
+    def __init__(
+        self,
+        stream: ChatModelStream,
+        attr: str,
+        pump_one: Callable[[], bool],
+    ) -> None:
+        self._stream = stream
+        self._attr = attr
+        self._pump_one = pump_one
+
+    def __iter__(self) -> Iterator[str]:
+        prev_len = 0
+        while True:
+            cur = getattr(self._stream, self._attr)
+            if len(cur) > prev_len:
+                yield cur[prev_len:]
+                prev_len = len(cur)
+            if self._stream._done:
+                return
+            if not self._pump_one():
+                # Source exhausted — yield any remaining
+                cur = getattr(self._stream, self._attr)
+                if len(cur) > prev_len:
+                    yield cur[prev_len:]
+                return
+
+    def __str__(self) -> str:
+        while not self._stream._done:
+            if not self._pump_one():
+                break
+        return getattr(self._stream, self._attr)
+
+    def __repr__(self) -> str:
+        return repr(getattr(self._stream, self._attr))
+
+    def __bool__(self) -> bool:
+        return bool(getattr(self._stream, self._attr))
+
+
+# ---------------------------------------------------------------------------
+# Sync variant
+# ---------------------------------------------------------------------------
+
+
+class ChatModelStream:
+    """Synchronous per-message object for a single LLM response.
+
+    Created by :class:`~langgraph.stream.transformers.MessagesTransformer`
+    and yielded by ``GraphRunStream.messages``.  By the time the sync
+    iterator yields a ``ChatModelStream``, the message lifecycle is
+    complete and all properties contain their final values.
+
+    Projections:
+
+    - ``.text`` — accumulated text content (``str``)
+    - ``.reasoning`` — accumulated reasoning content (``str``)
+    - ``.usage`` — :class:`UsageInfo` or ``None``
+    - ``.namespace`` / ``.node`` — provenance metadata
+    """
+
+    def __init__(
+        self,
+        *,
+        namespace: list[str] | None = None,
+        node: str | None = None,
+        message_id: str | None = None,
+    ) -> None:
+        self._namespace = namespace or []
+        self._node = node
+        self._message_id = message_id
+
+        # Accumulated state
+        self._text_acc = ""
+        self._reasoning_acc = ""
+        self._usage_value: UsageInfo | None = None
+        self._done = False
+
+        # Optional pump for sync streaming (set via _bind_pump)
+        self._pump_one: Callable[[], bool] | None = None
+
+    # -- Pump binding (called by GraphRunStream) ---------------------------
+
+    def _bind_pump(self, pump_one: Callable[[], bool]) -> None:
+        """Bind a pump function for sync token-by-token streaming.
+
+        When bound, ``.text`` and ``.reasoning`` return
+        :class:`_SyncDualProjection` instances that drive the pump and
+        yield deltas as the LLM produces tokens.
+        """
+        self._pump_one = pump_one
+
+    # -- Public projections ------------------------------------------------
+
+    @property
+    def text(self) -> str | _SyncDualProjection:
+        """Text content.
+
+        When a pump is bound (sync streaming), returns a
+        :class:`_SyncDualProjection` — iterable of deltas,
+        ``str()`` for the full accumulated text.  Otherwise returns
+        the accumulated text string directly.
+        """
+        if self._pump_one is not None and not self._done:
+            return _SyncDualProjection(self, "_text_acc", self._pump_one)
+        return self._text_acc
+
+    @property
+    def reasoning(self) -> str | _SyncDualProjection:
+        """Reasoning content.
+
+        Same dual behavior as :attr:`text`.
+        """
+        if self._pump_one is not None and not self._done:
+            return _SyncDualProjection(self, "_reasoning_acc", self._pump_one)
+        return self._reasoning_acc
+
+    @property
+    def usage(self) -> UsageInfo | None:
+        """Usage info, available after the message finishes."""
+        if self._pump_one is not None and not self._done:
+            while not self._done:
+                if not self._pump_one():
+                    break
+        return self._usage_value
+
+    @property
+    def namespace(self) -> list[str]:
+        return self._namespace
+
+    @property
+    def node(self) -> str | None:
+        return self._node
+
+    @property
+    def message_id(self) -> str | None:
+        return self._message_id
+
+    @property
+    def done(self) -> bool:
+        return self._done
+
+    # -- Internal API (called by MessagesTransformer) ----------------------
+
+    def _push_content_block_delta(self, data: dict[str, Any]) -> None:
+        """Process a ``content-block-delta`` event."""
+        block = data.get("content_block", {})
+        btype = block.get("type", "")
+
+        if btype == "text":
+            delta_text = block.get("text", "")
+            if delta_text:
+                self._text_acc += delta_text
+        elif btype == "reasoning":
+            delta_r = block.get("reasoning", "")
+            if delta_r:
+                self._reasoning_acc += delta_r
+
+    def _push_content_block_finish(self, data: dict[str, Any]) -> None:
+        """Process a ``content-block-finish`` event."""
+        block = data.get("content_block", {})
+        btype = block.get("type", "")
+
+        if btype == "text":
+            full_text = block.get("text", "")
+            if full_text and full_text != self._text_acc:
+                self._text_acc = full_text
+        elif btype == "reasoning":
+            full_r = block.get("reasoning", "")
+            if full_r and full_r != self._reasoning_acc:
+                self._reasoning_acc = full_r
+
+    def _finish(self, data: dict[str, Any]) -> None:
+        """Process a ``message-finish`` event."""
+        self._done = True
+        self._usage_value = data.get("usage")
+
+    def _fail(self, error: BaseException) -> None:
+        """Process a ``message-error`` event."""
+        self._done = True
+
+
+# ---------------------------------------------------------------------------
+# Async dual-projection helpers
+# ---------------------------------------------------------------------------
+
+
+class _DualProjection:
+    """Async iterable of deltas that is also awaitable for the final value.
+
+    When iterated, yields delta values (e.g. text fragments) as they arrive.
+    When awaited, returns the accumulated final value (e.g. full text string).
+    """
+
+    def __init__(self) -> None:
+        self._deltas: list[Any] = []
+        self._done = False
+        self._error: BaseException | None = None
+        self._waiters: list[asyncio.Future[None]] = []
+        self._final_value: Any = None
+        self._final_set = False
+
+    # -- Producer API (called by AsyncChatModelStream) ---------------------
+
+    def _push(self, delta: Any) -> None:
+        """Add a new delta value."""
+        self._deltas.append(delta)
+        self._wake()
+
+    def _finish(self, accumulated: Any) -> None:
+        """Set the final accumulated value and mark as done."""
+        self._final_value = accumulated
+        self._final_set = True
+        self._done = True
+        self._wake()
+
+    def _fail(self, error: BaseException) -> None:
+        self._error = error
+        self._done = True
+        self._wake()
+
+    def _wake(self) -> None:
+        for fut in self._waiters:
+            if not fut.done():
+                try:
+                    fut.get_loop().call_soon_threadsafe(fut.set_result, None)
+                except RuntimeError:
+                    pass
+        self._waiters.clear()
+
+    # -- Async iterable (yields deltas) ------------------------------------
+
+    def __aiter__(self) -> _DualProjectionIterator:
+        return _DualProjectionIterator(self)
+
+    # -- Awaitable (returns final value) -----------------------------------
+
+    def __await__(self) -> Generator[Any, None, Any]:
+        return self._await_impl().__await__()
+
+    async def _await_impl(self) -> Any:
+        while not self._final_set:
+            if self._error is not None:
+                raise self._error
+            loop = asyncio.get_running_loop()
+            fut: asyncio.Future[None] = loop.create_future()
+            self._waiters.append(fut)
+            await fut
+        if self._error is not None:
+            raise self._error
+        return self._final_value
+
+
+class _DualProjectionIterator:
+    """Async iterator over a :class:`_DualProjection`'s deltas."""
+
+    __slots__ = ("_proj", "_offset")
+
+    def __init__(self, proj: _DualProjection) -> None:
+        self._proj = proj
+        self._offset = 0
+
+    def __aiter__(self) -> _DualProjectionIterator:
+        return self
+
+    async def __anext__(self) -> Any:
+        while True:
+            if self._offset < len(self._proj._deltas):
+                item = self._proj._deltas[self._offset]
+                self._offset += 1
+                return item
+            if self._proj._error is not None:
+                raise self._proj._error
+            if self._proj._done:
+                raise StopAsyncIteration
+            loop = asyncio.get_running_loop()
+            fut: asyncio.Future[None] = loop.create_future()
+            self._proj._waiters.append(fut)
+            await fut
+
+
+# ---------------------------------------------------------------------------
+# Async variant
+# ---------------------------------------------------------------------------
+
+
+class AsyncChatModelStream(ChatModelStream):
+    """Asynchronous per-message streaming object for a single LLM response.
+
+    Created by :class:`~langgraph.stream.transformers.MessagesTransformer`
+    and yielded by ``AsyncGraphRunStream.messages``.  Content-block events
+    are fed into this object until ``message-finish``.
+
+    Projections:
+
+    - ``.text`` — async iterable of text deltas; awaitable for full text
+    - ``.reasoning`` — async iterable of reasoning deltas; awaitable for
+      full reasoning text
+    - ``.usage`` — awaitable for :class:`UsageInfo`
+    - ``.namespace`` / ``.node`` — provenance metadata
+    """
+
+    def __init__(
+        self,
+        *,
+        namespace: list[str] | None = None,
+        node: str | None = None,
+        message_id: str | None = None,
+    ) -> None:
+        super().__init__(namespace=namespace, node=node, message_id=message_id)
+        self._text_proj = _DualProjection()
+        self._reasoning_proj = _DualProjection()
+        self._usage_proj = _DualProjection()
+
+    # -- Public projections (override sync properties) ---------------------
+
+    @property
+    def text(self) -> _DualProjection:
+        """Text content — async iterable of deltas, awaitable for full text."""
+        return self._text_proj
+
+    @property
+    def reasoning(self) -> _DualProjection:
+        """Reasoning content — async iterable of deltas, awaitable for full text."""
+        return self._reasoning_proj
+
+    @property
+    def usage(self) -> _DualProjection:
+        """Usage info — awaitable for :class:`UsageInfo`."""
+        return self._usage_proj
+
+    # -- Internal API (extend base to also drive projections) --------------
+
+    def _push_content_block_delta(self, data: dict[str, Any]) -> None:
+        """Process a ``content-block-delta`` event."""
+        super()._push_content_block_delta(data)
+        block = data.get("content_block", {})
+        btype = block.get("type", "")
+
+        if btype == "text":
+            delta_text = block.get("text", "")
+            if delta_text:
+                self._text_proj._push(delta_text)
+        elif btype == "reasoning":
+            delta_r = block.get("reasoning", "")
+            if delta_r:
+                self._reasoning_proj._push(delta_r)
+
+    def _finish(self, data: dict[str, Any]) -> None:
+        """Process a ``message-finish`` event."""
+        super()._finish(data)
+        self._text_proj._finish(self._text_acc)
+        self._reasoning_proj._finish(self._reasoning_acc)
+        self._usage_proj._finish(self._usage_value)
+
+    def _fail(self, error: BaseException) -> None:
+        """Process a ``message-error`` event."""
+        super()._fail(error)
+        self._text_proj._fail(error)
+        self._reasoning_proj._fail(error)
+        self._usage_proj._fail(error)
+
+
+__all__ = ["AsyncChatModelStream", "ChatModelStream", "_SyncDualProjection"]

libs/langgraph/langgraph/stream/run_stream.py

@@ -0,0 +1,783 @@
+"""GraphRunStream and AsyncGraphRunStream for StreamingHandler.
+
+These are the top-level objects returned by
+``StreamingHandler.stream()`` / ``StreamingHandler.astream()``.
+They wrap a :class:`StreamMux` and expose named
+projections (``.values``, ``.messages``, ``.subgraphs``, ``.output``)
+for ergonomic consumption.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator, Callable, Iterator
+from typing import Any
+
+from langgraph.stream._convert import convert_to_protocol_event
+from langgraph.stream._event_log import EventLog
+from langgraph.stream._mux import AsyncStreamMux, StreamMux
+from langgraph.stream._types import InterruptPayload, ProtocolEvent, StreamTransformer
+from langgraph.stream.chat_model_stream import AsyncChatModelStream, ChatModelStream
+from langgraph.stream.transformers import MessagesTransformer, ValuesTransformer
+
+# ---------------------------------------------------------------------------
+# Values projection — dual async-iterable + awaitable
+# ---------------------------------------------------------------------------
+
+
+class _ValuesProjection:
+    """Async iterable of intermediate state snapshots; awaitable for final."""
+
+    def __init__(
+        self,
+        mux: AsyncStreamMux,
+        values_transformer: ValuesTransformer,
+        ns: list[str],
+        mapper: Callable[[Any], Any] | None = None,
+    ) -> None:
+        self._mux = mux
+        self._values_transformer = values_transformer
+        self._ns = ns
+        self._mapper = mapper
+
+    def __aiter__(self) -> AsyncIterator[Any]:
+        return _ValuesIterator(self._values_transformer, self._ns, self._mapper)
+
+    def __await__(self) -> Any:
+        return self._await_impl().__await__()
+
+    async def _await_impl(self) -> Any:
+        value = await self._mux.get_output_future(self._ns)
+        if value is not None and self._mapper is not None:
+            return self._mapper(value)
+        return value
+
+
+class _ValuesIterator:
+    """Filters the values log to events matching a namespace."""
+
+    def __init__(
+        self,
+        transformer: ValuesTransformer,
+        ns: list[str],
+        mapper: Callable[[Any], Any] | None = None,
+    ) -> None:
+        self._cursor = aiter(transformer.values_log)
+        self._ns = ns
+        self._mapper = mapper
+
+    def __aiter__(self) -> _ValuesIterator:
+        return self
+
+    async def __anext__(self) -> Any:
+        while True:
+            item = await self._cursor.__anext__()
+            item_ns = item.get("namespace", [])
+            if item_ns == self._ns:
+                data = item["data"]
+                if data is not None and self._mapper is not None:
+                    return self._mapper(data)
+                return data
+
+
+# ---------------------------------------------------------------------------
+# Messages projection
+# ---------------------------------------------------------------------------
+
+
+class _MessagesProjection:
+    """Async iterable of :class:`AsyncChatModelStream` instances."""
+
+    def __init__(self, messages_transformer: MessagesTransformer) -> None:
+        self._transformer = messages_transformer
+
+    def __aiter__(self) -> AsyncIterator[AsyncChatModelStream]:
+        return aiter(self._transformer.messages_log)
+
+
+# ---------------------------------------------------------------------------
+# Subgraphs projection
+# ---------------------------------------------------------------------------
+
+
+class _SubgraphsProjection:
+    """Async iterable yielding :class:`AsyncSubgraphRunStream` for each discovered subgraph."""
+
+    def __init__(self, mux: AsyncStreamMux, ns: list[str]) -> None:
+        self._mux = mux
+        self._ns = ns
+
+    async def __aiter__(self) -> AsyncIterator[AsyncSubgraphRunStream]:
+        async for segment in self._mux.subscribe_subgraphs(self._ns):
+            child_ns = self._ns + [segment]
+            child_transformers: list[StreamTransformer] = [
+                ValuesTransformer(),
+                MessagesTransformer(
+                    namespace=child_ns, stream_cls=AsyncChatModelStream
+                ),
+            ]
+            for t in child_transformers:
+                t.init()
+                self._mux.register_transformer(t)
+
+            yield AsyncSubgraphRunStream(
+                mux=self._mux,
+                namespace=child_ns,
+                transformers=child_transformers,
+            )
+
+
+# ---------------------------------------------------------------------------
+# AsyncGraphRunStream
+# ---------------------------------------------------------------------------
+
+
+class AsyncGraphRunStream:
+    """The async run stream returned by ``StreamingHandler.astream()``.
+
+    Async-iterable over all :class:`ProtocolEvent` instances.  Named
+    projections provide ergonomic access to values, messages, subgraphs,
+    and output.
+    """
+
+    def __init__(
+        self,
+        *,
+        mux: AsyncStreamMux,
+        namespace: list[str] | None = None,
+        transformers: list[StreamTransformer],
+        abort_event: asyncio.Event | None = None,
+        output_mapper: Callable[[Any], Any] | None = None,
+    ) -> None:
+        self._mux = mux
+        self._ns = namespace or []
+        self._transformers = transformers
+        self._abort_event = abort_event or asyncio.Event()
+        self._output_mapper = output_mapper
+
+    # -- Transformer lookup -------------------------------------------------
+
+    def _find_transformer(self, name: str) -> StreamTransformer | None:
+        for t in self._transformers:
+            if getattr(t, "name", None) == name:
+                return t
+        return None
+
+    # -- Raw event iteration ------------------------------------------------
+
+    def __aiter__(self) -> AsyncIterator[ProtocolEvent]:
+        return self._mux.subscribe_events(self._ns)
+
+    # -- Named projections --------------------------------------------------
+
+    @property
+    def values(self) -> _ValuesProjection:
+        """Async iterable of state snapshots; awaitable for final state."""
+        t = self._find_transformer("values")
+        return _ValuesProjection(self._mux, t, self._ns, self._output_mapper)
+
+    @property
+    def output(self) -> _ValuesProjection:
+        """Awaitable for the final output state."""
+        t = self._find_transformer("values")
+        return _ValuesProjection(self._mux, t, self._ns, self._output_mapper)
+
+    @property
+    def messages(self) -> _MessagesProjection:
+        """Async iterable of :class:`AsyncChatModelStream` instances."""
+        t = self._find_transformer("messages")
+        return _MessagesProjection(t)
+
+    def messages_from(self, node: str) -> _MessagesProjection:
+        """Async iterable of messages from a specific node."""
+        filtered = MessagesTransformer(
+            namespace=self._ns,
+            node_filter=node,
+            stream_cls=AsyncChatModelStream,
+        )
+        self._mux.register_transformer(filtered)
+        return _MessagesProjection(filtered)
+
+    @property
+    def subgraphs(self) -> _SubgraphsProjection:
+        """Async iterable of :class:`AsyncSubgraphRunStream` for child graphs."""
+        return _SubgraphsProjection(self._mux, self._ns)
+
+    # -- State --------------------------------------------------------------
+
+    @property
+    def interrupted(self) -> bool:
+        return self._mux.interrupted
+
+    @property
+    def interrupts(self) -> list[InterruptPayload]:
+        return self._mux.interrupts
+
+    # -- Cancellation -------------------------------------------------------
+
+    def abort(self, reason: str | None = None) -> None:
+        """Signal cancellation of the run."""
+        self._abort_event.set()
+
+    @property
+    def signal(self) -> asyncio.Event:
+        """The underlying cancellation event."""
+        return self._abort_event
+
+    # -- Extensions ---------------------------------------------------------
+
+    @property
+    def extensions(self) -> dict[str, Any]:
+        """All transformer projections."""
+        result: dict[str, Any] = {}
+        for t in self._transformers:
+            name = getattr(t, "name", None)
+            value = getattr(t, "value", None)
+            if name is not None and value is not None:
+                result[name] = value
+        return result
+
+
+# ---------------------------------------------------------------------------
+# AsyncSubgraphRunStream
+# ---------------------------------------------------------------------------
+
+
+class AsyncSubgraphRunStream(AsyncGraphRunStream):
+    """An :class:`AsyncGraphRunStream` for a child subgraph.
+
+    Adds ``.name`` and ``.index`` parsed from the last namespace segment
+    (e.g. ``"researcher:2"`` → ``name="researcher"``, ``index=2``).
+    """
+
+    @property
+    def name(self) -> str:
+        if self._ns:
+            segment = self._ns[-1]
+            return segment.split(":")[0] if ":" in segment else segment
+        return ""
+
+    @property
+    def index(self) -> int:
+        if self._ns:
+            segment = self._ns[-1]
+            if ":" in segment:
+                try:
+                    return int(segment.split(":")[-1])
+                except ValueError:
+                    pass
+        return 0
+
+
+# ---------------------------------------------------------------------------
+# Async factory
+# ---------------------------------------------------------------------------
+
+
+async def create_async_graph_run_stream(
+    source: AsyncIterator[tuple[tuple[str, ...], str, Any]],
+    *,
+    transformers: list[StreamTransformer] | None = None,
+    abort_event: asyncio.Event | None = None,
+    output_mapper: Callable[[Any], Any] | None = None,
+) -> AsyncGraphRunStream:
+    """Create an :class:`AsyncGraphRunStream` from a raw async stream source.
+
+    1. Creates a :class:`StreamMux`
+    2. Registers built-in ``ValuesTransformer`` and ``MessagesTransformer``
+    3. Registers user-supplied transformers
+    4. Creates the root ``AsyncGraphRunStream``
+    5. Starts a background pump task that reads from *source*,
+       converts each chunk to a ``ProtocolEvent``, and pushes it
+       through the mux
+    6. Returns the ``AsyncGraphRunStream``
+    """
+    abort = abort_event or asyncio.Event()
+
+    # Built-in transformers first, then user-supplied
+    all_transformers: list[StreamTransformer] = [
+        ValuesTransformer(),
+        MessagesTransformer(stream_cls=AsyncChatModelStream),
+    ]
+    all_transformers.extend(transformers or [])
+
+    # Initialize transformers, collecting projections to wire after mux creation
+    projections: list[Any] = []
+    for t in all_transformers:
+        projection = t.init()
+        if projection is not None:
+            projections.append(projection)
+
+    mux = AsyncStreamMux(transformers=all_transformers)
+
+    # Wire any StreamChannel instances found in transformer projections
+    for projection in projections:
+        mux.wire_channels(projection)
+
+    # Create the root stream
+    run_stream = AsyncGraphRunStream(
+        mux=mux,
+        transformers=all_transformers,
+        abort_event=abort,
+        output_mapper=output_mapper,
+    )
+
+    # Start the pump task
+    async def pump() -> None:
+        try:
+            async for ns, mode, payload in source:
+                if abort.is_set():
+                    break
+                # Extract node name embedded by StreamProtocolMessagesHandler.
+                node: str | None = None
+                if (
+                    mode == "messages"
+                    and isinstance(payload, dict)
+                    and "__node__" in payload
+                ):
+                    payload = dict(payload)
+                    node = payload.pop("__node__")
+                event = convert_to_protocol_event(ns, mode, payload, node=node)
+                if event is not None:
+                    mux.push(event)
+            mux.close()
+        except Exception as exc:
+            mux.fail(exc)
+
+    asyncio.get_running_loop().create_task(pump())
+
+    return run_stream
+
+
+# ---------------------------------------------------------------------------
+# GraphRunStream — returned by StreamingHandler.stream()
+# ---------------------------------------------------------------------------
+
+
+class _PumpDrivenLog:
+    """Wraps an ``EventLog`` so that iteration drives the sync pump.
+
+    Used by :attr:`GraphRunStream.extensions` to make extension logs
+    iterable without requiring the caller to drain the stream first.
+    """
+
+    __slots__ = ("_log", "_pump_one")
+
+    def __init__(self, log: EventLog, pump_one: Callable[[], bool]) -> None:
+        self._log = log
+        self._pump_one = pump_one
+
+    def __iter__(self) -> Iterator[Any]:
+        cursor = 0
+        while True:
+            if cursor < len(self._log):
+                yield self._log[cursor]
+                cursor += 1
+            elif not self._pump_one():
+                return
+
+    def __len__(self) -> int:
+        return len(self._log)
+
+    def __getitem__(self, index: int) -> Any:
+        return self._log[index]
+
+
+class GraphRunStream:
+    """Synchronous run stream returned by ``StreamingHandler.stream()``.
+
+    All projections are blocking / sync-iterable.  Internally uses
+    the same ``StreamMux`` and transformer pipeline, but without an
+    async event loop.
+
+    The source iterator is consumed lazily: each projection pulls
+    events from the source on demand rather than eagerly buffering
+    everything upfront.  This means callers see events as soon as
+    they are produced by the underlying ``stream()`` call.
+    """
+
+    def __init__(
+        self,
+        *,
+        mux: StreamMux,
+        source: Iterator[tuple[tuple[str, ...], str, Any]],
+        namespace: list[str] | None = None,
+        transformers: list[StreamTransformer],
+        output_mapper: Callable[[Any], Any] | None = None,
+    ) -> None:
+        self._mux = mux
+        self._source = source
+        self._source_exhausted = False
+        self._ns = namespace or []
+        self._transformers = transformers
+        self._output_mapper = output_mapper
+
+    # -- Transformer lookup -------------------------------------------------
+
+    def _find_transformer(self, name: str) -> StreamTransformer | None:
+        for t in self._transformers:
+            if getattr(t, "name", None) == name:
+                return t
+        return None
+
+    # -- Lazy pump ----------------------------------------------------------
+
+    def _pump_one(self) -> bool:
+        """Pull one item from the source, convert it, and push through the mux.
+
+        Returns ``True`` if an item was consumed, ``False`` if the source
+        is exhausted (or was already exhausted).
+        """
+        if self._source_exhausted:
+            return False
+        try:
+            ns, mode, payload = next(self._source)
+        except StopIteration:
+            self._source_exhausted = True
+            self._mux.close()
+            return False
+        except Exception as exc:
+            self._source_exhausted = True
+            self._mux.fail(exc)
+            return False
+
+        node: str | None = None
+        if mode == "messages" and isinstance(payload, dict) and "__node__" in payload:
+            payload = dict(payload)
+            node = payload.pop("__node__")
+        event = convert_to_protocol_event(ns, mode, payload, node=node)
+        if event is not None:
+            self._mux.push(event)
+        return True
+
+    def _pump_all(self) -> None:
+        """Drain the source iterator completely."""
+        while self._pump_one():
+            pass
+
+    # -- Helpers ------------------------------------------------------------
+
+    def _map(self, value: Any) -> Any:
+        if value is not None and self._output_mapper is not None:
+            return self._output_mapper(value)
+        return value
+
+    # -- Raw event iteration (sync) -----------------------------------------
+
+    def __iter__(self) -> Iterator[ProtocolEvent]:
+        for event in _PumpDrivenLog(self._mux.event_log, self._pump_one):
+            ns = event["params"].get("namespace", [])
+            if not self._ns or ns[: len(self._ns)] == self._ns:
+                yield event
+
+    # -- Named projections (sync) -------------------------------------------
+
+    @property
+    def output(self) -> Any:
+        """The final output state (blocking). Drains the source."""
+        self._pump_all()
+        return self._map(self._mux.get_latest_values(self._ns))
+
+    @property
+    def values(self) -> Iterator[Any]:
+        """Sync iterable of intermediate state snapshots."""
+        t = self._find_transformer("values")
+        if t is None:
+            return
+        for item in _PumpDrivenLog(t.value, self._pump_one):
+            if item.get("namespace", []) == self._ns:
+                yield self._map(item["data"])
+
+    @property
+    def messages(self) -> Iterator[ChatModelStream]:
+        """Sync iterable of :class:`ChatModelStream` instances.
+
+        Each ``ChatModelStream`` is yielded as soon as the LLM begins
+        responding (on ``message-start``).  Its ``.text`` and
+        ``.reasoning`` properties are pump-driven
+        :class:`~langgraph.stream.chat_model_stream._SyncDualProjection`
+        instances that yield deltas as tokens arrive::
+
+            for msg in run.messages:
+                for delta in msg.text:
+                    print(delta, end="", flush=True)
+
+        If you don't need streaming, ``str(msg.text)`` pumps until
+        the message completes and returns the full text.
+
+        After each message is consumed, the pump advances through
+        non-message events (tool completions, values, etc.) so that
+        other transformer state is up-to-date before the next message
+        is yielded.  This means you can check
+        ``run.extensions["tools"]`` between messages and see inline
+        results.
+        """
+        t = self._find_transformer("messages")
+        if t is None:
+            return
+        log = t.value
+        for msg in _PumpDrivenLog(log, self._pump_one):
+            msg._bind_pump(self._pump_one)
+            yield msg
+            # Advance the pump past non-message events so other
+            # transformers have up-to-date state before the next
+            # message is yielded.
+            prev_count = len(log)
+            while len(log) == prev_count:
+                if not self._pump_one():
+                    break
+
+    # -- Subgraphs ----------------------------------------------------------
+
+    @property
+    def subgraphs(self) -> Iterator[SubgraphRunStream]:
+        """Sync iterable of :class:`SubgraphRunStream` for child graphs.
+
+        Namespaces are discovered lazily as events are pumped from the
+        source.  Each yielded stream has its own ``values``, ``messages``,
+        and ``output`` projections scoped to the child namespace.
+
+        After yielding a subgraph, the caller may consume its projections
+        (e.g. ``sub.values``), which pumps more events and can discover
+        new namespaces.  The loop re-checks for newly discovered
+        namespaces after each yield before attempting another pump.
+        """
+        yielded: set[str] = set()
+
+        while True:
+            # Yield any newly discovered namespaces.  Re-check after
+            # each yield because consuming a subgraph's projections
+            # can pump events that discover further namespaces.
+            found_new = False
+            for ns_segment in list(self._mux._discovered_ns):
+                if ns_segment in yielded:
+                    continue
+                found_new = True
+                yielded.add(ns_segment)
+                child_ns = self._ns + [ns_segment]
+                child_transformers: list[StreamTransformer] = [
+                    ValuesTransformer(),
+                    MessagesTransformer(namespace=child_ns),
+                ]
+                for t in child_transformers:
+                    t.init()
+                    self._mux.register_transformer(t)
+
+                yield SubgraphRunStream(
+                    mux=self._mux,
+                    namespace=child_ns,
+                    transformers=child_transformers,
+                    pump_one=self._pump_one,
+                    output_mapper=self._output_mapper,
+                )
+
+            if found_new:
+                continue  # re-check before pumping
+
+            # No new namespaces — pump one event
+            if not self._pump_one():
+                break
+
+    # -- State --------------------------------------------------------------
+
+    @property
+    def interrupted(self) -> bool:
+        return self._mux.interrupted
+
+    @property
+    def interrupts(self) -> list[InterruptPayload]:
+        return self._mux.interrupts
+
+    # -- Extensions ---------------------------------------------------------
+
+    @property
+    def extensions(self) -> dict[str, Any]:
+        """All transformer projections as pump-driven iterables."""
+        result: dict[str, Any] = {}
+        for t in self._transformers:
+            name = getattr(t, "name", None)
+            value = getattr(t, "value", None)
+            if name is not None and value is not None:
+                if isinstance(value, EventLog):
+                    result[name] = _PumpDrivenLog(value, self._pump_one)
+                else:
+                    result[name] = value
+        return result
+
+
+# ---------------------------------------------------------------------------
+# SubgraphRunStream — sync child stream
+# ---------------------------------------------------------------------------
+
+
+class SubgraphRunStream:
+    """Synchronous run stream for a child subgraph.
+
+    Shares the parent's :class:`StreamMux` and pump function.  Has its
+    own transformer set registered on the shared mux so that projections
+    (``values``, ``messages``, ``output``) are scoped to the child
+    namespace.
+
+    Adds ``.name`` and ``.index`` parsed from the last namespace segment
+    (e.g. ``"researcher:2"`` → ``name="researcher"``, ``index=2``).
+    """
+
+    def __init__(
+        self,
+        *,
+        mux: StreamMux,
+        namespace: list[str],
+        transformers: list[StreamTransformer],
+        pump_one: Callable[[], bool],
+        output_mapper: Callable[[Any], Any] | None = None,
+    ) -> None:
+        self._mux = mux
+        self._ns = namespace
+        self._transformers = transformers
+        self._pump_one = pump_one
+        self._output_mapper = output_mapper
+
+    # -- Identity -----------------------------------------------------------
+
+    @property
+    def name(self) -> str:
+        if self._ns:
+            segment = self._ns[-1]
+            return segment.split(":")[0] if ":" in segment else segment
+        return ""
+
+    @property
+    def index(self) -> int:
+        if self._ns:
+            segment = self._ns[-1]
+            if ":" in segment:
+                try:
+                    return int(segment.split(":")[-1])
+                except ValueError:
+                    pass
+        return 0
+
+    # -- Transformer lookup -------------------------------------------------
+
+    def _find_transformer(self, name: str) -> StreamTransformer | None:
+        for t in self._transformers:
+            if getattr(t, "name", None) == name:
+                return t
+        return None
+
+    # -- Helpers ------------------------------------------------------------
+
+    def _map(self, value: Any) -> Any:
+        if value is not None and self._output_mapper is not None:
+            return self._output_mapper(value)
+        return value
+
+    def _pump_all(self) -> None:
+        while self._pump_one():
+            pass
+
+    # -- Raw event iteration (sync) -----------------------------------------
+
+    def __iter__(self) -> Iterator[ProtocolEvent]:
+        for event in _PumpDrivenLog(self._mux.event_log, self._pump_one):
+            ns = event["params"].get("namespace", [])
+            if ns[: len(self._ns)] == self._ns:
+                yield event
+
+    # -- Named projections (sync) -------------------------------------------
+
+    @property
+    def output(self) -> Any:
+        """The final output state (blocking). Drains the source."""
+        self._pump_all()
+        return self._map(self._mux.get_latest_values(self._ns))
+
+    @property
+    def values(self) -> Iterator[Any]:
+        """Sync iterable of intermediate state snapshots."""
+        t = self._find_transformer("values")
+        if t is None:
+            return
+        for item in _PumpDrivenLog(t.value, self._pump_one):
+            if item.get("namespace", []) == self._ns:
+                yield self._map(item["data"])
+
+    @property
+    def messages(self) -> Iterator[ChatModelStream]:
+        """Sync iterable of :class:`ChatModelStream` instances.
+
+        Each ``ChatModelStream`` is yielded as soon as the LLM begins
+        responding.  See :attr:`GraphRunStream.messages` for usage.
+        """
+        t = self._find_transformer("messages")
+        if t is None:
+            return
+        log = t.value
+        for msg in _PumpDrivenLog(log, self._pump_one):
+            msg._bind_pump(self._pump_one)
+            yield msg
+            prev_count = len(log)
+            while len(log) == prev_count:
+                if not self._pump_one():
+                    break
+
+    # -- State --------------------------------------------------------------
+
+    @property
+    def interrupted(self) -> bool:
+        return self._mux.interrupted
+
+    @property
+    def interrupts(self) -> list[InterruptPayload]:
+        return self._mux.interrupts
+
+
+def create_graph_run_stream(
+    source: Iterator[tuple[tuple[str, ...], str, Any]],
+    *,
+    transformers: list[StreamTransformer] | None = None,
+    output_mapper: Callable[[Any], Any] | None = None,
+) -> GraphRunStream:
+    """Create a :class:`GraphRunStream` from a sync stream source.
+
+    The source iterator is stored on the returned stream and consumed
+    lazily as projections are iterated.
+
+    Built-in transformers (values, messages) are always registered first
+    so that user-supplied transformers see events after built-in
+    processing.
+    """
+    # Built-in transformers first, then user-supplied
+    all_transformers: list[StreamTransformer] = [
+        ValuesTransformer(),
+        MessagesTransformer(),
+    ]
+    all_transformers.extend(transformers or [])
+
+    projections: list[Any] = []
+    for t in all_transformers:
+        projection = t.init()
+        if projection is not None:
+            projections.append(projection)
+
+    mux = StreamMux(transformers=all_transformers)
+
+    for projection in projections:
+        mux.wire_channels(projection)
+
+    return GraphRunStream(
+        mux=mux,
+        source=source,
+        transformers=all_transformers,
+        output_mapper=output_mapper,
+    )
+
+
+__all__ = [
+    "AsyncGraphRunStream",
+    "AsyncSubgraphRunStream",
+    "GraphRunStream",
+    "SubgraphRunStream",
+    "create_async_graph_run_stream",
+    "create_graph_run_stream",
+]

libs/langgraph/langgraph/stream/stream_channel.py

@@ -0,0 +1,76 @@
+"""StreamChannel — typed push-based channel for StreamTransformer projections.
+
+A ``StreamChannel`` wraps an :class:`EventLog` and declares a protocol
+channel name.  When the :class:`StreamMux` detects a ``StreamChannel``
+in a transformer's ``init()`` return, it wires every ``push()`` call to
+inject a :class:`ProtocolEvent` into the main event stream using the
+channel's name as the ``method``.
+
+In-process consumers iterate the channel directly (it is an async
+iterable).  Remote SDK clients subscribe via
+``session.subscribe("custom:<channelName>")``.
+
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Callable
+from typing import Any, Generic, TypeVar
+
+from langgraph.stream._event_log import EventLog
+
+T = TypeVar("T")
+
+
+class StreamChannel(Generic[T]):
+    """A typed push-based channel that integrates with the mux.
+
+    Transformer authors create a ``StreamChannel`` in ``init()`` and
+    call ``push()`` inside ``process()`` to emit domain objects.  The
+    mux auto-wires pushes to protocol events and auto-closes/fails the
+    channel on run completion.
+    """
+
+    __slots__ = ("channel_name", "_log", "_on_push")
+
+    def __init__(self, name: str) -> None:
+        self.channel_name = name
+        self._log: EventLog[T] = EventLog()
+        self._on_push: Callable[[Any], None] | None = None
+
+    def push(self, item: T) -> None:
+        """Push an item to the channel.
+
+        If the mux has wired this channel, the push also injects a
+        protocol event into the main event stream.
+        """
+        self._log.append(item)
+        if self._on_push is not None:
+            self._on_push(item)
+
+    # -- Async iteration (in-process consumption) ---------------------------
+
+    def __aiter__(self) -> AsyncIterator[T]:
+        return aiter(self._log)
+
+    # -- Internal (called by the mux) ---------------------------------------
+
+    def _wire(self, fn: Callable[[Any], None]) -> None:
+        """Wire a callback invoked on every ``push()``.  Called by the mux."""
+        self._on_push = fn
+
+    def _close(self) -> None:
+        """Close the underlying log.  Called by the mux on normal completion."""
+        self._log.close()
+
+    def _fail(self, err: BaseException) -> None:
+        """Fail the underlying log.  Called by the mux on failure."""
+        self._log.fail(err)
+
+
+def is_stream_channel(value: object) -> bool:
+    """Check if *value* is a :class:`StreamChannel` instance."""
+    return isinstance(value, StreamChannel)
+
+
+__all__ = ["StreamChannel", "is_stream_channel"]

libs/langgraph/langgraph/stream/streaming_handler.py

@@ -0,0 +1,168 @@
+"""Experimental streaming wrapper for CompiledGraph.
+
+``StreamingHandler`` wraps a compiled graph and exposes the new streaming
+API without adding methods to the ``CompiledGraph`` class itself.
+
+Usage::
+
+    from langgraph.stream import StreamingHandler
+
+    s = StreamingHandler(graph)
+
+    # async
+    run = await s.astream(input)
+    async for msg in run.messages:
+        ...
+
+    # sync
+    run = s.stream(input)
+    for event in run:
+        ...
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Iterator, Sequence
+from typing import TYPE_CHECKING, Any, cast
+
+from langchain_core.runnables import RunnableConfig
+
+from langgraph._internal._config import patch_configurable
+from langgraph.stream._convert import STREAM_V2_MODES
+from langgraph.stream._types import StreamTransformer
+from langgraph.stream.run_stream import (
+    AsyncGraphRunStream,
+    GraphRunStream,
+    create_async_graph_run_stream,
+    create_graph_run_stream,
+)
+from langgraph.types import All
+
+if TYPE_CHECKING:
+    from langgraph.pregel import Pregel
+
+#: Config key that activates the protocol messages handler.
+#: Duplicated here to avoid a circular import with ``pregel._messages_v2``.
+PROTOCOL_MESSAGES_STREAM_KEY = "__protocol_messages_stream"
+
+
+class StreamingHandler:
+    """Experimental streaming wrapper around a compiled graph.
+
+    Provides ``.stream()`` and ``.astream()`` returning
+    :class:`GraphRunStream` / :class:`AsyncGraphRunStream` with
+    ergonomic projections (``run.values``, ``run.messages``,
+    ``run.subgraphs``, ``run.output``).
+
+    Args:
+        graph: A compiled LangGraph (``Pregel`` instance).
+    """
+
+    def __init__(self, graph: Pregel) -> None:
+        self._graph = graph
+
+    async def astream(
+        self,
+        input: Any,
+        config: RunnableConfig | None = None,
+        *,
+        context: Any | None = None,
+        interrupt_before: All | Sequence[str] | None = None,
+        interrupt_after: All | Sequence[str] | None = None,
+        debug: bool | None = None,
+        transformers: list[StreamTransformer] | None = None,
+    ) -> AsyncGraphRunStream:
+        """Stream graph execution, returning an
+        :class:`~langgraph.stream.run_stream.AsyncGraphRunStream`.
+
+        The returned stream provides ergonomic projections:
+
+        - ``await run.output`` -- final state
+        - ``async for v in run.values`` -- intermediate state snapshots
+        - ``async for msg in run.messages`` -- per-message
+          :class:`~langgraph.stream.chat_model_stream.AsyncChatModelStream`
+          objects
+        - ``async for sub in run.subgraphs`` -- child
+          :class:`~langgraph.stream.run_stream.AsyncSubgraphRunStream`
+          instances
+        - ``async for event in run`` -- raw
+          :class:`~langgraph.stream._types.ProtocolEvent` objects
+
+        Args:
+            input: The input to the graph.
+            config: The configuration to use for the run.
+            context: The static context to use for the run.
+            interrupt_before: Nodes to interrupt before.
+            interrupt_after: Nodes to interrupt after.
+            debug: Whether to emit debug events.
+            transformers: Optional user-supplied
+                :class:`~langgraph.stream._types.StreamTransformer` instances
+                for custom projections (available on ``run.extensions``).
+
+        Returns:
+            An :class:`~langgraph.stream.run_stream.AsyncGraphRunStream`.
+        """
+        merged_config = patch_configurable(config, {PROTOCOL_MESSAGES_STREAM_KEY: True})
+
+        source = cast(
+            AsyncIterator[tuple[tuple[str, ...], str, Any]],
+            self._graph.astream(
+                input,
+                merged_config,
+                context=context,
+                stream_mode=STREAM_V2_MODES,
+                subgraphs=True,
+                interrupt_before=interrupt_before,
+                interrupt_after=interrupt_after,
+                debug=debug,
+                version="v1",
+            ),
+        )
+
+        return await create_async_graph_run_stream(
+            source,
+            transformers=transformers,
+            output_mapper=self._graph._output_mapper,
+        )
+
+    def stream(
+        self,
+        input: Any,
+        config: RunnableConfig | None = None,
+        *,
+        context: Any | None = None,
+        interrupt_before: All | Sequence[str] | None = None,
+        interrupt_after: All | Sequence[str] | None = None,
+        debug: bool | None = None,
+        transformers: list[StreamTransformer] | None = None,
+    ) -> GraphRunStream:
+        """Synchronous variant of :meth:`astream`.
+
+        Returns a :class:`~langgraph.stream.run_stream.GraphRunStream`
+        immediately.  The underlying source is consumed lazily as
+        projections are iterated.
+
+        See :meth:`astream` for full documentation.
+        """
+        merged_config = patch_configurable(config, {PROTOCOL_MESSAGES_STREAM_KEY: True})
+
+        source = cast(
+            Iterator[tuple[tuple[str, ...], str, Any]],
+            self._graph.stream(
+                input,
+                merged_config,
+                context=context,
+                stream_mode=STREAM_V2_MODES,
+                subgraphs=True,
+                interrupt_before=interrupt_before,
+                interrupt_after=interrupt_after,
+                debug=debug,
+                version="v1",
+            ),
+        )
+
+        return create_graph_run_stream(
+            source,
+            transformers=transformers,
+            output_mapper=self._graph._output_mapper,
+        )

libs/langgraph/langgraph/stream/transformers.py

@@ -0,0 +1,179 @@
+"""Built-in stream transformers for StreamingHandler.
+
+``ValuesTransformer`` extracts ``values`` events and maintains the latest
+state per namespace.  ``MessagesTransformer`` groups ``messages`` events
+into :class:`ChatModelStream` instances.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from langgraph.stream._event_log import EventLog
+from langgraph.stream._types import ProtocolEvent
+from langgraph.stream.chat_model_stream import ChatModelStream
+
+# Type alias for the stream class constructor signature
+_StreamCls = type[ChatModelStream]
+
+
+class ValuesTransformer:
+    """Extracts ``values`` events and populates a values event log.
+
+    Maintains the latest state per namespace and provides a separate
+    event log that :class:`AsyncGraphRunStream` / :class:`GraphRunStream` uses for ``.values``
+    iteration.
+
+    Implements the :class:`StreamTransformer` protocol.
+    """
+
+    name = "values"
+
+    def __init__(self) -> None:
+        self._values_log: EventLog[dict[str, Any]] = EventLog()
+        self._latest: dict[str, Any] = {}
+
+    @property
+    def value(self) -> EventLog[dict[str, Any]]:
+        return self._values_log
+
+    @property
+    def values_log(self) -> EventLog[dict[str, Any]]:
+        return self._values_log
+
+    def get_latest(self, ns_key: str = "") -> Any:
+        return self._latest.get(ns_key)
+
+    def init(self) -> Any:
+        return None
+
+    def process(self, event: ProtocolEvent) -> bool:
+        if event["method"] != "values":
+            return True
+
+        ns = event["params"].get("namespace", [])
+        data = event["params"]["data"]
+        ns_key = "|".join(ns) if ns else ""
+        self._latest[ns_key] = data
+
+        # Append to the values log for iteration
+        self._values_log.append({"namespace": ns, "data": data})
+        return True
+
+    def finalize(self) -> None:
+        self._values_log.close()
+
+    def fail(self, err: BaseException) -> None:
+        self._values_log.fail(err)
+
+
+class MessagesTransformer:
+    """Groups ``messages`` events into :class:`ChatModelStream` instances.
+
+    One ``ChatModelStream`` is created per ``message-start`` event.
+    Content-block events are routed to the active stream until
+    ``message-finish`` or ``message-error`` closes it.
+
+    Implements the :class:`StreamTransformer` protocol.
+    """
+
+    name = "messages"
+
+    def __init__(
+        self,
+        *,
+        namespace: list[str] | None = None,
+        node_filter: str | None = None,
+        stream_cls: _StreamCls | None = None,
+    ) -> None:
+        self._namespace = namespace
+        self._node_filter = node_filter
+        self._stream_cls: _StreamCls = stream_cls or ChatModelStream
+
+        # Message log for .messages iteration
+        self._messages_log: EventLog[ChatModelStream] = EventLog()
+
+        # Current active stream per namespace key
+        self._active: dict[str, ChatModelStream] = {}
+
+    @property
+    def value(self) -> EventLog[ChatModelStream]:
+        return self._messages_log
+
+    @property
+    def messages_log(self) -> EventLog[ChatModelStream]:
+        return self._messages_log
+
+    def init(self) -> Any:
+        return None
+
+    def process(self, event: ProtocolEvent) -> bool:
+        if event["method"] != "messages":
+            return True
+
+        ns = event["params"].get("namespace", [])
+        node = event["params"].get("node")
+        data = event["params"]["data"]
+
+        # Apply namespace filter
+        if self._namespace is not None:
+            if ns[: len(self._namespace)] != self._namespace:
+                return True
+
+        # Apply node filter
+        if self._node_filter is not None and node != self._node_filter:
+            return True
+
+        ns_key = "|".join(ns) if ns else ""
+        event_type = data.get("event") if isinstance(data, dict) else None
+
+        if event_type == "message-start":
+            stream = self._stream_cls(
+                namespace=ns,
+                node=node,
+                message_id=data.get("message_id"),
+            )
+            self._active[ns_key] = stream
+            self._messages_log.append(stream)
+
+        elif event_type in ("content-block-delta", "content-block-start"):
+            active = self._active.get(ns_key)
+            if active is not None and event_type == "content-block-delta":
+                active._push_content_block_delta(data)
+
+        elif event_type == "content-block-finish":
+            active = self._active.get(ns_key)
+            if active is not None:
+                active._push_content_block_finish(data)
+
+        elif event_type == "message-finish":
+            active = self._active.pop(ns_key, None)
+            if active is not None:
+                active._finish(data)
+
+        elif event_type == "error":
+            active = self._active.pop(ns_key, None)
+            if active is not None:
+                msg = data.get("message", "Unknown error")
+                active._fail(RuntimeError(msg))
+
+        return True
+
+    def finalize(self) -> None:
+        # Close any remaining active streams
+        for stream in self._active.values():
+            stream._finish({"reason": "stop"})
+        self._active.clear()
+        self._messages_log.close()
+
+    def fail(self, err: BaseException) -> None:
+        for stream in self._active.values():
+            stream._fail(err)
+        self._active.clear()
+        self._messages_log.fail(err)
+
+
+__all__ = [
+    "MessagesTransformer",
+    "ValuesTransformer",
+]

libs/langgraph/tests/test_graph_run_stream.py

@@ -0,0 +1,558 @@
+import asyncio
+from typing import Annotated, Any
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage
+from pydantic import BaseModel
+from typing_extensions import TypedDict
+
+from langgraph.config import get_stream_writer
+from langgraph.graph import END, START, MessagesState, StateGraph
+from langgraph.stream import AsyncChatModelStream, StreamingHandler
+from langgraph.stream._types import ProtocolEvent
+from tests.fake_chat import FakeChatModel
+
+
+class State(TypedDict):
+    value: str
+    items: Annotated[list[str], lambda a, b: a + b]
+
+
+def make_simple_graph():
+    def node_a(state):
+        return {"value": state["value"] + "_a", "items": ["a"]}
+
+    def node_b(state):
+        return {"value": state["value"] + "_b", "items": ["b"]}
+
+    graph = StateGraph(State)
+    graph.add_node("node_a", node_a)
+    graph.add_node("node_b", node_b)
+    graph.add_edge(START, "node_a")
+    graph.add_edge("node_a", "node_b")
+    graph.add_edge("node_b", END)
+    return graph.compile()
+
+
+@pytest.mark.anyio
+async def test_output():
+    graph = make_simple_graph()
+    run = await StreamingHandler(graph).astream({"value": "x", "items": []})
+    await asyncio.sleep(0.1)
+    output = await run.output
+    assert output == {"value": "x_a_b", "items": ["a", "b"]}
+
+
+@pytest.mark.anyio
+async def test_values_iteration():
+    graph = make_simple_graph()
+    run = await StreamingHandler(graph).astream({"value": "x", "items": []})
+    await asyncio.sleep(0.1)
+
+    snapshots = []
+    async for v in run.values:
+        snapshots.append(v)
+
+    assert len(snapshots) == 3
+    assert snapshots[0]["value"] == "x"
+    assert snapshots[1]["value"] == "x_a"
+    assert snapshots[2]["value"] == "x_a_b"
+
+
+@pytest.mark.anyio
+async def test_updates_in_raw_events():
+    graph = make_simple_graph()
+    run = await StreamingHandler(graph).astream({"value": "x", "items": []})
+    await asyncio.sleep(0.1)
+
+    updates = []
+    async for event in run:
+        if event["method"] == "updates":
+            updates.append(event["params"]["data"])
+
+    assert len(updates) == 2
+    assert "node_a" in updates[0]
+    assert "node_b" in updates[1]
+
+
+@pytest.mark.anyio
+async def test_messages_with_chat_model():
+    model = FakeChatModel(messages=[AIMessage(content="Hello world")])
+
+    def agent(state):
+        return {"messages": [model.invoke(state["messages"])]}
+
+    graph = StateGraph(MessagesState)
+    graph.add_node("agent", agent)
+    graph.add_edge(START, "agent")
+    graph.add_edge("agent", END)
+    compiled = graph.compile()
+
+    run = await StreamingHandler(compiled).astream(
+        {"messages": [HumanMessage(content="hi")]}
+    )
+    await asyncio.sleep(0.1)
+
+    messages_seen = []
+    async for msg in run.messages:
+        messages_seen.append(msg)
+
+    assert len(messages_seen) >= 1
+    msg = messages_seen[0]
+    assert isinstance(msg, AsyncChatModelStream)
+    text = await msg.text
+    assert text == "Hello world"
+
+
+@pytest.mark.anyio
+async def test_custom_events():
+    def node(state):
+        writer = get_stream_writer()
+        writer("hello")
+        writer(42)
+        return {"value": state["value"] + "_a", "items": ["a"]}
+
+    graph = StateGraph(State)
+    graph.add_node("node_a", node)
+    graph.add_edge(START, "node_a")
+    graph.add_edge("node_a", END)
+    compiled = graph.compile()
+
+    run = await StreamingHandler(compiled).astream({"value": "x", "items": []})
+    await asyncio.sleep(0.1)
+
+    custom_payloads = []
+    async for event in run:
+        if event["method"] == "custom":
+            custom_payloads.append(event["params"]["data"])
+
+    assert "hello" in custom_payloads
+    assert 42 in custom_payloads
+
+
+@pytest.mark.anyio
+async def test_multiple_modes_present():
+    graph = make_simple_graph()
+    run = await StreamingHandler(graph).astream({"value": "x", "items": []})
+    await asyncio.sleep(0.1)
+
+    methods = set()
+    async for event in run:
+        methods.add(event["method"])
+
+    assert {"values", "updates", "tasks", "debug"} <= methods
+
+
+@pytest.mark.anyio
+async def test_interrupted_false():
+    graph = make_simple_graph()
+    run = await StreamingHandler(graph).astream({"value": "x", "items": []})
+    await asyncio.sleep(0.1)
+    async for _ in run:
+        pass
+    assert run.interrupted is False
+
+
+@pytest.mark.anyio
+async def test_regression_v1_stream_unchanged():
+    graph = make_simple_graph()
+    chunks = []
+    async for chunk in graph.astream(
+        {"value": "x", "items": []}, stream_mode="values", version="v1"
+    ):
+        chunks.append(chunk)
+    for chunk in chunks:
+        assert isinstance(chunk, dict)
+
+
+@pytest.mark.anyio
+async def test_regression_v2_stream_unchanged():
+    graph = make_simple_graph()
+    chunks = []
+    async for chunk in graph.astream(
+        {"value": "x", "items": []}, stream_mode="values", version="v2"
+    ):
+        chunks.append(chunk)
+    assert len(chunks) >= 1
+    for chunk in chunks:
+        assert isinstance(chunk, dict)
+        assert "type" in chunk
+        assert chunk["type"] == "values"
+
+
+@pytest.mark.anyio
+async def test_regression_invoke_unchanged():
+    graph = make_simple_graph()
+    result = await graph.ainvoke({"value": "x", "items": []})
+    assert result == {"value": "x_a_b", "items": ["a", "b"]}
+
+
+def test_sync_stream_output():
+    graph = make_simple_graph()
+    run = StreamingHandler(graph).stream({"value": "x", "items": []})
+    assert run.output == {"value": "x_a_b", "items": ["a", "b"]}
+
+
+def test_sync_stream_values():
+    graph = make_simple_graph()
+    run = StreamingHandler(graph).stream({"value": "x", "items": []})
+    snapshots = list(run.values)
+    assert len(snapshots) == 3
+    assert snapshots[0]["value"] == "x"
+    assert snapshots[2]["value"] == "x_a_b"
+
+
+def test_sync_stream_raw_events():
+    graph = make_simple_graph()
+    run = StreamingHandler(graph).stream({"value": "x", "items": []})
+    methods = {e["method"] for e in run}
+    assert {"values", "updates", "tasks", "debug"} <= methods
+
+
+# ---------------------------------------------------------------------------
+# Typed output (pydantic)
+# ---------------------------------------------------------------------------
+
+
+class ModelState(BaseModel):
+    value: str
+    items: Annotated[list[str], lambda a, b: a + b]
+
+
+def _make_model_state_graph():
+    def node_a(state):
+        return {"value": state.value + "_a", "items": ["a"]}
+
+    graph = StateGraph(ModelState)
+    graph.add_node("node_a", node_a)
+    graph.add_edge(START, "node_a")
+    graph.add_edge("node_a", END)
+    return graph.compile()
+
+
+@pytest.mark.anyio
+async def test_pydantic_output():
+    graph = _make_model_state_graph()
+    run = await StreamingHandler(graph).astream(ModelState(value="x", items=[]))
+    await asyncio.sleep(0.1)
+    output = await run.output
+    assert isinstance(output, ModelState)
+    assert output.value == "x_a"
+
+
+@pytest.mark.anyio
+async def test_pydantic_values():
+    graph = _make_model_state_graph()
+    run = await StreamingHandler(graph).astream(ModelState(value="x", items=[]))
+    await asyncio.sleep(0.1)
+    snapshots = []
+    async for v in run.values:
+        snapshots.append(v)
+    for v in snapshots:
+        assert isinstance(v, ModelState)
+
+
+def test_sync_pydantic_output():
+    graph = _make_model_state_graph()
+    run = StreamingHandler(graph).stream(ModelState(value="x", items=[]))
+    assert isinstance(run.output, ModelState)
+    assert run.output.value == "x_a"
+
+
+# ---------------------------------------------------------------------------
+# Interrupts
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.anyio
+async def test_interrupts():
+    from langgraph.checkpoint.memory import MemorySaver
+
+    from langgraph.types import interrupt
+
+    def ask_human(state: State):
+        answer = interrupt("what do you want?")
+        return {"value": state["value"] + f"_{answer}", "items": [answer]}
+
+    graph = StateGraph(State)
+    graph.add_node("ask", ask_human)
+    graph.add_edge(START, "ask")
+    graph.add_edge("ask", END)
+    compiled = graph.compile(checkpointer=MemorySaver())
+
+    config = {"configurable": {"thread_id": "t1"}}
+    run = await StreamingHandler(compiled).astream(
+        {"value": "x", "items": []}, config=config
+    )
+    await asyncio.sleep(0.1)
+    # Drain events
+    async for _ in run:
+        pass
+    assert run.interrupted is True
+    assert len(run.interrupts) > 0
+
+
+# ---------------------------------------------------------------------------
+# messages_from(node)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.anyio
+async def test_messages_from_node():
+    model = FakeChatModel(messages=[AIMessage(content="from agent")])
+
+    def agent(state):
+        return {"messages": [model.invoke(state["messages"])]}
+
+    def postprocess(state):
+        return {"messages": state["messages"]}
+
+    graph = StateGraph(MessagesState)
+    graph.add_node("agent", agent)
+    graph.add_node("postprocess", postprocess)
+    graph.add_edge(START, "agent")
+    graph.add_edge("agent", "postprocess")
+    graph.add_edge("postprocess", END)
+    compiled = graph.compile()
+
+    run = await StreamingHandler(compiled).astream(
+        {"messages": [HumanMessage(content="hi")]}
+    )
+    await asyncio.sleep(0.1)
+
+    # All messages
+    all_msgs = []
+    async for m in run.messages:
+        all_msgs.append(m)
+    assert len(all_msgs) >= 1
+    # Node provenance should be set
+    assert all_msgs[0].node == "agent"
+
+
+# ---------------------------------------------------------------------------
+# Subgraph child stream
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.anyio
+async def test_subgraph_child_output():
+    """AsyncSubgraphRunStream.output should contain the child graph's final state."""
+
+    class ChildState(TypedDict):
+        value: str
+
+    class ParentState(TypedDict):
+        value: str
+
+    def child_node(state):
+        return {"value": state["value"] + "_child"}
+
+    child_graph = StateGraph(ChildState)
+    child_graph.add_node("child_node", child_node)
+    child_graph.add_edge(START, "child_node")
+    child_graph.add_edge("child_node", END)
+    # Add the compiled child as a node — this triggers LangGraph's
+    # subgraph streaming mechanism and emits child namespace events.
+    child_compiled = child_graph.compile()
+
+    parent_graph = StateGraph(ParentState)
+    parent_graph.add_node("child_node", child_compiled)
+    parent_graph.add_edge(START, "child_node")
+    parent_graph.add_edge("child_node", END)
+    parent_compiled = parent_graph.compile()
+
+    run = await StreamingHandler(parent_compiled).astream({"value": "x"})
+    await asyncio.sleep(0.1)
+
+    subgraph_streams = []
+    async for sub in run.subgraphs:
+        subgraph_streams.append(sub)
+
+    assert len(subgraph_streams) >= 1
+    child_output = await subgraph_streams[0].output
+    assert child_output is not None
+    assert child_output["value"] == "x_child"
+
+
+# ---------------------------------------------------------------------------
+# Custom reducers / .extensions
+# ---------------------------------------------------------------------------
+
+
+class _CountTransformer:
+    """Counts events. Exposes count via .value for extensions."""
+
+    name = "event_count"
+
+    def __init__(self) -> None:
+        self.value = 0
+
+    def init(self) -> Any:
+        return None
+
+    def process(self, event: ProtocolEvent) -> bool:
+        self.value += 1
+        return True
+
+    def finalize(self) -> None:
+        pass
+
+    def fail(self, err: BaseException) -> None:
+        pass
+
+
+@pytest.mark.anyio
+async def test_custom_reducer_extensions():
+    graph = make_simple_graph()
+    counter = _CountTransformer()
+    run = await StreamingHandler(graph).astream(
+        {"value": "x", "items": []}, transformers=[counter]
+    )
+    await asyncio.sleep(0.1)
+    async for _ in run:
+        pass
+    assert counter.value > 0
+    assert run.extensions["event_count"] == counter.value
+
+
+def test_sync_custom_reducer_extensions():
+    graph = make_simple_graph()
+    counter = _CountTransformer()
+    run = StreamingHandler(graph).stream(
+        {"value": "x", "items": []}, transformers=[counter]
+    )
+    for _ in run:
+        pass
+    assert counter.value > 0
+    assert run.extensions["event_count"] == counter.value
+
+
+# ---------------------------------------------------------------------------
+# Tool transformer via extensions
+# ---------------------------------------------------------------------------
+
+
+class _ToolExecution:
+    def __init__(self, tool_call_id: str, tool_name: str, input: Any, output: Any):
+        self.tool_call_id = tool_call_id
+        self.tool_name = tool_name
+        self.input = input
+        self.output = output
+
+
+class _ToolsTransformer:
+    """Groups tool-started/tool-finished custom events into _ToolExecution objects."""
+
+    name = "tools"
+
+    def __init__(self) -> None:
+        from langgraph.stream._event_log import EventLog
+
+        self._log: EventLog[_ToolExecution] = EventLog()
+        self._pending: dict[str, dict] = {}
+        self.value = self._log
+
+    def init(self) -> Any:
+        return None
+
+    def process(self, event: ProtocolEvent) -> bool:
+        if event["method"] != "custom":
+            return True
+        data = event["params"]["data"]
+        if not isinstance(data, dict) or "event" not in data:
+            return True
+
+        tool_call_id = data.get("tool_call_id")
+        if tool_call_id is None:
+            return True
+
+        if data["event"] == "tool-started":
+            self._pending[tool_call_id] = data
+            return False
+
+        if data["event"] == "tool-finished":
+            started = self._pending.pop(tool_call_id, {})
+            self._log.append(_ToolExecution(
+                tool_call_id=tool_call_id,
+                tool_name=started.get("tool_name", ""),
+                input=started.get("input"),
+                output=data["output"],
+            ))
+            return False
+
+        return True
+
+    def finalize(self) -> None:
+        self._log.close()
+
+    def fail(self, err: BaseException) -> None:
+        self._log.fail(err)
+
+
+def _make_tool_graph():
+    """Graph: agent emits a tool call, custom_tools executes it with writer events."""
+    from langgraph.types import StreamWriter
+
+    def agent(state):
+        return {
+            "value": "called",
+            "items": ["agent"],
+        }
+
+    def custom_tools(state, *, writer: StreamWriter):
+        writer({
+            "event": "tool-started",
+            "tool_call_id": "call_1",
+            "tool_name": "get_weather",
+            "input": {"city": "SF"},
+        })
+        writer({
+            "event": "tool-finished",
+            "tool_call_id": "call_1",
+            "output": {"temp_f": 64},
+        })
+        return {"value": "done", "items": ["tools"]}
+
+    graph = StateGraph(State)
+    graph.add_node("agent", agent)
+    graph.add_node("custom_tools", custom_tools)
+    graph.add_edge(START, "agent")
+    graph.add_edge("agent", "custom_tools")
+    graph.add_edge("custom_tools", END)
+    return graph.compile()
+
+
+def test_sync_tool_transformer_via_extensions():
+    """Tool events flow through extensions and are iterable without draining raw events."""
+    graph = _make_tool_graph()
+    run = StreamingHandler(graph).stream(
+        {"value": "", "items": []},
+        transformers=[_ToolsTransformer()],
+    )
+
+    # Iterating extensions drives the pump — no need to drain raw events first
+    executions = list(run.extensions["tools"])
+    assert len(executions) == 1
+    assert executions[0].tool_name == "get_weather"
+    assert executions[0].input == {"city": "SF"}
+    assert executions[0].output == {"temp_f": 64}
+
+
+@pytest.mark.anyio
+async def test_async_tool_transformer_via_extensions():
+    """Tool events flow through extensions in async mode."""
+    graph = _make_tool_graph()
+    run = await StreamingHandler(graph).astream(
+        {"value": "", "items": []},
+        transformers=[_ToolsTransformer()],
+    )
+    await asyncio.sleep(0.1)
+
+    # Drain main stream so transformer processes all events
+    async for _ in run:
+        pass
+
+    tools_log = run.extensions["tools"]
+    assert len(tools_log) == 1
+    assert tools_log[0].tool_name == "get_weather"
+    assert tools_log[0].output == {"temp_f": 64}

libs/langgraph/tests/test_messages_v2.py

@@ -0,0 +1,531 @@
+from uuid import uuid4
+
+import pytest
+from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage
+from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, LLMResult
+
+from langgraph.constants import TAG_HIDDEN, TAG_NOSTREAM
+from langgraph.pregel._messages_v2 import StreamProtocolMessagesHandler
+from langgraph.types import Command
+
+META = {"langgraph_checkpoint_ns": "root:", "langgraph_node": "agent"}
+
+
+def make_handler(subgraphs=True):
+    events = []
+    handler = StreamProtocolMessagesHandler(events.append, subgraphs)
+    return handler, events
+
+
+def test_streamed_text():
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=META, tags=[]
+    )
+
+    for token_text in ("Hello", " ", "world"):
+        chunk = ChatGenerationChunk(
+            message=AIMessageChunk(content=token_text, id=f"run-{run_id}")
+        )
+        handler.on_llm_new_token(token_text, chunk=chunk, run_id=run_id)
+
+    final_msg = AIMessage(content="Hello world", id=f"run-{run_id}")
+    handler.on_llm_end(
+        LLMResult(generations=[[ChatGeneration(message=final_msg)]]),
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    assert data_events[0]["event"] == "message-start"
+    assert data_events[1]["event"] == "content-block-start"
+    assert data_events[1]["index"] == 0
+
+    deltas = [d for d in data_events if d["event"] == "content-block-delta"]
+    assert len(deltas) == 3
+    assert deltas[0]["content_block"]["text"] == "Hello"
+    assert deltas[1]["content_block"]["text"] == " "
+    assert deltas[2]["content_block"]["text"] == "world"
+
+    finish_blocks = [d for d in data_events if d["event"] == "content-block-finish"]
+    assert len(finish_blocks) == 1
+    assert finish_blocks[0]["content_block"]["text"] == "Hello world"
+
+    assert data_events[-1]["event"] == "message-finish"
+    assert data_events[-1]["reason"] == "stop"
+
+
+def test_tool_calls():
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=META, tags=[]
+    )
+
+    chunk1 = ChatGenerationChunk(
+        message=AIMessageChunk(
+            content="",
+            tool_call_chunks=[
+                {"name": "search", "args": '{"q', "id": "call_1", "index": 0}
+            ],
+            id=f"run-{run_id}",
+        )
+    )
+    handler.on_llm_new_token("", chunk=chunk1, run_id=run_id)
+
+    chunk2 = ChatGenerationChunk(
+        message=AIMessageChunk(
+            content="",
+            tool_call_chunks=[
+                {"name": None, "args": 'uery":"hi"}', "id": None, "index": 0}
+            ],
+            id=f"run-{run_id}",
+        )
+    )
+    handler.on_llm_new_token("", chunk=chunk2, run_id=run_id)
+
+    final_msg = AIMessage(
+        content="",
+        tool_calls=[{"name": "search", "args": {"query": "hi"}, "id": "call_1"}],
+        id=f"run-{run_id}",
+    )
+    handler.on_llm_end(
+        LLMResult(generations=[[ChatGeneration(message=final_msg)]]),
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    finish_blocks = [d for d in data_events if d["event"] == "content-block-finish"]
+    assert len(finish_blocks) == 1
+    fb = finish_blocks[0]["content_block"]
+    assert fb["type"] == "tool_call"
+    assert fb["args"] == {"query": "hi"}
+    assert fb["name"] == "search"
+    assert fb["id"] == "call_1"
+
+
+def test_invalid_tool_call_json():
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=META, tags=[]
+    )
+
+    chunk = ChatGenerationChunk(
+        message=AIMessageChunk(
+            content="",
+            tool_call_chunks=[
+                {
+                    "name": "search",
+                    "args": "{not valid json",
+                    "id": "call_2",
+                    "index": 0,
+                }
+            ],
+            id=f"run-{run_id}",
+        )
+    )
+    handler.on_llm_new_token("", chunk=chunk, run_id=run_id)
+
+    final_msg = AIMessage(content="", id=f"run-{run_id}")
+    handler.on_llm_end(
+        LLMResult(generations=[[ChatGeneration(message=final_msg)]]),
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    finish_blocks = [d for d in data_events if d["event"] == "content-block-finish"]
+    assert len(finish_blocks) == 1
+    fb = finish_blocks[0]["content_block"]
+    assert fb["type"] == "invalid_tool_call"
+    assert "Failed to parse" in fb["error"]
+
+
+def test_reasoning_blocks():
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=META, tags=[]
+    )
+
+    chunk = ChatGenerationChunk(
+        message=AIMessageChunk(
+            content=[{"type": "reasoning_content", "reasoning_content": "thinking..."}],
+            id=f"run-{run_id}",
+        )
+    )
+    handler.on_llm_new_token("", chunk=chunk, run_id=run_id)
+
+    final_msg = AIMessage(content="", id=f"run-{run_id}")
+    handler.on_llm_end(
+        LLMResult(generations=[[ChatGeneration(message=final_msg)]]),
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    block_starts = [d for d in data_events if d["event"] == "content-block-start"]
+    assert len(block_starts) == 1
+    assert block_starts[0]["content_block"]["type"] == "reasoning"
+
+    deltas = [d for d in data_events if d["event"] == "content-block-delta"]
+    assert len(deltas) == 1
+    assert deltas[0]["content_block"]["reasoning"] == "thinking..."
+
+
+def test_multiple_content_blocks():
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=META, tags=[]
+    )
+
+    chunk1 = ChatGenerationChunk(
+        message=AIMessageChunk(content="hello", id=f"run-{run_id}")
+    )
+    handler.on_llm_new_token("hello", chunk=chunk1, run_id=run_id)
+
+    chunk2 = ChatGenerationChunk(
+        message=AIMessageChunk(
+            content="",
+            tool_call_chunks=[
+                {"name": "lookup", "args": '{"x":1}', "id": "call_3", "index": 1}
+            ],
+            id=f"run-{run_id}",
+        )
+    )
+    handler.on_llm_new_token("", chunk=chunk2, run_id=run_id)
+
+    final_msg = AIMessage(
+        content="hello",
+        tool_calls=[{"name": "lookup", "args": {"x": 1}, "id": "call_3"}],
+        id=f"run-{run_id}",
+    )
+    handler.on_llm_end(
+        LLMResult(generations=[[ChatGeneration(message=final_msg)]]),
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    finish_blocks = [d for d in data_events if d["event"] == "content-block-finish"]
+    assert len(finish_blocks) == 2
+
+
+def test_usage_metadata():
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=META, tags=[]
+    )
+
+    chunk = ChatGenerationChunk(
+        message=AIMessageChunk(content="hi", id=f"run-{run_id}")
+    )
+    handler.on_llm_new_token("hi", chunk=chunk, run_id=run_id)
+
+    final_msg = AIMessage(
+        content="hi",
+        id=f"run-{run_id}",
+        usage_metadata={"input_tokens": 10, "output_tokens": 5, "total_tokens": 15},
+    )
+    handler.on_llm_end(
+        LLMResult(generations=[[ChatGeneration(message=final_msg)]]),
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    finish_event = [d for d in data_events if d["event"] == "message-finish"][0]
+    assert "usage" in finish_event
+    assert finish_event["usage"]["input_tokens"] == 10
+
+
+@pytest.mark.parametrize(
+    "raw_reason,expected",
+    [
+        ("stop", "stop"),
+        ("tool_calls", "tool_use"),
+        ("length", "length"),
+        ("content_filter", "content_filter"),
+        ("end_turn", "stop"),
+    ],
+)
+def test_finish_reason_normalization(raw_reason, expected):
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=META, tags=[]
+    )
+    chunk = ChatGenerationChunk(message=AIMessageChunk(content="x", id=f"run-{run_id}"))
+    handler.on_llm_new_token("x", chunk=chunk, run_id=run_id)
+
+    final_msg = AIMessage(
+        content="x",
+        id=f"run-{run_id}",
+        response_metadata={"finish_reason": raw_reason},
+    )
+    handler.on_llm_end(
+        LLMResult(generations=[[ChatGeneration(message=final_msg)]]),
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    finish_event = [d for d in data_events if d["event"] == "message-finish"][0]
+    assert finish_event["reason"] == expected
+
+
+def test_tag_nostream():
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=META, tags=[TAG_NOSTREAM]
+    )
+    chunk = ChatGenerationChunk(
+        message=AIMessageChunk(content="secret", id=f"run-{run_id}")
+    )
+    handler.on_llm_new_token("secret", chunk=chunk, run_id=run_id)
+
+    final_msg = AIMessage(content="secret", id=f"run-{run_id}")
+    handler.on_llm_end(
+        LLMResult(generations=[[ChatGeneration(message=final_msg)]]),
+        run_id=run_id,
+    )
+    assert events == []
+
+
+def test_tag_hidden_chain():
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chain_start(
+        serialized={},
+        inputs={},
+        run_id=run_id,
+        metadata=META,
+        tags=[TAG_HIDDEN],
+        name="agent",
+    )
+    handler.on_chain_end(
+        {"messages": [AIMessage(content="hidden", id="msg-1")]},
+        run_id=run_id,
+    )
+    assert events == []
+
+
+def test_subgraph_filtering():
+    handler, events = make_handler(subgraphs=False)
+    run_id = uuid4()
+
+    subgraph_meta = {
+        "langgraph_checkpoint_ns": "root:|child:",
+        "langgraph_node": "agent",
+    }
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=subgraph_meta, tags=[]
+    )
+    chunk = ChatGenerationChunk(
+        message=AIMessageChunk(content="sub", id=f"run-{run_id}")
+    )
+    handler.on_llm_new_token("sub", chunk=chunk, run_id=run_id)
+
+    final_msg = AIMessage(content="sub", id=f"run-{run_id}")
+    handler.on_llm_end(
+        LLMResult(generations=[[ChatGeneration(message=final_msg)]]),
+        run_id=run_id,
+    )
+    assert events == []
+
+
+def test_chain_emits_messages():
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chain_start(
+        serialized={}, inputs={}, run_id=run_id, metadata=META, tags=[], name="agent"
+    )
+    handler.on_chain_end(
+        {"messages": [AIMessage(content="hello", id="msg-chain-1")]},
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    assert len(data_events) > 0
+    assert data_events[0]["event"] == "message-start"
+    assert data_events[-1]["event"] == "message-finish"
+
+
+def test_llm_error_after_start():
+    """on_llm_error should emit a message-error event for a started stream."""
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=META, tags=[]
+    )
+
+    chunk = ChatGenerationChunk(
+        message=AIMessageChunk(content="partial", id=f"run-{run_id}")
+    )
+    handler.on_llm_new_token("partial", chunk=chunk, run_id=run_id)
+
+    handler.on_llm_error(RuntimeError("connection lost"), run_id=run_id)
+
+    data_events = [e[2] for e in events]
+    assert data_events[0]["event"] == "message-start"
+    error_events = [d for d in data_events if d["event"] == "error"]
+    assert len(error_events) == 1
+    assert "connection lost" in error_events[0]["message"]
+
+
+def test_llm_error_before_start_no_emit():
+    """on_llm_error before any tokens should not emit error events."""
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=META, tags=[]
+    )
+
+    # Error before any token — state.started is False
+    handler.on_llm_error(RuntimeError("immediate fail"), run_id=run_id)
+
+    data_events = [e[2] for e in events]
+    error_events = [d for d in data_events if d.get("event") == "error"]
+    assert len(error_events) == 0
+
+
+def test_non_streamed_model():
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id, metadata=META, tags=[]
+    )
+
+    final_msg = AIMessage(
+        content="full response",
+        id=f"run-{run_id}",
+        response_metadata={"finish_reason": "stop"},
+    )
+    handler.on_llm_end(
+        LLMResult(generations=[[ChatGeneration(message=final_msg)]]),
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    assert len(data_events) > 0
+    assert data_events[0]["event"] == "message-start"
+
+    deltas = [d for d in data_events if d["event"] == "content-block-delta"]
+    assert len(deltas) == 1
+    assert deltas[0]["content_block"]["text"] == "full response"
+
+    assert data_events[-1]["event"] == "message-finish"
+    assert data_events[-1]["reason"] == "stop"
+
+
+def test_chain_emits_command_with_message():
+    """on_chain_end should emit protocol events for messages inside a Command."""
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chain_start(
+        serialized={}, inputs={}, run_id=run_id, metadata=META, tags=[], name="agent"
+    )
+    handler.on_chain_end(
+        Command(update={"messages": [AIMessage(content="from command", id="cmd-1")]}),
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    assert len(data_events) > 0
+    assert data_events[0]["event"] == "message-start"
+    deltas = [d for d in data_events if d["event"] == "content-block-delta"]
+    assert len(deltas) == 1
+    assert deltas[0]["content_block"]["text"] == "from command"
+    assert data_events[-1]["event"] == "message-finish"
+
+
+def test_chain_emits_command_in_list():
+    """on_chain_end should handle a list containing Command objects."""
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chain_start(
+        serialized={}, inputs={}, run_id=run_id, metadata=META, tags=[], name="agent"
+    )
+    handler.on_chain_end(
+        [Command(update={"messages": [AIMessage(content="listed", id="cmd-2")]})],
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    starts = [d for d in data_events if d["event"] == "message-start"]
+    assert len(starts) == 1
+
+
+def test_chain_deduplicates_seen_messages():
+    """Messages already seen from LLM streaming should not be re-emitted by chain end."""
+    handler, events = make_handler()
+    run_id_llm = uuid4()
+    run_id_chain = uuid4()
+    msg_id = f"run-{run_id_llm}"
+
+    # Simulate LLM streaming
+    handler.on_chat_model_start(
+        serialized={}, messages=[[]], run_id=run_id_llm, metadata=META, tags=[]
+    )
+    chunk = ChatGenerationChunk(message=AIMessageChunk(content="hello", id=msg_id))
+    handler.on_llm_new_token("hello", chunk=chunk, run_id=run_id_llm)
+
+    final_msg = AIMessage(content="hello", id=msg_id)
+    handler.on_llm_end(
+        LLMResult(generations=[[ChatGeneration(message=final_msg)]]),
+        run_id=run_id_llm,
+    )
+
+    events_before = len(events)
+
+    # Now chain end with the same message ID
+    handler.on_chain_start(
+        serialized={},
+        inputs={},
+        run_id=run_id_chain,
+        metadata=META,
+        tags=[],
+        name="agent",
+    )
+    handler.on_chain_end(
+        {"messages": [AIMessage(content="hello", id=msg_id)]},
+        run_id=run_id_chain,
+    )
+
+    # No new events should have been emitted for the duplicate
+    data_events_after = [e[2] for e in events[events_before:]]
+    starts = [d for d in data_events_after if d.get("event") == "message-start"]
+    assert len(starts) == 0
+
+
+def test_chain_emits_human_message_role():
+    """Non-AI messages from chain output should have the correct role."""
+    handler, events = make_handler()
+    run_id = uuid4()
+
+    handler.on_chain_start(
+        serialized={}, inputs={}, run_id=run_id, metadata=META, tags=[], name="agent"
+    )
+    handler.on_chain_end(
+        {"messages": [HumanMessage(content="user msg", id="hmsg-1")]},
+        run_id=run_id,
+    )
+
+    data_events = [e[2] for e in events]
+    starts = [d for d in data_events if d["event"] == "message-start"]
+    assert len(starts) == 1
+    assert starts[0]["role"] == "human"

libs/langgraph/tests/test_pregel.py

@@ -1329,20 +1329,22 @@ def test_imp_nested(
     }
 
     thread1 = {"configurable": {"thread_id": "1"}}
-    assert [*graph.stream([0, 1], thread1, durability=durability)] == [
-        {"submapper": "0"},
+    result = [*graph.stream([0, 1], thread1, durability=durability)]
+    # nested tasks run concurrently so output order is non-deterministic
+    assert sorted(result[:-1], key=lambda d: str(d)) == [
         {"mapper": "00"},
-        {"submapper": "1"},
         {"mapper": "11"},
-        {
-            "__interrupt__": (
-                Interrupt(
-                    value="question",
-                    id=AnyStr(),
-                ),
-            )
-        },
+        {"submapper": "0"},
+        {"submapper": "1"},
     ]
+    assert result[-1] == {
+        "__interrupt__": (
+            Interrupt(
+                value="question",
+                id=AnyStr(),
+            ),
+        )
+    }
 
     assert graph.invoke(Command(resume="answer"), thread1, durability=durability) == [
         "00answera",

libs/langgraph/tests/test_stream_chat_model_stream.py

@@ -0,0 +1,245 @@
+import pytest
+
+from langgraph.stream.chat_model_stream import AsyncChatModelStream, ChatModelStream
+
+
+def _text_delta(text: str) -> dict:
+    return {"content_block": {"type": "text", "text": text}}
+
+
+def _reasoning_delta(text: str) -> dict:
+    return {"content_block": {"type": "reasoning", "reasoning": text}}
+
+
+# ---------------------------------------------------------------------------
+# Sync ChatModelStream tests
+# ---------------------------------------------------------------------------
+
+
+def test_sync_text_accumulates():
+    stream = ChatModelStream()
+    stream._push_content_block_delta(_text_delta("Hello"))
+    stream._push_content_block_delta(_text_delta(", world"))
+    stream._finish({"reason": "stop"})
+
+    assert stream.text == "Hello, world"
+    assert isinstance(stream.text, str)
+
+
+def test_sync_reasoning_accumulates():
+    stream = ChatModelStream()
+    stream._push_content_block_delta(_reasoning_delta("step 1"))
+    stream._push_content_block_delta(_reasoning_delta(" -> step 2"))
+    stream._finish({"reason": "stop"})
+
+    assert stream.reasoning == "step 1 -> step 2"
+    assert isinstance(stream.reasoning, str)
+
+
+def test_sync_usage():
+    stream = ChatModelStream()
+    usage = {"input_tokens": 10, "output_tokens": 5}
+    stream._finish({"reason": "stop", "usage": usage})
+    assert stream.usage == usage
+
+
+def test_sync_mixed_blocks():
+    stream = ChatModelStream()
+    stream._push_content_block_delta(_text_delta("answer"))
+    stream._push_content_block_delta(
+        {"content_block": {"type": "tool_call", "name": "search"}}
+    )
+    stream._push_content_block_delta(_text_delta(" here"))
+    stream._finish({"reason": "stop"})
+
+    assert stream.text == "answer here"
+
+
+def test_sync_tool_call_only_text_empty():
+    stream = ChatModelStream()
+    stream._push_content_block_delta(
+        {"content_block": {"type": "tool_call", "name": "search"}}
+    )
+    stream._finish({"reason": "stop"})
+    assert stream.text == ""
+
+
+def test_sync_fail_marks_done():
+    stream = ChatModelStream()
+    assert not stream.done
+    stream._fail(RuntimeError("err"))
+    assert stream.done
+
+
+def test_sync_namespace_and_node():
+    stream = ChatModelStream(
+        namespace=["agent:0", "tools:1"],
+        node="chat_model",
+        message_id="msg-123",
+    )
+    assert stream.namespace == ["agent:0", "tools:1"]
+    assert stream.node == "chat_model"
+    assert stream.message_id == "msg-123"
+
+
+def test_sync_content_block_finish_authoritative():
+    """content-block-finish with authoritative text overrides accumulated."""
+    stream = ChatModelStream()
+    stream._push_content_block_delta(_text_delta("partial"))
+    stream._push_content_block_finish(
+        {"content_block": {"type": "text", "text": "full text"}}
+    )
+    assert stream.text == "full text"
+
+
+# ---------------------------------------------------------------------------
+# Async ChatModelStream tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.anyio
+async def test_async_text_iterable_yields_deltas():
+    stream = AsyncChatModelStream()
+    stream._push_content_block_delta(_text_delta("Hello"))
+    stream._push_content_block_delta(_text_delta(", world"))
+    stream._finish({"reason": "stop"})
+
+    collected = []
+    async for delta in stream.text:
+        collected.append(delta)
+    assert collected == ["Hello", ", world"]
+
+
+@pytest.mark.anyio
+async def test_async_text_awaitable_returns_full():
+    stream = AsyncChatModelStream()
+    stream._push_content_block_delta(_text_delta("Hello"))
+    stream._push_content_block_delta(_text_delta(", world"))
+    stream._finish({"reason": "stop"})
+
+    result = await stream.text
+    assert result == "Hello, world"
+
+
+@pytest.mark.anyio
+async def test_async_reasoning_dual_pattern():
+    stream = AsyncChatModelStream()
+    stream._push_content_block_delta(_reasoning_delta("step 1"))
+    stream._push_content_block_delta(_reasoning_delta(" -> step 2"))
+    stream._finish({"reason": "stop"})
+
+    collected = []
+    async for delta in stream.reasoning:
+        collected.append(delta)
+    assert collected == ["step 1", " -> step 2"]
+
+    stream2 = AsyncChatModelStream()
+    stream2._push_content_block_delta(_reasoning_delta("thinking"))
+    stream2._finish({"reason": "stop"})
+    full = await stream2.reasoning
+    assert full == "thinking"
+
+
+@pytest.mark.anyio
+async def test_async_usage_resolves():
+    stream = AsyncChatModelStream()
+    usage = {"input_tokens": 10, "output_tokens": 5}
+    stream._finish({"reason": "stop", "usage": usage})
+    result = await stream.usage
+    assert result == usage
+
+
+@pytest.mark.anyio
+async def test_async_mixed_blocks_text_only():
+    stream = AsyncChatModelStream()
+    stream._push_content_block_delta(_text_delta("answer"))
+    stream._push_content_block_delta(
+        {"content_block": {"type": "tool_call", "name": "search"}}
+    )
+    stream._push_content_block_delta(_text_delta(" here"))
+    stream._finish({"reason": "stop"})
+
+    collected = []
+    async for delta in stream.text:
+        collected.append(delta)
+    assert collected == ["answer", " here"]
+
+
+@pytest.mark.anyio
+async def test_async_tool_call_only_text_empty():
+    stream = AsyncChatModelStream()
+    stream._push_content_block_delta(
+        {"content_block": {"type": "tool_call", "name": "search"}}
+    )
+    stream._finish({"reason": "stop"})
+    result = await stream.text
+    assert result == ""
+
+
+@pytest.mark.anyio
+async def test_async_fail_raises_on_text_await():
+    stream = AsyncChatModelStream()
+    stream._push_content_block_delta(_text_delta("partial"))
+    stream._fail(RuntimeError("model error"))
+
+    with pytest.raises(RuntimeError, match="model error"):
+        await stream.text
+
+
+@pytest.mark.anyio
+async def test_async_fail_raises_on_reasoning_await():
+    stream = AsyncChatModelStream()
+    stream._push_content_block_delta(_reasoning_delta("thinking"))
+    stream._fail(RuntimeError("model error"))
+
+    with pytest.raises(RuntimeError, match="model error"):
+        await stream.reasoning
+
+
+@pytest.mark.anyio
+async def test_async_fail_raises_on_usage_await():
+    stream = AsyncChatModelStream()
+    stream._fail(RuntimeError("model error"))
+
+    with pytest.raises(RuntimeError, match="model error"):
+        await stream.usage
+
+
+@pytest.mark.anyio
+async def test_async_fail_raises_during_text_iteration():
+    stream = AsyncChatModelStream()
+    stream._push_content_block_delta(_text_delta("partial"))
+    stream._fail(RuntimeError("model error"))
+
+    collected = []
+    with pytest.raises(RuntimeError, match="model error"):
+        async for delta in stream.text:
+            collected.append(delta)
+    assert collected == ["partial"]
+
+
+@pytest.mark.anyio
+async def test_async_fail_marks_done():
+    stream = AsyncChatModelStream()
+    assert not stream.done
+    stream._fail(RuntimeError("err"))
+    assert stream.done
+
+
+@pytest.mark.anyio
+async def test_async_namespace_and_node():
+    stream = AsyncChatModelStream(
+        namespace=["agent:0", "tools:1"],
+        node="chat_model",
+        message_id="msg-123",
+    )
+    assert stream.namespace == ["agent:0", "tools:1"]
+    assert stream.node == "chat_model"
+    assert stream.message_id == "msg-123"
+
+
+@pytest.mark.anyio
+async def test_async_inherits_from_sync():
+    """AsyncChatModelStream is a subclass of ChatModelStream."""
+    stream = AsyncChatModelStream()
+    assert isinstance(stream, ChatModelStream)

libs/langgraph/tests/test_stream_convert.py

@@ -0,0 +1,86 @@
+from langgraph.stream._convert import STREAM_V2_MODES, convert_to_protocol_event
+
+
+def test_values_mode():
+    evt = convert_to_protocol_event((), "values", {"x": 1})
+    assert evt is not None
+    assert evt["method"] == "values"
+    assert evt["params"]["data"] == {"x": 1}
+
+
+def test_updates_mode():
+    evt = convert_to_protocol_event((), "updates", {"node": "out"})
+    assert evt is not None
+    assert evt["method"] == "updates"
+
+
+def test_messages_mode():
+    evt = convert_to_protocol_event((), "messages", {"event": "msg"})
+    assert evt is not None
+    assert evt["method"] == "messages"
+
+
+def test_custom_mode():
+    evt = convert_to_protocol_event((), "custom", "hello")
+    assert evt is not None
+    assert evt["method"] == "custom"
+    assert evt["params"]["data"] == "hello"
+
+
+def test_debug_mode():
+    evt = convert_to_protocol_event((), "debug", {})
+    assert evt is not None
+    assert evt["method"] == "debug"
+
+
+def test_checkpoints_mode():
+    evt = convert_to_protocol_event((), "checkpoints", {})
+    assert evt is not None
+    assert evt["method"] == "checkpoints"
+
+
+def test_tasks_mode():
+    evt = convert_to_protocol_event((), "tasks", {})
+    assert evt is not None
+    assert evt["method"] == "tasks"
+
+
+def test_namespace_passthrough():
+    evt = convert_to_protocol_event(("agent", "0"), "values", {})
+    assert evt is not None
+    assert evt["params"]["namespace"] == ["agent", "0"]
+
+
+def test_timestamp_populated():
+    evt = convert_to_protocol_event((), "values", {})
+    assert evt is not None
+    assert isinstance(evt["params"]["timestamp"], int)
+    assert evt["params"]["timestamp"] > 0
+
+
+def test_unknown_mode_returns_none():
+    assert convert_to_protocol_event((), "unknown_mode", {}) is None
+
+
+def test_node_parameter():
+    evt = convert_to_protocol_event((), "values", {}, node="agent")
+    assert evt is not None
+    assert evt["params"]["node"] == "agent"
+
+
+def test_type_is_event():
+    evt = convert_to_protocol_event((), "values", {})
+    assert evt is not None
+    assert evt["type"] == "event"
+
+
+def test_stream_v2_modes_complete():
+    assert set(STREAM_V2_MODES) == {
+        "values",
+        "updates",
+        "messages",
+        "custom",
+        "checkpoints",
+        "tasks",
+        "debug",
+    }

libs/langgraph/tests/test_stream_event_log.py

@@ -0,0 +1,133 @@
+import asyncio
+
+import pytest
+
+from langgraph.stream._event_log import EventLog
+
+
+@pytest.mark.anyio
+async def test_push_and_iterate_in_order():
+    log = EventLog()
+    log.append("a")
+    log.append("b")
+    log.append("c")
+    log.close()
+    items = [item async for item in aiter(log)]
+    assert items == ["a", "b", "c"]
+
+
+@pytest.mark.anyio
+async def test_multiple_independent_cursors():
+    log = EventLog()
+    log.append("x")
+    log.append("y")
+    log.close()
+    items1 = [item async for item in aiter(log)]
+    items2 = [item async for item in aiter(log)]
+    assert items1 == ["x", "y"]
+    assert items2 == ["x", "y"]
+
+
+@pytest.mark.anyio
+async def test_close_ends_iteration():
+    log = EventLog()
+    log.close()
+    items = [item async for item in aiter(log)]
+    assert items == []
+
+
+@pytest.mark.anyio
+async def test_fail_raises_error():
+    log = EventLog()
+    log.fail(RuntimeError("boom"))
+    with pytest.raises(RuntimeError, match="boom"):
+        async for _ in aiter(log):
+            pass
+
+
+@pytest.mark.anyio
+async def test_concurrent_push_and_iterate():
+    log = EventLog()
+    received = []
+
+    async def consumer():
+        async for item in aiter(log):
+            received.append(item)
+
+    async def producer():
+        for i in range(5):
+            log.append(i)
+            await asyncio.sleep(0.01)
+        log.close()
+
+    await asyncio.gather(producer(), consumer())
+    assert received == [0, 1, 2, 3, 4]
+
+
+@pytest.mark.anyio
+async def test_items_before_cursor_visible():
+    log = EventLog()
+    log.append("a")
+    log.append("b")
+    cursor = aiter(log)
+    log.append("c")
+    log.close()
+    items = [item async for item in cursor]
+    assert items == ["a", "b", "c"]
+
+
+@pytest.mark.anyio
+async def test_empty_log_closed_yields_nothing():
+    log = EventLog()
+    log.close()
+    items = [item async for item in aiter(log)]
+    assert items == []
+
+
+@pytest.mark.anyio
+async def test_fail_mid_iteration():
+    """A cursor that has consumed some items should raise when fail() is called."""
+    log = EventLog()
+    received = []
+
+    async def consumer():
+        async for item in aiter(log):
+            received.append(item)
+
+    async def producer():
+        log.append("a")
+        log.append("b")
+        await asyncio.sleep(0.02)
+        log.fail(RuntimeError("mid-stream error"))
+
+    with pytest.raises(RuntimeError, match="mid-stream error"):
+        await asyncio.gather(producer(), consumer())
+    assert received == ["a", "b"]
+
+
+@pytest.mark.anyio
+async def test_abandoned_cursor_cleans_up_waiters():
+    """Abandoned async cursors should not leave stale futures in the
+    EventLog waiter list.
+
+    When a cursor's __anext__ is cancelled (e.g. consumer breaks out of
+    ``async for``), the Future it registered in ``_waiters`` should be
+    cleaned up.  Otherwise the list grows without bound until the next
+    append/close/fail triggers ``_wake_all()``.
+    """
+    log: EventLog[str] = EventLog()
+
+    for _ in range(10):
+        cursor = aiter(log)
+        task = asyncio.ensure_future(cursor.__anext__())
+        await asyncio.sleep(0)  # let task register its waiter
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    assert len(log._waiters) == 0, (
+        f"Expected 0 waiters after abandoning 10 cursors, "
+        f"got {len(log._waiters)}.  Abandoned cursors leak futures."
+    )

libs/langgraph/tests/test_stream_mux.py

@@ -0,0 +1,293 @@
+from typing import Any
+
+import pytest
+
+from langgraph.stream._convert import convert_to_protocol_event
+from langgraph.stream._mux import AsyncStreamMux, StreamMux
+from langgraph.stream._types import ProtocolEvent
+from langgraph.stream.stream_channel import StreamChannel
+
+
+def _event(mode: str, data: Any, ns: list[str] | None = None) -> ProtocolEvent:
+    ev = convert_to_protocol_event(tuple(ns or []), mode, data)
+    assert ev is not None
+    return ev
+
+
+class _MockTransformer:
+    def __init__(self, *, suppress: bool = False):
+        self.calls: list[ProtocolEvent] = []
+        self._suppress = suppress
+
+    def init(self) -> Any:
+        return None
+
+    def process(self, event: ProtocolEvent) -> bool:
+        self.calls.append(event)
+        return not self._suppress
+
+    def finalize(self) -> None:
+        pass
+
+    def fail(self, err: BaseException) -> None:
+        pass
+
+
+@pytest.mark.anyio
+async def test_events_through_reducer_pipeline():
+    reducer = _MockTransformer()
+    mux = StreamMux(transformers=[reducer])
+    event = _event("values", {"key": "val"})
+    mux.push(event)
+    assert len(reducer.calls) == 1
+    assert reducer.calls[0] is event
+
+
+@pytest.mark.anyio
+async def test_reducer_suppresses_event():
+    reducer = _MockTransformer(suppress=True)
+    mux = StreamMux(transformers=[reducer])
+    mux.push(_event("values", {"x": 1}))
+    mux.close()
+    assert len(reducer.calls) == 1
+    assert len(mux.event_log) == 0
+
+
+@pytest.mark.anyio
+async def test_namespace_discovery():
+    mux = StreamMux()
+    mux.push(_event("values", {"a": 1}, ns=["child:0"]))
+    assert "child:0" in mux._discovered_ns
+
+
+@pytest.mark.anyio
+async def test_top_level_ns_only():
+    mux = StreamMux()
+    mux.push(_event("values", {"a": 1}, ns=["agent:0", "tools:1"]))
+    assert "agent:0" in mux._discovered_ns
+    assert "tools:1" not in mux._discovered_ns
+
+
+@pytest.mark.anyio
+async def test_subscribe_events_filter():
+    mux = AsyncStreamMux()
+    mux.push(_event("values", {"a": 1}, ns=["child:0"]))
+    mux.push(_event("values", {"b": 2}, ns=["other:1"]))
+    mux.push(_event("values", {"c": 3}, ns=["child:0"]))
+    mux.close()
+
+    collected = []
+    async for ev in mux.subscribe_events(["child:0"]):
+        collected.append(ev)
+    assert len(collected) == 2
+    assert collected[0]["params"]["data"] == {"a": 1}
+    assert collected[1]["params"]["data"] == {"c": 3}
+
+
+@pytest.mark.anyio
+async def test_close_resolves_output():
+    mux = AsyncStreamMux()
+    fut = mux.get_output_future()
+    mux.push(_event("values", {"v": 1}))
+    mux.push(_event("values", {"v": 2}))
+    mux.close()
+    result = await fut
+    assert result == {"v": 2}
+
+
+@pytest.mark.anyio
+async def test_fail_rejects_output():
+    mux = AsyncStreamMux()
+    fut = mux.get_output_future()
+    mux.fail(ValueError("boom"))
+    with pytest.raises(ValueError, match="boom"):
+        await fut
+
+
+@pytest.mark.anyio
+async def test_latest_values_tracked():
+    mux = StreamMux()
+    mux.push(_event("values", {"v": 1}, ns=["child:0"]))
+    mux.push(_event("values", {"v": 2}, ns=["child:0"]))
+    assert mux.get_latest_values(["child:0"]) == {"v": 2}
+
+
+@pytest.mark.anyio
+async def test_interrupt_tracking():
+    """StreamMux should track __interrupt__ payloads in values events."""
+
+    class _FakeInterrupt:
+        def __init__(self, id: str, payload: Any):
+            self.id = id
+            self.payload = payload
+
+    mux = StreamMux()
+    interrupt_obj = _FakeInterrupt("int-1", "what do you want?")
+    mux.push(
+        _event(
+            "values",
+            {"__interrupt__": [interrupt_obj]},
+        )
+    )
+    assert mux.interrupted is True
+    assert len(mux.interrupts) == 1
+    assert mux.interrupts[0]["interrupt_id"] == "int-1"
+    assert mux.interrupts[0]["payload"] is interrupt_obj
+
+
+@pytest.mark.anyio
+async def test_no_interrupt_by_default():
+    mux = StreamMux()
+    mux.push(_event("values", {"x": 1}))
+    mux.close()
+    assert mux.interrupted is False
+    assert mux.interrupts == []
+
+
+@pytest.mark.anyio
+async def test_push_after_close_ignored():
+    mux = StreamMux()
+    mux.push(_event("values", {"a": 1}))
+    mux.close()
+    mux.push(_event("values", {"b": 2}))
+    assert len(mux.event_log) == 1
+
+
+@pytest.mark.anyio
+async def test_fail_rejects_all_futures():
+    mux = AsyncStreamMux()
+    fut1 = mux.get_output_future([])
+    fut2 = mux.get_output_future(["child:0"])
+    mux.fail(ValueError("boom"))
+    with pytest.raises(ValueError, match="boom"):
+        await fut1
+    with pytest.raises(ValueError, match="boom"):
+        await fut2
+
+
+@pytest.mark.anyio
+async def test_channel_events_bypass_transformer_pipeline():
+    """Events emitted via ``StreamChannel.push()`` are appended directly
+    to the event log, bypassing the transformer pipeline.  This matches
+    the JS implementation and avoids re-entrancy bugs.
+    """
+    mock = _MockTransformer()
+    mux = AsyncStreamMux(transformers=[mock])
+
+    channel: StreamChannel[str] = StreamChannel("my_channel")
+    mux.wire_channels({"ch": channel})
+
+    # Regular push — transformer sees it
+    mux.push(_event("values", {"a": 1}))
+    assert len(mock.calls) == 1
+
+    # Channel push — bypasses transformers, goes straight to event log
+    channel.push("hello from channel")
+
+    assert len(mock.calls) == 1, (
+        f"Transformer saw {len(mock.calls)} events (expected 1).  "
+        "Channel events should bypass the transformer pipeline."
+    )
+
+    # But the event IS in the log
+    mux.close()
+    events = []
+    async for ev in mux.subscribe_events():
+        events.append(ev)
+    assert len(events) == 2
+    assert events[1]["method"] == "my_channel"
+    assert events[1]["params"]["data"] == "hello from channel"
+
+
+@pytest.mark.anyio
+async def test_event_log_has_monotonic_seq_numbers():
+    """All events in the event log should have strictly monotonically
+    increasing seq numbers so consumers can reason about ordering.
+
+    Events from ``mux.push()`` carry seq numbers assigned by the pump
+    while channel-emitted events use a separate counter
+    (``_next_emit_seq``).  When interleaved, seq numbers can duplicate.
+    """
+    mux = AsyncStreamMux()
+    channel: StreamChannel[str] = StreamChannel("test_ch")
+    mux.wire_channels({"ch": channel})
+
+    mux.push(_event("values", {"a": 1}))  # log seq: 0
+    channel.push("from_channel")  # log seq: 0 (from _next_emit_seq)
+    mux.push(_event("values", {"b": 2}))  # log seq: 1
+    mux.close()
+
+    seqs: list[int] = []
+    async for event in mux.subscribe_events():
+        seqs.append(event["seq"])
+
+    assert len(seqs) == 3, f"Expected 3 events but got {len(seqs)}"
+
+    for i in range(1, len(seqs)):
+        assert seqs[i] > seqs[i - 1], (
+            f"Seq numbers not strictly monotonic: {seqs}.  "
+            f"seq[{i}]={seqs[i]} <= seq[{i - 1}]={seqs[i - 1]}.  "
+            "Channel events use a separate counter from push() events."
+        )
+
+
+@pytest.mark.anyio
+async def test_channel_push_during_process_preserves_namespace():
+    """When two transformers both call channel.push() during the same
+    outer mux.push(), the second transformer's channel event should
+    still carry the original event's namespace.
+
+    Bug: the first channel.push() re-enters mux.push(), which resets
+    ``_current_namespace`` to ``[]`` on exit.  The second transformer's
+    channel.push() then reads the clobbered value and its event gets
+    ``namespace: []`` instead of the original.
+    """
+
+    class _ChannelTransformer:
+        """Pushes to its channel whenever it sees a ``values`` event."""
+
+        def __init__(self, name: str) -> None:
+            self.name = name
+            self.channel: StreamChannel[str] = StreamChannel(name)
+
+        def init(self) -> Any:
+            return {self.name: self.channel}
+
+        def process(self, event: ProtocolEvent) -> bool:
+            if event["method"] == "values":
+                self.channel.push(f"from_{self.name}")
+            return True
+
+        def finalize(self) -> None:
+            pass
+
+        def fail(self, err: BaseException) -> None:
+            pass
+
+    t1 = _ChannelTransformer("first")
+    t2 = _ChannelTransformer("second")
+    mux = AsyncStreamMux(transformers=[t1, t2])
+    mux.wire_channels({"first": t1.channel})
+    mux.wire_channels({"second": t2.channel})
+
+    # Push a values event with a non-root namespace
+    mux.push(_event("values", {"x": 1}, ns=["agent:0"]))
+    mux.close()
+
+    # Collect channel events emitted by each transformer
+    channel_events: list[ProtocolEvent] = []
+    async for ev in mux.subscribe_events():
+        if ev["method"] in ("first", "second"):
+            channel_events.append(ev)
+
+    assert len(channel_events) == 2, (
+        f"Expected 2 channel events but got {len(channel_events)}"
+    )
+
+    for ev in channel_events:
+        assert ev["params"]["namespace"] == ["agent:0"], (
+            f"Channel event for method={ev['method']!r} has "
+            f"namespace={ev['params']['namespace']!r}, expected ['agent:0'].  "
+            "The nested mux.push() from the first channel.push() clobbered "
+            "_current_namespace before the second transformer ran."
+        )

libs/langgraph/tests/test_stream_reducers.py

@@ -0,0 +1,214 @@
+from typing import Any
+
+import pytest
+
+from langgraph.stream._convert import convert_to_protocol_event
+from langgraph.stream._types import ProtocolEvent
+from langgraph.stream.chat_model_stream import ChatModelStream
+from langgraph.stream.transformers import MessagesTransformer, ValuesTransformer
+
+
+def _event(
+    mode: str,
+    data: Any,
+    ns: list[str] | None = None,
+    node: str | None = None,
+) -> ProtocolEvent:
+    ev = convert_to_protocol_event(tuple(ns or []), mode, data, node=node)
+    assert ev is not None
+    return ev
+
+
+# -- ValuesTransformer ---------------------------------------------------------
+
+
+@pytest.mark.anyio
+async def test_values_captures_values_events():
+    reducer = ValuesTransformer()
+    reducer.init()
+    reducer.process(_event("values", {"a": 1}))
+    reducer.process(_event("values", {"b": 2}))
+    reducer.finalize()
+
+    collected = []
+    async for item in reducer.values_log:
+        collected.append(item)
+    assert len(collected) == 2
+    assert collected[0]["data"] == {"a": 1}
+    assert collected[1]["data"] == {"b": 2}
+
+
+@pytest.mark.anyio
+async def test_values_ignores_other_modes():
+    reducer = ValuesTransformer()
+    reducer.init()
+    reducer.process(_event("updates", {"x": 1}))
+    reducer.process(_event("messages", {"event": "message-start"}))
+    reducer.finalize()
+
+    collected = []
+    async for item in reducer.values_log:
+        collected.append(item)
+    assert len(collected) == 0
+
+
+@pytest.mark.anyio
+async def test_values_latest_per_namespace():
+    reducer = ValuesTransformer()
+    reducer.init()
+    reducer.process(_event("values", {"v": 1}, ns=["child:0"]))
+    reducer.process(_event("values", {"v": 2}, ns=["child:0"]))
+    assert reducer.get_latest("child:0") == {"v": 2}
+
+
+@pytest.mark.anyio
+async def test_values_finalize_closes_log():
+    reducer = ValuesTransformer()
+    reducer.init()
+    reducer.process(_event("values", {"a": 1}))
+    reducer.finalize()
+    assert reducer.values_log.closed
+
+
+# -- MessagesTransformer -------------------------------------------------------
+
+
+def _msg_start(ns=None, node=None, message_id="msg-1"):
+    return _event(
+        "messages",
+        {"event": "message-start", "message_id": message_id},
+        ns=ns,
+        node=node,
+    )
+
+
+def _content_delta(text, ns=None, node=None):
+    return _event(
+        "messages",
+        {
+            "event": "content-block-delta",
+            "content_block": {"type": "text", "text": text},
+        },
+        ns=ns,
+        node=node,
+    )
+
+
+def _msg_finish(ns=None, node=None):
+    return _event(
+        "messages",
+        {"event": "message-finish", "reason": "stop"},
+        ns=ns,
+        node=node,
+    )
+
+
+@pytest.mark.anyio
+async def test_messages_groups_lifecycle():
+    reducer = MessagesTransformer()
+    reducer.init()
+    reducer.process(_msg_start())
+    reducer.process(_content_delta("hi"))
+    reducer.process(_msg_finish())
+    reducer.finalize()
+
+    collected = []
+    async for stream in reducer.messages_log:
+        collected.append(stream)
+    assert len(collected) == 1
+    assert isinstance(collected[0], ChatModelStream)
+    assert collected[0].done
+
+
+@pytest.mark.anyio
+async def test_messages_multiple_sequential():
+    reducer = MessagesTransformer()
+    reducer.init()
+    reducer.process(_msg_start(message_id="m1"))
+    reducer.process(_msg_finish())
+    reducer.process(_msg_start(message_id="m2"))
+    reducer.process(_msg_finish())
+    reducer.finalize()
+
+    collected = []
+    async for stream in reducer.messages_log:
+        collected.append(stream)
+    assert len(collected) == 2
+
+
+@pytest.mark.anyio
+async def test_messages_namespace_filter():
+    reducer = MessagesTransformer(namespace=["root"])
+    reducer.init()
+    reducer.process(_msg_start(ns=["root"]))
+    reducer.process(_msg_finish(ns=["root"]))
+    reducer.process(_msg_start(ns=["other"], message_id="m2"))
+    reducer.process(_msg_finish(ns=["other"]))
+    reducer.finalize()
+
+    collected = []
+    async for stream in reducer.messages_log:
+        collected.append(stream)
+    assert len(collected) == 1
+
+
+@pytest.mark.anyio
+async def test_messages_node_filter():
+    reducer = MessagesTransformer(node_filter="agent")
+    reducer.init()
+    reducer.process(_msg_start(node="agent"))
+    reducer.process(_msg_finish(node="agent"))
+    reducer.process(_msg_start(node="tools", message_id="m2"))
+    reducer.process(_msg_finish(node="tools"))
+    reducer.finalize()
+
+    collected = []
+    async for stream in reducer.messages_log:
+        collected.append(stream)
+    assert len(collected) == 1
+
+
+@pytest.mark.anyio
+async def test_messages_error_event():
+    """An error event should fail the active ChatModelStream."""
+    reducer = MessagesTransformer()
+    reducer.init()
+    reducer.process(_msg_start())
+    reducer.process(_content_delta("partial"))
+    reducer.process(
+        _event("messages", {"event": "error", "message": "connection lost"}),
+    )
+    reducer.finalize()
+
+    collected: list[ChatModelStream] = []
+    async for stream in reducer.messages_log:
+        collected.append(stream)
+    assert len(collected) == 1
+    assert collected[0].done
+
+
+@pytest.mark.anyio
+async def test_messages_fail_propagates_to_active():
+    """transformer.fail() should propagate the error to any active streams."""
+    reducer = MessagesTransformer()
+    reducer.init()
+    reducer.process(_msg_start())
+    reducer.process(_content_delta("partial"))
+    reducer.fail(RuntimeError("graph failed"))
+
+    # The messages log should be failed too
+    with pytest.raises(RuntimeError, match="graph failed"):
+        async for _ in reducer.messages_log:
+            pass
+
+
+@pytest.mark.anyio
+async def test_values_fail_propagates():
+    reducer = ValuesTransformer()
+    reducer.init()
+    reducer.process(_event("values", {"a": 1}))
+    reducer.fail(RuntimeError("graph failed"))
+
+    with pytest.raises(RuntimeError, match="graph failed"):
+        async for _ in reducer.values_log:
+            pass

libs/langgraph/tests/test_stream_run_stream.py

@@ -0,0 +1,980 @@
+import asyncio
+from collections.abc import AsyncIterator, Iterator
+from typing import Any
+
+import pytest
+
+from langgraph.stream._mux import AsyncStreamMux
+from langgraph.stream._types import ProtocolEvent
+from langgraph.stream.chat_model_stream import ChatModelStream
+from langgraph.stream.run_stream import (
+    AsyncGraphRunStream,
+    AsyncSubgraphRunStream,
+    SubgraphRunStream,
+    create_async_graph_run_stream,
+    create_graph_run_stream,
+)
+from langgraph.stream.transformers import MessagesTransformer, ValuesTransformer
+
+
+async def _mock_source(
+    chunks: list[tuple[tuple[str, ...], str, Any]],
+) -> AsyncIterator[tuple[tuple[str, ...], str, Any]]:
+    for chunk in chunks:
+        yield chunk
+
+
+@pytest.mark.anyio
+async def test_aiter_yields_all_events():
+    chunks = [
+        ((), "values", {"step": 1}),
+        ((), "values", {"step": 2}),
+        ((), "updates", {"node": "a"}),
+    ]
+    run = await create_async_graph_run_stream(_mock_source(chunks))
+    await asyncio.sleep(0.05)
+
+    collected: list[ProtocolEvent] = []
+    async for event in run:
+        collected.append(event)
+    assert len(collected) == 3
+    assert collected[0]["method"] == "values"
+    assert collected[2]["method"] == "updates"
+
+
+@pytest.mark.anyio
+async def test_subgraph_name_and_index():
+    vr, mr = ValuesTransformer(), MessagesTransformer()
+    mux = AsyncStreamMux(transformers=[vr, mr])
+    sub = AsyncSubgraphRunStream(
+        mux=mux,
+        namespace=["researcher:2"],
+        transformers=[vr, mr],
+    )
+    assert sub.name == "researcher"
+    assert sub.index == 2
+
+
+@pytest.mark.anyio
+async def test_subgraph_name_no_index():
+    vr, mr = ValuesTransformer(), MessagesTransformer()
+    mux = AsyncStreamMux(transformers=[vr, mr])
+    sub = AsyncSubgraphRunStream(
+        mux=mux, namespace=["agent"], transformers=[vr, mr]
+    )
+    assert sub.name == "agent"
+    assert sub.index == 0
+
+
+@pytest.mark.anyio
+async def test_values_iterable():
+    chunks = [
+        ((), "values", {"v": 1}),
+        ((), "values", {"v": 2}),
+    ]
+    run = await create_async_graph_run_stream(_mock_source(chunks))
+    await asyncio.sleep(0.05)
+
+    collected = []
+    async for v in run.values:
+        collected.append(v)
+    assert len(collected) == 2
+    assert collected[0] == {"v": 1}
+    assert collected[1] == {"v": 2}
+
+
+@pytest.mark.anyio
+async def test_values_awaitable():
+    chunks = [
+        ((), "values", {"v": 1}),
+        ((), "values", {"v": 2}),
+    ]
+    run = await create_async_graph_run_stream(_mock_source(chunks))
+    await asyncio.sleep(0.05)
+
+    result = await run.values
+    assert result == {"v": 2}
+
+
+@pytest.mark.anyio
+async def test_output_resolves():
+    chunks = [((), "values", {"final": True})]
+    run = await create_async_graph_run_stream(_mock_source(chunks))
+    await asyncio.sleep(0.05)
+    result = await run.output
+    assert result == {"final": True}
+
+
+@pytest.mark.anyio
+async def test_messages_yields_streams():
+    chunks = [
+        ((), "messages", {"event": "message-start", "message_id": "m1"}),
+        (
+            (),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "hi"},
+            },
+        ),
+        ((), "messages", {"event": "message-finish", "reason": "stop"}),
+    ]
+    run = await create_async_graph_run_stream(_mock_source(chunks))
+    await asyncio.sleep(0.05)
+
+    collected: list[ChatModelStream] = []
+    async for stream in run.messages:
+        collected.append(stream)
+    assert len(collected) == 1
+    assert isinstance(collected[0], ChatModelStream)
+    assert collected[0].done
+
+
+@pytest.mark.anyio
+async def test_interrupted_false_by_default():
+    vr, mr = ValuesTransformer(), MessagesTransformer()
+    mux = AsyncStreamMux(transformers=[vr, mr])
+    run = AsyncGraphRunStream(mux=mux, transformers=[vr, mr])
+    assert run.interrupted is False
+
+
+@pytest.mark.anyio
+async def test_abort_sets_signal():
+    vr, mr = ValuesTransformer(), MessagesTransformer()
+    mux = AsyncStreamMux(transformers=[vr, mr])
+    run = AsyncGraphRunStream(mux=mux, transformers=[vr, mr])
+    assert not run.signal.is_set()
+    run.abort()
+    assert run.signal.is_set()
+
+
+@pytest.mark.anyio
+async def test_abort_stops_pump():
+    """Calling abort() should stop the pump from processing further chunks."""
+    gate = asyncio.Event()
+
+    async def _gated_source():
+        yield ((), "values", {"v": 1})
+        yield ((), "values", {"v": 2})
+        await gate.wait()  # Block until released
+        yield ((), "values", {"v": 3})  # Should not be processed
+
+    run = await create_async_graph_run_stream(_gated_source())
+    await asyncio.sleep(0.05)  # Let first two events through
+    run.abort()
+    gate.set()  # Unblock the source so the pump can check abort and exit
+    await asyncio.sleep(0.05)  # Let pump close the mux
+
+    collected = []
+    async for event in run:
+        if event["method"] == "values":
+            collected.append(event["params"]["data"])
+    # v:3 should not have been processed because abort was set
+    assert all(v.get("v") != 3 for v in collected)
+
+
+@pytest.mark.anyio
+async def test_messages_from_filters_by_node():
+    """messages_from(node) should only yield messages from the specified node."""
+    chunks = [
+        (
+            (),
+            "messages",
+            {"event": "message-start", "message_id": "m1", "__node__": "agent"},
+        ),
+        (
+            (),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "from agent"},
+                "__node__": "agent",
+            },
+        ),
+        (
+            (),
+            "messages",
+            {"event": "message-finish", "reason": "stop", "__node__": "agent"},
+        ),
+        (
+            (),
+            "messages",
+            {"event": "message-start", "message_id": "m2", "__node__": "tools"},
+        ),
+        (
+            (),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "from tools"},
+                "__node__": "tools",
+            },
+        ),
+        (
+            (),
+            "messages",
+            {"event": "message-finish", "reason": "stop", "__node__": "tools"},
+        ),
+    ]
+    run = await create_async_graph_run_stream(_mock_source(chunks))
+    await asyncio.sleep(0.05)
+
+    agent_msgs: list[ChatModelStream] = []
+    async for stream in run.messages_from("agent"):
+        agent_msgs.append(stream)
+    assert len(agent_msgs) == 1
+    assert agent_msgs[0].node == "agent"
+
+
+# ---------------------------------------------------------------------------
+# GraphRunStream / create_graph_run_stream
+# ---------------------------------------------------------------------------
+
+
+def _sync_source(
+    chunks: list[tuple[tuple[str, ...], str, Any]],
+) -> Iterator[tuple[tuple[str, ...], str, Any]]:
+    yield from chunks
+
+
+def test_sync_create_yields_all_events():
+    chunks = [
+        ((), "values", {"step": 1}),
+        ((), "values", {"step": 2}),
+        ((), "updates", {"node": "a"}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    collected = list(run)
+    assert len(collected) == 3
+    assert collected[0]["method"] == "values"
+    assert collected[2]["method"] == "updates"
+
+
+def test_sync_output():
+    chunks = [
+        ((), "values", {"v": 1}),
+        ((), "values", {"v": 2}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    assert run.output == {"v": 2}
+
+
+def test_sync_values_iteration():
+    chunks = [
+        ((), "values", {"v": 1}),
+        ((), "values", {"v": 2}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    collected = list(run.values)
+    assert len(collected) == 2
+    assert collected[0] == {"v": 1}
+    assert collected[1] == {"v": 2}
+
+
+def test_sync_messages():
+    chunks = [
+        ((), "messages", {"event": "message-start", "message_id": "m1"}),
+        (
+            (),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "hi"},
+            },
+        ),
+        ((), "messages", {"event": "message-finish", "reason": "stop"}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    collected = list(run.messages)
+    assert len(collected) == 1
+    assert isinstance(collected[0], ChatModelStream)
+    assert collected[0].done
+
+
+def test_sync_messages_text_streaming():
+    """Sync consumers can iterate msg.text for deltas."""
+    chunks = [
+        ((), "messages", {"event": "message-start", "message_id": "m1"}),
+        (
+            (),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "Hello"},
+            },
+        ),
+        (
+            (),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": " world"},
+            },
+        ),
+        ((), "messages", {"event": "message-finish", "reason": "stop"}),
+    ]
+
+    # Iterate deltas
+    run = create_graph_run_stream(_sync_source(chunks))
+    for msg in run.messages:
+        deltas = list(msg.text)
+        assert deltas == ["Hello", " world"]
+        assert msg.done
+
+    # str() returns full text
+    run = create_graph_run_stream(_sync_source(chunks))
+    for msg in run.messages:
+        assert str(msg.text) == "Hello world"
+
+    # After message is done, .text returns plain str
+    run = create_graph_run_stream(_sync_source(chunks))
+    for msg in run.messages:
+        list(msg.text)  # exhaust deltas
+        assert isinstance(msg.text, str)
+        assert msg.text == "Hello world"
+
+
+def test_sync_messages_multiple():
+    """Multiple sync messages each stream their own deltas."""
+    chunks = [
+        ((), "messages", {"event": "message-start", "message_id": "m1"}),
+        (
+            (),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "answer"},
+            },
+        ),
+        ((), "messages", {"event": "message-finish", "reason": "stop"}),
+        ((), "messages", {"event": "message-start", "message_id": "m2"}),
+        (
+            (),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "second"},
+            },
+        ),
+        ((), "messages", {"event": "message-finish", "reason": "stop"}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+
+    all_deltas = []
+    for msg in run.messages:
+        all_deltas.append(list(msg.text))
+    assert all_deltas == [["answer"], ["second"]]
+
+
+def test_sync_output_mapper():
+    chunks = [((), "values", {"v": 1})]
+    run = create_graph_run_stream(
+        _sync_source(chunks), output_mapper=lambda x: {"mapped": x["v"]}
+    )
+    assert run.output == {"mapped": 1}
+
+
+def test_sync_interrupted_false():
+    chunks = [((), "values", {"v": 1})]
+    run = create_graph_run_stream(_sync_source(chunks))
+    assert run.interrupted is False
+
+
+def test_sync_source_error():
+    """If the source raises, the mux should fail and the error should propagate."""
+
+    def _bad_source():
+        yield ((), "values", {"v": 1})
+        raise ValueError("source error")
+
+    run = create_graph_run_stream(_bad_source())
+    collected = list(run)
+    # Events before the error are still accessible
+    assert len(collected) >= 1
+    assert collected[0]["method"] == "values"
+    # The mux recorded the failure
+    assert run._mux._error is not None
+    assert isinstance(run._mux._error, ValueError)
+    assert "source error" in str(run._mux._error)
+
+
+# ---------------------------------------------------------------------------
+# GraphRunStream — lazy consumption tests
+# ---------------------------------------------------------------------------
+
+
+def test_sync_lazy_not_consumed_on_creation():
+    """Source iterator should not be consumed when the stream is created."""
+    consumed = 0
+
+    def counting_source():
+        nonlocal consumed
+        for chunk in [
+            ((), "values", {"v": 1}),
+            ((), "values", {"v": 2}),
+            ((), "values", {"v": 3}),
+        ]:
+            consumed += 1
+            yield chunk
+
+    create_graph_run_stream(counting_source())
+    assert consumed == 0
+
+
+def test_sync_lazy_values_pull_incrementally():
+    """Iterating .values should pull from the source one event at a time."""
+    consumed = 0
+
+    def counting_source():
+        nonlocal consumed
+        for chunk in [
+            ((), "values", {"v": 1}),
+            ((), "values", {"v": 2}),
+            ((), "values", {"v": 3}),
+        ]:
+            consumed += 1
+            yield chunk
+
+    run = create_graph_run_stream(counting_source())
+    assert consumed == 0
+
+    it = iter(run.values)
+    v = next(it)
+    assert v == {"v": 1}
+    assert consumed == 1
+
+    v = next(it)
+    assert v == {"v": 2}
+    assert consumed == 2
+
+    # Source not fully drained yet
+    assert consumed < 3
+
+
+def test_sync_lazy_output_drains_all():
+    """Accessing .output should drain the entire source."""
+    consumed = 0
+
+    def counting_source():
+        nonlocal consumed
+        for chunk in [((), "values", {"v": i}) for i in range(5)]:
+            consumed += 1
+            yield chunk
+
+    run = create_graph_run_stream(counting_source())
+    assert consumed == 0
+    assert run.output == {"v": 4}
+    assert consumed == 5
+
+
+def test_sync_lazy_early_break():
+    """Breaking out of a projection early should leave the source partially consumed."""
+    consumed = 0
+
+    def counting_source():
+        nonlocal consumed
+        for chunk in [((), "values", {"v": i}) for i in range(10)]:
+            consumed += 1
+            yield chunk
+
+    run = create_graph_run_stream(counting_source())
+    for v in run.values:
+        break  # consume only the first value
+    assert consumed == 1
+    assert consumed < 10
+
+
+def test_sync_lazy_interleaved_projections():
+    """Switching between projections replays buffered items then resumes pumping."""
+    consumed = 0
+
+    def counting_source():
+        nonlocal consumed
+        for chunk in [
+            ((), "values", {"v": 1}),
+            ((), "messages", {"event": "message-start", "message_id": "m1"}),
+            ((), "messages", {"event": "message-finish", "reason": "stop"}),
+            ((), "values", {"v": 2}),
+        ]:
+            consumed += 1
+            yield chunk
+
+    run = create_graph_run_stream(counting_source())
+
+    # Pull first value — consumes 1 source item
+    vit = iter(run.values)
+    assert next(vit) == {"v": 1}
+    assert consumed == 1
+
+    # Pull first message — yielded on message-start (item 2).
+    # Consuming str(msg.text) drives the pump to message-finish (item 3).
+    mit = iter(run.messages)
+    msg = next(mit)
+    assert isinstance(msg, ChatModelStream)
+    assert consumed == 2
+    assert not msg.done
+    str(msg.text)  # pump until message completes
+    assert msg.done
+    assert consumed == 3
+
+    # Pull second value — pumps values (item 4)
+    assert next(vit) == {"v": 2}
+    assert consumed == 4
+
+
+def test_sync_lazy_iter_pulls_incrementally():
+    """Raw __iter__ should pull from the source lazily."""
+    consumed = 0
+
+    def counting_source():
+        nonlocal consumed
+        for chunk in [
+            ((), "values", {"v": 1}),
+            ((), "updates", {"node": "a"}),
+            ((), "values", {"v": 2}),
+        ]:
+            consumed += 1
+            yield chunk
+
+    run = create_graph_run_stream(counting_source())
+    it = iter(run)
+    event = next(it)
+    assert event["method"] == "values"
+    assert consumed == 1
+
+    event = next(it)
+    assert event["method"] == "updates"
+    assert consumed == 2
+
+
+def test_sync_lazy_source_error():
+    """If the source raises mid-stream, earlier events are still accessible."""
+    consumed = 0
+
+    def bad_source():
+        nonlocal consumed
+        consumed += 1
+        yield ((), "values", {"v": 1})
+        raise ValueError("boom")
+
+    run = create_graph_run_stream(bad_source())
+    collected = list(run)
+    assert len(collected) >= 1
+    assert collected[0]["method"] == "values"
+
+
+@pytest.mark.anyio
+async def test_subgraph_child_values_receive_post_discovery_events():
+    """Child AsyncSubgraphRunStream.values iteration should include events
+    that arrive AFTER the subgraph namespace is first discovered.
+
+    ``_SubgraphsProjection`` creates a local ``ValuesTransformer`` for
+    each child and replays existing events, but never registers the
+    transformer with the mux.  Events that arrive after discovery are
+    not routed to it, and ``finalize()`` is not called (the mux wasn't
+    closed at discovery time), so the child's values_log is never
+    closed and iteration hangs.
+    """
+    gate = asyncio.Event()
+
+    async def _source() -> AsyncIterator[tuple[tuple[str, ...], str, Any]]:
+        # First event from child namespace — triggers discovery
+        yield (("child:0",), "values", {"v": 1})
+        await gate.wait()
+        # Second event from same child — arrives after discovery
+        yield (("child:0",), "values", {"v": 2})
+        # Root event so the mux tracks output
+        yield ((), "values", {"done": True})
+
+    run = await create_async_graph_run_stream(_source())
+    await asyncio.sleep(0.05)  # let pump process first event
+
+    # Get the first subgraph while the mux is still open
+    sub = None
+    async for s in run.subgraphs:
+        sub = s
+        break
+
+    assert sub is not None
+
+    # Release the gate so the pump finishes
+    gate.set()
+    await asyncio.sleep(0.05)  # let pump close mux
+
+    # ``await sub.output`` uses the mux's output future — works fine
+    output = await sub.output
+    assert output == {"v": 2}, "await sub.output should reflect the latest value"
+
+    # But ``async for v in sub.values`` only gets the replayed event
+    # and then hangs because the child's values_log is never closed.
+    values: list[Any] = []
+    try:
+        async with asyncio.timeout(1.0):
+            async for v in sub.values:
+                values.append(v)
+    except (asyncio.TimeoutError, TimeoutError):
+        pass
+
+    assert len(values) == 2, (
+        f"Expected 2 child value snapshots but got {len(values)}: {values}.  "
+        "Child transformer missed post-discovery events."
+    )
+
+
+# ---------------------------------------------------------------------------
+# SubgraphRunStream — sync subgraph tests
+# ---------------------------------------------------------------------------
+
+
+def test_sync_subgraphs_discovery():
+    """Iterating .subgraphs should discover child namespaces and yield
+    SubgraphRunStream instances with correct name and index.
+    """
+    chunks = [
+        (("agent:0",), "values", {"v": 1}),
+        (("agent:1",), "values", {"v": 2}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    subs = list(run.subgraphs)
+    assert len(subs) == 2
+    assert all(isinstance(s, SubgraphRunStream) for s in subs)
+    assert subs[0].name == "agent"
+    assert subs[0].index == 0
+    assert subs[1].name == "agent"
+    assert subs[1].index == 1
+
+
+def test_sync_subgraph_name_no_index():
+    """Subgraph without a colon-delimited index should have index=0."""
+    chunks = [
+        (("planner",), "values", {"v": 1}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    subs = list(run.subgraphs)
+    assert len(subs) == 1
+    assert subs[0].name == "planner"
+    assert subs[0].index == 0
+
+
+def test_sync_subgraph_no_subgraphs():
+    """When all events are root-level, .subgraphs should yield nothing."""
+    chunks = [
+        ((), "values", {"v": 1}),
+        ((), "values", {"v": 2}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    subs = list(run.subgraphs)
+    assert subs == []
+
+
+def test_sync_subgraph_values():
+    """SubgraphRunStream.values should yield only values from the child namespace."""
+    chunks = [
+        (("child:0",), "values", {"v": 1}),
+        ((), "values", {"root": True}),
+        (("child:0",), "values", {"v": 2}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        vals = list(sub.values)
+        assert vals == [{"v": 1}, {"v": 2}]
+
+
+def test_sync_subgraph_values_multiple_children():
+    """Each child stream should only see its own values."""
+    chunks = [
+        (("a:0",), "values", {"who": "a0"}),
+        (("b:0",), "values", {"who": "b0"}),
+        (("a:0",), "values", {"who": "a0-2"}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    children: dict[str, list[Any]] = {}
+    for sub in run.subgraphs:
+        children[f"{sub.name}:{sub.index}"] = list(sub.values)
+
+    assert children["a:0"] == [{"who": "a0"}, {"who": "a0-2"}]
+    assert children["b:0"] == [{"who": "b0"}]
+
+
+def test_sync_subgraph_output():
+    """SubgraphRunStream.output should return the last values for the child."""
+    chunks = [
+        (("child:0",), "values", {"v": 1}),
+        (("child:0",), "values", {"v": 2}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        assert sub.output == {"v": 2}
+
+
+def test_sync_subgraph_output_with_mapper():
+    """Output mapper should apply to subgraph output."""
+    chunks = [
+        (("child:0",), "values", {"v": 42}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(
+        _sync_source(chunks), output_mapper=lambda x: {"mapped": x.get("v")}
+    )
+    for sub in run.subgraphs:
+        assert sub.output == {"mapped": 42}
+
+
+def test_sync_subgraph_values_with_mapper():
+    """Output mapper should apply to each yielded value snapshot."""
+    chunks = [
+        (("child:0",), "values", {"v": 1}),
+        (("child:0",), "values", {"v": 2}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(
+        _sync_source(chunks), output_mapper=lambda x: {"m": x.get("v")}
+    )
+    for sub in run.subgraphs:
+        vals = list(sub.values)
+        assert vals == [{"m": 1}, {"m": 2}]
+
+
+def test_sync_subgraph_messages():
+    """SubgraphRunStream.messages should yield fully populated ChatModelStream instances."""
+    chunks = [
+        (
+            ("agent:0",),
+            "messages",
+            {"event": "message-start", "message_id": "m1", "__node__": "agent"},
+        ),
+        (
+            ("agent:0",),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "hello"},
+                "__node__": "agent",
+            },
+        ),
+        (
+            ("agent:0",),
+            "messages",
+            {"event": "message-finish", "reason": "stop", "__node__": "agent"},
+        ),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        msgs = list(sub.messages)
+        assert len(msgs) == 1
+        assert isinstance(msgs[0], ChatModelStream)
+        assert msgs[0].done
+        assert msgs[0].text == "hello"
+
+
+def test_sync_subgraph_messages_isolated():
+    """Messages from different subgraphs should not leak between children."""
+    chunks = [
+        (
+            ("a:0",),
+            "messages",
+            {"event": "message-start", "message_id": "m-a", "__node__": "a"},
+        ),
+        (
+            ("a:0",),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "from-a"},
+                "__node__": "a",
+            },
+        ),
+        (
+            ("a:0",),
+            "messages",
+            {"event": "message-finish", "reason": "stop", "__node__": "a"},
+        ),
+        (
+            ("b:0",),
+            "messages",
+            {"event": "message-start", "message_id": "m-b", "__node__": "b"},
+        ),
+        (
+            ("b:0",),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "from-b"},
+                "__node__": "b",
+            },
+        ),
+        (
+            ("b:0",),
+            "messages",
+            {"event": "message-finish", "reason": "stop", "__node__": "b"},
+        ),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    msg_texts: dict[str, list[str]] = {}
+    for sub in run.subgraphs:
+        msg_texts[sub.name] = [str(m.text) for m in sub.messages]
+
+    assert msg_texts["a"] == ["from-a"]
+    assert msg_texts["b"] == ["from-b"]
+
+
+def test_sync_subgraph_raw_iter():
+    """Iterating a SubgraphRunStream directly should yield events scoped
+    to the child namespace.
+    """
+    chunks = [
+        (("child:0",), "values", {"v": 1}),
+        ((), "values", {"root": True}),
+        (("child:0",), "updates", {"node": "x"}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        events = list(sub)
+        methods = [e["method"] for e in events]
+        assert "values" in methods
+        assert "updates" in methods
+        # Root events should not appear
+        for e in events:
+            assert e["params"]["namespace"] == ["child:0"]
+
+
+def test_sync_subgraph_events_after_discovery():
+    """Events arriving after a namespace is first discovered should still
+    be visible in the child's values iteration.
+    """
+    chunks = [
+        (("child:0",), "values", {"v": 1}),  # triggers discovery
+        ((), "values", {"root": 1}),
+        (("child:0",), "values", {"v": 2}),  # after discovery
+        (("child:0",), "values", {"v": 3}),  # after discovery
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        vals = list(sub.values)
+        assert vals == [{"v": 1}, {"v": 2}, {"v": 3}]
+
+
+def test_sync_subgraph_lazy_pump():
+    """Subgraph iteration should pump the source lazily."""
+    consumed = 0
+
+    def counting_source():
+        nonlocal consumed
+        for chunk in [
+            (("child:0",), "values", {"v": 1}),
+            (("child:0",), "values", {"v": 2}),
+            (("child:0",), "values", {"v": 3}),
+            ((), "values", {"done": True}),
+        ]:
+            consumed += 1
+            yield chunk
+
+    run = create_graph_run_stream(counting_source())
+    assert consumed == 0
+
+    for sub in run.subgraphs:
+        # Discovery pumped the first event
+        it = iter(sub.values)
+        v = next(it)
+        assert v == {"v": 1}
+        # Should not have consumed everything yet
+        assert consumed < 4
+        break  # don't exhaust subgraphs
+
+
+def test_sync_subgraph_interleave_parent_values():
+    """Parent values and subgraph values should both be accessible
+    when interleaving iteration.
+    """
+    chunks = [
+        ((), "values", {"root": 1}),
+        (("child:0",), "values", {"child": 1}),
+        ((), "values", {"root": 2}),
+        (("child:0",), "values", {"child": 2}),
+        ((), "values", {"root": 3}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+
+    # First drain parent values
+    root_vals = list(run.values)
+    assert root_vals == [{"root": 1}, {"root": 2}, {"root": 3}]
+
+    # Source is exhausted, but subgraph transformers were registered
+    # via replay — subgraph iteration should still see buffered events
+    # Note: subgraphs must be iterated while source is being pumped
+    # to discover namespaces. Since we drained via values, namespace
+    # "child:0" was already discovered. But subgraphs iteration also
+    # needs to pump — and the source is exhausted. Let's verify it
+    # yields the discovered child.
+    subs = list(run.subgraphs)
+    assert len(subs) == 1
+    assert subs[0].name == "child"
+    # The child transformer was registered via replay, so it saw the events
+    vals = list(subs[0].values)
+    assert vals == [{"child": 1}, {"child": 2}]
+
+
+def test_sync_subgraph_interrupted():
+    """Subgraph .interrupted should reflect the mux's interrupt state."""
+
+    class _FakeInterrupt:
+        def __init__(self, id: str):
+            self.id = id
+
+    chunks = [
+        (("child:0",), "values", {"__interrupt__": [_FakeInterrupt("i1")]}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        # Pump to process the interrupt
+        _ = sub.output
+        assert sub.interrupted is True
+        assert len(sub.interrupts) == 1
+
+
+def test_sync_subgraph_source_error():
+    """If the source raises mid-stream, subgraphs that were already
+    discovered should still have their buffered data.
+    """
+
+    def bad_source():
+        yield (("child:0",), "values", {"v": 1})
+        yield (("child:0",), "values", {"v": 2})
+        raise ValueError("boom")
+
+    run = create_graph_run_stream(bad_source())
+    for sub in run.subgraphs:
+        vals = list(sub.values)
+        assert vals == [{"v": 1}, {"v": 2}]
+        assert run._mux._error is not None
+
+
+def test_sync_subgraph_output_drains_source():
+    """Accessing subgraph .output should drain the full source."""
+    consumed = 0
+
+    def counting_source():
+        nonlocal consumed
+        for chunk in [
+            (("child:0",), "values", {"v": 1}),
+            (("child:0",), "values", {"v": 2}),
+            ((), "values", {"done": True}),
+        ]:
+            consumed += 1
+            yield chunk
+
+    run = create_graph_run_stream(counting_source())
+    for sub in run.subgraphs:
+        result = sub.output
+        assert result == {"v": 2}
+        assert consumed == 3

libs/langgraph/tests/test_streaming_comparison.py

@@ -0,0 +1,420 @@
+"""Prove V1 and StreamingHandler APIs expose identical information.
+
+Each test runs the same graph through both APIs and asserts data
+equivalence — same state snapshots, same messages, same custom events,
+same interrupts.  Sync APIs are used where possible; async tests cover
+features without sync equivalents (subgraphs projection, messages_from).
+
+Run with:
+    TEST=tests/test_streaming_comparison.py make test
+"""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage
+from langgraph.checkpoint.memory import MemorySaver
+from typing_extensions import TypedDict
+
+from langgraph.config import get_stream_writer
+from langgraph.graph import END, START, MessagesState, StateGraph
+from langgraph.stream import StreamingHandler
+from langgraph.stream._convert import STREAM_V2_MODES
+from langgraph.types import interrupt
+from tests.fake_chat import FakeChatModel
+
+# ---------------------------------------------------------------------------
+# Graph factories
+# ---------------------------------------------------------------------------
+
+
+class State(TypedDict):
+    value: str
+    items: Annotated[list[str], lambda a, b: a + b]
+
+
+def _linear_graph(n_nodes: int = 3):
+    """Chain of *n_nodes* that concatenate strings."""
+    g = StateGraph(State)
+    names = [f"node_{i}" for i in range(n_nodes)]
+    for name in names:
+
+        def make_fn(n):
+            def fn(state: State) -> dict:
+                return {"value": state["value"] + f"_{n}", "items": [n]}
+
+            return fn
+
+        g.add_node(name, make_fn(name))
+
+    g.add_edge(START, names[0])
+    for i in range(len(names) - 1):
+        g.add_edge(names[i], names[i + 1])
+    g.add_edge(names[-1], END)
+    return g.compile()
+
+
+def _chat_graph():
+    """Single agent node with a FakeChatModel."""
+    model = FakeChatModel(messages=[AIMessage(content="Hello from agent")])
+
+    def agent(state: dict) -> dict:
+        return {"messages": [model.invoke(state["messages"])]}
+
+    g = StateGraph(MessagesState)
+    g.add_node("agent", agent)
+    g.add_edge(START, "agent")
+    g.add_edge("agent", END)
+    return g.compile()
+
+
+def _multi_node_chat_graph():
+    """Two LLM nodes: agent -> reviewer."""
+    agent_model = FakeChatModel(messages=[AIMessage(content="Agent reply")])
+    reviewer_model = FakeChatModel(messages=[AIMessage(content="Reviewer reply")])
+
+    def agent(state: dict) -> dict:
+        return {"messages": [agent_model.invoke(state["messages"])]}
+
+    def reviewer(state: dict) -> dict:
+        return {"messages": [reviewer_model.invoke(state["messages"])]}
+
+    g = StateGraph(MessagesState)
+    g.add_node("agent", agent)
+    g.add_node("reviewer", reviewer)
+    g.add_edge(START, "agent")
+    g.add_edge("agent", "reviewer")
+    g.add_edge("reviewer", END)
+    return g.compile()
+
+
+def _custom_events_graph():
+    """Node that emits custom events via StreamWriter."""
+
+    def worker(state: State) -> dict:
+        writer = get_stream_writer()
+        writer({"step": 1, "msg": "started"})
+        writer({"step": 2, "msg": "processing"})
+        writer({"step": 3, "msg": "done"})
+        return {"value": state["value"] + "_done", "items": ["done"]}
+
+    g = StateGraph(State)
+    g.add_node("worker", worker)
+    g.add_edge(START, "worker")
+    g.add_edge("worker", END)
+    return g.compile()
+
+
+def _interrupt_graph():
+    """Graph that interrupts for human input."""
+
+    def ask_human(state: State) -> dict:
+        answer = interrupt("What next?")
+        return {"value": state["value"] + f"_{answer}", "items": [answer]}
+
+    g = StateGraph(State)
+    g.add_node("ask", ask_human)
+    g.add_edge(START, "ask")
+    g.add_edge("ask", END)
+    return g.compile(checkpointer=MemorySaver())
+
+
+def _subgraph():
+    """Parent with a compiled child subgraph."""
+
+    class ChildState(TypedDict):
+        value: str
+
+    class ParentState(TypedDict):
+        value: str
+
+    def child_node(state: ChildState) -> dict:
+        return {"value": state["value"] + "_child"}
+
+    child = StateGraph(ChildState)
+    child.add_node("inner", child_node)
+    child.add_edge(START, "inner")
+    child.add_edge("inner", END)
+    child_compiled = child.compile()
+
+    parent = StateGraph(ParentState)
+    parent.add_node("child", child_compiled)
+    parent.add_edge(START, "child")
+    parent.add_edge("child", END)
+    return parent.compile()
+
+
+# ===================================================================
+# 1. Final output
+# ===================================================================
+
+
+def test_output():
+    """graph.invoke() produces the same result as StreamingHandler().stream().output."""
+    graph = _linear_graph()
+    inp = {"value": "x", "items": []}
+
+    v1 = graph.invoke(inp)
+
+    run = StreamingHandler(graph).stream(inp)
+    v2 = run.output
+
+    assert v1 == v2
+
+
+# ===================================================================
+# 2. Intermediate state snapshots (values mode)
+# ===================================================================
+
+
+def test_values():
+    """stream(mode='values') snapshots == StreamingHandler().stream().values snapshots."""
+    graph = _linear_graph()
+    inp = {"value": "x", "items": []}
+
+    v1 = list(graph.stream(inp, stream_mode="values"))
+
+    run = StreamingHandler(graph).stream(inp)
+    v2 = list(run.values)
+
+    assert v1 == v2
+
+
+# ===================================================================
+# 3. Per-node updates (updates mode)
+# ===================================================================
+
+
+def test_updates():
+    """stream(mode='updates') data == StreamingHandler raw events[method=updates]."""
+    graph = _linear_graph()
+    inp = {"value": "x", "items": []}
+
+    v1 = list(graph.stream(inp, stream_mode="updates"))
+
+    run = StreamingHandler(graph).stream(inp)
+    v2 = [
+        e["params"]["data"]
+        for e in run
+        if e["method"] == "updates" and not e["params"]["namespace"]
+    ]
+
+    assert v1 == v2
+
+
+# ===================================================================
+# 4. Message text and node attribution
+# ===================================================================
+
+
+def test_messages():
+    """Reassembled V1 message text per node == V2 .messages text per node."""
+    graph = _multi_node_chat_graph()
+    inp = {"messages": [HumanMessage(content="hi")]}
+
+    # V1: collect (chunk, metadata) pairs, group text by node
+    v1_text_by_node: dict[str, list[str]] = {}
+    for chunk, metadata in graph.stream(inp, stream_mode="messages"):
+        node = metadata["langgraph_node"]
+        v1_text_by_node.setdefault(node, []).append(chunk.content)
+    v1_text = {k: "".join(v) for k, v in v1_text_by_node.items()}
+
+    # V2: each ChatModelStream has .text and .node
+    run = StreamingHandler(graph).stream(inp)
+    v2_text: dict[str, str] = {}
+    for msg in run.messages:
+        assert msg.done is True
+        v2_text[msg.node] = msg.text
+
+    assert v1_text == v2_text
+
+
+# ===================================================================
+# 5. Custom events
+# ===================================================================
+
+
+def test_custom_events():
+    """stream(mode='custom') payloads == StreamingHandler raw events[method=custom]."""
+    graph = _custom_events_graph()
+    inp = {"value": "x", "items": []}
+
+    v1 = list(graph.stream(inp, stream_mode="custom"))
+
+    run = StreamingHandler(graph).stream(inp)
+    v2 = [
+        e["params"]["data"]
+        for e in run
+        if e["method"] == "custom" and not e["params"]["namespace"]
+    ]
+
+    assert v1 == v2
+
+
+# ===================================================================
+# 6. Mode coverage
+# ===================================================================
+
+
+def test_mode_coverage():
+    """V2 produces events for the same set of modes as V1."""
+    graph = _chat_graph()
+    inp = {"messages": [HumanMessage(content="hi")]}
+
+    # V1: request all modes, collect which ones appear
+    v1_modes: set[str] = set()
+    for ns, mode, _ in graph.stream(
+        inp, stream_mode=STREAM_V2_MODES, subgraphs=True, version="v1"
+    ):
+        if not ns:
+            v1_modes.add(mode)
+
+    # V2: iterate raw events, collect methods
+    run = StreamingHandler(graph).stream(inp)
+    v2_modes = {e["method"] for e in run if not e["params"]["namespace"]}
+
+    assert v1_modes == v2_modes
+
+
+# ===================================================================
+# 7. Interrupt detection
+# ===================================================================
+
+
+def test_interrupts():
+    """V1 __interrupt__ value == V2 .interrupted and .interrupts payload."""
+    graph = _interrupt_graph()
+    inp = {"value": "x", "items": []}
+
+    # V1: detect __interrupt__ in values stream
+    config1 = {"configurable": {"thread_id": "equiv-1"}}
+    v1_interrupt_value = None
+    for chunk in graph.stream(inp, config1, stream_mode="values"):
+        if isinstance(chunk, dict) and "__interrupt__" in chunk:
+            info = chunk["__interrupt__"]
+            if info:
+                v1_interrupt_value = info[0].value
+
+    assert v1_interrupt_value is not None
+
+    # V2: .interrupted and .interrupts (fresh thread)
+    config2 = {"configurable": {"thread_id": "equiv-2"}}
+    run = StreamingHandler(graph).stream(inp, config=config2)
+    for _ in run:
+        pass
+
+    assert run.interrupted is True
+    assert len(run.interrupts) > 0
+    v2_interrupt_value = run.interrupts[0]["payload"].value
+
+    assert v1_interrupt_value == v2_interrupt_value
+
+
+# ===================================================================
+# 8. Subgraph state snapshots
+# ===================================================================
+
+
+def test_subgraph_values():
+    """V1 child namespace values == V2 child namespace values."""
+    graph = _subgraph()
+    inp = {"value": "x"}
+
+    # V1: stream with subgraphs=True, collect child values
+    v1_child_values = []
+    for ns, data in graph.stream(inp, stream_mode="values", subgraphs=True):
+        if ns:
+            v1_child_values.append(data)
+
+    # V2: filter raw events for child namespace + values mode
+    run = StreamingHandler(graph).stream(inp)
+    v2_child_values = [
+        e["params"]["data"]
+        for e in run
+        if e["method"] == "values" and e["params"]["namespace"]
+    ]
+
+    assert v1_child_values == v2_child_values
+
+
+# ===================================================================
+# 9. Node filtering on messages
+# ===================================================================
+
+
+def test_messages_node_filtering():
+    """V1 manual metadata filter == V2 .messages filtered by .node."""
+    graph = _multi_node_chat_graph()
+    inp = {"messages": [HumanMessage(content="hi")]}
+
+    # V1: manual filter for "agent" node only
+    v1_agent_text: list[str] = []
+    for chunk, metadata in graph.stream(inp, stream_mode="messages"):
+        if metadata.get("langgraph_node") == "agent":
+            v1_agent_text.append(chunk.content)
+    v1_text = "".join(v1_agent_text)
+
+    # V2: filter .messages by .node
+    run = StreamingHandler(graph).stream(inp)
+    v2_agent_msgs = [msg for msg in run.messages if msg.node == "agent"]
+    assert len(v2_agent_msgs) == 1
+    v2_text = v2_agent_msgs[0].text
+
+    assert v1_text == v2_text
+
+
+# ===================================================================
+# 10. Async: subgraphs projection
+# ===================================================================
+
+
+@pytest.mark.anyio
+async def test_async_subgraph_projection():
+    """V2 .subgraphs child output matches V1 child namespace output."""
+    graph = _subgraph()
+    inp = {"value": "x"}
+
+    # V1
+    v1_child_output = None
+    async for ns, data in graph.astream(inp, stream_mode="values", subgraphs=True):
+        if ns:
+            v1_child_output = data
+
+    # V2: .subgraphs yields typed child stream objects
+    run = await StreamingHandler(graph).astream(inp)
+    v2_child_output = None
+    async for sub in run.subgraphs:
+        v2_child_output = await sub.output
+
+    assert v1_child_output == v2_child_output
+
+
+# ===================================================================
+# 11. Async: messages_from projection
+# ===================================================================
+
+
+@pytest.mark.anyio
+async def test_async_messages_from():
+    """V2 .messages_from('agent') text matches V1 filtered by metadata."""
+    graph = _multi_node_chat_graph()
+    inp = {"messages": [HumanMessage(content="hi")]}
+
+    # V1: manual filter for agent node
+    v1_agent_text: list[str] = []
+    async for chunk, metadata in graph.astream(inp, stream_mode="messages"):
+        if metadata.get("langgraph_node") == "agent":
+            v1_agent_text.append(chunk.content)
+    v1_text = "".join(v1_agent_text)
+
+    # V2: declarative node filtering
+    run = await StreamingHandler(graph).astream(inp)
+    v2_texts: list[str] = []
+    async for msg in run.messages_from("agent"):
+        v2_texts.append(await msg.text)
+    assert len(v2_texts) == 1
+    v2_text = v2_texts[0]
+
+    assert v1_text == v2_text

libs/langgraph/uv.lock

@@ -524,61 +524,61 @@ toml = [
 
 [[package]]
 name = "cryptography"
-version = "46.0.6"
+version = "46.0.7"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "cffi", marker = "python_full_version >= '3.11' and python_full_version < '3.14' and platform_python_implementation != 'PyPy'" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/a4/ba/04b1bd4218cbc58dc90ce967106d51582371b898690f3ae0402876cc4f34/cryptography-46.0.6.tar.gz", hash = "sha256:27550628a518c5c6c903d84f637fbecf287f6cb9ced3804838a1295dc1fd0759", size = 750542, upload-time = "2026-03-25T23:34:53.396Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/47/93/ac8f3d5ff04d54bc814e961a43ae5b0b146154c89c61b47bb07557679b18/cryptography-46.0.7.tar.gz", hash = "sha256:e4cfd68c5f3e0bfdad0d38e023239b96a2fe84146481852dffbcca442c245aa5", size = 750652, upload-time = "2026-04-08T01:57:54.692Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/47/23/9285e15e3bc57325b0a72e592921983a701efc1ee8f91c06c5f0235d86d9/cryptography-46.0.6-cp311-abi3-macosx_10_9_universal2.whl", hash = "sha256:64235194bad039a10bb6d2d930ab3323baaec67e2ce36215fd0952fad0930ca8", size = 7176401, upload-time = "2026-03-25T23:33:22.096Z" },
-    { url = "https://files.pythonhosted.org/packages/60/f8/e61f8f13950ab6195b31913b42d39f0f9afc7d93f76710f299b5ec286ae6/cryptography-46.0.6-cp311-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:26031f1e5ca62fcb9d1fcb34b2b60b390d1aacaa15dc8b895a9ed00968b97b30", size = 4275275, upload-time = "2026-03-25T23:33:23.844Z" },
-    { url = "https://files.pythonhosted.org/packages/19/69/732a736d12c2631e140be2348b4ad3d226302df63ef64d30dfdb8db7ad1c/cryptography-46.0.6-cp311-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:9a693028b9cbe51b5a1136232ee8f2bc242e4e19d456ded3fa7c86e43c713b4a", size = 4425320, upload-time = "2026-03-25T23:33:25.703Z" },
-    { url = "https://files.pythonhosted.org/packages/d4/12/123be7292674abf76b21ac1fc0e1af50661f0e5b8f0ec8285faac18eb99e/cryptography-46.0.6-cp311-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:67177e8a9f421aa2d3a170c3e56eca4e0128883cf52a071a7cbf53297f18b175", size = 4278082, upload-time = "2026-03-25T23:33:27.423Z" },
-    { url = "https://files.pythonhosted.org/packages/5b/ba/d5e27f8d68c24951b0a484924a84c7cdaed7502bac9f18601cd357f8b1d2/cryptography-46.0.6-cp311-abi3-manylinux_2_28_ppc64le.whl", hash = "sha256:d9528b535a6c4f8ff37847144b8986a9a143585f0540fbcb1a98115b543aa463", size = 4926514, upload-time = "2026-03-25T23:33:29.206Z" },
-    { url = "https://files.pythonhosted.org/packages/34/71/1ea5a7352ae516d5512d17babe7e1b87d9db5150b21f794b1377eac1edc0/cryptography-46.0.6-cp311-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:22259338084d6ae497a19bae5d4c66b7ca1387d3264d1c2c0e72d9e9b6a77b97", size = 4457766, upload-time = "2026-03-25T23:33:30.834Z" },
-    { url = "https://files.pythonhosted.org/packages/01/59/562be1e653accee4fdad92c7a2e88fced26b3fdfce144047519bbebc299e/cryptography-46.0.6-cp311-abi3-manylinux_2_31_armv7l.whl", hash = "sha256:760997a4b950ff00d418398ad73fbc91aa2894b5c1db7ccb45b4f68b42a63b3c", size = 3986535, upload-time = "2026-03-25T23:33:33.02Z" },
-    { url = "https://files.pythonhosted.org/packages/d6/8b/b1ebfeb788bf4624d36e45ed2662b8bd43a05ff62157093c1539c1288a18/cryptography-46.0.6-cp311-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:3dfa6567f2e9e4c5dceb8ccb5a708158a2a871052fa75c8b78cb0977063f1507", size = 4277618, upload-time = "2026-03-25T23:33:34.567Z" },
-    { url = "https://files.pythonhosted.org/packages/dd/52/a005f8eabdb28df57c20f84c44d397a755782d6ff6d455f05baa2785bd91/cryptography-46.0.6-cp311-abi3-manylinux_2_34_ppc64le.whl", hash = "sha256:cdcd3edcbc5d55757e5f5f3d330dd00007ae463a7e7aa5bf132d1f22a4b62b19", size = 4890802, upload-time = "2026-03-25T23:33:37.034Z" },
-    { url = "https://files.pythonhosted.org/packages/ec/4d/8e7d7245c79c617d08724e2efa397737715ca0ec830ecb3c91e547302555/cryptography-46.0.6-cp311-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:d4e4aadb7fc1f88687f47ca20bb7227981b03afaae69287029da08096853b738", size = 4457425, upload-time = "2026-03-25T23:33:38.904Z" },
-    { url = "https://files.pythonhosted.org/packages/1d/5c/f6c3596a1430cec6f949085f0e1a970638d76f81c3ea56d93d564d04c340/cryptography-46.0.6-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:2b417edbe8877cda9022dde3a008e2deb50be9c407eef034aeeb3a8b11d9db3c", size = 4405530, upload-time = "2026-03-25T23:33:40.842Z" },
-    { url = "https://files.pythonhosted.org/packages/7e/c9/9f9cea13ee2dbde070424e0c4f621c091a91ffcc504ffea5e74f0e1daeff/cryptography-46.0.6-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:380343e0653b1c9d7e1f55b52aaa2dbb2fdf2730088d48c43ca1c7c0abb7cc2f", size = 4667896, upload-time = "2026-03-25T23:33:42.781Z" },
-    { url = "https://files.pythonhosted.org/packages/ad/b5/1895bc0821226f129bc74d00eccfc6a5969e2028f8617c09790bf89c185e/cryptography-46.0.6-cp311-abi3-win32.whl", hash = "sha256:bcb87663e1f7b075e48c3be3ecb5f0b46c8fc50b50a97cf264e7f60242dca3f2", size = 3026348, upload-time = "2026-03-25T23:33:45.021Z" },
-    { url = "https://files.pythonhosted.org/packages/c3/f8/c9bcbf0d3e6ad288b9d9aa0b1dee04b063d19e8c4f871855a03ab3a297ab/cryptography-46.0.6-cp311-abi3-win_amd64.whl", hash = "sha256:6739d56300662c468fddb0e5e291f9b4d084bead381667b9e654c7dd81705124", size = 3483896, upload-time = "2026-03-25T23:33:46.649Z" },
-    { url = "https://files.pythonhosted.org/packages/01/41/3a578f7fd5c70611c0aacba52cd13cb364a5dee895a5c1d467208a9380b0/cryptography-46.0.6-cp314-cp314t-macosx_10_9_universal2.whl", hash = "sha256:2ef9e69886cbb137c2aef9772c2e7138dc581fad4fcbcf13cc181eb5a3ab6275", size = 7117147, upload-time = "2026-03-25T23:33:48.249Z" },
-    { url = "https://files.pythonhosted.org/packages/fa/87/887f35a6fca9dde90cad08e0de0c89263a8e59b2d2ff904fd9fcd8025b6f/cryptography-46.0.6-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:7f417f034f91dcec1cb6c5c35b07cdbb2ef262557f701b4ecd803ee8cefed4f4", size = 4266221, upload-time = "2026-03-25T23:33:49.874Z" },
-    { url = "https://files.pythonhosted.org/packages/aa/a8/0a90c4f0b0871e0e3d1ed126aed101328a8a57fd9fd17f00fb67e82a51ca/cryptography-46.0.6-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:d24c13369e856b94892a89ddf70b332e0b70ad4a5c43cf3e9cb71d6d7ffa1f7b", size = 4408952, upload-time = "2026-03-25T23:33:52.128Z" },
-    { url = "https://files.pythonhosted.org/packages/16/0b/b239701eb946523e4e9f329336e4ff32b1247e109cbab32d1a7b61da8ed7/cryptography-46.0.6-cp314-cp314t-manylinux_2_28_aarch64.whl", hash = "sha256:aad75154a7ac9039936d50cf431719a2f8d4ed3d3c277ac03f3339ded1a5e707", size = 4270141, upload-time = "2026-03-25T23:33:54.11Z" },
-    { url = "https://files.pythonhosted.org/packages/0f/a8/976acdd4f0f30df7b25605f4b9d3d89295351665c2091d18224f7ad5cdbf/cryptography-46.0.6-cp314-cp314t-manylinux_2_28_ppc64le.whl", hash = "sha256:3c21d92ed15e9cfc6eb64c1f5a0326db22ca9c2566ca46d845119b45b4400361", size = 4904178, upload-time = "2026-03-25T23:33:55.725Z" },
-    { url = "https://files.pythonhosted.org/packages/b1/1b/bf0e01a88efd0e59679b69f42d4afd5bced8700bb5e80617b2d63a3741af/cryptography-46.0.6-cp314-cp314t-manylinux_2_28_x86_64.whl", hash = "sha256:4668298aef7cddeaf5c6ecc244c2302a2b8e40f384255505c22875eebb47888b", size = 4441812, upload-time = "2026-03-25T23:33:57.364Z" },
-    { url = "https://files.pythonhosted.org/packages/bb/8b/11df86de2ea389c65aa1806f331cae145f2ed18011f30234cc10ca253de8/cryptography-46.0.6-cp314-cp314t-manylinux_2_31_armv7l.whl", hash = "sha256:8ce35b77aaf02f3b59c90b2c8a05c73bac12cea5b4e8f3fbece1f5fddea5f0ca", size = 3963923, upload-time = "2026-03-25T23:33:59.361Z" },
-    { url = "https://files.pythonhosted.org/packages/91/e0/207fb177c3a9ef6a8108f234208c3e9e76a6aa8cf20d51932916bd43bda0/cryptography-46.0.6-cp314-cp314t-manylinux_2_34_aarch64.whl", hash = "sha256:c89eb37fae9216985d8734c1afd172ba4927f5a05cfd9bf0e4863c6d5465b013", size = 4269695, upload-time = "2026-03-25T23:34:00.909Z" },
-    { url = "https://files.pythonhosted.org/packages/21/5e/19f3260ed1e95bced52ace7501fabcd266df67077eeb382b79c81729d2d3/cryptography-46.0.6-cp314-cp314t-manylinux_2_34_ppc64le.whl", hash = "sha256:ed418c37d095aeddf5336898a132fba01091f0ac5844e3e8018506f014b6d2c4", size = 4869785, upload-time = "2026-03-25T23:34:02.796Z" },
-    { url = "https://files.pythonhosted.org/packages/10/38/cd7864d79aa1d92ef6f1a584281433419b955ad5a5ba8d1eb6c872165bcb/cryptography-46.0.6-cp314-cp314t-manylinux_2_34_x86_64.whl", hash = "sha256:69cf0056d6947edc6e6760e5f17afe4bea06b56a9ac8a06de9d2bd6b532d4f3a", size = 4441404, upload-time = "2026-03-25T23:34:04.35Z" },
-    { url = "https://files.pythonhosted.org/packages/09/0a/4fe7a8d25fed74419f91835cf5829ade6408fd1963c9eae9c4bce390ecbb/cryptography-46.0.6-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:8e7304c4f4e9490e11efe56af6713983460ee0780f16c63f219984dab3af9d2d", size = 4397549, upload-time = "2026-03-25T23:34:06.342Z" },
-    { url = "https://files.pythonhosted.org/packages/5f/a0/7d738944eac6513cd60a8da98b65951f4a3b279b93479a7e8926d9cd730b/cryptography-46.0.6-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:b928a3ca837c77a10e81a814a693f2295200adb3352395fad024559b7be7a736", size = 4651874, upload-time = "2026-03-25T23:34:07.916Z" },
-    { url = "https://files.pythonhosted.org/packages/cb/f1/c2326781ca05208845efca38bf714f76939ae446cd492d7613808badedf1/cryptography-46.0.6-cp314-cp314t-win32.whl", hash = "sha256:97c8115b27e19e592a05c45d0dd89c57f81f841cc9880e353e0d3bf25b2139ed", size = 3001511, upload-time = "2026-03-25T23:34:09.892Z" },
-    { url = "https://files.pythonhosted.org/packages/c9/57/fe4a23eb549ac9d903bd4698ffda13383808ef0876cc912bcb2838799ece/cryptography-46.0.6-cp314-cp314t-win_amd64.whl", hash = "sha256:c797e2517cb7880f8297e2c0f43bb910e91381339336f75d2c1c2cbf811b70b4", size = 3471692, upload-time = "2026-03-25T23:34:11.613Z" },
-    { url = "https://files.pythonhosted.org/packages/c4/cc/f330e982852403da79008552de9906804568ae9230da8432f7496ce02b71/cryptography-46.0.6-cp38-abi3-macosx_10_9_universal2.whl", hash = "sha256:12cae594e9473bca1a7aceb90536060643128bb274fcea0fc459ab90f7d1ae7a", size = 7162776, upload-time = "2026-03-25T23:34:13.308Z" },
-    { url = "https://files.pythonhosted.org/packages/49/b3/dc27efd8dcc4bff583b3f01d4a3943cd8b5821777a58b3a6a5f054d61b79/cryptography-46.0.6-cp38-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:639301950939d844a9e1c4464d7e07f902fe9a7f6b215bb0d4f28584729935d8", size = 4270529, upload-time = "2026-03-25T23:34:15.019Z" },
-    { url = "https://files.pythonhosted.org/packages/e6/05/e8d0e6eb4f0d83365b3cb0e00eb3c484f7348db0266652ccd84632a3d58d/cryptography-46.0.6-cp38-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:ed3775295fb91f70b4027aeba878d79b3e55c0b3e97eaa4de71f8f23a9f2eb77", size = 4414827, upload-time = "2026-03-25T23:34:16.604Z" },
-    { url = "https://files.pythonhosted.org/packages/2f/97/daba0f5d2dc6d855e2dcb70733c812558a7977a55dd4a6722756628c44d1/cryptography-46.0.6-cp38-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:8927ccfbe967c7df312ade694f987e7e9e22b2425976ddbf28271d7e58845290", size = 4271265, upload-time = "2026-03-25T23:34:18.586Z" },
-    { url = "https://files.pythonhosted.org/packages/89/06/fe1fce39a37ac452e58d04b43b0855261dac320a2ebf8f5260dd55b201a9/cryptography-46.0.6-cp38-abi3-manylinux_2_28_ppc64le.whl", hash = "sha256:b12c6b1e1651e42ab5de8b1e00dc3b6354fdfd778e7fa60541ddacc27cd21410", size = 4916800, upload-time = "2026-03-25T23:34:20.561Z" },
-    { url = "https://files.pythonhosted.org/packages/ff/8a/b14f3101fe9c3592603339eb5d94046c3ce5f7fc76d6512a2d40efd9724e/cryptography-46.0.6-cp38-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:063b67749f338ca9c5a0b7fe438a52c25f9526b851e24e6c9310e7195aad3b4d", size = 4448771, upload-time = "2026-03-25T23:34:22.406Z" },
-    { url = "https://files.pythonhosted.org/packages/01/b3/0796998056a66d1973fd52ee89dc1bb3b6581960a91ad4ac705f182d398f/cryptography-46.0.6-cp38-abi3-manylinux_2_31_armv7l.whl", hash = "sha256:02fad249cb0e090b574e30b276a3da6a149e04ee2f049725b1f69e7b8351ec70", size = 3978333, upload-time = "2026-03-25T23:34:24.281Z" },
-    { url = "https://files.pythonhosted.org/packages/c5/3d/db200af5a4ffd08918cd55c08399dc6c9c50b0bc72c00a3246e099d3a849/cryptography-46.0.6-cp38-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:7e6142674f2a9291463e5e150090b95a8519b2fb6e6aaec8917dd8d094ce750d", size = 4271069, upload-time = "2026-03-25T23:34:25.895Z" },
-    { url = "https://files.pythonhosted.org/packages/d7/18/61acfd5b414309d74ee838be321c636fe71815436f53c9f0334bf19064fa/cryptography-46.0.6-cp38-abi3-manylinux_2_34_ppc64le.whl", hash = "sha256:456b3215172aeefb9284550b162801d62f5f264a081049a3e94307fe20792cfa", size = 4878358, upload-time = "2026-03-25T23:34:27.67Z" },
-    { url = "https://files.pythonhosted.org/packages/8b/65/5bf43286d566f8171917cae23ac6add941654ccf085d739195a4eacf1674/cryptography-46.0.6-cp38-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:341359d6c9e68834e204ceaf25936dffeafea3829ab80e9503860dcc4f4dac58", size = 4448061, upload-time = "2026-03-25T23:34:29.375Z" },
-    { url = "https://files.pythonhosted.org/packages/e0/25/7e49c0fa7205cf3597e525d156a6bce5b5c9de1fd7e8cb01120e459f205a/cryptography-46.0.6-cp38-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:9a9c42a2723999a710445bc0d974e345c32adfd8d2fac6d8a251fa829ad31cfb", size = 4399103, upload-time = "2026-03-25T23:34:32.036Z" },
-    { url = "https://files.pythonhosted.org/packages/44/46/466269e833f1c4718d6cd496ffe20c56c9c8d013486ff66b4f69c302a68d/cryptography-46.0.6-cp38-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:6617f67b1606dfd9fe4dbfa354a9508d4a6d37afe30306fe6c101b7ce3274b72", size = 4659255, upload-time = "2026-03-25T23:34:33.679Z" },
-    { url = "https://files.pythonhosted.org/packages/0a/09/ddc5f630cc32287d2c953fc5d32705e63ec73e37308e5120955316f53827/cryptography-46.0.6-cp38-abi3-win32.whl", hash = "sha256:7f6690b6c55e9c5332c0b59b9c8a3fb232ebf059094c17f9019a51e9827df91c", size = 3010660, upload-time = "2026-03-25T23:34:35.418Z" },
-    { url = "https://files.pythonhosted.org/packages/1b/82/ca4893968aeb2709aacfb57a30dec6fa2ab25b10fa9f064b8882ce33f599/cryptography-46.0.6-cp38-abi3-win_amd64.whl", hash = "sha256:79e865c642cfc5c0b3eb12af83c35c5aeff4fa5c672dc28c43721c2c9fdd2f0f", size = 3471160, upload-time = "2026-03-25T23:34:37.191Z" },
-    { url = "https://files.pythonhosted.org/packages/2e/84/7ccff00ced5bac74b775ce0beb7d1be4e8637536b522b5df9b73ada42da2/cryptography-46.0.6-pp311-pypy311_pp73-macosx_11_0_arm64.whl", hash = "sha256:2ea0f37e9a9cf0df2952893ad145fd9627d326a59daec9b0802480fa3bcd2ead", size = 3475444, upload-time = "2026-03-25T23:34:38.944Z" },
-    { url = "https://files.pythonhosted.org/packages/bc/1f/4c926f50df7749f000f20eede0c896769509895e2648db5da0ed55db711d/cryptography-46.0.6-pp311-pypy311_pp73-manylinux_2_28_aarch64.whl", hash = "sha256:a3e84d5ec9ba01f8fd03802b2147ba77f0c8f2617b2aff254cedd551844209c8", size = 4218227, upload-time = "2026-03-25T23:34:40.871Z" },
-    { url = "https://files.pythonhosted.org/packages/c6/65/707be3ffbd5f786028665c3223e86e11c4cda86023adbc56bd72b1b6bab5/cryptography-46.0.6-pp311-pypy311_pp73-manylinux_2_28_x86_64.whl", hash = "sha256:12f0fa16cc247b13c43d56d7b35287ff1569b5b1f4c5e87e92cc4fcc00cd10c0", size = 4381399, upload-time = "2026-03-25T23:34:42.609Z" },
-    { url = "https://files.pythonhosted.org/packages/f3/6d/73557ed0ef7d73d04d9aba745d2c8e95218213687ee5e76b7d236a5030fc/cryptography-46.0.6-pp311-pypy311_pp73-manylinux_2_34_aarch64.whl", hash = "sha256:50575a76e2951fe7dbd1f56d181f8c5ceeeb075e9ff88e7ad997d2f42af06e7b", size = 4217595, upload-time = "2026-03-25T23:34:44.205Z" },
-    { url = "https://files.pythonhosted.org/packages/9e/c5/e1594c4eec66a567c3ac4400008108a415808be2ce13dcb9a9045c92f1a0/cryptography-46.0.6-pp311-pypy311_pp73-manylinux_2_34_x86_64.whl", hash = "sha256:90e5f0a7b3be5f40c3a0a0eafb32c681d8d2c181fc2a1bdabe9b3f611d9f6b1a", size = 4380912, upload-time = "2026-03-25T23:34:46.328Z" },
-    { url = "https://files.pythonhosted.org/packages/1a/89/843b53614b47f97fe1abc13f9a86efa5ec9e275292c457af1d4a60dc80e0/cryptography-46.0.6-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:6728c49e3b2c180ef26f8e9f0a883a2c585638db64cf265b49c9ba10652d430e", size = 3409955, upload-time = "2026-03-25T23:34:48.465Z" },
+    { url = "https://files.pythonhosted.org/packages/0b/5d/4a8f770695d73be252331e60e526291e3df0c9b27556a90a6b47bccca4c2/cryptography-46.0.7-cp311-abi3-macosx_10_9_universal2.whl", hash = "sha256:ea42cbe97209df307fdc3b155f1b6fa2577c0defa8f1f7d3be7d31d189108ad4", size = 7179869, upload-time = "2026-04-08T01:56:17.157Z" },
+    { url = "https://files.pythonhosted.org/packages/5f/45/6d80dc379b0bbc1f9d1e429f42e4cb9e1d319c7a8201beffd967c516ea01/cryptography-46.0.7-cp311-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:b36a4695e29fe69215d75960b22577197aca3f7a25b9cf9d165dcfe9d80bc325", size = 4275492, upload-time = "2026-04-08T01:56:19.36Z" },
+    { url = "https://files.pythonhosted.org/packages/4a/9a/1765afe9f572e239c3469f2cb429f3ba7b31878c893b246b4b2994ffe2fe/cryptography-46.0.7-cp311-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:5ad9ef796328c5e3c4ceed237a183f5d41d21150f972455a9d926593a1dcb308", size = 4426670, upload-time = "2026-04-08T01:56:21.415Z" },
+    { url = "https://files.pythonhosted.org/packages/8f/3e/af9246aaf23cd4ee060699adab1e47ced3f5f7e7a8ffdd339f817b446462/cryptography-46.0.7-cp311-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:73510b83623e080a2c35c62c15298096e2a5dc8d51c3b4e1740211839d0dea77", size = 4280275, upload-time = "2026-04-08T01:56:23.539Z" },
+    { url = "https://files.pythonhosted.org/packages/0f/54/6bbbfc5efe86f9d71041827b793c24811a017c6ac0fd12883e4caa86b8ed/cryptography-46.0.7-cp311-abi3-manylinux_2_28_ppc64le.whl", hash = "sha256:cbd5fb06b62bd0721e1170273d3f4d5a277044c47ca27ee257025146c34cbdd1", size = 4928402, upload-time = "2026-04-08T01:56:25.624Z" },
+    { url = "https://files.pythonhosted.org/packages/2d/cf/054b9d8220f81509939599c8bdbc0c408dbd2bdd41688616a20731371fe0/cryptography-46.0.7-cp311-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:420b1e4109cc95f0e5700eed79908cef9268265c773d3a66f7af1eef53d409ef", size = 4459985, upload-time = "2026-04-08T01:56:27.309Z" },
+    { url = "https://files.pythonhosted.org/packages/f9/46/4e4e9c6040fb01c7467d47217d2f882daddeb8828f7df800cb806d8a2288/cryptography-46.0.7-cp311-abi3-manylinux_2_31_armv7l.whl", hash = "sha256:24402210aa54baae71d99441d15bb5a1919c195398a87b563df84468160a65de", size = 3990652, upload-time = "2026-04-08T01:56:29.095Z" },
+    { url = "https://files.pythonhosted.org/packages/36/5f/313586c3be5a2fbe87e4c9a254207b860155a8e1f3cca99f9910008e7d08/cryptography-46.0.7-cp311-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:8a469028a86f12eb7d2fe97162d0634026d92a21f3ae0ac87ed1c4a447886c83", size = 4279805, upload-time = "2026-04-08T01:56:30.928Z" },
+    { url = "https://files.pythonhosted.org/packages/69/33/60dfc4595f334a2082749673386a4d05e4f0cf4df8248e63b2c3437585f2/cryptography-46.0.7-cp311-abi3-manylinux_2_34_ppc64le.whl", hash = "sha256:9694078c5d44c157ef3162e3bf3946510b857df5a3955458381d1c7cfc143ddb", size = 4892883, upload-time = "2026-04-08T01:56:32.614Z" },
+    { url = "https://files.pythonhosted.org/packages/c7/0b/333ddab4270c4f5b972f980adef4faa66951a4aaf646ca067af597f15563/cryptography-46.0.7-cp311-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:42a1e5f98abb6391717978baf9f90dc28a743b7d9be7f0751a6f56a75d14065b", size = 4459756, upload-time = "2026-04-08T01:56:34.306Z" },
+    { url = "https://files.pythonhosted.org/packages/d2/14/633913398b43b75f1234834170947957c6b623d1701ffc7a9600da907e89/cryptography-46.0.7-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:91bbcb08347344f810cbe49065914fe048949648f6bd5c2519f34619142bbe85", size = 4410244, upload-time = "2026-04-08T01:56:35.977Z" },
+    { url = "https://files.pythonhosted.org/packages/10/f2/19ceb3b3dc14009373432af0c13f46aa08e3ce334ec6eff13492e1812ccd/cryptography-46.0.7-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:5d1c02a14ceb9148cc7816249f64f623fbfee39e8c03b3650d842ad3f34d637e", size = 4674868, upload-time = "2026-04-08T01:56:38.034Z" },
+    { url = "https://files.pythonhosted.org/packages/1a/bb/a5c213c19ee94b15dfccc48f363738633a493812687f5567addbcbba9f6f/cryptography-46.0.7-cp311-abi3-win32.whl", hash = "sha256:d23c8ca48e44ee015cd0a54aeccdf9f09004eba9fc96f38c911011d9ff1bd457", size = 3026504, upload-time = "2026-04-08T01:56:39.666Z" },
+    { url = "https://files.pythonhosted.org/packages/2b/02/7788f9fefa1d060ca68717c3901ae7fffa21ee087a90b7f23c7a603c32ae/cryptography-46.0.7-cp311-abi3-win_amd64.whl", hash = "sha256:397655da831414d165029da9bc483bed2fe0e75dde6a1523ec2fe63f3c46046b", size = 3488363, upload-time = "2026-04-08T01:56:41.893Z" },
+    { url = "https://files.pythonhosted.org/packages/7b/56/15619b210e689c5403bb0540e4cb7dbf11a6bf42e483b7644e471a2812b3/cryptography-46.0.7-cp314-cp314t-macosx_10_9_universal2.whl", hash = "sha256:d151173275e1728cf7839aaa80c34fe550c04ddb27b34f48c232193df8db5842", size = 7119671, upload-time = "2026-04-08T01:56:44Z" },
+    { url = "https://files.pythonhosted.org/packages/74/66/e3ce040721b0b5599e175ba91ab08884c75928fbeb74597dd10ef13505d2/cryptography-46.0.7-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:db0f493b9181c7820c8134437eb8b0b4792085d37dbb24da050476ccb664e59c", size = 4268551, upload-time = "2026-04-08T01:56:46.071Z" },
+    { url = "https://files.pythonhosted.org/packages/03/11/5e395f961d6868269835dee1bafec6a1ac176505a167f68b7d8818431068/cryptography-46.0.7-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:ebd6daf519b9f189f85c479427bbd6e9c9037862cf8fe89ee35503bd209ed902", size = 4408887, upload-time = "2026-04-08T01:56:47.718Z" },
+    { url = "https://files.pythonhosted.org/packages/40/53/8ed1cf4c3b9c8e611e7122fb56f1c32d09e1fff0f1d77e78d9ff7c82653e/cryptography-46.0.7-cp314-cp314t-manylinux_2_28_aarch64.whl", hash = "sha256:b7b412817be92117ec5ed95f880defe9cf18a832e8cafacf0a22337dc1981b4d", size = 4271354, upload-time = "2026-04-08T01:56:49.312Z" },
+    { url = "https://files.pythonhosted.org/packages/50/46/cf71e26025c2e767c5609162c866a78e8a2915bbcfa408b7ca495c6140c4/cryptography-46.0.7-cp314-cp314t-manylinux_2_28_ppc64le.whl", hash = "sha256:fbfd0e5f273877695cb93baf14b185f4878128b250cc9f8e617ea0c025dfb022", size = 4905845, upload-time = "2026-04-08T01:56:50.916Z" },
+    { url = "https://files.pythonhosted.org/packages/c0/ea/01276740375bac6249d0a971ebdf6b4dc9ead0ee0a34ef3b5a88c1a9b0d4/cryptography-46.0.7-cp314-cp314t-manylinux_2_28_x86_64.whl", hash = "sha256:ffca7aa1d00cf7d6469b988c581598f2259e46215e0140af408966a24cf086ce", size = 4444641, upload-time = "2026-04-08T01:56:52.882Z" },
+    { url = "https://files.pythonhosted.org/packages/3d/4c/7d258f169ae71230f25d9f3d06caabcff8c3baf0978e2b7d65e0acac3827/cryptography-46.0.7-cp314-cp314t-manylinux_2_31_armv7l.whl", hash = "sha256:60627cf07e0d9274338521205899337c5d18249db56865f943cbe753aa96f40f", size = 3967749, upload-time = "2026-04-08T01:56:54.597Z" },
+    { url = "https://files.pythonhosted.org/packages/b5/2a/2ea0767cad19e71b3530e4cad9605d0b5e338b6a1e72c37c9c1ceb86c333/cryptography-46.0.7-cp314-cp314t-manylinux_2_34_aarch64.whl", hash = "sha256:80406c3065e2c55d7f49a9550fe0c49b3f12e5bfff5dedb727e319e1afb9bf99", size = 4270942, upload-time = "2026-04-08T01:56:56.416Z" },
+    { url = "https://files.pythonhosted.org/packages/41/3d/fe14df95a83319af25717677e956567a105bb6ab25641acaa093db79975d/cryptography-46.0.7-cp314-cp314t-manylinux_2_34_ppc64le.whl", hash = "sha256:c5b1ccd1239f48b7151a65bc6dd54bcfcc15e028c8ac126d3fada09db0e07ef1", size = 4871079, upload-time = "2026-04-08T01:56:58.31Z" },
+    { url = "https://files.pythonhosted.org/packages/9c/59/4a479e0f36f8f378d397f4eab4c850b4ffb79a2f0d58704b8fa0703ddc11/cryptography-46.0.7-cp314-cp314t-manylinux_2_34_x86_64.whl", hash = "sha256:d5f7520159cd9c2154eb61eb67548ca05c5774d39e9c2c4339fd793fe7d097b2", size = 4443999, upload-time = "2026-04-08T01:57:00.508Z" },
+    { url = "https://files.pythonhosted.org/packages/28/17/b59a741645822ec6d04732b43c5d35e4ef58be7bfa84a81e5ae6f05a1d33/cryptography-46.0.7-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:fcd8eac50d9138c1d7fc53a653ba60a2bee81a505f9f8850b6b2888555a45d0e", size = 4399191, upload-time = "2026-04-08T01:57:02.654Z" },
+    { url = "https://files.pythonhosted.org/packages/59/6a/bb2e166d6d0e0955f1e9ff70f10ec4b2824c9cfcdb4da772c7dd69cc7d80/cryptography-46.0.7-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:65814c60f8cc400c63131584e3e1fad01235edba2614b61fbfbfa954082db0ee", size = 4655782, upload-time = "2026-04-08T01:57:04.592Z" },
+    { url = "https://files.pythonhosted.org/packages/95/b6/3da51d48415bcb63b00dc17c2eff3a651b7c4fed484308d0f19b30e8cb2c/cryptography-46.0.7-cp314-cp314t-win32.whl", hash = "sha256:fdd1736fed309b4300346f88f74cd120c27c56852c3838cab416e7a166f67298", size = 3002227, upload-time = "2026-04-08T01:57:06.91Z" },
+    { url = "https://files.pythonhosted.org/packages/32/a8/9f0e4ed57ec9cebe506e58db11ae472972ecb0c659e4d52bbaee80ca340a/cryptography-46.0.7-cp314-cp314t-win_amd64.whl", hash = "sha256:e06acf3c99be55aa3b516397fe42f5855597f430add9c17fa46bf2e0fb34c9bb", size = 3475332, upload-time = "2026-04-08T01:57:08.807Z" },
+    { url = "https://files.pythonhosted.org/packages/a7/7f/cd42fc3614386bc0c12f0cb3c4ae1fc2bbca5c9662dfed031514911d513d/cryptography-46.0.7-cp38-abi3-macosx_10_9_universal2.whl", hash = "sha256:462ad5cb1c148a22b2e3bcc5ad52504dff325d17daf5df8d88c17dda1f75f2a4", size = 7165618, upload-time = "2026-04-08T01:57:10.645Z" },
+    { url = "https://files.pythonhosted.org/packages/a5/d0/36a49f0262d2319139d2829f773f1b97ef8aef7f97e6e5bd21455e5a8fb5/cryptography-46.0.7-cp38-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:84d4cced91f0f159a7ddacad249cc077e63195c36aac40b4150e7a57e84fffe7", size = 4270628, upload-time = "2026-04-08T01:57:12.885Z" },
+    { url = "https://files.pythonhosted.org/packages/8a/6c/1a42450f464dda6ffbe578a911f773e54dd48c10f9895a23a7e88b3e7db5/cryptography-46.0.7-cp38-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:128c5edfe5e5938b86b03941e94fac9ee793a94452ad1365c9fc3f4f62216832", size = 4415405, upload-time = "2026-04-08T01:57:14.923Z" },
+    { url = "https://files.pythonhosted.org/packages/9a/92/4ed714dbe93a066dc1f4b4581a464d2d7dbec9046f7c8b7016f5286329e2/cryptography-46.0.7-cp38-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:5e51be372b26ef4ba3de3c167cd3d1022934bc838ae9eaad7e644986d2a3d163", size = 4272715, upload-time = "2026-04-08T01:57:16.638Z" },
+    { url = "https://files.pythonhosted.org/packages/b7/e6/a26b84096eddd51494bba19111f8fffe976f6a09f132706f8f1bf03f51f7/cryptography-46.0.7-cp38-abi3-manylinux_2_28_ppc64le.whl", hash = "sha256:cdf1a610ef82abb396451862739e3fc93b071c844399e15b90726ef7470eeaf2", size = 4918400, upload-time = "2026-04-08T01:57:19.021Z" },
+    { url = "https://files.pythonhosted.org/packages/c7/08/ffd537b605568a148543ac3c2b239708ae0bd635064bab41359252ef88ed/cryptography-46.0.7-cp38-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:1d25aee46d0c6f1a501adcddb2d2fee4b979381346a78558ed13e50aa8a59067", size = 4450634, upload-time = "2026-04-08T01:57:21.185Z" },
+    { url = "https://files.pythonhosted.org/packages/16/01/0cd51dd86ab5b9befe0d031e276510491976c3a80e9f6e31810cce46c4ad/cryptography-46.0.7-cp38-abi3-manylinux_2_31_armv7l.whl", hash = "sha256:cdfbe22376065ffcf8be74dc9a909f032df19bc58a699456a21712d6e5eabfd0", size = 3985233, upload-time = "2026-04-08T01:57:22.862Z" },
+    { url = "https://files.pythonhosted.org/packages/92/49/819d6ed3a7d9349c2939f81b500a738cb733ab62fbecdbc1e38e83d45e12/cryptography-46.0.7-cp38-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:abad9dac36cbf55de6eb49badd4016806b3165d396f64925bf2999bcb67837ba", size = 4271955, upload-time = "2026-04-08T01:57:24.814Z" },
+    { url = "https://files.pythonhosted.org/packages/80/07/ad9b3c56ebb95ed2473d46df0847357e01583f4c52a85754d1a55e29e4d0/cryptography-46.0.7-cp38-abi3-manylinux_2_34_ppc64le.whl", hash = "sha256:935ce7e3cfdb53e3536119a542b839bb94ec1ad081013e9ab9b7cfd478b05006", size = 4879888, upload-time = "2026-04-08T01:57:26.88Z" },
+    { url = "https://files.pythonhosted.org/packages/b8/c7/201d3d58f30c4c2bdbe9b03844c291feb77c20511cc3586daf7edc12a47b/cryptography-46.0.7-cp38-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:35719dc79d4730d30f1c2b6474bd6acda36ae2dfae1e3c16f2051f215df33ce0", size = 4449961, upload-time = "2026-04-08T01:57:29.068Z" },
+    { url = "https://files.pythonhosted.org/packages/a5/ef/649750cbf96f3033c3c976e112265c33906f8e462291a33d77f90356548c/cryptography-46.0.7-cp38-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:7bbc6ccf49d05ac8f7d7b5e2e2c33830d4fe2061def88210a126d130d7f71a85", size = 4401696, upload-time = "2026-04-08T01:57:31.029Z" },
+    { url = "https://files.pythonhosted.org/packages/41/52/a8908dcb1a389a459a29008c29966c1d552588d4ae6d43f3a1a4512e0ebe/cryptography-46.0.7-cp38-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:a1529d614f44b863a7b480c6d000fe93b59acee9c82ffa027cfadc77521a9f5e", size = 4664256, upload-time = "2026-04-08T01:57:33.144Z" },
+    { url = "https://files.pythonhosted.org/packages/4b/fa/f0ab06238e899cc3fb332623f337a7364f36f4bb3f2534c2bb95a35b132c/cryptography-46.0.7-cp38-abi3-win32.whl", hash = "sha256:f247c8c1a1fb45e12586afbb436ef21ff1e80670b2861a90353d9b025583d246", size = 3013001, upload-time = "2026-04-08T01:57:34.933Z" },
+    { url = "https://files.pythonhosted.org/packages/d2/f1/00ce3bde3ca542d1acd8f8cfa38e446840945aa6363f9b74746394b14127/cryptography-46.0.7-cp38-abi3-win_amd64.whl", hash = "sha256:506c4ff91eff4f82bdac7633318a526b1d1309fc07ca76a3ad182cb5b686d6d3", size = 3472985, upload-time = "2026-04-08T01:57:36.714Z" },
+    { url = "https://files.pythonhosted.org/packages/63/0c/dca8abb64e7ca4f6b2978769f6fea5ad06686a190cec381f0a796fdcaaba/cryptography-46.0.7-pp311-pypy311_pp73-macosx_11_0_arm64.whl", hash = "sha256:fc9ab8856ae6cf7c9358430e49b368f3108f050031442eaeb6b9d87e4dcf4e4f", size = 3476879, upload-time = "2026-04-08T01:57:38.664Z" },
+    { url = "https://files.pythonhosted.org/packages/3a/ea/075aac6a84b7c271578d81a2f9968acb6e273002408729f2ddff517fed4a/cryptography-46.0.7-pp311-pypy311_pp73-manylinux_2_28_aarch64.whl", hash = "sha256:d3b99c535a9de0adced13d159c5a9cf65c325601aa30f4be08afd680643e9c15", size = 4219700, upload-time = "2026-04-08T01:57:40.625Z" },
+    { url = "https://files.pythonhosted.org/packages/6c/7b/1c55db7242b5e5612b29fc7a630e91ee7a6e3c8e7bf5406d22e206875fbd/cryptography-46.0.7-pp311-pypy311_pp73-manylinux_2_28_x86_64.whl", hash = "sha256:d02c738dacda7dc2a74d1b2b3177042009d5cab7c7079db74afc19e56ca1b455", size = 4385982, upload-time = "2026-04-08T01:57:42.725Z" },
+    { url = "https://files.pythonhosted.org/packages/cb/da/9870eec4b69c63ef5925bf7d8342b7e13bc2ee3d47791461c4e49ca212f4/cryptography-46.0.7-pp311-pypy311_pp73-manylinux_2_34_aarch64.whl", hash = "sha256:04959522f938493042d595a736e7dbdff6eb6cc2339c11465b3ff89343b65f65", size = 4219115, upload-time = "2026-04-08T01:57:44.939Z" },
+    { url = "https://files.pythonhosted.org/packages/f4/72/05aa5832b82dd341969e9a734d1812a6aadb088d9eb6f0430fc337cc5a8f/cryptography-46.0.7-pp311-pypy311_pp73-manylinux_2_34_x86_64.whl", hash = "sha256:3986ac1dee6def53797289999eabe84798ad7817f3e97779b5061a95b0ee4968", size = 4385479, upload-time = "2026-04-08T01:57:46.86Z" },
+    { url = "https://files.pythonhosted.org/packages/20/2a/1b016902351a523aa2bd446b50a5bc1175d7a7d1cf90fe2ef904f9b84ebc/cryptography-46.0.7-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:258514877e15963bd43b558917bc9f54cf7cf866c38aa576ebf47a77ddbc43a4", size = 3412829, upload-time = "2026-04-08T01:57:48.874Z" },
 ]
 
 [[package]]

libs/sdk-py/uv.lock

@@ -262,7 +262,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.2.22"
+version = "1.2.28"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -274,9 +274,9 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/b1/a3/c4cd6827a1df46c821e7214b7f7b7a28b189e6c9b84ef15c6d629c5e3179/langchain_core-1.2.22.tar.gz", hash = "sha256:8d8f726d03d3652d403da915126626bb6250747e8ba406537d849e68b9f5d058", size = 842487, upload-time = "2026-03-24T18:48:44.9Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/a4/317a1a3ac1df33a64adb3670bf88bbe3b3d5baa274db6863a979db472897/langchain_core-1.2.28.tar.gz", hash = "sha256:271a3d8bd618f795fdeba112b0753980457fc90537c46a0c11998516a74dc2cb", size = 846119, upload-time = "2026-04-08T18:19:34.867Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c7/a6/2ffacf0f1a3788f250e75d0b52a24896c413be11be3a6d42bcdf46fbea48/langchain_core-1.2.22-py3-none-any.whl", hash = "sha256:7e30d586b75918e828833b9ec1efc25465723566845dd652c277baf751e9c04b", size = 506829, upload-time = "2026-03-24T18:48:43.286Z" },
+    { url = "https://files.pythonhosted.org/packages/a8/92/32f785f077c7e898da97064f113c73fbd9ad55d1e2169cf3a391b183dedb/langchain_core-1.2.28-py3-none-any.whl", hash = "sha256:80764232581eaf8057bcefa71dbf8adc1f6a28d257ebd8b95ba9b8b452e8c6ac", size = 508727, upload-time = "2026-04-08T18:19:32.823Z" },
 ]
 
 [[package]]