"""Session management commands for CleverAgents CLI.

The ``agents session`` command group manages interactive sessions — persistent
conversation threads tied to orchestrator actors.

Session operations are routed through the :class:`A2aLocalFacade` to comply
with the A2A protocol data flow::

    CLI ──▸ A2aLocalFacade.dispatch() ──▸ SessionService ──▸ Domain

Based on specification.md ``agents session`` section and implementation_plan.md
task A7.cli.
"""

from __future__ import annotations

import json
import logging
import re
import threading
from collections import OrderedDict
from pathlib import Path
from typing import Annotated, Any, cast

import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table

from cleveragents.a2a.models import A2aRequest
from cleveragents.application.container import get_container
from cleveragents.application.services.session_workflow import SessionWorkflow
from cleveragents.application.services.strategy_resolution import (
    build_actor_resolver,
)
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.core.exceptions import DatabaseError
from cleveragents.domain.models.core.session import (
    MessageRole,
    Session,
    SessionActorNotConfiguredError,
    SessionExportError,
    SessionImportError,
    SessionMessage,
    SessionNotFoundError,
    SessionService,
)
from cleveragents.providers.registry import ProviderRegistry

# Create sub-app for session commands
app = typer.Typer(help="Manage interactive sessions.")
console = Console()

_log = logging.getLogger(__name__)

# Reusable --format option description
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"

# MCP logger name constant — used in create() and list_sessions() to suppress
# MCP health-check output during structured (non-Rich) CLI output.
_MCP_LOGGER_NAME = "cleveragents.mcp"

# Thread lock for MCP logger level mutations to prevent race conditions when
# multiple CLI commands execute concurrently (e.g., in parallel test runners).
_mcp_logger_lock = threading.Lock()

# ---------------------------------------------------------------------------
# Module-level service and workflow accessors (patchable in tests)
# ---------------------------------------------------------------------------
_service: SessionService | None = None


def _get_session_service() -> SessionService:
    """Get or create the SessionService instance.

    Production usage goes through the DI container's ``session_service``
    provider; tests can patch ``_service`` or this function directly.
    """
    global _service
    if _service is not None:
        return _service

    container = get_container()
    svc = cast(SessionService, container.session_service())
    _service = svc
    return _service


def _reset_session_service() -> None:
    """Reset the module-level service (used by tests)."""
    global _service
    _service = None


def _build_session_workflow() -> SessionWorkflow:
    """Build a :class:`SessionWorkflow` wired to the CLI session service.

    Returns a fully configured workflow using the same ``SessionService``
    that the CLI commands use.  Tests can replace this function or its
    dependencies by patching ``_service`` (for the service layer) and
    ``_get_provider_registry`` (for the LLM layer).
    """
    service = _get_session_service()
    provider_registry = _get_provider_registry()
    actor_resolver = _build_actor_resolver()
    actor_options_resolver = _build_actor_options_resolver()
    return SessionWorkflow(
        session_service=service,
        provider_registry=provider_registry,
        actor_resolver=actor_resolver,
        actor_options_resolver=actor_options_resolver,
    )


def _get_provider_registry() -> ProviderRegistry | None:
    """Attempt to load the provider registry; return ``None`` on failure.

    Isolated in its own function so tests can patch it independently
    to avoid network / credential errors in CI.
    """
    try:
        from cleveragents.providers.registry import get_provider_registry

        return get_provider_registry()
    except (ImportError, RuntimeError, OSError) as exc:
        _log.warning("Provider registry unavailable: %s", exc)
        return None


def _build_actor_resolver():
    """Build a resolver callable for namespace/name -> provider/model resolution.

    Returns a callable ``(actor_name: str) -> str | None`` that looks up
    a namespace/name actor reference (e.g. ``"local/my-strategist"``) in
    the actor registry and returns the ``"provider/model"`` string, or
    ``None`` when the name is already in provider/model format, the actor
    is unknown, or the registry is unavailable.

    When no DI container is configured the function returns a resolver
    that always returns ``None`` (graceful degradation).
    """
    try:
        container = get_container()
        actor_service = container.actor_service()
        if actor_service is None:
            return _null_actor_resolver

        return build_actor_resolver(actor_service)
    except Exception:
        _log.warning(
            "actor_resolver_unavailable",
            exc_info=True,
        )
        return _null_actor_resolver


def _null_actor_resolver(_actor_name: str) -> None:
    """Null-object resolver — always returns ``None``."""
    return None


def _null_actor_options_resolver(_actor_name: str) -> None:
    """Null-object options resolver — always returns ``None``."""
    return None


def _build_actor_options_resolver():
    """Build a resolver callable for namespace/name -> actor options dict.

    Returns a callable ``(actor_name: str) -> dict | None`` that looks up
    a namespace/name actor reference (e.g. ``"local/my-strategist"``) in
    the actor registry and returns the actor's ``options`` dict from its
    config blob, or ``None`` when the name is in provider/model format,
    the actor is unknown, or the registry is unavailable.
    """
    try:
        container = get_container()
        actor_service = container.actor_service()
        if actor_service is None:
            return _null_actor_options_resolver

        from cleveragents.application.services.strategy_resolution import (
            build_actor_options_resolver,
        )

        return build_actor_options_resolver(actor_service)
    except Exception:
        _log.warning(
            "actor_options_resolver_unavailable",
            exc_info=True,
        )
        return _null_actor_options_resolver


