diff --git a/src/polygateway/middleware/cache.py b/src/polygateway/middleware/cache.py
index 52f0fa1..2e386bf 100644
--- a/src/polygateway/middleware/cache.py
+++ b/src/polygateway/middleware/cache.py
@@ -28,6 +28,31 @@ _KEY_PREFIX = "pgw:cache:"
 _RESPONSE_FIELDS = {f.name for f in dataclasses.fields(LLMResponse)}
 
 
+def _coerce_observation(raw: Any) -> ThinkingObservation:
+    """缓存里的三态取值 → 枚举;域外取值降级为 `UNKNOWN`,**不作废整条缓存**。
+
+    方向选择的理由: `_rehydrate` 对 JSON 里的**新字段**已经是宽容的(先按
+    `_RESPONSE_FIELDS` 过滤),对同一字段的**新取值**却不该是致命的。真实场景是
+    多个项目共用一个 Redis,先升级的那个写入了本版没有的取值,未升级的项目若把
+    这些条目判成未命中,就会每次真打网关、随后覆写回旧值,两个版本互相打对方的
+    缓存(表现是命中率莫名腰斩,而通用的"重建失败"文案给不出任何线索)。一个纯
+    可观测性字段不该有能力废掉内容完好的缓存响应——"整条作废"留给真正破坏内容
+    完整性的失败(JSON 坏了、结构化重建不过)。
+
+    降级到 `UNKNOWN` 而不是别的态: 它的语义恰好就是"本次判不出来",对一个本库
+    读不懂的取值,这是唯一诚实的说法。
+    """
+    try:
+        return ThinkingObservation(raw)
+    except ValueError:
+        logger.warning(
+            "缓存条目的 thinking_observation 取值 {!r} 不在本版取值域内(多半由更新版本的"
+            "进程写入),已降级为 UNKNOWN;响应内容照常复活——可观测性字段不作废缓存",
+            raw,
+        )
+        return ThinkingObservation.UNKNOWN
+
+
 def digest_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
     """多模态 content part 先各自 sha256 摘要再参与序列化;文本原文参与。
 
@@ -134,10 +159,9 @@ class CacheMW:
             structured_data = self._rebuild_structured(fields.get("content", ""), request)
             # JSON 里存的是 StrEnum 的字符串值,不转就复活成裸 str,与字段注解分叉
             # (下游 `is ThinkingObservation.OBSERVED` 会在命中路径上静默为 False);
-            # 键缺失即升级前写入的旧条目,交给 dataclass 默认值。域外值抛
-            # ValueError,由下方 except 吞成"按未命中回源"——降级方向正确。
+            # 键缺失即升级前写入的旧条目,交给 dataclass 默认值
             if "thinking_observation" in fields:
-                fields["thinking_observation"] = ThinkingObservation(fields["thinking_observation"])
+                fields["thinking_observation"] = _coerce_observation(fields["thinking_observation"])
             fields.update(
                 cache_hit=True,
                 latency_ms=0,
diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py
index 04251de..b330199 100644
--- a/src/polygateway/middleware/telemetry.py
+++ b/src/polygateway/middleware/telemetry.py
@@ -55,6 +55,31 @@ def _canonical_meta_json(meta: Mapping[str, Any]) -> str:
     return json.dumps(dict(meta), sort_keys=True, ensure_ascii=False, allow_nan=False)
 
 
+def _normalize_observation(raw: object) -> str:
+    """三态裁定 → 落库用的裸 str;不是枚举也不在取值域时降级为 `unknown` 并告警。
+
+    **不写 `raw.value`**: `LLMResponse` 是无运行时校验的 frozen dataclass,下游
+    (尤其迁移期的测试替身)写 `LLMResponse(..., thinking_observation="observed")`
+    完全自然、`==` 比较照常成立,而 `.value` 会当场抛 `AttributeError`,被 `_record`
+    的 `except Exception` 吞成一条泛化 warning —— 丢的不是这一列,是**整行**,而
+    "遥测必录"是铁律。
+
+    域外取值同样只降级不抛: 直接 `ThinkingObservation(raw)` 会抛 `ValueError`,
+    落到同一个 `except` 上、同样丢整行,那只修好了裸 str 一半(口误值对测试替身
+    一样自然)。降级到 `unknown` 是诚实的——库确实判不出这个取值的含义,而单独
+    一条点名取值的 warning 保证它不被掩盖(P5 不许默认值掩盖错误)。
+    """
+    try:
+        return ThinkingObservation(raw).value
+    except ValueError:
+        logger.warning(
+            "thinking_observation 取值 {!r} 不在取值域内,本行降级记为 unknown"
+            "(其余列照常落库);调用方应传 ThinkingObservation 成员",
+            raw,
+        )
+        return ThinkingObservation.UNKNOWN.value
+
+
 def _cap_text(text: str, cap: int | None) -> str:
     """超出 cap 时头部硬切并附省略标记 `…(略 N 字)`;cap 为 None 原样返回。"""
     if cap is None or len(text) <= cap:
@@ -285,7 +310,9 @@ class TelemetryEmitter:
         model_reported: str | None,
         sampling: str | None,
         reasoning_tokens: int | None,
-        # issue #16: 枚举形态进来,取 `.value` 后才下沉(归一化同样在本方法内收口)
+        # issue #16: 枚举形态进来,归一化成裸 str 后才下沉(收口在 `_record` 内)。
+        # 注解是契约,但 `LLMResponse` 无运行时校验,故 `_normalize_observation`
+        # 仍按外部输入防御——违约的代价不该是丢掉整行遥测
         thinking_observation: ThinkingObservation,
         # issue #11: 未归一化的调用方维度,归一化在本方法内收口(recorder 只落库)
         tenant_id: str | None,
@@ -339,10 +366,10 @@ class TelemetryEmitter:
                 # 对所有人永久不可见,空串则可用一条 SQL 审计出未归属的行
                 tenant_id=tenant_id or "",
                 meta=_canonical_meta_json(meta),
-                # 取 `.value` 落裸 str: `StrEnum` 虽是 `str` 子类,asyncpg 的参数
-                # 编码对子类不保证接受,而遥测写失败只降级成一条 warning——不会当场
-                # 炸,只会让 Postgres 那一路悄悄少一列数据
-                thinking_observation=thinking_observation.value,
+                # 落裸 str: `StrEnum` 虽是 `str` 子类,asyncpg 的参数编码对子类不
+                # 保证接受,而遥测写失败只降级成一条 warning——不会当场炸,只会让
+                # Postgres 那一路悄悄少一列数据
+                thinking_observation=_normalize_observation(thinking_observation),
             )
         except asyncio.CancelledError:
             raise
diff --git a/src/polygateway/thinking.py b/src/polygateway/thinking.py
index 36c17d0..7806d8c 100644
--- a/src/polygateway/thinking.py
+++ b/src/polygateway/thinking.py
@@ -32,7 +32,11 @@ def observe_thinking(*, thinking: str, reasoning_tokens: int | None) -> Thinking
     """
     if thinking.strip():
         return ThinkingObservation.OBSERVED
