ratatoskr/tests/test_tui.py

"""Tests for ratatoskr.tui per docs/contracts/issues/4.contract.md."""

from pathlib import Path
from unittest.mock import MagicMock

import httpx
import pytest
import respx
from textual.widgets import RichLog

from ratatoskr.cli import ParsedArgs
from ratatoskr.sessions import BifrostBinding
from ratatoskr.sse_client import (
    Cancelled,
    Done,
    SseId,
    Text,
    Thinking,
    ToolResult,
    ToolStart,
    WorkerPhase,
)
from ratatoskr.tui import RatatoskrApp, _cancel_via_sse

_CANCEL_OK_RESP = {"turn_id": 42, "cancelled": True, "reason": None, "partial_message_id": None}
_CREATE_OK_RESP = {
    "session_id": "s-new12345",
    "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": {},
}


def _args_new(**overrides) -> ParsedArgs:
    base = dict(
        send_content=None,
        session_id=None,
        new=True,
        agent_id="mimir",
        api_key="k",
        server_url="https://w.example",
        raw=False,
    )
    base.update(overrides)
    return ParsedArgs(**base)


def _args_existing(session_id: str = "s-1existing", **overrides) -> ParsedArgs:
    base = dict(
        send_content=None,
        session_id=session_id,
        new=False,
        agent_id=None,
        api_key="k",
        server_url="https://w.example",
        raw=False,
    )
    base.update(overrides)
    return ParsedArgs(**base)


def _spy_writes(monkeypatch) -> list:
    """Patch RichLog.write AND VerticalScroll.mount to record every renderable
    or mounted-widget content into a single list (returned).

    v0.9.0: transcript content is mounted into a VerticalScroll, not written
    to a RichLog. The spy captures both shapes — for each mounted Static, the
    Static's `renderable` (Markdown / RichText / str) lands in the list,
    indistinguishably from RichLog.write entries. Integration tests assert
    on substrings or types in `writes` so the merged shape is the right
    abstraction.

    Accepts *args/**kwargs so Textual's internal deferred-render paths still
    work after a write-during-mount + Resize sequence.
    """
    from textual.containers import VerticalScroll
    from textual.widgets import Static

    writes: list = []

    original_write = RichLog.write

    def spy_write(self, content, *args, **kw):
        writes.append(content)
        return original_write(self, content, *args, **kw)

    monkeypatch.setattr(RichLog, "write", spy_write)

    original_mount = VerticalScroll.mount

    def spy_mount(self, *children, **kw):
        for child in children:
            if isinstance(child, Static):
                writes.append(child.content)
            else:
                writes.append(child)
        return original_mount(self, *children, **kw)

    monkeypatch.setattr(VerticalScroll, "mount", spy_mount)
    return writes


def _resolved_app(
    args: ParsedArgs,
    *,
    session_id: str | None = None,
    agent_id: str | None = None,
    client: httpx.AsyncClient | None = None,
) -> RatatoskrApp:
    """Construct RatatoskrApp with pre-resolved state (issue #6 lifecycle).

    Production path: `run_tui` → `_resolve_then_run` opens AsyncClient, mints
    or attaches session, then constructs the App with the resolved tuple. This
    helper inlines that shape so tests bypass the pre-flight without
    re-implementing it. The client is opened here (and leaks at test teardown
    — acceptable; respx mocks all network calls and pytest exits cleanly).
    """
    sid = session_id if session_id is not None else (args.session_id or "s-default")
    aid = args.agent_id if agent_id is None else agent_id
    if client is None:
        client = httpx.AsyncClient(
            base_url=args.server_url,
            headers={"Authorization": f"Bearer {args.api_key}"},
            timeout=httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0),
        )
    return RatatoskrApp(args, session_id=sid, agent_id=aid, client=client)


SID = SseId(42, 5)


# Issue #12 — TuiPresenterState replaces _render_event_to_log with a stateful
# per-turn presenter. (Pre-amendment TestRenderEventToLog class and
# `_render_event_to_log` function have been removed under the project's
# no-backwards-compatibility rule.)


class TestTuiPresenterState:
    """Tests for the new TuiPresenterState — per issue #12 contract."""

    def test_thinking_coalesces_until_newline(self) -> None:
        """thinking_coalesces_until_newline [happy,tracer, v0.7.1]:
        Per-token deltas accumulate in the buffer; flush only on `\\n`.
        Three short token-shaped deltas without `\\n` → thinking_log gets
        ONLY Rule(start); content stays buffered.
        """
        from rich.rule import Rule

        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        thinking_log = MagicMock()
        state = TuiPresenterState()
        for chunk in ("Let", " me", " think"):
            state.render(
                Thinking(sse_id=SID, content=chunk),
                transcript=transcript,
                tools_log=MagicMock(),
                debug_log=MagicMock(),
                thinking_log=thinking_log,
                raw=False,
            )
        writes = [c[0][0] for c in thinking_log.write.call_args_list]
        # Only Rule(start) — content stays buffered (no `\n` seen).
        assert len(writes) == 1
        assert isinstance(writes[0], Rule)
        assert state.thinking_chunk_buffer == "Let me think"
        assert transcript.mount.call_count == 0

    def test_thinking_flushes_on_newline(self) -> None:
        """thinking_flushes_on_newline [happy, v0.7.1]:
        Delta carrying `\\n` flushes the accumulated buffer as ONE line.
        """
        from ratatoskr.tui import TuiPresenterState

        thinking_log = MagicMock()
        state = TuiPresenterState()
        for chunk in ("Hello", " world", "\n"):
            state.render(
                Thinking(sse_id=SID, content=chunk),
                transcript=MagicMock(),
                tools_log=MagicMock(),
                debug_log=MagicMock(),
                thinking_log=thinking_log,
                raw=False,
            )
        writes = [c[0][0] for c in thinking_log.write.call_args_list]
        # Rule(start) + "Hello world" (one coalesced line) = 2 writes
        assert len(writes) == 2
        assert writes[1] == "Hello world"
        assert state.thinking_chunk_buffer == ""

    def test_thinking_closes_to_thinking_log(self) -> None:
        """thinking_closes_to_thinking_log [happy, v0.7.1]: 2x Thinking + WorkerPhase →
        v0.7.1 coalesces "a"+"b" into one buffered string; the close flushes
        "ab" as a single line before Rule(end). Result: Rule(start) + "ab" +
        Rule(end) = 3 writes. debug_log gets worker_phase; transcript untouched.
        """
        from rich.rule import Rule

        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        debug_log = MagicMock()
        thinking_log = MagicMock()
        state = TuiPresenterState()
        for content in ("a", "b"):
            state.render(
                Thinking(sse_id=SID, content=content),
                transcript=transcript,
                tools_log=MagicMock(),
                debug_log=debug_log,
                thinking_log=thinking_log,
                raw=False,
            )
        state.render(
            WorkerPhase(sse_id=SID, phase="streaming", turn_id=42),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=debug_log,
            thinking_log=thinking_log,
            raw=False,
        )
        thinking_writes = [c[0][0] for c in thinking_log.write.call_args_list]
        # v0.7.1: 1 Rule(start) + 1 coalesced "ab" tail-flush + 1 Rule(end) = 3 writes
        assert len(thinking_writes) == 3
        assert isinstance(thinking_writes[0], Rule)
        assert thinking_writes[1] == "ab"
        assert isinstance(thinking_writes[2], Rule)
        # worker_phase still goes to debug_log; transcript untouched.
        assert "· worker_phase:" in _text_of(debug_log.write.call_args_list[-1][0][0])
        assert not transcript.mount.called

    # v0.6.5: thinking-current Static removed; test_thinking_widget_truncation
    # and test_thinking_widget_visibility_lifecycle deleted (no longer apply).

    def test_multiple_thinking_runs_each_get_thinking_log_section(self) -> None:
        """multiple_thinking_runs_each_get_section [scenario, v0.8.1]:
        Thinking → Text → Thinking → Done → TWO start/end Rule pairs in
        thinking_log (deltas coalesced into tail-flushes per run).
        Text deltas now stream into the transcript via coalesce-on-newline
        (no current-text Static); "hi" with no `\\n` stays buffered until
        Done's tail-flush.
        """
        from rich.rule import Rule

        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        thinking_log = MagicMock()
        state = TuiPresenterState()
        for evt in (
            Thinking(sse_id=SID, content="first"),
            Text(sse_id=SID, content="hi"),
            Thinking(sse_id=SID, content="second"),
        ):
            state.render(
                evt, transcript=transcript,
                tools_log=MagicMock(), debug_log=MagicMock(),
                thinking_log=thinking_log, raw=False,
            )
        state.render(
            _make_tui_done(),
            transcript=transcript,
            tools_log=MagicMock(), debug_log=MagicMock(),
            thinking_log=thinking_log, raw=False,
        )
        # thinking_log: 4 Rules (start+end per run) + 2 tail-flush strings.
        thinking_writes = [c[0][0] for c in thinking_log.write.call_args_list]
        rules = [w for w in thinking_writes if isinstance(w, Rule)]
        delta_strs = [w for w in thinking_writes if isinstance(w, str)]
        assert len(rules) == 4, f"expected 4 Rules (2 start + 2 end), got {len(rules)}"
        assert "first" in delta_strs
        assert "second" in delta_strs
        # v0.8.1: Text "hi" flushes as a line in transcript on Done.
        transcript_renderables = [_text_of(r) for r in _mounted_renderables(transcript)]
        assert "hi" in transcript_renderables
        assert any(w.startswith("[done]") for w in transcript_renderables if isinstance(w, str))

    def test_render_exception_fallback(self) -> None:
        """render_exception_fallback [adversarial, v0.6.5]:
        thinking_log.write raises → catch in presenter, write plain-label
        fallback + render_error line via INV-009 fallback path (routing
        preservation: thinking events still route to thinking_log).
        """
        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        thinking_log = MagicMock()
        # First call (Rule write) raises; subsequent calls succeed for fallback.
        thinking_log.write.side_effect = [
            AttributeError("rule write failed (msg should NOT leak)"),
            None,
            None,
        ]
        state = TuiPresenterState()
        state.render(
            Thinking(sse_id=SID, content="x"),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=MagicMock(),
            thinking_log=thinking_log,
            raw=False,
        )
        writes = [c[0][0] for c in thinking_log.write.call_args_list if isinstance(c[0][0], str)]
        assert any(w.startswith("[thinking]") for w in writes), writes
        assert any(w == "[render_error] AttributeError" for w in writes), writes
        assert not any("rule write failed" in w for w in writes), writes
        assert not transcript.mount.called

    def test_state_reset_per_worker(self) -> None:
        """state_reset_per_worker [trace]: fresh TuiPresenterState() starts no thinking open."""
        from ratatoskr.tui import TuiPresenterState

        s1 = TuiPresenterState()
        s1.render(
            Thinking(sse_id=SID, content="x"),
            transcript=MagicMock(),
            tools_log=MagicMock(),
            debug_log=MagicMock(),
            thinking_log=MagicMock(),
            raw=False,
        )
        s2 = TuiPresenterState()
        assert s1.thinking_open is True
        assert s2.thinking_open is False

    def test_cancelled_mid_thinking_closes(self) -> None:
        """cancelled_mid_thinking_closes [scenario, v0.5.0]:
        Thinking, Cancelled → ONE closed thinking entry in debug_log + a
        [cancelled] entry in transcript; widget hidden.
        """
        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        thinking_log = MagicMock()
        state = TuiPresenterState()
        state.render(
            Thinking(sse_id=SID, content="partial"),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=MagicMock(), thinking_log=thinking_log, raw=False,
        )
        state.render(
            Cancelled(
                sse_id=SID, phase="cancelled", turn_id=42, reason="user", partial_message_id=None
            ),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=MagicMock(), thinking_log=thinking_log, raw=False,
        )
        # v0.6.5: streamed thinking + Rule(end) in thinking_log; [cancelled] in transcript.
        transcript_renderables = [_text_of(r) for r in _mounted_renderables(transcript)]
        assert any(w.startswith("[cancelled]") for w in transcript_renderables)
        # thinking_log got at least Rule(start) + "partial" delta + Rule(end)
        assert thinking_log.write.call_count >= 3

    def test_text_then_done_mounts_widget_and_finalizes(self) -> None:
        """text_then_done_mounts_widget_and_finalizes [happy, v0.9.0]:
        First Text delta mounts a Static(Markdown(buffer)) into the transcript;
        Done finalizes the widget reference and mounts a styled [done] label.
        No duplicate content (v0.9.0 replaces v0.8.x's flush-on-Done with
        live in-place Markdown updates).
        """
        from rich.markdown import Markdown

        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        state = TuiPresenterState()
        state.render(
            Text(sse_id=SID, content="hi"),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=MagicMock(),
            thinking_log=MagicMock(),
            raw=False,
        )
        # v0.9.0: response widget mounted on first Text delta with Markdown wrapper.
        assert transcript.mount.called
        first_widget = transcript.mount.call_args_list[0][0][0]
        assert isinstance(first_widget.content, Markdown)
        assert first_widget.content.markup == "hi"
        assert state.text_chunk_buffer == "hi"
        # Done finalizes: text_chunk_buffer cleared, widget ref released, label mounted.
        state.render(
            _make_tui_done(),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=MagicMock(),
            thinking_log=MagicMock(),
            raw=False,
        )
        writes = _mounted_renderables(transcript)
        assert any(_text_of(w).startswith("[done]") for w in writes)
        # v0.9.0: response Markdown rendered live during stream — only ONE
        # Markdown renderable lands in the transcript (no post-Done re-render).
        markdowns = [w for w in writes if isinstance(w, Markdown)]
        assert len(markdowns) == 1
        assert state.text_chunk_buffer == ""
        assert state.current_response_widget is None

    def test_raw_flag_skips_markdown(self) -> None:
        """raw_flag_skips_markdown [v0.9.0]: raw=True → response widget holds
        plain str instead of Markdown. Live in-place update still happens;
        only the wrapper differs.
        """
        from rich.markdown import Markdown

        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        state = TuiPresenterState()
        state.render(
            Text(sse_id=SID, content="hi"),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=MagicMock(), thinking_log=MagicMock(), raw=True,
        )
        state.render(
            _make_tui_done(),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=MagicMock(), thinking_log=MagicMock(), raw=True,
        )
        writes = _mounted_renderables(transcript)
        # Raw mode bypasses Markdown entirely — content lives as plain str.
        assert not any(isinstance(w, Markdown) for w in writes)
        assert "hi" in writes

    def test_worker_phase_demoted_to_debug_log(self) -> None:
        """worker_phase_demoted_to_debug_log [trace, v0.5.0]: WorkerPhase → debug_log
        "· worker_phase:" prefix rendered with Australis dark-60 Rich style.
        Transcript receives nothing.
        """
        from rich.text import Text as RichText

        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        debug_log = MagicMock()
        state = TuiPresenterState()
        state.render(
            WorkerPhase(sse_id=SID, phase="streaming", turn_id=42),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=debug_log, thinking_log=MagicMock(), raw=False,
        )
        # v0.5.0: WorkerPhase routes to debug_log, NOT transcript.
        assert not transcript.mount.called
        renderable = debug_log.write.call_args[0][0]
        # INV-003: must be a styled Rich Text renderable, not a plain str.
        # v0.4.1 retheme: style is now Australis Sea dark-60 ("#86929d") instead
        # of the terminal-dim filter "dim". Assert non-empty styling either way.
        assert isinstance(renderable, RichText), type(renderable)
        assert renderable.style, "demoted telemetry must carry SOME style"
        text = renderable.plain
        assert text.startswith("· worker_phase:")
        assert "[worker_phase]" not in text

    # v0.6.5: test_terminal_events_belt_and_braces_widget_cleanup deleted.
    # The thinking-current Static is gone, so there's no widget to clean up
    # on terminal events. The corresponding Volva F3 invariant is obsoleted
    # by the streaming-into-thinking_log architecture.

    def test_tool_start_routes_to_tools_log(self) -> None:
        """tool_start_routes_to_tools_log [INV-014]: ToolStart writes to tools_log, NOT transcript.

        Issue #13: tool events route to the dedicated Tools pane (right column).
        Pre-#13 wrote them to the main transcript with `· tool_start:` prefix.
        Post-#13 the prefix is preserved but the destination shifts.
        """
        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        tools_log = MagicMock()
        state = TuiPresenterState()
        state.render(
            ToolStart(sse_id=SID, name="read_file", arguments={"path": "/x"}),
            transcript=transcript,
            tools_log=tools_log,
            debug_log=MagicMock(), thinking_log=MagicMock(), raw=False,
        )
        # INV-014: write went to tools_log
        assert tools_log.write.called
        assert _text_of(tools_log.write.call_args[0][0]).startswith("· tool_start:")
        # INV-014: transcript was NOT written to
        assert not transcript.mount.called

    def test_tool_result_routes_to_tools_log(self) -> None:
        """tool_result_routes_to_tools_log [INV-014]: ToolResult → tools_log, NOT transcript."""
        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        tools_log = MagicMock()
        state = TuiPresenterState()
        state.render(
            ToolResult(sse_id=SID, name="read_file", result="ok", duration_ms=12),
            transcript=transcript,
            tools_log=tools_log,
            debug_log=MagicMock(), thinking_log=MagicMock(), raw=False,
        )
        assert tools_log.write.called
        assert _text_of(tools_log.write.call_args[0][0]).startswith("· tool_result:")
        assert not transcript.mount.called

    def test_text_first_delta_mounts_response_widget(self) -> None:
        """text_first_delta_mounts_response_widget [v0.9.0]: first Text delta
        mounts a Static carrying Markdown(buffer) into the transcript. The
        text_chunk_buffer holds the accumulated content for the next delta's
        in-place update.
        """
        from rich.markdown import Markdown

        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        tools_log = MagicMock()
        state = TuiPresenterState()
        state.render(
            Text(sse_id=SID, content="hello"),
            transcript=transcript,
            tools_log=tools_log,
            debug_log=MagicMock(),
            thinking_log=MagicMock(),
            raw=False,
        )
        assert state.text_chunk_buffer == "hello"
        assert transcript.mount.call_count == 1
        widget = transcript.mount.call_args[0][0]
        assert isinstance(widget.content, Markdown)
        assert widget.content.markup == "hello"
        assert state.current_response_widget is widget
        assert not tools_log.write.called

    def test_text_subsequent_deltas_update_in_place(self) -> None:
        """text_subsequent_deltas_update_in_place [v0.9.0]: deltas after the
        first do NOT mount a new widget — they update the existing widget's
        Markdown content in place. The text_chunk_buffer accumulates.
        """
        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        state = TuiPresenterState()
        for tok in ("Hel", "lo", " ", "world"):
            state.render(
                Text(sse_id=SID, content=tok),
                transcript=transcript,
                tools_log=MagicMock(),
                debug_log=MagicMock(),
                thinking_log=MagicMock(),
                raw=False,
            )
        # Exactly ONE mount (the first delta); subsequent deltas update.
        assert transcript.mount.call_count == 1
        assert state.text_chunk_buffer == "Hello world"
        # Widget reference held; buffer is the source of truth re-rendered
        # into Markdown(...) for each Static.update call.
        assert state.current_response_widget is not None

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

        transcript = MagicMock()
        state = TuiPresenterState()
        state.render(
            _make_tui_done(duration_ms=5467),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=MagicMock(), thinking_log=MagicMock(), raw=True,
        )
        done_line = next(
            _text_of(r)
            for r in _mounted_renderables(transcript)
            if _text_of(r).startswith("[done]")
        )
        assert "duration=5.5s" in done_line
        assert "duration_ms=5467" not in done_line

    def test_usage_format_unicode_arrow(self) -> None:
        """usage_format_unicode_arrow [trace]: TUI Done label uses → (Unicode), not -> (ASCII)."""
        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        state = TuiPresenterState()
        usage = {
            "prompt_tokens": 6756,
            "completion_tokens": 126,
            "total_tokens": 6882,
            "cached_input_tokens": 0,
        }
        state.render(
            _make_tui_done(usage=usage),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=MagicMock(), thinking_log=MagicMock(), raw=True,
        )
        done_line = next(
            _text_of(r)
            for r in _mounted_renderables(transcript)
            if _text_of(r).startswith("[done]")
        )
        assert "usage 6756 in → 126 out (6882 total, 0 cached)" in done_line