def _facade_dispatch(operation: str, params: dict[str, Any]) -> dict[str, Any]:
    """Route an operation through the A2A local facade.

    Returns the response data dict on success.  Domain errors are mapped
    back to their original exception types so CLI error handlers work
    correctly.  Unknown / infrastructure errors raise ``RuntimeError``.

    stdout/stderr are redirected during facade construction to prevent
    structlog or Rich output from polluting CLI output captured by
    test runners.
    """
    import contextlib
    import io

    with (
        contextlib.redirect_stdout(io.StringIO()),
        contextlib.redirect_stderr(io.StringIO()),
    ):
        from cleveragents.a2a.cli_bootstrap import get_facade

        facade = get_facade()
        # Wire the CLI's session service and workflow into the facade
        # so that message/send and message/stream handlers use the same
        # SessionService and SessionWorkflow as the rest of the CLI.
        # Tests that patch _get_session_service() and
        # _build_session_workflow() therefore also control the facade.
        with contextlib.suppress(Exception):
            facade.register_service("session_service", _get_session_service())
        with contextlib.suppress(Exception):
            facade.register_service("session_workflow", _build_session_workflow())
        request = A2aRequest(method=operation, params=params)
        response = facade.dispatch(request)
    if response.error is not None:
        # Map A2A error codes back to domain exceptions so CLI error
        # handlers work correctly when routing through the facade.
        # Domain exceptions that handlers explicitly re-raise
        # (SessionNotFoundError, SessionActorNotConfiguredError,
        # DatabaseError) propagate directly through the facade
        # dispatch, so they never reach this point.  Any error here
        # is an unknown/infrastructure failure.
        raise RuntimeError(response.error.message)
    return dict(response.result or {})


# ---------------------------------------------------------------------------
# Helpers — dict builders and rich printing
# ---------------------------------------------------------------------------


def _resolve_actor_details(actor_name: str | None) -> OrderedDict[str, Any] | None:
    """Return spec-compliant actor details when an actor binding exists."""

    if not actor_name:
        return None

    actor = None
    try:
        container = get_container()
        registry = container.actor_registry()
        actor = registry.get_actor(actor_name)
    except Exception:  # pragma: no cover - defensive: missing registry or actor
        actor = None

    if actor is None:
        provider, _, model = actor_name.partition("/")
        fallback = OrderedDict()
        if provider:
            fallback["provider"] = provider
        if model:
            fallback["model"] = model
        elif provider:
            fallback["model"] = provider
        if fallback:
            return fallback
        return None  # pragma: no cover - degenerate actor_name

    details: OrderedDict[str, Any] = OrderedDict()
    details["provider"] = actor.provider
    details["model"] = actor.model

    config_blob = actor.config_blob if isinstance(actor.config_blob, dict) else {}
    options = config_blob.get("options")

    temperature: Any | None = None
    if isinstance(options, dict):
        temperature = options.get("temperature")
    if temperature is None:
        temperature = getattr(actor, "temperature", None)
    if temperature is not None:
        details["temperature"] = temperature

    context_window: Any | None = None
    if "context_window" in config_blob:
        context_window = config_blob.get("context_window")
    else:
        graph_descriptor = config_blob.get("graph_descriptor")
        if isinstance(graph_descriptor, dict):
            context_window = graph_descriptor.get("context_window")
    if context_window is not None:
        details["context_window"] = context_window

    return details


def _session_create_payload(session: Session) -> OrderedDict[str, Any]:
    """Build the spec-required payload for session create output."""

    session_block: OrderedDict[str, Any] = OrderedDict()
    session_block["id"] = session.session_id
    session_block["actor"] = session.actor_name or None
    session_block["created"] = session.created_at.isoformat()
    session_block["namespace"] = session.namespace

    settings_block: OrderedDict[str, Any] = OrderedDict(
        automation="review",
        streaming="off",
        context="default",
        memory="enabled",
        max_history=50,
    )

    payload: OrderedDict[str, Any] = OrderedDict()
    payload["session"] = session_block
    payload["settings"] = settings_block

    actor_details = _resolve_actor_details(session.actor_name)
    if actor_details is not None:
        payload["actor_details"] = actor_details

    return payload


def _build_session_create_command(actor: str | None, fmt: str | None) -> str:
    """Construct the command string used in JSON/YAML envelopes."""

    parts: list[str] = ["agents session create"]
    if actor:
        parts.append(f"--actor {actor}")
    if fmt:
        parts.append(f"--format {fmt}")
    return " ".join(parts)


