"""Ratatoskr CLI — non-interactive `--send` stdout presenter.

Implements docs/contracts/issues/3.contract.md.
"""

from __future__ import annotations

import argparse
import asyncio
import hashlib
import os
import signal
import sys
from dataclasses import dataclass, field
from importlib.metadata import PackageNotFoundError, version
from typing import Any, TextIO

import httpx

from ratatoskr.first_message import seed_preset_first_message
from ratatoskr.sessions import (
    AgentNotFound,
    AuthoredHistoryUnavailable,
    BifrostBinding,
    BifrostConsumerKeyMissing,
    BifrostHandshakeFailed,
    SessionApiFailed,
    create_character,
    create_session,
    delete_character,
    endpoint_for_plane,
    get_capabilities,
    get_character_state,
    get_me,
    get_session_messages,
    list_character_models,
    set_persona_state,
    write_authored_history,
)
from ratatoskr.sse_client import (
    AffectUpdate,
    AwaitingLlmFirstToken,
    CancelAlreadyCompleted,
    CancelFailed,
    Cancelled,
    CancelTurnNotFound,
    Done,
    Error,
    Event,
    MalformedSseData,
    MalformedSseId,
    SseConnectFailed,
    SseConnectionDropped,
    Text,
    TextBoundary,
    Thinking,
    ToolResult,
    ToolStart,
    TurnIdFlip,
    WorkerPhase,
    cancel_turn,
    stream_turn_resilient,
)


def _resolve_user_agent() -> str:
    """Compose the User-Agent header — `ratatoskr/<version> (<contact>)`.

    Per worldtree-dev's 2026-05-23 guidance: identify the consumer in outbound
    requests so server-side logs can distinguish ratatoskr traffic. Falls back
    to "0.0.0" if the package isn't installed (test contexts, editable installs
    without metadata).
    """
    try:
        ver = version("ratatoskr")
    except PackageNotFoundError:
        ver = "0.0.0"
    return f"ratatoskr/{ver} (vh@phasefinal.com)"


USER_AGENT = _resolve_user_agent()


class UsageError(Exception):
    """Raised on argument violations; mapped to exit code 10 by main()."""


class _AuthError(Exception):
    """Raised when no API key is resolvable; mapped to exit code 11 by main()."""


@dataclass(frozen=True)
class ParsedArgs:
    """Resolved CLI invocation. Post-validation: exactly one of session_id / new is set."""

    send_content: str | None
    session_id: str | None
    new: bool
    agent_id: str | None
    api_key: str
    server_url: str
    raw: bool
    # Per issue #5: optional `--end-user-id` for per-end-user agents (lofn etc.).
    # Default None preserves the pre-#5 baseline for agents that don't require it (mimir).
    end_user_id: str | None = None
    # Issue #17: optional Bifrost binding (one plane) + its consumer key. None on
    # the unbound pre-#17 path. `bifrost_plane` is the human label for the
    # bound-state indicator (None when --bifrost-url supplies the endpoint directly).
    bifrost: BifrostBinding | None = None
    bifrost_plane: str | None = None
    consumer_key: str | None = None
    # Standalone boot-time orientation probe: GET /me + GET /capabilities, print,
    # exit. Mutually exclusive with the session/turn flags (opens no session).
    whoami: bool = False
    # Optional admin-tier key (RATATOSKR_ADMIN_API_KEY / --admin-key) for the
    # admin-scoped inspection reads (BifrostState pane, GET /admin/sessions/…).
    # None when unset — the BifrostState pane then shows "admin key not configured".
    admin_key: str | None = None
    # Tier-2 one-shot probes (like --whoami). --characters runs the transient-
    # character CRUD lifecycle; --set-persona-pad "p,a,d" (with --session) writes
    # a session's persona state (affect injection).
    characters: bool = False
    set_persona_pad: str | None = None
    # #347 authored-history-write reference-consumer probe: create a fresh
    # session bound to --agent, seed an authored assistant first-message (seq-0).
    seed_first_message: str | None = None


class _ArgparseError(Exception):
    """Internal: re-raise marker so SystemExit from argparse becomes UsageError."""


class _UsageErrorParser(argparse.ArgumentParser):
    """ArgumentParser that raises instead of calling sys.exit on parse errors."""

    def error(self, message: str) -> None:  # type: ignore[override]
        raise _ArgparseError(message)


