feat: cap telemetry bodies at a configurable length

Chat rows stored full message and response text with no upper bound, so downstream contracts and tenders lived in llm_calls indefinitely. Add _cap_text/_cap_messages in the single telemetry exit (_record), applied after digest_messages and before json.dumps, plus to response/thinking. Capping is per text, not over the serialized JSON: cutting the whole string would emit invalid JSON into an unvalidated TEXT column. The cap builds new dicts and never mutates in place — digest_messages passes non-list content straight through as the same object, so an in-place cut would silently poison the caller's messages and the cache key. text_cap is required on TelemetryEmitter (internal class, three known construction sites) and defaults to None on the three public clients, so the default behaviour stays byte-for-byte identical. Settings wiring lands separately.
2026-08-19 13:39:17 -04:00
parent e0a33ecf93
commit 33ed7ecdfc
9 changed files with 352 additions and 51 deletions
@@ -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,174 @@ 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  # 落库那份确已截断
+
+
+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"] == "<vectors…(略 11 字)"  # `<vectors n=1 dim=1>` 共 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]
+        # 占位串 `<ocr:text image_bytes=3>` 共 24 字
+        assert json.loads(row["messages"])[0]["content"] == "<ocr:tex…(略 16 字)"
+        assert row["response"] == "识别结果识别结果…(略 32 字)"  # 先经 OCR 自有的 200 字上限