"""Read-only session browsing, latest-descendant lookup and owner maintenance.
"""

import logging
import asyncio
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Dict, Optional

from hermes_state_common import _RESET_CHILD_SQL, _sql_json_extract

# Same logger the code used before extraction (record parity).
_log = logging.getLogger("hermes_cli.web_server")

_DESCENDANTS_SQL = f"""
            WITH RECURSIVE descendants(id, parent_session_id, started_at) AS (
                SELECT id, parent_session_id, started_at FROM sessions WHERE id = ?
                UNION
                SELECT s.id, s.parent_session_id, s.started_at
                FROM sessions s
                JOIN descendants d ON s.parent_session_id = d.id
                -- Continuation edges only (same predicate as the session list's chain CTE): a subagent run,
                -- a /branch fork, a /new reset child or a tool-owned row is its own conversation, and resuming
                -- INTO one parks the user's chat in a row the sidebar never lists (#115092).
                WHERE {_sql_json_extract('s.model_config', '$._delegate_from')} IS NULL
                  AND {_sql_json_extract('s.model_config', '$._branched_from')} IS NULL
                  AND NOT ({_RESET_CHILD_SQL.format(a='s')})
                  AND COALESCE(s.source, '') != 'tool'
            )
            SELECT id, parent_session_id, started_at FROM descendants
            """


# Desktop 0.22.2's ``readableError`` keeps the HTTP ``detail`` and shows its "app and backend are
# on different versions" toast (with an Update action) for "out of sync (different versions)".
CLIENT_UPGRADE_REQUIRED = ('client_upgrade_required: this app is out of sync (different versions) '
                           'with the Hermes backend; update the app to edit sessions')


def _session_latest_descendant(session_id: str, db):
    """Resolve a session id to the newest child leaf session.

    /model may create child sessions; a dashboard refresh should continue the
    newest child instead of reopening the old parent. Returns ``(leaf, path)``.
    """
    sid = db.resolve_session_id(session_id)
    if not sid or not db.get_session(sid):
        return None, []

    conn = getattr(db, "_conn", None)
    if conn is not None:
        keys = ("id", "parent_session_id", "started_at")
        rows = [dict(zip(keys, row)) for row in conn.execute(_DESCENDANTS_SQL, (sid,)).fetchall()]
    else:
        rows = db.list_sessions_rich(limit=10000, offset=0, compact_rows=True)

    children = {}
    for row in rows:
        rid = row.get("id")
        parent = row.get("parent_session_id")
        if rid and parent:
            children.setdefault(parent, []).append(row)

    def started(row):
        try:
            return float(row.get("started_at") or 0)
        except Exception:
            return 0.0

    current = sid
    path = [sid]
    seen = {sid}
    while children.get(current):
        candidates = [r for r in children[current] if r.get("id") not in seen]
        if not candidates:
            break
        candidates.sort(key=started, reverse=True)
        current = candidates[0]["id"]
        path.append(current)
        seen.add(current)
    return current, path