def _parse_args(argv: list[str] | None) -> ParsedArgs:
    """argparse + env-fallback + xor-validation per the contract."""
    assert argv is None or all(isinstance(a, str) for a in argv)
    parser = _UsageErrorParser(prog="ratatoskr", description="Worldtree CLI presenter.")
    parser.add_argument("--send", default=None)
    parser.add_argument("--session")
    parser.add_argument("--new", action="store_true")
    parser.add_argument("--agent")
    parser.add_argument("--api-key", dest="api_key")
    parser.add_argument("--server")
    parser.add_argument("--raw", action="store_true")
    parser.add_argument("--whoami", action="store_true")
    parser.add_argument("--admin-key", dest="admin_key")
    parser.add_argument("--characters", action="store_true")
    parser.add_argument("--set-persona-pad", dest="set_persona_pad", default=None)
    parser.add_argument("--seed-first-message", dest="seed_first_message", default=None)
    # Issue #5: required for per-end-user agents (lofn etc.); optional otherwise (mimir).
    parser.add_argument("--end-user-id", dest="end_user_id", default=None)
    # Issue #17: bind the created session to our own Bifrost provider plane.
    parser.add_argument(
        "--bifrost-plane", dest="bifrost_plane", choices=("memory", "affect"),
        default=None,
    )
    parser.add_argument("--bifrost-host", dest="bifrost_host", default=None)
    parser.add_argument("--bifrost-url", dest="bifrost_url", default=None)
    try:
        ns = parser.parse_args(argv)
    except _ArgparseError as exc:
        raise UsageError(str(exc)) from exc

    # --send empty-string is still invalid; --send omitted (None) is the TUI-mode marker.
    if ns.send is not None and not ns.send:
        raise UsageError("--send content must be non-empty")
    # Issue #5 INV-001: --end-user-id, if passed, MUST be non-empty (mirrors --send).
    if ns.end_user_id is not None and not ns.end_user_id:
        raise UsageError("--end-user-id must be non-empty when passed")
    if sum([ns.whoami, ns.characters, bool(ns.set_persona_pad), bool(ns.seed_first_message)]) > 1:
        raise UsageError(
            "--whoami / --characters / --set-persona-pad / --seed-first-message "
            "are mutually exclusive"
        )
    if ns.whoami or ns.characters:
        # Standalone one-shot probes: open no session.
        if ns.send is not None or ns.session or ns.new or ns.agent:
            raise UsageError(
                "--whoami / --characters are standalone probes "
                "(no --send/--session/--new/--agent)"
            )
    elif ns.set_persona_pad is not None:
        # Session-scoped write probe: needs a target session, nothing else.
        if not ns.set_persona_pad:
            raise UsageError("--set-persona-pad must be non-empty (e.g. '0.4,0.1,-0.2')")
        if not ns.session:
            raise UsageError("--set-persona-pad requires --session <id>")
        if ns.send is not None or ns.new or ns.agent:
            raise UsageError("--set-persona-pad takes only --session")
    elif ns.seed_first_message is not None:
        # #347 first-message probe: creates a fresh session bound to --agent,
        # then seeds an authored assistant turn as seq-0 — manages its own session.
        if not ns.seed_first_message:
            raise UsageError("--seed-first-message must be non-empty")
        if not ns.agent:
            raise UsageError("--seed-first-message requires --agent <id>")
        if ns.send is not None or ns.session or ns.new:
            raise UsageError(
                "--seed-first-message manages its own session (no --send/--session/--new)"
            )
    else:
        if ns.session and ns.new:
            raise UsageError("--session and --new are mutually exclusive")
        if not ns.session and not ns.new:
            # Bare TUI mode → startup session picker (design-brief §4). --send is
            # non-interactive (no picker can open), so it still requires one flag;
            # --agent belongs with --new (bare mode resumes, it doesn't create).
            if ns.send is not None:
                raise UsageError("--send requires --session or --new (no interactive picker)")
            if ns.agent:
                raise UsageError(
                    "--agent belongs with --new; bare TUI mode opens the session picker"
                )
        if ns.session and ns.agent:
            raise UsageError("--agent is required with --new and forbidden with --session")
        if ns.new and not ns.agent and ns.send is not None:
            # Issue #8: --agent stays required for --send --new (non-interactive,
            # cannot prompt). Bare --new (TUI mode) accepts None — picker drives
            # the choice via list_agents in _resolve_then_run.
            raise UsageError("--agent is required when --new is passed in --send mode")

    api_key = ns.api_key or os.environ.get("WORLDTREE_API_KEY") or ""
    if not api_key:
        raise _AuthError("no API key (set --api-key or WORLDTREE_API_KEY)")

    server_url = ns.server or os.environ.get("WORLDTREE_API_URL") or "http://localhost:8000"
    # Issue #5 (post-2026-05-23 amendment): env-var fallback for end_user_id.
    # Resolution: --end-user-id flag > $RATATOSKR_END_USER_ID > None. env.sh
    # ships the project-stable default "ratatoskr-tui" so a sourced session
    # gets a stable partition without papering over the explicit-flag override.
    end_user_id = ns.end_user_id or os.environ.get("RATATOSKR_END_USER_ID") or None

    # Issue #17: resolve the optional Bifrost binding. --bifrost-url (direct,
    # HTTPS/prod) and --bifrost-plane (dev shortcut → endpoint_for_plane) are
    # mutually exclusive; a binding is a session-CREATE concern (forbidden with
    # --session). The consumer key — the privileged handshake identity, distinct
    # from the canary key — comes from the env (never a CLI flag).
    bifrost: BifrostBinding | None = None
    bifrost_plane: str | None = None
    if ns.bifrost_url and ns.bifrost_plane:
        raise UsageError("--bifrost-url and --bifrost-plane are mutually exclusive")
    if (ns.bifrost_url or ns.bifrost_plane) and not ns.new:
        raise UsageError("a bifrost binding requires --new (it binds at session create)")
    if ns.bifrost_url:
        bifrost = BifrostBinding(endpoint_url=ns.bifrost_url)
    elif ns.bifrost_plane:
        host = ns.bifrost_host or os.environ.get("RATATOSKR_PROVIDER_VISIBLE_HOST")
        if not host:
            raise UsageError(
                "--bifrost-plane requires --bifrost-host "
                "(or RATATOSKR_PROVIDER_VISIBLE_HOST) — the Worldtree-visible provider host"
            )
        bifrost = BifrostBinding(endpoint_url=endpoint_for_plane(ns.bifrost_plane, host))
        bifrost_plane = ns.bifrost_plane
    consumer_key = os.environ.get("RATATOSKR_BIFROST_CONSUMER_KEY") or None
    # Admin-tier key for the admin-scoped inspection reads (BifrostState pane).
    # Flag > env > None; None leaves the admin panes showing "not configured".
    admin_key = ns.admin_key or os.environ.get("RATATOSKR_ADMIN_API_KEY") or None

    return ParsedArgs(
        send_content=ns.send,
        session_id=ns.session,
        new=ns.new,
        agent_id=ns.agent,
        api_key=api_key,
        server_url=server_url,
        raw=ns.raw,
        end_user_id=end_user_id,
        bifrost=bifrost,
        bifrost_plane=bifrost_plane,
        consumer_key=consumer_key,
        whoami=ns.whoami,
        admin_key=admin_key,
        characters=ns.characters,
        set_persona_pad=ns.set_persona_pad,
        seed_first_message=ns.seed_first_message,
    )


