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

"""Tests for SyncThreadStream — Tasks 9.1 through 9.6."""

from __future__ import annotations

import queue
import re
import threading
import time
import uuid
from collections.abc import Iterator
from concurrent.futures import ThreadPoolExecutor
from typing import Any

import httpx
import orjson
import pytest

import langgraph_sdk.stream.sync_controller as _ctrl_mod
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.threads import SyncThreadsClient
from langgraph_sdk.stream.sync_controller import SyncStreamController
from langgraph_sdk.stream.transport.sync_http import (
    SyncEventStreamHandle,
    SyncProtocolSseTransport,
)
from streaming._events import (
    checkpoints_event,
    custom_event,
    lifecycle_completed_event,
    lifecycle_event,
    lifecycle_started_event,
    message_finish_event,
    message_start_event,
    message_text_delta_event,
    message_text_finish_event,
    tasks_start_event,
    tool_finished_event,
    tool_started_event,
    updates_event,
    values_event,
)
from streaming._sync_fake_server import SyncFakeServer, SyncStreamScript

# ---------------------------------------------------------------------------
# Task 9.1 — run_start_gate
# ---------------------------------------------------------------------------


def test_sync_subscribe_before_run_start_waits_on_gate():
    """A subscribe issued before run.start completes must block until the
    gate is set, mirroring async behavior."""
    fake = SyncFakeServer()
    # Lifecycle + fanout streams: empty so threads terminate cleanly.
    fake.script_sequence(
        [
            SyncStreamScript(events=[]),  # lifecycle watcher
            SyncStreamScript(events=[]),  # first subscribe
        ]
    )

    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:
            controller = thread._controller
            assert controller is not None

            started = threading.Event()

            def slow_subscriber() -> None:
                started.set()
                list(thread.subscribe(["values"]))

            t = threading.Thread(target=slow_subscriber)
            t.start()
            started.wait(timeout=0.5)

            # Set the gate manually (simulating run.start completing)
            time.sleep(0.05)
            assert controller._run_start_gate is not None
            controller._run_start_gate.set()

            t.join(timeout=2.0)

    # The subscriber should have unblocked and terminated cleanly.
    assert not t.is_alive(), "subscriber thread should have terminated"


# ---------------------------------------------------------------------------
# Task 9.2 — reconnect backoff + ready check
# ---------------------------------------------------------------------------


def test_sync_reconnect_uses_backoff_between_attempts(monkeypatch):
    """_reconnect_shared_stream sleeps between retry attempts with exp+jitter
    backoff, mirroring the async reconnect behavior."""

    sleeps: list[float] = []
    monkeypatch.setattr(_ctrl_mod.time, "sleep", lambda d: sleeps.append(d))

    class _FailingTransport(SyncProtocolSseTransport):
        """Transport that always raises on open_event_stream."""

        def open_event_stream(self, params: dict) -> SyncEventStreamHandle:  # noqa: ARG002
            raise RuntimeError("scripted transport failure")

    with httpx.Client(base_url="http://test") as raw:
        transport = _FailingTransport(client=raw, thread_id="t-1")
        controller = SyncStreamController(transport, max_reconnect_attempts=5)
        controller._shared_stream_filter = {"channels": ["values"]}
        result = controller._reconnect_shared_stream()

    assert result is False, "all attempts should have failed"
    # Attempts 0..4 → sleeps before attempts 1..4 → 4 sleeps
    assert len(sleeps) == 4, f"Expected 4 sleeps, got {sleeps}"
    # Backoff should grow (each delay is larger than previous, ignoring jitter)
    delays_without_jitter = [0.1 * (2**i) for i in range(4)]
    for i, (sleep, expected_base) in enumerate(
        zip(sleeps, delays_without_jitter, strict=False)
    ):
        assert sleep >= expected_base, (
            f"sleep[{i}]={sleep} < expected base {expected_base}"
        )


# ---------------------------------------------------------------------------
# Task 9.3 — rotation drains buffered events from old stream
# ---------------------------------------------------------------------------


