ratatoskr/tests/test_sse_client.py

"""Tests for ratatoskr.sse_client per docs/contracts/issues/1.contract.md."""

import httpx
import pytest
import respx

from ratatoskr.sse_client import (
    AdminEvent,
    AffectUpdate,
    AgentNotAvailable,
    AwaitingLlmFirstToken,
    CancelAlreadyCompleted,
    Cancelled,
    CancelResult,
    CancelTurnNotFound,
    Done,
    Error,
    InvalidLastEventId,
    MalformedSseId,
    ResumeBufferExpired,
    ResumeTurnFinished,
    SseConnectFailed,
    SseConnectionDropped,
    SseId,
    Text,
    TurnIdFlip,
    TurnLaunchUnavailable,
    _parse_sse_id,
    cancel_turn,
    reconnect_turn,
    stream_admin_events,
    stream_turn,
    stream_turn_resilient,
)

_DONE_42_6 = {
    "type": "done",
    "phase": "succeeded",
    "response": "hello",
    "model": "m",
    "duration_ms": 1,
    "usage": {
        "prompt_tokens": 0,
        "completion_tokens": 0,
        "total_tokens": 0,
        "cached_input_tokens": 0,
    },
}


def _sse_chunk(sse_id: str, body: dict[str, object]) -> bytes:
    """Compose one SSE event in wire format. Trailing blank line per spec."""
    import json

    return f"id: {sse_id}\ndata: {json.dumps(body)}\n\n".encode()


_EVENT_STREAM = {"content-type": "text/event-stream"}


class _DropStream(httpx.AsyncByteStream):
    """Yield the given chunks, then raise a mid-stream drop (RemoteProtocolError).

    Mirrors the inline `_DropAfter` used by TestStreamTurn.test_connection_drop;
    hoisted to module scope because the resilient-wrapper tests reuse it.
    """

    def __init__(self, chunks: list[bytes]) -> None:
        self._chunks = chunks

    async def __aiter__(self):  # type: ignore[no-untyped-def]
        for c in self._chunks:
            yield c
        raise httpx.RemoteProtocolError("simulated mid-stream drop")

    async def aclose(self) -> None:
        return None


def _drop_response(chunks: list[bytes]) -> httpx.Response:
    return httpx.Response(200, headers=_EVENT_STREAM, stream=_DropStream(chunks))


def _stream_response(content: bytes) -> httpx.Response:
    return httpx.Response(200, headers=_EVENT_STREAM, content=content)


class TestParseSseId:
    def test_happy_simple(self) -> None:
        """happy_simple [happy,tracer]: '42:3' -> SseId(turn_id=42, seq=3)."""
        assert _parse_sse_id("42:3") == SseId(turn_id=42, seq=3)

    def test_happy_seq_one(self) -> None:
        """happy_seq_one: smallest valid id per spec — first event of first turn."""
        assert _parse_sse_id("1:1") == SseId(turn_id=1, seq=1)

    def test_empty(self) -> None:
        """empty [adversarial]: '' -> ValueError."""
        with pytest.raises(ValueError):
            _parse_sse_id("")

    def test_no_colon(self) -> None:
        """no_colon [adversarial]: '42' -> ValueError with '42' in the message."""
        with pytest.raises(ValueError, match="42"):
            _parse_sse_id("42")

    def test_too_many_colons(self) -> None:
        """too_many_colons [adversarial]: '42:3:7' -> ValueError."""
        with pytest.raises(ValueError):
            _parse_sse_id("42:3:7")

    def test_alpha_turn_id(self) -> None:
        """alpha_turn_id [adversarial]: 'foo:3' -> ValueError."""
        with pytest.raises(ValueError):
            _parse_sse_id("foo:3")

    def test_alpha_seq(self) -> None:
        """alpha_seq [adversarial]: '42:bar' -> ValueError."""
        with pytest.raises(ValueError):
            _parse_sse_id("42:bar")

    def test_zero_turn_id(self) -> None:
        """zero_turn_id [adversarial]: '0:3' -> ValueError (turn_id ≥ 1)."""
        with pytest.raises(ValueError):
            _parse_sse_id("0:3")

    def test_zero_seq(self) -> None:
        """zero_seq [adversarial]: '1:0' -> ValueError (seq ≥ 1, spec line 705)."""
        with pytest.raises(ValueError):
            _parse_sse_id("1:0")

    def test_negative_turn_id(self) -> None:
        """negative_turn_id [adversarial]: '-1:3' -> ValueError."""
        with pytest.raises(ValueError):
            _parse_sse_id("-1:3")

    def test_negative_seq(self) -> None:
        """negative_seq [adversarial]: '42:-1' -> ValueError."""
        with pytest.raises(ValueError):
            _parse_sse_id("42:-1")

    def test_trailing_whitespace(self) -> None:
        """trailing_whitespace [adversarial]: '42:3 ' -> ValueError (strict; no strip)."""
        with pytest.raises(ValueError):
            _parse_sse_id("42:3 ")

    def test_truncation(self) -> None:
        """truncation [security]: 5000-char no-colon -> ValueError msg contains only raw[:64]."""
        raw = "a" * 5000
        with pytest.raises(ValueError) as exc_info:
            _parse_sse_id(raw)
        assert raw not in str(exc_info.value), "full 5000-char input must not appear in message"
        assert raw[:64] in str(exc_info.value), "first 64 chars must appear in message"

    def test_non_string_input(self) -> None:
        """PRE-001 hard: raw is a string -- isinstance check before parse."""
        with pytest.raises(AssertionError):
            _parse_sse_id(42)  # type: ignore[arg-type]
        with pytest.raises(AssertionError):
            _parse_sse_id(None)  # type: ignore[arg-type]


