feat(server): 参考 BidEngine 补五项：原因码、截断重跑、项目信息逐项表态、占位符、提示词规则对应校验

- 待确认记原因码（extract.FAILURES，存在 ai_values.failures），解析结果按原因计数
  （verify_failures）；前端待确认条目显示具体原因
- 相邻分块重叠约 800 字；模型输出被截断或不是合法 JSON 时对半重跑（最多 3 层），
  拆到底仍不完整记 output_incomplete。修复：以前截断时整块条目被当成「没有结果」悄悄丢掉
- 项目基本信息提示词 v2：12 个字段逐项表态 found / ambiguous / not_found，跨分块合并为
  found / absent / needs_review（info_fields）
- 「____元」「××」「（ ）」「待填」等占位符不算取值
- 提示词 substantive v4，qualification / scoring / schedule / tech_param v3：每条禁止事项
  对应一个原因码，对照表在 prompts/README.md
- 顾问（Fable）深读参考项目的结论补进 ref-bid-engine-analysis.md §5；架构 §13 SK12
- 待确认：国动 4 → 2，S02 13 → 3，S06 57 → 55

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

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

@@ -12,6 +12,8 @@ class Completion:
     tokens_in: int | None = None
     tokens_out: int | None = None
     output_mode: str = "json_mode"
+    # 输出不完整：被截断（finish_reason=length）或不是合法 JSON。调用方要拆小输入重跑，不能当成「没有结果」
+    incomplete: bool = False
 
 
 class TransientError(Exception):

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

@@ -135,15 +135,25 @@ def _tech_param(lines: list[tuple[str, str, str, str]]) -> dict[str, Any]:
     return {"items": items}
 
 
+INFO_KEYS = (
+    "name", "code", "purchaser", "agency", "procurement_method", "legal_regime", "budget_total", "ceiling_total",
+    "eval_method", "accept_consortium", "accept_imported", "sme_only",
+)  # fmt: skip
+
+
 def _project_info(lines: list[tuple[str, str, str, str]]) -> dict[str, Any]:
+    """v2 的格式：每个字段都表态 found / not_found。"""
     fields: dict[str, Any] = {}
     for sid, _, p, _ in lines:
         for key, label in (("name", "项目名称"), ("code", "项目编号"), ("purchaser", "采购人")):
             if key not in fields and (m := re.search(label + r"[：:]\s*([^，。；]+)", p)):
-                fields[key] = {"value": m.group(1).strip(), "sid": sid, "quote": p[:30]}
+                fields[key] = {"status": "found", "value": m.group(1).strip(), "sid": sid, "quote": p[:30]}
         if "budget_total" not in fields and (m := re.search(r"预算[^0-9]{0,8}(\d+(?:\.\d+)?)\s*(万元|元)", p)):
-            fields["budget_total"] = {"value": m.group(1), "unit": m.group(2), "sid": sid, "quote": p[:30]}
-    return {"fields": fields, "lots": []}
+            fields["budget_total"] = {
+                "status": "found", "value": m.group(1), "unit": m.group(2), "sid": sid, "quote": p[:30],
+            }  # fmt: skip
+    none = {"status": "not_found", "value": None, "unit": None, "sid": None, "quote": None}
+    return {"fields": {k: fields.get(k, none) for k in INFO_KEYS}, "lots": []}
 
 
 TASKS: dict[str, Callable[[list[tuple[str, str, str, str]]], dict[str, Any]]] = {

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

@@ -41,16 +41,20 @@ class OpenAICompatProvider:
             raise TransientError("server", f"模型接口 {r.status_code}")
         r.raise_for_status()
         payload = r.json()
-        raw = payload["choices"][0]["message"]["content"]
+        choice = payload["choices"][0]
+        raw = choice["message"]["content"] or ""
         usage = payload.get("usage") or {}
+        incomplete = choice.get("finish_reason") == "length"
         try:
             data = json.loads(raw)
         except json.JSONDecodeError:
-            data = {"items": []}  # 内容质量问题：不算任务失败，相关条目由规则兜底进待确认区（架构 §3.6）
+            # 以前当成「没有结果」：输出被截断时，这一整块的条目会悄悄丢掉（2026-11-03 审计发现，输出正好 8192 token）
+            data, incomplete = {}, True
         return Completion(
             data=data,
             raw=raw,
             model=payload.get("model", model),
             tokens_in=usage.get("prompt_tokens"),
             tokens_out=usage.get("completion_tokens"),
+            incomplete=incomplete,
         )

server/src/bid_doctor/ai/service.py

@@ -20,6 +20,10 @@ from bid_doctor.core.ids import uuid7
 RETRIES = 2
 
 
+class OutputIncomplete(Exception):
+    """模型输出被截断或不是合法 JSON：调用方把输入拆小重跑（extract.run），不能当成「没有结果」。"""
+
+
 def provider() -> Provider:
     return OpenAICompatProvider() if get_settings().ai_provider == "openai_compat" else FakeProvider()
 
@@ -61,10 +65,12 @@ async def run_json(
                 tokens_out=result.tokens_out if result else None,
                 latency_ms=int((time.monotonic() - started) * 1000),
                 output_mode=result.output_mode if result else None,
-                error_class=error.error_class if error else None,
+                error_class=error.error_class if error else "bad_output" if result and result.incomplete else None,
             )
         )
     if error:
         raise error
     assert result is not None
