"""Model switching for a live session: persist, snapshot/restore runtime, /model apply with
guards, bot-capability + config sync. Bodies are rebound onto server.py's globals at install
time (method_ctx.bind_module), so they reference server.py globals bare."""

from __future__ import annotations

import contextlib
import copy

from .method_ctx import HandlerRegistry, bind_module

_registry = HandlerRegistry()


_RUNTIME_KEYS = ("model", "provider", "api_key", "base_url", "api_mode")


def _snapshot_agent_model_runtime(agent) -> dict:
    """Capture the current agent model runtime for a one-turn restore."""
    return {**{k: getattr(agent, k, "") for k in _RUNTIME_KEYS},
            "reasoning_config": copy.deepcopy(getattr(agent, "reasoning_config", None)),
            "primary_runtime": copy.deepcopy(getattr(agent, "_primary_runtime", None))}


def _restore_agent_model_runtime(agent, snapshot: dict | None) -> None:
    """Restore an agent model runtime captured before a one-turn override."""
    if not snapshot or agent is None:
        return
    # `/model X --reasoning high --once`: the effort leaves with the model. Set before the
    # runtime restore paths below (primary_runtime may predate a session /reasoning change).
    if "reasoning_config" in snapshot:
        agent.reasoning_config = snapshot["reasoning_config"]
    primary = snapshot.get("primary_runtime")
    if primary and hasattr(agent, "_restore_primary_runtime"):
        try:
            agent._primary_runtime = copy.deepcopy(primary)
            agent._fallback_activated = True
            agent._rate_limited_until = 0
            if agent._restore_primary_runtime():
                if "reasoning_config" in snapshot:
                    agent.reasoning_config = snapshot["reasoning_config"]
                return
        except Exception:
            logger.debug("TUI one-turn model restore via primary runtime failed", exc_info=True)
    if hasattr(agent, "switch_model"):
        model, provider, api_key, base_url, api_mode = (snapshot.get(k, "") for k in _RUNTIME_KEYS)
        agent.switch_model(
            new_model=model, new_provider=provider, api_key=api_key, base_url=base_url,
            api_mode=api_mode, capabilities=snapshot.get("capabilities"))
        if "reasoning_config" in snapshot:
            agent.reasoning_config = snapshot["reasoning_config"]


def _profile_runtime_scope_tokens(profile_home, *, hydrate_secrets: bool = True) -> "_TurnScopes":
    """Bind HERMES_HOME + secret + terminal scope for ``profile_home`` (None = launch profile) and
    return the reset tokens. The launch profile's SECRET scope is always bound — its ``.env`` over
    the launch env (live while single-profile, frozen at activation afterwards; never live
    ``os.environ`` once a secondary context may have written to it, #107422) — so the credential
    source is fixed at entry and an in-flight launch body survives a concurrent first-secondary
    activation instead of hitting ``UnscopedSecretError`` mid-request. Its terminal policy is bound
    only once multiplexing is active: single-profile terminal execution keeps the standalone
    ``os.environ`` bridge.

    ``hydrate_secrets=False`` skips resolving the profile's EXTERNAL secret sources (``op run``,
    ``bws``, ``sh -c`` — a subprocess with a 30s CLI budget holding a process-global lock). Pass it
    from any body that persists state rather than calls a provider: the exit-flush worker has a 5s
    TOTAL budget, and one slow source there costs the transcript the flush exists to save.
    """
    from agent.secret_scope import is_multiplex_active
    scopes = _TurnScopes()
    try:
        if profile_home:
            home = Path(profile_home)
            # External sources first: the requested profile may never have been served in this process.
            if hydrate_secrets:
                from hermes_cli.env_loader import hydrate_profile_secret_sources
                hydrate_profile_secret_sources(home)
            secrets = build_profile_secret_scope(home)
            overlay = None
            scopes.home = set_hermes_home_override(str(home))
        else:
            # The launch home IS get_hermes_home() (``_profile_home`` answers None for "already the
            # launch profile"); single-profile, only its secrets need binding. Once multiplexing is
            # active the override is bound too: an unset override is the "unbound context" signal
            # plugin runtime bindings and per-home slots fail closed on (#118538).
            # Resolve at call time like the launch state.db handle: a harness that
            # re-homes the process after import must not read the old home's .env.
            from tui_gateway.launch_profile_policy import launch_secret_scope, launch_terminal_env
            home = _launch_home()
            secrets = launch_secret_scope(home)
            # No home stamp: this IS the process's own profile, and the stamp exists only to
            # mark a FOREIGN home for serves_routed_profile().
            scopes.secret = set_secret_scope(secrets)
            if not is_multiplex_active():
                return scopes
            scopes.home = set_hermes_home_override(str(home))
            overlay = launch_terminal_env()
        if scopes.secret is None:
            scopes.secret = set_secret_scope(secrets, profile_home=str(home) if profile_home else None)
        # Same terminal policy the gateway binds per turn: a docker-configured profile
        # must never resolve the launch process's pinned env. Failure → refusal scope.
        from tools.terminal_scope import install_profile_terminal_scope
        scopes.terminal = install_profile_terminal_scope(home, env_overlay=overlay)
        return scopes
    except Exception:
        # A raise mid-bind leaves no return value for the caller to release —
        # undo whatever was bound before propagating.
        _release_profile_runtime_scope_tokens(scopes)
        raise


