"""遥测子系统测试: SQLiteRecorder(22 列)+ TelemetryEmitter 单一 helper + TelemetryMW。"""

import asyncio
import copy
import json
import os
import sqlite3
import subprocess
from pathlib import Path

import pytest

from polygateway.backends.memory.breaker import InMemoryGate
from polygateway.backends.memory.limiter import InMemoryLimiter
from polygateway.embedding import EmbeddingClient
from polygateway.errors import CircuitOpenError, RequestRejectedError
from polygateway.middleware.cache import digest_messages
from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW
from polygateway.ocr import OcrClient
from polygateway.pricing import ModelPrice, PricingTable
from polygateway.sources import RoundRobinSelector
from polygateway.telemetry.sqlite import SQLiteRecorder
from polygateway.types import (
    BackpressurePolicy,
    BreakerConfig,
    ChatRequest,
    EmbeddingTransportResult,
    GlobalLimits,
    LLMResponse,
    OcrTextTransportResult,
    RetryPolicy,
    SourceConfig,
)

_REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}], session_id="sess-1")

_EXPECTED_COLUMNS = [
    "call_id",
    "parent_call_id",
    "session_id",
    "model",
    "provider",
    "source_name",
    "messages",
    "response",
    "thinking",
    "prompt_tokens",
    "completion_tokens",
    "usage_source",
    "latency_ms",
    "ttft_ms",
    "max_inter_token_ms",
    "cache_hit",
    "error",
    "cost",
    "created_at",
    "cached_prompt_tokens",
    "model_reported",
    "sampling",
    "reasoning_tokens",
    "tenant_id",
    "meta",
]


def _resp(**overrides):
    base = {
        "content": "ok",
        "thinking": "",
        "model": "m",
        "provider": "p",
        "prompt_tokens": 1,
        "completion_tokens": 2,
        "latency_ms": 30,
        "ttft_ms": None,
        "max_inter_token_ms": None,
        "cache_hit": False,
        "call_id": "cid-1",
        "source_name": "s1",
        "usage_source": "measured",
    }
    base.update(overrides)
    return LLMResponse(**base)


def _source(**overrides):
    base = {
        "name": "s1",
        "provider": "p",
        "base_url": "https://gw.example/v1",
        "api_key": "sk",
        "model": "m",
        "timeout_s": 10.0,
    }
    base.update(overrides)
    return SourceConfig(**base)


# 输出单价 8 元/百万: 改前 `unavailable` 行按兜底的 0/4000 换算恰好是 0.032
_PRICING = PricingTable({"m": ModelPrice(input_per_1m=1.0, output_per_1m=8.0)})


async def _record_minimal(recorder, call_id="c1", **overrides):
    fields = {
        "call_id": call_id,
        "parent_call_id": None,
        "session_id": "sess-1",
        "model": "m",
        "provider": "p",
        "source_name": "s1",
        "messages": "[]",
        "response": "ok",
        "thinking": "",
        "prompt_tokens": 1,
        "completion_tokens": 2,
        "usage_source": "measured",
        "latency_ms": 10,
        "ttft_ms": None,
        "max_inter_token_ms": None,
        "cache_hit": False,
        "error": None,
        "cost": None,
        "cached_prompt_tokens": None,
        "model_reported": None,
        "sampling": None,
        "reasoning_tokens": None,
        # 到达 recorder 时已由 emitter 归一化: None → '',空 dict → '{}'
        "tenant_id": "",
        "meta": "{}",
    }
    fields.update(overrides)
    await recorder.record_llm_call(**fields)


@pytest.fixture
def captured_warnings():
    """捕获库发出的 WARNING;loguru 不经标准 logging,pytest 的 caplog 抓不到。

    名字避开裸 `warnings`: 那会遮蔽标准库模块名,本文件将来任何一次
    `import warnings` 都会与它静默互相顶掉,而报错点离真因很远。
    """
    from loguru import logger

    messages: list[str] = []
    sink_id = logger.add(messages.append, level="WARNING")
    yield messages
    logger.remove(sink_id)


# 搬迁前(1.2.1)两个 recorder 各自持有的 INSERT 常量原文,逐字冻结在此。
# 这两条字符串是"纯搬迁不改行为"的机械证据: 构造逻辑换了地方,产物必须一字不差。
_FROZEN_SQLITE_INSERT = (
    "INSERT OR IGNORE INTO llm_calls (call_id, parent_call_id, session_id, model, provider, "
    "source_name, messages, response, thinking, prompt_tokens, completion_tokens, usage_source, "
    "latency_ms, ttft_ms, max_inter_token_ms, cache_hit, error, cost, cached_prompt_tokens, "
    "model_reported, sampling, reasoning_tokens, tenant_id, meta) "
    "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
)
_FROZEN_PG_INSERT = (
    "INSERT INTO llm_calls (call_id, parent_call_id, session_id, model, provider, source_name, "
    "messages, response, thinking, prompt_tokens, completion_tokens, usage_source, latency_ms, "
    "ttft_ms, max_inter_token_ms, cache_hit, error, cost, cached_prompt_tokens, model_reported, "
    "sampling, reasoning_tokens, tenant_id, meta) "
    "VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, "
    "$19, $20, $21, $22, $23, $24) "
    # 无冲突目标(issue #13 Task 2): 带 `(call_id)` 的版本在按 created_at 分区、
    # 主键为 (call_id, created_at) 的表上匹配不到约束,PG 直接拒收整条写入
    "ON CONFLICT DO NOTHING"
)


def _first_occurrence_order(text: str, names: list[str]) -> list[str]:
    """按各列名在 text 中首次出现的位置排序,用于比对"列名出现顺序"。"""
    found = [(text.index(name), name) for name in names if name in text]
    return [name for _, name in sorted(found)]


class TestSchemaModule:
    """`telemetry/schema.py` 是 schema 单一事实源(issue #13 Task 1)。

    库执行的 DDL 与打印给下游的 SQL 必须同源: 常量分散在两个 recorder 里各存一份时,
    公共函数再写一份就是三份,漂移的表现是"下游照打印的 SQL 建完表,库仍报缺列"。
    """

    def test_columns_and_ddl_are_frozen(self):
        """列序与两端 DDL 逐字未变(搬迁不得改动任何一个字符)。"""
        from polygateway.telemetry.schema import COLUMNS, PG_BACKFILL, PG_DDL, SQLITE_DDL

        # COLUMNS 是 INSERT 字段序,不含数据库自填的 created_at
        assert list(COLUMNS) == [c for c in _EXPECTED_COLUMNS if c != "created_at"]
        assert len(COLUMNS) == 24
        # 两端 DDL 的列出现顺序 == 物理列序(created_at 在第 19 位)
        for ddl in (SQLITE_DDL, PG_DDL):
            assert _first_occurrence_order(ddl, _EXPECTED_COLUMNS) == _EXPECTED_COLUMNS
        assert "CREATE TABLE IF NOT EXISTS llm_calls" in SQLITE_DDL
        assert "created_at         TEXT    NOT NULL DEFAULT (datetime('now'))" in SQLITE_DDL
        assert "created_at         TIMESTAMPTZ NOT NULL DEFAULT now()" in PG_DDL
        assert "meta                 JSONB NOT NULL DEFAULT '{}'::jsonb" in PG_DDL
        # 库内执行的补列语句不带 IF NOT EXISTS(它即便列已存在也先取 ACCESS EXCLUSIVE 锁)
        assert PG_BACKFILL[0] == (
            "cached_prompt_tokens",
            "ALTER TABLE llm_calls ADD COLUMN cached_prompt_tokens INTEGER",
        )
        assert PG_BACKFILL[-1] == (
            "meta",
            "ALTER TABLE llm_calls ADD COLUMN meta JSONB NOT NULL DEFAULT '{}'::jsonb",
        )
        assert all("IF NOT EXISTS" not in stmt for _, stmt in PG_BACKFILL)

    def test_insert_sql_reproduces_the_frozen_statements(self):
        """`insert_sql(backend, COLUMNS)` 与搬迁前的 `_INSERT` 一致(PG 侧去掉冲突目标)。"""
        from polygateway.telemetry.schema import COLUMNS, insert_sql

        assert insert_sql("sqlite", COLUMNS) == _FROZEN_SQLITE_INSERT
        assert insert_sql("postgres", COLUMNS) == _FROZEN_PG_INSERT
        # 裁剪列表按位置占位符重新编号,不留空洞
        assert insert_sql("postgres", ["call_id", "model"]) == (
            "INSERT INTO llm_calls (call_id, model) VALUES ($1, $2) ON CONFLICT DO NOTHING"
        )
        # 冲突目标不得被"顺手"补回: 分区表上它会让每一条遥测都被 PG 拒收
        assert "ON CONFLICT (" not in insert_sql("postgres", COLUMNS)

    def test_insert_sql_rejects_foreign_columns_and_backends(self):
        """列名来自数据库探测结果而非常量,子集校验是唯一的注入面闸门。"""
        from polygateway.telemetry.schema import COLUMNS, insert_sql

        with pytest.raises(ValueError, match="call_id_x"):
            insert_sql("sqlite", ["call_id_x"])
        with pytest.raises(ValueError):
            insert_sql("sqlite", ["call_id", "meta); DROP TABLE llm_calls; --"])
        with pytest.raises(ValueError, match="mysql"):
            insert_sql("mysql", COLUMNS)

    def test_insert_sql_rejects_an_empty_column_set(self):
        """空列集合两端都拼出语法非法的 SQL,构造器自己拒,不押在调用方的不变量上。

        入参来自数据库探测结果: 探测到一张与本库毫无共同列的同名表,`effective`
        就是空的。真放行会产出 `INSERT OR IGNORE INTO llm_calls () VALUES ()`,
        错误要到执行时才由数据库报,离真因很远。
        """
        from polygateway.telemetry.schema import insert_sql

        for backend in ("sqlite", "postgres"):
            with pytest.raises(ValueError, match="至少需要一列"):
                insert_sql(backend, [])

    def test_schema_sql_is_paste_ready_and_same_source(self):
        """打印给下游的脚本与库执行的 DDL 同源,且对人可重复执行。"""
        from polygateway.telemetry.schema import PG_BACKFILL, SQLITE_BACKFILL, telemetry_schema_sql

        pg = telemetry_schema_sql("postgres")
        lite = telemetry_schema_sql("sqlite")
        for script in (pg, lite):
            # 24 个 INSERT 字段 + created_at 全在,且首次出现顺序与建表 DDL 一致
            assert _first_occurrence_order(script, _EXPECTED_COLUMNS) == _EXPECTED_COLUMNS
            assert "CREATE TABLE IF NOT EXISTS llm_calls" in script
        # 人执行的那份必须幂等: PG 用 ADD COLUMN IF NOT EXISTS(与库内那份有意不同)
        for column, _ in PG_BACKFILL:
            assert f"ALTER TABLE llm_calls ADD COLUMN IF NOT EXISTS {column} " in pg
        # SQLite 无该语法(写上去直接语法错误),只能以注释交代执行前提
        lite_alters = [line for line in lite.splitlines() if line.startswith("ALTER TABLE")]
        assert len(lite_alters) == len(SQLITE_BACKFILL)
        assert all("IF NOT EXISTS" not in line for line in lite_alters)
        for column, _ in SQLITE_BACKFILL:
            assert f"ALTER TABLE llm_calls ADD COLUMN {column} " in lite
        assert "不存在" in lite
        with pytest.raises(ValueError, match="mysql"):
            telemetry_schema_sql("mysql")