def test_sync_rotation_does_not_lose_buffered_events():
    """When the shared stream rotates, old-stream events already in the queue
    are not dropped.  _drain_and_close dispatches remaining events from the
    old handle to subscribers before closing it."""

    event_a = values_event(seq=1, counter=1)

    class _ScriptedTransport(SyncProtocolSseTransport):
        """First call produces event_a; second call produces an empty stream."""

        def open_event_stream(self, params: dict) -> SyncEventStreamHandle:  # noqa: ARG002
            def _gen_a() -> Iterator[Any]:
                yield event_a

            def _gen_empty() -> Iterator[Any]:
                return
                yield  # pragma: no cover

            # Alternate: first call → a, second → empty.
            if not hasattr(self, "_call_count"):
                self._call_count = 0
            self._call_count += 1
            events_gen: Iterator[Any] = (
                _gen_a() if self._call_count == 1 else _gen_empty()
            )
            return SyncEventStreamHandle(
                events=events_gen,
                error=lambda: None,
                close=lambda: None,
            )

    with httpx.Client(base_url="http://test") as raw:
        transport = _ScriptedTransport(client=raw, thread_id="t-1")
        controller = SyncStreamController(transport)
        sub = controller.register_subscription({"channels": ["values"]})

        # First reconcile — opens old stream (event_a available immediately).
        controller.reconcile_stream({"channels": ["values"]})
        # Do NOT start fanout; let reconcile_stream cause a rotation directly.

        # Second reconcile: rotates to empty stream; drain thread handles old.
        controller.reconcile_stream({"channels": ["values", "updates"]})

        # Start fanout AFTER rotation (picks up the new empty stream).
        controller.ensure_fanout_running()

        # Allow drain thread to finish before collecting results.
        controller.close()

        received = []
        while True:
            try:
                item = sub.queue.get_nowait()
                if item is None:
                    continue
                received.append(item)
            except queue.Empty:
                break

    seqs = [e.get("seq") for e in received]
    assert 1 in seqs, f"event_a (seq=1) not received via drain; got seqs={seqs}"


# ---------------------------------------------------------------------------
# Task 9.4 — _next_command_id lock
# ---------------------------------------------------------------------------


def test_sync_concurrent_commands_do_not_share_command_id():
    """50 concurrent threads calling _send_command must each get a unique id."""

    captured_ids: list[int] = []
    ids_lock = threading.Lock()

    class _CapturingTransport(SyncProtocolSseTransport):
        """Captures command ids; always returns success."""

        def send_command(self, command: dict) -> dict:
            with ids_lock:
                captured_ids.append(command["id"])
            return {"type": "success", "id": command["id"], "result": {}}

        def open_event_stream(self, params: dict) -> SyncEventStreamHandle:  # noqa: ARG002
            def _gen() -> Iterator[Any]:
                return
                yield

            return SyncEventStreamHandle(
                events=_gen(), error=lambda: None, close=lambda: None
            )

    fake = SyncFakeServer()
    fake.script_sequence([SyncStreamScript(events=[])])

    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
        threads_client = SyncThreadsClient(SyncHttpClient(raw))
        with threads_client.stream(thread_id="t-cmd", assistant_id="agent") as stream:
            # Pre-set gate so _send_command doesn't wait.
            if stream._controller and stream._controller._run_start_gate:
                stream._controller._run_start_gate.set()
            # Replace transport with capturing transport.
            capture_transport = _CapturingTransport(client=raw, thread_id="t-cmd")
            stream._transport = capture_transport

            with ThreadPoolExecutor(max_workers=50) as ex:
                futures = [
                    ex.submit(stream._send_command, "noop", {}) for _ in range(50)
                ]
                for f in futures:
                    f.result()

    assert len(set(captured_ids)) == 50, (
        f"Expected 50 unique command ids, got {len(set(captured_ids))} unique "
        f"out of {len(captured_ids)} total: {sorted(captured_ids)}"
    )


# ---------------------------------------------------------------------------
# Task 9.5 — sync events returns fresh iterator per access
# ---------------------------------------------------------------------------


def test_sync_events_returns_fresh_iterator_each_access():
    """Two accesses of `thread.events` yield independent subscriptions,
    mirroring the async semantics where each access opens a new subscriber."""
    fake = SyncFakeServer()

    event_1 = values_event(seq=1, counter=1)
    fake.script_sequence(
        [
            SyncStreamScript(events=[]),  # lifecycle watcher
            SyncStreamScript(events=[event_1]),  # first events access
            SyncStreamScript(events=[event_1]),  # second events access
        ]
    )

    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
        threads_client = SyncThreadsClient(SyncHttpClient(raw))
        with threads_client.stream(thread_id="t-5", assistant_id="agent") as thread:
            # Pre-set gate.
            if thread._controller and thread._controller._run_start_gate:
                thread._controller._run_start_gate.set()

            iter1 = thread.events
            iter2 = thread.events

            # They must be independent objects (different subscription iterators).
            assert iter1 is not iter2


