"""slash.exec helpers: live-session command output + side-effect mirroring after a worker slash command.

Bodies are rebound onto server.py's globals at install time (see
method_ctx.bind_module), so they reference server.py globals bare.
"""

from __future__ import annotations

import contextlib

from .method_ctx import HandlerRegistry, bind_module

_registry = HandlerRegistry()


# ── Live-session slash output ────────────────────────────────────────

# Answered from the live session ONLY when the agent lives on a compute host.
_ISOLATED_SESSION_READ_COMMANDS = frozenset({"context", "tools", "help"})

# /context on a LOCAL session is answered in-process: the slash worker is a CLI
# subprocess whose ``self.agent`` is built lazily by the first chat prompt and
# never by a slash command, so its /context answers "(._.) No active agent" for
# any local session, active or idle (#81251 / #93280). With a live agent we render
# the full Cursor-style breakdown (the desktop gauge's engine); without one we
# fall back to the persisted-state view. ``tools``/``help`` keep the old routing:
# ``_format_live_tools_output`` reads ``session["agent"]``, so serving it here
# for an agentless session would answer worse, not better.
_CONTEXT_LOCAL_IN_PROCESS = frozenset({"context"})

_NO_AGENT_USAGE = "(._.) No active agent -- send a message first."
_NO_AGENT = "No active agent -- send a message first."


def _format_live_review_output(sid: str, session: Optional[dict], arg: str) -> str:
    """Dispatch /review against the live session's agent. The reviewer runs on the async
    delegation rail; its completion is stamped with the parent's durable session_id, which
    ``_session_owns_notification_event`` matches to drain it back into this chat."""
    if session is None:
        return "Nothing to review yet — send a message first."
    if _session_uses_compute_host(session):
        return "/review runs on the local agent only for now — this session's agent lives on a remote compute host."
    if (agent := session.get("agent")) is None:
        return "Nothing to review yet — send a message first."
    if session.get("running"):
        return "session busy — wait for the current turn to finish, then /review"
    with session.get("history_lock") or contextlib.nullcontext():
        snapshot = list(session.get("history", []))
    snapshot = snapshot or list(getattr(agent, "_session_messages", None) or [])
    # slash.exec runs on the RPC pool, not inside a turn: bind the same session identity a turn binds
    # (HERMES_UI_SESSION_ID + steer authority), or delegate_task registers the reviewer with no owner
    # and `subagent.list` hides it — the Desktop status stack then shows nothing for /review.
    tokens = _set_session_context(session["session_key"], ui_session_id=sid)
    runtime_token = _current_runtime_session_record.set(session)
    try:
        from agent.review_engine import format_dispatch_note, start_review
        # slash.exec is off-turn (RPC pool). start_review → resolve_runtime_provider
        # reads HERMES_CODEX_BASE_URL via get_secret; under multiplex that raises
        # UnscopedSecretError unless the same runtime scope a turn binds is here
        # (#117544; same wrap as _compress_live_with_feedback / #116611).
        with _session_profile_runtime_scope(session):
            result = start_review(agent, snapshot, arg or "")
    except ValueError as exc:
        return str(exc)
    except Exception as exc:
        return f"/review failed to start: {exc}"
    finally:
        _current_runtime_session_record.reset(runtime_token)
        _clear_session_context(tokens)
    return format_dispatch_note(result, arg or "")


