hermes_state_common.py

"""Shared constants and helpers for the SessionDB family of modules.  Lives outside hermes_state so
the mixin modules can import it without a cycle."""

import contextlib
import errno
import json
import logging
import os
import sys
import time
from typing import Any

from hermes_cli.timefmt import EPOCH_MAX, EPOCH_MIN
from hermes_state_keys import CRON_PRINCIPAL_ID
from hermes_state_pidns import persistent_record_pidns_checkable, pid_namespace_id
from agent.skill_commands import AUTO_LOAD_SCAFFOLD_SQL_LIKE, SKILL_EXCERPT_JOINT, SKILL_SCAFFOLD_SQL_LIKE, describe_skill_invocation
from agent.context_compressor import (LEGACY_SUMMARY_PREFIX, SUMMARY_PREFIX, _MERGED_PRIOR_CONTEXT_HEADER,
    _MERGED_SUMMARY_DELIMITER, _SUMMARY_END_MARKER)


# Persisted title provenance: automatic display labels are not user-selected identities.
TITLE_SOURCE_DERIVED = "derived"
TITLE_SOURCE_LLM = "llm"
TITLE_SOURCE_USER = "user"


# Session preview = head of the first user message (shown when a session has no title).  A /skill invocation
# embeds the whole skill body, so scaffolded rows take a wider excerpt (whole message under budget, else head +
# tail where the typed instruction lands) and ``_shape_preview`` recovers ``/work — fix ...`` from it.
_PREVIEW_HEAD_CHARS = 63
_PREVIEW_SCAFFOLD_WINDOW = 400
_PREVIEW_MAX_CHARS = 60


def routed_sessions_setting(key: str, env_var: str) -> Any:
    """``sessions.<key>`` for the profile whose state.db this process is touching.

    ``gateway/run.py`` bridges the LAUNCH profile's ``sessions.*`` into ``env_var`` (the cross-process
    carrier CLI/cron children read). Under a multiplexer a routed turn runs with a HERMES_HOME override
    and that env slot holds the default profile's value, so a served profile with different
    ``sessions.*`` settings must read its own config.yaml. Unscoped: the env bridge, as before.
    Returns ``None`` when neither source sets the key.
    """
    from hermes_constants import get_hermes_home_override

    if get_hermes_home_override():
        try:
            from hermes_cli.config import load_config_readonly
            return (load_config_readonly().get("sessions") or {}).get(key)
        except Exception:
            return None
    return os.environ.get(env_var)


def escape_like(text: str) -> str:
    """Escape LIKE wildcards (``%``, ``_``) so derived text matches literally; pair with ``ESCAPE '\\'``.
    ``_`` is common in branch names/titles/paths and a substring match must not silently widen."""
    return text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


_PREVIEW_CONTENT_SQL = "REPLACE(REPLACE(m.content, X'0A', ' '), X'0D', ' ')"
# User-invoked and gateway auto-load scaffolds both get the head+tail excerpt window:
# the typed request sits at the tail (auto-load) or behind the instruction marker.
# No LIKE wildcards in either prefix, so no ESCAPE clause is needed.
_PREVIEW_SCAFFOLDED_SQL = (
    f"(m.content LIKE '{SKILL_SCAFFOLD_SQL_LIKE}'"
    f" OR m.content LIKE '{AUTO_LOAD_SCAFFOLD_SQL_LIKE}')")
_SQL_WHITESPACE = "CHAR(9) || CHAR(10) || CHAR(13) || CHAR(32)"


def _sql_literal(text: str) -> str:
    return "'" + text.replace("'", "''") + "'"


# ``display_metadata`` flag stamped on the user row a busy-queue accept writes (#111868/#121374
# accept-time durability). The in-memory queue drain re-places an unmarked row at the transcript
# end and deactivates this one; a backend restart discards the queue with nothing left to retire
# the accept-time row, so ``reopen_session`` deactivates still-marked rows (#125577) — the marker
# is exactly what distinguishes a never-drained accept row from a dispatched-then-interrupted
# turn's row (which #123532 deliberately keeps active).
QUEUED_PROMPT_METADATA_KEY = "_queued_prompt"


# ``display_metadata = <this>`` merges a JSON object parameter into the column (its keys win, a null value
# removes a key): a row-addressed rewrite must not wipe a key another writer added meanwhile (a reaction).
_MERGE_DISPLAY_META = "json_patch(CASE WHEN json_valid(display_metadata) THEN display_metadata ELSE '{}' END, COALESCE(?, '{}'))"


def _sql_json_extract(expression: str, path: str) -> str:
    """Build a non-throwing JSON marker lookup for a JSON TEXT column."""

    safe_json = (
        f"(CASE WHEN json_valid({expression}) "
        f"THEN {expression} ELSE json_object() END)"
    )
    return f"json_extract({safe_json}, {_sql_literal(path)})"


def _sql_ltrim_whitespace(expression: str) -> str:
    return f"LTRIM({expression}, {_SQL_WHITESPACE})"


def _sql_trim_whitespace(expression: str) -> str:
    return f"TRIM({expression}, {_SQL_WHITESPACE})"


def _sql_starts_with(expression: str, prefixes: tuple[str, ...]) -> str:
    trimmed = _sql_ltrim_whitespace(expression)
    return "(" + " OR ".join(f"SUBSTR({trimmed}, 1, {len(p)}) = {_sql_literal(p)}" for p in prefixes) + ")"


def _sql_after_marker(marker: str) -> str:
    """``m.content`` after the first occurrence of *marker*."""
    return f"SUBSTR(m.content, INSTR(m.content, {_sql_literal(marker)}) + {len(marker)})"


# Match the whole introduction shared by current and legacy prefixes, so a message that merely starts with the
# bracketed label is not a compaction carrier.
_PREVIEW_LONG_FORM_PREFIX = SUMMARY_PREFIX.split("Do NOT answer", 1)[0]
_PREVIEW_SUMMARY_PREFIXES = (_PREVIEW_LONG_FORM_PREFIX, LEGACY_SUMMARY_PREFIX)
_PREVIEW_STANDALONE_SUMMARY_SQL = _sql_starts_with("m.content", _PREVIEW_SUMMARY_PREFIXES)
_PREVIEW_MERGED_AFTER_SQL = _sql_after_marker(_MERGED_SUMMARY_DELIMITER)
_PREVIEW_MERGED_SUMMARY_SQL = (f"(INSTR(m.content, {_sql_literal(_MERGED_SUMMARY_DELIMITER)}) > 0"
    f" AND {_sql_starts_with(_PREVIEW_MERGED_AFTER_SQL, _PREVIEW_SUMMARY_PREFIXES)})")
_PREVIEW_MERGED_PRIOR_SQL = _sql_trim_whitespace(
    f"SUBSTR(m.content, 1, INSTR(m.content, {_sql_literal(_MERGED_SUMMARY_DELIMITER)}) - 1)")
_PREVIEW_MERGED_PRIOR_LTRIMMED_SQL = _sql_ltrim_whitespace(_PREVIEW_MERGED_PRIOR_SQL)
_PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL = (f"CASE WHEN SUBSTR({_PREVIEW_MERGED_PRIOR_LTRIMMED_SQL}, 1,"
    f" {len(_MERGED_PRIOR_CONTEXT_HEADER)}) = {_sql_literal(_MERGED_PRIOR_CONTEXT_HEADER)}"
    f" THEN {_sql_ltrim_whitespace(f'SUBSTR({_PREVIEW_MERGED_PRIOR_LTRIMMED_SQL}, {len(_MERGED_PRIOR_CONTEXT_HEADER) + 1})')}"
    f" ELSE {_PREVIEW_MERGED_PRIOR_SQL} END")
_PREVIEW_FORCE_USER_REMAINDER_SQL = _sql_after_marker(_SUMMARY_END_MARKER)

# Pure compaction rows are ineligible; force-user-leading and merged carriers only when authentic content survives.
# A display_kind="hidden" row is model-facing scaffolding the gateway never paints; the preview must not paint it either.
_PREVIEW_ELIGIBLE_SQL = (f"(COALESCE(m.display_kind, '') <> 'hidden'"
    f" AND ((NOT {_PREVIEW_STANDALONE_SUMMARY_SQL} AND NOT {_PREVIEW_MERGED_SUMMARY_SQL})"
    f" OR ({_PREVIEW_STANDALONE_SUMMARY_SQL} AND INSTR(m.content, {_sql_literal(_SUMMARY_END_MARKER)}) > 0"
    f" AND LENGTH({_sql_trim_whitespace(_PREVIEW_FORCE_USER_REMAINDER_SQL)}) > 0)"
    f" OR ({_PREVIEW_MERGED_SUMMARY_SQL}"
    f" AND LENGTH({_sql_trim_whitespace(_PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL)}) > 0)))")

# ``_preview_raw`` SELECT for every listing query (scaffolded rows: head + tail around SKILL_EXCERPT_JOINT).
_PREVIEW_RAW_SELECT = (
    f"CASE WHEN {_PREVIEW_STANDALONE_SUMMARY_SQL} THEN {_PREVIEW_FORCE_USER_REMAINDER_SQL}"
    f" WHEN {_PREVIEW_MERGED_SUMMARY_SQL} THEN {_PREVIEW_MERGED_PRIOR_UNWRAPPED_SQL}"
    f" WHEN {_PREVIEW_SCAFFOLDED_SQL} AND LENGTH(m.content) > {_PREVIEW_SCAFFOLD_WINDOW * 2}"
    f" THEN SUBSTR({_PREVIEW_CONTENT_SQL}, 1, {_PREVIEW_SCAFFOLD_WINDOW}) || '{SKILL_EXCERPT_JOINT}'"
    f" || SUBSTR({_PREVIEW_CONTENT_SQL}, -{_PREVIEW_SCAFFOLD_WINDOW})"
    f" WHEN {_PREVIEW_SCAFFOLDED_SQL} THEN SUBSTR({_PREVIEW_CONTENT_SQL}, 1, {_PREVIEW_SCAFFOLD_WINDOW * 2})"
    f" ELSE SUBSTR({_PREVIEW_CONTENT_SQL}, 1, {_PREVIEW_HEAD_CHARS}) END")


