From 3eb22d2a55a5be687f7d12b96ec94072c0fb201a Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 9 Sep 2026 03:34:48 -0400 Subject: [PATCH] test: fix structured reask evidence and live coverage conclusions --- ...09-09-134-thinking-contracts-validation.md | 32 +- .../2026-09-09-134-thinking-contracts.md | 6 + tests/e2e/conftest.py | 48 ++- tests/e2e/test_smoke_gateway.py | 8 +- tests/e2e/test_thinking_live.py | 155 +++++++-- tests/live_evidence.py | 12 +- tests/unit/test_live_evidence.py | 307 ++++++++++++++++++ 7 files changed, 540 insertions(+), 28 deletions(-) diff --git a/research-wiki/findings/2026-09-09-134-thinking-contracts-validation.md b/research-wiki/findings/2026-09-09-134-thinking-contracts-validation.md index 7f5c7bd..5f9b1a5 100644 --- a/research-wiki/findings/2026-09-09-134-thinking-contracts-validation.md +++ b/research-wiki/findings/2026-09-09-134-thinking-contracts-validation.md @@ -106,9 +106,39 @@ date: 2026-09-09 | make check初报SIM117/B017 | 合并测试上下文，按真实解析异常指定类型，不加ignore；最终静态门通过 | | pi-lens解释器/StrEnum噪音 | 记录 `t56-diagnostics.txt`，conda内真实导入与ruff为门，不改任务外枚举；pytest wrapper generator的return report是协议必需，独立next/send/StopIteration.value测试通过 | | T10型号→400机器字段基线不存在 | 父会话明确确认：不编造白名单，实际400默认FAIL并逐轮留证。纯负向契约精确类型/状态/type单独测试；具体live预期拒绝未验证、需人工基线 | -| 结构化反馈重问 | 请求摘要预期固定，发生反馈重问更改messages时保守FAIL，不从待测payload补齐预期；未改生产结构化行为，不声称该取证分支已取得live覆盖 | +| 结构化反馈重问（已被本次审查修复替代） | 原固定摘要会误拒正常反馈重问，独立审查判 P1；不再保留为可接受限制，修复与真实 StructuredMW 离线两响应证据见下节 | | 发布/集成/live/下游 | 本任务未执行，M2空wire、M3非流式UNKNOWN、身份不足、三项目实际配置缺失仍保留为证据门 | 文档已同步README M1–M9、CHANGELOG未发布段、env注释、ARCH D11/5.1/7.5/7.8、旧设计替代指针及既有schema/metric；无版本bump、无新生产字段/DDL。Wiki站已下线，不虚报线上页更新。 续作提交：`73008ad test: apply evidence-based live checks without hiding regressions`。最终提交前实际门：`t56-precommit-unit.log/.exit` **1357 passed/0**，`t56-precommit-affected.log` **331 passed**（包含生产默认factory节点），`t56-precommit-check.log/.exit` **make check通过/0**，`t56-precommit-collect.log` **90 collected**，`t56-precommit-compile.log`通过；`git diff --check`通过，`git diff --quiet -- src`确认生产零差异。T8仅文档部分完成，不勾选完整验收门。 + +## 独立审查四项修复（起点 d332287） + +按 receiving-code-review 对照实际调用链核验 `verify134/live-contracts.md`：四项均成立。此处沿用户限定仅更新既有 finding／plan，不新建图实体或扩写设计。生产／版本零差异；没有 live、付费请求或子代理。本节是实现者核验，不冒充新一轮独立复审。 + +| 审查项／核验依据 | 最小修复与守卫 | +| --- | --- | +| P1 结构化重问：StructuredMW._with_feedback 追加两消息，旧 hook 固定完整摘要必错 | 仅结构化模型 smoke 启用原提示词前缀摘要、成对 assistant/user 字符串与已有预算；首个 attempt 仍精确原消息。薄委托保存本次摘要只校验 HTTP 保真，不从 wire 反填预期，不关闭重问。真实 GatewayClient＋StructuredMW＋MockTransport 缺字段→合法响应恰两 HTTP PASS；破坏前缀、角色、内容类型、配对、预算、wire 均 FAIL；首轮凭空反馈另有 FAIL 守卫 | +| P1 未登记候选：旧 None 分支被置 cannot_disable，ABSENT 假失败 | 明确 observation-only，先全轮请求／身份资格，再 UNCOVERED；ABSENT／OBSERVED／UNKNOWN 与资格 FAIL 四组运行真实 T10 消费者（剔除 .env 读取语句），不调用模型、不改长复核条件或轮次 | +| P1 结论重新 UUID，多个型号失去对应关系 | 每用例显式传同一 run/model；子运行用该 run 下唯一 matrix_id 关联，结论含完整计划／完成分母。L1–L8、T10短长／各档／默认全部复用；两型号一 PASS 一 UNCOVERED，在 NONE、tiers、L8 三组逐文件验证关联和轮数 | +| P2 机器字段未落盘 | 仅精确 model_not_found 保留；其他字符串（含假凭据 sentinel）为 omitted；写入端再拒未知机器值，绝不输出任意上游 type | + +### 本轮先红后绿与验证 + +日志位于 `tests/outputs/134/`，各命令结果后立即保存 `.exit`，原命令不接管道。 + +| 命令／节点 | 红证据 | 绿证据 | +| --- | --- | --- | +| `pytest tests/unit/test_live_evidence.py -k structured_reask -q` | `review-f1-red`：1 failed／6 passed，合法两响应误判 FAIL，exit 1 | `review-f1-green`：7 passed，exit 0 | +| `pytest tests/unit/test_live_evidence.py -k first_attempt_requires -q` | `review-f1-first-red`：首轮多反馈被放行，1 failed，exit 1 | `review-final-affected`：含该节点共129 passed，exit 0 | +| `pytest tests/unit/test_live_evidence.py -k unregistered_candidate -q` | `review-f2-red`：ABSENT 被误判FAIL，1 failed／3 passed，exit 1 | `review-f2-green`：4 passed，exit 0 | +| `pytest tests/unit/test_live_evidence.py -k capability_conclusions -q` | `review-f3-red`：三组结论缺型号断言红，3 failed，exit 1 | `review-f3-green`：三组＋未登记四组共7 passed，exit 0 | +| `pytest tests/unit/test_live_evidence.py -k safe_machine_type -q` | `review-f4-red`：三组缺机器字段，3 failed，exit 1 | `review-f4-green`：3 passed，exit 0 | +| `conda run --no-capture-output -n PolyGateway pytest tests/unit/ -q` | — | `review-final-unit`：**1375 passed，exit 0** | +| `make check` | — | `review-final-check`：格式／ruff／import-linter 1 kept，exit 0 | +| `conda run --no-capture-output -n PolyGateway pytest tests/e2e/ -m slow --collect-only -q` | — | `review-final-collect`：**90 collected，exit 0**；不是90通过 | + +上表节点命令也均加 `conda run --no-capture-output -n PolyGateway`；未使用缺 import 的伪红。最终受影响文件129单测通过，相比111新增18节点。pi-lens仍提示非conda缺httpx/pytest/pydantic及StrEnum等旧噪音，按已批准方向记录后继续conda门；新增测试命名空间显式 Any 类型，不抑制真实错误。 + +**尚未完成**：修复后独立复审、集成／make test覆盖率、真实能力取证、下游迁移和发布；T8/T9保持未勾选。M2空wire、M3非流UNKNOWN、真实机器拒绝白名单及外部服务状态不因离线绿变成已覆盖。 diff --git a/research-wiki/plans/2026-09-09-134-thinking-contracts.md b/research-wiki/plans/2026-09-09-134-thinking-contracts.md index eec7893..afa5fa5 100644 --- a/research-wiki/plans/2026-09-09-134-thinking-contracts.md +++ b/research-wiki/plans/2026-09-09-134-thinking-contracts.md @@ -385,3 +385,9 @@ T0–T4／T7 的命令、实际失败与修复、11 个隔离变异及 1241 项 父会话确认无已批准型号→400机器type白名单：不编造，缺机器证据400默认FAIL；精确预期拒绝契约离线守卫，具体live负向缺基线记录未验证。不可关闭命题完整合格轮次有OBSERVED支持本条件下未关闭，全ABSENT证伪，无OBSERVED但UNKNOWN未覆盖。T8复选框保持未勾选，因为独立verifier与全量/live证据门未执行；本轮仅其文档同步部分完成，禁止发布。 T5/T6实现提交：`73008ad`。最终日常单元1357、受影响含factory331、make check、compileall、e2e collect-only90通过；完整T8/T9仍未执行。日志路径及8项红→还原绿详见同一finding。 + +### 独立审查修复续作（d332287 后） + +已按 receiving-code-review 核验四项并仅修改测试及本计划／finding：结构化重问采用先验前缀／反馈角色与预算＋委托摘要的 wire 保真；未登记候选先资格再观测未覆盖；同一用例 run/model 关联所有子运行并保留计划／完成分母；落盘机器字段只准 model_not_found／omitted。没有生产／版本修改、slow执行或额外调用预算。 + +新增18个离线节点，四项及首轮精确消息守卫均有目标断言先红→绿。最终129项取证单测、1375全单元、make check及e2e collect-only90通过；命令日志／退出码详见既有finding“独立审查四项修复”。修复后独立复审、集成与live尚未完成，**T8/T9仍不勾选，不放行发布**。原结构化重问“保守FAIL”说明已标为被本次修复替代，不能再当成可接受限制。 diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index eef1582..c5a65cd 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -51,6 +51,8 @@ class _Attempt: call_id: str exchanges: list[_Exchange] = field(default_factory=list) + messages_digest: str | None = None + messages_valid: bool = False class LiveCapture: @@ -66,6 +68,14 @@ class LiveCapture: raise ValueError("取证矩阵缺少必需预期或混用 chat/embed") if not isinstance(expected["control"], dict): raise ValueError("control 必须是显式对象") + if "structured_max_retries" in expected and ( + "stream" not in expected + or type(expected["structured_max_retries"]) is not int + or expected["structured_max_retries"] < 0 + or type(expected.get("messages_prefix_length")) is not int + or expected["messages_prefix_length"] < 1 + ): + raise ValueError("结构化预期缺少合法前缀长度或重问预算") self._expectations = {name: dict(value) for name, value in expectations.items()} self._round: ContextVar[tuple[str, str]] = ContextVar("live_round") self._attempt: ContextVar[_Attempt] = ContextVar("live_attempt") @@ -146,6 +156,29 @@ class LiveCapture: self._notes[key].append("原始 JSON 身份无法独立解析") return HttpEvidence(call_id, exchange.checks, status, body, identity) + def observe_messages(self, source: SourceConfig, messages: list[dict[str, Any]]) -> None: + """先验前缀／反馈契约与委托摘要分开；摘要仅验证 HTTP 序列化保真。""" + expected = self._expectations[source.name] + if "structured_max_retries" not in expected: + return + attempt = self._attempt.get() + prefix_length = expected["messages_prefix_length"] + feedback = messages[prefix_length:] + attempt.messages_digest = messages_digest(messages) + attempt.messages_valid = ( + (not feedback or bool(self._records[self._round.get()])) + and messages_digest(messages[:prefix_length]) == expected["messages_digest"] + and len(feedback) % 2 == 0 + and len(feedback) <= 2 * expected["structured_max_retries"] + and all( + isinstance(message, dict) + and set(message) == {"role", "content"} + and message["role"] == ("assistant" if index % 2 == 0 else "user") + and isinstance(message["content"], str) + for index, message in enumerate(feedback) + ) + ) + def client_factory(self, source: SourceConfig) -> httpx.AsyncClient: """鉴权仅内存比较；沿已校验源 timeout/trust_env。""" expected = self._expectations[source.name] @@ -176,8 +209,13 @@ class LiveCapture: "model": payload.get("model") == expected["model"], "authorization": request.headers.get("Authorization") == f"Bearer {source.api_key}", "control": control == expected["control"], - "messages_digest": messages_digest(payload.get("messages", payload.get("input"))) - == expected["messages_digest"], + "messages_digest": ( + attempt.messages_valid + and messages_digest(payload.get("messages")) == attempt.messages_digest + if "structured_max_retries" in expected + else messages_digest(payload.get("messages", payload.get("input"))) + == expected["messages_digest"] + ), } if "stream" in expected: checks["stream"] = payload.get("stream") is expected["stream"] and ( @@ -255,6 +293,7 @@ class ObservedTransport: ) -> TransportResult: """与生产端口逐参数同签名。""" with self._capture.attempt_context(call_id): + self._capture.observe_messages(source, messages) return await self._transport.complete( messages=messages, source=source, @@ -315,6 +354,7 @@ def chat_expectations( messages: list[dict[str, Any]], stream: bool, controls: Mapping[str, dict[str, Any]], + structured_max_retries: int | None = None, ) -> dict[str, dict[str, Any]]: """URL 从源配置声明，控制片段必须由矩阵独立给出。""" result = {} @@ -330,6 +370,10 @@ def chat_expectations( "control": controls[source.name], "messages_digest": messages_digest(messages), } + if structured_max_retries is not None: + result[source.name].update( + messages_prefix_length=len(messages), structured_max_retries=structured_max_retries + ) return result diff --git a/tests/e2e/test_smoke_gateway.py b/tests/e2e/test_smoke_gateway.py index 208ad20..e0e22d8 100644 --- a/tests/e2e/test_smoke_gateway.py +++ b/tests/e2e/test_smoke_gateway.py @@ -40,7 +40,13 @@ async def _smoke(matrix, prompt, validate, *, stream=True, structured=None): controls = source_controls(settings) capture = LiveCapture( expectations=chat_expectations( - settings, messages=messages, stream=stream, controls=controls + settings, + messages=messages, + stream=stream, + controls=controls, + structured_max_retries=( + settings.structured_max_retries if isinstance(structured, type) else None + ), ) ) async with observed_client(settings, capture) as client: diff --git a/tests/e2e/test_thinking_live.py b/tests/e2e/test_thinking_live.py index 6a24c42..6c26e28 100644 --- a/tests/e2e/test_thinking_live.py +++ b/tests/e2e/test_thinking_live.py @@ -125,8 +125,36 @@ def _tier_settings(model): return dataclasses.replace(base, sources=(source,)) +@dataclasses.dataclass +class _CaseRun: + """单型号用例关联；子运行以同一 run_id 下的 matrix_id 唯一定位原件。""" + + run_id: str + model: str + subruns: list[dict] = dataclasses.field(default_factory=list) + + def report_fields(self): + """计划分母在收集前登记，完成数量只按实际回收轮次填写。""" + return { + "session_id": self.run_id, + "requested_model": self.model, + "subruns": self.subruns, + "planned_rounds": sum(group["planned_rounds"] for group in self.subruns), + "completed_rounds": sum(group["completed_rounds"] for group in self.subruns), + } + + async def _collect_rounds( - settings, *, rounds, stream, prompt, matrix_id, effort=None, capabilities=None, concurrency=1 + settings, + *, + run, + rounds, + stream, + prompt, + matrix_id, + effort=None, + capabilities=None, + concurrency=1, ): """保留所有失败轮，不把可用轮集合偷偷当新分母。""" if rounds < 1 or concurrency < 1: @@ -142,7 +170,13 @@ async def _collect_rounds( settings, messages=messages, stream=stream, controls=controls ) ) - run_id = uuid4().hex + if any(source.model != run.model for source in settings.sources): + raise ValueError("用例型号与收集源不一致") + if any(group["matrix_id"] == matrix_id for group in run.subruns): + raise ValueError("用例子运行标识重复") + group = {"matrix_id": matrix_id, "planned_rounds": rounds, "completed_rounds": 0} + run.subruns.append(group) + run_id = run.run_id semaphore = asyncio.Semaphore(concurrency) async with observed_client(settings, capture, capabilities=capabilities) as client: @@ -188,21 +222,29 @@ async def _collect_rounds( if isinstance(value, BaseException): raise value values.append(value) + group["completed_rounds"] = len(values) counts = summarize_verdicts([value["verdict"] for value in values], planned_rounds=rounds) write_live_round( _OUT_DIR, run_id=run_id, matrix_id=matrix_id + "-rounds", round_index=0, - safe_fields={"counts": counts, "completed_rounds": len(values), "planned_rounds": rounds}, + safe_fields={ + "session_id": run_id, + "requested_model": run.model, + "counts": counts, + "completed_rounds": len(values), + "planned_rounds": rounds, + }, ) return values -async def _run_rounds(rounds, *, stream=True, matrix_id="thinking", **source_overrides): +async def _run_rounds(rounds, *, run, stream=True, matrix_id="thinking", **source_overrides): """L1–L8 的资格证据出口，不作整类 skip。""" return await _collect_rounds( _settings(**source_overrides), + run=run, rounds=rounds, stream=stream, prompt=_PROMPT, @@ -218,6 +260,8 @@ def _qualified(rows, *, planned_rounds): def _coverage(rows, *, planned_rounds, proposition): """只有全轮资格通过才进入推理观测命题。""" verdict = _qualified(rows, planned_rounds=planned_rounds) + if verdict.status == "PASS" and proposition == "observation-only": + return LiveVerdict("UNCOVERED", "未登记候选只保留观测，不自动登记能力") if verdict.status == "PASS": verdict = assess_thinking_coverage( [row["response"].thinking_observation for row in rows], @@ -227,17 +271,18 @@ def _coverage(rows, *, planned_rounds, proposition): return verdict -def _conclude(matrix, verdict, *, proposition=None): +def _conclude(matrix, verdict, *, run, proposition=None): """命题汇总先落盘再交给 pytest，不覆盖逐轮原件。""" write_live_round( _OUT_DIR, - run_id=uuid4().hex, + run_id=run.run_id, matrix_id=matrix, round_index=0, safe_fields={ "status": verdict.status, "reason": verdict.reason, "proposition": proposition, + **run.report_fields(), }, ) enforce_verdict(verdict) @@ -247,18 +292,27 @@ class TestMiniMaxM3: """AUTO 拒绝已移至离线契约；真实开启明确请求 medium。""" async def test_l1_disable_actually_disables(self): + run = _CaseRun(uuid4().hex, "MiniMax-M3") rows = await _run_rounds( - _ROUNDS, matrix_id="L1", provider="minimax", model="MiniMax-M3", enable_thinking=False + _ROUNDS, + run=run, + matrix_id="L1", + provider="minimax", + model="MiniMax-M3", + enable_thinking=False, ) _conclude( "L1", _coverage(rows, planned_rounds=_ROUNDS, proposition="disabled"), + run=run, proposition="disabled", ) async def test_l2_enable_actually_enables(self): + run = _CaseRun(uuid4().hex, "MiniMax-M3") rows = await _run_rounds( _ROUNDS, + run=run, matrix_id="L2", provider="minimax", model="MiniMax-M3", @@ -267,14 +321,17 @@ class TestMiniMaxM3: _conclude( "L2", _coverage(rows, planned_rounds=_ROUNDS, proposition="enabled"), + run=run, proposition="enabled", ) async def test_l2b_off_and_on_are_distinguishable_without_magic_numbers(self): """指定历史 prompt 锚点回归，不宣称关闭能力已覆盖。""" + run = _CaseRun(uuid4().hex, "MiniMax-M3") rounds = max(3, _ROUNDS // 3) off = await _run_rounds( rounds, + run=run, matrix_id="L2b-off", provider="minimax", model="MiniMax-M3", @@ -282,6 +339,7 @@ class TestMiniMaxM3: ) on = await _run_rounds( rounds, + run=run, matrix_id="L2b-on", provider="minimax", model="MiniMax-M3", @@ -295,22 +353,27 @@ class TestMiniMaxM3: verdict = LiveVerdict( "PASS" if distinct else "FAIL", "指定历史 prompt 锚点比较；不是关闭证明" ) - _conclude("L2b", verdict, proposition="historical-prompt-anchor") + _conclude("L2b", verdict, run=run, proposition="historical-prompt-anchor") async def test_l3_no_opinion_is_the_model_default(self): - rows = await _run_rounds(_ROUNDS, matrix_id="L3", provider="minimax", model="MiniMax-M3") + run = _CaseRun(uuid4().hex, "MiniMax-M3") + rows = await _run_rounds( + _ROUNDS, run=run, matrix_id="L3", provider="minimax", model="MiniMax-M3" + ) verdict = _qualified(rows, planned_rounds=_ROUNDS) if verdict.status == "PASS" and any( row["response"].applied_effort is not None for row in rows ): verdict = LiveVerdict("FAIL", "不表态路径擅自记录档位") - _conclude("L3", verdict, proposition="no-opinion-not-capability") + _conclude("L3", verdict, run=run, proposition="no-opinion-not-capability") async def test_l3b_none_is_recognised_not_silently_dropped(self): """保留原非法 raw 值对照预算，但不提升 UNKNOWN。""" + run = _CaseRun(uuid4().hex, "MiniMax-M3") rounds = max(3, _ROUNDS // 3) bogus = await _run_rounds( rounds, + run=run, matrix_id="L3b-bogus", provider="minimax", model="MiniMax-M3", @@ -318,6 +381,7 @@ class TestMiniMaxM3: ) off = await _run_rounds( rounds, + run=run, matrix_id="L3b-off", provider="minimax", model="MiniMax-M3", @@ -327,13 +391,15 @@ class TestMiniMaxM3: _coverage(bogus, planned_rounds=rounds, proposition="enabled"), _coverage(off, planned_rounds=rounds, proposition="disabled"), ] - _conclude("L3b", _combine(verdicts), proposition="raw-counterexample") + _conclude("L3b", _combine(verdicts), run=run, proposition="raw-counterexample") async def test_l4_raw_only_explicit_high(self): """退出受管意图后才保留 raw high；双来源拒绝在 unit 守卫。""" + run = _CaseRun(uuid4().hex, "MiniMax-M3") rounds = max(3, _ROUNDS // 2) rows = await _run_rounds( rounds, + run=run, matrix_id="L4", provider="minimax", model="MiniMax-M3", @@ -342,14 +408,17 @@ class TestMiniMaxM3: _conclude( "L4", _coverage(rows, planned_rounds=rounds, proposition="enabled"), + run=run, proposition="enabled", ) async def test_l5_non_stream_path_is_distinguishable_and_honestly_unknown(self): """保留流／非流预算；UNKNOWN 是明确未覆盖而非长度锚点成功。""" + run = _CaseRun(uuid4().hex, "MiniMax-M3") rounds = max(3, _ROUNDS // 2) off = await _run_rounds( rounds, + run=run, matrix_id="L5-off", stream=False, provider="minimax", @@ -358,6 +427,7 @@ class TestMiniMaxM3: ) on = await _run_rounds( rounds, + run=run, matrix_id="L5-on", stream=False, provider="minimax", @@ -372,6 +442,7 @@ class TestMiniMaxM3: _coverage(on, planned_rounds=rounds, proposition="enabled"), ] ), + run=run, proposition="nonstream-enabled-disabled", ) @@ -389,22 +460,36 @@ class TestOtherProviders: [("L6", "qwen", "qwen3.7-plus"), ("L7", "deepseek", "deepseek-v4-pro")], ) async def test_existing_profiles_still_disable(self, matrix, provider, model): + run = _CaseRun(uuid4().hex, model) rows = await _run_rounds( - _ROUNDS, matrix_id=matrix, provider=provider, model=model, enable_thinking=False + _ROUNDS, + run=run, + matrix_id=matrix, + provider=provider, + model=model, + enable_thinking=False, ) _conclude( matrix, _coverage(rows, planned_rounds=_ROUNDS, proposition="disabled"), + run=run, proposition="disabled", ) async def test_qwen_enabled_is_observed(self): + run = _CaseRun(uuid4().hex, "qwen3.7-plus") rows = await _run_rounds( - _ROUNDS, matrix_id="L6b", provider="qwen", model="qwen3.7-plus", enable_thinking=True + _ROUNDS, + run=run, + matrix_id="L6b", + provider="qwen", + model="qwen3.7-plus", + enable_thinking=True, ) _conclude( "L6b", _coverage(rows, planned_rounds=_ROUNDS, proposition="enabled"), + run=run, proposition="enabled", ) @@ -419,9 +504,11 @@ class TestCapabilityDrift: ), ) async def test_declared_capability_matches_reality(self, model): + run = _CaseRun(uuid4().hex, model) rounds = max(3, _ROUNDS // 2) rows = await _run_rounds( rounds, + run=run, matrix_id="L8", provider=_MODEL_PROVIDER[model], model=model, @@ -430,14 +517,16 @@ class TestCapabilityDrift: _conclude( "L8", _coverage(rows, planned_rounds=rounds, proposition="disabled"), + run=run, proposition="disabled", ) -async def _probe_effort(model, effort, *, rounds, prompt, prompt_kind): +async def _probe_effort(model, effort, *, run, rounds, prompt, prompt_kind): """临时全档表仅用于 T10 探测，不写回 DEFAULT，也不生成预期 wire。""" return await _collect_rounds( _tier_settings(model), + run=run, rounds=rounds, stream=True, prompt=prompt, @@ -453,10 +542,22 @@ class TestTierProbe: @pytest.mark.parametrize("model", sorted(_MODEL_PROVIDER)) async def test_t10_none_direction_matches_declaration(self, model): + run = _CaseRun(uuid4().hex, model) capability = DEFAULT_CAPABILITIES.get(model) - proposition = "disabled" if capability and capability.can_disable else "cannot_disable" + proposition = ( + "observation-only" + if capability is None + else "disabled" + if capability.can_disable + else "cannot_disable" + ) short = await _probe_effort( - model, Effort.NONE, rounds=_TIER_ROUNDS, prompt=_TIER_PROMPT, prompt_kind="none-short" + model, + Effort.NONE, + run=run, + rounds=_TIER_ROUNDS, + prompt=_TIER_PROMPT, + prompt_kind="none-short", ) verdict = _coverage(short, planned_rounds=_TIER_ROUNDS, proposition=proposition) # 沿既有矩阵：短档没有 OBSERVED 才做长上下文复核；不新增锚点调用。 @@ -466,6 +567,7 @@ class TestTierProbe: long_rows = await _probe_effort( model, Effort.NONE, + run=run, rounds=_TIER_LONG_ROUNDS, prompt=_TIER_LONG_PROMPT, prompt_kind="none-long", @@ -475,18 +577,18 @@ class TestTierProbe: planned_rounds=_TIER_ROUNDS + _TIER_LONG_ROUNDS, proposition=proposition, ) - if capability is None and verdict.status != "FAIL": - verdict = LiveVerdict("UNCOVERED", "未登记候选只保留观测，不自动登记能力") - _conclude("T10-none", verdict, proposition=proposition) + _conclude("T10-none", verdict, run=run, proposition=proposition) @pytest.mark.parametrize("model", sorted(DEFAULT_CAPABILITIES)) async def test_t10_declared_tiers_actually_reason(self, model): + run = _CaseRun(uuid4().hex, model) tiers = [e for e in DEFAULT_CAPABILITIES[model].supported_efforts if e is not Effort.NONE] verdicts = [] for tier in tiers: rows = await _probe_effort( model, tier, + run=run, rounds=_TIER_ROUNDS, prompt=_TIER_PROMPT, prompt_kind="tier-" + tier.value, @@ -495,22 +597,31 @@ class TestTierProbe: verdicts.append(verdict) write_live_round( _OUT_DIR, - run_id=uuid4().hex, + run_id=run.run_id, matrix_id="T10-tier", round_index=0, safe_fields={ "requested_model": model, + "session_id": run.run_id, + "proposition": "enabled", + "subruns": [run.subruns[-1]], + "planned_rounds": _TIER_ROUNDS, + "completed_rounds": len(rows), "requested_effort": tier.value, "status": verdict.status, "reason": verdict.reason, }, ) - _conclude("T10-tiers", _combine(verdicts), proposition="enabled-all-declared-tiers") + _conclude( + "T10-tiers", _combine(verdicts), run=run, proposition="enabled-all-declared-tiers" + ) @pytest.mark.parametrize("model", ["gemini-3.1-pro", "gpt-5.5", "glm-5.3"]) async def test_t10_no_opinion_stays_no_opinion(self, model): + run = _CaseRun(uuid4().hex, model) rows = await _collect_rounds( _tier_settings(model), + run=run, rounds=_TIER_ROUNDS, stream=True, prompt=_TIER_PROMPT, @@ -522,4 +633,4 @@ class TestTierProbe: row["response"].applied_effort is not None for row in rows ): verdict = LiveVerdict("FAIL", "默认基线擅自推定档位") - _conclude("T10-default", verdict, proposition="no-opinion-not-capability") + _conclude("T10-default", verdict, run=run, proposition="no-opinion-not-capability") diff --git a/tests/live_evidence.py b/tests/live_evidence.py index e870d7c..f121d32 100644 --- a/tests/live_evidence.py +++ b/tests/live_evidence.py @@ -225,11 +225,12 @@ _SAFE_FIELDS = frozenset( "error_type", "error_status", "evidence_notes", + "subruns", } ) _ATTEMPT_FIELDS = frozenset({"call_id", "error_type", "http"}) _HTTP_FIELDS = frozenset( - {"status_code", "request_checks", "identity_captured", "error_body_complete"} + {"status_code", "request_checks", "identity_captured", "error_body_complete", "machine_type"} ) @@ -243,6 +244,8 @@ def _validate_safe(fields: Mapping[str, Any]) -> None: for event in attempt["http"]: if not isinstance(event, dict) or set(event) != _HTTP_FIELDS: raise ValueError("非法 HTTP 报告") + if event["machine_type"] not in ("model_not_found", "omitted"): + raise ValueError("报告机器字段不是认可枚举") if not isinstance(event["request_checks"], dict) or any( type(v) is not bool for v in event["request_checks"].values() ): @@ -273,7 +276,7 @@ def write_live_round( def safe_attempts(attempts: Sequence[AttemptEvidence]) -> list[dict[str, Any]]: - """只导出事实布尔值、异常类和状态；不落盘任何上游正文。""" + """只导出事实、异常类与认可机器枚举；任意上游字符串一律省略。""" return [ { "call_id": attempt.call_id, @@ -284,6 +287,11 @@ def safe_attempts(attempts: Sequence[AttemptEvidence]) -> list[dict[str, Any]]: "request_checks": dict(event.request_checks), "identity_captured": event.raw_identity[0], "error_body_complete": event.error_body is not None, + "machine_type": ( + "model_not_found" + if error_machine_type(event) == "model_not_found" + else "omitted" + ), } for event in attempt.http ], diff --git a/tests/unit/test_live_evidence.py b/tests/unit/test_live_evidence.py index 7a4d86c..3b9c28c 100644 --- a/tests/unit/test_live_evidence.py +++ b/tests/unit/test_live_evidence.py @@ -4,6 +4,7 @@ import asyncio import inspect import json from dataclasses import replace +from typing import Any import httpx import pytest @@ -814,3 +815,309 @@ def test_pytest_report_hook_preserves_report_and_uses_safe_fallback(tmp_path, mo assert finished.value.value is report paths = list(tmp_path.rglob("*.md")) assert len(paths) == 1 and '"status": "FAIL"' in paths[0].read_text() + + +@pytest.mark.parametrize( + "damage", [None, "prefix", "role", "content", "unpaired", "budget", "wire"] +) +async def test_structured_reask_preserves_message_contract(tmp_path, damage, monkeypatch): + """真实 StructuredMW 缺字段后重问成功；前缀、反馈结构及 wire 破坏均失败。""" + from pydantic import BaseModel + + from polygateway import GatewaySettings + from polygateway.middleware.structured import StructuredMW + from tests.e2e.conftest import captured_chat_round, observed_client + from tests.unit.test_config import _BASE_ENV + + class Answer(BaseModel): + """离线最小结构化契约。""" + + answer: int + reason: str + + settings = replace( + GatewaySettings.from_env("LLM", env=_BASE_ENV), + sources=(_source(),), + structured_max_retries=1, + ) + capture = _capture(messages_prefix_length=1, structured_max_retries=1) + original_feedback = StructuredMW._with_feedback + + def feedback(self, *args): + """只破坏重问产物，不替代生产阶梯或解析。""" + request = original_feedback(self, *args) + messages = [dict(message) for message in request.messages] + if damage == "prefix": + messages[0]["content"] = "changed" + elif damage == "role": + messages[-1]["role"] = "assistant" + elif damage == "content": + messages[-1]["content"] = ["wrong-type"] + elif damage == "unpaired": + messages.pop() + elif damage == "budget": + messages.extend(messages[-2:]) + return replace(request, messages=messages) + + monkeypatch.setattr(StructuredMW, "_with_feedback", feedback) + requests = [] + + def handler(request): + requests.append(json.loads(request.content)) + body = _response() + body["choices"][0]["message"]["content"] = ( + '{"answer":5}' if len(requests) == 1 else '{"answer":5,"reason":"sum"}' + ) + return httpx.Response(200, json=body) + + original_factory = capture.client_factory + + def factory(source): + client = original_factory(source) + client._transport = httpx.MockTransport(handler) + if damage == "wire": + + async def corrupt(request): + payload = json.loads(request.content) + if len(payload["messages"]) > 1: + payload["messages"][-1]["content"] = "well-shaped-but-corrupted" + request._content = json.dumps(payload).encode() + + client.event_hooks["request"].insert(0, corrupt) + return client + + capture.client_factory = factory + async with observed_client(settings, capture) as client: + response, verdict = await captured_chat_round( + client, + capture, + run_id="structured", + matrix_id="reask", + round_index=1, + output_dir=tmp_path, + messages=_MESSAGES, + models={"source": "gpt-5.5"}, + aliases={}, + stream=False, + structured=Answer, + ) + assert response.structured_data.answer == 5 + assert len(requests) == 2 + assert verdict.status == ("PASS" if damage is None else "FAIL") + text = "".join(path.read_text() for path in tmp_path.rglob("*.md")) + assert _SECRET not in text and _PROMPT not in text + + +def _thinking_consumer(): + """仅加载 live 消费者定义与字面矩阵，跳过所有环境读取语句。""" + import ast + from pathlib import Path + from types import SimpleNamespace + + path = Path(__file__).parents[1] / "e2e/test_thinking_live.py" + tree = ast.parse(path.read_text()) + excluded = { + "_ENV", + "_HAS_SOURCE", + "pytestmark", + "_ROUNDS", + "_TIER_ROUNDS", + "_TIER_LONG_ROUNDS", + "_TIER_CONCURRENCY", + } + tree.body = [ + node + for node in tree.body + if not ( + isinstance(node, ast.Assign) + and any( + isinstance(target, ast.Name) and target.id in excluded for target in node.targets + ) + ) + ] + namespace: dict[str, Any] = { + "_ROUNDS": 3, + "_TIER_ROUNDS": 2, + "_TIER_LONG_ROUNDS": 1, + "_TIER_CONCURRENCY": 1, + } + exec(compile(tree, str(path), "exec"), namespace) + return SimpleNamespace(**namespace), namespace + + +@pytest.mark.parametrize( + ("observation", "qualification", "expected"), + [ + (O.ABSENT, "PASS", "UNCOVERED"), + (O.OBSERVED, "PASS", "UNCOVERED"), + (O.UNKNOWN, "PASS", "UNCOVERED"), + (O.ABSENT, "FAIL", "FAIL"), + ], +) +async def test_unregistered_candidate_has_no_disable_declaration( + observation, qualification, expected +): + """执行真实 T10 消费者，未登记不等于不可关闭，资格失败仍红。""" + from types import SimpleNamespace + + live, namespace = _thinking_consumer() + conclusions = [] + calls = [] + + async def probe(model, effort, *, rounds, **kwargs): + calls.append(rounds) + return [ + { + "verdict": LiveVerdict(qualification, "safe"), + "response": SimpleNamespace(thinking_observation=observation), + } + for _ in range(rounds) + ] + + namespace["_probe_effort"] = probe + namespace["_conclude"] = lambda matrix, verdict, **kwargs: conclusions.append(verdict) + await live.TestTierProbe().test_t10_none_direction_matches_declaration("claude-haiku-5") + assert conclusions[0].status == expected + assert calls == ([2, 1] if observation is not O.OBSERVED and qualification == "PASS" else [2]) + + +@pytest.mark.parametrize("case", ["none", "tiers", "L8"]) +async def test_capability_conclusions_link_models_and_all_subruns(tmp_path, case): + """两个型号的 PASS／UNCOVERED 结论必须关联原件和完整短长／档位轮数。""" + from contextlib import asynccontextmanager + from types import SimpleNamespace + + from polygateway import GatewaySettings + from tests.unit.test_config import _BASE_ENV + + live, namespace = _thinking_consumer() + settings = GatewaySettings.from_env("LLM", env=_BASE_ENV) + namespace["_OUT_DIR"] = tmp_path + namespace["_tier_settings"] = lambda model: replace( + settings, sources=(replace(_source(), model=model),) + ) + namespace["_settings"] = lambda **kwargs: replace( + settings, sources=(replace(_source(), **kwargs),) + ) + namespace["enforce_verdict"] = lambda verdict: None + + @asynccontextmanager + async def client(*args, **kwargs): + yield None + + async def round_call(client, capture, **kwargs): + model = kwargs["models"]["source"] + observation = ( + O.UNKNOWN if model == "gpt-5.4" else O.OBSERVED if case == "tiers" else O.ABSENT + ) + write_live_round( + kwargs["output_dir"], + run_id=kwargs["run_id"], + matrix_id=kwargs["matrix_id"], + round_index=kwargs["round_index"], + safe_fields={ + "requested_model": model, + "session_id": kwargs["run_id"], + "status": "PASS", + "thinking_observation": observation, + }, + ) + return SimpleNamespace(thinking_observation=observation), LiveVerdict("PASS", "safe") + + namespace["observed_client"] = client + namespace["captured_chat_round"] = round_call + for model in ("gpt-5.5", "gpt-5.4"): + if case == "none": + await live.TestTierProbe().test_t10_none_direction_matches_declaration(model) + elif case == "tiers": + await live.TestTierProbe().test_t10_declared_tiers_actually_reason(model) + else: + await live.TestCapabilityDrift().test_declared_capability_matches_reality(model) + matrix = {"none": "T10-none", "tiers": "T10-tiers", "L8": "L8"}[case] + finals = list(tmp_path.rglob(f"{matrix}-0-*.md")) + assert len(finals) == 2 + seen = set() + for path in finals: + row = json.loads(path.read_text().split("```json\n")[1].split("\n```")[0]) + assert "requested_model" in row, "结论缺型号，无法关联逐轮原件" + model = row["requested_model"] + seen.add(model) + assert row["session_id"] == path.parent.name + assert row["status"] == ("PASS" if model == "gpt-5.5" else "UNCOVERED") + assert row["proposition"] + subruns = row["subruns"] + expected_groups = ( + 2 + if case == "none" + else len( + [ + effort + for effort in live.DEFAULT_CAPABILITIES[model].supported_efforts + if effort is not live.Effort.NONE + ] + ) + if case == "tiers" + else 1 + ) + assert len(subruns) == expected_groups + assert row["planned_rounds"] == sum(subrun["planned_rounds"] for subrun in subruns) + assert row["completed_rounds"] == row["planned_rounds"] + for subrun in subruns: + originals = [ + p + for p in path.parent.glob(f"{subrun['matrix_id']}-*.md") + if p.name[len(subrun["matrix_id"]) + 1 :].split("-", 1)[0].isdigit() + and not p.name.startswith(subrun["matrix_id"] + "-0-") + ] + assert len(originals) == subrun["planned_rounds"] == subrun["completed_rounds"] + for original in originals: + data = json.loads(original.read_text().split("```json\n")[1].split("\n```")[0]) + assert data["requested_model"] == model + assert data["session_id"] == row["session_id"] + assert seen == {"gpt-5.5", "gpt-5.4"} + + +@pytest.mark.parametrize("machine_type", ["model_not_found", _SECRET, "arbitrary-upstream-text"]) +def test_safe_machine_type_retains_only_known_enum(tmp_path, machine_type): + """认可机器枚举可复核；任意机器正文与 sentinel 不得落盘。""" + _, attempts = _failure(body=json.dumps({"error": {"type": machine_type}}).encode()) + fields = {"attempts": safe_attempts(attempts)} + event = fields["attempts"][0]["http"][0] + assert event.get("machine_type") == ( + "model_not_found" if machine_type == "model_not_found" else "omitted" + ) + path = write_live_round( + tmp_path, run_id="safe", matrix_id="machine", round_index=1, safe_fields=fields + ) + text = path.read_text() + assert _SECRET not in text and "arbitrary-upstream-text" not in text + if machine_type != "model_not_found": + event["machine_type"] = machine_type + with pytest.raises(ValueError, match="机器"): + write_live_round( + tmp_path, run_id="unsafe", matrix_id="machine", round_index=1, safe_fields=fields + ) + + +async def test_structured_first_attempt_requires_exact_initial_messages(): + """结构化窄规则不能允许首轮凭空带入一对反馈。""" + capture = _capture(messages_prefix_length=1, structured_max_retries=1) + real = _real_transport(capture, lambda request: httpx.Response(200, json=_response())) + try: + with capture.round_context(session_id="first", parent_call_id="parent"): + await ObservedTransport(real, capture).complete( + messages=[ + *_MESSAGES, + {"role": "assistant", "content": "old"}, + {"role": "user", "content": "retry"}, + ], + source=_source(), + stream=False, + overlay={}, + call_id="first", + reasoning_effort=None, + ) + event = capture.attempts(session_id="first", parent_call_id="parent")[0].http[0] + assert not request_is_valid(event) + finally: + await real.aclose()