feat: map Worldtree b1 eager turn-launch statuses (409/503) in stream_turn

Worldtree v1.0.0b1 (#331) decoupled turn execution from the SSE connection,
so turn-launch failures now arrive EAGERLY as an HTTP status before any
stream: 409 agent_not_available (pre-b1 was a 200 + in-stream error event)
and 503 (retryable turn-launch / infra failure). stream_turn previously
funneled both into a generic SseConnectFailed.

Map them to typed SseConnectFailed subclasses — AgentNotAvailable (409) and
TurnLaunchUnavailable (503, retryable=True) — carrying the parsed
error_code/message from the {detail:{error_code,message}} envelope.
Subclassing keeps existing `except SseConnectFailed` handlers working with
zero changes (POST-003 preserved — no synthetic event yielded; raise mirrors
reconnect_turn's 400/410/412 pattern).

worldtree-dev confirmed 409/503 are real runtime statuses; the OpenAPI 2.1.0
gap (not enumerating them) is theirs to fix (doc-completeness, not a wire
break). The 503 error_code is being re-pinned upstream (today internal_error
-> likely not_ready); our handling keys on STATUS so it's robust to the final
code — tighten the 503 default once they confirm.

Body shape live-confirmed against demo b1's 404/401 responses. Suite 509 green.
Contract docs/contracts/issues/1.contract.md updated.
This commit is contained in:
vh
2026-06-30 13:11:33 -07:00
parent af67ad995c
commit b2e4901264
5 changed files with 126 additions and 5 deletions
+56 -1
View File
@@ -6,6 +6,7 @@ import respx
from ratatoskr.sse_client import (
AffectUpdate,
AgentNotAvailable,
AwaitingLlmFirstToken,
CancelAlreadyCompleted,
Cancelled,
@@ -21,6 +22,7 @@ from ratatoskr.sse_client import (
SseId,
Text,
TurnIdFlip,
TurnLaunchUnavailable,
_parse_sse_id,
cancel_turn,
reconnect_turn,
@@ -415,8 +417,9 @@ class TestStreamTurn:
async def test_connect_failed_body_truncated(self) -> None:
"""ERROR_ROUTING: SseConnectFailed.body is truncated to <= 1024 bytes."""
big_body = b"x" * 5000
# 500 (not 409/503 — those are now eager turn-launch carve-outs, #331).
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(503, content=big_body)
return_value=httpx.Response(500, content=big_body)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SseConnectFailed) as exc_info:
@@ -424,6 +427,58 @@ class TestStreamTurn:
assert len(exc_info.value.body) <= 1024
assert exc_info.value.body == big_body[:1024]
@respx.mock
async def test_eager_409_agent_not_available(self) -> None:
"""b1 #331: eager 409 -> AgentNotAvailable (SseConnectFailed subclass) with
typed error_code; the turn never streams."""
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
409,
json={
"detail": {
"error_code": "agent_not_available",
"message": "agent ratatoskr:sindra is unavailable",
}
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AgentNotAvailable) as exc:
_ = [e async for e in stream_turn(client, "s1", "hi")]
assert exc.value.status == 409
assert exc.value.error_code == "agent_not_available"
assert "unavailable" in exc.value.message
assert isinstance(exc.value, SseConnectFailed) # existing handlers still catch
@respx.mock
async def test_eager_503_turn_launch_unavailable_retryable(self) -> None:
"""b1 #331: eager 503 -> TurnLaunchUnavailable (retryable, SseConnectFailed subclass)."""
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
503,
json={"error_code": "turn_launch_failed", "message": "resource exhausted"},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(TurnLaunchUnavailable) as exc:
_ = [e async for e in stream_turn(client, "s1", "hi")]
assert exc.value.status == 503
assert exc.value.retryable is True
assert exc.value.error_code == "turn_launch_failed"
assert isinstance(exc.value, SseConnectFailed)
@respx.mock
async def test_eager_409_non_json_body_defaults(self) -> None:
"""b1 #331: eager 409 with a non-JSON body -> AgentNotAvailable with the
status-derived default error_code."""
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(409, content=b"<html>nope</html>")
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(AgentNotAvailable) as exc:
_ = [e async for e in stream_turn(client, "s1", "hi")]
assert exc.value.error_code == "agent_not_available"
@respx.mock
async def test_no_text_aggregation(self) -> None:
"""no_text_aggregation: consumer yields each text event separately; no concat."""