feat: record the reasoning verdict in telemetry
This issue surfaced only because someone ran a slow suite that is excluded by default and had not been run for eighteen days. As a column it becomes a query: which model stopped being observable, and when. The emitter unwraps the enum to a plain str at the single _record exit. asyncpg makes no promise about encoding a str subclass, and a telemetry write that fails is downgraded to one warning — it would not crash, it would just quietly cost the Postgres path a column. Normalising at the emitter follows what tenant_id, meta and sampling already do. The column is appended last in COLUMNS and in both DDLs. An existing table can only take ALTER at the end, so putting it anywhere else forks the physical column order between a freshly built database and a backfilled one.
This commit is contained in:
@@ -244,7 +244,7 @@ class TestTelemetryRecorderSignature:
|
||||
params = inspect.signature(TelemetryRecorder.record_llm_call).parameters
|
||||
assert {"tenant_id", "meta"} <= set(params)
|
||||
|
||||
@pytest.mark.parametrize("name", ["tenant_id", "meta"])
|
||||
@pytest.mark.parametrize("name", ["tenant_id", "meta", "thinking_observation"])
|
||||
def test_caller_dimensions_have_no_default(self, name):
|
||||
import inspect
|
||||
|
||||
|
||||
+100
-16
@@ -30,6 +30,7 @@ from polygateway.types import (
|
||||
OcrTextTransportResult,
|
||||
RetryPolicy,
|
||||
SourceConfig,
|
||||
ThinkingObservation,
|
||||
)
|
||||
|
||||
_REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}], session_id="sess-1")
|
||||
@@ -60,6 +61,7 @@ _EXPECTED_COLUMNS = [
|
||||
"reasoning_tokens",
|
||||
"tenant_id",
|
||||
"meta",
|
||||
"thinking_observation",
|
||||
]
|
||||
|
||||
|
||||
@@ -127,6 +129,8 @@ async def _record_minimal(recorder, call_id="c1", **overrides):
|
||||
# 到达 recorder 时已由 emitter 归一化: None → '',空 dict → '{}'
|
||||
"tenant_id": "",
|
||||
"meta": "{}",
|
||||
# 同样已由 emitter 归一化: 枚举取 .value 后才下沉,recorder 只见裸 str
|
||||
"thinking_observation": "unknown",
|
||||
}
|
||||
fields.update(overrides)
|
||||
await recorder.record_llm_call(**fields)
|
||||
@@ -172,16 +176,16 @@ _FROZEN_SQLITE_INSERT = (
|
||||
"INSERT OR IGNORE INTO llm_calls (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, cached_prompt_tokens, "
|
||||
"model_reported, sampling, reasoning_tokens, tenant_id, meta) "
|
||||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
|
||||
"model_reported, sampling, reasoning_tokens, tenant_id, meta, thinking_observation) "
|
||||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
|
||||
)
|
||||
_FROZEN_PG_INSERT = (
|
||||
"INSERT INTO llm_calls (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, cached_prompt_tokens, model_reported, "
|
||||
"sampling, reasoning_tokens, tenant_id, meta) "
|
||||
"sampling, reasoning_tokens, tenant_id, meta, thinking_observation) "
|
||||
"VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, "
|
||||
"$19, $20, $21, $22, $23, $24) "
|
||||
"$19, $20, $21, $22, $23, $24, $25) "
|
||||
# 无冲突目标(issue #13 Task 2): 带 `(call_id)` 的版本在按 created_at 分区、
|
||||
# 主键为 (call_id, created_at) 的表上匹配不到约束,PG 直接拒收整条写入
|
||||
"ON CONFLICT DO NOTHING"
|
||||
@@ -207,7 +211,7 @@ class TestSchemaModule:
|
||||
|
||||
# COLUMNS 是 INSERT 字段序,不含数据库自填的 created_at
|
||||
assert list(COLUMNS) == [c for c in _EXPECTED_COLUMNS if c != "created_at"]
|
||||
assert len(COLUMNS) == 24
|
||||
assert len(COLUMNS) == 25
|
||||
# 两端 DDL 的列出现顺序 == 物理列序(created_at 在第 19 位)
|
||||
for ddl in (SQLITE_DDL, PG_DDL):
|
||||
assert _first_occurrence_order(ddl, _EXPECTED_COLUMNS) == _EXPECTED_COLUMNS
|
||||
@@ -221,8 +225,8 @@ class TestSchemaModule:
|
||||
"ALTER TABLE llm_calls ADD COLUMN cached_prompt_tokens INTEGER",
|
||||
)
|
||||
assert PG_BACKFILL[-1] == (
|
||||
"meta",
|
||||
"ALTER TABLE llm_calls ADD COLUMN meta JSONB NOT NULL DEFAULT '{}'::jsonb",
|
||||
"thinking_observation",
|
||||
"ALTER TABLE llm_calls ADD COLUMN thinking_observation TEXT",
|
||||
)
|
||||
assert all("IF NOT EXISTS" not in stmt for _, stmt in PG_BACKFILL)
|
||||
|
||||
@@ -270,7 +274,7 @@ class TestSchemaModule:
|
||||
pg = telemetry_schema_sql("postgres")
|
||||
lite = telemetry_schema_sql("sqlite")
|
||||
for script in (pg, lite):
|
||||
# 24 个 INSERT 字段 + created_at 全在,且首次出现顺序与建表 DDL 一致
|
||||
# 25 个 INSERT 字段 + created_at 全在,且首次出现顺序与建表 DDL 一致
|
||||
assert _first_occurrence_order(script, _EXPECTED_COLUMNS) == _EXPECTED_COLUMNS
|
||||
assert "CREATE TABLE IF NOT EXISTS llm_calls" in script
|
||||
# 人执行的那份必须幂等: PG 用 ADD COLUMN IF NOT EXISTS(与库内那份有意不同)
|
||||
@@ -304,11 +308,11 @@ class TestBackendColumnParity:
|
||||
assert sqlite.COLUMNS is COLUMNS
|
||||
assert postgres.COLUMNS is COLUMNS
|
||||
|
||||
def test_caller_dimensions_are_appended_last(self):
|
||||
def test_new_columns_are_appended_last(self):
|
||||
"""新列只能追加在末尾: 旧表经 ALTER 补列必落末尾,插在中间会让两条路径分叉。"""
|
||||
from polygateway.telemetry.schema import COLUMNS
|
||||
|
||||
assert COLUMNS[-2:] == ("tenant_id", "meta")
|
||||
assert COLUMNS[-3:] == ("tenant_id", "meta", "thinking_observation")
|
||||
|
||||
|
||||
class TestSQLiteRecorder:
|
||||
@@ -378,6 +382,28 @@ class TestSQLiteRecorder:
|
||||
assert rows["r-zero"] == 0 # 上报了且确实没推理
|
||||
assert rows["r-none"] is None # 本次调用未上报
|
||||
|
||||
async def test_thinking_observation_column_round_trips(self, tmp_path):
|
||||
"""issue #16: 三态裁定结果落库,事后才能按"这次到底推没推理"分组统计。
|
||||
|
||||
断言的是裸字符串 `"observed"` 而非枚举: 归一化在 emitter 侧完成
|
||||
(`_record` 取 `.value`),recorder 拿到的必须已经是 `str`——`StrEnum`
|
||||
虽是 `str` 子类,asyncpg 的参数编码对子类不保证接受,而遥测写失败只
|
||||
降级成一条 warning,PG 那一路会悄无声息地少一列数据。
|
||||
"""
|
||||
recorder = SQLiteRecorder(tmp_path / "t.db", auto_migrate=True)
|
||||
await _record_minimal(recorder, call_id="t-obs", thinking_observation="observed")
|
||||
await _record_minimal(recorder, call_id="t-absent", thinking_observation="absent")
|
||||
await _record_minimal(recorder, call_id="t-unknown")
|
||||
recorder.close()
|
||||
rows = dict(
|
||||
sqlite3.connect(tmp_path / "t.db")
|
||||
.execute("SELECT call_id, thinking_observation FROM llm_calls")
|
||||
.fetchall()
|
||||
)
|
||||
assert rows["t-obs"] == "observed"
|
||||
assert rows["t-absent"] == "absent" # 观测到"确实没推理",与"看不出来"不是一回事
|
||||
assert rows["t-unknown"] == "unknown"
|
||||
|
||||
async def test_sampling_column_round_trips(self, tmp_path):
|
||||
"""issue #4: 采样参数落库,否则事后无法证明某批数据跑在什么温度下。"""
|
||||
recorder = SQLiteRecorder(tmp_path / "t.db", auto_migrate=True)
|
||||
@@ -541,7 +567,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 → 24 个 recorder 字段(+ created_at 共 25 物理列)
|
||||
assert cols == _EXPECTED_COLUMNS # 22 → 25 个 recorder 字段(+ created_at 共 26 物理列)
|
||||
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 静默吞掉
|
||||
@@ -583,7 +609,7 @@ class TestSQLiteCallerDimensionsAcceptance:
|
||||
|
||||
stale = sqlite3.connect(db)
|
||||
assert [r[1] for r in stale.execute("PRAGMA table_info(llm_calls)")] == (
|
||||
_EXPECTED_COLUMNS[:-2]
|
||||
_EXPECTED_COLUMNS[:-3]
|
||||
) # 补列确实没成功,用例不是在只读库上空转
|
||||
|
||||
|
||||
@@ -591,7 +617,7 @@ class TestSQLiteSchemaMode:
|
||||
"""issue #13: `auto_migrate` 两档——auto 保持自动补列,manual 只裁剪写入不发 DDL。
|
||||
|
||||
列数断言一律按**物理列数**写: 旧表 22 个 INSERT 字段 + `created_at` = 23,
|
||||
补齐后 24 + `created_at` = 25。混用 INSERT 字段数与物理列数是本处最易错的地方。
|
||||
补齐后 25 + `created_at` = 26。混用 INSERT 字段数与物理列数是本处最易错的地方。
|
||||
"""
|
||||
|
||||
def _physical_columns(self, db: Path) -> list[str]:
|
||||
@@ -630,7 +656,7 @@ class TestSQLiteSchemaMode:
|
||||
assert "ALTER TABLE" in message # 给出可直接执行的补列 SQL
|
||||
|
||||
async def test_auto_mode_still_upgrades_the_legacy_table(self, tmp_path):
|
||||
"""auto + 同款旧表: 现状回归,补列后物理列数 23 → 25。"""
|
||||
"""auto + 同款旧表: 现状回归,补列后物理列数 23 → 26。"""
|
||||
db = tmp_path / "auto_legacy.db"
|
||||
_make_pre_tenant_db(db)
|
||||
|
||||
@@ -639,10 +665,10 @@ class TestSQLiteSchemaMode:
|
||||
recorder.close()
|
||||
|
||||
assert self._physical_columns(db) == _EXPECTED_COLUMNS
|
||||
assert len(self._physical_columns(db)) == 25
|
||||
assert len(self._physical_columns(db)) == 26
|
||||
|
||||
async def test_manual_mode_still_creates_a_fresh_table(self, tmp_path):
|
||||
"""manual 只管 ALTER,不管 CREATE: 全新库照建,25 个物理列齐全(设计 §4.2)。"""
|
||||
"""manual 只管 ALTER,不管 CREATE: 全新库照建,26 个物理列齐全(设计 §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")
|
||||
@@ -866,6 +892,7 @@ class TestPostgresBackfillDiscipline:
|
||||
"reasoning_tokens",
|
||||
"tenant_id",
|
||||
"meta",
|
||||
"thinking_observation",
|
||||
]
|
||||
|
||||
def _recorder(self, conn):
|
||||
@@ -926,6 +953,7 @@ class TestPostgresTableProbe:
|
||||
"reasoning_tokens",
|
||||
"tenant_id",
|
||||
"meta",
|
||||
"thinking_observation",
|
||||
]
|
||||
|
||||
def _recorder(self, conn):
|
||||
@@ -1118,6 +1146,62 @@ class TestEmitterRecorderContract:
|
||||
assert set(rec.rows[0]) == set(COLUMNS)
|
||||
|
||||
|
||||
class TestEmitterThinkingObservation:
|
||||
"""issue #16: 三态裁定经 emitter 落库,且落的是**裸 str** 而非枚举实例。
|
||||
|
||||
类型断言不是洁癖: `StrEnum` 虽是 `str` 子类,asyncpg 的参数编码对 `str`
|
||||
子类不保证接受,而遥测写失败只降级成一条 warning——PG 那一路会静默少一列
|
||||
数据,本地 SQLite 测试全绿也发现不了。归一化因此固定在 emitter 侧,与
|
||||
`tenant_id`/`meta`/`sampling` 同一先例。
|
||||
"""
|
||||
|
||||
async def test_attempt_carries_the_verdict_as_a_plain_string(self):
|
||||
rec = _MemoryRecorder()
|
||||
await TelemetryEmitter(rec, text_cap=None).emit_attempt(
|
||||
request=_REQ,
|
||||
source=_source(),
|
||||
call_id="c",
|
||||
latency_ms=1,
|
||||
response=_resp(thinking_observation=ThinkingObservation.OBSERVED),
|
||||
error=None,
|
||||
)
|
||||
value = rec.rows[0]["thinking_observation"]
|
||||
assert value == "observed"
|
||||
assert type(value) is str # 不是 ThinkingObservation: 子类实例不得下沉到 recorder
|
||||
|
||||
async def test_cache_hit_replays_the_recorded_verdict(self):
|
||||
"""缓存命中回放历史那次的裁定: 与 model/prompt_tokens 同一口径。"""
|
||||
rec = _MemoryRecorder()
|
||||
await TelemetryEmitter(rec, text_cap=None).emit_cache_hit(
|
||||
request=_REQ,
|
||||
response=_resp(cache_hit=True, thinking_observation=ThinkingObservation.ABSENT),
|
||||
)
|
||||
assert rec.rows[0]["thinking_observation"] == "absent"
|
||||
|
||||
async def test_terminal_failure_records_unknown(self):
|
||||
"""终态失败无响应可言,记 `unknown`——它恰好就是"观测不到",不撒谎。"""
|
||||
rec = _MemoryRecorder()
|
||||
await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure(
|
||||
request=_REQ, call_id="c", latency_ms=1, error="dead"
|
||||
)
|
||||
value = rec.rows[0]["thinking_observation"]
|
||||
assert value == "unknown"
|
||||
assert type(value) is str
|
||||
|
||||
async def test_failed_attempt_records_unknown(self):
|
||||
"""失败尝试(response=None)同理: 默认视图即 UNKNOWN。"""
|
||||
rec = _MemoryRecorder()
|
||||
await TelemetryEmitter(rec, text_cap=None).emit_attempt(
|
||||
request=_REQ,
|
||||
source=_source(),
|
||||
call_id="c",
|
||||
latency_ms=1,
|
||||
response=None,
|
||||
error="boom",
|
||||
)
|
||||
assert rec.rows[0]["thinking_observation"] == "unknown"
|
||||
|
||||
|
||||
class TestEmitterObservabilityFields:
|
||||
"""issue #3: 三个入口各自的取值口径(设计 §5 表)。"""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user