def _format_live_refine_output(sid: str, session: Optional[dict], arg: str) -> str:
    """Dispatch /refine against the live session's agent: spawn the background
    memory/skills review directly instead of forking the isolated slash worker."""
    if session is None:
        return "Refine unavailable (no session)."
    if session.get("running"):
        return "Agent is running — wait for the turn to finish, then /refine."
    if _session_uses_compute_host(session):
        command = "/refine" + (f" {arg}" if arg.strip() else "")
        try:
            ack = _send_compute_host_control(
                sid,
                route_name="slash.refine",
                command=command,
                wait=True,
            )
        except Exception as exc:
            return f"compute-host slash.refine failed: {exc}"
        if ack.get("type") in {"control.error", "error"}:
            return str(ack.get("message") or "compute-host slash.refine failed")
        _apply_compute_host_metadata_mirror(session, ack)
        return str(ack.get("output") or "")
    agent = session.get("agent")
    if agent is None:
        return "Nothing to refine yet — send a message first."

    snapshot = []
    transcript_ref = str(session.get("session_key") or "")
    if transcript_ref:
        try:
            with _session_db(session) as db:
                if db is not None:
                    snapshot = db.get_messages_as_conversation(
                        transcript_ref, include_ancestors=True
                    )
        except Exception:
            logger.debug("failed to load persisted /refine transcript", exc_info=True)
    if not snapshot:
        with session["history_lock"]:
            snapshot = list(session.get("history", []))
    if not snapshot:
        return "Nothing to refine yet — the conversation is empty."

    review_skills = "skill_manage" in getattr(agent, "valid_tool_names", set())
    try:
        agent._spawn_background_review(
            messages_snapshot=list(snapshot),
            review_memory=True,
            review_skills=review_skills,
            focus=arg.strip() or None,
            explicit=True,
        )
    except Exception as exc:
        return f"/refine failed to start: {exc}"
    tail = f" (focus: {arg.strip()})" if arg.strip() else ""
    return (
        f"⚗ Reviewing this conversation in the background{tail} — "
        "any memory/skill updates will be reported when done."
    )


def _format_live_usage_output(sid: str, session: dict, arg: str) -> str:
    agent = session.get("agent")
    usage = _session_usage_snapshot(session)
    if agent is None and not usage:
        return _NO_AGENT_USAGE
    if session.get("_metadata_message_count") is not None:
        message_count = int(session.get("_metadata_message_count") or 0)
    else:
        with session["history_lock"]:
            message_count = len(session.get("history", []))

    def n(key: str) -> str:
        return f"{int(usage.get(key) or 0):,}"
    rows = [("Input tokens:", n("input")), ("Output tokens:", n("output"))]
    if int(usage.get("reasoning") or 0):
        rows.append(("Reasoning tokens:", n("reasoning")))
    rows += [("Prompt tokens:", n("prompt")), ("Completion tokens:", n("completion")),
             ("Total tokens:", n("total")), ("API calls:", n("calls"))]
    if usage.get("context_max"):
        pct = int(usage.get("context_percent") or 0)
        mark = "~" if usage.get("context_estimated") else ""
        rows.append(("Current context:", f"{mark}{n('context_used')} / {n('context_max')} ({mark}{pct}%)"))
    rows += [("Messages:", f"{message_count:,}"), ("Compressions:", n("compressions"))]
    model = usage.get("model") or _metadata_mirror(session).get("model") or getattr(agent, "model", "") or "(unknown)"
    lines = ["Session Token Usage", "────────────────────────────────────────", f"Model: {model}"]
    return "\n".join(lines + [f"{label:<30}{value}" for label, value in rows])


def _live_session_messages(session: dict) -> Optional[list]:
    """Session-scoped transcript read; None when no db/key or the read fails. Uses
    ``_session_db`` (not ``_get_db()``): a profile session's rows live in its own
    profile's state.db, and through the launch handle this read comes back empty."""
    with _session_db(session) as db:
        if db is not None and session.get("session_key"):
            with contextlib.suppress(Exception):
                return db.get_messages_as_conversation(
                    session["session_key"], include_ancestors=True, include_row_ids=True)
    return None