def _release_profile_runtime_scope_tokens(scopes: "_TurnScopes | None") -> None:
    """Release terminal → secret → home. Each reset is independent: a failing terminal reset must
    not leave the previous profile's secrets / HERMES_HOME installed for the next body in this
    context (a fail-open scope leak on the teardown path). The first failure is re-raised after
    every scope has been released."""
    if scopes is None:
        return
    from tools.terminal_scope import reset_terminal_scope
    first_error: BaseException | None = None
    for token, reset in ((scopes.terminal, reset_terminal_scope), (scopes.secret, reset_secret_scope),
                         (scopes.home, reset_hermes_home_override)):
        if token is None:
            continue
        try:
            reset(token)
        except Exception as exc:  # noqa: BLE001 — keep releasing the remaining scopes
            first_error = first_error or exc
    if first_error is not None:
        raise first_error


@contextlib.contextmanager
def _session_profile_runtime_scope(session: dict, *, hydrate_secrets: bool = True):
    """Bind model resolution to the session's profile config and secrets (launch profile included
    once the process multiplexes; see ``_profile_runtime_scope_tokens``)."""
    scopes = _profile_runtime_scope_tokens(session.get("profile_home"), hydrate_secrets=hydrate_secrets)
    try:
        yield
    finally:
        _release_profile_runtime_scope_tokens(scopes)


def _session_default_model(session: dict) -> str:
    """The configured default model of the session's OWN profile. Bare ``_resolve_model()`` reads the
    LAUNCH profile's config, so a secondary session's reply or first state.db row carried the launch
    profile's model id."""
    with _session_profile_runtime_scope({"profile_home": session.get("profile_home") or None},
                                        hydrate_secrets=False):
        return _resolve_model()


def _restart_completed_failed_agent_build(sid: str, session: dict, failed_ready: threading.Event | None) -> bool:
    """Replace one completed failed build generation and start its retry."""
    if failed_ready is None:
        return False
    with session.setdefault("agent_build_lock", threading.Lock()):
        if (session.get("agent") is not None or session.get("agent_error") is None
                or session.get("agent_ready") is not failed_ready or not failed_ready.is_set()):
            return False
        model_override = session.get("model_override")
        resume_overrides = session.get("resume_runtime_overrides")
        if isinstance(model_override, dict) and isinstance(resume_overrides, dict):
            resume_overrides = {**resume_overrides, "model_override": model_override}
            if provider := model_override.get("provider"):
                resume_overrides["provider_override"] = provider
            else:
                resume_overrides.pop("provider_override", None)
            session["resume_runtime_overrides"] = resume_overrides
        session["agent_error"] = None
        session["agent_ready"] = threading.Event()
        session.pop("agent_build_started", None)
        session.pop("_agent_build_thread", None)
    _start_agent_build(sid, session)
    return True


