diff --git a/.env.example b/.env.example index 08ca2b9..7a45996 100644 --- a/.env.example +++ b/.env.example @@ -66,6 +66,13 @@ PGW_TELEMETRY_BACKEND=none # sqlite | postgres | none(必填) # # sqlite 则是下游自己的本地文件(runs/*.db):没有 DBA、没有迁移工具、 # # 没有第二个系统碰它,ALTER 是毫秒级元数据操作,强加手工 SQL 步骤是净损失。 # PGW_TELEMETRY_PG_DSN=postgresql://user:pass@host:5432/polygateway # postgres 时必填;严禁指向在用业务库(实验室约定: 专用库 polygateway) +# PGW_TELEMETRY_TEXT_CAP=2000 # 遥测落库正文的字符上限,须 > 0;**不设 = 不截断**(缺省,逐字节留全文)。 +# # 作用于 messages 的每条文本 content、多模态 text part、response 与 thinking; +# # 超出部分头部保留、尾部换成 `…(略 N 字)`。多模态 image_url 的 sha256 摘要不受影响。 +# # 缺省为何是"不截断": 遥测被下游当**审计证据**用——出了问题要回答"当时到底发了什么", +# # 也要能拿原样的请求复现与重放;截断后这两件事都做不成,而既有下游正依赖这一行为。 +# # 反面同样要看清: 不截断意味着客户合同、标书全文无限期留在 llm_calls 里, +# # 多租户下还混在同一张表。真在意留存面的部署应显式设一个上限,并配保留期与访问控制。 # PGW_PRICING_PATH=config/prices.json # 可选: {"": {"input_per_1m": x, "output_per_1m": y}};缺省 cost 恒 None # # 可选第三档 "cached_input_per_1m": z —— 供应商 prompt cache 命中部分的单价; # # 不填即命中部分也按 input 全额计(库不猜折扣率),cost 会偏高 diff --git a/CHANGELOG.md b/CHANGELOG.md index 21638a8..0d3fce6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,13 +4,39 @@ 遥测表 `llm_calls` 的结构变更从此**由下游掌控**(issue #13)。此前两个后端都会在初始化期对下游数据库发 DDL:表不存在则建表,表存在但缺列则逐列 `ALTER TABLE ADD COLUMN`,而补列**没有任何开关**——库一升级、下次调用即自动执行。在共享的生产 Postgres 上这有三重问题:`ALTER` 取 ACCESS EXCLUSIVE 锁会排在长事务后阻塞该表其后的所有查询(而遥测是业务路径上的内联 `await`),多进程多版本共存时谁先补列是竞态,且这些 DDL 不进任何迁移记录、事后无从审计。调研过的 11 个同类系统(Celery / APScheduler / Alembic / Django contrib / Hangfire / Quartz.NET / dbt / Airbyte / Fivetran / Prefect / Airflow)里没有一个把它作为默认行为。 -### 请先读这一条:三处破坏性变更 +同一版里,issue #12 补上这条边界的另一半——**删数据**,并把它落成三样**手段**: 遥测正文的可配置上限、`tools/` 下的独立保留期脚本、README 里的一份生产部署 DDL 模板。三样**没有一样改变缺省行为**——不设 `PGW_TELEMETRY_TEXT_CAP` 即逐字节存全文,与今天完全一致。缺省不截断是刻意取舍: 截断之后的遥测不再是审计证据,也无法拿原样的请求复现与重放,而这正是既有下游在依赖的用法;代价是 issue 那句"无限期保留全部租户全文不应是默认状态"只被解决了一半——默认仍是全文,但下游第一次有了不写全文的手段。库本体同样不因此持有 `DELETE`/`DROP` 权限: 保留期是 `tools/` 下的独立脚本,库不 import 它。 + +### 请先读这一条: 照抄过 1.2.1 那份 RLS 模板的 Postgres 部署,遥测表很可能是空的 + +1.2.1 的 README 给的 RLS 模板把**写侧**也绑在了 `app.tenant_id` 这个 GUC 上: + +```sql +-- 1.2.1 的模板,有缺陷,勿用 +CREATE POLICY llm_calls_tenant_isolation ON llm_calls TO polygateway_app + USING (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')) + WITH CHECK (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')); +``` + +但 `PostgresRecorder` 用**一个连接池给所有租户**写遥测,源码里从不发 `set_config('app.tenant_id', ...)`——库既拿不到也不该猜租户上下文该怎么设。于是 `WITH CHECK` 里的 `current_setting` 恒为 NULL、等值比较恒不为真,**库的每一条 `INSERT` 都被 policy 拒绝**。而遥测的失败方向是静默降级,所以表现不是报错,是**整张表零行**——业务调用一切正常,不看日志根本发现不了。 + +照抄过就请现在查这两条: + +| 查什么 | 中招的样子 | +|---|---| +| `SELECT count(*) FROM llm_calls;`,且必须用能**绕过 RLS** 的角色(superuser 或带 `BYPASSRLS` 属性的角色)——`FORCE` 之下表属主自己也受 policy 管,用它查出的 0 行分不清是"没数据"还是"读不到" | 启用 RLS 之后一直是 0,或从某个时刻起不再增长 | +| 应用日志里遥测写入的降级告警,前缀 `Postgres 遥测写入失败(丢弃该行):` | 每次调用刷一条,附带的 PG 原话是 `new row violates row-level security policy for table "llm_calls"` | + +本版的新模板把写侧改为 `WITH CHECK (true)`,隔离交由**读侧**的 `USING` 承担: 在这个模型里写入方是库自己(可信),要隔离的是读取方。若你的调用点保证每次调用都带 `tenant_id`,可把写侧收紧成 `WITH CHECK (tenant_id <> '')`,代价是漏传 `tenant_id` 的调用点会**丢遥测行**(同样只留一条 warning)。完整理由与四个陷阱见 README「生产部署 DDL 模板(PostgreSQL)」第 4 小节。 + +### 破坏性变更(五项) | # | 变更 | 影响与应对 | |---|---|---| | ① | **Postgres 侧不再自动补列**(缺省转为 manual 档) | 库升级带来新列时,旧表不会被自动 `ALTER`:库改为发**一条** warning 点名缺失的维度并附上可直接执行的 SQL,同时按现有列裁剪 `INSERT` 继续写入——**缺的那几列静默不落库**,直到有人执行那几条 SQL。要恢复旧行为设 `PGW_TELEMETRY_SCHEMA_MODE=auto`。SQLite 侧缺省不变(仍 auto),理由见下 | | ② | 两个 recorder 新增 **keyword-only 必填**参数 `auto_migrate` | `SQLiteRecorder(db_path, *, auto_migrate)` 与 `PostgresRecorder(dsn, *, pool=None, auto_migrate)`;直接构造 recorder 的调用点必须补这个参数,不传即 `TypeError`。**故意不给默认值**:缺省规则只写在 config 一处,不与类签名漂移 | | ③ | `GatewaySettings` 新增**必填**字段 `telemetry_auto_migrate: bool` | 只影响「构造函数全量注入」这条装配路(测试/高级用法);`from_env()` / `from_settings()` 的用户零改动。`telemetry_backend="none"` 时该字段在 `__post_init__` 归一为 `False` | +| ④ | `GatewaySettings` 再新增**必填**字段 `telemetry_text_cap: int \| None` | 同 ③,只影响直接构造这条路。`None`(不截断)是**取值**而不是默认值——字段本身没有默认值;`<= 0` 在 `__post_init__` 直接 `ValueError`,不会被当成"不截断" | +| ⑤ | `TelemetryEmitter` 新增 **keyword-only 必填**参数 `text_cap` | 库内部类,库内唯一构造者是三个公共 Client(本版已全部接通);直接构造过它的测试/高级用法不传即 `TypeError`。同样**故意不给默认值**: 漏传会静默改变落库正文。它也是值域校验的收口处——三个 Client 的 `text_cap` 全汇流到这里,而 `GatewaySettings` 那道只管 env 一条路 | ### 新增 @@ -18,6 +44,14 @@ - **公共函数 `telemetry_schema_sql(backend) -> str`**(已进顶层 `__all__`):返回可直接粘进迁移文件的完整脚本——注释头 + `CREATE TABLE IF NOT EXISTS`(全量列)+ 各补列语句。PG 变体带 `ADD COLUMN IF NOT EXISTS`,整段**可重复执行**;SQLite 无该语法,以注释标明"仅当该列不存在时执行"。非法 `backend` 抛 `ValueError`。 - **manual 档的缺列告警**逐列点名并写明后果(「以下维度不会被记录: tenant_id, meta」),附上可直接执行的 ALTER,且**只在准备期发一次**,不逐行刷屏。只说"缺列"是不够的:静默丢维度的后果是多租户账目全归空串且无任何报错。 +issue #12 交付的三样手段列在下表——它们改变的是**能做什么**,不是**默认做什么**: + +| 手段 | 内容 | +|---|---| +| **`PGW_TELEMETRY_TEXT_CAP`**(可选正整数键) | 遥测落库正文的字符上限;**不设 = 不截断**(缺省)。作用面正好四处: `messages` 里每条消息的字符串 `content`、多模态 content 数组中 `type == "text"` 的 part 的 `text`,以及 `response` 与 `thinking` 两列;超出部分头部保留、尾部换成 `…(略 N 字)`。**按每条文本切,而不是切整串 JSON**——后者会往不做任何校验的 TEXT 列里写进非法 JSON,让此后一切按 JSON 解析该列的分析全废。**覆盖面到此为止**: 调用方塞进 `tool_calls.function.arguments`、`name` 等 `content` 之外字段的内容不在其中,开了 cap 不等于表里没有全文残留 | +| **`tools/telemetry_retention.py`**(独立运维脚本) | 按 `created_at` 清理过期行。**默认 dry-run**: 先打出将删行数、`created_at` 窗口与按 `tenant_id` 的分布,让运维先判断"要删的是不是我想删的",给了 `--apply` 才真动手。退出码是与调度器(cron/systemd)的契约: `0` 正常(含 dry-run)、`1` 参数错误、`2` 连接/权限/目标表不可用(**含缺 `asyncpg`**——明确报错退出,绝不静默变成"删了 0 行")、`3` 目标是 PostgreSQL 分区表,此时脚本**拒绝 DELETE**,让路给 O(1) 的 `DETACH` + `DROP PARTITION`。请用维护角色跑,不要用应用账号(模板已对它 `REVOKE UPDATE, DELETE`) | +| **README 新增「生产部署 DDL 模板(PostgreSQL)」一节** | 三角色、`created_at` RANGE 分区与 `pg_partman` retention、`REVOKE UPDATE, DELETE` 加触发器兜底、RLS、**库自己需要的最小权限**、合规下游可直接照抄的组合配置、SQLite 侧按天轮转库文件。7 个 SQL 块带 `` 锚点,由 `tests/integration/test_postgres_telemetry.py` 从 README 解析出来在真实 PG 上逐条执行——**模板只有这一份**,不会与测试各自漂移。上面那条 RLS 缺陷正是"文档里的 SQL 从没被执行过"的产物 | + ### 变更 - **Postgres 的写入去掉了冲突目标**:`ON CONFLICT (call_id) DO NOTHING` → `ON CONFLICT DO NOTHING`。普通表上语义**逐字等价**(表上只有主键这一个唯一约束),但带目标的版本要求恰好匹配 `(call_id)` 的唯一约束,而 PostgreSQL 要求分区表的唯一约束必须包含分区键——按 `created_at` 分区后主键变成 `(call_id, created_at)`,该语句会被 PG 直接拒收,且失败只逐行 warning,表现为分区部署下遥测全线静默丢数据。SQLite 的 `INSERT OR IGNORE` 本就无目标,未动。 @@ -29,6 +63,8 @@ - **manual 档仍然建表**。issue 把建表列为现状描述而非指控(它已在 #9 收口为"PG 侧先 `to_regclass` 探测、表在就不发 DDL")。新建表没有既有数据、没有并发访问者,不存在锁队列与数据风险,而停掉它会让"零配置起步"这条路彻底断掉。 - **auto 档行为与从前逐字相同**,包括补列失败时**不裁剪**:该档承诺的是"把列补上",补不上就让缺列以逐行 warning 暴露;要降级写入请显式选 manual。 - 降级方向不变:缺列、补列失败、写入失败一律只 warning,绝不冒泡打断业务调用;列名与列序不变;错误面零变更。 +- **遥测缺省不截断**: 不设 `PGW_TELEMETRY_TEXT_CAP` 时落库正文与今天逐字节相同。`digest_messages`(缓存 key 与遥测共用的那个摘要函数)一个字节没改,截断只发生在遥测分支、缓存路径不经过它;且截断**只产出新对象、绝不就地修改**——`digest_messages` 对非 list 的 `content` 是原样透传**同一个 dict 对象**,就地改会一并污染调用方持有的 messages、后续重试的请求体与缓存写入的 key,而且全程没有任何报错。两条红线测试分别钉死这两件事: 同一组 messages 在 cap 生效前后 `build_cache_key` 的输出逐字节相同、落库那份被截断而调用方持有的那份(含嵌套 part)一字未改。 +- embedding 与 OCR 两条链路各自既有的 200 字符上限**保留不动**,与新 cap 是"取更严者"的关系;多模态 `image_url` 早已是 sha256 摘要,不受 cap 影响。 ### 库对下游数据库的承诺(Expand/Contract,本版成文) @@ -36,10 +72,14 @@ 合起来它们保证:你可以自行给 `llm_calls` 加列、加索引、挂 RLS,乃至把它建成 `PARTITION BY RANGE (created_at)` 的分区表,库的探测、补列与写入都照常工作。完整说明见 README「遥测表 schema 与升级纪律」——那份随包分发,`research-wiki/` 不在 sdist 内。 +同一条边界的另一半是**删数据**: 库不持有 `DELETE`/`DROP` 权限,保留期与访问控制以 README 模板加 `tools/` 独立脚本交付。这不是保守,是两条诉求的权限张力逼出来的唯一解——模板建议对应用角色 `REVOKE UPDATE, DELETE ON llm_calls`(按不可变审计表对待),那么过期清理就不可能再由应用角色的 `DELETE` 完成,只能是属主对 `created_at` RANGE 分区的 `DETACH` + `DROP PARTITION`(那是 DDL,同样不触发不可变性触发器)。分区在这里**不可替代**,不是性能偏好。 + ### 升级提示 - 用 `from_env()` / `from_settings()` 装配的下游**无需改代码**;Postgres 下游升级后建议执行一次 `python -c "import polygateway; print(polygateway.telemetry_schema_sql('postgres'))"` 的输出,把新列补齐(不补则新维度不落库,库会在首次写入前用一条 warning 点名)。 - 直接构造 `SQLiteRecorder` / `PostgresRecorder` 或直接构造 `GatewaySettings` 的调用点必须补上新参数/新字段,否则 `TypeError`。 +- **截断不需要任何升级动作**: 不设 `PGW_TELEMETRY_TEXT_CAP` 就维持全文。真在意留存面的部署应显式设一个上限,并同时配上保留期与访问控制——三件事要一起上才有意义,README 给了可直接照抄的组合。 +- 已按 1.2.1 的 RLS 模板部署过 Postgres 的,请先做本版开头那两条自查,再换用新模板。 ## 1.2.1(2026-08-18) diff --git a/README.md b/README.md index 90eaf0d..8f4c084 100644 --- a/README.md +++ b/README.md @@ -127,30 +127,7 @@ resp = await client.chat( 存储上 `tenant_id` 两端都是 `TEXT NOT NULL DEFAULT ''`,`meta` 在 Postgres 是 `JSONB`、在 SQLite 是 `TEXT`;老表要补上这两列(补列是否由库自动执行取决于 `PGW_TELEMETRY_SCHEMA_MODE`,见[遥测表 schema 与升级纪律](#遥测表-schema-与升级纪律)),**补列后老行读出是空串而非 NULL**(NULL 在任何 RLS policy 下都对所有人不可见,空串则可用一条 SQL 审出还有多少行待归属)。 -**库只提供列,不启用 RLS、不建索引。** 要数据库层的强制隔离,以下 DDL 是**下游 DBA 的职责,库不会代劳**;不执行则 `tenant_id` 只是一个可查可过滤的普通列,没有任何数据库层强制。库不代劳的原因是 default-deny:启用 RLS 而没有匹配的 policy = 零行可写且静默不报错,会让非多租户部署的遥测全量写失败。 - -```sql -ALTER TABLE llm_calls ENABLE ROW LEVEL SECURITY; -ALTER TABLE llm_calls FORCE ROW LEVEL SECURITY; -- 属主不豁免 -CREATE POLICY llm_calls_tenant_isolation ON llm_calls TO polygateway_app - USING (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')) - WITH CHECK (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')); -``` - -```sql -CREATE INDEX CONCURRENTLY idx_llm_calls_tenant_created - ON llm_calls (tenant_id, created_at); -``` - -`current_setting(..., true)` 的第二参数令 GUC 未设时返回 NULL 而非抛错,外层 `NULLIF` 把空串归一为 NULL——合起来使**未设租户 = 零行**(fail-closed)而不是全部行。索引列序不可颠倒:启用 RLS 后 policy 给每条查询隐式追加 `tenant_id` 等值谓词,它出现在 100% 的谓词里,必然是前导列。 - -三个陷阱,每一个的失败形态都是**静默的**: - -| 陷阱 | 后果 | -|---|---| -| 表属主默认**豁免** RLS | 只写 `ENABLE` 而漏 `FORCE`,用属主角色连库时隔离形同虚设,且查询一切正常看不出来 | -| 租户上下文必须在**显式事务内**用 `set_config('app.tenant_id', ..., true)` | asyncpg 默认 autocommit,单发 `SET LOCAL` 会当场失效,而 PG **只发 warning 不报错**;表现是 policy 永远拿不到租户 → fail-closed 到零行 | -| policy 必须同时写 `USING` 与 `WITH CHECK` | 只写前者则租户 A 读不到 B 的行,却**能插入标着 B 的行**——污染发生在写入侧,读侧查不出来 | +**库只提供列,不启用 RLS、不建索引。** 数据库层的强制隔离是**下游 DBA 的职责,库不会代劳**;不执行则 `tenant_id` 只是一个可查可过滤的普通列,没有任何数据库层强制。库不代劳的原因是 default-deny:启用 RLS 而没有匹配的 policy = 零行可写且静默不报错,会让非多租户部署的遥测全量写失败。三角色、RLS policy、分区与保留期的完整可执行模板见[生产部署 DDL 模板](#生产部署-ddl-模板postgresql)。 ## 遥测表 schema 与升级纪律 @@ -210,6 +187,176 @@ PG 变体的补列语句带 `ADD COLUMN IF NOT EXISTS`,**整段可重复执行** | 库从不 `SELECT *`,也从不读回这张表的数据 | 库侧根本没有读路径,你加索引、加自己的列、挂 RLS 都影响不到它 | | 写入的冲突处理**不绑定具体约束** | 你可以把 `llm_calls` 建成 `PARTITION BY RANGE (created_at)` 的分区表(此时主键必须是 `(call_id, created_at)`,PG 要求分区表唯一约束含分区键),库的探测、补列与写入照常工作 | +## 生产部署 DDL 模板(PostgreSQL) + +上一节讲的是**库怎么对待这张表**(只探测、只 INSERT、可选建表);本节讲的是**你该把这张表部署成什么样**:谁能读、谁能写、写进去的行能不能被改、存多久。这些库一件都不代劳——它没有、也不该有这些权限。 + + + +模板按下表顺序执行,标识符(角色名、schema、分区月份、密码)按你的环境改;`llm_calls` 一律不写 schema 限定,靠 `search_path` 解析,与库的写入口径一致。 + +| # | 锚点 | 做什么 | +|---|---|---| +| 1 | `roles` | 建三角色并授 schema 级权限 | +| 2 | `table` | 把 `llm_calls` 改造成按 `created_at` 的 RANGE 分区表,属主归 `polygateway_owner` | +| 3 | `partition` | 建一个月分区(生产用 `pg_partman` 自动滚动) | +| 4 | `grants` | 授表级权限并 `REVOKE UPDATE, DELETE` | +| 5 | `immutable` | 触发器兜底(只防误操作) | +| 6 | `rls` | 启用并 `FORCE` RLS + 两条 policy | +| 7 | `index` | `(tenant_id, created_at)` 复合索引 | + +### 1. 三角色 + +| 角色 | 拿到什么 | 谁在用 | +|---|---|---| +| `polygateway_owner` | 表属主:DDL、加分区、删分区 | DBA / 定时任务;**不用它连库跑业务** | +| `polygateway_app` | `INSERT` + 受 RLS 约束的 `SELECT` | 库的连接串用这个 | +| `polygateway_report` | 受 RLS 约束的 `SELECT` | BI、对账、成本报表 | + + + +```sql +CREATE ROLE polygateway_owner NOLOGIN; +CREATE ROLE polygateway_app LOGIN PASSWORD 'CHANGE_ME_APP'; +CREATE ROLE polygateway_report LOGIN PASSWORD 'CHANGE_ME_REPORT'; +GRANT polygateway_owner TO CURRENT_USER; -- 下一块要把表属主改过去,须先成为它的成员 +GRANT USAGE ON SCHEMA public TO polygateway_owner, polygateway_app, polygateway_report; +GRANT CREATE ON SCHEMA public TO polygateway_owner; -- 滚动分区要在该 schema 里建表 +``` + +### 2. 分区表 + +分区表**必须下游先手工建**:库的 `CREATE TABLE` 只会建普通表。列不在这里重抄一份——抄了就会漂移,故先用库自带脚本建出普通表,再原地改造: + +```bash +python -c "import polygateway; print(polygateway.telemetry_schema_sql('postgres'))" \ + | psql "$PGW_TELEMETRY_PG_DSN" +``` + + + +```sql +ALTER TABLE llm_calls RENAME TO llm_calls_seed; -- 上一步建出的普通表当模子 +CREATE TABLE llm_calls ( + LIKE llm_calls_seed INCLUDING DEFAULTS, -- 列/类型/NOT NULL/DEFAULT 全照搬 + PRIMARY KEY (call_id, created_at) -- 分区表的唯一约束必须含分区键 +) PARTITION BY RANGE (created_at); +DROP TABLE llm_calls_seed; +ALTER TABLE llm_calls OWNER TO polygateway_owner; +``` + + + +```sql +CREATE TABLE llm_calls_2026_01 PARTITION OF llm_calls + FOR VALUES FROM ('2026-01-01 00:00:00+00') TO ('2026-02-01 00:00:00+00'); +ALTER TABLE llm_calls_2026_01 OWNER TO polygateway_owner; +``` + +生产不要手工滚月份,交给 [`pg_partman`](https://github.com/pgpartman/pg_partman):5.x 用 `create_parent(p_parent_table := 'public.llm_calls', p_control := 'created_at', p_interval := '1 month')`(4.x 的参数序不同,以你装的版本文档为准),再把 `part_config.retention` 设成 `'6 months'`、`retention_keep_table` 设成 `false`,`run_maintenance_proc()` 就会到期 `DROP` 整个分区。清理必须走 `DETACH`/`DROP PARTITION` 而**不是** `DELETE`——这不是性能偏好,是权限张力的唯一解:下一块要对应用角色 `REVOKE DELETE`,而 `DROP PARTITION` 是属主的 DDL,两者不冲突,`DELETE` 则必然冲突。 + +**分区部署改变了幂等键**,按 `cache_hit` 出报表的下游必须知道:普通表上主键是 `call_id`,分区表上是 `(call_id, created_at)`。库的写入是无冲突目标的 `ON CONFLICT DO NOTHING`,两种表形态都合法;但 `emit_cache_hit` 复用的是响应里的**历史** `call_id`,于是同一次缓存命中的重复回放,在普通表上第二次起被 `DO NOTHING` 吞掉、在分区表上**每次都落一行**(`created_at` 由 `DEFAULT now()` 生成,主键不再重复)。逐次尝试行不受影响(每次尝试都是新 `call_id`)。 + +### 3. 权限与不可变性 + +`llm_calls` 按**不可变审计表**对待:写进去的行谁都不许改、不许删,过期数据靠 `DROP PARTITION` 整块消失。 + + + +```sql +GRANT INSERT, SELECT ON llm_calls TO polygateway_app; +GRANT SELECT ON llm_calls TO polygateway_report; +REVOKE UPDATE, DELETE, TRUNCATE ON llm_calls FROM polygateway_app, polygateway_report; +``` + + + +```sql +CREATE FUNCTION llm_calls_reject_mutation() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + RAISE EXCEPTION 'llm_calls 是不可变审计表,% 被拒绝', TG_OP; +END; +$$; +CREATE TRIGGER llm_calls_immutable BEFORE UPDATE OR DELETE ON llm_calls + FOR EACH ROW EXECUTE FUNCTION llm_calls_reject_mutation(); +``` + +触发器**只防误操作,不防恶意**:表属主可以 `ALTER TABLE llm_calls DISABLE TRIGGER llm_calls_immutable` 把它关掉。真正的强制是上一块的 `REVOKE`——权限检查发生在触发器之前,应用角色连触发器都碰不到。要防属主本人,需要的是数据库之外的手段(WAL 归档、只追加的外部存证),不是本表能解决的。 + +`DROP PARTITION` 与 `DETACH PARTITION` 是 DDL,**不会触发**行级触发器,故保留期清理不受这一块影响。 + +### 4. 行级安全与多租户隔离 + + + +```sql +ALTER TABLE llm_calls ENABLE ROW LEVEL SECURITY; +ALTER TABLE llm_calls FORCE ROW LEVEL SECURITY; -- 属主不豁免 +CREATE POLICY llm_calls_app_write ON llm_calls FOR INSERT TO polygateway_app + WITH CHECK (true); +CREATE POLICY llm_calls_app_read ON llm_calls FOR SELECT TO polygateway_app + USING (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')); +CREATE POLICY llm_calls_report_read ON llm_calls FOR SELECT TO polygateway_report + USING (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')); +``` + + + +```sql +CREATE INDEX idx_llm_calls_tenant_created ON llm_calls (tenant_id, created_at); +``` + +`current_setting(..., true)` 的第二参数令 GUC 未设时返回 NULL 而非抛错,外层 `NULLIF` 把空串归一为 NULL——合起来使**未设租户 = 零行**(fail-closed)而不是全部行。索引列序不可颠倒:启用 RLS 后 policy 给每条查询隐式追加 `tenant_id` 等值谓词,它出现在 100% 的谓词里,必然是前导列。分区表上**不能**用 `CREATE INDEX CONCURRENTLY`(PG 不支持在分区父表上并发建索引);父表此时还没有数据,直接建即可,给已有数据的普通表补索引才需要逐个分区 `CONCURRENTLY`。 + +**写侧 policy 为什么是 `WITH CHECK (true)` 而不是等值比较**:库用一个连接池给**所有**租户写遥测,且从不发 `set_config('app.tenant_id', ...)`(源码里没有这条语句)。把写侧也绑到 GUC 上,库的每一条 `INSERT` 都会被 policy 拒绝——而遥测的失败方向是静默降级,表现是逐行 warning + 整表零行。隔离在这个模型里由**读侧**承担:写入方是库自己(可信),读取方才是要隔离的人。若你的调用点保证每次调用都带 `tenant_id`,可把写侧收紧成 `WITH CHECK (tenant_id <> '')`,代价是漏传 `tenant_id` 的调用点会**丢遥测行**(只留一条 warning)。 + +四个陷阱,每一个的失败形态都是**静默的**: + +| 陷阱 | 后果 | +|---|---| +| 表属主默认**豁免** RLS | 只写 `ENABLE` 而漏 `FORCE`,用属主角色连库时隔离形同虚设,且查询一切正常看不出来 | +| `FORCE` 之后属主自己也被 policy 管 | 模板没给 `polygateway_owner` 任何 policy,故它读不到、也写不进任何行——这是有意的(它只用来做 DDL),但别拿它跑报表 | +| 租户上下文必须在**显式事务内**用 `set_config('app.tenant_id', ..., true)` | asyncpg 默认 autocommit,单发 `SET LOCAL` 会当场失效,而 PG **只发 warning 不报错**;表现是 policy 永远拿不到租户 → fail-closed 到零行 | +| 读侧 policy 漏写 `USING` | `FOR SELECT` 的 policy 只认 `USING`;写成 `WITH CHECK` 不报错也不生效,隔离直接落空 | + +### 5. 库本身需要的最小权限 + +按上面的模板部署后,库的连接串用 `polygateway_app`,它需要的权限恰好是下表这些——多一分都不必给: + +| 库会发的语句 | 需要什么 | +|---|---| +| 连库 | 数据库 `CONNECT` + schema `USAGE` | +| `SELECT to_regclass('llm_calls')`、查 `pg_attribute`(列探测) | 无需额外授权(系统 catalog 默认对 `PUBLIC` 可读) | +| `INSERT INTO llm_calls (...)` | 表 `INSERT`;RLS 打开后还须有一条允许写的 policy | +| `CREATE TABLE IF NOT EXISTS`(**仅当表不存在**) | schema `CREATE`。生产建议**不给**:表由 `owner` 先建好,库探测到表在就不发这条 | +| `ALTER TABLE ADD COLUMN`(**仅 `PGW_TELEMETRY_SCHEMA_MODE=auto`**) | 表**属主**——PG 的 `ALTER TABLE` 只认属主,这一项无法单独 `GRANT`。PG 侧缺省就是 `manual`,补列交给 DBA | + +### 6. 合规下游的推荐配置 + +三件事(截断、保留期、访问控制)要一起上才有意义,故给一份可直接照抄的组合,而不是让你自己拼: + +```dotenv +PGW_TELEMETRY_BACKEND=postgres +PGW_TELEMETRY_PG_DSN=postgresql://polygateway_app:...@db:5432/telemetry +PGW_TELEMETRY_SCHEMA_MODE=manual # PG 侧本就是缺省;写出来是为了不依赖缺省 +PGW_TELEMETRY_TEXT_CAP=2000 # 落库正文的字符上限;不设 = 存全文 +``` + +| 层 | 配置 | +|---|---| +| 正文体量 | `PGW_TELEMETRY_TEXT_CAP=2000`(按需调);超出部分头部硬切并附 `…(略 N 字)` | +| 保留期 | 上面的分区模板 + `pg_partman` 的 `retention`,过期分区整块 `DROP` | +| 访问控制 | 上面的三角色 + `REVOKE UPDATE, DELETE` + `FORCE` RLS | +| 存量兜底 | 已经攒成一张大普通表、来不及改造分区时,用 `tools/telemetry_retention.py`(默认 dry-run,`--apply` 才动手;探测到分区表会直接退出让路给 `DROP PARTITION`) | + +**`PGW_TELEMETRY_TEXT_CAP` 的覆盖面必须说清,否则合规判断会出错。** cap 落在四处:`messages` 里每条消息的字符串 `content`、多模态 content 数组中 `type == "text"` 的 part 的 `text`,以及 `response` 与 `thinking` 两列。消息侧的这个面与缓存摘要函数 `digest_messages` 一致——**只碰 `content`**,消息里别的字段一概不碰。所以调用方自己塞进 `tool_calls.function.arguments`、`name` 等字段的内容**不在覆盖范围内**:开了 cap 不等于表里没有全文残留。另需知道:缺省是**不截断**(存全文),而截断之后遥测不再是可复现重放的证据。 + +### 7. SQLite 侧的保留期 + +SQLite 侧**不建议**对着一个大库文件跑 `DELETE` + `VACUUM`,而应**按天/按实验轮转库文件**——`runs/.db`、`runs/.db` 这样,到期直接删文件。这是三个现有下游(Video-Tree-TRM5 / CHSAnalyzer / dissect)天然就有的形态,比删行省事也安全得多:删文件是 O(1) 且不可能删错行,而 `VACUUM` 会重写整库、期间需要一倍磁盘空间,还会把并发写入方挡在外面。 + +`tools/telemetry_retention.py` 的 SQLite 分支是给**存量场景**兜底的——已经攒成一个大库、来不及改轮转时用它,不是推荐路径。 + ## 错误模型(四分类) 一切失败在 transport 层翻译为四类之一,治理行为由分类决定,业务侧不需要判断状态码: @@ -254,6 +401,7 @@ PG 变体的补列语句带 `ADD COLUMN IF NOT EXISTS`,**整段可重复执行** | `PGW_CACHE_BACKEND` | `none` / `memory` / `redis`;非 `none` 时需 `PGW_CACHE_NAMESPACE` + `PGW_CACHE_TTL_S`(须 > 0) | | `PGW_TELEMETRY_BACKEND` | `none` / `sqlite`(需 `PGW_TELEMETRY_SQLITE_PATH`)/ `postgres`(需 `PGW_TELEMETRY_PG_DSN`) | | `PGW_TELEMETRY_SCHEMA_MODE` | 可选:`auto` / `manual`;**不设则按后端派生**(sqlite→`auto`、postgres→`manual`),显式设置则两侧都可覆盖。决定库是否给已存在的旧表自动 `ALTER` 补列,详见[遥测表 schema 与升级纪律](#遥测表-schema-与升级纪律) | +| `PGW_TELEMETRY_TEXT_CAP` | 可选正整数:遥测落库正文的字符上限(作用于每条消息的文本 `content`、多模态 part 的 `text`、`response`、`thinking`);**不设 = 不截断**,详见[合规下游的推荐配置](#6-合规下游的推荐配置) | | `PGW_PRICING_PATH` / `PGW_STRUCTURED_MAX_RETRIES` / `PGW_LEASE_TTL_S` | 可选:价格表(缺省则成本恒 `None`)/ 结构化重问上限(缺省 2)/ permit 租约秒数(缺省 1500,须 ≥ 最大源 `TIMEOUT_S`) | 两个易被忽略的源级键:`MISSING_DONE` 决定 SSE 缺 `[DONE]` 时的处置(`retry` 默认判瞬时重试 / `salvage` 收下已收内容并把用量可信度降为 `estimated`;零内容恒 `retry`,不受该键影响);`EXTRA_BODY` 是该源**恒定**的采样参数(JSON 对象串,并入请求体,优先级低于 `chat(overlay=...)`),禁用键 `model` / `messages` / `stream` / `stream_options` 配了直接报错,OCR 与 EMBED scope 不消费该键(配了忽略并 warning)。 diff --git a/research-wiki/ARCHITECTURE.md b/research-wiki/ARCHITECTURE.md index 3d4653b..9f6b281 100644 --- a/research-wiki/ARCHITECTURE.md +++ b/research-wiki/ARCHITECTURE.md @@ -248,7 +248,7 @@ HTTP API → arq 队列 → worker 协程 脚本 → asyncio.gather 协 **影响**: §5.2 `structured` 参数三档语义、§7.9 重写为阶梯、§6.1 ResultInvalid 行注 D14;缓存写入发生在阶梯通过之后(§7.5 "不固化坏结果"的执行点);反馈模板与策略升级细则留 M1 设计文档。 -### D15 库对下游数据库只做 SELECT/INSERT + 可选 CREATE;改结构与删数据归下游(2026-08-19,issue #13) +### D15 库对下游数据库只做 SELECT/INSERT + 可选 CREATE;改结构与删数据归下游(2026-08-19,issue #13 立,issue #12 补删数据一面) **决策**: 遥测表 `llm_calls` 是**下游的表**,不是库的私有存储。库对它发出的语句只有三类——catalog 探测(PG `to_regclass` + `pg_attribute`,SQLite `PRAGMA table_info`)、显式列名的 `INSERT`、以及表不存在时的 `CREATE TABLE IF NOT EXISTS`;**改结构(`ALTER`)与删数据(`UPDATE`/`DELETE`/`TRUNCATE`/`DROP`)一律归下游**。`ALTER` 保留唯一一个受控出口:`PGW_TELEMETRY_SCHEMA_MODE=auto` 时给已存在的旧表补列,而该档在 PG 侧**不是缺省**(缺省按后端派生: sqlite→auto、postgres→manual)。配套五条 Expand/Contract 承诺:新列只增不删不改名且追加在既有列之后、新列必可空或带非易失常量默认值、`INSERT` 永远显式列名、库从不 `SELECT *` 也从不读回该表数据、写入的冲突处理不绑定具体约束。 @@ -258,9 +258,11 @@ HTTP API → arq 队列 → worker 协程 脚本 → asyncio.gather 协 五条承诺本身是既有实现的**成文化**(零代码变更),但成文后才可被下游依赖——它同时是遥测保留期方案(issue #12)能成立的前提: 下游拿这份 schema 自己加 `PARTITION BY RANGE (created_at)` 建成分区表后,库的 `to_regclass` 探测、列探测与 `INSERT` 路由都照常工作。第五条(冲突处理不绑定约束)是审查带出的**新增**承诺,并伴随一处真实修复,见 §7.8。 +**删数据这一半(2026-08-19,issue #12)**: D15 里 `DELETE`/`TRUNCATE`/`DROP` 归下游,不只是"库不去做",是库连**手段**都不该持有——保留期与访问控制因此以 README 的 DDL 模板加 `tools/telemetry_retention.py` 独立脚本交付,库本体不 import 该脚本,连接串上也不需要任何删权限。这是 (b) 保留期与 (c) 不可变性两条诉求的**权限张力**逼出来的唯一解: 模板建议对应用角色 `REVOKE UPDATE, DELETE ON llm_calls`(按不可变审计表对待),那么过期清理就不可能再由应用角色的 `DELETE` 完成,只能是属主对 `created_at` RANGE 分区的 `DETACH` + `DROP PARTITION`——分区在这里**不可替代**,不是性能偏好(`DROP PARTITION` 是 DDL,同样不触发行级的不可变性触发器,且 O(1)、不留膨胀)。脚本只是存量普通表的兜底: 默认 dry-run,探测到分区表即以退出码 3 让路。库本体在 #12 里唯一的代码面是**预防性**的正文截断(§7.8)——没写进去的数据不需要删,这也是三个子问题里唯一能靠库解决的那个。 + **被否决的备选**: 两侧统一缺省 manual(语义最一致,但现有 SQLite 下游升级即需人工干预,而这些场景根本没有承接手工 SQL 的角色);保持 auto 缺省只加关闭档(默认状态仍是"库在下游生产表上发不受控 DDL",issue 的核心诉求未被满足);Celery 式"自动建表但永不 ALTER、无开关"(SQLite 场景纯净损失,且真想要自动补列的下游没有出路);APScheduler 4.x 式"schema 不认识就拒绝启动"(与「遥测初始化失败必须静默降级」的库铁律正面冲突,不可选)。 -**影响**: §7.8 补列一节按档位重写;新增配置键 `PGW_TELEMETRY_SCHEMA_MODE`(§9)与公共函数 `telemetry_schema_sql`;两个 recorder 新增 keyword-only 必填参数 `auto_migrate`、`GatewaySettings` 新增必填字段 `telemetry_auto_migrate`(缺省规则只写在 config 一处,不与类签名漂移);五条承诺进 README(随包分发)。 +**影响**: §7.8 补列一节按档位重写;新增配置键 `PGW_TELEMETRY_SCHEMA_MODE`(§9)与公共函数 `telemetry_schema_sql`;两个 recorder 新增 keyword-only 必填参数 `auto_migrate`、`GatewaySettings` 新增必填字段 `telemetry_auto_migrate`(缺省规则只写在 config 一处,不与类签名漂移);五条承诺进 README(随包分发)。issue #12 实现同一条边界的"删数据"一面: 新增可选键 `PGW_TELEMETRY_TEXT_CAP` 与遥测正文截断(§7.8、§9),保留期与访问控制走文档模板 + `tools/` 脚本,库的权限面不扩大。 --- @@ -511,6 +513,8 @@ flowchart TB **schema 单一事实源、档位与冲突目标(2026-08-19,issue #13,决策见 D15)**: 列序、两端 DDL、两端补列语句、`INSERT` 构造与缺列告警收敛进 `telemetry/schema.py`——此前在两个 recorder 各存一份,而公共函数 `telemetry_schema_sql` 打印给下游的 SQL 必须与库真正执行的 DDL **同源**,三份必然漂移,漂移的表现是"下游照打印的 SQL 建完表,库仍报缺列"。补列自此由 `PGW_TELEMETRY_SCHEMA_MODE` 控制(三态: 不设按后端派生 sqlite→auto / postgres→manual,显式设置两侧均可覆盖): manual 档一条 DDL 都不发,改为按探测到的现有列**裁剪 `INSERT`**(裁剪是关掉 ALTER 的前提,否则缺列旧表每行写入都被拒 = 遥测全失)并发**一条**点名缺列、附可执行 SQL 的 warning;auto 档行为不变,且补列失败时**不裁剪**(该档承诺"把列补上",补不上就让缺列以逐行 warning 暴露)。**库内执行的补列语句与打印给人的那份是两套文本**: 库内不用 `ADD COLUMN IF NOT EXISTS`(它即便列已存在也先取 ACCESS EXCLUSIVE 锁,故库侧一律先探测后 ALTER),打印的那份带,以保证下游可重复执行。同批把 PG 写入的 `ON CONFLICT (call_id) DO NOTHING` 改为**无冲突目标**的 `ON CONFLICT DO NOTHING`: 带目标的语句要求恰好匹配 `(call_id)` 的唯一约束,而 PG 要求分区表的唯一约束必须包含分区键——按 `created_at` 分区(issue #12)后主键变成 `(call_id, created_at)`,该语句被 PG 直接拒收,而写失败只逐行 warning,表现为分区部署下遥测全线静默丢数据;无目标版本在两种表形态上都合法,普通表上语义逐字等价(表上只有主键这一个唯一约束),SQLite 的 `INSERT OR IGNORE` 本就无目标。 +**正文截断(2026-08-19,issue #12)**: `PGW_TELEMETRY_TEXT_CAP` 给落库正文一个可配置的字符上限,**缺省不设 = 不截断**(人类决策 E-a): 截断后的遥测不再是审计证据,也无法拿原样的请求复现与重放,而这正是既有下游在依赖的行为,默认改动即破坏;代价是 issue 那句"无限期保留全部租户全文不应是默认状态"只被解决一半——默认仍是全文,但下游第一次有了不写全文的手段。截断落在 `TelemetryEmitter._record`(全库唯一遥测出口,单一 helper 铁律)内,位于 `digest_messages` 之后、`json.dumps` 之前,作用面四处: 每条消息的字符串 `content`、多模态 part 中 `type == "text"` 的 `text`、`response`、`thinking`;超出部分头部硬切并附 `…(略 N 字)`。**按每条文本切而不是切整串 JSON**——后者会往不做任何校验的 TEXT 列里写进非法 JSON,让此后一切按 JSON 解析该列的分析全废。**且只产出新对象、绝不就地修改**: `digest_messages` 对非 list 的 `content` 原样透传同一个 dict 对象,就地截断会同时污染调用方持有的 messages、后续重试的请求体与缓存写入的 key 且全程无报错——红线由"cap 开与关两态下 `build_cache_key` 输出逐字节相同"的测试钉死。覆盖面须诚实声明: 只碰 `content`(与 `digest_messages` 处理面一致),调用方放进 `tool_calls.function.arguments` 等字段的内容不在其中。embedding 与 OCR 两条链路各自既有的 200 字符上限保留不动,与新 cap 是取更严者的关系。 + - 后端: `SQLiteRecorder`(默认;WAL + busy_timeout、`INSERT OR IGNORE` 幂等、`asyncio.to_thread` 桥接、初始化/写入失败全降级不冒泡)与 `PostgresRecorder`。 - **单一 helper 铁律**: 遥测调用点收敛为一个内部函数/上下文管理器;Video-Tree 与 GovDoc 各有 4-5 处逐字复制的 `record_llm_call(15 个参数)` 是本条的直接教训。 - 成本: `pricing.py` 维护 model → (input 单价, output 单价, **可选** cached_input 单价) 表,遥测时换算 `cost` 字段;查不到价格记 None 并 warning,**不阻塞调用**。缓存读取单价(2026-07-31,issue #3)只在配置了该档且本次有命中时启用,按 `(prompt - cached) × input + cached × cached_input` 分段计价;**未配该档绝不按经验折扣率猜**,退化为全额输入价(P5)。命中数超过输入总数时按总数夹取并 warning,不产生负成本。 @@ -575,6 +579,7 @@ src/polygateway/ - **装配只有两条路**: `GatewayClient.from_env()`/`from_settings(settings)`(工厂,覆盖 90% 用户;补上三项目每次手写、GovDoc 缺失的"配置→client"一段)或构造函数全量依赖注入(测试/高级用户)。库内部任何组件**不得自读环境变量**(显式优于隐式)。 - 后端选择即配置: 如 `PGW_LIMITER_BACKEND=memory|redis`、`PGW_TELEMETRY_BACKEND=sqlite|postgres`、`PGW_QUOTA_FULL=wait|fail_fast`(命名待 M1 设计文档定稿)。 - **`PGW_TELEMETRY_SCHEMA_MODE=auto|manual`(2026-08-19,issue #13,D15)**: 可选键、**三态**——不设 = 按后端派生(sqlite→auto、postgres→manual),显式设置则两侧都可覆盖。派生只发生在 config 层一处,落到 `GatewaySettings.telemetry_auto_migrate`(无默认值,与既有全部字段一致;`telemetry_backend=none` 时无人消费,归一为 `False`),recorder 的 `auto_migrate` 是 keyword-only **必填**参数——关键行为参数不给默认值(P4),缺省规则也就不会与类签名漂移。 +- **`PGW_TELEMETRY_TEXT_CAP`(2026-08-19,issue #12)**: 可选正整数键、**二态**——不设 = 不截断(缺省)。与相邻的 `SCHEMA_MODE` 不同,这里"未设"本身就是最终答案,没有需要按后端派生的第二种缺省。落到 `GatewaySettings.telemetry_text_cap: int | None`(同样无默认值),`TelemetryEmitter.text_cap` 是 keyword-only 必填参数。值域(`> 0`)在 settings 与 emitter **两处**校验: 前者只管 env 一条路,而"构造函数全量注入"是库承诺的另一条公共装配路,`text_cap=0` 会让每条正文只剩一个省略标记(P5 不得静默)。 --- diff --git a/src/polygateway/client.py b/src/polygateway/client.py index f8352a6..1bfbd0b 100644 --- a/src/polygateway/client.py +++ b/src/polygateway/client.py @@ -136,6 +136,7 @@ class GatewayClient: quota_full: str = "wait", telemetry: TelemetryRecorder | None = None, pricing: PricingTable | None = None, + text_cap: int | None = None, cache: CacheBackend | None = None, cache_namespace: str | None = None, cache_ttl_s: int | None = None, @@ -146,7 +147,11 @@ class GatewayClient: sleep: Any = asyncio.sleep, rng: Any = random.random, ) -> None: - emitter = TelemetryEmitter(telemetry, pricing=pricing) if telemetry is not None else None + emitter = ( + TelemetryEmitter(telemetry, pricing=pricing, text_cap=text_cap) + if telemetry is not None + else None + ) terminal = RetryMW( scope=scope, sources=sources, @@ -312,6 +317,7 @@ class GatewayClient: pricing=PricingTable.from_file(settings.pricing_path) if settings.pricing_path is not None else None, + text_cap=settings.telemetry_text_cap, cache=cache if cache is not None else _build_cache(settings), cache_namespace=settings.cache_namespace, cache_ttl_s=settings.cache_ttl_s, diff --git a/src/polygateway/config.py b/src/polygateway/config.py index 7ab8e8b..8dfe5b2 100644 --- a/src/polygateway/config.py +++ b/src/polygateway/config.py @@ -58,6 +58,8 @@ _TELEMETRY_BACKENDS = frozenset({"sqlite", "postgres", "none"}) # 遥测 schema 档位(issue #13): auto 允许 recorder 给旧表 ALTER 补列,manual 不发 DDL _SCHEMA_MODES = frozenset({"auto", "manual"}) _SCHEMA_MODE_KEY = "PGW_TELEMETRY_SCHEMA_MODE" +# 遥测正文字符上限(issue #12);二态键,未设 = 不截断 +_TEXT_CAP_KEY = "PGW_TELEMETRY_TEXT_CAP" _REDIS_DEPENDENT_BACKENDS = ("limiter_backend", "breaker_backend", "cache_backend") # 背压默认(M1 仅 poll 生效;CHS _BACKOFF_S=0.05 同源) _DEFAULT_STALL_WINDOW_S = 300.0 @@ -139,6 +141,11 @@ class GatewaySettings: # backend=none 时恒 False 这条跨字段不变量则由 `_validate_telemetry` # 把关,对直接构造与 `dataclasses.replace` 同样生效 telemetry_auto_migrate: bool + # 遥测落库正文的字符上限(issue #12);None = 不截断,与本字段出现之前逐字节相同。 + # 缺省不截断是人类决策: 截断后的遥测不再是审计证据、也无法用于复现与重放,而 + # 既有下游正依赖这一行为。值域(> 0)由 `_validate_telemetry` 把关,直接构造、 + # `dataclasses.replace` 与 env 三条路一并覆盖 + telemetry_text_cap: int | None redis_url: str | None pricing_path: str | None structured_max_retries: int @@ -224,7 +231,16 @@ class GatewaySettings: recorder 消费它,True 是个自相矛盾却无害的状态。`from_env` 那条路的派生 已经给出 False,归一化是为了直接构造与 `dataclasses.replace` 也一致—— 不变量挂在构造期,才不用每加一个装配工厂就多一处要同步。 + + `telemetry_text_cap` 的值域则是**报错**而非归一化: 0 与负数都不是"不截断" + 的写法(不截断写 None),把它们悄悄改成 None 等于用默认值掩盖调用方的错误。 + 报错文本同时点出字段名与 env 键名,两条装配路的调用方各看得懂自己那套。 """ + if self.telemetry_text_cap is not None and self.telemetry_text_cap <= 0: + raise ValueError( + f"telemetry_text_cap({_TEXT_CAP_KEY})必须 > 0: {self.telemetry_text_cap};" + "不截断请不设该键(None),0 只会让每条正文退化成一个省略标记" + ) if self.telemetry_backend == "none" and self.telemetry_auto_migrate: object.__setattr__(self, "telemetry_auto_migrate", False) if self.telemetry_backend == "sqlite" and not self.telemetry_sqlite_path: @@ -461,6 +477,7 @@ def _load_pgw(env: Mapping[str, str]) -> dict[str, object]: else None, "telemetry_pg_dsn": _load_pg_dsn(env) if telemetry_backend == "postgres" else None, "telemetry_auto_migrate": auto_migrate, + "telemetry_text_cap": _load_text_cap(env), "redis_url": redis_url, "pricing_path": env.get("PGW_PRICING_PATH") or None, "structured_max_retries": _load_structured_retries(env), @@ -496,6 +513,28 @@ def _load_schema_mode(env: Mapping[str, str], telemetry_backend: str) -> bool: return _load_choice(env, _SCHEMA_MODE_KEY, _SCHEMA_MODES, "auto") == "auto" +def _load_text_cap(env: Mapping[str, str]) -> int | None: + """读 `PGW_TELEMETRY_TEXT_CAP`(issue #12);键未设即 None = 不截断。 + + 与相邻的 `PGW_TELEMETRY_SCHEMA_MODE` 不同,这个键是**二态**而非三态: + "未设"本身就是最终答案(不截断),没有需要按后端派生的第二种缺省,故不必像 + 那边一样先探"设没设"再分两条路取值,读到什么解什么即可。 + + 值域(> 0)刻意不在此处判: 构造期守卫那道同时覆盖直接构造与 + `dataclasses.replace`,而报错文本已点出本键名,env 路的调用方不会看丢。 + + Args: + env: 已合并的环境映射。 + + Returns: + 遥测正文的字符上限;键未设或为空串时返回 None(不截断)。 + """ + found = _first(env, _TEXT_CAP_KEY) + if found is None: + return None + return int(_cast(found[1], "int", found[0])) + + def _strip_dsn_driver(dsn: str) -> str: """剥 SQLAlchemy 风格的 `+driver` 后缀(asyncpg 不认);已干净的原样返回。""" scheme, sep, rest = dsn.partition("://") diff --git a/src/polygateway/embedding.py b/src/polygateway/embedding.py index 18ad334..2616f6c 100644 --- a/src/polygateway/embedding.py +++ b/src/polygateway/embedding.py @@ -104,6 +104,7 @@ class EmbeddingClient: quota_full: str = "wait", telemetry: TelemetryRecorder | None = None, pricing: PricingTable | None = None, + text_cap: int | None = None, batch_size: int, normalize: bool = False, expected_dim: int | None = None, @@ -128,7 +129,9 @@ class EmbeddingClient: self._retry = retry self._bp = backpressure self._quota_full = quota_full - self._emitter = TelemetryEmitter(telemetry, pricing=pricing) if telemetry else None + self._emitter = ( + TelemetryEmitter(telemetry, pricing=pricing, text_cap=text_cap) if telemetry else None + ) self._telemetry = telemetry self._pricing = pricing self._batch_size = batch_size @@ -560,6 +563,9 @@ class EmbeddingClient: pricing=PricingTable.from_file(gw.pricing_path) if gw.pricing_path is not None else None, + # embed 行与 chat 行写同一张 llm_calls;漏传这一条,同表内就一半受控 + # 一半不受控(issue #12) + text_cap=gw.telemetry_text_cap, batch_size=settings.batch_size, normalize=settings.normalize, expected_dim=settings.expected_dim, diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py index 335a495..058637f 100644 --- a/src/polygateway/middleware/telemetry.py +++ b/src/polygateway/middleware/telemetry.py @@ -55,6 +55,44 @@ def _canonical_meta_json(meta: Mapping[str, Any]) -> str: return json.dumps(dict(meta), sort_keys=True, ensure_ascii=False, allow_nan=False) +def _cap_text(text: str, cap: int | None) -> str: + """超出 cap 时头部硬切并附省略标记 `…(略 N 字)`;cap 为 None 原样返回。""" + if cap is None or len(text) <= cap: + return text + return f"{text[:cap]}…(略 {len(text) - cap} 字)" + + +def _cap_part(part: Any, cap: int) -> Any: + """多模态 part 的文本截断;非 `type == "text"` 的 part 原样返回同一对象。""" + if isinstance(part, dict) and part.get("type") == "text" and isinstance(part.get("text"), str): + return {**part, "text": _cap_text(part["text"], cap)} + return part + + +def _cap_messages(messages: list[dict[str, Any]], cap: int | None) -> list[dict[str, Any]]: + """对每条消息的文本 content 与多模态 part 中 type == "text" 的 text 逐条施加 cap。 + + 非字符串 content 原样放行(外部输入形状不可控,遥测路径不得因此抛错)。 + + **只产出新对象,严禁就地修改**: `digest_messages` 对 content 非 list 的消息是 + 原样透传**同一个 dict 对象**(`cache.py:43`),多模态里非 image_url 的 part 同理。 + 就地改它会一并污染调用方持有的 messages、后续重试尝试的请求体与缓存写入的 key, + 且全程无任何报错。 + """ + if cap is None: + return messages + capped: list[dict[str, Any]] = [] + for msg in messages: + content = msg.get("content") + if isinstance(content, str): + capped.append({**msg, "content": _cap_text(content, cap)}) + elif isinstance(content, list): + capped.append({**msg, "content": [_cap_part(part, cap) for part in content]}) + else: + capped.append(msg) + return capped + + @dataclass(frozen=True) class _AttemptUsage: """一次尝试的用量视图;默认值即"失败尝试"档(无用量可言,记 0 并标 unavailable)。 @@ -96,9 +134,25 @@ class _AttemptUsage: class TelemetryEmitter: """从请求与结果组装 24 字段并写入 recorder;一切写失败降级 warning。""" - def __init__(self, recorder: TelemetryRecorder, *, pricing: PricingTable | None = None) -> None: + def __init__( + self, + recorder: TelemetryRecorder, + *, + pricing: PricingTable | None = None, + text_cap: int | None, + ) -> None: + """`text_cap` 无默认值是有意的: 它是关键行为参数,漏传即静默改变落库正文。 + + 本类是库内部类,唯一构造者是三个公共 Client,必填能保证没有一处漏传。 + 同理,值域校验也放在这一处: 三个 Client 的 `text_cap` 全部汇流到这里, + `GatewaySettings` 那道只管 env 一条路,而直接构造 Client 是库承诺的另一 + 条公共装配路——`text_cap=0` 会让每条正文只剩一个省略标记(P5 不得静默)。 + """ + if text_cap is not None and text_cap <= 0: + raise ValueError(f"text_cap 必须 > 0(不截断请传 None): {text_cap}") self._recorder = recorder self._pricing = pricing + self._text_cap = text_cap async def emit_attempt( self, @@ -241,8 +295,12 @@ class TelemetryEmitter: ) else: cost = None - # messages 落库前多模态摘要,与缓存 key 共用同一函数(VT R12) - messages_json = json.dumps(digest_messages(request.messages), ensure_ascii=False) + # messages 落库前多模态摘要,与缓存 key 共用同一函数(VT R12); + # 截断只发生在摘要之后、序列化之前的遥测分支,缓存路径不经过它(issue #12) + messages_json = json.dumps( + _cap_messages(digest_messages(request.messages), self._text_cap), + ensure_ascii=False, + ) await self._recorder.record_llm_call( call_id=call_id, parent_call_id=request.parent_call_id, @@ -251,8 +309,8 @@ class TelemetryEmitter: provider=provider, source_name=source_name, messages=messages_json, - response=response_text, - thinking=thinking, + response=_cap_text(response_text, self._text_cap), + thinking=_cap_text(thinking, self._text_cap), prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, usage_source=usage_source, diff --git a/src/polygateway/ocr.py b/src/polygateway/ocr.py index 5a1b1d1..7f1b151 100644 --- a/src/polygateway/ocr.py +++ b/src/polygateway/ocr.py @@ -109,6 +109,7 @@ class OcrClient: backpressure: BackpressurePolicy, quota_full: str = "wait", telemetry: TelemetryRecorder | None = None, + text_cap: int | None = None, now: Callable[[], float] = time.monotonic, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, rng: Callable[[], float] = random.random, @@ -127,7 +128,7 @@ class OcrClient: self._retry = retry self._bp = backpressure self._quota_full = quota_full - self._emitter = TelemetryEmitter(telemetry) if telemetry else None + self._emitter = TelemetryEmitter(telemetry, text_cap=text_cap) if telemetry else None self._telemetry = telemetry self._memo = SourceCooldownMemo(now=now) self._now = now @@ -572,6 +573,9 @@ class OcrClient: backpressure=gw.backpressure, quota_full=gw.quota_full, telemetry=telemetry if telemetry is not None else _build_telemetry(gw), + # OCR 行与 chat 行写同一张 llm_calls;漏传这一条,同表内就一半受控 + # 一半不受控(issue #12) + text_cap=gw.telemetry_text_cap, ) @classmethod diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index dd427cd..e301c2d 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -14,7 +14,9 @@ import asyncio import json import os import re +from dataclasses import dataclass from datetime import UTC, datetime, timedelta +from pathlib import Path from uuid import uuid4 import pytest @@ -913,3 +915,297 @@ class TestPublishedSchemaScript: await _execute_script(fresh_dsn, script) # 可重复执行: 第二遍不得抛 rerun = [r["column_name"] for r in await _fetch(fresh_dsn, _PHYSICAL_COLUMNS_SQL, schema)] assert rerun == actual # 且第二遍没有偷偷改动表结构 + + +# --------------------------------------------------------------------------- +# issue #12 Task 4: README 的生产部署 DDL 模板,逐条在真实 PG 上执行 +# --------------------------------------------------------------------------- + +# 模板 SQL **只有一份**,在 README 里。测试从 README 解析出来跑,而不是在这里另抄 +# 一份: 抄一份就是两份会各自漂移的东西,而"README 里的 SQL 能跑"这个承诺恰恰只在 +# 同源时才成立(doctest / Rust doc tests / mdbook test 都是这个范式)。 +_README = Path(__file__).resolve().parents[2] / "README.md" + +# 锚点写成 HTML 注释,渲染时不可见,比按章节标题或代码块序号定位稳固得多。 +_TEMPLATE_BLOCK = re.compile(r"\s*\n```sql\n(.*?)\n```", re.DOTALL) + +# 顺序即执行顺序;数量与名字都钉死——解析不到或多出一块必须当场红, +# 绝不能退化成空列表让这条测试变成永远绿的摆设。 +_EXPECTED_TEMPLATE_BLOCKS = ( + "roles", + "table", + "partition", + "grants", + "immutable", + "rls", + "index", +) + +# README 里必须原样保留、由本测试做受控替换的标识符。README 那份是给下游照抄的, +# 故占位符是**合法可执行的具体值**而不是 `` 之类的尖括号洞。 +_TEMPLATE_PLACEHOLDERS = ( + "polygateway_owner", + "polygateway_app", + "polygateway_report", + "CHANGE_ME_APP", + "CHANGE_ME_REPORT", + "SCHEMA public", + "llm_calls_2026_01", + "'2026-01-01 00:00:00+00'", + "'2026-02-01 00:00:00+00'", +) + +# 应用角色在生产里能发的唯一一类写语句(与库的 INSERT 同形,只列 NOT NULL 列) +_TEMPLATE_INSERT = ( + "INSERT INTO llm_calls (call_id, model, provider, source_name, messages, response, " + "prompt_tokens, completion_tokens, usage_source, latency_ms, tenant_id) " + "VALUES ($1, 'm', 'p', 's1', '[]', 'ok', 1, 2, 'measured', 10, $2)" +) + + +def _template_blocks() -> dict[str, str]: + """从 README 解析带锚点的 SQL 块;顺序即文中出现顺序。""" + return dict(_TEMPLATE_BLOCK.findall(_README.read_text(encoding="utf-8"))) + + +@dataclass(frozen=True) +class _TemplateEnv: + """模板部署完成后的现场句柄:三个角色各自的连接串 + 当月分区名。""" + + admin_dsn: str + app_dsn: str + report_dsn: str + schema: str + partition: str + seeded: tuple[str, str] # (tenant-a 的行, tenant-b 的行) + + +def _localize(sql: str, schema: str, roles: dict[str, str], month: datetime) -> str: + """把 README 里给下游照抄的标识符换成本次运行专属的临时对象。 + + 替换规则写在测试里而不是让 README 变得不可直接复制: README 里那份必须是 + 下游 `pip install` 后照抄就能用的,占位符因此都是合法 SQL 值。 + """ + start = month.strftime("%Y-%m-%d %H:%M:%S%z") + end = (month + timedelta(days=32)).replace(day=1).strftime("%Y-%m-%d %H:%M:%S%z") + for placeholder, actual in ( + # 长名在前: 三个角色名互不为前缀,但顺序稳定便于排查 + ("polygateway_owner", roles["owner"]), + ("polygateway_report", roles["report"]), + ("polygateway_app", roles["app"]), + ("CHANGE_ME_APP", _PROBE_PASSWORD), + ("CHANGE_ME_REPORT", _PROBE_PASSWORD), + ("SCHEMA public", f"SCHEMA {schema}"), + ("llm_calls_2026_01", f"llm_calls_{month:%Y_%m}"), + ("'2026-01-01 00:00:00+00'", f"'{start}'"), + ("'2026-02-01 00:00:00+00'", f"'{end}'"), + ): + sql = sql.replace(placeholder, actual) + return sql + + +def _role_dsn(dsn: str, role: str, schema: str) -> str: + low = re.sub(r"//[^@/]+@", f"//{role}:{_PROBE_PASSWORD}@", dsn, count=1) + return _search_path_dsn(low, schema) + + +async def _drop_template_objects(dsn: str, schema: str, roles: dict[str, str]) -> None: + """删净临时 schema 与三个角色(角色是**全局**对象,漏删会跨 run 残留)。""" + import asyncpg + + admin = await asyncpg.connect(dsn, timeout=10) + try: + await admin.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE") + for role in roles.values(): + await admin.execute(f"DROP OWNED BY {role}") + await admin.execute(f"DROP ROLE IF EXISTS {role}") + finally: + await admin.close() + + +@pytest.fixture +async def production_template(dsn): + """在临时 schema + 临时角色上跑完 README 的整套模板,产出可用的三条连接串。 + + 隔离纪律(M4 事故教训)同 `least_privilege_dsn`: 共享的 `public.llm_calls` + 一个字节都不碰,建的 schema / 角色 / 函数 / 分区在 teardown 里删净。 + """ + import asyncpg + + suffix = uuid4().hex[:8] + schema = f"pgwtpl_{suffix}" + roles = { + "owner": f"pgwtpl_owner_{suffix}", + "app": f"pgwtpl_app_{suffix}", + "report": f"pgwtpl_report_{suffix}", + } + month = datetime.now(UTC).replace(day=1, hour=0, minute=0, second=0, microsecond=0) + blocks = _template_blocks() + # 解析不到就地红: 空 dict 会让下面的 for 一句不执行,测试变成"只验证了能连上库" + assert list(blocks) == list(_EXPECTED_TEMPLATE_BLOCKS), ( + f"README 的模板锚点与预期不符: {list(blocks)}" + ) + + seeded = (_cid("tpl-a"), _cid("tpl-b")) + admin_dsn = _search_path_dsn(dsn, schema) + admin = await asyncpg.connect(dsn, timeout=10) + # 权限门放在建任何对象**之前**: `pytest.skip` 抛的是 BaseException, + # 若它在下面的清理块内触发,清理会去 DROP 从未建过的角色而把 skip 盖掉 + can_create = await admin.fetchval( + "SELECT rolcreaterole OR rolsuper FROM pg_roles WHERE rolname = current_user" + ) + if not can_create: + await admin.close() + pytest.skip("当前账号无权建临时角色,跳过生产模板用例") + try: + await admin.execute(f"CREATE SCHEMA {schema}") + await admin.execute(f"SET search_path = {schema}") + # README §2 写明的前置步骤: 先用库自带脚本建出普通表当模子 + await admin.execute(telemetry_schema_sql("postgres")) + for name in _EXPECTED_TEMPLATE_BLOCKS: + await admin.execute(_localize(blocks[name], schema, roles, month)) + # 种两个租户的行(超级用户绕过 RLS,属于布景不属于被测行为) + for call_id, tenant in zip(seeded, ("tenant-a", "tenant-b"), strict=True): + await admin.execute(_TEMPLATE_INSERT, call_id, tenant) + except BaseException: + # 模板 SQL 出错时也必须删净: 建到一半的 schema 会残留一张 llm_calls, + # 而 `TestSchema` 那条按 table_name 查 information_schema 的用例不带 + # schema 过滤,会被残留物在**下一次运行**里以列数不符的形态误伤 + await admin.close() + await _drop_template_objects(dsn, schema, roles) + raise + finally: + if not admin.is_closed(): + await admin.close() + + yield _TemplateEnv( + admin_dsn=admin_dsn, + app_dsn=_role_dsn(dsn, roles["app"], schema), + report_dsn=_role_dsn(dsn, roles["report"], schema), + schema=schema, + partition=f"llm_calls_{month:%Y_%m}", + seeded=seeded, + ) + + admin = await asyncpg.connect(dsn, timeout=10) + try: + await admin.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE") + for role in roles.values(): + await admin.execute(f"DROP OWNED BY {role}") + await admin.execute(f"DROP ROLE IF EXISTS {role}") + finally: + await admin.close() + + +class TestProductionTemplate: + """issue #12: README 的生产部署 DDL 模板必须逐条可执行,且行为与文中描述一致。 + + 模板出错的代价全部落在下游身上(照抄就中招),而人工核对不构成回归保护—— + 改一次 README 就会悄悄失去它。故这里从 README **直接解析** SQL 来执行。 + """ + + def test_readme_exposes_exactly_the_expected_template_blocks(self): + """先钉死解析本身: 锚点没了、改名了、块数变了,这条当场红。 + + 没有它,`production_template` 里解析出空 dict 时下面每条用例都会以 + "表不存在"之类的间接形态失败,真因(README 结构变了)要靠猜。 + """ + blocks = _template_blocks() + assert list(blocks) == list(_EXPECTED_TEMPLATE_BLOCKS) + assert all(sql.strip() for sql in blocks.values()) + joined = "\n".join(blocks.values()) + for placeholder in _TEMPLATE_PLACEHOLDERS: + # 占位符没了 = 受控替换静默失效,测试会去打真实的 polygateway_* 角色 + assert placeholder in joined, f"README 模板缺占位符 {placeholder!r}" + + async def test_app_can_insert_but_cannot_mutate(self, production_template): + """应用角色: INSERT 通过,UPDATE / DELETE 被权限层拒绝(不是被触发器拒)。 + + 权限检查早于行级触发器,故这里拿到的必须是 InsufficientPrivilegeError—— + 若换成触发器的 RaiseError,说明 REVOKE 那一块没生效,而"不可变"就只剩 + 一层属主随手可关的兜底。 + """ + import asyncpg + + env = production_template + conn = await asyncpg.connect(env.app_dsn, timeout=10) + try: + await conn.execute(_TEMPLATE_INSERT, _cid("tpl-app"), "tenant-a") + with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError): + await conn.execute("DELETE FROM llm_calls WHERE call_id = $1", _cid("tpl-app")) + with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError): + await conn.execute("UPDATE llm_calls SET response = 'x'") + finally: + await conn.close() + rows = await _fetch( + env.admin_dsn, "SELECT call_id FROM llm_calls WHERE call_id = $1", _cid("tpl-app") + ) + assert [r["call_id"] for r in rows] == [_cid("tpl-app")] # 写入真落库了 + + async def test_report_can_read_but_cannot_write(self, production_template): + """报表角色: 带租户上下文读得到自己的行,任何写入都被拒。""" + import asyncpg + + env = production_template + conn = await asyncpg.connect(env.report_dsn, timeout=10) + try: + with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError): + await conn.execute(_TEMPLATE_INSERT, _cid("tpl-rpt"), "tenant-a") + async with conn.transaction(): + await conn.execute("SELECT set_config('app.tenant_id', 'tenant-a', true)") + rows = await conn.fetch("SELECT call_id, tenant_id FROM llm_calls") + assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(env.seeded[0], "tenant-a")] + finally: + await conn.close() + + async def test_reads_are_fail_closed_until_the_tenant_guc_is_set(self, production_template): + """未设 `app.tenant_id` → 零行(fail-closed);设了 → 只看得到本租户。 + + 两个断言缺一不可: 只验"设了能看到自己的"漏掉了 GUC 未设时全表泄露, + 只验"未设是零行"则一条永远返回 false 的 policy 也能通过。 + """ + import asyncpg + + env = production_template + conn = await asyncpg.connect(env.app_dsn, timeout=10) + try: + async with conn.transaction(): + assert await conn.fetch("SELECT call_id FROM llm_calls") == [] + async with conn.transaction(): + await conn.execute("SELECT set_config('app.tenant_id', 'tenant-b', true)") + rows = await conn.fetch("SELECT call_id, tenant_id FROM llm_calls") + assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(env.seeded[1], "tenant-b")] + finally: + await conn.close() + + async def test_rows_land_in_the_current_month_partition(self, production_template): + """分区表写入成功,且行确实落进当月分区(不是落进某个兜底分区)。""" + env = production_template + rows = await _fetch( + env.admin_dsn, + "SELECT tableoid::regclass::text AS part FROM llm_calls WHERE call_id = $1", + env.seeded[0], + ) + assert [r["part"].split(".")[-1] for r in rows] == [env.partition] + + async def test_trigger_blocks_delete_while_drop_partition_still_works( + self, production_template + ): + """兜底触发器拦得住 DELETE(连超级用户也拦),却拦不住 DROP PARTITION。 + + 这正是 README 说"清理只能走 DROP PARTITION 而不是 DELETE"的机械化依据: + 既要对应用角色 REVOKE DELETE、又要能清理过期数据,分区是唯一不冲突的解。 + """ + import asyncpg + + env = production_template + conn = await asyncpg.connect(env.admin_dsn, timeout=10) + try: + with pytest.raises(asyncpg.exceptions.RaiseError) as exc: + await conn.execute("DELETE FROM llm_calls WHERE call_id = $1", env.seeded[0]) + assert "不可变审计表" in str(exc.value) + await conn.execute(f"ALTER TABLE llm_calls DETACH PARTITION {env.partition}") + await conn.execute(f"DROP TABLE {env.partition}") + assert await conn.fetchval("SELECT count(*) FROM llm_calls") == 0 + finally: + await conn.close() diff --git a/tests/integration/test_retention_tool_pg.py b/tests/integration/test_retention_tool_pg.py new file mode 100644 index 0000000..7c95ba4 --- /dev/null +++ b/tests/integration/test_retention_tool_pg.py @@ -0,0 +1,298 @@ +"""`tools/telemetry_retention.py` 的 PostgreSQL 分支测试(issue #12 Task 3,真实 PG)。 + +DSN 走 .env `PGW_TELEMETRY_PG_DSN`,缺则 skip。 + +隔离纪律(M4 事故教训): `public.llm_calls` 是与真实批跑共享的表,而本测试跑的是 +一个**会删数据的脚本**——一律在自建的临时 schema 里操作(DSN 挂 search_path), +teardown 只 `DROP SCHEMA ... CASCADE`;分批删除那例另行断言 `public.llm_calls` +的行数前后不变,把"search_path 没生效"这种最坏情况钉成红灯而不是静默删库。 +""" + +from __future__ import annotations + +import os +import re +import subprocess +import sys +from datetime import UTC, datetime, timedelta +from pathlib import Path +from uuid import uuid4 + +import pytest +from dotenv import dotenv_values + +from polygateway.telemetry.schema import PG_DDL + +_ROOT = Path(__file__).resolve().parents[2] +_SCRIPT = _ROOT / "tools" / "telemetry_retention.py" + +_INSERT = ( + "INSERT INTO llm_calls (call_id, model, provider, source_name, messages, response, " + "prompt_tokens, completion_tokens, usage_source, latency_ms, tenant_id, created_at) " + "VALUES ($1, 'm', 'p', 's1', '[]', 'ok', 1, 2, 'measured', 10, $2, $3)" +) + + +def _partitioned_ddl() -> str: + """由库的真实 `PG_DDL` 派生一份 RANGE 分区版建表语句。 + + 不另抄一份 DDL: 抄的那份与库的 schema 必然漂移,而漂移后本测试验的就不再是 + "库建的表被做成分区后脚本认不认得"。两处改动都是分区表的**硬性要求**—— + 分区表上的唯一约束必须包含分区键,故 `call_id` 单列主键不再合法。 + """ + body, count = re.subn( + r"call_id(\s+)TEXT PRIMARY KEY", r"call_id\1TEXT NOT NULL", PG_DDL, count=1 + ) + if count != 1: + raise AssertionError("PG_DDL 的 call_id 主键声明形态已变,分区版 DDL 需同步") + body = body.strip().rstrip(";").strip() + if not body.endswith(")"): + raise AssertionError("PG_DDL 结尾形态已变,分区版 DDL 需同步") + return ( + f"{body[:-1].rstrip()},\n" + " PRIMARY KEY (call_id, created_at)\n" + ") PARTITION BY RANGE (created_at)" + ) + + +def _dsn_value() -> str | None: + merged = {**dotenv_values(".env"), **os.environ} + raw = merged.get("PGW_TELEMETRY_PG_DSN") + if not raw: + return None + scheme, sep, rest = raw.partition("://") + return f"{scheme.partition('+')[0]}{sep}{rest}" + + +def _search_path_dsn(dsn: str, schema: str) -> str: + sep = "&" if "?" in dsn else "?" + return f"{dsn}{sep}options=-csearch_path%3D{schema}" + + +def _stamp(delta: timedelta) -> datetime: + return datetime.now(UTC) + delta + + +def _run(*args: str, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, str(_SCRIPT), *args], + capture_output=True, + text=True, + cwd=_ROOT, + env=env, + timeout=120, + ) + + +@pytest.fixture +async def dsn(): + value = _dsn_value() + if value is None: + pytest.skip("PGW_TELEMETRY_PG_DSN 未配置") + # 隔离守卫: 该实例有 app/chs_prod 等在用库,只许打 polygateway 专用库 + if not value.rstrip("/").endswith("/polygateway"): + pytest.fail(f"保留期脚本测试只允许连 polygateway 专用库,当前 DSN 库名不符: {value!r}") + return value + + +async def _make_schema(dsn_value: str, prefix: str, ddl: str, extra: tuple[str, ...] = ()) -> str: + import asyncpg + + name = f"pgwret_{prefix}_{uuid4().hex[:8]}" + conn = await asyncpg.connect(dsn_value, timeout=10) + try: + await conn.execute(f"CREATE SCHEMA {name}") + await conn.execute(f"SET search_path = {name}") + await conn.execute(ddl) + for statement in extra: + await conn.execute(statement) + finally: + await conn.close() + return name + + +async def _drop_schema(dsn_value: str, name: str) -> None: + import asyncpg + + conn = await asyncpg.connect(dsn_value, timeout=10) + try: + await conn.execute(f"DROP SCHEMA {name} CASCADE") + finally: + await conn.close() + + +async def _seed(schema_dsn: str, rows: list[tuple[str, str, datetime]]) -> None: + import asyncpg + + conn = await asyncpg.connect(schema_dsn, timeout=10) + try: + await conn.executemany(_INSERT, rows) + finally: + await conn.close() + + +async def _call_ids(schema_dsn: str) -> list[str]: + import asyncpg + + conn = await asyncpg.connect(schema_dsn, timeout=10) + try: + rows = await conn.fetch("SELECT call_id FROM llm_calls ORDER BY call_id") + finally: + await conn.close() + return [r["call_id"] for r in rows] + + +async def _public_count(dsn_value: str) -> int: + """共享表的行数;本测试全程不得让它变动一行。""" + import asyncpg + + conn = await asyncpg.connect(dsn_value, timeout=10) + try: + if await conn.fetchval("SELECT to_regclass('public.llm_calls')") is None: + return -1 + return await conn.fetchval("SELECT COUNT(*) FROM public.llm_calls") + finally: + await conn.close() + + +@pytest.fixture +async def partitioned_schema(dsn): + """临时 schema 内的**分区表**: 脚本必须认出它并让路给 DROP PARTITION。""" + name = await _make_schema( + dsn, + "part", + _partitioned_ddl(), + extra=( + "CREATE TABLE llm_calls_all PARTITION OF llm_calls " + "FOR VALUES FROM ('2000-01-01') TO ('2100-01-01')", + ), + ) + yield _search_path_dsn(dsn, name), name + await _drop_schema(dsn, name) + + +@pytest.fixture +async def plain_schema(dsn): + """临时 schema 内的普通表: 存量场景,脚本的分批 DELETE 兜底路径。""" + name = await _make_schema(dsn, "plain", PG_DDL) + yield _search_path_dsn(dsn, name), name + await _drop_schema(dsn, name) + + +class TestPartitionedTarget: + async def test_partitioned_table_exits_three_without_deleting_anything( + self, partitioned_schema + ): + schema_dsn, schema = partitioned_schema + await _seed( + schema_dsn, + [ + ("part-old-1", "", _stamp(timedelta(days=-30))), + ("part-old-2", "acme", _stamp(timedelta(days=-20))), + ], + ) + + # 带 --apply 跑: 危险的那条路径必须在真正删之前就被分区探测拦住 + result = _run( + "--backend", "postgres", "--dsn", schema_dsn, "--older-than-days", "7", "--apply" + ) + + assert result.returncode == 3, (result.stdout, result.stderr) + combined = result.stdout + result.stderr + assert "DROP PARTITION" in combined + assert "DETACH" in combined + assert await _call_ids(schema_dsn) == ["part-old-1", "part-old-2"] + # 脚本必须报出它解析到的**限定表名**: 这是"我删的到底是哪张表"的唯一凭据 + assert f"{schema}.llm_calls" in result.stdout + + +class TestPlainTableBatches: + async def test_apply_deletes_only_expired_rows_in_batches(self, plain_schema, dsn): + schema_dsn, schema = plain_schema + before_public = await _public_count(dsn) + await _seed( + schema_dsn, + [ + ("old-1", "", _stamp(timedelta(days=-40))), + ("old-2", "acme", _stamp(timedelta(days=-30))), + ("old-3", "acme", _stamp(timedelta(days=-20))), + ("old-4", "acme", _stamp(timedelta(days=-15))), + ("old-5", "", _stamp(timedelta(days=-10))), + ("fresh-1", "acme", _stamp(timedelta(days=-1))), + ("fresh-2", "", _stamp(timedelta(hours=-1))), + ], + ) + + result = _run( + "--backend", + "postgres", + "--dsn", + schema_dsn, + "--older-than-days", + "7", + "--apply", + "--batch-size", + "2", + ) + + assert result.returncode == 0, (result.stdout, result.stderr) + assert await _call_ids(schema_dsn) == ["fresh-1", "fresh-2"] + assert f"{schema}.llm_calls" in result.stdout + assert "将删除行数: 5" in result.stdout + assert "'acme': 3" in result.stdout + # 5 行 / 每批 2 行 = 3 批,每批各自提交;批次行必须真的出现三条 + assert "批次 1" in result.stdout + assert "批次 3" in result.stdout + assert "批次 4" not in result.stdout + assert "已删除 5 行" in result.stdout + assert await _public_count(dsn) == before_public + + async def test_dry_run_on_a_plain_table_deletes_nothing(self, plain_schema): + schema_dsn, _ = plain_schema + await _seed(schema_dsn, [("old-1", "acme", _stamp(timedelta(days=-30)))]) + + result = _run("--backend", "postgres", "--dsn", schema_dsn, "--older-than-days", "7") + + assert result.returncode == 0, (result.stdout, result.stderr) + assert "将删除行数: 1" in result.stdout + assert "dry-run" in result.stdout + assert await _call_ids(schema_dsn) == ["old-1"] + + +class TestMissingAsyncpg: + async def test_missing_asyncpg_exits_two_without_touching_rows(self, plain_schema, tmp_path): + """缺 asyncpg 必须明确报错退出(码 2),不静默降级——这是运维工具不是库路径。 + + 用一个只 `raise ImportError` 的临时 `asyncpg.py` 挂进子进程的 PYTHONPATH 构造该 + 场景: 脚本跑在子进程里,monkeypatch 对它无效。DSN 用**真实可连**的临时 schema, + 这样"没有导入守卫"的实现会走通并退出 0,而不是碰巧也退出 2 而假绿。 + """ + schema_dsn, _ = plain_schema + await _seed(schema_dsn, [("old-1", "acme", _stamp(timedelta(days=-30)))]) + stub = tmp_path / "stub" + stub.mkdir() + (stub / "asyncpg.py").write_text( + 'raise ImportError("asyncpg 未安装(测试构造)")\n', encoding="utf-8" + ) + env = { + **os.environ, + "PYTHONPATH": os.pathsep.join( + [str(stub), *([p] if (p := os.environ.get("PYTHONPATH")) else [])] + ), + } + + result = _run( + "--backend", + "postgres", + "--dsn", + schema_dsn, + "--older-than-days", + "7", + "--apply", + env=env, + ) + + assert result.returncode == 2, (result.stdout, result.stderr) + assert "asyncpg" in result.stderr + assert "pip install" in result.stderr + assert await _call_ids(schema_dsn) == ["old-1"] diff --git a/tests/unit/test_cache.py b/tests/unit/test_cache.py index c23340c..1b28405 100644 --- a/tests/unit/test_cache.py +++ b/tests/unit/test_cache.py @@ -9,7 +9,8 @@ import pytest from polygateway.backends.memory.cache import InMemoryCache from polygateway.errors import ResultInvalidError, TransientError from polygateway.middleware.cache import CacheMW, build_cache_key, digest_messages -from polygateway.types import ChatRequest, LLMResponse +from polygateway.middleware.telemetry import TelemetryEmitter +from polygateway.types import ChatRequest, LLMResponse, SourceConfig _MSGS = [{"role": "user", "content": "hi"}] @@ -327,3 +328,51 @@ class TestStructuredRehydration: key = build_cache_key("m", _MSGS, "proj", None) raw = await backend.get(key) assert raw is not None and "structured_data" not in json.loads(raw) + + +class TestTelemetryCapDoesNotPoisonTheCacheKey: + """红线之一(issue #12): 遥测截断绝不能改到缓存 key。 + + `digest_messages` 对 content 非 list 的消息**原样透传同一个 dict 对象** + (本文件上方公式测试依赖的也是这份对象),遥测拿到的与算 key 用的是同一份。 + 就地截断会让同一组 messages 在遥测前后算出两个不同的 key——全量 miss、 + 且没有任何报错。故这里测的是"截断没有就地改掉调用方的对象",不只是 + "截断函数是纯的"。 + """ + + class _Rows: + def __init__(self): + self.rows = [] + + async def record_llm_call(self, **fields): + self.rows.append(fields) + + async def test_key_is_byte_identical_across_a_capped_emit(self): + messages = [ + {"role": "user", "content": "合同正文" * 31}, + {"role": "user", "content": [{"type": "text", "text": "标书正文" * 30}]}, + ] + before = build_cache_key("m", messages, "proj", None) + + rec = self._Rows() + await TelemetryEmitter(rec, text_cap=8).emit_attempt( + request=ChatRequest(messages=messages), + source=SourceConfig( + name="s1", + provider="p", + base_url="https://gw.example/v1", + api_key="sk", + model="m", + timeout_s=10.0, + ), + call_id="c", + latency_ms=1, + response=_resp(), + error=None, + ) + # 截断确实发生了(否则本用例恒真) + logged = json.loads(rec.rows[0]["messages"]) + assert "(略 116 字)" in logged[0]["content"] + assert "(略 112 字)" in logged[1]["content"][0]["text"] + + assert build_cache_key("m", messages, "proj", None) == before diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 6575071..302e644 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -351,6 +351,97 @@ class TestFactories: assert isinstance(client, GatewayClient) +class TestTelemetryTextCapWiring: + """`PGW_TELEMETRY_TEXT_CAP` 必须走通全部三条 `from_settings` 装配路(issue #12)。 + + 三条链路写的是**同一张** `llm_calls` 表:只接通 chat,embed 与 OCR 的行就 + 永远不受 cap 约束,同表内一半受控一半不受控——那正是本 issue 要消灭的状态。 + """ + + _CAP_ENV = dict(_ENV, PGW_TELEMETRY_TEXT_CAP="8") + _OCR_CAP_ENV = { + "OCR__MONKEY__1__BASE_URL": "http://10.77.0.20:7866", + "OCR__MONKEY__1__API_KEY": "none", + "OCR__MONKEY__1__MODEL": "monkey-ocr", + "OCR__MONKEY__1__TIMEOUT_S": "120", + "LLM_MAX_RETRIES": "3", + "LLM_RETRY_BASE_DELAY": "2.0", + "LLM_RETRY_MAX_DELAY": "30.0", + "LLM_CIRCUIT_BREAKER_THRESHOLD": "5", + "LLM_CIRCUIT_BREAKER_COOLDOWN": "60", + "PGW_CACHE_BACKEND": "none", + "PGW_TELEMETRY_BACKEND": "none", + "PGW_TELEMETRY_TEXT_CAP": "8", + } + + def test_gateway_from_settings_wires_the_cap(self): + settings = GatewaySettings.from_env("LLM", env=self._CAP_ENV) + client = GatewayClient.from_settings(settings, telemetry=_MemoryRecorder()) + assert client._terminal._emitter._text_cap == 8 + # 对照组: 不设该键时 emitter 拿到的必须是 None,否则 8 可能是硬编码来的 + unset = GatewayClient.from_settings( + GatewaySettings.from_env("LLM", env=_ENV), telemetry=_MemoryRecorder() + ) + assert unset._terminal._emitter._text_cap is None + + def test_embedding_from_settings_wires_the_cap(self): + from polygateway.config import EmbeddingSettings + from polygateway.embedding import EmbeddingClient + + gateway = GatewaySettings.from_env("LLM", env=self._CAP_ENV) + client = EmbeddingClient.from_settings( + EmbeddingSettings(gateway=gateway, batch_size=2), telemetry=_MemoryRecorder() + ) + assert client._emitter._text_cap == 8 + unset = EmbeddingClient.from_settings( + EmbeddingSettings(gateway=GatewaySettings.from_env("LLM", env=_ENV), batch_size=2), + telemetry=_MemoryRecorder(), + ) + assert unset._emitter._text_cap is None + + def test_ocr_from_settings_wires_the_cap(self): + from polygateway.config import OcrSettings + from polygateway.ocr import OcrClient + + settings = OcrSettings.from_env("OCR", env=dict(self._OCR_CAP_ENV)) + client = OcrClient.from_settings(settings, telemetry=_MemoryRecorder()) + assert client._emitter._text_cap == 8 + no_cap = dict(self._OCR_CAP_ENV) + no_cap.pop("PGW_TELEMETRY_TEXT_CAP") + unset = OcrClient.from_settings( + OcrSettings.from_env("OCR", env=no_cap), telemetry=_MemoryRecorder() + ) + assert unset._emitter._text_cap is None + + async def test_capped_body_reaches_the_recorder_end_to_end(self, monkeypatch): + """装配路通了还不够: 真跑一次 chat,落库的 messages 与 response 确已截断。 + + `from_settings` 自建 transport(没有 client_factory 入口),故在装配点 + 换掉该类以接上 MockTransport——洋葱其余各层仍是 `from_settings` 装的真件。 + """ + recorder = _MemoryRecorder() + long_text = "甲乙丙丁戊己庚辛壬癸" # 10 字,cap=8 → 略 2 字 + monkeypatch.setattr( + "polygateway.client.OpenAICompatTransport", + lambda **kwargs: OpenAICompatTransport( + client_factory=lambda source: httpx.AsyncClient( + transport=httpx.MockTransport(lambda request: _sse(content=long_text)) + ) + ), + ) + settings = GatewaySettings.from_env("LLM", env=self._CAP_ENV) + async with GatewayClient.from_settings(settings, telemetry=recorder) as client: + await client.chat([{"role": "user", "content": long_text}]) + row = recorder.rows[-1] + assert json.loads(row["messages"])[0]["content"] == "甲乙丙丁戊己庚辛…(略 2 字)" + assert row["response"] == "甲乙丙丁戊己庚辛…(略 2 字)" + + def test_non_positive_cap_rejected_on_the_direct_construction_path(self): + """直接构造是库承诺的另一条公共装配路;cap=0 会让每条正文只剩省略标记。""" + with pytest.raises(ValueError, match="text_cap"): + _client(telemetry=_MemoryRecorder(), text_cap=0) + + class TestSharedBackend: async def test_two_clients_share_global_concurrency_gate(self): """VT R5: 两个逻辑角色显式注入同一 limiter → 共享全局并发闸。""" diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 0bf1d27..4f9772f 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -386,6 +386,33 @@ class TestTelemetrySchemaMode: ) +class TestTelemetryTextCap: + """`PGW_TELEMETRY_TEXT_CAP`(issue #12): 二态键,未设即不截断。 + + 与 `PGW_TELEMETRY_SCHEMA_MODE` 的三态不同,这里"未设"本身就是最终答案 + (不截断),没有需要按后端派生的第二种缺省,故不走 `_load_choice` 那套。 + """ + + def test_unset_key_means_no_truncation(self): + """缺省不截断是人类决策: 截断后的遥测不再是审计证据、无法复现重放。""" + assert GatewaySettings.from_env("LLM", env=_env()).telemetry_text_cap is None + + def test_positive_value_is_parsed_as_int(self): + s = GatewaySettings.from_env("LLM", env=_env(PGW_TELEMETRY_TEXT_CAP="2000")) + assert s.telemetry_text_cap == 2000 + + @pytest.mark.parametrize("raw", ["0", "-1"]) + def test_non_positive_rejected(self, raw): + """0 会把每条正文退化成一个省略标记,负数无意义;都不是"不截断"的写法。""" + with pytest.raises(ValueError, match="PGW_TELEMETRY_TEXT_CAP"): + GatewaySettings.from_env("LLM", env=_env(PGW_TELEMETRY_TEXT_CAP=raw)) + + def test_non_integer_rejected_naming_the_env_key(self): + """报错须点出 env 键名: 这条路的调用方看得懂的是键名,不是字段名。""" + with pytest.raises(ValueError, match="PGW_TELEMETRY_TEXT_CAP"): + GatewaySettings.from_env("LLM", env=_env(PGW_TELEMETRY_TEXT_CAP="2k")) + + class TestOcrSettings: """M3 OcrSettings(设计 §3.4): 复用 GatewaySettings,无 OCR 专用键。""" @@ -622,6 +649,12 @@ class TestCrossFieldInvariants: # —— 标量域 —— + def test_non_positive_text_cap_rejected(self): + """env 路只覆盖 from_env;直接构造与 replace 同样能把 0 传进来(issue #12)。""" + base = self._base() + with pytest.raises(ValueError, match="telemetry_text_cap"): + dataclasses.replace(base, telemetry_text_cap=0) + def test_negative_structured_retries_rejected(self): base = self._base() with pytest.raises(ValueError, match="structured_max_retries"): diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py index 2e2067c..01f2845 100644 --- a/tests/unit/test_openai_compat.py +++ b/tests/unit/test_openai_compat.py @@ -113,7 +113,7 @@ async def _recorded_cost(result, source): source_name=source.name, usage_source=result.usage_source, ) - await TelemetryEmitter(recorder, pricing=_PRICING).emit_attempt( + await TelemetryEmitter(recorder, pricing=_PRICING, text_cap=None).emit_attempt( request=ChatRequest(messages=[{"role": "user", "content": "hi"}]), source=source, call_id="cid-1", diff --git a/tests/unit/test_pricing.py b/tests/unit/test_pricing.py index b4bb24d..cdd76b4 100644 --- a/tests/unit/test_pricing.py +++ b/tests/unit/test_pricing.py @@ -169,7 +169,7 @@ def _source(model="qwen-max"): class TestEmitterCost: async def test_success_row_costed(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, pricing=_TABLE) + emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None) await emitter.emit_attempt( request=_REQ, source=_source(), @@ -182,13 +182,13 @@ class TestEmitterCost: async def test_cache_hit_row_costs_zero(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, pricing=_TABLE) + emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None) await emitter.emit_cache_hit(request=_REQ, response=_resp(cache_hit=True)) assert rec.rows[0]["cost"] == 0.0 async def test_failure_row_cost_none(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, pricing=_TABLE) + emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None) await emitter.emit_attempt( request=_REQ, source=_source(), @@ -201,7 +201,7 @@ class TestEmitterCost: async def test_unknown_model_none_without_blocking(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, pricing=_TABLE) + emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None) await emitter.emit_attempt( request=_REQ, source=_source(model="mystery"), @@ -215,7 +215,7 @@ class TestEmitterCost: async def test_no_pricing_keeps_none(self): """未注入价格表 = M1 现状: cost 恒 None(回归)。""" rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec) + emitter = TelemetryEmitter(rec, text_cap=None) await emitter.emit_attempt( request=_REQ, source=_source(), diff --git a/tests/unit/test_retention_tool.py b/tests/unit/test_retention_tool.py new file mode 100644 index 0000000..249ea5a --- /dev/null +++ b/tests/unit/test_retention_tool.py @@ -0,0 +1,273 @@ +"""`tools/telemetry_retention.py` 的 SQLite 分支测试(issue #12 Task 3)。 + +一律经 `subprocess` 跑真实脚本 + 真实临时 SQLite 库文件: 脚本是独立运维工具、 +不被库 import,用 monkeypatch 或直接 import 私有函数测出来的"通过"与运维实际 +执行的那条路径不是同一条(退出码、argparse 行为、stdout 全都测不到)。 +""" + +from __future__ import annotations + +import sqlite3 +import subprocess +import sys +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from polygateway.telemetry.schema import SQLITE_DDL + +_ROOT = Path(__file__).resolve().parents[2] +_SCRIPT = _ROOT / "tools" / "telemetry_retention.py" +_TIME_FORMAT = "%Y-%m-%d %H:%M:%S" + +# 库写入 SQLite 的 created_at 是 UTC 的 'YYYY-MM-DD HH:MM:SS' 文本(schema 的 +# DEFAULT (datetime('now'))),测试数据必须同款,否则字符串比较的口径就假了 +_INSERT = ( + "INSERT INTO llm_calls (call_id, model, provider, source_name, messages, response, " + "prompt_tokens, completion_tokens, usage_source, latency_ms, tenant_id, created_at) " + "VALUES (?, 'm', 'p', 's1', '[]', 'ok', 1, 2, 'measured', 10, ?, ?)" +) + + +def _stamp(delta: timedelta) -> str: + return (datetime.now(UTC) + delta).strftime(_TIME_FORMAT) + + +def _make_db(tmp_path: Path, rows: list[tuple[str, str, str]]) -> Path: + """按库的真实 DDL 建临时库并灌入 (call_id, tenant_id, created_at) 三元组。""" + path = tmp_path / "telemetry.db" + conn = sqlite3.connect(path) + try: + conn.executescript(SQLITE_DDL) + conn.executemany(_INSERT, rows) + conn.commit() + finally: + conn.close() + return path + + +def _run(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, str(_SCRIPT), *args], + capture_output=True, + text=True, + cwd=_ROOT, + timeout=120, + ) + + +def _rows(path: Path) -> list[str]: + conn = sqlite3.connect(path) + try: + return [r[0] for r in conn.execute("SELECT call_id FROM llm_calls ORDER BY call_id")] + finally: + conn.close() + + +def _aged_db(tmp_path: Path) -> Path: + return _make_db( + tmp_path, + [ + ("old-1", "", _stamp(timedelta(days=-30))), + ("old-2", "acme", _stamp(timedelta(days=-20))), + ("old-3", "acme", _stamp(timedelta(days=-10))), + ("fresh-1", "acme", _stamp(timedelta(days=-1))), + ("fresh-2", "", _stamp(timedelta(hours=-1))), + ], + ) + + +class TestSqliteDryRun: + def test_dry_run_deletes_nothing_and_reports_counts_range_and_tenants(self, tmp_path): + """缺省(不带 --apply)是 dry-run: 一行不删,且报出足以判断"删的是不是我想删的"的三样。""" + path = _aged_db(tmp_path) + + result = _run("--backend", "sqlite", "--path", str(path), "--older-than-days", "7") + + assert result.returncode == 0, result.stderr + assert _rows(path) == ["fresh-1", "fresh-2", "old-1", "old-2", "old-3"] + assert "将删除行数: 3" in result.stdout + assert "created_at 范围:" in result.stdout + assert "按 tenant_id 分布" in result.stdout + # 空串是"未归属"的哨兵而非 NULL,repr 让它在输出里不被误读成缺失 + assert "'acme': 2" in result.stdout + assert "'': 1" in result.stdout + assert "dry-run" in result.stdout + + def test_dry_run_reports_the_actual_created_at_window(self, tmp_path): + """时间范围报的必须是**命中行**的窗口,不是全表的。""" + path = _aged_db(tmp_path) + + result = _run("--backend", "sqlite", "--path", str(path), "--older-than-days", "7") + + conn = sqlite3.connect(path) + try: + low, high = conn.execute( + "SELECT MIN(created_at), MAX(created_at) FROM llm_calls WHERE call_id LIKE 'old-%'" + ).fetchone() + finally: + conn.close() + assert f"{low} ~ {high}" in result.stdout + + +class TestSqliteApply: + def test_apply_removes_only_expired_rows(self, tmp_path): + path = _aged_db(tmp_path) + + result = _run( + "--backend", "sqlite", "--path", str(path), "--older-than-days", "7", "--apply" + ) + + assert result.returncode == 0, result.stderr + assert _rows(path) == ["fresh-1", "fresh-2"] + assert "已删除 3 行" in result.stdout + + def test_older_than_days_zero_deletes_everything_before_now(self, tmp_path): + """N=0 的边界: 截止时刻即"此刻",此刻之前的全删、之后的(未来戳)留下。""" + path = _make_db( + tmp_path, + [ + ("past", "", _stamp(timedelta(seconds=-5))), + ("future", "", _stamp(timedelta(hours=1))), + ], + ) + + result = _run( + "--backend", "sqlite", "--path", str(path), "--older-than-days", "0", "--apply" + ) + + assert result.returncode == 0, result.stderr + assert _rows(path) == ["future"] + + def test_vacuum_with_apply_rewrites_the_file(self, tmp_path): + path = _aged_db(tmp_path) + + result = _run( + "--backend", + "sqlite", + "--path", + str(path), + "--older-than-days", + "7", + "--apply", + "--vacuum", + ) + + assert result.returncode == 0, result.stderr + assert "VACUUM" in result.stdout + assert _rows(path) == ["fresh-1", "fresh-2"] + + def test_deleting_from_a_db_without_the_table_is_a_backend_failure(self, tmp_path): + """连得上但没有 llm_calls: 属"目标不可用",退出码 2 且**不**静默当成 0 行。""" + path = tmp_path / "empty.db" + sqlite3.connect(path).close() + + result = _run("--backend", "sqlite", "--path", str(path), "--older-than-days", "7") + + assert result.returncode == 2 + assert "llm_calls" in result.stderr + + def test_missing_db_file_exits_two(self, tmp_path): + result = _run( + "--backend", "sqlite", "--path", str(tmp_path / "nope.db"), "--older-than-days", "7" + ) + + assert result.returncode == 2 + assert "nope.db" in result.stderr + + +class TestUsageErrors: + """参数层的一切错误都是退出码 1(argparse 默认的 2 已被本脚本改写,2 留给连接失败)。""" + + def test_sqlite_with_dsn_exits_one(self, tmp_path): + result = _run( + "--backend", + "sqlite", + "--path", + str(tmp_path / "x.db"), + "--dsn", + "postgresql://x/y", + "--older-than-days", + "7", + ) + + assert result.returncode == 1 + assert "--dsn" in result.stderr + + def test_sqlite_without_path_exits_one(self): + result = _run("--backend", "sqlite", "--older-than-days", "7") + + assert result.returncode == 1 + assert "--path" in result.stderr + + def test_sqlite_with_batch_size_exits_one(self, tmp_path): + result = _run( + "--backend", + "sqlite", + "--path", + str(tmp_path / "x.db"), + "--older-than-days", + "7", + "--batch-size", + "10", + ) + + assert result.returncode == 1 + assert "--batch-size" in result.stderr + + def test_postgres_with_vacuum_exits_one(self): + result = _run( + "--backend", + "postgres", + "--dsn", + "postgresql://x/y", + "--older-than-days", + "7", + "--apply", + "--vacuum", + ) + + assert result.returncode == 1 + assert "--vacuum" in result.stderr + + def test_vacuum_without_apply_exits_one(self, tmp_path): + result = _run( + "--backend", + "sqlite", + "--path", + str(tmp_path / "x.db"), + "--older-than-days", + "7", + "--vacuum", + ) + + assert result.returncode == 1 + assert "--apply" in result.stderr + + def test_missing_older_than_days_exits_one(self, tmp_path): + result = _run("--backend", "sqlite", "--path", str(tmp_path / "x.db")) + + assert result.returncode == 1 + + def test_negative_older_than_days_exits_one(self, tmp_path): + result = _run( + "--backend", "sqlite", "--path", str(tmp_path / "x.db"), "--older-than-days", "-1" + ) + + assert result.returncode == 1 + assert "--older-than-days" in result.stderr + + def test_unknown_backend_exits_one(self, tmp_path): + result = _run("--backend", "mysql", "--path", str(tmp_path / "x.db")) + + assert result.returncode == 1 + + +class TestHelp: + def test_help_names_the_maintenance_role_and_the_recommended_path(self): + """帮助文本是运维唯一会读的文档,权限口径与"推荐不是 DELETE"必须在里面。""" + result = _run("--help") + + assert result.returncode == 0 + assert "维护角色" in result.stdout + assert "REVOKE" in result.stdout + assert "PARTITION" in result.stdout diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index a04fb0d..30d938b 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -1,6 +1,7 @@ """遥测子系统测试: SQLiteRecorder(22 列)+ TelemetryEmitter 单一 helper + TelemetryMW。""" import asyncio +import copy import json import os import sqlite3 @@ -9,11 +10,27 @@ from pathlib import Path import pytest +from polygateway.backends.memory.breaker import InMemoryGate +from polygateway.backends.memory.limiter import InMemoryLimiter +from polygateway.embedding import EmbeddingClient from polygateway.errors import CircuitOpenError, RequestRejectedError +from polygateway.middleware.cache import digest_messages from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW +from polygateway.ocr import OcrClient from polygateway.pricing import ModelPrice, PricingTable +from polygateway.sources import RoundRobinSelector from polygateway.telemetry.sqlite import SQLiteRecorder -from polygateway.types import ChatRequest, LLMResponse, SourceConfig +from polygateway.types import ( + BackpressurePolicy, + BreakerConfig, + ChatRequest, + EmbeddingTransportResult, + GlobalLimits, + LLMResponse, + OcrTextTransportResult, + RetryPolicy, + SourceConfig, +) _REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}], session_id="sess-1") @@ -966,7 +983,7 @@ class TestEmitterRecorderContract: from polygateway.telemetry.schema import COLUMNS rec = _MemoryRecorder() - await TelemetryEmitter(rec).emit_attempt( + await TelemetryEmitter(rec, text_cap=None).emit_attempt( request=_REQ, source=_source(), call_id="cid-1", @@ -981,7 +998,7 @@ class TestEmitterRecorderContract: from polygateway.telemetry.schema import COLUMNS rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec) + emitter = TelemetryEmitter(rec, text_cap=None) if emit == "attempt": await emitter.emit_attempt( request=_REQ, @@ -1005,7 +1022,7 @@ class TestEmitterObservabilityFields: async def test_attempt_carries_the_response_values(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec).emit_attempt( + await TelemetryEmitter(rec, text_cap=None).emit_attempt( request=_REQ, source=_source(), call_id="cid-1", @@ -1019,7 +1036,7 @@ class TestEmitterObservabilityFields: async def test_failed_attempt_has_no_provider_facts(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec).emit_attempt( + await TelemetryEmitter(rec, text_cap=None).emit_attempt( request=_REQ, source=_source(), call_id="cid-2", @@ -1034,7 +1051,7 @@ class TestEmitterObservabilityFields: async def test_cache_hit_replays_the_recorded_values(self): """决策 B1: 命中行原样回放,故命中率统计必须带 WHERE cache_hit = false。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec).emit_cache_hit( + await TelemetryEmitter(rec, text_cap=None).emit_cache_hit( request=_REQ, response=_resp(cached_prompt_tokens=64, model_reported="m-real", reasoning_tokens=7), ) @@ -1045,7 +1062,7 @@ class TestEmitterObservabilityFields: async def test_terminal_failure_records_none(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec).emit_terminal_failure( + await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure( request=_REQ, call_id="c", latency_ms=1, error="dead" ) assert rec.rows[0]["cached_prompt_tokens"] is None @@ -1069,7 +1086,7 @@ class TestEmitterSamplingColumn: async def test_attempt_merges_source_extra_body(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec).emit_attempt( + await TelemetryEmitter(rec, text_cap=None).emit_attempt( request=self._SAMPLED, source=_source(extra_body={"temperature": 0}), call_id="c", @@ -1082,7 +1099,7 @@ class TestEmitterSamplingColumn: async def test_response_format_never_leaks_into_the_column(self): """三行都不得出现 response_format——它不是采样参数。""" rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec) + emitter = TelemetryEmitter(rec, text_cap=None) await emitter.emit_attempt( request=self._SAMPLED, source=_source(), @@ -1103,7 +1120,7 @@ class TestEmitterSamplingColumn: async def test_sourceless_entries_record_call_level_only(self, emit): """两个最外层入口没有"生效源"可言,与 model/source_name 置空同一先例。""" rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec) + emitter = TelemetryEmitter(rec, text_cap=None) if emit == "cache_hit": await emitter.emit_cache_hit(request=self._SAMPLED, response=_resp()) else: @@ -1115,7 +1132,7 @@ class TestEmitterSamplingColumn: async def test_absent_sampling_is_null(self): """无采样参数时为 NULL,而非空字符串或 "{}"——便于 SQL 过滤。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec).emit_attempt( + await TelemetryEmitter(rec, text_cap=None).emit_attempt( request=_REQ, source=_source(), call_id="c", @@ -1145,7 +1162,7 @@ class TestEmitterCallerDimensions: async def test_every_entry_point_carries_the_dimensions(self, emit): """三条路径写出的行都必须带维度: 漏掉任一条,该租户的账就永远对不上。""" rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec) + emitter = TelemetryEmitter(rec, text_cap=None) if emit == "attempt": await emitter.emit_attempt( request=self._REQ_A, @@ -1178,7 +1195,7 @@ class TestEmitterCallerDimensions: meta={"batch": "old-batch"}, ) rec = _MemoryRecorder() - mw = TelemetryMW(TelemetryEmitter(rec)) + mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None)) async def terminal(request): # 缓存层回放的是历史那次的响应对象(其 call_id 属于 historical 那次) @@ -1200,7 +1217,7 @@ class TestEmitterCallerDimensions: JSON 函数直接查询,NULL 则要每条查询都额外判空。 """ rec = _MemoryRecorder() - await TelemetryEmitter(rec).emit_attempt( + await TelemetryEmitter(rec, text_cap=None).emit_attempt( request=_REQ, # tenant_id=None, meta={} source=_source(), call_id="c", @@ -1215,7 +1232,7 @@ class TestEmitterCallerDimensions: async def test_meta_is_serialized_with_sorted_keys(self): """键序固定,同一份维度在任意两行里字节一致,可直接做等值比对与去重。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec).emit_terminal_failure( + await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure( request=self._REQ_A, call_id="c", latency_ms=1, error="dead" ) assert list(json.loads(rec.rows[0]["meta"])) == ["a_first", "m_mid", "z_last"] @@ -1224,7 +1241,7 @@ class TestEmitterCallerDimensions: """`ensure_ascii=False`: 中文维度按原文落库,而非 `\\uXXXX` 转义串。""" rec = _MemoryRecorder() req = ChatRequest(messages=[{"role": "user", "content": "hi"}], meta={"dept": "研发"}) - await TelemetryEmitter(rec).emit_terminal_failure( + await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure( request=req, call_id="c", latency_ms=1, error="dead" ) assert "研发" in rec.rows[0]["meta"] @@ -1243,7 +1260,7 @@ class TestEmitterCallerDimensions: """ rec = _MemoryRecorder() req = ChatRequest(messages=[{"role": "user", "content": "hi"}], meta={"k": float("nan")}) - await TelemetryEmitter(rec).emit_terminal_failure( + await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure( request=req, call_id="c", latency_ms=1, error="dead" ) assert rec.rows == [] @@ -1258,7 +1275,7 @@ class TestCostWithCachedTier: async def test_cached_hit_lowers_the_recorded_cost(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, pricing=self._TABLE) + emitter = TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None) full = _resp(prompt_tokens=1_000_000, completion_tokens=0) await emitter.emit_attempt( request=_REQ, @@ -1284,7 +1301,7 @@ class TestCostWithCachedTier: async def test_cache_hit_row_still_costs_zero(self): """缓存命中未产生新调用 → cost 恒 0.0,该短路必须排在任何换算之前。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=self._TABLE).emit_cache_hit( + await TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None).emit_cache_hit( request=_REQ, response=_resp(prompt_tokens=1_000_000, cached_prompt_tokens=600_000), ) @@ -1292,7 +1309,7 @@ class TestCostWithCachedTier: async def test_unavailable_usage_still_costs_none(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=self._TABLE).emit_attempt( + await TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None).emit_attempt( request=_REQ, source=_source(), call_id="c", @@ -1306,7 +1323,7 @@ class TestCostWithCachedTier: class TestEmitter: async def test_attempt_success_row(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec) + emitter = TelemetryEmitter(rec, text_cap=None) await emitter.emit_attempt( request=_REQ, source=_source(), @@ -1322,7 +1339,7 @@ class TestEmitter: async def test_attempt_failure_row(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec) + emitter = TelemetryEmitter(rec, text_cap=None) await emitter.emit_attempt( request=_REQ, source=_source(), @@ -1339,7 +1356,7 @@ class TestEmitter: async def test_terminal_failure_row_is_unavailable(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=_PRICING).emit_terminal_failure( + await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_terminal_failure( request=_REQ, call_id="cid-t", latency_ms=5, error="cancelled" ) row = rec.rows[0] @@ -1352,7 +1369,7 @@ class TestEmitter: 参数第二组是改前兜底写出的 `0/4000` 形态: 那时换算出 0.032 的假金额。 """ rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=_PRICING).emit_attempt( + await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_attempt( request=_REQ, source=_source(), call_id="cid-u", @@ -1367,7 +1384,7 @@ class TestEmitter: async def test_measured_row_still_priced(self): """对照组: 同一价格表下 measured 行照常换算,证明 None 不是价格表没接上。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=_PRICING).emit_attempt( + await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_attempt( request=_REQ, source=_source(), call_id="cid-m", @@ -1380,7 +1397,7 @@ class TestEmitter: async def test_cache_hit_keeps_zero_cost_even_when_unavailable(self): """缓存命中未产生新调用,0.0 是事实而非未知 → 短路必须排在 cache_hit 之后。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=_PRICING).emit_cache_hit( + await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_cache_hit( request=_REQ, response=_resp(cache_hit=True, usage_source="unavailable", completion_tokens=4000), ) @@ -1388,7 +1405,7 @@ class TestEmitter: async def test_multimodal_messages_digested_before_storage(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec) + emitter = TelemetryEmitter(rec, text_cap=None) big = "data:image/png;base64," + "A" * 100_000 req = ChatRequest( messages=[ @@ -1415,7 +1432,7 @@ class TestEmitter: async def record_llm_call(self, **fields): raise OSError("disk full") - emitter = TelemetryEmitter(Broken()) + emitter = TelemetryEmitter(Broken(), text_cap=None) await emitter.emit_attempt( request=_REQ, source=_source(), @@ -1429,7 +1446,7 @@ class TestEmitter: class TestTelemetryMW: async def test_cache_hit_recorded(self): rec = _MemoryRecorder() - mw = TelemetryMW(TelemetryEmitter(rec)) + mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None)) async def terminal(request): return _resp(cache_hit=True, latency_ms=0, call_id="cache-cid") @@ -1442,7 +1459,7 @@ class TestTelemetryMW: async def test_normal_success_not_double_recorded(self): """成功尝试由 RetryMW 逐次记录;最外层不得重复记。""" rec = _MemoryRecorder() - mw = TelemetryMW(TelemetryEmitter(rec)) + mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None)) async def terminal(request): return _resp(cache_hit=False) @@ -1452,7 +1469,7 @@ class TestTelemetryMW: async def test_scope_level_failure_recorded(self): rec = _MemoryRecorder() - mw = TelemetryMW(TelemetryEmitter(rec)) + mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None)) async def terminal(request): raise CircuitOpenError(scope="llm", retry_after_s=30.0) @@ -1464,7 +1481,7 @@ class TestTelemetryMW: async def test_attempt_level_failure_not_double_recorded(self): """RequestRejected 已被 RetryMW 逐次记录 → 最外层跳过。""" rec = _MemoryRecorder() - mw = TelemetryMW(TelemetryEmitter(rec)) + mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None)) async def terminal(request): raise RequestRejectedError("400") @@ -1488,3 +1505,184 @@ def test_single_emitter_discipline(): if not p.endswith(("ports.py", "telemetry/sqlite.py", "telemetry/postgres.py")) ] assert callers == ["src/polygateway/middleware/telemetry.py"] + + +# —— issue #12 (a): 遥测正文可配置上限 —— + +_LONG = "甲乙丙丁戊己庚辛壬癸" * 5 # 50 字,cap=8 时省略 42 字 +_CAPPED = "甲乙丙丁戊己庚辛…(略 42 字)" + + +def _long_messages(): + """一条纯文本 + 一条多模态(text part + image_url part)。""" + return [ + {"role": "system", "content": _LONG}, + { + "role": "user", + "content": [ + {"type": "text", "text": _LONG}, + {"type": "image_url", "image_url": {"url": "https://gw.example/a.png"}}, + ], + }, + ] + + +async def _emit_with_cap(messages, *, cap, response=_LONG, thinking=_LONG): + rec = _MemoryRecorder() + await TelemetryEmitter(rec, text_cap=cap).emit_attempt( + request=ChatRequest(messages=messages, session_id="s"), + source=_source(), + call_id="c", + latency_ms=1, + response=_resp(content=response, thinking=thinking), + error=None, + ) + return rec.rows[0] + + +class TestTelemetryTextCap: + """截断发生在唯一遥测出口 `_record`(设计 §5.2);缺省 None = 不截断。""" + + async def test_cap_none_keeps_the_body_byte_for_byte(self): + """缺省不截断是人类决策(设计 §2 E-a): 落库正文与改前逐字节相同。""" + messages = _long_messages() + row = await _emit_with_cap(messages, cap=None) + assert row["messages"] == json.dumps(digest_messages(messages), ensure_ascii=False) + assert row["response"] == _LONG + assert row["thinking"] == _LONG + + async def test_cap_truncates_each_content_and_keeps_the_json_parsable(self): + """按每条文本切而非切整串 JSON: 否则该 TEXT 列此后无法按 JSON 解析。""" + row = await _emit_with_cap(_long_messages(), cap=8) + parsed = json.loads(row["messages"]) # 不抛 = 整串仍是合法 JSON + assert parsed[0]["content"] == _CAPPED + assert parsed[1]["content"][0]["text"] == _CAPPED + assert "(略 42 字)" in parsed[0]["content"] # 标记须含省略字数 + + async def test_image_digest_is_untouched_by_the_cap(self): + """多模态 image_url 的 sha256 摘要不是正文,不得被截断改形。""" + messages = _long_messages() + expected = digest_messages(messages)[1]["content"][1] + assert expected["type"] == "image_url" and len(expected["sha256"]) == 64 + row = await _emit_with_cap(messages, cap=8) + assert json.loads(row["messages"])[1]["content"][1] == expected + + async def test_non_string_content_passes_through_without_raising(self): + """外部输入形状不可控,遥测路径不得因此抛错(P5 + 降级方向)。 + + 同时钉住设计 §5.2 的覆盖面诚实声明: 只覆盖文本 content 与 text part, + 嵌套 dict 里的长文本**不在**覆盖范围内。 + """ + messages = [ + {"role": "user", "content": 123}, + {"role": "user", "content": None}, + {"role": "user", "content": {"nested": _LONG}}, + {"role": "user", "content": [{"type": "text", "text": 7}, "bare-part"]}, + ] + row = await _emit_with_cap(messages, cap=8) + assert json.loads(row["messages"]) == messages + + async def test_response_and_thinking_are_capped(self): + row = await _emit_with_cap([{"role": "user", "content": "hi"}], cap=8) + assert row["response"] == _CAPPED + assert row["thinking"] == _CAPPED + + async def test_cap_never_mutates_the_caller_messages(self): + """红线之二: 落库那份被截断,调用方持有的那份(含嵌套 part)一字未改。 + + `digest_messages` 对 content 非 list 的消息原样透传**同一个 dict 对象** + (`cache.py:43`),就地截断会连调用方的 messages、后续重试的请求体与缓存 + 写入的 key 一起改掉,且全程无任何报错。 + """ + messages = _long_messages() + snapshot = copy.deepcopy(messages) + row = await _emit_with_cap(messages, cap=8) + assert messages == snapshot + assert messages[0]["content"] == _LONG + assert messages[1]["content"][0]["text"] == _LONG + assert json.loads(row["messages"])[0]["content"] == _CAPPED # 落库那份确已截断 + + def test_non_positive_cap_rejected_at_construction(self): + """emitter 是三个 Client 唯一的汇合点,值域校验放这一处即覆盖全部装配路。 + + settings 层那道只管 env;直接构造 `GatewayClient(..., text_cap=0)` 是库 + 承诺的另一条公共装配路,没有这道闸就会把每条正文写成一个光秃秃的省略标记。 + """ + for bad in (0, -1): + with pytest.raises(ValueError, match="text_cap"): + TelemetryEmitter(_MemoryRecorder(), text_cap=bad) + + +class _StubEmbedTransport: + async def embed(self, *, texts, source, call_id): + return EmbeddingTransportResult( + vectors=[[1.0] for _ in texts], + dim=1, + prompt_tokens=1, + usage_source="measured", + raw={}, + ) + + +class _StubOcrTransport: + async def recognize_text(self, *, image, source, call_id): + return OcrTextTransportResult(text="识别结果" * 10, raw={"task_type": "text"}) + + async def parse_layout(self, *, image, source, call_id): + raise NotImplementedError + + +def _governance(scope, sources): + """embed/OCR 两条链路共用的最小治理装配(真实内存后端,不 mock)。""" + return { + "scope": scope, + "sources": sources, + "selector": RoundRobinSelector(), + "limiter": InMemoryLimiter( + scope=scope, + sources={s.name: s for s in sources}, + global_limits=GlobalLimits(max_concurrency=0, rpm=0, tpm=0), + lease_ttl_s=100.0, + ), + "breaker": InMemoryGate( + config=BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0) + ), + "retry": RetryPolicy(max_attempts=3, backoff_base_s=0.001, backoff_max_s=0.01), + "backpressure": BackpressurePolicy(stall_window_s=300.0, poll_interval_s=0.001), + } + + +class TestTextCapCoversEmbedAndOcrChains: + """`_record` 是三条链路共同的出口,cap 自然覆盖全部三条(设计 §5.2)。 + + 同一张表不该一半受控一半不受控;embed/OCR 各自的 200 字上限保留不动, + 与新 cap 是"取更严者"的关系。 + """ + + async def test_embed_rows_are_capped(self): + rec = _MemoryRecorder() + client = EmbeddingClient( + **_governance("embed", [_source(name="e1", model="embed-1")]), + transport=_StubEmbedTransport(), + batch_size=2, + telemetry=rec, + text_cap=8, + ) + await client.embed([_LONG]) + row = rec.rows[0] + assert json.loads(row["messages"])[0]["content"] == _CAPPED + assert row["response"] == "` 共 19 字 + + async def test_ocr_rows_are_capped(self): + rec = _MemoryRecorder() + client = OcrClient( + **_governance("ocr", [_source(name="m1", model="monkey-ocr")]), + transport=_StubOcrTransport(), + telemetry=rec, + text_cap=8, + ) + await client.recognize_text(b"jpg") + row = rec.rows[0] + # 占位串 `` 共 24 字 + assert json.loads(row["messages"])[0]["content"] == ".db),到期直接删文件。 + +时间口径: 截止时刻 = 当前 UTC 时刻 - N 天,删除 created_at < 截止时刻 的行; + --older-than-days 0 即"删除此刻之前的全部行"。 + +示例: + python tools/telemetry_retention.py --backend sqlite --path runs/telemetry.db \\ + --older-than-days 90 # dry-run,只看会删什么 + python tools/telemetry_retention.py --backend postgres --dsn "$DSN" \\ + --older-than-days 90 --apply --batch-size 1000 +""" + + +class _Parser(argparse.ArgumentParser): + """把 argparse 的参数错误退出码从 2 改成 1。 + + 2 在本脚本的契约里留给"连接/权限失败",两者混用会让调度器分不清"我写错了参数" + 与"数据库连不上"——后者要告警重试,前者不该重试。 + """ + + def error(self, message: str) -> NoReturn: + self.print_usage(sys.stderr) + print(f"{self.prog}: 参数错误: {message}", file=sys.stderr) + raise SystemExit(EXIT_USAGE) + + +def _build_parser() -> _Parser: + """构造 CLI 解析器(参数契约见设计 §6.2)。""" + parser = _Parser( + prog="telemetry_retention.py", + description="按 created_at 清理 PolyGateway 遥测表 llm_calls 的过期行(默认 dry-run)。", + epilog=_EPILOG, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("--backend", required=True, choices=("sqlite", "postgres")) + parser.add_argument("--path", help="SQLite 库文件路径(--backend sqlite 必填)") + parser.add_argument("--dsn", help="PostgreSQL DSN(--backend postgres 必填)") + parser.add_argument( + "--older-than-days", + type=int, + required=True, + metavar="N", + help="删除 created_at 早于 N 天前的行;N >= 0", + ) + parser.add_argument( + "--apply", + action="store_true", + help="真正执行删除;不给则只统计不删(默认)", + ) + parser.add_argument( + "--batch-size", + type=int, + metavar="N", + help="仅 postgres: 每批删除的行数,每批一个事务(默认 1000)", + ) + parser.add_argument( + "--vacuum", + action="store_true", + help="仅 sqlite: 删除后执行 VACUUM 回收文件空间;须与 --apply 同时给", + ) + return parser + + +def _validate(parser: _Parser, args: argparse.Namespace) -> None: + """校验参数组合;任何不合法组合以退出码 1 结束(P5: 不给默认值掩盖错误)。 + + **校验链的顺序就是错误消息的优先级**: 先两端通用,再按 backend 分支——同时给出 + 多个错误参数时,报出的是链上最先命中的那条。 + """ + _validate_shared(parser, args) + if args.backend == "sqlite": + _validate_sqlite(parser, args) + return + _validate_postgres(parser, args) + + +def _validate_shared(parser: _Parser, args: argparse.Namespace) -> None: + """两端通用的校验。 + + `--vacuum` 与 `--apply` 的联动归在这里(而不是 SQLite 分支): 它是"别在只想看看的 + 时候重写整个库"这条安全约束,先于"这个参数属于哪个 backend"成立。 + """ + if args.older_than_days < 0: + parser.error("--older-than-days 必须 >= 0") + if args.vacuum and not args.apply: + parser.error("--vacuum 会重写整个库文件,必须与 --apply 同时给") + + +def _validate_sqlite(parser: _Parser, args: argparse.Namespace) -> None: + """SQLite 分支: 必须有 --path,且拒绝一切 postgres 专属参数(不静默忽略)。""" + if args.path is None: + parser.error("--backend sqlite 需要 --path") + if args.dsn is not None: + parser.error("--backend sqlite 不接受 --dsn") + if args.batch_size is not None: + parser.error("--batch-size 仅用于 --backend postgres") + + +def _validate_postgres(parser: _Parser, args: argparse.Namespace) -> None: + """Postgres 分支: 必须有 --dsn,拒绝 sqlite 专属参数,并在此落 --batch-size 缺省值。""" + if args.dsn is None: + parser.error("--backend postgres 需要 --dsn") + if args.path is not None: + parser.error("--backend postgres 不接受 --path") + if args.vacuum: + parser.error("--vacuum 仅用于 --backend sqlite") + if args.batch_size is None: + args.batch_size = 1000 + elif args.batch_size < 1: + parser.error("--batch-size 必须 >= 1") + + +def _print_stats(total: int, low: object, high: object, tenants: Sequence[tuple[str, int]]) -> None: + """打印将删除行数、created_at 窗口与按 tenant_id 的分布。 + + tenant_id 用 repr 打: 空串是"未归属"的哨兵(不是 NULL),裸打会与缺失混淆。 + """ + print(f"将删除行数: {total}") + print(f"created_at 范围: {low} ~ {high}" if total else "created_at 范围: (无匹配行)") + print("按 tenant_id 分布:") + if not tenants: + print(" (无匹配行)") + for tenant, count in tenants: + print(f" {tenant!r}: {count}") + + +# --------------------------------------------------------------------------- SQLite + + +def _run_sqlite(path: str, cutoff: str, apply_: bool, vacuum: bool) -> int: + """SQLite 分支: 单条 DELETE(本地文件无长事务与锁膨胀问题),VACUUM 须显式要。""" + file = Path(path) + if not file.is_file(): + print(f"SQLite 库文件不存在: {file}", file=sys.stderr) + return EXIT_BACKEND + try: + conn = sqlite3.connect(f"file:{file}?mode=rw", uri=True) + except sqlite3.Error as exc: + print(f"打开 SQLite 库失败: {file}: {exc}", file=sys.stderr) + return EXIT_BACKEND + try: + exists = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", (TABLE,) + ).fetchone() + if exists is None: + print(f"目标库里没有表 {TABLE}: {file}", file=sys.stderr) + return EXIT_BACKEND + print(f"目标表: {file}::{TABLE}") + total, low, high = conn.execute( + f"SELECT COUNT(*), MIN(created_at), MAX(created_at) FROM {TABLE} WHERE created_at < ?", + (cutoff,), + ).fetchone() + tenants = conn.execute( + f"SELECT tenant_id, COUNT(*) FROM {TABLE} WHERE created_at < ? " + "GROUP BY tenant_id ORDER BY COUNT(*) DESC, tenant_id", + (cutoff,), + ).fetchall() + _print_stats(total, low, high, tenants) + if not apply_: + print("模式 dry-run: 未删除任何行。确认无误后加 --apply 才会真正删除。") + return EXIT_OK + cursor = conn.execute(f"DELETE FROM {TABLE} WHERE created_at < ?", (cutoff,)) + conn.commit() + print(f"已删除 {cursor.rowcount} 行。") + if vacuum: + print("执行 VACUUM(重写整个库文件,需要与库等量的空闲磁盘)…") + conn.execute("VACUUM") + conn.commit() + print("VACUUM 完成。") + except sqlite3.Error as exc: + print(f"SQLite 操作失败: {exc}", file=sys.stderr) + return EXIT_BACKEND + finally: + conn.close() + return EXIT_OK + + +# --------------------------------------------------------------------------- PostgreSQL + + +def _quote(identifier: str) -> str: + """把 catalog 取回的 schema/表名包成合法标识符(库名含大写或特殊字符时必需)。""" + escaped = identifier.replace('"', '""') + return f'"{escaped}"' + + +async def _run_postgres(dsn: str, cutoff: datetime, apply_: bool, batch_size: int) -> int: + """PostgreSQL 分支: 分区表让路,普通表分批 DELETE(每批一个事务)。""" + try: + import asyncpg + except ImportError as exc: + print( + f"--backend postgres 需要 asyncpg,当前不可用({exc});" + "请 pip install 'polygateway[postgres]' 或 pip install asyncpg 后重试。", + file=sys.stderr, + ) + return EXIT_BACKEND + try: + conn = await asyncpg.connect(dsn, timeout=10) + except (OSError, asyncpg.PostgresError) as exc: + print(f"连接 PostgreSQL 失败: {exc}", file=sys.stderr) + return EXIT_BACKEND + try: + return await _purge_postgres(conn, cutoff, apply_, batch_size) + except asyncpg.PostgresError as exc: + print(f"PostgreSQL 操作失败: {exc}", file=sys.stderr) + return EXIT_BACKEND + finally: + await conn.close() + + +async def _purge_postgres(conn: Any, cutoff: datetime, apply_: bool, batch_size: int) -> int: + """已连上后的清理主体(conn 是 asyncpg.Connection,不 import 类型以免脚本硬依赖)。""" + # 先解析目标: to_regclass 走连接自己的 search_path,故必须把解析结果打出来—— + # "我删的到底是哪张表"是这个脚本唯一不能猜的事(共享库里另有同名表的场景常见)。 + target = await conn.fetchrow( + "SELECT n.nspname AS schema, c.relname AS name, " + "EXISTS (SELECT 1 FROM pg_partitioned_table p WHERE p.partrelid = c.oid) AS partitioned " + "FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace " + "WHERE c.oid = to_regclass($1)", + TABLE, + ) + if target is None: + print(f"目标库的 search_path 下找不到表 {TABLE}", file=sys.stderr) + return EXIT_BACKEND + schema, name = target["schema"], target["name"] + qualified = f"{_quote(schema)}.{_quote(name)}" + print(f"目标表: {schema}.{name}") + if target["partitioned"]: + print( + f"{schema}.{name} 是分区表: 本脚本拒绝对分区表执行 DELETE。\n" + "请改用 DETACH/DROP PARTITION —— ALTER TABLE ... DETACH PARTITION <子表> 后 " + "DROP TABLE <子表>(或交给 pg_partman 的 retention)。\n" + "那是 O(1) 的,而 DELETE 会全表扫描并留下等量膨胀。" + ) + return EXIT_PARTITIONED + + stats = await conn.fetchrow( + f"SELECT COUNT(*) AS total, MIN(created_at) AS low, MAX(created_at) AS high " + f"FROM {qualified} WHERE created_at < $1", + cutoff, + ) + tenants = await conn.fetch( + f"SELECT tenant_id, COUNT(*) AS total FROM {qualified} WHERE created_at < $1 " + "GROUP BY tenant_id ORDER BY COUNT(*) DESC, tenant_id", + cutoff, + ) + _print_stats( + stats["total"], stats["low"], stats["high"], [(r["tenant_id"], r["total"]) for r in tenants] + ) + if not apply_: + print("模式 dry-run: 未删除任何行。确认无误后加 --apply 才会真正删除。") + return EXIT_OK + + # 分批: 一条大 DELETE 会撑出长事务(阻塞 autovacuum、堆积 WAL、锁膨胀), + # 中断后还得整批回滚重来。每批独立提交,中断只影响未删批次。 + deleted = 0 + batches = 0 + statement = ( + f"DELETE FROM {qualified} WHERE ctid IN " + f"(SELECT ctid FROM {qualified} WHERE created_at < $1 ORDER BY created_at LIMIT $2)" + ) + while True: + async with conn.transaction(): + status = await conn.execute(statement, cutoff, batch_size) + count = int(status.rsplit(" ", 1)[-1]) + if count == 0: + break + deleted += count + batches += 1 + print(f" 批次 {batches}: 删除 {count} 行(已提交)") + print(f"已删除 {deleted} 行,共 {batches} 批。") + return EXIT_OK + + +# --------------------------------------------------------------------------- 入口 + + +def main(argv: Sequence[str] | None = None) -> int: + """解析参数并分派到对应后端;返回值即进程退出码。""" + parser = _build_parser() + args = parser.parse_args(argv) + _validate(parser, args) + + cutoff = datetime.now(UTC) - timedelta(days=args.older_than_days) + print(f"后端: {args.backend}") + print( + f"截止时间(UTC): {cutoff.strftime(_SQLITE_TIME_FORMAT)}" + f"(--older-than-days {args.older_than_days};删除 created_at 早于该时刻的行)" + ) + print(f"模式: {'apply(将真正删除)' if args.apply else 'dry-run(只统计,不删除)'}") + if args.backend == "sqlite": + return _run_sqlite(args.path, cutoff.strftime(_SQLITE_TIME_FORMAT), args.apply, args.vacuum) + return asyncio.run(_run_postgres(args.dsn, cutoff, args.apply, args.batch_size)) + + +if __name__ == "__main__": + sys.exit(main())