-    if reasoning_tokens is None:
+    # 负数与 None 同档: `ABSENT` 是"上游明确上报未推理"这个最强的正面结论,坏
+    # 数据给不出它。当前 transport 已在边界把负数归 None,这里仍要自己闭合——本
+    # 函数对外承诺"外部输入校验后使用",第二个 transport 直接填该值时,漏判会
+    # 给出一个方向相反的强结论(P5)
+    if reasoning_tokens is None or reasoning_tokens < 0:
         return ThinkingObservation.UNKNOWN
     return ThinkingObservation.OBSERVED if reasoning_tokens > 0 else ThinkingObservation.ABSENT
 
diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py
index bc34bb6..01c879c 100644
--- a/src/polygateway/transports/openai_compat.py
+++ b/src/polygateway/transports/openai_compat.py
@@ -322,9 +322,12 @@ class OpenAICompatTransport:
         # 未登记模型只喊一次: 装配期已喊过,逐次调用再喊是日志洪水。
         # 实例级而非模块级 —— 模块级可变状态违反纯 asyncio 中立铁律
         self._warned_models: set[str] = set()
-        # 对账告警独立节流,**不复用** `_warned_models`: 那个 set 的语义是"未登记
-        # 能力已告警过",两件事共用一个开关会互相压制——一方喊过就把另一方静音
-        self._warned_mismatches: set[tuple[str, bool | None]] = set()
+        # 对账告警独立节流,**不复用** `_warned_models`: 两者语义不同(那个 set 记
+        # 的是"未登记能力已告警过",这个记的是"某源某方向的矛盾已告警过"),共用
+        # 一个容器会让两种告警的生命周期纠缠在一起——将来任一侧想加清空/过期策略,
+        # 都会连带改掉另一侧的行为。(键空间恰好不相交,故当下**不会**互相压制;
+        # 分开维护的理由是语义,不是碰撞)
+        self._warned_mismatches: set[tuple[str, str, bool | None]] = set()
         self._client_factory = client_factory or _default_client_factory
         self._clients: dict[str, httpx.AsyncClient] = {}
 
