ratatoskr/tests/test_cli.py

"""Tests for ratatoskr.cli per docs/contracts/issues/3.contract.md."""

import asyncio
import io
import json

import httpx
import pytest
import respx

from ratatoskr import cli as cli_mod
from ratatoskr.cli import (
    ParsedArgs,
    UsageError,
    _amain,
    _AuthError,
    _cancel_and_log,
    _parse_args,
    _render_event,
    _run_turn,
    main,
)
from ratatoskr.sse_client import (
    Cancelled,
    Done,
    Error,
    SseId,
    Text,
    TextBoundary,
    Thinking,
    ToolResult,
    ToolStart,
    WorkerPhase,
)


class _FlushCountingIO(io.StringIO):
    """StringIO subclass that counts flush() calls — used to verify INV-010."""

    def __init__(self) -> None:
        super().__init__()
        self.flush_count = 0

    def flush(self) -> None:
        self.flush_count += 1
        super().flush()


def _sse_chunk(sse_id: str, body: dict[str, object]) -> bytes:
    return f"id: {sse_id}\ndata: {json.dumps(body)}\n\n".encode()


def _sse_resp(body: bytes | httpx.AsyncByteStream) -> httpx.Response:
    """Wrap an SSE response body (bytes or stream) with the right content-type."""
    headers = {"content-type": "text/event-stream"}
    if isinstance(body, bytes):
        return httpx.Response(200, headers=headers, content=body)
    return httpx.Response(200, headers=headers, stream=body)


_DONE_BODY = {
    "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,
    },
}
_CANCELLED_BODY = {
    "type": "cancelled",
    "phase": "cancelled",
    "turn_id": 42,
    "reason": "user_cancel",
    "partial_message_id": None,
}
_CANCEL_OK_RESP = {"turn_id": 42, "cancelled": True, "reason": None, "partial_message_id": None}


@pytest.fixture(autouse=True)
def _clear_env(monkeypatch: pytest.MonkeyPatch) -> None:
    """Tests assert env-resolution behavior; default to a clean slate per test."""
    monkeypatch.delenv("WORLDTREE_API_KEY", raising=False)
    monkeypatch.delenv("WORLDTREE_API_URL", raising=False)


