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,
    _run_turn,
    main,
)
from ratatoskr.sessions import BifrostBinding
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)
    monkeypatch.delenv("RATATOSKR_END_USER_ID", raising=False)
    monkeypatch.delenv("RATATOSKR_BIFROST_CONSUMER_KEY", raising=False)
    monkeypatch.delenv("RATATOSKR_PROVIDER_VISIBLE_HOST", 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: --send with neither flag → UsageError.

        --send is non-interactive (no picker can open), so a session must be
        named. Bare TUI mode (no --send) is now valid → session picker (§4).
        """
        with pytest.raises(UsageError, match="--send requires"):
            _parse_args(["--send", "hi", "--api-key", "k"])

    def test_bare_tui_mode_accepted(self) -> None:
        """bare_tui_mode (slice b2): no --send, no --session, no --new → valid;
        _resolve_then_run drives the startup session picker (design-brief §4)."""
        args = _parse_args(["--api-key", "k"])
        assert args.send_content is None
        assert args.session_id is None
        assert args.new is False
        assert args.agent_id is None

    def test_usage_bare_tui_with_agent(self) -> None:
        """bare_tui_with_agent (slice b2): bare TUI + --agent → UsageError
        (--agent belongs with --new; bare mode opens the resume picker)."""
        with pytest.raises(UsageError, match="belongs with --new"):
            _parse_args(["--agent", "mimir", "--api-key", "k"])

    def test_usage_send_new_without_agent(self) -> None:
        """send_new_without_agent (issue #8): --send --new without --agent → UsageError.

        --send mode is non-interactive — cannot prompt; --agent stays required.
        """
        with pytest.raises(UsageError, match="--agent is required when --new"):
            _parse_args(["--send", "hi", "--new", "--api-key", "k"])

    def test_parse_bare_new_without_agent_accepted(self) -> None:
        """bare_new_without_agent (issue #8): --new without --send or --agent → agent_id=None.

        TUI mode CAN prompt; startup picker handles the choice. _parse_args
        accepts None here and the TUI's _resolve_then_run drives the picker.
        """
        args = _parse_args(["--new", "--api-key", "k"])
        assert args.new is True
        assert args.agent_id is None
        assert args.send_content is None

    def test_parse_bare_new_with_agent_accepted(self) -> None:
        """bare_new_with_agent (issue #8): --new --agent mimir (no --send) → picker skipped.

        Existing TUI launch path with an explicit agent_id continues to work
        — _resolve_then_run sees `args.agent_id is not None` and skips the
        picker entirely.
        """
        args = _parse_args(["--new", "--agent", "mimir", "--api-key", "k"])
        assert args.new is True
        assert args.agent_id == "mimir"
        assert args.send_content is None

    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"])

    def test_happy_new_with_end_user_id(self) -> None:
        """happy_new_with_end_user_id [happy]: --end-user-id alice → end_user_id='alice'."""
        args = _parse_args(
            ["--send", "hi", "--new", "--agent", "lofn", "--api-key", "k",
             "--end-user-id", "alice"]
        )
        assert args.end_user_id == "alice"
        assert args.agent_id == "lofn"

    def test_end_user_id_default_none(self) -> None:
        """end_user_id_default_none [trace]: omit --end-user-id → ParsedArgs.end_user_id is None."""
        args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
        assert args.end_user_id is None

    def test_empty_end_user_id(self) -> None:
        """empty_end_user_id [adversarial]: --end-user-id '' → UsageError (mirrors empty --send)."""
        with pytest.raises(UsageError, match="--end-user-id"):
            _parse_args(
                ["--send", "hi", "--new", "--agent", "m", "--api-key", "k", "--end-user-id", ""]
            )

    def test_end_user_id_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """end_user_id_from_env [trace]: $RATATOSKR_END_USER_ID fills when flag omitted."""
        monkeypatch.setenv("RATATOSKR_END_USER_ID", "ratatoskr-tui")
        args = _parse_args(["--send", "hi", "--new", "--agent", "m", "--api-key", "k"])
        assert args.end_user_id == "ratatoskr-tui"

    def test_end_user_id_flag_beats_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """end_user_id_flag_beats_env [trace]: explicit --end-user-id wins over env."""
        monkeypatch.setenv("RATATOSKR_END_USER_ID", "from-env")
        args = _parse_args(
            ["--send", "hi", "--new", "--agent", "m", "--api-key", "k",
             "--end-user-id", "from-flag"]
        )
        assert args.end_user_id == "from-flag"


SID = SseId(42, 5)


# Issue #12 — presenter contract semantics amendment.
# CliPresenterState replaces the stateless _render_event with a stateful per-turn
# presenter that coalesces thinking runs and demotes telemetry events.
# (Pre-amendment TestRenderEvent class and `_render_event` function have been
# removed under the project's no-backwards-compatibility rule.)

SID42 = SseId(42, 1)


class TestCliPresenterState:
    """Tests for the new CliPresenterState — per issue #12 contract."""

    def test_thinking_coalesce_single_run(self) -> None:
        """thinking_coalesce_single_run [happy,tracer]:
        Thinking("hello") + Thinking(" world") + Done →
        stderr has ". thinking: hello world\\n" followed by the [done] line.
        """
        from ratatoskr.cli import CliPresenterState

        stdout = io.StringIO()
        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(Thinking(sse_id=SID42, content="hello"), stdout=stdout, stderr=stderr)
        # After first delta: stderr has the open prefix + content, no \n yet.
        assert stderr.getvalue() == ". thinking: hello"
        state.render(Thinking(sse_id=SID42, content=" world"), stdout=stdout, stderr=stderr)
        # After second delta: still the same growing logical line, still no \n.
        assert stderr.getvalue() == ". thinking: hello world"
        # Now a Done event closes the thinking run with \n then writes the terminal label.
        done = Done(
            sse_id=SID42,
            phase="succeeded",
            response="hi",
            model="m",
            duration_ms=1,
            usage={
                "prompt_tokens": 0,
                "completion_tokens": 0,
                "total_tokens": 0,
                "cached_input_tokens": 0,
            },
        )
        state.render(done, stdout=stdout, stderr=stderr)
        captured = stderr.getvalue()
        # Thinking run closed with \n; terminal label landed; no demotion prefix on [done].
        assert captured.startswith(". thinking: hello world\n")
        assert "[done]" in captured
        # stdout untouched (no Text events were rendered)
        assert stdout.getvalue() == ""

    def test_thinking_closes_on_first_non_thinking_event(self) -> None:
        """thinking_closes_on_first_non_thinking_event [happy]:
        Thinking → WorkerPhase → stderr has ". thinking: ...\\n" then ". worker_phase: ..."
        """
        from ratatoskr.cli import CliPresenterState

        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=stderr)
        state.render(
            WorkerPhase(sse_id=SID42, phase="streaming", turn_id=42),
            stdout=io.StringIO(),
            stderr=stderr,
        )
        out = stderr.getvalue()
        # Thinking run closed; worker_phase rendered with demotion prefix.
        assert ". thinking: x\n" in out
        assert ". worker_phase:" in out
        # `[worker_phase]` (bracketed, pre-amendment shape) MUST NOT appear.
        assert "[worker_phase]" not in out

    def test_thinking_closes_on_error(self) -> None:
        """thinking_closes_on_error [error]:
        Thinking → Error → thinking line closes with \\n, then [error] line rendered
        (partial thinking content is NOT discarded — observability requirement).
        """
        from ratatoskr.cli import CliPresenterState

        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=stderr)
        state.render(
            Error(sse_id=SID42, phase="failed", message="boom", error_code="bad"),
            stdout=io.StringIO(),
            stderr=stderr,
        )
        out = stderr.getvalue()
        # Partial thinking preserved with closing \n; error rendered without demotion prefix.
        assert ". thinking: x\n" in out
        assert "[error]" in out
        # Demotion prefix MUST NOT precede [error]: it's load-bearing.
        assert ". [error]" not in out

    def test_cancelled_mid_thinking(self) -> None:
        """cancelled_mid_thinking [scenario]: Thinking → Cancelled → thinking closes with \\n;
        then [cancelled] (no demotion prefix, partial thinking preserved).
        """
        from ratatoskr.cli import CliPresenterState

        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=stderr)
        state.render(
            Cancelled(
                sse_id=SID42, phase="cancelled", turn_id=42, reason="user", partial_message_id=None
            ),
            stdout=io.StringIO(),
            stderr=stderr,
        )
        out = stderr.getvalue()
        assert ". thinking: x\n" in out
        assert "[cancelled]" in out
        assert ". [cancelled]" not in out

    def test_text_then_done_newline_boundary(self) -> None:
        """text_then_done_newline_boundary [trace]: Text("answer") → Done;
        stdout receives "answer\\n" (the \\n is the INV-005 boundary), stderr has "[done] ...".
        """
        from ratatoskr.cli import CliPresenterState

        stdout = io.StringIO()
        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(Text(sse_id=SID42, content="answer"), stdout=stdout, stderr=stderr)
        state.render(_make_done(), stdout=stdout, stderr=stderr)
        # INV-005: text without trailing \n → exactly one \n gets injected before terminal label
        assert stdout.getvalue() == "answer\n"
        assert "[done]" in stderr.getvalue()

    def test_no_text_then_done_no_extra_newline(self) -> None:
        """no_text_then_done_no_extra_newline [trace]: Done with no preceding Text →
        stdout untouched; stderr receives only "[done] ...".
        """
        from ratatoskr.cli import CliPresenterState

        stdout = io.StringIO()
        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(_make_done(), stdout=stdout, stderr=stderr)
        # INV-005 boundary fires ONLY when text was written; no text → no \n injection.
        assert stdout.getvalue() == ""
        assert "[done]" in stderr.getvalue()

    def test_newline_terminated_text_then_done(self) -> None:
        """newline_terminated_text_then_done [trace]: Text("answer\\n") → Done;
        stdout receives "answer\\n" exactly ONCE (no double-newline before [done]).
        Tests the F4 Volva fix: text_written_since_newline tracks last-char-was-\\n.
        """
        from ratatoskr.cli import CliPresenterState

        stdout = io.StringIO()
        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(Text(sse_id=SID42, content="answer\n"), stdout=stdout, stderr=stderr)
        state.render(_make_done(), stdout=stdout, stderr=stderr)
        # POST-003: content ends with \n → state.text_written_since_newline = False
        #         → INV-005 does NOT inject an extra \n before [done].
        assert stdout.getvalue() == "answer\n"

    def test_multiple_thinking_runs(self) -> None:
        """multiple_thinking_runs [scenario]: Thinking → Text → Thinking → Done →
        TWO separate ". thinking: ..." runs in stderr; stdout has the text + INV-005 boundary.
        """
        from ratatoskr.cli import CliPresenterState

        stdout = io.StringIO()
        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(Thinking(sse_id=SID42, content="first"), stdout=stdout, stderr=stderr)
        state.render(Text(sse_id=SID42, content="answer"), stdout=stdout, stderr=stderr)
        state.render(Thinking(sse_id=SID42, content="second"), stdout=stdout, stderr=stderr)
        state.render(_make_done(), stdout=stdout, stderr=stderr)
        err = stderr.getvalue()
        # Each thinking RUN gets its own ". thinking: " prefix.
        assert err.count(". thinking: ") == 2
        assert ". thinking: first" in err
        assert ". thinking: second" in err
        assert stdout.getvalue() == "answer\n"
        assert "[done]" in err


    def test_tool_start_demoted(self) -> None:
        """tool_start_demoted [trace]: ToolStart → stderr line starts with ". tool_start:" """
        from ratatoskr.cli import CliPresenterState

        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(
            ToolStart(sse_id=SID42, name="read_file", arguments={"path": "/x"}),
            stdout=io.StringIO(),
            stderr=stderr,
        )
        line = stderr.getvalue()
        assert line.startswith(". tool_start:")
        assert "[tool_start]" not in line

    def test_tool_result_truncated(self) -> None:
        """tool_result_truncated [trace]: long result repr truncates to ≤200 chars."""
        from ratatoskr.cli import CliPresenterState

        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(
            ToolResult(sse_id=SID42, name="x", result="b" * 500, duration_ms=42),
            stdout=io.StringIO(),
            stderr=stderr,
        )
        line = stderr.getvalue()
        assert line.startswith(". tool_result:")
        # Full 500-char result MUST NOT fit; truncation applied.
        assert "b" * 500 not in line

    def test_text_boundary_demoted(self) -> None:
        """text_boundary_demoted [trace]: TextBoundary → stderr ". text_boundary:" prefix."""
        from ratatoskr.cli import CliPresenterState

        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(
            TextBoundary(sse_id=SID42, kind="sentence", char_offset=128, ts="2026-05-21T00:00:00Z"),
            stdout=io.StringIO(),
            stderr=stderr,
        )
        line = stderr.getvalue()
        assert line.startswith(". text_boundary:")
        assert "[text_boundary]" not in line

    def test_duration_format_seconds(self) -> None:
        """duration_format_seconds [trace]: Done(duration_ms=5467) → "duration=5.5s"."""
        from ratatoskr.cli import CliPresenterState

        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(_make_done(duration_ms=5467), stdout=io.StringIO(), stderr=stderr)
        assert "duration=5.5s" in stderr.getvalue()
        assert "duration_ms=5467" not in stderr.getvalue()

    def test_duration_format_subsecond(self) -> None:
        """duration_format_subsecond [trace]: Done(duration_ms=347) → "duration=347ms"."""
        from ratatoskr.cli import CliPresenterState

        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(_make_done(duration_ms=347), stdout=io.StringIO(), stderr=stderr)
        assert "duration=347ms" in stderr.getvalue()

    def test_duration_format_minutes(self) -> None:
        """duration_format_minutes [trace]: Done(duration_ms=72000) → "duration=1.2m"."""
        from ratatoskr.cli import CliPresenterState

        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(_make_done(duration_ms=72000), stdout=io.StringIO(), stderr=stderr)
        assert "duration=1.2m" in stderr.getvalue()

    def test_usage_format_ascii_arrow(self) -> None:
        """usage_format_ascii_arrow [trace]: stderr label contains the natural-language
        usage shape with ASCII arrow (-> not →) for CLI scriptability.
        """
        from ratatoskr.cli import CliPresenterState

        stderr = io.StringIO()
        state = CliPresenterState()
        state.render(
            _make_done(
                usage={
                    "prompt_tokens": 6756,
                    "completion_tokens": 126,
                    "total_tokens": 6882,
                    "cached_input_tokens": 0,
                }
            ),
            stdout=io.StringIO(),
            stderr=stderr,
        )
        out = stderr.getvalue()
        assert "usage 6756 in -> 126 out (6882 total, 0 cached)" in out
        # Raw dict shape MUST NOT leak through.
        assert "'prompt_tokens'" not in out

    def test_state_reset_per_amain(self) -> None:
        """state_reset_per_amain [trace]: fresh CliPresenterState() starts with no thinking open."""
        from ratatoskr.cli import CliPresenterState

        # Simulate two _amain calls by constructing two independent states.
        s1 = CliPresenterState()
        s2 = CliPresenterState()
        # Run thinking into s1 — it should NOT bleed into s2.
        s1.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=io.StringIO())
        assert s1.thinking_open is True
        assert s2.thinking_open is False
        # s2's first render produces its own ". thinking: " prefix.
        e2 = io.StringIO()
        s2.render(Thinking(sse_id=SID42, content="y"), stdout=io.StringIO(), stderr=e2)
        assert e2.getvalue() == ". thinking: y"