def _session_mutation_context(request, profile):
    """Use the same principal and profile boundary for prepare and apply."""
    from fastapi import HTTPException
    from gateway.session_contract import Principal
    from hermes_cli.web_server import _has_valid_session_token
    from hermes_cli.web_server_cron import _cron_profile_home
    from hermes_state import _default_db_path

    native = getattr(request.state, 'native_http_principal', None)
    session = getattr(request.state, 'session', None)
    if native is not None:
        subject = native['subject']
    elif session is not None:
        from gateway.session_identity import authenticated_subject
        subject = authenticated_subject({"user_id": session.user_id,
            "provider": session.provider, "issuer": session.issuer})
    elif not getattr(request.app.state, 'auth_required', False) and _has_valid_session_token(request):
        from gateway.session_identity import authenticated_subject
        subject = authenticated_subject({'user_id': 'legacy-token-owner', 'provider': 'session-token'})
    else:
        raise HTTPException(status_code=401, detail='Unauthorized')
    authority = getattr(request.app.state, 'session_authority', None)
    home = Path(_cron_profile_home(profile)[1]) if profile else Path(_default_db_path()).parent
    capabilities = frozenset({'session:read', 'session:control', 'session:create', 'session:operator'})
    if authority is None:
        if getattr(request.app.state, 'gateway_runner', None) is not None:
            raise HTTPException(status_code=503, detail='session_authority_unavailable')
        # Standalone ``serve``/``dashboard`` (no composed gateway; its owner cutover is a
        # follow-up): the store is the owner. Same principal id form as a gateway authority's
        # (``str(home)``), so import bindings written offline stay valid once one starts.
        return None, Principal(subject, str(home.resolve()), capabilities, 'http')
    # ``?profile=`` selects a served home; its own authority (not the launch one) owns the mutation.
    from gateway.session_authorities import authority_for_home
    runner = getattr(request.app.state, 'gateway_runner', None)
    owner = authority_for_home(runner, home) if runner is not None else None
    if owner is None and home.resolve() == Path(authority.db.db_path).parent.resolve():
        owner = authority
    if owner is None:
        raise HTTPException(status_code=403, detail='profile_mismatch')
    authority = owner
    if native is not None and native['profile_id'] != authority.profile_id:
        raise HTTPException(status_code=403, detail='profile_mismatch')
    if native is not None:
        # One transport per verified native request, so the grant's own profile can claim its old
        # local history exactly as over the native socket (``require_history_claim``); a shared
        # 'http' id never can.
        import uuid
        return authority, Principal(subject, authority.profile_id, capabilities, 'native-http:' + uuid.uuid4().hex)
    return authority, Principal(subject, authority.profile_id, capabilities, 'http')


@contextmanager
def _native_history_transport(request, authority, actor):
    """Bind a verified native HTTP actor as a native legacy transport for one mutation."""
    native = None
    if authority is not None and str(getattr(actor, 'transport_id', '')).startswith('native-http:'):
        native = getattr(request.state, 'native_http_principal', None)
    if native is None:
        yield
        return
    from gateway.session_local_migration import bind_native_transport, unbind_native_transport
    bind_native_transport(authority, actor, {'provider': 'local', 'profile_id': authority.profile_id,
                                             'instance_id': native['instance_id']})
    try:
        yield
    finally:
        unbind_native_transport(authority, actor)


def _mutate_offline_store(profile, actor, params):
    """Standalone edit: the receipt transaction on this process's single shared writer.

    Runs under the exact owner lock (409 while any gateway owns the store, never a
    second writer beside it); revision/generation/idle guards are the gateway's own.
    """
    from fastapi import HTTPException
    from hermes_state_runtime import begin_runtime_epoch, mutate_runtime_session

    def run():
        db = _open_session_db_for_profile(profile, read_only=False)
        try:
            write_guard = None
            if params['operation'] == 'import':
                normalized, errors = db._validate_import_payload(params['payload']['sessions'])
                if errors:
                    raise HTTPException(status_code=400, detail={'errors': errors})
                from functools import partial
                from hermes_state_mutation_binding import import_history_control
                write_guard = partial(import_history_control, actor=actor,
                                      session_ids=tuple(item['session']['id'] for item in normalized))
            with db._read_ctx() as conn:
                row = conn.execute('SELECT epoch FROM runtime_epoch WHERE singleton=1').fetchone()
            # The reservation held here is the ownership lock begin_runtime_epoch requires.
            epoch = row[0] if row else begin_runtime_epoch(db, instance_id='dashboard-offline')
            return mutate_runtime_session(db, epoch=epoch, principal_id=actor.subject,
                expected_generation=params.get('expected_generation'), _authorize_write=write_guard,
                **{key: params[key] for key in ('session_id', 'request_id', 'expected_revision',
                                                'operation', 'payload')})
        finally:
            db.close()
    return _with_session_maintenance(profile, run)