# ---- Issue #12 presenter contract semantics amendment -------------------------
#
# CliPresenterState replaces the stateless _render_event with a stateful per-turn
# presenter that coalesces thinking runs and demotes telemetry events. One
# instance per `_amain` call.


def _format_duration_ms(ms: int) -> str:
    """Auto-scale duration formatting per issue #12 INV-006. Locale-blind."""
    assert isinstance(ms, int) and ms >= 0
    if ms < 1000:
        return f"{ms}ms"
    if ms < 60_000:
        return f"{ms / 1000:.1f}s"
    return f"{ms / 60_000:.1f}m"


def _format_usage(usage: dict[str, int], *, arrow: str) -> str:
    """Natural-language usage formatting per issue #12 INV-007.

    `arrow="->"` for CLI (ASCII scriptability); `arrow="→"` for TUI.
    """
    # PRE-001: usage has the four expected keys (issue #12 contract).
    assert all(
        k in usage
        for k in ("prompt_tokens", "completion_tokens", "total_tokens", "cached_input_tokens")
    )
    p = usage["prompt_tokens"]
    c = usage["completion_tokens"]
    t = usage["total_tokens"]
    ci = usage["cached_input_tokens"]
    return f"{p} in {arrow} {c} out ({t} total, {ci} cached)"