class TestPresenterAuditLogging:
    """v0.10.0 — per-event audit lines + turn-summary in the debug pane.

    The presenter emits one debug-pane line per arriving event (Text and
    Thinking are aggregated into the turn-summary instead of per-delta to
    avoid drowning the pane at token rate).
    """

    def test_worker_phase_emits_audit_line(self) -> None:
        """worker_phase_emits_audit_line: WorkerPhase arrival adds an audit
        line to debug_log alongside the existing `· worker_phase:` entry.
        Audit line shape: `[HH:MM:SS.fff] workerphase sse_id=N:M …`.
        """
        from ratatoskr.tui import TuiPresenterState

        debug_log = MagicMock()
        state = TuiPresenterState()
        state.render(
            WorkerPhase(sse_id=SID, phase="streaming", turn_id=42),
            transcript=MagicMock(),
            tools_log=MagicMock(),
            debug_log=debug_log,
            thinking_log=MagicMock(),
            raw=False,
        )
        # Two writes: audit line + worker_phase telemetry.
        assert debug_log.write.call_count == 2
        audit_line = _text_of(debug_log.write.call_args_list[0][0][0])
        assert "workerphase" in audit_line
        assert "sse_id=42:5" in audit_line
        assert "phase=streaming" in audit_line

    def test_tool_start_emits_audit_line(self) -> None:
        """tool_start_emits_audit_line: ToolStart adds one audit line to
        debug_log even though the tool event itself routes to tools_log.
        """
        from ratatoskr.tui import TuiPresenterState

        debug_log = MagicMock()
        state = TuiPresenterState()
        state.render(
            ToolStart(sse_id=SID, name="read_file", arguments={"path": "/x"}),
            transcript=MagicMock(),
            tools_log=MagicMock(),
            debug_log=debug_log,
            thinking_log=MagicMock(),
            raw=False,
        )
        assert debug_log.write.call_count == 1
        audit_line = _text_of(debug_log.write.call_args[0][0])
        assert "toolstart" in audit_line
        assert "sse_id=42:5" in audit_line
        assert "name=read_file" in audit_line

    def test_text_delta_counted_not_per_event_audit_line(self) -> None:
        """text_delta_counted_not_per_event_audit_line: a Text delta does
        NOT emit a per-event audit line (token-rate would drown the pane);
        instead it bumps text_delta_count / text_byte_count for the turn-
        summary at Done.
        """
        from ratatoskr.tui import TuiPresenterState

        debug_log = MagicMock()
        state = TuiPresenterState()
        state.render(
            Text(sse_id=SID, content="hello world"),
            transcript=MagicMock(),
            tools_log=MagicMock(),
            debug_log=debug_log,
            thinking_log=MagicMock(),
            raw=False,
        )
        # No debug-pane writes — text deltas are silent at token rate.
        assert not debug_log.write.called
        assert state.text_delta_count == 1
        assert state.text_byte_count == len("hello world")

    def test_affect_update_routes_to_audit_only(self) -> None:
        """affect_update_routes_to_audit_only [v0.11.0]: AffectUpdate emits ONE
        debug-pane audit line (with dominant_emotion + PAD for status=current)
        and touches NO other pane (no transcript mount, no tools_log, no
        thinking_log). UX shape (persona pane / sticky header) is deferred.
        """
        from ratatoskr.sse_client import AffectUpdate
        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        tools_log = MagicMock()
        thinking_log = MagicMock()
        debug_log = MagicMock()
        state = TuiPresenterState()
        snapshot = {
            "agent_id": "mimir",
            "pad": {"pleasure": 0.5, "arousal": 0.4, "dominance": 0.5},
            "dominant_emotion": "curiosity",
        }
        state.render(
            AffectUpdate(sse_id=SID, status="current", turn_id=42, snapshot=snapshot),
            transcript=transcript,
            tools_log=tools_log,
            debug_log=debug_log,
            thinking_log=thinking_log,
            raw=False,
        )
        assert debug_log.write.call_count == 1
        audit = _text_of(debug_log.write.call_args[0][0])
        assert "affectupdate" in audit
        assert "status=current" in audit
        assert "dominant_emotion='curiosity'" in audit
        assert "pad=(0.5,0.4,0.5)" in audit
        assert not transcript.mount.called
        assert not tools_log.write.called
        assert not thinking_log.write.called

    def test_awaiting_llm_first_token_mounts_indicator(self) -> None:
        """awaiting_llm_first_token_mounts_indicator [v0.14.0]: first heartbeat
        mounts a Static into the transcript and bumps heartbeat_count;
        debug-pane audit line carries turn_id + elapsed in seconds.
        """
        from ratatoskr.sse_client import AwaitingLlmFirstToken
        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        debug_log = MagicMock()
        state = TuiPresenterState()
        state.render(
            AwaitingLlmFirstToken(
                sse_id=SID, turn_id=42, elapsed_ms_since_building_prompt=5012.3
            ),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=debug_log,
            thinking_log=MagicMock(),
            raw=False,
        )
        assert state.heartbeat_count == 1
        assert state.awaiting_widget is not None
        assert transcript.mount.call_count == 1
        audit = _text_of(debug_log.write.call_args[0][0])
        assert "awaitingllmfirsttoken" in audit
        assert "turn_id=42" in audit
        assert "elapsed=5.0s" in audit

    def test_awaiting_subsequent_heartbeats_update_in_place(self) -> None:
        """awaiting_subsequent_heartbeats_update_in_place [v0.14.0]: second+
        heartbeats reuse the existing Static (no new mount); heartbeat_count
        tracks the total.
        """
        from ratatoskr.sse_client import AwaitingLlmFirstToken
        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        state = TuiPresenterState()
        for elapsed in (5000.0, 10005.4, 15011.8):
            state.render(
                AwaitingLlmFirstToken(
                    sse_id=SID, turn_id=42, elapsed_ms_since_building_prompt=elapsed
                ),
                transcript=transcript,
                tools_log=MagicMock(),
                debug_log=MagicMock(),
                thinking_log=MagicMock(),
                raw=False,
            )
        assert state.heartbeat_count == 3
        assert transcript.mount.call_count == 1  # mounted once on first

    def test_awaiting_indicator_removed_when_gap_closes(self) -> None:
        """awaiting_indicator_removed_when_gap_closes [v0.14.0]: any non-
        heartbeat event after one or more heartbeats removes the indicator
        and clears the awaiting_widget reference. Text event simulates the
        gap closing (CallingLLM fires, text begins).
        """
        from ratatoskr.sse_client import AwaitingLlmFirstToken
        from ratatoskr.tui import TuiPresenterState

        transcript = MagicMock()
        state = TuiPresenterState()
        state.render(
            AwaitingLlmFirstToken(
                sse_id=SID, turn_id=42, elapsed_ms_since_building_prompt=5000.0
            ),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=MagicMock(),
            thinking_log=MagicMock(),
            raw=False,
        )
        widget = state.awaiting_widget
        assert widget is not None
        state.render(
            Text(sse_id=SID, content="hello"),
            transcript=transcript,
            tools_log=MagicMock(),
            debug_log=MagicMock(),
            thinking_log=MagicMock(),
            raw=False,
        )
        # State reference cleared (the widget itself is a real Static whose
        # .remove() schedules removal — we verify the cleanup intent via
        # the state field, which is the contract callers actually observe).
        assert state.awaiting_widget is None

    def test_affect_update_scheduled_has_no_pad_detail(self) -> None:
        """affect_update_scheduled_has_no_pad_detail [v0.11.0]: status=scheduled
        carries no snapshot — the audit line omits dominant_emotion / pad and
        contains only status + turn_id.
        """
        from ratatoskr.sse_client import AffectUpdate
        from ratatoskr.tui import TuiPresenterState

        debug_log = MagicMock()
        state = TuiPresenterState()
        state.render(
            AffectUpdate(sse_id=SID, status="scheduled", turn_id=42, snapshot=None),
            transcript=MagicMock(),
            tools_log=MagicMock(),
            debug_log=debug_log,
            thinking_log=MagicMock(),
            raw=False,
        )
        audit = _text_of(debug_log.write.call_args[0][0])
        assert "status=scheduled" in audit
        assert "turn_id=42" in audit
        assert "dominant_emotion" not in audit
        assert "pad=" not in audit

    def test_done_emits_turn_summary_line(self) -> None:
        """done_emits_turn_summary_line: when Done arrives the presenter
        emits a `turn_summary` line aggregating per-delta Text + Thinking
        counters. The shape exposes the totals that per-event audit lines
        elided.
        """
        from ratatoskr.tui import TuiPresenterState

        debug_log = MagicMock()
        state = TuiPresenterState()
        # 3 Text deltas + 2 Thinking deltas, then Done.
        state.render(
            Text(sse_id=SID, content="a"),
            transcript=MagicMock(), tools_log=MagicMock(),
            debug_log=debug_log, thinking_log=MagicMock(), raw=False,
        )
        state.render(
            Text(sse_id=SID, content="bc"),
            transcript=MagicMock(), tools_log=MagicMock(),
            debug_log=debug_log, thinking_log=MagicMock(), raw=False,
        )
        state.render(
            Thinking(sse_id=SID, content="thought\n"),
            transcript=MagicMock(), tools_log=MagicMock(),
            debug_log=debug_log, thinking_log=MagicMock(), raw=False,
        )
        state.render(
            _make_tui_done(),
            transcript=MagicMock(), tools_log=MagicMock(),
            debug_log=debug_log, thinking_log=MagicMock(), raw=False,
        )
        writes = [_text_of(c[0][0]) for c in debug_log.write.call_args_list]
        summary = next(w for w in writes if "turn_summary" in w)
        assert "text_deltas=2" in summary
        assert "text_bytes=3" in summary  # "a" + "bc"
        assert "thinking_deltas=1" in summary
        assert "elapsed_ms=" in summary