def _session_list_dict(sessions: list[Session]) -> dict[str, Any]:
    """Build the list output with summary stats per spec."""
    items = []
    for s in sessions:
        items.append(
            {
                "id": s.session_id,
                "name": s.name or None,
                "actor": s.actor_name or "(none)",
                "messages": s.message_count,
                "updated": s.updated_at.isoformat(),
            }
        )

    # Build summary section per spec
    total_messages = sum(s.message_count for s in sessions)

    # Find most recent and oldest sessions
    if sessions:
        sorted_sessions = sorted(sessions, key=lambda x: x.updated_at, reverse=True)
        most_recent = sorted_sessions[0].name or sorted_sessions[0].session_id
        oldest = sorted_sessions[-1].name or sorted_sessions[-1].session_id
    else:
        most_recent = None
        oldest = None

    summary = {
        "total": len(sessions),
        "most_recent": most_recent,
        "oldest": oldest,
        "total_messages": total_messages,
        "storage": "0 KB",  # Placeholder - actual storage calculation not implemented
    }

    return {
        "sessions": items,
        "summary": summary,
    }


# ---------------------------------------------------------------------------
# CLI Commands
# ---------------------------------------------------------------------------


@app.command()
def create(
    actor: Annotated[
        str | None,
        typer.Option("--actor", help="Orchestrator actor name (namespace/name)"),
    ] = None,
    fmt: Annotated[
        str,
        typer.Option("--format", "-f", help=_FORMAT_HELP),
    ] = "rich",
) -> None:
    """Create a new interactive session.

    Optionally bind the session to an orchestrator actor via ``--actor``.

    Examples:
        agents session create
        agents session create --actor openai/gpt-4
        agents session create --format json
    """
    # Suppress MCP daemon logger during JSON/YAML output to prevent health check
    # messages from interfering with structured output.
    mcp_logger = logging.getLogger(_MCP_LOGGER_NAME)
    orig_level = mcp_logger.level
    if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
        with _mcp_logger_lock:
            mcp_logger.setLevel(logging.CRITICAL)

    try:
        # Create session through the service, then notify the A2A
        # facade for protocol compliance and telemetry.
        service = _get_session_service()
        session = service.create(actor_name=actor)

        # Notify the facade layer for A2A protocol bookkeeping.
        # Pass session_id so the facade handler acknowledges the already-
        # persisted session instead of creating a duplicate (#1141).
        try:
            _facade_dispatch(
                "session.create",
                {"actor_name": actor or "", "session_id": session.session_id},
            )
        except Exception as _exc:
            _log.warning(
                "session_create_facade_dispatch_failed",
                extra={"session_id": session.session_id, "error": str(_exc)},
                exc_info=True,
            )

        if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
            payload = _session_create_payload(session)
            command_string = _build_session_create_command(actor, fmt)
            output = format_output(
                payload,
                fmt,
                command=command_string,
                messages=[{"level": "ok", "text": "Session created"}],
            )
            if output:
                typer.echo(output)
            return

        details = (
            f"[bold]Session ID:[/bold] {session.session_id}\n"
            f"[bold]Actor:[/bold] {session.actor_name or '(none)'}\n"
            f"[bold]Namespace:[/bold] {session.namespace}\n"
            f"[bold]Created:[/bold] {session.created_at.strftime('%Y-%m-%d %H:%M')}"
        )
        console.print(Panel(details, title="Session", expand=False))

        # Settings panel
        settings_text = (
            "[yellow]Automation:[/yellow] default\n"
            "[yellow]Streaming:[/yellow] off\n"
            "[yellow]Context:[/yellow] default\n"
            "[yellow]Memory:[/yellow] enabled\n"
            "[yellow]Max History:[/yellow] 50 turns"
        )
        console.print(Panel(settings_text, title="Settings", expand=False))

        # Actor Details panel (if actor is bound)
        if session.actor_name:
            try:
                container = get_container()
                registry = container.actor_registry()
                actor_obj = registry.get_actor(session.actor_name)
                actor_details = (
                    f"[blue]Provider:[/blue] {actor_obj.provider}\n"
                    f"[blue]Model:[/blue] {actor_obj.model}\n"
                    f"[blue]Temperature:[/blue] "
                    f"{getattr(actor_obj, 'temperature', 0.7)}\n"
                    "[blue]Context Window:[/blue] 200K tokens"
                )
                console.print(Panel(actor_details, title="Actor Details", expand=False))
            except Exception:
                pass  # Actor details unavailable

        console.print("[green]✓ OK[/green] Session created")

    except SessionNotFoundError as exc:
        console.print(f"[red]Error:[/red] {exc}")
        raise typer.Exit(1) from exc
    except DatabaseError as exc:
        _log.debug("session create failed", exc_info=True)
        console.print(
            f"[red]Error:[/red] Database unavailable: {exc}\n"
            "Hint: run 'agents init' to initialise the database."
        )
        raise typer.Exit(1) from exc
    finally:
        # Restore original MCP logger level
        with _mcp_logger_lock:
            mcp_logger.setLevel(orig_level)