class TestFormatDurationMs:
    """Unit tests for _format_duration_ms per INV-006."""

    def test_subsecond(self) -> None:
        from ratatoskr.cli import _format_duration_ms
        assert _format_duration_ms(347) == "347ms"

    def test_exact_one_second(self) -> None:
        from ratatoskr.cli import _format_duration_ms
        assert _format_duration_ms(1000) == "1.0s"

    def test_fractional_seconds(self) -> None:
        from ratatoskr.cli import _format_duration_ms
        assert _format_duration_ms(5467) == "5.5s"

    def test_exact_one_minute(self) -> None:
        from ratatoskr.cli import _format_duration_ms
        assert _format_duration_ms(60000) == "1.0m"

    def test_fractional_minutes(self) -> None:
        from ratatoskr.cli import _format_duration_ms
        assert _format_duration_ms(72000) == "1.2m"

    def test_zero(self) -> None:
        from ratatoskr.cli import _format_duration_ms
        assert _format_duration_ms(0) == "0ms"


class TestFormatUsage:
    """Unit tests for _format_usage per INV-007."""

    def test_ascii_arrow(self) -> None:
        from ratatoskr.cli import _format_usage
        usage = {
            "prompt_tokens": 6756,
            "completion_tokens": 126,
            "total_tokens": 6882,
            "cached_input_tokens": 0,
        }
        assert (
            _format_usage(usage, arrow="->")
            == "6756 in -> 126 out (6882 total, 0 cached)"
        )

    def test_unicode_arrow(self) -> None:
        from ratatoskr.cli import _format_usage
        usage = {
            "prompt_tokens": 6756,
            "completion_tokens": 126,
            "total_tokens": 6882,
            "cached_input_tokens": 0,
        }
        assert (
            _format_usage(usage, arrow="→")
            == "6756 in → 126 out (6882 total, 0 cached)"
        )


