PolyGateway/tests/integration/test_postgres_telemetry.py

"""PostgresRecorder 集成测试(M2 设计 §5;真实实验室 Postgres,polygateway 专用库)。

DSN 走 .env `PGW_TELEMETRY_PG_DSN`,缺则 skip。该实例上有 app/chs_prod 等
在用库——本测试只允许连 polygateway 专用库(fixture 里守卫)。

隔离纪律(M4 事故教训): `llm_calls` 是与真实批跑/迁移项目共享的表,
**严禁 DROP/TRUNCATE**——本测试以 run 级 call_id 前缀隔离,断言只看
自己写入的行,teardown 只删自己的行。
"""

from __future__ import annotations

import asyncio
import json
import os
import re
import time
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from uuid import uuid4

import pytest
from dotenv import dotenv_values

from polygateway.telemetry.postgres import PostgresRecorder
from polygateway.telemetry.schema import COLUMNS, telemetry_schema_sql

_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",
    "thinking_observation",
]

# run 级前缀: 同库并存的其他运行(迁移批跑/另一开发机)互不可见
_RUN_PREFIX = f"pgwtest-{uuid4().hex[:8]}"


def _cid(suffix: str) -> str:
    return f"{_RUN_PREFIX}-{suffix}"


def _dsn() -> str | None:
    merged = {**dotenv_values(".env"), **os.environ}
    raw = merged.get("PGW_TELEMETRY_PG_DSN")
    if not raw:
        return None
    scheme, sep, rest = raw.partition("://")
    return f"{scheme.partition('+')[0]}{sep}{rest}"


@pytest.fixture
async def dsn():
    value = _dsn()
    if value is None:
        pytest.skip("PGW_TELEMETRY_PG_DSN 未配置")
    # 隔离守卫: 该实例有 app/chs_prod/mimiciv 等在用库,只许打 polygateway 专用库
    if not value.rstrip("/").endswith("/polygateway"):
        pytest.fail(f"遥测测试只允许连 polygateway 专用库,当前 DSN 库名不符: {value!r}")
    yield value
    # teardown: 只删本 run 写入的行;表可能尚不存在(全新库)则忽略
    import asyncpg

    conn = await asyncpg.connect(value, timeout=10)
    try:
        if await conn.fetchval("SELECT to_regclass('llm_calls')") is not None:
            await conn.execute("DELETE FROM llm_calls WHERE call_id LIKE $1", f"{_RUN_PREFIX}-%")
    finally:
        await conn.close()


async def _record_minimal(
    recorder: PostgresRecorder, call_id: str | None = None, **overrides
) -> dict[str, object]:
    """记一行最小遥测,并**返回实际提交的字段**供调用方逐列比对回读结果。

    返回值不是顺手加的: 逐列断言若在测试里另抄一份期望值,抄错的那一列会以
    "库写错列位"的形态误报,而漏抄的列则悄悄不被验证。
    """
    fields: dict[str, object] = {
        "call_id": call_id if call_id is not None else _cid("c1"),
        "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": "{}",
        # 同样已由 emitter 归一化: 枚举取 .value 后才下沉,recorder 只见裸 str
        "thinking_observation": "unknown",
    }
    fields.update(overrides)
    await recorder.record_llm_call(**fields)
    return fields


# 集成用例统一的池上限与写入预算(issue #15;两者是 recorder 的必填 keyword-only)。
# 池上限取 config 的生产缺省(4),让本文件跑的就是下游真实会跑的那个形状。
#
# 预算却**远比生产的 5s 宽**,这不是抄错: `test_concurrent_writes_all_land` 一次
# 发 50 行,50 行共享 4 条连接,实测跨内网 RTT 123ms 下整批约 3.2s——而那 50 个
# `record_llm_call` 的预算是**同时**起算的,批越慢离预算越近。这个实例被多项目
# 共用,别人的一次负载尖峰就能让批耗时翻几倍,于是"丢行"变成掷硬币(pool_max=2
# 时实测批耗时 5.3s/15s 预算,已经在全套件里红过一次)。给它 60s 是把余量拉到
# 近 20 倍,让这个用例只在真出 bug 时红(CLAUDE.md §4.6: 重跑一次就绿的测试是
# 信号污染源)。突发排队本身超预算即丢行是设计上的既定取舍(设计 §6),不在此改。
_POOL_MAX = 4
_WRITE_TIMEOUT_S = 60.0


def _recorder(dsn: str, *, auto_migrate: bool) -> PostgresRecorder:
    """本文件唯一的 recorder 构造点: 池参数只写一遍,免得 16 处各抄一份。"""
    return PostgresRecorder(
        dsn, auto_migrate=auto_migrate, pool_max=_POOL_MAX, write_timeout_s=_WRITE_TIMEOUT_S
    )


async def _fetch(dsn: str, sql: str, *args):
    import asyncpg

    conn = await asyncpg.connect(dsn, timeout=10)
    try:
        return await conn.fetch(sql, *args)
    finally:
        await conn.close()


async def _execute_script(dsn: str, sql: str) -> None:
    """整段执行多语句脚本(不带参数,走简单查询协议)——模拟下游把脚本贴进 psql。"""
    import asyncpg

    conn = await asyncpg.connect(dsn, timeout=10)
    try:
        await conn.execute(sql)
    finally:
        await conn.close()


_LEGACY_DDL = """
CREATE TABLE {schema}.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            DOUBLE PRECISION,
    max_inter_token_ms DOUBLE PRECISION,
    cache_hit          BOOLEAN NOT NULL DEFAULT FALSE,
    error              TEXT,
    cost               DOUBLE PRECISION,
    created_at         TIMESTAMPTZ NOT NULL DEFAULT now()
)
"""


@pytest.fixture
async def legacy_schema(dsn):
    """在**自建的临时 schema** 里造一张 18 列旧表,验证补列(issue #3)。

    绝不碰共享的 public.llm_calls: 用 search_path 把 recorder 指向临时 schema,
    teardown 只 DROP 自己建的 schema。
    """
    import asyncpg

    name = f"pgwtest_{uuid4().hex[:8]}"
    conn = await asyncpg.connect(dsn, timeout=10)
    try:
        await conn.execute(f"CREATE SCHEMA {name}")
        await conn.execute(_LEGACY_DDL.format(schema=name))
    finally:
        await conn.close()
    sep = "&" if "?" in dsn else "?"
    yield f"{dsn}{sep}options=-csearch_path%3D{name}", name
    conn = await asyncpg.connect(dsn, timeout=10)
    try:
        await conn.execute(f"DROP SCHEMA {name} CASCADE")
    finally:
        await conn.close()


