"""Exception types and error-classification predicates for the state store.
Shared by hermes_state and its mixins; string predicates match wrapped RPC
strings as well as live sqlite3 exceptions."""

import errno
import re
import sqlite3

# Malformed schema: ``sqlite_master`` itself is inconsistent (typically a DUPLICATE
# ``CREATE VIRTUAL TABLE messages_fts`` row). SQLite parses the whole schema while
# preparing the FIRST statement, so EVERY statement raises (even ``PRAGMA
# journal_mode`` during __init__); only ``PRAGMA writable_schema=ON`` +
# sqlite_master surgery still work. Canonical rows are intact; recovery rebuilds
# only the FTS layer.
_MALFORMED_SCHEMA_MARKERS = ("malformed database schema",)
_MALFORMED_DB_MARKERS = (*_MALFORMED_SCHEMA_MARKERS, "database disk image is malformed")


def is_malformed_db_error(exc: BaseException) -> bool:
    """Malformed-schema OR generic corrupt-image error. Diagnostics / offline
    recovery only — runtime repair must use :func:`is_malformed_schema_error`."""
    return isinstance(exc, sqlite3.DatabaseError) and any(
        marker in str(exc).lower() for marker in _MALFORMED_DB_MARKERS
    )


# SQLITE_IOERR as a substring (wrapped strings still classify).
_DISK_IO_ERROR_MARKER = "disk i/o error"

# "Store BUSY, not gone" — HTTP callers map these to 503 instead of 500. Corruption
# is deliberately absent: a malformed store must surface, not be retried into a timeout.
_TRANSIENT_SQLITE_MARKERS = (
    _DISK_IO_ERROR_MARKER, "database is locked", "database table is locked", "busy",
)


# Lock contention by result code. SQLite keeps SQLITE_BUSY when FTS5's xConnect loses the race
# on its %_config read but replaces the text with "vtable constructor failed: messages_fts",
# so a phrase match read a busy store as a hard failure.
_SQLITE_LOCK_CODES = (sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED)


def _sqlite_primary_code(exc_or_str) -> int | None:
    """Primary result code (extended codes keep it in the low byte); None when unknown."""
    code = getattr(exc_or_str, "sqlite_errorcode", None)
    return code & 0xFF if isinstance(code, int) else None


def is_sqlite_lock_error(exc_or_str) -> bool:
    """SQLITE_BUSY / SQLITE_LOCKED: wait and retry, never treat as damage. A known result code
    decides; only without one (our own re-raised messages, RPC-wrapped strings) does the text."""
    code = _sqlite_primary_code(exc_or_str)
    if code is not None:
        return code in _SQLITE_LOCK_CODES
    text = str(exc_or_str).lower()
    return "locked" in text or "busy" in text


def _is_no_more_rows(exc: sqlite3.Error) -> bool:
    """Transient engine error on contended WAL appends (retries like locked/busy);
    message-scoped because some builds raise it as InterfaceError."""
    return "no more rows available" in str(exc).lower()


def is_transient_sqlite_error(exc: BaseException) -> bool:
    """"Busy right now", not "damaged": one predicate so retry and the HTTP
    503-vs-500 split cannot drift apart."""
    return isinstance(exc, sqlite3.OperationalError) and (
        is_sqlite_lock_error(exc) or any(marker in str(exc).lower() for marker in _TRANSIENT_SQLITE_MARKERS)
    )


def is_malformed_schema_error(exc: BaseException) -> bool:
    """Only SQLite's explicit malformed-schema text: a generic "disk image is
    malformed" may be any B-tree page, so runtime repair must fail closed on it."""
    return isinstance(exc, sqlite3.DatabaseError) and any(
        marker in str(exc).lower() for marker in _MALFORMED_SCHEMA_MARKERS
    )


# "Filesystem cannot accept another write" substrings (OSError, sqlite3, wrapped RPC strings).
_DISK_FULL_MARKERS = (
    "no space left on device", "not enough space", "database or disk is full",  # SQLITE_FULL
    "disk full", "full disk", "enospc",
)


