"""Ink's ``/bg`` (``prompt.background``) and ``/btw`` (``prompt.btw``) on the shared owner.

Each verb replies ``{task_id}`` at once and the answer lands later on the parent session's event
stream as ``background.complete`` ``{task_id, text}`` / ``btw.complete`` ``{task_id, question,
text}`` (``error: <exc>`` on failure) — the sidecar's contract (``tui_gateway/methods_prompt.py``).
The side agent is detached: never an admission, never a FIFO turn, never a row in the parent
transcript. It runs on a daemon thread
under the owning profile's scope and the session's frozen launch policy (cwd, terminal, model
route, toolsets), so a session that never ran a turn (no resident agent) works the same.
"""
from functools import partial
import asyncio
import contextlib
import logging
import threading
import uuid

from gateway.session_busy_controls import authorize
from hermes_state_runtime import RuntimeStoreError

logger = logging.getLogger(__name__)

_RUNTIME_KEYS = ('api_key', 'base_url', 'provider', 'requested_provider', 'api_mode', 'command', 'args',
                 'credential_pool', 'max_tokens', 'capabilities')


def handlers(connection):
    return {'prompt.background': partial(background, connection), 'prompt.btw': partial(btw, connection),
            'preview.restart': partial(preview_restart, connection)}


def _admit(connection, ref, params, prefix, required='text', optional=()):
    """Shared admission: ``(live, policy, text, task_id)`` or a refusal before any work starts."""
    text = params.get(required)
    if (set(params) - {'session_id', 'profile', required, *optional} or not isinstance(ref.session_id, str)
            or not ref.session_id or not isinstance(text, str) or not text.strip()
            or any(params.get(key) is not None and not isinstance(params[key], str) for key in optional)):
        raise RuntimeStoreError('invalid_params')
    authority = connection.authority
    if ref.session_id not in authority.sessions:
        raise RuntimeStoreError('not_found')
    authorize(connection, ref, params, 'session:submit')
    # The answer rides this connection's event subscription, like a submit's reply.
    if ref.session_id not in connection.subscriptions:
        raise RuntimeStoreError('permission_denied')
    authority._require_admission_open()
    from gateway.session_policy import policy_for_source
    live = authority.sessions[ref.session_id]
    policy = policy_for_source(authority.runner, live.source)
    if policy is None:
        raise RuntimeStoreError('permission_denied')
    return live, policy, text, f'{prefix}_{uuid.uuid4().hex[:6]}'


@contextlib.contextmanager
def _side_scope(authority, policy, task_id):
    """The owning profile's home/secrets, the session's frozen policy, and the task's own session
    identity (approval key, HERMES_SESSION_ID) for the side agent's thread."""
    from gateway.session_authorities import owner_scope, served_profile_name
    from gateway.session_context import clear_session_vars, set_session_vars
    from gateway.session_policy import bind_turn_workspace, policy_scope, unbind_turn_workspace
    from tools.environments.local import _is_routed_home
    # The owning profile, as its own turns bind it: a served secondary by its name, so persistent
    # Docker keys ITS workspace container (never the launch profile's live one for the same folder);
    # the launch home binds none, like its ``agent:main`` turns.
    home = authority.profile_id
    profile = served_profile_name(home) if _is_routed_home(home) else ''
    with owner_scope(authority, hydrate_secrets=True), policy_scope(policy, authority=authority):
        tokens = set_session_vars(session_key=task_id, session_id=task_id, cwd=policy.cwd, profile=profile)
        try:
            # The session's workspace, as its own turns get it (#140665): mounted at /workspace in a
            # per-task container, or the session's own persistent workspace container (same digest).
            bind_turn_workspace(policy, task_id)
            yield
        finally:
            unbind_turn_workspace(task_id)
            clear_session_vars(tokens)


def _resolve_runtime(authority, policy):
    """``(model, runtime)`` the session's next turn would use: its frozen policy route."""
    from gateway.run_turn_prepare import _resolve_policy_agent_runtime
    return _resolve_policy_agent_runtime(authority.runner, policy)