class TestObservabilityColumns:
    """issue #3: 两列写入可回读,且已存在的 18 列旧表会被自动补列。"""

    async def test_values_round_trip(self, dsn):
        recorder = _recorder(dsn, auto_migrate=True)
        try:
            await _record_minimal(recorder, call_id=_cid("hit"), cached_prompt_tokens=64)
            await _record_minimal(recorder, call_id=_cid("zero"), cached_prompt_tokens=0)
            await _record_minimal(recorder, call_id=_cid("model"), model_reported="MiniMax-01")
            await _record_minimal(
                recorder, call_id=_cid("samp"), sampling='{"seed": 42, "temperature": 0}'
            )
            rows = await _fetch(
                dsn,
                "SELECT call_id, cached_prompt_tokens, model_reported, sampling FROM llm_calls "
                "WHERE call_id LIKE $1",
                f"{_RUN_PREFIX}-%",
            )
            by_id = {r["call_id"]: r for r in rows}
            assert by_id[_cid("hit")]["cached_prompt_tokens"] == 64
            assert by_id[_cid("zero")]["cached_prompt_tokens"] == 0  # 真实零命中 ≠ NULL
            assert by_id[_cid("model")]["cached_prompt_tokens"] is None
            assert by_id[_cid("model")]["model_reported"] == "MiniMax-01"
            # issue #4: PG 侧也须验非空 sampling 能读回原值(不只是列存在)
            assert json.loads(by_id[_cid("samp")]["sampling"]) == {"seed": 42, "temperature": 0}
            assert by_id[_cid("hit")]["sampling"] is None
        finally:
            await recorder.aclose()

    async def test_legacy_table_is_upgraded_in_place(self, legacy_schema):
        """18 列旧表不补列的话,每行写入都会被逐行 warning 丢弃(遥测静默全失)。"""
        schema_dsn, schema = legacy_schema
        recorder = _recorder(schema_dsn, auto_migrate=True)
        try:
            await _record_minimal(
                recorder, call_id=_cid("legacy"), cached_prompt_tokens=7, model_reported="m-real"
            )
            cols = await _fetch(
                schema_dsn,
                "SELECT column_name FROM information_schema.columns "
                "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
                schema,
            )
            # ALTER 只能追加到末尾: 与新建库的列序一致才不会分叉
            assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS
            rows = await _fetch(
                schema_dsn,
                "SELECT cached_prompt_tokens, model_reported FROM llm_calls WHERE call_id = $1",
                _cid("legacy"),
            )
            assert (rows[0]["cached_prompt_tokens"], rows[0]["model_reported"]) == (7, "m-real")
        finally:
            await recorder.aclose()


class TestSchema:
    async def test_schema_has_frozen_columns_in_order(self, dsn):
        recorder = _recorder(dsn, auto_migrate=True)
        try:
            await _record_minimal(recorder)
            rows = await _fetch(
                dsn,
                "SELECT column_name FROM information_schema.columns "
                "WHERE table_name='llm_calls' ORDER BY ordinal_position",
            )
            assert [r["column_name"] for r in rows] == _EXPECTED_COLUMNS
        finally:
            await recorder.aclose()

    async def test_call_id_idempotent(self, dsn):
        recorder = _recorder(dsn, auto_migrate=True)
        try:
            await _record_minimal(recorder, call_id=_cid("dup"))
            await _record_minimal(recorder, call_id=_cid("dup"), response="second")
            rows = await _fetch(
                dsn, "SELECT response FROM llm_calls WHERE call_id = $1", _cid("dup")
            )
            assert [r["response"] for r in rows] == ["ok"]  # ON CONFLICT DO NOTHING
        finally:
            await recorder.aclose()

    async def test_concurrent_writes_all_land(self, dsn):
        recorder = _recorder(dsn, auto_migrate=True)
        try:
            await asyncio.gather(
                *(_record_minimal(recorder, call_id=_cid(f"c{i}")) for i in range(50))
            )
            rows = await _fetch(
                dsn,
                "SELECT count(*) AS n FROM llm_calls WHERE call_id LIKE $1",
                f"{_RUN_PREFIX}-c%",
            )
            assert rows[0]["n"] == 50
        finally:
            await recorder.aclose()


class _FakeClock:
    """可手动推进的单调时钟: 冷却窗口靠它测,用例里绝不真睡 60 秒。"""

    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


class TestDegradation:
    async def test_unreachable_server_degrades_silently(self):
        """服务端连不上 → warning 一次后降级,业务零感知(不抛、不拖)。"""
        recorder = _recorder("postgresql://u:p@127.0.0.1:1/x", auto_migrate=True)
        await _record_minimal(recorder)  # 不抛
        await _record_minimal(recorder, call_id=_cid("c2"))  # 已降级短路,同样不抛
        await recorder.aclose()

    async def test_refused_connection_cools_down_and_retries_after_cooldown(self):
        """连接被拒 → 冷却降级(**非 fatal**)→ 冷却期内零成本短路 → 到期真的重试。

        走**不可达 DSN** 而不是把共享实例的连接打满: 那台 PG 上还有 app/chs_prod
        等在用库,制造连接耗尽会伤到别人;而"连接被拒"与"连接耗尽"落的是同一档
        (环境级,`_classify_failure`),这条路验的是同一段状态机。

        **时序前提**(避免间歇红): 假时钟只驱动 tracker 的冷却窗口,与真实网络耗时
        完全无关,故三段断言都不依赖墙钟。`retry_after_s` 是"有没有真的重试过"的
        唯一外部信号——重试失败会给冷却窗口续期,而短路不会碰它。
        """
        clock = _FakeClock()
        recorder = PostgresRecorder(
            "postgresql://u:p@127.0.0.1:1/x",
            auto_migrate=True,
            pool_max=_POOL_MAX,
            write_timeout_s=_WRITE_TIMEOUT_S,
            now=clock,
        )
        try:
            await _record_minimal(recorder, call_id=_cid("deg1"))
            first = recorder.telemetry_status
            # 非 fatal 正是 issue #15 的核心: 连接被拒过去在建池那一步被一刀判死,
            # 整进程从此一行遥测都不落、只有重启能恢复
            assert (first.degraded, first.fatal) == (True, False)
            assert first.retry_after_s == pytest.approx(60.0)
            assert first.dropped_rows == 1
            # min_size=0 之后建池不再触库,连接被拒因此暴露在准备期而不是建池期
            assert "建表探测失败" in (first.reason or "")

            clock.advance(30.0)
            await _record_minimal(recorder, call_id=_cid("deg2"))
            mid = recorder.telemetry_status
            # 冷却窗口没被刷新 = 这次调用压根没去连库(降级期间零成本短路)
            assert mid.retry_after_s == pytest.approx(30.0)
            assert mid.dropped_rows == 2

            clock.advance(30.1)
            await _record_minimal(recorder, call_id=_cid("deg3"))
            after = recorder.telemetry_status
            # 冷却窗口被重新拉满 = 真的重连了一次(照旧被拒,故仍降级但仍可自愈)
            assert after.retry_after_s == pytest.approx(60.0)
            assert (after.degraded, after.fatal) == (True, False)
            assert after.dropped_rows == 3
        finally:
            await recorder.aclose()

    async def test_row_failure_does_not_poison_later_rows(self, dsn):
        """运行时单条写失败(NUL 字节文本被 PG 拒)→ 丢该行,后续行照常落库。"""
        recorder = _recorder(dsn, auto_migrate=True)
        try:
            await _record_minimal(recorder, call_id=_cid("bad"), response="nul\x00byte")
            await _record_minimal(recorder, call_id=_cid("good"))
            rows = await _fetch(
                dsn,
                "SELECT call_id FROM llm_calls WHERE call_id = ANY($1::text[]) ORDER BY call_id",
                [_cid("bad"), _cid("good")],
            )
            assert [r["call_id"] for r in rows] == [_cid("good")]
        finally:
            await recorder.aclose()

    async def test_aclose_idempotent(self, dsn):
        recorder = _recorder(dsn, auto_migrate=True)
        await _record_minimal(recorder)
        await recorder.aclose()
        await recorder.aclose()