def _format_live_history_output(sid: str, session: dict, arg: str) -> str:
    with session["history_lock"]:
        history = list(session.get("history", []))
    db_history = _live_session_messages(session)
    messages = _history_to_messages(history if db_history is None else db_history, profile_home=session.get("profile_home"))
    if not messages:
        return "No conversation history yet."
    lines = ["Conversation History", "────────────────────────────────────────"]
    for idx, message in enumerate(messages, start=1):
        role = str(message.get("role") or "unknown")
        label = {"user": "You", "assistant": "Hermes"}.get(role, role.title())
        text = str(message.get("text") or message.get("context") or "").strip()
        text = f"{text[:400]}..." if len(text) > 400 else text
        lines.append(f"[{label} #{idx}] {text or '(no text)'}")
    return "\n".join(lines)


def _format_live_context_output(sid: str, session: dict, arg: str) -> str:
    from collections import Counter
    try:
        messages = _history_to_messages(_live_session_messages(session) or [], profile_home=session.get("profile_home"))
    except Exception:
        messages = []  # malformed db rows fall back to the live history below
    if not messages:
        with session["history_lock"]:
            messages = _history_to_messages(list(session.get("history", [])), profile_home=session.get("profile_home"))

    breakdown_lines: list[str] = []
    if (agent := session.get("agent")) is not None:
        # Live agent: the desktop gauge's engine (agent.context_breakdown) —
        # full Cursor-style breakdown, provider-anchored, no provider calls.
        # `all` expands per-skill / per-toolset costs, matching the CLI.
        with session["history_lock"]:
            live_messages = list(session.get("history", []))
        try:
            # Bind the session context: on the RPC thread the session cwd is unset, so the
            # prompt build inside would key its workspace pin on the backend's cwd.
            tokens = _set_session_context(session["session_key"], cwd=_session_cwd(session))
            try:
                from agent.context_breakdown import (
                    compute_context_details,
                    compute_session_context_breakdown,
                    render_context_breakdown_lines)
                payload = compute_session_context_breakdown(agent, live_messages)
                details = None
                if (arg or "").strip().lower() in {"all", "full", "details"}:
                    with contextlib.suppress(Exception):
                        details = compute_context_details(agent)
                breakdown_lines = render_context_breakdown_lines(payload, details=details, grid=False)
            finally:
                _clear_session_context(tokens)
        except Exception:
            breakdown_lines = []  # fall back to the persisted-state view below

    usage = _session_usage_snapshot(session)
    mirror = _metadata_mirror(session)
    lines = [f"Conversation: {len(messages)} messages" if messages else "Conversation is empty (no messages yet)."]
    roles = Counter(str(msg.get("role") or "unknown") for msg in messages)
    lines.append("  " + ", ".join(f"{r}: {roles.get(r, 0)}" for r in ("user", "assistant", "tool", "system")))
    if model := mirror.get("model") or usage.get("model") or "":
        lines.append(f"Model: {model}")
    lines.append(f"Provider: {mirror.get('provider') or 'auto'}")
    if breakdown_lines:
        lines.extend(["", *breakdown_lines])
        return "\n".join(lines)
    context_used = int(usage.get("context_used") or 0)
    mark = "~" if usage.get("context_estimated") else ""
    context_max = int(usage.get("context_max") or 0)
    if context_used and context_max:
        lines.append(
            f"Context usage: {mark}{context_used:,} / {context_max:,} tokens ({mark}{(context_used / context_max) * 100:.1f}%)")
    elif context_used:
        lines.append(f"Context usage: {mark}{context_used:,} tokens")
    if usage.get("compressions"):
        lines.append(f"Compressions: {int(usage.get('compressions') or 0):,}")
    if (agent := session.get("agent")) is not None:
        from agent.context_file_sources import context_file_sources_for_agent, render_context_file_lines
        # RPC thread: bind the session cwd or the discovery walk keys on the backend's cwd, not the workspace.
        tokens = _set_session_context(session["session_key"], cwd=_session_cwd(session))
        try:
            file_lines = render_context_file_lines(context_file_sources_for_agent(agent))
        finally:
            _clear_session_context(tokens)
        if file_lines:
            lines += [""] + file_lines
    return "\n".join(lines)


