feat: 查标第 1 步——上传投标分册 → 检查 → 逐条处理 → 复查沿用

后端
- check 模块：检查预览、发起、进度、取消、记录；六个阶段（准备投标文件、响应匹配、规则、证据校验、
  状态沿用、汇总）；问题列表、处理、撤销、批量处理（只限提示级）。
- 每条清单项的检查结论 check_item_result（C9，迁移 0005）：「已响应」也带投标原文出处；
  没有校验通过的证据不能判已响应（C3）。级别从清单项算，不让模型定；说明不写整改建议。
- 响应匹配先只有第三层：内存 BM25 检索候选句 + check.match v1 提示词，检索不到不调模型，
  漏答记 missing_answer，模型临时错误只影响那一批。
- 计费：免费版不记流水；专业版 24 小时内复查免扣（记一行 0）；expected_charge 不一致时拒绝不扣点；
  失败、取消退回；投标文件读不出文字按输入问题失败、不扣点。
- 投标分册：类型（volume_kind）和版本；电子投标平台加密格式在上传时拦下。
- 解析流水线前几个阶段拆到 document/pipeline/normalize.py，解析和检查共用。
- ops 模块：反馈接口；致命问题确认无误的说明自动进误报反馈（U4）。
- SK1：Valkey 客户端短超时、SSE 连接池上限。SK2：CI 加 Gotenberg、直接跑 init-roles.sh。

前端
- 上传投标文件并检查页、查标页（进度、覆盖范围、按级别筛选、逐条处理和行内说明、撤销、原文抽屉、
  已处理、本次已解决）；项目头抽成公共组件，加「查标」标签页；工作台「下一步」覆盖查标阶段。

契约、文档：openapi.yaml 加 item-results、volume_kind、coverage 字段；data-model、api-design §10、
架构 §8 / §13（SK14）、AGENTS.md 同步。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

server/src/bid_doctor/ai/providers/fake.py

@@ -156,6 +156,46 @@ def _project_info(lines: list[tuple[str, str, str, str]]) -> dict[str, Any]:
     return {"fields": {k: fields.get(k, none) for k in INFO_KEYS}, "lots": []}
 
 
+ITEM = re.compile(r"^\[(I\d+)\] (.*)$")
+FRAGMENT = re.compile(r"^\[(d\d+p\d+s\d+)\]（[^）]*）(.*)$")
+CJK2 = re.compile(r"[\u4e00-\u9fff]")
+MATCH_RATIO = 0.35
+
+
+def _bigrams(text: str) -> set[str]:
+    chars = "".join(CJK2.findall(text)) + "".join(re.findall(r"[A-Za-z0-9]+", text)).lower()
+    return {chars[i : i + 2] for i in range(len(chars) - 1)}
+
+
+def _check_match(user: str) -> dict[str, Any]:
+    """候选句和要求原文的字 2-gram 重合比例够高就判「已响应」，引用那一句；否则「未找到」。"""
+    head, _, frags = user.partition("## 投标文件片段")
+    fragments = {m.group(1): m.group(2) for raw in frags.splitlines() if (m := FRAGMENT.match(raw.strip()))}
+    results: list[dict[str, Any]] = []
+    key, req, sids = "", "", []
+    for raw in [*head.splitlines(), "[I0] 结束"]:
+        line = raw.strip()
+        if m := ITEM.match(line):
+            if key:
+                want = _bigrams(req)
+                best = max(sids, key=lambda x: len(want & _bigrams(fragments.get(x, ""))), default=None)
+                hit = best is not None and want and len(want & _bigrams(fragments[best])) / len(want) >= MATCH_RATIO
+                results.append(
+                    {
+                        "item": key,
+                        "verdict": "responded" if hit else "not_found",
+                        "evidence": [{"sid": best, "quote": fragments[best]}] if hit and best else [],
+                        "response_value": None,
+                    }
+                )
+            key, req, sids = m.group(1), m.group(2), []
+        elif line.startswith("要求原文："):
+            req = line.removeprefix("要求原文：")
+        elif line.startswith("候选句："):
+            sids = [x for x in line.removeprefix("候选句：").split("、") if x]
+    return {"results": results}
+
+
 TASKS: dict[str, Callable[[list[tuple[str, str, str, str]]], dict[str, Any]]] = {
     "extract.substantive": _substantive,
     "extract.qualification": _qualification,
@@ -171,5 +211,8 @@ class FakeProvider:
 
     async def complete_json(self, system: str, user: str, capability: str, task: str) -> Completion:
         del system, capability
-        data = TASKS.get(task, lambda _lines: {"items": []})(_lines(user))
+        if task == "check.match":
+            data = _check_match(user)
+        else:
+            data = TASKS.get(task, lambda _lines: {"items": []})(_lines(user))
         return Completion(data=data, raw=json.dumps(data, ensure_ascii=False), model="fake-extractor")

server/src/bid_doctor/app.py

@@ -17,10 +17,12 @@ API_PREFIX = "/api/v1"
 
 def create_app() -> FastAPI:
     from bid_doctor.billing.api import router as billing_router
+    from bid_doctor.check.api import router as check_router
     from bid_doctor.checklist.api import router as checklist_router
     from bid_doctor.document.api import router as document_router
     from bid_doctor.identity.api import router as identity_router
     from bid_doctor.notification.api import router as notification_router
+    from bid_doctor.ops.api import router as ops_router
     from bid_doctor.project.api import router as project_router
 
     settings = get_settings()
@@ -52,7 +54,16 @@ def create_app() -> FastAPI:
         expose_headers=["X-Request-ID", "ETag"],
     )
 
-    for r in (identity_router, billing_router, project_router, document_router, checklist_router, notification_router):
+    for r in (
+        identity_router,
+        billing_router,
+        project_router,
+        document_router,
+        checklist_router,
+        check_router,
+        notification_router,
+        ops_router,
+    ):
         app.include_router(r, prefix=API_PREFIX)
 
     @app.get("/healthz", include_in_schema=False)

server/src/bid_doctor/billing/service.py