async def _mutate_via_gateway_owner(profile, actor, params):
    """Forward a standalone dashboard mutation to the live gateway owner.

    While a gateway serves the same home, a separately launched ``hermes
    dashboard`` is not the store owner, so the offline path's
    exclusive-maintenance reservation can never win and every write action
    dies on a bare 409 (#138915). The owner exposes the same revision-fenced
    verb over the control socket; with no discoverable owner return None so the
    caller falls back to the offline store.

    A dial, mint or RPC that dies on an endpoint discovery just identified is
    not owner-less (a slow first dial, or an owner restart between discovery
    and mint): the offline reservation would only lose to that same owner and
    answer with the bare 409 this forward exists to remove, so every transport
    failure surfaces a retryable 503. Only the owner's refusal frame
    (``GatewayRPCError``) is passed on as a verdict.

    The ticket is minted for the dashboard's authenticated principal, not the
    socket UID: ``authorize_history`` compares the actor subject against the
    row's stored owner, so a cold row owned by a dashboard account only takes
    the mutation when the forward acts as that account.
    """
    from fastapi import HTTPException
    from hermes_cli.gateway_client import GatewayClientError, GatewayRPCError, connect_gateway
    from hermes_cli.gateway_runtime import discover_gateway_endpoint
    from hermes_cli.gateway_runtime_discovery import DiscoveryError
    from hermes_cli.web_server_cron import _cron_profile_home
    from hermes_state import _default_db_path

    home = Path(_cron_profile_home(profile)[1]) if profile else Path(_default_db_path()).parent
    # Discovery is read-only and never starts a gateway.
    discovery = await asyncio.to_thread(discover_gateway_endpoint, home, timeout=1.5)
    endpoint = getattr(discovery, 'endpoint', None)
    if endpoint is None:
        return None
    if not getattr(endpoint, 'control_home', True):
        # The selected home answered on its own socket (a multiplexer-served profile already
        # names the multiplexer root). Pin it, so the ticket is minted on the socket that just
        # identified as this profile's owner, never on the dashboard launch home's.
        from dataclasses import replace
        endpoint = replace(endpoint, control_home=str(home.expanduser().resolve()))
    call = {key: params[key] for key in
            ('session_id', 'request_id', 'expected_revision', 'operation', 'payload') if key in params}
    if params.get('expected_generation') is not None:
        call['expected_generation'] = params['expected_generation']
    try:
        async with connect_gateway(endpoint=endpoint, subject=actor.subject) as client:
            return await client.rpc('session.mutate', **call)
    except GatewayRPCError:
        # The owner's own refusal frame: a verdict the caller maps to the direct path's status.
        raise
    except (GatewayClientError, DiscoveryError, OSError, TimeoutError) as exc:
        # Transport, not a verdict: a refused WebSocket dial (wrapped as GatewayClientError), a
        # control socket that refuses the mint after a restart (DiscoveryError), a raw socket error
        # or a timeout. Retrying is safe because the request reuses its request_id.
        raise HTTPException(status_code=503, detail='owner_unreachable') from exc


# One reason -> HTTP status map for a session mutation refusal, shared by the direct path
# (RuntimeStoreError) and the owner forward (GatewayClientError): a storage failure is a
# server fault (503), never a session conflict.
_MUTATION_REFUSAL_STATUS = {'permission_denied': 403, 'profile_mismatch': 403, 'not_found': 404,
                            'invalid_params': 400, 'runtime_draining': 503, 'storage_unavailable': 503}


