Blame for app/db_sync.py

# =============================================================================
# app/db_sync.py
# Internal SQLite IPC — app/* state & communication
# Universal MCP Hub (Sandboxed) - based on PyFundaments Architecture
# Copyright 2026 - Volkan Kücükbudak
# Apache License V. 2 + ESOL 1.1
# Repo: https://github.com/VolkanSah/Universal-MCP-Hub-sandboxed
# =============================================================================
# ARCHITECTURE NOTE:
#   This file lives exclusively in app/ and is ONLY started by app/app.py.
#   NO direct access to fundaments/*, .env, or Guardian (main.py).
#   DB path comes from app/.pyfun [DB_SYNC] → SQLITE_PATH via app/config.py.
#
# CRITICAL RULES:
#   - This is NOT postgresql.py — cloud DB is Guardian-only!
#   - db_sync ONLY manages its own tables (hub_state, tool_cache)
#   - NEVER touch Guardian tables (users, sessions) — those belong to user_handler.py
#   - SQLite path is shared with user_handler.py via SQLITE_PATH
#   - app/* modules call db_sync.write() / db_sync.read() — never aiosqlite directly
#
# TABLE OWNERSHIP:
#   users, sessions  → Guardian (fundaments/user_handler.py) — DO NOT TOUCH!
#   hub_state        → db_sync (app/* internal state)
#   tool_cache       → db_sync (app/* tool response cache)
#   persist_queue    → db_sync (restart-safe queue → PostgreSQL long-term memory)
#   usage_log        → db_sync (tokens + estimated cost per LLM call)
# =============================================================================

import asyncio
import re
import time
import aiosqlite
from pathlib import Path
import logging
import json
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional

from . import config

logger = logging.getLogger("db_sync")

# =============================================================================
# Internal State
# =============================================================================
_db_path: Optional[str] = None
_initialized: bool = False


# =============================================================================
# Initialization — called by app/app.py (parameterless, sandboxed)
# =============================================================================

async def initialize() -> None:
    global _db_path, _initialized

    if _initialized:
        return

    db_cfg   = config.get_db_sync()
    raw_path = db_cfg.get("SQLITE_PATH", "app/.hub_state.db")
    
    # HF Spaces: SPACE_ID is set → filesystem is read-only except /tmp/
    import os
    if os.getenv("SPACE_ID"):
        filename = os.path.basename(raw_path)
        _db_path = f"/tmp/{filename}"
        logger.info(f"HF Space detected — SQLite relocated to {_db_path}")
    else:
        _db_path = raw_path

    await _init_tables()

    _initialized = True
    logger.info(f"db_sync initialized — path: {_db_path}")


# =============================================================================
# SECTION 1 — Table Setup (app/* tables only!)
# =============================================================================

async def _init_tables() -> None:
    """
    Creates app/* internal tables if they don't exist.
    NEVER modifies Guardian tables (users, sessions).
    """
    async with aiosqlite.connect(_db_path) as db:

        # hub_state — generic key/value store for app/* modules
        await db.execute("""
            CREATE TABLE IF NOT EXISTS hub_state (
                key        TEXT PRIMARY KEY,
                value      TEXT,
                updated_at TEXT
            )
        """)

        # tool_cache — cached tool responses to reduce API calls
        await db.execute("""
            CREATE TABLE IF NOT EXISTS tool_cache (
                id         INTEGER PRIMARY KEY AUTOINCREMENT,
                tool_name  TEXT NOT NULL,
                prompt     TEXT NOT NULL,
                response   TEXT NOT NULL,
                provider   TEXT,
                model      TEXT,
                created_at TEXT
            )
        """)

        await db.execute("""
            CREATE INDEX IF NOT EXISTS idx_tool_cache_tool
            ON tool_cache(tool_name)
        """)

        # persist_queue — Dauer-Gedächtnis-Queue: überlebt Restarts,
        # bis der Eintrag erfolgreich nach PostgreSQL geflusht wurde.
        # score-Spalte = reservierter Slot für den Bullshit-Filter (Nutzer-eigenes
        # Scoring, später) — wird von db_sync NICHT interpretiert.
        await db.execute("""
            CREATE TABLE IF NOT EXISTS persist_queue (
                id           INTEGER PRIMARY KEY AUTOINCREMENT,
                target_table TEXT NOT NULL,
                payload      TEXT NOT NULL,
                source       TEXT,
                model        TEXT,
                score        REAL,
                queued_at    TEXT NOT NULL,
                attempts     INTEGER NOT NULL DEFAULT 0,
                last_error   TEXT
            )
        """)

        # usage_log — one row per successful LLM call (tokens + estimated cost).
        # cost_usd is NULL when no price is known. Read-only for everything else.
        await db.execute("""
            CREATE TABLE IF NOT EXISTS usage_log (
                id         INTEGER PRIMARY KEY AUTOINCREMENT,
                ts         TEXT NOT NULL,
                tool       TEXT,
                provider   TEXT,
                model      TEXT,
                tokens_in  INTEGER,
                tokens_out INTEGER,
                cost_usd   REAL,
                latency_ms INTEGER
            )
        """)
        await db.execute("""
            CREATE INDEX IF NOT EXISTS idx_usage_log_ts ON usage_log(ts)
        """)

        await db.commit()

    logger.info("db_sync tables ready.")