def _side_agent(authority, policy, live, task_id, **overrides):
    """A fresh ``AIAgent`` on the session's frozen route (call inside ``_side_scope``)."""
    from agent.skill_utils import parse_config_string_list
    from gateway.run import _checkpoint_agent_kwargs
    from run_agent import AIAgent
    runner = authority.runner
    model, runtime = _resolve_runtime(authority, policy)
    config = policy.config(authority)
    from gateway.session_worker_construct import session_routing
    pr = session_routing(authority, policy)
    refresh_fallback = getattr(runner, '_refresh_fallback_model', None)
    kwargs = dict(
        model=model, **{key: runtime.get(key) for key in _RUNTIME_KEYS if key in runtime},
        **_checkpoint_agent_kwargs(config),
        max_iterations=policy.max_turns, quiet_mode=True, verbose_logging=False,
        enabled_toolsets=list(policy.current_toolsets(config)),
        disabled_toolsets=parse_config_string_list((config.get('agent') or {}).get('disabled_toolsets')) or None,
        reasoning_config=policy.current_reasoning_config(runner, live.route),
        service_tier=policy.current_service_tier(runner, live.route),
        request_overrides=dict(runtime.get('request_overrides') or {}),
        providers_allowed=pr.get('only'), providers_ignored=pr.get('ignore'), providers_order=pr.get('order'),
        provider_sort=pr.get('sort'), provider_require_parameters=pr.get('require_parameters', False),
        provider_data_collection=pr.get('data_collection'),
        # Its own session in the owning profile's store, like the sidecar's ``bg_*`` rows.
        session_id=task_id, platform=policy.platform, session_db=authority.db,
        fallback_model=refresh_fallback() if callable(refresh_fallback) else None,
        load_soul_identity=not policy.ignore_rules, skip_memory=policy.ignore_rules,
        side_agent=True)
    kwargs.update(overrides)
    return AIAgent(**kwargs)


def _abandoned(holder):
    abandoned = getattr(holder[0], 'abandoned', None) if holder else None
    return bool(abandoned and abandoned())


def _spawn(authority, ref, task_id, event_type, body, extra=None, holder=None):
    """Run ``body()`` on a daemon thread; publish its text (or ``error: <exc>``) on the session.

    The thread is owner work until it unwinds: the authority's ``_side_agents`` ledger is counted by
    the restart wait, the shutdown drain, the SessionDB close gate and idle exit
    (``session_runtime_workers.uncounted_runtime_work``), and profile retirement interrupts
    ``holder[0]`` (the side agent, once built) and joins the thread."""
    from gateway.session_runtime_workers import persist_work_count
    extra = extra or {}
    ledger = getattr(authority, '_side_agents', None)
    if ledger is None:
        ledger = authority._side_agents = {}
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = None

    def run():
        # The whole target: a BaseException the agent re-raises (cancellation, exit) must not
        # leave a dead thread counted as owner work.
        try:
            try:
                text = body()
            except Exception as exc:
                logger.warning('Side agent %s failed', task_id, exc_info=True)
                # A refusal names the check that fired (``workspace_unavailable (cwd: <path>)``).
                field, detail = getattr(exc, 'field', None), getattr(exc, 'detail', None)
                named = f' ({field}: {detail})' if field and detail else ''
                text = f'error: {getattr(exc, "reason", None) or exc}{named}'
            live = authority.sessions.get(ref.session_id)
            if live is not None and not _abandoned(holder):
                live.event_stream.publish(ref.session_id, {'task_id': task_id, **extra, 'text': str(text or '')},
                                          event_type=event_type)
        finally:
            ledger.pop(thread, None)
            if loop is not None:
                # gateway_state.json's active_agents drops back, recounted on the owner loop.
                with contextlib.suppress(RuntimeError):  # the loop already closed
                    loop.call_soon_threadsafe(persist_work_count, authority)

    thread = threading.Thread(target=run, name=f'side-agent-{task_id}', daemon=True)
    ledger[thread] = holder
    try:
        thread.start()
    except BaseException:
        ledger.pop(thread, None)
        raise
    persist_work_count(authority)
    return {'task_id': task_id}


async def background(connection, ref, params):
    """``/bg <prompt>``: a fresh agent on the session's route runs the prompt to completion."""
    live, policy, text, task_id = _admit(connection, ref, params, 'bg')
    authority = connection.authority
    holder = [None]

    def body():
        from tui_gateway.methods_prompt import _final_response_text
        with _side_scope(authority, policy, task_id):
            agent = holder[0] = _side_agent(authority, policy, live, task_id)
            try:
                result = agent.run_conversation(user_message=text, task_id=task_id)
            finally:
                # AIAgent.close() is the owner boundary (memory, tool subprocesses, clients).
                with contextlib.suppress(Exception):
                    agent.close()
        return _final_response_text(result)

    return _spawn(authority, ref, task_id, 'background.complete', body, holder=holder)


