Prevent --json from prompting interactively on a 403

Per review: --json still went through _create_host_backend_client
and _call_host_backend_with_optional_tenant, both of which prompt
interactively (writing to stdout, reading stdin) when the module-
level _no_input flag is left at its default False. deploy_listeners_
list never set it, unlike the main deploy command, so a missing API
key or an org-scoped key without LANGSMITH_TENANT_ID would corrupt
the JSON output with warning text and could hang waiting for input
that a scripted invocation never provides.

Sets _no_input = json_output at the top of the command, same pattern
_deploy_cmd already uses, so both paths raise a ClickException to
stderr instead. Added a regression test that fails without the fix
(click.prompt is genuinely invoked) and passes with it.

libs/cli/langgraph_cli/deploy.py

@@ -2518,6 +2518,13 @@ def deploy_listeners() -> None:
 
 @OPT_HOST_API_KEY
 @OPT_HOST_URL
+@click.option(
+    "--json",
+    "json_output",
+    is_flag=True,
+    default=False,
+    help="Emit the listeners as a JSON array instead of a table.",
+)
 @deploy_listeners.command(
     "list",
     help=(
@@ -2526,11 +2533,18 @@ def deploy_listeners() -> None:
         "--listener-id <id>` to deploy through it."
     ),
 )
-def deploy_listeners_list(api_key: str | None, host_url: str | None) -> None:
+def deploy_listeners_list(
+    api_key: str | None, host_url: str | None, json_output: bool
+) -> None:
+    global _no_input
+    _no_input = json_output
     client = _create_host_backend_client(host_url, api_key)
     listeners = _call_host_backend_with_optional_tenant(
         client, lambda c: c.list_listeners()
     )
+    if json_output:
+        click.echo(json_mod.dumps(listeners))
+        return
     if not listeners:
         click.echo("No listeners found for this workspace.")
         return

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

@@ -14,6 +14,7 @@ import langgraph_cli.deploy as deploy_module
 from langgraph_cli.cli import cli, prepare_args_and_stdin
 from langgraph_cli.config import Config, _get_pip_cleanup_lines, validate_config
 from langgraph_cli.docker import DEFAULT_POSTGRES_URI, DockerCapabilities, Version
