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 loguru import logger

from polygateway.telemetry.schema import COLUMNS, SQLITE_BACKFILL, SQLITE_DDL, insert_sql

# 缺列 warning 要打印可直接执行的补列语句,列定义与库内 ALTER 同源(不许两份)
_BACKFILL_DECLS = dict(SQLITE_BACKFILL)


def _missing_columns_message(missing: list[str], *, alien_table: bool) -> str:
    """拼 manual 档的缺列告警: 逐列点名 + 讲清后果 + 给出可直接执行的 SQL。

    只说"缺列"是不够的: 静默丢维度的后果是多租户账目全归空串且无任何报错,
    看告警的人必须一眼看到丢的是哪几个维度、以及怎么补。

    Args:
        missing: 缺失的列名(按 `COLUMNS` 保序)。
        alien_table: 连主键列 `call_id` 都没有——该表多半不是本库的 `llm_calls`。

    Returns:
        单条 warning 的完整文本(库只在准备期发一次,不逐行发)。
    """
    statements = [
        f"ALTER TABLE llm_calls ADD COLUMN {column} {_BACKFILL_DECLS[column]};"
        for column in missing
        if column in _BACKFILL_DECLS
    ]
    unknown = [column for column in missing if column not in _BACKFILL_DECLS]
    if unknown:
        # 这些列本库从未经 ALTER 补过(建表即有),给不出单条 ALTER,指向完整脚本
        statements.append(
            f"-- 另缺 {', '.join(unknown)};完整建表脚本见 "
            'polygateway.telemetry_schema_sql("sqlite")'
        )
    head = (
        "SQLite 遥测表 llm_calls 缺主键列 call_id,很可能不是本库的遥测表"
        "(库不做二次判定,仍照常尝试写入)"
        if alien_table
        else "SQLite 遥测表 llm_calls 缺列,且 auto_migrate=False(库不发任何 DDL)"
    )
    return f"{head};以下维度不会被记录: {', '.join(missing)}。补列请自行执行:\n" + "\n".join(
        statements
    )


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
        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:
            logger.warning("SQLite 遥测初始化失败,后续记录降级为 no-op: {}", exc)
        self._prepare_columns()

    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 INTO llm_calls () VALUES ()` 这种语法非法的语句(`insert_sql` 拦
        不住——空集技术上是子集),必须在交给它之前拦下。
        """
        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_message(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
        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()