feat: add P7 OCR soak scenario and scoreboard

src/polygateway/ocr.py

@@ -404,13 +404,18 @@ class OcrClient:
                 source_name=source.name,
                 usage_source="measured",
             )
+        # 错误带异常类名前缀(metric ocr-call-success 注册口径: 按类名归组)
+        if error is None or isinstance(error, str):
+            error_text = error
+        else:
+            error_text = f"{type(error).__name__}: {error}"
         await self._emitter.emit_attempt(
             request=request,
             source=source,
             call_id=call_id,
             latency_ms=latency_ms,
             response=response,
-            error=None if error is None else str(error),
+            error=error_text,
         )
 
     @staticmethod

tests/unit/test_ocr_client.py

@@ -350,7 +350,7 @@ class TestTelemetry:
         for row in recorder.rows:
             assert "<ocr:text image_bytes=15>" in row["messages"]
             assert "RAW-IMAGE-BYTES" not in row["messages"]  # 图像字节绝不入库
-        assert recorder.rows[0]["error"] is not None
+        assert recorder.rows[0]["error"].startswith("TransientError:")  # 类名前缀口径
         assert recorder.rows[1]["error"] is None
         assert recorder.rows[1]["prompt_tokens"] == 0
 

tests/unit/test_soak_scoreboard.py

@@ -0,0 +1,54 @@
+"""P7 记分板新增纯函数测试(M3 计划 T8): 成功率/坏源占比/错误分类。"""
+
+import pytest
+
+from tools.soak.scoreboard import (
+    inv_errors_classified,
+    inv_fault_share,
+    inv_success_rate,
+)
+
+_KNOWN = ("TransientError", "AllSourcesExhausted", "cancelled")
+
+
+def _row(source="monkey_1", error=None, cache_hit=0):
+    return {"source_name": source, "error": error, "cache_hit": cache_hit}
+
+
+class TestSuccessRate:
+    def test_pass_at_threshold(self):
+        inv_success_rate(98, 100, min_rate=0.98)
+
+    def test_fail_below(self):
+        with pytest.raises(AssertionError, match="成功率"):
+            inv_success_rate(97, 100, min_rate=0.98)
+
+    def test_zero_total_rejected(self):
+        with pytest.raises(AssertionError):
+            inv_success_rate(0, 0, min_rate=0.98)
+
+
+class TestFaultShare:
+    def test_share_within_threshold(self):
+        rows = [_row()] * 9 + [_row("monkey_3")]
+        inv_fault_share(rows, ["monkey_3", "monkey_4"], max_share=0.15)
+
+    def test_share_exceeds(self):
+        rows = [_row()] * 4 + [_row("monkey_3")] * 2
+        with pytest.raises(AssertionError, match="坏源尝试占比"):
+            inv_fault_share(rows, ["monkey_3"], max_share=0.15)
+
+    def test_cache_rows_excluded(self):
+        rows = [_row(cache_hit=1)] * 100 + [_row("monkey_3")] + [_row()] * 9
+        inv_fault_share(rows, ["monkey_3"], max_share=0.15)  # 缓存行不入分母
+
+
+class TestErrorsClassified:
+    def test_known_prefixes_pass(self):
+        rows = [_row(error="TransientError: m1 超时"), _row(error="cancelled"), _row()]
+        inv_errors_classified(rows, _KNOWN)
+
+    def test_unknown_prefix_fails(self):
+        rows = [_row(error="KeyError: 'oops'")]
+        with pytest.raises(AssertionError, match="未分类"):
+            inv_errors_classified(rows, _KNOWN)

tools/soak/run_soak.py

@@ -122,14 +122,18 @@ async def _worker_async(args: argparse.Namespace, worker_idx: int) -> None:
     """单 worker: 消费场景生成器,信号量限并发,双预算任一命中即停。"""
     from polygateway import GatewayClient
     from polygateway.telemetry.sqlite import SQLiteRecorder
-
     from tools.soak.scenarios import SCENARIOS, SoakCorpus
 
     env = _merged_env()
     run_id = args.run_id
     telemetry_path = _ROOT / f"data/soak/telemetry_{run_id}_{worker_idx}.db"
     recorder = SQLiteRecorder(telemetry_path)
