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.

docs/contracts/issues/1.contract.md

@@ -106,7 +106,7 @@ POST: [POST-002 return_value] AsyncIterator yields ≥1 event ending in exactly
 POST: [POST-003 state_change] every yielded Event has a populated sse_id with both fields >= 1 -- assert all(e.sse_id.turn_id >= 1 and e.sse_id.seq >= 1 for e in events)
 ERROR_ROUTING:
   httpx.HTTPStatusError:
-    local_handling: re-raise as SseConnectFailed(status=resp.status_code, body=resp.read()[:1024]) — server returned non-2xx before stream started (e.g., 404 session_not_found)
+    local_handling: re-raise as SseConnectFailed(status=resp.status_code, body=resp.read()[:1024]) — server returned non-2xx before stream started (e.g., 404 session_not_found). EXCEPT the two eager turn-launch failures (Worldtree v1.0.0b1 #331), checked BEFORE raise_for_status and raised as typed SseConnectFailed SUBCLASSES carrying error_code: 409 -> AgentNotAvailable (agent unavailable; pre-b1 this was a 200 + in-stream `error` event), 503 -> TurnLaunchUnavailable (transient turn-launch failure; retryable=True). Subclassing keeps existing `except SseConnectFailed` handlers working with zero changes.
     flow_control: abort
     state_recovery: none (no events yielded yet)
   httpx.ReadError | httpx.RemoteProtocolError | httpx.ReadTimeout:
@@ -129,7 +129,8 @@ ERROR_ROUTING:
 STEPS:
   1. [setup, flexibility=prescriptive] Validate inputs per PRE-001..PRE-003
   2. [sequential, flexibility=prescriptive] Open SSE connection via httpx_sse.aconnect_sse with method="POST", url=f"/sessions/{session_id}/messages", json={"content": content}
-     ON httpx.HTTPStatusError before stream opens:
+     2a. [branch, flexibility=prescriptive] IF response.status_code in (409, 503) (b1 #331 eager turn-launch failures): read the body, parse (error_code, message) from the `{"detail": {...}}` envelope OR a flat `{error_code, message}` body (status-derived default code when absent), then RAISE AgentNotAvailable (409) / TurnLaunchUnavailable (503).
+     ON httpx.HTTPStatusError before stream opens (any other non-2xx):
        RAISE SseConnectFailed
   3. [loop, flexibility=prescriptive] FOR EACH sse_event in event_source.aiter_sse():
      0. [branch, flexibility=prescriptive] IF sse_event.data == "":
@@ -162,6 +163,9 @@ TESTS:
   error_terminal [error]: mock emits one `text` then `error` with `error_code: "llm_output_invalid"` → consumer yields Text then Error; iteration ends; Error.message and Error.error_code are populated
   cancelled_terminal [error]: mock emits `cancelled` with phase=cancelled → consumer yields Cancelled with turn_id; iteration ends
   session_not_found [error]: mock returns 404 before stream opens → consumer raises SseConnectFailed(status=404)
+  eager_409_agent_not_available [error]: mock returns 409 {detail:{error_code:"agent_not_available", message}} before stream → consumer raises AgentNotAvailable(status=409, error_code="agent_not_available", retryable absent); isinstance SseConnectFailed
+  eager_503_retryable [error]: mock returns 503 before stream → consumer raises TurnLaunchUnavailable(status=503, retryable=True); isinstance SseConnectFailed
+  eager_409_non_json_body [adversarial]: mock returns 409 with a non-JSON body → consumer raises AgentNotAvailable with the status-derived default error_code "agent_not_available"
   malformed_id_no_seq [adversarial]: mock event has `id: 42` (missing `:seq`) → consumer raises MalformedSseId; no event yielded
   malformed_id_alpha [adversarial]: mock event has `id: foo:bar` (non-integer parts) → consumer raises MalformedSseId
   turn_id_flip [adversarial]: mock emits text events with ids `42:1` then `99:2` → consumer raises TurnIdFlip; only the first event was yielded

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "ratatoskr"
-version = "0.18.2"
+version = "0.18.3"
 description = "Worldtree Conversation API debug TUI — multi-pane observability dashboard"
 readme = "README.md"
 requires-python = ">=3.12"

src/ratatoskr/sse_client.py

@@ -215,6 +215,55 @@ class SseConnectFailed(Exception):
         self.body = body
 
 
+class AgentNotAvailable(SseConnectFailed):
+    """Eager 409 from the turn POST (Worldtree v1.0.0b1, #331): the session's
+    agent is unavailable, so the turn never launched. Pre-b1 this arrived as a
+    200 stream + an in-stream `error` event; b1 surfaces it eagerly. Subclass of
+    SseConnectFailed so existing `except SseConnectFailed` handlers still catch
+    it — this type just adds the parsed `error_code` + `message`."""
+
+    def __init__(self, *, body: bytes, error_code: str, message: str) -> None:
+        super().__init__(status=409, body=body)
+        self.error_code = error_code
+        self.message = message
+
+
+class TurnLaunchUnavailable(SseConnectFailed):
+    """Eager 503 from the turn POST (Worldtree v1.0.0b1, #331): a transient
+    turn-launch failure (loop shutdown / resource exhaustion). RETRYABLE.
+    Subclass of SseConnectFailed; adds `error_code`, `message`, `retryable`."""
+
+    retryable = True
+
+    def __init__(self, *, body: bytes, error_code: str, message: str) -> None:
+        super().__init__(status=503, body=body)
+        self.error_code = error_code
+        self.message = message
+
+
+_EAGER_TURN_FAILURE_CODE = {409: "agent_not_available", 503: "turn_launch_unavailable"}
+
+
+def _eager_failure_fields(body: bytes, status: int) -> tuple[str, str]:
+    """Extract (error_code, message) from an eager turn-launch failure body
+    (#331). Accepts the Worldtree `{"detail": {...}}` envelope OR a flat
+    `{error_code, message}`; falls back to a status-derived default code and a
+    generic message when the body is absent / non-JSON / malformed."""
+    try:
+        parsed: Any = json.loads(body)
+    except (json.JSONDecodeError, ValueError):
+        parsed = None
+    src: dict[str, Any] = {}
+    if isinstance(parsed, dict):
+        detail = parsed.get("detail")
+        src = detail if isinstance(detail, dict) else parsed
+    code = src.get("error_code") or _EAGER_TURN_FAILURE_CODE[status]
+    message = src.get("message")
+    if not isinstance(message, str):
+        message = f"turn launch failed (HTTP {status})"
+    return str(code), message
+
+
 class SseConnectionDropped(Exception):
     """Raised when the HTTP/SSE connection dropped mid-stream."""
 
@@ -434,6 +483,19 @@ async def stream_turn(
         f"/sessions/{session_id}/messages",
         json={"content": content},
     ) as event_source:
+        # Worldtree v1.0.0b1 (#331): turn-launch failures arrive EAGERLY as a
+        # status before any stream — 409 agent_not_available (pre-b1 this was a
+        # 200 + in-stream `error` event), 503 a transient retryable launch
+        # failure. Surface them as typed SseConnectFailed subclasses carrying
+        # error_code; request-level non-2xx (404 session_not_found, etc.) stay
+        # generic SseConnectFailed.
+        status = event_source.response.status_code
+        if status in (409, 503):
+            body = await event_source.response.aread()
+            code, message = _eager_failure_fields(body, status)
+            if status == 409:
+                raise AgentNotAvailable(body=body, error_code=code, message=message)
+            raise TurnLaunchUnavailable(body=body, error_code=code, message=message)
         try:
             event_source.response.raise_for_status()
         except httpx.HTTPStatusError as exc:

tests/test_sse_client.py

@@ -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."""

uv.lock

@@ -1052,7 +1052,7 @@ wheels = [
 
 [[package]]
 name = "ratatoskr"
-version = "0.18.2"
+version = "0.18.3"
 source = { editable = "." }
 dependencies = [
     { name = "httpx" },