PolyGateway/src/polygateway/telemetry/sqlite.py

"""SQLite 遥测后端(默认): WAL + 单持久连接 + to_thread 桥接。

蓝本 VT `adapters/telemetry.py`: 构造期建连接与表,失败降级为 no-op
(记录基础设施不得拖垮业务调用);`INSERT OR IGNORE` 幂等(call_id 主键);
写入经 threading.Lock 串行化后由 `asyncio.to_thread` 执行,不阻塞事件循环。

**这里不做 postgres.py 那样的建表前探测,是实测后的有意不对称**(issue #9):
SQLite 对已存在的表在**解析期**就把 `CREATE TABLE IF NOT EXISTS` 短路掉,
既不抢写锁也不检查可写性——实测同一时刻另一连接持 `BEGIN EXCLUSIVE`、或
文件 `chmod 444`,该语句均通过,而同条件下的 `INSERT` 与新表名建表分别报
database is locked / readonly database。故 PG 侧"权限检查早于存在性判断"
的坑在此不存在,加探测零收益。**别为了代码对称把它加回来**;需要对称的是
保证(表存在就不该因建表失败而失能),这一条两侧都已满足。
"""

from __future__ import annotations

import asyncio
import sqlite3
import threading
from pathlib import Path
from typing import TYPE_CHECKING

from loguru import logger

from polygateway.telemetry.schema import (
    COLUMNS,
    SQLITE_BACKFILL,
    SQLITE_DDL,
    insert_sql,
    missing_columns_warning,
)
from polygateway.telemetry.status import TelemetryStatusTracker

if TYPE_CHECKING:
    from polygateway.types import TelemetryStatus


