diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index 380e223..3e5055b 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -28,7 +28,7 @@ import pytest from dotenv import dotenv_values from polygateway.telemetry.postgres import PostgresRecorder -from polygateway.telemetry.schema import COLUMNS, telemetry_schema_sql +from polygateway.telemetry.schema import COLUMNS, insert_sql, telemetry_schema_sql _EXPECTED_COLUMNS = [ "call_id", @@ -58,8 +58,31 @@ _EXPECTED_COLUMNS = [ "meta", "thinking_observation", "reasoning_effort", + # —— 1.3.5 逻辑调用统计与结构化失败诊断的十列(issue #19/#23)—— + "scope", + "operation", + "logical_call_id", + "event_kind", + "http_status_code", + "error_type", + "cause_type", + "error_body", + "attempts", + "total_latency_ms", ] +# 1.3.5 之前那张表的 27 个物理列(26 个 INSERT 字段 + created_at)。写成固定切片 +# 而非 `[:-10]`: 后者会随下一次补列静默漂移到另一张表上,而漂移的表现是 +# "旧表补列"用例悄悄改测了别的形态。 +_PRE_135_COLUMNS = _EXPECTED_COLUMNS[:27] + +# 1.3.4 版本的 recorder 实际写入的列(物理列去掉库从不显式写的 created_at) +_PRE_135_WRITTEN_COLUMNS = [c for c in _PRE_135_COLUMNS if c != "created_at"] + +# 1.3.5 新增的十列,按 `COLUMNS`(即 INSERT 字段序)排列: manual 档告警逐字比对与 +# "旧行新列为 NULL"两处共用同一份,免得两边各抄一份后各自漂移。 +_CALL_OBSERVABILITY_COLUMNS = [c for c in COLUMNS if c not in _PRE_135_WRITTEN_COLUMNS] + def _dsn() -> str | None: """读 `.env` 的 DSN 并剥掉 SQLAlchemy 风格的 `+driver` 后缀;未配置返回 None。""" @@ -92,13 +115,13 @@ async def template_admin_dsn() -> str: return value -async def _record_minimal( - recorder: PostgresRecorder, call_id: str | None = None, **overrides -) -> dict[str, object]: - """记一行最小遥测,并**返回实际提交的字段**供调用方逐列比对回读结果。 +def _minimal_fields(call_id: str | None = None, **overrides) -> dict[str, object]: + """一行最小遥测的**完整字段字典**(不写库),供 recorder 写入与旧版本进程模拟共用。 - 返回值不是顺手加的: 逐列断言若在测试里另抄一份期望值,抄错的那一列会以 - "库写错列位"的形态误报,而漏抄的列则悄悄不被验证。 + 独立出来不是顺手拆的: "新旧进程混写"那条用例要以 1.3.4 的列集直接发 INSERT, + 若它另抄一份取值,抄错的那一列会以"库写错列位"的形态误报。 + + 调用方随后逐列比对回读结果,故返回的就是实际提交的那一份。 """ fields: dict[str, object] = { "call_id": call_id if call_id is not None else "c1", @@ -130,8 +153,30 @@ async def _record_minimal( "thinking_observation": "unknown", # 同理: `Effort` 归一成裸 str,不表态则是 None(与 'low' 必须分得开) "reasoning_effort": None, + # —— 1.3.5 十列: 默认形态即"一次普通尝试行"(与单测 `_record_minimal` 同款)—— + "scope": "LLM", + "operation": "chat", + # 库内现场构造的请求没有上下文 → NULL,不造 ID + "logical_call_id": None, + "event_kind": "attempt", + # 诊断四列只在失败的 attempt 行上非空;成功行不统一填 200 + "http_status_code": None, + "error_type": None, + "cause_type": None, + "error_body": None, + # 逻辑快照两列只属终态行 + "attempts": None, + "total_latency_ms": None, } fields.update(overrides) + return fields + + +async def _record_minimal( + recorder: PostgresRecorder, call_id: str | None = None, **overrides +) -> dict[str, object]: + """记一行最小遥测,并**返回实际提交的字段**供调用方逐列比对回读结果。""" + fields = _minimal_fields(call_id, **overrides) await recorder.record_llm_call(**fields) return fields @@ -167,6 +212,17 @@ async def _fetch(dsn: str, sql: str, *args): await conn.close() +async def _execute_args(dsn: str, sql: str, *args) -> None: + """带参数执行单条语句(扩展协议);用于模拟旧版本进程按旧列集发出的 INSERT。""" + import asyncpg + + conn = await asyncpg.connect(dsn, timeout=10) + try: + await conn.execute(sql, *args) + finally: + await conn.close() + + async def _execute_script(dsn: str, sql: str) -> None: """整段执行多语句脚本(不带参数,走简单查询协议)——模拟下游把脚本贴进 psql。""" import asyncpg @@ -588,14 +644,24 @@ _PRE_TENANT_INSERT = ( ) -# `_PRE_TENANT_DDL` 的物理列(23 个): 由 `_EXPECTED_COLUMNS` 去掉此后新增的四列 +# `_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", "reasoning_effort") -] +# 去掉后的顺序与 DDL 逐字一致(这些列在 DDL 里本就排在末尾)。 +_PRE_TENANT_ABSENT = ( + "tenant_id", + "meta", + "thinking_observation", + "reasoning_effort", + *_CALL_OBSERVABILITY_COLUMNS, +) + +_PRE_TENANT_COLUMNS = [c for c in _EXPECTED_COLUMNS if c not in _PRE_TENANT_ABSENT] + +# 这张表缺的 14 个维度,按 `COLUMNS`(即告警的排列序)列出: manual 档告警逐字比对用。 +# 逐字而非前缀断言,是为了让"将来漏进告警的新列"当场红(设计 §4.2 的纪律)。 +_PRE_TENANT_MISSING_NOTICE = ( + f"以下维度不会被记录: {', '.join(c for c in COLUMNS if c in _PRE_TENANT_ABSENT)}。" +) # 回读要逐列比对的字段: 物理列去掉库从不显式写的 created_at,恰好 22 个 _PRE_TENANT_WRITTEN_COLUMNS = [c for c in _PRE_TENANT_COLUMNS if c != "created_at"] @@ -703,7 +769,7 @@ class TestCallerDimensionsAcceptance: "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position", schema, ) - # 22 → 26 个 recorder 字段(加 created_at 共 27 个物理列),且新列追加在末尾 + # 22 → 36 个 recorder 字段(加 created_at 共 37 个物理列),且新列追加在末尾 assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS rows = await _fetch( schema_dsn, @@ -908,11 +974,8 @@ class TestManualSchemaModeAcceptance: 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, thinking_observation, reasoning_effort。" - in notices[0] - ) + # 逐字钉住缺的每一个维度: 前缀断言会让将来漏进告警的新列照样绿 + assert _PRE_TENANT_MISSING_NOTICE in notices[0] finally: await recorder.aclose() @@ -938,11 +1001,8 @@ class TestManualSchemaModeAcceptance: 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, thinking_observation, reasoning_effort。" - in notices[0] - ) + # 逐字钉住缺的每一个维度: 前缀断言会让将来漏进告警的新列照样绿 + assert _PRE_TENANT_MISSING_NOTICE in notices[0] # 提示里的 SQL 必须可直接粘贴执行,而不是只报个列名 assert ( "ALTER TABLE llm_calls ADD COLUMN tenant_id TEXT NOT NULL DEFAULT '';" in notices[0] @@ -974,6 +1034,283 @@ class TestManualSchemaModeAcceptance: await recorder.aclose() +# 1.3.5 之前(1.3.4 发布形态)的表: 26 个 recorder 字段 + created_at = 27 个物理列, +# 没有本版新增的任何一列。裸表名: 由 `pg_sandbox` 在沙箱 schema 的 search_path 下执行。 +_PRE_135_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 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, + thinking_observation TEXT, + reasoning_effort TEXT +) +""" + +# 一行 1.3.5 之前写下的历史数据(只列 NOT NULL 列,与当年 recorder 的写入等价)。 +# 工厂的 `extra` 逐条裸执行、不接受查询参数,故 call_id 内联成字面量。 +_PRE_135_INSERT = ( + "INSERT INTO llm_calls (call_id, model, provider, source_name, messages, response, " + "prompt_tokens, completion_tokens, usage_source, latency_ms) " + "VALUES ('pre135', 'm', 'p', 's1', '[]', 'old body', 1, 2, 'measured', 10)" +) + +# 这张表缺的正是本版十列;manual 档告警要逐字比对的那句。 +_PRE_135_MISSING_NOTICE = f"以下维度不会被记录: {', '.join(_CALL_OBSERVABILITY_COLUMNS)}。" + + +@pytest.fixture +async def pre_135_schema(pg_sandbox) -> tuple[str, str]: + """一次性沙箱里造一张 **1.3.4 形态的 27 列表**,并留一行本版之前的历史数据。 + + 共享表 `llm_calls` 一个字节都不碰: 本机那张表升级一次就再也回不到旧形态, + 指望它还是旧形态的测试第二次跑就会空转(与 `pre_tenant_schema` 同款理由)。 + """ + sandbox = await pg_sandbox(ddl=_PRE_135_DDL, extra=(_PRE_135_INSERT,)) + return sandbox.dsn, sandbox.schema + + +class TestCallObservabilityColumnsAcceptance: + """1.3.5(issue #19/#23)的 PG 存储兼容验收: auto 追加 / manual 裁剪 / 旧行 NULL / 混写。 + + 单元层在 SQLite 上断的是同一族语义,这里断的是**真实 PG 上确实如此**—— + 两端的 DDL、补列语句与列序是两份文本(`SQLITE_BACKFILL` 与 `_PG_BACKFILL_DECLS`), + 只有真表能证明它们没有分叉。 + """ + + async def test_pre_135_table_gains_the_ten_columns_and_old_rows_stay_null(self, pre_135_schema): + """27 列旧表 auto 补齐到 37 列,新行两类取值读得回,**历史行十列一律 NULL**。 + + 旧行不回填是本版的明示决策(设计 §5): NULL 表达的是"补列之前根本没记过 + 这件事",与任何哨兵值都不是一回事。若哪天有人给这十列加了 DEFAULT,历史行 + 会被就地改写成"记过且值为 X",归因查询从此分不清真实缺口——故这条断言的 + 方向是 NULL,不是空串也不是 0。 + """ + schema_dsn, schema = pre_135_schema + recorder = _recorder(schema_dsn, auto_migrate=True) + try: + # attempt 行: 诊断四列非空、逻辑快照两列 NULL + await _record_minimal( + recorder, + call_id="att", + scope="LLM", + operation="chat", + logical_call_id="lcid-1", + event_kind="attempt", + http_status_code=503, + error_type="TransientError", + cause_type="ReadTimeout", + error_body="upstream said 503", + ) + # 终态行: 逻辑快照两列非空、诊断三列 NULL(不搬最后一次 attempt 的现场) + await _record_minimal( + recorder, + call_id="term", + scope="LLM", + operation="chat", + logical_call_id="lcid-1", + event_kind="terminal_failure", + error="AllSourcesExhausted: 全部源已耗尽", + error_type="AllSourcesExhausted", + attempts=3, + total_latency_ms=4200, + ) + 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, + ) + # 26 → 36 个 recorder 字段(加 created_at 共 37 个物理列),新列追加在末尾: + # 列序与新建库一致才不会让两条升级路径分叉 + assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS + + names = ", ".join(_CALL_OBSERVABILITY_COLUMNS) + rows = await _fetch( + schema_dsn, + f"SELECT call_id, {names} FROM llm_calls " + "WHERE call_id = ANY($1::text[]) ORDER BY call_id", + ["att", "pre135", "term"], + ) + by_id = {r["call_id"]: r for r in rows} + + assert dict(by_id["att"]) == { + "call_id": "att", + "scope": "LLM", + "operation": "chat", + "logical_call_id": "lcid-1", + "event_kind": "attempt", + "http_status_code": 503, + "error_type": "TransientError", + "cause_type": "ReadTimeout", + "error_body": "upstream said 503", + "attempts": None, + "total_latency_ms": None, + } + assert dict(by_id["term"]) == { + "call_id": "term", + "scope": "LLM", + "operation": "chat", + "logical_call_id": "lcid-1", + "event_kind": "terminal_failure", + "http_status_code": None, + "error_type": "AllSourcesExhausted", + "cause_type": None, + "error_body": None, + "attempts": 3, + "total_latency_ms": 4200, + } + # 历史行: 十列逐列 NULL(整体比对,漏掉任一列都红) + assert dict(by_id["pre135"]) == { + "call_id": "pre135", + **dict.fromkeys(_CALL_OBSERVABILITY_COLUMNS), + } + finally: + await recorder.aclose() + + async def test_manual_trims_the_insert_on_a_pre_135_table( + self, pre_135_schema, captured_warnings + ): + """27 列旧表 + manual: 一条 DDL 都不发,写入按现有列裁剪后照样落库。 + + 与上一条恰成对照: 同一张表、同一份负载,只有 `auto_migrate` 不同,列数就必须是 + 27 与 37 之别。裁剪是关掉 ALTER 的前提——不裁剪的话每行 INSERT 都撞缺列 + (SQLSTATE 42703)而被整行丢弃,那是把自动补列换成遥测静默全失。 + """ + schema_dsn, schema = pre_135_schema + recorder = _recorder(schema_dsn, auto_migrate=False) + try: + recorded = await _record_minimal( + recorder, call_id="man135", scope="LLM", logical_call_id="lcid-x" + ) + 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, + ) + # 表结构逐字不动: 既没多出十列,也没被顺手改了列序 + assert [r["column_name"] for r in cols] == _PRE_135_COLUMNS + + names = ", ".join(_PRE_135_WRITTEN_COLUMNS) + rows = await _fetch( + schema_dsn, f"SELECT {names} FROM llm_calls WHERE call_id = $1", "man135" + ) + assert len(rows) == 1 # 裁剪后的 INSERT 真写进去了,不是被 PG 拒收 + # 其余 26 列逐列与提交值相等: 少写十列最容易引发的错是剩下的值整体错位 + assert dict(rows[0]) == {c: recorded[c] for c in _PRE_135_WRITTEN_COLUMNS} + + 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 _PRE_135_MISSING_NOTICE in notices[0] + # 提示里的 SQL 必须可直接粘贴执行(首列与末列各验一条,含类型) + assert "ALTER TABLE llm_calls ADD COLUMN scope TEXT;" in notices[0] + assert "ALTER TABLE llm_calls ADD COLUMN total_latency_ms INTEGER;" in notices[0] + finally: + await recorder.aclose() + + async def test_old_and_new_writers_share_one_table(self, fresh_schema, captured_warnings): + """滚动升级期的混写: 已补列的表上,旧版本进程按 26 列写、新版本按 36 列写。 + + 这是升级窗口里必然出现的形态(先升一个 worker,其余仍是 1.3.4),而它的失败 + 方式是静默的: 若新列带了 NOT NULL 或旧列集的 INSERT 被拒,旧 worker 的遥测 + 会整段消失而只留逐行 warning。故这里既断三行都在、也断没有写入失败 warning。 + + 旧进程用 `insert_sql("postgres", 旧列集)` 而不是另抄一条 SQL: 1.3.4 的 + recorder 发出的正是同一函数按当年列集拼出的语句,另抄一份只会各自漂移。 + """ + fresh_dsn, schema = fresh_schema + recorder = _recorder(fresh_dsn, auto_migrate=True) + try: + # 新版本进程: 建表(37 列)并写一条带完整新列的终态行 + await _record_minimal( + recorder, + call_id="new-1", + scope="LLM", + operation="chat", + logical_call_id="lcid-mix", + event_kind="terminal_failure", + error_type="AllSourcesExhausted", + attempts=2, + total_latency_ms=1500, + ) + + # 旧版本进程: 同一张表,按 1.3.4 的 26 列集写入 + legacy_fields = _minimal_fields("old-1", response="from a 1.3.4 worker") + await _execute_args( + fresh_dsn, + insert_sql("postgres", _PRE_135_WRITTEN_COLUMNS), + *(legacy_fields[c] for c in _PRE_135_WRITTEN_COLUMNS), + ) + + # 新版本进程继续写: 旧进程的写入不得污染后续(列集是每进程各自探测的) + await _record_minimal( + recorder, call_id="new-2", scope="LLM", operation="embed", event_kind="attempt" + ) + + 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 call_id, response, scope, operation, event_kind, attempts, " + "total_latency_ms FROM llm_calls ORDER BY call_id", + ) + assert [r["call_id"] for r in rows] == ["new-1", "new-2", "old-1"] + by_id = {r["call_id"]: r for r in rows} + assert by_id["old-1"]["response"] == "from a 1.3.4 worker" + # 旧进程写下的行,新列一律 NULL——它没写,也不该被谁替它填 + assert (by_id["old-1"]["scope"], by_id["old-1"]["event_kind"]) == (None, None) + assert (by_id["old-1"]["attempts"], by_id["old-1"]["total_latency_ms"]) == (None, None) + assert (by_id["new-1"]["attempts"], by_id["new-1"]["total_latency_ms"]) == (2, 1500) + assert by_id["new-2"]["operation"] == "embed" + assert [m for m in captured_warnings if "写入失败" in m] == [] + + # 下游可见变化的机械化依据: 新口径"计失败调用"按 event_kind 过滤, + # 混写期旧行(event_kind 为 NULL)既不会被误计成失败,也不会被误计成成功 + terminal = await _fetch( + fresh_dsn, + "SELECT count(*) AS n FROM llm_calls WHERE event_kind = 'terminal_failure'", + ) + assert terminal[0]["n"] == 1 + unclassified = await _fetch( + fresh_dsn, "SELECT count(*) AS n FROM llm_calls WHERE event_kind IS NULL" + ) + assert unclassified[0]["n"] == 1 + 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" @@ -1000,7 +1337,7 @@ class TestPublishedSchemaScript: await _execute_script(fresh_dsn, script) actual = [r["column_name"] for r in await _fetch(fresh_dsn, _PHYSICAL_COLUMNS_SQL, schema)] - # 物理列 = 26 个 INSERT 字段 + 库从不显式写的 created_at;对着库常量比,不另抄一份 + # 物理列 = 36 个 INSERT 字段 + 库从不显式写的 created_at;对着库常量比,不另抄一份 assert set(actual) == set(COLUMNS) | {"created_at"} # 列序也不许漂: 新列必须排在 created_at 之后,否则新建库与 ALTER 升级的列序分叉 assert actual == _EXPECTED_COLUMNS