@@ -410,11 +413,20 @@ class OpenAICompatTransport:
         return result
 
     def _warn_on_thinking_mismatch(self, source: SourceConfig, result: TransportResult) -> None:
-        """声明与观测矛盾即 warning;按 (model, direction) 节流,同组合只喊一次。
+        """声明与观测矛盾即 warning;按 (source, model, direction) 节流,同组合只喊一次。
 
-        节流键必须含方向: 同一模型的开、关两档是两个独立的矛盾,合并键会让先出现
-        的那一档把另一档永久静音。逐次调用刷屏会把告警变成噪声,噪声等于没有告警。
+        三段缺一不可。**方向**: 同一模型的开、关两档是两个独立的矛盾。**源名**:
+        多源多账号是本库的核心场景,同一 model 跨 N 个源是常态,而每个源背后是
+        独立的账号/网关,一个源的行为不代表另一个——漏掉源名,5 个源里第一个出
+        问题的喊完一次,其余四个永久静音。逐次调用刷屏会把告警变成噪声,噪声等于
+        没有告警。
+
+        **先判键再对账**: `reconcile_thinking` 会拼含完整 `evidence` 的长字符串,
+        而非流式档每次调用都命中这一分支,节流后再拼是纯粹的热路径浪费。
         """
+        key = (source.name, source.model, source.enable_thinking)
+        if key in self._warned_mismatches:
+            return
         message = reconcile_thinking(
             enable_thinking=source.enable_thinking,
             observation=result.thinking_observation,
@@ -423,11 +435,13 @@ class OpenAICompatTransport:
         )
         if message is None:
             return
-        key = (source.model, source.enable_thinking)
-        if key in self._warned_mismatches:
-            return
         self._warned_mismatches.add(key)
-        logger.warning(message)
+        # 源名拼在调用点而不是加进 `reconcile_thinking` 的签名: 那是纯判定函数,
+        # 输入只该含判定依据(声明/观测/能力/模型),源名是**定位信息**,进不了判据。
+        # 单参数传入 loguru: 文案里带 `thinking:{type:disabled}` 这类字面花括号
+        # (能力表 evidence),将来有人给这行加个格式化参数就会炸在成功调用的返回
+        # 路径上(与 telemetry/sqlite.py 的缺列告警同一先例)
+        logger.warning("源 {} —— {}", source.name, message)
 
     async def embed(
         self, *, texts: list[str], source: SourceConfig, call_id: str
diff --git a/tests/e2e/test_thinking_live.py b/tests/e2e/test_thinking_live.py
index b146915..db35e85 100644
--- a/tests/e2e/test_thinking_live.py
+++ b/tests/e2e/test_thinking_live.py
@@ -413,6 +413,31 @@ class TestOtherProviders:
         )
         assert len(offs) == len(obs), f"{provider} 关闭方向未满足: {obs}"
 
+    async def test_qwen_enabled_is_observed(self):
+        """设计 §14 验收: qwen 开启档必须裁定为 `OBSERVED`,不是 `UNKNOWN`。
+
+        本条是三态裁定的**跨供应商对照组**: MiniMax 这一路两个信号都可能缺失
+        (非流式档整片 `UNKNOWN`),若只按它调判据,很容易把"观测不到"当成常态;
+        qwen 在同一网关同一 key 上照常返回推理信号(findings 2026-08-25 §2),
+        故这里能且必须要求正面结论——它一旦掉成 `UNKNOWN`,说明的是库的组装路径
+        丢了信号,而不是上游行为变了。
+        """
+        matrix, provider, model = "L6b", "qwen", "qwen3.7-plus"
+        desc = f"{provider} enable_thinking=True"
+        try:
+            obs = await _run_rounds(_ROUNDS, provider=provider, model=model, enable_thinking=True)
+        except (AllSourcesExhausted, SourceDeadError, TransientError) as exc:
+            _skip_if_unreachable(exc, matrix, desc)
+        ons = [o for o in obs if _reasoning_on(o)]
+        _record(
+            matrix,
+            desc,
+            "PASS" if len(ons) * 2 > len(obs) else "FAIL",
+            f"{len(ons)}/{len(obs)} 轮观测到推理(OBSERVED)",
+            obs,
+        )
+        assert len(ons) * 2 > len(obs), f"{provider} 开启方向要求多数轮 OBSERVED: {obs}"
+
 
 class TestCapabilityDrift:
     """L8 漂移哨兵: 能力表过期是必然事件,这里是它的过期告警。"""
diff --git a/tests/unit/test_cache.py b/tests/unit/test_cache.py
index 156e6a1..290ba64 100644
--- a/tests/unit/test_cache.py
+++ b/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。"""
diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py
index e2f365b..2791828 100644
--- a/tests/unit/test_openai_compat.py
+++ b/tests/unit/test_openai_compat.py
@@ -547,10 +547,11 @@ class TestThinkingObservationVerdict:
 
 
 class TestThinkingReconciliation:
