fix: isolate test server and wait for root lifecycle completion

Run the locked dev server on supported Python separately from the core test matrix. Ignore child terminal lifecycle events until the root finishes committing state.

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
John Kennedy
2026-10-17 01:44:27 +00:00
co-authored by open-swe[bot] <open-swe@users.noreply.github.com>
parent b49bfeab05
commit 89fad2ab8b
4 changed files with 82 additions and 1 deletions
+7 -1
View File
@@ -43,9 +43,15 @@ start-services:
stop-services:
docker compose -f tests/compose-postgres.yml -f tests/compose-redis.yml down -v
# The test matrix includes 3.10 and 3.14, but the locked in-memory server
# dependencies are available on 3.11-3.13. Keep its interpreter separate.
DEV_SERVER_PYTHON ?= 3.12
DEV_SERVER_ENV = $(CURDIR)/.venvs/dev-server
start-dev-server:
uv sync --frozen --group test --no-dev
@LOG_LEVEL=warning uv run --no-sync langgraph dev --config tests/example_app/langgraph.json --no-browser --no-reload & \
UV_PROJECT_ENVIRONMENT=$(DEV_SERVER_ENV) uv sync --frozen --group test --no-dev --python $(DEV_SERVER_PYTHON)
@LOG_LEVEL=warning UV_PROJECT_ENVIRONMENT=$(DEV_SERVER_ENV) uv run --no-sync --python $(DEV_SERVER_PYTHON) langgraph dev --config tests/example_app/langgraph.json --no-browser --no-reload & \
PID=$$!; echo "$$PID" > .devserver.pid; \
for ATTEMPT in $$(seq 1 60); do \
if ! kill -0 $$PID 2>/dev/null; then break; fi; \
@@ -1972,6 +1972,10 @@ class AsyncThreadStream:
# knows a run exists (handles reattach without run.start).
self._run_seen = True
elif phase in ("completed", "failed"):
# A child may finish before its parent commits the child's update.
# Only the root lifecycle can make thread.output terminal.
if isinstance(params, dict) and params.get("namespace"):
return
# 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
@@ -1614,6 +1614,10 @@ class SyncThreadStream:
if phase in ("started", "running"):
self._run_seen = True
elif phase in ("completed", "failed"):
# A child may finish before its parent commits the child's update.
# Only the root lifecycle can make thread.output terminal.
if isinstance(params, dict) and params.get("namespace"):
return
self.interrupted = False
self.interrupts = []
run_done = self._run_done
@@ -0,0 +1,67 @@
"""Child lifecycle completion must not finish the parent thread's output."""
from __future__ import annotations
import asyncio
from typing import cast
import httpx
import pytest
from langchain_protocol import Event
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
from langgraph_sdk._sync.http import SyncHttpClient
from langgraph_sdk._sync.stream import _BlockingResult
from langgraph_sdk._sync.threads import SyncThreadsClient
from streaming._events import input_requested_event, lifecycle_event
@pytest.mark.parametrize("phase", ["completed", "failed"])
async def test_async_child_terminal_does_not_finish_root(phase):
async with httpx.AsyncClient(base_url="http://test") as raw:
thread = ThreadsClient(HttpClient(raw)).stream(
thread_id="t-1", assistant_id="agent"
)
thread._run_done = asyncio.get_running_loop().create_future()
await thread._apply_lifecycle_event(
cast(Event, input_requested_event(namespace=["sub:1"]))
)
await thread._apply_lifecycle_event(
cast(Event, lifecycle_event(namespace=["sub:1"], event=phase))
)
assert not thread._run_done.done()
assert thread.interrupted
assert thread.interrupts[0]["namespace"] == ["sub:1"]
await thread._apply_lifecycle_event(
cast(Event, lifecycle_event(event="completed"))
)
assert thread._run_done.done()
assert (await thread._run_done).error is None
assert not thread.interrupted
assert thread.interrupts == []
@pytest.mark.parametrize("phase", ["completed", "failed"])
def test_sync_child_terminal_does_not_finish_root(phase):
with httpx.Client(base_url="http://test") as raw:
thread = SyncThreadsClient(SyncHttpClient(raw)).stream(
thread_id="t-1", assistant_id="agent"
)
thread._run_done = _BlockingResult()
thread._apply_lifecycle_event(
cast(Event, input_requested_event(namespace=["sub:1"]))
)
thread._apply_lifecycle_event(
cast(Event, lifecycle_event(namespace=["sub:1"], event=phase))
)
assert not thread._run_done.done()
assert thread.interrupted
assert thread.interrupts[0]["namespace"] == ["sub:1"]
thread._apply_lifecycle_event(cast(Event, lifecycle_event(event="completed")))
assert thread._run_done.done()
assert thread._run_done.result().error is None
assert not thread.interrupted
assert thread.interrupts == []