def _tagged(dsn: str, app_name: str) -> str:
    """给 DSN 挂上 `application_name` 查询参数,让本池的连接在服务端可被点名。

    走 DSN 参数而不是给 recorder 加 `server_settings` 入口: 纯测试便利不值得
    扩公共 API(P1)。也**不能**改成"测试自建池后以 `pool=` 注入"——那会走
    `_external_pool` 分支、完全绕过被测的建池路径,而本节要验的恰恰是它。
    """
    sep = "&" if "?" in dsn else "?"
    return f"{dsn}{sep}application_name={app_name}"


async def _pool_backend_count(dsn: str, app_name: str) -> int:
    """数**本池**在服务端的连接数(只读查询,不改实例任何状态)。

    只按 run 级唯一的 `application_name` 过滤: 这台实例被多项目共用,按库名或
    用户名计数会把别人的连接算进来,做出的是设计上就会间歇红的用例
    (CLAUDE.md §4.6)。本查询自己那条连接走未打 tag 的 DSN,故不会数到自己。
    """
    rows = await _fetch(
        dsn, "SELECT count(*) AS n FROM pg_stat_activity WHERE application_name = $1", app_name
    )
    return rows[0]["n"]


async def _settled_backend_count(dsn: str, app_name: str, *, timeout_s: float = 5.0) -> int:
    """等本 tag 的连接数归零并返回最终值;超时则返回当下值,交给断言去红。

    轮询而不是一次采样: 客户端 `close()` 返回与服务端后台进程从
    `pg_stat_activity` 消失之间没有同步保证(实测立即归零,5s 余量只是不赌它)。
    """
    deadline = time.monotonic() + timeout_s
    while True:
        count = await _pool_backend_count(dsn, app_name)
        if count == 0 or time.monotonic() >= deadline:
            return count
        await asyncio.sleep(0.1)


class TestPoolFootprint:
    """issue #15 的直接回归钉子: 池不预连接,占用不超过库自己声明的上限。

    单元层断的是"`min_size`/`max_size` 传对了",这里断的是"服务端真的只开了
    那么多连接"——两件事,只有真实 PG 能证后者。
    """

    async def test_pool_does_not_preconnect_and_stays_within_pool_max(self, dsn):
        app_name = f"{_RUN_PREFIX}-pool"  # run 级唯一,与并跑的其他运行互不可见
        recorder = _recorder(_tagged(dsn, app_name), auto_migrate=True)
        try:
            # 构造只记参数、不触库: 这一条与下一条合起来才是钉子——修复前
            # `create_pool` 继承 asyncpg 的 min_size=10,首次写入后下面会是 10
            assert await _pool_backend_count(dsn, app_name) == 0

            await _record_minimal(recorder, call_id=_cid("fp1"))
            # **时序前提**: 写入已 await 到返回,连接必然已建立(没建立就写不成功),
            # 归还只是还进池而不断开,asyncpg 空闲回收是 300s 不会在用例内触发。
            # 故这是个确定值,不是"某一刻恰好的采样"
            assert await _pool_backend_count(dsn, app_name) == 1

            await asyncio.gather(
                *(_record_minimal(recorder, call_id=_cid(f"fp{i}")) for i in range(2, 22))
            )
            steady = await _pool_backend_count(dsn, app_name)
            # 上界由 max_size 保证;下界 ≥1 不是凑数——它确保过滤条件真的命中了本池,
            # 否则 tag 一旦拼错,上面那条 ==0 会以"永远绿"的形态通过
            assert 1 <= steady <= _POOL_MAX
        finally:
            await recorder.aclose()
        assert await _settled_backend_count(dsn, app_name) == 0  # 关闭即归还全部连接


_PROBE_PASSWORD = "pgw_issue9_probe"  # 临时角色,teardown 删除;非任何真实凭据


@pytest.fixture
async def least_privilege_dsn(dsn):
    """临时 schema + 临时角色: 只授表级 SELECT/INSERT,**不授 schema CREATE**。

    这是 issue #9 的现场——最小权限部署的标准形态。fixture 建的一切
    (schema、表、角色)都在 teardown 里删净,共享的 public.llm_calls 不受影响;
    连不上或无权建角色(非超级用户)时 skip,不让 CI 假绿。
    """
    import asyncpg

    from polygateway.telemetry.schema import PG_DDL

    name = f"pgwtest_lp_{uuid4().hex[:8]}"
    admin = await asyncpg.connect(dsn, timeout=10)
    try:
        if not await admin.fetchval(
            "SELECT rolcreaterole OR rolsuper FROM pg_roles WHERE rolname = current_user"
        ):
            pytest.skip("当前账号无权建临时角色,跳过最小权限用例")
        await admin.execute(f"CREATE ROLE {name} LOGIN PASSWORD '{_PROBE_PASSWORD}'")
        await admin.execute(f"CREATE SCHEMA {name}")
        await admin.execute(f"SET search_path = {name}")
        await admin.execute(PG_DDL)  # 表由**别的账号**建好,与现场一致
        await admin.execute(f"GRANT USAGE ON SCHEMA {name} TO {name}")
        await admin.execute(f"GRANT SELECT, INSERT ON {name}.llm_calls TO {name}")
        # 关键: 绝不 GRANT CREATE ON SCHEMA —— 缺的正是这一项
    finally:
        await admin.close()
    low = re.sub(r"//[^@/]+@", f"//{name}:{_PROBE_PASSWORD}@", dsn, count=1)
    sep = "&" if "?" in low else "?"
    yield f"{low}{sep}options=-csearch_path%3D{name}", name
    admin = await asyncpg.connect(dsn, timeout=10)
    try:
        await admin.execute(f"DROP SCHEMA IF EXISTS {name} CASCADE")
        await admin.execute(f"DROP OWNED BY {name}")
        await admin.execute(f"DROP ROLE IF EXISTS {name}")
    finally:
        await admin.close()


class TestLeastPrivilegeDeployment:
    """issue #9: 只有表级写权限的账号,遥测必须照常落库而不是整体判死。"""

    async def test_create_table_if_not_exists_is_denied_for_this_role(self, least_privilege_dsn):
        """库外事实先钉死: 表存在、写得进去,DDL 仍被拒——PG 的权限检查早于 IF NOT EXISTS。

        修复依赖的是这条 PG 语义;若某天它变了,这里先红,而不是让下面那条
        用例悄悄变成"永远通过"的空断言。
        """
        import asyncpg

        low_dsn, _ = least_privilege_dsn
        conn = await asyncpg.connect(low_dsn, timeout=10)
        try:
            assert await conn.fetchval("SELECT to_regclass('llm_calls')") is not None
            with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError):
                await conn.execute("CREATE TABLE IF NOT EXISTS llm_calls (call_id TEXT)")
        finally:
            await conn.close()

    async def test_records_land_without_schema_create_privilege(self, least_privilege_dsn):
        """修复前: 建表被拒 → 整体判死 → 整个进程一条不落(下游 150 次调用全丢)。"""
        low_dsn, schema = least_privilege_dsn
        recorder = _recorder(low_dsn, auto_migrate=True)
        try:
            await _record_minimal(recorder, call_id=_cid("lp1"))
            await _record_minimal(recorder, call_id=_cid("lp2"), cost=1.5)
            assert recorder.telemetry_status.degraded is False  # 建表权限不得触发降级
            rows = await _fetch(
                low_dsn,
                "SELECT call_id, cost FROM llm_calls WHERE call_id LIKE $1 ORDER BY call_id",
                f"{_RUN_PREFIX}-lp%",
            )
            assert [(r["call_id"], r["cost"]) for r in rows] == [
                (_cid("lp1"), None),
                (_cid("lp2"), 1.5),
            ]
            assert schema  # teardown 会连表带角色删净
        finally:
            await recorder.aclose()