def _switch_request(raw_input: str, parsed_flags, persist_override) -> tuple[str, str, bool, bool, str]:
    """Normalize /model flags → (model_input, explicit_provider, one_turn, persist_global, reasoning_effort)."""
    from hermes_cli.model_switch import (
        MODEL_SWITCH_ERR_ONCE_WITH_GLOBAL, MODEL_SWITCH_ERROR_TEXT, parse_model_switch_args,
        resolve_persist_behavior)

    f = parse_model_switch_args(raw_input) if parsed_flags is None else parsed_flags
    model_input, explicit_provider, is_global_flag, is_session, one_turn = (
        f.model_input, f.explicit_provider, f.is_global, f.is_session, f.is_once)
    # Conflict validation is the shared parser's; surface it with the canonical copy.
    for code in getattr(f, "errors", ()):
        raise ValueError(MODEL_SWITCH_ERROR_TEXT[code])
    if is_global_flag and one_turn:
        raise ValueError(MODEL_SWITCH_ERROR_TEXT[MODEL_SWITCH_ERR_ONCE_WITH_GLOBAL])
    if persist_override is None:
        persist_override = resolve_persist_behavior(
            is_global_flag, is_session, is_once=one_turn, explicit_provider=explicit_provider)
    if not model_input:
        raise ValueError("model value required")
    return model_input, explicit_provider, one_turn, persist_override, getattr(f, "reasoning_effort", "") or ""


def _current_model_runtime(agent, explicit_provider: str) -> tuple:
    """(provider, model, base_url, api_key) to switch from: live agent, else configured runtime."""
    if agent:
        return tuple(
            getattr(agent, k, "") or "" for k in ("provider", "model", "base_url", "api_key"))
    current_model = _resolve_model()
    if explicit_provider:
        return explicit_provider.strip(), current_model, "", ""
    from hermes_cli.runtime_provider import resolve_runtime_provider
    runtime = resolve_runtime_provider(requested=None, target_model=current_model or None)
    # Keep a callable api_key (Azure Entra bearer) unchanged: ``str()`` would
    # yield "<function ...>" and poison switch_model validation.
    key = runtime.get("api_key", "")
    if not (callable(key) and not isinstance(key, str)):
        key = str(key or "")
    provider = str(runtime.get("provider", "") or "")
    return provider, current_model, str(runtime.get("base_url", "") or ""), key


def _switch_away_provider(agent, explicit_provider: str, current_provider: str) -> str | None:
    """The provider of the model the user leaves. Agent-less with ``--provider``, ``current_provider``
    is the TARGET (what switch_model wants), so the launch route names it instead; None when only a
    credential resolve could tell, which the metric reads as ``unknown`` and so reports the model as
    ``custom``."""
    if agent or not explicit_provider:
        return current_provider
    if env_provider := os.environ.get("HERMES_TUI_PROVIDER", "").strip():
        return env_provider
    if _env_model_seed():
        return None
    return _config_model_target()[1] or None


def _merge_preflight_warning(result, agent, session: dict, cfg, custom_provs) -> None:
    """Fold the context-compression preflight warning into ``result`` (best-effort)."""
    try:
        from hermes_cli.context_switch_guard import merge_preflight_compression_warning
        cfg_ctx = None
        mc = cfg.get("model", {}) if isinstance(cfg, dict) else None
        if isinstance(mc, dict) and mc.get("context_length") is not None:
            cfg_ctx = int(mc["context_length"])
        merge_preflight_compression_warning(
            result, agent=agent, messages=list(session.get("history", [])),
            custom_providers=custom_provs, config_context_length=cfg_ctx)
    except Exception as exc:
        logger.debug("preflight-compression switch warning failed: %s", exc)