async def _mutate_session_request(request, profile, session_id, *, request_id,
                                  expected_revision, operation, payload, expected_generation=None):
    """Resolve the authenticated principal and owner; never open a second writer."""
    import sqlite3
    from fastapi import HTTPException
    from gateway.session_contract import SessionRef
    from gateway.session_mutations import mutate_session
    from hermes_cli.gateway_client import GatewayClientError
    from hermes_state_mutation_guards import MUTATION_GUARD_REFUSALS
    from hermes_state_runtime import RuntimeStoreError

    authority, actor = _session_mutation_context(request, profile)
    if operation == 'import' and authority is not None:
        _, errors = authority.db._validate_import_payload(payload['sessions'])
        if errors:
            raise HTTPException(status_code=400, detail={'errors': errors})
    if request_id is None and expected_revision is None and expected_generation is None:
        # A pre-fence client (Desktop <= 0.22.2) sends none of the identity. The owner never
        # commits an unfenced write, so name the upgrade; the wording is what those builds'
        # toast summariser maps to "different versions, update" (#139742).
        raise HTTPException(status_code=409, detail=CLIENT_UPGRADE_REQUIRED)
    if request_id is None or expected_revision is None:
        raise HTTPException(status_code=409, detail='mutation_identity_required')
    params = dict(session_id=session_id, request_id=request_id, expected_revision=expected_revision,
        operation=operation, payload=payload)
    if expected_generation is not None:
        params['expected_generation'] = expected_generation
    try:
        if authority is None:
            # A live gateway owner answers first; only an owner-less home takes
            # the offline store (whose reservation loses to that same owner).
            result = await _mutate_via_gateway_owner(profile, actor, params)
            if result is None:
                result = await asyncio.to_thread(_mutate_offline_store, profile, actor, params)
        else:
            with _native_history_transport(request, authority, actor):
                result = await mutate_session(authority, actor, SessionRef(authority.profile_id, session_id), params)
        return {'ok': True, **result}
    except RuntimeStoreError as exc:
        raise HTTPException(status_code=_MUTATION_REFUSAL_STATUS.get(exc.reason, 409), detail=exc.reason) from exc
    except GatewayClientError as exc:
        # The owner's own refusal verdicts (transport failures already became 503 above),
        # wire-shaped like RuntimeStoreError's reasons, so a forwarded write answers as the direct path.
        raise HTTPException(status_code=_MUTATION_REFUSAL_STATUS.get(str(exc), 409), detail=str(exc)) from exc
    except MUTATION_GUARD_REFUSALS as exc:
        raise HTTPException(status_code=409, detail='session_busy') from exc
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except sqlite3.Error as exc:
        raise HTTPException(status_code=503, detail='storage_unavailable') from exc


def _with_session_maintenance(profile, operation, *args):
    """Offline bulk actions reserve the exact owner lock through final DB close."""
    from fastapi import HTTPException
    from gateway.runtime_ownership import OwnershipConflict, exclusive_maintenance
    from hermes_cli.web_server_cron import _cron_profile_home
    from hermes_state import _default_db_path

    home = Path(_cron_profile_home(profile)[1]) if profile else Path(_default_db_path()).parent
    try:
        with exclusive_maintenance([home]):
            return operation(*args)
    except OwnershipConflict as exc:
        # uvicorn runs at log_level=warning, so a handled 4xx leaves no server-side trace and the
        # refusal reads as a UI bug (#138915). Bulk store maintenance has no owner verb: name it.
        _log.warning('session write refused for %s: %s', home, exc)
        raise HTTPException(status_code=409, detail=str(exc)) from exc


def _session_db_read_probe_statements() -> tuple:
    """Probe the declared schema without reconciling it on a browsing request."""
    from hermes_state_schema import schema_read_probe_statements

    return schema_read_probe_statements()


