feat: warn when the capability table and reality disagree

The M3 evidence sat at 08-02 for twenty-three days while nobody could
tell whether it still held. A declaration that goes stale in silence is
the failure this issue is really about, so the library now compares
what it declared against what it just observed and says so when the two
part ways.

Judgement is separated from logging: reconcile_thinking returns the
warning text, so tests assert on the text instead of parsing logs.
Two cases that look alike are kept apart — a model whose capability is
registered gets a drift warning quoting its evidence, an unregistered
one is never told the table said anything, because it never did.

False x UNKNOWN stays silent on purpose. UNKNOWN cannot falsify
anything, and warning on it would fire on every disabled call M3 makes
over the plain endpoint. A warning that always fires is not a warning.
This commit is contained in:
2026-08-26 00:23:57 -04:00
parent 3e869b9b39
commit 20a4a9ae47
4 changed files with 260 additions and 2 deletions
+50
View File
@@ -546,6 +546,56 @@ class TestThinkingObservationVerdict:
assert result.thinking_observation is ThinkingObservation.ABSENT
class TestThinkingReconciliation:
"""对账告警按 (model, direction) 节流(设计 §5)。
节流键必须含方向: 同一模型的开、关两档是两个独立的矛盾,合并键会让先出现
的那一档把另一档永久静音。
"""
def _handler(self, request):
payload = json.loads(request.content)
if payload.get("reasoning_effort") == "none":
# 关闭档却回了推理正文 → OBSERVED,与"要求关闭"矛盾
return _sse_stream(
_chunk(reasoning="偷偷想了"), _chunk(content="ok"), _chunk(usage=_USAGE)
)
# 开启档却零信号 → UNKNOWN,无法确认是否生效(M3 实测形态)
return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE))
def _minimax(self, enable_thinking):
return _source(
name="mm", provider="minimax", model="MiniMax-M3", enable_thinking=enable_thinking
)
async def test_same_model_and_direction_warns_only_once(self):
transport = _transport_for(self._handler)
source = self._minimax(False)
messages: list[str] = []
sink_id = logger.add(messages.append, level="WARNING")
try:
await _complete(transport, source)
await _complete(transport, source)
finally:
logger.remove(sink_id)
hits = [m for m in messages if "MiniMax-M3" in m]
assert len(hits) == 1, f"同一 (model, direction) 应只告警一次,实得 {len(hits)}"
async def test_switching_direction_earns_a_second_warning(self):
transport = _transport_for(self._handler)
messages: list[str] = []
sink_id = logger.add(messages.append, level="WARNING")
try:
await _complete(transport, self._minimax(False))
await _complete(transport, self._minimax(False))
await _complete(transport, self._minimax(True))
await _complete(transport, self._minimax(True))
finally:
logger.remove(sink_id)
hits = [m for m in messages if "MiniMax-M3" in m]
assert len(hits) == 2, f"两个方向各应告警一次,实得 {len(hits)}"
class TestNonStreamFastPath:
async def test_non_stream_parses_message(self):
def handler(request):
+117
View File
@@ -14,6 +14,7 @@ from polygateway.thinking import (
ThinkingCapability,
get_capability,
observe_thinking,
reconcile_thinking,
register_capability,
resolve_thinking,
)
@@ -160,3 +161,119 @@ class TestResolveThinking:
cap = ThinkingCapability(can_disable=False, evidence="构造")
with pytest.raises(ValueError, match="register_provider"):
resolve_thinking(get_provider("openai"), cap, False, model="whatever")
class TestReconcileThinking:
"""声明 × 观测对账(设计 §5): 矛盾出文案,不表态出 None。
文案本身是被断言对象——判定与日志分离正是为此: 告警内容可直接比对,不必
去解析日志格式。
"""
_CAP = ThinkingCapability(
can_disable=True, evidence="2026-08-02 实测 reasoning_effort=none 可关闭"
)
def test_off_but_observed_with_a_registered_capability_blames_the_table(self):
"""已登记却实测推理了 = 能力表漂移: 必须附 evidence 与更新指路。"""
msg = reconcile_thinking(
enable_thinking=False,
observation=ThinkingObservation.OBSERVED,
capability=self._CAP,
model="MiniMax-M3",
)
assert msg is not None
assert "MiniMax-M3" in msg
assert "2026-08-02 实测 reasoning_effort=none 可关闭" in msg
assert "register_capability" in msg
def test_off_but_observed_unregistered_never_claims_a_table_entry(self):
"""未登记模型没有"能力表声称"这回事——说它就是撒谎。"""
msg = reconcile_thinking(
enable_thinking=False,
observation=ThinkingObservation.OBSERVED,
capability=None,
model="MiniMax-M9",
)
assert msg is not None
assert "MiniMax-M9" in msg
assert "能力表" not in msg
assert "register_capability" in msg
def test_registered_and_unregistered_wordings_differ(self):
registered = reconcile_thinking(
enable_thinking=False,
observation=ThinkingObservation.OBSERVED,
capability=self._CAP,
model="MiniMax-M3",
)
unregistered = reconcile_thinking(
enable_thinking=False,
observation=ThinkingObservation.OBSERVED,
capability=None,
model="MiniMax-M3",
)
assert registered != unregistered
@pytest.mark.parametrize("capability", [None, _CAP])
def test_on_but_absent_is_a_contradiction(self, capability):
"""上游明确上报未推理: 这是唯一的正面证伪,与能力表登记与否无关。"""
msg = reconcile_thinking(
enable_thinking=True,
observation=ThinkingObservation.ABSENT,
capability=capability,
model="qwen3.7-plus",
)
assert msg is not None
assert "qwen3.7-plus" in msg
@pytest.mark.parametrize("capability", [None, _CAP])
def test_on_but_unknown_admits_it_cannot_confirm(self, capability):
"""issue #17 的诚实版本: 明说"我注入了,但我看不见结果""""
msg = reconcile_thinking(
enable_thinking=True,
observation=ThinkingObservation.UNKNOWN,
capability=capability,
model="MiniMax-M3",
)
assert msg is not None
assert "MiniMax-M3" in msg
def test_off_and_unknown_stays_silent(self):
"""UNKNOWN 没有证伪力: 拿它报警等于每次关闭调用都喊(M3 关闭档恒落此档)。"""
assert (
reconcile_thinking(
enable_thinking=False,
observation=ThinkingObservation.UNKNOWN,
capability=self._CAP,
model="MiniMax-M3",
)
is None
)
@pytest.mark.parametrize(
"observation",
[ThinkingObservation.OBSERVED, ThinkingObservation.ABSENT, ThinkingObservation.UNKNOWN],
)
def test_no_request_no_grievance(self, observation):
"""调用方不表态,就无从谈"违背""""
assert (
reconcile_thinking(
enable_thinking=None,
observation=observation,
capability=self._CAP,
model="MiniMax-M3",
)
is None
)
def test_on_and_observed_is_exactly_what_was_asked_for(self):
assert (
reconcile_thinking(
enable_thinking=True,
observation=ThinkingObservation.OBSERVED,
capability=self._CAP,
model="MiniMax-M3",
)
is None
)