def _shape_preview(raw: Any) -> str:
    """Turn a ``_preview_raw`` column into the short preview callers show."""
    text = str(raw or "").strip()
    if not text:
        return ""
    text = text.replace("\n", " ").replace("\r", " ")
    described = describe_skill_invocation(text)
    text = described if described is not None else text.split(SKILL_EXCERPT_JOINT)[0]
    return text[:_PREVIEW_MAX_CHARS] + "..." if len(text) > _PREVIEW_MAX_CHARS else text


def _sql_preview_raw(session_id_expr: str = "s.id") -> str:
    """Correlated ``_preview_raw`` column: the first eligible user message of *session_id_expr*.
    ``INDEXED BY`` walks the session in timestamp order and stops at the first eligible row; with planner
    stats older than idx_messages_session_id, SQLite otherwise reads and sorts the whole session (#119403)."""
    return (f"COALESCE((SELECT {_PREVIEW_RAW_SELECT} FROM messages m INDEXED BY idx_messages_session"
        f" WHERE m.session_id = {session_id_expr} AND m.role = 'user' AND m.content IS NOT NULL"
        f" AND {_PREVIEW_ELIGIBLE_SQL} ORDER BY m.timestamp, m.id LIMIT 1), '') AS _preview_raw")

# ── Session lineage predicates ({a} = sessions alias) ───────────────────────

# /branch child (kept visible, never cascade-deleted): stable marker OR legacy end_reason heuristic.
_BRANCH_CHILD_SQL = (f"{_sql_json_extract('{a}.model_config', '$._branched_from')} IS NOT NULL"
    " OR EXISTS (SELECT 1 FROM sessions p            WHERE p.id = {a}.parent_session_id"
    "            AND p.end_reason = 'branched'            AND {a}.started_at >= p.ended_at)")
_COMPRESSION_CHILD_SQL = ("EXISTS (SELECT 1 FROM sessions p        WHERE p.id = {a}.parent_session_id"
    "        AND p.end_reason = 'compression')")

# 'session_switch' creates no child row today, but pre-marker DBs hold legacy reset children whose parent
# ended that way.  Must stay identical to the recovery fence in find_latest_gateway_session_for_peer.
_RESET_END_REASONS = ("session_reset", "session_switch", "idle", "daily", "suspended", "resume_pending_expired")
_RESET_END_REASONS_SQL = ", ".join(f"'{reason}'" for reason in _RESET_END_REASONS)
# Deliberate conversation boundaries: the reset set plus CLI /new, which ends the predecessor as
# 'new_session' (hermes_cli/cli_session_mixin.py) without a reset child row.  A compression rotation must
# never heal one of these (#106459); tools/session_search_tool.py derives its fresh-reset set from it.
_BOUNDARY_END_REASONS = frozenset(_RESET_END_REASONS) | {"new_session"}

# Accidental end reasons recovery treats as resumable (website/docs/developer-guide/gateway-session-lifecycle.md); single source of truth for
# recovery SQL and SessionDB.RECOVERABLE_END_REASONS.  superseded_by_resume = sentinel-parked runtime replaced
# by a fresh session.resume; startup_orphan_reap = dead-gateway sweep, same class as ws_orphan_reap but kept
# distinct for forensics.
_RECOVERABLE_END_REASONS = ("agent_close", "ws_orphan_reap", "superseded_by_resume", "startup_orphan_reap")
# Startup sweep of rows orphaned by a dead gateway process (#65194): the in-process ws-orphan grace timer
# died with the process, so the row was closed at the next boot instead.
_RECOVERABLE_END_REASONS_SQL = ", ".join(f"'{reason}'" for reason in _RECOVERABLE_END_REASONS)

# End reasons written by AUTOMATIC cleanup (shutdown, orphan reapers, idle/LRU eviction), not a deliberate
# conversation boundary: "some runtime went away", so a writer that can prove liveness (e.g. a compression
# rotation holding the lease) may clear it.  Recoverable set plus the TUI gateway's automatic reasons.
# Superset of the recoverable set: those are already resumable accidents; the extra TUI reasons are the same
# accident class but were historically only known to tui_gateway's _AUTOMATIC_SESSION_END_REASONS. See
# #88197.
_AUTOMATIC_END_REASONS = frozenset(_RECOVERABLE_END_REASONS) | {
    "tui_shutdown", "ws_disconnect", "idle_timeout", "lru_evict"}


def is_automatic_end_reason(reason) -> bool:
    """True when *reason* is an automatic-cleanup end stamp; compression-liveness sites must call this.

    Single owner of the "accidental vs deliberate end" predicate — every compression-liveness site must call
    this instead of re-implementing the reason taxonomy (#88197, never-patch-predicates).
    """
    return isinstance(reason, str) and reason in _AUTOMATIC_END_REASONS


def _legacy_reset_child_sql(alias: str, reasons_sql: str) -> str:
    """Pre-marker reset-continuation heuristic: child rides its parent's exact non-empty routing key and the
    parent ended at a reset boundary.  Shared by ``_RESET_CHILD_SQL`` and ``reopen_session()`` so the two
    cannot drift; ``reasons_sql`` is a literal or placeholder list."""
    return (f"EXISTS (SELECT 1 FROM sessions p            WHERE p.id = {alias}.parent_session_id"
        f"            AND p.end_reason IN ({reasons_sql})            AND {alias}.session_key IS NOT NULL"
        f"            AND {alias}.session_key != ''            AND {alias}.session_key = p.session_key)")


def _non_continuation_child_sql(child: str = "", parent: str = "?") -> str:
    """``  AND ...`` clauses rejecting children that are NOT compression continuations of *parent*
    (branch/delegate/reset forks, tool sessions).  Markers are bound to the parent id: continuations
    inherit ``model_config`` verbatim, so a marker naming another row is inherited, not a fork.
    ``child`` is the column prefix (``""``, ``"c."``); single owner so prune and compression agree."""
    return "".join(
        f"  AND COALESCE({_sql_json_extract(f'{child}model_config', f'$.{marker}')}, '') != {parent}\n"
        for marker in ("_branched_from", "_delegate_from", "_reset_from")
    ) + f"  AND COALESCE({child}source, '') != 'tool'\n"


# A reset starts a separate user-visible conversation though rows keep parent_session_id for lineage.
# Stable marker, or the same-key fallback for pre-marker rows (exact key keeps subagent children out).
_RESET_CHILD_SQL = (f"{_sql_json_extract('{a}.model_config', '$._reset_from')} IS NOT NULL"
    " OR " + _legacy_reset_child_sql("{a}", _RESET_END_REASONS_SQL))

# Picker-visible rows: roots + branch/reset children (not subagent runs or compression continuations).
_LISTABLE_CHILD_SQL = (f"(s.parent_session_id IS NULL OR {_BRANCH_CHILD_SQL.format(a='s')}"
    f" OR {_RESET_CHILD_SQL.format(a='s')})")


def _ephemeral_child_sql(alias: str = "s") -> str:
    """Subagent runs, not branch, reset, or compression children."""
    return (f"({alias}.parent_session_id IS NOT NULL AND NOT ({_BRANCH_CHILD_SQL.format(a=alias)})"
        f" AND NOT ({_COMPRESSION_CHILD_SQL.format(a=alias)}) AND NOT ({_RESET_CHILD_SQL.format(a=alias)}))")


_SQL_IN_WINDOW = f"BETWEEN {EPOCH_MIN!r} AND {EPOCH_MAX!r}"


def _sql_in_window(expr: str) -> str:
    """*expr* when it is inside the ``coerce_epoch`` window, else NULL."""
    return f"(SELECT _win.v FROM (SELECT {expr} AS v) _win WHERE _win.v {_SQL_IN_WINDOW})"


def _sql_freshest_of(activity: str, session_id_expr: str, started: str) -> str:
    """Freshest of *activity* and the latest message timestamp for *session_id_expr*, else *started*.
    Heartbeats are rate-limited (~60s) so ``last_activity_at`` can lag a newer message; never use it alone.
    Cells outside the ``coerce_epoch`` window (garbage doubles salvaged from a damaged page, TEXT) are
    skipped, fallback included, or one bad row pins the session's recency to ``5e+246`` (#91536); a
    session with no trusted cell at all is NULL."""
    # Pinned for the same reason as _sql_preview_raw (#119403).
    msg_max = (f"(SELECT MAX(_act_m.timestamp) FROM messages _act_m INDEXED BY idx_messages_session"
        f" WHERE _act_m.session_id = {session_id_expr}"
        f" AND _act_m.timestamp {_SQL_IN_WINDOW})")
    return (f"COALESCE((SELECT MAX(_act_v.v) FROM (SELECT {activity} AS v UNION ALL SELECT {msg_max}) _act_v"
        f" WHERE _act_v.v {_SQL_IN_WINDOW}), {_sql_in_window(started)})")


def _sql_session_last_active(alias: str = "s") -> str:
    """Session recency expression for a ``sessions {alias}`` row."""
    return _sql_freshest_of(f"{alias}.last_activity_at", f"{alias}.id", f"{alias}.started_at")


def _sql_session_last_active_by_id(session_id_expr: str) -> str:
    """Same freshest-of expression keyed by a session-id SQL expression."""
    return _sql_freshest_of(
        f"(SELECT last_activity_at FROM sessions _act_s WHERE _act_s.id = {session_id_expr})", session_id_expr,
        f"(SELECT started_at FROM sessions _act_s WHERE _act_s.id = {session_id_expr})")


