diff --git a/docs/contracts/issues/1.contract.md b/docs/contracts/issues/1.contract.md index dcc5ac8..48680be 100644 --- a/docs/contracts/issues/1.contract.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index dba0473..b5a29d1 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/ratatoskr/sse_client.py b/src/ratatoskr/sse_client.py index 1cd298a..345d0c1 100644 --- a/src/ratatoskr/sse_client.py +++ b/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: diff --git a/tests/test_sse_client.py b/tests/test_sse_client.py index dc4ef60..d7ea595 100644 --- a/tests/test_sse_client.py +++ b/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"nope") + ) + 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.""" diff --git a/uv.lock b/uv.lock index 6db7864..abb0edf 100644 --- a/uv.lock +++ b/uv.lock @@ -1052,7 +1052,7 @@ wheels = [ [[package]] name = "ratatoskr" -version = "0.18.2" +version = "0.18.3" source = { editable = "." } dependencies = [ { name = "httpx" },