class TestStreamTurn:
    @respx.mock
    async def test_happy_one_text_done(self) -> None:
        """happy_one_text_done: text then done; same turn_id; iter ends after done."""
        stream = _sse_chunk("42:1", {"type": "text", "content": "hello"}) + _sse_chunk(
            "42:2",
            {
                "type": "done",
                "phase": "succeeded",
                "response": "hello",
                "model": "glm5-turbo",
                "duration_ms": 100,
                "usage": {
                    "prompt_tokens": 0,
                    "completion_tokens": 0,
                    "total_tokens": 0,
                    "cached_input_tokens": 0,
                },
            },
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn(client, "s1", "hi")]
        assert len(events) == 2
        assert isinstance(events[0], Text)
        assert events[0].content == "hello"
        assert events[0].sse_id == SseId(42, 1)
        assert isinstance(events[1], Done)
        assert events[1].sse_id == SseId(42, 2)
        assert events[0].sse_id.turn_id == events[1].sse_id.turn_id

    @respx.mock
    async def test_full_event_vocab(self) -> None:
        """full_event_vocab: one of each variant; all carry parsed sse_id."""
        stream = b"".join(
            [
                _sse_chunk(
                    "42:1",
                    {"type": "worker_phase", "phase": "BuildingPrompt", "turn_id": 42},
                ),
                _sse_chunk("42:2", {"type": "thinking", "content": "thinking out loud"}),
                _sse_chunk("42:3", {"type": "text", "content": "hello"}),
                _sse_chunk(
                    "42:4",
                    {
                        "type": "text_boundary",
                        "kind": "sentence",
                        "char_offset": 5,
                        "ts": "2026-05-21T00:00:00Z",
                    },
                ),
                _sse_chunk(
                    "42:5",
                    {"type": "tool_start", "name": "search", "arguments": {"q": "x"}},
                ),
                _sse_chunk(
                    "42:6",
                    {
                        "type": "tool_result",
                        "name": "search",
                        "result": {"n": 1},
                        "duration_ms": 12,
                    },
                ),
                _sse_chunk("42:7", {"type": "text", "content": " world"}),
                _sse_chunk(
                    "42:8",
                    {
                        "type": "done",
                        "phase": "succeeded",
                        "response": "hello world",
                        "model": "m",
                        "duration_ms": 50,
                        "usage": {
                            "prompt_tokens": 0,
                            "completion_tokens": 0,
                            "total_tokens": 0,
                            "cached_input_tokens": 0,
                        },
                    },
                ),
            ]
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn(client, "s1", "hi")]
        types_seen = [type(e).__name__ for e in events]
        assert types_seen == [
            "WorkerPhase",
            "Thinking",
            "Text",
            "TextBoundary",
            "ToolStart",
            "ToolResult",
            "Text",
            "Done",
        ]
        # INV-002: every event carries parsed sse_id with turn_id, seq >= 1
        for e in events:
            assert e.sse_id.turn_id == 42
            assert e.sse_id.seq >= 1

    @respx.mock
    async def test_error_terminal(self) -> None:
        """error_terminal: text then error; iteration ends; error_code populated."""
        stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
            "42:2",
            {
                "type": "error",
                "phase": "failed",
                "error_code": "llm_output_invalid",
                "message": "boom",
            },
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn(client, "s1", "hi")]
        assert len(events) == 2
        assert isinstance(events[1], Error)
        assert events[1].error_code == "llm_output_invalid"
        assert events[1].message == "boom"

    @respx.mock
    async def test_cancelled_terminal(self) -> None:
        """cancelled_terminal: cancelled with phase=cancelled, turn_id; iteration ends."""
        stream = _sse_chunk(
            "42:1",
            {
                "type": "cancelled",
                "phase": "cancelled",
                "turn_id": 42,
                "reason": "user_cancel",
                "partial_message_id": None,
            },
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn(client, "s1", "hi")]
        assert len(events) == 1
        assert isinstance(events[0], Cancelled)
        assert events[0].turn_id == 42
        assert events[0].reason == "user_cancel"

    @respx.mock
    async def test_session_not_found(self) -> None:
        """session_not_found: 404 -> SseConnectFailed(status=404)."""
        respx.post("https://w.example/sessions/missing/messages").mock(
            return_value=httpx.Response(404, json={"error": "session_not_found"})
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(SseConnectFailed) as exc_info:
                _ = [e async for e in stream_turn(client, "missing", "hi")]
        assert exc_info.value.status == 404

    @respx.mock
    async def test_malformed_id_no_seq(self) -> None:
        """malformed_id_no_seq: id `42` (missing :seq) -> MalformedSseId; no event yielded."""
        stream = b"id: 42\ndata: {\"type\":\"text\",\"content\":\"x\"}\n\n"
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events: list[object] = []
            with pytest.raises(MalformedSseId):
                async for e in stream_turn(client, "s1", "hi"):
                    events.append(e)
        assert events == []

    @respx.mock
    async def test_malformed_id_alpha(self) -> None:
        """malformed_id_alpha: id `foo:bar` -> MalformedSseId."""
        stream = b"id: foo:bar\ndata: {\"type\":\"text\",\"content\":\"x\"}\n\n"
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(MalformedSseId):
                _ = [e async for e in stream_turn(client, "s1", "hi")]

    @respx.mock
    async def test_turn_id_flip(self) -> None:
        """turn_id_flip: 42:1 then 99:2 -> TurnIdFlip; only first yielded."""
        stream = _sse_chunk("42:1", {"type": "text", "content": "a"}) + _sse_chunk(
            "99:2", {"type": "text", "content": "b"}
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            yielded: list[Text] = []
            with pytest.raises(TurnIdFlip) as exc_info:
                async for e in stream_turn(client, "s1", "hi"):
                    yielded.append(e)  # type: ignore[arg-type]
        assert len(yielded) == 1
        assert exc_info.value.established == 42
        assert exc_info.value.got == 99

    @respx.mock
    async def test_connection_drop(self) -> None:
        """connection_drop: RemoteProtocolError after one text -> SseConnectionDropped((42,1))."""
        from ratatoskr.sse_client import SseConnectionDropped

        class _DropAfter(httpx.AsyncByteStream):
            def __init__(self, chunks: list[bytes]) -> None:
                self._chunks = chunks

            async def __aiter__(self):  # type: ignore[no-untyped-def]
                for c in self._chunks:
                    yield c
                raise httpx.RemoteProtocolError("simulated mid-stream drop")

            async def aclose(self) -> None:
                return None

        first = _sse_chunk("42:1", {"type": "text", "content": "x"})
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200,
                headers={"content-type": "text/event-stream"},
                stream=_DropAfter([first]),
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            yielded: list[Text] = []
            with pytest.raises(SseConnectionDropped) as exc_info:
                async for e in stream_turn(client, "s1", "hi"):
                    yielded.append(e)  # type: ignore[arg-type]
        assert len(yielded) == 1
        assert exc_info.value.last_seen_sse_id == SseId(42, 1)

    @respx.mock
    async def test_clean_eof_before_terminal(self) -> None:
        """INV-001: clean EOF without Done/Error/Cancelled -> SseConnectionDropped."""
        from ratatoskr.sse_client import SseConnectionDropped

        # One text event, then stream ends cleanly (no terminal).
        stream = _sse_chunk("42:1", {"type": "text", "content": "x"})
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            yielded: list[Text] = []
            with pytest.raises(SseConnectionDropped) as exc_info:
                async for e in stream_turn(client, "s1", "hi"):
                    yielded.append(e)  # type: ignore[arg-type]
        assert len(yielded) == 1
        assert exc_info.value.last_seen_sse_id == SseId(42, 1)

    @respx.mock
    async def test_zero_event_eof(self) -> None:
        """INV-001: empty stream (no events at all) -> SseConnectionDropped(None)."""
        from ratatoskr.sse_client import SseConnectionDropped

        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=b""
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(SseConnectionDropped) as exc_info:
                _ = [e async for e in stream_turn(client, "s1", "hi")]
        assert exc_info.value.last_seen_sse_id is None

    @respx.mock
    async def test_connect_failed_body_truncated(self) -> None:
        """ERROR_ROUTING: SseConnectFailed.body is truncated to <= 1024 bytes."""
        big_body = b"x" * 5000
        # 500 (not 409/503 — those are now eager turn-launch carve-outs, #331).
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(500, content=big_body)
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(SseConnectFailed) as exc_info:
                _ = [e async for e in stream_turn(client, "s1", "hi")]
        assert len(exc_info.value.body) <= 1024
        assert exc_info.value.body == big_body[:1024]

    @respx.mock
    async def test_eager_409_agent_not_available(self) -> None:
        """b1 #331: eager 409 -> AgentNotAvailable (SseConnectFailed subclass) with
        typed error_code; the turn never streams."""
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                409,
                json={
                    "detail": {
                        "error_code": "agent_not_available",
                        "message": "agent ratatoskr:sindra is unavailable",
                    }
                },
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(AgentNotAvailable) as exc:
                _ = [e async for e in stream_turn(client, "s1", "hi")]
        assert exc.value.status == 409
        assert exc.value.error_code == "agent_not_available"
        assert "unavailable" in exc.value.message
        assert isinstance(exc.value, SseConnectFailed)  # existing handlers still catch

    @respx.mock
    async def test_eager_503_turn_launch_unavailable_retryable(self) -> None:
        """b1 #331: eager 503 -> TurnLaunchUnavailable (retryable, SseConnectFailed subclass)."""
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                503,
                json={"error_code": "turn_launch_failed", "message": "resource exhausted"},
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(TurnLaunchUnavailable) as exc:
                _ = [e async for e in stream_turn(client, "s1", "hi")]
        assert exc.value.status == 503
        assert exc.value.retryable is True
        assert exc.value.error_code == "turn_launch_failed"
        assert isinstance(exc.value, SseConnectFailed)

    @respx.mock
    async def test_eager_409_non_json_body_defaults(self) -> None:
        """b1 #331: eager 409 with a non-JSON body -> AgentNotAvailable with the
        status-derived default error_code."""
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(409, content=b"<html>nope</html>")
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(AgentNotAvailable) as exc:
                _ = [e async for e in stream_turn(client, "s1", "hi")]
        assert exc.value.error_code == "agent_not_available"

    @respx.mock
    async def test_eager_503_non_json_body_defaults_not_ready(self) -> None:
        """b2: eager 503 with a non-JSON body -> TurnLaunchUnavailable with the
        canonical default error_code `not_ready`."""
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(503, content=b"<html>nope</html>")
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(TurnLaunchUnavailable) as exc:
                _ = [e async for e in stream_turn(client, "s1", "hi")]
        assert exc.value.error_code == "not_ready"
        assert exc.value.retryable is True

    @respx.mock
    async def test_no_text_aggregation(self) -> None:
        """no_text_aggregation: consumer yields each text event separately; no concat."""
        stream = (
            _sse_chunk("42:1", {"type": "text", "content": "hello "})
            + _sse_chunk("42:2", {"type": "text", "content": "world"})
            + _sse_chunk(
                "42:3",
                {
                    "type": "done",
                    "phase": "succeeded",
                    "response": "hello world",
                    "model": "m",
                    "duration_ms": 1,
                    "usage": {
                        "prompt_tokens": 0,
                        "completion_tokens": 0,
                        "total_tokens": 0,
                        "cached_input_tokens": 0,
                    },
                },
            )
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn(client, "s1", "hi")]
        texts = [e for e in events if isinstance(e, Text)]
        assert [t.content for t in texts] == ["hello ", "world"]


class TestReconnectTurn:
    @respx.mock
    async def test_happy_resume_from_seq_3(self) -> None:
        """happy_resume_from_seq_3 [scenario,tracer]: replay 42:4,42:5 then live 42:6 done."""
        stream = (
            _sse_chunk("42:4", {"type": "text", "content": "d"})
            + _sse_chunk("42:5", {"type": "text", "content": "e"})
            + _sse_chunk("42:6", _DONE_42_6)
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in reconnect_turn(client, "s1", "original", "42:3")]
        assert len(events) == 3
        assert [e.sse_id.seq for e in events] == [4, 5, 6]
        assert all(e.sse_id.turn_id == 42 for e in events)
        assert isinstance(events[-1], Done)

    @respx.mock
    async def test_header_verbatim(self) -> None:
        """header_verbatim: outbound carries Last-Event-ID: 42:3 exactly."""
        stream = _sse_chunk("42:4", _DONE_42_6)
        route = respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            _ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
        req = route.calls[0].request
        assert req.headers["Last-Event-ID"] == "42:3"

    @respx.mock
    async def test_body_threads_content(self) -> None:
        """body_threads_content: outbound JSON body is {'content': <content>} byte-for-byte."""
        import json as _json

        stream = _sse_chunk("42:4", _DONE_42_6)
        route = respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            _ = [
                e async for e in reconnect_turn(client, "s1", "hello mimir", "42:3")
            ]
        body = _json.loads(route.calls[0].request.content)
        assert body == {"content": "hello mimir"}

    @respx.mock
    async def test_buffer_expired(self) -> None:
        """buffer_expired: 412 -> ResumeBufferExpired(turn_id=42, buffered_from_seq=10)."""
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                412,
                json={"error": "buffer_expired", "buffered_from_seq": 10, "turn_id": 42},
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(ResumeBufferExpired) as exc_info:
                _ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
        assert exc_info.value.turn_id == 42
        assert exc_info.value.buffered_from_seq == 10

    @respx.mock
    async def test_turn_finished(self) -> None:
        """turn_finished: 410 -> ResumeTurnFinished(turn_id=42)."""
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                410, json={"error": "turn_finished", "turn_id": 42}
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(ResumeTurnFinished) as exc_info:
                _ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
        assert exc_info.value.turn_id == 42

    @respx.mock
    async def test_invalid_last_event_id_server(self) -> None:
        """invalid_last_event_id_server: 400 -> InvalidLastEventId."""
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(400, json={"error": "invalid_last_event_id"})
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(InvalidLastEventId):
                _ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]

    @respx.mock
    async def test_malformed_input(self) -> None:
        """malformed_input: caller passes '42' (no seq) -> ValueError; no HTTP issued."""
        route = respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(200, content=b"")
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(ValueError):
                _ = [e async for e in reconnect_turn(client, "s1", "x", "42")]
        assert route.call_count == 0

    @respx.mock
    async def test_turn_id_flip_on_first_event(self) -> None:
        """turn_id_flip_on_first_event: server's first event 99:4 -> TurnIdFlip pre-yield."""
        stream = _sse_chunk("99:4", {"type": "text", "content": "wrong turn"})
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            yielded: list[Text] = []
            with pytest.raises(TurnIdFlip) as exc_info:
                async for e in reconnect_turn(client, "s1", "x", "42:3"):
                    yielded.append(e)  # type: ignore[arg-type]
        assert yielded == []
        assert exc_info.value.established == 42
        assert exc_info.value.got == 99


class TestCancelTurn:
    @respx.mock
    async def test_happy_cancel(self) -> None:
        """happy_cancel [happy,tracer]: 200 -> CancelResult(42, True, None, None)."""
        respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
            return_value=httpx.Response(
                200,
                json={
                    "turn_id": 42,
                    "cancelled": True,
                    "reason": None,
                    "partial_message_id": None,
                },
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            result = await cancel_turn(client, "s1", 42)
        assert result == CancelResult(
            turn_id=42, cancelled=True, reason=None, partial_message_id=None
        )

    @respx.mock
    async def test_persist_partial_query(self) -> None:
        """persist_partial_query [trace]: persist_partial=True -> URL has ?persist_partial=true."""
        route = respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
            return_value=httpx.Response(
                200,
                json={
                    "turn_id": 42,
                    "cancelled": True,
                    "reason": None,
                    "partial_message_id": None,
                },
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            await cancel_turn(client, "s1", 42, persist_partial=True)
        assert "persist_partial=true" in str(route.calls[0].request.url)

    @respx.mock
    async def test_default_no_persist(self) -> None:
        """default_no_persist [trace]: default call -> URL has no persist_partial param."""
        route = respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
            return_value=httpx.Response(
                200,
                json={
                    "turn_id": 42,
                    "cancelled": True,
                    "reason": None,
                    "partial_message_id": None,
                },
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            await cancel_turn(client, "s1", 42)
        assert "persist_partial" not in str(route.calls[0].request.url)

    @respx.mock
    async def test_cancel_already_completed(self) -> None:
        """cancel_already_completed [error]: 409 -> CancelAlreadyCompleted(turn_id=42)."""
        respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
            return_value=httpx.Response(409, json={"error": "turn_already_completed"})
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(CancelAlreadyCompleted) as exc_info:
                await cancel_turn(client, "s1", 42)
        assert exc_info.value.turn_id == 42

    @respx.mock
    async def test_cancel_turn_not_found(self) -> None:
        """cancel_turn_not_found [error]: 404 -> CancelTurnNotFound(turn_id=42)."""
        respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
            return_value=httpx.Response(404, json={"error": "turn_not_found"})
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(CancelTurnNotFound) as exc_info:
                await cancel_turn(client, "s1", 42)
        assert exc_info.value.turn_id == 42

    @respx.mock
    async def test_cancel_failed_truncates_body(self) -> None:
        """ERROR_ROUTING: 5xx -> CancelFailed; body truncated to <= 1024 bytes."""
        from ratatoskr.sse_client import CancelFailed

        big_body = b"x" * 5000
        respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
            return_value=httpx.Response(503, content=big_body)
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(CancelFailed) as exc_info:
                await cancel_turn(client, "s1", 42)
        assert exc_info.value.status == 503
        assert len(exc_info.value.body) <= 1024
        assert exc_info.value.body == big_body[:1024]

    @respx.mock
    async def test_cancel_idempotent(self) -> None:
        """cancel_idempotent: first 200 -> happy; second 409 -> CancelAlreadyCompleted."""
        respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
            side_effect=[
                httpx.Response(
                    200,
                    json={
                        "turn_id": 42,
                        "cancelled": True,
                        "reason": None,
                        "partial_message_id": None,
                    },
                ),
                httpx.Response(409, json={"error": "turn_already_completed"}),
            ]
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            first = await cancel_turn(client, "s1", 42)
            assert first.cancelled is True
            with pytest.raises(CancelAlreadyCompleted):
                await cancel_turn(client, "s1", 42)


# ============================================================================
# Issue #7: empty-data skip + MalformedSseData raise
# ============================================================================


def _sse_empty_chunk(sse_id: str) -> bytes:
    """SSE frame with id but empty data (server-emitted keepalive shape)."""
    return f"id: {sse_id}\ndata:\n\n".encode()


def _sse_raw_chunk(sse_id: str, raw_data: str) -> bytes:
    """SSE frame with id + arbitrary raw data (for testing malformed JSON)."""
    return f"id: {sse_id}\ndata: {raw_data}\n\n".encode()


def _sse_no_id_chunk(data: str) -> bytes:
    """SSE frame with NO id line + arbitrary data (v0.8.1: keepalive shape)."""
    return f"data: {data}\n\n".encode()


class TestEmptyIdSkipped:
    @respx.mock
    async def test_empty_id_on_first_event_skipped(self) -> None:
        """empty_id_on_first_event_skipped [v0.8.1]: stream starts with an
        event carrying NO `id:` line → httpx_sse exposes sse.id == ''
        (no prior id to inherit). Pre-v0.8.1: MalformedSseId raw='' crashed
        the turn. v0.8.1: treat same as empty-data keepalive — skip silently.

        Observed 2026-05-25 on Worldtree's qwen3.6-35-a3b-heretic provider:
        the first stream frame had no id line, every turn died with
        `[malformed_sse_id] raw=''`.
        """
        from ratatoskr.sse_client import Done as _Done
        from ratatoskr.sse_client import Text as _Text

        # First frame: no id line (httpx_sse → sse.id = ""). Skip it.
        # Subsequent frames have ids; normal processing resumes.
        stream = (
            _sse_no_id_chunk('{"type":"keepalive"}')  # ← skipped (sse.id == "")
            + _sse_chunk("42:1", {"type": "text", "content": "first"})
            + _sse_chunk("42:2", _DONE_42_6)
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn(client, "s1", "hi")]
        # 2 events — the no-id frame is invisible (no MalformedSseId crash).
        assert len(events) == 2
        assert isinstance(events[0], _Text)
        assert events[0].content == "first"
        assert isinstance(events[1], _Done)


class TestEmptyDataSkipped:
    @respx.mock
    async def test_empty_data_skipped(self) -> None:
        """empty_data_skipped [trace]: 4 frames in, 3 events out; skip preserves last_sse_id."""
        from ratatoskr.sse_client import Done as _Done
        from ratatoskr.sse_client import Text as _Text

        stream = (
            _sse_chunk("42:1", {"type": "text", "content": "first"})
            + _sse_empty_chunk("42:2")  # ← skipped silently
            + _sse_chunk("42:3", {"type": "text", "content": "second"})
            + _sse_chunk("42:4", _DONE_42_6)
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn(client, "s1", "hi")]
        # Exactly 3 events: Text, Text, Done — empty-data frame at 42:2 is invisible
        assert len(events) == 3
        assert isinstance(events[0], _Text) and events[0].sse_id == SseId(42, 1)
        assert isinstance(events[1], _Text) and events[1].sse_id == SseId(42, 3)
        assert isinstance(events[2], _Done) and events[2].sse_id == SseId(42, 4)
        # Per INV-003: skip MUST NOT advance through 42:2. The second Text's sse_id
        # is (42, 3) — directly verifies the skip didn't bookkeep 42:2.
        assert events[1].sse_id.seq == 3, "skip advanced through 42:2"

    @respx.mock
    async def test_empty_data_skip_preserves_last_seen_sse_id(self) -> None:
        """empty_skip_does_not_advance [trace]: drop-after-empty → last_seen is last real event."""
        from ratatoskr.sse_client import SseConnectionDropped

        # Stream: text(42:1), empty(42:2), then drop. Per INV-003, the skipped
        # 42:2 must NOT advance internal last_sse_id. If the consumer caught a
        # drop, SseConnectionDropped.last_seen_sse_id should be (42, 1) — the
        # last *real* event — NOT (42, 2).
        first = _sse_chunk("42:1", {"type": "text", "content": "x"})
        empty = _sse_empty_chunk("42:2")

        class _DropAfterEmpty(httpx.AsyncByteStream):
            async def __aiter__(self):  # type: ignore[no-untyped-def]
                yield first
                yield empty
                raise httpx.RemoteProtocolError("drop after skip")

            async def aclose(self) -> None:
                return None

        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200,
                headers={"content-type": "text/event-stream"},
                stream=_DropAfterEmpty(),
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(SseConnectionDropped) as exc_info:
                async for _ in stream_turn(client, "s1", "hi"):
                    pass
        assert exc_info.value.last_seen_sse_id == SseId(42, 1), (
            f"skip advanced last_sse_id through 42:2; got {exc_info.value.last_seen_sse_id}"
        )

    @respx.mock
    async def test_malformed_data_raises(self) -> None:
        """malformed_data_raises [error]: text + bad-JSON → yields Text then MalformedSseData."""
        from ratatoskr.sse_client import MalformedSseData
        from ratatoskr.sse_client import Text as _Text

        stream = (
            _sse_chunk("42:1", {"type": "text", "content": "hi"})
            + _sse_raw_chunk("42:2", "not-json")
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        yielded: list = []
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(MalformedSseData) as exc_info:
                async for e in stream_turn(client, "s1", "hi"):
                    yielded.append(e)
        assert len(yielded) == 1
        assert isinstance(yielded[0], _Text)
        assert exc_info.value.raw == "not-json"

    @respx.mock
    async def test_whitespace_data_raises(self) -> None:
        """whitespace_data_raises [adv]: single-space data → MalformedSseData (NOT skipped)."""
        from ratatoskr.sse_client import MalformedSseData

        stream = _sse_raw_chunk("42:1", " ")  # single space — non-empty, JSON-invalid
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(MalformedSseData):
                async for _ in stream_turn(client, "s1", "hi"):
                    pass

    @respx.mock
    async def test_malformed_data_truncation(self) -> None:
        """malformed_data_truncation [security]: 5000-char bad data → raw truncated to 200."""
        from ratatoskr.sse_client import MalformedSseData

        huge_bad = "x" * 5000  # not JSON; very long
        stream = _sse_raw_chunk("42:1", huge_bad)
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(MalformedSseData) as exc_info:
                async for _ in stream_turn(client, "s1", "hi"):
                    pass
        assert len(exc_info.value.raw) == 200
        assert exc_info.value.raw == "x" * 200
        # Exception message also only contains the truncated form
        assert "x" * 5000 not in str(exc_info.value)


class TestAffectUpdate:
    """Worldtree #204 / v0.28.0 — persona-state observability SSE event.

    Two emissions per qualifying turn (persona-enabled agent, non-ephemeral
    session): `status: "current"` at turn start with full snapshot, then
    `status: "scheduled"` near turn end (lightweight, no snapshot).

    See docs/conversation-api-spec.md § affect_update.
    """

    @respx.mock
    async def test_current_status_parsed_with_snapshot(self) -> None:
        """current_status_parsed_with_snapshot [tracer]: status=current carries
        the full snapshot dict; AffectUpdate.snapshot is populated with the
        nested PAD / dominant_emotion / emotions_active fields.
        """
        snapshot = {
            "agent_id": "mimir",
            "pad": {"pleasure": 0.52, "arousal": 0.47, "dominance": 0.50},
            "dominant_emotion": "curiosity",
            "emotions_active": [
                {"type": "curiosity", "intensity": 0.6, "decay_remaining_s": 202.7}
            ],
            "baseline_pad": {"pleasure": 0.50, "arousal": 0.40, "dominance": 0.50},
            "mood_drift": {"valence_delta": 0.02, "arousal_delta": 0.07},
            "last_updated_at": "2026-05-25T22:30:18+00:00",
        }
        stream = _sse_chunk(
            "42:1",
            {
                "type": "affect_update",
                "status": "current",
                "turn_id": 42,
                "snapshot": snapshot,
            },
        ) + _sse_chunk("42:2", _DONE_42_6)
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn(client, "s1", "hi")]
        affect = events[0]
        assert isinstance(affect, AffectUpdate)
        assert affect.status == "current"
        assert affect.turn_id == 42
        assert affect.snapshot == snapshot
        assert affect.sse_id == SseId(42, 1)

    @respx.mock
    async def test_scheduled_status_parsed_no_snapshot(self) -> None:
        """scheduled_status_parsed_no_snapshot [trace]: status=scheduled carries
        no snapshot field; AffectUpdate.snapshot is None.
        """
        stream = (
            _sse_chunk("42:1", {"type": "text", "content": "x"})
            + _sse_chunk(
                "42:2",
                {"type": "affect_update", "status": "scheduled", "turn_id": 42},
            )
            + _sse_chunk("42:3", _DONE_42_6)
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn(client, "s1", "hi")]
        affect = next(e for e in events if isinstance(e, AffectUpdate))
        assert affect.status == "scheduled"
        assert affect.turn_id == 42
        assert affect.snapshot is None
        assert affect.sse_id == SseId(42, 2)


class TestAwaitingLlmFirstToken:
    """Worldtree #201 / v0.29.0 — `awaiting_llm_first_token` SSE heartbeat.

    Top-level event (not a worker_phase extension) fired during the
    BuildingPrompt → CallingLLM gap at the configured interval (default
    5s). Server-authoritative elapsed_ms is time.monotonic()-based and
    monotonically increasing across the heartbeat sequence.

    See docs/conversation-api-spec.md § awaiting_llm_first_token.
    """

    @respx.mock
    async def test_single_heartbeat_parsed(self) -> None:
        """single_heartbeat_parsed [tracer]: type=awaiting_llm_first_token →
        AwaitingLlmFirstToken(turn_id, elapsed_ms_since_building_prompt).
        """
        stream = _sse_chunk(
            "42:1",
            {
                "type": "awaiting_llm_first_token",
                "turn_id": 42,
                "elapsed_ms_since_building_prompt": 5012.3,
            },
        ) + _sse_chunk("42:2", _DONE_42_6)
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn(client, "s1", "hi")]
        beat = events[0]
        assert isinstance(beat, AwaitingLlmFirstToken)
        assert beat.turn_id == 42
        assert beat.elapsed_ms_since_building_prompt == 5012.3
        assert beat.sse_id == SseId(42, 1)

    @respx.mock
    async def test_heartbeat_sequence_monotonic(self) -> None:
        """heartbeat_sequence_monotonic [scenario]: three consecutive heartbeats
        in one turn — elapsed_ms_since_building_prompt monotonically increases,
        all carry the same turn_id.
        """
        stream = (
            _sse_chunk(
                "42:1",
                {
                    "type": "awaiting_llm_first_token",
                    "turn_id": 42,
                    "elapsed_ms_since_building_prompt": 5000.0,
                },
            )
            + _sse_chunk(
                "42:2",
                {
                    "type": "awaiting_llm_first_token",
                    "turn_id": 42,
                    "elapsed_ms_since_building_prompt": 10005.4,
                },
            )
            + _sse_chunk(
                "42:3",
                {
                    "type": "awaiting_llm_first_token",
                    "turn_id": 42,
                    "elapsed_ms_since_building_prompt": 15011.8,
                },
            )
            + _sse_chunk("42:4", _DONE_42_6)
        )
        respx.post("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn(client, "s1", "hi")]
        beats = [e for e in events if isinstance(e, AwaitingLlmFirstToken)]
        assert len(beats) == 3
        elapsed = [b.elapsed_ms_since_building_prompt for b in beats]
        assert elapsed == sorted(elapsed)  # monotonically increasing
        assert all(b.turn_id == 42 for b in beats)


_URL = "https://w.example/sessions/s1/messages"


class TestStreamTurnResilient:
    """docs/contracts/issues/1.contract.md FN stream_turn_resilient (amendment 2026-06-30)."""

    @respx.mock
    async def test_happy_no_drop(self) -> None:
        """happy_no_drop [happy]: clean stream passes through; no reconnect issued."""
        stream = _sse_chunk("42:1", {"type": "text", "content": "a"}) + _sse_chunk(
            "42:2", _DONE_42_6
        )
        route = respx.post(_URL).mock(return_value=_stream_response(stream))
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn_resilient(client, "s1", "hi")]
        assert [e.sse_id for e in events] == [SseId(42, 1), SseId(42, 2)]
        assert isinstance(events[-1], Done)
        assert route.call_count == 1  # POST-001: no reconnect on a clean stream

    @respx.mock
    async def test_resume_after_one_drop(self) -> None:
        """resume_after_one_drop [tracer]: a mid-stream drop resumes via reconnect; one stream."""
        first = _drop_response([_sse_chunk("42:1", {"type": "text", "content": "a"})])
        resume = _stream_response(
            _sse_chunk("42:2", {"type": "text", "content": "b"})
            + _sse_chunk("42:3", _DONE_42_6)
        )
        route = respx.post(_URL).mock(side_effect=[first, resume])
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn_resilient(client, "s1", "hi")]
        assert [e.sse_id for e in events] == [SseId(42, 1), SseId(42, 2), SseId(42, 3)]
        assert isinstance(events[-1], Done)
        assert route.call_count == 2
        # POST-003: reconnect carries the last yielded pre-drop event's id.
        assert route.calls[1].request.headers.get("Last-Event-ID") == "42:1"
        # PRE/wire: first attempt does NOT carry a Last-Event-ID.
        assert route.calls[0].request.headers.get("Last-Event-ID") is None

    @respx.mock
    async def test_resume_after_clean_eof(self) -> None:
        """resume_after_clean_eof: a clean EOF before terminal also triggers resume (INV-001)."""
        first = _stream_response(_sse_chunk("42:1", {"type": "text", "content": "a"}))
        resume = _stream_response(_sse_chunk("42:2", _DONE_42_6))
        route = respx.post(_URL).mock(side_effect=[first, resume])
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn_resilient(client, "s1", "hi")]
        assert [e.sse_id for e in events] == [SseId(42, 1), SseId(42, 2)]
        assert isinstance(events[-1], Done)
        assert route.calls[1].request.headers.get("Last-Event-ID") == "42:1"

    @respx.mock
    async def test_two_drops_then_done(self) -> None:
        """two_drops_then_done: two transient drops, third attempt completes; ids thread through."""
        a1 = _drop_response([_sse_chunk("42:1", {"type": "text", "content": "a"})])
        a2 = _drop_response([_sse_chunk("42:2", {"type": "text", "content": "b"})])
        a3 = _stream_response(_sse_chunk("42:3", _DONE_42_6))
        route = respx.post(_URL).mock(side_effect=[a1, a2, a3])
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_turn_resilient(client, "s1", "hi")]
        assert [e.sse_id for e in events] == [SseId(42, 1), SseId(42, 2), SseId(42, 3)]
        assert route.call_count == 3
        assert route.calls[1].request.headers.get("Last-Event-ID") == "42:1"
        assert route.calls[2].request.headers.get("Last-Event-ID") == "42:2"

    @respx.mock
    async def test_unresumable_zero_event_drop(self) -> None:
        """unresumable_zero_event_drop [adversarial]: drop before any event → propagate."""
        route = respx.post(_URL).mock(side_effect=[_drop_response([])])
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(SseConnectionDropped):
                _ = [e async for e in stream_turn_resilient(client, "s1", "hi")]
        assert route.call_count == 1  # no id to resume from → no reconnect

    @respx.mock
    async def test_max_reconnects_exhausted(self) -> None:
        """max_reconnects_exhausted [adversarial]: every attempt drops; budget caps reconnects."""
        side = [
            _drop_response([_sse_chunk(f"42:{n}", {"type": "text", "content": "x"})])
            for n in (1, 2, 3)
        ]
        route = respx.post(_URL).mock(side_effect=side)
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(SseConnectionDropped):
                _ = [
                    e
                    async for e in stream_turn_resilient(
                        client, "s1", "hi", max_reconnects=2
                    )
                ]
        assert route.call_count == 3  # initial + 2 reconnects, then give up

    @respx.mock
    async def test_zero_budget_no_resume(self) -> None:
        """zero_budget_no_resume [adversarial]: max_reconnects=0 → first drop propagates."""
        first = _drop_response([_sse_chunk("42:1", {"type": "text", "content": "a"})])
        route = respx.post(_URL).mock(side_effect=[first])
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(SseConnectionDropped):
                _ = [
                    e
                    async for e in stream_turn_resilient(
                        client, "s1", "hi", max_reconnects=0
                    )
                ]
        assert route.call_count == 1

    @respx.mock
    async def test_buffer_expired_propagates(self) -> None:
        """buffer_expired_propagates [error]: a 412 on reconnect surfaces, not retried."""
        first = _drop_response([_sse_chunk("42:1", {"type": "text", "content": "a"})])
        resume = httpx.Response(412, json={"turn_id": 42, "buffered_from_seq": 7})
        route = respx.post(_URL).mock(side_effect=[first, resume])
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            collected: list[object] = []
            with pytest.raises(ResumeBufferExpired):
                async for e in stream_turn_resilient(client, "s1", "hi"):
                    collected.append(e)
        assert [e.sse_id for e in collected] == [SseId(42, 1)]  # type: ignore[attr-defined]
        assert route.call_count == 2