@app.command("list")
def list_sessions(
    fmt: Annotated[
        str,
        typer.Option("--format", "-f", help=_FORMAT_HELP),
    ] = "rich",
) -> None:
    """List all sessions.

    Displays session IDs, bound actors, message counts, and last update times.

    Examples:
        agents session list
        agents session list --format json
        agents session list --format table
    """
    # Suppress MCP daemon logger during JSON/YAML output to prevent health check
    # messages from interfering with structured output.
    mcp_logger = logging.getLogger(_MCP_LOGGER_NAME)
    orig_level = mcp_logger.level
    if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
        with _mcp_logger_lock:
            mcp_logger.setLevel(logging.CRITICAL)

    try:
        service = _get_session_service()
        sessions = service.list()
    except DatabaseError as exc:
        _log.debug("session list failed", exc_info=True)
        console.print(
            f"[red]Error:[/red] Database unavailable: {exc}\n"
            "Hint: run 'agents init' to initialise the database."
        )
        raise typer.Exit(1) from exc
    finally:
        # Restore original MCP logger level
        with _mcp_logger_lock:
            mcp_logger.setLevel(orig_level)

    if not sessions:
        # For machine-readable formats, always emit a structured empty list
        # so that callers parsing JSON/YAML receive valid output.
        if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
            typer.echo(
                format_output(
                    {"sessions": [], "total": 0},
                    fmt,
                    messages=[{"level": "ok", "text": "0 sessions listed"}],
                )
            )
            return
        console.print("[yellow]No sessions found.[/yellow]")
        console.print("Create one with 'agents session create'")
        return

    data = _session_list_dict(sessions)

    if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
        typer.echo(
            format_output(
                data,
                fmt,
                messages=[{"level": "ok", "text": f"{len(sessions)} sessions listed"}],
            )
        )
        return

    # Rich table
    table = Table(title="Sessions", show_header=True, border_style="blue")
    table.add_column("ID", style="cyan")
    table.add_column("Name", style="magenta")
    table.add_column("Actor", style="blue")
    table.add_column("Messages", justify="right")
    table.add_column("Updated", style="green")

    for s in sessions:
        table.add_row(
            s.session_id,  # Full ULID for copy-paste compatibility with session tell
            s.name or "(unnamed)",
            s.actor_name or "(none)",
            str(s.message_count),
            s.updated_at.strftime("%Y-%m-%d %H:%M"),
        )

    console.print(table)
    console.print()

    # Reuse the summary dict already computed by _session_list_dict to avoid
    # duplicating the total_msgs / sorted_sessions / most_recent / oldest logic.
    summary = data["summary"]

    summary_table = Table.grid(padding=(0, 1))
    summary_table.add_column(style="cyan bold", justify="left")
    summary_table.add_column(style="white", justify="left")
    summary_table.add_row("Total:", str(summary["total"]))
    summary_table.add_row("Most Recent:", summary["most_recent"] or "")
    summary_table.add_row("Oldest:", summary["oldest"] or "")
    summary_table.add_row("Total Messages:", str(summary["total_messages"]))
    summary_table.add_row("Storage:", summary["storage"])

    console.print(Panel(summary_table, title="Summary", border_style="blue"))
    console.print()
    console.print(f"[green]✓ OK[/green] {len(sessions)} sessions listed")


@app.command()
def show(
    session_id: Annotated[
        str,
        typer.Argument(help="Session ULID to show"),
    ],
    fmt: Annotated[
        str,
        typer.Option("--format", "-f", help=_FORMAT_HELP),
    ] = "rich",
) -> None:
    """Show session details and recent messages.

    Displays session metadata, recent messages, linked plans, and token usage.

    Examples:
        agents session show 01HXYZ...
        agents session show 01HXYZ... --format json
    """
    try:
        service = _get_session_service()
        session = service.get(session_id)
        data = session.as_cli_dict()

        if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
            typer.echo(
                format_output(
                    dict(data),
                    fmt,
                    messages=[{"level": "ok", "text": "Session details loaded"}],
                )
            )
            return

        # Session summary panel — field order per spec: ID, Actor, Messages,
        # Created, Updated, Automation (docs/specification.md §agents session show)
        details = (
            f"[bold]ID:[/bold] {session.session_id}\n"
            f"[bold]Actor:[/bold] {session.actor_name or '(none)'}\n"
            f"[bold]Messages:[/bold] {session.message_count}\n"
            f"[bold]Created:[/bold] {session.created_at.strftime('%Y-%m-%d %H:%M')}\n"
            f"[bold]Updated:[/bold] {session.updated_at.strftime('%Y-%m-%d %H:%M')}\n"
            f"[bold]Automation:[/bold] {session.automation or '(none)'}"
        )
        console.print(Panel(details, title="Session Summary", expand=False))

        # Recent messages
        if session.messages:
            recent = session.messages[-5:]
            msg_table = Table(title="Recent Messages", show_header=True)
            msg_table.add_column("Role", style="cyan")
            msg_table.add_column("Text")
            for msg in recent:
                text = msg.content
                if len(text) > 80:
                    text = text[:77] + "..."
                msg_table.add_row(msg.role.value, text)
            console.print(msg_table)

        # Linked plans — spec requires Plan ID / Phase / State columns
        if session.linked_plans:
            plan_table = Table(title="Linked Plans", show_header=True)
            plan_table.add_column("Plan ID", style="cyan")
            plan_table.add_column("Phase")
            plan_table.add_column("State")
            for lp in session.linked_plans:
                plan_table.add_row(lp.plan_id, lp.phase, lp.state)
            console.print(Panel(plan_table, title="Linked Plans", expand=False))
        elif session.linked_plan_ids:
            # Fallback: only flat IDs available
            plan_text = "\n".join(f"  • {pid}" for pid in session.linked_plan_ids)
            console.print(Panel(plan_text, title="Linked Plans", expand=False))

        # Token usage
        tu = session.token_usage
        usage_text = (
            f"[blue]Input Tokens:[/blue] {tu.input_tokens:,}\n"
            f"[blue]Output Tokens:[/blue] {tu.output_tokens:,}\n"
            f"[yellow]Estimated Cost:[/yellow] ${tu.estimated_cost:.4f}"
        )
        console.print(Panel(usage_text, title="Token Usage", expand=False))

        # Cost budget (per-session budget cap, #584)
        if session.cost_budget is not None:
            cb = session.cost_budget
            util = cb.utilization()
            util_display = f"{util * 100:.1f}%" if util is not None else "N/A"
            remaining = cb.remaining()
            remaining_display = (
                f"${remaining:.4f}" if remaining is not None else "unlimited"
            )
            max_display = (
                f"${cb.max_cost_usd:.4f}"
                if cb.max_cost_usd is not None
                else "unlimited"
            )
            budget_text = (
                f"[blue]Total Cost:[/blue] ${cb.total_cost:.4f}\n"
                f"[blue]Max Cost:[/blue] {max_display}\n"
                f"[yellow]Utilization:[/yellow] {util_display}\n"
                f"[green]Remaining:[/green] {remaining_display}"
            )
            console.print(Panel(budget_text, title="Cost Budget", expand=False))

        console.print("[green bold]✓ OK[/green bold] Session details loaded")

    except SessionNotFoundError as exc:
        console.print(f"[red]Session not found:[/red] {session_id}")
        raise typer.Exit(1) from exc
    except DatabaseError as exc:
        _log.debug("session show failed", exc_info=True)
        console.print(
            f"[red]Error:[/red] Database unavailable: {exc}\n"
            "Hint: run 'agents init' to initialise the database."
        )
        raise typer.Exit(1) from exc