class TestBackendColumnParity:
    """两个后端的列清单必须逐字同名同序(issue #11)。

    emitter 只组装一份 `fields`,两个后端各按自己的清单取值;两份清单一旦分叉,
    同一次调用在 SQLite 上写得进、在 PG 上抛 KeyError 被降级吞掉,差异只在换后端时
    才暴露。issue #13 起两端共用 `schema.COLUMNS`,故这里断言的是"共用"本身
    (同一个对象则永远无从分叉),列**序**仍单独断言: INSERT 用位置占位符,
    顺序错位会把值写进错误的列而不报错。
    """

    def test_two_backends_agree_on_columns(self):
        from polygateway.telemetry import postgres, sqlite
        from polygateway.telemetry.schema import COLUMNS

        assert sqlite.COLUMNS is COLUMNS
        assert postgres.COLUMNS is COLUMNS

    def test_caller_dimensions_are_appended_last(self):
        """新列只能追加在末尾: 旧表经 ALTER 补列必落末尾,插在中间会让两条路径分叉。"""
        from polygateway.telemetry.schema import COLUMNS

        assert COLUMNS[-2:] == ("tenant_id", "meta")


class TestSQLiteRecorder:
    async def test_schema_has_frozen_columns(self, tmp_path):
        recorder = SQLiteRecorder(tmp_path / "t.db", auto_migrate=True)
        await _record_minimal(recorder)
        recorder.close()
        cols = [
            r[1] for r in sqlite3.connect(tmp_path / "t.db").execute("PRAGMA table_info(llm_calls)")
        ]
        assert cols == _EXPECTED_COLUMNS

    async def test_call_id_idempotent(self, tmp_path):
        recorder = SQLiteRecorder(tmp_path / "t.db", auto_migrate=True)
        await _record_minimal(recorder, call_id="dup")
        await _record_minimal(recorder, call_id="dup", response="second")
        recorder.close()
        rows = (
            sqlite3.connect(tmp_path / "t.db")
            .execute("SELECT response FROM llm_calls WHERE call_id='dup'")
            .fetchall()
        )
        assert rows == [("ok",)]  # INSERT OR IGNORE: 第二次静默忽略

    async def test_concurrent_writes_all_land(self, tmp_path):
        recorder = SQLiteRecorder(tmp_path / "t.db", auto_migrate=True)
        await asyncio.gather(*(_record_minimal(recorder, call_id=f"c{i}") for i in range(50)))
        recorder.close()
        (count,) = (
            sqlite3.connect(tmp_path / "t.db").execute("SELECT COUNT(*) FROM llm_calls").fetchone()
        )
        assert count == 50

    async def test_unwritable_path_degrades_silently(self):
        recorder = SQLiteRecorder(Path("/nonexistent-root/deep/t.db"), auto_migrate=True)
        await _record_minimal(recorder)  # 不抛
        recorder.close()

    async def test_observability_columns_round_trip(self, tmp_path):
        recorder = SQLiteRecorder(tmp_path / "t.db", auto_migrate=True)
        await _record_minimal(recorder, call_id="c-hit", cached_prompt_tokens=64)
        await _record_minimal(recorder, call_id="c-zero", cached_prompt_tokens=0)
        await _record_minimal(recorder, call_id="c-none", model_reported="MiniMax-Text-01")
        recorder.close()
        rows = dict(
            sqlite3.connect(tmp_path / "t.db")
            .execute("SELECT call_id, cached_prompt_tokens FROM llm_calls")
            .fetchall()
        )
        assert rows["c-hit"] == 64
        assert rows["c-zero"] == 0  # 真实零命中,读回仍是 0 而非 NULL
        assert rows["c-none"] is None

    async def test_reasoning_tokens_column_round_trip(self, tmp_path):
        """issue #6: 7 / 0 / None 三种值各自如实落库,0 与 NULL 不得混同。"""
        recorder = SQLiteRecorder(tmp_path / "t.db", auto_migrate=True)
        await _record_minimal(recorder, call_id="r-some", reasoning_tokens=7)
        await _record_minimal(recorder, call_id="r-zero", reasoning_tokens=0)
        await _record_minimal(recorder, call_id="r-none", reasoning_tokens=None)
        recorder.close()
        rows = dict(
            sqlite3.connect(tmp_path / "t.db")
            .execute("SELECT call_id, reasoning_tokens FROM llm_calls")
            .fetchall()
        )
        assert rows["r-some"] == 7
        assert rows["r-zero"] == 0  # 上报了且确实没推理
        assert rows["r-none"] is None  # 本次调用未上报

    async def test_sampling_column_round_trips(self, tmp_path):
        """issue #4: 采样参数落库,否则事后无法证明某批数据跑在什么温度下。"""
        recorder = SQLiteRecorder(tmp_path / "t.db", auto_migrate=True)
        await _record_minimal(recorder, call_id="c-s", sampling='{"seed": 42, "temperature": 0}')
        await _record_minimal(recorder, call_id="c-plain")
        recorder.close()
        rows = dict(
            sqlite3.connect(tmp_path / "t.db")
            .execute("SELECT call_id, sampling FROM llm_calls")
            .fetchall()
        )
        assert json.loads(rows["c-s"]) == {"seed": 42, "temperature": 0}
        assert rows["c-plain"] is None  # 无采样参数为 NULL,便于 SQL 过滤


