feat: wire the telemetry text cap through settings

`PGW_TELEMETRY_TEXT_CAP` now reaches the emitter on every assembly path.
Unset means no truncation, which stays the default: a truncated row is
no longer audit evidence and cannot be replayed, and downstreams rely on
that today. The flip side — contracts and bids sitting in `llm_calls`
indefinitely, multi-tenant — is spelled out in `.env.example` so readers
can weigh both.

All three `from_settings` paths are wired (chat, embedding, OCR): they
write the same table, so capping only chat would leave half of it
uncontrolled. `TelemetryEmitter.__init__` now rejects `text_cap <= 0`;
it is the single point where the three clients converge, so the direct
construction path — a public assembly route the settings guard never
sees — is covered too. `0` would otherwise reduce every body to a bare
elision marker.
This commit is contained in:
2026-08-19 13:57:15 -04:00
parent 33ed7ecdfc
commit c26b34e854
9 changed files with 192 additions and 0 deletions
+91
View File
@@ -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 → 共享全局并发闸。"""
+33
View File
@@ -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"):
+10
View File
@@ -1602,6 +1602,16 @@ class TestTelemetryTextCap:
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):