@app.command()
def delete(
    session_id: Annotated[
        str,
        typer.Argument(help="Session ULID to delete"),
    ],
    yes: Annotated[
        bool,
        typer.Option("--yes", "-y", help="Skip confirmation prompt"),
    ] = False,
    fmt: Annotated[
        OutputFormat,
        typer.Option("--format", help=_FORMAT_HELP),
    ] = OutputFormat.RICH,
) -> None:
    """Delete a session permanently.

    A confirmation prompt is shown unless ``--yes`` is provided.

    Examples:
        agents session delete 01HXYZ...
        agents session delete 01HXYZ... --yes
    """
    try:
        service = _get_session_service()

        # Verify session exists before prompting and capture message count
        session_to_delete = service.get(session_id)
        message_count = session_to_delete.message_count

        if not yes:
            confirm = typer.confirm(f"Delete session {session_id}?", default=False)
            if not confirm:
                console.print("[yellow]Aborted.[/yellow]")
                raise typer.Abort()

        service.delete(session_id)

        # Rich output: render Deletion Summary and Cleanup panels
        if fmt == OutputFormat.RICH:
            # Deletion Summary panel
            summary_table = Table.grid(padding=(0, 1))
            summary_table.add_column(style="cyan bold", justify="left")
            summary_table.add_column(style="white", justify="left")
            summary_table.add_row("Session:", session_id)
            summary_table.add_row("ID:", session_id)
            summary_table.add_row("Messages:", f"{message_count} removed")
            summary_table.add_row("Storage:", "0 KB freed")
            summary_table.add_row("Plans Orphaned:", "0")

            console.print(
                Panel(summary_table, title="Deletion Summary", border_style="blue")
            )
            console.print()

            # Cleanup panel
            cleanup_table = Table.grid(padding=(0, 1))
            cleanup_table.add_column(style="cyan bold", justify="left")
            cleanup_table.add_column(style="white", justify="left")
            cleanup_table.add_row("Backups:", "none")
            cleanup_table.add_row("Logs:", "preserved")
            cleanup_table.add_row("Context:", "cleared")
            cleanup_table.add_row("Checkpoints:", "none")

            console.print(Panel(cleanup_table, title="Cleanup", border_style="blue"))
            console.print()
            console.print("[green]✓ OK[/green] Session deleted")
        elif fmt == OutputFormat.COLOR:
            # Color format: human-readable Rich-styled line, no envelope.
            console.print(f"[green]✓ OK[/green] Session {session_id} deleted")
        else:
            # Machine-readable formats (json/yaml/plain): emit a structured
            # envelope so callers can parse the success message reliably.
            typer.echo(
                format_output(
                    {
                        "session_id": session_id,
                        "messages_removed": message_count,
                    },
                    fmt.value,
                    command=f"agents session delete {session_id}",
                    messages=[{"level": "ok", "text": "Session deleted"}],
                )
            )

    except SessionNotFoundError as exc:
        console.print(f"[red]Session not found:[/red] {session_id}")
        raise typer.Exit(1) from exc
    except DatabaseError as exc:
        _log.debug("session delete failed", exc_info=True)
        console.print(
            f"[red]Error:[/red] Database unavailable: {exc}\n"
            "Hint: run 'agents init' to initialise the database."
        )
        raise typer.Exit(1) from exc