@dataclass(slots=True)
class CliPresenterState:
    """Per-turn presenter state for `--send` mode (issue #12).

    See `docs/contracts/issues/12.contract.md` for the full spec.
    """

    thinking_buffer: list[str] = field(default_factory=list)
    thinking_open: bool = False
    text_written_since_newline: bool = False

    def render(self, event: Event, *, stdout: TextIO, stderr: TextIO) -> None:
        """Render one Worldtree SSE event with editorial hierarchy + coalescing."""
        assert isinstance(
            event,
            (
                WorkerPhase, Thinking, Text, TextBoundary,
                ToolStart, ToolResult, Done, Error, Cancelled,
                AffectUpdate, AwaitingLlmFirstToken,
            ),
        )
        # Thinking events accumulate into the open run.
        if isinstance(event, Thinking):
            if not self.thinking_open:
                stderr.write(". thinking: ")
                self.thinking_open = True
            stderr.write(event.content)
            stderr.flush()
            self.thinking_buffer.append(event.content)
            return
        # Non-thinking event: close any open thinking run first.
        if self.thinking_open:
            stderr.write("\n")
            stderr.flush()
            self.thinking_open = False
            self.thinking_buffer.clear()
        # Now render the new event.
        if isinstance(event, Text):
            stdout.write(event.content)
            stdout.flush()
            # POST-003: only set if cursor is mid-line (no trailing newline).
            self.text_written_since_newline = not event.content.endswith("\n")
            return
        if isinstance(event, (Done, Error, Cancelled)):
            # INV-005: ensure stdout newline boundary before stderr terminal label.
            if self.text_written_since_newline:
                stdout.write("\n")
                stdout.flush()
                self.text_written_since_newline = False
        if isinstance(event, Done):
            stderr.write(
                f"[done] turn_id={event.sse_id.turn_id} model={event.model} "
                f"duration={_format_duration_ms(event.duration_ms)} "
                f"usage {_format_usage(event.usage, arrow='->')}\n"
            )
            return
        if isinstance(event, WorkerPhase):
            stderr.write(
                f". worker_phase: phase={event.phase} turn_id={event.turn_id}\n"
            )
            return
        if isinstance(event, Error):
            stderr.write(
                f"[error] turn_id={event.sse_id.turn_id} code={event.error_code} "
                f"message={event.message!r}\n"
            )
            return
        if isinstance(event, Cancelled):
            stderr.write(
                f"[cancelled] turn_id={event.turn_id} reason={event.reason!r} "
                f"partial_message_id={event.partial_message_id}\n"
            )
            return
        if isinstance(event, ToolStart):
            stderr.write(
                f". tool_start: name={event.name} args={event.arguments!r}\n"
            )
            return
        if isinstance(event, ToolResult):
            stderr.write(
                f". tool_result: name={event.name} duration_ms={event.duration_ms} "
                f"result={event.result!r:.200}\n"
            )
            return
        if isinstance(event, TextBoundary):
            stderr.write(
                f". text_boundary: kind={event.kind} char_offset={event.char_offset}\n"
            )
            return
        if isinstance(event, AffectUpdate):
            # Worldtree #204 / v0.28.0. CLI surface is debug telemetry —
            # one line to stderr with status + (for current) dominant_emotion.
            if event.snapshot is not None:
                dom = event.snapshot.get("dominant_emotion")
                stderr.write(
                    f". affect_update: status={event.status} turn_id={event.turn_id} "
                    f"dominant_emotion={dom!r}\n"
                )
            else:
                stderr.write(
                    f". affect_update: status={event.status} turn_id={event.turn_id}\n"
                )
            return
        if isinstance(event, AwaitingLlmFirstToken):
            # Worldtree #201 / v0.29.0. Heartbeat during BuildingPrompt →
            # CallingLLM gap. Stderr surface, one line per heartbeat.
            secs = event.elapsed_ms_since_building_prompt / 1000.0
            stderr.write(
                f". awaiting_llm_first_token: turn_id={event.turn_id} elapsed={secs:.1f}s\n"
            )
            return