+    if result.incomplete:
+        raise OutputIncomplete(task)
     return result.data

server/src/bid_doctor/checklist/extract.py

@@ -15,6 +15,7 @@ from __future__ import annotations
 import asyncio
 import itertools
 import re
+from collections import Counter
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -29,6 +30,9 @@ from bid_doctor.document.pipeline import structure
 from bid_doctor.document.pipeline.text_layer import Sentence, norm
 
 CHUNK_CHARS = 6000
+OVERLAP_CHARS = 800  # 相邻分块重叠的字数：跨块的条款在后一块里还能看到开头
+OVERLAP_SENTENCES = 8
+SPLIT_DEPTH = 3  # 模型输出被截断时把分块对半重跑，最多拆这么多层
 CONCURRENCY = 8  # 每个解析任务同时调用模型的分块数；TODO(路由表)：按「供应商 × 租户」统一限流
 INFO_CHUNKS = 3  # 项目基本信息只看文件开头几个分块（招标公告、须知前附表）
 CN_TZ = ZoneInfo("Asia/Shanghai")
@@ -54,6 +58,29 @@ CN_DIGIT = {"零": 0, "〇": 0, "一": 1, "二": 2, "两": 2, "三": 3, "四": 4
 CN_UNIT = {"十": 10, "百": 100, "千": 1000}
 END_PUNCT = "。；;，,、：:！!？?"
 OUR_MARKUP = re.compile(r"</?(?:iu|u|i)>|〔(?:续下页|上级：[^〕]*)〕")
+SID = re.compile(r"^p\d+s\d+$")
+# 招标文件里待投标人填写的空白：「____元」「××」「（  ）」「待填」——不是取值
+PLACEHOLDER = re.compile(r"_{2,}|＿{2,}|×{2,}|[xX]{2,}|\*{2,}|（\s*）|\(\s*\)|待填|TBD", re.IGNORECASE)
+
+# 没通过校验的原因码（记在清单项 ai_values.failures 里，解析结果按原因计数）。每一条对应提示词里的一条规则，
+# 见 prompts/README.md「提示词规则和服务端校验」
+FAILURES = {
+    "evidence_missing": "没有给证据",
+    "sid_format": "句子编号格式不对",
+    "sid_unknown": "句子编号不在本分块里",
+    "quote_too_short": "引文太短（不足 8 个字，短句没有整句复制）",
+    "quote_not_original": "引文不是这句话的原文",
+    "basis_unmatched": "认定为实质性条款的依据在原文里找不到",
+    "value_unverified": "取值在原文里找不到",
+    "points_unverified": "分值在原文里找不到",
+    "date_unverified": "日期时间在原文里找不到",
+    "amount_unverified": "金额在原文里找不到",
+    "placeholder_value": "取值是待填写的空白",
+    "marker_uncovered": "标注了实质性条款，但模型没有抽到",
+    "not_found_with_evidence": "说没找到却带了证据",
+    "ambiguous": "模型认为有歧义",
+    "missing_answer": "模型漏答了这一项",
+}
 
 
 @dataclass(slots=True)
@@ -72,6 +99,7 @@ class Extracted:
     evidence_required: str | None = None
     lot_no: str | None = None
     ai_values: dict[str, Any] = field(default_factory=dict)
+    failures: list[str] = field(default_factory=list)  # 没通过校验的原因码（FAILURES）
 
 
 @dataclass(slots=True)
@@ -92,6 +120,8 @@ class InfoField:
 class ProjectInfo:
     fields: dict[str, InfoField] = field(default_factory=dict)
     lots: list[dict[str, Any]] = field(default_factory=list)
+    # 每个字段的结论：found / absent（看过的分块都说没有）/ needs_review（有歧义、漏答或证据没通过），带原因码
+    status: dict[str, dict[str, Any]] = field(default_factory=dict)
 
 
 @dataclass(slots=True)
@@ -184,7 +214,36 @@ def chunks(sentences: list[Sentence], continuations: dict[str, Sentence]) -> lis
         prev = s
     if cur:
         out.append(cur)
-    return out
+    return [out[0], *(_tail(a, continuations) + b for a, b in itertools.pairwise(out))] if out else out
+
+
+def _tail(chunk: list[Sentence], continuations: dict[str, Sentence]) -> list[Sentence]:
+    """上一块的结尾几句，放到下一块开头作重叠：跨块的条款（列表的前几条、表格的前几行）在后一块里也看得到。
+    重叠部分抽到的结果按首句合并，不会重复。跨页接续的两句不拆开。"""
+    out: list[Sentence] = []
+    size = 0
+    for s in reversed(chunk):
+        line = len(s.marked) + 12
+        if out and (size + line > OVERLAP_CHARS or len(out) >= OVERLAP_SENTENCES):
+            break
+        out.insert(0, s)
+        size += line
+    while len(out) < len(chunk) and chunk[len(chunk) - len(out) - 1].key in continuations:
+        out.insert(0, chunk[len(chunk) - len(out) - 1])
+    return out if len(out) < len(chunk) else []
+
+
+def is_placeholder(value: Any) -> bool:
+    return isinstance(value, str) and (not value.strip() or PLACEHOLDER.search(value) is not None)
+
+
+def value_issue(value: Any, sents: list[Sentence], code: str = "value_unverified") -> str | None:
+    """结构化取值的问题：待填写的空白 → placeholder_value；原文里找不到 → code；没问题返回 None。"""
+    if value in (None, "", []):
+        return None
+    if is_placeholder(str(value)):
+        return "placeholder_value"
+    return None if check_value(value, sents) else code
 
 
 def _render(chunk: list[Sentence], doc: Doc, marks: bool = False) -> str:
@@ -205,19 +264,21 @@ def _render(chunk: list[Sentence], doc: Doc, marks: bool = False) -> str:
 def check_evidence(
     ev: dict[str, Any], by_id: dict[str, Sentence], nxt: dict[str, Sentence] | None = None
 ) -> tuple[str | None, list[Sentence]]:
-    """第一、二道：编号在本分块里；引文是原句的子串，且不少于 8 个字。返回 (没通过的那一道, 引文所在的句子)。
+    """第一、二道：编号在本分块里；引文是原句的子串，且不少于 8 个字。返回 (没通过的原因码, 引文所在的句子)。
     比对前去掉我们自己加进输入的格式标记和〔…〕附注（不是原文，模型偶尔会照抄）；原文文字本身不做任何放宽。
 
     表格单元格、跨页会把一句话切成两段（S3），模型摘引文时常把两段连起来。所以引文可以从这一句接续到紧跟着的
     下一句（同一分块里的下一句，或跨页接续句），但必须从这一句开始，两句都作为出处。
     """
-    sid, quote = str(ev.get("sid", "")), norm(OUR_MARKUP.sub("", str(ev.get("quote", ""))))
+    sid, quote = str(ev.get("sid", "")).strip(), norm(OUR_MARKUP.sub("", str(ev.get("quote", ""))))
+    if not SID.match(sid):
+        return "sid_format", []
     if sid not in by_id:
-        return "id", []
+        return "sid_unknown", []
     s = by_id[sid]
     text = norm(s.text)
     if not quote or len(quote) < min(8, len(text.rstrip(END_PUNCT))):  # 不足 8 个字的短句要整句复制，句末标点不算
-        return "quote", []
+        return "quote_too_short" if quote and quote in text else "quote_not_original", []
     if quote in text:
         return None, [s]
     following = (nxt or {}).get(s.key)
@@ -226,22 +287,25 @@ def check_evidence(
         at = joined.find(quote)
         if 0 <= at < len(text) and at + len(quote) > len(text):
             return None, [s, following]
-    return "quote", []
+    return "quote_not_original", []
 
 
 def _evidence(
     it: dict[str, Any], by_id: dict[str, Sentence], nxt: dict[str, Sentence] | None = None
-) -> tuple[list[Sentence], list[str], bool]:
-    """返回 (通过前两道的句子, 通过了哪几道, 是否全部证据都通过)。"""
+) -> tuple[list[Sentence], list[str], bool, list[str]]:
+    """返回 (通过前两道的句子, 通过了哪几道, 是否全部证据都通过, 没通过的原因码)。"""
     evs = [e for e in it.get("evidence") or [] if isinstance(e, dict)]
+    if not evs:
+        return [], [], False, ["evidence_missing"]
     results = [check_evidence(e, by_id, nxt) for e in evs]
     ok: list[Sentence] = []
     for fail, sents in results:
         if fail is None:
             ok += [x for x in sents if x not in ok]
-    fails = [f for f, _ in results]
-    passed = [c for c in ("id", "quote") if ok and c not in fails]
-    return ok, passed, bool(ok) and all(f is None for f in fails)
+    fails = [f for f, _ in results if f]
+    id_failed = any(f.startswith("sid_") for f in fails)
+    passed = [c for c, bad in (("id", id_failed), ("quote", bool(fails))) if ok and not bad]
+    return ok, passed, bool(ok) and not fails, [*dict.fromkeys(fails)]
 
 
 def _next_map(chunk: list[Sentence], doc: Doc) -> dict[str, Sentence]:
@@ -385,8 +449,10 @@ def _item(
     value_ok: bool,
     raw: dict[str, Any],
     doc: Doc,
+    value_fail: str | None = None,
     **kw: Any,
 ) -> Extracted:
+    """value_fail：第三道没通过的原因码；value_ok 为 False 时必须给。"""
     if value_ok and sents:
         passed = [*passed, "basis"]
     accepted = ok and value_ok
@@ -408,6 +474,7 @@ def _item(
         importance=importance,
         importance_basis=imp_basis,
         ai_values={task: raw},
+        failures=[] if value_ok or not sents else [value_fail or "value_unverified"],
         **kw,
     )
 
@@ -426,6 +493,7 @@ def _substantive(
         bool(sents) and check_basis(basis, sents, chunk, doc),
         it,
         doc,
+        "basis_unmatched",
         basis=basis,
         spec={"invalidity": {"source": "substantive_marker" if basis != "explicit_invalid" else "tender_text"}},
     )
@@ -463,6 +531,7 @@ def _scoring(
     points = it.get("points")
     unit = "" if not points or str(points).endswith(("%", "分")) else " 分"  # 价格扣除是百分比，不加「分」
     title = f"{it.get('factor') or '评分项'}" + (f"（{points}{unit}）" if points else "")
+    issue = value_issue(points, ctx, "points_unverified")
     return _item(
         "scoring",
         ["scoring"],
@@ -470,9 +539,10 @@ def _scoring(
         sents,
         passed,
         ok,
-        check_value(points, ctx),
+        issue is None,
         it,
         doc,
+        issue,
         spec={
             "scoring": {
                 "factor": it.get("factor"),
@@ -499,6 +569,7 @@ def _schedule(
         else None
     )
     spec = {"date_kind": kind, "at": at, "deposit_amount": deposit, "submission_mode": mode}
+    issue = "date_unverified" if not at_ok else value_issue(deposit, ctx, "amount_unverified")
     return _item(
         "schedule",
         ["schedule"],
@@ -506,9 +577,10 @@ def _schedule(
         sents,
         passed,
         ok,
-        at_ok and check_value(deposit, ctx),
+        issue is None,
         it,
         doc,
+        issue,
         spec={"schedule": {k: v for k, v in spec.items() if v is not None}},
     )
 
@@ -533,22 +605,23 @@ def _tech_param(
         sents,
         passed,
         ok,
-        check_value(it.get("value"), ctx),
+        (issue := value_issue(it.get("value"), ctx)) is None,
         it,
         doc,
+        issue,
         spec={"tech_param": {k: v for k, v in spec.items() if v is not None}},
         lot_no=it.get("lot_no"),
     )
 
 
 TASKS = [
-    Task("extract.substantive", 3, _substantive, marks=True),
-    Task("extract.qualification", 2, _qualification),
-    Task("extract.scoring", 2, _scoring),
-    Task("extract.schedule", 2, _schedule),
-    Task("extract.tech_param", 2, _tech_param),
+    Task("extract.substantive", 4, _substantive, marks=True),
+    Task("extract.qualification", 3, _qualification),
+    Task("extract.scoring", 3, _scoring),
+    Task("extract.schedule", 3, _schedule),
+    Task("extract.tech_param", 3, _tech_param),
 ]
-INFO_TASK = ("extract.project_info", 1)
+INFO_TASK = ("extract.project_info", 2)
 INFO_KEYS = (
     "name",
     "code",
@@ -573,54 +646,84 @@ def prompt_versions() -> dict[str, str]:
 
 
 Progress = Callable[[float], Awaitable[None]]
+Parts = list[tuple[list[Sentence], dict[str, Any]]]  # 一次调用（可能被对半拆开）的 (分块, 模型输出) 列表
 
 
 async def run(
     ctx: TenantContext, job_id: UUID, doc: Doc, progress: Progress | None = None
-) -> tuple[list[Extracted], ProjectInfo]:
+) -> tuple[list[Extracted], ProjectInfo, dict[str, Any]]:
+    """返回 (清单项, 项目基本信息, 统计)。统计里有对半重跑的次数、拆到底仍不完整的分块数和各原因码的条数。"""
     parts = chunks(doc.sentences, doc.continuations)
     gate = asyncio.Semaphore(CONCURRENCY)
     total = len(parts) * len(TASKS) + min(INFO_CHUNKS, len(parts))
     done = 0
+    stats: Counter[str] = Counter()
 
-    async def call(task: str, version: int, chunk: list[Sentence], marks: bool = False) -> dict[str, Any]:
-        nonlocal done
-        async with gate:
-            data = await ai.run_json(
-                ctx,
-                job_type="parse_job",
-                job_id=job_id,
-                task=task,
-                prompt_id=task,
-                version=version,
-                user=_render(chunk, doc, marks),
+    async def ask(task: str, version: int, chunk: list[Sentence], marks: bool, depth: int = 0) -> Parts:
+        """一个分块调用一次；输出被截断或不是合法 JSON 时对半拆开重跑（参考项目 BidEngine 的做法），
+        拆到底还不完整就记下来，不能当成「这一块没有结果」。"""
+        try:
+            async with gate:
+                data = await ai.run_json(
+                    ctx,
+                    job_type="parse_job",
+                    job_id=job_id,
+                    task=task,
+                    prompt_id=task,
+                    version=version,
+                    user=_render(chunk, doc, marks),
+                )
+            return [(chunk, data)]
+        except ai.OutputIncomplete:
+            if len(chunk) < 2 or depth >= SPLIT_DEPTH:
+                stats["output_incomplete"] += 1
+                return [(chunk, {})]
+            stats["chunks_split"] += 1
+            mid = len(chunk) // 2
+            while mid < len(chunk) - 1 and chunk[mid - 1].key in doc.continuations:
+                mid += 1  # 跨页接续的两句不拆开
+            a, b = await asyncio.gather(
+                ask(task, version, chunk[:mid], marks, depth + 1), ask(task, version, chunk[mid:], marks, depth + 1)
             )
+            return a + b
+
+    async def call(task: str, version: int, chunk: list[Sentence], marks: bool = False) -> Parts:
+        nonlocal done
+        got = await ask(task, version, chunk, marks)
         done += 1
         if progress:
             await progress(done / total)
-        return data
+        return got
 
-    jobs = [(t, c, call(t.name, t.prompt_version, c, t.marks)) for t in TASKS for c in parts]
-    info_jobs = [(c, call(*INFO_TASK, c)) for c in parts[:INFO_CHUNKS]]
-    results = await asyncio.gather(*(j for _, _, j in jobs), *(j for _, j in info_jobs))
+    jobs = [(t, call(t.name, t.prompt_version, c, t.marks)) for t in TASKS for c in parts]
+    info_jobs = [call(*INFO_TASK, c) for c in parts[:INFO_CHUNKS]]
+    results = await asyncio.gather(*(j for _, j in jobs), *info_jobs)
 
     out: list[Extracted] = []
-    for (task, chunk, _), data in zip(jobs, results[: len(jobs)], strict=True):
-        by_id = {_sid(s): s for s in chunk}
-        nxt = _next_map(chunk, doc)
-        for it in data.get("items") or []:
-            if not isinstance(it, dict):
-                continue
-            sents, passed, ok = _evidence(it, by_id, nxt)
-            sents = without_headings(sents, doc)
-            if sents and all(s.key in doc.note_keys or is_marker_talk(s) for s in sents):
-                continue  # 只是「标注规则说明」或表头，不是条款（提示词要求不输出，这里再兜一道）
-            if (e := task.build(it, sents, passed, ok, chunk, doc)) is not None:
-                out += split_listed(e, doc)
-    info = _project_info([(c, d) for (c, _), d in zip(info_jobs, results[len(jobs) :], strict=True)])
+    for (task, _), got in zip(jobs, results[: len(jobs)], strict=True):
+        for chunk, data in got:
+            by_id = {_sid(s): s for s in chunk}
+            nxt = _next_map(chunk, doc)
+            for it in data.get("items") or []:
+                if not isinstance(it, dict):
+                    continue
+                sents, passed, ok, fails = _evidence(it, by_id, nxt)
+                sents = without_headings(sents, doc)
+                if sents and all(s.key in doc.note_keys or is_marker_talk(s) for s in sents):
+                    continue  # 只是「标注规则说明」或表头，不是条款（提示词要求不输出，这里再兜一道）
+                if (e := task.build(it, sents, passed, ok, chunk, doc)) is not None:
+                    e.failures = [*dict.fromkeys([*fails, *e.failures])] if e.zone == "pending" else []
+                    out += split_listed(e, doc)
+    info = _project_info([pair for got in results[len(jobs) :] for pair in got])
     items = merge_items(out)
     items += fallback(items, doc)
-    return items, info
+    for e in items:
+        if e.failures:
+            e.ai_values = {**e.ai_values, "failures": e.failures}
+            stats.update(f"fail:{f}" for f in e.failures)
+    for f in info.status.values():
+        stats[f"info:{f['status']}"] += 1
+    return items, info, dict(stats)
 
 
 def without_headings(sents: list[Sentence], doc: Doc) -> list[Sentence]:
@@ -657,6 +760,7 @@ def split_listed(e: Extracted, doc: Doc) -> list[Extracted]:
                 sentences=sents,
                 zone="formal" if ok else "pending",
                 pending_reason=None if ok else "verify_failed",
+                failures=[] if ok else (e.failures or ["basis_unmatched"]),
                 checks_passed=e.checks_passed,
                 importance="substantive",
                 importance_basis="marker" if basis != "explicit_invalid" else "tender_text",
@@ -693,13 +797,22 @@ def merge_items(items: list[Extracted]) -> list[Extracted]:
             head.ai_values = {**e.ai_values, **head.ai_values}
             head.evidence_required = head.evidence_required or e.evidence_required
             head.basis = head.basis or e.basis
+            if head.zone == "pending":
+                head.failures = [*dict.fromkeys([*head.failures, *e.failures])]
             head.checks_passed = [*dict.fromkeys([*head.checks_passed, *e.checks_passed])]
             keys = {s.key for s in head.sentences}
             head.sentences = sorted(
                 [*head.sentences, *(s for s in e.sentences if s.key not in keys)], key=lambda s: (s.page_no, s.seq)
             )
         merged.append(head)
-    return merged + loose
+    # 没有出处的结果没法按首句对齐：同一任务、同一概括的只留一条（相邻分块重叠时会抽到两次）
+    seen: set[tuple[str, str]] = set()
+    unique = []
+    for e in loose:
+        if (e.task, e.summary) not in seen:
+            seen.add((e.task, e.summary))
+            unique.append(e)
+    return merged + unique
 
 
 def fallback(items: list[Extracted], doc: Doc) -> list[Extracted]:
@@ -737,6 +850,7 @@ def fallback(items: list[Extracted], doc: Doc) -> list[Extracted]:
                 sentences=sents,
                 zone="pending",
                 pending_reason="marker_uncovered",
+                failures=["marker_uncovered"],
                 checks_passed=["id", "quote", "basis"],
                 importance="substantive",
                 importance_basis="marker" if marker != "explicit_invalid" else "tender_text",
@@ -747,33 +861,69 @@ def fallback(items: list[Extracted], doc: Doc) -> list[Extracted]:
     return out
 
 
-def _project_info(results: list[tuple[list[Sentence], dict[str, Any]]]) -> ProjectInfo:
-    """项目基本信息：每个字段取第一个通过校验的值（文件开头的公告最权威）。"""
+def _project_info(results: Parts) -> ProjectInfo:
+    """项目基本信息（参考 BidEngine 的固定字段车道）：每个字段在每个分块里都要表态
+    found / ambiguous / not_found；取第一个通过校验的 found（文件开头的公告最权威）。
+    最后没有 found 的字段分两种：看过的分块都干净地说 not_found → absent（文件本身没写，至少开头几块没写）；
+    有歧义、漏答、说没找到却带证据、证据或取值没通过 → needs_review（抽取出了问题，要人核对），带原因码。"""
     info = ProjectInfo()
+    codes: dict[str, list[str]] = {k: [] for k in INFO_KEYS}
     for chunk, data in results:
         by_id = {_sid(s): s for s in chunk}
-        fields = data.get("fields") or {}
+        fields = data.get("fields")
+        if not isinstance(fields, dict):
+            for key in INFO_KEYS:
+                codes[key].append("missing_answer")
+            continue
         for key in INFO_KEYS:
-            f = fields.get(key)
-            if key in info.fields or not isinstance(f, dict) or f.get("value") in (None, ""):
+            if key in info.fields:
                 continue
-            sents, _, ok = _evidence({"evidence": [{"sid": f.get("sid"), "quote": f.get("quote")}]}, by_id)
-            value = f["value"]
+            if key not in fields:
+                codes[key].append("missing_answer")
+                continue
+            f = fields[key]
+            status = f.get("status") if isinstance(f, dict) else None
+            if not isinstance(f, dict) or status == "not_found" or (status is None and f.get("value") in (None, "")):
+                if isinstance(f, dict) and (f.get("sid") or f.get("quote")):
+                    codes[key].append("not_found_with_evidence")
+                continue
+            if status == "ambiguous":
+                codes[key].append("ambiguous")
+                continue
+            value = f.get("value")
+            if value in (None, ""):
+                codes[key].append("missing_answer")
+                continue
+            if is_placeholder(value):
+                codes[key].append("placeholder_value")
+                continue
+            sents, _, ok, fails = _evidence({"evidence": [{"sid": f.get("sid"), "quote": f.get("quote")}]}, by_id)
             if not ok:
+                codes[key] += fails
                 continue
             if isinstance(value, bool):
                 if not check_bool(key, value, sents):
+                    codes[key].append("value_unverified")
                     continue
             elif key != "legal_regime" and not check_value(value, sents):
+                codes[key].append("value_unverified")
                 continue
             info.fields[key] = InfoField(value=value, unit=f.get("unit"), sentence=sents[0])
         if not info.lots:
             for lot in data.get("lots") or []:
                 if not isinstance(lot, dict) or not lot.get("lot_no"):
                     continue
-                sents, _, ok = _evidence({"evidence": [{"sid": lot.get("sid"), "quote": lot.get("quote")}]}, by_id)
+                ev = {"evidence": [{"sid": lot.get("sid"), "quote": lot.get("quote")}]}
+                sents, _, ok, _ = _evidence(ev, by_id)
                 if ok:
                     info.lots.append({**lot, "sentence": sents[0]})
+    for key in INFO_KEYS:
+        if key in info.fields:
+            info.status[key] = {"status": "found"}
+        elif codes[key]:
+            info.status[key] = {"status": "needs_review", "codes": [*dict.fromkeys(codes[key])]}
+        else:
+            info.status[key] = {"status": "absent"}
     return info
 
 

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

@@ -109,7 +109,7 @@ async def _stage(ctx: TenantContext, job_id: UUID, stage: str) -> dict[str, Any]
     return checkpoints
 
 
-async def _stages(ctx: TenantContext, job_id: UUID) -> dict[str, int]:
+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()
@@ -189,7 +189,7 @@ async def _stages(ctx: TenantContext, job_id: UUID) -> dict[str, int]:
             last[0] = pct
             await _progress(ctx, job_id, "extract", pct)
 
-    items, info = await extract.run(ctx, job_id, doc_info, progress)
+    items, info, run_stats = await extract.run(ctx, job_id, doc_info, progress)
 
     await _stage(ctx, job_id, "verify")  # 三道校验在抽取时逐条做完；这里只是进度节点
     await _stage(ctx, job_id, "merge")
@@ -197,7 +197,15 @@ async def _stages(ctx: TenantContext, job_id: UUID) -> dict[str, int]:
         d = (await s.execute(select(SourceDocument).where(SourceDocument.id == doc.id))).scalar_one()
         lots = await project_extracted.apply(s, ctx, d, info, extract.legal_regime_by_rule(body), items)
         stats = await checklist.merge(s, ctx, d, job_id, items, notes, lots)
-    return {**stats, "lots_found": len([k for k in lots if k != "默认"])}
+    return {
+        **stats,
+        "lots_found": len([k for k in lots if k != "默认"]),
+        # 质量统计（评测和排查用）：没通过校验的原因码条数、对半重跑次数、拆到底仍不完整的分块、项目信息各字段的结论
+        "verify_failures": {k.removeprefix("fail:"): n for k, n in run_stats.items() if k.startswith("fail:")},
+        "chunks_split": run_stats.get("chunks_split", 0),
+        "output_incomplete": run_stats.get("output_incomplete", 0),
+        "info_fields": {k: v["status"] for k, v in info.status.items()},
+    }
 
 
 async def _finish(
@@ -206,7 +214,7 @@ async def _finish(
     status: str,
     error_class: str | None,
     code: str | None,
-    result: dict[str, int] | None = None,
+    result: dict[str, Any] | None = None,
 ) -> None:
     async with tenant_tx(ctx) as s:
         job = (await s.execute(select(ParseJob).where(ParseJob.id == job_id))).scalar_one()

server/tests/test_units.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import itertools
+
 import pytest
 
 from bid_doctor.checklist.extract import (
@@ -30,9 +32,10 @@ def test_evidence_checks() -> None:
     s = sent("★投标人须具备有效的信息系统集成服务资质证书。")
     by_id = {"p1s1": s}
     assert check_evidence({"sid": "p1s1", "quote": "投标人须具备有效的信息"}, by_id) == (None, [s])
-    assert check_evidence({"sid": "p9s9", "quote": "投标人须具备有效的信息"}, by_id)[0] == "id"
-    assert check_evidence({"sid": "p1s1", "quote": "投标人须具备"}, by_id)[0] == "quote"  # 少于 8 个字
-    assert check_evidence({"sid": "p1s1", "quote": "这句话是模型改写的"}, by_id)[0] == "quote"
+    assert check_evidence({"sid": "p9s9", "quote": "投标人须具备有效的信息"}, by_id)[0] == "sid_unknown"
+    assert check_evidence({"sid": "第1页", "quote": "投标人须具备有效的信息"}, by_id)[0] == "sid_format"
+    assert check_evidence({"sid": "p1s1", "quote": "投标人须具备"}, by_id)[0] == "quote_too_short"  # 少于 8 个字
+    assert check_evidence({"sid": "p1s1", "quote": "这句话是模型改写的"}, by_id)[0] == "quote_not_original"
 
 
 def test_quote_may_run_into_next_sentence() -> None:
@@ -41,9 +44,9 @@ def test_quote_may_run_into_next_sentence() -> None:
     b = sent("用户，连续30分钟压力测试成功率不低于99%。", seq=2)
     by_id = {"p1s1": a, "p1s2": b}
     assert check_evidence({"sid": "p1s1", "quote": "支持不少于200个并发用户"}, by_id, {a.key: b}) == (None, [a, b])
-    assert check_evidence({"sid": "p1s1", "quote": "支持不少于200个并发用户"}, by_id)[0] == "quote"
+    assert check_evidence({"sid": "p1s1", "quote": "支持不少于200个并发用户"}, by_id)[0] == "quote_not_original"
     # 必须从这一句开始，不能整段都在下一句里
-    assert check_evidence({"sid": "p1s1", "quote": "连续30分钟压力测试"}, by_id, {a.key: b})[0] == "quote"
+    assert check_evidence({"sid": "p1s1", "quote": "连续30分钟压力测试"}, by_id, {a.key: b})[0] == "quote_not_original"
 
 
 def test_split_listed_conditions() -> None:
@@ -249,8 +252,8 @@ def test_quote_ignores_our_markup_and_end_punctuation() -> None:
     assert check_evidence({"sid": "p1s1", "quote": "用扣除后的价格参与评审〔续下页〕"}, by_id)[0] is None
     short = sent("1.软件开发；")
     assert check_evidence({"sid": "p1s1", "quote": "1.软件开发"}, {"p1s1": short})[0] is None
-    assert check_evidence({"sid": "p1s1", "quote": "软件开发"}, {"p1s1": short})[0] == "quote"
-    assert check_evidence({"sid": "p1s1", "quote": "价格扣除"}, by_id)[0] == "quote"  # 只摘关键词照样退回
+    assert check_evidence({"sid": "p1s1", "quote": "软件开发"}, {"p1s1": short})[0] == "quote_too_short"
+    assert check_evidence({"sid": "p1s1", "quote": "价格扣除"}, by_id)[0] == "quote_too_short"  # 只摘关键词照样退回
 
 
 def test_chinese_numbers_in_values() -> None:
@@ -302,3 +305,100 @@ def test_model_input_marks_only_for_marking_task() -> None:
     d = doc(sentences=[a, b], continuations={a.key: b})
     assert _render([a, b], d).splitlines()[-2:] == ["[p1s1] 须提供授权书，采购人有权〔续下页〕", "[p2s1] 永久使用。"]
     assert "<iu>授权书</iu>" in _render([a, b], d, marks=True)
+
+
+# ---------------------------------------------------------------- 2026-11-03 参考 BidEngine 的五项改进
+
+
+def test_placeholder_is_not_a_value() -> None:
+    from bid_doctor.checklist.extract import is_placeholder, value_issue
+
+    s = sent("最高限价：____万元，投标保证金：（  ）元。")
+    assert is_placeholder("____") and is_placeholder("××万元") and is_placeholder("（ ）") and is_placeholder("待填")
+    assert not is_placeholder("45") and not is_placeholder("0～5")
+    assert value_issue("____", [s]) == "placeholder_value"  # 空白在原文里「找得到」，但不是值
+    assert value_issue("45", [s], "amount_unverified") == "amount_unverified"
+    assert value_issue("45", [sent("最高限价：45万元。")]) is None
+
+
+def test_chunks_overlap_and_keep_continuations() -> None:
+    from bid_doctor.checklist import extract as x
+
+    ss = [sent(f"第{i}条要求的内容写在这里，长度大约四十个字左右，用来撑满分块。" * 3, seq=i) for i in range(1, 200)]
+    parts = x.chunks(ss, {})
+    assert len(parts) > 2
+    for a, b in itertools.pairwise(parts):
+        tail = b[: len(b) - len([s for s in b if s not in a])]
+        assert tail and tail == a[-len(tail) :]  # 下一块开头是上一块的结尾几句
+        assert len(tail) <= x.OVERLAP_SENTENCES
+    # 每一句都至少在一个分块里
+    assert {s.key for p in parts for s in p} == {s.key for s in ss}
+
+
+async def test_truncated_output_is_split_and_rerun(monkeypatch: pytest.MonkeyPatch) -> None:
+    """模型输出被截断：把分块对半重跑，拆到底还不完整就记下来；不能当成「这一块没有结果」。"""
+    from uuid import UUID
+
+    from bid_doctor.ai import service as ai
+    from bid_doctor.checklist import extract as x
+    from bid_doctor.core.db import TenantContext
+
+    ss = [sent(f"★第{i}条：供应商须具备第{i}项能力，否则投标无效。", seq=i) for i in range(1, 9)]
+    sizes: list[int] = []
+
+    async def fake_run_json(ctx: object, **kw: object) -> dict[str, object]:
+        user = str(kw["user"])
+        n = user.count("\n[p")
+        sizes.append(n)
+        if kw["task"] == "extract.substantive" and n > 2:
+            raise ai.OutputIncomplete("truncated")
+        if kw["task"] == "extract.substantive":
+            lines = [ln for ln in user.splitlines() if ln.startswith("[p")]
+            items = [
+                {"summary": ln[8:20], "basis": "star", "evidence": [{"sid": ln[1:5].rstrip("]"), "quote": ln[7:]}]}
+                for ln in lines
+            ]
+            return {"items": items}
+        return {"items": []} if kw["task"] != "extract.project_info" else {"fields": {}}
+
+    monkeypatch.setattr(x.ai, "run_json", fake_run_json)
+    d = doc(declared=set(), sentences=ss)
+    items, _, stats = await x.run(TenantContext(tenant_id=UUID(int=1), user_id=UUID(int=1)), UUID(int=2), d)
+    assert stats["chunks_split"] >= 3 and stats.get("output_incomplete", 0) == 0
+    assert {s.key for e in items if e.zone == "formal" for s in e.sentences} == {s.key for s in ss}
+
+
+def test_project_info_status() -> None:
+    from bid_doctor.checklist.extract import _project_info  # pyright: ignore[reportPrivateUsage]
+
+    a = sent("项目名称：某市智慧政务平台建设项目。", seq=1)
+    b = sent("最高限价：____万元。", seq=2)
+    chunk = [a, b]
+    found = {
+        "status": "found",
+        "value": "某市智慧政务平台建设项目",
+        "sid": "p1s1",
+        "quote": "项目名称：某市智慧政务平台建设项目。",
+    }
+    nf = {"status": "not_found", "value": None, "sid": None, "quote": None}
+    fields = {
+        "name": found,
+        "ceiling_total": {
+            "status": "found",
+            "value": "____",
+            "unit": "万元",
+            "sid": "p1s2",
+            "quote": "最高限价：____万元。",
+        },
+        "budget_total": {"status": "ambiguous", "value": None},
+        "code": {**nf, "sid": "p1s1"},
+        **{k: nf for k in ("purchaser", "agency", "procurement_method", "legal_regime", "eval_method")},
+        **{k: nf for k in ("accept_consortium", "accept_imported")},
+    }  # sme_only 漏答
+    info = _project_info([(chunk, {"fields": fields})])
+    st = {k: v["status"] for k, v in info.status.items()}
+    assert st["name"] == "found" and st["purchaser"] == "absent"
+    assert info.status["ceiling_total"] == {"status": "needs_review", "codes": ["placeholder_value"]}
+    assert info.status["budget_total"]["codes"] == ["ambiguous"]
+    assert info.status["code"]["codes"] == ["not_found_with_evidence"]
+    assert info.status["sme_only"]["codes"] == ["missing_answer"]