release(langgraph): 1.2.16 (#9334 )

Bumps langgraph from 1.2.15 to 1.2.16. Includes: - Keep a DeltaChannel's value when its field changes to a plain reducer (#9302) - Save a resumed exit run's pending writes once (#9314) - Stop a subgraph from rerunning its finished tasks on resume (#9315) - Apply a resumed `Command`'s update once (#9322) - Let other tasks finish when a resume value is invalid (#9265)
release(sdk-py): 0.4.8 (#9335 )
2026-10-21 23:41:53 +02:00 · 2026-10-16 18:02:08 -04:00 · 2026-10-16 18:01:53 -04:00 · 2026-10-16 15:29:34 -04:00 · 2026-10-16 11:33:53 -07:00 · 2026-10-16 12:29:42 -04:00
27 changed files with 1109 additions and 110 deletions
@@ -1 +1 @@
-__version__ = "0.4.35"
+__version__ = "0.4.36"
@@ -1209,6 +1209,33 @@ class DockerBuildCommand:
        return cls(("docker", "buildx", "build"), flags)


+def _active_docker_host(runner: CommandRunner) -> str | None:
+    """Return the daemon endpoint of the user's active Docker context.
+
+    None when `DOCKER_HOST` is set, which Docker honors with any config
+    directory. Also None when the lookup fails (for example, `DOCKER_CONTEXT`
+    names a context that doesn't exist) or the context has no Docker endpoint
+    (Docker prints `<no value>`); that keeps Docker's own default in place.
+    """
+    if os.environ.get("DOCKER_HOST"):
+        return None
+    try:
+        stdout, _ = runner.run(
+            subp_exec(
+                "docker",
+                "context",
+                "inspect",
+                "--format",
+                "{{.Endpoints.docker.Host}}",
+                collect=True,
+            )
+        )
+    except (click.exceptions.Exit, OSError):
+        return None
+    host = (stdout or "").strip()
+    return host if "://" in host else None
+
+
 def _docker_argv(docker_config_dir: str | None) -> tuple[str, ...]:
    if docker_config_dir is None:
        return ("docker",)
@@ -1242,17 +1269,13 @@ def _push_image(
    runner: CommandRunner,
    image: str,
    *,
-    docker_config_dir: str | None,
+    docker: tuple[str, ...],
    verbose: bool,
 ) -> None:
    for attempt in range(1, _PUSH_ATTEMPTS + 1):
        try:
            with Progress(message="Pushing...", elapsed=not verbose):
-                runner.run(
-                    subp_exec(
-                        *_docker_argv(docker_config_dir), "push", image, verbose=verbose
-                    )
-                )
+                runner.run(subp_exec(*docker, "push", image, verbose=verbose))
            return
        except click.exceptions.Exit:
            if attempt == _PUSH_ATTEMPTS:
@@ -1337,6 +1360,11 @@ def _run_local_build(
        registry_host = repository.split("/")[0]

        with _docker_config_for_token(registry_host, deployment_token, username) as cfg:
+            # A config directory has no current context, so Docker would fall
+            # back to /var/run/docker.sock; keep using the daemon the build used.
+            docker_host = _active_docker_host(runner)
+            host_args = ("--host", docker_host) if docker_host else ()
+            docker = (*_docker_argv(cfg), *host_args)
            _log_deploy_step(step, f"Logging into {registry_host}")
            token_input = (
                deployment_token
@@ -1345,7 +1373,7 @@ def _run_local_build(
            )
            runner.run(
                subp_exec(
-                    *_docker_argv(cfg),
+                    *docker,
                    "login",
                    "-u",
                    username,
@@ -1361,7 +1389,7 @@ def _run_local_build(
            runner.run(
                subp_exec("docker", "tag", image_to_push, remote_image, verbose=verbose)
            )
-            _push_image(runner, remote_image, docker_config_dir=cfg, verbose=verbose)
+            _push_image(runner, remote_image, docker=docker, verbose=verbose)
        step += 1

        resolved_image = _resolve_pushed_image_digest(
@@ -1826,7 +1854,7 @@ class CustomerRegistrySource:
                _build_image(runner, ctx.spec, image, verbose=ctx.verbose)
            step += 1
            _log_deploy_step(step, f"Pushing image {image}")
-            _push_image(runner, image, docker_config_dir=None, verbose=ctx.verbose)
+            _push_image(runner, image, docker=_docker_argv(None), verbose=ctx.verbose)
            step += 1
            digest = _resolve_pushed_image_digest(
                runner, remote_image=image, docker_config_dir=None, verbose=ctx.verbose
@@ -2185,11 +2213,13 @@ def _deploy_base_options(
            ),
            click.option(
                "--deployment-type",
-                type=click.Choice(["dev", "prod"]),
+                type=click.Choice(["serverless", "dedicated", "dev", "prod"]),
                default="dev",
                show_default=True,
                help=(
-                    "Deployment type (used when creating a new deployment). "
+                    "Deployment type (used when creating a new deployment): "
+                    "serverless (scale to zero) or dedicated (prod); "
+                    "legacy dev and prod are also accepted. "
                    "Ignored with --push-to."
                ),
            ),
@@ -2480,7 +2510,9 @@ def _deploy_cmd(
            ),
            verbose=verbose,
            selector=selector,
-            deployment_type=deployment_type,
+            deployment_type={"serverless": "dev_zero", "dedicated": "prod"}.get(
+                deployment_type, deployment_type
+            ),
            secrets=secrets,
            tracked_packages=tracked_packages,
        )
@@ -52,6 +52,8 @@ ARCHIVE = ("/tmp/src.tgz", 2048, "langgraph.json")
 OBJECT_PATH = "tarballs/src.tgz"
 PLATFORM_FORMAT = "{{.Os}}/{{.Architecture}}"
 DIGESTS_FORMAT = "{{json .RepoDigests}}"
+CONTEXT_HOST_FORMAT = "{{.Endpoints.docker.Host}}"
+CONTEXT_HOST = "unix:///Users/dev/.docker/run/docker.sock"
 NOT_A_CLI_DEPLOYMENT = (
    "push token is only available for 'internal_docker' source deployments"
 )
@@ -192,6 +194,8 @@ class DockerCommand:
 class DockerDouble:
    timeline: list[str]
    failing_pushes: int = 0
+    context_lookup_fails: bool = False
+    context_host: str = CONTEXT_HOST
    builds: list[dict] = field(default_factory=list)
    commands: list[DockerCommand] = field(default_factory=list)

@@ -235,6 +239,10 @@ class DockerDouble:
        if "push" in args and self.failing_pushes > 0:
            self.failing_pushes -= 1
            raise click.exceptions.Exit(1)
+        if CONTEXT_HOST_FORMAT in args:
+            if self.context_lookup_fails:
+                raise click.exceptions.Exit(1)
+            return f"{self.context_host}\n", None
        if PLATFORM_FORMAT in args:
            return "linux/amd64\n", None
        if DIGESTS_FORMAT in args:
@@ -244,6 +252,8 @@ class DockerDouble:

    @staticmethod
    def _verb(args: tuple[str, ...]) -> str:
+        if CONTEXT_HOST_FORMAT in args:
+            return "inspect-context"
        if PLATFORM_FORMAT in args:
            return "inspect-platform"
        if DIGESTS_FORMAT in args:
@@ -292,7 +302,12 @@ def deploy_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> DeployPro
        json.dumps({"graphs": {"agent": "agent.py:graph"}, "dependencies": ["."]})
    )
    monkeypatch.chdir(tmp_path)
-    for name in ("LANGSMITH_TENANT_ID", "LANGSMITH_ENDPOINT", "LANGGRAPH_HOST_URL"):
+    for name in (
+        "LANGSMITH_TENANT_ID",
+        "LANGSMITH_ENDPOINT",
+        "LANGGRAPH_HOST_URL",
+        "DOCKER_HOST",
+    ):
        monkeypatch.delenv(name, raising=False)

    timeline: list[str] = []
@@ -340,6 +355,7 @@ def test_first_local_deploy_creates_then_builds_pushes_and_updates_in_order(
        CREATE_DEPLOYMENT,
        "docker build",
        _push_token(CREATED_ID),
+        "docker inspect-context",
        "docker login",
        "docker tag",
        "docker push",
@@ -349,6 +365,27 @@ def test_first_local_deploy_creates_then_builds_pushes_and_updates_in_order(
    assert "Deployment updated" in result.output


+@pytest.mark.parametrize(
+    ("deployment_type", "wire_type"),
+    [
+        ("serverless", "dev_zero"),
+        ("dedicated", "prod"),
+        ("dev", "dev"),
+        ("prod", "prod"),
+    ],
+)
+@pytest.mark.parametrize("build_mode", ["--no-remote", "--remote"])
+def test_deployment_type_aliases(
+    deploy_project: DeployProject, deployment_type: str, wire_type: str, build_mode: str
+) -> None:
+    result = deploy_project.run(build_mode, "--deployment-type", deployment_type)
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "deployment_type": wire_type
+    }
+
+
 def test_first_local_deploy_creates_an_internal_docker_deployment(
    deploy_project: DeployProject,
 ) -> None:
@@ -406,7 +443,7 @@ def test_local_deploy_logs_in_with_the_control_plane_push_token(

    login = deploy_project.docker.command("login")
    assert login.args[:2] == ("docker", "--config")
-    assert login.args[3:] == (
+    assert login.args[-5:] == (
        "login",
        "-u",
        "oauth2accesstoken",
@@ -416,6 +453,53 @@ def test_local_deploy_logs_in_with_the_control_plane_push_token(
    assert login.kwargs["input"] == f"{PUSH_TOKEN}\n"


+def test_local_deploy_logs_in_and_pushes_through_the_active_docker_context(
+    deploy_project: DeployProject,
+) -> None:
+    """The temporary config has no current context, so without the active
+    context's endpoint Docker would push through /var/run/docker.sock."""
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    for verb in ("login", "push"):
+        args = deploy_project.docker.command(verb).args
+        assert args[3:5] == ("--host", CONTEXT_HOST), verb
+
+
+def test_local_deploy_leaves_an_explicit_docker_host_to_docker(
+    deploy_project: DeployProject, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv("DOCKER_HOST", "tcp://127.0.0.1:2375")
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    for verb in ("login", "push"):
+        assert "--host" not in deploy_project.docker.command(verb).args, verb
+
+
+def test_local_deploy_keeps_dockers_default_when_the_context_lookup_fails(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.docker.context_lookup_fails = True
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    assert "--host" not in deploy_project.docker.command("push").args
+
+
+def test_local_deploy_ignores_a_context_without_a_docker_endpoint(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.docker.context_host = "<no value>"
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    assert "--host" not in deploy_project.docker.command("push").args
+
+
 def test_local_deploy_tells_the_control_plane_it_honors_registry_usernames(
    deploy_project: DeployProject,
 ) -> None:
@@ -438,7 +522,7 @@ def test_local_deploy_logs_in_with_the_username_the_control_plane_returns(
    deploy_project.run("--no-remote")

    login = deploy_project.docker.command("login")
-    assert login.args[3:] == (
+    assert login.args[-5:] == (
        "login",
        "-u",
        "AWS",
@@ -506,6 +590,7 @@ def test_prebuilt_image_is_validated_and_pushed_without_a_build(
    assert deploy_project.docker.builds == []
    assert deploy_project.docker.verbs() == [
        "docker inspect-platform",
+        "docker inspect-context",
        "docker login",
        "docker tag",
        "docker push",
@@ -257,8 +257,10 @@ class PregelLoop:
    _resume_loaded_writes: Sequence[PendingWrite] = ()
    # Tasks `_reapply_writes_to_succeeded_nodes` handed loaded writes back to.
    _reapplied_task_ids: set[str]
-    # An update or fork checkpoint addressed by `checkpoint_id` that the thread
-    # has moved past, so it may already have children.
+    # Tasks that wrote during this run, the only ones exit mode saves at exit.
+    _written_task_ids: set[str]
+    # A checkpoint addressed by `checkpoint_id` that the thread has moved past,
+    # so it may already have children.
    _addressed_checkpoint_superseded: bool = False

    # The checkpoint_config that points at the parent loaded at `__enter__`
@@ -349,7 +351,6 @@ class PregelLoop:
        self.interrupt_before = interrupt_before
        self.manager = manager
        self.is_nested = CONFIG_KEY_TASK_ID in self.config.get(CONF, {})
-        self.is_replaying = CONFIG_KEY_CHECKPOINT_ID in config[CONF]
        self._migrate_checkpoint = migrate_checkpoint
        self.trigger_to_nodes = trigger_to_nodes
        self.retry_policy = retry_policy
@@ -407,6 +408,14 @@ class PregelLoop:
            else ()
        )
        self.prev_checkpoint_config = None
+        # Whether the run points this loop at a checkpoint: a truthy
+        # `checkpoint_id`, this graph's `checkpoint_map` entry, or a replaying
+        # parent. Every task config carries `checkpoint_id: None`, so a key
+        # check would count every subgraph.
+        self._addressed = bool(
+            self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID)
+            or self.config[CONF].get(CONFIG_KEY_REPLAY_STATE)
+        )
        runtime = self.config[CONF].get(CONFIG_KEY_RUNTIME)
        self.control = runtime.control if isinstance(runtime, Runtime) else None

@@ -455,6 +464,7 @@ class PregelLoop:
        """Put writes for a task, to be read by the next tick."""
        if not writes:
            return
+        self._written_task_ids.add(task_id)
        # deduplicate writes to special channels, last write wins
        if all(w[0] in WRITES_IDX_MAP for w in writes):
            writes = list({w[0]: w for w in writes}.values())
@@ -468,6 +478,11 @@ class PregelLoop:
                for w in self.checkpoint_pending_writes
                if w[0] != task_id or w[1] not in WRITES_IDX_MAP
            ]
+            # Savers key a regular write by its position in the list, and the
+            # special ones dropped above won't be there next time, so they go
+            # last: a resume saved ahead of an update would move it to a new
+            # slot once the list is saved again, and it would apply twice.
+            writes = sorted(writes, key=lambda w: w[0] in WRITES_IDX_MAP)
            writes_to_save: WritesT = [
                w[1:] for w in self.checkpoint_pending_writes if w[0] == task_id
            ] + list(writes)
@@ -556,10 +571,12 @@ class PregelLoop:
                CONFIG_KEY_CHECKPOINT_ID: self.checkpoint["id"],
            },
        )
-        # group by task id
+        # Writes loaded with the checkpoint are stored on it already. Saving them
+        # again can file them under new indexes, and then they apply twice.
        by_task = defaultdict(list)
        for task_id, channel, value in self.checkpoint_pending_writes:
-            by_task[task_id].append((channel, value))
+            if task_id in self._written_task_ids:
+                by_task[task_id].append((channel, value))
        # submit writes to checkpointer
        for task_id, writes in by_task.items():
            if self.checkpointer_put_writes_accepts_task_path and hasattr(
@@ -904,6 +921,7 @@ class PregelLoop:
            if tid != NULL_TASK_ID and isinstance(self.specs.get(ch), DeltaChannel)
        }
        self._exit_command_writes = []
+        self._written_task_ids = set()
        # Resuming from a previous checkpoint requires two things:
        # 1. A prior checkpoint exists (channel_versions is non-empty)
        # 2. The input signals continuation (not a fresh run with new input)
@@ -931,7 +949,7 @@ class PregelLoop:
        # writes so that interrupt() calls re-fire instead of returning
        # stale values. But if we're actively resuming, keep them —
        # multi-interrupt scenarios need previously resolved values preserved.
-        is_time_traveling = self.is_replaying and (
+        is_time_traveling = self._addressed and (
            # Time-travel to a subgraph checkpoint: the parent sets
            # RESUMING=True (it can't distinguish time-travel from resume),
            # so we check if this subgraph's own ns is in checkpoint_map.
@@ -950,6 +968,10 @@ class PregelLoop:
                or configurable.get(CONFIG_KEY_RESUMING, False)
            )
        )
+        # Finished tasks rerun only on time travel, or on a resume at a
+        # checkpoint the thread has moved past. A resume addressed at the
+        # thread's latest checkpoint hands their writes back, as without an id.
+        self.is_replaying = is_time_traveling or self._addressed_checkpoint_superseded
        if is_time_traveling:
            self.checkpoint_pending_writes = [
                w for w in self.checkpoint_pending_writes if w[1] != RESUME
@@ -1165,11 +1187,11 @@ class PregelLoop:
            # checkpoint ID since the fork was created after the subgraph's
            # checkpoints from the original execution.
            #
-            # Only gate on is_time_traveling (not is_replaying). When the
+            # Only gate on is_time_traveling (not _addressed). When the
            # client resumes with an explicit checkpoint_id that happens to
            # point at the current head (e.g. LangGraph Studio sending
            # `checkpoint: {checkpoint_id}` alongside Command(resume=...)),
-            # is_replaying is True but is_time_traveling is False. In that
+            # the run is addressed but not time traveling. In that
            # case subgraphs should load their latest checkpoint normally,
            # not go through ReplayState's before-bound lookup which would
            # miss subgraph checkpoints created during processing of the
@@ -1789,7 +1811,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
            # This covers both normal replay and subgraphs resolved via
            # checkpoint_map during time-travel.
            saved = self.checkpointer.get_tuple(self.checkpoint_config)
-            if saved is not None and saved.metadata.get("source") in ("update", "fork"):
+            if saved is not None:
                self._addressed_checkpoint_superseded = checkpoint_superseded(
                    self.checkpointer, self.checkpoint_config, saved
                )
@@ -2045,7 +2067,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
            # This covers both normal replay and subgraphs resolved via
            # checkpoint_map during time-travel.
            saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
-            if saved is not None and saved.metadata.get("source") in ("update", "fork"):
+            if saved is not None:
                self._addressed_checkpoint_superseded = await acheckpoint_superseded(
                    self.checkpointer, self.checkpoint_config, saved
                )
@@ -41,7 +41,7 @@ from langgraph._internal._future import chain_future, run_coroutine_threadsafe
 from langgraph._internal._scratchpad import PregelScratchpad
 from langgraph._internal._typing import MISSING
 from langgraph.constants import TAG_HIDDEN
-from langgraph.errors import GraphBubbleUp, GraphInterrupt
+from langgraph.errors import GraphBubbleUp, GraphInterrupt, is_invalid_resume
 from langgraph.pregel._algo import Call
 from langgraph.pregel._executor import Submit
 from langgraph.pregel._retry import arun_with_retry, run_with_retry
@@ -621,7 +621,11 @@ def _should_stop_others(
    handled_exception_ids: set[int] | None = None,
 ) -> bool:
    """Check if any task failed, if so, cancel all other tasks.
-    GraphInterrupts are not considered failures."""
+    GraphInterrupts are not considered failures.
+
+    Neither is an invalid resume value: it's raised once the other tasks finish, so a
+    task answered in the same resume isn't cut off after its side effects and then run
+    again with its saved answer."""
    for fut in done:
        if fut.cancelled():
            continue
@@ -629,6 +633,7 @@ def _should_stop_others(
            if (
                id(exc) not in (handled_exception_ids or set())
                and not isinstance(exc, GraphBubbleUp)
+                and not is_invalid_resume(exc)
                and fut not in SKIP_RERAISE_SET
            ):
                return True
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"

 [project]
 name = "langgraph"
-version = "1.2.15"
+version = "1.2.16"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"
@@ -521,10 +521,11 @@ def test_addressed_resume_keeps_a_rerun_tasks_write_to_a_new_channel(
    graph = builder.compile(checkpointer=sync_checkpointer)
    config = {"configurable": {"thread_id": "t"}}
    graph.invoke(_both("in"), config, durability=durability)
-    head = graph.get_state(config).config
+    paused = graph.get_state(config).config
+    graph.update_state(config, None)

    live = graph.invoke(
-        Command(resume="yes", update={"flag": True}), head, durability=durability
+        Command(resume="yes", update={"flag": True}), paused, durability=durability
    )

    state = graph.get_state(config)
@@ -152,6 +152,83 @@ def test_finished_task_does_not_report_answered_interrupt(
        assert asked == [["A1", "B"]]


+def test_finished_task_output_applies_once(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    builder = StateGraph(State)
+    builder.add_node("a", lambda s: {"log": [f"a:{interrupt('A')}"]})
+    builder.add_node("b", lambda s: {"log": [f"b:{interrupt('B')}"]})
+    builder.add_node("c", lambda s: {"log": [f"c:{interrupt('C')}"]})
+    builder.add_edge(START, "a")
+    builder.add_edge(START, "b")
+    builder.add_edge(START, "c")
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = _config()
+
+    result = graph.invoke({"log": []}, config, durability=durability)
+    while pending := result.get("__interrupt__"):
+        result = graph.invoke(
+            Command(resume={pending[0].id: "yes"}), config, durability=durability
+        )
+
+    assert sorted(result["log"]) == ["a:yes", "b:yes", "c:yes"]
+
+
+def test_resume_with_update_applies_the_update_once(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    def ask_three_times(state: State) -> State:
+        answers = [interrupt(str(question)) for question in range(1, 4)]
+        return {"log": ["a:" + "".join(answers)]}
+
+    graph = (
+        StateGraph(State)
+        .add_node("a", ask_three_times)
+        .add_edge(START, "a")
+        .compile(checkpointer=sync_checkpointer)
+    )
+    config = _config()
+
+    result = graph.invoke({"log": []}, config, durability=durability)
+    for i in range(1, 4):
+        result = graph.invoke(
+            Command(resume=str(i), update={"log": [f"u{i}"]}),
+            config,
+            durability=durability,
+        )
+
+    assert sorted(result["log"]) == ["a:123", "u1", "u2", "u3"]
+    assert sorted(graph.get_state(config).values["log"]) == sorted(result["log"])
+
+
+@pytest.mark.parametrize("checkpoint_id", ["none", "head"])
+def test_resume_at_the_latest_checkpoint_does_not_rerun_finished_task(
+    sync_checkpointer: BaseCheckpointSaver, checkpoint_id: str
+) -> None:
+    calls: Counter[str] = Counter()
+    graph = _build_parallel(sync_checkpointer, calls)
+    config = _config()
+    graph.invoke({"log": []}, config)
+
+    def resume(question: str, answer: str) -> Any:
+        snapshot = graph.get_state(config)
+        target = (
+            snapshot.config
+            if checkpoint_id == "head"
+            else {"configurable": {**config["configurable"], "checkpoint_id": None}}
+        )
+        return graph.invoke(
+            Command(resume={_interrupt_by_value(snapshot, question).id: answer}),
+            target,
+        )
+
+    resume("A1", "yes")
+    result = resume("B", "ok")
+
+    assert sorted(result["log"]) == ["a:yes", "b:ok"]
+    assert calls == {"a": 2, "b": 3}, "resuming B ran the finished A again"
+
+
@NEEDS_CONTEXTVARS
 async def test_finished_task_does_not_report_answered_interrupt_async(
    async_checkpointer: BaseCheckpointSaver,
@@ -474,6 +551,64 @@ def test_functional_task_finished_with_none_is_not_rerun(
    assert calls == {"a": 2, "b": 3}


+def test_subgraph_finished_task_is_not_rerun(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    calls: Counter[str] = Counter()
+    builder = StateGraph(State)
+    builder.add_node("child", _build_parallel(None, calls))
+    builder.add_edge(START, "child")
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = _config()
+
+    graph.invoke({"log": []}, config, durability=durability)
+    snapshot = graph.get_state(config)
+    graph.invoke(
+        Command(resume={_interrupt_by_value(snapshot, "A1").id: "yes"}),
+        config,
+        durability=durability,
+    )
+    snapshot = graph.get_state(config)
+    result = graph.invoke(
+        Command(resume={_interrupt_by_value(snapshot, "B").id: "ok"}),
+        config,
+        durability=durability,
+    )
+
+    assert sorted(result["log"]) == ["a:yes", "b:ok"]
+    assert calls == {"a": 2, "b": 3}, "resuming B ran the finished A again"
+
+
+def test_nested_entrypoint_finished_task_is_not_rerun(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    calls: Counter[str] = Counter()
+
+    @task
+    def step_1(query: str) -> str:
+        calls["step_1"] += 1
+        return f"{query} bar"
+
+    @task
+    def human_feedback(query: str) -> str:
+        calls["human_feedback"] += 1
+        return f"{query} {interrupt('feedback?')}"
+
+    @entrypoint()
+    def child(query: str) -> str:
+        return human_feedback(step_1(query).result()).result()
+
+    @entrypoint(checkpointer=sync_checkpointer)
+    def parent(query: str) -> str:
+        return child.invoke(query)
+
+    config = _config()
+    parent.invoke("foo", config)
+
+    assert parent.invoke(Command(resume="baz"), config) == "foo bar baz"
+    assert calls == {"step_1": 1, "human_feedback": 2}
+
+
 # --- Reading task status from recorded writes ---


@@ -1,5 +1,8 @@
+import asyncio
+import operator
+import sys
 from dataclasses import dataclass
-from typing import Any
+from typing import Annotated, Any

 import pytest
 from langgraph.checkpoint.base import BaseCheckpointSaver
@@ -251,3 +254,61 @@ def test_interrupt_response_schema_invalid_resume_after_earlier_interrupt(
    assert graph.invoke(resume({"approved": True}), config) == {
        "answer": ["ok", Decision(approved=True)]
    }
+
+
+@pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="Python 3.11+ is required for async contextvars support",
+)
+async def test_invalid_resume_lets_other_answered_tasks_finish(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """An invalid answer fails the resume without cutting off the other answered tasks.
+
+    A task cancelled after its side effect keeps its answer and runs again on the next
+    resume, so the side effect would happen twice.
+    """
+    sent: list[str] = []
+
+    class State(TypedDict):
+        log: Annotated[list[str], operator.add]
+
+    async def send(state: State) -> State:
+        interrupt("send?", response_schema=Decision)
+        sent.append("email")
+        await asyncio.sleep(0.05)  # waiting on the reply to a request already sent
+        return {"log": ["sent"]}
+
+    async def check(state: State) -> State:
+        interrupt("check?", response_schema=Decision)
+        return {"log": ["checked"]}
+
+    graph = (
+        StateGraph(State)
+        .add_node("send", send)
+        .add_node("check", check)
+        .add_edge(START, "send")
+        .add_edge(START, "check")
+        .compile(checkpointer=async_checkpointer)
+    )
+    config = {"configurable": {"thread_id": "1"}}
+    await graph.ainvoke({"log": []}, config)
+    ids = {i.value: i.id for i in (await graph.aget_state(config)).interrupts}
+
+    with pytest.raises(ValidationError, match="approved"):
+        await graph.ainvoke(
+            Command(
+                resume={
+                    ids["send?"]: {"approved": True},
+                    ids["check?"]: {"approved": "nope"},
+                }
+            ),
+            config,
+        )
+    pending = [i.value for i in (await graph.aget_state(config)).interrupts]
+    assert (pending, sent) == (["check?"], ["email"])
+
+    result = await graph.ainvoke(
+        Command(resume={ids["check?"]: {"approved": True}}), config
+    )
+    assert (sorted(result["log"]), sent) == (["checked", "sent"], ["email"])
@@ -7737,6 +7737,8 @@ def test_parallel_interrupts(sync_checkpointer: BaseCheckpointSaver) -> None:
 def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> None:
    # --- CHILD GRAPH ---

+    answered: Counter[str] = Counter()
+
    class ChildState(BaseModel):
        prompt: str = Field(..., description="What is going to be asked to the user?")
        human_input: str | None = Field(None, description="What the human said")
@@ -7746,6 +7748,7 @@ def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> N

    def get_human_input(state: ChildState):
        human_input = interrupt(state.prompt)
+        answered["human"] += 1

        return dict(
            human_inputs=[human_input],  # update parent state
@@ -7753,6 +7756,7 @@ def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> N

    def get_dolphin_input(state: ChildState):
        human_input = interrupt(state.prompt)
+        answered["dolphin"] += 1

        return dict(
            human_inputs=[human_input],  # update parent state
@@ -7846,6 +7850,7 @@ def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> N

    assert invokes == 5
    assert len(events) == 5
+    assert answered == {"human": 2, "dolphin": 2}


 def test_pregel_loop_refcount():
@@ -1177,6 +1177,43 @@ def test_subgraph_interrupt_resume_with_explicit_head_checkpoint_id(
    assert result["value"] == ["sub_a", "human:answer", "sub_b"]


+def test_subgraph_replay_reasks_answered_parallel_interrupt(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Replaying a subgraph checkpoint asks its step's questions again, including
+    one already answered. An update checkpoint replays in place instead of
+    forking, so its tasks keep their ids and its saved answers could be reused."""
+
+    subgraph = (
+        StateGraph(State)
+        .add_node("a", lambda s: {"value": [f"a:{interrupt('a')}"]})
+        .add_node("b", lambda s: {"value": [f"b:{interrupt('b')}"]})
+        .add_edge(START, "a")
+        .add_edge(START, "b")
+        .compile(checkpointer=True)
+    )
+    graph = (
+        StateGraph(State)
+        .add_node("subgraph_node", subgraph)
+        .add_edge(START, "subgraph_node")
+        .compile(checkpointer=sync_checkpointer)
+    )
+    config = {"configurable": {"thread_id": "1"}}
+
+    def subgraph_head() -> RunnableConfig:
+        return graph.get_state(config, subgraphs=True).tasks[0].state.config
+
+    graph.invoke({"value": []}, config)
+    graph.update_state(subgraph_head(), {"value": ["u"]}, as_node="__start__")
+    result = graph.invoke(None, config)
+    question_a = next(i for i in result["__interrupt__"] if i.value == "a")
+    graph.invoke(Command(resume={question_a.id: "yes"}), config)
+
+    result = graph.invoke(None, subgraph_head())
+
+    assert sorted(i.value for i in result["__interrupt__"]) == ["a", "b"]
+
+
 def test_subgraph_replay_loads_accumulated_state_then_resume(
    sync_checkpointer: BaseCheckpointSaver,
 ) -> None:
@@ -1451,7 +1451,7 @@ wheels = [

 [[package]]
 name = "langgraph"
-version = "1.2.15"
+version = "1.2.16"
 source = { editable = "." }
 dependencies = [
    { name = "langchain-core" },
@@ -1835,7 +1835,7 @@ test = [

 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "../prebuilt" }
 dependencies = [
    { name = "langchain-core" },
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"

 [project]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 description = "Library with high-level APIs for creating and executing LangGraph agents and tools."
 authors = []
 requires-python = ">=3.10"
@@ -286,7 +286,7 @@ wheels = [

 [[package]]
 name = "langgraph"
-version = "1.2.15"
+version = "1.2.16"
 source = { editable = "../langgraph" }
 dependencies = [
    { name = "langchain-core" },
@@ -512,7 +512,7 @@ test = [

 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "." }
 dependencies = [
    { name = "langchain-core" },
@@ -3,7 +3,7 @@ from langgraph_sdk.client import get_client, get_sync_client
 from langgraph_sdk.encryption import Encryption
 from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext

-__version__ = "0.4.7"
+__version__ = "0.4.8"

 __all__ = [
    "Auth",
@@ -15,6 +15,7 @@ from __future__ import annotations

 import asyncio
 import contextlib
+import logging
 import random
 from collections.abc import AsyncGenerator, AsyncIterator, Generator, Mapping
 from dataclasses import dataclass, field
@@ -36,6 +37,11 @@ from langgraph_sdk.stream.decoders import (
    ToolCallsDecoder,
    validate_interleave_channels,
 )
+from langgraph_sdk.stream.errors import (
+    is_non_retryable_stream_error,
+    protocol_error,
+    stream_error_frame,
+)
 from langgraph_sdk.stream.subscription import compute_union_filter, infer_channel
 from langgraph_sdk.stream.transport import (
    AsyncProtocolTransport,
@@ -44,6 +50,8 @@ from langgraph_sdk.stream.transport import (
    ProtocolWebSocketTransport,
 )

+logger = logging.getLogger(__name__)
+

 class InterruptPayload(TypedDict):
    """Payload surfaced when the server requests human input for a thread."""
@@ -193,6 +201,7 @@ class RunModule:
            if not gate.done():
                gate.set_result(None)
            self._owner._run_seen = True
+            self._owner._ensure_lifecycle_watcher_running()
            return result
        except BaseException as err:
            # Why: gate MUST reject on any exit type, including CancelledError,
@@ -1203,6 +1212,7 @@ class AsyncThreadStream:
        # observes a `completed`/`errored` event.
        self._interrupts_lock = asyncio.Lock()
        self._lifecycle_watcher_task: asyncio.Task[None] | None = None
+        self._lifecycle_watcher_rejected = False
        self._lifecycle_watcher_handle: EventStreamHandle | None = None
        self._lifecycle_cursor: int | None = None
        self._lifecycle_max_reconnect_attempts = 5
@@ -1261,8 +1271,11 @@ class AsyncThreadStream:
        # Create the run-done future here (async context guarantees a running loop).
        self._run_done = asyncio.get_running_loop().create_future()
        # Start the lifecycle watcher immediately so reattach and thread.output
-        # work without a preceding run.start call.
-        self._ensure_lifecycle_watcher_running()
+        # work without a preceding run.start call. A minted thread_id names no
+        # thread until run.start creates it, and with custom auth a stream for a
+        # thread that doesn't exist yet fails, so run.start starts the watcher.
+        if self._explicit_thread_id:
+            self._ensure_lifecycle_watcher_running()
        return self

    async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
@@ -1630,6 +1643,13 @@ class AsyncThreadStream:
                async for event in self._dedup_iter(shared.events):
                    if self._closed:
                        break
+                    if (frame_error := stream_error_frame(event)) is not None:
+                        logger.warning(
+                            "Thread %s event stream failed: %s",
+                            self.thread_id,
+                            frame_error,
+                        )
+                        continue
                    self._observe_event(event)
                    for sub in list(self._subscriptions.values()):
                        if matches_subscription(event, sub.params):
@@ -1655,6 +1675,7 @@ class AsyncThreadStream:
                if (
                    err is not None
                    and not isinstance(err, asyncio.CancelledError)
+                    and not is_non_retryable_stream_error(err)
                    and not self._closed
                ):
                    with contextlib.suppress(Exception):
@@ -1681,7 +1702,8 @@ class AsyncThreadStream:

        Returns:
            `True` if a new stream was opened (caller should resume fanout),
-            `False` if all reconnect attempts were exhausted or the controller
+            `False` if all reconnect attempts were exhausted, the server
+            rejected the stream with a non-retryable 4xx, or the controller
            was closed in the meantime.
        """
        if self._transport is None:
@@ -1702,7 +1724,9 @@ class AsyncThreadStream:
                await new_stream.ready
            except asyncio.CancelledError:
                raise
-            except Exception:
+            except Exception as err:
+                if is_non_retryable_stream_error(err):
+                    return False
                await self._reconnect_sleep(attempt)
                continue
            self._shared_stream = new_stream
@@ -1800,9 +1824,7 @@ class AsyncThreadStream:
            # 202/204 — no body. Caller gets an empty result.
            return {}
        if response.get("type") == "error":
-            code = response.get("error", "unknown")
-            message = response.get("message", "")
-            raise RuntimeError(f"Protocol error [{code}]: {message}")
+            raise protocol_error(response)
        meta = response.get("meta")
        if isinstance(meta, dict):
            applied_through_seq = meta.get("applied_through_seq")
@@ -1826,8 +1848,15 @@ class AsyncThreadStream:
            await asyncio.wait_for(asyncio.shield(gate), timeout=timeout)

    def _ensure_lifecycle_watcher_running(self) -> None:
-        if self._lifecycle_watcher_task is not None:
+        task = self._lifecycle_watcher_task
+        if task is not None and not self._lifecycle_watcher_rejected:
            return
+        if task is not None:
+            # A 4xx stopped the watcher before this run started, typically the
+            # 404 for a thread this run.start just created, so the result it
+            # left isn't this run's.
+            self._run_done = asyncio.get_running_loop().create_future()
+            self._lifecycle_watcher_rejected = False
        self._lifecycle_watcher_task = asyncio.create_task(
            self._run_lifecycle_watcher()
        )
@@ -1850,7 +1879,11 @@ class AsyncThreadStream:
        if self._transport is None:
            return
        reconnect_attempts = 0
+        frame_error: RuntimeError | None = None
        while not self._closed:
+            # A stream opened before run.start can 404 for the thread that
+            # run.start creates, so it gets one more try once a run has started.
+            opened_before_run = not self._run_seen
            try:
                handle = self._transport.open_event_stream(
                    self._lifecycle_stream_params()
@@ -1860,6 +1893,7 @@ class AsyncThreadStream:
                async for event in handle.events:
                    if self._closed:
                        return
+                    frame_error = stream_error_frame(event) or frame_error
                    self._observe_lifecycle_event(event)
                    await self._apply_lifecycle_event(event)
                err = await handle.done
@@ -1868,38 +1902,43 @@ class AsyncThreadStream:
                    # event. Resolve `_run_done` as errored so awaiters of
                    # `thread.output` don't hang.
                    if err is None:
-                        run_done = self._run_done
-                        if run_done is not None and not run_done.done():
-                            run_done.set_result(
-                                _RunTerminal(
-                                    status="errored",
-                                    error=RuntimeError(
-                                        "lifecycle stream ended before terminal event"
-                                    ),
-                                )
+                        self._fail_run_done(
+                            frame_error
+                            or RuntimeError(
+                                "lifecycle stream ended before terminal event"
                            )
+                        )
                    return
                reconnect_attempts += 1
-                if reconnect_attempts > self._lifecycle_max_reconnect_attempts:
+                if (
+                    is_non_retryable_stream_error(err)
+                    or reconnect_attempts > self._lifecycle_max_reconnect_attempts
+                ):
                    raise err
                await asyncio.sleep(0.05)
            except asyncio.CancelledError:
                raise
            except Exception as exc:
+                if is_non_retryable_stream_error(exc):
+                    if opened_before_run and self._run_seen:
+                        continue
+                    self._lifecycle_watcher_rejected = True
+                    self._fail_run_done(exc)
+                    return
                reconnect_attempts += 1
                if reconnect_attempts <= self._lifecycle_max_reconnect_attempts:
                    await asyncio.sleep(0.05)
                    continue
-                run_done = self._run_done
-                if run_done is not None and not run_done.done():
-                    run_done.set_result(
-                        _RunTerminal(
-                            status="errored",
-                            error=RuntimeError(f"Lifecycle transport failed: {exc}"),
-                        )
-                    )
+                self._fail_run_done(
+                    frame_error or RuntimeError(f"Lifecycle transport failed: {exc}")
+                )
                return

+    def _fail_run_done(self, error: BaseException) -> None:
+        run_done = self._run_done
+        if run_done is not None and not run_done.done():
+            run_done.set_result(_RunTerminal(status="errored", error=error))
+
    async def _fetch_state(self) -> dict[str, Any]:
        """Fetch the current thread state from the REST endpoint."""
        return await self._http.get(
@@ -34,6 +34,11 @@ from langgraph_sdk.stream.decoders import (
    ToolCallsDecoder,
    validate_interleave_channels,
 )
+from langgraph_sdk.stream.errors import (
+    is_non_retryable_stream_error,
+    protocol_error,
+    stream_error_frame,
+)
 from langgraph_sdk.stream.subscription import compute_union_filter, infer_channel
 from langgraph_sdk.stream.sync_controller import (
    SyncStreamController,
@@ -237,6 +242,7 @@ class SyncRunModule:
        controller = self._owner._controller
        if controller is not None and controller._run_start_gate is not None:
            controller._run_start_gate.set()
+        self._owner._ensure_lifecycle_watcher_running()
        return result

    def respond(
@@ -1111,6 +1117,7 @@ class SyncThreadStream:
        self.interrupted: bool = False
        self.interrupts: list[InterruptPayload] = []
        self._lifecycle_watcher_thread: threading.Thread | None = None
+        self._lifecycle_watcher_lock = threading.Lock()
        self._lifecycle_watcher_handle: SyncEventStreamHandle | None = None
        self._lifecycle_cursor: int | None = None
        self._lifecycle_max_reconnect_attempts = 5
@@ -1149,7 +1156,11 @@ class SyncThreadStream:
            self._transport, run_start_gate=run_start_gate
        )
        self._run_done = _BlockingResult()
-        self._ensure_lifecycle_watcher_running()
+        # A minted thread_id names no thread until run.start creates it, and
+        # with custom auth a stream for a thread that doesn't exist yet fails,
+        # so run.start starts the watcher instead.
+        if self._explicit_thread_id:
+            self._ensure_lifecycle_watcher_running()
        return self

    def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
@@ -1485,9 +1496,7 @@ class SyncThreadStream:
        if response is None:
            return {}
        if response.get("type") == "error":
-            code = response.get("error", "unknown")
-            message = response.get("message", "")
-            raise RuntimeError(f"Protocol error [{code}]: {message}")
+            raise protocol_error(response)
        meta = response.get("meta")
        if isinstance(meta, dict):
            applied_through_seq = meta.get("applied_through_seq")
@@ -1496,14 +1505,20 @@ class SyncThreadStream:
        return response.get("result", {})

    def _ensure_lifecycle_watcher_running(self) -> None:
-        if self._lifecycle_watcher_thread is not None:
-            return
-        self._lifecycle_watcher_thread = threading.Thread(
-            target=self._run_lifecycle_watcher,
-            name="langgraph-sdk-sync-lifecycle",
-            daemon=True,
-        )
-        self._lifecycle_watcher_thread.start()
+        with self._lifecycle_watcher_lock:
+            if self._lifecycle_watcher_thread is not None:
+                return
+            if self._run_done is not None and self._run_done.done():
+                # A 4xx stopped the watcher before this run started, typically
+                # the 404 for a thread this run.start just created, so the
+                # result it left isn't this run's.
+                self._run_done = _BlockingResult()
+            self._lifecycle_watcher_thread = threading.Thread(
+                target=self._run_lifecycle_watcher,
+                name="langgraph-sdk-sync-lifecycle",
+                daemon=True,
+            )
+            self._lifecycle_watcher_thread.start()

    def _observe_lifecycle_event(self, event: Event) -> None:
        seq = event.get("seq")
@@ -1523,7 +1538,11 @@ class SyncThreadStream:
        if self._transport is None:
            return
        reconnect_attempts = 0
+        frame_error: RuntimeError | None = None
        while not self._closed:
+            # A stream opened before run.start can 404 for the thread that
+            # run.start creates, so it gets one more try once a run has started.
+            opened_before_run = not self._run_seen
            try:
                handle = self._transport.open_event_stream(
                    self._lifecycle_stream_params()
@@ -1532,28 +1551,64 @@ class SyncThreadStream:
                for event in handle.events:
                    if self._closed:
                        return
+                    frame_error = stream_error_frame(event) or frame_error
                    self._observe_lifecycle_event(event)
                    self._apply_lifecycle_event(event)
                err = handle.error()
                if err is None:
+                    # Clean EOF: stream ended without a terminal lifecycle
+                    # event. Resolve `_run_done` as errored so `thread.output`
+                    # doesn't block forever.
+                    if not self._closed:
+                        self._fail_run_done(
+                            frame_error
+                            or RuntimeError(
+                                "lifecycle stream ended before terminal event"
+                            )
+                        )
                    return
                reconnect_attempts += 1
-                if reconnect_attempts > self._lifecycle_max_reconnect_attempts:
+                if (
+                    is_non_retryable_stream_error(err)
+                    or reconnect_attempts > self._lifecycle_max_reconnect_attempts
+                ):
                    raise err
            except Exception as exc:
+                if is_non_retryable_stream_error(exc):
+                    if self._stop_rejected_lifecycle_watcher(
+                        exc, unless_run_started=opened_before_run
+                    ):
+                        return
+                    continue
                reconnect_attempts += 1
                if reconnect_attempts <= self._lifecycle_max_reconnect_attempts:
                    continue
-                run_done = self._run_done
-                if run_done is not None and not run_done.done():
-                    run_done.set_result(
-                        _RunTerminal(
-                            status="errored",
-                            error=RuntimeError(f"Lifecycle transport failed: {exc}"),
-                        )
-                    )
+                self._fail_run_done(
+                    frame_error or RuntimeError(f"Lifecycle transport failed: {exc}")
+                )
                return

+    def _stop_rejected_lifecycle_watcher(
+        self, error: BaseException, *, unless_run_started: bool = False
+    ) -> bool:
+        """Fail the run with a 4xx `error` and clear the watcher, so run.start starts another.
+
+        Returns False, changing nothing, when `unless_run_started` is set and a
+        run has started. Holding the lock keeps a run.start from seeing this
+        watcher as still running once it has decided to stop.
+        """
+        with self._lifecycle_watcher_lock:
+            if unless_run_started and self._run_seen:
+                return False
+            self._fail_run_done(error)
+            self._lifecycle_watcher_thread = None
+            return True
+
+    def _fail_run_done(self, error: BaseException) -> None:
+        run_done = self._run_done
+        if run_done is not None and not run_done.done():
+            run_done.set_result(_RunTerminal(status="errored", error=error))
+
    def _fetch_state(self) -> dict[str, Any]:
        """Fetch the current thread state from the REST endpoint."""
        return self._http.get(
@@ -0,0 +1,46 @@
+"""Thread-stream failures, shared by the async and sync clients."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+from websockets.exceptions import ConnectionClosed, InvalidStatus
+
+
+def protocol_error(response: dict[str, Any]) -> RuntimeError:
+    """Build the error for a protocol `{"type": "error"}` envelope."""
+    code = response.get("error", "unknown")
+    message = response.get("message", "")
+    return RuntimeError(f"Protocol error [{code}]: {message}")
+
+
+def stream_error_frame(event: Any) -> RuntimeError | None:
+    """Return the error in an unsolicited `{"type": "error", "id": null}` frame.
+
+    The server sends one just before it closes a thread stream whose read failed.
+    """
+    if (
+        isinstance(event, dict)
+        and event.get("type") == "error"
+        and event.get("id") is None
+    ):
+        return protocol_error(event)
+    return None
+
+
+def is_non_retryable_stream_error(error: BaseException | None) -> bool:
+    """Whether reopening the stream would fail the same way.
+
+    True for a 4xx other than 408 and 429: the status of a refused stream request
+    or WebSocket upgrade, or a WebSocket close code of 4000 plus that status. A
+    server that rejects a WebSocket after accepting it closes with such a code,
+    because a browser can't read the status of a refused upgrade.
+    """
+    if isinstance(error, (httpx.HTTPStatusError, InvalidStatus)):
+        status = error.response.status_code
+    elif isinstance(error, ConnectionClosed) and error.rcvd is not None:
+        status = error.rcvd.code - 4000
+    else:
+        return False
+    return 400 <= status < 500 and status not in (408, 429)
@@ -13,6 +13,10 @@ from typing import Any

 from langchain_protocol import Event, SubscribeParams

+from langgraph_sdk.stream.errors import (
+    is_non_retryable_stream_error,
+    stream_error_frame,
+)
 from langgraph_sdk.stream.subscription import compute_union_filter, filter_covers
 from langgraph_sdk.stream.transport import (
    SyncEventStreamHandle,
@@ -167,6 +171,13 @@ class SyncStreamController:
                return
            try:
                for event in self._dedup_iter(shared.events):
+                    if (frame_error := stream_error_frame(event)) is not None:
+                        _logger.warning(
+                            "Thread %s event stream failed: %s",
+                            self._transport.thread_id,
+                            frame_error,
+                        )
+                        continue
                    with self._lock:
                        if self._closed:
                            break
@@ -190,7 +201,11 @@ class SyncStreamController:
                    continue  # rotation happened; pick up new stream
                # No rotation — check if this was a transport drop
            err = shared.error()
-            if err is not None and not self._closed:
+            if (
+                err is not None
+                and not is_non_retryable_stream_error(err)
+                and not self._closed
+            ):
                if self._reconnect_shared_stream():
                    continue
            break
@@ -199,25 +214,6 @@ class SyncStreamController:
            for sub in self._subscriptions.values():
                sub.queue.put(None)

-    def _reconnect_shared_stream(self) -> bool:
-        with self._lock:
-            base_filter = self._shared_stream_filter
-        if base_filter is None:
-            return False
-        for _ in range(self._max_reconnect_attempts):
-            with self._lock:
-                if self._closed:
-                    return False
-                params = self._filter_with_since(base_filter)
-            try:
-                new_stream = self._transport.open_event_stream(params)
-            except Exception:
-                continue
-            with self._lock:
-                self._shared_stream = new_stream
-            return True
-        return False
-
    def _compute_current_union(
        self, extra: SubscribeParams | None = None
    ) -> dict[str, Any]:
@@ -296,7 +292,8 @@ class SyncStreamController:
        """Attempt to reopen the shared stream after a transport drop.

        Returns True if a new stream was successfully opened, False if all
-        reconnect attempts were exhausted or the controller was closed.
+        reconnect attempts were exhausted, the server rejected the stream with a
+        non-retryable 4xx, or the controller was closed.
        """
        base_filter = self._shared_stream_filter
        if base_filter is None:
@@ -317,6 +314,8 @@ class SyncStreamController:
                        old.close()
                return True
            except Exception as err:
+                if is_non_retryable_stream_error(err):
+                    return False
                _logger.debug("sync reconnect attempt %d failed: %r", attempt, err)
        return False

@@ -69,6 +69,16 @@ def lifecycle_errored_event(
    return _base(seq, "lifecycle", namespace or [], {"event": "failed", "error": error})


+def thread_stream_failed_frame() -> dict[str, Any]:
+    """Unsolicited error frame the server sends before closing a failed stream."""
+    return {
+        "type": "error",
+        "id": None,
+        "error": "unknown_error",
+        "message": "Thread event stream failed.",
+    }
+
+
 def values_event(
    seq: int = 0, namespace: list[str] | None = None, **data: Any
 ) -> dict[str, Any]:
@@ -60,6 +60,7 @@ class _StreamScript:
    events: list[dict[str, Any]]
    delay: float = 0.0
    fail_after: int | None = None
+    status: int = 200


 class FakeServer:
@@ -263,6 +264,8 @@ class FakeServer:
                    events=list(self.scripted_events), delay=self._stream_delay
                )
            )
+            if script.status != 200:
+                return httpx.Response(script.status, json={"detail": "scripted"})
            self.open_event_streams += 1
            self._open_event_streams_max = max(
                self._open_event_streams_max, self.open_event_streams
@@ -14,6 +14,7 @@ import orjson
 class SyncStreamScript:
    events: list[dict[str, Any]]
    fail_after: int | None = None
+    status: int = 200


 class _SseByteStream(httpx.SyncByteStream):
@@ -113,6 +114,8 @@ class SyncFakeServer:
                if self._stream_scripts
                else SyncStreamScript(events=list(self.scripted_events))
            )
+            if script.status != 200:
+                return httpx.Response(script.status, json={"detail": "scripted"})
            return httpx.Response(
                200,
                headers={"content-type": "text/event-stream"},
@@ -17,6 +17,7 @@ from streaming._events import (
    lifecycle_completed_event,
    lifecycle_errored_event,
    lifecycle_event,
+    thread_stream_failed_frame,
 )
 from streaming._fake_server import FakeServer, _StreamScript

@@ -48,7 +49,7 @@ async def test_interrupts_populated_from_input_requested_event():


 async def test_aenter_starts_lifecycle_watcher():
-    """Entering AsyncThreadStream opens lifecycle/input SSE before run.start."""
+    """Entering with an explicit thread_id opens the lifecycle/input SSE before run.start."""
    fake = FakeServer()
    fake.script([lifecycle_event(seq=0, phase="started")])
    asgi = httpx.ASGITransport(app=fake.app)
@@ -274,3 +275,137 @@ async def test_lifecycle_watcher_reconnects_with_since_after_transport_drop():
    assert fake.stream_request_bodies[0]["channels"] == ["lifecycle", "input"]
    assert fake.stream_request_bodies[1]["channels"] == ["lifecycle", "input"]
    assert fake.stream_request_bodies[1]["since"] == 1
+
+
+async def test_lifecycle_error_frame_becomes_the_run_error():
+    fake = FakeServer()
+    fake.script([lifecycle_event(seq=0, phase="running"), thread_stream_failed_frame()])
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+    assert terminal.status == "errored"
+    assert (
+        str(terminal.error)
+        == "Protocol error [unknown_error]: Thread event stream failed."
+    )
+
+
+async def test_lifecycle_watcher_stops_on_a_404_without_retrying():
+    fake = FakeServer()
+    fake.script_sequence([_StreamScript(events=[], status=404)])
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+    assert isinstance(terminal.error, httpx.HTTPStatusError)
+    assert terminal.error.response.status_code == 404
+    assert len(fake.stream_request_bodies) == 1
+
+
+async def test_minted_thread_opens_the_lifecycle_stream_after_run_start():
+    fake = FakeServer()
+    fake.script([lifecycle_completed_event(seq=1)])
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(assistant_id="agent") as thread:
+            assert thread._lifecycle_watcher_task is None
+            await thread.run.start(input={})
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+    assert terminal.status == "completed"
+
+
+def _new_thread_scripts() -> list[_StreamScript]:
+    return [
+        _StreamScript(events=[], status=404),
+        _StreamScript(events=[lifecycle_completed_event(seq=1)]),
+    ]
+
+
+async def test_run_start_restarts_a_watcher_stopped_by_a_404_for_a_new_thread():
+    fake = FakeServer()
+    fake.script_sequence(_new_thread_scripts())
+    fake.set_state({"ok": True})
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-new", assistant_id="agent") as thread:
+            rejected = thread._run_done
+            assert rejected is not None
+            await asyncio.wait_for(rejected, timeout=2.0)
+            await thread.run.start(input={})
+            output = await thread.output.with_timeout(2.0)
+
+    assert output == {"ok": True}
+    assert len(fake.stream_request_bodies) == 2
+
+
+async def test_lifecycle_watcher_retries_a_404_that_arrives_after_run_start():
+    fake = FakeServer()
+    fake.script_sequence(_new_thread_scripts())
+    fake.set_state({"ok": True})
+    stream_opened = asyncio.Event()
+    run_started = asyncio.Event()
+
+    async def handle(request: httpx.Request) -> httpx.Response:
+        if (
+            request.url.path.endswith("/stream/events")
+            and not fake.stream_request_bodies
+        ):
+            stream_opened.set()
+            await run_started.wait()
+        return await fake._handle_request(request)
+
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(handle), base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-new", assistant_id="agent") as thread:
+            await asyncio.wait_for(stream_opened.wait(), timeout=2.0)
+            await thread.run.start(input={})
+            run_started.set()
+            output = await thread.output.with_timeout(2.0)
+
+    assert output == {"ok": True}
+    assert len(fake.stream_request_bodies) == 2
+
+
+async def test_lifecycle_watcher_reports_the_error_frame_when_reconnects_run_out():
+    fake = FakeServer()
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        stream = threads.stream(thread_id="t-1", assistant_id="agent")
+        stream._lifecycle_max_reconnect_attempts = 1
+        fake.script_sequence(
+            [
+                _StreamScript(events=[thread_stream_failed_frame()], fail_after=1)
+                for _ in range(stream._lifecycle_max_reconnect_attempts + 1)
+            ]
+        )
+        async with stream as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+    assert (
+        str(terminal.error)
+        == "Protocol error [unknown_error]: Thread event stream failed."
+    )
@@ -1,11 +1,13 @@
 from __future__ import annotations

 import asyncio
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, AsyncIterator
 from typing import Any, cast
 from unittest.mock import MagicMock

 import httpx
+from websockets.exceptions import ConnectionClosedError
+from websockets.frames import Close

 from langgraph_sdk._async.http import HttpClient
 from langgraph_sdk._async.threads import ThreadsClient
@@ -348,3 +350,66 @@ async def test_send_command_applied_through_seq_seeds_shared_stream_since():
    ]
    assert len(values_requests) == 1
    assert values_requests[0]["since"] == 17
+
+
+async def _drain(events: AsyncIterator[Any]) -> list[Any]:
+    return [event async for event in events]
+
+
+async def test_shared_stream_stops_reconnecting_on_a_404():
+    fake = FakeServer()
+    fake.script_sequence(
+        [
+            _StreamScript(events=[values_event(seq=1)], fail_after=1),
+            _StreamScript(events=[], status=404),
+        ]
+    )
+    async with httpx.AsyncClient(
+        transport=fake.transport, base_url="http://test"
+    ) as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(assistant_id="agent") as thread:
+            received = await asyncio.wait_for(
+                _drain(thread.subscribe(["values"])), timeout=5.0
+            )
+
+    assert [event["seq"] for event in received] == [1]
+    assert len(fake.stream_request_bodies) == 2
+
+
+async def test_shared_stream_does_not_reconnect_after_a_4404_close(
+    monkeypatch: Any,
+) -> None:
+    opened: list[dict[str, Any]] = []
+
+    def open_then_close_with_4404(
+        _self: ProtocolSseTransport, params: dict[str, Any]
+    ) -> EventStreamHandle:
+        opened.append(params)
+        loop = asyncio.get_running_loop()
+        ready: asyncio.Future[None] = loop.create_future()
+        ready.set_result(None)
+        done: asyncio.Future[BaseException | None] = loop.create_future()
+        done.set_result(ConnectionClosedError(Close(4404, "not found"), None))
+
+        async def events() -> AsyncGenerator[Any, None]:
+            await asyncio.sleep(0)
+            yield values_event(seq=1)
+
+        async def close() -> None:
+            return None
+
+        return EventStreamHandle(events=events(), ready=ready, done=done, close=close)
+
+    monkeypatch.setattr(
+        ProtocolSseTransport, "open_event_stream", open_then_close_with_4404
+    )
+    async with httpx.AsyncClient(base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(assistant_id="agent") as thread:
+            received = await asyncio.wait_for(
+                _drain(thread.subscribe(["values"])), timeout=5.0
+            )
+
+    assert [event["seq"] for event in received] == [1]
+    assert len(opened) == 1
@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import httpx
+import pytest
+from websockets.datastructures import Headers
+from websockets.exceptions import ConnectionClosedError, InvalidStatus
+from websockets.frames import Close
+from websockets.http11 import Response
+
+from langgraph_sdk.stream.errors import (
+    is_non_retryable_stream_error,
+    stream_error_frame,
+)
+
+
+def _http_error(status: int) -> httpx.HTTPStatusError:
+    request = httpx.Request("POST", "http://test/threads/t-1/stream/events")
+    response = httpx.Response(status, request=request)
+    return httpx.HTTPStatusError("scripted", request=request, response=response)
+
+
+def _ws_close(code: int) -> ConnectionClosedError:
+    return ConnectionClosedError(Close(code, "scripted"), None)
+
+
+@pytest.mark.parametrize(
+    ("error", "expected"),
+    [
+        pytest.param(_http_error(403), True, id="http-403"),
+        pytest.param(_http_error(404), True, id="http-404"),
+        pytest.param(_http_error(408), False, id="http-408"),
+        pytest.param(_http_error(429), False, id="http-429"),
+        pytest.param(_http_error(503), False, id="http-503"),
+        pytest.param(_ws_close(4404), True, id="ws-close-4404"),
+        pytest.param(_ws_close(4422), True, id="ws-close-4422"),
+        pytest.param(_ws_close(4429), False, id="ws-close-4429"),
+        pytest.param(_ws_close(4503), False, id="ws-close-4503"),
+        pytest.param(_ws_close(1011), False, id="ws-close-1011"),
+        pytest.param(
+            InvalidStatus(Response(403, "Forbidden", Headers())),
+            True,
+            id="ws-upgrade-403",
+        ),
+        pytest.param(RuntimeError("network down"), False, id="no-status"),
+        pytest.param(None, False, id="none"),
+    ],
+)
+def test_is_non_retryable_stream_error(
+    error: BaseException | None, expected: bool
+) -> None:
+    assert is_non_retryable_stream_error(error) is expected
+
+
+def test_stream_error_frame_reads_only_unsolicited_error_frames() -> None:
+    frame = {
+        "type": "error",
+        "id": None,
+        "error": "unknown_error",
+        "message": "Thread event stream failed.",
+    }
+
+    assert (
+        str(stream_error_frame(frame))
+        == "Protocol error [unknown_error]: Thread event stream failed."
+    )
+    assert stream_error_frame({**frame, "id": 3}) is None
+    assert stream_error_frame({"type": "event", "method": "values"}) is None
@@ -35,6 +35,7 @@ from streaming._events import (
    message_text_delta_event,
    message_text_finish_event,
    tasks_start_event,
+    thread_stream_failed_frame,
    tool_finished_event,
    tool_started_event,
    updates_event,
@@ -868,3 +869,196 @@ def test_interleave_projections_data_channel_scoped_to_root_namespace():
            ]
    assert {"scope": "root"} in checkpoints
    assert {"scope": "child"} not in checkpoints
+
+
+def test_sync_lifecycle_clean_eof_fails_the_run_instead_of_blocking():
+    fake = SyncFakeServer()
+    fake.script([lifecycle_started_event(seq=0)])
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = run_done.result(timeout=2.0)
+
+    assert terminal.status == "errored"
+    assert "ended before terminal" in str(terminal.error)
+
+
+def test_sync_lifecycle_error_frame_becomes_the_run_error():
+    fake = SyncFakeServer()
+    fake.script([lifecycle_event(seq=0, phase="running"), thread_stream_failed_frame()])
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = run_done.result(timeout=2.0)
+
+    assert terminal.status == "errored"
+    assert (
+        str(terminal.error)
+        == "Protocol error [unknown_error]: Thread event stream failed."
+    )
+
+
+def test_sync_lifecycle_watcher_stops_on_a_404_without_retrying():
+    fake = SyncFakeServer()
+    fake.script_sequence([SyncStreamScript(events=[], status=404)])
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = run_done.result(timeout=2.0)
+
+    assert isinstance(terminal.error, httpx.HTTPStatusError)
+    assert terminal.error.response.status_code == 404
+    assert len(fake.stream_request_bodies) == 1
+
+
+def test_sync_minted_thread_opens_the_lifecycle_stream_after_run_start():
+    fake = SyncFakeServer()
+    fake.script([lifecycle_completed_event(seq=1)])
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(assistant_id="agent") as thread:
+            assert thread._lifecycle_watcher_thread is None
+            thread.run.start(input={})
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = run_done.result(timeout=2.0)
+
+    assert terminal.status == "completed"
+
+
+def _sync_new_thread_scripts() -> list[SyncStreamScript]:
+    return [
+        SyncStreamScript(events=[], status=404),
+        SyncStreamScript(events=[lifecycle_completed_event(seq=1)]),
+    ]
+
+
+def test_sync_run_start_restarts_a_watcher_stopped_by_a_404_for_a_new_thread():
+    fake = SyncFakeServer()
+    fake.script_sequence(_sync_new_thread_scripts())
+    fake.set_state({"ok": True})
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-new", assistant_id="agent") as thread:
+            rejected = thread._run_done
+            assert rejected is not None
+            rejected.result(timeout=2.0)
+            thread.run.start(input={})
+            output = thread.output
+
+    assert output == {"ok": True}
+    assert len(fake.stream_request_bodies) == 2
+
+
+def test_sync_lifecycle_watcher_retries_a_404_that_arrives_after_run_start():
+    fake = SyncFakeServer()
+    fake.script_sequence(_sync_new_thread_scripts())
+    fake.set_state({"ok": True})
+    stream_opened = threading.Event()
+    run_started = threading.Event()
+
+    def handle(request: httpx.Request) -> httpx.Response:
+        if (
+            request.url.path.endswith("/stream/events")
+            and not fake.stream_request_bodies
+        ):
+            stream_opened.set()
+            run_started.wait(timeout=2.0)
+        return fake._handle(request)
+
+    with httpx.Client(
+        transport=httpx.MockTransport(handle), base_url="http://test"
+    ) as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-new", assistant_id="agent") as thread:
+            assert stream_opened.wait(timeout=2.0)
+            thread.run.start(input={})
+            run_started.set()
+            output = thread.output
+
+    assert output == {"ok": True}
+    assert len(fake.stream_request_bodies) == 2
+
+
+def test_sync_lifecycle_watcher_reports_the_error_frame_when_reconnects_run_out():
+    fake = SyncFakeServer()
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        stream = threads.stream(thread_id="t-1", assistant_id="agent")
+        stream._lifecycle_max_reconnect_attempts = 1
+        fake.script_sequence(
+            [
+                SyncStreamScript(events=[thread_stream_failed_frame()], fail_after=1)
+                for _ in range(stream._lifecycle_max_reconnect_attempts + 1)
+            ]
+        )
+        with stream as thread:
+            run_done = thread._run_done
+            assert run_done is not None
+            terminal = run_done.result(timeout=2.0)
+
+    assert (
+        str(terminal.error)
+        == "Protocol error [unknown_error]: Thread event stream failed."
+    )
+
+
+def test_sync_concurrent_run_starts_start_one_lifecycle_watcher(monkeypatch):
+    real_thread = threading.Thread
+    created_watchers: list[str] = []
+
+    class _SlowToCreateThread(real_thread):
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            if kwargs.get("name") == "langgraph-sdk-sync-lifecycle":
+                created_watchers.append(kwargs["name"])
+                time.sleep(0.05)
+            super().__init__(*args, **kwargs)
+
+    monkeypatch.setattr(threading, "Thread", _SlowToCreateThread)
+    fake = SyncFakeServer()
+    fake.script([lifecycle_completed_event(seq=1)])
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(assistant_id="agent") as thread:
+            barrier = threading.Barrier(2)
+
+            def start_run() -> None:
+                barrier.wait()
+                thread.run.start(input={})
+
+            runners = [real_thread(target=start_run) for _ in range(2)]
+            for runner in runners:
+                runner.start()
+            for runner in runners:
+                runner.join()
+
+    assert len(created_watchers) == 1
+
+
+def test_sync_reconnect_stops_on_a_404(monkeypatch):
+    monkeypatch.setattr(_ctrl_mod.time, "sleep", lambda _delay: None)
+    opened: list[dict[str, Any]] = []
+
+    class _NotFoundTransport(SyncProtocolSseTransport):
+        def open_event_stream(self, params: dict) -> SyncEventStreamHandle:
+            opened.append(params)
+            request = httpx.Request("POST", "http://test/threads/t-1/stream/events")
+            raise httpx.HTTPStatusError(
+                "not found",
+                request=request,
+                response=httpx.Response(404, request=request),
+            )
+
+    with httpx.Client(base_url="http://test") as raw:
+        transport = _NotFoundTransport(client=raw, thread_id="t-1")
+        controller = SyncStreamController(transport, max_reconnect_attempts=5)
+        controller._shared_stream_filter = {"channels": ["values"]}
+        assert controller._reconnect_shared_stream() is False
+
+    assert len(opened) == 1
@@ -301,7 +301,7 @@ wheels = [

 [[package]]
 name = "langgraph"
-version = "1.2.15"
+version = "1.2.16"
 source = { editable = "../langgraph" }
 dependencies = [
    { name = "langchain-core" },
@@ -433,7 +433,7 @@ test = [

 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "../prebuilt" }
 dependencies = [
    { name = "langchain-core" },
Author	SHA1	Message	Date
Elior Nataf LackritzandGitHub	41c9c22fdc	release(langgraph): 1.2.16 (#9334 ) Bumps langgraph from 1.2.15 to 1.2.16. Includes: - Keep a DeltaChannel's value when its field changes to a plain reducer (#9302) - Save a resumed exit run's pending writes once (#9314) - Stop a subgraph from rerunning its finished tasks on resume (#9315) - Apply a resumed `Command`'s update once (#9322) - Let other tasks finish when a resume value is invalid (#9265)	2026-10-16 18:02:08 -04:00
Elior Nataf LackritzandGitHub	51018652a6	release(sdk-py): 0.4.8 (#9335 ) Bumps the Python SDK version from 0.4.7 to 0.4.8. Thread streams now report why they ended: a server's error frame becomes the error `thread.output` raises, the sync client no longer blocks forever when the server closes the stream, and a stream the server rejects with a 4xx ends at once instead of being retried (#9317).	2026-10-16 18:01:53 -04:00
Elior Nataf LackritzandGitHub	9e6fc45c6f	fix(sdk-py): surface thread stream errors and stop retrying a rejected stream (#9317 ) ## Summary Thread streams in `langgraph_sdk` now report why they ended. An `id: null` error frame becomes the error `thread.output` raises, the sync client no longer blocks forever when the server closes the stream, and a stream the server rejects with a 4xx ends at once instead of being retried. ## Problem - A server sends `{"type": "error", "id": null, ...}` just before it closes a thread stream whose read failed. The SDK dropped the frame, so `thread.output` raised a generic "lifecycle stream ended before terminal event" (async) or blocked forever (sync: the lifecycle watcher returned without settling the run). - A 403 or 404 on the stream request was retried like a network error, then reported as `RuntimeError("Lifecycle transport failed: ...")` with no status. A server that rejects a WebSocket after accepting it closes with 4000 plus the HTTP status (4403, 4404, 4422), since a browser can't read a refused upgrade's status. The shared stream counted each such reconnect as a success and reconnected forever. ## Fix - The lifecycle watcher fails the run with the frame's error on a clean close (both clients). The shared stream logs the frame instead of dropping it: projections still end when the stream closes, as before, and `thread.output` carries the error. - A 4xx other than 408 and 429, as a status or as a 4000+status close code, ends the stream without retries. `thread.output` now raises that `httpx.HTTPStatusError` or `websockets.exceptions.ConnectionClosedError` instead of the generic `RuntimeError`. - For a minted `thread_id`, the lifecycle watcher starts after `run.start` creates the thread. Without this, the 4xx stop would turn a stream opened under custom auth for a not-yet-created thread from retried into failed. The sync client starts it under a lock, since `run.start` can be called from several threads. - Drops the first of two `SyncStreamController._reconnect_shared_stream` definitions. It was shadowed by the second and never ran. Not in this PR: over WebSocket the server closes with 1011 after an error frame, which is still retried, while over SSE the close is clean and fails the stream. Reconnecting after a clean close, as `langchain-ai/langgraphjs#2812` does for the JS SDK, is a separate change. ## Test plan - [x] The 10 new behavior tests (async and sync: error frame, clean close, 404, 4404 close, reconnect on 404, minted `thread_id`) fail on `main` and pass here; a race test checks that two concurrent `run.start` calls start one watcher. - [x] `make test` in `libs/sdk-py`: 541 passed. `make lint` clean, `check_sdk_methods.py` passes. - [x] `RemoteGraph` v3 tests in `libs/langgraph` (`test_remote_graph_v3.py`, `test_stream_events_v3_kwarg_forwarding.py`): 94 passed against this SDK. - [x] Against a local server with custom auth: a thread the caller can't read fails at once with 404 (SSE) or 4404 (WebSocket), and a new thread works with a 1 s pause before `run.start`, which fails on `main`.	2026-10-16 15:29:34 -04:00
Baihao WangandGitHub	b0cabd93d0	fix(langgraph): let other tasks finish when a resume value is invalid (#9265 ) Found while testing per-call human-in-the-loop (langchain-ai/langchain#41062). The code depends only on #9232, which is already on `main`. When one answer in a multi-answer resume fails `response_schema` validation, the runner cancelled the other tasks in the step, so in async runs a task cut off after its side effect kept its saved answer and ran again on the next resume. Invalid resume values (already tagged by `is_invalid_resume` from #9232) no longer stop the other tasks: the error is raised once they finish, so their results are saved. Hit by per-call human-in-the-loop: approving one tool call while giving another an invalid edit, in one resume, sent the approved call's email twice once the second answer was fixed. ## How to verify - `test_invalid_resume_lets_other_answered_tasks_finish` fails on `main` (the approved task is still pending after its side effect) and passes with this change. Like the other async interrupt tests, it is skipped before Python 3.11, where async nodes can't call `interrupt()`. - The per-call HITL reproduction against this branch: the approved async tool runs once (twice on `main` whenever it awaits after its side effect). - `make lint` passes. With `NO_DOCKER=true`, the full `libs/langgraph` suite has the same failures as `main` locally (`test_prebuilt_tool_chat` and `test_stream_mode_messages_command`, sync and async), and nothing new.	2026-10-16 11:33:53 -07:00
raghavverma-langchainandGitHub	4ac46ba186	release(cli): 0.4.36 (#9330 ) ## Summary <!-- Write the summary yourself --> ## Notes for reviewers - Includes [push through the active Docker context](https://github.com/langchain-ai/langgraph/pull/9321) and [accept serverless and dedicated deployment aliases](https://github.com/langchain-ai/langgraph/pull/9325), the two CLI changes merged since 0.4.35. - Merging doesn't publish by itself: after it merges, run the `release` workflow with `working-directory: libs/cli`, as was done for 0.4.35. ## Test Plan - [x] Only `__version__` in `libs/cli/langgraph_cli/__init__.py` changes (0.4.35 → 0.4.36), like the 0.4.35 release PR - [ ] After merge, the `release` workflow for `libs/cli` succeeds and PyPI shows `langgraph-cli` 0.4.36 - [ ] `uvx --from langgraph-cli==0.4.36 langgraph deploy` pushes on Docker Desktop without `DOCKER_HOST` set ## Demo <!-- Drag a screen recording or screenshot here -->	2026-10-16 12:29:42 -04:00
Elior Nataf LackritzandGitHub	e14f4277dc	fix(langgraph): apply a resumed Command's update once (#9322 ) `Command(resume=..., update=...)` applied its update twice once the resumed node called `interrupt()` again, with sync or async durability (`invoke`'s default). A node that asks three times, resumed with `Command(resume=str(i), update={"log": [f"u{i}"]})`, ends with `["u1", "u1", "u2", "u3", "a:123"]`. ## Cause The null task's writes are saved as one list, and savers key a regular write by its position in that list. The Command's resume came first, so its update was stored at index 1. The next Command saved the null task's earlier writes again without the resume, which put the update at index 0, so it was stored twice and every later resume applied both copies. `durability="exit"` saves after the resume is consumed, so it wasn't affected. ## Fix The null task's special writes (a resume) now go after its regular ones, so each regular write keeps its index as the list grows. `test_resume_with_update_applies_the_update_once` fails on `main` for sync and async durability. Found by @hntrl, and flagged by @soarez on #9314.	2026-10-16 12:07:59 -04:00
hari-dhanushkodiandGitHub	4e2dfe4a24	fix(cli): accept serverless and dedicated deployment aliases (#9325 )	2026-10-16 08:02:43 -04:00
raghavverma-langchainandGitHub	e77a5cb08c	fix(cli): push through the active Docker context in langgraph deploy (#9321 ) ## Summary <!-- Write the summary yourself --> ## Notes for reviewers - Known gap: `--host` carries only the context's endpoint, not its TLS settings, so a remote context that uses client certificates still can't push. That case was broken before this change too, and it's rare for local builds. - A context with no Docker endpoint (Docker prints `<no value>`) or a missing `docker` binary falls back to today's behavior. ## Test Plan - [x] `make lint` and `make test` pass in `libs/cli` (556 tests) - [x] New tests: `login` and `push` get the active context's endpoint as `--host`; with `DOCKER_HOST` set there's no lookup and no `--host`; when the context lookup fails, or the context has no Docker endpoint, there's no `--host` and the deploy still succeeds. The first fails without the change. - [x] On a Mac running Docker Desktop with no `/var/run/docker.sock`: `docker --config <temp dir> version` fails with `failed to connect to the docker API at unix:///var/run/docker.sock`, and adding `--host <active context endpoint>` reaches the daemon - [x] Reproduced end to end on LangSmith AWS: `langgraph deploy` (0.4.35) failed at the push with that error, then passed with `DOCKER_HOST` set to the context's endpoint, which this change does automatically - [ ] After release, `langgraph deploy` pushes on Docker Desktop without `DOCKER_HOST` set ## Demo <!-- Drag a screen recording or screenshot here -->	2026-10-15 16:14:01 -04:00
Elior Nataf Lackritz GitHub Connor Imes	b0482d0627	fix(langgraph): stop a subgraph from rerunning its finished tasks on resume (#9315 ) Fixes #9106 When a subgraph pauses in several parallel interrupts and they're answered one at a time, each resume ran the nodes that were already answered again. Their saved answers carried them straight through `interrupt()`, so whatever they do after it (an API write, say) ran once more per later answer, while the final state still looked right. A `@task` that finished before an interrupt in a nested graph reran the same way, which is the rerun reported in #6792. ## Cause Every task config sets `checkpoint_id` to None, and `PregelLoop` treats a config with that key as a replay, so a nested graph replayed on every resume. A replaying loop doesn't hand saved writes back to the tasks that made them, so finished tasks ran again. ## Fix `PregelLoop` now notes whether the run points it at a checkpoint (a truthy `checkpoint_id`, its own `checkpoint_map` entry, or a parent that is replaying), the test `__enter__` already uses to pick the checkpoint to load. `_first` then decides once whether finished tasks rerun: only on time travel, or on a resume at a checkpoint the thread has moved past. A plain resume of a subgraph is neither, so it hands finished tasks their writes back the way a top-level resume does. The same rule fixes the top level, which @hntrl pointed out: a resume sent with the head's `checkpoint_id`, as Studio sends it, or with `checkpoint_id: None` reran finished siblings too. A resume at an older checkpoint still reruns them, as the checkpoint migration tests expect. Telling the two apart costs one more checkpoint read for a run sent with a `checkpoint_id`. `test_addressed_resume_keeps_a_rerun_tasks_write_to_a_new_channel` relied on a resume at the head rerunning a task, so it now resumes a checkpoint the thread has moved past. Most subgraph time travel forks first, which gives tasks new ids, so saved writes can't match them anyway. `test_subgraph_replay_reasks_answered_parallel_interrupt` replays an update checkpoint, which runs in place: it passes on `main` and fails if subgraphs never replay. The counters added to `test_parallel_interrupts_double` and the three new tests in `test_interrupt_state.py` fail on `main`. ## Not in this PR - With `durability="exit"` and three or more parallel interrupts, a subgraph now hits a bug top-level graphs already have: a finished task's output is applied twice. #9314 fixes it. - #6792 also reports the interrupt of a `@task` inside a nested `@entrypoint` coming back twice. That's a separate bug. Thanks @connorimes for the report, the root cause and the fix, which this applies as proposed. The same change was opened as #9113 by @AnvitDevadiga, and @morehardy found the root cause for #6792 in #6793. JS: langchain-ai/langgraphjs#3028 Co-authored-by: Connor Imes <5701850+connorimes@users.noreply.github.com>	2026-10-15 15:51:21 -04:00
Elior Nataf LackritzandGitHub	889a5ff16f	fix(langgraph): save a resumed exit run's pending writes once (#9314 ) With `durability="exit"`, resuming three or more parallel interrupts one at a time applied the earlier answers twice. Nodes `a`, `b` and `c` each return `{"log": [f"{name}:{interrupt(name)}"]}`, and the run ends with `{"log": ["a:yes", "a:yes", "b:yes", "c:yes"]}`. ## Cause An exit-mode run saves its pending writes when it ends, and that save took all of them, including the ones loaded with the checkpoint. Savers key a task's writes by their position in the list they're given. A task that finished in an earlier run comes back with its old interrupt next to its output, so saving its list again put the output under a new index, and the next load applied both copies. ## Fix The save at exit keeps only the writes of tasks that wrote during this run. The rest are already stored on the checkpoint, which is what sync and async durability do. `test_finished_task_output_applies_once` fails on `main` in exit mode for every saver and passes for sync and async. The #9106 fix needs this too: once a subgraph stops rerunning its finished tasks, it hits this bug in exit mode. JS: langchain-ai/langgraphjs#3027	2026-10-15 15:26:01 -04:00
Baihao WangandGitHub	5eb60aef6f	release(prebuilt): 1.1.1 (#9316 ) Bumps langgraph-prebuilt from 1.1.0 to 1.1.1. Includes: - Stop `ToolNode` from swallowing a resume value that fails an interrupt's `response_schema` (#9232). This needs langgraph 1.2.15 or later, which marks those errors; with an older langgraph, `ToolNode` keeps today's behavior. - Keep tool results out of v3 messages (#7838) Managed Deep Agents (langchain-ai/managed-deepagents-sdk#710) will raise its `langgraph-prebuilt` minimum to this release, so an invalid answer to an interrupt raised inside a tool, such as a subagent, fails the resume instead of reaching the model. After merge, run the Release workflow with `working-directory=libs/prebuilt`. --- This PR was written with help from an AI coding agent (Claude Code); a human reviewed and directed the changes. 🤖 Generated with [Claude Code](https://claude.com/claude-code)	2026-10-14 18:31:09 -07:00