"""CacheMW 与缓存 key 公式测试(ARCH §7.5: 防毒化 key、命中重建、静默降级)。""" import dataclasses import hashlib import json import pytest from loguru import logger 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, Effort, LLMResponse, SourceConfig, ThinkingObservation, ) _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 ) def test_request_tier_changes_key(self): """同 messages 跑 low 与 max 不得互相命中(issue #20;issue #4 的逐字翻版)。 请求级档位必须**独立于** `model_fingerprint` 进 key: 后者是装配期算出的 集合级指纹,一次调用改档位不会让它变一个字节。 """ k_low = build_cache_key("m", _MSGS, "proj", None, reasoning_effort=Effort.LOW) k_max = build_cache_key("m", _MSGS, "proj", None, reasoning_effort=Effort.MAX) assert k_low != k_max def test_explicit_none_tier_is_not_the_absent_tier(self): """`None`(不表态)与 `Effort.NONE`(要求不推理)是两个 key。 二者合并即毒化: "没写档位"的调用会读到"明确关掉推理"那次的响应, 而后者的内容恰恰是缺推理过程的。 """ assert build_cache_key("m", _MSGS, "proj", None) != build_cache_key( "m", _MSGS, "proj", None, reasoning_effort=Effort.NONE ) def test_absent_tier_keeps_legacy_key(self): """不表态档位时键形逐字不变,存量缓存不被本次升级全量作废。 golden 值与 `test_empty_sampling_keeps_legacy_key` 同源,取自加 `reasoning_effort` 维度之前的实现,不得随实现漂移。 """ assert build_cache_key( "qwen-max", [{"role": "user", "content": "hi"}], "proj", None, reasoning_effort=None, ) == ("pgw:cache:c54544e8672f4c91373b4a72716a88497445b440b89445aa5379b356b228f58b") 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_differing_reasoning_effort_does_not_hit(self): """接线门: `CacheMW` 必须把 `request.reasoning_effort` 传进 key 公式。 只测 `build_cache_key` 不够——参数加了却没人传是本改动最可能的落地方式, 那种缺口在公式层的用例里完全看不见。 """ backend = InMemoryCache() mw = _mw(backend) terminal = _Terminal(_resp()) await mw(ChatRequest(messages=_MSGS, reasoning_effort=Effort.LOW), terminal) await mw(ChatRequest(messages=_MSGS, reasoning_effort=Effort.MAX), terminal) assert terminal.calls == 2 # 两档各自回源 third = await mw(ChatRequest(messages=_MSGS, reasoning_effort=Effort.LOW), 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 TestThinkingObservationRehydration: """issue #16/#17: 命中回放必须复活成枚举实例,而不是 JSON 里的裸 str。 裸 str 与字段注解分叉,下游拿 `resp.thinking_observation is ThinkingObservation.OBSERVED` 判等会在缓存命中路径上静默为 False。 """ async def test_hit_replays_enum_instance_not_bare_str(self): backend = InMemoryCache() mw = _mw(backend) terminal = _Terminal(_resp(thinking_observation=ThinkingObservation.OBSERVED)) await mw(ChatRequest(messages=_MSGS), terminal) hit = await mw(ChatRequest(messages=_MSGS), terminal) assert hit.cache_hit is True and terminal.calls == 1 assert isinstance(hit.thinking_observation, ThinkingObservation) assert hit.thinking_observation is ThinkingObservation.OBSERVED async def test_unknown_value_degrades_to_unknown_and_still_hits(self): """域外取值降级为 UNKNOWN,内容照常复活——不得因此作废整条缓存。 真实场景: 三项目共用一个 Redis,先升级的项目写入了本版没有的第四态, 未升级的两个项目若把它判成未命中,就会在这些 key 上每次真打网关、随后 覆写回旧值,两个版本互相打对方的缓存(表现是命中率莫名腰斩)。一个纯 可观测性字段不该有能力废掉内容完好的缓存响应。 """ backend = InMemoryCache() mw = _mw(backend) key = build_cache_key("m", _MSGS, "proj", None) poisoned = dataclasses.asdict(_resp(content="from-a-newer-version")) poisoned["thinking_observation"] = "partially_observed" poisoned.pop("structured_data", None) await backend.set(key, json.dumps(poisoned), 3600) terminal = _Terminal(_resp()) messages: list[str] = [] sink_id = logger.add(messages.append, level="WARNING") try: resp = await mw(ChatRequest(messages=_MSGS), terminal) finally: logger.remove(sink_id) assert terminal.calls == 0 and resp.cache_hit is True assert resp.content == "from-a-newer-version" # 内容完好,照常复活 assert resp.thinking_observation is ThinkingObservation.UNKNOWN # 单独一条讲清原因的 warning: 通用的"重建失败"没有任何线索指向真因 hits = [m for m in messages if "partially_observed" in m] assert len(hits) == 1, f"域外取值必须单独告警一次,实得 {len(hits)} 条: {messages}" assert "thinking_observation" in hits[0] assert [m for m in messages if "重建失败" in m] == [] async def test_a_broken_payload_still_falls_back_to_source(self): """对照组: 内容完整性真被破坏时,仍必须按未命中回源(降级方向不变)。""" backend = InMemoryCache() mw = _mw(backend) key = build_cache_key("m", _MSGS, "proj", None) await backend.set(key, "{not json at all", 3600) terminal = _Terminal(_resp()) resp = await mw(ChatRequest(messages=_MSGS), terminal) assert terminal.calls == 1 and resp.cache_hit is False assert resp.content == "cached" async def test_legacy_entry_without_key_rehydrates_to_default(self): """升级前写入的条目没有该键,必须照常复活并落到默认 UNKNOWN。""" backend = InMemoryCache() mw = _mw(backend) key = build_cache_key("m", _MSGS, "proj", None) legacy = dataclasses.asdict(_resp(content="legacy")) legacy.pop("thinking_observation") legacy.pop("structured_data", None) await backend.set(key, json.dumps(legacy), 3600) terminal = _Terminal(_resp()) hit = await mw(ChatRequest(messages=_MSGS), terminal) assert hit.content == "legacy" and terminal.calls == 0 assert hit.thinking_observation is ThinkingObservation.UNKNOWN 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