def _open_session_db_at_path(db_path: Path, *, read_only: bool):
    """Browsing never initializes or repairs the owner's session store.

    Owner startup reconciles schema. The one write a browse may do: a store that lags
    SCHEMA_SQL (missing table/column, not corruption) with NO gateway owning it gets the
    missing declared tables/columns once, in one short locked transaction (#140315); an
    owned store is refused by name instead. Explicit mutation APIs retain their writable
    acquisition until their separate owner-RPC migration.
    """
    import sqlite3

    from fastapi import HTTPException
    from hermes_state import SessionDB
    from hermes_state_errors import is_transient_sqlite_error
    from hermes_state_registry import acquire

    if not read_only:
        return acquire(db_path)

    try:
        initialized = db_path.stat().st_size > 0
    except FileNotFoundError:
        initialized = False
    if not initialized:
        raise HTTPException(
            status_code=503,
            detail="Session store is not initialized. Start this profile's gateway to initialize it.")

    from hermes_state_reconcile import (
        DeclarativeDriftError, is_declarative_drift_error, reconcile_drift_for_browse,
    )

    def _open_probed():
        db = SessionDB(db_path=db_path, read_only=True)
        try:
            conn = getattr(db, "_conn", None)
            if conn is not None:
                for statement in _session_db_read_probe_statements():
                    conn.execute(statement).fetchone()
            return db
        except BaseException:
            db.close()
            raise

    try:
        try:
            return _open_probed()
        except sqlite3.OperationalError as exc:
            # A missing declared table/column is upgrade drift, not corruption (#140315): catch
            # an unowned store up once under its runtime lock, refuse by name when a gateway
            # owns it. Never destructive: additive DDL only, no repair from browse.
            if not is_declarative_drift_error(exc):
                raise
            try:
                reconcile_drift_for_browse(db_path)
            except DeclarativeDriftError as drift:
                _log.warning("session store at %s is behind SCHEMA_SQL: %s", db_path, drift)
                raise HTTPException(status_code=503, detail=str(drift)) from exc
            return _open_probed()
    except (sqlite3.DatabaseError, UnicodeDecodeError) as exc:
        from hermes_state_errors import is_malformed_db_error
        if isinstance(exc, sqlite3.DatabaseError) and is_malformed_db_error(exc):
            # Same structured 503 the analytics routers publish (``corrupt_store_as_status``): a
            # corrupt image names the repair (`hermes doctor`), never a generic "schema unavailable".
            from hermes_cli.web_routers._common import corrupt_store_status
            raise corrupt_store_status(db_path, exc) from exc
        if isinstance(exc, sqlite3.OperationalError) and is_transient_sqlite_error(exc):
            detail = "Session store is busy (disk I/O or lock). Retry; the list was not cleared."
        else:
            detail = (
                "Session store schema is unavailable or corrupt. Start or restart this profile's "
                "gateway to reconcile it; if it persists, run `hermes doctor` for diagnosis. "
                "Browsing does not repair the store.")
        raise HTTPException(status_code=503, detail=detail) from exc


def _session_db_path_for_profile(profile: Optional[str]) -> Path:
    """state.db path for ``profile`` (None/empty = this process's own)."""
    from hermes_cli.web_server_cron import _cron_profile_home
    from hermes_state import _default_db_path

    if profile:
        _name, home = _cron_profile_home(profile)
        return Path(home) / "state.db"
    return Path(_default_db_path())


def _open_session_db_for_profile(profile: Optional[str], *, read_only: bool):
    """Open a SessionDB for ``profile`` (None/empty = this process's own state.db).

    Access-mode semantics: see :func:`_open_session_db_at_path`.
    """
    return _open_session_db_at_path(_session_db_path_for_profile(profile), read_only=read_only)


# In-process throttle for the opportunistic auto-archive trigger, keyed by
# profile: bounds the config.yaml read to once per window; the sweep itself is
# throttled far more coarsely by state_meta (sessions.min_interval_hours).
_AUTO_ARCHIVE_CHECK_INTERVAL_S = 300.0
_last_auto_archive_check: dict[str, float] = {}