# ---------------------------------------------------------------------------
# Task 9.6 — close ordering: fail active streams before controller close
# ---------------------------------------------------------------------------


def test_close_unblocks_active_subscription_before_lifecycle_join():
    """close() must send None to active subscriptions BEFORE joining the
    lifecycle watcher thread, so callers wake quickly even if the watcher
    thread blocks for up to 1s."""

    # Gate that keeps the lifecycle watcher thread alive for 0.4s.
    lifecycle_block = threading.Event()
    unblock_times: list[float] = []
    close_times: list[float] = []

    class _BlockingFakeServer(SyncFakeServer):
        """Lifecycle stream blocks until gate set; subscribe stream is empty."""

        def _handle(self, request: httpx.Request) -> httpx.Response:
            path = request.url.path
            if path.endswith("/stream/events"):
                body = orjson.loads(request.content)
                channels = body.get("channels", [])
                if "lifecycle" in channels:
                    # Block lifecycle watcher for 0.4s.
                    lifecycle_block.wait(timeout=0.4)
            return super()._handle(request)

    fake = _BlockingFakeServer()
    fake.script_sequence(
        [
            SyncStreamScript(events=[]),  # lifecycle watcher
            SyncStreamScript(events=[]),  # subscribe fanout stream
        ]
    )

    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
        threads_client = SyncThreadsClient(SyncHttpClient(raw))
        with threads_client.stream(thread_id="t-6", assistant_id="agent") as thread:
            if thread._controller and thread._controller._run_start_gate:
                thread._controller._run_start_gate.set()

            assert thread._controller is not None
            sub = thread._controller.register_subscription({"channels": ["values"]})
            thread._controller.reconcile_stream({"channels": ["values"]})
            thread._controller.ensure_fanout_running()

            consumer_ready = threading.Event()

            def _consume() -> None:
                consumer_ready.set()
                while True:
                    try:
                        item = sub.queue.get(timeout=2.0)
                        if item is None:
                            unblock_times.append(time.monotonic())
                            return
                    except queue.Empty:
                        return

            t = threading.Thread(target=_consume)
            t.start()
            consumer_ready.wait(timeout=1.0)
            time.sleep(0.02)

            close_times.append(time.monotonic())
            # __exit__ calls close() here.

    lifecycle_block.set()  # Unblock watcher so test can finish.
    t.join(timeout=2.0)
    assert not t.is_alive(), "consumer thread should have unblocked"
    assert unblock_times, "consumer never received sentinel"
    elapsed = unblock_times[0] - close_times[0]
    # With controller closed BEFORE lifecycle join, sentinel arrives fast.
    # Lifecycle watcher blocks for 0.4s but that should not delay the sentinel.
    assert elapsed < 0.3, (
        f"consumer woke {elapsed:.3f}s after close() — "
        "controller.close() should precede the lifecycle thread join"
    )


def test_sync_thread_agent_get_tree_fetches_assistant_graph():
    fake = SyncFakeServer()
    fake.set_graph(
        {
            "nodes": [{"id": "agent", "type": "runnable", "data": {"name": "agent"}}],
            "edges": [{"source": "agent", "target": "__end__"}],
        }
    )
    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",
            headers={"X-Custom-Header": "my-value"},
        ) as thread:
            graph = thread.agent.get_tree(xray=True)

    assert graph["nodes"][0]["id"] == "agent"
    assert graph["edges"] == [{"source": "agent", "target": "__end__"}]
    assert fake.graph_request_params == [{"xray": "true"}]
    assert fake.graph_request_headers[0].get("x-custom-header") == "my-value"


def test_sync_thread_agent_get_tree_raises_after_close():
    with httpx.Client(base_url="http://test") as raw:
        threads = SyncThreadsClient(SyncHttpClient(raw))
        stream = threads.stream(thread_id="t-1", assistant_id="agent")
        stream.close()
        with pytest.raises(RuntimeError, match="closed"):
            stream.agent.get_tree()


def test_sync_extensions_projection_empty_name_raises():
    with httpx.Client(base_url="http://test") as raw:
        threads = SyncThreadsClient(SyncHttpClient(raw))
        stream = threads.stream(thread_id="t-1", assistant_id="agent")
        with pytest.raises(ValueError, match="non-empty"):
            stream.extensions[""]


