From e3a10ad80e696c40ac93b5080d878c1df4ae13da Mon Sep 17 00:00:00 2001 From: Vuong Hoang Date: Sun, 19 Jul 2026 06:07:50 -0700 Subject: [PATCH] feat(#20): rewire the CLI turn path onto the wt adapter (slice-2, part 2b-i) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The --send turn path (_amain create + _run_turn stream + _cancel_and_log) now goes through ratatoskr.wt over the worldtree-sdk; external CLI behavior (output, exit codes) is preserved. No hand-rolled path is deleted yet — web/server.py still uses them (part 2b-ii), so the deletions + live smoke come after web is rewired. - _amain builds one WorldtreeClient via wt.build_client over a ratatoskr-owned transport (INV-CUT-1); create → wt.create_session (reads the SDK's open create dict); the transport keeps the default bearer so the not-yet-migrated hand-rolled seed_preset_first_message (slice-3) still authenticates. - _run_turn drives wt.stream_turn and consumes SDK TurnEvents; the mid-stream cancel target is parsed from the composite sse_id ("{turn}:{seq}") — the SDK's top-level turn_id is the body field and is absent on text/thinking frames. - CliPresenterState.render consumes the SDK TurnEvent union with None-hardening on the now-optional fields (usage degrades to "(n/a)" rather than crashing). - The SDK normalizes a pre-response transport failure to ConnectFailed(status=0); _amain (network → exit 21) and _cancel_and_log (swallow, INV-009) catch it. - build_client gains max_reconnects (SDK default 5; tests pass 0 to surface drops immediately). test_cli: SDK-event factories keep the render-test bodies intact; client constructions wrap in build_client; cancel-race mocks carry the SDK's (status, error_code) pair. Suite 570 green; cli.py + wt.py mypy + ruff clean (the pre-existing send_content arg-type note is unchanged). Patch (internal; external CLI behavior preserved). --- pyproject.toml | 2 +- src/ratatoskr/cli.py | 193 +++++++++++++++++++++++++++++-------------- src/ratatoskr/wt.py | 14 ++-- tests/test_cli.py | 165 ++++++++++++++++++++++++++++-------- uv.lock | 2 +- 5 files changed, 272 insertions(+), 104 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 614a6f8..4a7cece 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "ratatoskr" -version = "0.21.5" +version = "0.21.6" description = "Worldtree Conversation API debug console (web + headless CLI) — multi-pane observability" readme = "README.md" requires-python = ">=3.12" diff --git a/src/ratatoskr/cli.py b/src/ratatoskr/cli.py index 3df1599..3082ba2 100644 --- a/src/ratatoskr/cli.py +++ b/src/ratatoskr/cli.py @@ -11,12 +11,30 @@ import hashlib import os import signal import sys +from collections.abc import Mapping from dataclasses import dataclass, field from importlib.metadata import PackageNotFoundError, version from typing import Any, TextIO import httpx +from worldtree_sdk import ( + AffectUpdateEvent, + AwaitingLlmFirstTokenEvent, + CancelledEvent, + ConnectFailed, + DoneEvent, + ErrorEvent, + TextBoundaryEvent, + TextEvent, + ThinkingEvent, + ToolResultEvent, + ToolStartEvent, + TurnEvent, + WorkerPhaseEvent, + WorldtreeClient, +) +from ratatoskr import wt from ratatoskr.first_message import seed_preset_first_message from ratatoskr.sessions import ( AgentNotFound, @@ -37,29 +55,21 @@ from ratatoskr.sessions import ( set_persona_state, write_authored_history, ) + +# The turn path (create / stream / cancel) is served by the worldtree-sdk adapter +# (`wt.*`); these caller-semantic exceptions are what the adapter raises, so the +# presenter keeps catching ratatoskr's own types (DEC-2). The hand-rolled probes +# (--whoami / --characters / --set-persona / --seed-first-message) stay on the +# `sessions` wrappers until their own slices. from ratatoskr.sse_client import ( - AffectUpdate, - AwaitingLlmFirstToken, CancelAlreadyCompleted, CancelFailed, - Cancelled, CancelTurnNotFound, - Done, - Error, - Event, MalformedSseData, MalformedSseId, SseConnectFailed, SseConnectionDropped, - Text, - TextBoundary, - Thinking, - ToolResult, - ToolStart, TurnIdFlip, - WorkerPhase, - cancel_turn, - stream_turn_resilient, ) @@ -337,6 +347,29 @@ def _format_usage(usage: dict[str, int], *, arrow: str) -> str: return f"{p} in {arrow} {c} out ({t} total, {ci} cached)" +def _format_usage_safe(usage: Mapping[str, int] | None) -> str: + """Tolerant wrapper over `_format_usage` for the SDK's open-world + `DoneEvent.usage` (typed optional): the canonical four-key usage formats; + anything absent or malformed degrades to `(n/a)` rather than crashing the + presenter (same posture as `_format_whoami`).""" + keys = ("prompt_tokens", "completion_tokens", "total_tokens", "cached_input_tokens") + if usage is not None and all(k in usage for k in keys): + return _format_usage(dict(usage), arrow="->") + return "(n/a)" + + +def _turn_id_from_sse_id(sse_id: str) -> int | None: + """The turn component of the SDK's composite sse_id (`"{turn}:{seq}"`). This is + the mid-stream cancel target: it is present on EVERY frame, unlike the SDK's + top-level `turn_id`, which is the body field (absent on text/thinking events).""" + head, _, _ = sse_id.partition(":") + try: + turn = int(head) + except ValueError: + return None + return turn if turn > 0 else None + + @dataclass(slots=True) class CliPresenterState: """Per-turn presenter state for `--send` mode (issue #12). @@ -348,24 +381,31 @@ class CliPresenterState: thinking_open: bool = False text_written_since_newline: bool = False - def render(self, event: Event, *, stdout: TextIO, stderr: TextIO) -> None: - """Render one Worldtree SSE event with editorial hierarchy + coalescing.""" + def render(self, event: TurnEvent, *, stdout: TextIO, stderr: TextIO) -> None: + """Render one Worldtree SSE event with editorial hierarchy + coalescing. + + Consumes the worldtree-sdk `TurnEvent` union. The SDK types the de-facto + fields as OPTIONAL (open-world), so every read is hardened: a + malformed/partial event degrades to a placeholder rather than crashing the + presenter — the same posture as `_format_whoami`. + """ assert isinstance( event, ( - WorkerPhase, Thinking, Text, TextBoundary, - ToolStart, ToolResult, Done, Error, Cancelled, - AffectUpdate, AwaitingLlmFirstToken, + WorkerPhaseEvent, ThinkingEvent, TextEvent, TextBoundaryEvent, + ToolStartEvent, ToolResultEvent, DoneEvent, ErrorEvent, CancelledEvent, + AffectUpdateEvent, AwaitingLlmFirstTokenEvent, ), ) # Thinking events accumulate into the open run. - if isinstance(event, Thinking): + if isinstance(event, ThinkingEvent): + content = event.content or "" if not self.thinking_open: stderr.write(". thinking: ") self.thinking_open = True - stderr.write(event.content) + stderr.write(content) stderr.flush() - self.thinking_buffer.append(event.content) + self.thinking_buffer.append(content) return # Non-thinking event: close any open thinking run first. if self.thinking_open: @@ -374,59 +414,60 @@ class CliPresenterState: self.thinking_open = False self.thinking_buffer.clear() # Now render the new event. - if isinstance(event, Text): - stdout.write(event.content) + if isinstance(event, TextEvent): + content = event.content or "" + stdout.write(content) stdout.flush() # POST-003: only set if cursor is mid-line (no trailing newline). - self.text_written_since_newline = not event.content.endswith("\n") + self.text_written_since_newline = not content.endswith("\n") return - if isinstance(event, (Done, Error, Cancelled)): + if isinstance(event, (DoneEvent, ErrorEvent, CancelledEvent)): # INV-005: ensure stdout newline boundary before stderr terminal label. if self.text_written_since_newline: stdout.write("\n") stdout.flush() self.text_written_since_newline = False - if isinstance(event, Done): + if isinstance(event, DoneEvent): stderr.write( - f"[done] turn_id={event.sse_id.turn_id} model={event.model} " - f"duration={_format_duration_ms(event.duration_ms)} " - f"usage {_format_usage(event.usage, arrow='->')}\n" + f"[done] turn_id={event.turn_id} model={event.model} " + f"duration={_format_duration_ms(event.duration_ms or 0)} " + f"usage {_format_usage_safe(event.usage)}\n" ) return - if isinstance(event, WorkerPhase): + if isinstance(event, WorkerPhaseEvent): stderr.write( f". worker_phase: phase={event.phase} turn_id={event.turn_id}\n" ) return - if isinstance(event, Error): + if isinstance(event, ErrorEvent): stderr.write( - f"[error] turn_id={event.sse_id.turn_id} code={event.error_code} " + f"[error] turn_id={event.turn_id} code={event.error_code} " f"message={event.message!r}\n" ) return - if isinstance(event, Cancelled): + if isinstance(event, CancelledEvent): stderr.write( f"[cancelled] turn_id={event.turn_id} reason={event.reason!r} " f"partial_message_id={event.partial_message_id}\n" ) return - if isinstance(event, ToolStart): + if isinstance(event, ToolStartEvent): stderr.write( f". tool_start: name={event.name} args={event.arguments!r}\n" ) return - if isinstance(event, ToolResult): + if isinstance(event, ToolResultEvent): stderr.write( f". tool_result: name={event.name} duration_ms={event.duration_ms} " f"result={event.result!r:.200}\n" ) return - if isinstance(event, TextBoundary): + if isinstance(event, TextBoundaryEvent): stderr.write( f". text_boundary: kind={event.kind} char_offset={event.char_offset}\n" ) return - if isinstance(event, AffectUpdate): + if isinstance(event, AffectUpdateEvent): # Worldtree #204 / v0.28.0. CLI surface is debug telemetry — # one line to stderr with status + (for current) dominant_emotion. if event.snapshot is not None: @@ -440,10 +481,10 @@ class CliPresenterState: f". affect_update: status={event.status} turn_id={event.turn_id}\n" ) return - if isinstance(event, AwaitingLlmFirstToken): + if isinstance(event, AwaitingLlmFirstTokenEvent): # Worldtree #201 / v0.29.0. Heartbeat during BuildingPrompt → # CallingLLM gap. Stderr surface, one line per heartbeat. - secs = event.elapsed_ms_since_building_prompt / 1000.0 + secs = (event.elapsed_ms_since_building_prompt or 0) / 1000.0 stderr.write( f". awaiting_llm_first_token: turn_id={event.turn_id} elapsed={secs:.1f}s\n" ) @@ -451,23 +492,31 @@ class CliPresenterState: async def _cancel_and_log( - client: httpx.AsyncClient, + client: WorldtreeClient, session_id: str, turn_id: int, *, stderr: TextIO, ) -> None: - """Spawn-and-forget cancel that never raises (INV-009).""" + """Spawn-and-forget cancel that never raises (INV-009). The adapter maps the + cancel races onto ratatoskr's typed exceptions; a benign late-cancel (200, + cancelled=False) returns a result and logs nothing.""" assert client is not None assert isinstance(turn_id, int) and turn_id > 0 try: - await cancel_turn(client, session_id, turn_id) - except (CancelFailed, CancelTurnNotFound, CancelAlreadyCompleted, httpx.RequestError) as exc: + await wt.cancel_turn(client, session_id, turn_id) + except ( + CancelFailed, + CancelTurnNotFound, + CancelAlreadyCompleted, + ConnectFailed, # SDK normalizes a transport drop to ConnectFailed(status=0) + httpx.RequestError, + ) as exc: stderr.write(f"[cancel_failed] {type(exc).__name__}: {exc}\n") async def _run_turn( - client: httpx.AsyncClient, + client: WorldtreeClient, session_id: str, content: str, sigint_event: asyncio.Event, @@ -493,7 +542,10 @@ async def _run_turn( cancelling = False sigint_task: asyncio.Task[bool] | None = None cancel_task: asyncio.Task[None] | None = None # strong ref to fire-and-forget cancel - aiter_obj = stream_turn_resilient(client, session_id, content).__aiter__() + # wt.stream_turn is an async generator — it is already its own iterator, so no + # explicit __aiter__(); keeping the concrete type lets __anext__() type as a + # coroutine for asyncio.create_task. + aiter_obj = wt.stream_turn(client, session_id, content) try: while True: @@ -537,17 +589,21 @@ async def _run_turn( except TurnIdFlip as exc: stderr.write(f"[turn_id_flip] expected={exc.established} got={exc.got}\n") return 22 - last_turn_id = event.sse_id.turn_id + # The cancel target is the turn from the composite sse_id (present on + # every frame); the body's turn_id is absent on text/thinking events. + tid = _turn_id_from_sse_id(event.sse_id) + if tid is not None: + last_turn_id = tid state.render(event, stdout=stdout, stderr=stderr) - if isinstance(event, Done): + if isinstance(event, DoneEvent): if sigint_task is not None and not cancelling: sigint_task.cancel() return 0 - if isinstance(event, Error): + if isinstance(event, ErrorEvent): if sigint_task is not None and not cancelling: sigint_task.cancel() return 2 - if isinstance(event, Cancelled): + if isinstance(event, CancelledEvent): if sigint_task is not None and not cancelling: sigint_task.cancel() return 3 @@ -564,6 +620,11 @@ async def _run_turn( async def _amain(args: ParsedArgs) -> int: """Async orchestrator: create-session (if --new) → SIGINT install → _run_turn → cleanup.""" assert isinstance(args, ParsedArgs) + # ratatoskr owns the transport (INV-CUT-1): the SDK is injected with it and + # never closes it. The transport carries base_url / User-Agent / timeout AND the + # default bearer — the SDK overrides Authorization per request (so a bound create + # still uses its consumer_key), while the not-yet-migrated hand-rolled + # `seed_preset_first_message` reuses the transport's default bearer directly. async with httpx.AsyncClient( base_url=args.server_url, headers={ @@ -575,7 +636,8 @@ async def _amain(args: ParsedArgs) -> int: # connect/write/pool keep modest timeouts so true network failures # still surface promptly. timeout=httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0), - ) as client: + ) as transport: + client = wt.build_client(args.server_url, api_key=args.api_key, transport=transport) if args.new: assert args.agent_id is not None try: @@ -586,7 +648,7 @@ async def _amain(args: ParsedArgs) -> int: if args.system_prompt is not None else None ) - info = await create_session( + info = await wt.create_session( client, args.agent_id, end_user_id=args.end_user_id, @@ -616,23 +678,32 @@ async def _amain(args: ParsedArgs) -> int: "(RATATOSKR_BIFROST_CONSUMER_KEY), not WORLDTREE_API_KEY\n" ) return 23 - except SessionApiFailed as exc: + except wt.SessionApiFailed as exc: sys.stderr.write(f"[session_api_failed] status={exc.status} body={exc.body!r}\n") return 20 - except (httpx.ConnectError, httpx.ReadTimeout, httpx.TransportError) as exc: + except ( + httpx.ConnectError, + httpx.ReadTimeout, + httpx.TransportError, + ConnectFailed, # SDK normalizes a pre-response transport failure here + ) as exc: sys.stderr.write(f"[network_error] {type(exc).__name__}: {exc}\n") return 21 # Issue #12: demoted lifecycle line — written directly here (NOT via - # state.render, which only accepts SSE Event variants per PRE-001). - kind_suffix = f" kind={info.kind}" if info.kind else "" + # state.render, which only accepts SSE Event variants per PRE-001). The + # adapter returns the SDK's open-world create dict; read it as a mapping. + session_id = info["session_id"] + kind = info.get("kind") + kind_suffix = f" kind={kind}" if kind else "" sys.stderr.write( - f". create_session: session_id={info.session_id} " - f"agent_id={info.agent_id}{kind_suffix}\n" + f". create_session: session_id={session_id} " + f"agent_id={info['agent_id']}{kind_suffix}\n" ) # #347 authored first-message: seed the agent's preset opening (best-effort). - if await seed_preset_first_message(client, info.session_id, args.agent_id): + # Uses the transport directly — first_message is a slice-3 hand-rolled path. + if await seed_preset_first_message(transport, session_id, args.agent_id): sys.stderr.write( - f". first_message: seeded preset opening for {info.agent_id}\n" + f". first_message: seeded preset opening for {args.agent_id}\n" ) # Issue #17 bound-state indicator: plane + endpoint + status, so the # operator sees WHICH identity/endpoint bound (not a bare boolean). @@ -642,7 +713,7 @@ async def _amain(args: ParsedArgs) -> int: f". bifrost: status=bound plane={plane} " f"endpoint={args.bifrost.endpoint_url}\n" ) - session_id = info.session_id + # session_id was bound above from the create dict. else: assert args.session_id is not None session_id = args.session_id diff --git a/src/ratatoskr/wt.py b/src/ratatoskr/wt.py index 469135e..b4d9803 100644 --- a/src/ratatoskr/wt.py +++ b/src/ratatoskr/wt.py @@ -28,7 +28,7 @@ carries the SDK's parsed `error_code`). from __future__ import annotations import json -from collections.abc import AsyncIterator, Mapping +from collections.abc import AsyncGenerator, Mapping from typing import Any import httpx @@ -81,6 +81,7 @@ def build_client( api_key: AuthProvider, admin_key: AuthProvider | None = None, transport: httpx.AsyncClient, + max_reconnects: int = 5, ) -> WorldtreeClient: """Construct the adapter's `WorldtreeClient` over a ratatoskr-owned transport. @@ -88,15 +89,18 @@ def build_client( `_owns_client=False`, so `WorldtreeClient.aclose()` never closes it — ratatoskr owns the lifecycle exactly as today (INV-CUT-1). ratatoskr's `api_key` / `admin_key` map to the SDK's per-request `auth` / `admin_auth` providers; the - injected transport carries ratatoskr's User-Agent / timeout (wired by the - caller in slice-2), NOT the Authorization header — the SDK adds auth per - request. + injected transport carries ratatoskr's User-Agent / timeout, and (transitionally) + the default bearer — the SDK adds auth per request, overriding it. + + `max_reconnects` is the resilient turn-stream's reconnect budget (SDK default 5); + pass 0 to surface a transport drop immediately without auto-resume. """ return WorldtreeClient( base_url, auth=api_key, admin_auth=admin_key, transport=transport, + max_reconnects=max_reconnects, ) @@ -249,7 +253,7 @@ async def get_session_tools( async def stream_turn( client: WorldtreeClient, session_id: str, content: str -) -> AsyncIterator[wtsdk.TurnEvent]: +) -> AsyncGenerator[wtsdk.TurnEvent, None]: """Drive the resilient turn stream (auto-resume; absorbs the old `reconnect_turn`) and yield the SDK's `TurnEvent`s, re-wrapping the stream's TERMINAL SDK errors into ratatoskr's caller-semantic exceptions (INV-CUT-2 / DEC-2 — the presenter diff --git a/tests/test_cli.py b/tests/test_cli.py index 166c181..14e211a 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -7,8 +7,10 @@ import json import httpx import pytest import respx +from worldtree_sdk.events import build_event from ratatoskr import cli as cli_mod +from ratatoskr import wt from ratatoskr.cli import ( ParsedArgs, UsageError, @@ -20,18 +22,7 @@ from ratatoskr.cli import ( main, ) from ratatoskr.sessions import BifrostBinding -from ratatoskr.sse_client import ( - Cancelled, - Done, - Error, - SseId, - Text, - TextBoundary, - Thinking, - ToolResult, - ToolStart, - WorkerPhase, -) +from ratatoskr.sse_client import SseId class _FlushCountingIO(io.StringIO): @@ -333,6 +324,83 @@ SID = SseId(42, 5) SID42 = SseId(42, 1) +# ── SDK-event factories ────────────────────────────────────────────────────── +# The presenter now consumes worldtree-sdk `TurnEvent`s. These build them exactly +# as the SDK's parser does (via `build_event` from the raw envelope), preserving +# the old dataclass call shapes so the render-test bodies stay unchanged. `sse_id` +# is a parsed `SseId` here purely to keep the terse SID42 idiom; the SDK carries the +# composite id as a string and turn_id top-level. +def _sid_str(sse_id: SseId) -> str: + return f"{sse_id.turn_id}:{sse_id.seq}" + + +def Thinking(*, sse_id: SseId, content: str) -> object: + return build_event("thinking", _sid_str(sse_id), sse_id.turn_id, {"content": content}) + + +def Text(*, sse_id: SseId, content: str) -> object: + return build_event("text", _sid_str(sse_id), sse_id.turn_id, {"content": content}) + + +def WorkerPhase(*, sse_id: SseId, phase: str, turn_id: int) -> object: + return build_event("worker_phase", _sid_str(sse_id), turn_id, {"phase": phase}) + + +def TextBoundary(*, sse_id: SseId, kind: str, char_offset: int, ts: str) -> object: + return build_event( + "text_boundary", _sid_str(sse_id), sse_id.turn_id, + {"kind": kind, "char_offset": char_offset, "ts": ts}, + ) + + +def ToolStart(*, sse_id: SseId, name: str, arguments: object) -> object: + return build_event( + "tool_start", _sid_str(sse_id), sse_id.turn_id, {"name": name, "arguments": arguments} + ) + + +def ToolResult(*, sse_id: SseId, name: str, result: object, duration_ms: int) -> object: + return build_event( + "tool_result", _sid_str(sse_id), sse_id.turn_id, + {"name": name, "result": result, "duration_ms": duration_ms}, + ) + + +def Done( + *, sse_id: SseId, phase: str, response: str, model: str, duration_ms: int, usage: object +) -> object: + return build_event( + "done", _sid_str(sse_id), sse_id.turn_id, + {"phase": phase, "response": response, "model": model, + "duration_ms": duration_ms, "usage": usage}, + ) + + +def Error(*, sse_id: SseId, phase: str, message: str, error_code: str) -> object: + return build_event( + "error", _sid_str(sse_id), sse_id.turn_id, + {"phase": phase, "message": message, "error_code": error_code}, + ) + + +def Cancelled( + *, sse_id: SseId, phase: str, turn_id: int, reason: object, partial_message_id: object +) -> object: + return build_event( + "cancelled", _sid_str(sse_id), turn_id, + {"phase": phase, "reason": reason, "partial_message_id": partial_message_id}, + ) + + +def _wtc(transport: httpx.AsyncClient) -> object: + """The adapter's WorldtreeClient over a respx-mocked transport. Reconnects are + disabled (max_reconnects=0) so a transport drop surfaces immediately instead of + burning the resilient retry budget with real backoff sleeps.""" + return wt.build_client( + "https://w.example", api_key="k", transport=transport, max_reconnects=0 + ) + + class TestCliPresenterState: """Tests for the new CliPresenterState — per issue #12 contract.""" @@ -687,7 +755,7 @@ _USAGE_ZERO: dict[str, int] = { } -def _make_done(*, duration_ms: int = 1, usage: dict[str, int] | None = None) -> Done: +def _make_done(*, duration_ms: int = 1, usage: dict[str, int] | None = None) -> object: return Done( sse_id=SID42, phase="succeeded", @@ -709,7 +777,8 @@ class TestCancelAndLog: ) ) stderr = io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) result = await _cancel_and_log(client, "s-1", 42, stderr=stderr) assert result is None assert stderr.getvalue() == "" @@ -721,7 +790,8 @@ class TestCancelAndLog: return_value=httpx.Response(500, content=b"boom") ) stderr = io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) await _cancel_and_log(client, "s-1", 42, stderr=stderr) out = stderr.getvalue() assert "[cancel_failed]" in out @@ -730,11 +800,14 @@ class TestCancelAndLog: @respx.mock async def test_cancel_already_completed(self) -> None: """cancel_already_completed [scenario]: …""" + # SDK gates the race on the (status, error_code) PAIR (B-CAN-3): 409 alone is + # a generic CancelFailed; 409 + turn_finished is the double-cancel race. respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( - return_value=httpx.Response(409) + return_value=httpx.Response(409, json={"error_code": "turn_finished"}) ) stderr = io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) await _cancel_and_log(client, "s-1", 42, stderr=stderr) out = stderr.getvalue() assert "[cancel_failed]" in out @@ -743,11 +816,13 @@ class TestCancelAndLog: @respx.mock async def test_cancel_turn_not_found(self) -> None: """cancel_turn_not_found [scenario]: 404 → returns None; stderr CancelTurnNotFound.""" + # 404 + turn_not_found is the benign "finished before cancel arrived" race. respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock( - return_value=httpx.Response(404) + return_value=httpx.Response(404, json={"error_code": "turn_not_found"}) ) stderr = io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) await _cancel_and_log(client, "s-1", 42, stderr=stderr) out = stderr.getvalue() assert "[cancel_failed]" in out @@ -760,11 +835,14 @@ class TestCancelAndLog: side_effect=httpx.ConnectError("network down") ) stderr = io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) await _cancel_and_log(client, "s-1", 42, stderr=stderr) out = stderr.getvalue() + # The SDK normalizes a transport drop to ConnectFailed(status=0); _cancel_and_log + # swallows it (INV-009) and logs the normalized type. assert "[cancel_failed]" in out - assert "ConnectError" in out + assert "ConnectFailed" in out class _GatedStream(httpx.AsyncByteStream): @@ -797,7 +875,8 @@ class TestRunTurn: sigint = asyncio.Event() stdout = io.StringIO() stderr = io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) assert exit_code == 0 assert stdout.getvalue() == "hello\n" @@ -820,7 +899,8 @@ class TestRunTurn: ) sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) assert exit_code == 2 assert "[error]" in stderr.getvalue() @@ -836,7 +916,8 @@ class TestRunTurn: ) sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) assert exit_code == 3 assert "[cancelled]" in stderr.getvalue() @@ -849,7 +930,8 @@ class TestRunTurn: ) sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) exit_code = await _run_turn( client, "missing", "hi", sigint, stdout=stdout, stderr=stderr ) @@ -882,7 +964,8 @@ class TestRunTurn: ) sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) assert exit_code == 21 assert "[connection_dropped]" in stderr.getvalue() @@ -896,7 +979,8 @@ class TestRunTurn: ) sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) assert exit_code == 22 assert "[malformed_sse_id]" in stderr.getvalue() @@ -913,7 +997,8 @@ class TestRunTurn: ) sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) assert exit_code == 22 out = stderr.getvalue() @@ -933,7 +1018,8 @@ class TestRunTurn: ) sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) assert exit_code == 22 out = stderr.getvalue() @@ -955,7 +1041,8 @@ class TestRunTurn: ) sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) assert exit_code == 22 out = stderr.getvalue() @@ -978,7 +1065,8 @@ class TestRunTurn: sigint = asyncio.Event() sigint.set() # SIGINT before _run_turn even starts stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) exit_code = await asyncio.wait_for( _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr), timeout=2.0 ) @@ -1007,7 +1095,8 @@ class TestRunTurn: sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) task = asyncio.create_task( _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) ) @@ -1045,7 +1134,8 @@ class TestRunTurn: sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) task = asyncio.create_task( _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) ) @@ -1091,7 +1181,8 @@ class TestRunTurn: monkeypatch.setattr(sigint, "wait", counting_wait) stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) task = asyncio.create_task( _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) ) @@ -1129,7 +1220,8 @@ class TestRunTurn: sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) task = asyncio.create_task( _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) ) @@ -1176,7 +1268,8 @@ class TestRunTurn: sigint = asyncio.Event() stdout, stderr = io.StringIO(), io.StringIO() - async with httpx.AsyncClient(base_url="https://w.example") as client: + async with httpx.AsyncClient(base_url="https://w.example") as _tp: + client = _wtc(_tp) exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr) assert exit_code == 0 assert call_count == 3 diff --git a/uv.lock b/uv.lock index c243697..bee1b9d 100644 --- a/uv.lock +++ b/uv.lock @@ -472,7 +472,7 @@ wheels = [ [[package]] name = "ratatoskr" -version = "0.21.5" +version = "0.21.6" source = { editable = "." } dependencies = [ { name = "httpx" },