# v32 (#140315): no data migration. #106742 added runtime_epoch, session_admissions, worker_executions,
# worker_receipts and three columns to SCHEMA_SQL under v31. Drift is detected by the schema read probe
# (schema_read_probe_statements), never by this number, and its catch-up is the declarative
# executescript(SCHEMA_SQL) + _reconcile_columns() every writable open already runs (additive,
# idempotent). The bump only records that the declared shape changed; a new table or column needs no bump.
# v33: data-only — re-tag delegate subagent rows that lost ``_delegate_from`` (see _run_data_migrations).
SCHEMA_VERSION = 33

# Auto-maintenance VACUUMs only above this freelist fraction; below it a rewrite costs more I/O than it returns.
# Auto-maintenance only VACUUMs when at least this fraction of the database file is reclaimable (``PRAGMA
# freelist_count / PRAGMA page_count``). Below it a full rewrite costs more I/O than it returns — pruning a
# handful of small sessions on a dense multi-GB state.db should never rewrite the whole file to reclaim a
# few MB (#54189). Composes with ``min_vacuum_interval_days``.
AUTO_VACUUM_MIN_FREELIST_RATIO = 0.25

# FTS storage-layout version, tracked INDEPENDENTLY of SCHEMA_VERSION in the
# state_meta key ``fts_storage_version``. The main schema version advances
# freely on open (so future migrations always land); the FTS *layout* only
# reaches the current version when a DB is either born fresh or explicitly
# optimized via ``hermes sessions optimize-storage``. A legacy DB sits at
# layout 0 (marker absent) with a working inline index until the user opts in.
#   1 = v23 external-content layout with a tool-row-excluded trigram
#   2 = trigram also excludes structured tool_calls JSON
#   3 = messages_fts source aligned to a stable projection view
#       (``messages_fts_src``): always-truncate tool rows to the prefix, no
#       moving high-water boundary. The external-content source now reads
#       back EXACTLY what the triggers indexed, so the rank=1
#       'integrity-check' probe cannot drift from the stored index (the
#       recurring fts5 "checksum mismatch" / leaked-token failures).
FTS_STORAGE_VERSION = 3

# Tool results are often multi-megabyte machine payloads. The base FTS index
# stores only a bounded prefix of every tool row; tool rows are skipped by
# default in search, and explicit tool-only search uses a LIKE fallback over
# the full stored content, so no search capability is lost. The projection
# below is STABLE — it depends only on the row being written, never on
# mutable ``state_meta`` markers — which is what keeps the external-content
# integrity checker and the trigger 'delete'/'update' commands in agreement
# with the stored index forever.
FTS_TOOL_CONTENT_PREFIX_CHARS = 8_192


def _fts_indexed_content_sql(alias: str) -> str:
    return f"""CASE WHEN {alias}.role = 'tool'
         THEN substr(COALESCE({alias}.content, ''), 1, {FTS_TOOL_CONTENT_PREFIX_CHARS})
         ELSE {alias}.content END"""


_FTS_NEW_INDEXED_CONTENT_SQL = _fts_indexed_content_sql("new")
_FTS_OLD_INDEXED_CONTENT_SQL = _fts_indexed_content_sql("old")

# Cap on user-controlled FTS5 query input before sanitizer processing.
MAX_FTS5_QUERY_CHARS = 2_048


def stat_db_file_identity(path) -> tuple[int, int] | None:
    """``(st_dev, st_ino)`` for *path*, or None.  st_ino=0 (Windows, some network FS) would false-positive
    every replaced-file check, so it counts as unknown."""
    try:
        st = os.stat(path)
    except OSError:
        return None
    return (st.st_dev, st.st_ino) if st.st_dev and st.st_ino else None


# Row probes shared by the messages / compression mixins.
_ENDED_ROW_SQL = "SELECT ended_at, end_reason FROM sessions WHERE id = ?"
_COMPRESSION_LOCK_ROW_SQL = "SELECT holder, expires_at FROM compression_locks WHERE session_id = ?"


def _ended_by_compression(row) -> bool:
    return row is not None and row["ended_at"] is not None and row["end_reason"] == "compression"


def _placeholders(items) -> str:
    """``?,?,?`` for one bound parameter per element of *items* (a sequence or an int count)."""
    return ",".join("?" for _ in range(items if isinstance(items, int) else len(items)))


# A first prompt is durable in ``session_admissions`` from admission on, while its transcript row lands
# only at turn start; a restart in between leaves the admission ``unknown`` and the session at 0
# messages. Only a session with neither messages nor an admission is an empty launch stub.
_NOT_A_LAUNCH_STUB_SQL = (
    "(s.message_count > 0 OR EXISTS (SELECT 1 FROM session_admissions a WHERE a.target_session_id = s.id))")


def launch_stub_filter_sql(db) -> str:
    """``sessions s`` predicate that drops empty launch stubs (the CLI crashed or a resume failed before
    the first turn). A read-only open does not migrate, so a store from before the runtime tables may
    lack ``session_admissions``; messages alone decide there."""
    if db._read_all("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'session_admissions'"):
        return _NOT_A_LAUNCH_STUB_SQL
    return "s.message_count > 0"


def _source_match_clause(sources, local_created_sources=None) -> tuple[str, list]:
    """``sessions s`` source predicate. With ``local_created_sources``, a ``source='local'`` row
    matches only when ``created_source`` is one of those surfaces; the placeholder ``'unknown'``
    is not a surface and stays unmatched until an upsert stamps the real creator."""
    labels = list(sources or [])
    if not labels:
        return "", []
    surfaces = list(local_created_sources or ())
    if not surfaces:
        return f"s.source IN ({_placeholders(labels)})", labels
    non_local = [item for item in labels if item != "local"]
    local_sql = f"(s.source = 'local' AND s.created_source IN ({_placeholders(surfaces)}))"
    if not non_local:
        return local_sql, list(surfaces)
    return (
        f"(s.source IN ({_placeholders(non_local)}) OR {local_sql})",
        [*non_local, *surfaces],
    )


# Ids per ``IN (?,...)`` list: SQLite caps bound parameters at SQLITE_MAX_VARIABLE_NUMBER (999 on builds
# < 3.32, 32766 after); a bulk prune of a cron-heavy store bound tens of thousands of ids into one list and
# died with "too many SQL variables". Every IN-list over session ids goes through ``_id_chunks``.
_SQL_IN_CHUNK = 900


def _id_chunks(ids, size: int = _SQL_IN_CHUNK):
    """Yield *ids* (any iterable) as lists of at most *size* elements."""
    ids = list(ids)
    for start in range(0, len(ids), size):
        yield ids[start:start + size]


def emit_session_deleted(db, deleted_ids, *, reason, primary=None, sessions_dir=None):
    """Fire ``on_session_delete`` after a committed delete or deliberate archive (#124511).

    Call right after the write transaction commits (before follow-up cleanup that can raise), with
    the ids it removed or, for ``reason="archive"``, hid. *primary* is the session the caller asked
    for; without it the first id stands in (bulk sweeps). ``deleted_ids`` rides along when more
    than one row went. Dispatch runs under the store's own profile home, so a process touching
    another profile's store (``sessions repair-profiles``) reaches THAT profile's plugins.
    Late-bound and fail-open: never breaks the delete.
    """
    ids = sorted({sid for sid in (deleted_ids or ()) if isinstance(sid, str) and sid})
    if not ids:
        return
    try:
        from hermes_cli.lifecycle import invoke_hook
        from hermes_constants import (
            get_hermes_home, profile_name_for_home, reset_hermes_home_override, set_hermes_home_override,
        )

        home = os.path.dirname(os.path.abspath(os.fspath(db.db_path)))
        profile = profile_name_for_home(home)
        payload: dict = {"session_id": primary if primary in ids else ids[0], "reason": reason, "profile": profile}
        if len(ids) > 1:
            payload["deleted_ids"] = ids
        if sessions_dir is not None:
            payload["sessions_dir"] = str(sessions_dir)
        rebind = profile is not None and os.path.realpath(home) != os.path.realpath(get_hermes_home())
        token = set_hermes_home_override(home) if rebind else None
        try:
            invoke_hook("on_session_delete", **payload)
        finally:
            if token is not None:
                reset_hermes_home_override(token)
    except Exception:
        logger.warning("on_session_delete dispatch failed", exc_info=True)


_FTS_TRIGGERS = ("messages_fts_insert", "messages_fts_delete", "messages_fts_update",
                 "messages_fts_trigram_insert", "messages_fts_trigram_delete", "messages_fts_trigram_update")

SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS schema_version (
    version INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS system_prompts (
    hash TEXT PRIMARY KEY,
    prompt TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS sessions (
    id TEXT PRIMARY KEY,
    runtime_revision INTEGER NOT NULL DEFAULT 0,
    runtime_generation INTEGER NOT NULL DEFAULT 0,
    source TEXT NOT NULL,
    created_source TEXT,
    user_id TEXT,
    session_key TEXT,
    chat_id TEXT,
    chat_type TEXT,
    thread_id TEXT,
    display_name TEXT,
    origin_json TEXT,
    expiry_finalized INTEGER DEFAULT 0,
    model TEXT,
    model_config TEXT,
    system_prompt TEXT,
    system_prompt_hash TEXT,
    parent_session_id TEXT,
    started_at REAL NOT NULL,
    ended_at REAL,
    end_reason TEXT,
    message_count INTEGER DEFAULT 0,
    tool_call_count INTEGER DEFAULT 0,
    input_tokens INTEGER DEFAULT 0,
    output_tokens INTEGER DEFAULT 0,
    cache_read_tokens INTEGER DEFAULT 0,
    cache_write_tokens INTEGER DEFAULT 0,
    reasoning_tokens INTEGER DEFAULT 0,
    cwd TEXT,
    git_branch TEXT,
    git_repo_root TEXT,
    git_metadata_generation INTEGER NOT NULL DEFAULT 0,
    billing_provider TEXT,
    billing_base_url TEXT,
    billing_mode TEXT,
    estimated_cost_usd REAL,
    actual_cost_usd REAL,
    cost_status TEXT,
    cost_source TEXT,
    pricing_version TEXT,
    title TEXT,
    title_source TEXT,
    last_activity_at REAL,
    last_activity_description TEXT,
    last_activity_provenance TEXT,
    api_call_count INTEGER DEFAULT 0,
    handoff_state TEXT,
    handoff_platform TEXT,
    handoff_error TEXT,
    compression_failure_cooldown_until REAL,
    compression_failure_error TEXT,
    compression_fallback_streak INTEGER NOT NULL DEFAULT 0,
    compression_ineffective_count INTEGER NOT NULL DEFAULT 0,
    compression_recovery_deadline REAL,
    compression_overload_streak INTEGER NOT NULL DEFAULT 0,
    profile_name TEXT,
    transport_profile TEXT,
    rewind_count INTEGER NOT NULL DEFAULT 0,
    archived INTEGER NOT NULL DEFAULT 0,
    auto_archived INTEGER NOT NULL DEFAULT 0,
    pinned INTEGER NOT NULL DEFAULT 0,
    hidden INTEGER NOT NULL DEFAULT 0,
    last_read_at REAL,
    tool_names TEXT,
    transcript_revision INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (parent_session_id) REFERENCES sessions(id),
    FOREIGN KEY (system_prompt_hash) REFERENCES system_prompts(hash)
);

CREATE TABLE IF NOT EXISTS messages (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id TEXT NOT NULL REFERENCES sessions(id),
    role TEXT NOT NULL,
    content TEXT,
    tool_call_id TEXT,
    tool_calls TEXT,
    tool_name TEXT,
    effect_disposition TEXT,
    timestamp REAL NOT NULL,
    token_count INTEGER,
    finish_reason TEXT,
    reasoning TEXT,
    reasoning_content TEXT,
    reasoning_details TEXT,
    codex_reasoning_items TEXT,
    codex_message_items TEXT,
    platform_message_id TEXT,
    observed INTEGER DEFAULT 0,
    _compressed_summary INTEGER NOT NULL DEFAULT 0,
    active INTEGER NOT NULL DEFAULT 1,
    compacted INTEGER NOT NULL DEFAULT 0,
    api_content TEXT,
    display_kind TEXT,
    display_metadata TEXT,
    display_identity BLOB,
    display_order INTEGER,
    message_uid TEXT,
    absorbed_message_uids TEXT,
    tool_call_uids TEXT,
    tool_call_uid TEXT
);

CREATE TABLE IF NOT EXISTS session_model_usage (
    session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
    model TEXT NOT NULL,
    billing_provider TEXT NOT NULL DEFAULT '',
    billing_base_url TEXT NOT NULL DEFAULT '',
    billing_mode TEXT NOT NULL DEFAULT '',
    task TEXT NOT NULL DEFAULT '',
    api_call_count INTEGER NOT NULL DEFAULT 0,
    input_tokens INTEGER NOT NULL DEFAULT 0,
    output_tokens INTEGER NOT NULL DEFAULT 0,
    cache_read_tokens INTEGER NOT NULL DEFAULT 0,
    cache_write_tokens INTEGER NOT NULL DEFAULT 0,
    reasoning_tokens INTEGER NOT NULL DEFAULT 0,
    estimated_cost_usd REAL NOT NULL DEFAULT 0,
    actual_cost_usd REAL NOT NULL DEFAULT 0,
    cost_status TEXT,
    cost_source TEXT,
    first_seen REAL,
    last_seen REAL,
    PRIMARY KEY (session_id, model, billing_provider, billing_base_url, billing_mode, task)
);

CREATE TABLE IF NOT EXISTS runtime_epoch (
    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
    epoch INTEGER NOT NULL CHECK (epoch > 0),
    instance_id TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS session_admissions (
    seq INTEGER PRIMARY KEY AUTOINCREMENT,
    admission_id TEXT NOT NULL UNIQUE,
    request_id TEXT NOT NULL,
    principal_id TEXT NOT NULL,
    target_session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE RESTRICT,
    lineage_json TEXT NOT NULL,
    payload_json TEXT NOT NULL,
    payload_digest TEXT NOT NULL,
    intent TEXT NOT NULL CHECK (intent IN ('queue','steer','redirect')),
    status TEXT NOT NULL CHECK (status IN ('queued','started','unknown','terminal')),
    outcome TEXT,
    owner_epoch INTEGER,
    generation INTEGER,
    UNIQUE (principal_id, target_session_id, request_id),
    CHECK (status != 'started' OR (owner_epoch IS NOT NULL AND generation IS NOT NULL))
);
CREATE INDEX IF NOT EXISTS session_admissions_pending
    ON session_admissions(target_session_id, status, seq);
-- Cron run history finds a job's unified-runtime runs by the scheduler principal's request ids
-- (``cron:<job>:<execution>``); without this a lookup walked
-- every cron admission of every job. Partial: interactive turns, the bulk of admissions, pay
-- nothing for it. The constant leading column gives the planner equality + range over a covering
-- index, so it beats the UNIQUE autoindex's bare principal_id prefix even without ANALYZE stats.
-- The WHERE is built from CRON_PRINCIPAL_ID, the literal the run queries inline, so they cannot drift.
CREATE INDEX IF NOT EXISTS idx_session_admissions_cron_request
    ON session_admissions(principal_id, request_id, target_session_id) WHERE principal_id = '""" + CRON_PRINCIPAL_ID + """';
CREATE TABLE IF NOT EXISTS worker_executions (
    execution_id TEXT PRIMARY KEY,
    session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE RESTRICT,
    kind TEXT NOT NULL CHECK (kind IN ('cron','child','compute','kanban')),
    owner_epoch INTEGER NOT NULL,
    generation INTEGER NOT NULL,
    status TEXT NOT NULL CHECK (status IN ('registered','running','unknown','terminal')),
    adoption_digest TEXT NOT NULL,
    last_sequence INTEGER NOT NULL DEFAULT 0 CHECK (last_sequence >= 0)
);
-- Lineage-wide worker exclusion (claim/register/reset) probes every row of a conversation's lineage
-- inside the write transaction; without this each probe scanned the whole (only-growing) table.
CREATE INDEX IF NOT EXISTS idx_worker_executions_session_status
    ON worker_executions(session_id, status);
CREATE TABLE IF NOT EXISTS worker_receipts (
    execution_id TEXT NOT NULL REFERENCES worker_executions(execution_id) ON DELETE RESTRICT,
    sequence INTEGER NOT NULL CHECK (sequence > 0),
    payload_digest TEXT NOT NULL,
    result_json TEXT NOT NULL,
    PRIMARY KEY (execution_id, sequence)
);

CREATE TABLE IF NOT EXISTS state_meta (
    key TEXT PRIMARY KEY,
    value TEXT
);

CREATE TABLE IF NOT EXISTS gateway_routing (
    scope TEXT NOT NULL DEFAULT '',
    session_key TEXT NOT NULL,
    entry_json TEXT NOT NULL,
    updated_at REAL NOT NULL,
    PRIMARY KEY (scope, session_key)
);

CREATE TABLE IF NOT EXISTS gateway_hygiene_state (
    session_key TEXT PRIMARY KEY,
    failure_streak INTEGER NOT NULL DEFAULT 0
);

-- Monotonic conversation generation per routing peer (#96811).
--
-- A host-declared conversation key (X-Hermes-Session-Key / build_session_key)
-- is per-CHAT and outlives any single conversation on it, so the prompt-cache
-- affinity scope derived from it must be qualified by which conversation is
-- currently live. Deriving that from the session rows themselves
-- (COUNT/MAX over _RESET_END_REASONS boundaries) cannot prove non-reuse:
-- delete_session() and bulk pruning remove ended rows, so an aggregate can
-- return a pair it already emitted and hand a new conversation a retired
-- affinity identity.
--
-- This counter lives outside prunable session history and only ever
-- increments, once per boundary actually written, so a generation can never
-- be reused for a peer even if every session row behind it is deleted.
--
-- These rows are deliberately NEVER garbage-collected, including when every
-- session row for the peer is gone. Collecting one resets that peer to "no
-- generation", so its next boundary writes generation = 1 again and re-issues
-- a gwk_ scope a retired conversation already used — exactly the ABA this
-- table exists to close. Do not add it to delete_session()'s cascade or to any
-- prune sweep. One (TEXT, TEXT, INTEGER) row per routing peer is the intended,
-- bounded cost.
CREATE TABLE IF NOT EXISTS conversation_generations (
    source TEXT NOT NULL,
    session_key TEXT NOT NULL,
    generation INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (source, session_key)
);

-- Per-backend liveness heartbeat (#94895). Each serve / tui_gateway process
-- registers a row at startup and refreshes ``last_heartbeat`` periodically.
-- The startup orphan sweep (sessions.startup_orphan_reap) consults this
-- table to avoid reaping rows whose owning backend is still alive but
-- just idle (multi-backend state.db shared by isolated serve processes).
-- A backend whose ``last_heartbeat`` is older than the heartbeat staleness
-- window is treated as dead; rows without ANY matching heartbeat fall back
-- to the original staleness predicate so legacy deployments keep working.
CREATE TABLE IF NOT EXISTS gateway_heartbeats (
    backend_id TEXT PRIMARY KEY,
    pid INTEGER NOT NULL,
    started_at REAL NOT NULL,
    last_heartbeat REAL NOT NULL,
    profile TEXT NOT NULL DEFAULT '',
    host TEXT NOT NULL DEFAULT ''
);

CREATE TABLE IF NOT EXISTS compression_locks (
    session_id TEXT PRIMARY KEY,
    holder TEXT NOT NULL,
    acquired_at REAL NOT NULL,
    expires_at REAL NOT NULL
);

CREATE TABLE IF NOT EXISTS session_turn_leases (
    conversation_id TEXT PRIMARY KEY,
    holder TEXT NOT NULL,
    acquired_at REAL NOT NULL,
    expires_at REAL NOT NULL
);

CREATE TABLE IF NOT EXISTS async_delegations (
    delegation_id TEXT PRIMARY KEY,
    origin_session TEXT NOT NULL,
    origin_ui_session_id TEXT NOT NULL DEFAULT '',
    parent_session_id TEXT,
    state TEXT NOT NULL,
    dispatched_at REAL NOT NULL,
    completed_at REAL,
    updated_at REAL NOT NULL,
    event_json TEXT,
    result_json TEXT,
    delivery_state TEXT NOT NULL DEFAULT 'pending',
    delivery_attempts INTEGER NOT NULL DEFAULT 0,
    delivered_at REAL,
    owner_pid INTEGER,
    owner_started_at INTEGER,
    task_json TEXT,
    -- Owner-side execution fence (unified runtime).
    owner_execution_id TEXT,
    delivery_claim TEXT,
    delivery_claimed_at REAL,
    -- Mirrors the delegation tool's own CREATE TABLE (tools/async_delegation.py
    -- _initialize_schema). Keeping the canonical fresh-install shape identical
    -- to the tool's avoids a silent schema drift: the tool's lazy
    -- ALTER TABLE ADD COLUMN used to be the only source of this column, so two
    -- databases at the same schema_version had different
    -- async_delegations shapes depending on whether the delegation tool had
    -- ever run, breaking rebuild/replay pipelines that reconstruct state.db
    -- from the canonical schema (#94691).
    origin_session_id TEXT NOT NULL DEFAULT ''
);

CREATE INDEX IF NOT EXISTS idx_sessions_source ON sessions(source);
CREATE INDEX IF NOT EXISTS idx_sessions_source_id ON sessions(source, id);
CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id);
CREATE INDEX IF NOT EXISTS idx_sessions_started ON sessions(started_at DESC);
CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, timestamp);
CREATE INDEX IF NOT EXISTS idx_messages_session_id ON messages(session_id, id);
-- Partial index for the Insights assistant tool-call scan
-- (agent/insights.py _get_tool_usage / _get_skill_usage): those queries filter
-- messages by role='assistant' AND tool_calls IS NOT NULL, a small fraction of
-- rows on a large state.db. role and tool_calls are base columns, so this can
-- live in SCHEMA_SQL rather than DEFERRED_INDEX_SQL.
CREATE INDEX IF NOT EXISTS idx_messages_assistant_calls_by_session
    ON messages(session_id)
    WHERE role = 'assistant' AND tool_calls IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_compression_locks_expires ON compression_locks(expires_at);
CREATE INDEX IF NOT EXISTS idx_session_turn_leases_expires ON session_turn_leases(expires_at);
CREATE INDEX IF NOT EXISTS idx_session_model_usage_session ON session_model_usage(session_id);
CREATE INDEX IF NOT EXISTS idx_session_model_usage_model ON session_model_usage(model);
CREATE INDEX IF NOT EXISTS idx_async_delegations_delivery
    ON async_delegations(delivery_state, completed_at);
"""

# Indexes on later-added columns must run AFTER _reconcile_columns(), or executescript fails on legacy DBs.
# sessions_runtime_ledger_delete_guard: every delete path in this release retires a session's ledger rows
# (session_admissions / worker_executions, ON DELETE RESTRICT) first, so only a release that predates the
# gateway runtime reaches it. Such a release (schema_version 31 too, and it never refuses a newer schema)
# otherwise dies on the bare FK with no hint. It prints the RAISE text verbatim and buckets errors by phrase,
# so the text must stay free of the words it reads as storage damage, lock contention or disk trouble.
DEFERRED_INDEX_SQL = """
DROP TRIGGER IF EXISTS sessions_runtime_ledger_delete_guard;
CREATE TRIGGER IF NOT EXISTS sessions_runtime_ledger_delete_guard
BEFORE DELETE ON sessions
WHEN EXISTS (SELECT 1 FROM session_admissions WHERE target_session_id = old.id)
  OR EXISTS (SELECT 1 FROM worker_executions WHERE session_id = old.id)
BEGIN
    SELECT RAISE(ABORT, 'Refused to delete a session that has gateway runtime history: state.db was upgraded by a newer Hermes, and this older Hermes cannot delete such sessions. Downgrading is unsupported. Run hermes update to go back to the newer release, or restore the pre-update snapshot: https://hermes-agent.nousresearch.com/docs/getting-started/updating#rolling-back-to-an-older-release');
END;
DROP TRIGGER IF EXISTS messages_transcript_revision_update;
CREATE TRIGGER IF NOT EXISTS messages_transcript_revision_update
AFTER UPDATE OF role, content, api_content, tool_calls, tool_call_id, tool_name, reasoning, reasoning_content,
    reasoning_details ON messages
BEGIN
    -- An in-place rewrite (prompt expansion, multimodal context, a redaction) moves neither the
    -- active row count nor the highest id; this write-side revision is what a transcript-derived
    -- cache keys on (conversation_revision). Whoever wrote the row, old builds included.
    UPDATE sessions SET transcript_revision = transcript_revision + 1 WHERE id = new.session_id;
END;
DROP TRIGGER IF EXISTS messages_display_revision_update;
CREATE TRIGGER IF NOT EXISTS messages_display_revision_update
AFTER UPDATE OF codex_reasoning_items, codex_message_items, display_kind, display_metadata ON messages
BEGIN
    -- A reaction, a display rewrite or checkpoint pruning: a transcript poller's cursor
    -- (transcript_cursor_state) keys on these too. Its own name, so an older build's open, which
    -- recreates the trigger above with that build's column list, leaves this one in place.
    UPDATE sessions SET transcript_revision = transcript_revision + 1 WHERE id = new.session_id;
END;
CREATE INDEX IF NOT EXISTS idx_messages_session_active
    ON messages(session_id, active, timestamp);
CREATE INDEX IF NOT EXISTS idx_messages_display_page
    ON messages(session_id, display_order, active DESC, id DESC)
    WHERE active = 1 OR compacted = 1;
CREATE INDEX IF NOT EXISTS idx_messages_display_backfill
    ON messages(session_id) WHERE (display_order IS NULL OR display_identity IS NULL)
    AND (active = 1 OR compacted = 1);
CREATE INDEX IF NOT EXISTS idx_messages_display_identity
    ON messages(session_id, display_identity, display_order)
    WHERE display_identity IS NOT NULL AND (active = 1 OR compacted = 1);
DROP TRIGGER IF EXISTS messages_message_uid_insert;
CREATE TRIGGER IF NOT EXISTS messages_message_uid_insert
AFTER INSERT ON messages WHEN new.message_uid IS NULL
BEGIN
    -- Every row carries a message_uid, whoever wrote it: a build that predates the column binds NULL, so
    -- the store mints one on its behalf (the current build always binds a uid; this never fires for it).
    -- The note lives inside the body: a comment before DROP/CREATE would defeat the settled-trigger skip.
    UPDATE messages SET message_uid = lower(hex(randomblob(16))) WHERE id = new.id;
END;
DROP TRIGGER IF EXISTS messages_display_order_insert;
CREATE TRIGGER IF NOT EXISTS messages_display_order_insert
AFTER INSERT ON messages WHEN new.display_order IS NULL
BEGIN
    UPDATE messages SET display_order = COALESCE((
        SELECT display_order FROM messages
        WHERE session_id = new.session_id AND id <> new.id
          AND (active = 1 OR compacted = 1)
          AND display_identity = new.display_identity AND display_order IS NOT NULL
        ORDER BY display_order LIMIT 1
    ), new.id) WHERE id = new.id;
END;
DROP TRIGGER IF EXISTS messages_display_visibility_update;
CREATE TRIGGER IF NOT EXISTS messages_display_visibility_update
AFTER UPDATE OF active, compacted ON messages
WHEN (new.active = 1 OR new.compacted = 1) <> (old.active = 1 OR old.compacted = 1)
BEGIN
    UPDATE messages SET display_order = MIN(new.id, COALESCE((
        SELECT display_order FROM messages
        WHERE session_id = new.session_id AND id <> new.id
          AND (active = 1 OR compacted = 1)
          AND display_identity = new.display_identity AND display_order IS NOT NULL
        ORDER BY display_order LIMIT 1
    ), new.id)) WHERE id = new.id
      AND (new.active = 1 OR new.compacted = 1);
    UPDATE messages SET display_order = (SELECT display_order FROM messages WHERE id = new.id)
    WHERE session_id = new.session_id AND id <> new.id AND (active = 1 OR compacted = 1)
      AND display_identity = new.display_identity
      AND (new.active = 1 OR new.compacted = 1);
    UPDATE messages SET display_order = (
        SELECT MIN(peer.id) FROM messages AS peer
        WHERE peer.session_id = old.session_id AND (peer.active = 1 OR peer.compacted = 1)
          AND peer.display_identity = old.display_identity
    ) WHERE session_id = old.session_id AND (active = 1 OR compacted = 1)
      AND display_identity = old.display_identity
      AND NOT (new.active = 1 OR new.compacted = 1);
END;
DROP TRIGGER IF EXISTS messages_display_identity_update;
CREATE TRIGGER IF NOT EXISTS messages_display_identity_update
AFTER UPDATE OF role, content, timestamp, tool_call_id, tool_calls, tool_name,
                display_kind ON messages
WHEN new.role IS NOT old.role
  OR new.content IS NOT old.content
  OR new.timestamp IS NOT old.timestamp
  OR new.tool_call_id IS NOT old.tool_call_id
  OR new.tool_calls IS NOT old.tool_calls
  OR new.tool_name IS NOT old.tool_name
  OR new.display_kind IS NOT old.display_kind
BEGIN
    UPDATE messages SET display_identity = NULL, display_order = NULL
    WHERE id = new.id OR (
        session_id = old.session_id AND display_identity = old.display_identity
        AND (active = 1 OR compacted = 1)
    );
END;
DROP TRIGGER IF EXISTS messages_display_identity_delete;
CREATE TRIGGER IF NOT EXISTS messages_display_identity_delete
AFTER DELETE ON messages WHEN old.active = 1 OR old.compacted = 1
BEGIN
    UPDATE messages SET display_order = (
        SELECT MIN(peer.id) FROM messages AS peer
        WHERE peer.session_id = old.session_id AND (peer.active = 1 OR peer.compacted = 1)
          AND peer.display_identity = old.display_identity
    ) WHERE session_id = old.session_id AND (active = 1 OR compacted = 1)
      AND display_identity = old.display_identity;
END;
CREATE INDEX IF NOT EXISTS idx_messages_active_null
    ON messages(active) WHERE active IS NULL;
-- Logical-identity recovery (agent.transcript_repair._active_logical_message_row) runs inside the write lock.
CREATE INDEX IF NOT EXISTS idx_messages_session_uid
    ON messages(session_id, message_uid) WHERE active = 1;
CREATE INDEX IF NOT EXISTS idx_sessions_session_key
    ON sessions(session_key, started_at DESC);
CREATE INDEX IF NOT EXISTS idx_sessions_gateway_peer
    ON sessions(source, user_id, chat_id, chat_type, thread_id, started_at DESC);
CREATE INDEX IF NOT EXISTS idx_sessions_handoff_state
    ON sessions(handoff_state, started_at);
CREATE INDEX IF NOT EXISTS idx_sessions_system_prompt_hash
    ON sessions(system_prompt_hash);
CREATE INDEX IF NOT EXISTS idx_sessions_tool_names
    ON sessions(tool_names);
-- Recent-session browsing must never derive recency by scanning messages.
-- This expression is the durable, indexable approximation used to preselect
-- a small candidate set before compression-chain and preview hydration.
CREATE INDEX IF NOT EXISTS idx_sessions_effective_activity
    ON sessions(COALESCE(last_activity_at, started_at) DESC, started_at DESC);
-- CLI/TUI listings match gateway rows stored as source='local' by created_source,
-- before LIMIT. created_source is reconciled onto legacy DBs, so this index lives
-- here rather than in SCHEMA_SQL. A source-only index would walk every local row
-- (Desktop and TUI included) to find one surface.
CREATE INDEX IF NOT EXISTS idx_sessions_local_created_started
    ON sessions(created_source, started_at DESC)
    WHERE source = 'local';
"""


# ── Deferred FTS rebuild bookkeeping (schema v23) ──
# While a background index rebuild is pending, two state_meta keys define
# which message rows are currently IN the FTS indexes:
#
#   fts_rebuild_high_water  H — MAX(messages.id) at the moment the old
#                                indexes were dropped
#   fts_rebuild_progress    P — highest id the chunked backfill has indexed
#
# A row is indexed iff  id <= P  (backfilled)  OR  id > H  (inserted after
# the drop; ids are AUTOINCREMENT so new rows are always > H and the insert
# triggers index them live).  Rows in (P, H] are not yet indexed.
#
# Every trigger below gates on that same predicate: firing an FTS5
# external-content 'delete' for a row that is NOT in the index corrupts the
# index, and skipping it for a row that IS indexed leaves a stale entry.
# When no rebuild is pending both keys are absent and COALESCE turns the
# predicate into a tautology (id > -1 OR id <= -1), i.e. normal operation.
# The two state_meta PK probes per write are negligible next to the FTS
# insert itself.
#
# messages_fts_src: the base word index no longer reads raw `messages` as its
# external content. Tool rows are indexed as a bounded prefix, so the index
# must read that SAME projection back or FTS5's 'integrity-check' / 'delete'
# commands disagree with the stored tokens and corrupt the index (the
# recurring fts5 checksum-mismatch drift: the projection used to depend on a
# moving state_meta high-water key). The view/trigger/backfill all share the
# one expression in `_fts_indexed_content_sql` — a fixed per-row function
# with no marker lookups — so the boundary can never move again.
FTS_SQL = f"""
-- Stable projection the base word index reads and writes through: the view
-- computes EXACTLY what the triggers/backfill insert, so 'rebuild' and the
-- integrity checker always agree with the stored index.
CREATE VIEW IF NOT EXISTS messages_fts_src AS
    SELECT id,
           CASE WHEN role = 'tool'
                THEN substr(COALESCE(content, ''), 1, {FTS_TOOL_CONTENT_PREFIX_CHARS})
                ELSE content END AS content,
           tool_name, tool_calls
    FROM messages;

CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
    content,
    tool_name,
    tool_calls,
    content='messages_fts_src',
    content_rowid='id'
);

CREATE TRIGGER IF NOT EXISTS messages_fts_insert AFTER INSERT ON messages
WHEN (new.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                         WHERE key = 'fts_rebuild_high_water'), -1)
   OR new.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                          WHERE key = 'fts_rebuild_progress'), -1))
