fix: address M3 verifier findings on soak invariants and probe tests

src/polygateway/transports/monkey_ocr.py

@@ -142,7 +142,9 @@ def _parse_pages(payload: object) -> tuple[list[OcrLayoutElement], list[tuple[fl
             raise ValueError("页面结构无效")
         page_sizes.append(_parse_page_size(page))
         raw_index = page.get("page_idx", fallback_index)
-        page_index = raw_index if isinstance(raw_index, int) else fallback_index
+        # bool 是 int 子类、负页号无意义——与 _finite_number 拒 bool 的口径一致
+        valid_idx = isinstance(raw_index, int) and not isinstance(raw_index, bool) and raw_index >= 0
+        page_index = raw_index if valid_idx else fallback_index
         blocks = page.get("para_blocks", [])
         if not isinstance(blocks, list):
             raise ValueError("para_blocks 无效")

tests/integration/test_monkey_ocr_live.py

@@ -55,7 +55,7 @@ class TestLiveEndpoints:
     async def test_recognize_text_returns_content(self):
         async with OcrClient.from_env("OCR", env=_env({1: _PRIMARY})) as client:
             result = await client.recognize_text(_TABLE_IMAGE.read_bytes())
-        assert result.text.strip()  # 真实护理记录图必有文字
+        assert result.text.strip()  # 取证样本已核: 该图含可识别文字
         assert result.source_name == "monkey_1"
         assert result.usage.prompt_tokens == 0 and result.latency_ms > 0
 

tests/unit/test_monkey_ocr.py

@@ -280,6 +280,14 @@ class TestMiddleJsonDefense:
         elements, _ = _parse_middle_json(_zip_bytes(_middle_bytes([page])))
         assert elements[0].page_index == 0
 
+    @pytest.mark.parametrize("bad_idx", [True, -3, "2", 1.5])
+    def test_page_idx_non_int_falls_back(self, bad_idx):
+        # bool/负数/非 int 一律回退枚举序(与 _finite_number 拒 bool 的防御口径一致)
+        page = _page([_block("text", (1, 2, 30, 40))])
+        page["page_idx"] = bad_idx
+        elements, _ = _parse_middle_json(_zip_bytes(_middle_bytes([page])))
+        assert elements[0].page_index == 0
+
 
 class TestErrorTranslation:
     @pytest.mark.parametrize(

tests/unit/test_ocr_client.py

@@ -292,7 +292,49 @@ class TestBackpressure:
             await permit.release()
 
 
+class FakeClock:
+    def __init__(self, start=1000.0):
+        self.t = start
+
+    def __call__(self):
+        return self.t
+
+    def advance(self, seconds):
+        self.t += seconds
+
+
 class TestCancellation:
+    async def test_cancel_during_probe_returns_probe(self):
+        # 探针取消归还(铁律分支;verifier I3): 开路→冷却过后半开探针→
+        # transport 挂起中取消 → release_probe 必须发生,后续可再探
+        clock = FakeClock()
+        inner = InMemoryGate(config=_BREAKER, now=clock)  # 门与 client 共用注入钟
+        gate = RecordingGate(inner)
+        client, _, _ = _client([_src()], ["hang"], now=clock, breaker=gate)
+        entry = await inner.try_enter("m1", "setup")
+        await inner.record_failure(entry, "source_dead", True)  # force_open → OPEN
+        clock.advance(61.0)  # 越过 cooldown_s=60 → HALF_OPEN
+        task = asyncio.create_task(client.recognize_text(b"jpg"))
+        await asyncio.sleep(0.05)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert gate.probe_releases >= 1  # 探针已归还,不悬挂
+        probe = await inner.try_enter("m1", "again")
+        assert probe.allowed and probe.is_probe  # 可再探 = 未悬挂的行为证据
+
+    async def test_cancel_during_backoff_sleep_propagates(self):
+        async def cancelling_sleep(delay):
+            raise asyncio.CancelledError()
+
+        client, limiter, _ = _client(
+            [_src()], [TransientError("boom", status_code=500)], sleep=cancelling_sleep
+        )
+        with pytest.raises(asyncio.CancelledError):
+            await client.recognize_text(b"jpg")
+        stats = await limiter.source_stats("m1")
+        assert stats.inflight == 0  # 退避期取消: permit 已在 finally 释放
+
     async def test_cancel_during_transport_releases_permit(self):
         src = _src(max_concurrency=1)
         client, limiter, _ = _client([src], ["hang"])

tests/unit/test_soak_scoreboard.py

@@ -52,3 +52,22 @@ class TestErrorsClassified:
         rows = [_row(error="KeyError: 'oops'")]
         with pytest.raises(AssertionError, match="未分类"):
             inv_errors_classified(rows, _KNOWN)
+
+
+class TestRssAbsolute:
+    def test_within_bound(self):
+        from tools.soak.scoreboard import inv_rss_absolute
+
+        inv_rss_absolute([120.0, 180.0, 150.0], max_mb=500.0)
+
+    def test_peak_exceeds(self):
+        from tools.soak.scoreboard import inv_rss_absolute
+
+        with pytest.raises(AssertionError, match="峰值"):
+            inv_rss_absolute([120.0, 690.0, 200.0], max_mb=500.0)  # verifier I1 实测形态
+
+    def test_empty_rejected(self):
+        from tools.soak.scoreboard import inv_rss_absolute
+
+        with pytest.raises(AssertionError):
+            inv_rss_absolute([], max_mb=500.0)

tools/soak/run_soak.py

@@ -139,6 +139,8 @@ async def _worker_async(args: argparse.Namespace, worker_idx: int) -> None:
         telemetry_db=_ROOT / "data/soak/generate_questions_telemetry.db",
         frames_root=_ROOT / "data/soak/vt_frames",
         images_root=_ROOT / "data/soak/chs_images",
+        # P7 只需 images: chains/replays 的 base64 负载曾把起跑 RSS 顶到 ~690MB(verifier I1)
+        modalities=("images",) if args.scenario == "P7" else ("chains", "replays", "images"),
     )
     generator = SCENARIOS[args.scenario](corpus, f"{run_id}-w{worker_idx}")
     sem = asyncio.Semaphore(args.concurrency)
@@ -297,6 +299,8 @@ def _p7_checks(args, env, results, rows, calls, verdicts, check) -> None:
     check("坏源尝试占比 ≤15%(P7)", sb.inv_fault_share, rows, fault_names, max_share=0.15)
     check("故障混编生效(P7)", sb.inv_fault_errors_present, rows, fault_source_names=fault_names)
     check("错误全部可分类(P7)", sb.inv_errors_classified, rows, _KNOWN_ERROR_PREFIXES)
+    for r in results:
+        check("RSS 绝对值 <500MB(P7)", sb.inv_rss_absolute, r["rss_current_mb"], max_mb=500.0)
     cancelled = sum(r["stats"]["cancelled"] for r in results)
     verdicts.append(
         ("零取消泄漏(P7)", "PASS" if cancelled == 0 else f"FAIL — cancelled={cancelled}")
@@ -388,11 +392,18 @@ async def _live_checks(
         lease_ttl_s=settings.lease_ttl_s,
     )
     gate = RedisGate.from_url(env["REDIS_URL"], config=settings.breaker, scope=settings.scope)
+    checks = [
+        ("记账归零(inflight)", sb.inv_accounting_zeroed(limiter, names)),
+        ("gate 可再准入(探针不悬挂)", sb.inv_gate_reenterable(gate, names)),
+    ]
+    if args.scenario == "P7":
+        fault = {
+            s.strip() for s in env.get(f"{args.scope}_FAULT_SOURCES", "").split(",") if s.strip()
+        }
+        healthy = [n for n in names if n not in fault]
+        checks.append(("真源门态 CLOSED(P7 零误熔)", sb.inv_healthy_gates_closed(gate, healthy)))
     try:
-        for name, invariant in (
+        for name, invariant in checks:
-            ("记账归零(inflight)", sb.inv_accounting_zeroed(limiter, names)),
-            ("gate 可再准入(探针不悬挂)", sb.inv_gate_reenterable(gate, names)),
-        ):
             try:
                 await invariant
                 verdicts.append((name, "PASS"))

tools/soak/scenarios.py

@@ -47,19 +47,31 @@ class ChsExtraction(BaseModel):
 
 @dataclass
 class SoakCorpus:
-    """一次 run 的语料句柄(装载一次,场景间共享)。"""
+    """一次 run 的语料句柄(装载一次,场景间共享)。
+
+    modalities 控制装载面: P7 只需 images——chains/replays 含整段 base64
+    负载,全量装载曾使 P7 起跑 RSS 冲到 ~690MB(verifier I1,2026-07-22)。
+    """
 
     harness_db: Path
     telemetry_db: Path
     frames_root: Path
     images_root: Path
+    modalities: tuple[str, ...] = ("chains", "replays", "images")
 
     def __post_init__(self) -> None:
-        self.chains = load_trace_chains(self.harness_db)
+        self.chains = load_trace_chains(self.harness_db) if "chains" in self.modalities else []
-        self.replays = load_replay_payloads(self.telemetry_db)
+        self.replays = (
-        self.images = sorted(Path(self.images_root).glob("chs_*.jpg"))
+            load_replay_payloads(self.telemetry_db) if "replays" in self.modalities else []
-        if not (self.chains and self.replays and self.images):
+        )
-            raise ValueError("语料不完整: 请确认 data/soak/ 已按 findings §6 拉取")
+        self.images = (
+            sorted(Path(self.images_root).glob("chs_*.jpg"))
+            if "images" in self.modalities
+            else []
+        )
+        missing = [m for m in self.modalities if not getattr(self, m)]
+        if missing:
+            raise ValueError(f"语料不完整({missing}): 请确认 data/soak/ 已按 findings §6 拉取")
 
 
 def weighted_mix(weights: dict[str, float], rng) -> str:

tools/soak/scoreboard.py

@@ -268,6 +268,26 @@ def inv_errors_classified(rows: list[Row], known_prefixes: tuple[str, ...]) -> N
     assert not unknown, f"未分类错误前缀: {dict(unknown)}"
 
 
+def inv_rss_absolute(samples_mb: list[float], *, max_mb: float) -> None:
+    """P7 不变量⑤(计划 T8): 全程 RSS 绝对值有界(与增长口径互补)。"""
+    assert samples_mb, "RSS 采样为空"
+    peak = max(samples_mb)
+    assert peak < max_mb, f"RSS 峰值 {peak:.1f}MB ≥ 绝对上限 {max_mb}MB"
+
+
+async def inv_healthy_gates_closed(gate, healthy_names: list[str]) -> None:
+    """P7 不变量③(计划 T8): 跑后真源熔断门 CLOSED 可准入 = 真源零误熔的
+    机械化终态判据(误熔必由该源失败计数驱动,配合遥测"真源零错误行"覆盖全程)。
+    拿到探针当场归还,不留悬挂。"""
+    for name in healthy_names:
+        decision = await gate.try_enter(name, "scoreboard-healthy-probe")
+        if decision.allowed and decision.is_probe:
+            await gate.release_probe(decision)
+        assert decision.allowed and str(decision.state) == "closed", (
+            f"真源 {name} 跑后门态 {decision.state}(allowed={decision.allowed})——疑似误熔"
+        )
+
+
 def inv_any_errors(rows: list[Row]) -> None:
     """不变量 2c 的兜底形态(P5/P6): 故障源混编池下错误行必然存在。