def is_disk_full_error(exc: BaseException | str | None) -> bool:
    """Disk-full / ENOSPC: OSError(ENOSPC), SQLITE_FULL, or matching strings."""
    if exc is None:
        return False
    if isinstance(exc, OSError) and getattr(exc, "errno", None) == errno.ENOSPC:
        return True
    lowered = (exc if isinstance(exc, str) else str(exc)).lower()
    return any(marker in lowered for marker in _DISK_FULL_MARKERS)


# Every classify_persistence_error bucket; consumers enumerate this tuple.
PERSISTENCE_ERROR_CAUSES = (
    "locked", "compression", "compression_closed", "turn_lease", "corrupt", "fts_index",
    "replaced", "deleted_wal", "disk", "session_row_missing", "unknown",
)


# "Database FILE structurally damaged" substrings. "database disk image is
# malformed" contains "disk", so this check MUST run before the disk bucket in
# classify_persistence_error or B-tree corruption reads as "free some disk space".
# Kept as plain substrings so sqlite3.DatabaseError, wrapped RPC strings, and logged message text all match
# the same helper. See #77386.
_DB_CORRUPTION_MARKERS = (
    "malformed", "file is not a database", "not a database", "database corruption",
)

# The module constant exists on Python 3.11+; the numeric value is stable across SQLite releases.
SQLITE_CORRUPT_VTAB = getattr(sqlite3, "SQLITE_CORRUPT_VTAB", 267)

# Every FTS object hangs off this prefix: the virtual tables and their _data/_idx/_content/
# _docsize/_config shadow b-trees. FTS5 names the table in its own corruption reports.
_FTS_OBJECT_RE = re.compile(r"\bmessages_fts\w*")


def is_fts_scoped_corruption_error(exc_or_str) -> bool:
    """Corruption SQLite itself attributes to the FTS index layer: the ONE provenance rule
    shared by the write-repair gate (``SessionDB._is_fts_write_corruption_error``), the
    gateway transcript retry and :func:`classify_persistence_error` (#96038, #97794).

    A known result code outranks prose: ``SQLITE_CORRUPT_VTAB`` is FTS-scoped even with
    the generic malformed-image text older SQLite builds emit, while bare ``SQLITE_CORRUPT``
    / ``SQLITE_NOTADB`` carry no object scope and any other known code contradicts
    FTS-looking prose, so both fail closed. Only without a code (Python < 3.11, RPC-wrapped
    strings) does the text decide, and then only an ``fts5:`` corruption report or a
    corruption marker that names a ``messages_fts*`` object counts.
    """
    if exc_or_str is None:
        return False
    code = getattr(exc_or_str, "sqlite_errorcode", None)
    if code is not None:
        return code == SQLITE_CORRUPT_VTAB
    text = (exc_or_str if isinstance(exc_or_str, str) else str(exc_or_str)).lower()
    if not _FTS_OBJECT_RE.search(text):
        return False
    if text.startswith("fts5:") and "corrupt" in text:
        return True
    return any(marker in text for marker in _DB_CORRUPTION_MARKERS)


class CompressionSessionClosedError(RuntimeError):
    """A durable write targeted a parent already closed by compression."""

    def __init__(self, session_id: str):
        self.session_id = session_id
        super().__init__(
            f"Session {session_id!r} is closed by compression; "
            "adopt its live continuation before appending messages"
        )


class CompressionSessionBusyError(RuntimeError):
    """A non-owner tried to write while compression owns the session."""


class SessionCompressionInProgressError(CompressionSessionBusyError):
    """A concurrent writer collided with a *live* compression lock — transient
    (the compressor publishes in seconds; ``_execute_write`` waits), unlike the
    parent class's other case (a compressor whose own lease is gone: permanent,
    fail fast). Subclassing keeps every existing handler working."""


class PruneRowUnresolvedError(RuntimeError):
    """An in-place prune rewrite could not name the exact live row of a changed message. Raised before
    any write, so the caller commits through ``archive_and_compact`` instead."""


class PruneRowStaleError(PruneRowUnresolvedError):
    """The named live row no longer holds the body the pruner is pruning: another writer already
    rewrote this generation. Unlike plain ambiguity this must NOT fall back to the full writer —
    that writer archives rows it never compared and would republish the stale held text over the
    newer one. Subclassing keeps callers that only know the parent failing closed-enough (they stop
    the in-place path); the prune caller tells the two apart and skips the pass entirely."""


