"""Session-DB access for the dashboard: per-profile SessionDB opening with schema
heal, latest-descendant lookup and the auto-archive ticker.
"""

import logging
import asyncio
import threading
import time
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
            """


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


# Serialises the one-time writable schema bootstrap for read-only opens, so
# concurrent first-load polls don't open mode=ro against a half-written schema
# ("no such table: sessions").
_session_db_bootstrap_lock = threading.Lock()


def _session_db_read_probe_statements() -> tuple:
    """Stale-schema probes for read-only opens (which skip _reconcile_columns()).
    Derived from SCHEMA_SQL so a new column is probed automatically — a
    hand-written list once went stale and emptied the sidebar after update."""
    from hermes_state_schema import schema_read_probe_statements

    return schema_read_probe_statements()


# Stores where a heal WRITABLE OPEN SUCCEEDED but the read probe still failed:
# one reconciliation cannot fix them (e.g. a NOT-NULL-without-default column),
# so they fall back to the raw read-only open until restart instead of paying
# a writable init per poll. A FAILED writable open (transient lock) is NOT
# recorded — the next poll retries the heal.
_session_db_heal_exhausted: set = set()

# Deduplicates the heal-failure warning per store per process.
_session_db_heal_warned: set = set()


def _is_stale_schema_error(exc: BaseException) -> bool:
    message = str(exc).lower()
    return "no such table" in message or "no such column" in message


def _open_session_db_at_path(db_path: Path, *, read_only: bool):
    """Open a SessionDB at an explicit path with an explicit access mode.

    Read-only opens bootstrap a missing/zero-byte store once and heal a stale or
    malformed schema through ONE writable open before reopening read-only; the
    healthy read path never takes a write lock.  Tables outside SCHEMA_SQL
    (telemetry ``tel_*``, FTS shadow tables) are outside both probe and heal.
    """
    import sqlite3

    from hermes_state import SessionDB, is_malformed_schema_error
    from hermes_state_registry import acquire, release_or_close

    # Read-only file/sidecar preflight (port of kilocode#12508): repair-or-refuse BEFORE the first
    # connection so users get an actionable message instead of an opaque "attempt to write a readonly
    # database" from deep inside _init_schema.
    if not read_only:
        return acquire(db_path)

    def _needs_bootstrap() -> bool:
        try:
            return db_path.stat().st_size == 0
        except FileNotFoundError:
            return True
        except OSError:
            return False

    if _needs_bootstrap():
        with _session_db_bootstrap_lock:
            if _needs_bootstrap():
                db = acquire(db_path)
                release_or_close(db)

    def _open_probed():
        db = SessionDB(db_path=db_path, read_only=True)
        # Unit-test fakes may replace SessionDB without exposing a raw
        # connection. Probe only real connections.
        conn = getattr(db, "_conn", None)
        if conn is not None and str(db_path) not in _session_db_heal_exhausted:
            try:
                for statement in _session_db_read_probe_statements():
                    conn.execute(statement).fetchone()
            except BaseException:
                db.close()
                raise
        return db

    try:
        return _open_probed()
    except (sqlite3.DatabaseError, UnicodeDecodeError) as exc:
        # UnicodeDecodeError = pysqlite could not decode SQLite's own error
        # message because corrupt file bytes were embedded in it; the
        # one-writable-open heal is the only repair path, so treat it as
        # malformed schema.
        if not (
            _is_stale_schema_error(exc)
            or is_malformed_schema_error(exc)
            or isinstance(exc, UnicodeDecodeError)):
            raise
        db = acquire(db_path)
        release_or_close(db)
        try:
            return _open_probed()
        except (sqlite3.DatabaseError, UnicodeDecodeError) as still_stale:
            if not _is_stale_schema_error(still_stale):
                raise
            # Writable open succeeded but the store is STILL behind the probe:
            # serve reads without the probe (only queries touching the broken
            # part fail) and stop paying the writable init per poll.
            _session_db_heal_exhausted.add(str(db_path))
            if str(db_path) not in _session_db_heal_warned:
                _session_db_heal_warned.add(str(db_path))
                _log.warning(
                    "state.db at %s is missing schema that a writable "
                    "reconcile could not add (%s); read paths may partially "
                    "fail until the store is repaired",
                    db_path,
                    still_stale)
            return _open_probed()


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 stale-session auto-archive for ``profile``; never raises.
    ``hermes serve`` runs neither CLI nor gateway startup hooks, so this
    session-list trigger is what makes ``sessions.auto_archive`` work there."""
    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
    from tools.skills_sync_client import maybe_pull_skills
    from tools.skills_sync_client_org import maybe_pull_org_skills

    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)
    for pull in (maybe_pull_skills, maybe_pull_org_skills):
        try:
            pull()
        except Exception as exc:
            _log.debug("serve skill sync 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)