_USAGE_ZERO: dict[str, int] = {
    "prompt_tokens": 0,
    "completion_tokens": 0,
    "total_tokens": 0,
    "cached_input_tokens": 0,
}


def _make_done(*, duration_ms: int = 1, usage: dict[str, int] | None = None) -> Done:
    return Done(
        sse_id=SID42,
        phase="succeeded",
        response="r",
        model="m",
        duration_ms=duration_ms,
        usage=usage if usage is not None else _USAGE_ZERO,
    )


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_malformed_sse_data(self) -> None:
        """malformed_sse_data [error]: text + non-JSON → exit 22; [malformed_sse_data] raw='...'."""
        stream = (
            _sse_chunk("42:1", {"type": "text", "content": "x"})
            + b"id: 42:2\ndata: not-json\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
        out = stderr.getvalue()
        # Exact label + `raw='X'` shape — the stderr format the contract specifies
        assert "[malformed_sse_data] raw='not-json'" in out

    @respx.mock
    async def test_malformed_sse_data_truncation(self) -> None:
        """malformed_sse_data_truncation [security]: 5000-char bad data → raw truncated."""
        huge_bad = "x" * 5000
        stream = (
            _sse_chunk("42:1", {"type": "text", "content": "x"})
            + f"id: 42:2\ndata: {huge_bad}\n\n".encode()
        )
        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 "[malformed_sse_data]" in out
        # MalformedSseData.raw was truncated to 200 chars at the exception layer;
        # presenter's `repr()` rendering of that 200-char string carries through.
        # Full 5000-char payload MUST NOT appear in stderr.
        assert "x" * 5000 not in out
        assert "x" * 200 in out  # the truncated form IS in the rendered raw='...'

    @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 on render; call_count == event count.

        Per issue #12: rendering went from stateless `_render_event` to
        `CliPresenterState.render`; the spy moves accordingly.
        """
        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)
            )
        )

        from ratatoskr.cli import CliPresenterState

        call_count = 0
        original = CliPresenterState.render

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

        monkeypatch.setattr(CliPresenterState, "render", 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_NEW_WITH_END_USER = ParsedArgs(
    send_content="hi",
    session_id=None,
    new=True,
    agent_id="mimir",
    api_key="k",
    server_url="https://w.example",
    raw=False,
    end_user_id="alice",
)
_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_user_agent_header_sent(self) -> None:
        """user_agent_header_sent [trace]: outbound requests carry the ratatoskr User-Agent.

        Worldtree-dev (althing 2026-05-23) requested consumers send User-Agent
        so server logs can distinguish ratatoskr traffic.
        """
        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-new/messages").mock(
            return_value=_sse_resp(sse_body)
        )
        await _amain(_PARSED_NEW)
        ua = sessions_route.calls[0].request.headers["User-Agent"]
        assert ua.startswith("ratatoskr/")
        assert "vh@phasefinal.com" in ua

    @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
        # Per issue #12: [create_session] lifecycle line demoted to `. create_session:`.
        assert ". create_session:" in err
        assert "[create_session]" not in err  # pre-amendment shape forbidden
        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]: …"""
        # INV-001 import-only boundary: cli.py must not import textual/rich at the
        # source level. The load-bearing check is a static source grep (NOT a live
        # `importlib.reload`, which would mutate the shared module in place and break
        # class identity — isinstance / pytest.raises — for every later test).
        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}'"


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_empty_argv_fails_on_auth(
        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """empty argv → exit 11 [auth_error]; _amain never called.

        Since slice b2 bare TUI mode (no --send/--session/--new) is VALID (it
        opens the session picker), so empty argv is no longer a usage error —
        it now fails on the missing API key instead (still before _amain).
        """
        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 == 11
        assert "[auth_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 == []


class TestBifrostBindCli:
    """Issue #17 slice 3a — the CLI Bifrost-bind trigger (INV-008, one of three)."""

    def test_plane_and_host_build_binding(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """tracer: --bifrost-plane + --bifrost-host resolve a BifrostBinding via
        endpoint_for_plane; the consumer key comes from the env."""
        monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
        args = _parse_args(
            [
                "--send", "hi", "--new", "--agent", "ratatoskr:sindra", "--api-key", "k",
                "--bifrost-plane", "memory", "--bifrost-host", "10.100.10.50",
            ]
        )
        assert args.bifrost == BifrostBinding(endpoint_url="http://10.100.10.50:8391")
        assert args.bifrost_plane == "memory"
        assert args.consumer_key == "ck"

    def test_host_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """--bifrost-host falls back to RATATOSKR_PROVIDER_VISIBLE_HOST."""
        monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
        monkeypatch.setenv("RATATOSKR_PROVIDER_VISIBLE_HOST", "10.0.0.9")
        args = _parse_args(
            ["--send", "hi", "--new", "--agent", "a", "--api-key", "k",
             "--bifrost-plane", "affect"]
        )
        assert args.bifrost == BifrostBinding(endpoint_url="http://10.0.0.9:8390")

    def test_direct_url_bypasses_plane(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """--bifrost-url is the direct (HTTPS/prod) endpoint, bypassing the plane
        shortcut; no plane label."""
        monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
        args = _parse_args(
            ["--send", "hi", "--new", "--agent", "a", "--api-key", "k",
             "--bifrost-url", "https://prov.example:8391"]
        )
        assert args.bifrost == BifrostBinding(endpoint_url="https://prov.example:8391")
        assert args.bifrost_plane is None

    def test_no_bifrost_flags_leaves_binding_none(self) -> None:
        """regression: no bifrost flags → bifrost/consumer_key None (pre-#17 path)."""
        args = _parse_args(
            ["--send", "hi", "--new", "--agent", "mimir", "--api-key", "k"]
        )
        assert args.bifrost is None
        assert args.consumer_key is None

    def test_plane_and_url_mutually_exclusive(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
        with pytest.raises(UsageError):
            _parse_args(
                ["--send", "hi", "--new", "--agent", "a", "--api-key", "k",
                 "--bifrost-plane", "memory", "--bifrost-host", "h",
                 "--bifrost-url", "https://x:8391"]
            )

    def test_plane_without_host_is_usage_error(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
        with pytest.raises(UsageError):
            _parse_args(
                ["--send", "hi", "--new", "--agent", "a", "--api-key", "k",
                 "--bifrost-plane", "memory"]
            )

    def test_bind_with_existing_session_is_usage_error(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """A binding is a session-CREATE concern; --session (existing) + bind is
        a usage error."""
        monkeypatch.setenv("RATATOSKR_BIFROST_CONSUMER_KEY", "ck")
        with pytest.raises(UsageError):
            _parse_args(
                ["--send", "hi", "--session", "s-1", "--api-key", "k",
                 "--bifrost-plane", "memory", "--bifrost-host", "h"]
            )

    @respx.mock
    async def test_amain_bound_create_carries_binding_and_routes_502(
        self, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """_amain on a bound create sends the bifrost body + the consumer-key
        bearer; a 502 auth_rejected routes to BifrostHandshakeFailed with the
        consumer-key-mismatch hint (INV-001/002, 401-message scoping)."""
        route = respx.post("http://w/sessions").mock(
            return_value=httpx.Response(
                502,
                json={
                    "error_code": "bifrost_handshake_failed",
                    "detail": {"bifrost_error": "bifrost.auth_rejected"},
                },
            )
        )
        args = ParsedArgs(
            send_content="hi", session_id=None, new=True, agent_id="ratatoskr:sindra",
            api_key="canary", server_url="http://w", raw=False, end_user_id="smoke-user",
            bifrost=BifrostBinding(endpoint_url="http://10.100.10.50:8391"),
            bifrost_plane="memory", consumer_key="ck",
        )
        rc = await _amain(args)
        assert rc == 23
        body = json.loads(route.calls[0].request.content)
        assert body["bifrost"] == {
            "endpoint_url": "http://10.100.10.50:8391", "scope": None
        }
        assert route.calls[0].request.headers["Authorization"] == "Bearer ck"
        err = capsys.readouterr().err
        assert "bifrost.auth_rejected" in err
        assert "consumer key" in err  # the 401-scoping hint

    async def test_amain_bind_without_consumer_key_exits(self) -> None:
        """_amain on a bind with no consumer key raises BifrostConsumerKeyMissing
        (before HTTP) → a clean exit code, never a canary fallback."""
        args = ParsedArgs(
            send_content="hi", session_id=None, new=True, agent_id="a",
            api_key="canary", server_url="http://w", raw=False, end_user_id=None,
            bifrost=BifrostBinding(endpoint_url="http://x:8391"),
            bifrost_plane="memory", consumer_key=None,
        )
        rc = await _amain(args)
        assert rc == 22


class TestWhoami:
    """--whoami one-shot probe (slice: capabilities+me): GET /me + GET /capabilities."""

    def test_whoami_standalone_accepted(self) -> None:
        """whoami_standalone_accepted: --whoami alone → valid; whoami=True, no turn flags."""
        args = _parse_args(["--whoami", "--api-key", "k"])
        assert args.whoami is True
        assert args.send_content is None
        assert args.session_id is None
        assert args.new is False

    def test_whoami_with_send_rejected(self) -> None:
        """whoami_with_send_rejected [adversarial]: --whoami + --send → UsageError."""
        with pytest.raises(UsageError, match="standalone probe"):
            _parse_args(["--whoami", "--send", "hi", "--api-key", "k"])

    def test_whoami_with_new_rejected(self) -> None:
        """whoami_with_new_rejected [adversarial]: --whoami + --new → UsageError."""
        with pytest.raises(UsageError, match="standalone probe"):
            _parse_args(["--whoami", "--new", "--agent", "m", "--api-key", "k"])

    @respx.mock
    def test_whoami_mode_prints_report(self, capsys: pytest.CaptureFixture[str]) -> None:
        """whoami_mode_prints_report [happy,tracer]: /me + /capabilities → stdout report; exit 0."""
        respx.get("https://w.example/me").mock(
            return_value=httpx.Response(
                200,
                json={
                    "user_id": "alice",
                    "scopes": ["conversations.read", "conversations.write"],
                    "tier": "user",
                    "key_id": "a1b2c3d4",
                },
            )
        )
        respx.get("https://w.example/capabilities").mock(
            return_value=httpx.Response(
                200,
                json={
                    "ephemeral_templates": {
                        "echo": {
                            "allowed_models": ["glm5-turbo"],
                            "default_model": "glm5-turbo",
                            "system_prompt_max_bytes": 32768,
                        }
                    }
                },
            )
        )
        rc = main(["--whoami", "--api-key", "k", "--server", "https://w.example"])
        assert rc == 0
        out = capsys.readouterr().out
        assert "user_id: alice" in out
        assert "tier: user" in out
        assert "key_id: a1b2c3d4" in out
        assert "ephemeral_template echo" in out
        assert "glm5-turbo" in out

    @respx.mock
    def test_whoami_me_auth_failure_exits_20(self, capsys: pytest.CaptureFixture[str]) -> None:
        """whoami_me_auth_failure [error]: /me 401 → exit 20 [session_api_failed]."""
        respx.get("https://w.example/me").mock(
            return_value=httpx.Response(401, json={"detail": "auth_invalid"})
        )
        rc = main(["--whoami", "--api-key", "k", "--server", "https://w.example"])
        assert rc == 20
        assert "[session_api_failed]" in capsys.readouterr().err


class TestTier2Probes:
    """--characters + --set-persona-pad one-shot probes (Tier-2: #161 + persona_state-write)."""

    def test_characters_standalone_accepted(self) -> None:
        """characters_standalone: --characters alone → valid."""
        args = _parse_args(["--characters", "--api-key", "k"])
        assert args.characters is True
        assert args.session_id is None

    def test_set_persona_requires_session(self) -> None:
        """set_persona_requires_session [adversarial]: --set-persona-pad needs --session."""
        with pytest.raises(UsageError, match="requires --session"):
            _parse_args(["--set-persona-pad", "0.4,0.1,-0.2", "--api-key", "k"])

    def test_probes_mutually_exclusive(self) -> None:
        """probes_mutually_exclusive [adversarial]: --whoami + --characters → UsageError."""
        with pytest.raises(UsageError, match="mutually exclusive"):
            _parse_args(["--whoami", "--characters", "--api-key", "k"])

    @respx.mock
    def test_characters_probe_lifecycle(self, capsys: pytest.CaptureFixture[str]) -> None:
        """characters_probe [happy,tracer]: models → create → state → delete; report to stdout."""
        respx.get("https://w.example/models/available-for-characters").mock(
            return_value=httpx.Response(200, json={"items": [{"name": "fast"}]})
        )
        respx.post("https://w.example/characters").mock(
            return_value=httpx.Response(201, json={"character_id": "char_z", "ttl_expires_at": "t"})
        )
        respx.get("https://w.example/characters/char_z/state").mock(
            return_value=httpx.Response(200, json={"schema_version": "1", "pad": [0.1, 0.2, 0.3]})
        )
        del_route = respx.delete("https://w.example/characters/char_z").mock(
            return_value=httpx.Response(204)
        )
        rc = main(["--characters", "--api-key", "k", "--server", "https://w.example"])
        assert rc == 0
        out = capsys.readouterr().out
        assert "character models: fast" in out
        assert "created: char_z" in out
        assert "pad=[0.1, 0.2, 0.3]" in out
        assert "deleted: char_z" in out
        assert del_route.call_count == 1  # lifecycle cleaned up

    @respx.mock
    def test_set_persona_probe(self, capsys: pytest.CaptureFixture[str]) -> None:
        """set_persona_probe [happy,tracer]: POST pad to /sessions/{id}/persona_state; 204."""
        import json as _json

        route = respx.post("https://w.example/sessions/s1/persona_state").mock(
            return_value=httpx.Response(204)
        )
        rc = main(
            ["--set-persona-pad", "0.4,0.1,-0.2", "--session", "s1",
             "--api-key", "k", "--server", "https://w.example"]
        )
        assert rc == 0
        assert "persona_state set" in capsys.readouterr().out
        # canonical POST /sessions/{id}/persona_state body: named-key dict, NOT a list
        assert _json.loads(route.calls[0].request.content) == {
            "pad": {"pleasure": 0.4, "arousal": 0.1, "dominance": -0.2}
        }

    def test_set_persona_wrong_count(self) -> None:
        """set_persona_wrong_count [adversarial]: not exactly 3 floats → exit 10, no HTTP."""
        rc = main(
            ["--set-persona-pad", "0.4,0.1", "--session", "s1",
             "--api-key", "k", "--server", "https://w.example"]
        )
        assert rc == 10


class TestSeedFirstMessageProbe:
    """--seed-first-message one-shot (#347 authored-history-write reference-consumer probe)."""

    def test_seed_requires_agent(self) -> None:
        """seed_requires_agent [adversarial]: --seed-first-message needs --agent."""
        with pytest.raises(UsageError, match="requires --agent"):
            _parse_args(["--seed-first-message", "hello", "--api-key", "k"])

    def test_seed_forbids_session(self) -> None:
        """seed_forbids_session [adversarial]: manages its own session — no --session."""
        with pytest.raises(UsageError, match="manages its own session"):
            _parse_args(
                ["--seed-first-message", "hi", "--agent", "m", "--session", "s1", "--api-key", "k"]
            )

    def test_seed_mutually_exclusive(self) -> None:
        """seed_mutually_exclusive [adversarial]: --seed-first-message + --whoami → UsageError."""
        with pytest.raises(UsageError, match="mutually exclusive"):
            _parse_args(["--seed-first-message", "hi", "--whoami", "--api-key", "k"])

    def test_seed_empty_rejected(self) -> None:
        """seed_empty_rejected [adversarial]: empty content → UsageError."""
        with pytest.raises(UsageError, match="non-empty"):
            _parse_args(["--seed-first-message", "", "--agent", "m", "--api-key", "k"])

    def test_seed_accepted(self) -> None:
        """seed_accepted [happy]: --seed-first-message + --agent → parses."""
        args = _parse_args(["--seed-first-message", "hi", "--agent", "mimir", "--api-key", "k"])
        assert args.seed_first_message == "hi"
        assert args.agent_id == "mimir"
        assert args.session_id is None and args.new is False

    @respx.mock
    def test_seed_probe_happy(self, capsys: pytest.CaptureFixture[str]) -> None:
        """seed_probe [happy,tracer]: create session → seed → read-back; report to stdout."""
        respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(
                201,
                json={
                    "session_id": "s1",
                    "agent_id": "mimir",
                    "message_count": 0,
                    "created_at": "2026-07-06T12:00:00+00:00",
                    "last_active": "2026-07-06T12:00:00+00:00",
                    "metadata": {},
                },
            )
        )
        hist_route = respx.post("https://w.example/sessions/s1/history").mock(
            return_value=httpx.Response(
                201,
                json={
                    "author": "assistant",
                    "content_chars": 5,
                    "injected_at": "2026-07-06T12:00:01+00:00",
                    "phase": "seeded",
                    "seq": 0,
                    "session_id": "s1",
                    "turn_id": "t1",
                },
            )
        )
        respx.get("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200,
                json={
                    "session_id": "s1",
                    "items": [{"seq": 0, "role": "assistant", "content": "hello"}],
                    "next_cursor": None,
                },
            )
        )
        rc = main(
            ["--seed-first-message", "hello", "--agent", "mimir",
             "--api-key", "k", "--server", "https://w.example"]
        )
        assert rc == 0
        out = capsys.readouterr().out
        assert "session: s1" in out
        assert "seeded: seq=0 phase=seeded" in out
        assert "read-back: 1 message" in out
        assert "role=assistant" in out
        assert hist_route.call_count == 1

    @respx.mock
    def test_seed_probe_feature_absent(self, capsys: pytest.CaptureFixture[str]) -> None:
        """feature_absent [error-path]: 404 hide-existence → benign report, exit 0, no read-back."""
        respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(
                201,
                json={
                    "session_id": "s1",
                    "agent_id": "mimir",
                    "message_count": 0,
                    "created_at": "2026-07-06T12:00:00+00:00",
                    "last_active": "2026-07-06T12:00:00+00:00",
                    "metadata": {},
                },
            )
        )
        respx.post("https://w.example/sessions/s1/history").mock(
            return_value=httpx.Response(404, json={"error_code": "session_not_found"})
        )
        msgs_route = respx.get("https://w.example/sessions/s1/messages").mock(
            return_value=httpx.Response(
                200, json={"session_id": "s1", "items": [], "next_cursor": None}
            )
        )
        rc = main(
            ["--seed-first-message", "hello", "--agent", "mimir",
             "--api-key", "k", "--server", "https://w.example"]
        )
        assert rc == 0
        assert "feature-absent" in capsys.readouterr().out
        assert msgs_route.call_count == 0  # never capability-probes past the 404