class HeldRowRewrittenError(RuntimeError):
    """A full-writer commit found one of the held rows it was about to archive already rewritten in
    place by another writer. Raised INSIDE the commit's write transaction, after the caller's
    precheck but before any row is archived: a write that lands in that window is exactly the lost
    race the precheck cannot see (#124102). Callers treat it like ``StaleHeldHistory`` — nothing
    written, the pass is a no-op and the newer generation keeps the session."""


class SessionTurnLeaseLostError(RuntimeError):
    """A transcript write presented a turn-lease holder that no longer owns it.
    Fail-fast fencing (no ``_execute_write`` retry): a later writer may already
    be persisting a newer turn, and landing this one would interleave a stale reply."""


class SessionActiveWriteGuardError(RuntimeError):
    """Raised when an active turn lease or compression lock rejects session deletion."""


class StateDbReplacedError(RuntimeError):
    """The state.db path no longer names the file this SessionDB opened
    (out-of-band cp/mv/restore). In-place FTS repair and fail-open trigger
    dropping cannot fix a generation mismatch; they amplify it."""


class DeletedWalGenerationError(StateDbReplacedError):
    """A live process holds a deleted state.db-wal / -shm generation. Opening or
    writing through this handle would mint a second WAL inode (split-brain ->
    intermittent SQLITE_CORRUPT / IOERR). Stop the writers; never unlink the WAL
    yourself. Subclasses StateDbReplacedError so every consumer that diverts
    transcripts on a replaced store handles this identically."""


# SQLite header application_id (offset 68). Distinct from inode: ``cp`` onto the
# same path keeps st_ino and truncates+rewrites.
_STATE_DB_APPLICATION_ID_OFFSET = 68
_STATE_DB_GENERATION_KEY = "db_file_generation"
_STATE_DB_REPLACED_MSG = (
    "FATAL: state.db was replaced underneath the gateway; refusing further "
    "writes to this file. Divert transcripts to sessions/<id>.jsonl (and the "
    "gateway pending_messages spool) and restore or reopen after operator intervention."
)
STORAGE_RECOVERY_DOCS_URL = "https://hermes-agent.nousresearch.com/docs/user-guide/session-storage-recovery"

# Two layers (#110054): the first sentence is for the person reading a chat bubble or a banner (what
# happened, nothing is lost, the one thing to do); the rest is the operator detail. The phrase
# "deleted state.db-wal or state.db-shm" is the classifier's RPC-wrapped fingerprint — keep it.
_DELETED_WAL_GENERATION_MSG = (
    "FATAL: session storage stopped writing because another Hermes process still holds a deleted "
    "state.db-wal or state.db-shm inode (an old copy of the write-ahead log). Nothing is lost: quit "
    "every Hermes process on this profile (Desktop app, gateway, dashboard, cron), run `hermes doctor` "
    "(it names the processes still holding the log), then start Hermes again. Do not delete the WAL "
    "yourself and do not run `hermes doctor --fix` while they are running. "
    f"Guide: {STORAGE_RECOVERY_DOCS_URL} "
    "Detail: the path names a different (or missing) generation than the one this process holds "
    "open; opening or writing through it would mint a second WAL (split-brain). "
    "database.journal_mode: delete is operator containment, not a new default."
)


class StateDbCorruptError(sqlite3.DatabaseError):
    """A live SessionDB observed structural (non-FTS, non-replaced) corruption and
    is quarantined: sticky for the handle's life — writes fail fast, no reopen,
    no close-time checkpoint (a handle that kept writing after the first error
    checkpointed 15 pages under wrong page numbers and turned a readable file
    into "file is not a database"; SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE on 3.12+
    also stops SQLite's own). Subclasses sqlite3.DatabaseError so every degrade
    path keeps working. Recovery boundary: restart on a repaired/restored file.

    Stopping the writes is what prevents that; skipping the explicit checkpoint is the second line of
    defence. SQLite still runs its own last-connection checkpoint inside ``close()`` (and deletes the
    ``-wal`` sidecar) unless ``SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE`` is set — Python exposes it via
    ``Connection.setconfig()`` on 3.12+, so quarantine disables the close-time checkpoint there and the WAL
    survives on disk for forensics; on 3.11 the internal checkpoint is unavoidable (post-quarantine it can
    only carry pre-corruption committed frames, since no further writes are accepted). See #90837.
    """