@app.command("export")
def export_session(
    session_id: Annotated[
        str,
        typer.Argument(help="Session ULID to export"),
    ],
    output: Annotated[
        Path | None,
        typer.Option("--output", "-o", help="Output file path (default: stdout)"),
    ] = None,
    force: Annotated[
        bool,
        typer.Option("--force", help="Overwrite existing output file"),
    ] = False,
    fmt: Annotated[
        str,
        typer.Option(
            "--format",
            help="Export format: json (default) or md (Markdown transcript)",
        ),
    ] = "json",
    output_format: Annotated[
        str | None,
        typer.Option(
            "--output-format",
            help=(
                "CLI output presentation: rich (default), json, yaml, or plain. "
                "Machine-readable formats emit a structured envelope and "
                "suppress Rich panels."
            ),
        ),
    ] = None,
) -> None:
    """Export a session as JSON or Markdown.

    Writes the full session data to a file or stdout.  The default format is
    JSON (canonical, importable).  Use ``--format md`` for a human-readable
    Markdown transcript (lossy — cannot be re-imported).

    On success, three Rich panels are displayed: "Session Export" (session ID,
    output path, message count, file size, format), "Contents" (messages, plan
    references, metadata keys, actor config, schema version), and "Integrity"
    (checksum, encrypted flag).

    Examples:
        agents session export 01HXYZ...
        agents session export 01HXYZ... -o session.json
        agents session export 01HXYZ... -o session.json --force
        agents session export 01HXYZ... --format md -o session.md
        agents session export 01HXYZ... --output-format json
    """
    if fmt not in ("json", "md"):
        console.print(f"[red]Invalid format:[/red] {fmt!r}. Use 'json' or 'md'.")
        raise typer.Exit(1)
    structured_output = output_format in ("json", "yaml", "plain")

    try:
        service = _get_session_service()
        json_data: dict[str, Any] = {}

        if fmt == "md":
            # Markdown export: load full session with messages
            session = service.get(session_id)
            # Load messages via export_session to populate session.messages
            json_data = service.export_session(session_id)
            messages = [
                SessionMessage(
                    message_id=m["message_id"],
                    role=MessageRole(m["role"]),
                    content=m["content"],
                    sequence=m["sequence"],
                    timestamp=m["timestamp"],
                    metadata=m.get("metadata", {}),
                    tool_call_id=m.get("tool_call_id"),
                )
                for m in json_data.get("messages", [])
            ]
            session.messages = messages
            content = session.as_export_markdown()
        else:
            json_data = service.export_session(session_id)
            content = json.dumps(json_data, indent=2, default=str)

        if output is not None:
            if output.exists() and not force:
                console.print(
                    f"[red]File already exists:[/red] {output}\n"
                    "Use --force to overwrite."
                )
                raise typer.Exit(1)
            # Create parent directories if needed
            output.parent.mkdir(parents=True, exist_ok=True)
            output.write_text(content, encoding="utf-8")
        elif not structured_output:
            # Structured output formats suppress the raw content emission so
            # the envelope is the only thing on stdout (and remains valid JSON).
            typer.echo(content)

        if structured_output:
            assert output_format is not None  # for type-checker
            envelope_data: dict[str, Any] = {
                "session_id": session_id,
                "output": str(output) if output is not None else None,
                "format": fmt,
                "messages_exported": len(json_data.get("messages", [])),
                "schema_version": json_data.get("schema_version", "v1"),
            }
            typer.echo(
                format_output(
                    envelope_data,
                    output_format,
                    command=f"agents session export {session_id}",
                    messages=[{"level": "ok", "text": "Export completed"}],
                )
            )
        else:
            # Render Rich panels for both file and stdout export paths
            _render_export_panels(
                session_id=session_id,
                output=output,
                content=content,
                export_data=json_data,
                fmt=fmt,
            )

    except SessionNotFoundError as exc:
        console.print(f"[red]Session not found:[/red] {session_id}")
        raise typer.Exit(1) from exc
    except SessionExportError as exc:
        console.print(f"[red]Export error:[/red] {exc}")
        raise typer.Exit(1) from exc
    except DatabaseError as exc:
        _log.debug("session export failed", exc_info=True)
        console.print(
            f"[red]Error:[/red] Database unavailable: {exc}\n"
            "Hint: run 'agents init' to initialise the database."
        )
        raise typer.Exit(1) from exc


