diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py index ec730f0..fb49c18 100644 --- a/src/polygateway/middleware/telemetry.py +++ b/src/polygateway/middleware/telemetry.py @@ -221,6 +221,9 @@ class _AttemptUsage: # 两字段留在此处只为让"尝试视图"形状完整,默认 None 即"无沉默观测" stall_ms: int | None = None stall_count: int | None = None + # 1.3.10(#31): 与 stall 相反,结束方式**从 response 取**——LLMResponse 本次有 + # 该字段,免去约 50 处 emit_attempt 调用点加参;失败尝试 None = 无流可言 + stream_end: str | None = None @classmethod def of(cls, response: LLMResponse | None) -> _AttemptUsage: @@ -239,6 +242,7 @@ class _AttemptUsage: model_reported=response.model_reported, reasoning_tokens=response.reasoning_tokens, thinking_observation=response.thinking_observation, + stream_end=response.stream_end, ) @@ -370,7 +374,7 @@ def _assert_recorder_shape(recorder: TelemetryRecorder) -> None: class TelemetryEmitter: - """从请求与结果组装 38 字段并写入 recorder;一切写失败降级 warning。""" + """从请求与结果组装 39 字段并写入 recorder;一切写失败降级 warning。""" def __init__( self, @@ -464,6 +468,7 @@ class TelemetryEmitter: total_latency_ms=None, stall_ms=stall_ms, stall_count=stall_count, + stream_end=usage.stream_end, ) async def emit_cache_hit( @@ -514,6 +519,8 @@ class TelemetryEmitter: # 缓存命中行未产生 transport 观测: 无沉默可言,显式 None stall_ms=None, stall_count=None, + # 命中/终态行无流可言(缓存命中已在 CacheMW 置 None,此处显式不信任) + stream_end=None, ) async def emit_terminal_failure( @@ -569,6 +576,8 @@ class TelemetryEmitter: # 终态行无具体源与 transport 观测: 显式 None stall_ms=None, stall_count=None, + # 命中/终态行无流可言(缓存命中已在 CacheMW 置 None,此处显式不信任) + stream_end=None, ) async def _record( @@ -615,6 +624,8 @@ class TelemetryEmitter: # 本方法只搬运(同 reasoning_effort 先例)—— stall_ms: int | None, stall_count: int | None, + # —— 1.3.10: 结束方式(#31);入口已定型,本方法只搬运 —— + stream_end: str | None, ) -> None: try: # 成本换算(M2 §6): 成功行按单价换算;缓存命中 0.0(未产生新调用); @@ -685,6 +696,7 @@ class TelemetryEmitter: total_latency_ms=total_latency_ms, stall_ms=stall_ms, stall_count=stall_count, + stream_end=stream_end, ) except asyncio.CancelledError: raise diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py index 1eae955..b8b9610 100644 --- a/src/polygateway/ports.py +++ b/src/polygateway/ports.py @@ -290,7 +290,7 @@ class TelemetryStatusProvider(Protocol): @runtime_checkable class TelemetryRecorder(Protocol): - """遥测后端;38 字段冻结(M1 设计 §4.4 + issue #3/#4/#11/#16/#20 + 1.3.5/1.3.8),唯一调用点是 TelemetryEmitter。 + """遥测后端;39 字段冻结(M1 设计 §4.4 + issue #3/#4/#11/#16/#20 + 1.3.5/1.3.8/1.3.10),唯一调用点是 TelemetryEmitter。 新增参数不设默认值: 库外无第三方实现者(三项目迁移时删除了各自的同名 Protocol),完整签名的成本为零,而少写一列会被 emitter 的降级吞成 warning。 @@ -357,4 +357,5 @@ class TelemetryRecorder(Protocol): total_latency_ms: int | None, stall_ms: int | None, stall_count: int | None, + stream_end: str | None, ) -> None: ... diff --git a/src/polygateway/telemetry/schema.py b/src/polygateway/telemetry/schema.py index 9ed9615..47ec510 100644 --- a/src/polygateway/telemetry/schema.py +++ b/src/polygateway/telemetry/schema.py @@ -5,8 +5,8 @@ 多处各存一份必然漂移,而漂移的表现是"下游照打印的 SQL 建完表,库仍报缺列"。 **`COLUMNS` 是 INSERT 字段序,不是物理列序**: 数据库自填的 `created_at` 不在其中(它带 -`DEFAULT now()` / `datetime('now')`,库从不显式写它)。物理表列 = 38 个 INSERT 字段 + -`created_at` = 39;列数断言一律按物理列数写,两套口径混用是最易错处。 +`DEFAULT now()` / `datetime('now')`,库从不显式写它)。物理表列 = 39 个 INSERT 字段 + +`created_at` = 40;列数断言一律按物理列数写,两套口径混用是最易错处。 本模块只依赖标准库: `telemetry/` 与 `backends/`、`transports/`、`structured/` 同层且 互不依赖(import-linter 契约执法)。 @@ -64,7 +64,8 @@ CREATE TABLE IF NOT EXISTS llm_calls ( attempts INTEGER, total_latency_ms INTEGER, stall_ms INTEGER, - stall_count INTEGER + stall_count INTEGER, + stream_end TEXT ); """ @@ -108,7 +109,8 @@ CREATE TABLE IF NOT EXISTS llm_calls ( attempts INTEGER, total_latency_ms INTEGER, stall_ms INTEGER, - stall_count INTEGER + stall_count INTEGER, + stream_end TEXT ); """ @@ -145,6 +147,8 @@ SQLITE_BACKFILL = ( # 没记过这件事",与"观测过且零沉默"(0)是两回事 ("stall_ms", "INTEGER"), ("stall_count", "INTEGER"), + # 1.3.10 结束方式(#31): 可空无默认,旧行 NULL = 补列之前没记过;值域见 StreamEnd + ("stream_end", "TEXT"), ) # PG 补列的列定义。语句由此派生成两份文本(见下),使"库内执行的那份"与"打印给 @@ -174,6 +178,8 @@ _PG_BACKFILL_DECLS = ( # 1.3.8 两列,列序与 SQLITE_BACKFILL 逐项对齐(理由同上) ("stall_ms", "INTEGER"), ("stall_count", "INTEGER"), + # 1.3.10 结束方式(#31): 可空无默认,旧行 NULL = 补列之前没记过;值域见 StreamEnd + ("stream_end", "TEXT"), ) # 新列排在 created_at 之后: 与旧表 ALTER 追加的位置一致(见 SQLITE_BACKFILL 同款注释)。 @@ -226,6 +232,8 @@ COLUMNS = ( # —— 1.3.8 沉默归因(issue #27)—— "stall_ms", "stall_count", + # —— 1.3.10 结束方式(#31)—— + "stream_end", ) _COLUMN_SET = frozenset(COLUMNS) diff --git a/src/polygateway/telemetry/sqlite.py b/src/polygateway/telemetry/sqlite.py index 05f8107..309fee4 100644 --- a/src/polygateway/telemetry/sqlite.py +++ b/src/polygateway/telemetry/sqlite.py @@ -143,7 +143,7 @@ class SQLiteRecorder: logger.warning("SQLite 遥测补列失败(写入将逐行降级): {}", exc) async def record_llm_call(self, **fields: object) -> None: - """写一行遥测;字段集合即 38 字段冻结签名(ports.TelemetryRecorder)。 + """写一行遥测;字段集合即 39 字段冻结签名(ports.TelemetryRecorder)。 取值按 `self._columns`(manual 档可能已被裁剪),与 `self._insert` 的 占位符同序——两者必须一起改,分开改就是把值写进错位的列。 diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index 3ce3b8c..3a17582 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -72,6 +72,8 @@ _EXPECTED_COLUMNS = [ # —— 1.3.8 沉默归因两列(issue #27)—— "stall_ms", "stall_count", + # —— 1.3.10 结束方式(#31)—— + "stream_end", ] # 1.3.5 之前那张表的 27 个物理列(26 个 INSERT 字段 + created_at)。写成固定切片 @@ -173,6 +175,8 @@ def _minimal_fields(call_id: str | None = None, **overrides) -> dict[str, object # —— 1.3.8 两列: 沉默归因只属流式 chat 成功 attempt 行(issue #27),普通行 NULL —— "stall_ms": None, "stall_count": None, + # —— 1.3.10: 结束方式只属流式成功 attempt 行,普通行 NULL —— + "stream_end": None, } fields.update(overrides) return fields @@ -775,7 +779,7 @@ class TestCallerDimensionsAcceptance: "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position", schema, ) - # 22 → 38 个 recorder 字段(加 created_at 共 39 个物理列),且新列追加在末尾 + # 22 → 39 个 recorder 字段(加 created_at 共 40 个物理列),且新列追加在末尾 assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS rows = await _fetch( schema_dsn, @@ -1106,7 +1110,7 @@ class TestCallObservabilityColumnsAcceptance: """ async def test_pre_135_table_gains_the_ten_columns_and_old_rows_stay_null(self, pre_135_schema): - """27 列旧表 auto 补齐到 39 列(1.3.5 十列 + 1.3.8 两列),新行两类取值读得回,**历史行新增列一律 NULL**。 + """27 列旧表 auto 补齐到 40 列(1.3.5 十列 + 1.3.8 两列 + 1.3.10 一列),新行两类取值读得回,**历史行新增列一律 NULL**。 旧行不回填是本版的明示决策(设计 §5): NULL 表达的是"补列之前根本没记过 这件事",与任何哨兵值都不是一回事。若哪天有人给这十列加了 DEFAULT,历史行 @@ -1148,7 +1152,7 @@ class TestCallObservabilityColumnsAcceptance: "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position", schema, ) - # 26 → 38 个 recorder 字段(加 created_at 共 39 个物理列),新列追加在末尾: + # 26 → 39 个 recorder 字段(加 created_at 共 40 个物理列),新列追加在末尾: # 列序与新建库一致才不会让两条升级路径分叉 assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS @@ -1176,6 +1180,8 @@ class TestCallObservabilityColumnsAcceptance: # 1.3.8 两列: 普通成功 attempt 行无沉默观测 → NULL "stall_ms": None, "stall_count": None, + # 1.3.10: 该行未传结束方式 → NULL + "stream_end": None, } assert dict(by_id["term"]) == { "call_id": "term", @@ -1192,6 +1198,8 @@ class TestCallObservabilityColumnsAcceptance: # 1.3.8 两列: 终态行无 transport 观测 → NULL "stall_ms": None, "stall_count": None, + # 1.3.10: 终态行无流可言 → NULL + "stream_end": None, } # 历史行: 十列逐列 NULL(整体比对,漏掉任一列都红) assert dict(by_id["pre135"]) == { @@ -1349,7 +1357,7 @@ class TestPublishedSchemaScript: await _execute_script(fresh_dsn, script) actual = [r["column_name"] for r in await _fetch(fresh_dsn, _PHYSICAL_COLUMNS_SQL, schema)] - # 物理列 = 38 个 INSERT 字段 + 库从不显式写的 created_at;对着库常量比,不另抄一份 + # 物理列 = 39 个 INSERT 字段 + 库从不显式写的 created_at;对着库常量比,不另抄一份 assert set(actual) == set(COLUMNS) | {"created_at"} # 列序也不许漂: 新列必须排在 created_at 之后,否则新建库与 ALTER 升级的列序分叉 assert actual == _EXPECTED_COLUMNS diff --git a/tests/unit/test_ports.py b/tests/unit/test_ports.py index be90f5e..7e352b5 100644 --- a/tests/unit/test_ports.py +++ b/tests/unit/test_ports.py @@ -150,6 +150,7 @@ class _DummyRecorder: total_latency_ms, stall_ms, stall_count, + stream_end, ) -> None: ... @@ -322,7 +323,7 @@ class TestTelemetryRecorderSignature: "attempts", "total_latency_ms", } <= set(params) - assert len(params) - 1 == 38 # 减掉 self + assert len(params) - 1 == 39 # 减掉 self @pytest.mark.parametrize( "name", @@ -341,6 +342,7 @@ class TestTelemetryRecorderSignature: "error_body", "attempts", "total_latency_ms", + "stream_end", ], ) def test_caller_dimensions_have_no_default(self, name): diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index cd87404..8778747 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -92,6 +92,8 @@ _EXPECTED_COLUMNS = [ # —— 1.3.8 沉默归因两列(issue #27)—— "stall_ms", "stall_count", + # —— 1.3.10 结束方式(#31)—— + "stream_end", ] # 1.3.5 之前的 26 个 INSERT 字段(旧表形态),供冻结 INSERT 语句与旧表补列用例复用 @@ -187,6 +189,8 @@ async def _record_minimal(recorder, call_id="c1", **overrides): # —— 1.3.8 沉默归因两列: 默认 None = 非流式/无沉默观测(合法事实)—— "stall_ms": None, "stall_count": None, + # —— 1.3.10 结束方式: 默认 None = 非流式/缓存命中/失败行 —— + "stream_end": None, } fields.update(overrides) await recorder.record_llm_call(**fields) @@ -268,7 +272,7 @@ class TestSchemaModule: # COLUMNS 是 INSERT 字段序,不含数据库自填的 created_at assert list(COLUMNS) == [c for c in _EXPECTED_COLUMNS if c != "created_at"] - assert len(COLUMNS) == 38 + assert len(COLUMNS) == 39 # 两端 DDL 的列出现顺序 == 物理列序(created_at 在第 19 位) for ddl in (SQLITE_DDL, PG_DDL): assert _first_occurrence_order(ddl, _EXPECTED_COLUMNS) == _EXPECTED_COLUMNS @@ -282,8 +286,8 @@ class TestSchemaModule: "ALTER TABLE llm_calls ADD COLUMN cached_prompt_tokens INTEGER", ) assert PG_BACKFILL[-1] == ( - "stall_count", - "ALTER TABLE llm_calls ADD COLUMN stall_count INTEGER", + "stream_end", + "ALTER TABLE llm_calls ADD COLUMN stream_end TEXT", ) assert all("IF NOT EXISTS" not in stmt for _, stmt in PG_BACKFILL) @@ -297,10 +301,10 @@ class TestSchemaModule: assert insert_sql("sqlite", _PRE_135_COLUMNS) == _FROZEN_SQLITE_INSERT assert insert_sql("postgres", _PRE_135_COLUMNS) == _FROZEN_PG_INSERT - # 全量 38 列: 占位符随列数增长,且不留空洞 - assert insert_sql("postgres", COLUMNS).count("$") == 38 - assert "$38)" in insert_sql("postgres", COLUMNS) - assert insert_sql("sqlite", COLUMNS).count("?") == 38 + # 全量 39 列: 占位符随列数增长,且不留空洞 + assert insert_sql("postgres", COLUMNS).count("$") == 39 + assert "$39)" in insert_sql("postgres", COLUMNS) + assert insert_sql("sqlite", COLUMNS).count("?") == 39 # 裁剪列表按位置占位符重新编号,不留空洞 assert insert_sql("postgres", ["call_id", "model"]) == ( "INSERT INTO llm_calls (call_id, model) VALUES ($1, $2) ON CONFLICT DO NOTHING" @@ -377,13 +381,14 @@ class TestBackendColumnParity: """新列只能追加在末尾: 旧表经 ALTER 补列必落末尾,插在中间会让两条路径分叉。""" from polygateway.telemetry.schema import COLUMNS - assert COLUMNS[-6:] == ( + assert COLUMNS[-7:] == ( "cause_type", "error_body", "attempts", "total_latency_ms", "stall_ms", "stall_count", + "stream_end", ) @@ -776,7 +781,7 @@ class TestSQLiteCallerDimensionsAcceptance: conn = sqlite3.connect(db) cols = [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")] - assert cols == _EXPECTED_COLUMNS # 22 → 38 个 recorder 字段(+ created_at 共 39 物理列) + assert cols == _EXPECTED_COLUMNS # 22 → 39 个 recorder 字段(+ created_at 共 40 物理列) rows = dict(conn.execute("SELECT call_id, tenant_id FROM llm_calls").fetchall()) assert rows["new-row"] == "tenant-a" assert rows["old-row"] == "" # 不是 None: NULL 会被 RLS 静默吞掉 @@ -826,7 +831,7 @@ class TestSQLiteSchemaMode: """issue #13: `auto_migrate` 两档——auto 保持自动补列,manual 只裁剪写入不发 DDL。 列数断言一律按**物理列数**写: 旧表 22 个 INSERT 字段 + `created_at` = 23, - 补齐后 38 + `created_at` = 39。混用 INSERT 字段数与物理列数是本处最易错的地方。 + 补齐后 39 + `created_at` = 40。混用 INSERT 字段数与物理列数是本处最易错的地方。 """ def _physical_columns(self, db: Path) -> list[str]: @@ -865,7 +870,7 @@ class TestSQLiteSchemaMode: assert "ALTER TABLE" in message # 给出可直接执行的补列 SQL async def test_auto_mode_still_upgrades_the_legacy_table(self, tmp_path): - """auto + 同款旧表: 现状回归,补列后物理列数 23 → 39。""" + """auto + 同款旧表: 现状回归,补列后物理列数 23 → 40。""" db = tmp_path / "auto_legacy.db" _make_pre_tenant_db(db) @@ -874,10 +879,10 @@ class TestSQLiteSchemaMode: recorder.close() assert self._physical_columns(db) == _EXPECTED_COLUMNS - assert len(self._physical_columns(db)) == 39 + assert len(self._physical_columns(db)) == 40 async def test_manual_mode_still_creates_a_fresh_table(self, tmp_path): - """manual 只管 ALTER,不管 CREATE: 全新库照建,39 个物理列齐全(设计 §4.2)。""" + """manual 只管 ALTER,不管 CREATE: 全新库照建,40 个物理列齐全(设计 §4.2)。""" db = tmp_path / "manual_fresh.db" recorder = SQLiteRecorder(db, auto_migrate=False) await _record_minimal(recorder, call_id="c-fresh", tenant_id="tenant-a") @@ -1117,6 +1122,8 @@ class TestPostgresBackfillDiscipline: # 1.3.8 两列: 稳态定义随补列前移,否则每进程首写都去抢锁 "stall_ms", "stall_count", + # 1.3.10 一列,理由同上 + "stream_end", ] def _recorder(self, conn): @@ -3586,3 +3593,92 @@ class TestTerminalRowSqlSemantics: ).fetchone()[0] == 0 ) + + +# 1.3.8 形态(1.3.10 补列前)的表: 38 个 INSERT 字段 + created_at = 39 物理列。 +# 冻结字面量(由 _PRE_138_DDL 末行补两列),不从常量切片派生,理由同 _PRE_138_DDL。 +_PRE_1310_DDL = _PRE_138_DDL.replace( + " total_latency_ms INTEGER\n);", + " total_latency_ms INTEGER,\n stall_ms INTEGER,\n" + " stall_count INTEGER\n);", +) + + +class TestStreamEndColumn: + """1.3.10 结束方式列(#31): 落行、缺省 NULL、补列幂等、manual 告警、各入口取值。""" + + async def test_value_round_trips_and_default_is_null(self, tmp_path): + db = tmp_path / "stream_end.db" + recorder = SQLiteRecorder(db, auto_migrate=True) + await _record_minimal(recorder, call_id="c-eof", stream_end="eof_after_finish") + await _record_minimal(recorder, call_id="c-none") + recorder.close() + conn = sqlite3.connect(db) + rows = dict(conn.execute("SELECT call_id, stream_end FROM llm_calls").fetchall()) + conn.close() + assert rows == {"c-eof": "eof_after_finish", "c-none": None} + + async def test_auto_migrate_backfills_from_138_table(self, tmp_path): + db = tmp_path / "pre1310.db" + conn = sqlite3.connect(db) + conn.execute(_PRE_1310_DDL) + conn.commit() + conn.close() + for call_id in ("c1", "c2"): # 二次初始化幂等 + recorder = SQLiteRecorder(db, auto_migrate=True) + await _record_minimal(recorder, call_id=call_id, stream_end="done") + recorder.close() + conn = sqlite3.connect(db) + assert [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")] == _EXPECTED_COLUMNS + conn.close() + + async def test_manual_mode_names_the_new_alter(self, tmp_path, captured_warnings): + db = tmp_path / "manual1310.db" + conn = sqlite3.connect(db) + conn.execute(_PRE_1310_DDL) + conn.commit() + conn.close() + recorder = SQLiteRecorder(db, auto_migrate=False) + await _record_minimal(recorder, call_id="c-manual", stream_end="done") + recorder.close() + conn = sqlite3.connect(db) + assert len([r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")]) == 39 + assert conn.execute("SELECT call_id FROM llm_calls").fetchone() == ("c-manual",) + conn.close() + assert "ALTER TABLE llm_calls ADD COLUMN stream_end TEXT" in captured_warnings[0] + + async def test_attempt_takes_stream_end_from_response(self): + rec = _MemoryRecorder() + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( + request=_REQ, + source=_source(), + call_id="c", + latency_ms=1, + response=_resp(stream_end="eof_after_finish"), + error=None, + reasoning_applies=True, + operation="chat", + ) + assert rec.rows[0]["stream_end"] == "eof_after_finish" + + async def test_failed_hit_and_terminal_rows_are_null(self): + rec = _MemoryRecorder() + emitter = TelemetryEmitter(rec, text_cap=None, scope="LLM") + await emitter.emit_attempt( + request=_REQ, + source=_source(), + call_id="c", + latency_ms=1, + response=None, + error="boom", + reasoning_applies=True, + operation="chat", + ) + # 命中行显式不信任 response 上的值(即使上游误带) + await emitter.emit_cache_hit( + request=_REQ, response=_resp(cache_hit=True, stream_end="done"), operation="chat" + ) + await emitter.emit_terminal_failure( + request=_REQ, call_id="c", error="dead", operation="chat", stats=_MIGRATED_STATS + ) + assert [row["stream_end"] for row in rec.rows] == [None, None, None]