+from langgraph_cli.host_backend import HostBackendError
 from langgraph_cli.util import clean_empty_lines
 
 FORMATTED_CLEANUP_LINES = _get_pip_cleanup_lines(
@@ -535,6 +536,131 @@ def test_deploy_listeners_list_command_no_results(monkeypatch) -> None:
     assert result.output.strip() == "No listeners found for this workspace."
 
 
+def test_deploy_listeners_list_command_json(monkeypatch) -> None:
+    # --json sets the module-level _no_input global as a side effect; pin
+    # the "before" value so monkeypatch restores it after this test.
+    monkeypatch.setattr(deploy_module, "_no_input", False)
+    runner = CliRunner()
+
+    listeners = [
+        {
+            "id": "listener-1",
+            "compute_id": "prod-cluster",
+            "compute_config": {"k8s_namespaces": ["agents"]},
+        },
+        {
+            "id": "listener-2",
+            "compute_id": "multi-cluster",
+            "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
+        },
+    ]
+
+    class FakeClient:
+        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
+            pass
+
+        def list_listeners(self):
+            return listeners
+
+    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
+
+    result = runner.invoke(
+        cli,
+        [
+            "deploy",
+            "listeners",
+            "list",
+            "--json",
+            "--api-key",
+            "test-key",
+            "--host-url",
+            "https://api.example.com",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert json.loads(result.output) == listeners
+
+
+def test_deploy_listeners_list_command_json_no_results(monkeypatch) -> None:
+    # --json sets the module-level _no_input global as a side effect; pin
+    # the "before" value so monkeypatch restores it after this test.
+    monkeypatch.setattr(deploy_module, "_no_input", False)
+    runner = CliRunner()
+
+    class FakeClient:
+        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
+            pass
+
+        def list_listeners(self):
+            return []
+
+    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
+
+    result = runner.invoke(
+        cli,
+        [
+            "deploy",
+            "listeners",
+            "list",
+            "--json",
+            "--api-key",
+            "test-key",
+            "--host-url",
+            "https://api.example.com",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert json.loads(result.output) == []
+
+
+def test_deploy_listeners_list_command_json_org_scoped_403(monkeypatch) -> None:
+    """With --json, an org-scoped 403 must raise cleanly instead of
+    prompting for a workspace ID, which would write to stdout and hang
+    waiting for input that a scripted invocation never provides."""
+    # deploy_listeners_list sets the module-level _no_input global as a
+    # side effect; pin its "before" value so monkeypatch restores it after
+    # this test instead of leaking True into whichever test runs next.
+    monkeypatch.setattr(deploy_module, "_no_input", False)
+    runner = CliRunner()
+
+    class FakeClient:
+        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
+            pass
+
+        def list_listeners(self):
+            raise HostBackendError("requires workspace specification", status_code=403)
+
+        def set_tenant(self, tenant_id: str) -> None:
+            raise AssertionError("must not retry with a prompted tenant with --json")
+
+    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
+
+    prompt_calls = []
+    monkeypatch.setattr(
+        click, "prompt", lambda *a, **k: prompt_calls.append((a, k)) or "unreachable"
+    )
+
+    result = runner.invoke(
+        cli,
+        [
+            "deploy",
+            "listeners",
+            "list",
+            "--json",
+            "--api-key",
+            "test-key",
+            "--host-url",
+            "https://api.example.com",
+        ],
+    )
+
+    assert prompt_calls == [], "must not prompt interactively with --json"
+    assert result.exit_code != 0
+    assert "LANGSMITH_TENANT_ID" in result.output
+
+
 def test_deploy_revisions_list_command(monkeypatch) -> None:
     runner = CliRunner()
     captured: dict[str, str] = {}

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -26,6 +26,7 @@ from langgraph._internal._constants import (
     CONFIG_KEY_CHECKPOINT_ID,
     NS_END,
     NS_SEP,
+    NULL_TASK_ID,
     PUSH,
     SNAPSHOT_BUMPS,
 )
@@ -64,10 +65,14 @@ def exit_delta_task_id(step: int, task_id: str) -> str:
 
     Embeds the superstep in the first UUID group so `ORDER BY task_id, idx`
     preserves chronological order while remaining a valid RFC UUID (required by
-    Postgres `checkpoint_writes.task_id uuid` columns).
+    Postgres `checkpoint_writes.task_id uuid` columns). Never `NULL_TASK_ID`:
+    readers apply writes under it as the anchor checkpoint's own pending writes.
     """
     parts = str(uuid.UUID(task_id)).split("-")
-    return f"{step:08d}-{parts[1]}-{parts[2]}-{parts[3]}-{parts[4]}"
+    synthetic = f"{step:08d}-{parts[1]}-{parts[2]}-{parts[3]}-{parts[4]}"
+    if synthetic == NULL_TASK_ID:
+        return f"{step:08d}-0000-0000-0000-000000000001"
+    return synthetic
 
 
 def exit_delta_late_task_id(step: int, task_id: str) -> str:

libs/langgraph/langgraph/pregel/main.py

@@ -3714,16 +3714,15 @@ class Pregel(
         config: RunnableConfig | None = None,
         *,
         version: Literal["v1", "v2", "v3"] = "v2",
-        interrupt_before: All | Sequence[str] | None = None,
-        interrupt_after: All | Sequence[str] | None = None,
-        control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
         **kwargs: Any,
     ) -> Any:
         """Stream events from this graph.
 
-        For `version="v1"` / `"v2"`, yields `StreamEvent` dicts (see
-        `Runnable.stream_events`). For `version="v3"`, returns a
+        For `version="v1"` / `"v2"`, delegates to
+        `Runnable.(a)stream_events`; synchronous v1/v2 event streaming is
+        not implemented in langchain-core, so use `astream_events` for
+        those versions. For `version="v3"`, returns a
         `GraphRunStream` whose typed projections the caller drives by
         iterating — no background thread.
 
@@ -3753,20 +3752,27 @@ class Pregel(
             config: Optional runnable config.
             version: Streaming-event schema version. `"v3"` selects the
                 content-block-centric streaming protocol.
-            interrupt_before: Nodes to interrupt before, if any. Only
-                used for `version="v3"`.
-            interrupt_after: Nodes to interrupt after, if any. Only
-                used for `version="v3"`.
+            interrupt_before: Nodes to interrupt before, if any.
+                Honored on every version that can run; type-checked
+                only on the `version="v3"` overloads.
+            interrupt_after: Nodes to interrupt after, if any. Honored
+                on every version that can run; type-checked only on
+                the `version="v3"` overloads.
             control: Optional run control used to request cooperative
-                drain. Only used for `version="v3"`.
+                drain. Honored on every version that can run;
+                type-checked only on the `version="v3"` overloads.
             transformers: Extra transformer classes or configured
                 factories appended after compile-time
                 `stream_transformers`. Factories are called as
                 `factory(scope)` so they can propagate to subgraph
                 scopes. Only used for `version="v3"`.
-            **kwargs: For `version="v1"`/`"v2"`, forwarded to
-                `Runnable.stream_events`. For `version="v3"`, forwarded
-                to the underlying `stream(...)` call (e.g. `context`,
+            **kwargs: For `version="v1"`/`"v2"` on `astream_events`,
+                forwarded to `Runnable.astream_events`, which passes
+                them through to `astream` — so execution kwargs such as
+                `context`, `durability`, `interrupt_before`,
+                `interrupt_after` and `control` are honored on every
+                version that can run. For `version="v3"`, forwarded to the
+                underlying `stream(...)` call (e.g. `context`,
                 `durability`, `output_keys`, `print_mode`, `debug`).
                 `stream_mode` and `subgraphs` are not accepted under
                 `version="v3"` and raise `TypeError` if supplied; v3
@@ -3774,16 +3780,15 @@ class Pregel(
 
         Returns:
             For `version="v3"`, a `GraphRunStream` the caller iterates
-            to drive the run. Otherwise an `Iterator[StreamEvent]`.
+            to drive the run. For `version="v1"`/`"v2"`,
+            `astream_events` yields `StreamEvent` dicts; the synchronous
+            v1/v2 path is not implemented in langchain-core.
         """
         if version == "v3":
             _reject_v3_invariant_kwargs(kwargs)
             return self._pregel_stream_v3(
                 input,
                 config,
-                interrupt_before=interrupt_before,
-                interrupt_after=interrupt_after,
-                control=control,
                 transformers=transformers,
                 **kwargs,
             )
@@ -3819,9 +3824,6 @@ class Pregel(
         config: RunnableConfig | None = None,
         *,
         version: Literal["v1", "v2", "v3"] = "v2",
-        interrupt_before: All | Sequence[str] | None = None,
-        interrupt_after: All | Sequence[str] | None = None,
-        control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
         **kwargs: Any,
     ) -> AsyncIterator[StreamEvent] | Awaitable[Any]:
@@ -3844,9 +3846,6 @@ class Pregel(
             return self._apregel_stream_v3(
                 input,
                 config,
-                interrupt_before=interrupt_before,
-                interrupt_after=interrupt_after,
-                control=control,
                 transformers=transformers,
                 **kwargs,
             )

libs/langgraph/tests/test_delta_channel_exit_mode.py

@@ -17,6 +17,7 @@ from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import TypedDict
 
+from langgraph._internal._constants import NULL_TASK_ID
 from langgraph.channels.delta import DeltaChannel
 from langgraph.graph import START, StateGraph
 from langgraph.graph.message import _messages_delta_reducer
@@ -38,6 +39,7 @@ def test_exit_delta_task_id_is_valid_uuid_and_ordered() -> None:
     assert id1.split("-")[0] == "00000001"
     assert id7.split("-")[0] == "00000007"
     assert id1.endswith("-0270-bf16-1ef8-fb321bef9f3d")
+    assert exit_delta_task_id(0, NULL_TASK_ID) != NULL_TASK_ID
 
     with pytest.raises(ValueError):
         uuid.UUID(f"00000001-{tid}")
@@ -463,6 +465,43 @@ def test_resume_with_a_command_update_replays_its_write_once(
     assert state.values["log"] == state.values["plain"] == ["in", "cmd", "ask", "done"]
 
 
+def test_command_update_on_an_input_checkpoint_matches_a_plain_channel(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    builder = StateGraph(_ResumeState)
+    builder.add_node("node", lambda state: _both("node"))
+    builder.add_edge(START, "node")
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = {"configurable": {"thread_id": "t"}}
+    graph.update_state(config, _both("in"), as_node="__input__")
+
+    graph.invoke(Command(update=_both("cmd")), config, durability=durability)
+
+    history = list(graph.get_state_history(config))
+    assert [s.values.get("log", []) for s in history] == [
+        s.values.get("plain", []) for s in history
+    ]
+    replayed = graph.invoke(None, history[-1].config, durability=durability)
+    assert replayed["log"] == replayed["plain"]
+
+
+def test_exit_command_update_on_a_new_thread_matches_a_plain_channel(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    builder = StateGraph(_ResumeState)
+    builder.add_node("node", lambda state: _both("node"))
+    builder.add_edge(START, "node")
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = {"configurable": {"thread_id": "t"}}
+
+    graph.invoke(Command(update=_both("cmd")), config, durability="exit")
+
+    history = list(graph.get_state_history(config))
+    assert [s.values.get("log", []) for s in history] == [
+        s.values.get("plain", []) for s in history
+    ]
+
+
 class _FlagState(_ResumeState, total=False):
     extra: Annotated[list, DeltaChannel(_append)]
     flag: bool

libs/langgraph/tests/test_stream_events_v3_kwarg_forwarding.py

@@ -8,20 +8,33 @@ caller kwargs to the inner ``(a)stream`` call but rejects ``stream_mode`` and
 ``subgraphs`` since v3 owns them (``stream_mode`` is built from the
 transformer mux; ``subgraphs`` is forced True so nested namespaces flow
 through scoped muxes).
+
+A second regression is pinned here: #7677 (first released in 1.2.0a3)
+declared `interrupt_before` / `interrupt_after` / `control` as named
+parameters on the `Pregel.stream_events` / `astream_events` dispatchers
+but forwarded them only on the v3 branch, silently dropping them on v1/v2
+(where they had reached `(a)stream` through `**kwargs` before).
+`TestAstreamKwargsForwardedOnEveryVersion` and friends pin that they
+reach `(a)stream` on every version, with the v1/v2 passthrough semantics
+restored (exactly what the caller passed, including an explicit `None`)
+and v3's explicit-default semantics preserved.
 """
 
 from __future__ import annotations
 
 import sys
+from collections.abc import AsyncIterator, Callable
 from dataclasses import dataclass
 from typing import Any
 
 import pytest
+from langgraph.checkpoint.memory import InMemorySaver
 from typing_extensions import TypedDict
 
 from langgraph.constants import END, START
+from langgraph.errors import GraphDrained
 from langgraph.graph import StateGraph
-from langgraph.runtime import Runtime
+from langgraph.runtime import RunControl, Runtime
 
 NEEDS_CONTEXTVARS = pytest.mark.skipif(
     sys.version_info < (3, 11),
@@ -106,3 +119,274 @@ class TestKwargForwardingAsync:
                 version="v3",
                 subgraphs=False,
             )
+
+
+_KWARG_NAMES = ("control", "interrupt_before", "interrupt_after")
+
+
+def _build_two_step_graph(
+    first: Callable[[_State], dict[str, Any]] | None = None,
+) -> Any:
+    """A `first -> second` graph with a checkpointer, for interrupt/drain tests."""
+
+    def default_first(state: _State) -> dict[str, Any]:
+        return {"message": state["message"] + " first"}
+
+    def second(state: _State) -> dict[str, Any]:
+        return {"message": state["message"] + " second"}
+
+    builder = StateGraph(_State)
+    builder.add_node("first", first or default_first)
+    builder.add_node("second", second)
+    builder.add_edge(START, "first")
+    builder.add_edge("first", "second")
+    builder.add_edge("second", END)
+    return builder.compile(checkpointer=InMemorySaver())
+
+
+async def _drive_astream_events(
+    graph: Any, config: dict[str, Any], version: str, **kwargs: Any
+) -> None:
+    """Consume an astream_events run for `version` to completion."""
+    if version == "v3":
+        run = await graph.astream_events(
+            {"message": "hi"}, config, version="v3", **kwargs
+        )
+        await run.output()
+    else:
+        async for _ in graph.astream_events(
+            {"message": "hi"}, config, version=version, **kwargs
+        ):
+            pass
+
+
+@pytest.mark.anyio
+@pytest.mark.filterwarnings("ignore:astream_events version='v1' is deprecated")
+@pytest.mark.parametrize("version", ["v1", "v2", "v3"])
+class TestAstreamKwargsForwardedOnEveryVersion:
+    """`interrupt_before`/`interrupt_after`/`control` reach `astream` on every
+    version.
+
+    Regression test for #7677 (first released in 1.2.0a3): the dispatchers
+    captured these parameters as named arguments but forwarded them only on
+    the v3 branch, silently dropping them on v1/v2.
+    """
+
+    async def test_interrupt_before(self, version: str) -> None:
+        graph = _build_two_step_graph()
+        config = {"configurable": {"thread_id": "ib"}}
+        await _drive_astream_events(graph, config, version, interrupt_before=["second"])
+        state = await graph.aget_state(config)
+        assert state.next == ("second",)
+        assert state.values == {"message": "hi first"}
+
+    async def test_interrupt_after(self, version: str) -> None:
+        graph = _build_two_step_graph()
+        config = {"configurable": {"thread_id": "ia"}}
+        await _drive_astream_events(graph, config, version, interrupt_after=["first"])
+        state = await graph.aget_state(config)
+        assert state.next == ("second",)
+        assert state.values == {"message": "hi first"}
+
+    async def test_pre_drained_control(self, version: str) -> None:
+        graph = _build_two_step_graph()
+        config = {"configurable": {"thread_id": "drain"}}
+        control = RunControl()
+        control.request_drain("sigterm")
+        with pytest.raises(GraphDrained, match="sigterm"):
+            await _drive_astream_events(graph, config, version, control=control)
+
+
+class TestStreamEventsV3SyncInterrupts:
+    """Sync v3 static interrupts reach `stream()` after the kwargs rewire."""
+
+    @pytest.mark.parametrize(
+        ("kwarg", "node"),
+        [("interrupt_before", "second"), ("interrupt_after", "first")],
+    )
+    def test_static_interrupt(self, kwarg: str, node: str) -> None:
+        graph = _build_two_step_graph()
+        config = {"configurable": {"thread_id": "sync"}}
+        run = graph.stream_events(
+            {"message": "hi"}, config, version="v3", **{kwarg: [node]}
+        )
+        list(run.values)
+        state = graph.get_state(config)
+        assert state.next == ("second",)
+        assert state.values == {"message": "hi first"}
+
+
+@pytest.mark.anyio
+@pytest.mark.filterwarnings("ignore:astream_events version='v1' is deprecated")
+@pytest.mark.parametrize("version", ["v1", "v2", "v3"])
+class TestAstreamMidRunDrain:
+    """A drain requested from inside a node propagates out of `astream_events`.
+
+    This is the graceful-shutdown scenario: `request_drain()` called while
+    the run is in flight (e.g. from a signal handler), with v1/v2 running
+    inside core's event-stream task. The caller's own `RunControl` is used
+    (the drain reason proves identity), `GraphDrained` propagates to the
+    consumer, and the checkpoint keeps the pending step.
+    """
+
+    async def test_drain_requested_inside_first_node(self, version: str) -> None:
+        control = RunControl()
+
+        def first(state: _State) -> dict[str, Any]:
+            control.request_drain("sigterm-mid")
+            return {"message": state["message"] + " first"}
+
+        graph = _build_two_step_graph(first)
+        config = {"configurable": {"thread_id": "midrun"}}
+
+        with pytest.raises(GraphDrained, match="sigterm-mid"):
+            await _drive_astream_events(graph, config, version, control=control)
+        state = await graph.aget_state(config)
+        assert state.next == ("second",)
+        assert state.values == {"message": "hi first"}
+
+
+def _record_astream_kwargs(graph: Any) -> list[dict[str, Any]]:
+    """Patch `graph.astream` to record the kwargs each call receives."""
+    received: list[dict[str, Any]] = []
+    original = graph.astream
+
+    async def recording_astream(
+        input: Any, config: Any = None, **kwargs: Any
+    ) -> AsyncIterator[Any]:
+        received.append(kwargs)
+        async for chunk in original(input, config, **kwargs):
+            yield chunk
+
+    graph.astream = recording_astream  # type: ignore[method-assign]
+    return received
+
+
+@pytest.mark.anyio
+@pytest.mark.filterwarnings("ignore:astream_events version='v1' is deprecated")
+@pytest.mark.parametrize("version", ["v1", "v2"])
+class TestAstreamV1V2KwargsPassthrough:
+    """v1/v2 forward to `astream` exactly what the caller passed.
+
+    Pre-#7677 semantics: an explicit `None` is forwarded as `None`, and an
+    omitted argument is not forwarded at all (so an override's own default
+    would apply).
+    """
+
+    @pytest.mark.parametrize("name", _KWARG_NAMES)
+    async def test_passed_values_reach_astream(self, version: str, name: str) -> None:
+        graph = _build_two_step_graph()
+        received = _record_astream_kwargs(graph)
+        value: Any = RunControl() if name == "control" else ["second"]
+        await _drive_astream_events(
+            graph, {"configurable": {"thread_id": "rec"}}, version, **{name: value}
+        )
+        assert len(received) == 1
+        assert received[0][name] == value
+        if name == "control":
+            assert received[0][name] is value
+
+    @pytest.mark.parametrize("name", _KWARG_NAMES)
+    async def test_explicit_none_is_forwarded(self, version: str, name: str) -> None:
+        graph = _build_two_step_graph()
+        received = _record_astream_kwargs(graph)
+        await _drive_astream_events(
+            graph, {"configurable": {"thread_id": "rec-none"}}, version, **{name: None}
+        )
+        assert len(received) == 1
+        assert received[0][name] is None
+
+    @pytest.mark.parametrize("name", _KWARG_NAMES)
+    async def test_omitted_values_are_absent(self, version: str, name: str) -> None:
+        graph = _build_two_step_graph()
+        received = _record_astream_kwargs(graph)
+        await _drive_astream_events(
+            graph, {"configurable": {"thread_id": "rec-omit"}}, version
+        )
+        assert len(received) == 1
+        assert name not in received[0]
+
+
+@pytest.mark.anyio
+class TestAstreamV3KwargsDefaults:
+    """v3 keeps its since-inception explicit-default semantics (#7519).
+
+    Omitted `interrupt_before`/`interrupt_after`/`control` are supplied to
+    `astream` as `None`; passed values are forwarded as-is.
+    """
+
+    async def test_omitted_values_arrive_as_none(self) -> None:
+        graph = _build_two_step_graph()
+        received = _record_astream_kwargs(graph)
+        await _drive_astream_events(
+            graph, {"configurable": {"thread_id": "v3-rec"}}, "v3"
+        )
+        assert len(received) == 1
+        assert received[0]["control"] is None
+        assert received[0]["interrupt_before"] is None
+        assert received[0]["interrupt_after"] is None
+
+    @pytest.mark.parametrize("name", _KWARG_NAMES)
+    async def test_passed_values_reach_astream(self, name: str) -> None:
+        graph = _build_two_step_graph()
+        received = _record_astream_kwargs(graph)
+        value: Any = RunControl() if name == "control" else ["second"]
+        await _drive_astream_events(
+            graph, {"configurable": {"thread_id": "v3-rec-2"}}, "v3", **{name: value}
+        )
+        assert len(received) == 1
+        assert received[0][name] == value
+        if name == "control":
+            assert received[0][name] is value
+
+
+def _record_stream_kwargs(graph: Any) -> list[dict[str, Any]]:
+    """Patch `graph.stream` to record the kwargs each call receives."""
+    received: list[dict[str, Any]] = []
+    original = graph.stream
+
+    def recording_stream(input: Any, config: Any = None, **kwargs: Any) -> Any:
+        received.append(kwargs)
+        yield from original(input, config, **kwargs)
+
+    graph.stream = recording_stream  # type: ignore[method-assign]
+    return received
+
+
+def _drive_stream_events_v3(graph: Any, config: dict[str, Any], **kwargs: Any) -> None:
+    """Consume a sync v3 stream_events run to completion."""
+    run = graph.stream_events({"message": "hi"}, config, version="v3", **kwargs)
+    list(run.values)
+
+
+class TestStreamEventsV3SyncKwargsDefaults:
+    """Sync v3 keeps its since-inception explicit-default semantics (#7519).
+
+    Mirror of `TestAstreamV3KwargsDefaults`: omitted
+    `interrupt_before`/`interrupt_after`/`control` are supplied to `stream`
+    as `None`; passed values are forwarded as-is. Pins the sync helper
+    against a kwargs-only "simplification" that would change subclass
+    default handling.
+    """
+
+    def test_omitted_values_arrive_as_none(self) -> None:
+        graph = _build_two_step_graph()
+        received = _record_stream_kwargs(graph)
+        _drive_stream_events_v3(graph, {"configurable": {"thread_id": "s-rec"}})
+        assert len(received) == 1
+        assert received[0]["control"] is None
+        assert received[0]["interrupt_before"] is None
+        assert received[0]["interrupt_after"] is None
+
+    @pytest.mark.parametrize("name", _KWARG_NAMES)
+    def test_passed_values_reach_stream(self, name: str) -> None:
+        graph = _build_two_step_graph()
+        received = _record_stream_kwargs(graph)
+        value: Any = RunControl() if name == "control" else ["second"]
+        _drive_stream_events_v3(
+            graph, {"configurable": {"thread_id": "s-rec-2"}}, **{name: value}
+        )
+        assert len(received) == 1
+        assert received[0][name] == value
+        if name == "control":
+            assert received[0][name] is value

libs/sdk-py/langgraph_sdk/_async/stream.py

@@ -1972,7 +1972,7 @@ class AsyncThreadStream:
                 # Mark that we have observed an active run so thread.output
                 # knows a run exists (handles reattach without run.start).
                 self._run_seen = True
-            elif phase in ("completed", "failed"):
+            elif _is_root_terminal_lifecycle(event):
                 # Why: interrupts describe current-run state. Clear on terminal
                 # lifecycle so a subsequent run.respond() can't fire against a
                 # stale prior-run interrupt_id. Acquire `_interrupts_lock` so

libs/sdk-py/langgraph_sdk/_sync/stream.py

@@ -35,7 +35,11 @@ from langgraph_sdk.stream.decoders import (
     validate_interleave_channels,
 )
 from langgraph_sdk.stream.subscription import compute_union_filter, infer_channel
-from langgraph_sdk.stream.sync_controller import SyncStreamController, _SyncSubscription
+from langgraph_sdk.stream.sync_controller import (
+    SyncStreamController,
+    _is_root_terminal_lifecycle,
+    _SyncSubscription,
+)
 from langgraph_sdk.stream.transport import (
     SyncEventStreamHandle,
     SyncProtocolSseTransport,
@@ -1614,7 +1618,7 @@ class SyncThreadStream:
             phase = data.get("event") if isinstance(data, dict) else None
             if phase in ("started", "running"):
                 self._run_seen = True
-            elif phase in ("completed", "failed"):
+            elif _is_root_terminal_lifecycle(event):
                 self.interrupted = False
                 self.interrupts = []
                 run_done = self._run_done

libs/sdk-py/tests/streaming/test_lifecycle_watcher.py

@@ -15,6 +15,7 @@ from langgraph_sdk.stream.transport import EventStreamHandle, ProtocolSseTranspo
 from streaming._events import (
     input_requested_event,
     lifecycle_completed_event,
+    lifecycle_errored_event,
     lifecycle_event,
 )
 from streaming._fake_server import FakeServer, _StreamScript
@@ -115,6 +116,25 @@ async def test_terminal_lifecycle_clears_interrupts():
     assert thread.interrupts == []
 
 
+async def test_subgraph_completed_event_does_not_end_run():
+    fake = FakeServer()
+    fake.script(
+        [
+            lifecycle_completed_event(seq=0, namespace=["child:1"]),
+            lifecycle_errored_event(seq=1, error="root failed"),
+        ]
+    )
+    asgi = httpx.ASGITransport(app=fake.app)
+    async with httpx.AsyncClient(transport=asgi, 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", "a subgraph's completed event ended the run"
+    assert "root failed" in str(terminal.error)
+
+
 async def test_lifecycle_error_captured_for_output():
     """Lifecycle error terminal state is captured in _run_done with error set."""
     fake = FakeServer()

libs/sdk-py/tests/streaming/test_sync_thread_stream.py

@@ -27,6 +27,7 @@ from streaming._events import (
     checkpoints_event,
     custom_event,
     lifecycle_completed_event,
+    lifecycle_errored_event,
     lifecycle_event,
     lifecycle_started_event,
     message_finish_event,
@@ -475,6 +476,23 @@ def test_sync_lifecycle_watcher_reconnects_with_since_after_transport_drop():
     assert fake.stream_request_bodies[1]["since"] == 1
 
 
+def test_sync_subgraph_completed_event_does_not_end_run():
+    fake = SyncFakeServer()
+    fake.script(
+        [
+            lifecycle_completed_event(seq=1, namespace=["child:1"]),
+            lifecycle_errored_event(seq=2, error="root failed"),
+        ]
+    )
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="existing", assistant_id="agent") as thread:
+            terminal = thread._wait_for_run_done()
+
+    assert terminal.status == "errored", "a subgraph's completed event ended the run"
+    assert "root failed" in str(terminal.error)
+
+
 def test_sync_threads_stream_accepts_websocket_transport_option():
     with httpx.Client(base_url="http://test") as raw:
         threads = SyncThreadsClient(SyncHttpClient(raw))