class TestParseArgs:
    def test_happy_new(self) -> None:
        """happy_new [happy,tracer]: --send --new --agent --api-key → full ParsedArgs."""
        args = _parse_args(["--send", "hi", "--new", "--agent", "mimir", "--api-key", "k"])
        assert args == ParsedArgs(
            send_content="hi",
            session_id=None,
            new=True,
            agent_id="mimir",
            api_key="k",
            server_url="http://localhost:8000",
            raw=False,
        )

    def test_happy_existing_session(self) -> None:
        """happy_existing_session: --send --session --api-key → ParsedArgs with session_id."""
        args = _parse_args(["--send", "hi", "--session", "s-1", "--api-key", "k"])
        assert args == ParsedArgs(
            send_content="hi",
            session_id="s-1",
            new=False,
            agent_id=None,
            api_key="k",
            server_url="http://localhost:8000",
            raw=False,
        )

    def test_api_key_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """api_key_from_env: WORLDTREE_API_KEY env var fills in when --api-key omitted."""
        monkeypatch.setenv("WORLDTREE_API_KEY", "from-env")
        args = _parse_args(["--send", "hi", "--new", "--agent", "mimir"])
        assert args.api_key == "from-env"

    def test_api_key_flag_beats_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """api_key_flag_beats_env: explicit --api-key wins over WORLDTREE_API_KEY."""
        monkeypatch.setenv("WORLDTREE_API_KEY", "env")
        args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "flag"])
        assert args.api_key == "flag"

    def test_server_default(self) -> None:
        """server_default: no --server, no WORLDTREE_API_URL → http://localhost:8000."""
        args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
        assert args.server_url == "http://localhost:8000"

    def test_server_env_fallback(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """server_env_fallback: WORLDTREE_API_URL fills in when --server omitted."""
        monkeypatch.setenv("WORLDTREE_API_URL", "http://t.local:9000")
        args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
        assert args.server_url == "http://t.local:9000"

    def test_server_flag_beats_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """server_flag_beats_env: explicit --server wins over WORLDTREE_API_URL."""
        monkeypatch.setenv("WORLDTREE_API_URL", "env")
        args = _parse_args(
            ["--send", "hi", "--new", "--agent", "m", "--api-key", "k", "--server", "flag"]
        )
        assert args.server_url == "flag"

    def test_no_send_marks_tui_mode(self) -> None:
        """no_send_marks_tui_mode: missing --send → send_content=None (TUI marker)."""
        args = _parse_args(["--new", "--agent", "mimir", "--api-key", "k"])
        assert args.send_content is None
        # Other fields still populate normally
        assert args.new is True
        assert args.agent_id == "mimir"

    def test_raw_flag_default_false(self) -> None:
        """raw_flag_default_false: --raw absent → ParsedArgs.raw == False."""
        args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
        assert args.raw is False

    def test_raw_flag_set(self) -> None:
        """raw_flag_set: --raw → ParsedArgs.raw == True."""
        args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k", "--raw"])
        assert args.raw is True

    def test_usage_both_session_and_new(self) -> None:
        """usage_both_session_and_new: --session AND --new → UsageError('mutually exclusive')."""
        with pytest.raises(UsageError, match="mutually exclusive"):
            _parse_args(
                ["--send", "hi", "--session", "s", "--new", "--agent", "m", "--api-key", "k"]
            )

    def test_usage_neither_session_nor_new(self) -> None:
        """usage_neither_session_nor_new: neither flag → UsageError('pass exactly one')."""
        with pytest.raises(UsageError, match="pass exactly one"):
            _parse_args(["--send", "hi", "--api-key", "k"])

    def test_usage_new_without_agent(self) -> None:
        """usage_new_without_agent: --new without --agent → UsageError."""
        with pytest.raises(UsageError, match="--agent is required when --new"):
            _parse_args(["--send", "hi", "--new", "--api-key", "k"])

    def test_usage_session_with_agent(self) -> None:
        """usage_session_with_agent: --session AND --agent → UsageError."""
        with pytest.raises(UsageError, match="forbidden with --session"):
            _parse_args(["--send", "hi", "--session", "s-1", "--agent", "x", "--api-key", "k"])

    def test_auth_missing(self) -> None:
        """auth_missing: no --api-key and no env → _AuthError."""
        with pytest.raises(_AuthError, match="no API key"):
            _parse_args(["--send", "hi", "--new", "--agent", "mimir"])

    def test_empty_send(self) -> None:
        """empty_send: --send '' → UsageError (non-empty enforced)."""
        with pytest.raises(UsageError):
            _parse_args(["--send", "", "--new", "--agent", "x", "--api-key", "k"])


SID = SseId(42, 5)


class TestRenderEvent:
    def test_text_to_stdout_only(self) -> None:
        """text_to_stdout_only [happy,tracer]: Text → stdout=="hello"; stderr empty; flushed."""
        stdout = _FlushCountingIO()
        stderr = io.StringIO()
        _render_event(Text(sse_id=SID, content="hello"), stdout=stdout, stderr=stderr)
        assert stdout.getvalue() == "hello"
        assert stderr.getvalue() == ""
        assert stdout.flush_count == 1  # INV-010: per-chunk flush

    def test_done_writes_newline_and_label(self) -> None:
        """done_writes_newline_and_label: stdout=="\\n" (flushed); stderr "[done]" labels."""
        stdout = _FlushCountingIO()
        stderr = io.StringIO()
        evt = Done(
            sse_id=SID,
            phase="completed",
            response="hi",
            model="glm5-turbo",
            duration_ms=1234,
            usage={"prompt": 10, "completion": 5},
        )
        _render_event(evt, stdout=stdout, stderr=stderr)
        assert stdout.getvalue() == "\n"
        assert stdout.flush_count == 1  # INV-010: post-Done newline flushed
        out_err = stderr.getvalue()
        assert out_err.startswith("[done]")
        assert "turn_id=42" in out_err
        assert "model=glm5-turbo" in out_err
        assert "duration_ms=1234" in out_err

    def test_error_to_stderr_only(self) -> None:
        """error_to_stderr_only: Error → stderr "[error]" with code; stdout empty."""
        stdout = io.StringIO()
        stderr = io.StringIO()
        evt = Error(sse_id=SID, phase="failed", message="boom", error_code="llm_output_invalid")
        _render_event(evt, stdout=stdout, stderr=stderr)
        assert stdout.getvalue() == ""
        out_err = stderr.getvalue()
        assert out_err.startswith("[error]")
        assert "turn_id=42" in out_err
        assert "code=llm_output_invalid" in out_err

    def test_cancelled_to_stderr_only(self) -> None:
        """cancelled_to_stderr_only: Cancelled → stderr "[cancelled]" + reason + partial id."""
        stdout = io.StringIO()
        stderr = io.StringIO()
        evt = Cancelled(
            sse_id=SID, phase="cancelled", turn_id=42, reason="user", partial_message_id=7
        )
        _render_event(evt, stdout=stdout, stderr=stderr)
        assert stdout.getvalue() == ""
        out_err = stderr.getvalue()
        assert out_err.startswith("[cancelled]")
        assert "reason='user'" in out_err
        assert "partial_message_id=7" in out_err

    def test_worker_phase_to_stderr(self) -> None:
        """worker_phase_to_stderr: WorkerPhase → stderr "[worker_phase]"; stdout empty."""
        stdout = io.StringIO()
        stderr = io.StringIO()
        evt = WorkerPhase(sse_id=SID, phase="streaming", turn_id=42)
        _render_event(evt, stdout=stdout, stderr=stderr)
        assert stdout.getvalue() == ""
        assert stderr.getvalue().startswith("[worker_phase]")

    def test_thinking_truncated(self) -> None:
        """thinking_truncated [trace]: …"""
        stdout = io.StringIO()
        stderr = io.StringIO()
        _render_event(Thinking(sse_id=SID, content="a" * 500), stdout=stdout, stderr=stderr)
        out_err = stderr.getvalue()
        assert out_err.startswith("[thinking]")
        assert "a" * 500 not in out_err
        assert "a" * 200 in out_err

    def test_tool_start_to_stderr(self) -> None:
        """tool_start_to_stderr: ToolStart → stderr "[tool_start] name=... args=..."."""
        stdout = io.StringIO()
        stderr = io.StringIO()
        evt = ToolStart(sse_id=SID, name="read_file", arguments={"path": "/x"})
        _render_event(evt, stdout=stdout, stderr=stderr)
        assert stdout.getvalue() == ""
        out_err = stderr.getvalue()
        assert out_err.startswith("[tool_start] name=read_file args=")

    def test_tool_result_truncated(self) -> None:
        """tool_result_truncated [trace]: ToolResult.result repr truncated to ≤200 chars."""
        stdout = io.StringIO()
        stderr = io.StringIO()
        evt = ToolResult(sse_id=SID, name="x", result="b" * 500, duration_ms=42)
        _render_event(evt, stdout=stdout, stderr=stderr)
        out_err = stderr.getvalue()
        assert out_err.startswith("[tool_result]")
        # the contract uses `{event.result!r:.200}` — 200 chars max of repr output
        assert "b" * 500 not in out_err

    def test_text_boundary_to_stderr(self) -> None:
        """text_boundary_to_stderr: TextBoundary → stderr "[text_boundary]"; stdout empty."""
        stdout = io.StringIO()
        stderr = io.StringIO()
        evt = TextBoundary(sse_id=SID, kind="sentence", char_offset=128, ts="2026-05-21T00:00:00Z")
        _render_event(evt, stdout=stdout, stderr=stderr)
        assert stdout.getvalue() == ""
        out_err = stderr.getvalue()
        assert out_err.startswith("[text_boundary]")
        assert "kind=sentence" in out_err
        assert "char_offset=128" in out_err

    def test_invariant_inv003_stderr_only(self) -> None:
        """invariant_inv003_stderr_only [scenario]: …"""
        for evt in [
            WorkerPhase(sse_id=SID, phase="x", turn_id=42),
            Thinking(sse_id=SID, content="x"),
            TextBoundary(sse_id=SID, kind="x", char_offset=0, ts="t"),
            ToolStart(sse_id=SID, name="x", arguments={}),
            ToolResult(sse_id=SID, name="x", result=None, duration_ms=0),
            Error(sse_id=SID, phase="failed", message="m", error_code="e"),
            Cancelled(
                sse_id=SID, phase="cancelled", turn_id=42, reason="r", partial_message_id=None
            ),
        ]:
            stdout = io.StringIO()
            stderr = io.StringIO()
            _render_event(evt, stdout=stdout, stderr=stderr)
            assert stdout.getvalue() == "", f"INV-002 violated for {type(evt).__name__}"


class TestCancelAndLog:
    @respx.mock
    async def test_happy_cancel(self) -> None:
        """happy_cancel [happy,tracer]: 200 OK → returns None; stderr empty."""
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            return_value=httpx.Response(
                200,
                json={"turn_id": 42, "cancelled": True, "reason": None, "partial_message_id": None},
            )
        )
        stderr = io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            result = await _cancel_and_log(client, "s-1", 42, stderr=stderr)
        assert result is None
        assert stderr.getvalue() == ""

    @respx.mock
    async def test_cancel_failed_500(self) -> None:
        """cancel_failed_500 [error]: …"""
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            return_value=httpx.Response(500, content=b"boom")
        )
        stderr = io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            await _cancel_and_log(client, "s-1", 42, stderr=stderr)
        out = stderr.getvalue()
        assert "[cancel_failed]" in out
        assert "CancelFailed" in out

    @respx.mock
    async def test_cancel_already_completed(self) -> None:
        """cancel_already_completed [scenario]: …"""
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            return_value=httpx.Response(409)
        )
        stderr = io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            await _cancel_and_log(client, "s-1", 42, stderr=stderr)
        out = stderr.getvalue()
        assert "[cancel_failed]" in out
        assert "CancelAlreadyCompleted" in out

    @respx.mock
    async def test_cancel_turn_not_found(self) -> None:
        """cancel_turn_not_found [scenario]: 404 → returns None; stderr CancelTurnNotFound."""
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            return_value=httpx.Response(404)
        )
        stderr = io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            await _cancel_and_log(client, "s-1", 42, stderr=stderr)
        out = stderr.getvalue()
        assert "[cancel_failed]" in out
        assert "CancelTurnNotFound" in out

    @respx.mock
    async def test_transport_error_swallowed(self) -> None:
        """transport_error_swallowed [error]: …"""
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            side_effect=httpx.ConnectError("network down")
        )
        stderr = io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            await _cancel_and_log(client, "s-1", 42, stderr=stderr)
        out = stderr.getvalue()
        assert "[cancel_failed]" in out
        assert "ConnectError" in out