def _render_export_panels(
    *,
    session_id: str,
    output: Path | None,
    content: str,
    export_data: dict[str, Any],
    fmt: str,
) -> None:
    """Render the three spec-required Rich panels for ``agents session export``.

    Displays:
    - "Session Export" panel: session ID, output path, message count, size, format
    - "Contents" panel: messages, plan references, metadata keys, actor config,
      schema version
    - "Integrity" panel: checksum, encrypted flag
    - ``✓ OK Export completed`` success line
    """
    # Compute file size from content
    size_bytes = len(content.encode("utf-8"))
    if size_bytes < 1024:
        size_str = f"{size_bytes} B"
    elif size_bytes < 1024 * 1024:
        size_str = f"{size_bytes // 1024} KB"
    else:
        size_str = f"{size_bytes // (1024 * 1024)} MB"

    # Derive display values from export data (JSON format) or defaults (md)
    message_count = len(export_data.get("messages", []))
    plan_refs = len(export_data.get("linked_plan_ids", []))
    metadata_keys = len(export_data.get("metadata", {}))
    actor_config = "included" if export_data.get("actor_name") else "none"
    schema_version = export_data.get("schema_version", "v1")
    checksum_raw = export_data.get("checksum", "")
    checksum_display = (
        f"sha256:{checksum_raw[:4]}...{checksum_raw[-4:]}"
        if len(checksum_raw) >= 8
        else checksum_raw or "n/a"
    )
    output_display = str(output) if output is not None else "(stdout)"
    format_display = "JSON" if fmt == "json" else "Markdown"

    # Session Export panel
    export_table = Table.grid(padding=(0, 1))
    export_table.add_column(style="cyan bold", justify="left")
    export_table.add_column(style="white", justify="left")
    export_table.add_row("Session:", session_id)
    export_table.add_row("Output:", output_display)
    export_table.add_row("Messages:", str(message_count))
    export_table.add_row("Size:", size_str)
    export_table.add_row("Format:", format_display)
    console.print(Panel(export_table, title="Session Export", border_style="blue"))
    console.print()

    # Contents panel
    contents_table = Table.grid(padding=(0, 1))
    contents_table.add_column(style="blue bold", justify="left")
    contents_table.add_column(style="white", justify="left")
    contents_table.add_row("Messages:", str(message_count))
    contents_table.add_row("Plan References:", str(plan_refs))
    contents_table.add_row("Metadata Keys:", str(metadata_keys))
    contents_table.add_row("Actor Config:", actor_config)
    contents_table.add_row("Schema Version:", str(schema_version))
    console.print(Panel(contents_table, title="Contents", border_style="blue"))
    console.print()

    # Integrity panel
    integrity_table = Table.grid(padding=(0, 1))
    integrity_table.add_column(style="magenta bold", justify="left")
    integrity_table.add_column(style="white", justify="left")
    integrity_table.add_row("Checksum:", checksum_display)
    integrity_table.add_row("Encrypted:", "no")
    console.print(Panel(integrity_table, title="Integrity", border_style="blue"))
    console.print()

    console.print("[green]✓ OK[/green] Export completed")


@app.command("import")
def import_session(
    input_file: Annotated[
        Path,
        typer.Option("--input", "-i", help="Input JSON file path"),
    ],
    fmt: Annotated[
        str,
        typer.Option("--format", "-f", help=_FORMAT_HELP),
    ] = "rich",
) -> None:
    """Import a session from a JSON file.

    The file must have been produced by ``agents session export`` and must
    contain a valid schema version and checksum.

    Examples:
        agents session import -i session.json
        agents session import -i session.json --format json
    """
    if not input_file.exists():
        console.print(f"[red]File not found:[/red] {input_file}")
        raise typer.Exit(1)

    try:
        raw = input_file.read_text(encoding="utf-8")
        data = json.loads(raw)
    except json.JSONDecodeError as exc:
        console.print(f"[red]Invalid JSON:[/red] {exc}")
        raise typer.Exit(1) from exc

    try:
        service = _get_session_service()
        schema_version = data.get("schema_version", "unknown")
        actor_name = data.get("actor_name")
        session = service.import_session(data)

        if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
            envelope_data: dict[str, Any] = {
                "session_id": session.session_id,
                "input": str(input_file),
                "message_count": session.message_count,
                "schema_version": schema_version,
                "actor_ref": "resolved" if actor_name else "none",
            }
            typer.echo(
                format_output(
                    envelope_data,
                    fmt,
                    command=f"agents session import --input {input_file}",
                    messages=[{"level": "ok", "text": "Import completed"}],
                )
            )
            return

        # Session Import panel
        session_details = (
            f"[bold]Input:[/bold] {input_file}\n"
            f"[bold]Session ID:[/bold] {session.session_id}\n"
            f"[bold]Messages:[/bold] {session.message_count}\n"
            f"[bold]Schema:[/bold] {schema_version}"
        )
        console.print(Panel(session_details, title="Session Import", expand=False))

        # Validation panel
        actor_ref_status = "resolved" if actor_name else "none"
        validation_details = (
            f"[bold]Checksum:[/bold] verified\n"
            f"[bold]Schema:[/bold] compatible\n"
            f"[bold]Actor Ref:[/bold] {actor_ref_status}"
        )
        console.print(Panel(validation_details, title="Validation", expand=False))

        # Merge panel
        merge_details = "[bold]Existing:[/bold] none\n[bold]Strategy:[/bold] create new"
        console.print(Panel(merge_details, title="Merge", expand=False))

        console.print("[green]✓ OK[/green] Import completed")

    except SessionImportError as exc:
        console.print(f"[red]Import error:[/red] {exc}")
        raise typer.Exit(1) from exc
    except DatabaseError as exc:
        _log.debug("session import failed", exc_info=True)
        console.print(
            f"[red]Error:[/red] Database unavailable: {exc}\n"
            "Hint: run 'agents init' to initialise the database."
        )
        raise typer.Exit(1) from exc