class TestSQLiteColumnBackfill:
    """issue #3: 已存在的 18 列旧表必须自动补列,否则每行写入都被丢弃。"""

    _LEGACY_DDL = """
    CREATE TABLE llm_calls (
        call_id            TEXT PRIMARY KEY,
        parent_call_id     TEXT,
        session_id         TEXT,
        model              TEXT    NOT NULL,
        provider           TEXT    NOT NULL,
        source_name        TEXT    NOT NULL,
        messages           TEXT    NOT NULL,
        response           TEXT    NOT NULL,
        thinking           TEXT    NOT NULL DEFAULT '',
        prompt_tokens      INTEGER NOT NULL,
        completion_tokens  INTEGER NOT NULL,
        usage_source       TEXT    NOT NULL,
        latency_ms         INTEGER NOT NULL,
        ttft_ms            REAL,
        max_inter_token_ms REAL,
        cache_hit          INTEGER NOT NULL DEFAULT 0,
        error              TEXT,
        cost               REAL,
        created_at         TEXT    NOT NULL DEFAULT (datetime('now'))
    );
    """

    async def test_legacy_table_is_upgraded_in_place(self, tmp_path):
        db = tmp_path / "legacy.db"
        legacy = sqlite3.connect(db)
        legacy.execute(self._LEGACY_DDL)
        legacy.commit()
        legacy.close()

        recorder = SQLiteRecorder(db, auto_migrate=True)
        await _record_minimal(recorder, cached_prompt_tokens=7, model_reported="m-real")
        recorder.close()

        conn = sqlite3.connect(db)
        cols = [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")]
        assert cols == _EXPECTED_COLUMNS  # ALTER 追加到末尾,与新建库列序一致
        assert conn.execute(
            "SELECT cached_prompt_tokens, model_reported FROM llm_calls"
        ).fetchone() == (7, "m-real")

    async def test_backfill_failure_keeps_the_recorder_usable(self, tmp_path):
        """补列失败只能逐行降级,绝不能把 recorder 整体变成 no-op(设计 D1 纪律)。

        把 llm_calls 建成同名 view: `CREATE TABLE IF NOT EXISTS` 遇 view 静默
        no-op(不抛),随后的 ALTER 才抛 "Cannot add a column to a view"——正是
        补列失败这条分支。`_conn` 必须保持非 None,否则整个 recorder 永久失能。
        """
        db = tmp_path / "view.db"
        conn = sqlite3.connect(db)
        conn.execute("CREATE TABLE real_rows (call_id TEXT)")
        conn.execute("CREATE VIEW llm_calls AS SELECT call_id FROM real_rows")
        conn.commit()
        conn.close()

        recorder = SQLiteRecorder(db, auto_migrate=True)  # 不得抛
        assert recorder._conn is not None  # 补列失败 ≠ recorder 失能(D1 纪律)
        await _record_minimal(recorder)  # 不得抛
        recorder.close()


# issue #11 之前的表形态: 22 个 recorder 字段 + created_at = 23 个物理列,没有任何租户维度
_PRE_TENANT_DDL = """
CREATE TABLE llm_calls (
    call_id            TEXT PRIMARY KEY,
    parent_call_id     TEXT,
    session_id         TEXT,
    model              TEXT    NOT NULL,
    provider           TEXT    NOT NULL,
    source_name        TEXT    NOT NULL,
    messages           TEXT    NOT NULL,
    response           TEXT    NOT NULL,
    thinking           TEXT    NOT NULL DEFAULT '',
    prompt_tokens      INTEGER NOT NULL,
    completion_tokens  INTEGER NOT NULL,
    usage_source       TEXT    NOT NULL,
    latency_ms         INTEGER NOT NULL,
    ttft_ms            REAL,
    max_inter_token_ms REAL,
    cache_hit          INTEGER NOT NULL DEFAULT 0,
    error              TEXT,
    cost               REAL,
    created_at         TEXT    NOT NULL DEFAULT (datetime('now')),
    cached_prompt_tokens INTEGER,
    model_reported       TEXT,
    sampling             TEXT,
    reasoning_tokens     INTEGER
);
"""

_PRE_TENANT_INSERT = (
    "INSERT INTO llm_calls (call_id, model, provider, source_name, messages, response, "
    "prompt_tokens, completion_tokens, usage_source, latency_ms) "
    "VALUES ('old-row', 'm', 'p', 's1', '[]', 'old body', 1, 2, 'measured', 10)"
)


def _make_pre_tenant_db(path: Path) -> None:
    """造一个 issue #11 之前的库: 22 字段旧表 + 一行没有租户归属的历史数据。"""
    conn = sqlite3.connect(path)
    conn.execute(_PRE_TENANT_DDL)
    conn.execute(_PRE_TENANT_INSERT)
    conn.commit()
    conn.close()


class TestSQLiteCallerDimensionsAcceptance:
    """issue #11 的机械化验收(SQLite 侧,真实临时文件): 新建库 / 旧表补列 / 补列失败方向。"""

    async def test_fresh_db_round_trips_the_dimensions(self, tmp_path):
        """新建库: 列齐全,且维度值原样读回——只验列存在会漏掉写错列位的错。"""
        db = tmp_path / "fresh.db"
        recorder = SQLiteRecorder(db, auto_migrate=True)
        await _record_minimal(
            recorder, call_id="c-dim", tenant_id="tenant-a", meta='{"batch": "b7"}'
        )
        recorder.close()

        conn = sqlite3.connect(db)
        assert [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")] == _EXPECTED_COLUMNS
        row = conn.execute(
            "SELECT tenant_id, meta FROM llm_calls WHERE call_id = 'c-dim'"
        ).fetchone()
        assert row[0] == "tenant-a"
        assert json.loads(row[1]) == {"batch": "b7"}

    async def test_pre_tenant_table_gains_columns_and_old_rows_stay_auditable(self, tmp_path):
        """22 字段旧表补列后,新行带维度,而**老行的 tenant_id 是空串而非 NULL**。

        这条直接验收 issue #11 的核心论点(先启用落库、后加列,补列之前的行没有
        租户归属)。断言方向必须是空串: PG 的 RLS `USING` 表达式对返回 false **或
        NULL** 的行一律隐藏且不报错,故 NULL 的 `tenant_id` 不是"未归属",而是对
        所有人永久不可见的黑洞;哨兵空串则能被一条 `COUNT(*) WHERE tenant_id = ''`
        审计出来,历史欠账是可见、可量化、可补录的。
        """
        db = tmp_path / "pre_tenant.db"
        _make_pre_tenant_db(db)

        recorder = SQLiteRecorder(db, auto_migrate=True)
        await _record_minimal(recorder, call_id="new-row", tenant_id="tenant-a", meta='{"k": 1}')
        recorder.close()

        conn = sqlite3.connect(db)
        cols = [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")]
        assert cols == _EXPECTED_COLUMNS  # 22 → 24 个 recorder 字段(+ created_at 共 25 物理列)
        rows = dict(conn.execute("SELECT call_id, tenant_id FROM llm_calls").fetchall())
        assert rows["new-row"] == "tenant-a"
        assert rows["old-row"] == ""  # 不是 None: NULL 会被 RLS 静默吞掉
        assert (
            conn.execute("SELECT meta FROM llm_calls WHERE call_id = 'old-row'").fetchone()[0]
            == "{}"
        )

    async def test_readonly_file_backfill_failure_keeps_the_recorder_alive(self, tmp_path):
        """补列失败的降级方向(SQLite 等价构造: 文件只读)。

        SQLite 没有角色权限模型,与 PG「只有 SELECT/INSERT 权限的角色」等价的构造
        是文件本身只读。库文件必须**预先置为 WAL 且干净关闭**,否则 `__init__` 的
        `PRAGMA journal_mode=WAL` 会先撞上只读而让失败点跑到补列之前,测不到本用例
        要测的那条分支(实测: 非 WAL 库 chmod 444 后该 PRAGMA 报 readonly database)。
        只读库连 INSERT 都做不了,故这里**只断言**补列失败不清空 `_conn`、不抛出
        `__init__`(sqlite.py `_backfill_columns` 那条纪律),不断言"写入仍成功"。
        """
        if os.geteuid() == 0:
            pytest.skip("root 无视文件权限位,只读构造不成立")
        db = tmp_path / "readonly.db"
        conn = sqlite3.connect(db)
        conn.execute("PRAGMA journal_mode=WAL")  # 预置 WAL: 让只读连接不必改日志模式
        conn.execute(_PRE_TENANT_DDL)
        conn.execute(_PRE_TENANT_INSERT)
        conn.commit()
        conn.close()
        db.chmod(0o444)

        # finally 还原权限位: 任一断言先失败时,不还原会让 tmp_path 清理连带报错,
        # 把"某条断言失败"的真因盖成一个无关的 PermissionError
        try:
            recorder = SQLiteRecorder(db, auto_migrate=True)  # 不得抛
            assert recorder._conn is not None  # 补列失败 ≠ recorder 失能
            await _record_minimal(recorder, call_id="doomed")  # 只读库写不进,但不得抛
            recorder.close()
        finally:
            db.chmod(0o644)

        stale = sqlite3.connect(db)
        assert [r[1] for r in stale.execute("PRAGMA table_info(llm_calls)")] == (
            _EXPECTED_COLUMNS[:-2]
        )  # 补列确实没成功,用例不是在只读库上空转


class TestSQLiteSchemaMode:
    """issue #13: `auto_migrate` 两档——auto 保持自动补列,manual 只裁剪写入不发 DDL。

    列数断言一律按**物理列数**写: 旧表 22 个 INSERT 字段 + `created_at` = 23,
    补齐后 24 + `created_at` = 25。混用 INSERT 字段数与物理列数是本处最易错的地方。
    """

    def _physical_columns(self, db: Path) -> list[str]:
        conn = sqlite3.connect(db)
        try:
            return [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")]
        finally:
            conn.close()

    async def test_manual_mode_trims_the_insert_instead_of_altering(
        self, tmp_path, captured_warnings
    ):
        """manual + 22 字段旧表: 一条 ALTER 都不发,写入按现有列裁剪后照样落库。

        裁剪是关掉 ALTER 的前提: 不裁剪的话每行 INSERT 都撞 `no column named
        tenant_id` 而被整行丢弃——那是把自动补列换成静默全失能。
        """
        db = tmp_path / "manual_legacy.db"
        _make_pre_tenant_db(db)

        recorder = SQLiteRecorder(db, auto_migrate=False)
        await _record_minimal(recorder, call_id="new-row", tenant_id="tenant-a", meta='{"k": 1}')
        recorder.close()

        assert len(self._physical_columns(db)) == 23  # 未 ALTER: 物理列数原封不动
        conn = sqlite3.connect(db)
        assert conn.execute(
            "SELECT response, model FROM llm_calls WHERE call_id = 'new-row'"
        ).fetchone() == ("ok", "m")  # 裁剪后的列值仍对得上位
        conn.close()

        assert len(captured_warnings) == 1  # 缺列只讲一次,不逐行刷屏
        message = captured_warnings[0]
        assert "tenant_id" in message and "meta" in message  # 逐列点名
        assert "不会被记录" in message  # 讲清后果
        assert "ALTER TABLE" in message  # 给出可直接执行的补列 SQL

    async def test_auto_mode_still_upgrades_the_legacy_table(self, tmp_path):
        """auto + 同款旧表: 现状回归,补列后物理列数 23 → 25。"""
        db = tmp_path / "auto_legacy.db"
        _make_pre_tenant_db(db)

        recorder = SQLiteRecorder(db, auto_migrate=True)
        await _record_minimal(recorder, call_id="new-row", tenant_id="tenant-a")
        recorder.close()

        assert self._physical_columns(db) == _EXPECTED_COLUMNS
        assert len(self._physical_columns(db)) == 25

    async def test_manual_mode_still_creates_a_fresh_table(self, tmp_path):
        """manual 只管 ALTER,不管 CREATE: 全新库照建,25 个物理列齐全(设计 §4.2)。"""
        db = tmp_path / "manual_fresh.db"
        recorder = SQLiteRecorder(db, auto_migrate=False)
        await _record_minimal(recorder, call_id="c-fresh", tenant_id="tenant-a")
        recorder.close()

        assert self._physical_columns(db) == _EXPECTED_COLUMNS
        conn = sqlite3.connect(db)
        assert (
            conn.execute("SELECT tenant_id FROM llm_calls WHERE call_id = 'c-fresh'").fetchone()[0]
            == "tenant-a"
        )
        conn.close()

    async def test_table_without_call_id_escalates_the_wording(self, tmp_path, captured_warnings):
        """缺主键列 call_id = 该表压根不是本库的 llm_calls: 措辞升级,但库不做二次判定。"""
        db = tmp_path / "alien.db"
        conn = sqlite3.connect(db)
        conn.execute("CREATE TABLE llm_calls (model TEXT, provider TEXT)")
        conn.commit()
        conn.close()

        recorder = SQLiteRecorder(db, auto_migrate=False)  # 不得抛
        await _record_minimal(recorder)  # 照常尝试写入
        recorder.close()

        message = "\n".join(captured_warnings)
        assert "call_id" in message
        assert "不是本库" in message

    async def test_no_recognizable_column_falls_back_to_the_full_column_set(
        self, tmp_path, captured_warnings
    ):
        """探测结果与 COLUMNS 毫无交集视同探测异常: 保守回落全量列。

        `insert_sql` 自己拒空列集合(见 `test_insert_sql_rejects_an_empty_column_set`),
        故这里回落不发生就不是"拼出空语句",而是 ValueError 逃出 `__init__` ——
        遥测初始化失败必须静默降级,崩溃比丢维度严重得多。
        """
        from polygateway.telemetry.schema import COLUMNS

        db = tmp_path / "foreign.db"
        conn = sqlite3.connect(db)
        conn.execute("CREATE TABLE llm_calls (foo TEXT, bar TEXT)")
        conn.commit()
        conn.close()

        recorder = SQLiteRecorder(db, auto_migrate=False)  # 不得抛
        assert recorder._columns == COLUMNS
        await _record_minimal(recorder)  # 写不进去,但只逐行 warning,不抛
        recorder.close()
        assert captured_warnings  # 沉默地退化成空语句是最坏结果,必须有声

    async def test_empty_probe_result_degrades_instead_of_raising(
        self, tmp_path, captured_warnings
    ):
        """探测返回空集合时走回落,绝不让 `insert_sql` 的 ValueError 逃出去。

        SQLite 建不出零列的表,故直接喂空探测结果调那条分支——它正是
        `insert_sql` 拒空之后唯一可能把"静默降级"变成崩溃的入口。
        """
        from polygateway.telemetry.schema import COLUMNS

        db = tmp_path / "empty_probe.db"
        recorder = SQLiteRecorder(db, auto_migrate=False)
        recorder._adopt_existing_columns(set())  # 不得抛
        assert recorder._columns == COLUMNS
        await _record_minimal(recorder, call_id="c-after")  # 写入照常
        recorder.close()

        conn = sqlite3.connect(db)
        assert conn.execute(
            "SELECT response FROM llm_calls WHERE call_id = 'c-after'"
        ).fetchone() == ("ok",)
        conn.close()
        assert [m for m in captured_warnings if "没有任何本库认识的列" in m]  # 只有 warning

    async def test_auto_migrate_is_required_keyword_only(self, tmp_path):
        """关键行为参数不给默认值(P4): 缺省规则只写在 config 一处,不与类签名漂移。"""
        with pytest.raises(TypeError):
            SQLiteRecorder(tmp_path / "t.db")  # type: ignore[call-arg]


# 假池用例的池上限与写入预算: 两者都是必填 keyword-only(缺省只写在 config 一处),
# 本文件统一取这一份,免得每个 helper 各写一个数字
_TEST_POOL_MAX = 2
_TEST_WRITE_TIMEOUT_S = 5.0
_PG_DSN = "postgresql://u:p@h:5432/polygateway"


class _FakePgConn:
    """记录执行过的语句;可让 ALTER/CREATE/探测抛错以模拟权限不足与抖动。

    `existing` 为空列表即表示**表不存在**(与真实 PG 一致: `to_regclass` 为 NULL
    时列探测必然零行),故 `fetchval` 与 `fetch` 共用同一份事实。
    """

    def __init__(
        self,
        existing: list[str],
        *,
        fail_alter: bool = False,
        fail_create: bool = False,
        probe_errors: int = 0,
        hang_insert: bool = False,
        fail_terminate: bool = False,
    ):
        self.existing = existing
        self.fail_alter = fail_alter
        self.fail_create = fail_create
        self.probe_errors = probe_errors
        # 只挂 INSERT: 准备期照常完成,挂住的才是业务路径上那次内联 await
        self.hang_insert = hang_insert
        self.fail_terminate = fail_terminate
        self.terminated = False
        self.statements: list[str] = []

    def terminate(self):
        self.terminated = True
        if self.fail_terminate:
            raise RuntimeError("connection is already closed")

    async def execute(self, sql, *args):
        self.statements.append(sql)
        if sql.startswith("INSERT INTO") and self.hang_insert:
            await asyncio.sleep(3600)
        if sql.startswith("ALTER TABLE") and self.fail_alter:
            raise RuntimeError("must be owner of table llm_calls")
        if sql.lstrip().startswith("CREATE TABLE"):
            if self.fail_create:
                raise RuntimeError("permission denied for schema public")
            self.existing = list(_EXPECTED_COLUMNS)

    async def fetchval(self, sql, *args):
        self.statements.append(sql)
        if self.probe_errors > 0:
            self.probe_errors -= 1
            raise RuntimeError("connection was closed in the middle of operation")
        return "llm_calls" if self.existing else None

    async def fetch(self, sql, *args):
        self.statements.append(sql)
        return [{"attname": name} for name in self.existing]


class _FakePgPool:
    """假池: 记 acquire/release 的配对次数与实参 timeout(issue #15 T3)。

    形状跟着被测代码走: recorder 改用**显式** `acquire(timeout=)` /
    `release(conn, timeout=)`,不再用 `async with pool.acquire()`(那条路
    的 shielded release 会把写入的真实上界撑成 ≈2× 预算,设计 §3.1),
    故这里也不再提供上下文管理器。
    """

    def __init__(
        self,
        conn,
        *,
        hang_acquire: bool = False,
        fail_release: bool = False,
        hang_close: bool = False,
    ):
        self._conn = conn
        self.hang_acquire = hang_acquire
        self.fail_release = fail_release
        # 模拟 asyncpg 的 `Pool.close()` 在 in-flight 连接未归还时**无限等**
        # (pool.py:939-948, 961-972 只在 60s 发一条 warning)
        self.hang_close = hang_close
        self.acquired = 0
        self.released = 0
        self.close_calls = 0
        self.terminated = False
        self.acquire_timeouts: list[object] = []
        self.release_timeouts: list[object] = []

    async def acquire(self, *, timeout=None):
        self.acquire_timeouts.append(timeout)
        if self.hang_acquire:
            await asyncio.sleep(3600)
        self.acquired += 1
        return self._conn

    async def release(self, conn, *, timeout=None):
        assert conn is self._conn
        self.release_timeouts.append(timeout)
        if self.fail_release:
            raise RuntimeError("connection reset failed")
        self.released += 1

    async def close(self):
        self.close_calls += 1
        if self.hang_close:
            await asyncio.sleep(3600)

    def terminate(self):
        self.terminated = True


class TestPostgresBackfillDiscipline:
    """PG 补列必须与 SQLite 侧对称: 失败只逐行降级,且稳态不抢排他锁(issue #3)。"""

    _LEGACY = ["call_id", "cost", "created_at"]
    _CURRENT = [
        "call_id",
        "cost",
        "created_at",
        "cached_prompt_tokens",
        "model_reported",
        "sampling",
        "reasoning_tokens",
        "tenant_id",
        "meta",
    ]

    def _recorder(self, conn):
        from polygateway.telemetry.postgres import PostgresRecorder

        return PostgresRecorder(
            "postgresql://u:p@h:5432/polygateway",
            pool=_FakePgPool(conn),
            auto_migrate=True,
            pool_max=_TEST_POOL_MAX,
            write_timeout_s=_TEST_WRITE_TIMEOUT_S,
        )

    async def test_alter_failure_does_not_disable_the_recorder(self):
        """ALTER 失败(如账号只有 INSERT 权限)不得置 _failed —— 那会让遥测全灭。"""
        conn = _FakePgConn(self._LEGACY, fail_alter=True)
        recorder = self._recorder(conn)
        await _record_minimal(recorder)  # 不得抛
        assert recorder._failed is False
        assert any(s.startswith("INSERT INTO llm_calls") for s in conn.statements)

    async def test_no_alter_when_columns_already_exist(self):
        """ADD COLUMN IF NOT EXISTS 即使列已存在也会先抢 ACCESS EXCLUSIVE 锁,

        而遥测是内联 await——稳态下必须一条 ALTER 都不发,否则每个进程的首次
        写入都会去锁共享审计表。
        """
        conn = _FakePgConn(self._CURRENT)
        await _record_minimal(self._recorder(conn))
        assert not [s for s in conn.statements if s.startswith("ALTER TABLE")]

    async def test_missing_columns_are_added_once(self):
        conn = _FakePgConn(self._LEGACY)
        await _record_minimal(self._recorder(conn))
        from polygateway.telemetry.schema import PG_BACKFILL

        altered = [s for s in conn.statements if s.startswith("ALTER TABLE")]
        assert len(altered) == len(PG_BACKFILL)  # 旧表缺全部补列,故一列一条 ALTER
        assert all("IF NOT EXISTS" not in s for s in altered)  # 探测已确认缺列,无需再判


class TestPostgresTableProbe:
    """建表必须先探测,且"判死"只认"确定写不进去"(issue #9)。

    实测(PostgreSQL 16.14,只有表级 SELECT/INSERT 的角色): `CREATE TABLE IF NOT
    EXISTS` 被拒 permission denied for schema,而同一连接的 `INSERT` 通过——
    PG 对 schema 的 CREATE 权限检查早于 `IF NOT EXISTS` 的存在性判断。无条件发
    DDL 会让这类最小权限部署的整个进程静默失遥测。
    """

    _CURRENT = [
        "call_id",
        "cost",
        "created_at",
        "cached_prompt_tokens",
        "model_reported",
        "sampling",
        "reasoning_tokens",
        "tenant_id",
        "meta",
    ]

    def _recorder(self, conn):
        from polygateway.telemetry.postgres import PostgresRecorder

        return PostgresRecorder(
            "postgresql://u:p@h:5432/polygateway",
            pool=_FakePgPool(conn),
            auto_migrate=True,
            pool_max=_TEST_POOL_MAX,
            write_timeout_s=_TEST_WRITE_TIMEOUT_S,
        )

    def _created(self, conn):
        return [s for s in conn.statements if s.lstrip().startswith("CREATE TABLE")]

    async def test_existing_table_is_never_recreated(self):
        """表已存在就一条 DDL 都不发——这是权限被拒的唯一根治办法。"""
        conn = _FakePgConn(self._CURRENT)
        await _record_minimal(self._recorder(conn))
        assert not self._created(conn)

    async def test_create_denied_on_existing_table_keeps_recording(self):
        """就算 DDL 仍被发出并被拒,表存在时也不得判死整个 recorder。"""
        conn = _FakePgConn(self._CURRENT, fail_create=True)
        recorder = self._recorder(conn)
        await _record_minimal(recorder)  # 不得抛
        assert recorder._failed is False
        assert any(s.startswith("INSERT INTO llm_calls") for s in conn.statements)

    async def test_missing_table_is_created_and_not_backfilled(self):
        """表不存在→建表;新建表列已齐全,不得再发补列 ALTER。"""
        conn = _FakePgConn([])
        recorder = self._recorder(conn)
        await _record_minimal(recorder)
        assert len(self._created(conn)) == 1
        assert not [s for s in conn.statements if s.startswith("ALTER TABLE")]
        assert recorder._failed is False
        assert any(s.startswith("INSERT INTO llm_calls") for s in conn.statements)

    async def test_create_failure_on_missing_table_degrades_to_noop(self):
        """表确定不存在且建不出来 = 确定写不进去: 此时才允许永久 no-op。"""
        conn = _FakePgConn([], fail_create=True)
        recorder = self._recorder(conn)
        await _record_minimal(recorder)  # 不得抛
        assert recorder._failed is True
        assert not [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]

    async def test_probe_failure_is_transient_not_terminal(self):
        """探测失败多为连接抖动: 跳过本次,下次调用必须重试,绝不永久判死。"""
        conn = _FakePgConn(self._CURRENT, probe_errors=1)
        recorder = self._recorder(conn)
        await _record_minimal(recorder, call_id="first")  # 不得抛
        assert recorder._failed is False
        assert not [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]
        await _record_minimal(recorder, call_id="second")
        assert [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")]


class TestPostgresSchemaMode:
    """issue #13: PG 侧 manual 档一条 ALTER 都不发,改按现有列裁剪 INSERT。

    真实 PG 的验收在 `tests/integration/test_postgres_telemetry.py`;这里用 fake 连接
    锁住"发了哪些语句",无 DSN 环境下集成用例被 skip 时仍有回归保护。
    """

    _LEGACY = ["call_id", "cost", "created_at"]

    def _recorder(self, conn, *, auto_migrate):
        from polygateway.telemetry.postgres import PostgresRecorder

        return PostgresRecorder(
            "postgresql://u:p@h:5432/polygateway",
            pool=_FakePgPool(conn),
            auto_migrate=auto_migrate,
            pool_max=_TEST_POOL_MAX,
            write_timeout_s=_TEST_WRITE_TIMEOUT_S,
        )

    async def test_manual_mode_trims_the_insert_instead_of_altering(self, captured_warnings):
        conn = _FakePgConn(self._LEGACY)
        recorder = self._recorder(conn, auto_migrate=False)
        await _record_minimal(recorder)

        assert not [s for s in conn.statements if s.startswith("ALTER TABLE")]
        assert "INSERT INTO llm_calls (call_id, cost) VALUES ($1, $2) ON CONFLICT DO NOTHING" in (
            conn.statements
        )
        assert recorder._columns == ("call_id", "cost")
        message = "\n".join(captured_warnings)
        assert "tenant_id" in message and "meta" in message  # 逐列点名
        assert "不会被记录" in message  # 讲清后果
        assert "ALTER TABLE llm_calls ADD COLUMN tenant_id" in message  # 可直接执行的 SQL

    async def test_manual_mode_still_creates_a_missing_table(self):
        """manual 只管 ALTER 不管 CREATE: 新建表列已齐全,写入照发全量列。"""
        from polygateway.telemetry.schema import COLUMNS

        conn = _FakePgConn([])
        recorder = self._recorder(conn, auto_migrate=False)
        await _record_minimal(recorder)

        assert [s for s in conn.statements if s.lstrip().startswith("CREATE TABLE")]
        assert recorder._columns == COLUMNS

    async def test_no_recognizable_column_falls_back_to_the_full_column_set(
        self, captured_warnings
    ):
        """PG 侧同款回落(SQLite 侧对称用例见 TestSQLiteSchemaMode)。

        表存在(`to_regclass` 非空)但列与 `COLUMNS` 毫无交集: 裁剪结果为空,
        必须回落全量而不是把空列集交给 `insert_sql`——`_prepare_schema` 里那次
        调用在 try 之外,ValueError 会顺着 `record_llm_call` 冒给业务调用方。
        """
        from polygateway.telemetry.schema import COLUMNS

        conn = _FakePgConn(["foo", "bar"])
        recorder = self._recorder(conn, auto_migrate=False)
        await _record_minimal(recorder)  # 不得抛

        assert recorder._columns == COLUMNS
        assert not [s for s in conn.statements if s.startswith("ALTER TABLE")]
        assert [m for m in captured_warnings if "没有任何本库认识的列" in m]

    async def test_auto_mode_still_backfills(self):
        """auto 档现状回归: 缺列照补,补完写全量列。"""
        from polygateway.telemetry.schema import COLUMNS, PG_BACKFILL

        conn = _FakePgConn(self._LEGACY)
        recorder = self._recorder(conn, auto_migrate=True)
        await _record_minimal(recorder)

        assert len([s for s in conn.statements if s.startswith("ALTER TABLE")]) == len(PG_BACKFILL)
        assert recorder._columns == COLUMNS


class _MemoryRecorder:
    def __init__(self):
        self.rows = []

    async def record_llm_call(self, **fields):
        self.rows.append(fields)


class TestEmitterRecorderContract:
    """emitter 的实参键集合必须与后端的 `schema.COLUMNS` 完全一致(issue #3)。

    两个后端的 `row = tuple(fields[col] for col in COLUMNS)` 都在 try **之外**,
    emitter 漏传一个键就抛 KeyError,被 `_record` 的 except Exception 吞成 warning
    → 遥测静默全丢。而 8 个 `**fields` 形态的 fake 一个都拦不住,故显式断言。
    """

    async def test_emitter_supplies_exactly_the_backend_columns(self):
        from polygateway.telemetry.schema import COLUMNS

        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, text_cap=None).emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="cid-1",
            latency_ms=42,
            response=_resp(),
            error=None,
        )
        assert set(rec.rows[0]) == set(COLUMNS)

    @pytest.mark.parametrize("emit", ["attempt", "cache_hit", "terminal_failure"])
    async def test_every_entry_point_supplies_the_same_keys(self, emit):
        from polygateway.telemetry.schema import COLUMNS

        rec = _MemoryRecorder()
        emitter = TelemetryEmitter(rec, text_cap=None)
        if emit == "attempt":
            await emitter.emit_attempt(
                request=_REQ,
                source=_source(),
                call_id="c",
                latency_ms=1,
                response=None,
                error="boom",
            )
        elif emit == "cache_hit":
            await emitter.emit_cache_hit(request=_REQ, response=_resp())
        else:
            await emitter.emit_terminal_failure(
                request=_REQ, call_id="c", latency_ms=1, error="dead"
            )
        assert set(rec.rows[0]) == set(COLUMNS)


class TestEmitterObservabilityFields:
    """issue #3: 三个入口各自的取值口径(设计 §5 表)。"""

    async def test_attempt_carries_the_response_values(self):
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, text_cap=None).emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="cid-1",
            latency_ms=42,
            response=_resp(cached_prompt_tokens=64, model_reported="m-real", reasoning_tokens=7),
            error=None,
        )
        assert rec.rows[0]["cached_prompt_tokens"] == 64
        assert rec.rows[0]["model_reported"] == "m-real"
        assert rec.rows[0]["reasoning_tokens"] == 7

    async def test_failed_attempt_has_no_provider_facts(self):
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, text_cap=None).emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="cid-2",
            latency_ms=7,
            response=None,
            error="boom",
        )
        assert rec.rows[0]["cached_prompt_tokens"] is None
        assert rec.rows[0]["model_reported"] is None
        assert rec.rows[0]["reasoning_tokens"] is None

    async def test_cache_hit_replays_the_recorded_values(self):
        """决策 B1: 命中行原样回放,故命中率统计必须带 WHERE cache_hit = false。"""
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, text_cap=None).emit_cache_hit(
            request=_REQ,
            response=_resp(cached_prompt_tokens=64, model_reported="m-real", reasoning_tokens=7),
        )
        row = rec.rows[0]
        assert row["cache_hit"] is True
        assert row["cached_prompt_tokens"] == 64 and row["model_reported"] == "m-real"
        assert row["reasoning_tokens"] == 7  # 与 cached 同口径原样回放

    async def test_terminal_failure_records_none(self):
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure(
            request=_REQ, call_id="c", latency_ms=1, error="dead"
        )
        assert rec.rows[0]["cached_prompt_tokens"] is None
        assert rec.rows[0]["model_reported"] is None
        assert rec.rows[0]["reasoning_tokens"] is None