_STATE_DB_CORRUPT_MSG = (
    "FATAL: state.db reported structural corruption (database disk image is "
    "malformed outside the FTS shadow tables) on a live handle; refusing further "
    "writes, automatic reopen, and the close-time WAL checkpoint on this file. "
    "Stop the gateway, then run `hermes sessions recover --source <state.db> "
    "--inspect-only` or restore a snapshot. Unwritten transcripts are diverted to "
    "sessions/<id>.jsonl (and the gateway pending_messages spool)."
)


_PERSISTENCE_CAUSE_BY_TYPE = (
    (SessionTurnLeaseLostError, "turn_lease"),
    (CompressionSessionClosedError, "compression_closed"),
    (CompressionSessionBusyError, "compression"),
    # The WAL-generation error subclasses StateDbReplacedError so existing write diversion keeps
    # working; classify it first because its recovery artifact and operator action are different.
    (DeletedWalGenerationError, "deleted_wal"),
    (StateDbReplacedError, "replaced"),
    (StateDbCorruptError, "corrupt"),
)
_PERSISTENCE_CAUSE_BY_PHRASE = (
    (("turn lease",), "turn_lease"),
    (("closed by compression",), "compression_closed"),
    (("being compressed", "compression lease"), "compression"),
    # RPC-wrapped errors lose their exception type; retain the same sidecar/main-file split.
    (("deleted state.db-wal", "deleted state.db-shm"), "deleted_wal"),
    (("was replaced underneath",), "replaced"),
    (_DB_CORRUPTION_MARKERS, "corrupt"),
    (("locked", "busy"), "locked"),
    # A flush rejected by the session-row FK: the row was removed under a live agent (#123583).
    (("foreign key constraint failed",), "session_row_missing"),
)


def classify_persistence_error(exc_or_str) -> str:
    """Coarse cause bucket (PERSISTENCE_ERROR_CAUSES) so the user's guidance
    matches: "locked" = busy, retry; "disk" = full/read-only/permissions;
    "compression" = a live lease refused the write; "compression_closed" = adopt
    the rotated session id; "turn_lease" = fencing, not storage; "corrupt" =
    file damage (repair path, not disk space); "fts_index" = SQLite scoped the
    corruption to the FTS index (the transcript store is not damaged); "replaced" =
    main-file replacement; "deleted_wal" = a retired sidecar generation requiring
    capture inspection; "session_row_missing" = the session row was deleted under a
    live agent (FK rejection; the flush recreates it)."""
    if exc_or_str is None:
        return "unknown"
    # Lease refusals contain neither "locked" nor "busy": match by type first,
    # then by phrase for strings that survived RPC wrapping. Order matters:
    # StateDbReplacedError covers DeletedWalGenerationError; corruption comes
    # BEFORE the lock/disk buckets ("disk image is malformed" contains "disk").
    for exc_type, cause in _PERSISTENCE_CAUSE_BY_TYPE:
        if isinstance(exc_or_str, exc_type):
            return cause
    # Provenance before prose: an FTS-scoped result code (or, without one, an fts5 report
    # naming messages_fts*) is index damage, never whole-file corruption (#97794).
    if is_fts_scoped_corruption_error(exc_or_str):
        return "fts_index"
    if _sqlite_primary_code(exc_or_str) in _SQLITE_LOCK_CODES:
        return "locked"
    if getattr(exc_or_str, "sqlite_errorcode", None) == sqlite3.SQLITE_CONSTRAINT_FOREIGNKEY:
        return "session_row_missing"
    text = str(exc_or_str).lower()
    for markers, cause in _PERSISTENCE_CAUSE_BY_PHRASE:
        if any(marker in text for marker in markers):
            return cause
    if is_disk_full_error(exc_or_str) or any(m in text for m in ("disk", "readonly", "read-only")):
        return "disk"
    return "unknown"