BEGIN
    INSERT INTO messages_fts(rowid, content, tool_name, tool_calls)
    VALUES (
        new.id,
        {_FTS_NEW_INDEXED_CONTENT_SQL},
        new.tool_name,
        new.tool_calls
    );
END;

CREATE TRIGGER IF NOT EXISTS messages_fts_delete AFTER DELETE ON messages
WHEN (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                         WHERE key = 'fts_rebuild_high_water'), -1)
   OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                          WHERE key = 'fts_rebuild_progress'), -1))
BEGIN
    INSERT INTO messages_fts(messages_fts, rowid, content, tool_name, tool_calls)
    VALUES (
        'delete',
        old.id,
        {_FTS_OLD_INDEXED_CONTENT_SQL},
        old.tool_name,
        old.tool_calls
    );
END;

-- UPDATE OF skips the trigger entirely for non-content column writes
-- (status/compacted/observed/etc.), which is stronger than the WHEN gate
-- alone and avoids FTS I/O saturation on large state.db (#68858 / #73639).
CREATE TRIGGER IF NOT EXISTS messages_fts_update
AFTER UPDATE OF content, tool_name, tool_calls, role ON messages
WHEN (old.content IS NOT new.content
    OR old.tool_name IS NOT new.tool_name
    OR old.tool_calls IS NOT new.tool_calls
    OR old.role IS NOT new.role)
   AND (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                           WHERE key = 'fts_rebuild_high_water'), -1)
     OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                            WHERE key = 'fts_rebuild_progress'), -1))