def _text_of(write_arg: object) -> str:
    """Extract plain text from a RichLog.write() arg (str or rich.text.Text).

    Issue #12 wraps demoted-telemetry entries in `rich.text.Text(..., style="dim")`
    so the RichLog can apply dim styling; non-demoted writes stay as plain str.
    Tests that want to assert against content need both shapes flattened.

    v0.9.0: also extracts plain text from Markdown wrappers (the streaming-text
    response path uses Markdown(buffer) now; tests assert against the source
    markup, which lives in `Markdown.markup`).
    """
    from rich.markdown import Markdown
    from rich.text import Text as RichText

    if isinstance(write_arg, RichText):
        return write_arg.plain
    if isinstance(write_arg, Markdown):
        return write_arg.markup
    if isinstance(write_arg, str):
        return write_arg
    return ""  # Rule / etc. — not text content


def _mounted_renderables(transcript_mock: MagicMock) -> list:
    """v0.9.0: TuiPresenterState now mounts Static widgets into the transcript
    VerticalScroll instead of writing renderables to a RichLog. Tests using a
    MagicMock transcript inspect `transcript.mount.call_args_list`; each call's
    first positional arg is the Static child whose `.content` carries the
    Markdown / RichText / str that pre-v0.9.0 would have been the write arg.
    Returns those renderables in mount-call order so tests can assert on them
    with the same shape they used for `log.write.call_args_list` previously.
    """
    out: list = []
    for call in transcript_mock.mount.call_args_list:
        for child in call.args:
            renderable = getattr(child, "content", child)
            out.append(renderable)
    return out


def _make_tui_done(*, duration_ms: int = 1, usage: dict[str, int] | None = None) -> Done:
    return Done(
        sse_id=SID,
        phase="succeeded",
        response="r",
        model="m",
        duration_ms=duration_ms,
        usage=usage
        if usage is not None
        else {
            "prompt_tokens": 0,
            "completion_tokens": 0,
            "total_tokens": 0,
            "cached_input_tokens": 0,
        },
    )


class TestCancelViaSse:
    @respx.mock
    async def test_happy_cancel(self) -> None:
        """happy_cancel [happy,tracer]: 200 OK → returns None; transcript has no [cancel_failed]."""
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            return_value=httpx.Response(200, json=_CANCEL_OK_RESP)
        )
        transcript = MagicMock()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            result = await _cancel_via_sse(client, "s-1", 42, transcript=transcript)
        assert result is None
        transcript.mount.assert_not_called()

    @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")
        )
        transcript = MagicMock()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            await _cancel_via_sse(client, "s-1", 42, transcript=transcript)
        line = transcript.mount.call_args[0][0].content
        assert "[cancel_failed]" in line
        assert "CancelFailed" in line

    @respx.mock
    async def test_cancel_already_completed(self) -> None:
        """cancel_already_completed [scenario]: 409 → '[cancel_failed] CancelAlreadyCompleted:'."""
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            return_value=httpx.Response(409)
        )
        transcript = MagicMock()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            await _cancel_via_sse(client, "s-1", 42, transcript=transcript)
        line = transcript.mount.call_args[0][0].content
        assert "[cancel_failed]" in line
        assert "CancelAlreadyCompleted" in line

    @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")
        )
        transcript = MagicMock()
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            await _cancel_via_sse(client, "s-1", 42, transcript=transcript)
        line = transcript.mount.call_args[0][0].content
        assert "[cancel_failed]" in line
        assert "ConnectError" in line

    @respx.mock
    async def test_audit_callback_records_lifecycle(self) -> None:
        """audit_callback_records_lifecycle [v0.10.0]: when the caller passes
        an `audit` callback, _cancel_via_sse emits two lines on the happy
        path (`cancel_post issued …` + `cancel_post ok …`) and two lines on
        the failure path (`issued` + `failed …`). Gives the debug pane a
        complete cancel-POST timeline.
        """
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            return_value=httpx.Response(200, json=_CANCEL_OK_RESP)
        )
        transcript = MagicMock()
        audit_lines: list = []
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            await _cancel_via_sse(
                client, "s-1", 42, transcript=transcript, audit=audit_lines.append
            )
        assert len(audit_lines) == 2
        assert audit_lines[0].startswith("cancel_post issued ")
        assert "session_id=s-1" in audit_lines[0]
        assert "turn_id=42" in audit_lines[0]
        assert audit_lines[1] == "cancel_post ok turn_id=42"

    @respx.mock
    async def test_audit_callback_records_failure(self) -> None:
        """audit_callback_records_failure [v0.10.0]: failure path emits
        `cancel_post issued` then `cancel_post failed …` with exception type.
        """
        respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
            return_value=httpx.Response(500, content=b"boom")
        )
        transcript = MagicMock()
        audit_lines: list = []
        async with httpx.AsyncClient(base_url="https://w.example") as client:
            await _cancel_via_sse(
                client, "s-1", 42, transcript=transcript, audit=audit_lines.append
            )
        assert len(audit_lines) == 2
        assert audit_lines[0].startswith("cancel_post issued ")
        assert audit_lines[1].startswith("cancel_post failed turn_id=42 CancelFailed")


class TestPersonaFormatters:
    """v0.13.0 — _format_persona_header / _format_persona_detail rendering."""

    def test_header_compact_summary(self) -> None:
        """header_compact_summary: agent_id · dominant_emotion · pad(P,A,D) · N emotions."""
        from ratatoskr.tui import _format_persona_header

        snapshot = {
            "agent_id": "mimir",
            "dominant_emotion": "curiosity",
            "pad": {"pleasure": 0.52, "arousal": 0.47, "dominance": 0.50},
            "emotions_active": [
                {"type": "curiosity"}, {"type": "joy"},
            ],
        }
        line = _format_persona_header(snapshot)
        assert "mimir" in line
        assert "curiosity" in line
        assert "pad(0.52, 0.47, 0.5)" in line
        assert "2 emotions active" in line

    def test_header_singular_emotion(self) -> None:
        """header_singular_emotion: single emotion → '1 emotion active' (no 's')."""
        from ratatoskr.tui import _format_persona_header

        line = _format_persona_header(
            {
                "agent_id": "mimir",
                "dominant_emotion": "calm",
                "pad": {"pleasure": 0.5, "arousal": 0.4, "dominance": 0.5},
                "emotions_active": [{"type": "calm"}],
            }
        )
        assert "1 emotion active" in line
        assert "1 emotions" not in line

    def test_header_no_emotions_drops_count(self) -> None:
        """header_no_emotions_drops_count: empty emotions list → no count suffix."""
        from ratatoskr.tui import _format_persona_header

        line = _format_persona_header(
            {
                "agent_id": "mimir",
                "dominant_emotion": "?",
                "pad": {"pleasure": 0.5, "arousal": 0.4, "dominance": 0.5},
                "emotions_active": [],
            }
        )
        assert "emotion" not in line  # neither "1 emotion" nor "N emotions"

    def test_detail_renders_full_snapshot(self) -> None:
        """detail_renders_full_snapshot: PAD axes + baseline + delta + drift +
        emotions + last_updated_at all surface in the multi-line render.
        """
        from ratatoskr.tui import _format_persona_detail

        snapshot = {
            "agent_id": "mimir",
            "pad": {"pleasure": 0.52, "arousal": 0.47, "dominance": 0.50},
            "dominant_emotion": "curiosity",
            "emotions_active": [
                {"type": "curiosity", "intensity": 0.6, "decay_remaining_s": 202.7}
            ],
            "baseline_pad": {"pleasure": 0.50, "arousal": 0.40, "dominance": 0.50},
            "mood_drift": {"valence_delta": 0.02, "arousal_delta": 0.07},
            "last_updated_at": "2026-05-25T22:30:18+00:00",
        }
        detail = _format_persona_detail(snapshot)
        assert "Persona snapshot · mimir" in detail
        assert "Dominant emotion: curiosity" in detail
        assert "pleasure" in detail
        assert "baseline 0.5" in detail
        assert "+0.02" in detail or "0.02" in detail
        assert "valence_delta" in detail
        assert "curiosity" in detail
        assert "intensity 0.6" in detail
        assert "decay 3.4m" in detail
        assert "2026-05-25T22:30:18+00:00" in detail


class TestPresenterPersonaCallback:
    """v0.13.0 — presenter wires AffectUpdate snapshots into a callback."""

    def test_current_invokes_callback_with_snapshot(self) -> None:
        """current_invokes_callback_with_snapshot: AffectUpdate(current, snapshot)
        calls on_persona_snapshot(snapshot)."""
        from ratatoskr.sse_client import AffectUpdate
        from ratatoskr.tui import TuiPresenterState

        captured: list = []
        state = TuiPresenterState()
        snapshot = {"agent_id": "mimir", "pad": {"pleasure": 0.5}}
        state.render(
            AffectUpdate(sse_id=SID, status="current", turn_id=42, snapshot=snapshot),
            transcript=MagicMock(),
            tools_log=MagicMock(),
            debug_log=MagicMock(),
            thinking_log=MagicMock(),
            raw=False,
            on_persona_snapshot=captured.append,
        )
        assert captured == [snapshot]

    def test_scheduled_does_not_invoke_callback(self) -> None:
        """scheduled_does_not_invoke_callback: status=scheduled has no snapshot,
        so the callback is skipped (would be called with None otherwise)."""
        from ratatoskr.sse_client import AffectUpdate
        from ratatoskr.tui import TuiPresenterState

        captured: list = []
        state = TuiPresenterState()
        state.render(
            AffectUpdate(sse_id=SID, status="scheduled", turn_id=42, snapshot=None),
            transcript=MagicMock(),
            tools_log=MagicMock(),
            debug_log=MagicMock(),
            thinking_log=MagicMock(),
            raw=False,
            on_persona_snapshot=captured.append,
        )
        assert captured == []

    def test_callback_exception_swallowed(self) -> None:
        """callback_exception_swallowed: a raising callback does NOT crash the
        presenter — the audit line still landed (it precedes the callback).
        """
        from ratatoskr.sse_client import AffectUpdate
        from ratatoskr.tui import TuiPresenterState

        def boom(_snap: dict) -> None:
            raise RuntimeError("widget tearing down")

        debug_log = MagicMock()
        state = TuiPresenterState()
        # Should NOT raise
        state.render(
            AffectUpdate(
                sse_id=SID, status="current", turn_id=42, snapshot={"agent_id": "x"}
            ),
            transcript=MagicMock(),
            tools_log=MagicMock(),
            debug_log=debug_log,
            thinking_log=MagicMock(),
            raw=False,
            on_persona_snapshot=boom,
        )
        # Audit line still emitted (it runs before the callback)
        assert debug_log.write.called


class TestAppMount:
    """on_mount narrows per issue #6: only identity-widget population.

    Session resolution + AsyncClient open + error-on-resolve are exercised at
    the `_resolve_then_run` layer (see TestResolveThenRun); only happy mount
    paths remain here, exercised with pre-resolved state via _resolved_app.
    """

    async def test_happy_new_session_mount(self) -> None:
        """happy_new_session_mount [happy,tracer]: identity populated from pre-resolved state."""
        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            assert app.session_id == "s-new12345"
            assert app.agent_id == "mimir"
            assert app.state == "idle"
            # INV-002: session identity visible — agent_id + last 8 of session_id
            assert "mimir" in (app.sub_title or "")
            assert app.session_id[-8:] in (app.sub_title or "")

    async def test_happy_existing_session_mount(self) -> None:
        """happy_existing_session_mount: identity shows <unknown> when agent_id is None."""
        app = _resolved_app(_args_existing(session_id="s-existing-tail8x"))
        async with app.run_test() as pilot:
            await pilot.pause()
            assert app.session_id == "s-existing-tail8x"
            assert app.state == "idle"
            # INV-002 carve-out: agent unknown → <unknown> · …<tail>
            assert "<unknown>" in (app.sub_title or "")
            assert app.session_id[-8:] in (app.sub_title or "")

    async def test_footer_identity_visible_first_frame(self) -> None:
        """footer_identity_visible_first_frame [trace]: identity widget rendered first frame."""
        from textual.widgets import Static

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            identity_widget = app.query_one("#identity", Static)
            rendered = str(identity_widget.render())
            assert "mimir" in rendered
            assert "·" in rendered
            assert app.session_id[-8:] in rendered