class _GatedStream(httpx.AsyncByteStream):
    """SSE byte stream: list of (bytes | asyncio.Event); Event entries pause until set."""

    def __init__(self, items: list[bytes | asyncio.Event]) -> None:
        self._items = items

    async def __aiter__(self):  # type: ignore[no-untyped-def]
        for item in self._items:
            if isinstance(item, asyncio.Event):
                await item.wait()
            else:
                yield item

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


class TestRunTurn:
    @respx.mock
    async def test_happy_text_then_done(self) -> None:
        """happy_text_then_done [happy,tracer]: …"""
        stream = _sse_chunk(
            "42:1", {"type": "text", "content": "hello"}
        ) + _sse_chunk("42:2", _DONE_BODY)
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=_sse_resp(stream)
        )
        sigint = asyncio.Event()
        stdout = io.StringIO()
        stderr = io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 0
        assert stdout.getvalue() == "hello\n"
        assert "[done]" in stderr.getvalue()

    @respx.mock
    async def test_error_terminal(self) -> None:
        """error_terminal: text + error → exit 2; stderr has [error]."""
        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/s-1/messages").mock(
            return_value=_sse_resp(stream)
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 2
        assert "[error]" in stderr.getvalue()

    @respx.mock
    async def test_cancelled_terminal_server(self) -> None:
        """cancelled_terminal_server: text + cancelled → exit 3; stderr has [cancelled]."""
        stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
            "42:2", _CANCELLED_BODY
        )
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=_sse_resp(stream)
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 3
        assert "[cancelled]" in stderr.getvalue()

    @respx.mock
    async def test_sse_connect_failed_404(self) -> None:
        """sse_connect_failed_404 [error]: 404 → exit 20; stderr [sse_connect_failed] status=404."""
        respx.post("https://w.example/sessions/missing/messages").mock(
            return_value=httpx.Response(404, json={"error": "session_not_found"})
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            exit_code = await _run_turn(
                client, "missing", "hi", sigint, stdout=stdout, stderr=stderr
            )
        assert exit_code == 20
        out = stderr.getvalue()
        assert "[sse_connect_failed]" in out
        assert "status=404" in out

    @respx.mock
    async def test_connection_dropped(self) -> None:
        """connection_dropped [error]: RemoteProtocolError mid-stream → exit 21."""

        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/s-1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, stream=_DropAfter([first])
            )
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 21
        assert "[connection_dropped]" in stderr.getvalue()

    @respx.mock
    async def test_malformed_sse_id(self) -> None:
        """malformed_sse_id [error]: id without seq → exit 22; stderr [malformed_sse_id]."""
        stream = b"id: 42\ndata: {\"type\": \"text\", \"content\": \"x\"}\n\n"
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=_sse_resp(stream)
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 22
        assert "[malformed_sse_id]" in stderr.getvalue()

    @respx.mock
    async def test_turn_id_flip(self) -> None:
        """turn_id_flip [error]: …"""
        stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
            "99:2", {"type": "text", "content": "y"}
        )
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=_sse_resp(stream)
        )
        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 22
        out = stderr.getvalue()
        assert "[turn_id_flip]" in out
        assert "expected=42" in out
        assert "got=99" in out

    @respx.mock
    async def test_sigint_before_first_event(self) -> None:
        """sigint_before_first_event [scenario]: …"""
        gate = asyncio.Event()
        # Stream never yields anything until gate (the gate is never set; the test exits via sigint)
        stream = _GatedStream([gate])
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=_sse_resp(stream)
        )
        cancel_route = respx.post("https://w.example/sessions/s-1/turns").mock(
            return_value=httpx.Response(200, json=_CANCEL_OK_RESP)
        )
        sigint = asyncio.Event()
        sigint.set()  # SIGINT before _run_turn even starts
        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            exit_code = await asyncio.wait_for(
                _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr), timeout=2.0
            )
        assert exit_code == 3
        assert "[cancelled] (before any event arrived)" in stderr.getvalue()
        assert cancel_route.call_count == 0

    @respx.mock
    async def test_sigint_mid_stream_drains_to_cancelled(self) -> None:
        """sigint_mid_stream_drains_to_cancelled [scenario,tracer]: …"""
        cancel_observed = asyncio.Event()

        def cancel_handler(req: httpx.Request) -> httpx.Response:
            cancel_observed.set()
            return httpx.Response(200, json=_CANCEL_OK_RESP)

        cancel_route = respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            side_effect=cancel_handler
        )
        text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
        cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY)
        stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk])
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=_sse_resp(stream)
        )

        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            task = asyncio.create_task(
                _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
            )
            # Wait for the first event to flush to stdout (signals last_turn_id is set)
            for _ in range(50):
                if "x" in stdout.getvalue():
                    break
                await asyncio.sleep(0.01)
            else:
                task.cancel()
                pytest.fail("text event never reached stdout")
            sigint.set()
            exit_code = await asyncio.wait_for(task, timeout=2.0)
        assert exit_code == 3
        assert cancel_route.call_count == 1

    @respx.mock
    async def test_sigint_twice_issues_one_cancel(self) -> None:
        """sigint_twice_issues_one_cancel [scenario]: sigint set twice → one cancel POST."""
        cancel_observed = asyncio.Event()

        def cancel_handler(req: httpx.Request) -> httpx.Response:
            cancel_observed.set()
            return httpx.Response(200, json=_CANCEL_OK_RESP)

        cancel_route = respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            side_effect=cancel_handler
        )
        text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
        cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY)
        stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk])
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=_sse_resp(stream)
        )

        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            task = asyncio.create_task(
                _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
            )
            for _ in range(50):
                if "x" in stdout.getvalue():
                    break
                await asyncio.sleep(0.01)
            sigint.set()
            # Set again — should be no-op (event is already set; idempotent)
            sigint.set()
            exit_code = await asyncio.wait_for(task, timeout=2.0)
        assert exit_code == 3
        assert cancel_route.call_count == 1

    @respx.mock
    async def test_no_busy_loop_after_cancel(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """no_busy_loop_after_cancel [trace]: only one sigint_event.wait()-task created."""
        cancel_observed = asyncio.Event()

        def cancel_handler(req: httpx.Request) -> httpx.Response:
            cancel_observed.set()
            return httpx.Response(200, json=_CANCEL_OK_RESP)

        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            side_effect=cancel_handler
        )
        text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
        cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY)
        stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk])
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=_sse_resp(stream)
        )

        sigint = asyncio.Event()
        wait_call_count = 0
        original_wait = sigint.wait

        async def counting_wait() -> bool:
            nonlocal wait_call_count
            wait_call_count += 1
            return await original_wait()

        monkeypatch.setattr(sigint, "wait", counting_wait)

        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            task = asyncio.create_task(
                _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
            )
            for _ in range(50):
                if "x" in stdout.getvalue():
                    break
                await asyncio.sleep(0.01)
            sigint.set()
            exit_code = await asyncio.wait_for(task, timeout=2.0)
        assert exit_code == 3
        # INV-007 + busy-loop fix: sigint.wait() created at most once per pre-cancelling
        # iteration. For text → sigint → cancelled, that's iter 1 (raced w/ text) and
        # iter 2 (raced w/ sigint; flipped cancelling=True). Iter 3+ MUST skip wait()
        # creation entirely — the busy-loop bug would make this number grow unbounded.
        assert wait_call_count == 2

    @respx.mock
    async def test_cancel_failed_drains_anyway(self) -> None:
        """cancel_failed_drains_anyway [scenario]: …"""
        cancel_observed = asyncio.Event()

        def cancel_handler(req: httpx.Request) -> httpx.Response:
            cancel_observed.set()
            return httpx.Response(500, content=b"boom")

        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            side_effect=cancel_handler
        )
        text_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
        cancelled_chunk = _sse_chunk("42:2", _CANCELLED_BODY)
        stream = _GatedStream([text_chunk, cancel_observed, cancelled_chunk])
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=_sse_resp(stream)
        )

        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            task = asyncio.create_task(
                _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
            )
            for _ in range(50):
                if "x" in stdout.getvalue():
                    break
                await asyncio.sleep(0.01)
            sigint.set()
            exit_code = await asyncio.wait_for(task, timeout=2.0)
        assert exit_code == 3
        # INV-009: cancel POST failed but stream still drained to cancelled terminal
        assert "[cancel_failed]" in stderr.getvalue()
        assert "[cancelled]" in stderr.getvalue()

    @respx.mock
    async def test_render_called_once_per_event(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """render_called_once_per_event [trace]: spy → _render_event call_count == event count."""
        chunks = [
            _sse_chunk("42:1", {"type": "worker_phase", "phase": "streaming", "turn_id": 42}),
            _sse_chunk("42:2", {"type": "text", "content": "hi"}),
            _sse_chunk("42:3", _DONE_BODY),
        ]
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=httpx.Response(
                200, headers={"content-type": "text/event-stream"}, content=b"".join(chunks)
            )
        )

        call_count = 0
        from ratatoskr import cli as cli_mod

        original = cli_mod._render_event

        def spy(event, **kw):  # type: ignore[no-untyped-def]
            nonlocal call_count
            call_count += 1
            return original(event, **kw)

        monkeypatch.setattr(cli_mod, "_render_event", spy)

        sigint = asyncio.Event()
        stdout, stderr = io.StringIO(), io.StringIO()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            exit_code = await _run_turn(client, "s-1", "hi", sigint, stdout=stdout, stderr=stderr)
        assert exit_code == 0
        assert call_count == 3