# =============================================================================
# SECTION 2 — Key/Value Store (hub_state table)
# =============================================================================

async def write(key: str, value: Any) -> None:
    """
    Write a value to hub_state key/value store.
    Value is JSON-serialized — supports dicts, lists, strings, numbers.

    Args:
        key:   Unique key string (e.g. 'scheduler.last_run').
        value: Any JSON-serializable value.
    """
    _check_init()
    now = datetime.now(timezone.utc).isoformat()

    async with aiosqlite.connect(_db_path) as db:
        await db.execute("""
            INSERT OR REPLACE INTO hub_state (key, value, updated_at)
            VALUES (?, ?, ?)
        """, (key, json.dumps(value), now))
        await db.commit()


async def read(key: str, default: Any = None) -> Any:
    """
    Read a value from hub_state key/value store.
    Returns default if key does not exist.

    Args:
        key:     Key string to look up.
        default: Default value if key not found. Default: None.

    Returns:
        Deserialized value, or default if not found.
    """
    _check_init()

    async with aiosqlite.connect(_db_path) as db:
        cursor = await db.execute(
            "SELECT value FROM hub_state WHERE key = ?", (key,)
        )
        row = await cursor.fetchone()

    if row is None:
        return default

    try:
        return json.loads(row[0])
    except (json.JSONDecodeError, TypeError):
        return row[0]


async def delete(key: str) -> None:
    """
    Delete a key from hub_state.

    Args:
        key: Key string to delete.
    """
    _check_init()

    async with aiosqlite.connect(_db_path) as db:
        await db.execute("DELETE FROM hub_state WHERE key = ?", (key,))
        await db.commit()


# =============================================================================
# SECTION 3 — Tool Cache (tool_cache table)
# =============================================================================

async def cache_write(
    tool_name: str,
    prompt: str,
    response: str,
    provider: str = None,
    model: str = None,
) -> None:
    """
    Cache a tool response to reduce redundant API calls.

    Args:
        tool_name: Tool name (e.g. 'llm_complete', 'web_search').
        prompt:    The input prompt/query that was used.
        response:  The response to cache.
        provider:  Provider name used (optional).
        model:     Model name used (optional).
    """
    _check_init()

    db_cfg      = config.get_db_sync()
    max_entries = int(db_cfg.get("MAX_CACHE_ENTRIES", "1000"))
    now         = datetime.now(timezone.utc).isoformat()

    async with aiosqlite.connect(_db_path) as db:
        await db.execute("""
            INSERT INTO tool_cache (tool_name, prompt, response, provider, model, created_at)
            VALUES (?, ?, ?, ?, ?, ?)
        """, (tool_name, prompt, response, provider, model, now))

        # Enforce MAX_CACHE_ENTRIES — delete oldest if exceeded
        await db.execute("""
            DELETE FROM tool_cache WHERE id NOT IN (
                SELECT id FROM tool_cache ORDER BY created_at DESC LIMIT ?
            )
        """, (max_entries,))

        await db.commit()