class TestEmitterSamplingColumn:
    """issue #4: sampling 列在三个入口的口径(设计决策 D 表格)。

    列语义 = 「调用方采样意图 ⊎ 生效源 extra_body」,**不含**结构化注入的
    response_format(列名是采样参数,schema 不是;且数 KB schema 逐行落库会让
    审计表无谓膨胀)。三入口若各读各的层,同一列在不同行含义就不同。
    """

    _SAMPLED = ChatRequest(
        messages=[{"role": "user", "content": "hi"}],
        sampling={"seed": 42},
        overlay={"seed": 42, "response_format": {"type": "json_object"}},
    )

    async def test_attempt_merges_source_extra_body(self):
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, text_cap=None).emit_attempt(
            request=self._SAMPLED,
            source=_source(extra_body={"temperature": 0}),
            call_id="c",
            latency_ms=1,
            response=_resp(),
            error=None,
        )
        assert json.loads(rec.rows[0]["sampling"]) == {"seed": 42, "temperature": 0}

    async def test_response_format_never_leaks_into_the_column(self):
        """三行都不得出现 response_format——它不是采样参数。"""
        rec = _MemoryRecorder()
        emitter = TelemetryEmitter(rec, text_cap=None)
        await emitter.emit_attempt(
            request=self._SAMPLED,
            source=_source(),
            call_id="c",
            latency_ms=1,
            response=_resp(),
            error=None,
        )
        await emitter.emit_cache_hit(request=self._SAMPLED, response=_resp())
        await emitter.emit_terminal_failure(
            request=self._SAMPLED, call_id="c", latency_ms=1, error="dead"
        )
        assert len(rec.rows) == 3
        for row in rec.rows:
            assert "response_format" not in row["sampling"]

    @pytest.mark.parametrize("emit", ["cache_hit", "terminal_failure"])
    async def test_sourceless_entries_record_call_level_only(self, emit):
        """两个最外层入口没有"生效源"可言,与 model/source_name 置空同一先例。"""
        rec = _MemoryRecorder()
        emitter = TelemetryEmitter(rec, text_cap=None)
        if emit == "cache_hit":
            await emitter.emit_cache_hit(request=self._SAMPLED, response=_resp())
        else:
            await emitter.emit_terminal_failure(
                request=self._SAMPLED, call_id="c", latency_ms=1, error="dead"
            )
        assert json.loads(rec.rows[0]["sampling"]) == {"seed": 42}

    async def test_absent_sampling_is_null(self):
        """无采样参数时为 NULL,而非空字符串或 "{}"——便于 SQL 过滤。"""
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, text_cap=None).emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="c",
            latency_ms=1,
            response=_resp(),
            error=None,
        )
        assert rec.rows[0]["sampling"] is None