-    client = GatewayClient.from_env(args.scope, telemetry=recorder, env=env)
+    if args.scenario == "P7":
+        from polygateway.ocr import OcrClient
+
+        client = OcrClient.from_env(args.scope, telemetry=recorder, env=env)
+    else:
+        client = GatewayClient.from_env(args.scope, telemetry=recorder, env=env)
     corpus = SoakCorpus(
         harness_db=_ROOT / "data/soak/harness.db",
         telemetry_db=_ROOT / "data/soak/generate_questions_telemetry.db",
@@ -148,11 +152,16 @@ async def _worker_async(args: argparse.Namespace, worker_idx: int) -> None:
 
     async def _one(kind: str, kwargs: dict) -> None:
         try:
-            # run 级双层隔离: FLUSHDB(入口)+ 每 run 独立 namespace(此处)
-            kwargs.setdefault("cache_namespace", run_id)
-            resp = await client.chat(**kwargs)
+            if kind == "ocr_text":
+                await client.recognize_text(**kwargs)
+            elif kind == "ocr_layout":
+                await client.parse_layout(**kwargs)
+            else:
+                # run 级双层隔离: FLUSHDB(入口)+ 每 run 独立 namespace(此处)
+                kwargs.setdefault("cache_namespace", run_id)
+                resp = await client.chat(**kwargs)
+                stats["tokens"] += resp.prompt_tokens + resp.completion_tokens
             stats["ok"] += 1
-            stats["tokens"] += resp.prompt_tokens + resp.completion_tokens
         except asyncio.CancelledError:
             stats["cancelled"] += 1
             raise
@@ -263,8 +272,11 @@ def _scoreboard(args: argparse.Namespace, env: dict[str, str]) -> None:
     # 交报告"错误分布"人工核对(findings §4 条 2 的比例校准留待 P5 常态化)
     if args.scenario in ("P5", "P6"):
         _check("故障混编生效(P5/P6)", sb.inv_any_errors, rows)
-    rate = sb.structured_success_rate(rows, session_suffix="-p3")
-    verdicts.append(("P3 结构化成功率", f"{rate:.3f}(基线首跑建立)"))
+    if args.scenario == "P7":
+        _p7_checks(args, env, results, rows, calls, verdicts, _check)
+    else:
+        rate = sb.structured_success_rate(rows, session_suffix="-p3")
+        verdicts.append(("P3 结构化成功率", f"{rate:.3f}(基线首跑建立)"))
     report = sb.render_report(args.run_id, rows, verdicts)
     path = sb.write_report(args.run_id, report)
     print(report)
@@ -273,6 +285,78 @@ def _scoreboard(args: argparse.Namespace, env: dict[str, str]) -> None:
         raise SystemExit("硬不变量被击穿,见报告")
 
 
+def _p7_checks(args, env, results, rows, calls, verdicts, check) -> None:
+    """P7 专属裁决(M3 计划 T8): 成功率/坏源压制/故障生效/错误分类/取消泄漏/一致性。"""
+    from tools.soak import scoreboard as sb
+
+    ok_total = sum(r["stats"]["ok"] for r in results)
+    check("成功率 ≥98%(P7)", sb.inv_success_rate, ok_total, calls, min_rate=0.98)
+    fault_names = [
+        s.strip() for s in env.get(f"{args.scope}_FAULT_SOURCES", "").split(",") if s.strip()
+    ]
+    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)
+    cancelled = sum(r["stats"]["cancelled"] for r in results)
+    verdicts.append(
+        ("零取消泄漏(P7)", "PASS" if cancelled == 0 else f"FAIL — cancelled={cancelled}")
+    )
+    asyncio.run(_ocr_consistency_check(env, args.scope, verdicts))
+
+
+# P7 已知错误前缀(metric ocr-call-success 口径: 异常类名 + 取消哨兵)
+_KNOWN_ERROR_PREFIXES = (
+    "TransientError",
+    "SourceDeadError",
+    "RequestRejectedError",
+    "ResultInvalidError",
+    "AllSourcesExhausted",
+    "CircuitOpenError",
+    "GatewayUnavailableError",
+    "cancelled",
+)
+
+
+async def _ocr_consistency_check(
+    env: dict[str, str], scope: str, verdicts: list[tuple[str, str]], n: int = 20
+) -> None:
+    """P7 不变量⑥: tables/para_blocks 一致性抽查(跑后活检查,设计 §1.2 护栏)。"""
+    import io
+    import random
+    import zipfile
+
+    import httpx
+
+    from polygateway.transports.monkey_ocr import _parse_middle_json
+
+    base = env[f"{scope}__MONKEY__1__BASE_URL"]
+    images = sorted((_ROOT / "data/soak/chs_images").glob("chs_*.jpg"))
+    sample = random.sample(images, min(n, len(images)))
+    mismatches: list[str] = []
+    async with httpx.AsyncClient(base_url=base, trust_env=False, timeout=300) as client:
+        for path in sample:
+            resp = await client.post(
+                "/parse", files={"file": ("image.jpg", path.read_bytes(), "image/jpeg")}
+            )
+            resp.raise_for_status()
+            zip_resp = await client.get(resp.json()["download_url"])
+            zip_resp.raise_for_status()
+            elements, _ = _parse_middle_json(zip_resp.content)
+            lib = sorted(e.bbox for e in elements if e.type == "table")
+            with zipfile.ZipFile(io.BytesIO(zip_resp.content)) as archive:
+                member = next(m for m in archive.namelist() if m.endswith("_middle.json"))
+                payload = json.load(archive.open(member))
+            raw = sorted(
+                tuple(float(v) for v in t["bbox"])
+                for page in payload["pdf_info"]
+                for t in page.get("tables", [])
+            )
+            if lib != raw:
+                mismatches.append(path.name)
+    verdict = "PASS" if not mismatches else f"FAIL — 不一致: {mismatches[:5]}"
+    verdicts.append((f"tables/para_blocks 一致性抽查 n={len(sample)}", verdict))
+
+
 async def _redis_clock_offset_s(redis_url: str) -> float:
     """Redis 服务器钟相对本机钟的偏移(秒);限流窗口 id 以服务器钟为准。"""
     import redis.asyncio as aioredis