async def btw(connection, ref, params):
    """``/btw <question>``: a side question over a snapshot of the conversation
    (``agent/side_question.py``); the live history, FIFO and prompt cache stay untouched."""
    _live, policy, text, task_id = _admit(connection, ref, params, 'btw')
    authority = connection.authority
    # The resident agent's conversation (a cache-parity fork replays it); a session with no
    # resident agent (never ran a turn here, or evicted) answers from its stored transcript.
    agent = authority.agent(ref)
    snapshot = list(getattr(agent, '_session_messages', None) or [])
    from gateway.session_local_recovery import transcript_target
    target = None if snapshot else transcript_target(authority, ref)

    handle = _SideQuestionStop()

    def body():
        from agent.side_question import answer_side_question
        history = snapshot or authority.db.get_messages_as_conversation(target)
        with _side_scope(authority, policy, task_id):
            model, runtime = _resolve_runtime(authority, policy)
            main_runtime = {'model': model, 'session_id': ref.session_id,
                            **{key: runtime.get(key) for key in ('provider', 'base_url', 'api_key', 'api_mode')}}
            return answer_side_question(text, history, parent_agent=agent, main_runtime=main_runtime,
                                        on_fork=handle.set_fork, should_stop=handle.stopped.is_set) or ''

    return _spawn(authority, ref, task_id, 'btw.complete', body, extra={'question': text}, holder=[handle])


class _SideQuestionStop:
    """``/btw``'s holder: retirement and shutdown interrupt its cache-parity fork, and once stopped
    the read-only one-shot fallback (an uncancellable aux call of up to 180 s that writes nothing)
    no longer holds owner work, so it cannot keep a retiring profile's reservation."""

    def __init__(self):
        self.stopped = threading.Event()
        self._lock = threading.Lock()
        self._fork = None

    def set_fork(self, fork):
        with self._lock:
            self._fork = fork
        if fork is not None and self.stopped.is_set():
            self.hard_interrupt('Profile runtime is stopping')

    def hard_interrupt(self, message=None):
        self.stopped.set()
        with self._lock:
            fork = self._fork
        if fork is not None:
            from agent.interrupt_compat import request_hard_interrupt
            request_hard_interrupt(fork, message)

    def abandoned(self):
        """Stopped with no fork running: what is left publishes nothing and writes nothing."""
        with self._lock:
            return self.stopped.is_set() and self._fork is None


async def preview_restart(connection, ref, params):
    """Desktop preview pane "restart dev server": a hidden terminal+file agent brings the preview
    URL's server back up (rules: ``tui_gateway/methods_prompt.py`` ``_PREVIEW_RESTART_RULES``).
    Detached like ``/bg``; progress rides ``preview.restart.progress``, the answer
    ``preview.restart.complete`` ``{task_id, text}``. Context is the session's stored transcript."""
    live, policy, url, task_id = _admit(connection, ref, params, 'preview', required='url',
                                        optional=('cwd', 'context'))
    authority = connection.authority
    cwd, context = (str(params.get(key) or '').strip() for key in ('cwd', 'context'))
    from gateway.session_local_recovery import transcript_target
    target = transcript_target(authority, ref)
    holder = [None]

    def progress(text):
        if (text := str(text or '').strip()) and (stream := authority.sessions.get(ref.session_id)) is not None:
            stream.event_stream.publish(ref.session_id, {'task_id': task_id, 'text': text},
                                        event_type='preview.restart.progress')

    def body():
        from tui_gateway.agent_callbacks import _preview_restart_history
        from tui_gateway.methods_prompt import (_final_response_text, _preview_restart_prompt,
                                                _validated_preview_cwd)
        preview_cwd = _validated_preview_cwd(cwd)
        history = _preview_restart_history(authority.db.get_messages_as_conversation(target) if target else [])
        prompt = _preview_restart_prompt(url, cwd, context, bool(history))
        with _side_scope(authority, policy, task_id):
            if preview_cwd:
                from tools.terminal_tool import register_task_env_overrides
                register_task_env_overrides(task_id, {'cwd': preview_cwd})
            try:
                progress('Starting hidden restart agent')
                # Deliberately NOT closed via AIAgent.close(): it would kill the background
                # server this task exists to leave running.
                agent = _side_agent(authority, policy, live, task_id, enabled_toolsets=['terminal', 'file'],
                                    disabled_toolsets=None, session_db=None, skip_memory=True,
                                    tool_start_callback=lambda _id, name, _args: progress(f'Running {name}'))
                holder[0] = agent
                result = agent.run_conversation(user_message=prompt, task_id=task_id,
                                                conversation_history=history or None)
            finally:
                with contextlib.suppress(Exception):
                    from tools.terminal_tool import clear_task_env_overrides
                    clear_task_env_overrides(task_id)
        return _final_response_text(result)

    return _spawn(authority, ref, task_id, 'preview.restart.complete', body, holder=holder)