vh/ratatoskr
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(#20): rewire the CLI turn path onto the wt adapter (slice-2, part 2b-i)

Browse Source 
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).
This commit is contained in:
Vuong Hoang
2026-07-19 06:07:50 -07:00
parent b907a7b8a5
commit e3a10ad80e
5 changed files with 272 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1 -1
View File
@@ -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"
src/ratatoskr/cli.py
+132 -61
View File
@@ -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
src/ratatoskr/wt.py
+9 -5
View File
@@ -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
tests/test_cli.py
+129 -36
View File
@@ -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
uv.lock
Generated
+1 -1
View File
@@ -472,7 +472,7 @@ wheels = [
[[package]]
name = "ratatoskr"
version = "0.21.5"
version = "0.21.6"
source = { editable = "." }
dependencies = [
{ name = "httpx" },