@@ -90,6 +90,38 @@ async def hold(
     return entry.id
 
 
+async def balance(session: AsyncSession, ctx: TenantContext, unit: str) -> int:
+    acct = (
+        await session.execute(
+            select(QuotaAccount).where(QuotaAccount.tenant_id == ctx.tenant_id, QuotaAccount.unit == unit)
+        )
+    ).scalar_one_or_none()
+    return acct.balance if acct else 0
+
+
+async def waive(
+    session: AsyncSession, ctx: TenantContext, *, unit: str, ref_type: str, ref_id: UUID, project_id: UUID, reason: str
+) -> None:
+    """免扣也记一行 0（决策 Q13）：用户能在明细里核对这次确实没扣点。"""
+    acct = (
+        await session.execute(
+            select(QuotaAccount).where(QuotaAccount.tenant_id == ctx.tenant_id, QuotaAccount.unit == unit)
+        )
+    ).scalar_one()
+    await _append(
+        session,
+        ctx,
+        acct,
+        kind="waive",
+        amount=0,
+        key=f"{ref_type}:{ref_id}:waive",
+        ref_type=ref_type,
+        ref_id=ref_id,
+        project_id=project_id,
+        reason=reason,
+    )
+
+
 async def settle(session: AsyncSession, ctx: TenantContext, hold_id: UUID) -> None:
     await _finish(session, ctx, hold_id, "settle", 0)
 

server/src/bid_doctor/check/api.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+from typing import Any, Literal
+from uuid import UUID
+
+from fastapi import APIRouter
+from pydantic import BaseModel, Field
+
+from bid_doctor.check import service
+from bid_doctor.core.db import tenant_tx
+from bid_doctor.identity.deps import Ctx
+from bid_doctor.project import service as projects
+
+router = APIRouter(tags=["checks"])
+
+
+class CheckRunIn(BaseModel):
+    document_ids: list[UUID] = Field(min_length=1)
+    expected_charge: Literal[0, 1]
+
+
+class ResolveIn(BaseModel):
+    status: Literal["confirmed_ok", "modified", "ignored"]
+    note: str | None = Field(default=None, max_length=500)
+
+
+class BulkResolveIn(BaseModel):
+    finding_ids: list[UUID] = Field(min_length=1, max_length=500)
+    status: Literal["confirmed_ok", "ignored"]
+    note: str | None = Field(default=None, max_length=500)
+
+
+@router.get("/projects/{project_id}/check-preview")
+async def get_check_preview(project_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        return await service.preview(s, ctx, project_id)
+
+
+@router.get("/projects/{project_id}/check-runs")
+async def list_check_runs(project_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        await projects.get(s, ctx, project_id)
+        return {"items": [await service.run_to_api(s, ctx, r) for r in await service.list_runs(s, project_id)]}
+
+
+@router.post("/projects/{project_id}/check-runs", status_code=201)
+async def start_check_run(project_id: UUID, body: CheckRunIn, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        run = await service.start(s, ctx, project_id, body.document_ids, body.expected_charge)
+        return await service.run_to_api(s, ctx, run)
+
+
+@router.get("/check-runs/{run_id}")
+async def get_check_run(run_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        return await service.run_to_api(s, ctx, await service.get_run(s, run_id))
+
+
+@router.post("/check-runs/{run_id}/cancel", status_code=202)
+async def cancel_check_run(run_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        return await service.run_to_api(s, ctx, await service.request_cancel(s, run_id))
+
+
+@router.get("/check-runs/{run_id}/findings")
+async def list_findings(
+    run_id: UUID,
+    ctx: Ctx,
+    severity: Literal["fatal", "important", "hint"] | None = None,
+    status: Literal["pending", "confirmed_ok", "modified", "ignored"] | None = None,
+    presence: Literal["new", "carried", "resolved"] | None = None,
+) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        return {"items": await service.findings(s, run_id, severity=severity, status=status, presence=presence)}
+
+
+@router.get("/check-runs/{run_id}/item-results")
+async def list_item_results(run_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        return {"items": await service.item_results(s, run_id)}
+
+
+@router.post("/check-runs/{run_id}/findings/bulk-resolve")
+async def bulk_resolve_findings(run_id: UUID, body: BulkResolveIn, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        return await service.bulk_resolve(s, ctx, run_id, body.finding_ids, body.status, body.note)
+
+
+@router.post("/findings/{finding_id}/resolve")
+async def resolve_finding(finding_id: UUID, body: ResolveIn, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        return await service.resolve(s, ctx, finding_id, body.status, body.note)
+
+
+@router.post("/findings/{finding_id}/reopen")
+async def reopen_finding(finding_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        return await service.reopen(s, ctx, finding_id)

server/src/bid_doctor/check/findings.py

@@ -0,0 +1,85 @@
+"""由每条清单项的检查结论生成问题（FR-M3-03、m3-plan §2.3）。
+
+- 级别从清单项算，不让模型定（参考项目让模型定级别，模型可以把高危项降成低危）；
+- 说明只写事实和出处，不写整改建议（报价类问题的「建议」会写出数字，红线 6）；
+- 「已响应」不生成问题，结论留在 check_item_result 里。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from bid_doctor.check.match import Result
+
+FATAL_CATEGORIES = {"qualification", "invalidity"}
+REASONS = {
+    "evidence_missing": "没有给出投标原文",
+    "sid_format": "引用的句子编号写法不对",
+    "sid_unknown": "引用的句子不在候选范围内",
+    "quote_not_original": "引文和投标原文对不上",
+    "quote_too_short": "引文太短，无法核对",
+    "missing_answer": "这一条没有得到判断结果",
+    "model_error": "模型服务暂时不可用",
+}
+
+
+@dataclass(frozen=True, slots=True)
+class Spec:
+    type: str
+    severity: str
+    title: str
+    explanation: str
+    detail: dict[str, Any]
+
+
+def severity(item: dict[str, Any]) -> str:
+    """★ 条款、法规列明或原文写明无效的、资格条件 → 致命；▲ 参数、评分项（缺证明材料）→ 重要；其余 → 提示。"""
+    cats = set(item.get("categories") or [])
+    if item.get("importance") == "substantive" or cats & FATAL_CATEGORIES:
+        return "fatal"
+    if item.get("importance") == "important" or "scoring" in cats:
+        return "important"
+    return "hint"
+
+
+def _pages(pages: list[int]) -> str:
+    return "、".join(f"第 {p} 页" for p in pages)
+
+
+def spec(result: Result, item: dict[str, Any]) -> Spec | None:
+    sev = severity(item)
+    name = str(item.get("title") or "")
+    base = {"importance_basis": item.get("importance_basis"), **result.detail}
+    scope = "、".join(result.detail.get("search_scope") or [])
+    if result.verdict == "responded":
+        return None
+    if result.verdict == "not_found" and result.evidence_status == "unverified":
+        claimed = result.detail.get("claimed_pages") or []
+        where = f"{_pages(claimed)}有响应" if claimed else "有响应"
+        why = REASONS.get(result.reason_code or "", result.reason_code or "")
+        text = f"模型认为投标文件{where}，但出处没能核对（{why}），按「未找到响应」处理。请人工核对。"
+        return Spec("unresponded", sev, f"未找到响应：{name}", text, {**base, "evidence_status": "unverified"})
+    if result.verdict == "not_found":
+        text = f"在投标文件（{scope}）中没有找到对这条要求的响应。" if scope else "没有找到对这条要求的响应。"
+        return Spec("unresponded", sev, f"未找到响应：{name}", text, base)
+    if result.verdict == "negative_deviation":
+        return Spec(
+            "negative_deviation", sev, f"负偏离未说明：{name}", "投标文件写明了偏离或不满足，没有找到说明。", base
+        )
+    if result.verdict == "suspected_unmet":
+        value = result.detail.get("response_value")
+        text = (
+            f"投标文件给出的值（{value}）按要求值判断可能不满足。"
+            if value
+            else "投标文件给出的值按要求值判断可能不满足。"
+        )
+        return Spec("suspected_unmet", sev, f"疑似不满足：{name}", text, base)
+    if result.verdict == "no_value":
+        text = "投标文件只写了满足或响应，没有给出要求里要的具体值或材料。"
+        return Spec("suspected_unmet", "hint", f"未给出具体值：{name}", text, {**base, "kind": "no_value"})
+    if result.verdict == "scanned":
+        text = "候选内容所在的页面是扫描件，系统读不出文字，请人工核对。"
+        return Spec("manual_review", sev, f"请人工核对：{name}", text, base)
+    why = REASONS.get(result.reason_code or "", result.reason_code or "")
+    return Spec("manual_review", sev, f"未判定：{name}", f"这条要求本次没有完成判断（{why}），请人工核对。", base)

server/src/bid_doctor/check/match.py

@@ -0,0 +1,232 @@
+"""响应匹配（m3-plan §2.1）：对每条清单项在投标文件里找响应，给出带证据的结论。
+
+第 1 步只有第三层（检索 + 模型判断）：
+- 检索不到候选句的，不调模型，直接「未找到响应」（参考项目的做法）；
+- 其余按批交给模型，模型只能引用候选句的编号；每条必答，漏答记 missing_answer、结论「未判定」；
+- 投标侧证据过前两道校验（编号、原文）。**没有校验通过的证据，不能判为「已响应」**：
+  模型说已响应但证据没过的，记「未找到响应」并带原因码（C3）；
+- 模型输出被截断时对半重跑；临时错误只影响这一批，条目记「未判定（模型错误）」，全部失败才算任务失败。
+TODO(第 3 步)：偏离表直取、参数规则比较、按投标章节打批、token 预算、复查复用、重复条目分组、
+致命项重问一次、第三道校验（按类别核对响应值）。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+from typing import Any
+from uuid import UUID
+
+from bid_doctor.ai import service as ai
+from bid_doctor.ai.providers.base import TransientError
+from bid_doctor.check.retrieval import Index
+from bid_doctor.core.db import TenantContext
+from bid_doctor.document.pipeline.text_layer import Sentence, norm
+
+PROMPT = ("check.match", 1)
+CANDIDATES = 5  # 每条清单项送给模型的候选句数
+BATCH = 8  # 每次调用判断的清单项数
+SPLIT_DEPTH = 2
+CONCURRENCY = 4  # TODO(第 3 步)：和解析共用「供应商 × 租户」统一限流（extract.py 的同一个 TODO）
+SID = re.compile(r"^d(\d+)p(\d+)s(\d+)$")
+END_PUNCT = "。；;，,、：:）)"
+VERDICTS = {"responded", "not_found", "negative_deviation", "suspected_unmet", "no_value"}
+CATEGORY_LABELS = {
+    "qualification": "资格条件",
+    "invalidity": "无效 / 废标条款",
+    "scoring": "评分项",
+    "schedule": "日期与递交",
+    "tech_param": "技术参数",
+    "format_doc": "格式文件",
+    "price": "报价",
+}
+
+
+@dataclass(slots=True)
+class BidDoc:
+    id: UUID
+    no: int  # 在本次检查里的序号，组成短编号 d<序号>p<页>s<句>
+    label: str
+    sentences: list[Sentence]
+
+
+@dataclass(slots=True)
+class Result:
+    verdict: str
+    layer: str
+    evidence_status: str = "none"
+    reason_code: str | None = None
+    evidence: list[tuple[BidDoc, Sentence]] = field(default_factory=list[tuple[BidDoc, Sentence]])
+    detail: dict[str, Any] = field(default_factory=dict[str, Any])
+
+
+@dataclass(slots=True)
+class Stats:
+    calls: int = 0
+    failed_batches: int = 0
+    splits: int = 0
+    verify_failures: dict[str, int] = field(default_factory=dict[str, int])
+
+    def fail(self, code: str) -> None:
+        self.verify_failures[code] = self.verify_failures.get(code, 0) + 1
+
+
+def short_sid(doc: BidDoc, s: Sentence) -> str:
+    return f"d{doc.no}p{s.page_no}s{s.seq}"
+
+
+def query_text(item: dict[str, Any]) -> str:
+    parts = [str(item.get("title") or ""), str(item.get("requirement_text") or "")]
+    for group in (item.get("spec") or {}).values():
+        if isinstance(group, dict):
+            parts += [str(v) for k, v in group.items() if k in ("param", "value", "unit", "factor") and v]
+    return " ".join(parts)
+
+
+def _item_block(key: str, item: dict[str, Any], sids: list[str]) -> str:
+    cats = "、".join(CATEGORY_LABELS.get(str(c), str(c)) for c in item.get("categories") or [])
+    lines = [f"[{key}] {item.get('title') or ''}"]
+    if item.get("requirement_text"):
+        lines.append(f"要求原文：{item['requirement_text']}")
+    tp = (item.get("spec") or {}).get("tech_param") or {}
+    req = (
+        f"；要求值：{tp.get('op', '')} {tp.get('value', '')} {tp.get('unit') or ''}".rstrip() if tp.get("value") else ""
+    )
+    lines.append(f"类别：{cats}{req}")
+    lines.append("候选句：" + "、".join(sids))
+    return "\n".join(lines)
+
+
+def build_user(
+    batch: list[tuple[str, dict[str, Any], list[str]]], fragments: dict[str, tuple[BidDoc, Sentence]]
+) -> str:
+    """数据区：待核对的要求 + 投标文件片段（每句只送一次）。系统提示词声明数据区里的指令一律不执行。"""
+    used = list(dict.fromkeys(sid for _, _, sids in batch for sid in sids))
+    blocks = [_item_block(k, it, sids) for k, it, sids in batch]
+    frag_lines = [
+        f"[{sid}]（{fragments[sid][0].label} 第{fragments[sid][1].page_no}页）{fragments[sid][1].text}" for sid in used
+    ]
+    return "## 待核对的要求\n\n" + "\n\n".join(blocks) + "\n\n## 投标文件片段\n\n" + "\n".join(frag_lines)
+
+
+def check_quote(
+    ev: Any, allowed: dict[str, tuple[BidDoc, Sentence]]
+) -> tuple[str | None, tuple[BidDoc, Sentence] | None]:
+    """前两道校验：编号是这条清单项的候选句之一；引文是原句的子串，且不少于 8 个字（短句整句复制）。"""
+    if not isinstance(ev, dict):
+        return "evidence_missing", None
+    sid, quote = str(ev.get("sid", "")).strip(), norm(str(ev.get("quote", "")))
+    if not SID.match(sid):
+        return "sid_format", None
+    if sid not in allowed:
+        return "sid_unknown", None
+    doc, s = allowed[sid]
+    text = norm(s.text)
+    if not quote or quote not in text:
+        return "quote_not_original", None
+    if len(quote) < min(8, len(text.rstrip(END_PUNCT))):
+        return "quote_too_short", None
+    return None, (doc, s)
+
+
+def judge(answer: Any, allowed: dict[str, tuple[BidDoc, Sentence]], scope: list[str]) -> Result:
+    """把模型对一条清单项的回答变成结论。模型说「已响应 / 不满足」但证据没通过的，记「未找到响应」（C3）。"""
+    if not isinstance(answer, dict) or answer.get("verdict") not in VERDICTS:
+        return Result("undetermined", "model", reason_code="missing_answer", detail={"search_scope": scope})
+    verdict = str(answer["verdict"])
+    if verdict == "not_found":
+        return Result("not_found", "model", detail={"search_scope": scope})
+    evs = answer.get("evidence") or []
+    checked = [check_quote(e, allowed) for e in evs] if evs else [("evidence_missing", None)]
+    fails = [f for f, _ in checked if f]
+    ok = [x for f, x in checked if f is None and x is not None]
+    if fails or not ok:
+        pages = sorted(
+            {allowed[str(e.get("sid"))][1].page_no for e in evs if isinstance(e, dict) and str(e.get("sid")) in allowed}
+        )
+        return Result(
+            "not_found",
+            "model",
+            evidence_status="unverified",
+            reason_code=fails[0],
+            detail={"claimed_verdict": verdict, "claimed_pages": pages, "search_scope": scope},
+        )
+    detail: dict[str, Any] = {}
+    if answer.get("response_value"):
+        detail["response_value"] = str(answer["response_value"])[:100]
+    return Result(verdict, "model", evidence_status="verified", evidence=ok, detail=detail)
+
+
+async def run(
+    ctx: TenantContext,
+    run_id: UUID,
+    items: dict[UUID, dict[str, Any]],
+    docs: list[BidDoc],
+    progress: Callable[[float], Awaitable[None]],
+) -> tuple[dict[UUID, Result], Stats]:
+    stats = Stats()
+    scope = [d.label for d in docs]
+    index = Index([s for d in docs for s in d.sentences])
+    doc_of = {s.key: d for d in docs for s in d.sentences}
+    results: dict[UUID, Result] = {}
+    queue: list[tuple[UUID, dict[str, Any], dict[str, tuple[BidDoc, Sentence]]]] = []
+    for item_id, item in items.items():
+        hits = index.search(query_text(item), CANDIDATES)
+        if not hits:  # 检索不到候选句：不调模型（m3-plan §2.1）
+            results[item_id] = Result("not_found", "retrieval", detail={"search_scope": scope})
+            continue
+        allowed = {short_sid(doc_of[h.sentence.key], h.sentence): (doc_of[h.sentence.key], h.sentence) for h in hits}
+        queue.append((item_id, item, allowed))
+
+    sem = asyncio.Semaphore(CONCURRENCY)
+    total, done = max(len(queue), 1), [0]
+
+    async def call(batch: list[tuple[UUID, dict[str, Any], dict[str, tuple[BidDoc, Sentence]]]], depth: int) -> None:
+        keyed = [(f"I{i + 1}", item_id, item, allowed) for i, (item_id, item, allowed) in enumerate(batch)]
+        fragments = {sid: x for _, _, _, allowed in keyed for sid, x in allowed.items()}
+        user = build_user([(k, item, list(allowed)) for k, _, item, allowed in keyed], fragments)
+        try:
+            async with sem:
+                stats.calls += 1
+                data = await ai.run_json(
+                    ctx,
+                    job_type="check_run",
+                    job_id=run_id,
+                    task="check.match",
+                    prompt_id=PROMPT[0],
+                    version=PROMPT[1],
+                    user=user,
+                )
+        except ai.OutputIncomplete as e:
+            if depth < SPLIT_DEPTH and len(batch) > 1:
+                stats.splits += 1
+                mid = len(batch) // 2
+                await asyncio.gather(call(batch[:mid], depth + 1), call(batch[mid:], depth + 1))
+                return
+            data = e.data  # 拆到底仍不完整：用已经解析出来的部分，没答到的按漏答处理
+        except TransientError:
+            stats.failed_batches += 1
+            for _, item_id, _, _ in keyed:
+                results[item_id] = Result(
+                    "undetermined", "model", reason_code="model_error", detail={"search_scope": scope}
+                )
+            done[0] += len(batch)
+            await progress(done[0] / total)
+            return
+        answers = {str(a.get("item")): a for a in data.get("results") or [] if isinstance(a, dict)}
+        for key, item_id, _, allowed in keyed:
+            r = judge(answers.get(key), allowed, scope)
+            if r.reason_code and r.reason_code != "model_error":
+                stats.fail(r.reason_code)
+            results[item_id] = r
+        done[0] += len(batch)
+        await progress(done[0] / total)
+
+    await asyncio.gather(*(call(queue[i : i + BATCH], 0) for i in range(0, len(queue), BATCH)))
+    if queue and all(results[item_id].reason_code == "model_error" for item_id, _, _ in queue):
+        raise TransientError(
+            "model_error", "all check.match batches failed"
+        )  # 一批都没成功：按基础设施错误失败，退回额度
+    return results, stats

server/src/bid_doctor/check/models.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from sqlalchemy import FetchedValue
+from sqlalchemy.orm import Mapped, mapped_column
+
+from bid_doctor.core.columns import PK, Bytes, DbTs, Json, TextArray, Ts, UuidArray, UuidCol
+from bid_doctor.core.db import Base
+
+
+class CheckRun(Base):
+    """检查批次（FR-M3-02）：记下当时的清单快照、投标文件版本、规则集版本、计费方式。"""
+
+    __tablename__ = "check_run"
+    id: Mapped[PK]
+    tenant_id: Mapped[UuidCol]
+    project_id: Mapped[UuidCol]
+    run_no: Mapped[int]
+    checklist_version_id: Mapped[UuidCol]
+    checklist_confirmed: Mapped[bool]
+    rule_set_version: Mapped[str]
+    legal_regime: Mapped[str]
+    lot_ids: Mapped[UuidArray]
+    previous_run_id: Mapped[UuidCol | None]
+    status: Mapped[str] = mapped_column(server_default=FetchedValue())
+    stage: Mapped[str | None]
+    checkpoints: Mapped[Json] = mapped_column(server_default=FetchedValue())
+    progress: Mapped[int] = mapped_column(server_default=FetchedValue())
+    cancel_requested_at: Mapped[Ts | None]
+    error_class: Mapped[str | None]
+    error_code: Mapped[str | None]
+    error_detail: Mapped[str | None]
+    billing: Mapped[str]
+    quota_hold_id: Mapped[UuidCol | None]
+    prompt_versions: Mapped[Json | None]
+    coverage: Mapped[Json | None]
+    counts: Mapped[Json | None]
+    queue_job_id: Mapped[int | None]
+    created_by: Mapped[UuidCol]
+    created_at: Mapped[DbTs]
+    started_at: Mapped[Ts | None]
+    finished_at: Mapped[Ts | None]
+    updated_at: Mapped[DbTs]
+
+
+class CheckRunDocument(Base):
+    __tablename__ = "check_run_document"
+    tenant_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    check_run_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    document_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+
+
+class CheckItemResult(Base):
+    """每条清单项的检查结论，「已响应」也留（C9）。"""
+
+    __tablename__ = "check_item_result"
+    tenant_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    check_run_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    checklist_item_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    verdict: Mapped[str]
+    layer: Mapped[str]
+    evidence_status: Mapped[str]
+    reason_code: Mapped[str | None]
+    bid_citation_ids: Mapped[UuidArray]
+    detail: Mapped[Json] = mapped_column(server_default=FetchedValue())
+    group_key: Mapped[str | None]
+    cache_key: Mapped[Bytes | None]
+    reused_from_run_id: Mapped[UuidCol | None]
+    created_at: Mapped[DbTs]
+
+
+class Finding(Base):
+    __tablename__ = "finding"
+    id: Mapped[PK]
+    tenant_id: Mapped[UuidCol]
+    project_id: Mapped[UuidCol]
+    check_run_id: Mapped[UuidCol]
+    checklist_item_id: Mapped[UuidCol | None]
+    rule_code: Mapped[str | None]
+    match_key: Mapped[str]
+    type: Mapped[str]
+    severity: Mapped[str]
+    title: Mapped[str]
+    explanation: Mapped[str]
+    detector: Mapped[str]
+    detail: Mapped[Json] = mapped_column(server_default=FetchedValue())
+    search_scope: Mapped[Json | None]
+    checks_passed: Mapped[TextArray]
+    presence: Mapped[str]
+    carried_from_id: Mapped[UuidCol | None]
+    status: Mapped[str] = mapped_column(server_default=FetchedValue())
+    resolution_note: Mapped[str | None]
+    resolved_by: Mapped[UuidCol | None]
+    resolved_at: Mapped[Ts | None]
+    created_at: Mapped[DbTs]
+    updated_at: Mapped[DbTs]
+
+
+class FindingCitation(Base):
+    __tablename__ = "finding_citation"
+    tenant_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    finding_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    citation_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    side: Mapped[str]
+    position: Mapped[int]
+
+
+class ManualCheck(Base):
+    """需人工核对的勾选：按项目存，复查时沿用（P2）。"""
+
+    __tablename__ = "manual_check"
+    tenant_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    project_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    check_key: Mapped[str] = mapped_column(primary_key=True)
+    checked: Mapped[bool]
+    note: Mapped[str | None]
+    checked_by: Mapped[UuidCol]
+    checked_at: Mapped[DbTs]

server/src/bid_doctor/check/retrieval.py

@@ -0,0 +1,64 @@
+"""投标文件检索：内存里的 BM25（C11、m3-plan §2.2）。只在 worker 里、一次检查之内用，纯函数，便于测试和离线评测。
+
+- 索引单位是带编号的句子（含表格单元格），出处能直接回链、三道校验能复用；
+- 分词：汉字 2-gram，「数字 + 单位」（≥32GB → 32gb）和英文型号（10GE、BFD）保留成整词。
+TODO(第 3 步)：评测 jieba 分词 + 字 2/3-gram 混合是否更好；按类别构造查询、结构先验排序。
+"""
+
+from __future__ import annotations
+
+import math
+import re
+from collections import Counter
+from dataclasses import dataclass
+
+from bid_doctor.document.pipeline.text_layer import Sentence
+
+NUM_UNIT = re.compile(r"(\d+(?:\.\d+)?)\s*([A-Za-z]+|[个台套项条次人天年月日分秒毫兆核位路口])?")
+LATIN = re.compile(r"[A-Za-z][A-Za-z0-9\-]*")
+CJK = re.compile(r"[一-鿿]+")
+K1, B = 1.5, 0.75
+
+
+def tokens(text: str) -> list[str]:
+    out = [(n + (u or "")).lower() for n, u in NUM_UNIT.findall(text)]
+    out += [w.lower() for w in LATIN.findall(text)]
+    for run in CJK.findall(text):
+        out += [run[i : i + 2] for i in range(len(run) - 1)] or [run]
+    return out
+
+
+@dataclass(slots=True)
+class Hit:
+    sentence: Sentence
+    score: float
+
+
+class Index:
+    def __init__(self, sentences: list[Sentence]) -> None:
+        self.sentences = sentences
+        self.tf = [Counter(tokens(s.text)) for s in sentences]
+        self.lengths = [sum(c.values()) for c in self.tf]
+        self.avg = (sum(self.lengths) / len(self.lengths)) if self.lengths else 0.0
+        df: Counter[str] = Counter()
+        for c in self.tf:
+            df.update(c.keys())
+        n = len(sentences)
+        self.idf = {t: math.log(1 + (n - f + 0.5) / (f + 0.5)) for t, f in df.items()}
+
+    def search(self, query: str, k: int) -> list[Hit]:
+        q = set(tokens(query))
+        if not q or not self.sentences:
+            return []
+        scored: list[Hit] = []
+        for i, tf in enumerate(self.tf):
+            score = 0.0
+            norm = K1 * (1 - B + B * self.lengths[i] / (self.avg or 1))
+            for t in q:
+                f = tf.get(t)
+                if f:
+                    score += self.idf[t] * f * (K1 + 1) / (f + norm)
+            if score > 0:
+                scored.append(Hit(self.sentences[i], score))
+        scored.sort(key=lambda h: -h.score)
+        return scored[:k]

server/src/bid_doctor/check/runner.py

@@ -0,0 +1,424 @@
+"""检查批次的流水线（m3-plan §2）：准备投标文件 → 响应匹配 → 规则检查 → 证据校验 → 状态沿用 → 汇总。
+
+- 和解析任务同一套做法（架构 §3.4）：阶段之间检查取消标记；结局由错误类别决定扣不扣点，失败、取消都退回。
+- 健全性检查：投标文件几乎读不出文字时按输入问题失败，不扣点、不生成问题（参考项目的同类闸门）。
+- 认不出、查不了、没查的都写进检查覆盖范围（coverage），不静默跳过（m3-plan §2.5）。
+TODO(第 2 步)：规则检查（报价形式、形式要件、需人工核对）；投标侧结构识别（偏离表、开标一览表、格式文件）。
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime
+from typing import Any
+from uuid import UUID
+
+import httpx
+from sqlalchemy import delete, func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from bid_doctor.ai.models import AiCall
+from bid_doctor.ai.providers.base import TransientError
+from bid_doctor.billing import service as billing
+from bid_doctor.check import findings, match
+from bid_doctor.check.models import CheckItemResult, CheckRun, CheckRunDocument, Finding, FindingCitation
+from bid_doctor.check.service import RULE_SET_VERSION, open_fatal
+from bid_doctor.checklist import edit as checklist_edit
+from bid_doctor.checklist import service as checklist
+from bid_doctor.core import events
+from bid_doctor.core.db import TenantContext, tenant_tx
+from bid_doctor.core.errors import AppError
+from bid_doctor.core.ids import uuid7
+from bid_doctor.document import citations
+from bid_doctor.document.citations import document_label
+from bid_doctor.document.models import SourceDocument
+from bid_doctor.document.pipeline import normalize, text_layer
+from bid_doctor.notification.models import Notification
+from bid_doctor.project import service as projects
+
+log = logging.getLogger(__name__)
+PROGRESS = {"prepare_bid": 5, "match": 30, "rules": 85, "verify": 90, "carry_over": 93, "summarize": 96}
+MIN_BID_CHARS = 200  # 所有分册加起来读不出这么多字：按「投标文件几乎读不出文字」失败
+REGIMES_VERIFIED = {"gov_procurement"}  # 招标投标体系没有样本，规则未经验证（m3-plan §2.4）
+
+
+class Cancelled(Exception):
+    pass
+
+
+async def run(tenant_id: UUID, run_id: UUID) -> None:
+    ctx = TenantContext(tenant_id=tenant_id, user_id=UUID(int=0))
+    async with tenant_tx(ctx) as s:
+        r = (await s.execute(select(CheckRun).where(CheckRun.id == run_id))).scalar_one()
+        if r.status not in ("queued", "running"):
+            return  # 重复派发：已经有结局了
+        ctx = TenantContext(tenant_id=tenant_id, user_id=r.created_by, actor_name="系统")
+        r.status, r.started_at = "running", r.started_at or datetime.now(UTC)
+    try:
+        await _stages(ctx, run_id)
+    except Cancelled:
+        await _finish(ctx, run_id, "cancelled", "cancelled", "cancelled")
+    except AppError as e:
+        await _finish(ctx, run_id, "failed", "input", e.code)
+    except (TransientError, httpx.HTTPError) as e:
+        log.warning("check run %s infra error: %s", run_id, e)
+        code = "model_unavailable" if isinstance(e, TransientError) else "converter_unavailable"
+        await _finish(ctx, run_id, "failed", "infra", code)
+    except Exception:
+        log.exception("check run %s crashed", run_id)
+        await _finish(ctx, run_id, "failed", "infra", "internal_error")
+    else:
+        await _finish(ctx, run_id, "succeeded", None, None)
+
+
+async def _publish(ctx: TenantContext, r: CheckRun, status: str, stage: str | None, pct: int) -> None:
+    await events.publish(
+        ctx.tenant_id,
+        "job.progress" if status == "running" else "job.finished",
+        project_id=str(r.project_id),
+        job={"kind": "check_run", "id": str(r.id), "status": status, "stage": stage, "progress": pct},
+    )
+
+
+async def _stage(ctx: TenantContext, run_id: UUID, stage: str, pct: int | None = None) -> None:
+    """进入一个阶段（或更新阶段内的进度）：写进度、检查取消标记。"""
+    async with tenant_tx(ctx) as s:
+        r = (await s.execute(select(CheckRun).where(CheckRun.id == run_id))).scalar_one()
+        if r.cancel_requested_at:
+            raise Cancelled
+        r.stage, r.progress = stage, pct if pct is not None else PROGRESS[stage]
+    await _publish(ctx, r, "running", stage, r.progress)
+
+
+async def _stages(ctx: TenantContext, run_id: UUID) -> None:
+    async with tenant_tx(ctx) as s:
+        r = (await s.execute(select(CheckRun).where(CheckRun.id == run_id))).scalar_one()
+        doc_ids = list(
+            (
+                await s.execute(select(CheckRunDocument.document_id).where(CheckRunDocument.check_run_id == run_id))
+            ).scalars()
+        )
+        docs = list(
+            (
+                await s.execute(
+                    select(SourceDocument).where(SourceDocument.id.in_(doc_ids)).order_by(SourceDocument.created_at)
+                )
+            ).scalars()
+        )
+        r.prompt_versions = {match.PROMPT[0]: f"v{match.PROMPT[1]}"}
+
+    await _stage(ctx, run_id, "prepare_bid")
+
+    async def on_stage(_: str) -> None:
+        await _stage(ctx, run_id, "prepare_bid")  # 子阶段（转换、文字层）只用来检查取消标记
+
+    bid_docs: list[match.BidDoc] = []
+    pages_info: list[dict[str, Any]] = []
+    for no, d in enumerate(docs, start=1):
+        pages = await normalize.pages(ctx, d.id, on_stage, reuse=True, require_text=False)
+        label = document_label(d.role, d.seq_no, d.volume_name, d.volume_kind)
+        bid_docs.append(match.BidDoc(d.id, no, label, text_layer.body_sentences(pages)))
+        pages_info.append(
+            {
+                "id": str(d.id),
+                "label": label,
+                "pages": len(pages),
+                "pages_without_text": [p.page_no for p in pages if p.text_source == "empty"],
+                "chars": sum(p.char_count for p in pages),
+            }
+        )
+    if sum(x["chars"] for x in pages_info) < MIN_BID_CHARS:
+        raise AppError("bid_text_missing")
+
+    await _stage(ctx, run_id, "match")
+    async with tenant_tx(ctx) as s:
+        r = (await s.execute(select(CheckRun).where(CheckRun.id == run_id))).scalar_one()
+        snapshot = await checklist_edit.version_items(s, r.checklist_version_id)
+        lots = await projects.lot_labels(s, r.project_id)
+        pending = (await checklist.pending_counts(s, [r.project_id])).get(r.project_id, 0)
+    bidding = set(r.lot_ids)
+    items, other_lot = {}, 0
+    for item_id, payload in snapshot.items():
+        lot = UUID(payload["lot_id"]) if payload.get("lot_id") else None
+        if lot is None or lots.get(lot) is None or not bidding or lot in bidding:
+            items[item_id] = payload
+        else:
+            other_lot += 1  # 不在所投包里的条目不查，写进覆盖范围（C6）
+    last = [PROGRESS["match"]]
+
+    async def progress(ratio: float) -> None:
+        pct = PROGRESS["match"] + int((PROGRESS["rules"] - PROGRESS["match"]) * ratio)
+        if pct > last[0]:
+            last[0] = pct
+            await _stage(ctx, run_id, "match", pct)
+
+    results, stats = await match.run(ctx, run_id, items, bid_docs, progress)
+
+    await _stage(ctx, run_id, "rules")  # 第 2 步：报价形式、形式要件、需人工核对
+    await _stage(ctx, run_id, "verify")  # 前两道校验在匹配时逐条做完；这里只是进度节点
+    await _stage(ctx, run_id, "carry_over")
+    async with tenant_tx(ctx) as s:
+        r = (await s.execute(select(CheckRun).where(CheckRun.id == run_id))).scalar_one()
+        # 重新读：规范化 PDF 的哈希是 prepare_bid 阶段才写上的，出处要用它
+        by_doc = {
+            d.id: d for d in (await s.execute(select(SourceDocument).where(SourceDocument.id.in_(doc_ids)))).scalars()
+        }
+        await _write(s, ctx, r, items, results, by_doc)
+        await _carry_over(s, ctx, r, set(items))
+        _stage_nowait(r, "summarize")
+        tokens = (
+            await s.execute(
+                select(
+                    func.coalesce(func.sum(AiCall.tokens_in), 0), func.coalesce(func.sum(AiCall.tokens_out), 0)
+                ).where(AiCall.job_id == run_id)
+            )
+        ).one()
+        r.counts = await _counts(s, run_id)
+        r.coverage = _coverage(r, pages_info, items, results, pending, other_lot, stats, tokens)
+    await _publish(ctx, r, "running", "summarize", PROGRESS["summarize"])
+
+
+def _stage_nowait(r: CheckRun, stage: str) -> None:
+    r.stage, r.progress = stage, PROGRESS[stage]
+
+
+async def _write(
+    s: AsyncSession,
+    ctx: TenantContext,
+    r: CheckRun,
+    items: dict[UUID, dict[str, Any]],
+    results: dict[UUID, match.Result],
+    by_doc: dict[UUID, SourceDocument],
+) -> None:
+    """每条清单项留一行结论（C9）；没响应好的生成问题，挂招标侧和投标侧出处。阶段幂等：先清本批次的残留。"""
+    await s.execute(delete(Finding).where(Finding.check_run_id == r.id))
+    await s.execute(delete(CheckItemResult).where(CheckItemResult.check_run_id == r.id))
+    for item_id, item in items.items():
+        res = results[item_id]
+        bid_cites = []
+        for doc_id in dict.fromkeys(d.id for d, _ in res.evidence):
+            sents = [x for d, x in res.evidence if d.id == doc_id]
+            bid_cites += await citations.from_sentences(s, ctx, by_doc[doc_id], sents, "ai", ["id", "quote"])
+        s.add(
+            CheckItemResult(
+                tenant_id=ctx.tenant_id,
+                check_run_id=r.id,
+                checklist_item_id=item_id,
+                verdict=res.verdict,
+                layer=res.layer,
+                evidence_status=res.evidence_status,
+                reason_code=res.reason_code,
+                bid_citation_ids=[c.id for c in bid_cites],
+                detail=res.detail,
+            )
+        )
+        spec = findings.spec(res, item)
+        if spec is None:
+            continue
+        f = Finding(
+            id=uuid7(),
+            tenant_id=ctx.tenant_id,
+            project_id=r.project_id,
+            check_run_id=r.id,
+            checklist_item_id=item_id,
+            match_key=f"item:{item_id}",
+            type=spec.type,
+            severity=spec.severity,
+            title=spec.title[:200],
+            explanation=spec.explanation,
+            detector="ai",
+            detail={k: v for k, v in spec.detail.items() if k != "search_scope"},
+            search_scope={"documents": res.detail.get("search_scope") or [], "sections": []}
+            if spec.type == "unresponded"
+            else None,
+            checks_passed=["id", "quote"] if res.evidence_status == "verified" else [],
+            presence="new",
+        )
+        s.add(f)
+        await s.flush()
+        for pos, cid in enumerate(item.get("citation_ids") or []):
+            s.add(
+                FindingCitation(
+                    tenant_id=ctx.tenant_id, finding_id=f.id, citation_id=UUID(cid), side="tender", position=pos
+                )
+            )
+        for pos, c in enumerate(bid_cites):
+            s.add(FindingCitation(tenant_id=ctx.tenant_id, finding_id=f.id, citation_id=c.id, side="bid", position=pos))
+    await s.flush()
+
+
+async def _carry_over(s: AsyncSession, ctx: TenantContext, r: CheckRun, checked: set[UUID]) -> None:
+    """和上一次检查按「match_key + 问题类型」对齐（FR-M3-08）：
+    还在的沿用上次的处理状态（上次「已修改」的改回待处理）；消失的留一行「本次已解决」；其余是新增。"""
+    if r.previous_run_id is None:
+        return
+    prev = {
+        (f.match_key, f.type): f
+        for f in (
+            await s.execute(
+                select(Finding).where(Finding.check_run_id == r.previous_run_id, Finding.presence != "resolved")
+            )
+        ).scalars()
+    }
+    now = {
+        (f.match_key, f.type): f
+        for f in (await s.execute(select(Finding).where(Finding.check_run_id == r.id))).scalars()
+    }
+    for key, f in now.items():
+        old = prev.get(key)
+        if old is None:
+            continue
+        f.presence, f.carried_from_id = "carried", old.id
+        f.detail = {**(f.detail or {}), "carried_status": old.status}
+        if old.status not in ("pending", "modified"):  # 上次「已修改」又查出来了：说明没改好，回到待处理
+            f.status, f.resolution_note, f.resolved_by, f.resolved_at = (
+                old.status,
+                old.resolution_note,
+                old.resolved_by,
+                old.resolved_at,
+            )
+    for key, old in prev.items():
+        if key in now:
+            continue
+        gone = old.checklist_item_id is not None and old.checklist_item_id not in checked
+        f = Finding(
+            id=uuid7(),
+            tenant_id=ctx.tenant_id,
+            project_id=r.project_id,
+            check_run_id=r.id,
+            checklist_item_id=old.checklist_item_id,
+            rule_code=old.rule_code,
+            match_key=old.match_key,
+            type=old.type,
+            severity=old.severity,
+            title=old.title,
+            explanation=old.explanation,
+            detector=old.detector,
+            detail={
+                **(old.detail or {}),
+                "carried_status": old.status,
+                **({"resolved_reason": "item_removed"} if gone else {}),
+            },
+            search_scope=old.search_scope,
+            checks_passed=old.checks_passed,
+            presence="resolved",
+            carried_from_id=old.id,
+            status=old.status,
+            resolution_note=old.resolution_note,
+            resolved_by=old.resolved_by,
+            resolved_at=old.resolved_at,
+        )
+        s.add(f)
+        await s.flush()
+        links = (await s.execute(select(FindingCitation).where(FindingCitation.finding_id == old.id))).scalars()
+        for link in links:
+            s.add(
+                FindingCitation(
+                    tenant_id=ctx.tenant_id,
+                    finding_id=f.id,
+                    citation_id=link.citation_id,
+                    side=link.side,
+                    position=link.position,
+                )
+            )
+    await s.flush()
+
+
+async def _counts(s: AsyncSession, run_id: UUID) -> dict[str, int]:
+    rows = list(
+        (
+            await s.execute(
+                select(Finding.severity, Finding.presence, Finding.status).where(Finding.check_run_id == run_id)
+            )
+        ).tuples()
+    )
+    live = [(sev, st) for sev, pr, st in rows if pr != "resolved"]
+    return {
+        "fatal": sum(sev == "fatal" for sev, _ in live),
+        "important": sum(sev == "important" for sev, _ in live),
+        "hint": sum(sev == "hint" for sev, _ in live),
+        "open_fatal": sum(sev == "fatal" and st == "pending" for sev, st in live),
+        "resolved_this_run": sum(pr == "resolved" for _, pr, _ in rows),
+        "new": sum(pr == "new" for _, pr, _ in rows),
+    }
+
+
+def _coverage(
+    r: CheckRun,
+    pages_info: list[dict[str, Any]],
+    items: dict[UUID, dict[str, Any]],
+    results: dict[UUID, match.Result],
+    pending: int,
+    other_lot: int,
+    stats: match.Stats,
+    tokens: Any,
+) -> dict[str, Any]:
+    """检查覆盖范围：报告首页、查标页顶部都读这里（FR-M3-09、m3-plan §2.5）。"""
+    undetermined = sum(x.verdict == "undetermined" for x in results.values())
+    unverified = sum(x.evidence_status == "unverified" for x in results.values())
+    scanned = sum(len(x["pages_without_text"]) for x in pages_info)
+    checked = [f"响应完整性：正式清单 {len(items)} 条，逐条在投标文件里找响应"]
+    not_checked = []
+    if pending:
+        not_checked.append(f"待确认区的 {pending} 条（不参与检查）")
+    if other_lot:
+        not_checked.append(f"不在所投采购包里的 {other_lot} 条")
+    if undetermined:
+        not_checked.append(f"{undetermined} 条没有完成判断（模型漏答或暂时不可用），已列为「请人工核对」")
+    if scanned:
+        not_checked.append(f"{scanned} 页没有文字层（扫描件），这些页上的内容没有检查")
+    not_checked.append("报价形式、形式要件、资质匹配：规则检查还没有上线")
+    if not r.checklist_confirmed:
+        not_checked.append("清单未确认，检查结果以当前清单为准")
+    return {
+        "checked": checked,
+        "not_checked": not_checked,
+        "documents": pages_info,
+        "items": {
+            "formal": len(items) + other_lot,
+            "checked": len(items),
+            "pending_zone": pending,
+            "other_lot": other_lot,
+            "undetermined": undetermined,
+            "evidence_unverified": unverified,
+        },
+        "checklist_confirmed": r.checklist_confirmed,
+        "legal_regime": {"value": r.legal_regime, "rules_verified": r.legal_regime in REGIMES_VERIFIED},
+        "rule_set_version": RULE_SET_VERSION,
+        "prompt_versions": r.prompt_versions,
+        "tokens": {"in": int(tokens[0]), "out": int(tokens[1])},
+        "model_calls": stats.calls,
+        "verify_failures": stats.verify_failures,
+    }
+
+
+async def _finish(ctx: TenantContext, run_id: UUID, status: str, error_class: str | None, code: str | None) -> None:
+    async with tenant_tx(ctx) as s:
+        r = (await s.execute(select(CheckRun).where(CheckRun.id == run_id))).scalar_one()
+        r.status, r.finished_at, r.error_class, r.error_code = status, datetime.now(UTC), error_class, code
+        r.progress = 100 if status == "succeeded" else r.progress
+        p = await projects.get(s, ctx, r.project_id)
+        if r.quota_hold_id:
+            if status == "succeeded":
+                await billing.settle(s, ctx, r.quota_hold_id)
+                p.last_charged_check_at = r.finished_at  # 24 小时免扣只从真正扣了额度的检查算起（P1）
+            else:
+                await billing.release(s, ctx, r.quota_hold_id)  # 失败、超时、取消都不扣点（红线 9）
+        if status == "succeeded":
+            projects.mark_checked(p, r.id, await open_fatal(s, r.id))
+        else:
+            # 失败的批次不留半截结果（重复派发时阶段会重写，但失败后用户看到的应该是空的）
+            await s.execute(delete(Finding).where(Finding.check_run_id == r.id))
+            await s.execute(delete(CheckItemResult).where(CheckItemResult.check_run_id == r.id))
+        if status != "cancelled":
+            s.add(
+                Notification(
+                    id=uuid7(),
+                    tenant_id=ctx.tenant_id,
+                    user_id=r.created_by,
+                    kind="check_succeeded" if status == "succeeded" else "check_failed",
+                    project_id=r.project_id,
+                    payload={"run_id": str(r.id), **({"counts": r.counts} if r.counts else {})},
+                )
+            )
+    await _publish(ctx, r, status, r.stage, 100)

server/src/bid_doctor/check/service.py

@@ -0,0 +1,458 @@
+"""检查批次与问题处理（FR-M3-02、07、08，api-design §4.1、§4.5）。
+
+- 计费（data-model §4.3）：免费版不记流水；专业版 24 小时内复查免扣（记一行 0）；其余在解析额度上预扣 1 次。
+  预览和发起共用同一个计费函数；前端带来的 expected_charge 和服务端当下算的不一致时拒绝，不扣点（API4）。
+- 「预扣 → 建批次 → 入队」在同一个事务里（架构 §3.4）。
+- 只有最近一次检查的问题可以处理；致命问题「确认无误」也要写说明，说明同时进误报反馈（U4）。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from typing import Any
+from uuid import UUID
+
+from fastapi.encoders import jsonable_encoder
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from bid_doctor.audit import service as audit
+from bid_doctor.billing import service as billing
+from bid_doctor.check.models import CheckItemResult, CheckRun, CheckRunDocument, Finding, FindingCitation
+from bid_doctor.checklist import edit as checklist_edit
+from bid_doctor.checklist import service as checklist
+from bid_doctor.core import queue
+from bid_doctor.core.db import TenantContext
+from bid_doctor.core.errors import AppError
+from bid_doctor.core.ids import uuid7
+from bid_doctor.document import citations
+from bid_doctor.document import service as documents
+from bid_doctor.document.models import SourceDocument
+from bid_doctor.identity import service as identity
+from bid_doctor.ops import service as feedback
+from bid_doctor.project import service as projects
+from bid_doctor.project.models import Project
+
+CHECK_TASK = "check_project"
+RULE_SET_VERSION = "2026.11-r0"  # 第 1 步还没有规则；规则集有版本，检查批次记下当时用的版本（架构 §3.7）
+WAIVE_WINDOW = timedelta(hours=24)
+ERROR_MESSAGES = {
+    "bid_text_missing": "投标文件里几乎读不出文字（可能都是扫描件），扫描件识别还没有上线",
+    "file_corrupted": "投标文件无法转换，可能已加密或损坏",
+    "plan_limit_exceeded": "投标文件页数超出套餐上限",
+    "checklist_empty": "正式清单还没有条目",
+    "converter_unavailable": "文档转换服务暂时不可用",
+    "model_unavailable": "模型服务暂时不可用",
+    "internal_error": "系统出错了",
+}
+
+
+@dataclass(frozen=True, slots=True)
+class Billing:
+    mode: str  # charged / waived_recheck / free_plan
+    charge: int
+    waive_until: datetime | None
+
+
+async def billing_mode(s: AsyncSession, ctx: TenantContext, p: Project) -> Billing:
+    _, plan = await billing.current_plan(s, ctx)
+    if plan.code == "free":
+        return Billing("free_plan", 0, None)  # 免费版只限 1 个项目，项目内检查不限次数（01 §4）
+    last = p.last_charged_check_at
+    if last is not None and datetime.now(UTC) - last < WAIVE_WINDOW:
+        return Billing("waived_recheck", 0, last + WAIVE_WINDOW)  # 从上一次扣了额度的检查算起，不重新计时（P1）
+    return Billing("charged", 1, None)
+
+
+async def _running(s: AsyncSession, project_id: UUID) -> CheckRun | None:
+    return (
+        await s.execute(
+            select(CheckRun).where(CheckRun.project_id == project_id, CheckRun.status.in_(("queued", "running")))
+        )
+    ).scalar_one_or_none()
+
+
+async def _tender_ready(s: AsyncSession, project_id: UUID) -> bool:
+    n = (
+        await s.execute(
+            select(func.count()).where(
+                SourceDocument.project_id == project_id,
+                SourceDocument.role == "tender",
+                SourceDocument.status == "ready",
+            )
+        )
+    ).scalar_one()
+    return n > 0
+
+
+async def preview(s: AsyncSession, ctx: TenantContext, project_id: UUID) -> dict[str, Any]:
+    p = await projects.get(s, ctx, project_id)
+    mode = await billing_mode(s, ctx, p)
+    bids = await documents.latest_bid_volumes(s, project_id)
+    blockers: list[str] = []
+    if p.status in ("submitted", "archived"):
+        blockers.append("project_read_only")
+    if not await _tender_ready(s, project_id):
+        blockers.append("no_parsed_tender")
+    elif not await checklist.items(s, project_id, zone="formal"):
+        blockers.append("checklist_empty")
+    if not bids:
+        blockers.append("no_bid_documents")
+    if await _running(s, project_id):
+        blockers.append("job_already_running")
+    if mode.charge and await billing.balance(s, ctx, "parse") < 1:
+        blockers.append("quota_exhausted")
+    states = await checklist.statuses(s, [project_id])
+    return {
+        "can_start": not blockers,
+        "blockers": blockers,
+        "billing": mode.mode,
+        "charge": mode.charge,
+        "charge_unit": "parse",
+        "waive_until": mode.waive_until,
+        "checklist_confirmed": states.get(project_id) == "confirmed",
+        "pending_item_count": (await checklist.pending_counts(s, [project_id])).get(project_id, 0),
+        "bid_documents": [await documents.doc_to_api(s, d) for d in bids],
+        "lot_ids": await projects.bidding_lot_ids(s, project_id),
+    }
+
+
+async def start(
+    s: AsyncSession, ctx: TenantContext, project_id: UUID, document_ids: list[UUID], expected_charge: int
+) -> CheckRun:
+    p = await projects.get(s, ctx, project_id)
+    projects.ensure_editable(p)
+    if running := await _running(s, project_id):
+        raise AppError("job_already_running", job_id=str(running.id))
+    if not await _tender_ready(s, project_id):
+        raise AppError("no_parsed_tender")
+    docs = (
+        (await s.execute(select(SourceDocument).where(SourceDocument.id.in_(document_ids)))).scalars().all()
+        if document_ids
+        else []
+    )
+    if not docs or len(docs) != len(set(document_ids)):
+        raise AppError("no_bid_documents")
+    if any(d.project_id != project_id or d.role != "bid_volume" for d in docs):
+        raise AppError("not_found")
+    mode = await billing_mode(s, ctx, p)
+    if expected_charge != mode.charge:
+        raise AppError("billing_changed", preview=jsonable_encoder(await preview(s, ctx, project_id)))
+    version, confirmed = await checklist_edit.check_snapshot(s, ctx, project_id)
+    run_no = (
+        await s.execute(
+            select(func.coalesce(func.max(CheckRun.run_no), 0) + 1).where(CheckRun.project_id == project_id)
+        )
+    ).scalar_one()
+    run = CheckRun(
+        id=uuid7(),
+        tenant_id=ctx.tenant_id,
+        project_id=project_id,
+        run_no=run_no,
+        checklist_version_id=version.id,
+        checklist_confirmed=confirmed,
+        rule_set_version=RULE_SET_VERSION,
+        legal_regime=p.legal_regime,
+        lot_ids=await projects.bidding_lot_ids(s, project_id),
+        previous_run_id=p.latest_check_run_id,
+        billing=mode.mode,
+        created_by=ctx.user_id,
+    )
+    if mode.mode == "charged":
+        run.quota_hold_id = await billing.hold(
+            s, ctx, unit="parse", ref_type="check_run", ref_id=run.id, project_id=project_id
+        )
+    s.add(run)
+    await s.flush()
+    for d in docs:
+        s.add(CheckRunDocument(tenant_id=ctx.tenant_id, check_run_id=run.id, document_id=d.id))
+    if mode.mode == "waived_recheck":
+        await billing.waive(
+            s, ctx, unit="parse", ref_type="check_run", ref_id=run.id, project_id=project_id, reason="recheck_24h"
+        )
+    run.queue_job_id = await queue.enqueue(
+        s, CHECK_TASK, {"tenant_id": str(ctx.tenant_id), "run_id": str(run.id)}, lock=f"project:{project_id}"
+    )
+    await audit.record(
+        s,
+        ctx,
+        "check_run.start",
+        "check_run",
+        run.id,
+        project_id=project_id,
+        after={"run_no": run_no, "billing": mode.mode, "documents": [str(d.id) for d in docs]},
+    )
+    await s.flush()
+    await s.refresh(run)
+    return run
+
+
+async def get_run(s: AsyncSession, run_id: UUID) -> CheckRun:
+    run = (await s.execute(select(CheckRun).where(CheckRun.id == run_id))).scalar_one_or_none()
+    if run is None:
+        raise AppError("not_found")
+    return run
+
+
+async def request_cancel(s: AsyncSession, run_id: UUID) -> CheckRun:
+    run = await get_run(s, run_id)
+    if run.status not in ("queued", "running"):
+        raise AppError("job_finished")
+    run.cancel_requested_at = datetime.now(UTC)
+    await s.flush()
+    return run
+
+
+async def list_runs(s: AsyncSession, project_id: UUID) -> list[CheckRun]:
+    rows = await s.execute(select(CheckRun).where(CheckRun.project_id == project_id).order_by(CheckRun.run_no.desc()))
+    return list(rows.scalars())
+
+
+async def run_to_api(s: AsyncSession, ctx: TenantContext, run: CheckRun) -> dict[str, Any]:
+    p = await projects.get(s, ctx, run.project_id)
+    doc_ids = list(
+        (await s.execute(select(CheckRunDocument.document_id).where(CheckRunDocument.check_run_id == run.id))).scalars()
+    )
+    docs = (await s.execute(select(SourceDocument).where(SourceDocument.id.in_(doc_ids)))).scalars().all()
+    version_no = (await checklist_edit.version_numbers(s, [run.checklist_version_id])).get(run.checklist_version_id, 0)
+    names = await identity.display_names(s, [run.created_by])
+    error = None
+    if run.status in ("failed", "cancelled") and run.error_code:
+        error = {
+            "code": run.error_code,
+            "message": ERROR_MESSAGES.get(run.error_code, run.error_code),
+            "refunded": run.billing == "charged",
+        }
+    return {
+        "id": run.id,
+        "run_no": run.run_no,
+        "status": run.status,
+        "stage": run.stage,
+        "progress": run.progress,
+        "cancel_requested": run.cancel_requested_at is not None,
+        "error": error,
+        "billing": run.billing,
+        "is_latest": p.latest_check_run_id == run.id,
+        "checklist_version_no": version_no,
+        "checklist_confirmed": run.checklist_confirmed,
+        "rule_set_version": run.rule_set_version,
+        "legal_regime": run.legal_regime,
+        "documents": [await documents.doc_to_api(s, d) for d in docs],
+        "coverage": run.coverage,
+        "counts": run.counts,
+        "created_by": {"id": run.created_by, "name": names.get(run.created_by, "")},
+        "created_at": run.created_at,
+        "finished_at": run.finished_at,
+    }
+
+
+# ------------------------------------------------------------------ 问题
+
+
+async def _items_of(s: AsyncSession, run: CheckRun) -> dict[UUID, dict[str, Any]]:
+    return await checklist_edit.version_items(s, run.checklist_version_id)
+
+
+async def findings(
+    s: AsyncSession, run_id: UUID, *, severity: str | None, status: str | None, presence: str | None
+) -> list[dict[str, Any]]:
+    run = await get_run(s, run_id)
+    q = select(Finding).where(Finding.check_run_id == run_id)
+    if severity:
+        q = q.where(Finding.severity == severity)
+    if status:
+        q = q.where(Finding.status == status)
+    if presence:
+        q = q.where(Finding.presence == presence)
+    rows = list((await s.execute(q)).scalars())
+    return await _to_api_many(s, run, rows)
+
+
+ORDER = {"fatal": 0, "important": 1, "hint": 2}
+
+
+async def _to_api_many(s: AsyncSession, run: CheckRun, rows: list[Finding]) -> list[dict[str, Any]]:
+    """按级别、再按招标原文顺序（清单项的排序键）。"""
+    items = await _items_of(s, run)
+    links = (
+        await s.execute(
+            select(FindingCitation)
+            .where(FindingCitation.finding_id.in_([f.id for f in rows]))
+            .order_by(FindingCitation.position)
+        )
+    ).scalars()
+    by_finding: dict[UUID, list[FindingCitation]] = {}
+    for link in links:
+        by_finding.setdefault(link.finding_id, []).append(link)
+    cites = await citations.to_api_many(s, [link.citation_id for ls in by_finding.values() for link in ls])
+    names = await identity.display_names(s, [f.resolved_by for f in rows if f.resolved_by])
+    sort_keys = await checklist.sort_keys(s, [f.checklist_item_id for f in rows if f.checklist_item_id])
+
+    def api(f: Finding) -> dict[str, Any]:
+        item = items.get(f.checklist_item_id) if f.checklist_item_id else None
+        ls = by_finding.get(f.id, [])
+        return {
+            "id": f.id,
+            "type": f.type,
+            "severity": f.severity,
+            "title": f.title,
+            "explanation": f.explanation,
+            "detector": f.detector,
+            "checklist_item": {
+                "id": f.checklist_item_id,
+                "title": item.get("title"),
+                "importance": item.get("importance"),
+                "categories": item.get("categories"),
+            }
+            if item
+            else None,
+            "rule": {"code": f.rule_code, "description": ""} if f.rule_code and not item else None,
+            "detail": f.detail,
+            "tender_citations": [cites[x.citation_id] for x in ls if x.side == "tender" and x.citation_id in cites],
+            "bid_citations": [cites[x.citation_id] for x in ls if x.side == "bid" and x.citation_id in cites],
+            "search_scope": f.search_scope,
+            "presence": f.presence,
+            "carried_status": (f.detail or {}).get("carried_status"),
+            "status": f.status,
+            "resolution_note": f.resolution_note,
+            "resolved_by": {"id": f.resolved_by, "name": names.get(f.resolved_by, "")} if f.resolved_by else None,
+            "resolved_at": f.resolved_at,
+        }
+
+    rows.sort(key=lambda f: (ORDER[f.severity], sort_keys.get(f.checklist_item_id or UUID(int=0), "~"), f.title))
+    return [api(f) for f in rows]
+
+
+async def _get_finding(s: AsyncSession, finding_id: UUID) -> Finding:
+    f = (await s.execute(select(Finding).where(Finding.id == finding_id))).scalar_one_or_none()
+    if f is None:
+        raise AppError("not_found")
+    return f
+
+
+async def _editable_run(s: AsyncSession, ctx: TenantContext, run_id: UUID) -> tuple[CheckRun, Project]:
+    run = await get_run(s, run_id)
+    p = await projects.get(s, ctx, run.project_id)
+    projects.ensure_editable(p)
+    if p.latest_check_run_id != run.id:
+        raise AppError("run_superseded")
+    return run, p
+
+
+async def open_fatal(s: AsyncSession, run_id: UUID) -> int:
+    return (
+        await s.execute(
+            select(func.count()).where(
+                Finding.check_run_id == run_id,
+                Finding.severity == "fatal",
+                Finding.status == "pending",
+                Finding.presence != "resolved",
+            )
+        )
+    ).scalar_one()
+
+
+def _require_note(severity: str, status: str, note: str | None) -> None:
+    if (status == "ignored" or (severity == "fatal" and status == "confirmed_ok")) and not (note or "").strip():
+        raise AppError("note_required")
+
+
+async def _apply(s: AsyncSession, ctx: TenantContext, f: Finding, p: Project, status: str, note: str | None) -> None:
+    before = {"status": f.status, "note": f.resolution_note}
+    done = status != "pending"
+    f.status = status
+    f.resolution_note = ((note or "").strip() or None) if done else None
+    f.resolved_by, f.resolved_at = (ctx.user_id, datetime.now(UTC)) if done else (None, None)
+    if f.severity == "fatal" and status == "confirmed_ok":
+        await feedback.record(
+            s,
+            ctx,
+            project_id=p.id,
+            object_type="finding",
+            object_id=f.id,
+            kind="false_positive",
+            source="fatal_confirm",
+            note=f.resolution_note,
+        )
+    await audit.record(
+        s,
+        ctx,
+        "finding.resolve" if done else "finding.reopen",
+        "finding",
+        f.id,
+        project_id=p.id,
+        before=before,
+        after={"status": f.status, "note": f.resolution_note},
+    )
+
+
+async def resolve(
+    s: AsyncSession, ctx: TenantContext, finding_id: UUID, status: str, note: str | None
+) -> dict[str, Any]:
+    f = await _get_finding(s, finding_id)
+    run, p = await _editable_run(s, ctx, f.check_run_id)
+    if f.presence == "resolved":
+        raise AppError("finding_resolved_this_run")
+    _require_note(f.severity, status, note)
+    await _apply(s, ctx, f, p, status, note)
+    await s.flush()
+    p.open_fatal_count = await open_fatal(s, run.id)
+    await s.flush()
+    (out,) = await _to_api_many(s, run, [f])
+    return {"finding": out, "project": projects.readiness(p)}
+
+
+async def reopen(s: AsyncSession, ctx: TenantContext, finding_id: UUID) -> dict[str, Any]:
+    f = await _get_finding(s, finding_id)
+    run, p = await _editable_run(s, ctx, f.check_run_id)
+    if f.presence == "resolved":
+        raise AppError("finding_resolved_this_run")
+    await _apply(s, ctx, f, p, "pending", None)
+    await s.flush()
+    p.open_fatal_count = await open_fatal(s, run.id)
+    await s.flush()
+    (out,) = await _to_api_many(s, run, [f])
+    return {"finding": out, "project": projects.readiness(p)}
+
+
+async def bulk_resolve(
+    s: AsyncSession, ctx: TenantContext, run_id: UUID, finding_ids: list[UUID], status: str, note: str | None
+) -> dict[str, Any]:
+    """只限「提示」级（FR-M3-07）；混进了别的级别就整批拒绝，返回不符合的问题编号。"""
+    run, p = await _editable_run(s, ctx, run_id)
+    rows = list((await s.execute(select(Finding).where(Finding.id.in_(finding_ids)))).scalars())
+    if len(rows) != len(set(finding_ids)) or any(f.check_run_id != run.id for f in rows):
+        raise AppError("not_found")
+    bad = [str(f.id) for f in rows if f.severity != "hint" or f.presence == "resolved"]
+    if bad:
+        raise AppError("bulk_only_hint", finding_ids=bad)
+    _require_note("hint", status, note)
+    for f in rows:
+        await _apply(s, ctx, f, p, status, note)
+    await s.flush()
+    p.open_fatal_count = await open_fatal(s, run.id)
+    await s.flush()
+    return {"items": await _to_api_many(s, run, rows), "project": projects.readiness(p)}
+
+
+async def item_results(s: AsyncSession, run_id: UUID) -> list[dict[str, Any]]:
+    """每条清单项的检查结论（C9）：「已响应」也有出处，可以复核、可以反馈判断有误。"""
+    run = await get_run(s, run_id)
+    items = await _items_of(s, run)
+    rows = list((await s.execute(select(CheckItemResult).where(CheckItemResult.check_run_id == run_id))).scalars())
+    cites = await citations.to_api_many(s, [c for r in rows for c in r.bid_citation_ids])
+    return [
+        {
+            "checklist_item_id": r.checklist_item_id,
+            "title": (items.get(r.checklist_item_id) or {}).get("title"),
+            "verdict": r.verdict,
+            "layer": r.layer,
+            "evidence_status": r.evidence_status,
+            "reason_code": r.reason_code,
+            "bid_citations": [cites[c] for c in r.bid_citation_ids if c in cites],
+            "detail": r.detail,
+            "reused": r.reused_from_run_id is not None,
+        }
+        for r in rows
+    ]

server/src/bid_doctor/checklist/edit.py

@@ -374,6 +374,75 @@ def _snapshot(r: dict[str, Any]) -> dict[str, Any]:
     return {**{k: _json(r[k]) for k in keep}, "citation_ids": [str(c["id"]) for c in r["citations"]]}
 
 
+async def _freeze(
+    s: AsyncSession, ctx: TenantContext, cl: Checklist, rows: list[dict[str, Any]], kind: str
+) -> ChecklistVersion:
+    """把正式清单存成不可变版本；内容和已有版本一样就复用（确认：上一份确认版；检查：任何一份）。"""
+    payloads = {str(r["id"]): _snapshot(r) for r in rows}
+    digest = sha256(json.dumps(payloads, ensure_ascii=False, sort_keys=True, default=str))
+    q = select(ChecklistVersion).where(ChecklistVersion.checklist_id == cl.id)
+    if kind == "confirmed":
+        q = q.where(ChecklistVersion.kind == "confirmed")
+    last = (await s.execute(q.order_by(ChecklistVersion.version_no.desc()).limit(1))).scalar_one_or_none()
+    if kind == "check":
+        last = (await s.execute(q.where(ChecklistVersion.content_hash == digest).limit(1))).scalar_one_or_none() or last
+    if last is not None and last.content_hash == digest:
+        return last
+    cl.last_version_no += 1
+    version = ChecklistVersion(
+        id=uuid7(),
+        tenant_id=ctx.tenant_id,
+        checklist_id=cl.id,
+        version_no=cl.last_version_no,
+        kind=kind,
+        content_hash=digest,
+        item_count=len(payloads),
+        created_by=ctx.user_id,
+    )
+    s.add(version)
+    await s.flush()
+    for item_id, payload in payloads.items():
+        s.add(
+            ChecklistVersionItem(tenant_id=ctx.tenant_id, version_id=version.id, item_id=UUID(item_id), payload=payload)
+        )
+    await s.flush()
+    return version
+
+
+async def check_snapshot(s: AsyncSession, ctx: TenantContext, project_id: UUID) -> tuple[ChecklistVersion, bool]:
+    """检查开始前的清单快照（data-model §5.4）：已确认且没再改过的，用确认版；否则生成 kind = check 的快照。
+    返回 (版本, 是否已确认)。清单没确认也能检查（G7）。"""
+    cl = await service.get_or_create(s, ctx, project_id)
+    rows = await service.items(s, project_id, zone="formal")
+    if not rows:
+        raise AppError("checklist_empty")
+    if cl.status == "confirmed" and cl.confirmed_version_id:
+        confirmed = (
+            await s.execute(select(ChecklistVersion).where(ChecklistVersion.id == cl.confirmed_version_id))
+        ).scalar_one()
+        payloads = {str(r["id"]): _snapshot(r) for r in rows}
+        if confirmed.content_hash == sha256(json.dumps(payloads, ensure_ascii=False, sort_keys=True, default=str)):
+            return confirmed, True
+    return await _freeze(s, ctx, cl, rows, "check"), False
+
+
+async def version_items(s: AsyncSession, version_id: UUID) -> dict[UUID, dict[str, Any]]:
+    """快照里每条清单项当时的值和出处编号。"""
+    rows = await s.execute(
+        select(ChecklistVersionItem.item_id, ChecklistVersionItem.payload).where(
+            ChecklistVersionItem.version_id == version_id
+        )
+    )
+    return dict(rows.tuples().all())
+
+
+async def version_numbers(s: AsyncSession, version_ids: list[UUID]) -> dict[UUID, int]:
+    rows = await s.execute(
+        select(ChecklistVersion.id, ChecklistVersion.version_no).where(ChecklistVersion.id.in_(version_ids))
+    )
+    return dict(rows.tuples().all())
+
+
 async def confirm(s: AsyncSession, ctx: TenantContext, project_id: UUID) -> dict[str, Any]:
     """确认清单：把正式清单里未删除的条目存成不可变版本；内容和上一版一样就复用上一版（data-model §5）。
     待确认区的条目不进版本，也不参与检查。"""
@@ -382,38 +451,7 @@ async def confirm(s: AsyncSession, ctx: TenantContext, project_id: UUID) -> dict
     rows = await service.items(s, project_id, zone="formal")
     if not rows:
         raise AppError("checklist_empty")
-    payloads = {str(r["id"]): _snapshot(r) for r in rows}
-    digest = sha256(json.dumps(payloads, ensure_ascii=False, sort_keys=True, default=str))
-    last = (
-        await s.execute(
-            select(ChecklistVersion)
-            .where(ChecklistVersion.checklist_id == cl.id, ChecklistVersion.kind == "confirmed")
-            .order_by(ChecklistVersion.version_no.desc())
-            .limit(1)
-        )
-    ).scalar_one_or_none()
-    if last is not None and last.content_hash == digest:
-        version = last
-    else:
-        cl.last_version_no += 1
-        version = ChecklistVersion(
-            id=uuid7(),
-            tenant_id=ctx.tenant_id,
-            checklist_id=cl.id,
-            version_no=cl.last_version_no,
-            kind="confirmed",
-            content_hash=digest,
-            item_count=len(payloads),
-            created_by=ctx.user_id,
-        )
-        s.add(version)
-        await s.flush()
-        for item_id, payload in payloads.items():
-            s.add(
-                ChecklistVersionItem(
-                    tenant_id=ctx.tenant_id, version_id=version.id, item_id=UUID(item_id), payload=payload
-                )
-            )
+    version = await _freeze(s, ctx, cl, rows, "confirmed")
     if cl.status == "confirmed" and cl.confirmed_version_id == version.id:
         return await service.overview(s, project_id)  # 重复点确认：什么都不变
     cl.status, cl.confirmed_version_id = "confirmed", version.id

server/src/bid_doctor/checklist/service.py

@@ -332,6 +332,14 @@ async def statuses(s: AsyncSession, project_ids: list[UUID]) -> dict[UUID, str]:
     return dict(rows.tuples().all())
 
 
+async def sort_keys(s: AsyncSession, item_ids: list[UUID]) -> dict[UUID, str]:
+    """清单项在原文里的顺序（问题列表按原文顺序排，FR-M3-03）。"""
+    if not item_ids:
+        return {}
+    rows = await s.execute(select(ChecklistItem.id, ChecklistItem.sort_key).where(ChecklistItem.id.in_(item_ids)))
+    return {i: k or "~" for i, k in rows.tuples()}
+
+
 async def pending_counts(s: AsyncSession, project_ids: list[UUID]) -> dict[UUID, int]:
     if not project_ids:
         return {}

server/src/bid_doctor/core/columns.py

@@ -20,4 +20,5 @@ DbTs = Annotated[
 Bytes = Annotated[bytes, mapped_column(LargeBinary)]
 Json = Annotated[Any, mapped_column(JSONB)]
 TextArray = Annotated[list[str], mapped_column(ARRAY(Text), server_default=FetchedValue())]
+UuidArray = Annotated[list[UUID], mapped_column(ARRAY(Uuid), server_default=FetchedValue())]
 Inet = Annotated[str, mapped_column(INET)]

server/src/bid_doctor/core/errors.py

@@ -31,6 +31,10 @@ CODES: dict[str, tuple[int, str]] = {
     "upload_incomplete": (409, "文件还没有上传完"),
     "item_not_pending": (409, "这一条已经不在待确认区"),
     "checklist_empty": (409, "正式清单还没有条目"),
+    "no_parsed_tender": (409, "还没有解析成功的招标文件"),
+    "no_bid_documents": (409, "还没有上传投标文件"),
+    "run_superseded": (409, "这是历史检查，只能查看"),
+    "finding_resolved_this_run": (409, "这个问题本次已解决，不需要处理"),
     "precondition_failed": (412, "内容已被修改，请刷新后再改"),
     "validation_failed": (422, "填写的内容有误"),
     "sms_code_invalid": (422, "验证码不正确或已过期"),
@@ -40,6 +44,9 @@ CODES: dict[str, tuple[int, str]] = {
     "bid_file_platform_encrypted": (422, "这是电子投标平台加密后的文件，请上传加密前的 Word 或 PDF"),
     "upload_checksum_mismatch": (422, "文件校验不一致，请重新上传"),
     "citation_required": (422, "请补上出处，或者声明「人工确认、无出处」"),
+    "note_required": (422, "请填写说明"),
+    "bulk_only_hint": (422, "批量处理只限「提示」级问题"),
+    "bid_text_missing": (422, "投标文件里几乎读不出文字（可能都是扫描件），扫描件识别还没有上线"),
     "sentence_not_found": (422, "选中的句子不在这份文件里，或不在同一页"),
     "ocr_required": (422, "这份文件大部分页面没有文字（可能是扫描件），扫描件识别还没有上线"),
     "rate_limited": (429, "操作太频繁，请稍后再试"),

server/src/bid_doctor/core/valkey.py

@@ -8,16 +8,27 @@ from redis.asyncio import Redis
 
 from bid_doctor.core.config import get_settings
 
+TIMEOUT_S = 2  # Valkey 挂掉时发布事件、限流都要很快失败，不能卡住 worker 和请求（SK1）
+STREAM_MAX_CONNECTIONS = 500  # 每个 SSE 连接占一个；超出时新连接报错，前端退回轮询
+
 
 @lru_cache
 def valkey() -> Redis:
-    return Redis.from_url(get_settings().valkey_url, decode_responses=True)
+    return Redis.from_url(
+        get_settings().valkey_url, decode_responses=True, socket_timeout=TIMEOUT_S, socket_connect_timeout=TIMEOUT_S
+    )
 
 
 @lru_cache
 def stream_valkey() -> Redis:
     """事件流的长阻塞读（XREAD BLOCK）用单独的连接池：读超时要比阻塞时间长，否则连接被当成超时断开。"""
-    return Redis.from_url(get_settings().valkey_url, decode_responses=True, socket_timeout=60)
+    return Redis.from_url(
+        get_settings().valkey_url,
+        decode_responses=True,
+        socket_timeout=60,
+        socket_connect_timeout=TIMEOUT_S,
+        max_connections=STREAM_MAX_CONNECTIONS,
+    )
 
 
 async def hit_limit(key: str, limit: int, window_s: int) -> int | None:

server/src/bid_doctor/document/api.py

@@ -33,6 +33,8 @@ class DocumentIn(BaseModel):
     upload_id: UUID
     role: Literal["tender", "addendum", "clarification", "bid_volume"]
     volume_name: str | None = Field(default=None, max_length=40)
+    volume_kind: Literal["whole", "qualification", "commercial", "technical", "price", "other"] | None = None
+    replaces_document_id: UUID | None = None
 
 
 class ParseIn(BaseModel):
@@ -84,7 +86,16 @@ async def list_documents(project_id: UUID, ctx: Ctx) -> dict[str, Any]:
 @router.post("/projects/{project_id}/documents", status_code=201)
 async def create_document(project_id: UUID, body: DocumentIn, ctx: Ctx) -> dict[str, Any]:
     async with tenant_tx(ctx) as s:
-        doc = await service.create(s, ctx, project_id, body.upload_id, body.role, body.volume_name)
+        doc = await service.create(
+            s,
+            ctx,
+            project_id,
+            body.upload_id,
+            body.role,
+            body.volume_name,
+            volume_kind=body.volume_kind,
+            replaces=body.replaces_document_id,
+        )
         return await service.doc_to_api(s, doc)
 
 

server/src/bid_doctor/document/citations.py

@@ -16,7 +16,21 @@ from bid_doctor.document.pipeline.text_layer import Sentence
 LABELS = {"tender": "招标文件", "addendum": "补遗", "clarification": "澄清", "bid_volume": "投标文件"}
 
 
-def document_label(role: str, seq_no: int) -> str:
+VOLUME_KINDS = {
+    "whole": "投标文件",
+    "qualification": "资格证明分册",
+    "commercial": "商务分册",
+    "technical": "技术分册",
+    "price": "报价分册",
+    "other": "投标分册",
+}
+
+
+def document_label(role: str, seq_no: int, volume_name: str | None = None, volume_kind: str | None = None) -> str:
+    """「招标文件」「补遗 1」；投标分册是「技术分册 v2」，用户填了分册名时用分册名。"""
+    if role == "bid_volume":
+        name = volume_name or VOLUME_KINDS.get(volume_kind or "whole", "投标文件")
+        return f"{name} v{seq_no}" if seq_no > 1 else name
     return f"{LABELS[role]} {seq_no}" if role in ("addendum", "clarification") else LABELS.get(role, role)
 
 
@@ -68,8 +82,10 @@ async def to_api_many(s: AsyncSession, ids: list[UUID]) -> dict[UUID, dict[str,
     if not ids:
         return {}
     rows = await s.execute(
-        select(Citation, SourceDocument.role, SourceDocument.seq_no)
+        select(
+            Citation, SourceDocument.role, SourceDocument.seq_no, SourceDocument.volume_name, SourceDocument.volume_kind
+        )
         .join(SourceDocument, SourceDocument.id == Citation.document_id)
         .where(Citation.id.in_(ids))
     )
-    return {c.id: to_api(c, document_label(role, seq)) for c, role, seq in rows.tuples()}
+    return {c.id: to_api(c, document_label(role, seq, name, kind)) for c, role, seq, name, kind in rows.tuples()}

server/src/bid_doctor/document/formats.py

@@ -12,6 +12,8 @@ from bid_doctor.core.errors import AppError
 
 OLE_MAGIC = bytes.fromhex("D0CF11E0A1B11AE1")
 EXTENSIONS = {"pdf": {"pdf"}, "docx": {"docx", "doc", "wps"}, "doc": {"doc", "wps"}, "ofd": {"ofd"}}
+# 电子投标平台的专有 / 加密格式（FR-M3-01、C8）：只认有样本或公开资料确认过的写法，见过新的再补
+PLATFORM_FORMATS = {"kedt": "苏采云（江苏政府采购）"}
 
 
 @dataclass(frozen=True, slots=True)
@@ -23,6 +25,10 @@ class Detected:
 
 def detect(data: bytes, filename: str) -> Detected:
     ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
+    if platform := PLATFORM_FORMATS.get(ext):
+        raise AppError(
+            "bid_file_platform_encrypted", f"这是{platform}平台的文件，请上传加密前的 Word 或 PDF", platform=platform
+        )
     if data.startswith(b"%PDF"):
         fmt, pages = "pdf", _pdf_pages(data)
     elif data.startswith(b"PK\x03\x04"):
@@ -52,13 +58,17 @@ def _pdf_pages(data: bytes) -> int:
 
 def _zip_kind(data: bytes) -> str:
     try:
-        names = set(zipfile.ZipFile(io.BytesIO(data)).namelist())
+        infos = zipfile.ZipFile(io.BytesIO(data)).infolist()
     except zipfile.BadZipFile as e:
         raise AppError("file_corrupted") from e
+    names = {i.filename for i in infos}
     if "word/document.xml" in names:
         return "docx"
     if "OFD.xml" in names:
         return "ofd"
     if "EncryptionInfo" in names or "EncryptedPackage" in names:
         raise AppError("file_encrypted")
+    if infos and all(i.flag_bits & 0x1 for i in infos if not i.is_dir()):
+        # 条目全部加密的压缩包：电子投标客户端导出的加密投标文件常是这种（07 F3）
+        raise AppError("bid_file_platform_encrypted")
     raise AppError("file_format_unsupported", "支持 PDF、Word（DOCX / DOC）和 WPS 文件")

server/src/bid_doctor/document/models.py

@@ -15,6 +15,7 @@ class SourceDocument(Base):
     role: Mapped[str]
     seq_no: Mapped[int]
     volume_name: Mapped[str | None]
+    volume_kind: Mapped[str | None]
     lineage_id: Mapped[UuidCol]
     original_filename: Mapped[str]
     detected_format: Mapped[str]

server/src/bid_doctor/document/pipeline/normalize.py

@@ -0,0 +1,98 @@
+"""解析流水线的前几个阶段：格式核对 → 转换 → 文字层 → OCR 补全（架构 §3.5）。
+
+解析任务（招标文件）和检查批次（投标文件，m3-plan §2 prepare_bid）共用。阶段的进度、取消由调用方的 on_stage 负责。
+TODO(S5)：OCR 还没接，没有文字层的页 text_source = "empty"。
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Awaitable, Callable
+from uuid import UUID
+
+from sqlalchemy import delete, select
+
+from bid_doctor.billing import service as billing
+from bid_doctor.core import keys
+from bid_doctor.core.crypto import open_object, seal_object, sha256
+from bid_doctor.core.db import TenantContext, tenant_tx
+from bid_doctor.core.errors import AppError
+from bid_doctor.core.storage import storage
+from bid_doctor.document.models import DocumentPage, SourceDocument
+from bid_doctor.document.pipeline import convert, text_layer
+from bid_doctor.document.pipeline.text_layer import Page
+from bid_doctor.document.service import doc_prefix
+
+# text_layer v2：页眉页脚带单独成句，句子编号和 v1 不同（同一份规范化 PDF 在 v1、v2 下的编号不能混用）
+PROFILE = {"converter": "gotenberg-8", "fonts": "gotenberg-default", "text_layer": "v4", "sentence_split": "s3-v4"}
+OCR_REQUIRED_RATIO = 0.5  # 超过一半的页没有文字层：招标文件按「需要 OCR」失败
+
+OnStage = Callable[[str], Awaitable[None]]
+
+
+async def pages(
+    ctx: TenantContext, document_id: UUID, on_stage: OnStage, *, reuse: bool, require_text: bool
+) -> list[Page]:
+    """返回文档的文字层。reuse=True 时，已有同一版本的文字层就直接读出来（投标分册每个版本只解析一次）。
+
+    require_text：没有文字层的页太多时报 ocr_required（招标文件）。
+    投标文件不报，扫描页由检查按「请人工核对」处理（C10）。"""
+    async with tenant_tx(ctx) as s:
+        doc = (await s.execute(select(SourceDocument).where(SourceDocument.id == document_id))).scalar_one()
+        dk = await keys.data_key(s, ctx, doc.project_id)
+    if reuse and doc.text_layer_key and (doc.pipeline_profile or {}).get("text_layer") == PROFILE["text_layer"]:
+        raw = open_object(dk, storage().get(doc.text_layer_key), doc.text_layer_key)
+        return [Page.from_json(p) for p in json.loads(raw)]
+
+    await on_stage("detect")  # 上传完成时已识别过格式；这里只核对原件没被替换
+    original = open_object(dk, storage().get(doc.original_key), doc.original_key)
+    if sha256(original) != doc.original_sha256:
+        raise AppError("file_corrupted")
+
+    await on_stage("convert")
+    if doc.normalized_key is None:
+        pdf = await convert.to_pdf(original, doc.detected_format)
+        key = f"{doc_prefix(doc)}/normalized.pdf"
+        storage().put(key, seal_object(dk, pdf, key))
+        async with tenant_tx(ctx) as s:
+            d = (await s.execute(select(SourceDocument).where(SourceDocument.id == doc.id))).scalar_one()
+            d.normalized_key, d.normalized_sha256, d.status = key, sha256(pdf), "normalizing"
+            d.pipeline_profile = PROFILE
+        doc.normalized_key, doc.normalized_sha256 = key, sha256(pdf)
+    else:
+        pdf = open_object(dk, storage().get(doc.normalized_key), doc.normalized_key)
+
+    await on_stage("text_layer")
+    result = text_layer.extract(str(doc.id), pdf)
+    text_key = f"{doc_prefix(doc)}/text.json"
+    storage().put(
+        text_key, seal_object(dk, json.dumps([p.to_json() for p in result], ensure_ascii=False).encode(), text_key)
+    )
+    async with tenant_tx(ctx) as s:
+        await s.execute(delete(DocumentPage).where(DocumentPage.document_id == doc.id))  # 阶段幂等：先清残留
+        for p in result:
+            s.add(
+                DocumentPage(
+                    tenant_id=ctx.tenant_id,
+                    document_id=doc.id,
+                    page_no=p.page_no,
+                    width_pt=p.width,
+                    height_pt=p.height,
+                    rotation=p.rotation,
+                    text_source=p.text_source,
+                    char_count=p.char_count,
+                    sentence_count=len(p.sentences),
+                )
+            )
+        d = (await s.execute(select(SourceDocument).where(SourceDocument.id == doc.id))).scalar_one()
+        d.text_layer_key, d.page_count, d.status = text_key, len(result), "ready"
+        d.pipeline_profile = PROFILE
+        _, plan = await billing.current_plan(s, ctx)
+    if len(result) > plan.limits["max_pages"]:
+        raise AppError("plan_limit_exceeded")
+
+    await on_stage("ocr")
+    empty = sum(p.text_source == "empty" for p in result)
+    if require_text and result and empty / len(result) > OCR_REQUIRED_RATIO:
+        raise AppError("ocr_required")
+    return result

server/src/bid_doctor/document/pipeline/runner.py

@@ -8,28 +8,24 @@ TODO(结构识别)：章节、表格、分包识别还没做，现在只抽实
 
 from __future__ import annotations
 
-import json
 import logging
 from datetime import UTC, datetime
 from typing import Any
 from uuid import UUID
 
 import httpx
-from sqlalchemy import delete, select
+from sqlalchemy import select
 
 from bid_doctor.ai.providers.base import TransientError
 from bid_doctor.billing import service as billing
 from bid_doctor.checklist import extract
 from bid_doctor.checklist import service as checklist
-from bid_doctor.core import events, keys
-from bid_doctor.core.crypto import open_object, seal_object, sha256
+from bid_doctor.core import events
 from bid_doctor.core.db import TenantContext, tenant_tx
 from bid_doctor.core.errors import AppError
 from bid_doctor.core.ids import uuid7
-from bid_doctor.core.storage import storage
-from bid_doctor.document.models import DocumentPage, ParseJob, SourceDocument
-from bid_doctor.document.pipeline import convert, text_layer
-from bid_doctor.document.service import doc_prefix
+from bid_doctor.document.models import ParseJob, SourceDocument
+from bid_doctor.document.pipeline import normalize, text_layer
 from bid_doctor.notification.models import Notification
 from bid_doctor.project import extracted as project_extracted
 from bid_doctor.project import service as projects
@@ -45,8 +41,6 @@ PROGRESS = {
     "verify": 92,
     "merge": 95,
 }
-# text_layer v2：页眉页脚带单独成句，句子编号和 v1 不同（同一份规范化 PDF 在 v1、v2 下的编号不能混用）
-PROFILE = {"converter": "gotenberg-8", "fonts": "gotenberg-default", "text_layer": "v4", "sentence_split": "s3-v4"}
 
 
 class Cancelled(Exception):
@@ -113,58 +107,11 @@ async def _stages(ctx: TenantContext, job_id: UUID) -> dict[str, Any]:
     async with tenant_tx(ctx) as s:
         job = (await s.execute(select(ParseJob).where(ParseJob.id == job_id))).scalar_one()
         doc = (await s.execute(select(SourceDocument).where(SourceDocument.id == job.document_id))).scalar_one()
-        dk = await keys.data_key(s, ctx, doc.project_id)
 
-    await _stage(ctx, job_id, "detect")  # 上传完成时已识别过格式；这里只核对原件没被替换
-    original = open_object(dk, storage().get(doc.original_key), doc.original_key)
-    if sha256(original) != doc.original_sha256:
-        raise AppError("file_corrupted")
+    async def on_stage(stage: str) -> None:
+        await _stage(ctx, job_id, stage)
 
-    await _stage(ctx, job_id, "convert")
-    if doc.normalized_key is None:
-        pdf = await convert.to_pdf(original, doc.detected_format)
-        key = f"{doc_prefix(doc)}/normalized.pdf"
-        storage().put(key, seal_object(dk, pdf, key))
-        async with tenant_tx(ctx) as s:
-            d = (await s.execute(select(SourceDocument).where(SourceDocument.id == doc.id))).scalar_one()
-            d.normalized_key, d.normalized_sha256, d.status = key, sha256(pdf), "normalizing"
-            d.pipeline_profile = PROFILE
-        doc.normalized_key, doc.normalized_sha256 = key, sha256(pdf)
-    else:
-        pdf = open_object(dk, storage().get(doc.normalized_key), doc.normalized_key)
-
-    await _stage(ctx, job_id, "text_layer")
-    pages = text_layer.extract(str(doc.id), pdf)
-    text_key = f"{doc_prefix(doc)}/text.json"
-    storage().put(
-        text_key, seal_object(dk, json.dumps([p.to_json() for p in pages], ensure_ascii=False).encode(), text_key)
-    )
-    async with tenant_tx(ctx) as s:
-        await s.execute(delete(DocumentPage).where(DocumentPage.document_id == doc.id))  # 阶段幂等：先清残留
-        for p in pages:
-            s.add(
-                DocumentPage(
-                    tenant_id=ctx.tenant_id,
-                    document_id=doc.id,
-                    page_no=p.page_no,
-                    width_pt=p.width,
-                    height_pt=p.height,
-                    rotation=p.rotation,
-                    text_source=p.text_source,
-                    char_count=p.char_count,
-                    sentence_count=len(p.sentences),
-                )
-            )
-        d = (await s.execute(select(SourceDocument).where(SourceDocument.id == doc.id))).scalar_one()
-        d.text_layer_key, d.page_count, d.status = text_key, len(pages), "ready"
-        _, plan = await billing.current_plan(s, ctx)
-    if len(pages) > plan.limits["max_pages"]:
-        raise AppError("plan_limit_exceeded")
-
-    await _stage(ctx, job_id, "ocr")
-    empty = sum(p.text_source == "empty" for p in pages)
-    if pages and empty / len(pages) > 0.5:
-        raise AppError("ocr_required")
+    pages = await normalize.pages(ctx, doc.id, on_stage, reuse=False, require_text=True)
 
     await _stage(ctx, job_id, "structure")
     body = text_layer.body_sentences(pages)

server/src/bid_doctor/document/service.py

@@ -33,22 +33,42 @@ def doc_prefix(doc: SourceDocument) -> str:
 
 
 async def create(
-    s: AsyncSession, ctx: TenantContext, project_id: UUID, upload_id: UUID, role: str, volume_name: str | None
+    s: AsyncSession,
+    ctx: TenantContext,
+    project_id: UUID,
+    upload_id: UUID,
+    role: str,
+    volume_name: str | None,
+    *,
+    volume_kind: str | None = None,
+    replaces: UUID | None = None,
 ) -> SourceDocument:
+    """挂到项目上。投标分册重新上传时带 replaces，生成同一分册的新版本（FR-M3-01）。"""
     p = await projects.get(s, ctx, project_id)
     projects.ensure_editable(p)
+    doc_id = uuid7()
+    lineage, seq = doc_id, 0
+    if role == "bid_volume":
+        volume_kind = volume_kind or "whole"
+        seq = 1
+        if replaces is not None:
+            prev = await get(s, replaces)
+            if prev.project_id != project_id or prev.role != "bid_volume":
+                raise AppError("not_found")
+            lineage = prev.lineage_id
+            seq = await _next_version(s, lineage)
+            volume_name = volume_name or prev.volume_name
+            volume_kind = prev.volume_kind if volume_kind == "whole" and prev.volume_kind else volume_kind
+    elif role != "tender":
+        seq = (
+            await s.execute(
+                select(func.coalesce(func.max(SourceDocument.seq_no), 0) + 1).where(
+                    SourceDocument.project_id == project_id, SourceDocument.role == role
+                )
+            )
+        ).scalar_one()
     meta, data = await uploads.take(ctx, upload_id)
     assert meta.detected_format is not None
-    seq = (
-        await s.execute(
-            select(func.coalesce(func.max(SourceDocument.seq_no), -1) + 1).where(
-                SourceDocument.project_id == project_id, SourceDocument.role == role
-            )
-        )
-    ).scalar_one()
-    if role == "tender":
-        seq = 0
-    doc_id = uuid7()
     doc = SourceDocument(
         id=doc_id,
         tenant_id=ctx.tenant_id,
@@ -56,7 +76,8 @@ async def create(
         role=role,
         seq_no=seq,
         volume_name=volume_name,
-        lineage_id=doc_id,
+        volume_kind=volume_kind if role == "bid_volume" else None,
+        lineage_id=lineage,
         original_filename=meta.filename,
         detected_format=meta.detected_format,
         size_bytes=meta.size_bytes,
@@ -76,7 +97,7 @@ async def create(
         "source_document",
         doc.id,
         project_id=project_id,
-        after={"role": role, "filename": meta.filename},
+        after={"role": role, "filename": meta.filename, "version": seq if role == "bid_volume" else None},
     )
     await s.flush()
     await s.refresh(doc)
@@ -84,6 +105,34 @@ async def create(
     return doc
 
 
+async def _next_version(s: AsyncSession, lineage_id: UUID) -> int:
+    return (
+        await s.execute(
+            select(func.coalesce(func.max(SourceDocument.seq_no), 0) + 1).where(SourceDocument.lineage_id == lineage_id)
+        )
+    ).scalar_one()
+
+
+async def latest_bid_volumes(s: AsyncSession, project_id: UUID) -> list[SourceDocument]:
+    """每个投标分册的最新版本（检查默认用它们，CheckPreview.bid_documents）。"""
+    docs = (
+        (
+            await s.execute(
+                select(SourceDocument)
+                .where(SourceDocument.project_id == project_id, SourceDocument.role == "bid_volume")
+                .order_by(SourceDocument.created_at)
+            )
+        )
+        .scalars()
+        .all()
+    )
+    latest: dict[UUID, SourceDocument] = {}
+    for d in docs:
+        if d.lineage_id not in latest or d.seq_no > latest[d.lineage_id].seq_no:
+            latest[d.lineage_id] = d
+    return list(latest.values())
+
+
 async def get(s: AsyncSession, document_id: UUID) -> SourceDocument:
     doc = (await s.execute(select(SourceDocument).where(SourceDocument.id == document_id))).scalar_one_or_none()
     if doc is None:
@@ -189,14 +238,24 @@ async def doc_to_api(s: AsyncSession, doc: SourceDocument) -> dict[str, Any]:
             select(ParseJob).where(ParseJob.document_id == doc.id).order_by(ParseJob.attempt_no.desc()).limit(1)
         )
     ).scalar_one_or_none()
+    newer = None
+    if doc.role == "bid_volume":
+        newer = (
+            await s.execute(
+                select(SourceDocument.id)
+                .where(SourceDocument.lineage_id == doc.lineage_id, SourceDocument.seq_no > doc.seq_no)
+                .limit(1)
+            )
+        ).scalar_one_or_none()
     return {
         "id": doc.id,
         "role": doc.role,
-        "label": document_label(doc.role, doc.seq_no),
+        "label": document_label(doc.role, doc.seq_no, doc.volume_name, doc.volume_kind),
         "seq_no": doc.seq_no,
         "volume_name": doc.volume_name,
+        "volume_kind": doc.volume_kind,
         "lineage_id": doc.lineage_id,
-        "is_latest_version": True,
+        "is_latest_version": newer is None,
         "original_filename": doc.original_filename,
         "detected_format": doc.detected_format,
         "size_bytes": doc.size_bytes,
@@ -210,18 +269,18 @@ async def doc_to_api(s: AsyncSession, doc: SourceDocument) -> dict[str, Any]:
 
 
 async def summaries(s: AsyncSession, project_ids: list[UUID]) -> dict[UUID, dict[str, Any]]:
-    """工作台用：每个项目的招标文件数、正在跑的解析。"""
-    out: dict[UUID, dict[str, Any]] = {pid: {"tenders": 0, "job": None} for pid in project_ids}
+    """工作台用：每个项目的招标文件数、投标分册数、正在跑的解析。"""
+    out: dict[UUID, dict[str, Any]] = {pid: {"tenders": 0, "bids": 0, "job": None} for pid in project_ids}
     if not project_ids:
         return out
-    for pid, n in (
+    for pid, role, n in (
         await s.execute(
-            select(SourceDocument.project_id, func.count())
-            .where(SourceDocument.project_id.in_(project_ids), SourceDocument.role == "tender")
-            .group_by(SourceDocument.project_id)
+            select(SourceDocument.project_id, SourceDocument.role, func.count())
+            .where(SourceDocument.project_id.in_(project_ids), SourceDocument.role.in_(("tender", "bid_volume")))
+            .group_by(SourceDocument.project_id, SourceDocument.role)
         )
     ).tuples():
-        out[pid]["tenders"] = n
+        out[pid]["tenders" if role == "tender" else "bids"] = n
     for job in (
         await s.execute(
             select(ParseJob).where(ParseJob.project_id.in_(project_ids), ParseJob.status.in_(("queued", "running")))

server/src/bid_doctor/ops/api.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from typing import Literal
+from uuid import UUID
+
+from fastapi import APIRouter
+from pydantic import BaseModel, Field
+
+from bid_doctor.core.db import tenant_tx
+from bid_doctor.identity.deps import Ctx
+from bid_doctor.ops import service
+from bid_doctor.project import service as projects
+
+router = APIRouter(tags=["notifications"])
+
+
+class FeedbackIn(BaseModel):
+    project_id: UUID
+    object_type: Literal["finding", "checklist_item", "checklist"]
+    object_id: UUID | None = None
+    kind: Literal["false_positive", "false_negative"]
+    note: str | None = Field(default=None, max_length=1000)
+
+
+@router.post("/feedback", status_code=201)
+async def create_feedback(body: FeedbackIn, ctx: Ctx) -> None:
+    async with tenant_tx(ctx) as s:
+        await projects.get(s, ctx, body.project_id)
+        await service.record(
+            s,
+            ctx,
+            project_id=body.project_id,
+            object_type=body.object_type,
+            object_id=body.object_id,
+            kind=body.kind,
+            source="user",
+            note=body.note,
+        )

server/src/bid_doctor/ops/models.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from sqlalchemy import FetchedValue
+from sqlalchemy.orm import Mapped, mapped_column
+
+from bid_doctor.core.columns import PK, DbTs, Ts, UuidCol
+from bid_doctor.core.db import Base
+
+
+class Feedback(Base):
+    """误报 / 漏报（J5）：用户写，运营后台处理。致命问题「确认无误」的说明也自动进来（U4）。"""
+
+    __tablename__ = "feedback"
+    id: Mapped[PK]
+    tenant_id: Mapped[UuidCol]
+    project_id: Mapped[UuidCol]
+    object_type: Mapped[str]
+    object_id: Mapped[UuidCol | None]
+    kind: Mapped[str]
+    source: Mapped[str]
+    note: Mapped[str | None]
+    status: Mapped[str] = mapped_column(server_default=FetchedValue())
+    handled_by_ops: Mapped[UuidCol | None]
+    handled_at: Mapped[Ts | None]
+    created_by: Mapped[UuidCol]
+    created_at: Mapped[DbTs]
+    updated_at: Mapped[DbTs]

server/src/bid_doctor/ops/service.py

@@ -0,0 +1,38 @@
+"""误报 / 漏报反馈（FR-M3-07、J5）。运营后台的处理接口以后另做（FR-M6-09）。"""
+
+from __future__ import annotations
+
+from uuid import UUID
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from bid_doctor.core.db import TenantContext
+from bid_doctor.core.ids import uuid7
+from bid_doctor.ops.models import Feedback
+
+
+async def record(
+    s: AsyncSession,
+    ctx: TenantContext,
+    *,
+    project_id: UUID,
+    object_type: str,
+    object_id: UUID | None,
+    kind: str,
+    source: str,
+    note: str | None,
+) -> Feedback:
+    fb = Feedback(
+        id=uuid7(),
+        tenant_id=ctx.tenant_id,
+        project_id=project_id,
+        object_type=object_type,
+        object_id=object_id,
+        kind=kind,
+        source=source,
+        note=note,
+        created_by=ctx.user_id,
+    )
+    s.add(fb)
+    await s.flush()
+    return fb

server/src/bid_doctor/project/models.py

@@ -23,6 +23,7 @@ class Project(Base):
     status: Mapped[str] = mapped_column(server_default=FetchedValue())
     latest_check_run_id: Mapped[UuidCol | None]
     open_fatal_count: Mapped[int] = mapped_column(server_default=FetchedValue())
+    last_charged_check_at: Mapped[Ts | None]
     created_by: Mapped[UuidCol]
     created_at: Mapped[DbTs]
     updated_at: Mapped[DbTs]

server/src/bid_doctor/project/service.py

@@ -133,3 +133,27 @@ async def lot_labels(s: AsyncSession, project_id: UUID) -> dict[UUID, str | None
     """项目的分包编号（给清单项展示「采购包 N」、校验 lot_id 用）；默认包没有编号，表示各包通用。"""
     rows = await s.execute(select(Lot.id, Lot.lot_no, Lot.is_default).where(Lot.project_id == project_id))
     return {i: None if default else no for i, no, default in rows.tuples()}
+
+
+async def bidding_lot_ids(s: AsyncSession, project_id: UUID) -> list[UUID]:
+    """所投的包（不含默认包）；没有分包时为空，表示整个项目。"""
+    rows = await s.execute(
+        select(Lot.id).where(Lot.project_id == project_id, Lot.is_bidding, Lot.is_default.is_(False))
+    )
+    return list(rows.scalars())
+
+
+def mark_checked(p: Project, run_id: UUID, open_fatal: int) -> None:
+    """完成过至少一次检查：跟进中 / 编制中 → 已检查（FR-M6-02）；就绪标记随最近一次检查更新（data-model §5.6）。"""
+    if p.status in ("tracking", "drafting"):
+        p.status = "checked"
+    p.latest_check_run_id, p.open_fatal_count = run_id, open_fatal
+
+
+def readiness(p: Project) -> dict[str, Any]:
+    """ProjectReadiness：就绪状态只在「已检查」时有意义（G7）；未就绪也不阻止导出和递交（红线 4）。"""
+    return {
+        "status": p.status,
+        "open_fatal_count": p.open_fatal_count,
+        "ready": None if p.status != "checked" else p.open_fatal_count == 0,
+    }

server/src/bid_doctor/project/workbench.py

@@ -36,7 +36,6 @@ async def detail(s: AsyncSession, ctx: TenantContext, project_id: UUID) -> dict[
         "procurement_method": p.procurement_method,
         "fields": await projects.fields(s, p.id),
         "key_dates": await checklist.key_dates(s, p.id),
-        "latest_check_run_id": p.latest_check_run_id,
         "created_by": {"id": p.created_by, "name": ctx.actor_name},
         "created_at": p.created_at,
     }
@@ -57,18 +56,23 @@ def _row(
         "bidding_lots": lots,
         "pending_item_count": pending,
         "active_job": docs["job"],
-        "next_action": _next_action(p.status, docs["tenders"], docs["job"], checklist_status),
+        "next_action": _next_action(p, docs, checklist_status),
+        "latest_check_run_id": p.latest_check_run_id,
         "updated_at": p.updated_at,
     }
 
 
-def _next_action(status: str, tenders: int, job: Any, checklist_status: str | None) -> str:
-    """工作台「下一步」（api-design §6）。骨架阶段只覆盖解读这一段，查标阶段补齐。"""
-    if job:
+def _next_action(p: Project, docs: dict[str, Any], checklist_status: str | None) -> str:
+    """工作台「下一步」（api-design §6），和原型工作台一致。"""
+    if docs["job"]:
         return "wait_parse"
-    if status == "tracking":
-        return "upload_tender" if tenders == 0 else "wait_parse"
-    if status == "drafting":
+    if p.status == "tracking":
+        return "upload_tender" if docs["tenders"] == 0 else "wait_parse"
+    if p.status == "drafting":
         # 清单确认之前先确认清单（原型工作台的「下一步」）；待确认区有条目也算没确认完
-        return "upload_bid" if checklist_status == "confirmed" else "confirm_checklist"
+        if checklist_status != "confirmed":
+            return "confirm_checklist"
+        return "run_check" if docs["bids"] else "upload_bid"
+    if p.status == "checked":
+        return "handle_fatal" if p.open_fatal_count else "export_report"
     return "none"

server/src/bid_doctor/worker.py

@@ -28,6 +28,13 @@ async def parse_document(tenant_id: str, job_id: str) -> None:
     await runner.run(UUID(tenant_id), UUID(job_id))
 
 
+@app.task(name="check_project", retry=False)
+async def check_project(tenant_id: str, run_id: str) -> None:
+    from bid_doctor.check import runner
+
+    await runner.run(UUID(tenant_id), UUID(run_id))
+
+
 async def run_worker() -> None:
     async with app.open_async():
         await app.run_worker_async(concurrency=4, install_signal_handlers=True)