fix(cli): include build environment hash in Docker cache keys

- A node's `error_handler` ran, but the original error was still raised
when the node ran on the executor path: when streaming `messages` or
`custom`, with `subgraphs=True`, or next to a parallel node. The runner
marked the handled task's future, but the executors re-raised every
failed future on exit without checking the mark. The mark now lives in
the executor module, and both executors skip those futures. Fixes #8608.
- With only `tasks` or `debug` streaming, a failing task's result (with
its `error`) was never streamed: the single-task fast path raised before
the caller drained the stream queue. It now yields once before
re-raising. Fixes #5764.

The new tests fail on `main`.

Ideas from #8278 by @UdayKiranJadi, #8611 by @SnowingFox, #8823 by
@sailikhithk, #6984 by @Br1an67 and #8400 by @cms19859230182-lang.

libs/cli/langgraph_cli/config.py

@@ -1589,6 +1589,8 @@ ADD {relpath} /deps/{name}
     docker_file_contents.extend(
         [
             f"FROM {image_str}",
+            # BuildKit secrets do not invalidate RUN caches; hosted builds supply this hash.
+            "ARG USER_ENV_VARS_HASH",
             "",
             os.linesep.join(config["dockerfile_lines"]),
             "",
@@ -1679,6 +1681,8 @@ def node_config_to_docker(
 
     docker_file_contents = [
         f"FROM {image_str}",
+        # BuildKit secrets do not invalidate RUN caches; hosted builds supply this hash.
+        "ARG USER_ENV_VARS_HASH",
         "",
         os.linesep.join(config["dockerfile_lines"]),
         "",

libs/cli/langgraph_cli/uv_lock.py

@@ -955,6 +955,8 @@ def python_config_to_docker_uv_lock(
         docker_plan.add_blank()
 
     docker_plan.add_instruction("FROM", image_str)
+    # BuildKit secrets do not invalidate RUN caches; hosted builds supply this hash.
+    docker_plan.add_instruction("ARG", "USER_ENV_VARS_HASH")
     docker_plan.add_blank()
     docker_plan.extend_nonempty(config["dockerfile_lines"])
     if config["dockerfile_lines"]:

libs/cli/tests/unit_tests/cli/test_cli.py

@@ -148,6 +148,7 @@ services:
             dockerfile_inline: |
                 # syntax=docker/dockerfile:1.4
                 FROM langchain/langgraph-api:3.11
+                ARG USER_ENV_VARS_HASH
                 # -- Adding local package . --
                 ADD . /deps/cli
                 # -- End of local package . --

libs/cli/tests/unit_tests/test_config.py

@@ -910,6 +910,7 @@ def test_config_to_docker_simple():
     expected_docker_stdin = f"""\
 # syntax=docker/dockerfile:1.4
 FROM langchain/langgraph-api:3.11
+ARG USER_ENV_VARS_HASH
 # -- Installing local requirements --
 COPY --from=outer-requirements.txt requirements.txt /deps/outer-graphs_reqs_a/graphs_reqs_a/requirements.txt
 RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt -r /deps/outer-graphs_reqs_a/graphs_reqs_a/requirements.txt
@@ -974,6 +975,7 @@ def test_config_to_docker_outside_path():
         """\
 # syntax=docker/dockerfile:1.4
 FROM langchain/langgraph-api:3.11
+ARG USER_ENV_VARS_HASH
 # -- Adding non-package dependency unit_tests --
 ADD . /deps/outer-unit_tests/unit_tests
 RUN set -ex && \\
@@ -1034,6 +1036,7 @@ def test_config_to_docker_pipconfig():
     expected_docker_stdin = (
         """\
 FROM langchain/langgraph-api:3.11
+ARG USER_ENV_VARS_HASH
 ADD pipconfig.txt /pipconfig.txt
 # -- Adding non-package dependency unit_tests --
 ADD . /deps/outer-unit_tests/unit_tests
@@ -1097,6 +1100,7 @@ def test_config_to_docker_local_deps():
     )
     expected_docker_stdin = f"""\
 FROM langchain/langgraph-api-custom:3.11
+ARG USER_ENV_VARS_HASH
 # -- Adding non-package dependency graphs --
 ADD ./graphs /deps/outer-graphs/src
 RUN set -ex && \\
@@ -1144,6 +1148,7 @@ dependencies = ["langchain"]"""
     os.remove(pyproject_path)
     expected_docker_stdin = (
         """FROM langchain/langgraph-api:3.11
+ARG USER_ENV_VARS_HASH
 # -- Adding local package . --
 ADD . /deps/unit_tests
 # -- End of local package . --
@@ -1177,6 +1182,7 @@ def test_config_to_docker_end_to_end():
         base_image="langchain/langgraph-api",
     )
     expected_docker_stdin = f"""FROM langchain/langgraph-api:3.12
+ARG USER_ENV_VARS_HASH
 ARG meow
 ARG foo
 ADD pipconfig.txt /pipconfig.txt
@@ -1222,6 +1228,7 @@ def test_config_to_docker_nodejs():
         base_image="langchain/langgraphjs-api",
     )
     expected_docker_stdin = """FROM langchain/langgraphjs-api:20
+ARG USER_ENV_VARS_HASH
 ARG meow
 ARG foo
 ADD . /deps/unit_tests
@@ -1413,6 +1420,7 @@ def test_config_to_docker_nodejs_internal_docker_tag():
         base_image="langchain/langgraphjs-api",
     )
     expected_docker_stdin = """FROM langchain/langgraphjs-api:my-tag
+ARG USER_ENV_VARS_HASH
 ARG meow
 ARG foo
 ADD . /deps/unit_tests
@@ -1524,6 +1532,7 @@ def test_config_to_docker_gen_ui_python():
     )
 
     expected_docker_stdin = f"""FROM langchain/langgraph-api:3.11
+ARG USER_ENV_VARS_HASH
 RUN /storage/install-node.sh
 # -- Adding non-package dependency unit_tests --
 ADD . /deps/outer-unit_tests/unit_tests
@@ -1571,6 +1580,7 @@ def test_config_to_docker_multiplatform():
     )
 
     expected_docker_stdin = f"""FROM langchain/langgraph-api:3.11
+ARG USER_ENV_VARS_HASH
 RUN /storage/install-node.sh
 # -- Adding non-package dependency unit_tests --
 ADD . /deps/outer-unit_tests/unit_tests
@@ -2843,6 +2853,7 @@ def test_config_to_compose_simple_config():
             context: .
             dockerfile_inline: |
                 FROM langchain/langgraph-api:3.11
+                ARG USER_ENV_VARS_HASH
                 # -- Adding non-package dependency unit_tests --
                 ADD . /deps/outer-unit_tests/unit_tests
                 RUN set -ex && \\
@@ -2884,6 +2895,7 @@ def test_config_to_compose_env_vars():
             context: .
             dockerfile_inline: |
                 FROM langchain/langgraph-api-custom:3.11
+                ARG USER_ENV_VARS_HASH
                 # -- Adding non-package dependency unit_tests --
                 ADD . /deps/outer-unit_tests/unit_tests
                 RUN set -ex && \\
@@ -2929,6 +2941,7 @@ def test_config_to_compose_env_file():
             context: .
             dockerfile_inline: |
                 FROM langchain/langgraph-api:3.11
+                ARG USER_ENV_VARS_HASH
                 # -- Adding non-package dependency unit_tests --
                 ADD . /deps/outer-unit_tests/unit_tests
                 RUN set -ex && \\
@@ -2967,6 +2980,7 @@ def test_config_to_compose_watch():
             context: .
             dockerfile_inline: |
                 FROM langchain/langgraph-api:3.11
+                ARG USER_ENV_VARS_HASH
                 # -- Adding non-package dependency unit_tests --
                 ADD . /deps/outer-unit_tests/unit_tests
                 RUN set -ex && \\
@@ -3014,6 +3028,7 @@ def test_config_to_compose_end_to_end():
             context: .
             dockerfile_inline: |
                 FROM langchain/langgraph-api:3.11
+                ARG USER_ENV_VARS_HASH
                 # -- Adding non-package dependency unit_tests --
                 ADD . /deps/outer-unit_tests/unit_tests
                 RUN set -ex && \\

libs/cli/tests/unit_tests/test_env_cache.py

@@ -0,0 +1,142 @@
+"""Generated Dockerfile cache keys; opt into the real BuildKit test locally."""
+
+import hashlib
+import os
+import pathlib
+import subprocess
+import uuid
+
+import pytest
+
+from langgraph_cli.config import config_to_docker, validate_config
+
+
+def generated_prefix(tmp_path: pathlib.Path, runtime: str) -> str:
+    config = {
+        "graphs": {"agent": "agent:graph"},
+        "dockerfile_lines": [
+            f"RUN --mount=type=secret,id=build_marker echo {tmp_path.name} >/dev/null; "
+            "sha256sum /run/secrets/build_marker | cut -d ' ' -f 1 > /result"
+        ],
+    }
+    if runtime == "node":
+        config["node_version"] = "20"
+        config["graphs"] = {"agent": "./agent.js:graph"}
+    else:
+        config.update(python_version="3.11", dependencies=["requests"])
+        if runtime == "uv":
+            config.pop("dependencies")
+            config["source"] = {"kind": "uv", "root": "."}
+            (tmp_path / "uv.lock").write_text("version = 1\n")
+            (tmp_path / "pyproject.toml").write_text(
+                '[project]\nname = "agent"\nversion = "0.1.0"\n'
+                '[build-system]\nrequires = ["setuptools"]\n'
+                'build-backend = "setuptools.build_meta"\n'
+            )
+    dockerfile, _ = config_to_docker(
+        tmp_path / "langgraph.json",
+        validate_config(config),
+        base_image="langchain/langgraph-api",
+    )
+    # Keep the generated stage header and custom RUN exactly as emitted. Avoid
+    # installing real graph dependencies: this test exercises the cache boundary.
+    return dockerfile.split(" > /result", 1)[0] + " > /result\n"
+
+
+@pytest.mark.parametrize("runtime", ["python", "uv", "node"])
+def test_env_hash_in_build_stage_before_custom_run(tmp_path, runtime):
+    prefix = generated_prefix(tmp_path, runtime)
+    assert prefix.index("FROM ") < prefix.index("ARG USER_ENV_VARS_HASH")
+    assert prefix.index("ARG USER_ENV_VARS_HASH") < prefix.index("RUN ")
+
+
+@pytest.mark.skipif(
+    os.environ.get("LANGGRAPH_TEST_BUILDKIT") != "1",
+    reason="Set LANGGRAPH_TEST_BUILDKIT=1 with a local Docker BuildKit daemon",
+)
+@pytest.mark.parametrize("runtime", ["python", "uv", "node"])
+def test_secret_hash_invalidates_real_buildkit_cache(tmp_path, runtime, request):
+    prefix = generated_prefix(tmp_path, runtime)
+    lines = prefix.splitlines()
+    lines[0] = "FROM alpine:3.21"
+    prefix = "\n".join(lines) + "\n"
+    dockerfile = tmp_path / "Dockerfile"
+    marker = tmp_path / "build-marker"
+    (tmp_path / ".dockerignore").write_text("*\n!Dockerfile\n")
+    output_counter = 0
+    cache_image = "langgraph-cli-cache-test:" + uuid.uuid4().hex
+    request.addfinalizer(
+        lambda: subprocess.run(
+            ["docker", "image", "rm", cache_image], capture_output=True, check=False
+        )
+    )
+
+    def build(contents, value, *, pass_hash=True, export_image=False):
+        nonlocal output_counter
+        output_counter += 1
+        marker.write_text(value)
+        dockerfile.write_text(
+            contents + "FROM scratch\nCOPY --from=0 /result /result\n"
+        )
+        output = tmp_path / f"output-{output_counter}"
+        args = [
+            "docker",
+            "buildx",
+            "build",
+            "--progress=plain",
+            "--secret",
+            f"id=build_marker,src={marker}",
+            "--output",
+            f"type=local,dest={output}",
+            "-f",
+            str(dockerfile),
+            str(tmp_path),
+        ]
+        if pass_hash:
+            args.extend(
+                [
+                    "--build-arg",
+                    f"USER_ENV_VARS_HASH={hashlib.sha256(value.encode()).hexdigest()}",
+                ]
+            )
+        if output_counter > 1:
+            args.extend(["--cache-from", cache_image])
+        if export_image:
+            # Export a legacy image with inline cache, then explicitly import it.
+            image_args = args.copy()
+            output_index = image_args.index("--output")
+            del image_args[output_index : output_index + 2]
+            image_args.extend(
+                ["--load", "--tag", cache_image, "--cache-to", "type=inline"]
+            )
+            subprocess.run(image_args, capture_output=True, text=True, check=True)
+        # List arguments only; all inputs are local public test markers.
+        result = subprocess.run(args, capture_output=True, text=True, check=True)
+        return (output / "result").read_text().strip(), result.stderr
+
+    first = f"{runtime}-{tmp_path.name}-first"
+    second = f"{runtime}-{tmp_path.name}-second"
+    legacy = prefix.replace("ARG USER_ENV_VARS_HASH\n", "")
+    legacy_output, _ = build(legacy, first, export_image=True)
+    assert legacy_output == hashlib.sha256(first.encode()).hexdigest()
+    stale_output, _ = build(legacy, second)
+    assert stale_output == legacy_output  # Secret contents alone do not bust cache.
+
+    fixed_output, _ = build(prefix, second)
+    assert fixed_output == hashlib.sha256(second.encode()).hexdigest()
+    assert fixed_output != stale_output  # Old cache is safe when adopting the ARG.
+    repeated_output, repeated_log = build(prefix, second)
+    assert repeated_output == fixed_output
+    run_vertex = next(
+        line.split()[0]
+        for line in repeated_log.splitlines()
+        if "] RUN " in line and "sha256sum" in line
+    )
+    assert f"{run_vertex} CACHED" in repeated_log
+    changed_output, _ = build(prefix, first)
+    assert changed_output == hashlib.sha256(first.encode()).hexdigest()
+    assert changed_output != repeated_output
+
+    # Ordinary CLI callers can still build without supplying the optional hash.
+    unset_output, _ = build(prefix, first, pass_hash=False)
+    assert unset_output == hashlib.sha256(first.encode()).hexdigest()

libs/langgraph/langgraph/channels/delta.py

@@ -60,7 +60,7 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
         typ: The value type (e.g. `list`, `dict`). Inferred automatically
             from the outer type when used inside `Annotated[T, DeltaChannel(...)]`.
         snapshot_frequency: Every Nth update to this channel writes a snapshot
-            blob (default `1000`). Must be a positive int.
+            blob (default `100`). Must be a positive int.
     """
 
     __slots__ = ("value", "reducer", "snapshot_frequency")
@@ -71,7 +71,7 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
         reducer: Callable[[Any, Sequence[Any]], Any],
         typ: type[Value] | None = None,
         *,
-        snapshot_frequency: int = 1000,
+        snapshot_frequency: int = 100,
     ) -> None:
         if snapshot_frequency <= 0:
             raise ValueError(

libs/langgraph/langgraph/pregel/_executor.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import concurrent.futures
 import time
+import weakref
 from collections.abc import Awaitable, Callable, Coroutine
 from contextlib import AbstractAsyncContextManager, AbstractContextManager, ExitStack
 from contextvars import copy_context
@@ -37,6 +38,13 @@ class Submit(Protocol[P, T]):
     ) -> concurrent.futures.Future[T]: ...
 
 
+# Futures whose exception a node's error handler took over; the executors
+# don't re-raise them on exit.
+SKIP_RERAISE_SET: weakref.WeakSet[concurrent.futures.Future | asyncio.Future] = (
+    weakref.WeakSet()
+)
+
+
 class BackgroundExecutor(AbstractContextManager):
     """A context manager that runs sync tasks in the background.
     Uses a thread pool executor to delegate tasks to separate threads.
@@ -111,7 +119,7 @@ class BackgroundExecutor(AbstractContextManager):
         if exc_type is None:
             # re-raise the first exception that occurred in a task
             for task, (_, reraise) in tasks.items():
-                if not reraise:
+                if not reraise or task in SKIP_RERAISE_SET:
                     continue
                 try:
                     task.result()
@@ -202,7 +210,7 @@ class AsyncBackgroundExecutor(AbstractAsyncContextManager):
         if exc_type is None:
             # re-raise the first exception that occurred in a task
             for task, (_, reraise) in tasks.items():
-                if not reraise:
+                if not reraise or task in SKIP_RERAISE_SET:
                     continue
                 try:
                     if exc := task.exception():

libs/langgraph/langgraph/pregel/_runner.py

@@ -43,7 +43,7 @@ from langgraph._internal._typing import MISSING
 from langgraph.constants import TAG_HIDDEN
 from langgraph.errors import GraphBubbleUp, GraphInterrupt, is_invalid_resume
 from langgraph.pregel._algo import Call
-from langgraph.pregel._executor import Submit
+from langgraph.pregel._executor import SKIP_RERAISE_SET, Submit
 from langgraph.pregel._retry import arun_with_retry, run_with_retry
 from langgraph.pregel._task_status import CONTROL_WRITES
 from langgraph.types import (
@@ -68,10 +68,6 @@ EXCLUDED_FRAME_FNAMES = (
     "concurrent/futures/_base.py",
 )
 
-SKIP_RERAISE_SET: weakref.WeakSet[concurrent.futures.Future | asyncio.Future] = (
-    weakref.WeakSet()
-)
-
 
 class FuturesDict(Generic[F, E], dict[F, PregelExecutableTask | None]):
     event: E
@@ -239,6 +235,8 @@ class PregelRunner:
                         fut.set_exception(exc)
                         futures.done.add(fut)
                 elif reraise and id(exc) not in self._handled_exception_ids:
+                    # let the caller stream this task's result before it raises
+                    yield
                     if tb := exc.__traceback__:
                         while tb.tb_next is not None and any(
                             tb.tb_frame.f_code.co_filename.endswith(name)
@@ -430,6 +428,8 @@ class PregelRunner:
                         fut.set_exception(exc)
                         futures.done.add(fut)
                 elif reraise and id(exc) not in self._handled_exception_ids:
+                    # let the caller stream this task's result before it raises
+                    yield
                     if tb := exc.__traceback__:
                         while tb.tb_next is not None and any(
                             tb.tb_frame.f_code.co_filename.endswith(name)

libs/langgraph/tests/test_channels.py

@@ -20,6 +20,7 @@ from langgraph.errors import EmptyChannelError, InvalidUpdateError
 from langgraph.graph import START, StateGraph
 from langgraph.graph.message import _messages_delta_reducer
 from langgraph.graph.state import _get_channel
+from langgraph.pregel._checkpoint import delta_channels_to_snapshot
 from langgraph.types import Overwrite
 
 pytestmark = pytest.mark.anyio
@@ -342,6 +343,26 @@ def test_delta_channel_checkpoint_returns_missing() -> None:
 # ---------------------------------------------------------------------------
 
 
+@pytest.mark.parametrize("frequency", [None, 50, 1000])
+def test_delta_channel_snapshot_frequency(frequency: int | None) -> None:
+    spec = (
+        DeltaChannel(_messages_delta_reducer)
+        if frequency is None
+        else DeltaChannel(_messages_delta_reducer, snapshot_frequency=frequency)
+    )
+    expected = 100 if frequency is None else frequency
+    assert spec.snapshot_frequency == expected
+    channel = spec.from_checkpoint(MISSING)
+    channel.update([HumanMessage(content="hi", id="h1")])
+    assert channel.copy().snapshot_frequency == expected
+    for updates in (expected - 1, expected):
+        assert delta_channels_to_snapshot(
+            {"messages": channel},
+            {"messages": (updates, updates)},
+            {"messages": 1},
+        ) == ({"messages"} if updates == expected else set())
+
+
 def test_delta_channel_snapshot_version_based() -> None:
     """Snapshots fire when a channel accumulates `snapshot_frequency` updates.
 

libs/langgraph/tests/test_pregel.py

@@ -4346,6 +4346,25 @@ def test_enum_node_names():
     assert graph.invoke({"foo": "hello"}) == {"foo": "hello", "bar": "hello!"}
 
 
+@pytest.mark.parametrize("stream_mode", ["tasks", "debug"])
+def test_failing_task_result_is_streamed_before_the_error(stream_mode: str) -> None:
+    class State(TypedDict):
+        x: str
+
+    def failing(state: State) -> State:
+        raise ValueError("boom")
+
+    graph = StateGraph(State).add_node("failing", failing).add_edge(START, "failing")
+    errors = []
+    with pytest.raises(ValueError, match="boom"):
+        for chunk in graph.compile().stream({"x": ""}, stream_mode=stream_mode):
+            payload = chunk["payload"] if stream_mode == "debug" else chunk
+            if "error" in payload:
+                errors.append(payload["error"])
+
+    assert [type(error) for error in errors] == [ValueError]
+
+
 def test_debug_retry(sync_checkpointer: BaseCheckpointSaver):
     class State(TypedDict):
         messages: Annotated[list[str], operator.add]

libs/langgraph/tests/test_pregel_async.py

@@ -5809,6 +5809,60 @@ async def test_store_injected_async(
     )  # still overwriting the same one
 
 
+@pytest.mark.parametrize("stream_mode", ["tasks", "debug"])
+async def test_failing_task_result_is_streamed_before_the_error(
+    stream_mode: str,
+) -> None:
+    class State(TypedDict):
+        x: str
+
+    async def failing(state: State) -> State:
+        raise ValueError("boom")
+
+    graph = StateGraph(State).add_node("failing", failing).add_edge(START, "failing")
+    errors = []
+    with pytest.raises(ValueError, match="boom"):
+        async for chunk in graph.compile().astream({"x": ""}, stream_mode=stream_mode):
+            payload = chunk["payload"] if stream_mode == "debug" else chunk
+            if "error" in payload:
+                errors.append(payload["error"])
+
+    assert [type(error) for error in errors] == [ValueError]
+
+
+@pytest.mark.parametrize(
+    ("parallel", "stream_mode"), [(False, "custom"), (True, "values")]
+)
+async def test_graph_error_handler_recovers_on_the_executor_path(
+    parallel: bool, stream_mode: str
+) -> None:
+    class State(TypedDict):
+        x: str
+
+    async def failing(state: State) -> State:
+        raise ValueError("boom")
+
+    async def handler(state: State) -> Command:
+        return Command(update={"x": "handled"}, goto=END)
+
+    builder = (
+        StateGraph(State)
+        .add_node("failing", failing, error_handler=handler)
+        .add_edge(START, "failing")
+    )
+    if parallel:
+        builder.add_node("sibling", lambda state: {}).add_edge(START, "sibling")
+
+    chunks = [
+        chunk
+        async for chunk in builder.compile().astream(
+            {"x": ""}, stream_mode=[stream_mode, "values"]
+        )
+    ]
+
+    assert [c[-1] for c in chunks if c[0] == "values"][-1] == {"x": "handled"}
+
+
 async def test_debug_retry(async_checkpointer: BaseCheckpointSaver):
     class State(TypedDict):
         messages: Annotated[list[str], operator.add]

libs/langgraph/tests/test_retry.py

@@ -2097,6 +2097,47 @@ def test_graph_error_handler_can_route_with_command():
     assert result["foo"] == "handled_next"
 
 
+# Each case runs the failing node on the executor path instead of the
+# single-task fast path.
+@pytest.mark.parametrize(
+    ("parallel", "stream_mode", "subgraphs"),
+    [
+        (False, "custom", False),
+        (False, "messages", False),
+        (False, "values", True),
+        (True, "values", False),
+    ],
+)
+def test_graph_error_handler_recovers_on_the_executor_path(
+    parallel: bool, stream_mode: str, subgraphs: bool
+) -> None:
+    class State(TypedDict):
+        x: str
+
+    def failing(state: State) -> State:
+        raise ValueError("boom")
+
+    def handler(state: State) -> Command:
+        return Command(update={"x": "handled"}, goto=END)
+
+    builder = (
+        StateGraph(State)
+        .add_node("failing", failing, error_handler=handler)
+        .add_edge(START, "failing")
+    )
+    if parallel:
+        builder.add_node("sibling", lambda state: {}).add_edge(START, "sibling")
+    graph = builder.compile()
+
+    chunks = list(
+        graph.stream(
+            {"x": ""}, stream_mode=[stream_mode, "values"], subgraphs=subgraphs
+        )
+    )
+
+    assert [c[-1] for c in chunks if c[-2] == "values"][-1] == {"x": "handled"}
+
+
 def test_graph_error_handler_failure_fails_run():
     class State(TypedDict):
         foo: str