# issue #11 之前的表形态: 22 个 recorder 字段 + created_at = 23 个物理列,没有任何租户维度
_PRE_TENANT_DDL = """
CREATE TABLE {schema}.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            DOUBLE PRECISION,
    max_inter_token_ms DOUBLE PRECISION,
    cache_hit          BOOLEAN NOT NULL DEFAULT FALSE,
    error              TEXT,
    cost               DOUBLE PRECISION,
    created_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
    cached_prompt_tokens INTEGER,
    model_reported       TEXT,
    sampling             TEXT,
    reasoning_tokens     INTEGER
)
"""

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


# `_PRE_TENANT_DDL` 的物理列(23 个): 由 `_EXPECTED_COLUMNS` 去掉此后新增的三列
# 派生而非另抄一份——两份常量必然漂移,而漂移的表现是"manual 档没补列"这条断言假绿。
# 去掉后的顺序与 DDL 逐字一致(这三列在 DDL 里本就排在末尾)。
_PRE_TENANT_COLUMNS = [
    c for c in _EXPECTED_COLUMNS if c not in ("tenant_id", "meta", "thinking_observation")
]

# 回读要逐列比对的字段: 物理列去掉库从不显式写的 created_at,恰好 22 个
_PRE_TENANT_WRITTEN_COLUMNS = [c for c in _PRE_TENANT_COLUMNS if c != "created_at"]


def _search_path_dsn(dsn: str, schema: str) -> str:
    sep = "&" if "?" in dsn else "?"
    return f"{dsn}{sep}options=-csearch_path%3D{schema}"


@pytest.fixture
async 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)


@pytest.fixture
async def pre_tenant_schema(dsn):
    """自建临时 schema 里造一张 **22 字段的 issue #11 之前的表**,并留一行历史数据。

    绝不碰共享的 public.llm_calls——本机那张表早已被 `_BACKFILL` 真实补过列,
    指望它还是旧形态的测试第二次跑就会空转。schema 名带 uuid,可重复运行。
    """
    import asyncpg

    name = f"pgwtest_pre_{uuid4().hex[:8]}"
    conn = await asyncpg.connect(dsn, timeout=10)
    try:
        await conn.execute(f"CREATE SCHEMA {name}")
        await conn.execute(_PRE_TENANT_DDL.format(schema=name))
        await conn.execute(_PRE_TENANT_INSERT.format(schema=name), _cid("old"))
    finally:
        await conn.close()
    yield _search_path_dsn(dsn, name), name
    conn = await asyncpg.connect(dsn, timeout=10)
    try:
        await conn.execute(f"DROP SCHEMA {name} CASCADE")
    finally:
        await conn.close()


@pytest.fixture
async def fresh_schema(dsn):
    """空 schema: recorder 自己建表,验"新建库"这条路径而不依赖共享表的历史状态。"""
    import asyncpg

    name = f"pgwtest_new_{uuid4().hex[:8]}"
    conn = await asyncpg.connect(dsn, timeout=10)
    try:
        await conn.execute(f"CREATE SCHEMA {name}")
    finally:
        await conn.close()
    yield _search_path_dsn(dsn, name), name
    conn = await asyncpg.connect(dsn, timeout=10)
    try:
        await conn.execute(f"DROP SCHEMA {name} CASCADE")
    finally:
        await conn.close()


@pytest.fixture
async def least_privilege_pre_tenant_dsn(dsn):
    """22 字段旧表 + 只有 `SELECT, INSERT` 权限的角色: 补列必然失败的现场。

    与 `least_privilege_dsn` 分开而非复用: 那个 fixture 建的是列已齐全的当前表
    (测的是 CREATE 被拒),这里必须是缺列的旧表,才能让 `ALTER TABLE` 真的发出去
    并撞上 ownership 检查(该检查早于 `IF NOT EXISTS` 的存在性判断)。
    """
    import asyncpg

    name = f"pgwtest_lppre_{uuid4().hex[:8]}"
    admin = await asyncpg.connect(dsn, timeout=10)
    try:
        if not await admin.fetchval(
            "SELECT rolcreaterole OR rolsuper FROM pg_roles WHERE rolname = current_user"
        ):
            pytest.skip("当前账号无权建临时角色,跳过最小权限用例")
        await admin.execute(f"CREATE ROLE {name} LOGIN PASSWORD '{_PROBE_PASSWORD}'")
        await admin.execute(f"CREATE SCHEMA {name}")
        await admin.execute(_PRE_TENANT_DDL.format(schema=name))  # 表属主是 admin,不是应用账号
        await admin.execute(f"GRANT USAGE ON SCHEMA {name} TO {name}")
        await admin.execute(f"GRANT SELECT, INSERT ON {name}.llm_calls TO {name}")
    finally:
        await admin.close()
    low = re.sub(r"//[^@/]+@", f"//{name}:{_PROBE_PASSWORD}@", dsn, count=1)
    yield _search_path_dsn(low, name)
    admin = await asyncpg.connect(dsn, timeout=10)
    try:
        await admin.execute(f"DROP SCHEMA IF EXISTS {name} CASCADE")
        await admin.execute(f"DROP OWNED BY {name}")
        await admin.execute(f"DROP ROLE IF EXISTS {name}")
    finally:
        await admin.close()


