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
from datetime import UTC, datetime, timedelta
from uuid import uuid4

import pytest
from dotenv import dotenv_values

from polygateway.telemetry.postgres import PostgresRecorder

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

# 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": "{}",
    }
    fields.update(overrides)
    await recorder.record_llm_call(**fields)
    return fields


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()


_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 = PostgresRecorder(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 = PostgresRecorder(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 = PostgresRecorder(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 = PostgresRecorder(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 = PostgresRecorder(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 TestDegradation:
    async def test_unreachable_server_degrades_silently(self):
        """结构性失败(建池不通)→ warning 一次后永久降级,业务零感知。"""
        recorder = PostgresRecorder("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_row_failure_does_not_poison_later_rows(self, dsn):
        """运行时单条写失败(NUL 字节文本被 PG 拒)→ 丢该行,后续行照常落库。"""
        recorder = PostgresRecorder(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 = PostgresRecorder(dsn, auto_migrate=True)
        await _record_minimal(recorder)
        await recorder.aclose()
        await recorder.aclose()


_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):
        """修复前: 建表被拒 → _failed → 整个进程一条不落(下游 150 次调用全丢)。"""
        low_dsn, schema = least_privilege_dsn
        recorder = PostgresRecorder(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._failed 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` 去掉 issue #11 的两个新维度
# 派生而非另抄一份——两份常量必然漂移,而漂移的表现是"manual 档没补列"这条断言假绿。
# 去掉后的顺序与 DDL 逐字一致(tenant_id/meta 在 DDL 里本就排在末尾)。
_PRE_TENANT_COLUMNS = [c for c in _EXPECTED_COLUMNS if c not in ("tenant_id", "meta")]

# 回读要逐列比对的字段: 物理列去掉库从不显式写的 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 = PostgresRecorder(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 = PostgresRecorder(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 → 24 个 recorder 字段(加 created_at 共 25 个物理列),且新列追加在末尾
            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、不置 `_failed`、后续 INSERT 仍照发。

        置 `_failed` 会让整个进程从此一条遥测都不写(比逐行丢弃严重得多),
        且一旦 DBA 补上列也不会自愈——必须等重启。
        """
        recorder = PostgresRecorder(least_privilege_pre_tenant_dsn, auto_migrate=True)
        try:
            await _record_minimal(recorder, call_id=_cid("lpp1"))  # 不得抛
            assert recorder._failed 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 = PostgresRecorder(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 = PostgresRecorder(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 = PostgresRecorder(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 = PostgresRecorder(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._failed 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()