async def _cancel_and_log(
    client: httpx.AsyncClient,
    session_id: str,
    turn_id: int,
    *,
    stderr: TextIO,
) -> None:
    """Spawn-and-forget cancel that never raises (INV-009)."""
    assert client is not None
    assert isinstance(turn_id, int) and turn_id > 0
    try:
        await cancel_turn(client, session_id, turn_id)
    except (CancelFailed, CancelTurnNotFound, CancelAlreadyCompleted, httpx.RequestError) as exc:
        stderr.write(f"[cancel_failed] {type(exc).__name__}: {exc}\n")


async def _run_turn(
    client: httpx.AsyncClient,
    session_id: str,
    content: str,
    sigint_event: asyncio.Event,
    *,
    stdout: TextIO,
    stderr: TextIO,
    state: CliPresenterState | None = None,
) -> int:
    """Drive stream_turn, render events, race against sigint_event for mid-stream cancel.

    Per issue #12: a `CliPresenterState` is passed in by `_amain` for stateful
    coalescing + telemetry demotion. Callers that omit `state` get a fresh
    instance (transitional convenience; tests construct their own to inspect).
    """
    assert client is not None
    assert session_id and isinstance(session_id, str)
    assert content and isinstance(content, str)
    assert isinstance(sigint_event, asyncio.Event)
    if state is None:
        state = CliPresenterState()

    last_turn_id: int | None = None
    cancelling = False
    sigint_task: asyncio.Task[bool] | None = None
    cancel_task: asyncio.Task[None] | None = None  # strong ref to fire-and-forget cancel
    aiter_obj = stream_turn_resilient(client, session_id, content).__aiter__()

    try:
        while True:
            next_task = asyncio.create_task(aiter_obj.__anext__())
            if not cancelling:
                sigint_task = asyncio.create_task(sigint_event.wait())
                done, _pending = await asyncio.wait(
                    {next_task, sigint_task}, return_when=asyncio.FIRST_COMPLETED
                )
                if sigint_task in done:
                    if last_turn_id is not None:
                        cancel_task = asyncio.create_task(
                            _cancel_and_log(client, session_id, last_turn_id, stderr=stderr)
                        )
                        cancelling = True
                    else:
                        next_task.cancel()
                        stderr.write("[cancelled] (before any event arrived)\n")
                        return 3
                if next_task not in done:
                    await next_task
            else:
                await next_task
            try:
                event = next_task.result()
            except StopAsyncIteration:
                stderr.write("[connection_dropped] last_seen=<none>\n")
                return 21
            except SseConnectFailed as exc:
                stderr.write(f"[sse_connect_failed] status={exc.status} body={exc.body!r}\n")
                return 20
            except SseConnectionDropped as exc:
                stderr.write(f"[connection_dropped] last_seen={exc.last_seen_sse_id}\n")
                return 21
            except MalformedSseId as exc:
                stderr.write(f"[malformed_sse_id] raw={exc.raw!r}\n")
                return 22
            except MalformedSseData as exc:
                stderr.write(f"[malformed_sse_data] raw={exc.raw!r}\n")
                return 22
            except TurnIdFlip as exc:
                stderr.write(f"[turn_id_flip] expected={exc.established} got={exc.got}\n")
                return 22
            last_turn_id = event.sse_id.turn_id
            state.render(event, stdout=stdout, stderr=stderr)
            if isinstance(event, Done):
                if sigint_task is not None and not cancelling:
                    sigint_task.cancel()
                return 0
            if isinstance(event, Error):
                if sigint_task is not None and not cancelling:
                    sigint_task.cancel()
                return 2
            if isinstance(event, Cancelled):
                if sigint_task is not None and not cancelling:
                    sigint_task.cancel()
                return 3
    finally:
        if sigint_task is not None and not sigint_task.done():
            sigint_task.cancel()
        # Await the fire-and-forget cancel so INV-009 + _cancel_and_log POST-002 are
        # guaranteed: the [cancel_failed] line (if any) lands before _run_turn returns
        # AND before _amain closes the AsyncClient context. _cancel_and_log never raises.
        if cancel_task is not None:
            await cancel_task