def _format_live_tools_output(sid: str, session: dict, arg: str) -> str:
    info = _session_info(session.get("agent"), session)
    groups = info.get("tools") if isinstance(info, dict) else {}
    if not isinstance(groups, dict) or not groups:
        return "No tools available."
    names = sorted({str(n) for g in groups.values() if isinstance(g, list) for n in g})
    if not names:
        return "No tools available."
    return "Available tools ({}):\n{}".format(len(names), "\n".join(f"  {name}" for name in names))


def _format_live_help_output(sid: str, session: dict, arg: str) -> str:
    try:
        from hermes_cli.commands import COMMANDS_BY_CATEGORY
        lines = ["Available commands:", ""]
        for category, commands in COMMANDS_BY_CATEGORY.items():
            lines.append(f"{category}:")
            lines.extend(f"  {cmd:<15} {desc}" for cmd, desc in commands.items())
        return "\n".join(lines)
    except Exception as exc:
        return f"help unavailable: {exc}"


def _format_live_model_output(session: dict) -> str:
    agent = session.get("agent")
    model = getattr(agent, "model", "") if agent is not None else ""
    provider = getattr(agent, "provider", "") if agent is not None else ""
    if not model:
        return "Current model: (unknown)"
    return f"Current model: {model}" + (f" ({provider})" if provider else "")


def _format_live_status_output(sid: str, session: dict, arg: str) -> str:
    response = _methods["session.status"]("status", {"session_id": sid})
    if response.get("error"):
        return str(response["error"].get("message") or "status unavailable")
    return str(response.get("result", {}).get("output") or "")


# name → (reply when there is no session, formatter(sid, session, arg) or a fixed reply).
# A None no-session reply means the formatter handles a missing session itself.
_LIVE_SLASH_OUTPUT = {
    "compress": ("no active session for /compress",
                 lambda sid, session, arg: _mirror_slash_side_effects(sid, session, f"/compress {arg}".strip())),
    "usage": (_NO_AGENT_USAGE, _format_live_usage_output),
    "review": (None, _format_live_review_output),
    "history": ("No conversation history yet.", _format_live_history_output),
    "refine": (None, _format_live_refine_output),
    "status": (None, _format_live_status_output),
    "context": ("Conversation is empty (no messages yet).", _format_live_context_output),
    "tools": ("No tools available.", _format_live_tools_output),
    "help": (None, _format_live_help_output),
    "clear": (None, "Screen clear is terminal-only; desktop/TUI chat left unchanged."),
    "models": (None, "Use /model to view or switch the current model; desktop users can also open the model picker."),
    "rename": (None, "Use /title <name> to rename this session."),
    "effort": (None, "Use /reasoning <effort> to change reasoning effort.")}


def _live_slash_command_output(sid: str, session: Optional[dict], name: str, arg: str) -> Optional[str]:
    """Answer a slash command from the live session instead of the slash worker; None = not ours."""
    name = (name or "").lstrip("/").lower()
    arg = arg or ""
    if name == "model" and not arg.strip():
        return _format_live_model_output(session or {})
    if (
        name in _CONTEXT_LOCAL_IN_PROCESS
        and session is not None
        and not _session_uses_compute_host(session)
    ):
        return _format_live_context_output(sid, session, arg)
    if name in _ISOLATED_SESSION_READ_COMMANDS and not (session is not None and _session_uses_compute_host(session)):
        return None
    entry = _LIVE_SLASH_OUTPUT.get(name)
    if entry is None:
        return None
    no_session_reply, fmt = entry
    if session is None and no_session_reply is not None:
        return no_session_reply
    return fmt(sid, session, arg) if callable(fmt) else fmt


# ── Side-effect mirroring ────────────────────────────────────────────

# Read-then-mutate live agent/session state that a running turn is using; rejected
# while running (parity with session.compress / session.undo and the gateway's
# running-agent /model guard).
_MUTATES_WHILE_RUNNING = frozenset({"model", "personality", "compress"})