def test_sync_extensions_projection_closed_stream_yields_nothing():
    with httpx.Client(base_url="http://test") as raw:
        threads = SyncThreadsClient(SyncHttpClient(raw))
        # Enter and immediately exit so _controller is set but _closed is True.
        with threads.stream(thread_id="t-1", assistant_id="agent") as stream:
            pass
        payloads = list(stream.extensions["progress"])
    assert payloads == []


def test_sync_threads_stream_mints_uuid4_when_thread_id_none():
    with httpx.Client(base_url="http://test") as raw:
        threads = SyncThreadsClient(SyncHttpClient(raw))
        stream = threads.stream(assistant_id="agent")
        assert re.fullmatch(
            r"[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}",
            stream.thread_id,
        )
        assert uuid.UUID(stream.thread_id).version == 4


def test_sync_run_start_sends_command():

    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(thread_id="t-1", assistant_id="agent") as thread:
            result = thread.run.start(input={"x": 1})

    assert result == {"run_id": "run-1"}
    assert fake.received_commands[0]["method"] == "run.start"
    assert fake.received_commands[0]["params"]["assistant_id"] == "agent"


def test_sync_events_iterates_raw_events():

    fake = SyncFakeServer()
    fake.script([values_event(seq=1, counter=1)])
    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:
            thread.run.start(input={})
            events = list(thread.subscribe(["values"]))

    assert events == [values_event(seq=1, counter=1)]


def test_sync_lifecycle_watcher_reconnects_with_since_after_transport_drop():

    fake = SyncFakeServer()
    fake.set_state({"ok": True})
    fake.script_sequence(
        [
            SyncStreamScript(
                events=[lifecycle_event(seq=1, phase="running")],
                fail_after=1,
            ),
            SyncStreamScript(events=[lifecycle_completed_event(seq=2)]),
        ]
    )
    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 == "completed"
    assert terminal.error is None
    assert fake.stream_request_bodies[1]["since"] == 1


def test_sync_threads_stream_accepts_websocket_transport_option():
    with httpx.Client(base_url="http://test") as raw:
        threads = SyncThreadsClient(SyncHttpClient(raw))
        stream = threads.stream(
            thread_id="t-1",
            assistant_id="agent",
            transport="websocket",
        )
    assert stream._transport_kind == "websocket"


def test_sync_threads_stream_rejects_unknown_transport_option():

    with httpx.Client(base_url="http://test") as raw:
        threads = SyncThreadsClient(SyncHttpClient(raw))
        with pytest.raises(ValueError, match="transport"):
            threads.stream(
                thread_id="t-1",
                assistant_id="agent",
                transport="bogus",  # ty: ignore[invalid-argument-type]
            )


def test_v3_streaming_sync_surface_smoke():

    fake = SyncFakeServer()
    fake.set_state({"final": True})
    # Single script — projections consume events in parallel threads so all
    # subscriptions are registered before SSE rotation could drop events.
    # Mirrors the async smoke test's `asyncio.gather` pattern.
    fake.script(
        [
            values_event(seq=1, values={"step": 1}),
            message_start_event(seq=2, message_id="msg-1"),
            message_text_delta_event(seq=3, text="hi", message_id="msg-1"),
            message_text_finish_event(seq=4, text="hi", message_id="msg-1"),
            message_finish_event(seq=5, message_id="msg-1"),
            tool_started_event(seq=6, tool_call_id="call-1", tool_name="search"),
            tool_finished_event(seq=7, tool_call_id="call-1", output={"ok": True}),
            custom_event(seq=8, name="progress", step=1),
            lifecycle_completed_event(seq=9),
        ]
    )
    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:
            start = thread.run.start(
                input={"messages": [{"role": "user", "content": "hi"}]}
            )

            # Gate every reconcile_stream call on a barrier so that all four
            # projection threads register their subscriptions before any
            # reconcile widens (or rotates) the shared SSE. This mirrors the
            # async smoke test's `asyncio.gather` pattern: every subscription
            # is registered before the first SSE opens; one SSE covers all
            # consumers and `_seen_event_ids` covers any subsequent reconnect.
            controller = thread._controller
            assert controller is not None
            barrier = threading.Barrier(4)
            real_reconcile = controller.reconcile_stream

            def _gated_reconcile(candidate_filter):
                barrier.wait(timeout=10)
                return real_reconcile(candidate_filter)

            controller.reconcile_stream = _gated_reconcile  # ty: ignore[invalid-assignment]

            results: dict[str, object] = {}
            errors: list[BaseException] = []

            def _run_values() -> None:
                try:
                    for v in thread.values:
                        results["values"] = v
                        return
                except BaseException as err:  # pragma: no cover - propagated
                    errors.append(err)

            def _run_messages() -> None:
                try:
                    results["messages"] = list(thread.messages)
                except BaseException as err:  # pragma: no cover - propagated
                    errors.append(err)

            def _run_tools() -> None:
                try:
                    results["tools"] = list(thread.tool_calls)
                except BaseException as err:  # pragma: no cover - propagated
                    errors.append(err)

            def _run_progress() -> None:
                try:
                    results["progress"] = list(thread.extensions["progress"])
                except BaseException as err:  # pragma: no cover - propagated
                    errors.append(err)

            workers = [
                threading.Thread(target=_run_values),
                threading.Thread(target=_run_messages),
                threading.Thread(target=_run_tools),
                threading.Thread(target=_run_progress),
            ]
            for w in workers:
                w.start()
            for w in workers:
                w.join(timeout=10)
                assert not w.is_alive(), "smoke worker thread hung"
            controller.reconcile_stream = real_reconcile  # ty: ignore[invalid-assignment]
            assert not errors, errors
            final = thread.output

    assert start == {"run_id": "run-1"}
    assert results["values"] == fake.state["values"]
    messages_result = results["messages"]
    assert isinstance(messages_result, list)
    assert [str(m.text) for m in messages_result] == ["hi"]  # ty: ignore[unresolved-attribute]
    tools_result = results["tools"]
    assert isinstance(tools_result, list)
    assert tools_result[0].name == "search"  # ty: ignore[unresolved-attribute]
    assert results["progress"] == [{"name": "progress", "step": 1}]
    assert final == {"final": True}


