PolyGateway/research-wiki/plans/2026-08-19-issue12-telemetry-retention.md

实现计划: 遥测正文体量、保留期与访问控制(issue #12)

• 目标: 让下游第一次有手段控制遥测表里存什么、留多久、谁能读——正文可配置截断,保留期与访问控制以可执行模板 + 独立脚本交付,库本体不持有 DELETE/DROP 权限。
• 方案概述: 新增 PGW_TELEMETRY_TEXT_CAP(缺省 None 即不截断),截断只发生在 TelemetryEmitter._record 这个唯一遥测调用点,按每条文本切而非切整串 JSON;保留期走 README 的 RANGE 分区 + pg_partman 模板与 tools/telemetry_retention.py(默认 dry-run);访问控制是纯文档的三角色模板 + REVOKE UPDATE, DELETE。README 的模板 SQL 有真实 PG 集成测试逐条执行。
• 依据设计: research-wiki/designs/2026-08-19-issue12-telemetry-retention-design.md(已人类审批 2026-08-19)。
• 涉及技术: Python 3.11+、argparse、sqlite3、asyncpg、pytest、PostgreSQL 分区与 RLS。
• 保真校验: 本计划不涉及参考实现迁移,保真校验不适用。
• 前置依赖: issue #13 的计划须先合并,本分支必须从合并后的 main 开出(不可两条分支并行改再靠自动合并)。两者都动 config.py:118-137 的字段列表、config.py:423-451 的 _load_pgw 返回键与 client.py:396-407 的装配,字段顺序与返回键极易冲突且冲突后是静默的。两条分支都会改 config.py(新增 settings 字段)与 client.py(装配透传),且本计划 Task 4 的分区模板依赖 #13 的 telemetry_schema_sql() 与无冲突目标的写入。本分支从 #13 合并后的 main 起。

---

文件结构

文件	动作	职责
src/polygateway/middleware/telemetry.py	修改	_cap_text/_cap_messages;TelemetryEmitter 增 text_cap 必填
src/polygateway/config.py	修改	PGW_TELEMETRY_TEXT_CAP 解析与校验;GatewaySettings 增 telemetry_text_cap
src/polygateway/client.py	修改	client.py:149 的 emitter 构造点传参
src/polygateway/embedding.py	修改	embedding.py:131 同上(既有 200 上限保留不动)
src/polygateway/ocr.py	修改	ocr.py:130 同上(既有 200 上限保留不动)
tools/telemetry_retention.py	创建	独立清理脚本,不被库 import
tests/unit/test_telemetry.py	修改	截断行为、三链路覆盖
tests/unit/test_cache.py	修改	红线: 缓存 key 不受 cap 影响
tests/unit/test_config.py	修改	配置校验
tests/unit/test_retention_tool.py	创建	脚本 dry-run/apply(经 subprocess)
tests/integration/test_postgres_telemetry.py	修改	README 模板 SQL 逐条执行
README.md、CHANGELOG.md、.env.example	修改	生产部署模板、推荐配置组合、配置键

依赖顺序: Task 1 → Task 2 → (Task 3 ‖ Task 4) → Task 5。

---

关键接口(跨任务消费,此处定稿)

截断函数(middleware/telemetry.py 模块级私有,紧邻 _canonical_meta_json):

def _cap_text(text: str, cap: int | None) -> str:
    """超出 cap 时头部硬切并附省略标记 `…(略 N 字)`;cap 为 None 原样返回。"""

def _cap_messages(messages: list[dict[str, Any]], cap: int | None) -> list[dict[str, Any]]:
    """对每条消息的文本 content 与多模态 part 中 type == "text" 的 text 逐条施加 cap。

    非字符串 content 原样放行(外部输入形状不可控,遥测路径不得因此抛错)。
    """

TelemetryEmitter 构造签名(text_cap keyword-only 必填,无默认值):

class TelemetryEmitter:
    def __init__(
        self, recorder: TelemetryRecorder, *, pricing: PricingTable | None = None,
        text_cap: int | None,
    ) -> None: ...

三个公共 Client 的 __init__ 各增 keyword-only text_cap,带默认值 None(与既有全部可选参数同款,非破坏性):

class GatewayClient:      # client.py:130 起的构造签名
    def __init__(self, *, ..., text_cap: int | None = None) -> None: ...
# EmbeddingClient / OcrClient 同款

为什么 emitter 必填而 Client 带默认: TelemetryEmitter 是库内部类,唯一构造者是这三个 Client,必填能保证没有一处漏传;而三个 Client 是公共装配路(下游可直接构造并注入自己的 recorder),给它们加必填参数会破坏既有调用点,且默认 None 恰好等于全局缺省行为(不截断)。少了这一层,直接构造的下游要么撞 TypeError,要么永远没法启用 cap。

GatewaySettings 新字段(无默认值),排在 telemetry_auto_migrate 之后:

telemetry_text_cap: int | None

tools/telemetry_retention.py 的 CLI 契约:

--backend sqlite|postgres   必填
--path PATH | --dsn DSN     按 backend 二选一,必填
--older-than-days N         必填,N >= 0
--apply                     缺省不带即 dry-run(只统计不删)
--batch-size N              仅 postgres,缺省 1000
--vacuum                    仅 sqlite,须与 --apply 同时给
退出码: 0 正常;1 参数错误;2 连接/权限失败;3 目标是分区表(PG,提示改用 DROP PARTITION)

---

Task 1: 正文截断与 emitter 参数

• 文件: src/polygateway/middleware/telemetry.py、src/polygateway/client.py、src/polygateway/embedding.py、src/polygateway/ocr.py;tests/unit/test_telemetry.py、tests/unit/test_cache.py。
• 行为:
  • 按上文签名实现两个截断函数;_record 内在 digest_messages(...) 之后、json.dumps(...) 之前调用 _cap_messages,并对 response_text、thinking 调用 _cap_text。
  • TelemetryEmitter 增必填 text_cap;库内三个构造点(client.py:149、embedding.py:131、ocr.py:130)同步传参;三个 Client 的 __init__ 各增带默认值的 text_cap 参数(见上,否则直接构造路要么 TypeError 要么永远用不上 cap);测试内十余处 emitter 构造点一并补齐。
  • digest_messages 一个字节都不改(它是缓存 key 与遥测共用的函数,middleware/cache.py:31)。
  • _cap_messages 必须产出新对象,严禁就地修改。这是本任务最容易踩的坑: digest_messages 对 content 不是 list 的消息是原样 append 同一个 dict 对象(cache.py:43),即遥测拿到的 dict 与调用方传入的、以及缓存 key 计算用的是同一份。就地改它会同时污染调用方的 messages、后续重试尝试的请求体与缓存写入的 key,且全程无任何报错。多模态 part 同理(_digest_part 对非 image_url 的 part 也是原样返回)。
  • embedding.py:73 与 ocr.py:73 各自的 200 字符上限保留不动,与新 cap 是"取更严者"的关系。
• 验收:
  • cap=None → 落库正文与今天逐字节相同。
  • cap=N → 每条 content 被切且整串 messages JSON 仍可 json.loads;标记含省略字数。
  • 多模态消息: type == "text" 的 part 被切,image_url 的 sha256 摘要原样不动。
  • 非字符串 content(如 123、None、嵌套 dict)不抛异常。
  • response/thinking 同样受 cap。
  • OCR 与 embed 两条链路的行同样受 cap(它们共用 _record)。
• 测试:
  • 上述六条各一例(tests/unit/test_telemetry.py)。
  • 红线用例之一(tests/unit/test_cache.py): 取一组含长文本的 messages,先算一次 build_cache_key(...),再经 cap=8 的 emitter 走一遍遥测,然后用同一个 messages 对象再算一次 key —— 两次输出必须逐字节相同。这测的是"截断没有就地改掉调用方的对象",而不只是"截断函数是纯的"。
  • 红线用例之二(tests/unit/test_telemetry.py): cap=8 走一遍遥测后,断言传入的 messages 结构与内容完全未变(含嵌套的多模态 part),落库的那份则已被截断。
  • 先失败证据: 参数不存在时 TypeError;截断未实现时 cap=8 的用例读回全文;就地修改的实现会让两条红线用例直接失败(先写一版就地改的实现跑一遍,把失败输出留档,证明红线用例真的能抓住它)。
• 验证: conda run -n PolyGateway pytest tests/unit/test_telemetry.py tests/unit/test_cache.py tests/unit/test_ocr_client.py tests/unit/test_embedding.py -v → PASS。
• 提交: feat: cap telemetry bodies at a configurable length

Task 2: 配置与装配

• 文件: src/polygateway/config.py、.env.example;tests/unit/test_config.py。
• 行为: _load_pgw 解析 PGW_TELEMETRY_TEXT_CAP(未设 → None;设了则转 int);GatewaySettings 增 telemetry_text_cap: int | None,_validate_telemetry 内校验 <= 0 报 ValueError(错误信息含键名);client.py 把它传给 emitter;.env.example 加注释行,写明缺省不截断及其取舍(截断后遥测不再是审计证据、无法复现重放)。
• 验收: 未设 → None;"0" 与 "-1" 报 ValueError;非整数字符串报 ValueError;合法值透传到 emitter 并生效(端到端一例)。
• 测试: 上述四条各一例。先失败证据: 字段不存在时 AttributeError。
• 验证: conda run -n PolyGateway pytest tests/unit/test_config.py tests/unit/test_client.py -v → PASS。
• 提交: feat: wire the telemetry text cap through settings

Task 3: 保留期脚本

• 文件: 创建 tools/telemetry_retention.py;创建 tests/unit/test_retention_tool.py。
• 行为: 按上文 CLI 契约实现。
  • 缺省 dry-run: 不带 --apply 时只统计并打印将删除的行数、created_at 时间范围、按 tenant_id 的分布,一行不删。
  • SQLite: DELETE FROM llm_calls WHERE created_at < ?;--vacuum 才执行 VACUUM(它重写整库,不得默认)。
  • PG: 分批 DELETE(每批一个事务,--batch-size 控制),避免长事务与锁膨胀;先探测目标是否为分区表(pg_partitioned_table),是则打印"改用 DETACH/DROP PARTITION"并以退出码 3 结束,不执行 DELETE。
  • 脚本不被库 import(tools/ 规则);缺 asyncpg 时明确报错退出码 2,不静默降级(这是运维工具不是库路径)。
  • 文档串: 帮助文本写明"用维护角色跑,不要用应用账号(应用账号已被 REVOKE DELETE)"。
• 验收: 见测试。
• 测试(经 subprocess.run([sys.executable, "tools/telemetry_retention.py", ...]),真实临时 SQLite):
  • dry-run 后行数不变,stdout 含将删行数与时间范围。
  • --apply 后仅超期行被删,未超期行完好。
  • --older-than-days 0 的边界(删到"此刻之前")行为明确且与文档一致。
  • 参数缺失/冲突(如 backend=sqlite 却给 --dsn)退出码 1。
  • --vacuum 不带 --apply 时退出码 1。
  • PG 分支必须自带证据(集成,真实 PG,临时 schema 隔离): ① 临时 schema 内建分区表,脚本探测到后打印改用 DETACH/DROP PARTITION 的提示并以退出码 3 结束、一行都没删; ② 临时 schema 内建普通表灌入跨日期的行,--apply --batch-size 2 后仅超期行被删且分多批提交; ③ 缺 asyncpg 时退出码 2——用一个只含 raise ImportError 的临时 asyncpg.py 目录挂进 PYTHONPATH 跑 subprocess 来构造该场景,不要靠 monkeypatch(脚本走的是子进程)。
  • 先失败证据: 脚本不存在时 subprocess 返回非零且 stderr 含 No such file;PG 三例在脚本只实现 SQLite 分支时分别以"未知 backend"或退出码 1 失败。
• 验证: conda run -n PolyGateway pytest tests/unit/test_retention_tool.py tests/integration/test_retention_tool_pg.py -v → PASS(PG 三例须在有 PGW_TELEMETRY_PG_DSN 的环境实跑,skip 不算通过)。
• 提交: feat: add a retention script downstreams can schedule

Task 4: 生产部署模板与其机械化验收

• 文件: README.md;tests/integration/test_postgres_telemetry.py。
• 行为: README 现有多租户 RLS 段扩为完整的"生产部署 DDL 模板"一节,包含:
  • 三角色: owner(DDL 与清理)、app(INSERT + 受 RLS 约束读自己租户)、report(只读 + 受 RLS 约束)。
  • 不可变性: REVOKE UPDATE, DELETE ON llm_calls FROM app, report;触发器兜底明确标注"只防误操作,不防恶意(属主可 disable)"。
  • 分区: PARTITION BY RANGE (created_at)、主键 (call_id, created_at)、pg_partman retention;并写明分区部署下幂等键实际是 (call_id, created_at),emit_cache_hit 复用历史 call_id,故缓存命中行在普通表上第二次起会被吞掉、在分区表上每次都落一行——按 cache_hit 统计的下游必须知道。
  • 库需要的最小权限: catalog SELECT(探测)+ INSERT +(可选)CREATE;auto 档另需 ALTER。
  • 合规下游推荐配置: 一段可直接照抄的组合(PGW_TELEMETRY_TEXT_CAP + 分区 retention + 三角色),不把三件事散着让下游自己拼。
  • 截断覆盖面的诚实声明(设计 §5.2,不得省): cap 作用于消息的 content 文本与多模态 part 中 type == "text" 的 text,与 digest_messages 的处理面一致;调用方放进 tool_calls.function.arguments 等其他字段的内容不在覆盖范围内。漏写这条,下游会以为开了 cap 就没有全文残留,合规判断直接出错。
  • SQLite 侧的保留期(设计 §6,不得省): 给按天/按实验轮转库文件的建议——这是 VT / CHSAnalyzer / dissect 三家现成的形态,比对本地文件跑 DELETE + VACUUM 更省事也更安全;tools/telemetry_retention.py 的 SQLite 分支是给"已经攒成一个大库"的存量场景兜底,不是推荐路径。
  • 每个代码块 ≤15 行(输出规范),超长的拆成相邻多块。
• 验收: 模板 SQL 在真实 PG 上逐条可执行;README 里的行为描述与实测一致。
• 测试(集成,真实 PG,新建自己的 fixture,手法照搬 least_privilege_dsn 的临时 schema + 临时角色 + teardown 删净,严禁碰共享的 public.llm_calls): 新增一例,把 README 的模板 SQL 逐条执行后断言:
  • app 角色能 INSERT、不能 DELETE(报权限错)。
  • report 角色能读、不能写。
  • 未设 app.tenant_id 时查询为零行(fail-closed),设了则只看到本租户的行。
  • 分区表上写入成功且落进当月分区。
  • 先失败证据: 模板尚未写进 README 时该测试无 SQL 可读、直接失败。
• 验证: conda run -n PolyGateway pytest tests/integration/test_postgres_telemetry.py -v → PASS(必须在有 PGW_TELEMETRY_PG_DSN 且账号有 CREATEROLE 的环境实跑;无权限时 skip,skip 不算通过)。
• 提交: docs: ship a production deployment template with its own test

Task 5: CHANGELOG 与 wiki

• 文件: CHANGELOG.md、Gitea wiki(指南-遥测与成本/参考-配置键/参考-公共API)、research-wiki/ARCHITECTURE.md。
• 行为: CHANGELOG 写明新配置键、缺省不截断的取舍、保留期脚本与部署模板的位置;ARCHITECTURE 的 D15(库对下游库的权限边界)若 issue #13 已建,此处只补 #12 的一面;wiki 三页按 docs-convention §2 同步。
• 验收: 版本条目里能一眼看出"默认行为未变,新增的是手段";wiki 与 README 不重复叙述(深度内容只放指针)。
• 测试: 无自动化测试。
• 验证: conda run -n PolyGateway make ci → 全绿。
• 提交: docs: record the retention boundary and its knobs

---

完成判据

1. 五个任务的提交点全部落地,make ci 全绿。
2. 每条行为变更能出示先失败后通过的测试证据;Task 1 的缓存 key 红线用例与 Task 4 的模板 SQL 用例必须在本会话内实跑并留下输出。
3. 合并前派全新上下文 verifier subagent 独立验证(CLAUDE.md §3 硬门)。
4. 与 issue #13 合并后一起发 1.2.3,发布走 CLAUDE.md §4.4.1 九步——README 必须在构建之前定稿(sdist 会把当时那份固化进包)。