_PARSED_NEW = ParsedArgs(
    send_content="hi",
    session_id=None,
    new=True,
    agent_id="mimir",
    api_key="k",
    server_url="https://w.example",
    raw=False,
)
_PARSED_EXISTING = ParsedArgs(
    send_content="hi",
    session_id="s-1",
    new=False,
    agent_id=None,
    api_key="k",
    server_url="https://w.example",
    raw=False,
)
_CREATE_OK_RESP = {
    "session_id": "s-new",
    "agent_id": "mimir",
    "message_count": 0,
    "created_at": "2026-05-21T00:00:00+00:00",
    "last_active": "2026-05-21T00:00:00+00:00",
    "metadata": {},
}


class TestAmain:
    @respx.mock
    async def test_happy_new_session_then_stream(self, capsys: pytest.CaptureFixture[str]) -> None:
        """happy_new_session_then_stream [happy,tracer]: …"""
        respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(201, json=_CREATE_OK_RESP)
        )
        sse_body = _sse_chunk("42:1", {"type": "text", "content": "hello"}) + _sse_chunk(
            "42:2", _DONE_BODY
        )
        respx.post("https://w.example/sessions/s-new/messages").mock(
            return_value=_sse_resp(sse_body)
        )
        exit_code = await _amain(_PARSED_NEW)
        assert exit_code == 0
        captured = capsys.readouterr()
        err = captured.err
        assert "[create_session]" in err
        assert "[done]" in err
        assert err.index("[create_session]") < err.index("[done]")

    @respx.mock
    async def test_happy_existing_session(self, capsys: pytest.CaptureFixture[str]) -> None:
        """happy_existing_session: --session, no create POST; just SSE stream → exit 0."""
        sessions_route = respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(201, json=_CREATE_OK_RESP)
        )
        sse_body = _sse_chunk("42:1", {"type": "text", "content": "hi"}) + _sse_chunk(
            "42:2", _DONE_BODY
        )
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=_sse_resp(sse_body)
        )
        exit_code = await _amain(_PARSED_EXISTING)
        assert exit_code == 0
        assert sessions_route.call_count == 0

    @respx.mock
    async def test_agent_not_found_exits_12(self, capsys: pytest.CaptureFixture[str]) -> None:
        """agent_not_found_exits_12 [error]: POST /sessions → 404 → exit 12; no stream_turn."""
        respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(404, json={"error": "unknown_agent_id"})
        )
        stream_route = respx.post("https://w.example/sessions/s-new/messages").mock(
            return_value=httpx.Response(200)
        )
        exit_code = await _amain(_PARSED_NEW)
        assert exit_code == 12
        assert "[agent_not_found]" in capsys.readouterr().err
        assert stream_route.call_count == 0

    @respx.mock
    async def test_session_api_failed_exits_20(self, capsys: pytest.CaptureFixture[str]) -> None:
        """session_api_failed_exits_20: POST /sessions → 500 → exit 20; [session_api_failed]."""
        respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(500, content=b"server error")
        )
        exit_code = await _amain(_PARSED_NEW)
        assert exit_code == 20
        err = capsys.readouterr().err
        assert "[session_api_failed]" in err
        assert "status=500" in err

    @respx.mock
    async def test_connect_error_exits_21(self, capsys: pytest.CaptureFixture[str]) -> None:
        """connect_error_exits_21: httpx.ConnectError on POST /sessions → exit 21."""
        respx.post("https://w.example/sessions").mock(side_effect=httpx.ConnectError("down"))
        exit_code = await _amain(_PARSED_NEW)
        assert exit_code == 21
        assert "[network_error]" in capsys.readouterr().err

    @respx.mock
    async def test_sigint_handler_installed_and_removed(self) -> None:
        """sigint_handler_installed_and_removed [trace]: signal handler add/remove paired."""
        sse_body = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
            "42:2", _DONE_BODY
        )
        respx.post("https://w.example/sessions/s-1/messages").mock(
            return_value=_sse_resp(sse_body)
        )
        loop = asyncio.get_running_loop()
        original_add = loop.add_signal_handler
        original_remove = loop.remove_signal_handler
        add_calls: list[int] = []
        remove_calls: list[int] = []

        def add_spy(sig, callback, *args):  # type: ignore[no-untyped-def]
            add_calls.append(sig)
            return original_add(sig, callback, *args)

        def remove_spy(sig):  # type: ignore[no-untyped-def]
            remove_calls.append(sig)
            return original_remove(sig)

        loop.add_signal_handler = add_spy  # type: ignore[method-assign]
        loop.remove_signal_handler = remove_spy  # type: ignore[method-assign]
        try:
            exit_code = await _amain(_PARSED_EXISTING)
        finally:
            loop.add_signal_handler = original_add  # type: ignore[method-assign]
            loop.remove_signal_handler = original_remove  # type: ignore[method-assign]
        import signal as _sig

        assert exit_code == 0
        assert add_calls == [_sig.SIGINT]
        assert remove_calls == [_sig.SIGINT]

    def test_no_textual_import(self) -> None:
        """no_textual_import [scenario]: …"""
        import importlib
        import sys

        # Clear any prior textual import to make this test honest in isolation
        textual_was_imported = "textual" in sys.modules
        # We cannot reliably remove textual mid-suite (other tests might rely on it via dev deps),
        # so the assertion is: importing ratatoskr.cli does not REQUIRE textual.
        importlib.reload(__import__("ratatoskr.cli", fromlist=["_amain"]))
        # The boundary is the INV-001 import-only rule. If ratatoskr/cli.py grew an
        # `import textual` directly, the import would still succeed (textual is installed)
        # but the source-level boundary is the load-bearing check — covered by a static-grep
        # smoke test pattern. Do that here:
        import pathlib

        src = pathlib.Path(__file__).parent.parent / "src" / "ratatoskr" / "cli.py"
        text = src.read_text()
        for forbidden in ("import textual", "from textual", "import rich", "from rich"):
            assert forbidden not in text, f"INV-001 violation: cli.py contains '{forbidden}'"
        _ = textual_was_imported  # avoid unused warning