def _compress_live_with_feedback(sid: str, session: dict, agent, arg: str, *, snapshot_kwargs: bool) -> str:
    """Compress the live session; return the user-facing feedback text (shared by command.dispatch
    /compress and the slash mirror). ``snapshot_kwargs`` forwards the pre-read snapshot to
    ``_compress_session_history``; the raw arg goes through unparsed (the choke point parses
    ``here [N]`` / ``--keep N``). CompressionLockHeld is a clean no-op (skip note returned);
    other errors propagate to the caller, which finalizes the context-engine notification."""
    from agent.conversation_compression import finalize_context_engine_compression_notification
    from agent.conversation_compression_manual import (
        AGGRESSIVE_UNSUPPORTED, compress_now, parse_compress_args, render_compress_result)
    from agent.manual_compression_feedback import describe_compression_lock_skip, summarize_manual_compression
    from agent.model_metadata import estimate_request_tokens_rough
    with _session_profile_runtime_scope(session):
        with session["history_lock"]:
            before_messages = list(session.get("history", []))
            history_version = int(session.get("history_version", 0))
        request = parse_compress_args(arg)
        if request.aggressive:
            return AGGRESSIVE_UNSUPPORTED
        if request.preview:  # report only — history, agent and session key untouched
            return "\n".join(render_compress_result(compress_now(agent, before_messages, request)))
        sys_prompt = getattr(agent, "_cached_system_prompt", "") or ""
        tools = getattr(agent, "tools", None) or None

        def estimate(messages, prompt, tool_defs) -> int:
            return estimate_request_tokens_rough(messages, system_prompt=prompt, tools=tool_defs) if messages else 0
        before_tokens = estimate(before_messages, sys_prompt, tools)
        snapshot = {"approx_tokens": before_tokens, "before_messages": before_messages, "history_version": history_version}
        try:
            if snapshot_kwargs:
                _compress_session_history(session, arg.strip() or None, **snapshot)
            else:
                # The raw argument goes through unparsed: _compress_session_history (the choke point shared by
                # all three manual-compress routes) parses the boundary-aware forms (here [N], up to here,
                # --keep N) and does the partial head/tail split there (#35533).
                _compress_session_history(session, arg)
        except CompressionLockHeld as e:
            return describe_compression_lock_skip(e.holder)
        _sync_session_key_after_compress(sid, session)
        with session["history_lock"]:
            after_messages = list(session.get("history", []))
        after_tokens = estimate(
            after_messages, getattr(agent, "_cached_system_prompt", "") or sys_prompt, getattr(agent, "tools", None) or tools)
        _emit("session.info", sid, _session_info(agent, session))
        fb = summarize_manual_compression(
            before_messages, after_messages, before_tokens, after_tokens,
            compression_state=getattr(agent, "context_compressor", None))
        finalize_context_engine_compression_notification(agent, committed=True)
        return "\n".join(filter(None, [fb["headline"], fb["token_line"], fb.get("note")]))


def _mirror_approvals(sid, session, agent, arg) -> None:
    if arg:  # the worker already persisted approvals.mode; the bare read-only form needs no repaint
        broadcast_session_info()


def _mirror_personality(sid, session, agent, arg) -> None:
    if arg and agent:
        pname, new_prompt = _validate_personality(arg, _load_cfg())
        from hermes_cli.personality import persist_personality  # single owner: no surface drift
        persist_personality(pname)
        _apply_personality_to_session(sid, session, new_prompt, pname)


_FAST_TIERS = {"fast": "priority", "on": "priority", "normal": None, "off": None, "auto": "auto", "cold": "cold",
               "ultrafast": "ultrafast"}


def _mirror_fast(sid, session, agent, arg) -> None:
    if agent:
        if arg.lower() in _FAST_TIERS:
            agent.service_tier = _FAST_TIERS[arg.lower()]
        _emit("session.info", sid, _session_info(agent, session))