@@ -292,7 +376,6 @@ async def _live_checks(
     from polygateway.backends.redis.breaker import RedisGate
     from polygateway.backends.redis.limiter import RedisLimiter
     from polygateway.config import GatewaySettings
-
     from tools.soak import scoreboard as sb
 
     settings = GatewaySettings.from_env(args.scope, env=env)
@@ -322,7 +405,9 @@ async def _live_checks(
 
 def main() -> None:
     parser = argparse.ArgumentParser(description="PolyGateway 真实数据压测")
-    parser.add_argument("--scenario", required=True, choices=["P1", "P2", "P3", "P4", "P5", "P6"])
+    parser.add_argument(
+        "--scenario", required=True, choices=["P1", "P2", "P3", "P4", "P5", "P6", "P7"]
+    )
     parser.add_argument("--budget-calls", type=int)
     parser.add_argument("--budget-tokens", type=int)
     parser.add_argument("--workers", type=int, default=1)

tools/soak/scenarios.py

@@ -1,8 +1,9 @@
-"""P1-P6 场景请求生成器(findings §2 矩阵;async 生成器,产出调用参数)。
+"""P1-P7 场景请求生成器(findings §2 矩阵;async 生成器,产出调用参数)。
 
-每项产出 `(kind, kwargs)`: kind ∈ {"chat"};kwargs 直接喂
-`GatewayClient.chat(**kwargs)`。回放/组装场景一律掺 `cache_salt=run_id`
-破缓存(缓存行为归 P4);P4 子流量特意重复 messages 且不掺 salt。
+每项产出 `(kind, kwargs)`: kind ∈ {"chat", "ocr_text", "ocr_layout"};
+chat 喂 `GatewayClient.chat(**kwargs)`,ocr_* 喂 `OcrClient` 对应方法
+(M3 计划 T8)。回放/组装场景一律掺 `cache_salt=run_id` 破缓存(缓存
+行为归 P4);P4 子流量特意重复 messages 且不掺 salt。
 """
 
 from __future__ import annotations
@@ -10,10 +11,13 @@ from __future__ import annotations
 import random
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, AsyncIterator
+from typing import TYPE_CHECKING, Any
 
 from pydantic import BaseModel
 
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
 from tools.soak.corpus import (
     assemble_frame_messages,
     chs_image_messages,
@@ -182,6 +186,19 @@ async def p6_mixed_soak(
             }[key](corpus, f"{run_id}-r{rng()}", rng)
 
 
+async def p7_ocr(
+    corpus: SoakCorpus, run_id: str, rng=random.random, *, layout_ratio: float = 0.2
+) -> AsyncIterator[Item]:
+    """P7 OCR 双端点混合(M3 计划 T8): 真实图像语料循环,text/layout 8:2。
+
+    故障性来自 SOAK_OCR 源池配置(黑洞/坏端口),生成器本身不造故障。
+    """
+    while True:
+        image = corpus.images[int(rng() * len(corpus.images)) % len(corpus.images)]
+        kind = "ocr_layout" if rng() < layout_ratio else "ocr_text"
+        yield (kind, {"image": image.read_bytes(), "session_id": f"{run_id}-p7"})
+
+
 SCENARIOS = {
     "P1": p1_trace_chains,
     "P2": p2_multimodal_replay,
@@ -189,4 +206,5 @@ SCENARIOS = {
     "P4": p4_cache_bidirectional,
     "P5": p5_fault_mixed,
     "P6": p6_mixed_soak,
+    "P7": p7_ocr,
 }

tools/soak/scoreboard.py

@@ -9,7 +9,7 @@ from __future__ import annotations
 
 import sqlite3
 from collections import Counter
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
@@ -50,7 +50,7 @@ def _admit_second(row: Row, clock_offset_s: float) -> float:
     dt = datetime.fromisoformat(str(row["created_at"]))
     if dt.tzinfo is None:
         # SQLite datetime('now') 落库为 UTC naive;按本机时区解析会错位整时
-        dt = dt.replace(tzinfo=timezone.utc)
+        dt = dt.replace(tzinfo=UTC)
     latency_ms = row.get("latency_ms") or 0
     return dt.timestamp() - float(latency_ms) / 1000.0 + clock_offset_s
 
@@ -193,13 +193,14 @@ def write_report(run_id: str, content: str, out_dir: Path | str = "tests/outputs
 
 # —— 活后端检查与预算帽(2026-07-21 verifier I1/M6/M7 补齐)——
 
-SCENARIO_CALL_CAPS = {  # 设计 §8.1 签字的单场景请求数上限
+SCENARIO_CALL_CAPS = {  # 设计 §8.1 签字的单场景请求数上限(P7: M3 计划 T8)
     "P1": 500,
     "P2": 450,
     "P3": 2500,
     "P4": 2500,
     "P5": 1000,
     "P6": 8000,
+    "P7": 1500,
 }
 
 
@@ -239,6 +240,34 @@ def inv_fault_errors_present(rows: list[Row], *, fault_source_names: list[str])
     assert fault_errors, f"故障源 {fault_source_names} 零错误行——故障混编未生效"
 
 
+def inv_success_rate(ok: int, total: int, *, min_rate: float) -> None:
+    """P7 不变量①: 调用级成功率(worker stats 口径,与 M2.5 验收同源)。"""
+    assert total > 0, "零调用无法评估成功率"
+    rate = ok / total
+    assert rate >= min_rate, f"成功率 {rate:.4f}({ok}/{total})< 阈值 {min_rate}"
+
+
+def inv_fault_share(rows: list[Row], fault_source_names: list[str], *, max_share: float) -> None:
+    """P7 不变量②: 坏源尝试占比受健康选源压制(排除缓存行)。"""
+    attempts = [r for r in rows if not r.get("cache_hit")]
+    assert attempts, "无尝试行"
+    fault = sum(1 for r in attempts if r["source_name"] in fault_source_names)
+    share = fault / len(attempts)
+    assert share <= max_share, (
+        f"坏源尝试占比 {share:.3f}({fault}/{len(attempts)})> 阈值 {max_share}"
+    )
+
+
+def inv_errors_classified(rows: list[Row], known_prefixes: tuple[str, ...]) -> None:
+    """P7 不变量⑦/⑧: 一切错误行可归入已知异常类名前缀(零未分类异常)。"""
+    unknown = Counter(
+        str(r["error"]).split(":")[0]
+        for r in rows
+        if r.get("error") and not str(r["error"]).startswith(known_prefixes)
+    )
+    assert not unknown, f"未分类错误前缀: {dict(unknown)}"
+
+
 def inv_any_errors(rows: list[Row]) -> None:
     """不变量 2c 的兜底形态(P5/P6): 故障源混编池下错误行必然存在。