def _expensive_model_confirm(result, current_base_url: str, current_api_key, agent=None) -> dict | None:
    """Deferred-confirm response when the selection guards flag the target model (or, with a live
    ``agent``, the switch itself — large cached context), else None."""
    try:
        from hermes_cli.model_selection_guards import (
            combined_selection_warning, selection_context_for_agent)
        warning = combined_selection_warning(
            result.new_model, provider=result.target_provider, base_url=result.base_url or current_base_url,
            api_key=result.api_key or current_api_key, model_info=result.model_info,
            selection_context=selection_context_for_agent(agent))
    except Exception:
        warning = None
    if warning is None:
        return None
    msg = f"{warning.message}\n\n{result.warning_message}" if result.warning_message else warning.message
    # Same contract as _set_model's deferred branch: confirm_message is canonical, warning legacy.
    return {"value": result.new_model, "warning": msg, "confirm_required": True, "confirm_message": msg}


def _commit_agent_switch(sid: str, session: dict, agent, result, current_model: str, snapshot):
    """Swap the live agent in place, then restart/persist/mark/announce; a failed swap aborts."""
    try:
        agent.switch_model(
            new_model=result.new_model, new_provider=result.target_provider, api_key=result.api_key,
            base_url=result.base_url, api_mode=result.api_mode,
            capabilities=getattr(result, "runtime_capabilities", None))
    except Exception as exc:
        # The in-place swap rolled the agent back and re-raised. Abort the whole commit (worker
        # restart, persist, marker, override, config write) or the session pins a broken model.
        # Abort the commit: do NOT restart the slash worker, persist runtime, append the switch marker, set
        # a session model_override, or persist to config — all of which would otherwise leave the session
        # pinned to a broken model and kill the conversation on the next turn (#50163). A failed switch is a
        # no-op; surface a clean error to the client.
        logger.warning("In-place model switch failed for TUI agent: %s", exc)
        raise ValueError(f"Model switch to {result.new_model} failed ({exc}); "
                         f"staying on {getattr(agent, 'model', current_model)}.") from exc
    _restart_slash_worker(sid, session)
    _persist_live_session_runtime(session)
    _persist_live_session_system_prompt(session)
    _append_model_switch_marker(session, model=result.new_model, provider=result.target_provider)
    _emit_session_info(sid, session)
    if snapshot is not None:
        session["one_turn_model_restore"] = snapshot
    else:
        session.pop("one_turn_model_restore", None)