class TestCallerDimensionsAcceptance:
    """issue #11 的机械化验收(PG 侧,真实实例): 新建库 / 旧表补列 / 补列失败方向。"""

    async def test_fresh_schema_round_trips_the_dimensions(self, fresh_schema):
        """新建库: 列齐全,且维度值原样读回——只验列存在会漏掉写错列位的错。"""
        fresh_dsn, schema = fresh_schema
        recorder = _recorder(fresh_dsn, auto_migrate=True)
        try:
            await _record_minimal(
                recorder, call_id=_cid("dim"), tenant_id="tenant-a", meta='{"batch": "b7"}'
            )
            cols = await _fetch(
                fresh_dsn,
                "SELECT column_name FROM information_schema.columns "
                "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
                schema,
            )
            assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS
            rows = await _fetch(
                fresh_dsn,
                "SELECT tenant_id, meta FROM llm_calls WHERE call_id = $1",
                _cid("dim"),
            )
            assert rows[0]["tenant_id"] == "tenant-a"
            assert json.loads(rows[0]["meta"]) == {"batch": "b7"}
        finally:
            await recorder.aclose()

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

        这条直接验收 issue #11 的核心论点(先启用落库、后加列,补列之前的行没有
        租户归属)。断言方向必须是空串: PG 的 RLS `USING` 表达式对返回 false **或
        NULL** 的行一律隐藏且不报错,故 NULL 的 `tenant_id` 不是"未归属",而是对
        所有人永久不可见的黑洞;哨兵空串则能被一条 `COUNT(*) WHERE tenant_id = ''`
        审计出来,历史欠账是可见、可量化、可补录的。
        """
        schema_dsn, schema = pre_tenant_schema
        recorder = _recorder(schema_dsn, auto_migrate=True)
        try:
            await _record_minimal(
                recorder, call_id=_cid("new"), tenant_id="tenant-a", meta='{"k": 1}'
            )
            cols = await _fetch(
                schema_dsn,
                "SELECT column_name FROM information_schema.columns "
                "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
                schema,
            )
            # 22 → 25 个 recorder 字段(加 created_at 共 26 个物理列),且新列追加在末尾
            assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS
            rows = await _fetch(
                schema_dsn,
                "SELECT call_id, tenant_id, meta FROM llm_calls "
                "WHERE call_id = ANY($1::text[]) ORDER BY call_id",
                [_cid("new"), _cid("old")],
            )
            by_id = {r["call_id"]: r for r in rows}
            assert by_id[_cid("new")]["tenant_id"] == "tenant-a"
            assert json.loads(by_id[_cid("new")]["meta"]) == {"k": 1}
            assert by_id[_cid("old")]["tenant_id"] == ""  # 不是 None: NULL 会被 RLS 静默吞掉
            assert json.loads(by_id[_cid("old")]["meta"]) == {}
        finally:
            await recorder.aclose()

    async def test_alter_is_denied_for_a_role_that_can_still_insert(
        self, least_privilege_pre_tenant_dsn
    ):
        """库外事实先钉死: 表存在、写得进去,补列的 ALTER 仍被拒(ownership 检查早于存在性判断)。

        没有这条,下面那个降级用例可能因为 ALTER 其实成功了而变成"永远通过"的空断言。
        """
        import asyncpg

        conn = await asyncpg.connect(least_privilege_pre_tenant_dsn, timeout=10)
        try:
            assert await conn.fetchval("SELECT to_regclass('llm_calls')") is not None
            with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError):
                await conn.execute("ALTER TABLE llm_calls ADD COLUMN IF NOT EXISTS tenant_id TEXT")
        finally:
            await conn.close()

    async def test_backfill_failure_degrades_per_row_not_wholesale(
        self, least_privilege_pre_tenant_dsn, captured_warnings
    ):
        """补列失败的降级方向: 记 warning、不整体降级、后续 INSERT 仍照发。

        整体降级会让整个进程停写(比逐行丢弃严重得多),而缺列(SQLSTATE 42703)
        是判据的唯一具名例外: 必须逐行暴露,好让下游看见 schema 漂移(issue #13)。
        """
        recorder = _recorder(least_privilege_pre_tenant_dsn, auto_migrate=True)
        try:
            await _record_minimal(recorder, call_id=_cid("lpp1"))  # 不得抛
            assert recorder.telemetry_status.degraded is False
            assert any("补列失败" in m for m in captured_warnings)
            # 缺列的表上 INSERT 必然失败;逐行 warning 正是"INSERT 照发了"的证据
            assert any("写入失败" in m for m in captured_warnings)
        finally:
            await recorder.aclose()


# issue #12 的目标表形态: 按 created_at 做 RANGE 分区(过期清理 DROP PARTITION 而非 DELETE)。
# PG 强制分区表的唯一约束必须包含分区键,故主键只能是 (call_id, created_at) ——
# 这正是带目标的 `ON CONFLICT (call_id)` 再也匹配不到约束的现场。
_PARTITIONED_DDL = """
CREATE TABLE {schema}.llm_calls (
    call_id            TEXT NOT NULL,
    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            DOUBLE PRECISION,
    max_inter_token_ms DOUBLE PRECISION,
    cache_hit          BOOLEAN NOT NULL DEFAULT FALSE,
    error              TEXT,
    cost               DOUBLE PRECISION,
    created_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
    cached_prompt_tokens INTEGER,
    model_reported       TEXT,
    sampling             TEXT,
    reasoning_tokens     INTEGER,
    tenant_id            TEXT NOT NULL DEFAULT '',
    meta                 JSONB NOT NULL DEFAULT '{{}}'::jsonb,
    PRIMARY KEY (call_id, created_at)
) PARTITION BY RANGE (created_at)
"""

_PARTITION_DDL = (
    "CREATE TABLE {schema}.llm_calls_current PARTITION OF {schema}.llm_calls "
    "FOR VALUES FROM ('{start}') TO ('{end}')"
)


def _current_month_bounds() -> tuple[str, str]:
    """当前月的 [月初, 下月初) 边界字面量;分区键落在区间外会因找不到分区而写失败。"""
    now = datetime.now(UTC)
    start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    end = (start + timedelta(days=32)).replace(day=1)
    fmt = "%Y-%m-%d %H:%M:%S%z"
    return start.strftime(fmt), end.strftime(fmt)


@pytest.fixture
async def partitioned_schema(dsn):
    """自建临时 schema 里造一张按 created_at RANGE 分区的表 + 覆盖当前月的分区。

    与 legacy_schema 同款隔离: 绝不碰共享的 public.llm_calls,teardown 只 DROP
    自己建的 schema(CASCADE 连分区一并删)。
    """
    import asyncpg

    name = f"pgwtest_part_{uuid4().hex[:8]}"
    start, end = _current_month_bounds()
    conn = await asyncpg.connect(dsn, timeout=10)
    try:
        await conn.execute(f"CREATE SCHEMA {name}")
        await conn.execute(_PARTITIONED_DDL.format(schema=name))
        await conn.execute(_PARTITION_DDL.format(schema=name, start=start, end=end))
    finally:
        await conn.close()
    yield _search_path_dsn(dsn, name), name
    conn = await asyncpg.connect(dsn, timeout=10)
    try:
        await conn.execute(f"DROP SCHEMA {name} CASCADE")
    finally:
        await conn.close()


class TestConflictTargetFreeInsert:
    """issue #13: INSERT 不绑定冲突目标,普通表与分区表两种形态都写得进去。"""

    async def test_plain_table_still_dedupes_by_call_id(self, fresh_schema, captured_warnings):
        """普通表上语义不变: 重复 call_id 仍只落一行,且不是被拒后丢弃。

        表上只有主键这一个唯一约束,故无目标的 DO NOTHING 与 `(call_id)` 逐字等价;
        断言"无写入失败 warning"是为了区分"冲突被忽略"与"整条被 PG 拒收"。
        """
        fresh_dsn, _ = fresh_schema
        recorder = _recorder(fresh_dsn, auto_migrate=True)
        try:
            await _record_minimal(recorder, call_id=_cid("nodup"))
            await _record_minimal(recorder, call_id=_cid("nodup"), response="second")
            assert [m for m in captured_warnings if "写入失败" in m] == []
            rows = await _fetch(
                fresh_dsn, "SELECT response FROM llm_calls WHERE call_id = $1", _cid("nodup")
            )
            assert [r["response"] for r in rows] == ["ok"]  # 首行胜出,写入幂等
        finally:
            await recorder.aclose()

    async def test_partitioned_table_accepts_writes(self, partitioned_schema, captured_warnings):
        """分区表上写入成功且能读回——改动前这里必红。

        带目标的 `ON CONFLICT (call_id)` 在主键为 `(call_id, created_at)` 的表上
        匹配不到任何约束,PG 报 "there is no unique or exclusion constraint matching
        the ON CONFLICT specification";该错误被逐行降级吞成 warning,于是分区部署下
        遥测全线写不进去却一声不吭,只能靠"读不回来"暴露。
        """
        part_dsn, _ = partitioned_schema
        recorder = _recorder(part_dsn, auto_migrate=True)
        try:
            await _record_minimal(recorder, call_id=_cid("part"), tenant_id="tenant-p")
            assert [m for m in captured_warnings if "写入失败" in m] == []
            rows = await _fetch(
                part_dsn,
                "SELECT call_id, tenant_id FROM llm_calls WHERE call_id = $1",
                _cid("part"),
            )
            assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(_cid("part"), "tenant-p")]
        finally:
            await recorder.aclose()