async def cache_read(tool_name: str, prompt: str, max_age_sec: int = 0) -> Optional[str]:
    """
    Read a cached tool response.
    Returns None if no cache entry exists (or it is older than max_age_sec).

    Args:
        tool_name:   Tool name to look up.
        prompt:      The exact prompt/query to match.
        max_age_sec: Ignore entries older than this (0 = no age limit).

    Returns:
        Cached response string, or None if not found.
    """
    _check_init()

    cutoff = ""
    if max_age_sec and max_age_sec > 0:
        cutoff = (datetime.now(timezone.utc) - timedelta(seconds=max_age_sec)).isoformat()

    async with aiosqlite.connect(_db_path) as db:
        cursor = await db.execute("""
            SELECT response FROM tool_cache
            WHERE tool_name = ? AND prompt = ? AND created_at >= ?
            ORDER BY created_at DESC LIMIT 1
        """, (tool_name, prompt, cutoff))
        row = await cursor.fetchone()

    return row[0] if row else None


# =============================================================================
# SECTION 4 — Read-Only Query (for mcp.py db_query tool)
# =============================================================================

# Tables db_query must never touch (Guardian-owned) + SQLite internals.
# Identifiers must appear literally in SQL, so a word-boundary check is reliable.
_QUERY_BLOCKED = re.compile(
    r"\b(users|sessions|sqlite_master|sqlite_schema|sqlite_temp_master|pragma|attach|detach)\b",
    re.IGNORECASE,
)
_QUERY_MAX_ROWS = 1000


async def query(sql: str) -> List[Dict]:
    """
    Execute a read-only SELECT query on the internal hub state database.
    Called by mcp.py db_query tool when db_sync.py is active.

    Hardening (the docstring used to promise more than the code did):
      - exactly ONE statement, must start with SELECT
      - Guardian tables (users, sessions) + SQLite internals are rejected
      - connection is opened read-only at engine level (mode=ro) — writes are impossible
      - result is capped at 1000 rows

    Args:
        sql: SQL SELECT statement to execute.

    Returns:
        List of result rows as dicts.

    Raises:
        ValueError: If the query is not an allowed SELECT statement.
    """
    _check_init()

    stmt = sql.strip()
    if stmt.endswith(";"):
        stmt = stmt[:-1].rstrip()
    if not stmt.upper().startswith("SELECT"):
        raise ValueError("Only SELECT queries are permitted in db_query tool.")
    if ";" in stmt:
        raise ValueError("Only a single statement is permitted in db_query tool.")
    if _QUERY_BLOCKED.search(stmt):
        raise ValueError("Query touches a protected table or SQLite internals.")

    ro_uri = Path(_db_path).resolve().as_uri() + "?mode=ro"
    async with aiosqlite.connect(ro_uri, uri=True) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(stmt)
        rows   = await cursor.fetchmany(_QUERY_MAX_ROWS)
        return [dict(r) for r in rows]


# =============================================================================
# SECTION 4b — Usage / Cost Log (usage_log table)
# =============================================================================

async def log_usage(
    tool: str,
    provider: str,
    model: str,
    tokens_in: Optional[int],
    tokens_out: Optional[int],
    cost_usd: Optional[float],
    latency_ms: Optional[int] = None,
) -> None:
    """Append one usage row. Called by providers.py after every successful LLM call."""
    _check_init()
    now = datetime.now(timezone.utc).isoformat()
    async with aiosqlite.connect(_db_path) as db:
        await db.execute("""
            INSERT INTO usage_log (ts, tool, provider, model, tokens_in, tokens_out, cost_usd, latency_ms)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """, (now, tool, provider, model, tokens_in, tokens_out, cost_usd, latency_ms))
        await db.commit()


_USAGE_GROUPS = {"model": "model", "provider": "provider", "tool": "tool", "day": "substr(ts, 1, 10)"}


