From 3c9f306ea357d5a78547fc1526a5465fc8644090 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 22 Jul 2026 02:21:33 -0400 Subject: [PATCH] fix: address M3 verifier findings on soak invariants and probe tests --- src/polygateway/transports/monkey_ocr.py | 4 ++- tests/integration/test_monkey_ocr_live.py | 2 +- tests/unit/test_monkey_ocr.py | 8 +++++ tests/unit/test_ocr_client.py | 42 +++++++++++++++++++++++ tests/unit/test_soak_scoreboard.py | 19 ++++++++++ tools/soak/run_soak.py | 19 +++++++--- tools/soak/scenarios.py | 24 +++++++++---- tools/soak/scoreboard.py | 20 +++++++++++ 8 files changed, 126 insertions(+), 12 deletions(-) diff --git a/src/polygateway/transports/monkey_ocr.py b/src/polygateway/transports/monkey_ocr.py index 8f420dc..6426251 100644 --- a/src/polygateway/transports/monkey_ocr.py +++ b/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 无效") diff --git a/tests/integration/test_monkey_ocr_live.py b/tests/integration/test_monkey_ocr_live.py index 51ea58e..de8ac75 100644 --- a/tests/integration/test_monkey_ocr_live.py +++ b/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 diff --git a/tests/unit/test_monkey_ocr.py b/tests/unit/test_monkey_ocr.py index 7ea58ba..98d59a7 100644 --- a/tests/unit/test_monkey_ocr.py +++ b/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( diff --git a/tests/unit/test_ocr_client.py b/tests/unit/test_ocr_client.py index 5e904e5..2759b24 100644 --- a/tests/unit/test_ocr_client.py +++ b/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"]) diff --git a/tests/unit/test_soak_scoreboard.py b/tests/unit/test_soak_scoreboard.py index c37e2c9..8cb4031 100644 --- a/tests/unit/test_soak_scoreboard.py +++ b/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) diff --git a/tools/soak/run_soak.py b/tools/soak/run_soak.py index 9e83b55..53018e1 100644 --- a/tools/soak/run_soak.py +++ b/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 ( - ("记账归零(inflight)", sb.inv_accounting_zeroed(limiter, names)), - ("gate 可再准入(探针不悬挂)", sb.inv_gate_reenterable(gate, names)), - ): + for name, invariant in checks: try: await invariant verdicts.append((name, "PASS")) diff --git a/tools/soak/scenarios.py b/tools/soak/scenarios.py index b69c9dc..604e091 100644 --- a/tools/soak/scenarios.py +++ b/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.replays = load_replay_payloads(self.telemetry_db) - self.images = sorted(Path(self.images_root).glob("chs_*.jpg")) - if not (self.chains and self.replays and self.images): - raise ValueError("语料不完整: 请确认 data/soak/ 已按 findings §6 拉取") + self.chains = load_trace_chains(self.harness_db) if "chains" in self.modalities else [] + self.replays = ( + load_replay_payloads(self.telemetry_db) if "replays" in self.modalities else [] + ) + 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: diff --git a/tools/soak/scoreboard.py b/tools/soak/scoreboard.py index a9101e8..c3b5769 100644 --- a/tools/soak/scoreboard.py +++ b/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): 故障源混编池下错误行必然存在。