fix: close the failure modes review found in the new code

Three of them were the same shape as the bug this branch exists to fix:
something goes wrong, the library swallows it, and the caller is left
with a number that means the opposite of what happened.

The throttle key had no source in it. Five sources on one model is the
normal case here, so the first one to break would warn once and silence
the other four for the life of the process, and the message never said
which gateway to look at.

An unknown verdict in a cached entry threw away the whole response. The
rehydrator tolerates unknown fields but not unknown values of a known
field, so two library versions sharing a Redis would each invalidate
the other's entries: halved hit rate, and the only log line says the
cache rebuild failed. A purely observational field should not be able
to void a response whose content is intact.

Normalising for telemetry now degrades instead of raising, both for a
bare string and for a value outside the domain. Either one used to
reach the same except and cost the whole row, which is exactly how
1.3.0 lost nineteen calls without anyone noticing.

tests/unit/test_cache.py

@@ -5,6 +5,7 @@ import hashlib
 import json
 
 import pytest
+from loguru import logger
 
 from polygateway.backends.memory.cache import InMemoryCache
 from polygateway.errors import ResultInvalidError, TransientError
@@ -266,19 +267,47 @@ class TestThinkingObservationRehydration:
         assert isinstance(hit.thinking_observation, ThinkingObservation)
         assert hit.thinking_observation is ThinkingObservation.OBSERVED
 
-    async def test_illegal_value_falls_back_to_source(self):
-        """污染值(旧版本写入或人为篡改)按未命中回源,不得复活出域外取值。"""
+    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="poisoned"))
-        poisoned["thinking_observation"] = "bogus"
+        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"  # 回源结果,不是被污染的那条
+        assert resp.content == "cached"
 
     async def test_legacy_entry_without_key_rehydrates_to_default(self):
         """升级前写入的条目没有该键,必须照常复活并落到默认 UNKNOWN。"""