@app.command()
def tell(
    prompt: Annotated[
        str,
        typer.Argument(help="Message to send to the session"),
    ],
    session_id: Annotated[
        str,
        typer.Option("--session", help="Session ULID to send message to"),
    ],
    actor: Annotated[
        str | None,
        typer.Option("--actor", help="Override actor for this message"),
    ] = None,
    stream: Annotated[
        bool,
        typer.Option("--stream", help="Stream response in real-time"),
    ] = False,
    fmt: Annotated[
        OutputFormat,
        typer.Option(
            "--format",
            "-f",
            help=_FORMAT_HELP,
        ),
    ] = OutputFormat.RICH,
) -> None:
    """Send a message to a session.

    Appends a user message and invokes the session's bound orchestrator actor
    (or ``--actor`` override) via the A2A ``message/send`` operation.
    The actor's real response is persisted and token usage is tracked.

    Examples:
        agents session tell --session 01HXYZ... "Hello, world"
        agents session tell --session 01HXYZ... --actor openai/gpt-4 "Plan a feature"
        agents session tell --session 01HXYZ... --stream "Build tests"
        agents session tell --session 01HXYZ... --format json "Hello"
    """
    try:
        _do_session_tell(
            session_id=session_id,
            prompt=prompt,
            actor_override=actor,
            stream=stream,
            fmt=fmt,
        )
    except SessionActorNotConfiguredError as exc:
        console.print(f"[red]Error:[/red] {exc}")
        raise typer.Exit(1) from exc
    except SessionNotFoundError as exc:
        console.print(f"[red]Session not found:[/red] {session_id}")
        raise typer.Exit(1) from exc
    except DatabaseError as exc:
        _log.debug("session tell failed", exc_info=True)
        console.print(
            f"[red]Error:[/red] Database unavailable: {exc}\n"
            "Hint: run 'agents init' to initialise the database."
        )
        raise typer.Exit(1) from exc
    except Exception as exc:
        if not isinstance(exc, typer.Exit):
            _log.exception("session tell failed with unexpected error")
            console.print(
                "[red]Error:[/red] An unexpected error occurred. "
                "Check the logs for details."
            )
            raise typer.Exit(1) from exc
        raise


def _do_session_tell(
    session_id: str,
    prompt: str,
    actor_override: str | None,
    stream: bool,
    fmt: OutputFormat,
) -> None:
    """Core implementation of ``session tell``, extracted for testability.

    Both streaming and non-streaming paths route through
    :class:`~cleveragents.a2a.facade.A2aLocalFacade` per the spec's
    A2A protocol mapping (message/send, message/stream; addresses C3, C5).
    """
    # Validate --actor format before passing to the workflow (m8).
    if actor_override is not None and not re.match(
        r"^[a-z0-9][a-z0-9_-]*/[a-z0-9][a-z0-9_-]*$",
        actor_override,
    ):
        console.print(
            f"[red]Error:[/red] Invalid actor name {actor_override!r}. "
            "Expected format: 'namespace/name' "
            "(e.g. 'openai/gpt-4')."
        )
        raise typer.Exit(1)

    # Route through A2aLocalFacade per the spec's A2A protocol mapping.
    # message/send  → SessionWorkflow.tell() (non-streaming)
    # message/stream → falls back to non-streaming with streamed=false
    #   (true SSE-based streaming is deferred; acceptable per the spec).
    operation = "message/stream" if stream else "message/send"
    result_dict = _facade_dispatch(
        operation,
        {
            "session_id": session_id,
            "message": prompt,
            "actor": actor_override,
        },
    )

    assistant_content = result_dict.get("assistant_message", "")
    usage = result_dict.get("usage", {})

    if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
        data: dict[str, object] = {
            "session_id": session_id,
            "user_message": prompt,
            "assistant_message": assistant_content,
            "usage": usage,
        }
        typer.echo(format_output(data, fmt))
        return

    from rich.markup import escape

    console.print(f"[dim]user:[/dim] {escape(prompt)}")
    console.print(f"[cyan]assistant:[/cyan] {escape(assistant_content)}")
    _print_usage_panel(
        input_tokens=int(usage.get("input_tokens", 0)),
        output_tokens=int(usage.get("output_tokens", 0)),
        cost=float(usage.get("cost_usd", 0.0)),
        duration_ms=float(usage.get("duration_ms", 0.0)),
        tool_calls=int(usage.get("tool_calls", 0)),
    )


def _print_usage_panel(
    input_tokens: int,
    output_tokens: int,
    cost: float,
    duration_ms: float,
    tool_calls: int,
) -> None:
    """Render a Rich Usage panel summarising token and cost metrics."""
    lines = [
        f"[bold]Input tokens:[/bold]  {input_tokens}",
        f"[bold]Output tokens:[/bold] {output_tokens}",
        f"[bold]Est. cost:[/bold]     ${cost:.6f}",
        f"[bold]Duration:[/bold]      {duration_ms / 1000:.1f}s",
        f"[bold]Tool calls:[/bold]    {tool_calls}",
    ]
    console.print(Panel("\n".join(lines), title="Usage", expand=False))