class TestManualSchemaModeAcceptance:
    """issue #13 manual 档的真实实例验收: 旧表原样不动,写入照常,缺列只作提示。

    manual 档的承诺是"库一条 DDL 都不发"——单元测试只能验"没调用 execute",
    真表上才验得了"表结构确实没变"。两条用例分别覆盖有权补列却不补(纪律)与
    无权补列(现场),后者正是 auto 档会刷出 `补列失败` warning 的那张表。
    """

    async def test_manual_leaves_the_stale_table_untouched(
        self, pre_tenant_schema, captured_warnings
    ):
        """22 字段旧表 + manual: 列一个不加,行照常落库,缺的两维度静默不写。

        与 `test_pre_tenant_table_gains_columns_and_old_rows_stay_auditable` 恰成对照:
        同一张表、同一份负载,只有 `auto_migrate` 不同,列数就必须是 23 与 25 之别。
        """
        schema_dsn, schema = pre_tenant_schema
        recorder = _recorder(schema_dsn, auto_migrate=False)
        try:
            recorded = await _record_minimal(
                recorder, call_id=_cid("man"), tenant_id="tenant-a", meta='{"k": 1}'
            )
            cols = await _fetch(
                schema_dsn,
                "SELECT column_name FROM information_schema.columns "
                "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
                schema,
            )
            # 表结构逐字不动: 既没多出 tenant_id/meta,也没被顺手改了列序
            assert [r["column_name"] for r in cols] == _PRE_TENANT_COLUMNS

            names = ", ".join(_PRE_TENANT_WRITTEN_COLUMNS)
            rows = await _fetch(
                schema_dsn, f"SELECT {names} FROM llm_calls WHERE call_id = $1", _cid("man")
            )
            assert len(rows) == 1  # 裁剪后的 INSERT 真写进去了,不是被 PG 拒收
            # 其余 22 列逐列与提交值相等: 少写两列最容易引发的错是剩下的值整体错位
            assert dict(rows[0]) == {c: recorded[c] for c in _PRE_TENANT_WRITTEN_COLUMNS}

            assert [m for m in captured_warnings if "写入失败" in m] == []
            assert [m for m in captured_warnings if "补列失败" in m] == []
            notices = [m for m in captured_warnings if "auto_migrate=False" in m]
            assert len(notices) == 1  # 准备期一次讲清,不逐行刷屏
            assert "以下维度不会被记录: tenant_id, meta" in notices[0]
        finally:
            await recorder.aclose()

    async def test_manual_on_a_role_that_cannot_alter_emits_no_backfill_failure(
        self, least_privilege_pre_tenant_dsn, captured_warnings
    ):
        """缺列旧表 + 只授 SELECT/INSERT 的角色 + manual: 补列失败的 warning 彻底消失。

        auto 档在这张表上会刷出 `补列失败` 再刷 `写入失败`(见
        `test_backfill_failure_degrades_per_row_not_wholesale`)——那是 issue #13 要
        消灭的噪声。manual 档下 ALTER 压根不发,取而代之的是一条点名缺列并附可直接
        执行的 ALTER 的提示,而遥测照常落库。
        """
        recorder = _recorder(least_privilege_pre_tenant_dsn, auto_migrate=False)
        try:
            recorded = await _record_minimal(
                recorder, call_id=_cid("manlp1"), tenant_id="tenant-b", meta='{"k": 2}'
            )
            await _record_minimal(recorder, call_id=_cid("manlp2"), cost=2.5)

            assert [m for m in captured_warnings if "补列失败" in m] == []
            assert [m for m in captured_warnings if "写入失败" in m] == []
            assert recorder.telemetry_status.degraded is False
            notices = [m for m in captured_warnings if "auto_migrate=False" in m]
            assert len(notices) == 1  # 准备期一次,第二行不再重复
            assert "以下维度不会被记录: tenant_id, meta" in notices[0]
            # 提示里的 SQL 必须可直接粘贴执行,而不是只报个列名
            assert (
                "ALTER TABLE llm_calls ADD COLUMN tenant_id TEXT NOT NULL DEFAULT '';" in notices[0]
            )
            assert (
                "ALTER TABLE llm_calls ADD COLUMN meta JSONB NOT NULL DEFAULT '{}'::jsonb;"
                in notices[0]
            )

            # 该角色无权 ALTER,表必然还是旧形态: 缺的两列确实没被写
            cols = await _fetch(
                least_privilege_pre_tenant_dsn,
                "SELECT column_name FROM information_schema.columns "
                "WHERE table_schema = current_schema() AND table_name = 'llm_calls' "
                "ORDER BY ordinal_position",
            )
            assert [r["column_name"] for r in cols] == _PRE_TENANT_COLUMNS

            names = ", ".join(_PRE_TENANT_WRITTEN_COLUMNS)
            rows = await _fetch(
                least_privilege_pre_tenant_dsn,
                f"SELECT {names} FROM llm_calls WHERE call_id LIKE $1 ORDER BY call_id",
                f"{_RUN_PREFIX}-manlp%",
            )
            assert [r["call_id"] for r in rows] == [_cid("manlp1"), _cid("manlp2")]
            assert dict(rows[0]) == {c: recorded[c] for c in _PRE_TENANT_WRITTEN_COLUMNS}
            assert rows[1]["cost"] == 2.5
        finally:
            await recorder.aclose()


_PHYSICAL_COLUMNS_SQL = (
    "SELECT column_name FROM information_schema.columns "
    "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position"
)


class TestPublishedSchemaScript:
    """issue #13: README 叫下游执行的那份脚本,在真实实例上必须建得出、且可重复执行。

    这份脚本是 `telemetry_schema_sql("postgres")` 的输出,manual 档下游拿它建表,
    库随后靠列探测决定写哪些列——脚本与 `COLUMNS` 一旦漂移,表现是"照文档建完表,
    库仍报缺列"。人工核对不构成回归保护: 改一次 README 或 DDL 就会悄悄失去它。
    """

    async def test_script_builds_the_full_table_and_is_rerunnable(self, fresh_schema):
        """空 schema 里执行一遍建出全部物理列;再执行一遍不报错。

        第二遍是 `ADD COLUMN IF NOT EXISTS` 的幂等性验收: 去掉 IF NOT EXISTS 后,
        建表语句会被 `IF NOT EXISTS` 跳过而补列语句撞上 "column ... already exists",
        整段脚本第二次执行即失败——而"可重复执行"正是这份脚本对下游的承诺。
        """
        fresh_dsn, schema = fresh_schema
        script = telemetry_schema_sql("postgres")

        await _execute_script(fresh_dsn, script)
        actual = [r["column_name"] for r in await _fetch(fresh_dsn, _PHYSICAL_COLUMNS_SQL, schema)]
        # 物理列 = 25 个 INSERT 字段 + 库从不显式写的 created_at;对着库常量比,不另抄一份
        assert set(actual) == set(COLUMNS) | {"created_at"}
        # 列序也不许漂: 新列必须排在 created_at 之后,否则新建库与 ALTER 升级的列序分叉
        assert actual == _EXPECTED_COLUMNS

        await _execute_script(fresh_dsn, script)  # 可重复执行: 第二遍不得抛
        rerun = [r["column_name"] for r in await _fetch(fresh_dsn, _PHYSICAL_COLUMNS_SQL, schema)]
        assert rerun == actual  # 且第二遍没有偷偷改动表结构


