iomgaa/PolyGateway
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 17 Wiki Activity

feat: record stream end mode in telemetry

Browse Source 
llm_calls gains a nullable stream_end TEXT column (auto backfill,
manual-mode trimming as in 1.3.8). TelemetryRecorder.record_llm_call
takes a new required keyword stream_end; the emitter reads it from the
response on attempt rows and writes NULL for failed, cache-hit and
terminal rows.

Red: 18 tests in tests/unit/test_telemetry.py and test_ports.py failed
against the pre-change source (missing column, 38-field signature,
stream_end absent from emitted rows).
This commit is contained in:
iomgaa
2026-10-14 23:54:11 -04:00
parent 48ba36dd94
commit 779a9b29a3
7 changed files with 152 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_ports.py
+3 -1
View File
@@ -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):
tests/unit/test_telemetry.py
+109 -13
View File
@@ -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]