fix(#20): heid-code-review fixups — INV-CUT-2 completeness on cancel/stream (slice-2)

Triaged the heid-code-review panel (Gróa + Hulda substantive, Regin zero=weak).

Adopted (genuine adds):
- cancel_turn + stream_turn gain a defensive `except ApiError -> SessionApiFailed`
  default after their discriminated branches. INV-CUT-2 ("every ApiError is mapped;
  default SessionApiFailed") now holds STRUCTURALLY on those routes, not by coupling
  to the SDK's internal guarantee that it maps them to discriminated types. + tests.
- get_session_tools error-path test (symmetric with messages).
- Contract § Error map amended: added the stream ProtocolError rows
  (Malformed*/TurnIdFlip -> ratatoskr same-named), clarified the cancel row (the SDK
  RAISES the typed races -> ratatoskr exceptions, only a 200/cancelled=False is a
  CancelResult; caller surface stays exception-based per DEC-2), and noted the
  ApiError default holds on stream+cancel too.

Rejected (category-5, wrong-grounding) — 2/3 arms flagged create's bound-502 as
"should gate on error_code like list's 422+cursor_invalid". Verified against the SDK
parser (not in the arms' file set): the bound-502 body is
{"error_code":"bifrost_handshake_failed","detail":{"bifrost_error":...}}, and the
SDK's envelope parser PREFERS the nested detail (which lacks error_code), so
ApiError.error_code resolves to "unknown" — gating would REGRESS handshake detection
(the cli/web integration tests caught it). INV-002 also makes the handshake the sole
bound-502 cause. Kept the any-bound-502 mapping; documented WHY in code + contract.

Accepted-as-is: create_session -> Mapping annotation (intentional open-world
passthrough, already documented in the route-map note; category 3).

Suite 493 green; wt.py mypy + ruff clean. Patch.

docs/contracts/worldtree_sdk_cutover.contract.md

@@ -158,18 +158,23 @@ others, they get their own row here — the default is NOT a general "any 404
 | SDK `AgentNotAvailable` / `TurnLaunchUnavailable` / `SessionRetired` (stream-open) | ratatoskr `AgentNotAvailable` / `TurnLaunchUnavailable` / (retired → `SessionApiFailed`) — same names, passthrough |
 | SDK `ConnectionDropped` (mid-stream) | `SseConnectionDropped` |
 | SDK `ResumeError` subclasses (in resilient stream) | resilient `stream_turn` absorbs; terminal → `SseConnectFailed` |
-| SDK `Cancel*` (cancel_turn) | folded into `CancelResult`; late-cancel race (B-CAN-3) returns `cancelled=False`, never raises |
+| SDK `MalformedSseId` / `MalformedSseData` / `TurnIdFlip` (stream `ProtocolError`) | ratatoskr same-named types — same-name rewrap of the discriminated stream protocol errors |
+| SDK `Cancel*` (cancel_turn) — the SDK RAISES the typed races | 404 `turn_not_found` → `CancelTurnNotFound`; 409 `turn_finished` → `CancelAlreadyCompleted`; other `CancelError` → `CancelFailed`. A 200 (incl. `cancelled=False`, the B-CAN-3 late-cancel no-op) returns a `CancelResult` — never raises. The caller surface stays exception-based (DEC-2; matches the pre-cutover CLI/web handlers). |
 | `ApiError(404)` on `sessions.create` | `AgentNotFound` |
 | `ApiError(404)` on `sessions.write_history` | `AuthoredHistoryUnavailable` (hide-existence) |
-| `ApiError(422 cursor_invalid)` on `sessions.list` | `InvalidCursor` |
-| `ApiError(502 bifrost_handshake_failed)` on bound `sessions.create` | `BifrostHandshakeFailed` |
+| `ApiError(422 cursor_invalid)` on `sessions.list` | `InvalidCursor` (dual-key: status 422 AND error_code; the flat cursor body surfaces the code) |
+| `ApiError(502)` on bound `sessions.create` | `BifrostHandshakeFailed` — NOT gated on error_code (unlike list's 422): INV-002, the synchronous handshake is the SOLE bound-502 cause; and the SDK's envelope parser prefers the nested `detail` (which carries `bifrost_error`, not `error_code`), so no distinguishing top-level `error_code` surfaces. The route+status IS the discriminator. |
 | **`ApiError` (any other status/route) — the default** | `SessionApiFailed(status, error_code, body)` |
 
 The default row is load-bearing: any `ApiError` not matched above surfaces as the
 generic `SessionApiFailed` carrying the raw `status`/`error_code`/`body` — the
 adapter does NOT invent per-route semantics the contract doesn't list, and does NOT
-leave an `ApiError` un-mapped. Each slice adds/confirms its route's rows here before
-the old path is deleted.
+leave an `ApiError` un-mapped. **This default holds on EVERY route, including the
+stream and cancel** (each carries a defensive `except ApiError → SessionApiFailed`
+after its discriminated branches — the SDK maps those routes to discriminated types
+today, but the default guarantees INV-CUT-2 structurally, not by SDK-internal
+coupling). Each slice adds/confirms its route's rows here before the old path is
+deleted.
 
 ## Slice plan (incremental, DEC-4)
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "ratatoskr"
-version = "0.21.7"
+version = "0.21.9"
 description = "Worldtree Conversation API debug console (web + headless CLI) — multi-pane observability"
 readme = "README.md"
 requires-python = ">=3.12"

src/ratatoskr/sessions.py

@@ -38,12 +38,6 @@ class SessionInfo:
     config: dict[str, Any] | None = None
 
 
-@dataclass(frozen=True)
-class SessionPage:
-    """One page of GET /sessions results. `next_cursor=None` on the last page."""
-
-    items: list[SessionInfo]
-    next_cursor: str | None
 
 
 @dataclass(frozen=True)
@@ -204,53 +198,6 @@ class AuthoredHistoryUnavailable(Exception):
         self.session_id = session_id
 
 
-async def list_sessions(
-    client: httpx.AsyncClient,
-    *,
-    include_archived: bool = False,
-    limit: int = 50,
-    cursor: str | None = None,
-) -> SessionPage:
-    """GET /sessions. See contract FN list_sessions."""
-    assert client is not None
-    assert 1 <= limit <= 200
-    assert cursor is None or (isinstance(cursor, str) and cursor)
-
-    params: dict[str, str] = {"limit": str(limit)}
-    if include_archived:
-        params["include_archived"] = "true"
-    if cursor is not None:
-        params["cursor"] = cursor
-
-    resp = await client.get("/sessions", params=params)
-    if resp.status_code == 422:
-        try:
-            err = resp.json()
-        except ValueError:
-            err = {}
-        if err.get("error_code") == "cursor_invalid":
-            raise InvalidCursor(raw=cursor)
-        raise SessionApiFailed(status=422, body=resp.content)
-    if resp.status_code != 200:
-        raise SessionApiFailed(status=resp.status_code, body=resp.content)
-    body = resp.json()
-    items = [
-        SessionInfo(
-            session_id=item["session_id"],
-            agent_id=item["agent_id"],
-            created_at=item["created_at"],
-            last_active=item["last_active"],
-            metadata=item.get("metadata", {}),
-            message_count=None,
-            name=item.get("name"),
-            archived=item.get("archived") or False,
-            tags=item.get("tags") or [],
-            kind=item.get("kind"),      # INV-002 amendment (#161): present on list items
-            config=item.get("config"),  # forward-compat passthrough; None today
-        )
-        for item in body["items"]
-    ]
-    return SessionPage(items=items, next_cursor=body.get("next_cursor"))
 
 
 def endpoint_for_plane(plane: str, base_host: str) -> str:
@@ -572,22 +519,6 @@ async def get_session_bifrost(
     raise SessionApiFailed(status=resp.status_code, body=resp.content)
 
 
-async def get_session_tools(client: httpx.AsyncClient, session_id: str) -> dict[str, Any]:
-    """GET /sessions/{session_id}/tools — owner-scoped tool inventory (spec #183).
-
-    Returns the merged tool list the LLM saw at turn-fire: `{agent_id,
-    builtin_tools: [...], bifrost_tools: [{name, description, parameters}, ...]}`.
-    Owner-scoped (`ctx.user_id == session.user_id`) — reachable with the consumer
-    key, NO admin scope. Cross-owner access returns 404 `session_not_found`
-    (existence-hiding); a revoked session returns 401 `auth_revoked`. Parsed dict
-    verbatim; any non-200 → SessionApiFailed (mirrors get_persona_state).
-    """
-    assert client is not None
-    assert session_id and isinstance(session_id, str)
-    resp = await client.get(f"/sessions/{session_id}/tools")
-    if resp.status_code == 200:
-        return resp.json()
-    raise SessionApiFailed(status=resp.status_code, body=resp.content)
 
 
 async def get_capabilities(client: httpx.AsyncClient) -> dict[str, Any]:

src/ratatoskr/sse_client.py

@@ -6,7 +6,6 @@ Implements docs/contracts/issues/1.contract.md.
 from __future__ import annotations
 
 import json
-import re
 from collections.abc import AsyncIterator
 from dataclasses import dataclass
 from typing import Any, NamedTuple
@@ -14,8 +13,6 @@ from typing import Any, NamedTuple
 import httpx
 import httpx_sse
 
-_INT_RE = re.compile(r"^-?\d+$")
-
 
 class SseId(NamedTuple):
     """Parsed composite SSE wire `id:` per spec §SSE id format."""
@@ -24,155 +21,6 @@ class SseId(NamedTuple):
     seq: int
 
 
-@dataclass(frozen=True)
-class WorkerPhase:
-    """SSE event `worker_phase`: agent entered a new processing phase."""
-
-    sse_id: SseId
-    phase: str
-    turn_id: int
-
-
-@dataclass(frozen=True)
-class Thinking:
-    """SSE event `thinking`: incremental thinking content from thinking-enabled models."""
-
-    sse_id: SseId
-    content: str
-
-
-@dataclass(frozen=True)
-class Text:
-    """SSE event `text`: an incremental response-text delta."""
-
-    sse_id: SseId
-    content: str
-
-
-@dataclass(frozen=True)
-class TextBoundary:
-    """SSE event `text_boundary`: speakable breakpoint after a `text` event."""
-
-    sse_id: SseId
-    kind: str
-    char_offset: int
-    ts: str
-
-
-@dataclass(frozen=True)
-class ToolStart:
-    """SSE event `tool_start`: agent is about to execute a tool."""
-
-    sse_id: SseId
-    name: str
-    arguments: dict[str, Any]
-
-
-@dataclass(frozen=True)
-class ToolResult:
-    """SSE event `tool_result`: a tool call completed."""
-
-    sse_id: SseId
-    name: str
-    result: Any
-    duration_ms: int
-
-
-@dataclass(frozen=True)
-class Done:
-    """Terminal SSE event `done`: turn succeeded."""
-
-    sse_id: SseId
-    phase: str
-    response: str
-    model: str
-    duration_ms: int
-    usage: dict[str, int]
-
-
-@dataclass(frozen=True)
-class Error:
-    """Terminal SSE event `error`: turn failed."""
-
-    sse_id: SseId
-    phase: str
-    message: str
-    error_code: str | None
-
-
-@dataclass(frozen=True)
-class Cancelled:
-    """Terminal SSE event `cancelled`: turn was cancelled server-side."""
-
-    sse_id: SseId
-    phase: str
-    turn_id: int
-    reason: str | None
-    partial_message_id: int | None
-
-
-@dataclass(frozen=True)
-class AwaitingLlmFirstToken:
-    """SSE event `awaiting_llm_first_token`: heartbeat during slow first-token.
-
-    Fires at the configured interval (default 5s) during the gap between
-    `worker_phase` phase=BuildingPrompt and phase=CallingLLM. Lets clients
-    render a live "thinking for Ns…" indicator instead of a frozen line
-    during legitimate-slow first-token latency. Stops the moment CallingLLM
-    fires (defense-in-depth at three sites); no heartbeat after Cancelled
-    or stalled terminal events. Tool round-trip re-entries do NOT re-fire
-    heartbeats — INV-201-5 scopes the mechanism to the FIRST gap only.
-
-    `elapsed_ms_since_building_prompt` is server-authoritative
-    `time.monotonic()`-based — independent of network latency or clock
-    skew, monotonically increasing across the heartbeat sequence.
-
-    See docs/conversation-api-spec.md § awaiting_llm_first_token
-    (Worldtree #201, v0.29.0).
-    """
-
-    sse_id: SseId
-    turn_id: int
-    elapsed_ms_since_building_prompt: float
-
-
-@dataclass(frozen=True)
-class AffectUpdate:
-    """SSE event `affect_update`: persona-state observability snapshot.
-
-    Two emissions per qualifying turn (persona-enabled agent on non-
-    ephemeral session): `status="current"` at turn start carrying the full
-    snapshot, `status="scheduled"` after post-turn appraisal kicks off
-    (lightweight — `snapshot` is None). Suppressed entirely for persona-
-    disabled agents (e.g. `domari`, `muninn`), Tier 3 consumer-defined
-    agents (Phase 2.0), and ephemeral sessions.
-
-    Bootstrap reads available via `GET /agents/{agent_id}/persona_state`
-    (same `snapshot` shape, requires `persona.read` scope).
-
-    See docs/conversation-api-spec.md § affect_update (Worldtree #204,
-    v0.28.0).
-    """
-
-    sse_id: SseId
-    status: str  # "current" | "scheduled"
-    turn_id: int
-    snapshot: dict[str, Any] | None  # None when status="scheduled"
-
-
-Event = (
-    WorkerPhase
-    | Thinking
-    | Text
-    | TextBoundary
-    | ToolStart
-    | ToolResult
-    | Done
-    | Error
-    | Cancelled
-    | AffectUpdate
-    | AwaitingLlmFirstToken
-)
 
 
 @dataclass(frozen=True)
@@ -258,31 +106,6 @@ class TurnLaunchUnavailable(SseConnectFailed):
         self.message = message
 
 
-# Canonical error_codes (Worldtree #331 / v1.0.0b2): 409 -> agent_not_available,
-# 503 -> not_ready (retryable; re-pinned from internal_error). Used only as a
-# fallback default when the body omits error_code — the real code is surfaced
-# verbatim from the {detail:{error_code,message}} envelope.
-_EAGER_TURN_FAILURE_CODE = {409: "agent_not_available", 503: "not_ready"}
-
-
-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):
@@ -352,275 +175,6 @@ class CancelFailed(Exception):
         self.body = body
 
 
-@dataclass(frozen=True)
-class CancelResult:
-    """Response envelope from POST /sessions/{id}/turns/{turn_id}/cancel."""
-
-    turn_id: int
-    cancelled: bool
-    reason: str | None
-    partial_message_id: int | None
-
-
-def _envelope_for_type(body: dict[str, Any], sse_id: SseId) -> Event:
-    """Dispatch a parsed JSON body to its typed Event variant."""
-    t = body["type"]
-    if t == "text":
-        return Text(sse_id=sse_id, content=body["content"])
-    if t == "worker_phase":
-        return WorkerPhase(sse_id=sse_id, phase=body["phase"], turn_id=body["turn_id"])
-    if t == "thinking":
-        return Thinking(sse_id=sse_id, content=body["content"])
-    if t == "text_boundary":
-        return TextBoundary(
-            sse_id=sse_id,
-            kind=body["kind"],
-            char_offset=body["char_offset"],
-            ts=body["ts"],
-        )
-    if t == "tool_start":
-        return ToolStart(sse_id=sse_id, name=body["name"], arguments=body["arguments"])
-    if t == "tool_result":
-        return ToolResult(
-            sse_id=sse_id,
-            name=body["name"],
-            result=body["result"],
-            duration_ms=body["duration_ms"],
-        )
-    if t == "done":
-        return Done(
-            sse_id=sse_id,
-            phase=body["phase"],
-            response=body["response"],
-            model=body["model"],
-            duration_ms=body["duration_ms"],
-            usage=body["usage"],
-        )
-    if t == "error":
-        return Error(
-            sse_id=sse_id,
-            phase=body.get("phase", "failed"),
-            message=body.get("message", ""),
-            error_code=body.get("error_code"),
-        )
-    if t == "cancelled":
-        return Cancelled(
-            sse_id=sse_id,
-            phase=body["phase"],
-            turn_id=body["turn_id"],
-            reason=body.get("reason"),
-            partial_message_id=body.get("partial_message_id"),
-        )
-    if t == "awaiting_llm_first_token":
-        # Worldtree #201 / v0.29.0: top-level heartbeat during BuildingPrompt
-        # → CallingLLM gap. Lets clients render live elapsed-time indicators
-        # instead of frozen lines on legitimate-slow first-token latency.
-        return AwaitingLlmFirstToken(
-            sse_id=sse_id,
-            turn_id=body["turn_id"],
-            elapsed_ms_since_building_prompt=body["elapsed_ms_since_building_prompt"],
-        )
-    if t == "affect_update":
-        # Worldtree #204 / v0.28.0: persona-state observability event.
-        # status="current" carries full snapshot at turn start;
-        # status="scheduled" omits snapshot (lightweight post-appraisal-
-        # kickoff notification).
-        return AffectUpdate(
-            sse_id=sse_id,
-            status=body["status"],
-            turn_id=body["turn_id"],
-            snapshot=body.get("snapshot"),
-        )
-    raise ValueError(f"unknown SSE event type: {t!r}")
-
-
-async def _iter_events(
-    event_source: httpx_sse.EventSource,
-    *,
-    expected_turn_id: int | None,
-) -> AsyncIterator[Event]:
-    """Apply INV-002 (sse_id present + in range) and INV-003 (turn_id stable) per event.
-
-    `expected_turn_id=None` means "establish from the first event" (stream_turn semantics).
-    `expected_turn_id=N` means "every event must match N" (reconnect_turn semantics — the
-    first event is already a flip-candidate per INV-003).
-    """
-    established = expected_turn_id
-    last_sse_id: SseId | None = None
-    terminal_seen = False
-    try:
-        async for sse in event_source.aiter_sse():
-            # Issue #7 INV-001: empty-data frames are keepalives — skip silently.
-            # ORDERING: this branch fires BEFORE _parse_sse_id; an empty-data event
-            # with a malformed id is silently swallowed (intentional — a keepalive
-            # with a bad id is still a keepalive). Don't reorder.
-            if sse.data == "":
-                continue
-            # v0.8.1: empty-id frames are also treated as keepalives. Worldtree
-            # SOMETIMES emits events without an `id:` line (observed mid-stream
-            # on the qwen3.6-35-a3b-heretic provider, 2026-05-25). Per the SSE
-            # RFC, events without ids are legitimate (they just don't update
-            # Last-Event-ID); the previous strict behavior crashed every turn
-            # on the offending agent. Treat same as empty-data: skip silently.
-            if sse.id == "":
-                continue
-            try:
-                sse_id = _parse_sse_id(sse.id)
-            except ValueError as exc:
-                raise MalformedSseId(raw=sse.id) from exc
-            if established is None:
-                established = sse_id.turn_id
-            elif sse_id.turn_id != established:
-                raise TurnIdFlip(established=established, got=sse_id.turn_id)
-            try:
-                body = json.loads(sse.data)
-            except json.JSONDecodeError as exc:
-                raise MalformedSseData(raw=sse.data) from exc
-            event = _envelope_for_type(body, sse_id=sse_id)
-            yield event
-            last_sse_id = sse_id
-            if isinstance(event, (Done, Error, Cancelled)):
-                terminal_seen = True
-                return
-    except (httpx.ReadError, httpx.RemoteProtocolError, httpx.ReadTimeout) as exc:
-        # ReadTimeout covers idle gaps that exceed httpx's read timeout — the SSE
-        # stream went quiet long enough for httpx to give up. Treat the same as a
-        # raw read error: surface as SseConnectionDropped so the caller can decide
-        # whether to reconnect_turn. (Callers SHOULD configure a long-or-disabled
-        # read timeout on their AsyncClient for SSE; this is defense in depth.)
-        raise SseConnectionDropped(last_seen_sse_id=last_sse_id) from exc
-    if not terminal_seen:
-        # Clean EOF before terminal event — INV-001 says stream MUST NOT end
-        # without exactly one Done/Error/Cancelled. Surface as connection drop;
-        # caller may reconnect_turn if it holds last_sse_id.
-        raise SseConnectionDropped(last_seen_sse_id=last_sse_id)
-
-
-async def stream_turn(
-    client: httpx.AsyncClient, session_id: str, content: str
-) -> AsyncIterator[Event]:
-    """POST a message and yield typed Events. See contract FN stream_turn."""
-    assert client is not None
-    assert session_id and isinstance(session_id, str)
-    assert content and isinstance(content, str)
-
-    async with httpx_sse.aconnect_sse(
-        client,
-        "POST",
-        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:
-            body = await exc.response.aread()
-            raise SseConnectFailed(status=exc.response.status_code, body=body) from exc
-        async for event in _iter_events(event_source, expected_turn_id=None):
-            yield event
-
-
-async def reconnect_turn(
-    client: httpx.AsyncClient,
-    session_id: str,
-    content: str,
-    last_event_id: str,
-) -> AsyncIterator[Event]:
-    """Re-POST with Last-Event-ID to resume. See contract FN reconnect_turn."""
-    assert client is not None
-    assert session_id and isinstance(session_id, str)
-    assert isinstance(content, str)
-    expected = _parse_sse_id(last_event_id)
-
-    async with httpx_sse.aconnect_sse(
-        client,
-        "POST",
-        f"/sessions/{session_id}/messages",
-        json={"content": content},
-        headers={"Last-Event-ID": last_event_id},
-    ) as event_source:
-        status = event_source.response.status_code
-        if status != 200:
-            body_bytes = await event_source.response.aread()
-            try:
-                body = json.loads(body_bytes)
-            except json.JSONDecodeError:
-                body = {}
-            if status == 400:
-                raise InvalidLastEventId(raw=last_event_id)
-            if status == 410:
-                raise ResumeTurnFinished(turn_id=body.get("turn_id", expected.turn_id))
-            if status == 412:
-                raise ResumeBufferExpired(
-                    turn_id=body.get("turn_id", expected.turn_id),
-                    buffered_from_seq=body.get("buffered_from_seq", 0),
-                )
-            raise SseConnectFailed(status=status, body=body_bytes)
-        async for event in _iter_events(event_source, expected_turn_id=expected.turn_id):
-            yield event
-
-
-async def stream_turn_resilient(
-    client: httpx.AsyncClient,
-    session_id: str,
-    content: str,
-    *,
-    max_reconnects: int = 5,
-) -> AsyncIterator[Event]:
-    """Resume-orchestration wrapper over stream_turn + reconnect_turn.
-
-    Yields ONE continuous Event stream; on `SseConnectionDropped` (mid-stream
-    drop or clean EOF before a terminal), resumes from the last-seen `sse_id`
-    via `reconnect_turn`, up to `max_reconnects` times, until a terminal
-    Done/Error/Cancelled arrives. The single shared surface presenters consume
-    for resilient streaming (design-brief §8b: "share the consumer, branch the
-    presenter"). Cross-process resume stays deferred to v2 (§8d): `last_seen`
-    lives only in this generator's frame. See contract FN stream_turn_resilient
-    (amendment 2026-06-30).
-    """
-    assert client is not None
-    assert session_id and isinstance(session_id, str)
-    assert content and isinstance(content, str)
-    assert isinstance(max_reconnects, int) and max_reconnects >= 0
-
-    last_seen: SseId | None = None
-    reconnects = 0
-    gen = stream_turn(client, session_id, content)
-    while True:
-        try:
-            async for event in gen:
-                last_seen = event.sse_id
-                yield event
-            return  # generator completed cleanly → terminal event reached (INV-001)
-        except SseConnectionDropped as drop:
-            # Prefer the id we tracked from a yielded event; fall back to the one
-            # the drop carries (covers a drop on the very first frame). Non-drop
-            # reconnect failures (412/410/400/flip) are NOT caught here — they
-            # propagate per the contract's "surface, not recover" policy.
-            seen = last_seen or drop.last_seen_sse_id
-            if seen is None or reconnects >= max_reconnects:
-                raise
-            reconnects += 1
-            gen = reconnect_turn(
-                client,
-                session_id,
-                content,
-                # A str cursor is already the composite id; an SseId is formatted.
-                last_event_id=seen if isinstance(seen, str) else f"{seen.turn_id}:{seen.seq}",
-            )
-
 
 async def stream_admin_events(
     client: httpx.AsyncClient,
@@ -667,48 +221,3 @@ async def stream_admin_events(
             raise SseConnectionDropped(last_seen_sse_id=None) from exc
 
 
-def _parse_sse_id(raw: str) -> SseId:
-    """Parse the SSE wire `id:` as composite `{turn_id}:{seq}`. See contract FN _parse_sse_id."""
-    assert isinstance(raw, str)
-    parts = raw.split(":")
-    if len(parts) != 2:
-        raise ValueError(f"expected '{{turn_id}}:{{seq}}', got: {raw[:64]!r}")
-    turn_id_str, seq_str = parts
-    if not _INT_RE.match(turn_id_str) or not _INT_RE.match(seq_str):
-        raise ValueError(f"expected '{{turn_id}}:{{seq}}' with decimal ints, got: {raw[:64]!r}")
-    turn_id = int(turn_id_str)
-    seq = int(seq_str)
-    if turn_id < 1 or seq < 1:
-        raise ValueError(f"expected both ints >= 1 per spec, got: {raw[:64]!r}")
-    return SseId(turn_id=turn_id, seq=seq)
-
-
-async def cancel_turn(
-    client: httpx.AsyncClient,
-    session_id: str,
-    turn_id: int,
-    *,
-    persist_partial: bool = False,
-) -> CancelResult:
-    """POST /sessions/{id}/turns/{turn_id}/cancel. See contract FN cancel_turn."""
-    assert client is not None
-    assert session_id and isinstance(session_id, str)
-    assert isinstance(turn_id, int) and turn_id > 0
-
-    params = {"persist_partial": "true"} if persist_partial else None
-    resp = await client.post(
-        f"/sessions/{session_id}/turns/{turn_id}/cancel", params=params
-    )
-    if resp.status_code == 404:
-        raise CancelTurnNotFound(turn_id=turn_id)
-    if resp.status_code == 409:
-        raise CancelAlreadyCompleted(turn_id=turn_id)
-    if resp.status_code != 200:
-        raise CancelFailed(status=resp.status_code, body=resp.content)
-    body = resp.json()
-    return CancelResult(
-        turn_id=body["turn_id"],
-        cancelled=body["cancelled"],
-        reason=body.get("reason"),
-        partial_message_id=body.get("partial_message_id"),
-    )

src/ratatoskr/wt.py

@@ -197,6 +197,12 @@ async def create_session(
     except ApiError as exc:
         if exc.status == 404:
             raise AgentNotFound(agent_id=agent_id) from exc
+        # NOT gated on error_code (unlike list's 422+cursor_invalid): INV-002 — a 502
+        # on a BOUND create IS the synchronous Bifrost handshake failing, the sole
+        # bound-502 cause; and the SDK does not surface a distinguishing top-level
+        # error_code here (its envelope parser prefers the nested `detail`, which
+        # carries `bifrost_error`, not `error_code`). The nested bifrost_error is
+        # extracted for the exception; the route+status is the discriminator.
         if bifrost is not None and exc.status == 502:
             raise BifrostHandshakeFailed(
                 bifrost_error=_bifrost_error_from_body(exc.body),
@@ -299,6 +305,10 @@ async def stream_turn(
         raise MalformedSseData(raw=exc.raw) from exc
     except wtsdk.TurnIdFlip as exc:
         raise TurnIdFlip(established=exc.established, got=exc.got) from exc
+    except ApiError as exc:
+        # INV-CUT-2 default: an undiscriminated ApiError surfacing from the stream →
+        # SessionApiFailed (the discriminated stream errors are handled above).
+        raise translate_error(exc) from exc
 
 
 async def cancel_turn(
@@ -323,3 +333,6 @@ async def cancel_turn(
         raise CancelFailed(
             status=0, body=(getattr(exc, "message", "") or str(exc)).encode()
         ) from exc
+    except ApiError as exc:
+        # INV-CUT-2 default: an undiscriminated ApiError on this route → SessionApiFailed.
+        raise translate_error(exc) from exc

tests/test_sessions.py

@@ -13,10 +13,8 @@ from ratatoskr.sessions import (
     BifrostBinding,
     BifrostConsumerKeyMissing,
     BifrostHandshakeFailed,
-    InvalidCursor,
     PersonaNotConfigured,
     SessionApiFailed,
-    SessionPage,
     create_character,
     create_session,
     delete_character,
@@ -27,10 +25,8 @@ from ratatoskr.sessions import (
     get_persona_state,
     get_session_bifrost,
     get_session_messages,
-    get_session_tools,
     list_agents,
     list_character_models,
-    list_sessions,
     set_persona_state,
     write_authored_history,
 )
@@ -551,223 +547,6 @@ class TestEndpointForPlane:
             endpoint_for_plane("persona", "10.100.10.50")
 
 
-def _list_item(
-    *,
-    session_id: str = "s1",
-    agent_id: str = "mimir",
-    created_at: str = "2026-04-15T12:00:00+00:00",
-    last_active: str = "2026-04-15T12:05:00+00:00",
-    metadata: dict[str, object] | None = None,
-    name: str | None = "Research session",
-    archived: bool = False,
-    tags: list[str] | None = None,
-) -> dict[str, object]:
-    """Build a GET /sessions list-item body for tests."""
-    item: dict[str, object] = {
-        "session_id": session_id,
-        "agent_id": agent_id,
-        "created_at": created_at,
-        "last_active": last_active,
-        "metadata": metadata if metadata is not None else {},
-        "name": name,
-        "archived": archived,
-        "tags": tags if tags is not None else ["work"],
-    }
-    return item
-
-
-class TestListSessions:
-    @respx.mock
-    async def test_explicit_null_list_defaults(self) -> None:
-        """INV-002: explicit-null archived -> False; explicit-null tags -> []."""
-        raw_item = {
-            "session_id": "s1",
-            "agent_id": "mimir",
-            "created_at": "2026-04-15T12:00:00+00:00",
-            "last_active": "2026-04-15T12:05:00+00:00",
-            "metadata": {},
-            "name": None,
-            "archived": None,
-            "tags": None,
-        }
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200, json={"items": [raw_item], "next_cursor": None}
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            page = await list_sessions(client)
-        info = page.items[0]
-        assert info.archived is False, "explicit-null archived must default to False"
-        assert info.tags == [], "explicit-null tags must default to []"
-        assert info.name is None
-
-    @respx.mock
-    async def test_happy_first_page(self) -> None:
-        """happy_first_page [happy,tracer]: one item + next_cursor -> SessionPage shape."""
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200,
-                json={
-                    "items": [_list_item()],
-                    "next_cursor": "v1.eyJhYmMifQ",
-                },
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            page = await list_sessions(client)
-        assert isinstance(page, SessionPage)
-        assert len(page.items) == 1
-        assert page.next_cursor == "v1.eyJhYmMifQ"
-        info = page.items[0]
-        assert info.session_id == "s1"
-        assert info.message_count is None  # INV-002: not in list response
-        assert info.name == "Research session"
-        assert info.archived is False
-        assert info.tags == ["work"]
-
-    @respx.mock
-    async def test_happy_last_page(self) -> None:
-        """happy_last_page: next_cursor=null -> SessionPage(next_cursor=None)."""
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200,
-                json={"items": [_list_item()], "next_cursor": None},
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            page = await list_sessions(client)
-        assert page.next_cursor is None
-
-    @respx.mock
-    async def test_empty_results(self) -> None:
-        """empty_results: {items: [], next_cursor: null} -> SessionPage([], None)."""
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            page = await list_sessions(client)
-        assert page == SessionPage(items=[], next_cursor=None)
-
-    @respx.mock
-    async def test_include_archived_query(self) -> None:
-        """include_archived_query: True -> has param; default -> NO param at all."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200, json={"items": [], "next_cursor": None}
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            await list_sessions(client, include_archived=True)
-            await list_sessions(client)  # default
-        url_with = str(route.calls[0].request.url)
-        url_default = str(route.calls[1].request.url)
-        assert "include_archived=true" in url_with
-        assert "include_archived" not in url_default
-
-    @respx.mock
-    async def test_cursor_threaded(self) -> None:
-        """cursor_threaded: cursor=opaque -> URL has cursor=opaque."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200, json={"items": [], "next_cursor": None}
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            await list_sessions(client, cursor="opaque-from-prev-page")
-        assert "cursor=opaque-from-prev-page" in str(route.calls[0].request.url)
-
-    @respx.mock
-    async def test_limit_query(self) -> None:
-        """limit_query: limit=10 -> URL has limit=10."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                200, json={"items": [], "next_cursor": None}
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            await list_sessions(client, limit=10)
-        assert "limit=10" in str(route.calls[0].request.url)
-
-    @respx.mock
-    async def test_invalid_cursor_server(self) -> None:
-        """invalid_cursor_server: 422 cursor_invalid -> InvalidCursor(raw=<passed cursor>)."""
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                422,
-                json={"error_code": "cursor_invalid", "message": "bad cursor"},
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(InvalidCursor) as exc_info:
-                await list_sessions(client, cursor="bogus")
-        assert exc_info.value.raw == "bogus"
-
-    @respx.mock
-    async def test_other_validation_failed(self) -> None:
-        """other_validation_failed: 422 other error_code -> SessionApiFailed(422); truncated."""
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(
-                422,
-                json={"error_code": "validation_failed", "message": "limit out of range"},
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(SessionApiFailed) as exc_info:
-                await list_sessions(client)
-        assert exc_info.value.status == 422
-        assert len(exc_info.value.body) <= 1024
-
-    @respx.mock
-    async def test_unexpected_status_truncates(self) -> None:
-        """unexpected_status_truncates: 500 + 5000-byte body -> SessionApiFailed; body == 1024."""
-        big = b"x" * 5000
-        respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(500, content=big)
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(SessionApiFailed) as exc_info:
-                await list_sessions(client)
-        assert exc_info.value.status == 500
-        assert exc_info.value.body == big[:1024]
-
-    @respx.mock
-    async def test_limit_below_one(self) -> None:
-        """limit_below_one [adversarial]: limit=0 -> AssertionError; no HTTP."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(AssertionError):
-                await list_sessions(client, limit=0)
-        assert route.call_count == 0
-
-    @respx.mock
-    async def test_limit_above_max(self) -> None:
-        """limit_above_max [adversarial]: limit=300 -> AssertionError; no HTTP."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(AssertionError):
-                await list_sessions(client, limit=300)
-        assert route.call_count == 0
-
-    @respx.mock
-    async def test_empty_cursor(self) -> None:
-        """empty_cursor [adversarial]: cursor='' -> AssertionError; no HTTP."""
-        route = respx.get("https://w.example/sessions").mock(
-            return_value=httpx.Response(200, json={"items": [], "next_cursor": None})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(AssertionError):
-                await list_sessions(client, cursor="")
-        assert route.call_count == 0
-
-
-# ---- Issue #8: list_agents + AgentInfo --------------------------------------
-
-
 class TestListAgents:
     @respx.mock
     async def test_happy_full_shape(self) -> None:
@@ -1145,53 +924,6 @@ class TestGetCapabilities:
         assert exc.value.status == 500
 
 
-class TestGetSessionTools:
-    """docs/contracts/issues/2.contract.md — get_session_tools (GET /sessions/{id}/tools, #183)."""
-
-    @respx.mock
-    async def test_happy(self) -> None:
-        """happy [happy,tracer]: 200 → merged tool inventory dict verbatim."""
-        respx.get("https://w.example/sessions/s1/tools").mock(
-            return_value=httpx.Response(
-                200,
-                json={
-                    "agent_id": "alice:wizard",
-                    "builtin_tools": [],
-                    "bifrost_tools": [
-                        {"name": "bifrost.alice.set_field", "description": "d", "parameters": {}}
-                    ],
-                },
-            )
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            tools = await get_session_tools(client, "s1")
-        assert tools["agent_id"] == "alice:wizard"
-        assert tools["builtin_tools"] == []
-        assert tools["bifrost_tools"][0]["name"] == "bifrost.alice.set_field"
-
-    @respx.mock
-    async def test_cross_owner_404_raises(self) -> None:
-        """cross_owner_404 [error]: 404 session_not_found → SessionApiFailed(404)."""
-        respx.get("https://w.example/sessions/s1/tools").mock(
-            return_value=httpx.Response(404, json={"error_code": "session_not_found"})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(SessionApiFailed) as exc:
-                await get_session_tools(client, "s1")
-        assert exc.value.status == 404
-
-    @respx.mock
-    async def test_empty_session_id_asserts(self) -> None:
-        """empty_session_id [adversarial]: '' → AssertionError; no HTTP issued."""
-        route = respx.get("https://w.example/sessions//tools").mock(
-            return_value=httpx.Response(200, json={})
-        )
-        async with httpx.AsyncClient(base_url="https://w.example") as client:
-            with pytest.raises(AssertionError):
-                await get_session_tools(client, "")
-        assert route.call_count == 0
-
-
 class TestGetSessionBifrost:
     """#2 contract — get_session_bifrost (GET /admin/sessions/{id}/bifrost, #176)."""
 

tests/test_wt.py

@@ -299,6 +299,12 @@ class TestReadPassthroughs:
             await get_session_messages(_wt(fake), "s")
         assert ei.value.status == 401
 
+    async def test_tools_error_maps_to_session_api_failed(self) -> None:
+        fake = _FakeSessions(error=ApiError("auth_revoked", "no", status=401))
+        with pytest.raises(SessionApiFailed) as ei:
+            await get_session_tools(_wt(fake), "s")
+        assert ei.value.status == 401
+
 
 async def _drain(aiter: Any) -> list[Any]:
     out: list[Any] = []
@@ -376,6 +382,14 @@ class TestStreamTurn:
             await _drain(stream_turn(_wt(fake), "s", "hi"))
         assert (ei.value.established, ei.value.got) == (5, 7)
 
+    async def test_undiscriminated_api_error_maps_to_session_api_failed(self) -> None:
+        # INV-CUT-2 default: an undiscriminated ApiError surfacing from the stream
+        # (not a discriminated stream error) → SessionApiFailed.
+        fake = _FakeSessions(stream_error=ApiError("weird", "boom", status=500))
+        with pytest.raises(SessionApiFailed) as ei:
+            await _drain(stream_turn(_wt(fake), "s", "hi"))
+        assert ei.value.status == 500
+
 
 class TestCancelTurn:
     async def test_happy_returns_cancel_result(self) -> None:
@@ -410,3 +424,11 @@ class TestCancelTurn:
         fake = _FakeSessions(error=wtsdk.CancelFailed(42, error_code="boom", message="failed"))
         with pytest.raises(CancelFailed):
             await cancel_turn(_wt(fake), "s", 42)
+
+    async def test_undiscriminated_api_error_maps_to_session_api_failed(self) -> None:
+        # INV-CUT-2 default: an undiscriminated ApiError on the cancel route (not a
+        # typed Cancel* race) → SessionApiFailed, never leaked as a bare ApiError.
+        fake = _FakeSessions(error=ApiError("weird", "boom", status=500))
+        with pytest.raises(SessionApiFailed) as ei:
+            await cancel_turn(_wt(fake), "s", 42)
+        assert ei.value.status == 500

uv.lock

@@ -472,7 +472,7 @@ wheels = [
 
 [[package]]
 name = "ratatoskr"
-version = "0.21.7"
+version = "0.21.9"
 source = { editable = "." }
 dependencies = [
     { name = "httpx" },