# ---------------------------------------------------------------------------
# interleave_projections tests
# ---------------------------------------------------------------------------


def test_interleave_projections_single_channel_values():

    fake = SyncFakeServer()
    fake.set_state({"counter": 0})
    fake.script(
        [
            lifecycle_started_event(seq=0),
            values_event(seq=1, counter=1),
            values_event(seq=2, counter=2),
            lifecycle_completed_event(seq=3),
        ]
    )
    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:
            thread.run.start(input={})
            items = []
            for ch, item in thread.interleave_projections(["values"]):
                items.append((ch, item))
    assert ("values", {"counter": 1}) in items
    assert ("values", {"counter": 2}) in items
    assert all(ch == "values" for ch, _ in items)


def test_interleave_projections_values_and_messages_arrival_order():

    fake = SyncFakeServer()
    fake.set_state({"counter": 0})
    fake.script(
        [
            lifecycle_started_event(seq=0),
            values_event(seq=1, counter=1),
            message_start_event(seq=2, message_id="m-1"),
            values_event(seq=3, counter=2),
            message_finish_event(seq=4, message_id="m-1"),
            lifecycle_completed_event(seq=5),
        ]
    )
    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:
            thread.run.start(input={})
            order = []
            for ch, _ in thread.interleave_projections(["values", "messages"]):
                order.append(ch)
                if len(order) >= 3:
                    break
    assert order[:3] == ["values", "messages", "values"]


def test_interleave_projections_mixes_builtin_and_extension():

    fake = SyncFakeServer()
    fake.set_state({"counter": 0})
    fake.script(
        [
            lifecycle_started_event(seq=0),
            values_event(seq=1, counter=1),
            custom_event(seq=2, name="foo", hello="world"),
            lifecycle_completed_event(seq=3),
        ]
    )
    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:
            thread.run.start(input={})
            items = []
            for ch, item in thread.interleave_projections(["values", "foo"]):
                items.append((ch, item))
    assert ("values", {"counter": 1}) in items
    assert ("foo", {"name": "foo", "hello": "world"}) in items


def test_interleave_projections_tool_calls_uses_public_name():

    fake = SyncFakeServer()
    fake.set_state({})
    fake.script(
        [
            lifecycle_started_event(seq=0),
            tool_started_event(seq=1, tool_call_id="call-1", tool_name="search"),
            tool_finished_event(seq=2, tool_call_id="call-1", output={"ok": True}),
            lifecycle_completed_event(seq=3),
        ]
    )
    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:
            thread.run.start(input={})
            names = []
            handle = None
            for ch, item in thread.interleave_projections(["tool_calls"]):
                names.append(ch)
                if handle is None:
                    handle = item
                break
    assert names == ["tool_calls"]
    assert handle is not None
    assert handle.tool_call_id == "call-1"