BEGIN
    INSERT INTO messages_fts(messages_fts, rowid, content, tool_name, tool_calls)
    VALUES (
        'delete',
        old.id,
        {_FTS_OLD_INDEXED_CONTENT_SQL},
        old.tool_name,
        old.tool_calls
    );
    INSERT INTO messages_fts(rowid, content, tool_name, tool_calls)
    VALUES (
        new.id,
        {_FTS_NEW_INDEXED_CONTENT_SQL},
        new.tool_name,
        new.tool_calls
    );
END;
"""


# Trigram FTS5 table for CJK substring search.  The default unicode61
# tokenizer splits CJK characters into individual tokens, breaking phrase
# matching.  The trigram tokenizer creates overlapping 3-byte sequences so
# substring queries work natively for any script (CJK, Thai, etc.).
#
# The trigram index is the most expensive index in state.db (~2.6x the size
# of the text it covers). Tool output (~90% of message bytes, machine noise)
# and cron transcripts are excluded: the index reads through
# ``messages_fts_trigram_src``, a view that skips both classes. They stay
# fully stored in ``messages`` and searchable via the standard
# ``messages_fts`` index; they just don't get trigram (CJK substring)
# treatment. ``search_messages`` routes explicit tool/cron CJK searches to
# LIKE for the same reason. Structured ``tool_calls`` JSON likewise stays
# searchable through ``messages_fts``; excluding it here avoids indexing
# repetitive JSON syntax as trigrams (FTS_STORAGE_VERSION 2).
#
# Delegate-child (subagent) transcripts are excluded the same way (v30):
# on a fan-out-heavy install they were ~70% of all message bytes and
# ``session_search`` hides ``source='subagent'`` sessions anyway. A child
# is recognised by its source OR by the ``_delegate_from`` creation marker
# (children spawned under a gateway turn inherit the gateway's source).
# Compression/branch continuations of interactive sessions also carry
# ``parent_session_id`` but NOT the marker, so they stay trigram-indexed.
FTS_TRIGRAM_EXCLUDED_SOURCES = ("cron", "subagent")

def fts_trigram_session_sql(alias: str = "") -> str:
    """Predicate over a ``sessions`` row selecting sessions whose rows belong in
    the trigram index; ``alias`` qualifies every column for joins. Shared by the
    view, the sync triggers, and the deferred-backfill INSERT ... SELECTs so they
    can never disagree about the index boundary."""
    q = f"{alias}." if alias else ""
    return (
        f"{q}source NOT IN ("
        + ", ".join(f"'{src}'" for src in FTS_TRIGRAM_EXCLUDED_SOURCES)
        + f") AND {_sql_json_extract(q + 'model_config', '$._delegate_from')} IS NULL"
    )


FTS_TRIGRAM_SESSION_SQL = fts_trigram_session_sql()


FTS_TRIGRAM_SQL = f"""
CREATE VIEW IF NOT EXISTS messages_fts_trigram_src AS
    SELECT m.id, m.role, m.content, m.tool_name
    FROM messages AS m
    JOIN sessions AS s ON s.id = m.session_id
    WHERE m.role <> 'tool' AND {fts_trigram_session_sql('s')};

CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts_trigram USING fts5(
    content,
    tool_name,
    content='messages_fts_trigram_src',
    content_rowid='id',
    tokenize='trigram'
);

CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_insert AFTER INSERT ON messages
WHEN new.role <> 'tool'
   AND EXISTS (SELECT 1 FROM sessions
               WHERE id = new.session_id AND {FTS_TRIGRAM_SESSION_SQL})
   AND (new.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                           WHERE key = 'fts_rebuild_high_water'), -1)
     OR new.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                            WHERE key = 'fts_rebuild_progress'), -1))
BEGIN
    INSERT INTO messages_fts_trigram(rowid, content, tool_name)
    VALUES (new.id, new.content, new.tool_name);
END;

CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_delete AFTER DELETE ON messages
WHEN old.role <> 'tool'
   AND EXISTS (SELECT 1 FROM sessions
               WHERE id = old.session_id AND {FTS_TRIGRAM_SESSION_SQL})
   AND (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                           WHERE key = 'fts_rebuild_high_water'), -1)
     OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                            WHERE key = 'fts_rebuild_progress'), -1))
BEGIN
    INSERT INTO messages_fts_trigram(messages_fts_trigram, rowid, content, tool_name)
    VALUES ('delete', old.id, old.content, old.tool_name);
END;

CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_update
AFTER UPDATE OF content, tool_name, role ON messages
WHEN (old.content IS NOT new.content
    OR old.tool_name IS NOT new.tool_name
    OR old.role IS NOT new.role)
   AND (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                           WHERE key = 'fts_rebuild_high_water'), -1)
     OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
                            WHERE key = 'fts_rebuild_progress'), -1))
BEGIN
    INSERT INTO messages_fts_trigram(messages_fts_trigram, rowid, content, tool_name)
    SELECT 'delete', old.id, old.content, old.tool_name
    WHERE old.role <> 'tool'
      AND EXISTS (SELECT 1 FROM sessions
                  WHERE id = old.session_id AND {FTS_TRIGRAM_SESSION_SQL});
    INSERT INTO messages_fts_trigram(rowid, content, tool_name)
    SELECT new.id, new.content, new.tool_name
    WHERE new.role <> 'tool'
      AND EXISTS (SELECT 1 FROM sessions
                  WHERE id = new.session_id AND {FTS_TRIGRAM_SESSION_SQL});
END;
"""

# Set when a base/trigram FTS index was detached after runtime corruption; startup must rebuild the complete
# index before reinstalling sync triggers (rows written while they were absent leave an unknown gap).
FTS_STALE_KEY = "fts_stale"

# Durable diagnostic for stale FTS recovery blocked across process restarts.
FTS_REBUILD_DEFERRAL_KEY = "fts_rebuild_deferral"


# ── Legacy (v22 / inline-content) FTS DDL ──────────────────────────────
# Used ONLY to keep an existing pre-v23 install's search working and its
# triggers repairable UNTIL the user opts into `hermes db optimize`. This is
# the exact inline shape v11..v22 shipped: each virtual table stores its own
# copy of ``content || tool_name || tool_calls`` and the trigram table indexes
# every row (including role='tool'). We never CREATE these on a fresh install —
# fresh installs are born on the v23 external-content schema above. These
# constants exist so a legacy DB is never accidentally handed the v23 DDL
# (which would create the external-content trigram source VIEW and leave the
# DB in a mixed, broken state). `optimize_fts_storage()` is what migrates a
# legacy DB to the v23 shape.
LEGACY_FTS_SQL = f"""
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
    content
);

CREATE TRIGGER IF NOT EXISTS messages_fts_insert AFTER INSERT ON messages BEGIN
    INSERT INTO messages_fts(rowid, content) VALUES (
        new.id,
        COALESCE({_FTS_NEW_INDEXED_CONTENT_SQL}, '')
        || ' ' || COALESCE(new.tool_name, '') || ' ' || COALESCE(new.tool_calls, '')
    );
END;

CREATE TRIGGER IF NOT EXISTS messages_fts_delete AFTER DELETE ON messages BEGIN
    DELETE FROM messages_fts WHERE rowid = old.id;
END;

CREATE TRIGGER IF NOT EXISTS messages_fts_update
AFTER UPDATE OF content, tool_name, tool_calls, role ON messages BEGIN
    DELETE FROM messages_fts WHERE rowid = old.id;
    INSERT INTO messages_fts(rowid, content) VALUES (
        new.id,
        COALESCE({_FTS_NEW_INDEXED_CONTENT_SQL}, '')
        || ' ' || COALESCE(new.tool_name, '') || ' ' || COALESCE(new.tool_calls, '')
    );
END;
"""


LEGACY_FTS_TRIGRAM_SQL = f"""
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts_trigram USING fts5(
    content,
    tokenize='trigram'
);

CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_insert AFTER INSERT ON messages BEGIN
    INSERT INTO messages_fts_trigram(rowid, content) VALUES (
        new.id,
        COALESCE({_FTS_NEW_INDEXED_CONTENT_SQL}, '')
        || ' ' || COALESCE(new.tool_name, '') || ' ' || COALESCE(new.tool_calls, '')
    );
END;

CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_delete AFTER DELETE ON messages BEGIN
    DELETE FROM messages_fts_trigram WHERE rowid = old.id;
END;

CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_update
AFTER UPDATE OF content, tool_name, tool_calls, role ON messages BEGIN
    DELETE FROM messages_fts_trigram WHERE rowid = old.id;
    INSERT INTO messages_fts_trigram(rowid, content) VALUES (
        new.id,
        COALESCE({_FTS_NEW_INDEXED_CONTENT_SQL}, '')
        || ' ' || COALESCE(new.tool_name, '') || ' ' || COALESCE(new.tool_calls, '')
    );