class TestStreamAdminEvents:
    """docs/conversation-api-spec.md § Admin Event Stream — stream_admin_events (#11)."""

    @respx.mock
    async def test_happy_multi_event_admin_bearer(self) -> None:
        """happy [happy,tracer]: yields AdminEvent envelopes; request uses the ADMIN bearer."""
        env1 = {
            "id": 41, "type": "session.created", "timestamp": "2026-05-06T10:00:00.000Z",
            "data": {"session_id": "s1", "agent_id": "mimir", "user_id": None},
        }
        env2 = {
            "id": 42, "type": "turn.started", "timestamp": "2026-05-06T10:00:01.000Z",
            "data": {"session_id": "s1", "turn_id": 7, "agent_id": "mimir", "user_id": None},
        }
        stream = _sse_chunk("41", env1) + _sse_chunk("42", env2)
        route = respx.get("https://w.example/admin/events").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=stream
            )
        )
        async with httpx.AsyncClient(
            base_url="https://w.example", headers={"Authorization": "Bearer consumer"}
        ) as client:
            events = [e async for e in stream_admin_events(client, admin_key="admin-xyz")]
        assert [e.type for e in events] == ["session.created", "turn.started"]
        assert isinstance(events[0], AdminEvent)
        assert events[0].id == 41
        assert events[1].data["turn_id"] == 7
        assert route.calls[0].request.headers["Authorization"] == "Bearer admin-xyz"

    @respx.mock
    async def test_last_event_id_header(self) -> None:
        """last_event_id_header [trace]: empty stream → []; Last-Event-ID header sent."""
        route = respx.get("https://w.example/admin/events").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=b""
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_admin_events(client, admin_key="k", last_event_id=99)]
        assert events == []
        assert route.calls[0].request.headers["Last-Event-ID"] == "99"

    @respx.mock
    async def test_403_scope_denied(self) -> None:
        """403 [error]: key lacks admin.events.read → SseConnectFailed(403)."""
        respx.get("https://w.example/admin/events").mock(
            return_value=httpx.Response(403, json={"error_code": "auth_scope_denied"})
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            with pytest.raises(SseConnectFailed) as exc:
                _ = [e async for e in stream_admin_events(client, admin_key="k")]
        assert exc.value.status == 403

    @respx.mock
    async def test_skips_malformed_frame(self) -> None:
        """skips_malformed [adversarial]: a bad-JSON frame is skipped, not fatal."""
        good = _sse_chunk("41", {"id": 41, "type": "session.created", "data": {"session_id": "s1"}})
        bad = b"id: 42\ndata: not-json\n\n"
        good2 = _sse_chunk("43", {"id": 43, "type": "session.deleted", "data": {"session_id": "s1"}})
        respx.get("https://w.example/admin/events").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=good + bad + good2
            )
        )
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            events = [e async for e in stream_admin_events(client, admin_key="k")]
        assert [e.type for e in events] == ["session.created", "session.deleted"]