def test_interleave_projections_subgraphs_discovers_child():

    fake = SyncFakeServer()
    fake.set_state({})
    fake.script(
        [
            lifecycle_started_event(seq=0),
            lifecycle_started_event(seq=1, namespace=["child"]),
            lifecycle_completed_event(seq=2),
        ]
    )
    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:
            thread.run.start(input={})
            discovered = []
            for ch, handle in thread.interleave_projections(["subgraphs"]):
                discovered.append((ch, handle.path))
    assert ("subgraphs", ("child",)) in discovered


def test_interleave_projections_inflight_tool_call_failed_on_break():
    """A tool handle held past an early break is failed in teardown, never left hanging."""

    fake = SyncFakeServer()
    fake.set_state({})
    fake.script(
        [
            lifecycle_started_event(seq=0),
            tool_started_event(seq=1, tool_call_id="call-1", tool_name="search"),
            # no tool-finished: the call is still in flight when the consumer breaks
            lifecycle_completed_event(seq=2),
        ]
    )
    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:
            thread.run.start(input={})
            handle = None
            for _, item in thread.interleave_projections(["tool_calls"]):
                handle = item
                break
            assert handle is not None
            # Without teardown finalization this blocks forever; the bounded
            # timeout turns a regression into a TimeoutError, not a RuntimeError.
            with pytest.raises(RuntimeError):
                handle._result.result(timeout=2)


def test_interleave_projections_inflight_subgraph_finished_on_terminal():
    """A discovered subgraph child with no terminal tasks-result is force-completed."""

    fake = SyncFakeServer()
    fake.set_state({})
    fake.script(
        [
            lifecycle_started_event(seq=0),
            lifecycle_started_event(seq=1, namespace=["child"]),
            # no tasks-result for the child: it is still "started" at run end
            lifecycle_completed_event(seq=2),
        ]
    )
    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:
            thread.run.start(input={})
            child = None
            for _, handle in thread.interleave_projections(["subgraphs"]):
                child = handle
            assert child is not None
            assert child.status == "completed"


@pytest.mark.parametrize("channel", ["lifecycle", "tools", "input"])
def test_interleave_projections_rejects_reserved_channel(channel):
    """Reserved protocol channel names raise instead of silently no-op'ing.

    `infer_channel` treats these as first-class methods, but they have no
    interleave decoder, so routing them to the extension/`custom:` fallback
    would subscribe to a channel that never matches and yield nothing. Fail
    closed. (`updates`/`checkpoints`/`tasks` are supported and tested below.)
    """

    fake = SyncFakeServer()
    fake.set_state({})
    fake.script([lifecycle_started_event(seq=0), lifecycle_completed_event(seq=1)])
    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,
            pytest.raises(ValueError, match=channel),
        ):
            for _ in thread.interleave_projections([channel]):
                pass


def test_interleave_projections_data_channels_yield_payloads():
    """`updates`/`checkpoints`/`tasks` yield their raw `params.data` payloads."""

    fake = SyncFakeServer()
    fake.set_state({})
    fake.script(
        [
            lifecycle_started_event(seq=0),
            updates_event(seq=1, node={"v": 1}),
            checkpoints_event(seq=2, ts="t-0", v=4),
            tasks_start_event(seq=3, task_id="task-9"),
            lifecycle_completed_event(seq=4),
        ]
    )
    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:
            thread.run.start(input={})
            items = list(
                thread.interleave_projections(["updates", "checkpoints", "tasks"])
            )
    assert ("updates", {"node": {"v": 1}}) in items
    assert ("checkpoints", {"ts": "t-0", "v": 4}) in items
    assert any(ch == "tasks" and item.get("id") == "task-9" for ch, item in items)


def test_interleave_projections_data_channel_scoped_to_root_namespace():
    """A child-namespace checkpoint must not leak into a root interleave."""

    fake = SyncFakeServer()
    fake.set_state({"counter": 0})
    fake.script(
        [
            lifecycle_started_event(seq=0),
            checkpoints_event(seq=1, namespace=["child"], scope="child"),
            checkpoints_event(seq=2, scope="root"),
            lifecycle_completed_event(seq=3),
        ]
    )
    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:
            thread.run.start(input={})
            checkpoints = [
                item
                for ch, item in thread.interleave_projections(["values", "checkpoints"])
                if ch == "checkpoints"
            ]
    assert {"scope": "root"} in checkpoints
    assert {"scope": "child"} not in checkpoints