# ---------------------------------------------------------------------------
# issue #12 Task 4: README 的生产部署 DDL 模板,逐条在真实 PG 上执行
# ---------------------------------------------------------------------------

# 模板 SQL **只有一份**,在 README 里。测试从 README 解析出来跑,而不是在这里另抄
# 一份: 抄一份就是两份会各自漂移的东西,而"README 里的 SQL 能跑"这个承诺恰恰只在
# 同源时才成立(doctest / Rust doc tests / mdbook test 都是这个范式)。
_README = Path(__file__).resolve().parents[2] / "README.md"

# 锚点写成 HTML 注释,渲染时不可见,比按章节标题或代码块序号定位稳固得多。
_TEMPLATE_BLOCK = re.compile(r"<!-- pg-template:([a-z_]+) -->\s*\n```sql\n(.*?)\n```", re.DOTALL)

# 顺序即执行顺序;数量与名字都钉死——解析不到或多出一块必须当场红,
# 绝不能退化成空列表让这条测试变成永远绿的摆设。
_EXPECTED_TEMPLATE_BLOCKS = (
    "roles",
    "table",
    "partition",
    "grants",
    "immutable",
    "rls",
    "index",
)

# README 里必须原样保留、由本测试做受控替换的标识符。README 那份是给下游照抄的,
# 故占位符是**合法可执行的具体值**而不是 `<schema>` 之类的尖括号洞。
_TEMPLATE_PLACEHOLDERS = (
    "polygateway_owner",
    "polygateway_app",
    "polygateway_report",
    "CHANGE_ME_APP",
    "CHANGE_ME_REPORT",
    "SCHEMA public",
    "llm_calls_2026_01",
    "'2026-01-01 00:00:00+00'",
    "'2026-02-01 00:00:00+00'",
)

# 应用角色在生产里能发的唯一一类写语句(与库的 INSERT 同形,只列 NOT NULL 列)
_TEMPLATE_INSERT = (
    "INSERT INTO llm_calls (call_id, model, provider, source_name, messages, response, "
    "prompt_tokens, completion_tokens, usage_source, latency_ms, tenant_id) "
    "VALUES ($1, 'm', 'p', 's1', '[]', 'ok', 1, 2, 'measured', 10, $2)"
)


def _template_blocks() -> dict[str, str]:
    """从 README 解析带锚点的 SQL 块;顺序即文中出现顺序。"""
    return dict(_TEMPLATE_BLOCK.findall(_README.read_text(encoding="utf-8")))


@dataclass(frozen=True)
class _TemplateEnv:
    """模板部署完成后的现场句柄:三个角色各自的连接串 + 当月分区名。"""

    admin_dsn: str
    app_dsn: str
    report_dsn: str
    schema: str
    partition: str
    seeded: tuple[str, str]  # (tenant-a 的行, tenant-b 的行)


def _localize(sql: str, schema: str, roles: dict[str, str], month: datetime) -> str:
    """把 README 里给下游照抄的标识符换成本次运行专属的临时对象。

    替换规则写在测试里而不是让 README 变得不可直接复制: README 里那份必须是
    下游 `pip install` 后照抄就能用的,占位符因此都是合法 SQL 值。
    """
    start = month.strftime("%Y-%m-%d %H:%M:%S%z")
    end = (month + timedelta(days=32)).replace(day=1).strftime("%Y-%m-%d %H:%M:%S%z")
    for placeholder, actual in (
        # 长名在前: 三个角色名互不为前缀,但顺序稳定便于排查
        ("polygateway_owner", roles["owner"]),
        ("polygateway_report", roles["report"]),
        ("polygateway_app", roles["app"]),
        ("CHANGE_ME_APP", _PROBE_PASSWORD),
        ("CHANGE_ME_REPORT", _PROBE_PASSWORD),
        ("SCHEMA public", f"SCHEMA {schema}"),
        ("llm_calls_2026_01", f"llm_calls_{month:%Y_%m}"),
        ("'2026-01-01 00:00:00+00'", f"'{start}'"),
        ("'2026-02-01 00:00:00+00'", f"'{end}'"),
    ):
        sql = sql.replace(placeholder, actual)
    return sql


def _role_dsn(dsn: str, role: str, schema: str) -> str:
    low = re.sub(r"//[^@/]+@", f"//{role}:{_PROBE_PASSWORD}@", dsn, count=1)
    return _search_path_dsn(low, schema)


async def _drop_template_objects(dsn: str, schema: str, roles: dict[str, str]) -> None:
    """删净临时 schema 与三个角色(角色是**全局**对象,漏删会跨 run 残留)。"""
    import asyncpg

    admin = await asyncpg.connect(dsn, timeout=10)
    try:
        await admin.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE")
        for role in roles.values():
            await admin.execute(f"DROP OWNED BY {role}")
            await admin.execute(f"DROP ROLE IF EXISTS {role}")
    finally:
        await admin.close()


@pytest.fixture
async def production_template(dsn):
    """在临时 schema + 临时角色上跑完 README 的整套模板,产出可用的三条连接串。

    隔离纪律(M4 事故教训)同 `least_privilege_dsn`: 共享的 `public.llm_calls`
    一个字节都不碰,建的 schema / 角色 / 函数 / 分区在 teardown 里删净。
    """
    import asyncpg

    suffix = uuid4().hex[:8]
    schema = f"pgwtpl_{suffix}"
    roles = {
        "owner": f"pgwtpl_owner_{suffix}",
        "app": f"pgwtpl_app_{suffix}",
        "report": f"pgwtpl_report_{suffix}",
    }
    month = datetime.now(UTC).replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    blocks = _template_blocks()
    # 解析不到就地红: 空 dict 会让下面的 for 一句不执行,测试变成"只验证了能连上库"
    assert list(blocks) == list(_EXPECTED_TEMPLATE_BLOCKS), (
        f"README 的模板锚点与预期不符: {list(blocks)}"
    )

    seeded = (_cid("tpl-a"), _cid("tpl-b"))
    admin_dsn = _search_path_dsn(dsn, schema)
    admin = await asyncpg.connect(dsn, timeout=10)
    # 权限门放在建任何对象**之前**: `pytest.skip` 抛的是 BaseException,
    # 若它在下面的清理块内触发,清理会去 DROP 从未建过的角色而把 skip 盖掉
    can_create = await admin.fetchval(
        "SELECT rolcreaterole OR rolsuper FROM pg_roles WHERE rolname = current_user"
    )
    if not can_create:
        await admin.close()
        pytest.skip("当前账号无权建临时角色,跳过生产模板用例")
    try:
        await admin.execute(f"CREATE SCHEMA {schema}")
        await admin.execute(f"SET search_path = {schema}")
        # README §2 写明的前置步骤: 先用库自带脚本建出普通表当模子
        await admin.execute(telemetry_schema_sql("postgres"))
        for name in _EXPECTED_TEMPLATE_BLOCKS:
            await admin.execute(_localize(blocks[name], schema, roles, month))
        # 种两个租户的行(超级用户绕过 RLS,属于布景不属于被测行为)
        for call_id, tenant in zip(seeded, ("tenant-a", "tenant-b"), strict=True):
            await admin.execute(_TEMPLATE_INSERT, call_id, tenant)
    except BaseException:
        # 模板 SQL 出错时也必须删净: 建到一半的 schema 会残留一张 llm_calls,
        # 而 `TestSchema` 那条按 table_name 查 information_schema 的用例不带
        # schema 过滤,会被残留物在**下一次运行**里以列数不符的形态误伤
        await admin.close()
        await _drop_template_objects(dsn, schema, roles)
        raise
    finally:
        if not admin.is_closed():
            await admin.close()

    yield _TemplateEnv(
        admin_dsn=admin_dsn,
        app_dsn=_role_dsn(dsn, roles["app"], schema),
        report_dsn=_role_dsn(dsn, roles["report"], schema),
        schema=schema,
        partition=f"llm_calls_{month:%Y_%m}",
        seeded=seeded,
    )

    admin = await asyncpg.connect(dsn, timeout=10)
    try:
        await admin.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE")
        for role in roles.values():
            await admin.execute(f"DROP OWNED BY {role}")
            await admin.execute(f"DROP ROLE IF EXISTS {role}")
    finally:
        await admin.close()