-    """对账告警按 (model, direction) 节流(设计 §5)。
+    """对账告警按 (source, model, direction) 节流(设计 §5)。
 
-    节流键必须含方向: 同一模型的开、关两档是两个独立的矛盾,合并键会让先出现
-    的那一档把另一档永久静音。
+    键的三段缺一不可,理由同源: 合并任意一段,都会让先出现的那一组把另一组
+    永久静音——同一模型的开/关两档是两个独立的矛盾,同一模型的两个源背后是
+    两个独立的账号/网关。
     """
 
     def _handler(self, request):
@@ -563,9 +564,9 @@ class TestThinkingReconciliation:
         # 开启档却零信号 → UNKNOWN,无法确认是否生效(M3 实测形态)
         return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE))
 
-    def _minimax(self, enable_thinking):
+    def _minimax(self, enable_thinking, name="mm"):
         return _source(
-            name="mm", provider="minimax", model="MiniMax-M3", enable_thinking=enable_thinking
+            name=name, provider="minimax", model="MiniMax-M3", enable_thinking=enable_thinking
         )
 
     async def test_same_model_and_direction_warns_only_once(self):
@@ -581,6 +582,36 @@ class TestThinkingReconciliation:
         hits = [m for m in messages if "MiniMax-M3" in m]
         assert len(hits) == 1, f"同一 (model, direction) 应只告警一次,实得 {len(hits)} 次"
 
+    async def test_each_source_gets_its_own_warning(self):
+        """多源多账号是本库的核心场景: 同一 model 跨 N 个源不得只喊第一个。
+
+        节流键漏掉源标识时,5 个共用同一模型的源里第一个出问题的喊完一次,其余
+        四个**永久静音**——而每个源背后是独立的账号/网关,它们的行为互不代表。
+        """
+        transport = _transport_for(self._handler)
+        messages: list[str] = []
+        sink_id = logger.add(messages.append, level="WARNING")
+        try:
+            await _complete(transport, self._minimax(False, name="gw-a"))
+            await _complete(transport, self._minimax(False, name="gw-b"))
+        finally:
+            logger.remove(sink_id)
+        hits = [m for m in messages if "MiniMax-M3" in m]
+        assert len(hits) == 2, f"两个源各应告警一次,实得 {len(hits)} 次"
+
+    async def test_the_warning_names_the_source(self):
+        """拿到告警的人得知道该查哪个网关: 只报模型名定位不到源。"""
+        transport = _transport_for(self._handler)
+        messages: list[str] = []
+        sink_id = logger.add(messages.append, level="WARNING")
+        try:
+            await _complete(transport, self._minimax(False, name="gw-a"))
+        finally:
+            logger.remove(sink_id)
+        hits = [m for m in messages if "MiniMax-M3" in m]
+        assert len(hits) == 1
+        assert "gw-a" in hits[0], f"告警未点名出问题的源: {hits[0]}"
+
     async def test_switching_direction_earns_a_second_warning(self):
         transport = _transport_for(self._handler)
         messages: list[str] = []
diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py
index e2294e0..c91188e 100644
--- a/tests/unit/test_telemetry.py
+++ b/tests/unit/test_telemetry.py
@@ -9,6 +9,7 @@ import subprocess
 from pathlib import Path
 
 import pytest
+from loguru import logger
 
 from polygateway.backends.memory.breaker import InMemoryGate
 from polygateway.backends.memory.limiter import InMemoryLimiter
@@ -1169,6 +1170,57 @@ class TestEmitterThinkingObservation:
         assert value == "observed"
         assert type(value) is str  # 不是 ThinkingObservation: 子类实例不得下沉到 recorder
 
+    async def test_a_bare_string_verdict_still_lands(self):
+        """下游填裸 str 时**整行**不得丢失(遥测必录)。
+
+        `LLMResponse` 是无运行时校验的 frozen dataclass,写
+        `LLMResponse(..., thinking_observation="observed")` 完全自然且 `==` 比较
+        照常成立;若 emitter 直接取 `.value`,这里会抛 `AttributeError` 并被
+        `_record` 的 `except Exception` 吞成一条泛化 warning——丢的不是这一列,
+        是整行,正是 1.3.0 那次"19 次调用一行未落"的同款形态。
+        """
+        rec = _MemoryRecorder()
+        await TelemetryEmitter(rec, text_cap=None).emit_attempt(
+            request=_REQ,
+            source=_source(),
+            call_id="c",
+            latency_ms=1,
+            response=_resp(thinking_observation="observed"),
+            error=None,
+        )
+        assert len(rec.rows) == 1, "整行被吞了"
+        value = rec.rows[0]["thinking_observation"]
+        assert value == "observed"
+        assert type(value) is str
+
+    async def test_an_out_of_domain_verdict_degrades_but_keeps_the_row(self):
+        """域外取值挡在落库前,但**降级不丢行**: 列的取值域由库守,代价不是整行。
+
+        直接 `ThinkingObservation(x).value` 会在这里抛 `ValueError`,同样被
+        `_record` 的 `except Exception` 吞成丢整行——那只修好了裸 str 一半,
+        口误值(大小写不符、拼错)对测试替身同样自然。故降级为 `unknown`
+        (对库而言本次确实判不出来)并单独告警,与缓存回放的方向选择一致。
+        """
+        rec = _MemoryRecorder()
+        messages: list[str] = []
+        sink_id = logger.add(messages.append, level="WARNING")
+        try:
+            await TelemetryEmitter(rec, text_cap=None).emit_attempt(
+                request=_REQ,
+                source=_source(),
+                call_id="c",
+                latency_ms=1,
+                response=_resp(thinking_observation="OBSERVED"),  # 大小写不符即域外
+                error=None,
+            )
+        finally:
+            logger.remove(sink_id)
+        assert len(rec.rows) == 1, "整行被吞了"
+        assert rec.rows[0]["thinking_observation"] == "unknown"
+        hits = [m for m in messages if "OBSERVED" in m]
+        assert len(hits) == 1, f"域外取值必须单独告警: {messages}"
+        assert [m for m in messages if "遥测记录失败" in m] == []
+
     async def test_cache_hit_replays_the_recorded_verdict(self):
         """缓存命中回放历史那次的裁定: 与 model/prompt_tokens 同一口径。"""
         rec = _MemoryRecorder()
diff --git a/tests/unit/test_thinking.py b/tests/unit/test_thinking.py
index 75cacb6..c6fa423 100644
--- a/tests/unit/test_thinking.py
+++ b/tests/unit/test_thinking.py
@@ -66,6 +66,18 @@ class TestObserveThinking:
             observe_thinking(thinking="想了想", reasoning_tokens=0) is ThinkingObservation.OBSERVED
         )
 
+    @pytest.mark.parametrize("negative", [-1, -205])
+    def test_negative_token_count_is_not_evidence_of_absence(self, negative):
+        """负数是坏数据,不是"上游明确上报未推理"这个最强的正面结论。
+
+        当前 transport 已在边界把负数归 `None`,所以这条走不通;但本函数的
+        docstring 自称"外部输入校验后使用",第二个 transport 直接填该值时,
+        `> 0 else ABSENT` 会给出一个方向相反的强结论。函数自身必须闭合(P5)。
+        """
+        assert observe_thinking(thinking="", reasoning_tokens=negative) is (
+            ThinkingObservation.UNKNOWN
+        )
+
 
 class TestThinkingObservationEnum:
     def test_values_are_stable_strings(self):
@@ -85,7 +97,12 @@ class TestThinkingObservationEnum:
 
 @pytest.mark.parametrize("bogus", ["", "OBSERVED", "yes", "none"])
 def test_unknown_strings_are_rejected(bogus):
-    """非法值必须抛 ValueError: 缓存回放靠它把污染数据挡成"未命中"(设计 §6)。"""
+    """非法值必须抛 ValueError: 缓存回放与遥测归一化都靠它识别域外取值(设计 §6)。
+
+    两处接住这个 ValueError 后**降级而非作废**(缓存复活内容 + 记 UNKNOWN、遥测
+    照常落行),但降级的前提是构造器真的会拒绝——它一旦放行,域外取值就会一路
+    进到 `LLMResponse` 与遥测列里。
+    """
     with pytest.raises(ValueError):
         ThinkingObservation(bogus)
 
@@ -239,6 +256,23 @@ class TestReconcileThinking:
         assert msg is not None
         assert "MiniMax-M3" in msg
 
+    def test_off_and_absent_stays_silent(self):
+        """要求关闭 + 上游明确上报未推理 = 要求被满足,没有可报的矛盾。
+
+        这一格与 `test_off_and_unknown_stays_silent` 的沉默理由**不同**: 那里是
+        "没有证伪力",这里是"正面证实要求已满足"。两者都必须沉默,漏测哪一格,
+        把 Phase 2 的判据写成 `is ABSENT` 之类的反向条件都不会被抓住。
+        """
+        assert (
+            reconcile_thinking(
+                enable_thinking=False,
+                observation=ThinkingObservation.ABSENT,
+                capability=self._CAP,
+                model="qwen3.7-plus",
+            )
+            is None
+        )
+
     def test_off_and_unknown_stays_silent(self):
         """UNKNOWN 没有证伪力: 拿它报警等于每次关闭调用都喊(M3 关闭档恒落此档)。"""
         assert (