# Issue #13 — TUI layout reshape + Tools pane (§5 v1 entry point)


class TestLayoutShape:
    """INV-013 + INV-014 + INV-017: Horizontal two-column layout with Tools tab."""

    async def test_main_row_is_horizontal(self) -> None:
        """main_row_is_horizontal [tracer]: compose() yields Horizontal#main-row."""
        from textual.containers import Horizontal

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            row = app.query_one("#main-row", Horizontal)
            assert row is not None

    async def test_left_column_content_only(self) -> None:
        """left_column_content_only [v0.9.0]: left column = transcript-scroll
        VerticalScroll + prompt Input. thinking-current Static removed in
        v0.6.5; transcript RichLog replaced by VerticalScroll in v0.9.0.
        """
        from textual.containers import Vertical, VerticalScroll
        from textual.widgets import Input

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            left = app.query_one("#left-column", Vertical)
            transcript = app.query_one("#transcript-scroll", VerticalScroll)
            prompt = app.query_one("#prompt", Input)
            assert transcript in left.walk_children()
            assert prompt in left.walk_children()
            # v0.6.5: thinking-current Static removed; no longer in DOM at all.
            from textual.css.query import NoMatches
            try:
                app.query_one("#thinking-current")
                raise AssertionError("thinking-current should not exist in v0.6.5")
            except NoMatches:
                pass  # expected

    async def test_right_column_has_tabbed_content_with_tools_tab(self) -> None:
        """right_column_has_tabbed_content_with_tools_tab: #side-panes + TabPane#tools-tab."""
        from textual.widgets import TabbedContent, TabPane

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            tabbed = app.query_one("#side-panes", TabbedContent)
            assert tabbed is not None
            tools_tab = app.query_one("#tools-tab", TabPane)
            assert tools_tab is not None

    async def test_tools_log_inside_tools_tab(self) -> None:
        """tools_log_inside_tools_tab: tools-transcript RichLog is a descendant of tools-tab TabPane."""
        from textual.widgets import RichLog, TabPane

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            tools_tab = app.query_one("#tools-tab", TabPane)
            tools_log = app.query_one("#tools-log", RichLog)
            assert tools_log in tools_tab.walk_children()

    async def test_pane_name_widget_renders_tools(self) -> None:
        """pane_name_widget_renders_tools [INV-pane-name]: #pane-name == 'Tools' on first frame."""
        from textual.widgets import Static

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            pane_name = app.query_one("#pane-name", Static)
            rendered = str(pane_name.render())
            assert rendered == "Tools"

    async def test_ctrl_1_activates_tools_tab(self) -> None:
        """ctrl_1_activates_tools_tab [tracer]: Ctrl+1 → TabbedContent.active == 'tools-tab'."""
        from textual.widgets import TabbedContent

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            await pilot.press("ctrl+1")
            await pilot.pause()
            tabbed = app.query_one("#side-panes", TabbedContent)
            assert tabbed.active == "tools-tab"

    async def test_ctrl_1_preserves_input_focus(self) -> None:
        """ctrl_1_preserves_input_focus [INV-016]: Ctrl+1 does NOT steal focus from Input."""
        from textual.widgets import Input

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            prompt = app.query_one("#prompt", Input)
            prompt.focus()
            await pilot.pause()
            assert app.focused is prompt
            await pilot.press("ctrl+1")
            await pilot.pause()
            assert app.focused is prompt, (
                f"INV-016: Input focus must survive Ctrl+1 tab switch; got focused={app.focused}"
            )

    async def test_debug_tab_exists(self) -> None:
        """debug_tab_exists [v0.5.0]: right column has Debug TabPane + #debug-transcript RichLog."""
        from textual.widgets import RichLog, TabPane

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            debug_tab = app.query_one("#debug-tab", TabPane)
            debug_log = app.query_one("#debug-log", RichLog)
            assert debug_log in debug_tab.walk_children()

    async def test_ctrl_2_activates_debug_tab(self) -> None:
        """ctrl_2_activates_debug_tab [v0.5.0]: Ctrl+2 → TabbedContent.active == 'debug-tab'."""
        from textual.widgets import TabbedContent

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            await pilot.press("ctrl+2")
            await pilot.pause()
            assert app.query_one("#side-panes", TabbedContent).active == "debug-tab"

    async def test_persona_tab_exists(self) -> None:
        """persona_tab_exists [v0.13.0]: right column has Persona TabPane +
        #persona-log RichLog as descendant.
        """
        from textual.widgets import RichLog, TabPane

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            persona_tab = app.query_one("#persona-tab", TabPane)
            persona_log = app.query_one("#persona-log", RichLog)
            assert persona_log in persona_tab.walk_children()

    async def test_ctrl_4_activates_persona_tab(self) -> None:
        """ctrl_4_activates_persona_tab [v0.13.0]: Ctrl+4 → active == 'persona-tab'."""
        from textual.widgets import TabbedContent

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            await pilot.press("ctrl+4")
            await pilot.pause()
            assert app.query_one("#side-panes", TabbedContent).active == "persona-tab"

    async def test_persona_header_starts_empty(self) -> None:
        """persona_header_starts_empty [v0.13.0]: sticky persona-header widget
        exists, starts hidden (height collapsed via .empty class) until
        hydration succeeds.
        """
        from textual.widgets import Static

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            header = app.query_one("#persona-header", Static)
            assert "empty" in header.classes

    async def test_done_label_styled_success(self) -> None:
        """done_label_styled_success [v0.9.0]: [done] label mounts as Static
        carrying a RichText with Aurora green style. Inspect the mounted
        Static's `.content`.
        """
        from rich.text import Text as RichText
        from textual.containers import VerticalScroll
        from textual.widgets import RichLog

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            from ratatoskr.tui import TuiPresenterState
            transcript = app.query_one("#transcript-scroll", VerticalScroll)
            state = TuiPresenterState()
            mounted: list = []
            orig_mount = transcript.mount

            def spy_mount(*ch, **kw):
                mounted.extend(ch)
                return orig_mount(*ch, **kw)

            transcript.mount = spy_mount  # type: ignore[method-assign]
            state.render(
                _make_tui_done(),
                transcript=transcript,
                tools_log=app.query_one("#tools-log", RichLog),
                debug_log=app.query_one("#debug-log", RichLog),
                thinking_log=MagicMock(),
                raw=True,
            )
            done = next(
                w.content for w in mounted
                if isinstance(getattr(w, "content", None), RichText)
                and _text_of(w.content).startswith("[done]")
            )
            assert done.style == "#16B866"  # Aurora green

    async def test_empty_state_placeholders_present(self) -> None:
        """empty_state_placeholders_present [v0.5.1]: tools-transcript + debug-transcript show
        placeholder lines before any turn fires."""
        from textual.widgets import RichLog

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            # Activate Debug tab so its content actually renders.
            from textual.widgets import TabbedContent
            tabbed = app.query_one("#side-panes", TabbedContent)
            tabbed.active = "debug-tab"
            await pilot.pause()
            tabbed.active = "tools-tab"
            await pilot.pause()
            tools_log = app.query_one("#tools-log", RichLog)
            debug_log = app.query_one("#debug-log", RichLog)
            tools_text = " ".join(str(line) for line in tools_log.lines)
            tabbed.active = "debug-tab"
            await pilot.pause()
            debug_text = " ".join(str(line) for line in debug_log.lines)
            assert "no tool events" in tools_text
            assert "worker_phase" in debug_text

    async def test_pane_name_updates_on_tab_switch(self) -> None:
        """pane_name_updates_on_tab_switch [v0.5.0]: pane-name reflects active tab.

        Two tabs now (Tools / Debug); pane-name updates from "Tools" to "Debug"
        and back as the operator switches via Ctrl+1 / Ctrl+2.
        """
        from textual.widgets import Static

        app = _resolved_app(_args_new(), session_id="s-new12345", agent_id="mimir")
        async with app.run_test() as pilot:
            await pilot.pause()
            pane_name = app.query_one("#pane-name", Static)
            await pilot.press("ctrl+2")
            await pilot.pause()
            assert str(pane_name.render()) == "Debug"
            await pilot.press("ctrl+1")
            await pilot.pause()
            assert str(pane_name.render()) == "Tools"


import asyncio  # noqa: E402

from textual.widgets import Input  # noqa: E402


async def _noop_worker(self, content: str) -> None:
    """Fake _stream_turn_worker that never completes (lets state stay 'streaming')."""
    await asyncio.Future()  # await forever; cancelled when test exits