class TestMain:
    def test_happy_returns_amain_exit_code(
        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """happy_returns_amain_exit_code [happy,tracer]: …"""

        async def fake_amain(args: ParsedArgs) -> int:
            assert args.send_content == "hi"
            return 0

        monkeypatch.setattr(cli_mod, "_amain", fake_amain)
        rc = main(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
        assert rc == 0

    def test_usage_error_no_send(
        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """usage_error_no_send: empty argv → exit 10; stderr [usage_error]; _amain never called."""
        amain_calls: list[int] = []

        async def fake_amain(args: ParsedArgs) -> int:
            amain_calls.append(1)
            return 0

        monkeypatch.setattr(cli_mod, "_amain", fake_amain)
        rc = main([])
        assert rc == 10
        assert "[usage_error]" in capsys.readouterr().err
        assert amain_calls == []

    def test_usage_error_both_session_and_new(
        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """usage_error_both_session_and_new: both flags → exit 10; [usage_error]."""

        async def fake_amain(args: ParsedArgs) -> int:
            return 0

        monkeypatch.setattr(cli_mod, "_amain", fake_amain)
        rc = main(["--send", "hi", "--session", "s", "--new", "--agent", "m", "--api-key", "k"])
        assert rc == 10
        assert "[usage_error]" in capsys.readouterr().err

    def test_auth_error_missing_key(self, capsys: pytest.CaptureFixture[str]) -> None:
        """auth_error_missing_key: …"""
        # _clear_env fixture has already deleted WORLDTREE_API_KEY
        rc = main(["--send", "hi", "--new", "--agent", "m"])
        assert rc == 11
        assert "[auth_error]" in capsys.readouterr().err

    def test_no_argv_uses_sys_argv(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """no_argv_uses_sys_argv [trace]: argv=None → _parse_args reads sys.argv[1:]."""
        monkeypatch.setattr(
            "sys.argv",
            ["ratatoskr", "--send", "hi", "--new", "--agent", "m", "--api-key", "k"],
        )

        async def fake_amain(args: ParsedArgs) -> int:
            assert args.send_content == "hi"
            assert args.agent_id == "m"
            return 0

        monkeypatch.setattr(cli_mod, "_amain", fake_amain)
        rc = main(None)
        assert rc == 0

    def test_help_exits_cleanly(
        self,
        monkeypatch: pytest.MonkeyPatch,
        capsys: pytest.CaptureFixture[str],
    ) -> None:
        """help_exits_cleanly [happy]: --help → main returns 0; _amain never called."""
        amain_calls: list[int] = []

        async def fake_amain(args: ParsedArgs) -> int:
            amain_calls.append(1)
            return 0

        monkeypatch.setattr(cli_mod, "_amain", fake_amain)
        rc = main(["--help"])
        assert rc == 0
        assert amain_calls == []
        # argparse prints help text to stdout
        assert "ratatoskr" in capsys.readouterr().out

    def test_no_send_dispatches_to_tui(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """no_send_dispatches_to_tui: --send omitted → main calls run_tui, NOT _amain."""
        from ratatoskr import tui as tui_mod

        tui_calls: list[ParsedArgs] = []
        amain_calls: list[int] = []

        def fake_run_tui(args: ParsedArgs) -> int:
            tui_calls.append(args)
            return 0

        async def fake_amain(args: ParsedArgs) -> int:
            amain_calls.append(1)
            return 0

        monkeypatch.setattr(tui_mod, "run_tui", fake_run_tui)
        monkeypatch.setattr(cli_mod, "_amain", fake_amain)
        rc = main(["--session", "s-1", "--api-key", "k"])
        assert rc == 0
        assert len(tui_calls) == 1
        assert tui_calls[0].send_content is None
        assert tui_calls[0].session_id == "s-1"
        assert amain_calls == []