class TestEmitterCallerDimensions:
    """issue #11: 三个 emit 入口统一从 `request` 读维度,`_record` 落库前归一化。

    维度只有一个读取点(`request`),否则同一列在三种行里口径分叉——那正是
    "遥测调用点收敛为单一 helper"这条铁律要防的形态。
    """

    _META = {"z_last": "z", "a_first": 1, "m_mid": True}
    _REQ_A = ChatRequest(
        messages=[{"role": "user", "content": "hi"}],
        session_id="sess-1",
        tenant_id="tenant-a",
        meta=_META,
    )

    @pytest.mark.parametrize("emit", ["attempt", "cache_hit", "terminal_failure"])
    async def test_every_entry_point_carries_the_dimensions(self, emit):
        """三条路径写出的行都必须带维度: 漏掉任一条,该租户的账就永远对不上。"""
        rec = _MemoryRecorder()
        emitter = TelemetryEmitter(rec, text_cap=None)
        if emit == "attempt":
            await emitter.emit_attempt(
                request=self._REQ_A,
                source=_source(),
                call_id="c",
                latency_ms=1,
                response=_resp(),
                error=None,
            )
        elif emit == "cache_hit":
            await emitter.emit_cache_hit(request=self._REQ_A, response=_resp(cache_hit=True))
        else:
            await emitter.emit_terminal_failure(
                request=self._REQ_A, call_id="c", latency_ms=1, error="dead"
            )
        row = rec.rows[0]
        assert row["tenant_id"] == "tenant-a"
        assert json.loads(row["meta"]) == self._META

    async def test_cache_hit_records_the_current_caller_not_the_cached_one(self):
        """缓存命中行的维度是"本次由谁发起",不是历史那次——最容易实现反的一处。

        历史那次由租户 B 发起并把响应留在了缓存里;本次由租户 A 发起并命中。
        若读了历史那次的归属,租户 A 的调用会记到 B 头上,而 A 的账面凭空少一行
        ——两个租户的账同时错,且错得没有任何报错。
        """
        historical = ChatRequest(
            messages=[{"role": "user", "content": "hi"}],
            tenant_id="tenant-b",
            meta={"batch": "old-batch"},
        )
        rec = _MemoryRecorder()
        mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None))

        async def terminal(request):
            # 缓存层回放的是历史那次的响应对象(其 call_id 属于 historical 那次)
            return _resp(cache_hit=True, latency_ms=0, call_id="cache-cid")

        assert historical.tenant_id == "tenant-b"  # 历史归属确实不同,否则本用例是空转
        await mw(self._REQ_A, terminal)

        row = rec.rows[0]
        assert row["cache_hit"] is True
        assert row["tenant_id"] == "tenant-a"
        assert "old-batch" not in row["meta"]

    async def test_absent_dimensions_land_as_sentinels(self):
        """未传维度落哨兵值: `tenant_id` 空串、`meta` 字面量 `'{}'`,都不是 NULL。

        NULL 的 `tenant_id` 在 PG 的 RLS policy 下对所有人永久不可见(设计 §4.4),
        空串则可用一条 SQL 审计出还有多少行未归属;`meta` 同理,`'{}'` 可被
        JSON 函数直接查询,NULL 则要每条查询都额外判空。
        """
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, text_cap=None).emit_attempt(
            request=_REQ,  # tenant_id=None, meta={}
            source=_source(),
            call_id="c",
            latency_ms=1,
            response=_resp(),
            error=None,
        )
        row = rec.rows[0]
        assert row["tenant_id"] == ""
        assert row["meta"] == "{}"

    async def test_meta_is_serialized_with_sorted_keys(self):
        """键序固定,同一份维度在任意两行里字节一致,可直接做等值比对与去重。"""
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure(
            request=self._REQ_A, call_id="c", latency_ms=1, error="dead"
        )
        assert list(json.loads(rec.rows[0]["meta"])) == ["a_first", "m_mid", "z_last"]

    async def test_non_ascii_meta_stays_readable(self):
        """`ensure_ascii=False`: 中文维度按原文落库,而非 `\\uXXXX` 转义串。"""
        rec = _MemoryRecorder()
        req = ChatRequest(messages=[{"role": "user", "content": "hi"}], meta={"dept": "研发"})
        await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure(
            request=req, call_id="c", latency_ms=1, error="dead"
        )
        assert "研发" in rec.rows[0]["meta"]

    async def test_non_finite_meta_value_drops_the_row_instead_of_poisoning_it(self):
        """入口失守时 `allow_nan=False` 的真实结果: 整行降级丢弃,且不抛给调用方。

        直接构造带 `nan` 的 `ChatRequest`(绕过 `validate_caller_dimensions` 这道
        主防线,模拟将来某个新入口忘记校验)。没有 `allow_nan=False` 时,
        `json.dumps` 会写出裸 `NaN` 字面量——PG 的 JSONB 会拒收,但 **SQLite 的
        `meta` 是 TEXT 列不做校验**,那串非法 JSON 会被静默存进去,污染此后一切
        按 JSON 解析 meta 的分析。宁可丢一行遥测,也不要一行毒数据。

        同时断言不抛: 遥测的降级方向是"静默降级"(铁律),把调用方的一次正常
        业务调用因为一个维度值炸掉,方向反了。
        """
        rec = _MemoryRecorder()
        req = ChatRequest(messages=[{"role": "user", "content": "hi"}], meta={"k": float("nan")})
        await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure(
            request=req, call_id="c", latency_ms=1, error="dead"
        )
        assert rec.rows == []


