iomgaa
33ed7ecdfc
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.
379 lines
14 KiB
Python
379 lines
14 KiB
Python
"""CacheMW 与缓存 key 公式测试(ARCH §7.5: 防毒化 key、命中重建、静默降级)。"""
|
|
|
|
import dataclasses
|
|
import hashlib
|
|
import json
|
|
|
|
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.middleware.telemetry import TelemetryEmitter
|
|
from polygateway.types import ChatRequest, LLMResponse, SourceConfig
|
|
|
|
_MSGS = [{"role": "user", "content": "hi"}]
|
|
|
|
|
|
def _resp(content="cached", **overrides):
|
|
base = {
|
|
"content": content,
|
|
"thinking": "",
|
|
"model": "m",
|
|
"provider": "p",
|
|
"prompt_tokens": 1,
|
|
"completion_tokens": 2,
|
|
"latency_ms": 30,
|
|
"ttft_ms": 5.0,
|
|
"max_inter_token_ms": 2.0,
|
|
"cache_hit": False,
|
|
"call_id": "orig",
|
|
"source_name": "s1",
|
|
"usage_source": "measured",
|
|
}
|
|
base.update(overrides)
|
|
return LLMResponse(**base)
|
|
|
|
|
|
class TestKeyFormula:
|
|
def test_same_request_same_key(self):
|
|
k1 = build_cache_key("m", _MSGS, "proj", None)
|
|
k2 = build_cache_key("m", _MSGS, "proj", None)
|
|
assert k1 == k2 and k1.startswith("pgw:cache:")
|
|
|
|
@pytest.mark.parametrize(
|
|
("a", "b"),
|
|
[
|
|
(("m1", _MSGS, "proj", None), ("m2", _MSGS, "proj", None)),
|
|
(("m", _MSGS, "proj", None), ("m", _MSGS, "tenant2", None)),
|
|
(("m", _MSGS, "proj", None), ("m", _MSGS, "proj", "epoch2")),
|
|
(("m", _MSGS, "proj", "s1"), ("m", _MSGS, "proj", "s2")),
|
|
(("m", _MSGS, "proj", None), ("m", [{"role": "user", "content": "yo"}], "proj", None)),
|
|
],
|
|
)
|
|
def test_any_dimension_change_changes_key(self, a, b):
|
|
assert build_cache_key(*a) != build_cache_key(*b)
|
|
|
|
def test_empty_sampling_keeps_legacy_key(self):
|
|
"""空采样参数时键形逐字不变,存量缓存不被全量作废(issue #4 决策 C)。
|
|
|
|
golden 值取自加 sampling 维度之前的实现,不得随实现漂移。
|
|
"""
|
|
assert build_cache_key("qwen-max", [{"role": "user", "content": "hi"}], "proj", None) == (
|
|
"pgw:cache:c54544e8672f4c91373b4a72716a88497445b440b89445aa5379b356b228f58b"
|
|
)
|
|
assert build_cache_key("qwen-max", [{"role": "user", "content": "hi"}], "proj", "s1") == (
|
|
"pgw:cache:eed9cd9cc06acc0dedf4f337b74e06ed3482afdc30fa2acedd194f6cc1df33bf"
|
|
)
|
|
|
|
def test_differing_seed_changes_key(self):
|
|
"""issue #4 的直接回归: 5 个 seed 若共用一个 key,标准差会恒为 0。"""
|
|
k1 = build_cache_key("m", _MSGS, "proj", None, sampling={"seed": 1})
|
|
k2 = build_cache_key("m", _MSGS, "proj", None, sampling={"seed": 2})
|
|
assert k1 != k2
|
|
|
|
def test_sampling_key_order_irrelevant(self):
|
|
k1 = build_cache_key("m", _MSGS, "proj", None, sampling={"seed": 1, "temperature": 0})
|
|
k2 = build_cache_key("m", _MSGS, "proj", None, sampling={"temperature": 0, "seed": 1})
|
|
assert k1 == k2
|
|
|
|
def test_empty_sampling_equals_omitted(self):
|
|
"""空 dict 与不传须同键,否则升级后存量缓存全部 miss。"""
|
|
assert build_cache_key("m", _MSGS, "proj", None, sampling={}) == build_cache_key(
|
|
"m", _MSGS, "proj", None
|
|
)
|
|
|
|
def test_multimodal_part_digested_not_inlined(self):
|
|
big_b64 = "data:image/png;base64," + "A" * 1_000_000
|
|
messages = [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "image_url", "image_url": {"url": big_b64}},
|
|
{"type": "text", "text": "describe"},
|
|
],
|
|
}
|
|
]
|
|
digested = digest_messages(messages)
|
|
payload = json.dumps(digested, ensure_ascii=False)
|
|
assert len(payload) < 500 # 大图不进 canonical_json
|
|
expected = hashlib.sha256(big_b64.encode()).hexdigest()
|
|
assert expected in payload # 但字节变化仍改变 key
|
|
# 图像字节变化 → key 变
|
|
messages2 = [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "image_url", "image_url": {"url": big_b64[:-1] + "B"}},
|
|
{"type": "text", "text": "describe"},
|
|
],
|
|
}
|
|
]
|
|
assert build_cache_key("m", messages, "p", None) != build_cache_key(
|
|
"m", messages2, "p", None
|
|
)
|
|
|
|
|
|
class _Terminal:
|
|
def __init__(self, response):
|
|
self.response = response
|
|
self.calls = 0
|
|
|
|
async def __call__(self, request):
|
|
self.calls += 1
|
|
if isinstance(self.response, Exception):
|
|
raise self.response
|
|
return self.response
|
|
|
|
|
|
def _mw(backend, **kwargs):
|
|
defaults = {
|
|
"backend": backend,
|
|
"model_fingerprint": "m",
|
|
"default_namespace": "proj",
|
|
"ttl_s": 3600,
|
|
}
|
|
defaults.update(kwargs)
|
|
return CacheMW(**defaults)
|
|
|
|
|
|
class TestCacheFlow:
|
|
async def test_miss_then_hit_with_fresh_call_id(self):
|
|
backend = InMemoryCache()
|
|
mw = _mw(backend)
|
|
terminal = _Terminal(_resp())
|
|
first = await mw(ChatRequest(messages=_MSGS), terminal)
|
|
assert first.cache_hit is False and terminal.calls == 1
|
|
second = await mw(ChatRequest(messages=_MSGS), terminal)
|
|
assert second.cache_hit is True and second.latency_ms == 0
|
|
assert second.content == "cached"
|
|
assert second.call_id != first.call_id # 命中生成独立 cache_call_id
|
|
assert terminal.calls == 1 # 未再触达内层
|
|
|
|
async def test_differing_sampling_does_not_hit(self):
|
|
"""issue #4 的中间件层回归: 逐 rollout 变 seed 必须回源,不得复用响应。"""
|
|
backend = InMemoryCache()
|
|
mw = _mw(backend)
|
|
terminal = _Terminal(_resp())
|
|
await mw(ChatRequest(messages=_MSGS, sampling={"seed": 1}), terminal)
|
|
await mw(ChatRequest(messages=_MSGS, sampling={"seed": 2}), terminal)
|
|
assert terminal.calls == 2 # 两次都回源
|
|
# 同 seed 才命中
|
|
third = await mw(ChatRequest(messages=_MSGS, sampling={"seed": 1}), terminal)
|
|
assert third.cache_hit is True and terminal.calls == 2
|
|
|
|
async def test_structured_injection_does_not_pollute_key(self):
|
|
"""CacheMW 读 sampling 而非 overlay: 结构化注入不该改变缓存身份。"""
|
|
backend = InMemoryCache()
|
|
mw = _mw(backend)
|
|
terminal = _Terminal(_resp())
|
|
await mw(ChatRequest(messages=_MSGS, sampling={"seed": 1}), terminal)
|
|
polluted = ChatRequest(
|
|
messages=_MSGS,
|
|
sampling={"seed": 1},
|
|
overlay={"seed": 1, "response_format": {"type": "json_object"}},
|
|
)
|
|
assert (await mw(polluted, terminal)).cache_hit is True
|
|
assert terminal.calls == 1
|
|
|
|
async def test_per_call_namespace_overrides_default(self):
|
|
backend = InMemoryCache()
|
|
mw = _mw(backend)
|
|
terminal = _Terminal(_resp())
|
|
await mw(ChatRequest(messages=_MSGS, cache_namespace="tenant-a"), terminal)
|
|
# 另一租户不得命中
|
|
await mw(ChatRequest(messages=_MSGS, cache_namespace="tenant-b"), terminal)
|
|
assert terminal.calls == 2
|
|
|
|
async def test_failure_not_cached(self):
|
|
backend = InMemoryCache()
|
|
mw = _mw(backend)
|
|
failing = _Terminal(TransientError("boom"))
|
|
with pytest.raises(TransientError):
|
|
await mw(ChatRequest(messages=_MSGS), failing)
|
|
ok = _Terminal(_resp())
|
|
await mw(ChatRequest(messages=_MSGS), ok)
|
|
assert ok.calls == 1 # 失败未被固化,正常回源
|
|
|
|
async def test_ttl_expiry(self):
|
|
t = {"now": 0.0}
|
|
backend = InMemoryCache(now=lambda: t["now"])
|
|
mw = _mw(backend, ttl_s=100)
|
|
terminal = _Terminal(_resp())
|
|
await mw(ChatRequest(messages=_MSGS), terminal)
|
|
t["now"] = 101.0
|
|
await mw(ChatRequest(messages=_MSGS), terminal)
|
|
assert terminal.calls == 2
|
|
|
|
|
|
class TestObservabilityFieldsOnHit:
|
|
"""issue #3 决策 B1: 命中行原样回放,与 model/prompt_tokens 同一口径。"""
|
|
|
|
async def test_fields_replayed_on_hit(self):
|
|
backend = InMemoryCache()
|
|
mw = _mw(backend)
|
|
terminal = _Terminal(
|
|
_resp(cached_prompt_tokens=64, model_reported="MiniMax-Text-01-250321")
|
|
)
|
|
await mw(ChatRequest(messages=_MSGS), terminal)
|
|
hit = await mw(ChatRequest(messages=_MSGS), terminal)
|
|
assert hit.cache_hit is True
|
|
assert hit.cached_prompt_tokens == 64
|
|
assert hit.model_reported == "MiniMax-Text-01-250321"
|
|
|
|
async def test_legacy_cache_entry_without_new_keys_rehydrates(self):
|
|
"""旧格式条目(无这两个键)必须照常重建为 None,不得抛异常回源。"""
|
|
backend = InMemoryCache()
|
|
mw = _mw(backend)
|
|
key = build_cache_key("m", _MSGS, "proj", None)
|
|
legacy = {
|
|
"content": "legacy",
|
|
"thinking": "",
|
|
"model": "m",
|
|
"provider": "p",
|
|
"prompt_tokens": 1,
|
|
"completion_tokens": 2,
|
|
"latency_ms": 30,
|
|
"ttft_ms": 5.0,
|
|
"max_inter_token_ms": 2.0,
|
|
"cache_hit": False,
|
|
"call_id": "orig",
|
|
"source_name": "s1",
|
|
"cost": None,
|
|
"usage_source": "measured",
|
|
}
|
|
await backend.set(key, json.dumps(legacy), ttl_s=100)
|
|
terminal = _Terminal(_resp())
|
|
hit = await mw(ChatRequest(messages=_MSGS), terminal)
|
|
assert hit.content == "legacy" and terminal.calls == 0 # 真的走了缓存
|
|
assert hit.cached_prompt_tokens is None and hit.model_reported is None
|
|
|
|
|
|
class _BrokenBackend:
|
|
async def get(self, key):
|
|
raise ConnectionError("redis down")
|
|
|
|
async def set(self, key, value, ttl_s):
|
|
raise ConnectionError("redis down")
|
|
|
|
|
|
class TestDegradation:
|
|
async def test_backend_failure_degrades_silently(self):
|
|
mw = _mw(_BrokenBackend())
|
|
terminal = _Terminal(_resp())
|
|
resp = await mw(ChatRequest(messages=_MSGS), terminal)
|
|
assert resp.content == "cached" and terminal.calls == 1 # 读写全降级,调用照常
|
|
|
|
async def test_corrupt_cache_value_treated_as_miss(self):
|
|
backend = InMemoryCache()
|
|
mw = _mw(backend)
|
|
key = build_cache_key("m", _MSGS, "proj", None)
|
|
await backend.set(key, "{not json", 3600)
|
|
terminal = _Terminal(_resp())
|
|
resp = await mw(ChatRequest(messages=_MSGS), terminal)
|
|
assert terminal.calls == 1 and resp.cache_hit is False
|
|
|
|
|
|
class _FakeStrategy:
|
|
"""fake StructuredOutputStrategy(T2 冻结的 Protocol,不依赖 T11)。"""
|
|
|
|
def request_overlay(self, schema):
|
|
return {}
|
|
|
|
def parse(self, text):
|
|
data = json.loads(text) # 简化: 直接 json.loads
|
|
if not isinstance(data, dict):
|
|
raise ResultInvalidError("非对象", raw_text=text)
|
|
return data
|
|
|
|
|
|
class _StrictModel:
|
|
"""鸭子型 pydantic 模型: model_validate 要求含 answer 键。"""
|
|
|
|
@classmethod
|
|
def model_validate(cls, data):
|
|
if "answer" not in data:
|
|
raise ValueError("missing answer")
|
|
return {"validated": data["answer"]}
|
|
|
|
|
|
class TestStructuredRehydration:
|
|
async def test_hit_rebuilds_structured_data(self):
|
|
backend = InMemoryCache()
|
|
mw = _mw(backend, strategy=_FakeStrategy())
|
|
terminal = _Terminal(_resp(content='{"answer": 42}'))
|
|
req = ChatRequest(messages=_MSGS, structured=_StrictModel)
|
|
await mw(req, terminal)
|
|
hit = await mw(req, terminal)
|
|
assert hit.cache_hit is True
|
|
assert hit.structured_data == {"validated": 42}
|
|
assert terminal.calls == 1
|
|
|
|
async def test_schema_change_revalidation_failure_falls_back_to_source(self):
|
|
backend = InMemoryCache()
|
|
mw = _mw(backend, strategy=_FakeStrategy())
|
|
terminal = _Terminal(_resp(content='{"other": 1}')) # 缓存内容不含 answer
|
|
await mw(ChatRequest(messages=_MSGS), terminal) # 无 structured 写入
|
|
# 换 schema 读: 重校验失败 → 按未命中回源
|
|
again = await mw(ChatRequest(messages=_MSGS, structured=_StrictModel), terminal)
|
|
assert terminal.calls == 2 and again.cache_hit is False
|
|
|
|
async def test_structured_data_not_serialized_into_cache(self):
|
|
backend = InMemoryCache()
|
|
mw = _mw(backend, strategy=_FakeStrategy())
|
|
terminal = _Terminal(
|
|
dataclasses.replace(_resp(content='{"answer": 1}'), structured_data={"x": object()})
|
|
)
|
|
await mw(ChatRequest(messages=_MSGS), terminal) # 不可 JSON 的 structured_data 不阻塞写缓存
|
|
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
|