class TestProductionTemplate:
    """issue #12: README 的生产部署 DDL 模板必须逐条可执行,且行为与文中描述一致。

    模板出错的代价全部落在下游身上(照抄就中招),而人工核对不构成回归保护——
    改一次 README 就会悄悄失去它。故这里从 README **直接解析** SQL 来执行。
    """

    def test_readme_exposes_exactly_the_expected_template_blocks(self):
        """先钉死解析本身: 锚点没了、改名了、块数变了,这条当场红。

        没有它,`production_template` 里解析出空 dict 时下面每条用例都会以
        "表不存在"之类的间接形态失败,真因(README 结构变了)要靠猜。
        """
        blocks = _template_blocks()
        assert list(blocks) == list(_EXPECTED_TEMPLATE_BLOCKS)
        assert all(sql.strip() for sql in blocks.values())
        joined = "\n".join(blocks.values())
        for placeholder in _TEMPLATE_PLACEHOLDERS:
            # 占位符没了 = 受控替换静默失效,测试会去打真实的 polygateway_* 角色
            assert placeholder in joined, f"README 模板缺占位符 {placeholder!r}"

        # 再钉死列的**同源性**: `table` 块必须靠 `LIKE llm_calls_seed` 从库自建的表派生
        # 列,绝不能手抄一份列定义。手抄的那份会与 telemetry/schema.py 各自漂移,而漂移
        # 的表现是照模板部署的下游少掉新增列——manual 档下库按现有列裁剪写入,那一列
        # 就此静默消失,正是可观测性 issue 要消灭的那类静默。
        table_sql = blocks["table"]
        assert "LIKE llm_calls_seed" in table_sql, "生产模板的列必须由 LIKE 派生,不得手抄"
        inlined = [
            column
            for column in COLUMNS
            if re.search(rf"^\s*{column}\s+[A-Z]", table_sql, re.MULTILINE)
        ]
        assert not inlined, f"生产模板内联了列定义 {inlined},与 telemetry/schema.py 必然漂移"

    async def test_app_can_insert_but_cannot_mutate(self, production_template):
        """应用角色: INSERT 通过,UPDATE / DELETE 被权限层拒绝(不是被触发器拒)。

        权限检查早于行级触发器,故这里拿到的必须是 InsufficientPrivilegeError——
        若换成触发器的 RaiseError,说明 REVOKE 那一块没生效,而"不可变"就只剩
        一层属主随手可关的兜底。
        """
        import asyncpg

        env = production_template
        conn = await asyncpg.connect(env.app_dsn, timeout=10)
        try:
            await conn.execute(_TEMPLATE_INSERT, _cid("tpl-app"), "tenant-a")
            with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError):
                await conn.execute("DELETE FROM llm_calls WHERE call_id = $1", _cid("tpl-app"))
            with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError):
                await conn.execute("UPDATE llm_calls SET response = 'x'")
        finally:
            await conn.close()
        rows = await _fetch(
            env.admin_dsn, "SELECT call_id FROM llm_calls WHERE call_id = $1", _cid("tpl-app")
        )
        assert [r["call_id"] for r in rows] == [_cid("tpl-app")]  # 写入真落库了

    async def test_report_can_read_but_cannot_write(self, production_template):
        """报表角色: 带租户上下文读得到自己的行,任何写入都被拒。"""
        import asyncpg

        env = production_template
        conn = await asyncpg.connect(env.report_dsn, timeout=10)
        try:
            with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError):
                await conn.execute(_TEMPLATE_INSERT, _cid("tpl-rpt"), "tenant-a")
            async with conn.transaction():
                await conn.execute("SELECT set_config('app.tenant_id', 'tenant-a', true)")
                rows = await conn.fetch("SELECT call_id, tenant_id FROM llm_calls")
            assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(env.seeded[0], "tenant-a")]
        finally:
            await conn.close()

    async def test_reads_are_fail_closed_until_the_tenant_guc_is_set(self, production_template):
        """未设 `app.tenant_id` → 零行(fail-closed);设了 → 只看得到本租户。

        两个断言缺一不可: 只验"设了能看到自己的"漏掉了 GUC 未设时全表泄露,
        只验"未设是零行"则一条永远返回 false 的 policy 也能通过。
        """
        import asyncpg

        env = production_template
        conn = await asyncpg.connect(env.app_dsn, timeout=10)
        try:
            async with conn.transaction():
                assert await conn.fetch("SELECT call_id FROM llm_calls") == []
            async with conn.transaction():
                await conn.execute("SELECT set_config('app.tenant_id', 'tenant-b', true)")
                rows = await conn.fetch("SELECT call_id, tenant_id FROM llm_calls")
            assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(env.seeded[1], "tenant-b")]
        finally:
            await conn.close()

    async def test_rows_land_in_the_current_month_partition(self, production_template):
        """分区表写入成功,且行确实落进当月分区(不是落进某个兜底分区)。"""
        env = production_template
        rows = await _fetch(
            env.admin_dsn,
            "SELECT tableoid::regclass::text AS part FROM llm_calls WHERE call_id = $1",
            env.seeded[0],
        )
        assert [r["part"].split(".")[-1] for r in rows] == [env.partition]

    async def test_trigger_blocks_delete_while_drop_partition_still_works(
        self, production_template
    ):
        """兜底触发器拦得住 DELETE(连超级用户也拦),却拦不住 DROP PARTITION。

        这正是 README 说"清理只能走 DROP PARTITION 而不是 DELETE"的机械化依据:
        既要对应用角色 REVOKE DELETE、又要能清理过期数据,分区是唯一不冲突的解。
        """
        import asyncpg

        env = production_template
        conn = await asyncpg.connect(env.admin_dsn, timeout=10)
        try:
            with pytest.raises(asyncpg.exceptions.RaiseError) as exc:
                await conn.execute("DELETE FROM llm_calls WHERE call_id = $1", env.seeded[0])
            assert "不可变审计表" in str(exc.value)
            await conn.execute(f"ALTER TABLE llm_calls DETACH PARTITION {env.partition}")
            await conn.execute(f"DROP TABLE {env.partition}")
            assert await conn.fetchval("SELECT count(*) FROM llm_calls") == 0
        finally:
            await conn.close()