class TestCostWithCachedTier:
    """issue #3: 命中部分按缓存单价计费,避免 cost 系统性高估。"""

    _TABLE = PricingTable(
        {"m": ModelPrice(input_per_1m=10.0, output_per_1m=20.0, cached_input_per_1m=2.0)}
    )

    async def test_cached_hit_lowers_the_recorded_cost(self):
        rec = _MemoryRecorder()
        emitter = TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None)
        full = _resp(prompt_tokens=1_000_000, completion_tokens=0)
        await emitter.emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="c1",
            latency_ms=1,
            response=full,
            error=None,
        )
        await emitter.emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="c2",
            latency_ms=1,
            response=_resp(
                prompt_tokens=1_000_000, completion_tokens=0, cached_prompt_tokens=600_000
            ),
            error=None,
        )
        assert rec.rows[0]["cost"] == pytest.approx(10.0)
        assert rec.rows[1]["cost"] == pytest.approx(5.2)  # 400k×10 + 600k×2

    async def test_cache_hit_row_still_costs_zero(self):
        """缓存命中未产生新调用 → cost 恒 0.0,该短路必须排在任何换算之前。"""
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None).emit_cache_hit(
            request=_REQ,
            response=_resp(prompt_tokens=1_000_000, cached_prompt_tokens=600_000),
        )
        assert rec.rows[0]["cost"] == 0.0

    async def test_unavailable_usage_still_costs_none(self):
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None).emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="c",
            latency_ms=1,
            response=_resp(usage_source="unavailable", cached_prompt_tokens=5),
            error=None,
        )
        assert rec.rows[0]["cost"] is None


class TestEmitter:
    async def test_attempt_success_row(self):
        rec = _MemoryRecorder()
        emitter = TelemetryEmitter(rec, text_cap=None)
        await emitter.emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="cid-1",
            latency_ms=42,
            response=_resp(),
            error=None,
        )
        row = rec.rows[0]
        assert row["call_id"] == "cid-1" and row["error"] is None
        assert row["session_id"] == "sess-1" and row["source_name"] == "s1"
        assert row["response"] == "ok" and row["cost"] is None

    async def test_attempt_failure_row(self):
        rec = _MemoryRecorder()
        emitter = TelemetryEmitter(rec, text_cap=None)
        await emitter.emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="cid-2",
            latency_ms=7,
            response=None,
            error="TransientError: boom",
        )
        row = rec.rows[0]
        assert row["error"].startswith("TransientError")
        # 失败尝试没有任何用量信息可言 → unavailable(设计 §3.2 #6)
        assert row["response"] == "" and row["usage_source"] == "unavailable"
        assert row["cost"] is None

    async def test_terminal_failure_row_is_unavailable(self):
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_terminal_failure(
            request=_REQ, call_id="cid-t", latency_ms=5, error="cancelled"
        )
        row = rec.rows[0]
        assert row["usage_source"] == "unavailable" and row["cost"] is None

    @pytest.mark.parametrize(("prompt", "completion"), [(0, 0), (0, 4000)])
    async def test_unavailable_success_row_has_null_cost(self, prompt, completion):
        """产生了真实调用但用量不可得 → cost 记 NULL(设计 §3.1 不变式)。

        参数第二组是改前兜底写出的 `0/4000` 形态: 那时换算出 0.032 的假金额。
        """
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="cid-u",
            latency_ms=42,
            response=_resp(
                usage_source="unavailable", prompt_tokens=prompt, completion_tokens=completion
            ),
            error=None,
        )
        assert rec.rows[0]["cost"] is None

    async def test_measured_row_still_priced(self):
        """对照组: 同一价格表下 measured 行照常换算,证明 None 不是价格表没接上。"""
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="cid-m",
            latency_ms=42,
            response=_resp(prompt_tokens=0, completion_tokens=4000),
            error=None,
        )
        assert rec.rows[0]["cost"] == pytest.approx(0.032)

    async def test_cache_hit_keeps_zero_cost_even_when_unavailable(self):
        """缓存命中未产生新调用,0.0 是事实而非未知 → 短路必须排在 cache_hit 之后。"""
        rec = _MemoryRecorder()
        await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_cache_hit(
            request=_REQ,
            response=_resp(cache_hit=True, usage_source="unavailable", completion_tokens=4000),
        )
        assert rec.rows[0]["cache_hit"] is True and rec.rows[0]["cost"] == 0.0

    async def test_multimodal_messages_digested_before_storage(self):
        rec = _MemoryRecorder()
        emitter = TelemetryEmitter(rec, text_cap=None)
        big = "data:image/png;base64," + "A" * 100_000
        req = ChatRequest(
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "image_url", "image_url": {"url": big}},
                    ],
                }
            ]
        )
        await emitter.emit_attempt(
            request=req,
            source=_source(),
            call_id="c",
            latency_ms=1,
            response=None,
            error="x",
        )
        assert len(rec.rows[0]["messages"]) < 500  # base64 不整段进库(VT R12)

    async def test_recorder_failure_swallowed(self):
        class Broken:
            async def record_llm_call(self, **fields):
                raise OSError("disk full")

        emitter = TelemetryEmitter(Broken(), text_cap=None)
        await emitter.emit_attempt(
            request=_REQ,
            source=_source(),
            call_id="c",
            latency_ms=1,
            response=_resp(),
            error=None,
        )  # 不抛(降级不冒泡)


class TestTelemetryMW:
    async def test_cache_hit_recorded(self):
        rec = _MemoryRecorder()
        mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None))

        async def terminal(request):
            return _resp(cache_hit=True, latency_ms=0, call_id="cache-cid")

        resp = await mw(_REQ, terminal)
        assert resp.cache_hit
        assert len(rec.rows) == 1
        assert rec.rows[0]["cache_hit"] is True and rec.rows[0]["latency_ms"] == 0

    async def test_normal_success_not_double_recorded(self):
        """成功尝试由 RetryMW 逐次记录;最外层不得重复记。"""
        rec = _MemoryRecorder()
        mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None))

        async def terminal(request):
            return _resp(cache_hit=False)

        await mw(_REQ, terminal)
        assert rec.rows == []

    async def test_scope_level_failure_recorded(self):
        rec = _MemoryRecorder()
        mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None))

        async def terminal(request):
            raise CircuitOpenError(scope="llm", retry_after_s=30.0)

        with pytest.raises(CircuitOpenError):
            await mw(_REQ, terminal)
        assert len(rec.rows) == 1 and "circuit_open" in rec.rows[0]["error"]

    async def test_attempt_level_failure_not_double_recorded(self):
        """RequestRejected 已被 RetryMW 逐次记录 → 最外层跳过。"""
        rec = _MemoryRecorder()
        mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None))

        async def terminal(request):
            raise RequestRejectedError("400")

        with pytest.raises(RequestRejectedError):
            await mw(_REQ, terminal)
        assert rec.rows == []


def test_single_emitter_discipline():
    """铁律执法: record_llm_call 在 src/ 的调用点只允许出现在 telemetry emitter。"""
    out = subprocess.run(
        ["grep", "-rln", "record_llm_call(", "src/polygateway"],
        capture_output=True,
        text=True,
        cwd=Path(__file__).resolve().parents[2],
    ).stdout.splitlines()
    callers = [
        p
        for p in out
        if not p.endswith(("ports.py", "telemetry/sqlite.py", "telemetry/postgres.py"))
    ]
    assert callers == ["src/polygateway/middleware/telemetry.py"]


# —— issue #12 (a): 遥测正文可配置上限 ——

_LONG = "甲乙丙丁戊己庚辛壬癸" * 5  # 50 字,cap=8 时省略 42 字
_CAPPED = "甲乙丙丁戊己庚辛…(略 42 字)"


def _long_messages():
    """一条纯文本 + 一条多模态(text part + image_url part)。"""
    return [
        {"role": "system", "content": _LONG},
        {
            "role": "user",
            "content": [
                {"type": "text", "text": _LONG},
                {"type": "image_url", "image_url": {"url": "https://gw.example/a.png"}},
            ],
        },
    ]


async def _emit_with_cap(messages, *, cap, response=_LONG, thinking=_LONG):
    rec = _MemoryRecorder()
    await TelemetryEmitter(rec, text_cap=cap).emit_attempt(
        request=ChatRequest(messages=messages, session_id="s"),
        source=_source(),
        call_id="c",
        latency_ms=1,
        response=_resp(content=response, thinking=thinking),
        error=None,
    )
    return rec.rows[0]


class TestTelemetryTextCap:
    """截断发生在唯一遥测出口 `_record`(设计 §5.2);缺省 None = 不截断。"""

    async def test_cap_none_keeps_the_body_byte_for_byte(self):
        """缺省不截断是人类决策(设计 §2 E-a): 落库正文与改前逐字节相同。"""
        messages = _long_messages()
        row = await _emit_with_cap(messages, cap=None)
        assert row["messages"] == json.dumps(digest_messages(messages), ensure_ascii=False)
        assert row["response"] == _LONG
        assert row["thinking"] == _LONG

    async def test_cap_truncates_each_content_and_keeps_the_json_parsable(self):
        """按每条文本切而非切整串 JSON: 否则该 TEXT 列此后无法按 JSON 解析。"""
        row = await _emit_with_cap(_long_messages(), cap=8)
        parsed = json.loads(row["messages"])  # 不抛 = 整串仍是合法 JSON
        assert parsed[0]["content"] == _CAPPED
        assert parsed[1]["content"][0]["text"] == _CAPPED
        assert "(略 42 字)" in parsed[0]["content"]  # 标记须含省略字数

    async def test_image_digest_is_untouched_by_the_cap(self):
        """多模态 image_url 的 sha256 摘要不是正文,不得被截断改形。"""
        messages = _long_messages()
        expected = digest_messages(messages)[1]["content"][1]
        assert expected["type"] == "image_url" and len(expected["sha256"]) == 64
        row = await _emit_with_cap(messages, cap=8)
        assert json.loads(row["messages"])[1]["content"][1] == expected

    async def test_non_string_content_passes_through_without_raising(self):
        """外部输入形状不可控,遥测路径不得因此抛错(P5 + 降级方向)。

        同时钉住设计 §5.2 的覆盖面诚实声明: 只覆盖文本 content 与 text part,
        嵌套 dict 里的长文本**不在**覆盖范围内。
        """
        messages = [
            {"role": "user", "content": 123},
            {"role": "user", "content": None},
            {"role": "user", "content": {"nested": _LONG}},
            {"role": "user", "content": [{"type": "text", "text": 7}, "bare-part"]},
        ]
        row = await _emit_with_cap(messages, cap=8)
        assert json.loads(row["messages"]) == messages

    async def test_response_and_thinking_are_capped(self):
        row = await _emit_with_cap([{"role": "user", "content": "hi"}], cap=8)
        assert row["response"] == _CAPPED
        assert row["thinking"] == _CAPPED

    async def test_cap_never_mutates_the_caller_messages(self):
        """红线之二: 落库那份被截断,调用方持有的那份(含嵌套 part)一字未改。

        `digest_messages` 对 content 非 list 的消息原样透传**同一个 dict 对象**
        (`cache.py:43`),就地截断会连调用方的 messages、后续重试的请求体与缓存
        写入的 key 一起改掉,且全程无任何报错。
        """
        messages = _long_messages()
        snapshot = copy.deepcopy(messages)
        row = await _emit_with_cap(messages, cap=8)
        assert messages == snapshot
        assert messages[0]["content"] == _LONG
        assert messages[1]["content"][0]["text"] == _LONG
        assert json.loads(row["messages"])[0]["content"] == _CAPPED  # 落库那份确已截断

    def test_non_positive_cap_rejected_at_construction(self):
        """emitter 是三个 Client 唯一的汇合点,值域校验放这一处即覆盖全部装配路。

        settings 层那道只管 env;直接构造 `GatewayClient(..., text_cap=0)` 是库
        承诺的另一条公共装配路,没有这道闸就会把每条正文写成一个光秃秃的省略标记。
        """
        for bad in (0, -1):
            with pytest.raises(ValueError, match="text_cap"):
                TelemetryEmitter(_MemoryRecorder(), text_cap=bad)