async def _amain(args: ParsedArgs) -> int:
    """Async orchestrator: create-session (if --new) → SIGINT install → _run_turn → cleanup."""
    assert isinstance(args, ParsedArgs)
    async with httpx.AsyncClient(
        base_url=args.server_url,
        headers={
            "Authorization": f"Bearer {args.api_key}",
            "User-Agent": USER_AGENT,
        },
        # SSE streaming sits idle between events while the LLM thinks.
        # Default 5s read timeout would kill mid-stream; disable it.
        # connect/write/pool keep modest timeouts so true network failures
        # still surface promptly.
        timeout=httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0),
    ) as client:
        if args.new:
            assert args.agent_id is not None
            try:
                info = await create_session(
                    client,
                    args.agent_id,
                    end_user_id=args.end_user_id,
                    bifrost=args.bifrost,
                    consumer_key=args.consumer_key,
                )
            except AgentNotFound as exc:
                sys.stderr.write(f"[agent_not_found] agent_id={exc.agent_id}\n")
                return 12
            except BifrostConsumerKeyMissing as exc:
                # INV-001: never fall back to the canary key — fail loud.
                sys.stderr.write(
                    f"[bifrost_consumer_key_missing] {exc} "
                    f"(set RATATOSKR_BIFROST_CONSUMER_KEY)\n"
                )
                return 22
            except BifrostHandshakeFailed as exc:
                # INV-002: bind-time handshake failure fails session creation.
                sys.stderr.write(
                    f"[bifrost_handshake_failed] bifrost_error={exc.bifrost_error}\n"
                )
                # 401-message scoping: keyed on auth_rejected, name the key mismatch.
                if exc.bifrost_error == "bifrost.auth_rejected":
                    sys.stderr.write(
                        "  bound create requires the consumer key "
                        "(RATATOSKR_BIFROST_CONSUMER_KEY), not WORLDTREE_API_KEY\n"
                    )
                return 23
            except SessionApiFailed as exc:
                sys.stderr.write(f"[session_api_failed] status={exc.status} body={exc.body!r}\n")
                return 20
            except (httpx.ConnectError, httpx.ReadTimeout, httpx.TransportError) as exc:
                sys.stderr.write(f"[network_error] {type(exc).__name__}: {exc}\n")
                return 21
            # Issue #12: demoted lifecycle line — written directly here (NOT via
            # state.render, which only accepts SSE Event variants per PRE-001).
            sys.stderr.write(
                f". create_session: session_id={info.session_id} agent_id={info.agent_id}\n"
            )
            # #347 authored first-message: seed the agent's preset opening (best-effort).
            if await seed_preset_first_message(client, info.session_id, args.agent_id):
                sys.stderr.write(
                    f". first_message: seeded preset opening for {info.agent_id}\n"
                )
            # Issue #17 bound-state indicator: plane + endpoint + status, so the
            # operator sees WHICH identity/endpoint bound (not a bare boolean).
            if args.bifrost is not None:
                plane = args.bifrost_plane or "direct"
                sys.stderr.write(
                    f". bifrost: status=bound plane={plane} "
                    f"endpoint={args.bifrost.endpoint_url}\n"
                )
            session_id = info.session_id
        else:
            assert args.session_id is not None
            session_id = args.session_id

        sigint_event = asyncio.Event()
        loop = asyncio.get_running_loop()
        loop.add_signal_handler(signal.SIGINT, sigint_event.set)
        state = CliPresenterState()
        try:
            return await _run_turn(
                client,
                session_id,
                args.send_content,
                sigint_event,
                state=state,
                stdout=sys.stdout,
                stderr=sys.stderr,
            )
        finally:
            loop.remove_signal_handler(signal.SIGINT)


def _format_whoami(me: dict[str, Any], caps: dict[str, Any]) -> str:
    """Render the --whoami report: identity (GET /me) + server capabilities."""
    lines = ["identity:"]
    lines.append(f"  user_id: {me.get('user_id', '?')}")
    lines.append(f"  tier: {me.get('tier', '?')}")
    lines.append(f"  scopes: {', '.join(me.get('scopes', [])) or '(none)'}")
    for k in ("display_name", "key_id", "key_label"):
        if k in me:
            lines.append(f"  {k}: {me[k]}")
    lines.append("capabilities:")
    templates = caps.get("ephemeral_templates", {})
    if templates:
        for name, spec in templates.items():
            models = ", ".join(spec.get("allowed_models", []))
            lines.append(
                f"  ephemeral_template {name}: default={spec.get('default_model', '?')} "
                f"max_bytes={spec.get('system_prompt_max_bytes', '?')} models=[{models}]"
            )
    else:
        lines.append("  (no ephemeral templates advertised)")
    return "\n".join(lines) + "\n"