class TestOnInputSubmitted:
    @respx.mock
    async def test_happy_submit_echoes_and_spawns(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """happy_submit_echoes_and_spawns [happy,tracer]: …"""
        monkeypatch.setattr(RatatoskrApp, "_stream_turn_worker", _noop_worker)
        writes = _spy_writes(monkeypatch)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            inp = app.query_one("#prompt", Input)
            inp.value = "hello"
            await inp.action_submit()
            await pilot.pause()
            assert any("❯ hello" in str(w) for w in writes)  # noqa: RUF001
            assert inp.value == ""
            assert app.state == "streaming"
            assert app.stream_worker is not None

    @respx.mock
    async def test_empty_submit_no_op(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """empty_submit_no_op [trace]: '' + Enter → no change; no worker spawned."""
        monkeypatch.setattr(RatatoskrApp, "_stream_turn_worker", _noop_worker)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            # Spy AFTER mount so identity-widget writes (if any) aren't counted.
            writes = _spy_writes(monkeypatch)
            inp = app.query_one("#prompt", Input)
            inp.value = ""
            await inp.action_submit()
            await pilot.pause()
            assert app.state == "idle"
            assert app.stream_worker is None
            # POST: no RichLog write fires on empty submit
            assert writes == []

    @respx.mock
    async def test_submit_during_streaming_shows_busy_notice(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """submit_during_streaming_shows_busy_notice [adversarial]: …"""
        monkeypatch.setattr(RatatoskrApp, "_stream_turn_worker", _noop_worker)
        writes = _spy_writes(monkeypatch)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            inp = app.query_one("#prompt", Input)
            # First submit: enters streaming
            inp.value = "first"
            await inp.action_submit()
            await pilot.pause()
            first_worker = app.stream_worker
            assert app.state == "streaming"
            # Second submit while streaming → busy notice; no new worker
            writes.clear()
            inp.value = "second"
            await inp.action_submit()
            await pilot.pause()
            assert any("[busy] turn in flight; input ignored" in str(w) for w in writes)
            assert app.stream_worker is first_worker  # unchanged
            assert app.state == "streaming"
            assert inp.value == ""

    @respx.mock
    async def test_submit_during_cancelling_shows_busy_notice(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """submit_during_cancelling_shows_busy_notice [adversarial]: …"""
        monkeypatch.setattr(RatatoskrApp, "_stream_turn_worker", _noop_worker)
        writes = _spy_writes(monkeypatch)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            app.state = "cancelling"  # bypass the natural transition for the test
            assert app.stream_worker is None  # no live worker before non-idle submit
            inp = app.query_one("#prompt", Input)
            inp.value = "x"
            await inp.action_submit()
            await pilot.pause()
            assert any("[busy]" in str(w) for w in writes)
            assert app.state == "cancelling"
            # POST-005 (from issue #4 on_input_submitted contract):
            # input cleared; NO new worker spawned during non-idle submit.
            assert inp.value == ""
            assert app.stream_worker is None

    @respx.mock
    async def test_footer_hint_flips_to_cancel(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """footer_hint_flips_to_cancel [trace]: hint widget shows 'Ctrl-C to cancel'."""
        from textual.widgets import Static

        monkeypatch.setattr(RatatoskrApp, "_stream_turn_worker", _noop_worker)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            hint_widget = app.query_one("#hint", Static)
            assert str(hint_widget.render()) == RatatoskrApp.HINT_IDLE
            inp = app.query_one("#prompt", Input)
            inp.value = "hi"
            await inp.action_submit()
            await pilot.pause()
            assert str(hint_widget.render()) == RatatoskrApp.HINT_STREAMING


import json  # noqa: E402


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


_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_STREAM_BODY = {
    "type": "cancelled",
    "phase": "cancelled",
    "turn_id": 42,
    "reason": "user_cancel",
    "partial_message_id": None,
}


def _sse_resp(body: bytes | httpx.AsyncByteStream) -> httpx.Response:
    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)


async def _submit_and_wait(app: RatatoskrApp, pilot, content: str) -> None:
    """Type content into the input and submit; wait for worker to finish."""
    inp = app.query_one("#prompt", Input)
    inp.value = content
    await inp.action_submit()
    await pilot.pause()  # let the Input.Submitted message dispatch
    # Poll until the worker resolves (state returns to idle)
    for _ in range(100):
        if app.state == "idle" and app.stream_worker is not None:
            return
        await pilot.pause(0.02)


class TestStreamTurnWorker:
    @respx.mock
    async def test_happy_text_done_no_double_print(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """happy_text_done_no_double_print [happy,tracer, v0.9.0]:
        Text("hello") mounts a Static(Markdown("hello")) into the transcript;
        Done mounts a [done] label Static. The Markdown is rendered live (one
        widget for the whole stream, updated in place), so there is NO
        post-Done re-render — exactly ONE Markdown renderable lands in the
        transcript for the response body. v0.9.0 supersedes v0.8.2's
        drop-Markdown patch with proper live rendering.
        """
        stream = _sse_chunk("42:1", {"type": "text", "content": "hello"}) + _sse_chunk(
            "42:2", _DONE_BODY
        )
        respx.post("https://w.example/sessions/s-1existing/messages").mock(
            return_value=_sse_resp(stream)
        )

        writes = _spy_writes(monkeypatch)

        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            await _submit_and_wait(app, pilot, "hi")
            assert app.state == "idle"
            from rich.markdown import Markdown

            # The response body lives as ONE Markdown renderable mounted into
            # the transcript; live updates happen via Static.update, not via
            # re-mount, so there's exactly one Markdown in the spy stream.
            markdowns = [w for w in writes if isinstance(w, Markdown)]
            assert len(markdowns) == 1, (
                f"v0.9.0: expected exactly ONE Markdown mounted, got {len(markdowns)}"
            )
            assert markdowns[0].markup == "hello"
            # [done] label fires too.
            assert any("[done]" in _text_of(w) for w in writes)

    @respx.mock
    async def test_raw_flag_skips_markdown_render(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """raw_flag_skips_markdown_render [trace, v0.9.0]:
        With --raw, the response widget holds plain str instead of Markdown.
        Turn-header markers still appear in every pane: 3 RichLog panes
        receive a Rule, the transcript-scroll receives a Static-wrapped
        RichText (mounted, not written), giving 3 Rules in the captured
        writes list.
        """
        stream = _sse_chunk("42:1", {"type": "text", "content": "hi"}) + _sse_chunk(
            "42:2", _DONE_BODY
        )
        respx.post("https://w.example/sessions/s-1existing/messages").mock(
            return_value=_sse_resp(stream)
        )
        writes = _spy_writes(monkeypatch)
        app = _resolved_app(_args_existing(raw=True))
        async with app.run_test() as pilot:
            await pilot.pause()
            await _submit_and_wait(app, pilot, "x")
            from rich.markdown import Markdown
            from rich.rule import Rule

            # No Markdown in raw mode.
            assert not any(isinstance(w, Markdown) for w in writes)
            # 3 Rules — one per RichLog pane (tools / debug / thinking).
            # Transcript-scroll uses a Static turn-header Markdown alternative.
            rules = [w for w in writes if isinstance(w, Rule)]
            assert len(rules) == 3, f"expected 3 turn-header Rules, got {len(rules)}"
            # Accumulated text "hi" mounted as plain str into transcript.
            assert "hi" in writes

    @respx.mock
    async def test_error_terminal_returns_to_idle(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """error_terminal_returns_to_idle [happy]: …"""
        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-1existing/messages").mock(
            return_value=_sse_resp(stream)
        )
        writes = _spy_writes(monkeypatch)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            await _submit_and_wait(app, pilot, "x")
            assert app.state == "idle"
            assert any("[error]" in str(w) for w in writes)

    @respx.mock
    async def test_cancelled_terminal_returns_to_idle(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """cancelled_terminal_returns_to_idle [happy]: …"""
        stream = _sse_chunk("42:1", {"type": "text", "content": "x"}) + _sse_chunk(
            "42:2", _CANCELLED_STREAM_BODY
        )
        respx.post("https://w.example/sessions/s-1existing/messages").mock(
            return_value=_sse_resp(stream)
        )
        writes = _spy_writes(monkeypatch)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            await _submit_and_wait(app, pilot, "x")
            assert app.state == "idle"
            assert any("[cancelled]" in str(w) for w in writes)

    @respx.mock
    async def test_active_turn_id_set_on_first_event(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """active_turn_id_set_on_first_event [trace]: …"""
        # Use a gated stream: yield first event, then hold, so we can inspect mid-stream
        first = _sse_chunk("42:1", {"type": "text", "content": "x"})
        gate = asyncio.Event()

        class _GatedAfterFirst(httpx.AsyncByteStream):
            async def __aiter__(self):
                yield first
                await gate.wait()
                yield _sse_chunk("42:2", _DONE_BODY)

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

        respx.post("https://w.example/sessions/s-1existing/messages").mock(
            return_value=_sse_resp(_GatedAfterFirst())
        )

        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            inp = app.query_one("#prompt", Input)
            inp.value = "x"
            await inp.action_submit()
            # Wait for first event to be processed (active_turn_id set)
            for _ in range(50):
                if app.active_turn_id is not None:
                    break
                await pilot.pause(0.02)
            assert app.active_turn_id == 42
            # Release the gate so the worker can finish and the app can shut down cleanly
            gate.set()
            for _ in range(50):
                if app.state == "idle":
                    break
                await pilot.pause(0.02)

    @respx.mock
    async def test_sse_connect_failed_returns_to_idle(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """sse_connect_failed_returns_to_idle [error]: …"""
        respx.post("https://w.example/sessions/s-1existing/messages").mock(
            return_value=httpx.Response(404, json={"error": "session_not_found"})
        )
        writes = _spy_writes(monkeypatch)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            await _submit_and_wait(app, pilot, "x")
            assert app.state == "idle"
            assert any("[sse_connect_failed]" in str(w) for w in writes)
            assert app.return_value is None  # app NOT exited per INV-008

    @respx.mock
    async def test_connection_dropped_returns_to_idle(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """connection_dropped_returns_to_idle [error]: …"""

        class _DropAfter(httpx.AsyncByteStream):
            async def __aiter__(self):
                yield _sse_chunk("42:1", {"type": "text", "content": "x"})
                raise httpx.RemoteProtocolError("drop")

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

        respx.post("https://w.example/sessions/s-1existing/messages").mock(
            return_value=_sse_resp(_DropAfter())
        )
        writes = _spy_writes(monkeypatch)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            await _submit_and_wait(app, pilot, "x")
            assert app.state == "idle"
            assert any("[connection_dropped]" in str(w) for w in writes)

    @respx.mock
    async def test_malformed_sse_data_returns_to_idle(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """malformed_sse_data_returns_to_idle [error]: bad-JSON → [malformed_sse_data]; idle."""
        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-1existing/messages").mock(
            return_value=_sse_resp(stream)
        )
        writes = _spy_writes(monkeypatch)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            await _submit_and_wait(app, pilot, "x")
            assert app.state == "idle"
            assert any("[malformed_sse_data]" in str(w) for w in writes)
            assert any("not-json" in str(w) for w in writes)
            # INV-008: mid-session error does NOT exit the app
            assert app.return_value is None

    @respx.mock
    async def test_rendered_event_per_event(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """rendered_event_per_event [trace]: …"""
        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-1existing/messages").mock(
            return_value=_sse_resp(chunks)
        )

        # Per issue #12: rendering went from stateless _render_event_to_log to
        # TuiPresenterState.render; the spy moves to the new method.
        from ratatoskr.tui import TuiPresenterState

        call_count = 0
        original = TuiPresenterState.render

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

        monkeypatch.setattr(TuiPresenterState, "render", spy)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            await _submit_and_wait(app, pilot, "x")
            assert call_count == 3


class TestActionInterrupt:
    @respx.mock
    async def test_idle_ctrl_c_exits_zero(self) -> None:
        """idle_ctrl_c_exits_zero [happy,tracer]: state=idle; ctrl+c → exit(0)."""
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            assert app.state == "idle"
            await pilot.press("ctrl+c")
            await pilot.pause()
        assert app.return_value == 0

    @respx.mock
    async def test_streaming_first_ctrl_c_cancels(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """streaming_first_ctrl_c_cancels [scenario,tracer]: …"""
        # Stream that yields one text event (sets active_turn_id) then waits forever
        first_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
        gate = asyncio.Event()

        class _GatedAfterFirst(httpx.AsyncByteStream):
            async def __aiter__(self):
                yield first_chunk
                await gate.wait()

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

        respx.post("https://w.example/sessions/s-1existing/messages").mock(
            return_value=_sse_resp(_GatedAfterFirst())
        )
        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-1existing/turns/42/cancel").mock(
            side_effect=cancel_handler
        )

        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            inp = app.query_one("#prompt", Input)
            inp.value = "go"
            await inp.action_submit()
            await pilot.pause()
            # Wait for active_turn_id to be set (first event consumed)
            for _ in range(50):
                if app.active_turn_id == 42:
                    break
                await pilot.pause(0.02)
            assert app.active_turn_id == 42
            assert app.state == "streaming"
            await pilot.press("ctrl+c")
            # Wait for cancel POST to land
            for _ in range(50):
                if cancel_observed.is_set():
                    break
                await pilot.pause(0.02)
            assert cancel_route.call_count == 1
            assert app.state == "cancelling"
            from textual.widgets import Static

            hint_widget = app.query_one("#hint", Static)
            assert str(hint_widget.render()) == RatatoskrApp.HINT_CANCELLING
            # Release the gate so the stream worker can finish cleanly during teardown
            gate.set()

    @respx.mock
    async def test_streaming_no_turn_id_force_exits(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """streaming_no_turn_id_force_exits [scenario]: …"""
        cancel_route = respx.post("https://w.example/sessions/s-1existing/turns/0/cancel").mock(
            return_value=httpx.Response(200, json=_CANCEL_OK_RESP)
        )
        # Stream that hangs forever (no events to set active_turn_id)
        gate = asyncio.Event()

        class _NeverYields(httpx.AsyncByteStream):
            async def __aiter__(self):
                await gate.wait()
                if False:
                    yield b""

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

        respx.post("https://w.example/sessions/s-1existing/messages").mock(
            return_value=_sse_resp(_NeverYields())
        )

        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            inp = app.query_one("#prompt", Input)
            inp.value = "go"
            await inp.action_submit()
            await pilot.pause()
            assert app.state == "streaming"
            assert app.active_turn_id is None
            # Capture worker reference + spy on its .cancel() before ctrl+c
            worker_ref = app.stream_worker
            assert worker_ref is not None
            cancel_calls: list = []
            original_cancel = type(worker_ref).cancel
            monkeypatch.setattr(
                type(worker_ref),
                "cancel",
                lambda self: (cancel_calls.append(self), original_cancel(self))[-1],
            )
            await pilot.press("ctrl+c")
            await pilot.pause()
            gate.set()  # let the gated stream resolve so teardown is clean
        assert app.return_value == 3
        assert cancel_route.call_count == 0
        # action_interrupt MUST cancel the stream worker on the no-active_turn_id force-exit path
        assert worker_ref in cancel_calls

    @respx.mock
    async def test_cancelling_second_ctrl_c_force_exits(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """cancelling_second_ctrl_c_force_exits [scenario]: …"""
        # Set up a real live stream worker (gated, hangs forever) so we can
        # observe action_interrupt's cancel() call on the second-Ctrl-C path.
        monkeypatch.setattr(RatatoskrApp, "_stream_turn_worker", _noop_worker)
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            inp = app.query_one("#prompt", Input)
            inp.value = "go"
            await inp.action_submit()
            await pilot.pause()
            assert app.stream_worker is not None
            app.state = "cancelling"  # bypass the natural transition for the test
            worker_ref = app.stream_worker
            cancel_calls: list = []
            original_cancel = type(worker_ref).cancel
            monkeypatch.setattr(
                type(worker_ref),
                "cancel",
                lambda self: (cancel_calls.append(self), original_cancel(self))[-1],
            )
            await pilot.press("ctrl+c")
            await pilot.pause()
        assert app.return_value == 3
        # Second-Ctrl-C in cancelling state MUST cancel the in-flight worker
        assert worker_ref in cancel_calls

    @respx.mock
    async def test_cancel_failed_swallowed(self) -> None:
        """cancel_failed_swallowed [scenario]: …"""
        first_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
        stream_gate = asyncio.Event()

        class _GatedAfterFirst(httpx.AsyncByteStream):
            async def __aiter__(self):
                yield first_chunk
                await stream_gate.wait()

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

        respx.post("https://w.example/sessions/s-1existing/messages").mock(
            return_value=_sse_resp(_GatedAfterFirst())
        )
        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-1existing/turns/42/cancel").mock(
            side_effect=cancel_handler
        )

        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            inp = app.query_one("#prompt", Input)
            inp.value = "go"
            await inp.action_submit()
            await pilot.pause()
            for _ in range(50):
                if app.active_turn_id == 42:
                    break
                await pilot.pause(0.02)
            await pilot.press("ctrl+c")
            for _ in range(50):
                if cancel_observed.is_set():
                    break
                await pilot.pause(0.02)
            # Give _cancel_via_sse time to write the [cancel_failed] line
            await pilot.pause(0.05)
            from textual.containers import VerticalScroll
            from textual.widgets import Static
            transcript = app.query_one("#transcript-scroll", VerticalScroll)
            rendered = "\n".join(
                str(child.content)
                for child in transcript.children
                if isinstance(child, Static)
            )
            assert "[cancel_failed]" in rendered
            assert app.state == "cancelling"
            stream_gate.set()  # let stream finish for teardown


class TestActionQuit:
    @respx.mock
    async def test_idle_ctrl_d_exits_zero(self) -> None:
        """idle_ctrl_d_exits_zero [happy,tracer]: state=idle; ctrl+d → exit(0)."""
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            await pilot.press("ctrl+d")
            await pilot.pause()
        assert app.return_value == 0

    @respx.mock
    async def test_streaming_ctrl_d_force_exits(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """streaming_ctrl_d_force_exits [scenario]: …"""
        cancel_route = respx.post("https://w.example/sessions/s-1existing/turns/42/cancel").mock(
            return_value=httpx.Response(200, json=_CANCEL_OK_RESP)
        )
        first_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
        gate = asyncio.Event()

        class _GatedAfterFirst(httpx.AsyncByteStream):
            async def __aiter__(self):
                yield first_chunk
                await gate.wait()

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

        respx.post("https://w.example/sessions/s-1existing/messages").mock(
            return_value=_sse_resp(_GatedAfterFirst())
        )
        app = _resolved_app(_args_existing())
        async with app.run_test() as pilot:
            await pilot.pause()
            inp = app.query_one("#prompt", Input)
            inp.value = "go"
            await inp.action_submit()
            await pilot.pause()
            for _ in range(50):
                if app.active_turn_id == 42:
                    break
                await pilot.pause(0.02)
            # Capture worker + spy on cancel before ctrl+d
            worker_ref = app.stream_worker
            assert worker_ref is not None
            cancel_calls: list = []
            original_cancel = type(worker_ref).cancel
            monkeypatch.setattr(
                type(worker_ref),
                "cancel",
                lambda self: (cancel_calls.append(self), original_cancel(self))[-1],
            )
            await pilot.press("ctrl+d")
            await pilot.pause()
            gate.set()
        assert app.return_value == 0
        assert cancel_route.call_count == 0
        # POST-002: Ctrl-D MUST cancel the in-flight stream worker (abandon-and-exit)
        assert worker_ref in cancel_calls


from ratatoskr.tui import run_tui  # noqa: E402


class TestResolveThenRun:
    """Tests at the `_resolve_then_run` layer — pre-`App.run()` session
    resolution + AsyncClient ownership + stderr error routing per issue #6.
    """

    @respx.mock
    def test_happy_new_session_resolve(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """happy_new_session_resolve [happy]: --new path through _resolve_then_run.

        Verifies POST /sessions count + SessionInfo propagation to RatatoskrApp's
        pre-resolved state (session_id / agent_id / client). The corresponding
        TestAppMount.test_happy_new_session_mount uses _resolved_app and bypasses
        _resolve_then_run entirely; this test exercises the production resolve
        path with a real POST /sessions mock.
        """
        sessions_route = respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(201, json=_CREATE_OK_RESP)
        )
        snapshot: dict = {}

        async def capture_run_async(self, *a, **kw):
            snapshot["session_id"] = self.session_id
            snapshot["agent_id"] = self.agent_id
            snapshot["client"] = self.client
            snapshot["client_open"] = not self.client.is_closed
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", capture_run_async)
        rc = run_tui(_args_new())
        assert rc == 0
        # Exactly one POST /sessions invocation by _resolve_then_run
        assert sessions_route.call_count == 1
        # SessionInfo fields propagated into the constructed App
        assert snapshot["session_id"] == "s-new12345"
        assert snapshot["agent_id"] == "mimir"
        assert snapshot["client"] is not None
        assert snapshot["client_open"] is True

    @respx.mock
    def test_happy_new_with_end_user_id_resolve(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """happy_new_with_end_user_id_resolve [happy]: args.end_user_id threads into POST body.

        Issue #5 amends #4: _resolve_then_run's create_session call now forwards
        args.end_user_id (renamed from the contract's _mount target, since #6
        moved session resolution out of on_mount into _resolve_then_run).
        """
        import json as _json

        sessions_route = respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(201, json=_CREATE_OK_RESP)
        )

        async def fake_run_async(self, *a, **kw):
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", fake_run_async)
        rc = run_tui(_args_new(end_user_id="alice"))
        assert rc == 0
        assert sessions_route.call_count == 1
        body = _json.loads(sessions_route.calls[0].request.content)
        assert body == {"agent_id": "mimir", "end_user_id": "alice"}

    @respx.mock
    def test_user_agent_header_sent(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """user_agent_header_sent [trace]: outbound requests carry the ratatoskr User-Agent.

        Worldtree-dev (althing 2026-05-23) requested consumers send `User-Agent:
        ratatoskr/<version> (<contact>)` so server logs can distinguish ratatoskr
        traffic from other consumers.
        """
        sessions_route = respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(201, json=_CREATE_OK_RESP)
        )

        async def fake_run_async(self, *a, **kw):
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", fake_run_async)
        rc = run_tui(_args_new())
        assert rc == 0
        ua = sessions_route.calls[0].request.headers["User-Agent"]
        assert ua.startswith("ratatoskr/")
        assert "vh@phasefinal.com" in ua

    @respx.mock
    def test_alt_screen_never_opens_on_resolve_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """alt_screen_never_opens_on_resolve_error [trace]: 404 → run_tui=12; run_async unhit.

        Directly probes INV-001: session resolution failures MUST short-circuit
        BEFORE the alt-screen opens.
        """
        respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(404, json={"error": "unknown_agent_id"})
        )
        sentinel_called = False

        async def sentinel(self, *a, **kw):
            nonlocal sentinel_called
            sentinel_called = True
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", sentinel)
        rc = run_tui(_args_new())
        assert rc == 12
        assert not sentinel_called

    @respx.mock
    def test_agent_not_found_on_resolve(self, capsys: pytest.CaptureFixture[str]) -> None:
        """agent_not_found_on_resolve [error]: --new + 404 → stderr [agent_not_found]; exit 12."""
        respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(404, json={"error": "unknown_agent_id"})
        )
        rc = run_tui(_args_new())
        err = capsys.readouterr().err
        assert rc == 12
        assert "[agent_not_found]" in err
        assert "agent_id=mimir" in err

    @respx.mock
    def test_session_api_failed_on_resolve(self, capsys: pytest.CaptureFixture[str]) -> None:
        """session_api_failed_on_resolve [error]: --new + 500 → [session_api_failed] stderr."""
        respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(500, content=b"server error")
        )
        rc = run_tui(_args_new())
        err = capsys.readouterr().err
        assert rc == 20
        assert "[session_api_failed]" in err
        assert "status=500" in err

    @respx.mock
    def test_network_error_on_resolve(self, capsys: pytest.CaptureFixture[str]) -> None:
        """network_error_on_resolve [error]: --new + ConnectError → [network_error] stderr."""
        respx.post("https://w.example/sessions").mock(side_effect=httpx.ConnectError("down"))
        rc = run_tui(_args_new())
        err = capsys.readouterr().err
        assert rc == 21
        assert "[network_error]" in err
        assert "ConnectError" in err

    @respx.mock
    def test_stderr_label_format_matches_cli(self, capsys: pytest.CaptureFixture[str]) -> None:
        """stderr_label_format_matches_cli [trace]: cli._amain and _resolve_then_run produce
        identical stderr lines for AgentNotFound (INV-006).
        """
        # Re-fetch cli's ParsedArgs from the current module state — test_cli's
        # `importlib.reload(ratatoskr.cli)` rebinds the class, so the top-of-file
        # `from ratatoskr.cli import ParsedArgs` may now refer to a stale class.
        from ratatoskr import cli as cli_mod

        respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(404, json={"error": "unknown_agent_id"})
        )
        cli_args = cli_mod.ParsedArgs(
            send_content="x",
            session_id=None,
            new=True,
            agent_id="mimir",
            api_key="k",
            server_url="https://w.example",
            raw=False,
        )
        # Drive cli._amain's error path (--send mode)
        cli_rc = asyncio.run(cli_mod._amain(cli_args))
        cli_err = capsys.readouterr().err
        # Drive _resolve_then_run's error path (TUI mode); _args_new() uses the
        # pre-reload ParsedArgs which still matches tui.run_tui's isinstance check.
        tui_rc = run_tui(_args_new())
        tui_err = capsys.readouterr().err
        # Same exit code, same verbatim stderr line.
        assert cli_rc == 12
        assert tui_rc == 12
        assert cli_err == tui_err
        assert cli_err == "[agent_not_found] agent_id=mimir\n"

    def test_client_open_after_resolve(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """client_open_after_resolve [trace]: app.client is open at the time run_async runs."""
        snapshot: dict = {}

        async def capture_run_async(self, *a, **kw):
            snapshot["client_is"] = self.client
            snapshot["closed_during_run"] = self.client.is_closed
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", capture_run_async)
        rc = run_tui(_args_existing())
        assert rc == 0
        assert snapshot["client_is"] is not None
        assert snapshot["closed_during_run"] is False

    def test_client_lifetime_owned_by_run_tui(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """client_lifetime_owned_by_run_tui [trace]: open during run_async, closed after run_tui.

        Probes INV-002: the App is a consumer of an externally-owned client;
        the async-with in run_tui closes it, not on_unmount.
        """
        snapshot: dict = {}

        async def capture_run_async(self, *a, **kw):
            # During run_async (the alt-screen lifetime) the client is open.
            snapshot["client"] = self.client
            snapshot["closed_during_run"] = self.client.is_closed
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", capture_run_async)
        rc = run_tui(_args_existing())
        assert rc == 0
        client = snapshot["client"]
        assert client is not None
        # Open while the app was running; closed by run_tui's async-with after.
        assert snapshot["closed_during_run"] is False
        assert client.is_closed is True

    def test_run_tui_closes_client_on_app_exit(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """run_tui_closes_client_on_app_exit: async-with closes client after app.run_async ret."""
        seen_clients: list[httpx.AsyncClient] = []

        async def fake_run_async(self, *a, **kw):
            seen_clients.append(self.client)
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", fake_run_async)
        rc = run_tui(_args_existing())
        assert rc == 0
        assert len(seen_clients) == 1
        # After run_tui returns, the client should be closed by the async-with
        assert seen_clients[0].is_closed

    async def test_on_unmount_does_not_close_client(self) -> None:
        """on_unmount narrowed [trace]: probes INV-002 from the on_unmount side.

        The complementary check to test_client_lifetime_owned_by_run_tui (which
        patches run_async and so never exercises on_unmount). Here we DO run the
        real on_unmount via Pilot ctrl+d → app teardown, and assert the client
        is still open afterward (close site is run_tui's async-with, which is
        NOT entered in this Pilot-driven test).
        """
        client = httpx.AsyncClient(
            base_url="https://w.example",
            headers={"Authorization": "Bearer k"},
            timeout=httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0),
        )
        app = RatatoskrApp(
            _args_existing(),
            session_id="s-1existing",
            agent_id=None,
            client=client,
        )
        async with app.run_test() as pilot:
            await pilot.pause()
            assert client.is_closed is False
            await pilot.press("ctrl+d")
            await pilot.pause()
        # After app.run_test() teardown, on_unmount has fired. Per INV-002 the
        # client MUST still be open — only run_tui's async-with closes it.
        assert client.is_closed is False
        await client.aclose()  # test-side cleanup


class TestRunTui:
    def test_happy_returns_zero_on_quit(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """happy_returns_zero_on_quit [happy,tracer]: run_tui propagates app.run_async exit code."""
        captured: list[ParsedArgs] = []

        async def fake_run_async(self, *a, **kw):
            captured.append(self.args)
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", fake_run_async)
        rc = run_tui(_args_existing())
        assert rc == 0
        assert len(captured) == 1
        assert captured[0].send_content is None

    def test_precondition_send_content_none(self) -> None:
        """precondition_send_content_none [adversarial]: …"""
        bad_args = ParsedArgs(
            send_content="x",  # PRE-001 violation
            session_id="s-1",
            new=False,
            agent_id=None,
            api_key="k",
            server_url="https://w.example",
            raw=False,
        )
        with pytest.raises(AssertionError):
            run_tui(bad_args)


# ---- Issue #8: startup agent picker ----------------------------------------


def _args_new_no_agent(**overrides) -> ParsedArgs:
    """ParsedArgs for bare --new (no --agent) — TUI-mode picker entry."""
    base = dict(
        send_content=None,
        session_id=None,
        new=True,
        agent_id=None,  # Issue #8: bare --new, picker drives the choice
        api_key="k",
        server_url="https://w.example",
        raw=False,
    )
    base.update(overrides)
    return ParsedArgs(**base)


_AGENTS_RESP = [
    {
        "agent_id": "mimir",
        "name": "Mimir",
        "description": "Keeper of the Well of Knowledge.",
    },
    {
        "agent_id": "lofn",
        "name": "Lofn",
        "description": "Mediator of secret affairs.",
    },
]


class TestAgentPickerApp:
    def test_picker_renders_rows(self) -> None:
        """picker_renders_rows: AgentPickerApp composes one ListItem per agent."""
        from textual.widgets import ListView

        from ratatoskr.sessions import AgentInfo
        from ratatoskr.tui import AgentPickerApp

        agents = [
            AgentInfo(
                agent_id="a",
                name="A",
                description="x",
                version=None,
                capabilities=[],
                supported_models=[],
                persona_traits={},
                ui_hints={},
            ),
            AgentInfo(
                agent_id="b",
                name="B",
                description="y",
                version=None,
                capabilities=[],
                supported_models=[],
                persona_traits={},
                ui_hints={},
            ),
        ]
        app = AgentPickerApp(agents)

        async def probe() -> None:
            async with app.run_test() as pilot:
                lv = app.query_one("#agent-list", ListView)
                assert len(lv.children) == 2
                await pilot.pause()
                app.exit(None)

        import asyncio

        asyncio.run(probe())

    def test_picker_pick_returns_agent_id(self) -> None:
        """picker_pick_returns_agent_id: highlight idx 1 + Enter → exit value == 'b'."""
        from ratatoskr.sessions import AgentInfo
        from ratatoskr.tui import AgentPickerApp

        agents = [
            AgentInfo(
                agent_id="a",
                name="A",
                description="x",
                version=None,
                capabilities=[],
                supported_models=[],
                persona_traits={},
                ui_hints={},
            ),
            AgentInfo(
                agent_id="b",
                name="B",
                description="y",
                version=None,
                capabilities=[],
                supported_models=[],
                persona_traits={},
                ui_hints={},
            ),
        ]
        app = AgentPickerApp(agents)

        async def drive() -> str | None:
            async with app.run_test() as pilot:
                from textual.widgets import ListView

                lv = app.query_one("#agent-list", ListView)
                lv.index = 1
                await pilot.pause()
                await pilot.press("enter")
                await pilot.pause()
            return app.return_value

        import asyncio

        chosen = asyncio.run(drive())
        assert chosen == "b"

    def test_picker_esc_returns_none(self) -> None:
        """picker_esc_returns_none: Esc → exit value is None."""
        from ratatoskr.sessions import AgentInfo
        from ratatoskr.tui import AgentPickerApp

        agents = [
            AgentInfo(
                agent_id="a",
                name="A",
                description="x",
                version=None,
                capabilities=[],
                supported_models=[],
                persona_traits={},
                ui_hints={},
            ),
        ]
        app = AgentPickerApp(agents)

        async def drive() -> str | None:
            async with app.run_test() as pilot:
                await pilot.press("escape")
                await pilot.pause()
            return app.return_value

        import asyncio

        chosen = asyncio.run(drive())
        assert chosen is None


class TestResolveThenRunWithPicker:
    """Issue #8: picker integration in _resolve_then_run."""

    @respx.mock
    def test_picker_happy_path(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """picker_happy_path [happy,tracer]: agents fetched → picker exits with id → create_session.

        Patches AgentPickerApp.run_async to return 'lofn' (simulating user pick);
        asserts list_agents fired once, POST /sessions body carries agent_id=lofn,
        and RatatoskrApp opens with the chosen identity.
        """
        agents_route = respx.get("https://w.example/agents").mock(
            return_value=httpx.Response(200, json=_AGENTS_RESP)
        )
        sessions_route = respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(
                201,
                json={
                    **_CREATE_OK_RESP,
                    "agent_id": "lofn",
                },
            )
        )

        from ratatoskr.tui import AgentPickerApp

        async def picker_returns_lofn(self, *a, **kw):
            return "lofn"

        monkeypatch.setattr(AgentPickerApp, "run_async", picker_returns_lofn)

        snapshot: dict = {}

        async def capture_main(self, *a, **kw):
            snapshot["session_id"] = self.session_id
            snapshot["agent_id"] = self.agent_id
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", capture_main)
        from ratatoskr.tui import run_tui

        rc = run_tui(_args_new_no_agent())
        assert rc == 0
        assert agents_route.call_count == 1
        assert sessions_route.call_count == 1
        import json as _json

        body = _json.loads(sessions_route.calls[0].request.content)
        assert body == {"agent_id": "lofn"}
        assert snapshot["agent_id"] == "lofn"

    @respx.mock
    def test_picker_esc_clean_exit(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """picker_esc_clean_exit: picker returns None → exit 0; no create_session; no main App."""
        agents_route = respx.get("https://w.example/agents").mock(
            return_value=httpx.Response(200, json=_AGENTS_RESP)
        )
        sessions_route = respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(201, json=_CREATE_OK_RESP)
        )

        from ratatoskr.tui import AgentPickerApp

        async def picker_dismissed(self, *a, **kw):
            return None

        monkeypatch.setattr(AgentPickerApp, "run_async", picker_dismissed)

        main_called = False

        async def sentinel(self, *a, **kw):
            nonlocal main_called
            main_called = True
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", sentinel)
        from ratatoskr.tui import run_tui

        rc = run_tui(_args_new_no_agent())
        assert rc == 0
        assert agents_route.call_count == 1
        assert sessions_route.call_count == 0
        assert main_called is False

    @respx.mock
    def test_picker_skipped_when_agent_id_provided(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """picker_skipped_when_agent_id_provided: --new --agent mimir → list_agents NOT called."""
        agents_route = respx.get("https://w.example/agents").mock(
            return_value=httpx.Response(200, json=_AGENTS_RESP)
        )
        sessions_route = respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(201, json=_CREATE_OK_RESP)
        )

        async def fake_main(self, *a, **kw):
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", fake_main)
        from ratatoskr.tui import run_tui

        rc = run_tui(_args_new())  # agent_id="mimir"
        assert rc == 0
        assert agents_route.call_count == 0
        assert sessions_route.call_count == 1

    @respx.mock
    def test_picker_merges_local_tier3_agents(
        self,
        monkeypatch: pytest.MonkeyPatch,
        tmp_path: Path,
    ) -> None:
        """picker_merges_local_tier3_agents [v0.8.0]: local index entries
        appear in the picker's agent list alongside remote agents."""
        from ratatoskr.local_agents import LocalAgentEntry, add_local_agent

        # Isolate the local index in a tmp file.
        monkeypatch.setenv(
            "RATATOSKR_LOCAL_AGENTS", str(tmp_path / "local_agents.json")
        )
        add_local_agent(LocalAgentEntry(
            agent_id="ratatoskr:wizard",
            agent_name="wizard",
            model="qwen3.6-35-a3b",
            description="(tier 3) test wizard",
            defined_at="2026-05-25T00:00:00+00:00",
        ))
        respx.get("https://w.example/agents").mock(
            return_value=httpx.Response(200, json=_AGENTS_RESP)
        )
        respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(201, json=_CREATE_OK_RESP)
        )

        from ratatoskr.tui import AgentPickerApp

        captured: list = []

        async def capture_picker_init(self, *a, **kw):
            captured.append(list(self.agents))
            return "mimir"  # auto-pick something so the rest succeeds

        # Patch __init__ to capture the agent list passed to the picker.
        orig_init = AgentPickerApp.__init__

        def init_spy(self, agents):
            captured.append(list(agents))
            orig_init(self, agents)

        monkeypatch.setattr(AgentPickerApp, "__init__", init_spy)

        async def picker_returns_mimir(self):
            return "mimir"

        monkeypatch.setattr(AgentPickerApp, "run_async", picker_returns_mimir)

        async def fake_main(self, *a, **kw):
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", fake_main)
        from ratatoskr.tui import run_tui

        rc = run_tui(_args_new_no_agent())
        assert rc == 0
        # The local tier-3 agent should appear in the picker's agents list.
        assert captured, "AgentPickerApp.__init__ was never called"
        agent_ids = {a.agent_id for a in captured[0]}
        assert "ratatoskr:wizard" in agent_ids
        # Plus the remote agents.
        assert "mimir" in agent_ids
        assert "lofn" in agent_ids

    @respx.mock
    def test_picker_skipped_when_session_mode(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """picker_skipped_when_session_mode: --session s-1 → no list_agents, no create_session."""
        agents_route = respx.get("https://w.example/agents").mock(
            return_value=httpx.Response(200, json=_AGENTS_RESP)
        )
        sessions_route = respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(201, json=_CREATE_OK_RESP)
        )

        async def fake_main(self, *a, **kw):
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", fake_main)
        from ratatoskr.tui import run_tui

        rc = run_tui(_args_existing())
        assert rc == 0
        assert agents_route.call_count == 0
        assert sessions_route.call_count == 0

    @respx.mock
    def test_picker_list_agents_session_api_failed(
        self,
        monkeypatch: pytest.MonkeyPatch,
        capsys: pytest.CaptureFixture[str],
    ) -> None:
        """list_agents 500 → stderr [session_api_failed]; exit 20; picker NOT opened."""
        respx.get("https://w.example/agents").mock(
            return_value=httpx.Response(500, content=b"oops")
        )

        from ratatoskr.tui import AgentPickerApp

        picker_called = False

        async def sentinel(self, *a, **kw):
            nonlocal picker_called
            picker_called = True
            return None

        monkeypatch.setattr(AgentPickerApp, "run_async", sentinel)

        main_called = False

        async def main_sentinel(self, *a, **kw):
            nonlocal main_called
            main_called = True
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", main_sentinel)
        from ratatoskr.tui import run_tui

        rc = run_tui(_args_new_no_agent())
        assert rc == 20
        err = capsys.readouterr().err
        assert "[session_api_failed]" in err
        assert "status=500" in err
        assert picker_called is False
        assert main_called is False

    @respx.mock
    def test_picker_empty_list(
        self,
        monkeypatch: pytest.MonkeyPatch,
        capsys: pytest.CaptureFixture[str],
        tmp_path: Path,
    ) -> None:
        """list_agents returns [] AND no local tier-3 entries → stderr
        [no_agents]; exit 13; picker NOT opened. Isolate
        $RATATOSKR_LOCAL_AGENTS so the operator's real local index
        doesn't merge in and turn this into a non-empty list."""
        # v0.8.0 isolation: point local agents at an empty tmp file.
        monkeypatch.setenv(
            "RATATOSKR_LOCAL_AGENTS", str(tmp_path / "empty_local_agents.json")
        )
        respx.get("https://w.example/agents").mock(return_value=httpx.Response(200, json=[]))

        from ratatoskr.tui import AgentPickerApp

        picker_called = False

        async def sentinel(self, *a, **kw):
            nonlocal picker_called
            picker_called = True
            return None

        monkeypatch.setattr(AgentPickerApp, "run_async", sentinel)
        from ratatoskr.tui import run_tui

        rc = run_tui(_args_new_no_agent())
        assert rc == 13
        err = capsys.readouterr().err
        assert "[no_agents]" in err
        assert picker_called is False


class TestTuiBifrostBind:
    """Issue #17 slice 3b — TUI bind trigger: bind failures route to the real
    stderr BEFORE the alt-screen opens (INV-002, mirrors issue #6; same exit
    codes/vocabulary as cli._amain per INV-006)."""

    @respx.mock
    async def test_handshake_failure_routes_pre_altscreen(
        self, capsys: pytest.CaptureFixture[str]
    ) -> None:
        from ratatoskr.tui import _resolve_then_run

        respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(
                502,
                json={
                    "error_code": "bifrost_handshake_failed",
                    "detail": {"bifrost_error": "bifrost.auth_rejected"},
                },
            )
        )
        args = _args_new(
            agent_id="ratatoskr:sindra",
            bifrost=BifrostBinding(endpoint_url="http://10.100.10.50:8391"),
            bifrost_plane="memory",
            consumer_key="ck",
        )
        rc = await _resolve_then_run(args)
        assert rc == 23
        err = capsys.readouterr().err
        assert "bifrost.auth_rejected" in err
        assert "consumer key" in err  # the 401-scoping hint

    @respx.mock
    async def test_consumer_key_missing_routes_pre_altscreen(
        self, capsys: pytest.CaptureFixture[str]
    ) -> None:
        from ratatoskr.tui import _resolve_then_run

        args = _args_new(
            agent_id="a",
            bifrost=BifrostBinding(endpoint_url="http://x:8391"),
            consumer_key=None,
        )
        rc = await _resolve_then_run(args)
        assert rc == 22
        assert "bifrost_consumer_key_missing" in capsys.readouterr().err

    @respx.mock
    async def test_bound_create_carries_binding_and_consumer_key(
        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """A successful bound create sends the bifrost body + the consumer-key
        bearer and prints the bound-state indicator (run_async stubbed so no
        alt-screen opens)."""
        from ratatoskr import tui as tui_mod
        from ratatoskr.tui import _resolve_then_run

        route = respx.post("https://w.example/sessions").mock(
            return_value=httpx.Response(
                201,
                json={
                    "session_id": "s-bound",
                    "agent_id": "ratatoskr:sindra",
                    "message_count": 0,
                    "created_at": "2026-06-18T12:00:00+00:00",
                    "last_active": "2026-06-18T12:00:00+00:00",
                    "metadata": {},
                },
            )
        )
        # sindra is a preset agent → the TUI create path now auto-seeds a #347 first-message.
        respx.post("https://w.example/sessions/s-bound/history").mock(
            return_value=httpx.Response(201, json={})
        )

        async def fake_run_async(self) -> int:
            return 0

        monkeypatch.setattr(tui_mod.RatatoskrApp, "run_async", fake_run_async)
        args = _args_new(
            agent_id="ratatoskr:sindra",
            bifrost=BifrostBinding(endpoint_url="http://10.100.10.50:8391"),
            bifrost_plane="memory",
            consumer_key="ck",
        )
        rc = await _resolve_then_run(args)
        assert rc == 0
        import json as _json

        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: status=bound" in err
        assert "plane=memory" in err


class TestSessionPickerApp:
    """docs/contracts/issues/6.contract.md FN SessionPickerApp (amendment slice b2)."""

    @staticmethod
    def _two():
        from ratatoskr.sessions import SessionInfo

        return [
            SessionInfo(
                session_id="s-first-0001", agent_id="mimir", created_at="t0",
                last_active="t1", metadata={}, message_count=3, name=None,
                archived=False, tags=[],
            ),
            SessionInfo(
                session_id="s-second-002", agent_id="echo", created_at="t0",
                last_active="t2", metadata={}, message_count=None, name="probe",
                archived=False, tags=[],
            ),
        ]

    def test_pick_returns_session_id(self) -> None:
        """pick_returns_session_id [happy,tracer]: idx 1 + Enter → exit value == that session_id."""
        from ratatoskr.tui import SessionPickerApp

        app = SessionPickerApp(self._two())

        async def drive() -> str | None:
            async with app.run_test() as pilot:
                from textual.widgets import ListView

                lv = app.query_one("#session-list", ListView)
                lv.index = 1
                await pilot.pause()
                await pilot.press("enter")
                await pilot.pause()
            return app.return_value

        import asyncio

        assert asyncio.run(drive()) == "s-second-002"

    def test_esc_returns_none(self) -> None:
        """esc_returns_none [happy]: Esc → exit value is None (dismiss, resume nothing)."""
        from ratatoskr.tui import SessionPickerApp

        app = SessionPickerApp(self._two())

        async def drive() -> str | None:
            async with app.run_test() as pilot:
                await pilot.press("escape")
                await pilot.pause()
            return app.return_value

        import asyncio

        assert asyncio.run(drive()) is None

    def test_ctrl_d_returns_none(self) -> None:
        """ctrl_d_returns_none [adversarial]: Ctrl-D → None."""
        from ratatoskr.tui import SessionPickerApp

        app = SessionPickerApp(self._two())

        async def drive() -> str | None:
            async with app.run_test() as pilot:
                await pilot.press("ctrl+d")
                await pilot.pause()
            return app.return_value

        import asyncio

        assert asyncio.run(drive()) is None


class TestBareSessionPicker:
    """docs/contracts/issues/6.contract.md amendment (slice b2): _resolve_then_run bare mode."""

    @staticmethod
    def _bare_args() -> ParsedArgs:
        return ParsedArgs(
            send_content=None, session_id=None, new=False, agent_id=None,
            api_key="k", server_url="https://w.example", raw=False,
            end_user_id=None, bifrost=None, bifrost_plane=None, consumer_key=None,
        )

    @staticmethod
    def _sess(sid: str, agent: str = "mimir"):
        from ratatoskr.sessions import SessionInfo

        return SessionInfo(
            session_id=sid, agent_id=agent, created_at="t0", last_active="t1",
            metadata={}, message_count=1, name=None, archived=False, tags=[],
        )

    def test_bare_zero_sessions_errors(
        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """bare_zero_sessions_errors [error]: 0 sessions → exit 14 [no_sessions]; App not opened."""
        import ratatoskr.tui as tui_mod
        from ratatoskr.sessions import SessionPage

        async def fake_list(client, **kw):
            return SessionPage(items=[], next_cursor=None)

        monkeypatch.setattr(tui_mod, "list_sessions", fake_list)
        opened: list[int] = []

        async def spy(self, *a, **k):
            opened.append(1)
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", spy)
        rc = run_tui(self._bare_args())
        assert rc == 14
        assert "[no_sessions]" in capsys.readouterr().err
        assert not opened

    def test_bare_one_session_auto_resumes(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """bare_one_session_auto_resumes: exactly 1 → auto-resume, no picker (§4 >1 rule)."""
        import ratatoskr.tui as tui_mod
        from ratatoskr.sessions import SessionPage
        from ratatoskr.tui import SessionPickerApp

        async def fake_list(client, **kw):
            return SessionPage(items=[self._sess("s-solo")], next_cursor=None)

        monkeypatch.setattr(tui_mod, "list_sessions", fake_list)
        picker_used: list[int] = []

        async def spy_picker(self, *a, **k):
            picker_used.append(1)
            return None

        monkeypatch.setattr(SessionPickerApp, "run_async", spy_picker)
        snap: dict = {}

        async def cap(self, *a, **k):
            snap["sid"] = self.session_id
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", cap)
        rc = run_tui(self._bare_args())
        assert rc == 0
        assert snap["sid"] == "s-solo"
        assert not picker_used

    def test_bare_multi_opens_picker(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """bare_multi_opens_picker [scenario,tracer]: >1 → picker; its choice resumes."""
        import ratatoskr.tui as tui_mod
        from ratatoskr.sessions import SessionPage
        from ratatoskr.tui import SessionPickerApp

        async def fake_list(client, **kw):
            return SessionPage(items=[self._sess("s-a"), self._sess("s-b")], next_cursor=None)

        monkeypatch.setattr(tui_mod, "list_sessions", fake_list)

        async def pick_b(self, *a, **k):
            return "s-b"

        monkeypatch.setattr(SessionPickerApp, "run_async", pick_b)
        snap: dict = {}

        async def cap(self, *a, **k):
            snap["sid"] = self.session_id
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", cap)
        rc = run_tui(self._bare_args())
        assert rc == 0
        assert snap["sid"] == "s-b"

    def test_bare_picker_dismiss_exits_zero(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """bare_picker_dismiss_exits_zero [scenario]: picker None → exit 0; App not opened."""
        import ratatoskr.tui as tui_mod
        from ratatoskr.sessions import SessionPage
        from ratatoskr.tui import SessionPickerApp

        async def fake_list(client, **kw):
            return SessionPage(items=[self._sess("s-a"), self._sess("s-b")], next_cursor=None)

        monkeypatch.setattr(tui_mod, "list_sessions", fake_list)

        async def pick_none(self, *a, **k):
            return None

        monkeypatch.setattr(SessionPickerApp, "run_async", pick_none)
        opened: list[int] = []

        async def spy(self, *a, **k):
            opened.append(1)
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", spy)
        rc = run_tui(self._bare_args())
        assert rc == 0
        assert not opened

    def test_bare_list_sessions_api_failure(
        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """bare_list_sessions_api_failure [error]: list_sessions 500 → exit 20; App not opened."""
        import ratatoskr.tui as tui_mod
        from ratatoskr.sessions import SessionApiFailed

        async def fake_list(client, **kw):
            raise SessionApiFailed(status=500, body=b"boom")

        monkeypatch.setattr(tui_mod, "list_sessions", fake_list)
        opened: list[int] = []

        async def spy(self, *a, **k):
            opened.append(1)
            return 0

        monkeypatch.setattr(RatatoskrApp, "run_async", spy)
        rc = run_tui(self._bare_args())
        assert rc == 20
        assert "[session_api_failed]" in capsys.readouterr().err
        assert not opened


class TestSessionToolsHydration:
    """get_session_tools + the #183 Tools-pane inventory hydrate (GET /sessions/{id}/tools)."""

    def test_format_tool_inventory(self) -> None:
        """format_tool_inventory [unit]: header + builtin + bifrost lines."""
        from ratatoskr.tui import _format_tool_inventory

        lines = _format_tool_inventory(
            {
                "agent_id": "alice:wizard",
                "builtin_tools": [],
                "bifrost_tools": [{"name": "bifrost.x"}, {"name": "bifrost.y"}],
            }
        )
        joined = "\n".join(lines)
        assert "agent=alice:wizard" in joined
        assert "builtin=0 bifrost=2" in joined
        assert "builtin: (none)" in joined
        assert "bifrost.x, bifrost.y" in joined

    async def test_hydrate_writes_inventory_and_audits(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """hydrate_writes_inventory [scenario,tracer]: 200 → inventory in Tools pane + audit."""
        import ratatoskr.tui as tui_mod

        writes = _spy_writes(monkeypatch)

        async def fake_tools(client, session_id):
            return {
                "agent_id": "alice:wizard",
                "builtin_tools": [],
                "bifrost_tools": [{"name": "bifrost.set_field"}],
            }

        monkeypatch.setattr(tui_mod, "get_session_tools", fake_tools)
        app = _resolved_app(_args_existing(session_id="s-tools-01"))
        async with app.run_test() as pilot:
            await pilot.pause()
            await app._hydrate_session_tools()
            await pilot.pause()
        joined = " ".join(_text_of(w) for w in writes)
        assert "session tool inventory" in joined
        assert "bifrost.set_field" in joined
        assert "session_tools_hydrated" in joined  # audit line landed

    async def test_hydrate_failure_audits_no_crash(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """hydrate_failure [error]: get_session_tools raises → failure audit; no crash."""
        import ratatoskr.tui as tui_mod
        from ratatoskr.sessions import SessionApiFailed

        writes = _spy_writes(monkeypatch)

        async def boom(client, session_id):
            raise SessionApiFailed(status=404, body=b"session_not_found")

        monkeypatch.setattr(tui_mod, "get_session_tools", boom)
        app = _resolved_app(_args_existing(session_id="s-tools-02"))
        async with app.run_test() as pilot:
            await pilot.pause()
            await app._hydrate_session_tools()
            await pilot.pause()
        joined = " ".join(_text_of(w) for w in writes)
        assert "session_tools_hydration_failed" in joined


class TestBifrostStateHydration:
    """get_session_bifrost + the #176 BifrostState pane (GET /admin/sessions/{id}/bifrost)."""

    @staticmethod
    def _mute_tools(monkeypatch: pytest.MonkeyPatch) -> None:
        """Neutralize the on_mount Tools-pane worker so it makes no real call."""
        import ratatoskr.tui as tui_mod

        async def noop(client, session_id):
            return {"agent_id": "x", "builtin_tools": [], "bifrost_tools": []}

        monkeypatch.setattr(tui_mod, "get_session_tools", noop)

    def test_format_bifrost_state(self) -> None:
        """format_bifrost_state [unit]: connected / endpoint / caps / tools lines."""
        from ratatoskr.tui import _format_bifrost_state

        lines = _format_bifrost_state(
            {
                "endpoint_url": "https://b/mcp",
                "consumer_id": "alice",
                "connected": True,
                "capabilities_granted": ["tools:call", "tools:read"],
                "tools": [{"name": "bifrost.echo"}],
            }
        )
        joined = "\n".join(lines)
        assert "connected=True" in joined
        assert "consumer=alice" in joined
        assert "https://b/mcp" in joined
        assert "tools:call, tools:read" in joined
        assert "bifrost.echo" in joined

    async def test_hydrate_no_admin_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """hydrate_no_admin_key [scenario]: admin_key None → 'not configured' + skip audit."""
        self._mute_tools(monkeypatch)
        writes = _spy_writes(monkeypatch)
        app = _resolved_app(_args_existing(session_id="s-bf-1"))  # admin_key defaults None
        async with app.run_test() as pilot:
            await pilot.pause()
            await app._hydrate_bifrost_state()
            await pilot.pause()
        joined = " ".join(_text_of(w) for w in writes)
        assert "admin key not configured" in joined
        assert "bifrost_state_skipped" in joined

    async def test_hydrate_success(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """hydrate_success [scenario,tracer]: 200 → binding in BifrostState pane + audit."""
        import ratatoskr.tui as tui_mod

        self._mute_tools(monkeypatch)
        writes = _spy_writes(monkeypatch)

        async def fake_bifrost(client, session_id, *, admin_key):
            return {
                "endpoint_url": "https://b/mcp",
                "consumer_id": "alice",
                "connected": True,
                "capabilities_granted": ["tools:call"],
                "tools": [{"name": "bifrost.echo"}],
            }

        monkeypatch.setattr(tui_mod, "get_session_bifrost", fake_bifrost)
        app = _resolved_app(_args_existing(session_id="s-bf-2", admin_key="ak"))
        async with app.run_test() as pilot:
            await pilot.pause()
            await app._hydrate_bifrost_state()
            await pilot.pause()
        joined = " ".join(_text_of(w) for w in writes)
        assert "bifrost binding" in joined
        assert "bifrost.echo" in joined
        assert "bifrost_state_hydrated" in joined

    async def test_hydrate_404_not_bound(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """hydrate_404_not_bound [error]: 404 → 'not bound to Bifrost' + audit; no crash."""
        import ratatoskr.tui as tui_mod
        from ratatoskr.sessions import SessionApiFailed

        self._mute_tools(monkeypatch)
        writes = _spy_writes(monkeypatch)

        async def not_bound(client, session_id, *, admin_key):
            raise SessionApiFailed(status=404, body=b"session_not_bifrost_bound")

        monkeypatch.setattr(tui_mod, "get_session_bifrost", not_bound)
        app = _resolved_app(_args_existing(session_id="s-bf-3", admin_key="ak"))
        async with app.run_test() as pilot:
            await pilot.pause()
            await app._hydrate_bifrost_state()
            await pilot.pause()
        joined = " ".join(_text_of(w) for w in writes)
        assert "not bound to Bifrost" in joined
        assert "bifrost_state_unavailable" in joined


class TestAdminEventsStream:
    """stream_admin_events + the #11 AdminEvents pane (GET /admin/events, session-filtered)."""

    @staticmethod
    def _mute_hydrates(monkeypatch: pytest.MonkeyPatch) -> None:
        """Neutralize the other on_mount workers (tools + bifrost) — no real calls."""
        import ratatoskr.tui as tui_mod
        from ratatoskr.sessions import SessionApiFailed

        async def noop_tools(client, session_id):
            return {"agent_id": "x", "builtin_tools": [], "bifrost_tools": []}

        async def noop_bifrost(client, session_id, *, admin_key):
            raise SessionApiFailed(status=404, body=b"nb")

        monkeypatch.setattr(tui_mod, "get_session_tools", noop_tools)
        monkeypatch.setattr(tui_mod, "get_session_bifrost", noop_bifrost)

    def test_format_admin_event(self) -> None:
        """format_admin_event [unit]: HH:MM:SS + type + fields; session_id dropped."""
        from ratatoskr.sse_client import AdminEvent
        from ratatoskr.tui import _format_admin_event

        line = _format_admin_event(
            AdminEvent(
                42, "turn.completed", "2026-05-06T10:00:05.000Z",
                {"session_id": "s1", "turn_id": 7, "duration_ms": 1200, "phase": "succeeded"},
            )
        )
        assert "turn.completed" in line
        assert "[10:00:05]" in line
        assert "turn_id=7" in line
        assert "session_id" not in line  # dropped — pane is already session-scoped

    def test_admin_event_matches_filter(self) -> None:
        """admin_event_matches [unit]: active-session + non-heartbeat system.* pass (§6)."""
        from ratatoskr.sse_client import AdminEvent

        E = AdminEvent
        app = _resolved_app(_args_existing(session_id="s-match"))
        assert app._admin_event_matches(E(1, "session.created", "t", {"session_id": "s-match"}))
        assert not app._admin_event_matches(E(2, "turn.started", "t", {"session_id": "other"}))
        assert not app._admin_event_matches(E(0, "system.heartbeat", "t", {}))
        assert app._admin_event_matches(E(3, "system.events_dropped", "t", {"count": 5}))

    async def test_stream_writes_filtered_events(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """stream_filtered [scenario,tracer]: only active-session + non-heartbeat lines land."""
        import ratatoskr.tui as tui_mod
        from ratatoskr.sse_client import AdminEvent

        self._mute_hydrates(monkeypatch)
        writes = _spy_writes(monkeypatch)

        async def fake_stream(client, *, admin_key, last_event_id=None):
            yield AdminEvent(41, "session.created", "t", {"session_id": "s-ae-2"})
            yield AdminEvent(0, "system.heartbeat", "t", {})  # filtered (noise)
            yield AdminEvent(42, "turn.started", "t", {"session_id": "other"})  # diff session
            yield AdminEvent(43, "session.deleted", "t", {"session_id": "s-ae-2"})

        monkeypatch.setattr(tui_mod, "stream_admin_events", fake_stream)
        app = _resolved_app(_args_existing(session_id="s-ae-2", admin_key="ak"))
        async with app.run_test() as pilot:
            await pilot.pause()
            await app._stream_admin_events()
            await pilot.pause()
        joined = " ".join(_text_of(w) for w in writes)
        assert "session.created" in joined
        assert "session.deleted" in joined
        assert "system.heartbeat" not in joined
        assert "turn.started" not in joined  # different session → filtered

    async def test_stream_no_admin_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """stream_no_admin_key [scenario]: admin_key None → 'not configured' + skip audit."""
        self._mute_hydrates(monkeypatch)
        writes = _spy_writes(monkeypatch)
        app = _resolved_app(_args_existing(session_id="s-ae-3"))  # admin_key None
        async with app.run_test() as pilot:
            await pilot.pause()
            await app._stream_admin_events()
            await pilot.pause()
        joined = " ".join(_text_of(w) for w in writes)
        assert "admin key not configured" in joined
        assert "admin_events_skipped" in joined

    async def test_stream_403_unavailable(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """stream_403 [error]: 403 scope-denied → 'unavailable' + audit; no crash."""
        import ratatoskr.tui as tui_mod
        from ratatoskr.sse_client import SseConnectFailed

        self._mute_hydrates(monkeypatch)
        writes = _spy_writes(monkeypatch)

        async def denied(client, *, admin_key, last_event_id=None):
            raise SseConnectFailed(status=403, body=b"auth_scope_denied")
            yield  # unreachable — makes this an async generator

        monkeypatch.setattr(tui_mod, "stream_admin_events", denied)
        app = _resolved_app(_args_existing(session_id="s-ae-4", admin_key="ak"))
        async with app.run_test() as pilot:
            await pilot.pause()
            await app._stream_admin_events()
            await pilot.pause()
        joined = " ".join(_text_of(w) for w in writes)
        assert "admin events unavailable: HTTP 403" in joined
        assert "admin_events_unavailable" in joined