END;
"""

# Cross-process full-FTS-rebuild admission (single authority).  Several processes share one state.db and a
# structural rebuild (FTS5 'rebuild' or `_recover_stale_fts`'s drop/recreate) must run in ONE at a time —
# concurrent rebuilds corrupted state.db in production.  Gates `rebuild_fts()`, `_rebuild_fts_indexes()`,
# `_recover_stale_fts()`; the chunked backfill (`fts_rebuild_step`) is deliberately NOT routed through it (it
# claims progress under SQLite transaction authority).  Mirrors `hermes_state_repair._cross_process_repair_lock`:
# portable (msvcrt/flock), bounded wait, FAIL CLOSED; orphaned-fd holders (see `_acquire_db_flock`) are broken
# only when provably dead, indeterminate liveness defers.  `<db>.fts_rebuild.lock` is distinct from
# `<db>.repair.lock` (offline schema surgery, minutes in VACUUM).  Lives here: mixins cannot import hermes_state.

# ── Cross-process full-FTS-rebuild admission (single authority) ────────────── Several independent Hermes
# processes routinely share one state.db (gateway service, the Desktop app's `hermes serve` backend,
# interactive CLI sessions, the TUI slash worker). A full structural FTS rebuild — the FTS5 'rebuild'
# command or the drop/recreate script in `_recover_stale_fts` — must only ever run in ONE of them at a time:
# two concurrent rebuilds collide on write and have structurally corrupted state.db in production (PR
# #93200; the 2026-08-15 / 2026-08-23 incidents and issues #89293 / #90950). This is the single admission
# authority for every full structural rebuild entry point: `SessionSearchMixin.rebuild_fts()`,
# `SessionSchemaMixin._rebuild_fts_indexes()` (via `_init_schema`), and
# `SessionSchemaMixin._recover_stale_fts()`. The chunked deferred backfill (`fts_rebuild_step`) is
# deliberately NOT routed through it — it claims progress under `_execute_write`'s SQLite transaction
# authority and is intentionally multi-process. Semantics mirror `hermes_state_repair._cross_process_repair_lock`
# (the schema- surgery authority): portable (msvcrt on Windows, flock elsewhere), bounded wait, and FAIL
# CLOSED — a caller that cannot acquire the lock must NOT rebuild. The kernel drops both lock types when the
# holder dies — UNLESS a forked child inherited the lock fd (flock rides the open file description, which
# fork() duplicates), in which case the orphaned descriptor holds the lock forever (issue #100108).
# `_acquire_db_flock` therefore records the holder's pid + start time under the lock and, when the recorded
# holder is provably dead, breaks the orphaned lock by unlinking and retaking it on a fresh inode;
# indeterminate liveness still defers. It lives here (not hermes_state) because the search/schema mixins
# cannot import hermes_state (cycle). The lock file is `<db>.fts_rebuild.lock`, distinct from
# `<db>.repair.lock`: schema surgery runs on an EXCLUSIVE offline connection and can legitimately take
# minutes in VACUUM, while runtime rebuilds run on live connections. The timeout is sized for a full
# 'rebuild' of both indexes on a large DB.
logger = logging.getLogger("hermes_state")

_FTS_REBUILD_LOCK_TIMEOUT_SECONDS = 120.0
_FTS_REBUILD_LOCK_POLL_SECONDS = 0.1
_IS_WINDOWS = sys.platform == "win32"
# Post-break re-acquire budget: the fresh inode is contended only by live processes — never the full timeout.
_LOCK_BREAK_REACQUIRE_SECONDS = 5.0

# "Another process holds the lock": flock → EWOULDBLOCK/EAGAIN, msvcrt.locking → EACCES (EDEADLK when its retry
# gives up).  Anything else (ESTALE, ENOTSUP, ENOLCK, EIO) is a persistent failure polling cannot fix.
_LOCK_CONTENTION_ERRNOS = {errno.EAGAIN, errno.EACCES, errno.EWOULDBLOCK, errno.EDEADLK}


def is_advisory_lock_contention(exc: BaseException) -> bool:
    """True when *exc* means another process holds the lock; on any other ``OSError`` fail closed at once."""
    return isinstance(exc, BlockingIOError) or (isinstance(exc, OSError) and exc.errno in _LOCK_CONTENTION_ERRNOS)


def _proc_start_ticks(pid: int):
    """Kernel start time of *pid* (field 22 of ``/proc/<pid>/stat``; with the PID it identifies a process
    uniquely).  None off Linux or on any failure — callers must treat None as unknowable and FAIL CLOSED."""
    try:
        # comm (field 2) may contain spaces/parens; split after the LAST ')'.
        with open(f"/proc/{pid}/stat", "rb") as fh:
            return int(fh.read().rsplit(b")", 1)[1].split()[19])
    except (OSError, ValueError, IndexError):
        return None


def _read_lock_holder_record(handle):
    """Best-effort parse of the holder metadata JSON in a lock file."""
    try:
        handle.seek(0)
        raw = handle.read(4096)
        record = json.loads(raw.decode("utf-8", "replace")) if raw else None
    except (OSError, ValueError, UnicodeDecodeError):
        return None
    return record if isinstance(record, dict) else None


def _rewrite_lock_file(handle, payload: bytes) -> None:
    """Best-effort truncate-and-write of *payload* at offset 0."""
    with contextlib.suppress(OSError, ValueError):
        handle.seek(0)
        handle.truncate()
        if payload:
            handle.write(payload)
        handle.flush()


def _write_lock_holder_record(handle) -> None:
    """Record this process as holder (best effort) so timed-out contenders can tell an orphaned-fd holder
    from a live wedged one.

    Written under the flock so contenders that time out can tell an orphaned-fd holder (recorded process
    dead, flock inherited by a forked child — issue #100108) from a live wedged holder.  ``pidns``: see
    ``hermes_state_pidns``.
    """
    record = {"pid": os.getpid(), "pidns": pid_namespace_id(),
              "start_ticks": _proc_start_ticks(os.getpid()), "acquired_at": time.time()}
    _rewrite_lock_file(handle, json.dumps(record, sort_keys=True).encode("utf-8"))


def _clear_lock_holder_record(handle) -> None:
    """Erase holder metadata before a normal release: a surviving record means ABNORMAL exit (break allowed)."""
    _rewrite_lock_file(handle, b"")


def _lock_holder_provably_dead(record) -> bool:
    """True ONLY when the recorded holder is provably dead or PID-recycled.  Anything indeterminate
    (no/malformed record, PID owned by another user, /proc unavailable, foreign PID namespace)
    is False: FAIL CLOSED and defer.  Unstamped records keep probing: see ``hermes_state_pidns``."""
    try:
        pid = int(record["pid"])
    except (KeyError, TypeError, ValueError):
        return False
    if pid <= 0:
        return False
    if not persistent_record_pidns_checkable(record.get("pidns")):
        return False  # foreign namespace: a local reading is not proof — defer
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return True
    except OSError:
        return False  # PermissionError et al.: PID exists (or unknowable) — closed
    recorded_ticks = record.get("start_ticks")
    if recorded_ticks is None:
        return False
    current_ticks = _proc_start_ticks(pid)
    return current_ticks is not None and current_ticks != recorded_ticks  # different start time: PID recycled


def _acquire_db_flock(lock_path, handle, timeout_seconds, poll_seconds, description):
    """Bounded POSIX flock acquire with orphaned-holder break.  Returns ``(acquired, handle)``; *handle* may
    have been re-opened and the caller closes whichever comes back.  *acquired*: True, False (a holder kept
    the lock past the deadline) or None (non-contention ``OSError``, already logged: treat as not acquired
    without the held-by-another-process warning).  ``flock`` rides the open file DESCRIPTION, which ``fork()``
    duplicates, so a holder that forks then dies leaves the lock held forever; when the acquirer is provably
    dead the file is unlinked and retaken on a fresh inode (the orphan's flock excludes nobody).  Every
    acquire verifies its inode still names *lock_path*, so a racer on a dead inode retries.

    A holder that forks (multiprocessing worker, daemonized helper) and then dies leaves the flock held by a
    child that will never release it — the kernel's holder-death release never triggers, and every contender
    defers forever. Indeterminate liveness always defers (fail closed). See #100108.
    """
    import fcntl
    deadline = time.monotonic() + timeout_seconds
    broke_lock = False
    while True:
        try:
            fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except (BlockingIOError, OSError) as exc:
            if not is_advisory_lock_contention(exc):
                logger.warning("Could not acquire %s %s (%s) — deferring rather than "
                               "waiting out the %.0fs holder timeout on a non-contention error.",
                               description, lock_path, exc, timeout_seconds)
                return None, handle
            if time.monotonic() < deadline:
                time.sleep(poll_seconds)
                continue
            if broke_lock:
                return False, handle
            record = _read_lock_holder_record(handle)
            if not _lock_holder_provably_dead(record):
                return False, handle
            logger.warning("%s %s is held by an orphaned file descriptor (recorded holder pid %s is dead — a "
                           "forked child inherited the lock fd); breaking the stale lock and retaking it on a "
                           "fresh file.", description, lock_path, (record or {}).get("pid"))
            try:
                os.unlink(lock_path)
                handle.close()
                handle = open(lock_path, "a+b")
            except OSError as exc:
                logger.warning("Could not break stale %s %s (%s) — deferring.", description, lock_path, exc)
                return False, handle
            broke_lock = True
            deadline = time.monotonic() + _LOCK_BREAK_REACQUIRE_SECONDS
            continue
        # A breaker may have replaced the file while we waited; a lock on a dead inode excludes nobody.
        try:
            fd_stat, path_stat = os.fstat(handle.fileno()), os.stat(lock_path)
            same_file = fd_stat.st_dev == path_stat.st_dev and fd_stat.st_ino == path_stat.st_ino
        except OSError:
            same_file = False
        if same_file:
            _write_lock_holder_record(handle)
            return True, handle
        try:
            handle.close()
            handle = open(lock_path, "a+b")
        except OSError:
            return False, handle
        if time.monotonic() >= deadline:
            return False, handle


def _describe_lock_holder(record) -> str:
    """Human-readable holder identity for deferral warnings."""
    if not isinstance(record, dict) or "pid" not in record:
        return "unknown (no holder record; pre-fix writer or non-Hermes)"
    age = ""
    with contextlib.suppress(TypeError, ValueError):
        if record.get("acquired_at") is not None:
            age = f", acquired {time.time() - float(record['acquired_at']):.0f}s ago"
    return f"pid {record.get('pid')}{age}"


def _acquire_msvcrt_lock(lock_path, handle, timeout):
    """Windows counterpart of ``_acquire_db_flock`` (no orphan break); same True / False / None contract."""
    import msvcrt
    deadline = time.monotonic() + timeout
    while True:
        try:
            handle.seek(0)
            msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
            return True
        except (BlockingIOError, OSError) as exc:
            if not is_advisory_lock_contention(exc):
                logger.warning("Could not acquire FTS rebuild lock %s (%s) — deferring on a non-contention error.",
                               lock_path, exc)
                return None
            if time.monotonic() >= deadline:
                return False
            time.sleep(_FTS_REBUILD_LOCK_POLL_SECONDS)


@contextlib.contextmanager
def fts_rebuild_admission(db_path, *, timeout_seconds=None):
    """Serialize full structural FTS rebuilds on *db_path* across processes.  Yields True when this process
    holds the authority, False when the bounded acquire timed out or the lock file could not be opened: the
    caller must NOT rebuild (fail closed; the stale breadcrumb guarantees a retry).  ``db_path`` None
    (in-memory) yields True.  In-process retries pass ``timeout_seconds=0`` so a live holder never stalls a
    long-lived writer; the orphan break still applies."""
    if db_path is None:
        yield True
        return
    timeout = _FTS_REBUILD_LOCK_TIMEOUT_SECONDS if timeout_seconds is None else max(float(timeout_seconds), 0.0)
    lock_path = f"{db_path}.fts_rebuild.lock"
    try:
        handle = open(lock_path, "a+b")
    except OSError as exc:
        # Fail closed like a timed-out acquire: an unopenable lock file means the FS is out of
        # space/inodes/descriptors and a sibling that opened earlier may still be rebuilding — yielding True
        # gave every process on a full disk a concurrent rebuild.  Deferring is free: the breadcrumb retries.
        logger.warning("Could not open FTS rebuild lock %s (%s) — deferring this rebuild "
                       "rather than running it without cross-process authority.", lock_path, exc)
        yield False
        return
    acquired = False
    try:
        if _IS_WINDOWS:
            acquired = _acquire_msvcrt_lock(lock_path, handle, timeout)
        else:
            acquired, handle = _acquire_db_flock(
                lock_path, handle, timeout, _FTS_REBUILD_LOCK_POLL_SECONDS, "FTS rebuild lock")
        if acquired is None:
            # Already logged with the real errno; "held by another process" would be a lie.
            acquired = False
        elif not acquired:
            record = None if _IS_WINDOWS else _read_lock_holder_record(handle)
            if timeout <= 0:
                # Non-blocking probe from an in-process retry: keep it quiet.
                logger.info("FTS rebuild lock %s is busy — deferring this retry "
                            "(the stale-FTS breadcrumb keeps it retryable). Recorded holder: %s.",
                            lock_path, _describe_lock_holder(record))
            else:
                logger.warning("FTS rebuild lock %s held by another process for more than %.0fs — deferring "
                               "this rebuild to avoid racing the holder (the stale-FTS breadcrumb keeps it "
                               "retryable). Recorded holder: %s.", lock_path, timeout, _describe_lock_holder(record))
        yield acquired
    finally:
        try:
            with contextlib.suppress(OSError):  # best-effort release
                if acquired and _IS_WINDOWS:
                    import msvcrt
                    handle.seek(0)
                    msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
                elif acquired:
                    import fcntl
                    _clear_lock_holder_record(handle)
                    fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
        finally:
            handle.close()


def _json_or(raw: Any, fallback: Any, warning: str) -> Any:
    """``json.loads(raw)``; on failure log *warning* and return *fallback*."""
    try:
        return json.loads(raw)
    except (json.JSONDecodeError, TypeError):
        logger.warning(warning)
        return fallback