async def _whoami(args: ParsedArgs) -> int:
    """--whoami one-shot: GET /me + GET /capabilities, print a compact report, exit.

    A boot-time orientation probe (mirrors --send's non-interactive shape):
    "who am I against this server, and what does it offer." Opens no session.
    Errors land on stderr with the same [session_api_failed] / [network_error]
    vocab + exit codes as the other modes.
    """
    assert isinstance(args, ParsedArgs)
    async with httpx.AsyncClient(
        base_url=args.server_url,
        headers={"Authorization": f"Bearer {args.api_key}", "User-Agent": USER_AGENT},
        timeout=httpx.Timeout(connect=10.0, read=10.0, write=10.0, pool=10.0),
    ) as client:
        try:
            me = await get_me(client)
            caps = await get_capabilities(client)
        except SessionApiFailed as exc:
            sys.stderr.write(f"[session_api_failed] status={exc.status} body={exc.body!r}\n")
            return 20
        except (httpx.ConnectError, httpx.ReadTimeout, httpx.TransportError) as exc:
            sys.stderr.write(f"[network_error] {type(exc).__name__}: {exc}\n")
            return 21
        sys.stdout.write(_format_whoami(me, caps))
    return 0


def _probe_client(args: ParsedArgs) -> httpx.AsyncClient:
    """AsyncClient for the one-shot probes (--whoami / --characters / --set-persona-pad)."""
    return httpx.AsyncClient(
        base_url=args.server_url,
        headers={"Authorization": f"Bearer {args.api_key}", "User-Agent": USER_AGENT},
        timeout=httpx.Timeout(connect=10.0, read=10.0, write=10.0, pool=10.0),
    )


async def _characters_probe(args: ParsedArgs) -> int:
    """--characters one-shot: exercise the transient-character CRUD lifecycle
    (models → create → get-state → delete), print a report, exit. A reference-
    consumer smoke of the #161 character surface (needs character.read/write)."""
    assert isinstance(args, ParsedArgs)
    async with _probe_client(args) as client:
        try:
            models = await list_character_models(client)
            names = ", ".join(m.get("name", "?") for m in models.get("items", []))
            sys.stdout.write(f"character models: {names or '(none)'}\n")
            created = await create_character(
                client,
                {
                    "schema_version": "1",
                    "name": "ratatoskr-probe",
                    "ocean": {
                        "openness": 0.5, "conscientiousness": 0.5, "extraversion": 0.0,
                        "agreeableness": 0.5, "neuroticism": 0.5,
                    },
                    "description": "ratatoskr --characters lifecycle probe",
                    "narrative": "A throwaway probe character.",
                    "voice_profile_block": "plain",
                },
            )
            cid = created["character_id"]
            sys.stdout.write(f"created: {cid} (ttl {created.get('ttl_expires_at')})\n")
            state = await get_character_state(client, cid)
            sys.stdout.write(f"state: pad={state.get('pad')}\n")
            await delete_character(client, cid)
            sys.stdout.write(f"deleted: {cid}\n")
        except SessionApiFailed as exc:
            sys.stderr.write(f"[session_api_failed] status={exc.status} body={exc.body!r}\n")
            return 20
        except (httpx.ConnectError, httpx.ReadTimeout, httpx.TransportError) as exc:
            sys.stderr.write(f"[network_error] {type(exc).__name__}: {exc}\n")
            return 21
    return 0


async def _set_persona_probe(args: ParsedArgs) -> int:
    """--set-persona-pad one-shot: POST a PAD to /sessions/{id}/persona_state
    (affect injection), print the result, exit. Requires --session."""
    assert isinstance(args, ParsedArgs)
    assert args.session_id is not None and args.set_persona_pad is not None
    try:
        pad = [float(x) for x in args.set_persona_pad.split(",")]
    except ValueError:
        sys.stderr.write(
            "[usage_error] --set-persona-pad must be comma-separated floats "
            "(e.g. '0.4,0.1,-0.2')\n"
        )
        return 10
    if len(pad) != 3:
        sys.stderr.write(
            "[usage_error] --set-persona-pad needs exactly 3 floats "
            "(pleasure,arousal,dominance), e.g. '0.4,0.1,-0.2'\n"
        )
        return 10
    # Canonical POST /sessions/{id}/persona_state body (#317): a named-key dict,
    # NOT a bare list — {"pad": {"pleasure", "arousal", "dominance"}}.
    snapshot = {"pad": {"pleasure": pad[0], "arousal": pad[1], "dominance": pad[2]}}
    async with _probe_client(args) as client:
        try:
            await set_persona_state(client, args.session_id, snapshot)
        except SessionApiFailed as exc:
            sys.stderr.write(f"[session_api_failed] status={exc.status} body={exc.body!r}\n")
            return 20
        except (httpx.ConnectError, httpx.ReadTimeout, httpx.TransportError) as exc:
            sys.stderr.write(f"[network_error] {type(exc).__name__}: {exc}\n")
            return 21
        sys.stdout.write(
            f"persona_state set: session={args.session_id[-8:]} pad={pad} (204)\n"
        )
    return 0