class SQLiteRecorder:
    """TelemetryRecorder 端口的 SQLite 实现;初始化/写入失败全降级 warning。"""

    def __init__(self, db_path: Path | str, *, auto_migrate: bool) -> None:
        """建连接与表,并按探测到的列定型本实例的 INSERT 语句。

        Args:
            db_path: 库文件路径;父目录不存在会自动创建。
            auto_migrate: True 则给已存在的旧表自动补列(SQLite 侧的缺省档:
                下游本地文件,无 DBA 无迁移工具);False 则一条 ALTER 都不发,
                改为按现有列裁剪写入。keyword-only **必填**: 缺省规则只写在
                config 一处,不与本类签名漂移(设计 D-c)。
        """
        self._auto_migrate = auto_migrate
        # SQLite 侧本次只做可见性: 它的失败模式(目录不可写、文件损坏)在装配期
        # 就暴露给下游,不是"跑到一半悄悄断",故降级恒为 fatal,不做冷却重连
        self._status = TelemetryStatusTracker(backend="sqlite")
        self._lock = threading.Lock()
        self._conn: sqlite3.Connection | None = None
        # 先按全量列定型: 连接失败/探测失败时保守沿用全量(今天的行为)
        self._columns: tuple[str, ...] = COLUMNS
        self._insert = insert_sql("sqlite", COLUMNS)
        try:
            path = Path(db_path)
            path.parent.mkdir(parents=True, exist_ok=True)
            conn = sqlite3.connect(path, check_same_thread=False, timeout=10.0)
            conn.execute("PRAGMA journal_mode=WAL")
            conn.execute("PRAGMA busy_timeout=5000")
            conn.execute(SQLITE_DDL)
            conn.commit()
            self._conn = conn
        except (OSError, sqlite3.Error) as exc:
            self._status.enter_degraded(f"初始化失败: {exc}", fatal=True, cooldown_s=None)
        self._prepare_columns()

    @property
    def telemetry_status(self) -> TelemetryStatus:
        """当前可写状态快照(ports.TelemetryStatusProvider)。"""
        return self._status.snapshot()

    def _prepare_columns(self) -> None:
        """探测现有列后定型写入: auto 档补齐缺列,manual 档改为裁剪写入(issue #13)。

        必须放在 `self._conn` 赋值**之后**并先判空: 初始化失败时连接为 None,
        无守卫的探测会抛 AttributeError 逃出 `__init__`,把"静默降级"变成崩溃。
        探测失败保守沿用全量列(今天的行为): 猜不出真实列集合时,让写入照常尝试。
        """
        if self._conn is None:
            return
        try:
            existing = {row[1] for row in self._conn.execute("PRAGMA table_info(llm_calls)")}
        except sqlite3.Error as exc:
            logger.warning("SQLite 遥测列探测失败(沿用全量列,写入将逐行降级): {}", exc)
            return
        if self._auto_migrate:
            self._backfill_columns(existing)
            return
        self._adopt_existing_columns(existing)

    def _adopt_existing_columns(self, existing: set[str]) -> None:
        """manual 档: 不发任何 DDL,按现有列裁剪 INSERT,并把缺列一次讲清楚。

        裁剪是关掉 ALTER 的**前提**而非增强: 旧表缺列时仍发全量 INSERT,每一行
        都会因未知列被拒 → 遥测彻底丢失,比自动 ALTER 更严重地违反"遥测必录"。
        探测结果与 `COLUMNS` 毫无交集时视同探测异常保守回落全量: 空列集拼不出合法
        INSERT,`insert_sql` 会 ValueError,而遥测构造期抛异常就是把"初始化失败静默
        降级"的铁律破成崩溃——回落必须发生在把空列集交给它之前。
        """
        effective = tuple(column for column in COLUMNS if column in existing)
        if not effective:
            logger.warning(
                "SQLite 遥测表 llm_calls 没有任何本库认识的列(沿用全量列,写入将逐行降级);"
                "现有列: {}",
                sorted(existing),
            )
            return
        self._columns = effective
        self._insert = insert_sql("sqlite", effective)
        missing = [column for column in COLUMNS if column not in existing]
        if missing:
            # 单参数传入: 补列 SQL 里带 `'{}'` 字面量,拼进 format 模板会被当占位符
            logger.warning(
                "{}",
                missing_columns_warning("sqlite", missing, alien_table="call_id" not in existing),
            )

    def _backfill_columns(self, existing: set[str]) -> None:
        """auto 档: 给已存在的旧表补新列(issue #3);逐列独立 try,失败只降级为逐行丢弃。

        补列失败绝不清空 `self._conn`——那会让整个 recorder 永久 no-op,
        比逐行丢弃严重得多。失败后写入沿用全量列(今天的行为): auto 档承诺的是
        "把列补上",补不上就让缺列以逐行 warning 暴露;要降级写入请显式选 manual。
        """
        assert self._conn is not None  # 内部不变量: 调用方已判空
        for column, decl in SQLITE_BACKFILL:
            if column in existing:
                continue
            # 逐列独立 try: 一列撞上 duplicate 不得让后面的列漏补
            try:
                self._conn.execute(f"ALTER TABLE llm_calls ADD COLUMN {column} {decl}")
                self._conn.commit()
            except sqlite3.Error as exc:
                # duplicate column: 多进程共库时后到者必然撞上,属预期竞态,视为成功
                if "duplicate column" not in str(exc).lower():
                    logger.warning("SQLite 遥测补列失败(写入将逐行降级): {}", exc)

    async def record_llm_call(self, **fields: object) -> None:
        """写一行遥测;字段集合即 24 字段冻结签名(ports.TelemetryRecorder)。

        取值按 `self._columns`(manual 档可能已被裁剪),与 `self._insert` 的
        占位符同序——两者必须一起改,分开改就是把值写进错位的列。
        """
        if self._conn is None:
            # 改前这里是**裸 return**: 初始化失败后每一行都无声消失,长跑进程里
            # 与"遥测正常"外观上完全一致(设计 §1.4 的直接钉子)
            self._status.record_drop("遥测已降级")
            return
        row = tuple(fields[col] for col in self._columns)
        try:
            await asyncio.to_thread(self._write, row)
        except (OSError, sqlite3.Error) as exc:
            logger.warning("SQLite 遥测写入失败(降级不冒泡): {}", exc)

    def _write(self, row: tuple) -> None:
        assert self._conn is not None  # 内部不变量: 调用方已判空
        with self._lock:
            self._conn.execute(self._insert, row)
            self._conn.commit()

    def close(self) -> None:
        """幂等关闭持久连接。"""
        conn, self._conn = self._conn, None
        if conn is not None:
            with self._lock:
                conn.close()