def _apply_model_switch(
    sid: str, session: dict, raw_input: str, *, confirm_expensive_model: bool = False,
    pin_session_override: bool = True, parsed_flags: Any | None = None,
    persist_override: bool | None = None, count_switch: bool = True) -> dict:
    """``count_switch=False``: an internal swap (config adoption, MoA one-shot and its restore), not a
    user's /model pick, so it stays out of the shared-metrics switch count."""
    from hermes_cli.model_switch import switch_model
    model_input, explicit_provider, one_turn, persist_global, reasoning_effort = _switch_request(
        raw_input, parsed_flags, persist_override)
    agent = session.get("agent")
    if one_turn and not agent:
        raise ValueError("/model --once requires a live session")
    current_provider, current_model, current_base_url, current_api_key = _current_model_runtime(
        agent, explicit_provider)
    # User-defined providers let switch_model resolve named custom endpoints
    # (e.g. "ollama-launch") and validate against saved model lists.
    user_provs = custom_provs = cfg = None
    with contextlib.suppress(Exception):
        from hermes_cli.config import get_compatible_custom_providers, load_config
        cfg = load_config()
        user_provs = cfg.get("providers")
        custom_provs = get_compatible_custom_providers(cfg)
    result = switch_model(
        raw_input=model_input, current_provider=current_provider, current_model=current_model,
        current_base_url=current_base_url, current_api_key=current_api_key, is_global=persist_global,
        explicit_provider=explicit_provider, user_providers=user_provs,
        custom_providers=custom_provs)
    if not result.success:
        raise ValueError(result.error_message or "model switch failed")
    restore_snapshot = _snapshot_agent_model_runtime(agent) if (one_turn and agent) else None
    if agent:
        _merge_preflight_warning(result, agent, session, cfg, custom_provs)
    if not confirm_expensive_model:
        confirm = _expensive_model_confirm(result, current_base_url, current_api_key, agent)
        if confirm is not None:
            return confirm
    records_composer_override = (
        pin_session_override and isinstance(session, dict) and not one_turn
        and not persist_global and session.get("follow_profile_config"))
    had_composer_profile = "composer_override_profile" in session
    previous_composer_profile = session.get("composer_override_profile")
    if records_composer_override:
        profile_model, profile_provider = _config_model_target()
        session["composer_override_profile"] = {
            "model": profile_model, "provider": profile_provider}
    try:
        if agent:
            # Provenance must exist before this transaction persists the switched runtime.
            _commit_agent_switch(sid, session, agent, result, current_model, restore_snapshot)
    except Exception:
        if records_composer_override:
            if had_composer_profile:
                session["composer_override_profile"] = previous_composer_profile
            else:
                session.pop("composer_override_profile", None)
        raise
    # PER-SESSION override so a rebuild of THIS session (/new, resume) re-derives the model.
    # Deliberately NOT written to process-global env (HERMES_MODEL & co.): the desktop hosts
    # every same-profile session in one process, so os.environ would leak the switch to all.
    if pin_session_override and isinstance(session, dict) and not one_turn:
        session["model_override"] = {
            "model": result.new_model, "provider": result.target_provider,
            "base_url": result.base_url, "api_key": result.api_key, "api_mode": result.api_mode}
    if persist_global:
        from hermes_cli.model_switch import persist_model_selection
        persist_model_selection(result)
    if reasoning_effort:
        _apply_switch_reasoning(sid, session, agent, reasoning_effort, persist_global=persist_global, one_turn=one_turn)
    if count_switch:
        from hermes_cli.observability.shared_metrics_events import record_model_switch

        record_model_switch(
            from_provider=_switch_away_provider(agent, explicit_provider, current_provider),
            to_provider=result.target_provider, surface=_session_source(session), from_model=current_model,
            session_id=getattr(agent, "session_id", None))
    return {
        "value": result.new_model, "warning": result.warning_message or "",
        "confirm_required": False,
        "scope": "once" if one_turn else ("global" if persist_global else "session")}


def _apply_switch_reasoning(sid: str, session, agent, effort: str, *, persist_global: bool, one_turn: bool) -> None:
    """``/model X --reasoning <level>``: the effort rides with the pick and shares its scope. Runs
    AFTER ``agent.switch_model`` (which re-resolves ``reasoning_config`` from config.yaml, so an
    earlier write would be clobbered). ``--once`` restores through ``one_turn_model_restore`` —
    the snapshot's ``primary_runtime`` carries the pre-switch ``reasoning_config``."""
    from hermes_constants import parse_reasoning_effort
    parsed = parse_reasoning_effort(effort)
    if parsed is None:
        return
    if agent is not None:
        agent.reasoning_config = parsed
    if one_turn or not isinstance(session, dict):
        return
    if persist_global:
        _write_config_key("agent.reasoning_effort", effort)
        session.pop("create_reasoning_override", None)  # global wins; see _set_reasoning
    else:
        session["create_reasoning_override"] = parsed
    if agent is not None:
        _persist_live_session_runtime(session)
        _emit_session_info(sid, session)  # the switch's own emit predates the effort change


