mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-10-18 05:57:37 +02:00
Elior Nataf Lackritz
e901fc1746
An explicit thread_id can name a thread that run.start is about to create. Under custom auth its lifecycle stream gets a 404 first, which stopped the watcher for good, so thread.output raised that 404 after a successful run. run.start now starts a fresh watcher and run result when the watcher has stopped, and a 404 on a stream opened before run.start gets one more try once a run has started. Also reports the last error frame when lifecycle reconnects run out, and checks the stream error types with isinstance instead of getattr.
1065 lines
41 KiB
Python
1065 lines
41 KiB
Python
"""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_errored_event,
|
|
lifecycle_event,
|
|
lifecycle_started_event,
|
|
message_finish_event,
|
|
message_start_event,
|
|
message_text_delta_event,
|
|
message_text_finish_event,
|
|
tasks_start_event,
|
|
thread_stream_failed_frame,
|
|
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},
|
|
langsmith_tracing={"project_name": "replica-project"},
|
|
)
|
|
|
|
assert result == {"run_id": "run-1"}
|
|
assert fake.received_commands[0]["method"] == "run.start"
|
|
assert fake.received_commands[0]["params"]["assistant_id"] == "agent"
|
|
assert fake.received_commands[0]["params"]["langsmith_tracer"] == {
|
|
"project_name": "replica-project"
|
|
}
|
|
|
|
|
|
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_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))
|
|
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"]
|
|
tools_result = results["tools"]
|
|
assert isinstance(tools_result, list)
|
|
assert tools_result[0].name == "search"
|
|
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
|
|
|
|
|
|
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
|