def _mirror_reload_mcp(sid, session, agent, arg) -> None:
    if agent and hasattr(agent, "reload_mcp_tools"):
        agent.reload_mcp_tools()


def _mirror_stop(sid, session, agent, arg) -> None:
    from tools.process_registry import process_registry
    # Deliberate user stop: an explicit source keeps it reaching
    # persist_on_release jobs (#41225).
    process_registry.kill_all(source="slash.stop")


# name → mirror(sid, session, agent, arg); a falsy return means "no warning".
_SLASH_MIRRORS = {
    "model": lambda sid, session, agent, arg: (
        _apply_model_switch(sid, session, arg).get("warning", "") if arg and agent else ""),
    "approvals": _mirror_approvals, "personality": _mirror_personality,
    "compress": lambda sid, session, agent, arg: (
        _compress_live_with_feedback(sid, session, agent, arg, snapshot_kwargs=False) if agent else ""),
    "fast": _mirror_fast,
    "reload-mcp": _mirror_reload_mcp, "stop": _mirror_stop}


def _compute_host_slash(sid: str, session: dict, name: str, command: str) -> tuple[str, str]:
    """Forward a mutating slash command to the session's compute host → ``(status, text)``:
    ``pending`` (compress still running after the wait), ``failed`` (transport error/timeout),
    ``rejected`` (host control.error), ``ok`` (host output; metadata mirror applied). Compress
    waits longer and installs a late-ack adopter so a slow compression still lands here."""
    route_name = f"slash.{name}"
    is_compress = name == "compress"

    def _on_late_ack(late: dict, _sid=sid) -> None:
        _adopt_late_compute_host_compress_ack(_sid, session, late, route_name=route_name)
    try:
        ack = _send_compute_host_control(
            sid, route_name=route_name, command=command, wait=True,
            **({"timeout": _compute_host_compress_wait_seconds(), "on_late_ack": _on_late_ack} if is_compress else {}))
    except queue.Empty:
        if is_compress:
            return "pending", "compression still running in the background; the transcript will refresh when it finishes"
        return "failed", f"compute-host {route_name} failed: timed out"
    except Exception as exc:
        return "failed", f"compute-host {route_name} failed: {exc}"
    if ack.get("type") in {"control.error", "error"}:
        return "rejected", str(ack.get("message") or f"compute-host {route_name} failed")
    _apply_compute_host_metadata_mirror(session, ack)
    return "ok", str(ack.get("output") or "")


def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str:
    """Apply side effects that must also hit the gateway's live agent."""
    parts = command.lstrip("/").split(None, 1)
    if not parts:
        return ""
    name, arg, agent = parts[0], (parts[1].strip() if len(parts) > 1 else ""), session.get("agent")
    if name == "compact":  # /compact aliases /compress; the compute-host control forwards the raw alias
        name = "compress"
    if name in _MUTATES_WHILE_RUNNING:
        if _session_uses_compute_host(session):
            return _compute_host_slash(sid, session, name, command)[1]
        if session.get("running"):
            return busy_message(name)
    if (mirror := _SLASH_MIRRORS.get(name)) is None:
        return ""
    try:
        # Mirrors run OFF-turn (slash.exec RPC pool / compute-host control reader): bind the session's
        # profile scope or /model's credential read raises UnscopedSecretError under multiplex (#122655).
        with _session_profile_runtime_scope(session):
            return mirror(sid, session, agent, arg) or ""
    except Exception as e:
        if name == "compress" and agent:
            from agent.conversation_compression import finalize_context_engine_compression_notification
            finalize_context_engine_compression_notification(agent, committed=False)
        return f"live session sync failed: {e}"


def register(server) -> None:
    """Publish this module's helpers + handlers onto ``server``, rebound to its globals."""
    bind_module(globals(), server, skip=("_",))