def _maybe_auto_archive_for_profile(profile: Optional[str]) -> None:
    """Config-gated owner-lifetime maintenance, never invoked by browsing.

    The standalone serve ticker maintains its own profile; a composed HTTP
    listener relies on its gateway owner's startup and housekeeping hooks.
    """
    try:
        key = profile or ""
        now = time.monotonic()
        last = _last_auto_archive_check.get(key)
        if last is not None and now - last < _AUTO_ARCHIVE_CHECK_INTERVAL_S:
            return
        _last_auto_archive_check[key] = now

        from hermes_cli.config import load_config as _load_full_config
        from hermes_constants import reset_hermes_home_override, set_hermes_home_override

        # The config that governs a store is the one in that store's OWN home. A zero-arg
        # load_config() resolves through the PROCESS HERMES_HOME, so the dashboard swept every
        # profile's sessions with the launch profile's sessions.auto_archive/auto_archive_days —
        # one profile's retention silently decided another's.
        profile_home = _session_db_path_for_profile(profile).parent
        _home_token = set_hermes_home_override(str(profile_home))
        try:
            cfg = (_load_full_config().get("sessions") or {})
        finally:
            reset_hermes_home_override(_home_token)
        if not cfg.get("auto_archive", False):
            return
        from hermes_cli.profiles import _check_gateway_running

        # A live gateway owns this profile's store and runs the same sweep on its own
        # housekeeping tick ("state.db maintenance tick" in gateway/run.py, profile-scoped so a
        # multiplexed secondary's store is swept too). Opening it WRITABLE from `hermes
        # serve` adds a second writer to a database another process is already archiving,
        # for zero extra coverage (#110405). `_check_gateway_running` is the canonical
        # per-profile predicate (`_maybe_run_skill_maintenance` below uses it): its
        # multiplexer rung catches a served secondary, which owns no gateway.pid or lock
        # of its own and a bare lock-file probe would report stopped.
        if _check_gateway_running(profile_home):
            return
        db = _open_session_db_for_profile(profile, read_only=False)
        try:
            db.maybe_auto_archive(
                idle_days=float(cfg.get("auto_archive_days", 3)),
                min_interval_hours=int(cfg.get("min_interval_hours", 24)))
        finally:
            db.close()
    except Exception as exc:
        _log.debug("opportunistic auto-archive skipped: %s", exc)


def _skill_maintenance_idle_for(started_at: float) -> Optional[float]:
    """Measure chat inactivity, not socket inactivity (Desktop stays connected)."""
    import tui_gateway.server as gateway

    from hermes_constants import get_hermes_home

    home = get_hermes_home().resolve()
    with gateway._sessions_lock:
        sessions = [session for session in gateway._sessions.values()
                    if Path(session.get("profile_home") or home).resolve() == home]
        if any(session.get("running") for session in sessions):
            return None
        last_active = max(
            [started_at, gateway._closed_session_activity.get(str(home), 0)]
            + [float(session.get("last_active") or started_at) for session in sessions])
    return max(0.0, time.time() - last_active)


def _maybe_run_skill_maintenance(started_at: float) -> None:
    from hermes_constants import get_hermes_home
    from hermes_cli.profiles import _check_gateway_running

    # A live messaging gateway already owns these chores for this profile.
    if _check_gateway_running(get_hermes_home()):
        return

    from agent.curator import maybe_run_curator

    try:
        idle_for = _skill_maintenance_idle_for(started_at)
        if idle_for is not None:
            maybe_run_curator(idle_for_seconds=idle_for)
    except Exception as exc:
        _log.debug("serve curator tick skipped: %s", exc)


async def _auto_archive_ticker_loop(
    interval_s: float = 3600.0, initial_delay_s: float = 90.0) -> None:
    """Poll maintenance for this serve profile, including Desktop-only installs.

    Individual chores own their config/interval gates. Curator additionally uses
    real chat inactivity; merely keeping a Desktop WebSocket open is not activity.
    """
    started_at = time.time()

    def _sweep() -> None:
        _maybe_auto_archive_for_profile(None)
        _maybe_run_skill_maintenance(started_at)

    await asyncio.sleep(initial_delay_s)
    while True:
        try:
            await asyncio.to_thread(_sweep)
        except Exception as exc:
            _log.debug("auto-archive tick skipped: %s", exc)
        await asyncio.sleep(interval_s)