async def usage_stats(group_by: str = "model", days: Optional[int] = None) -> Dict[str, Any]:
    """
    Aggregate usage_log.

    Args:
        group_by: "model" | "provider" | "tool" | "day"
        days:     Only count the last N days (None = all time).

    Returns:
        {"group_by", "days", "rows": [...], "total": {...}}
        cost_usd is the sum of KNOWN costs; calls_unpriced counts calls without a price.
    """
    _check_init()
    col = _USAGE_GROUPS.get(str(group_by).lower())
    if not col:
        raise ValueError(f"group_by must be one of {sorted(_USAGE_GROUPS)}")

    where, args = "", []
    if days:
        cutoff = (datetime.now(timezone.utc) - timedelta(days=int(days))).isoformat()
        where, args = "WHERE ts >= ?", [cutoff]

    select = """
        COUNT(*)                                         AS calls,
        COALESCE(SUM(tokens_in), 0)                      AS tokens_in,
        COALESCE(SUM(tokens_out), 0)                     AS tokens_out,
        ROUND(COALESCE(SUM(cost_usd), 0), 6)             AS cost_usd,
        SUM(CASE WHEN cost_usd IS NULL THEN 1 ELSE 0 END) AS calls_unpriced
    """
    async with aiosqlite.connect(_db_path) as db:
        db.row_factory = aiosqlite.Row
        cur  = await db.execute(
            f"SELECT {col} AS grp, {select} FROM usage_log {where} GROUP BY grp ORDER BY cost_usd DESC, calls DESC",
            args)
        rows = [dict(r) for r in await cur.fetchall()]
        cur  = await db.execute(f"SELECT {select} FROM usage_log {where}", args)
        total = dict(await cur.fetchone())

    return {"group_by": group_by, "days": days, "rows": rows, "total": total}


# =============================================================================
# SECTION 5 — Helpers
# =============================================================================

def _check_init() -> None:
    """Raise RuntimeError if db_sync was not initialized."""
    if not _initialized or not _db_path:
        raise RuntimeError("db_sync not initialized — call initialize() first.")


def is_ready() -> bool:
    """Returns True if db_sync is initialized and ready."""
    return _initialized and _db_path is not None


# =============================================================================
# SECTION 6 — Memory Orchestration: SQLite (Kurzzeit) → PostgreSQL (Dauer)
# =============================================================================
# ARCHITECTURE (Gedächtnis-Design):
#   SQLite (hub_state / tool_cache / persist_queue) = Kurzzeit-Gedächtnis.
#   Alle app/* Tools lesen/schreiben hier — immer verfügbar, kein Key nötig.
#   PostgreSQL = Dauer-Gedächtnis — WICHTIGE Dinge wandern dorthin.
#
#   persist_result Tool → enqueue_persist()   (sofort, immer, kein Crash)
#                          ↓
#                    persist_queue (SQLite, überlebt Restarts)
#                          ↓
#            flush_persist_queue() → PostgreSQL (Guardian-Writer)
#            (Background-Retry bis die DB da ist — siehe start_persist_flusher)
#
#   Delivery-Semantik: at-least-once (PG-Insert + Queue-Delete sind nicht
#   atomar über zwei Systeme). Für ein Gedächtnis ist ein seltener Duplikat-
#   Eintrag unkritisch — Datenverlust wäre es.
#
#   HF-Spaces-Hinweis: SPACE_ID verschiebt SQLite nach /tmp — Queue überbrückt
#   dort DB-Ausfälle innerhalb der Container-Lebenszeit; dauerhaft ist nur PG.
# =============================================================================
_psql_writer = None
_flush_task: Optional[asyncio.Task] = None
_pg_tables_ensured: set = set()

# Defense: Tabellennamen kommen aus .pyfun, aber wir validieren trotzdem.
_SAFE_TABLE_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")

# Ziel-Schema in PostgreSQL — source/model/score als eigene Spalten,
# damit dein Bullshit-Filter später direkt auf score filtern kann.
_PG_DDL = """
    CREATE TABLE IF NOT EXISTS {table} (
        id          BIGSERIAL PRIMARY KEY,
        created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
        source      TEXT,
        model       TEXT,
        score       REAL,
        payload     JSONB NOT NULL
    )
"""


def set_psql_writer(writer_fn) -> None:
    """
    Receives the asyncpg execute callable from Guardian via app.py.
    Called once in start_application() if db_service is available.
    app/* never imports postgresql.py directly — this is the only bridge.
    """
    global _psql_writer
    _psql_writer = writer_fn
    _pg_tables_ensured.clear()  # neue Connection → CREATE TABLE erneut sicherstellen
    logger.info("PostgreSQL writer registered.")