def _sync_bot_capabilities(sid: str, session: dict) -> None:
    """Rebuild a Bot Chat session's agent when its capability surface changed. Bot Chats are
    eternal sessions with toolsets/MCP baked in at construction, so a capability edit would
    otherwise wait for /new: fingerprint at turn start and on change swap in a fresh agent for
    the SAME session (history is DB-backed)."""
    agent = session.get("agent")
    if agent is None:
        return
    try:
        title = str(getattr(agent, "_session_title_hint", "") or "").strip()
        if not title:
            db, key = getattr(agent, "_session_db", None), session.get("session_key") or ""
            title = str((db.get_session_title(key) if (db and key) else None) or "").strip()
        if title != "Bot Chat":
            return
        from tools.bot_mode_probe import capability_fingerprint
        current = capability_fingerprint(session.get("profile_home") or None)
        if current == "unavailable":
            return
        seen = session.get("bot_caps_seen")
        session["bot_caps_seen"] = current
        if seen is None or seen == current:
            return
    except Exception:
        return
    try:
        tokens = _set_session_context(sid, cwd=_session_cwd(session))
        try:
            new_agent = _rebuild_session_agent(sid, session, session_id=session["session_key"],
                                               platform_override=_session_source(session))
        finally:
            _clear_session_context(tokens)
        new_agent._session_title_hint = "Bot Chat"
        _emit("notice", sid, {"message": "Capabilities updated — this bot's tools and prompt were refreshed."})
    except Exception as e:
        logger.warning("Bot capability sync failed for %s: %s", sid, e)


def _sync_agent_model_with_config(sid: str, session: dict) -> None:
    """Adopt a config.yaml model change at turn start (like gateways do per message). Sessions
    pinned with /model keep their choice; a failed switch keeps the current model."""
    agent = session.get("agent")
    if agent is None:
        return
    target = _config_model_target()
    if not target[0]:
        return
    seen = session.get("config_model_seen")
    if target == seen:
        return
    superseded_pin = None
    if session.get("model_override"):
        composer_profile = session.get("composer_override_profile")
        pinned_profile = (
            str(composer_profile.get("model") or "").strip(),
            str(composer_profile.get("provider") or "").strip(),
        ) if isinstance(composer_profile, dict) else None
        if pinned_profile is None or pinned_profile == target:
            return
        # A later profile edit supersedes the canonical chat's explicit pick. Clearing both fields lets
        # the normal config-sync path switch now and prevents the old composer pick resurfacing on rebuild.
        superseded_pin = session.pop("model_override"), composer_profile
        session["composer_override_profile"] = None
    # Record first so a broken config gets one attempt per edit, not per turn.
    session["config_model_seen"] = target
    model, provider = target
    # Already on the configured model (resumed before first sync, or a config revert after
    # a failed switch): adopt without switching.
    if model == getattr(agent, "model", "") and (not provider or provider == getattr(agent, "provider", "")):
        if superseded_pin is not None:
            _persist_live_session_runtime(session)
        return
    raw = f"{model} --provider {provider}" if provider else model
    try:
        # This sync ADOPTS a config.yaml change; it must never write config back (that is
        # how `hermes --tui -m` once leaked into config.yaml).
        _apply_model_switch(
            sid, session, raw, confirm_expensive_model=True, pin_session_override=False,
            persist_override=False, count_switch=False)
    except Exception as e:
        logger.warning("Configured model %s could not be adopted for session %s: %s", model, sid, e)
        from gateway.warning_notifications import render_notification
        render_notification(
            lambda: _emit("error", sid, {"message": f"Could not switch to configured model {model}: {e}"}),
            platform="tui", user_config=getattr(session.get("agent"), "_notification_config", None))


def _pending_switch_selection_warning(model: str, provider: str) -> str | None:
    """Selection-guard message for a model queued mid-turn, or ``None``. Runs BEFORE the pick is
    stashed (the client can still turn the response into a confirm prompt); only pre-resolution
    inputs exist so it can only under-fire — ``_apply_model_switch`` is the backstop."""
    if not model:
        return None
    try:
        from hermes_cli.model_selection_guards import combined_selection_warning
        warning = combined_selection_warning(model, provider=provider or None)
    except Exception:
        return None
    return warning.message if warning is not None else None


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