async def _seed_first_message_probe(args: ParsedArgs) -> int:
    """--seed-first-message one-shot: create a fresh session bound to --agent,
    write an authored assistant first-message (#347 POST /sessions/{id}/history),
    read it back via GET /messages, print a report, exit. A reference-consumer
    smoke of the authored-history-write primitive.

    Hide-existence: a 404 (feature-absent OR the key lacks `session.history.write`)
    is reported as a benign 'feature-absent' result (exit 0) — the probe NEVER
    capability-probes to distinguish the causes (server INV-347-1). The probe
    seeds but does not generate, so the assistant-first provider constraint is
    inert here.
    """
    assert isinstance(args, ParsedArgs)
    assert args.agent_id is not None and args.seed_first_message is not None
    async with _probe_client(args) as client:
        try:
            session = await create_session(
                client, args.agent_id, end_user_id=args.end_user_id
            )
            sys.stdout.write(f"session: {session.session_id} (agent {session.agent_id})\n")
            key = "ratatoskr-first-message-" + hashlib.sha256(
                args.seed_first_message.encode("utf-8")
            ).hexdigest()[:12]
            try:
                ack = await write_authored_history(
                    client,
                    session.session_id,
                    content=args.seed_first_message,
                    idempotency_key=key,
                )
            except AuthoredHistoryUnavailable:
                sys.stdout.write(
                    "authored-history: feature-absent or ungranted (404 hide-existence) "
                    "— a production consumer falls back to a model-generated greeting; "
                    "no capability-probe attempted.\n"
                )
                return 0
            sys.stdout.write(
                f"seeded: seq={ack.get('seq')} phase={ack.get('phase')} "
                f"turn_id={ack.get('turn_id')} content_chars={ack.get('content_chars')}\n"
            )
            history = await get_session_messages(client, session.session_id)
            items = history.get("items", [])
            sys.stdout.write(f"read-back: {len(items)} message(s)\n")
            for m in items:
                sys.stdout.write(
                    f"  seq={m.get('seq')} role={m.get('role')} content={m.get('content')!r}\n"
                )
        except SessionApiFailed as exc:
            sys.stderr.write(f"[session_api_failed] status={exc.status} body={exc.body!r}\n")
            return 20
        except (httpx.ConnectError, httpx.ReadTimeout, httpx.TransportError) as exc:
            sys.stderr.write(f"[network_error] {type(exc).__name__}: {exc}\n")
            return 21
    return 0


def main(argv: list[str] | None = None) -> int:
    """Sync entry point. Maps UsageError/_AuthError to exit codes BEFORE the event loop."""
    assert argv is None or all(isinstance(a, str) for a in argv)
    try:
        args = _parse_args(argv)
    except UsageError as exc:
        sys.stderr.write(f"[usage_error] {exc}\n")
        return 10
    except _AuthError as exc:
        sys.stderr.write(f"[auth_error] {exc}\n")
        return 11
    except SystemExit as exc:
        # argparse's --help / --version short-circuit via SystemExit(0). Pass the code
        # through verbatim — argparse already printed help to stdout.
        return int(exc.code) if exc.code is not None else 0
    if args.whoami:
        return asyncio.run(_whoami(args))
    if args.characters:
        return asyncio.run(_characters_probe(args))
    if args.set_persona_pad is not None:
        return asyncio.run(_set_persona_probe(args))
    if args.seed_first_message is not None:
        return asyncio.run(_seed_first_message_probe(args))
    if args.send_content is None:
        # TUI mode — lazy import preserves INV-001 (no textual in cli at module scope).
        from ratatoskr.tui import run_tui

        return run_tui(args)
    return asyncio.run(_amain(args))