class _StubEmbedTransport:
    async def embed(self, *, texts, source, call_id):
        return EmbeddingTransportResult(
            vectors=[[1.0] for _ in texts],
            dim=1,
            prompt_tokens=1,
            usage_source="measured",
            raw={},
        )


class _StubOcrTransport:
    async def recognize_text(self, *, image, source, call_id):
        return OcrTextTransportResult(text="识别结果" * 10, raw={"task_type": "text"})

    async def parse_layout(self, *, image, source, call_id):
        raise NotImplementedError


def _governance(scope, sources):
    """embed/OCR 两条链路共用的最小治理装配(真实内存后端,不 mock)。"""
    return {
        "scope": scope,
        "sources": sources,
        "selector": RoundRobinSelector(),
        "limiter": InMemoryLimiter(
            scope=scope,
            sources={s.name: s for s in sources},
            global_limits=GlobalLimits(max_concurrency=0, rpm=0, tpm=0),
            lease_ttl_s=100.0,
        ),
        "breaker": InMemoryGate(
            config=BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
        ),
        "retry": RetryPolicy(max_attempts=3, backoff_base_s=0.001, backoff_max_s=0.01),
        "backpressure": BackpressurePolicy(stall_window_s=300.0, poll_interval_s=0.001),
    }


class TestTextCapCoversEmbedAndOcrChains:
    """`_record` 是三条链路共同的出口,cap 自然覆盖全部三条(设计 §5.2)。

    同一张表不该一半受控一半不受控;embed/OCR 各自的 200 字上限保留不动,
    与新 cap 是"取更严者"的关系。
    """

    async def test_embed_rows_are_capped(self):
        rec = _MemoryRecorder()
        client = EmbeddingClient(
            **_governance("embed", [_source(name="e1", model="embed-1")]),
            transport=_StubEmbedTransport(),
            batch_size=2,
            telemetry=rec,
            text_cap=8,
        )
        await client.embed([_LONG])
        row = rec.rows[0]
        assert json.loads(row["messages"])[0]["content"] == _CAPPED
        assert row["response"] == "<vectors…(略 11 字)"  # `<vectors n=1 dim=1>` 共 19 字

    async def test_ocr_rows_are_capped(self):
        rec = _MemoryRecorder()
        client = OcrClient(
            **_governance("ocr", [_source(name="m1", model="monkey-ocr")]),
            transport=_StubOcrTransport(),
            telemetry=rec,
            text_cap=8,
        )
        await client.recognize_text(b"jpg")
        row = rec.rows[0]
        # 占位串 `<ocr:text image_bytes=3>` 共 24 字
        assert json.loads(row["messages"])[0]["content"] == "<ocr:tex…(略 16 字)"
        assert row["response"] == "识别结果识别结果…(略 32 字)"  # 先经 OCR 自有的 200 字上限


# —— 遥测降级状态(issue #15 C 组): 共用 tracker + 只读快照 + 节流日志 ——


class _FakeClock:
    """可手动推进的单调时钟: 冷却与节流都靠它测,用例里绝不真睡。"""

    def __init__(self, start: float = 1_000.0) -> None:
        self.t = start

    def __call__(self) -> float:
        return self.t

    def advance(self, seconds: float) -> None:
        self.t += seconds


@pytest.fixture
def captured_infos():
    """捕获 INFO 及以上;恢复那条 info 是"降级已结束"的唯一外部信号。"""
    from loguru import logger

    messages: list[str] = []
    sink_id = logger.add(messages.append, level="INFO")
    yield messages
    logger.remove(sink_id)


class TestTelemetryStatusTracker:
    """状态机六字段逐个钉;它是两个 recorder 共用的降级事实源(设计 §3.3)。"""

    def _tracker(self, clock):
        from polygateway.telemetry.status import TelemetryStatusTracker

        return TelemetryStatusTracker(backend="postgres", now=clock)

    def test_fresh_tracker_reports_no_degradation(self):
        tracker = self._tracker(_FakeClock())
        status = tracker.snapshot()
        assert (status.degraded, status.fatal, status.reason) == (False, False, None)
        assert status.degraded_for_s is None and status.retry_after_s is None
        assert status.dropped_rows == 0
        assert tracker.should_retry() is True  # 未降级本就该正常走准备路径

    def test_entering_degraded_reports_reason_and_cooldown(self, captured_warnings):
        tracker = self._tracker(_FakeClock())
        tracker.enter_degraded("连接被拒", fatal=False, cooldown_s=60.0)
        status = tracker.snapshot()
        assert status.degraded is True and status.fatal is False
        assert status.reason == "连接被拒"
        assert status.degraded_for_s == pytest.approx(0.0)
        assert status.retry_after_s == pytest.approx(60.0)
        assert len(captured_warnings) == 1
        assert "连接被拒" in captured_warnings[0] and "60" in captured_warnings[0]

    def test_fake_clock_drains_the_cooldown(self):
        clock = _FakeClock()
        tracker = self._tracker(clock)
        tracker.enter_degraded("连接被拒", fatal=False, cooldown_s=60.0)
        clock.advance(25.0)
        status = tracker.snapshot()
        assert status.degraded_for_s == pytest.approx(25.0)
        assert status.retry_after_s == pytest.approx(35.0)
        assert tracker.should_retry() is False
        clock.advance(40.0)  # 越过冷却窗口
        assert tracker.snapshot().retry_after_s == pytest.approx(0.0)  # 不得为负
        assert tracker.should_retry() is True

    def test_recover_clears_degradation_but_keeps_dropped_rows(self, captured_infos):
        tracker = self._tracker(_FakeClock())
        tracker.enter_degraded("连接被拒", fatal=False, cooldown_s=60.0)
        for _ in range(3):
            tracker.record_drop("遥测已降级")
        tracker.recover()
        status = tracker.snapshot()
        assert (status.degraded, status.fatal, status.reason) == (False, False, None)
        assert status.degraded_for_s is None and status.retry_after_s is None
        # 进程生命周期内单调不减: 恢复不是"没丢过",下游要靠它对账
        assert status.dropped_rows == 3
        assert any("3" in message and "恢复" in message for message in captured_infos)

    def test_drop_warnings_are_throttled_by_the_row_constant(self, captured_warnings):
        from polygateway.telemetry.status import _DROP_REPEAT_EVERY_ROWS

        tracker = self._tracker(_FakeClock())  # 时钟不动: 只有行数阈值能触发复述
        tracker.enter_degraded("连接被拒", fatal=False, cooldown_s=60.0)
        captured_warnings.clear()  # 只数丢弃复述,不数进入降级那条
        total = _DROP_REPEAT_EVERY_ROWS * 2 + 1
        for _ in range(total):
            tracker.record_drop("遥测已降级")
        # 首条必报,其后每满一个阈值报一次 —— 关系由常量决定,不写死数字
        assert len(captured_warnings) == 1 + (total - 1) // _DROP_REPEAT_EVERY_ROWS
        assert tracker.snapshot().dropped_rows == total

    def test_drop_warnings_are_also_throttled_by_time(self, captured_warnings):
        from polygateway.telemetry.status import _DROP_REPEAT_EVERY_S

        clock = _FakeClock()
        tracker = self._tracker(clock)
        tracker.enter_degraded("连接被拒", fatal=False, cooldown_s=60.0)
        captured_warnings.clear()
        tracker.record_drop("遥测已降级")
        assert len(captured_warnings) == 1
        tracker.record_drop("遥测已降级")
        assert len(captured_warnings) == 1  # 同一窗口内不刷屏
        clock.advance(_DROP_REPEAT_EVERY_S)
        tracker.record_drop("遥测已降级")
        assert len(captured_warnings) == 2  # 长跑进程里也不会静默

    def test_fatal_degradation_never_retries(self, captured_warnings):
        clock = _FakeClock()
        tracker = self._tracker(clock)
        tracker.enter_degraded("DSN 不可解析", fatal=True, cooldown_s=None)
        clock.advance(1_000_000.0)
        status = tracker.snapshot()
        assert status.fatal is True and status.retry_after_s is None
        assert tracker.should_retry() is False
        assert "重启" in captured_warnings[0]  # 恢复条件必须写在日志里

    def test_repeating_the_same_reason_does_not_spam(self, captured_warnings):
        """冷却到期重试再失败会反复进入降级: 同一原因只讲一次,只刷新窗口。"""
        clock = _FakeClock()
        tracker = self._tracker(clock)
        tracker.enter_degraded("连接被拒", fatal=False, cooldown_s=60.0)
        clock.advance(60.0)
        tracker.enter_degraded("连接被拒", fatal=False, cooldown_s=60.0)
        assert len(captured_warnings) == 1
        assert tracker.snapshot().retry_after_s == pytest.approx(60.0)  # 窗口已刷新
        assert tracker.snapshot().degraded_for_s == pytest.approx(60.0)  # 但仍是同一段降级


class TestSQLiteStatusVisibility:
    """SQLite 侧今天初始化失败后写入静默 return,连一条日志都没有(设计 §1.4)。"""

    def _broken(self, tmp_path):
        blocker = tmp_path / "blocker"
        blocker.write_text("父目录是个文件,mkdir 必然失败")
        return SQLiteRecorder(blocker / "telemetry.db", auto_migrate=True)

    def test_init_failure_is_degraded_and_fatal(self, tmp_path, captured_warnings):
        recorder = self._broken(tmp_path)
        status = recorder.telemetry_status
        assert status.degraded is True and status.fatal is True
        assert captured_warnings  # 静默降级 ≠ 静默

    async def test_dropped_rows_are_counted_and_visible(self, tmp_path, captured_warnings):
        recorder = self._broken(tmp_path)
        captured_warnings.clear()
        await _record_minimal(recorder)  # 不得抛: 遥测绝不冒泡
        assert recorder.telemetry_status.dropped_rows == 1
        assert captured_warnings  # 丢的第一行必须出声

    def test_healthy_recorder_is_not_degraded(self, tmp_path):
        recorder = SQLiteRecorder(tmp_path / "ok.db", auto_migrate=True)
        assert recorder.telemetry_status.degraded is False
        recorder.close()


class TestPostgresStatusVisibility:
    """PG 侧的判死本任务不改判据,只让它经 tracker 变得可见(计划 T2)。"""

    def _recorder(self, conn):
        from polygateway.telemetry.postgres import PostgresRecorder

        return PostgresRecorder(
            "postgresql://u:p@h:5432/polygateway",
            pool=_FakePgPool(conn),
            auto_migrate=True,
            pool_max=_TEST_POOL_MAX,
            write_timeout_s=_TEST_WRITE_TIMEOUT_S,
        )

    async def test_unusable_table_shows_up_in_the_status(self, captured_warnings):
        """表确定不存在且建不出来 = 既有的判死档;现在它要能被下游查到。"""
        recorder = self._recorder(_FakePgConn([], fail_create=True))
        await _record_minimal(recorder)
        status = recorder.telemetry_status
        assert status.degraded is True and status.fatal is True
        assert status.dropped_rows == 1  # 判死那一次调用本身也丢了一行
        assert recorder._failed is True  # 过渡期两份状态并存(T5 收掉 `_failed`)

    async def test_healthy_recorder_is_not_degraded(self):
        recorder = self._recorder(_FakePgConn(list(_EXPECTED_COLUMNS)))
        await _record_minimal(recorder)
        status = recorder.telemetry_status
        assert status.degraded is False and status.dropped_rows == 0