async def enqueue_persist(
    table: str,
    payload: Any,
    source: str = "",
    model: str = "",
    score: Optional[float] = None,
) -> int:
    """
    Queue an entry for the PostgreSQL long-term memory.
    ALWAYS works (no writer needed) — this is the 'no crash' guarantee:
    persist_result ohne DATABASE_URL = Eintrag wartet restart-safe in der Queue.

    Args:
        table:   Target table in PostgreSQL (from .pyfun target_table).
        payload: JSON-serializable data to persist.
        source:  Which tool/component produced this entry.
        model:   Which model produced this entry.
        score:   Optional quality score — reserved slot for the bullshit filter.

    Returns:
        Queue depth after insert (int).
    """
    _check_init()
    max_q = int(config.get_db_sync().get("PERSIST_QUEUE_MAX", "500"))
    now  = datetime.now(timezone.utc).isoformat()

    async with aiosqlite.connect(_db_path) as db:
        await db.execute(
            """INSERT INTO persist_queue
               (target_table, payload, source, model, score, queued_at)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (table, json.dumps(payload, ensure_ascii=False, default=str),
             source, model, score, now),
        )
        # Queue begrenzen — älteste zuerst raus (gleiche Regel wie tool_cache)
        await db.execute(
            "DELETE FROM persist_queue WHERE id NOT IN "
            "(SELECT id FROM persist_queue ORDER BY id DESC LIMIT ?)",
            (max_q,),
        )
        await db.commit()
        cursor = await db.execute("SELECT COUNT(*) FROM persist_queue")
        return int((await cursor.fetchone())[0])


async def _flush_locked(limit: int = 50) -> Dict[str, Any]:
    """
    Try to move queued entries to PostgreSQL. NEVER raises.
    (Single-flight: always called through flush_persist_queue(), which holds the lock.)

    No writer  → no-op with reason (entries keep waiting — no crash).
    Writer set → CREATE TABLE IF NOT EXISTS + INSERT per entry + queue delete.
    Failure    → attempts+1 + sanitized last_error, entry stays queued.

    Args:
        limit: Max entries per flush run (bounded work per interval).

    Returns:
        Dict: {flushed, errors, remaining, reason}
    """
    _check_init()

    if not _psql_writer:
        return {
            "flushed": 0, "errors": 0,
            "remaining": await _queue_count(),
            "reason": "no_postgres_writer",
        }

    # Neon asleep / unreachable a moment ago → don't even try (entries stay queued, nothing blocks)
    if time.monotonic() < _pg_down_until:
        return {
            "flushed": 0, "errors": 0,
            "remaining": await _queue_count(),
            "reason": "pg_backoff",
        }

    flushed = errors = 0
    async with aiosqlite.connect(_db_path) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(
            "SELECT id, target_table, payload, source, model, score "
            "FROM persist_queue ORDER BY id ASC LIMIT ?",
            (limit,),
        )
        rows = await cursor.fetchall()

        for row in rows:
            table = row["target_table"]
            try:
                if not _SAFE_TABLE_NAME.match(table or ""):
                    raise ValueError("invalid table name")
                if table not in _pg_tables_ensured:
                    await _pg_exec(_PG_DDL.format(table=table))
                    _pg_tables_ensured.add(table)
                await _pg_exec(
                    f"INSERT INTO {table} (source, model, score, payload) "
                    "VALUES ($1, $2, $3, $4::jsonb)",
                    row["source"], row["model"], row["score"], row["payload"],
                )
                await db.execute("DELETE FROM persist_queue WHERE id = ?", (row["id"],))
                await db.commit()
                flushed += 1
            except Exception as e:
                errors += 1
                # Sanitized: nur Exception-TYP — niemals DSN/Payload in Logs oder DB
                await db.execute(
                    "UPDATE persist_queue SET attempts = attempts + 1, last_error = ? "
                    "WHERE id = ?",
                    (type(e).__name__, row["id"]),
                )
                await db.commit()
                if _is_conn_failure(e):
                    break   # DB unreachable: stop here — don't burn one timeout per remaining row

    remaining = await _queue_count()
    if flushed:
        logger.info(f"Persist flush: {flushed} → PostgreSQL, {remaining} queued.")
    if errors:
        logger.warning(f"Persist flush: {errors} failed — stay queued (retry next interval).")
    return {"flushed": flushed, "errors": errors, "remaining": remaining}


# =============================================================================
# SECTION 6b — Neon / PostgreSQL resilience (scale-to-zero friendly)
# Neon suspends idle compute; the first connection afterwards can be slow or time out.
# Principle: the hub NEVER waits for PostgreSQL on a request path and never crashes.
#   - every PG write has a hard time limit (PG_WRITE_TIMEOUT_SEC)
#   - after a connection failure PG is skipped for PG_BACKOFF_SEC (entries stay queued)
#   - request path waits at most PG_REQUEST_WAIT_SEC; the flush continues in background
#   - pg_status() = cheap "is the DB awake?" ping for health checks (it also wakes Neon)
#   - optional keepalive (PG_KEEPALIVE_SEC, default 0 = off): keeps Neon awake, costs compute time
# All knobs in .pyfun [DB_SYNC].
# =============================================================================

_pg_down_until: float = 0.0
_pg_last: Dict[str, Any] = {"at": -1e9}
_flush_lock: Optional[asyncio.Lock] = None
_bounded_task: Optional[asyncio.Task] = None
_keepalive_task: Optional[asyncio.Task] = None


def _pg_cfg(key: str, default: float) -> float:
    try:
        return float(config.get_db_sync().get(key, default))
    except (TypeError, ValueError):
        return float(default)


def _is_conn_failure(e: BaseException) -> bool:
    """Timeout / socket / asyncpg connection-class errors (matched by name — no asyncpg import here)."""
    name = type(e).__name__
    return isinstance(e, (asyncio.TimeoutError, OSError)) or "Connection" in name or "Interface" in name


def _pg_mark(ok: bool) -> None:
    global _pg_down_until
    _pg_down_until = 0.0 if ok else time.monotonic() + _pg_cfg("PG_BACKOFF_SEC", 30)


async def _pg_exec(query: str, *args):
    """One PG write with a hard time limit. Connection-class failures arm the backoff."""
    try:
        result = await asyncio.wait_for(_psql_writer(query, *args), timeout=_pg_cfg("PG_WRITE_TIMEOUT_SEC", 10))
        _pg_mark(True)
        return result
    except Exception as e:
        if _is_conn_failure(e):
            _pg_mark(False)
        raise


async def flush_persist_queue(limit: int = 50) -> Dict[str, Any]:
    """Single-flight flush (lock): timer + request path can never insert the same row twice."""
    global _flush_lock
    _check_init()
    if _flush_lock is None:
        _flush_lock = asyncio.Lock()
    async with _flush_lock:
        return await _flush_locked(limit)


async def flush_persist_queue_bounded(limit: int = 50, wait_sec: Optional[float] = None) -> Dict[str, Any]:
    """
    Request-path flush: waits at most PG_REQUEST_WAIT_SEC. If PostgreSQL is slow/asleep the
    flush keeps running in the background (shielded) and the caller gets reason "pg_slow".
    """
    global _bounded_task
    wait = _pg_cfg("PG_REQUEST_WAIT_SEC", 3) if wait_sec is None else wait_sec
    if _bounded_task is None or _bounded_task.done():
        _bounded_task = asyncio.create_task(flush_persist_queue(limit))
    try:
        return await asyncio.wait_for(asyncio.shield(_bounded_task), timeout=wait)
    except asyncio.TimeoutError:
        return {"flushed": 0, "errors": 0, "remaining": await _queue_count(), "reason": "pg_slow"}


async def pg_status(force: bool = False) -> Dict[str, Any]:
    """
    Is PostgreSQL awake? Cheap ping (SELECT 1) with a short limit, cached PG_STATUS_TTL_SEC.
    state: "up" | "no_response" (asleep/waking/unreachable) | "error"; bridge: active|inactive.
    The ping itself wakes a suspended Neon compute.
    """
    if not _psql_writer:
        return {"bridge": "inactive", "state": "n/a"}
    now = time.monotonic()
    if not force and now - _pg_last["at"] < _pg_cfg("PG_STATUS_TTL_SEC", 20):
        return {k: v for k, v in _pg_last.items() if k != "at"}

    t0 = time.monotonic()
    state, err = "up", None
    try:
        await asyncio.wait_for(_psql_writer("SELECT 1"), timeout=_pg_cfg("PG_PING_TIMEOUT_SEC", 5))
        _pg_mark(True)
    except asyncio.TimeoutError:
        state = "no_response"
        _pg_mark(False)
    except Exception as e:
        state, err = "error", type(e).__name__
        if _is_conn_failure(e):
            _pg_mark(False)
    _pg_last.clear()
    _pg_last.update({
        "at": time.monotonic(), "bridge": "active", "state": state,
        "latency_ms": int((time.monotonic() - t0) * 1000), "error": err,
        "backoff_active": time.monotonic() < _pg_down_until,
    })
    return {k: v for k, v in _pg_last.items() if k != "at"}


async def _keepalive_loop(interval: float) -> None:
    while True:
        await asyncio.sleep(interval)
        try:
            await pg_status(force=True)
        except Exception as e:
            logger.debug(f"Keepalive ping skipped: {type(e).__name__}")


async def _queue_count() -> int:
    async with aiosqlite.connect(_db_path) as db:
        cursor = await db.execute("SELECT COUNT(*) FROM persist_queue")
        return int((await cursor.fetchone())[0])


async def persist_queue_stats() -> Dict[str, Any]:
    """
    Zustand des Gedächtnisses — Observability für die Orchestrierung.
    Wird von persist_status (MCP + /api) gelesen.
    """
    _check_init()
    async with aiosqlite.connect(_db_path) as db:
        cursor = await db.execute(
            "SELECT COUNT(*), MIN(queued_at), "
            "SUM(CASE WHEN attempts > 0 THEN 1 ELSE 0 END) FROM persist_queue"
        )
        row = await cursor.fetchone()
    count, oldest, failed = row[0] or 0, row[1], row[2] or 0
    return {
        "queue_depth": count,
        "oldest_queued_at": oldest,
        "entries_with_failed_attempts": failed,
        "postgres_bridge": "active" if _psql_writer else "inactive (DATABASE_URL not set)",
        "note": "queue is restart-safe — flushes to PostgreSQL as soon as the bridge is active",
    }


def start_persist_flusher() -> None:
    """
    Background retry loop — started by app.py AFTER db_sync.initialize().
    Works with AND without writer: ohne PG bleibt alles einfach liegen.
    """
    global _flush_task
    if _flush_task is not None:
        return
    interval = float(config.get_db_sync().get("PERSIST_FLUSH_INTERVAL_SEC", "60"))
    _flush_task = asyncio.create_task(_persist_flush_loop(interval))
    logger.info(f"Persist flusher started (interval: {interval}s).")

    global _keepalive_task
    ka = _pg_cfg("PG_KEEPALIVE_SEC", 0)
    if ka > 0 and _keepalive_task is None:
        _keepalive_task = asyncio.create_task(_keepalive_loop(ka))
        logger.info(f"PostgreSQL keepalive ping every {int(ka)}s (keeps Neon awake — uses compute time).")


async def _persist_flush_loop(interval: float) -> None:
    # Erster Versuch kurz nach dem Boot — fängt Einträge der letzten Session ab
    # (genau dann, wenn PG wieder da ist, während er vorher down war).
    await asyncio.sleep(5)
    while True:
        try:
            result = await flush_persist_queue()
            if result.get("flushed") or result.get("errors"):
                logger.info(f"Persist flusher: {result}")
        except Exception as e:
            logger.warning(f"Persist flusher error (non-fatal): {type(e).__name__}")
        await asyncio.sleep(interval)


async def persist(table: str, data: dict) -> None:
    """
    Legacy direct-write — jetzt queue-safe: enqueue + sofortiger Flush-Versuch.
    Beißt nicht mehr ohne DATABASE_URL (früher: RuntimeError). Backward-kompatibel
    für externe Aufrufer; tools.py nutzt jetzt enqueue_persist() direkt.
    """
    await enqueue_persist(table, payload=data)
    await flush_persist_queue(limit=1)

# =============================================================================
# Direct execution guard
# =============================================================================

if __name__ == "__main__":
    print("WARNING: Run via main.py → app.py, not directly.")