class TestPostgresPoolResourceSemantics:
    """issue #15 A 组: 库必须自己声明池的资源占用,并给写入一个硬预算。

    建池这条路在本 issue 之前**零测试覆盖**(全部 PG 用例都经 `pool=` 注入,
    走的是外部池分支),`min_size=10` 因此潜伏至今: 4 个 client × 10 = 40 条
    常驻连接专用于写遥测,共享实例余量不足时先倒下的必然是它。
    """

    def _recorder(self, pool, **overrides):
        from polygateway.telemetry.postgres import PostgresRecorder

        kwargs: dict[str, object] = {
            "auto_migrate": True,
            "pool_max": _TEST_POOL_MAX,
            "write_timeout_s": _TEST_WRITE_TIMEOUT_S,
        }
        kwargs.update(overrides)
        return PostgresRecorder(_PG_DSN, pool=pool, **kwargs)

    async def test_pool_is_created_without_preconnecting(self, monkeypatch):
        """**主回归钉子**: `min_size=0` 且 `max_size` 取配置值。

        `min_size` 的语义是"预连接"而非"下限"(asyncpg `pool.py:457` 为 0 时
        一条连接都不连),故它是"建池要么全有要么全无"这个脆点的唯一来源。
        继承第三方默认值等于库对自己的资源占用不表态(P4),本条防的就是回归。
        """
        import asyncpg

        from polygateway.telemetry.postgres import PostgresRecorder

        captured: dict[str, object] = {}
        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)))

        async def fake_create_pool(dsn, **kwargs):
            captured["dsn"] = dsn
            captured.update(kwargs)
            return pool

        monkeypatch.setattr(asyncpg, "create_pool", fake_create_pool)
        recorder = PostgresRecorder(_PG_DSN, auto_migrate=True, pool_max=3, write_timeout_s=2.5)
        await _record_minimal(recorder)

        assert captured["min_size"] == 0
        assert captured["max_size"] == 3
        # connect 与单条语句都在同一份写入预算内,不留继承来的 10s 默认值
        assert captured["timeout"] == 2.5
        assert captured["command_timeout"] == 2.5

    async def test_acquire_gets_an_explicit_timeout(self):
        """`pool.acquire()` 无参 = 无限等(asyncpg 缺省 `timeout=None`)。

        池满时那是挂在业务路径上的无限期 await,`max_size` 收到个位数后必现。
        """
        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)))
        await _record_minimal(self._recorder(pool))
        assert pool.acquire_timeouts  # 准备期与写入期各一次
        assert all(t == _TEST_WRITE_TIMEOUT_S for t in pool.acquire_timeouts)

    async def test_write_budget_drops_the_row_instead_of_blocking_the_caller(
        self, captured_warnings
    ):
        """整次写入有硬预算: 后端挂住时丢一行,绝不把业务调用拖在那里。"""
        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS), hang_insert=True))
        recorder = self._recorder(pool, write_timeout_s=0.05)
        loop = asyncio.get_running_loop()
        started = loop.time()
        # 挂死就当场红,而不是把整个套件拖到 CI 超时
        await asyncio.wait_for(_record_minimal(recorder), timeout=5)
        assert loop.time() - started < 1.0
        assert any("预算" in m for m in captured_warnings)
        assert recorder.telemetry_status.dropped_rows == 1

    async def test_release_is_paired_even_when_the_budget_fires(self):
        """预算取消发生在 execute 上,连接照样要还回去——否则池被慢查询吃干。"""
        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS), hang_insert=True))
        recorder = self._recorder(pool, write_timeout_s=0.05)
        await asyncio.wait_for(_record_minimal(recorder), timeout=5)
        assert pool.acquired == 2  # 准备期一次 + 写入一次
        assert pool.released == pool.acquired
        # 归还有独立的小上限: 复用写入预算就等于允许再等一个预算(设计 §3.1)
        assert all(t is not None and t < _TEST_WRITE_TIMEOUT_S for t in pool.release_timeouts)

    async def test_acquire_timeout_drops_the_row_without_leaking(self, captured_warnings):
        """取连接本身挂住时同样丢行;没拿到的连接不许伪造一次 release。"""
        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)), hang_acquire=True)
        recorder = self._recorder(pool, write_timeout_s=0.05)
        await asyncio.wait_for(_record_minimal(recorder), timeout=5)
        assert pool.acquired == 0 and pool.released == 0
        assert captured_warnings

    async def test_external_cancellation_is_not_swallowed_as_a_timeout(self):
        """铁律"取消可穿透": `asyncio.timeout` 只把**自己**触发的 cancel 转成
        TimeoutError,外部取消必须照常以 CancelledError 冒出去。
        """
        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS), hang_insert=True))
        recorder = self._recorder(pool, write_timeout_s=30.0)
        task = asyncio.create_task(_record_minimal(recorder))
        await asyncio.sleep(0.05)  # 让它跑到挂住的那次 INSERT
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        assert pool.released == pool.acquired  # 取消路径上也不许泄漏连接


def _pg_recorder(*, pool=None, **overrides):
    """本文件统一的 PG recorder 构造口: 池上限与写入预算取同一份测试常量。"""
    from polygateway.telemetry.postgres import PostgresRecorder

    kwargs: dict[str, object] = {
        "auto_migrate": True,
        "pool_max": _TEST_POOL_MAX,
        "write_timeout_s": _TEST_WRITE_TIMEOUT_S,
    }
    kwargs.update(overrides)
    return PostgresRecorder(_PG_DSN, pool=pool, **kwargs)


class TestPostgresCloseIsBounded:
    """issue #15 B 组: 关闭动作本身必须有界,且"关了就是关了"(设计 §3.2 第 4 点)。

    两个缺口各钉一次: ① `Pool.close()` 会 await 每个 holder 的
    `wait_until_released()`,in-flight 未归还时无限等 —— 收尾路径上照样是
    "遥测拖垮业务";② 关完还能自己重建池的灰色状态 —— 关闭是所有权终结,
    恢复归运行时的冷却机制管,不该是关闭动作的副作用。
    """

    def _self_built(self, monkeypatch, pool):
        """让 recorder 走**自建池**那条路,并交回建池次数(复活的唯一证据)。"""
        import asyncpg

        created: list[str] = []

        async def fake_create_pool(dsn, **kwargs):
            created.append(dsn)
            return pool

        monkeypatch.setattr(asyncpg, "create_pool", fake_create_pool)
        return _pg_recorder(), created

    async def test_stuck_pool_close_falls_back_to_terminate(self, monkeypatch, captured_warnings):
        """**主回归钉子**: 池关不掉时超时即 terminate,绝不无限期挂在收尾路径上。"""
        from polygateway.telemetry import postgres

        monkeypatch.setattr(postgres, "_CLOSE_TIMEOUT_S", 0.05)
        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)), hang_close=True)
        recorder, _ = self._self_built(monkeypatch, pool)
        await _record_minimal(recorder)

        loop = asyncio.get_running_loop()
        started = loop.time()
        # 用例自带超时: 实现无界时这里要当场红,而不是挂死整个套件
        await asyncio.wait_for(recorder.aclose(), timeout=5)

        assert loop.time() - started < 1.0
        assert pool.close_calls == 1 and pool.terminated is True
        assert captured_warnings  # 强制拆池是异常路径,不许静默

    async def test_writes_after_close_do_not_rebuild_the_pool(self, monkeypatch):
        """关了就是关了: 后续写入短路丢行,**不**再建一个没人负责关的池。"""
        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)))
        recorder, created = self._self_built(monkeypatch, pool)
        await _record_minimal(recorder)
        assert len(created) == 1

        await recorder.aclose()
        dropped_before = recorder.telemetry_status.dropped_rows
        await _record_minimal(recorder, call_id="c2")  # 遥测绝不冒泡

        assert len(created) == 1  # 复活的唯一证据: 第二次 create_pool
        assert pool.acquired == 2  # 准备期 + 首次写入;关闭后一次都没有
        assert recorder.telemetry_status.dropped_rows == dropped_before + 1

    async def test_aclose_is_idempotent(self, monkeypatch):
        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)))
        recorder, _ = self._self_built(monkeypatch, pool)
        await _record_minimal(recorder)
        await recorder.aclose()
        await recorder.aclose()
        assert pool.close_calls == 1

    async def test_injected_pool_is_left_to_its_owner(self, monkeypatch):
        """注入的池既不关也不拆(既有纪律),但 recorder 自己照样"关了就是关了"。

        注入档的复活更隐蔽: 关闭把 `_pool` 置 None 后,下一次写入会拿 DSN
        **自建**一个池——注入方以为自己管着全部连接,实际早已不是。
        """
        import asyncpg

        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)))
        created: list[str] = []

        async def fake_create_pool(dsn, **kwargs):
            # 不能直接 raise: recorder 会把它当建池失败吞掉,用例就白测了
            created.append(dsn)
            return _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)))

        monkeypatch.setattr(asyncpg, "create_pool", fake_create_pool)
        recorder = _pg_recorder(pool=pool)
        await _record_minimal(recorder)
        await recorder.aclose()

        assert pool.close_calls == 0 and pool.terminated is False
        await _record_minimal(recorder, call_id="c2")
        assert created == []  # 注入档的复活: 拿 DSN 另起一个池
        assert pool.acquired == 2  # 关闭后也不再往注入的池上写

    async def test_external_cancellation_during_close_is_not_swallowed(self, monkeypatch):
        """铁律"取消可穿透": 有界关闭不得把外部取消吃成一次超时降级。"""
        pool = _FakePgPool(_FakePgConn(list(_EXPECTED_COLUMNS)), hang_close=True)
        recorder, _ = self._self_built(monkeypatch, pool)
        await _record_minimal(recorder)

        task = asyncio.create_task(recorder.aclose())
        await asyncio.sleep(0.05)  # 让它跑进那次挂住的 close()
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task


class TestPostgresReleaseDegradation:
    """归还连接失败时的防御路径(T3 未覆盖的缺口,借 T4 的假池顺带钉住)。

    留一条状态不明的连接在池里比断开更坏: 它会被下次 acquire 取到,
    把一次失败放大成持续失败。
    """

    async def test_failed_release_terminates_the_connection(self, captured_warnings):
        conn = _FakePgConn(list(_EXPECTED_COLUMNS))
        await _record_minimal(_pg_recorder(pool=_FakePgPool(conn, fail_release=True)))
        assert conn.terminated is True
        assert any("归还失败" in m for m in captured_warnings)

    async def test_terminate_failure_does_not_escape(self, captured_warnings):
        """断开本身再失败也只记 warning: 遥测绝不冒泡,剩下的交给池自行回收。"""
        conn = _FakePgConn(list(_EXPECTED_COLUMNS), fail_terminate=True)
        await _record_minimal(_pg_recorder(pool=_FakePgPool(conn, fail_release=True)))
        assert any("断开失败" in m for m in captured_warnings)
