ai.memory/MEMORY.md

@@ -6,7 +6,7 @@
 
 - [CADProject 图纸审查 Agent 工作](project_dwg_review_agent.md) — 基于 MCP 框架的图纸审查工具集与逐图框审查流程
 - [DeepSeek API Key 泄露与轮换（已处理）](project_api_key_rotation.md) — 旧 Key 曾入 git 历史；2026-10-14 已吊销并换新 Key，验证可用
-- [LLM 识图审图 Agent 架构决策](project_review_agent_impl.md) — 混合式自研 agent、基座 B 最小集、标准化先走 stdio 桥；含 dwg_review 包结构
+- [LLM 识图审图 Agent 架构决策](project_review_agent_impl.md) — 混合式自研 agent、基座 B 最小集、可信度三件套、标准化先走 stdio 桥；含 dwg_review 包结构
 - [stdio↔TCP 桥（MCP 标准化）](project_stdio_bridge.md) — mcp_bridge 包：基座协议缺口、读线程+单工作线程设计、MCP Inspector 验证方法与踩坑
 - [MCP 基座 c++_dll 加载与执行的坑](project_mcp_frame_dll_loader.md) — acrxGetApiVersion 有状态不可比对、写库需文档锁、同一 DLL 只加载一次
 - [用户协作偏好](feedback_preferences.md) — 调试用 Debug/R230、模型 deepseek-flash、DLL 集中在 cad_mcp_plugins 等

ai.memory/feedback_preferences.md

@@ -22,6 +22,19 @@ type: feedback
   `git push origin HEAD:refs/for/main/home-ai-dev`。每次提交后自动完成推送 + 建 PR，无需再问。
   注意 `home-ai-dev` 的 upstream 常显示 `[gone]`（合并后源分支被删），这不影响该推送方式。
 
+## CAD 没开就自己打开，别问（自动加载已配好）
+
+- 规则：需要 CAD 在线（慢层 `verify_*.py` / 端到端 `python -m dwg_review`）而
+  `127.0.0.1:8080` 未监听时，**自己启动 CAD，不要先问用户**。
+- **Why:** 用户（2026-10-14）明确要求：「以后遇到 cad 没有运行的情况，就直接打开，
+  运行你的测试就行，我已经设置好了自动加载」。自动加载确实已配好。
+- **How to apply:** PowerShell 里
+  `Start-Process "C:\Program Files\Autodesk\AutoCAD 2020\acad.exe"`，再轮询 8080
+  （实测约 10 秒内监听）。自动加载链路：`bin/Debug-R230/acad.rx`（内容一行
+  `cad_mcp_frame.arx`）+ 该目录已在 CAD 的搜索/信任路径中。
+  **注意**：CAD 默认打开的是**空模板**（0 图元、无图框），端到端审查会因
+  「未识别到任何图框」提前退出；要跑完整审查需另开一张真实图纸。
+
 ## 调试用 R230 + Debug，不用 Release
 
 - 规则：涉及 CAD 版本的调试**只用 R230**，且配置用 **Debug**（非 Release）。

ai.memory/project_review_agent_impl.md

@@ -1,6 +1,6 @@
 ---
 name: LLM 识图审图 Agent —— 架构决策（已定稿）
-description: cad_mcp_frame + 自研 Agent 的架构决策与分阶段路线——混合式（流水线骨架+受控只读取证子循环）、基座 B 最小集、标准化路线（先 stdio 桥）
+description: cad_mcp_frame + 自研 Agent 的架构决策与分阶段路线——混合式（流水线骨架+受控只读取证子循环）、基座 B 最小集、可信度三件套（确定性检查前置/结构化结论/handle 交叉校验）、标准化路线（先 stdio 桥）
 type: project
 ---
 
@@ -48,6 +48,29 @@ type: project
   （构造即 `list_tools()` 触网）；一律 `make_ctx()` 注入假件。
 - `verify_mcp_frame.py`：基座改动验证脚本（分帧 / 大报文 / string id / 通知 / isError）。
 
+## 可信度三件套（2026-10-14 落地）
+
+根因是**「LLM 从自由文本里回忆 handle」**：原 `review_frame` 输出无格式约束的自由文本，
+再用**第二次 LLM 调用**（`extract_issues`）转 JSON，handle 靠回忆 → 批注错位或丢失。
+
+改后：`review_frame` 直接输出受约束 JSON `{summary, issues[]}`（**删除**二次抽取）
++ `checks.run_deterministic_checks`（确定性，handle 来自工具/数据本身）
+→ `findings.build_finding`（合并 + handle 交叉校验）→ `annotator`（只批注 verified 的 handle）。
+
+- **确定性检查前置**（`dwg_review/checks.py`）：`find_zero_length` / `find_duplicate_entities`
+  （都传 `window`）+ 纯 Python 派生（空文字、未闭合多段线 `vertex_count>=3 and closed==false`）。
+  实机验证：四类全部检出。
+- **结论结构化**：`checklist.build_system_prompt` 强约束只输出
+  `{summary, issues:[{category,severity,message,handle?}]}`；`review_frame` 仅在
+  「不再请求工具」的回合解析（中间工具回合的 content 永不解析）；解析失败降级
+  `summary=原文, issues=[]`（信息不丢，只是不进批注）。实机验证 deepseek-flash 遵守契约。
+- **证据交叉校验**（`findings.normalize_issue`）：LLM 引用的 handle 不在已采集证据中时
+  **保留问题、剥掉 handle、`verified=False`**（永不批注）；handle 比对统一 `upper()`。
+- 开关（`config.py`，默认全开）：`DETERMINISTIC_CHECKS` / `VALIDATE_HANDLES` /
+  `CHECK_UNCLOSED_POLYLINES`。
+- 结构：issue = `{frame,category,severity,message,handle,handles?,source,verified}`；
+  findings 元素 = `{frame,sheet,summary,issues[],counts{deterministic,llm,unverified}}`。
+
 ## 环境事实
 
 - **Pillow 已安装**（12.3.0）。截图下采样策略（`dwg_review/llm_client.py: downsample_image_b64`）：
@@ -56,6 +79,10 @@ type: project
   缩放后默认输出**无损 PNG**（`IMAGE_FORMAT=png`，适合线稿），需要更小体积可设 `IMAGE_FORMAT=jpeg`。
   缺少 Pillow 时自动降级为原图直传。
 - 视觉模型 `deepseek-flash`。
+- **`find_zero_length` / `find_duplicate_entities` 的 `window` 参数：schema 未声明，但 C++ 侧
+  `parseCommonQuery` 实际支持**（2026-10-14 实机验证生效）。Python 侧直接传 `window` 即可，
+  不必改 JSON schema、不必重编 C++。注意 `find_duplicate_entities.limit` 语义是
+  **最大重复组数**（非条目数），返回 `{count, groups:[{reason,type,handles[]}]}`（非分页）。
 - **`DEEPSEEK_API_KEY` 设在 Windows `Machine`（系统）级**。因新设的系统变量**不会传播到已启动
   的进程**，`dwg_review/config.py` 的 `_read_env()` 在进程环境读不到时**回退读注册表**
   （`HKLM\SYSTEM\CurrentControlSet\Control\Session Manager\Environment` → `HKCU\Environment`）。

dwg_review/__main__.py

@@ -6,6 +6,8 @@ import sys
 from pathlib import Path
 
 from . import config
+from . import checks as checks_phase
+from . import findings as findings_mod
 from .annotator import annotate, clear
 from .checklist import build_system_prompt, checklist_summary
 from .context import ReviewContext
@@ -80,25 +82,35 @@ def run_review(ctx, args):
         print(f"[evidence] 已写入 {path}")
         return 0
 
-    print("[3/4] 审查：逐图框 LLM 审查（含受控只读取证子循环） ...")
-    findings = []
+    print("[3/4] 审查：确定性检查前置 + 逐图框 LLM 审查（含受控只读取证子循环） ...")
+    frame_findings = []
     for ev in evidences:
         print(f"  - 审查图框 {ev['index']} ...")
-        text = review.review_frame(ctx, ev, system_prompt)
-        findings.append(
-            {
-                "frame": ev["index"],
-                "sheet": (ev.get("frame") or {}).get("sheet_size"),
-                "finding": text,
-            }
+        checks = (
+            checks_phase.run_deterministic_checks(ctx, ev)
+            if config.DETERMINISTIC_CHECKS
+            else []
+        )
+        conclusion = review.review_frame(ctx, ev, system_prompt)
+        finding = findings_mod.build_finding(
+            ev, conclusion, checks, validate_handles=config.VALIDATE_HANDLES
+        )
+        frame_findings.append(finding)
+
+        counts = finding["counts"]
+        preview = finding["summary"]
+        if len(preview) > 200:
+            preview = preview[:200] + "..."
+        print(
+            f"    结论：{preview}"
+            f"（确定性 {counts['deterministic']} / LLM {counts['llm']}"
+            f" / 未核实 {counts['unverified']}）"
         )
-        preview = text if len(text) <= 200 else text[:200] + "..."
-        print(f"    结论：{preview}")
 
     print("[4/4] 汇总报告 ...")
-    markdown = report.summarize(ctx, findings)
-    issues = report.extract_issues(ctx, findings)
-    report.save_report(markdown, findings, plan, issues)
+    markdown = report.summarize(ctx, frame_findings)
+    issues = findings_mod.flatten_issues(frame_findings)
+    report.save_report(markdown, frame_findings, plan, issues)
     print()
     print(markdown)
 

dwg_review/annotator.py

@@ -36,11 +36,13 @@ def annotate(ctx, issues, *, dry_run=False, log=print):
     """按问题列表回写批注，返回 (created, highlighted) 两个 handle 列表。"""
     created = []
     highlighted = []
+    seen = set()  # 同一 handle 只处理一次，避免重复批注
 
     for issue in issues:
         handle = issue.get("handle")
-        if not handle:
+        if not handle or handle in seen:
             continue
+        seen.add(handle)
 
         text = f"[{issue.get('severity', '?')}] {issue.get('message', '')}"
 
@@ -81,8 +83,6 @@ def annotate(ctx, issues, *, dry_run=False, log=print):
             continue
 
         # 引线建好后再高亮原实体（失败不影响已创建的批注）
-        if handle in highlighted:
-            continue
         try:
             ctx.tools.call("highlight_entities", {"handle": handle, "color": 1})
             highlighted.append(handle)

dwg_review/checklist.py

@@ -62,9 +62,27 @@ def build_system_prompt(checklist, *, allow_tools=True):
         "要求：",
         "- 只报告你**有依据**的问题，不要臆测；数据与截图冲突时以数据为准。",
         "- 按清单条目组织结论，未命中的条目可省略。",
-        "- 若该图框无任何明显问题，明确说“未发现明显问题”。",
         "- 不要复述输入数据。",
+        "",
+        "输出格式（**必须严格遵守**）：",
+        "在你完成全部取证、准备给出该图框的最终结论时，**只输出一个 JSON 对象**，"
+        "不要输出任何解释文字，也不要用 markdown 代码块包裹。结构如下：",
+        "{",
+        '  "summary": "<该图框的一句话整体结论；若未发现问题，写“未发现明显问题”>",',
+        '  "issues": [',
+        "    {",
+        '      "category": "<图层/文字/标注/几何/图面整体 之一>",',
+        '      "severity": "<high|medium|low>",',
+        '      "message": "<问题描述，简洁具体>",',
+        '      "handle": "<该问题对应的实体句柄；无法定位到具体实体时省略此字段>"',
+        "    }",
+        "  ]",
+        "}",
+        "其中 `handle` 必须是你**在取证数据或工具返回里真实见过**的句柄，不要编造；"
+        "无法确定具体实体时省略 `handle`，该问题仍会被记录。",
     ]
+    if allow_tools:
+        lines.append("（需要进一步确认时先调用工具；拿到结果后，最终再输出上述 JSON。）")
     return "\n".join(lines)
 
 

dwg_review/checks.py

@@ -0,0 +1,164 @@
+"""确定性检查：MCP 工具（find_zero_length / find_duplicate_entities）+ 纯 Python 派生。
+
+产出的 issue 带 `source="deterministic"`，其 handle 来自工具/数据本身，
+因此被视为**可信**，可直接进入批注链路，不依赖 LLM 回忆。
+
+注意：`find_zero_length` / `find_duplicate_entities` 的 `window` 参数在 C++ 侧
+（`parseCommonQuery`）**实际生效**，只是 JSON schema 未声明——Python 侧显式传参即可。
+"""
+
+from . import config
+from .mcp_client import McpError, McpToolError, result_json
+
+
+def make_issue(category, severity, message, *, handle=None, handles=None,
+               source="deterministic"):
+    """构造统一 issue（`verified` 由 findings 层补齐）。"""
+    issue = {
+        "category": category,
+        "severity": severity,
+        "message": message,
+        "source": source,
+    }
+    if handle:
+        issue["handle"] = handle
+    if handles:
+        issue["handles"] = list(handles)
+    return issue
+
+
+def _iter_items(envelope):
+    """健壮迭代 items（`result_json` 解析失败会返回原始字符串）。"""
+    if not isinstance(envelope, dict):
+        return []
+    items = envelope.get("items")
+    if not isinstance(items, list):
+        return []
+    return [i for i in items if isinstance(i, dict)]
+
+
+# ---------------------------------------------------------------------------
+#  MCP 检查
+# ---------------------------------------------------------------------------
+def check_zero_length(ctx, window, *, tolerance, limit, log=print):
+    """find_zero_length(window) → 逐条零长度实体转 issue。"""
+    try:
+        resp = ctx.tools.call(
+            "find_zero_length",
+            {"window": window, "tolerance": tolerance, "limit": limit},
+        )
+    except (McpError, McpToolError) as exc:
+        log(f"      [确定性] find_zero_length 失败: {exc}")
+        return []
+
+    out = []
+    for item in _iter_items(result_json(resp, {})):
+        handle = item.get("handle")
+        if not handle:
+            continue
+        out.append(
+            make_issue(
+                "几何", "medium",
+                f"零长度/退化实体（类型 {item.get('type')}，长度 {item.get('length')}）",
+                handle=handle,
+            )
+        )
+    return out
+
+
+def check_duplicate_entities(ctx, window, *, tolerance, group_limit, log=print):
+    """find_duplicate_entities(window) → 每个重复组一条 issue。
+
+    `limit` 的语义是「最大重复组数」（非条目数）。一组重复是一个问题，
+    故只产出一条 issue，主 handle 取组内首个，全部 handle 放进 `handles`。
+    """
+    try:
+        resp = ctx.tools.call(
+            "find_duplicate_entities",
+            {"window": window, "tolerance": tolerance, "limit": group_limit},
+        )
+    except (McpError, McpToolError) as exc:
+        log(f"      [确定性] find_duplicate_entities 失败: {exc}")
+        return []
+
+    env = result_json(resp, {})
+    groups = env.get("groups") if isinstance(env, dict) else None
+    out = []
+    for group in groups or []:
+        if not isinstance(group, dict):
+            continue
+        handles = [h for h in (group.get("handles") or []) if h]
+        if not handles:
+            continue
+        out.append(
+            make_issue(
+                "几何", "low",
+                f"重复/重叠实体组（类型 {group.get('type')}，共 {len(handles)} 个几何相同）",
+                handle=handles[0],
+                handles=handles,
+            )
+        )
+    return out
+
+
+# ---------------------------------------------------------------------------
+#  纯 Python 派生（零额外 MCP 往返）
+# ---------------------------------------------------------------------------
+def _empty_texts(evidence):
+    out = []
+    for item in _iter_items(evidence.get("texts")):
+        content = item.get("content")
+        if content is None or str(content).strip() == "":
+            out.append(
+                make_issue("文字", "medium", "存在空文字或仅含空格的文字",
+                           handle=item.get("handle"))
+            )
+    return out
+
+
+def _unclosed_polylines(evidence):
+    out = []
+    for item in _iter_items(evidence.get("polylines")):
+        if item.get("vertex_count", 0) >= 3 and item.get("closed") is False:
+            out.append(
+                make_issue("几何", "low",
+                           "多段线顶点≥3但未闭合，可能是应当闭合的轮廓",
+                           handle=item.get("handle"))
+            )
+    return out
+
+
+def derive_from_evidence(evidence, *, check_unclosed=True):
+    """从已采集证据里做零成本派生检查。"""
+    issues = _empty_texts(evidence)
+    if check_unclosed:
+        issues += _unclosed_polylines(evidence)
+    return issues
+
+
+# ---------------------------------------------------------------------------
+#  入口
+# ---------------------------------------------------------------------------
+def run_deterministic_checks(ctx, evidence, *, log=print):
+    """对单个图框跑全部确定性检查，返回 issue 列表。
+
+    任何单项 MCP 失败都只记录并跳过，不抛出（不拖垮该图框）。
+    """
+    issues = derive_from_evidence(
+        evidence, check_unclosed=config.CHECK_UNCLOSED_POLYLINES
+    )
+    window = evidence.get("window")
+    if window:
+        issues += check_zero_length(
+            ctx, window,
+            tolerance=config.ZERO_LENGTH_TOLERANCE,
+            limit=config.FRAME_ENTITY_LIMIT,
+            log=log,
+        )
+        issues += check_duplicate_entities(
+            ctx, window,
+            tolerance=config.DUP_TOLERANCE,
+            group_limit=config.DUP_GROUP_LIMIT,
+            log=log,
+        )
+    return issues

dwg_review/config.py

@@ -47,6 +47,14 @@ def _read_env(name: str) -> str:
     return ""
 
 
+def _read_bool(name: str, default: bool = True) -> bool:
+    """读取布尔型环境变量；未设置或空串时返回 default。"""
+    value = os.environ.get(name)
+    if value is None or value.strip() == "":
+        return default
+    return value.strip().lower() in ("1", "true", "yes", "on")
+
+
 # ---------------------------------------------------------------------------
 # CAD MCP 基座（TCP JSON-RPC）
 # ---------------------------------------------------------------------------
@@ -87,6 +95,22 @@ REPORT_DIR = _resolve_path(os.environ.get("REPORT_DIR", "reports"))
 # 批注回写使用的图层
 ANNOTATE_LAYER = os.environ.get("ANNOTATE_LAYER", "AI_REVIEW")
 
+# ---------------------------------------------------------------------------
+# 可信度三件套：确定性检查前置 / 结论结构化 / 证据交叉校验
+# ---------------------------------------------------------------------------
+# 是否在 LLM 审查前跑确定性检查（零长度、重复实体、空文字、未闭合多段线）
+DETERMINISTIC_CHECKS = _read_bool("DETERMINISTIC_CHECKS", True)
+# 是否校验 LLM 引用的 handle 确实出现在已采集证据中（不在则剥掉 handle）
+VALIDATE_HANDLES = _read_bool("VALIDATE_HANDLES", True)
+# 是否检查未闭合多段线（中心线/开放轮廓多时误报，可关闭）
+CHECK_UNCLOSED_POLYLINES = _read_bool("CHECK_UNCLOSED_POLYLINES", True)
+# find_zero_length 的长度阈值
+ZERO_LENGTH_TOLERANCE = float(os.environ.get("ZERO_LENGTH_TOLERANCE", "0.001"))
+# find_duplicate_entities 的几何量化容差
+DUP_TOLERANCE = float(os.environ.get("DUP_TOLERANCE", "0.001"))
+# find_duplicate_entities 的 limit（语义是「最大重复组数」，非条目数）
+DUP_GROUP_LIMIT = int(os.environ.get("DUP_GROUP_LIMIT", "50"))
+
 # ---------------------------------------------------------------------------
 # 截图（视觉 token 成本控制）
 # ---------------------------------------------------------------------------

dwg_review/findings.py

@@ -0,0 +1,207 @@
+"""结构化结论：解析 LLM 的 JSON 输出、交叉校验 handle、装配 findings。
+
+设计要点：
+  - `review_frame` 的结论是**受约束的 JSON**（`{summary, issues[]}`），不再是自由文本，
+    因此不需要第二次 LLM 调用来「抽取」结构。
+  - 解析失败时降级为 `summary=原文, issues=[]`：信息不丢，只是不进批注链路。
+  - LLM 引用的 handle 若不在已采集证据中 → **保留问题、剥掉 handle**（verified=False），
+    既不会丢发现，也绝不会批注到不存在的实体。
+"""
+
+import json
+import re
+
+# evidence 中带 handle 的数据源（items[].handle）
+_SOURCES = ("entities", "texts", "dimensions", "blocks", "polylines")
+_SEVERITIES = ("high", "medium", "low")
+_FENCE = re.compile(r"```(?:json)?\s*(.*?)```", re.S)
+_MISS = object()  # 解析失败的哨兵（区别于合法的 JSON null）
+
+
+def _try_load(text):
+    try:
+        return json.loads(text)
+    except (json.JSONDecodeError, TypeError):
+        return _MISS
+
+
+def _iter_items(envelope):
+    """健壮迭代数据信封的 items。
+
+    `result_json` 在解析失败时会返回**原始字符串**，因此必须做 dict/list 守卫。
+    """
+    if not isinstance(envelope, dict):
+        return []
+    items = envelope.get("items")
+    if not isinstance(items, list):
+        return []
+    return [i for i in items if isinstance(i, dict)]
+
+
+def _outermost_json(text):
+    """抓文本中第一个 `{`/`[` 起、括号配平的整段 JSON（正确处理字符串内的括号）。"""
+    start = None
+    for i, ch in enumerate(text):
+        if ch in "{[":
+            start, open_c = i, ch
+            break
+    if start is None:
+        return None
+    close_c = "}" if open_c == "{" else "]"
+    depth = 0
+    in_str = False
+    esc = False
+    for i in range(start, len(text)):
+        ch = text[i]
+        if in_str:
+            if esc:
+                esc = False
+            elif ch == "\\":
+                esc = True
+            elif ch == '"':
+                in_str = False
+            continue
+        if ch == '"':
+            in_str = True
+        elif ch == open_c:
+            depth += 1
+        elif ch == close_c:
+            depth -= 1
+            if depth == 0:
+                cand = text[start : i + 1]
+                return cand if _try_load(cand) is not _MISS else None
+    return None
+
+
+def extract_json_block(text):
+    """从可能含围栏/前后缀的文本中抽出可解析的 JSON 字符串；抽不到返回 None。"""
+    t = (text or "").strip()
+    if not t:
+        return None
+    if _try_load(t) is not _MISS:  # 整体就是 JSON
+        return t
+    m = _FENCE.search(t)
+    if m:
+        inner = m.group(1).strip()
+        if _try_load(inner) is not _MISS:
+            return inner
+        t = inner
+    return _outermost_json(t)
+
+
+def parse_conclusion(text, *, log=print):
+    """把 LLM 结论解析为 {"summary": str, "issues": [dict], "raw": str}。
+
+    支持：`{summary, issues}` 对象、裸 issues 数组、围栏包裹、前后夹带解释文字。
+    解析不到结构化内容时降级为 summary=原文、issues=[]。
+    """
+    block = extract_json_block(text)
+    if block is not None:
+        data = json.loads(block)
+        if isinstance(data, dict):
+            raw_issues = data.get("issues")
+            issues = (
+                [i for i in raw_issues if isinstance(i, dict)]
+                if isinstance(raw_issues, list)
+                else []
+            )
+            summary = data.get("summary")
+            return {
+                "summary": str(summary).strip() if summary is not None else "",
+                "issues": issues,
+                "raw": text,
+            }
+        if isinstance(data, list):
+            return {
+                "summary": "",
+                "issues": [i for i in data if isinstance(i, dict)],
+                "raw": text,
+            }
+    log("[结论解析] 非结构化输出，降级为纯文本 summary")
+    return {"summary": (text or "").strip(), "issues": [], "raw": text}
+
+
+def collect_known_handles(evidence, extra_handles=None):
+    """汇总证据里出现过的全部 handle（统一大写），外加 extra_handles。"""
+    known = set()
+    for key in _SOURCES:
+        for item in _iter_items((evidence or {}).get(key)):
+            handle = item.get("handle")
+            if handle:
+                known.add(str(handle).upper())
+    for handle in extra_handles or ():
+        if handle:
+            known.add(str(handle).upper())
+    return known
+
+
+def normalize_issue(issue, *, frame_index, source, known_handles, validate_handles):
+    """补 frame/source/verified 并做 handle 交叉校验，返回统一 issue 结构。"""
+    issue = issue if isinstance(issue, dict) else {}
+    handle = issue.get("handle")
+    handle = str(handle) if handle else None
+    verified = True
+    # 确定性问题的 handle 来自工具/数据本身，恒可信；只校验 LLM 的引用
+    if source != "deterministic" and handle and validate_handles:
+        if handle.upper() not in known_handles:
+            verified, handle = False, None
+
+    severity = issue.get("severity")
+    if severity not in _SEVERITIES:
+        severity = "low"
+
+    out = {
+        "frame": frame_index,
+        "category": str(issue.get("category") or "图面整体"),
+        "severity": severity,
+        "message": str(issue.get("message") or "").strip(),
+        "handle": handle,
+        "source": source,
+        "verified": verified,
+    }
+    handles = issue.get("handles")
+    if isinstance(handles, list) and handles:
+        out["handles"] = [str(h) for h in handles if h]
+    return out
+
+
+def build_finding(evidence, conclusion, checks_issues=None, *,
+                  validate_handles=True, log=print):
+    """把确定性检查与 LLM 结论合并、校验后装配成一个 findings 元素。"""
+    evidence = evidence or {}
+    index = evidence.get("index")
+    checks = checks_issues or []
+    known = collect_known_handles(
+        evidence, extra_handles=[i.get("handle") for i in checks]
+    )
+
+    det = [
+        normalize_issue(i, frame_index=index, source="deterministic",
+                        known_handles=known, validate_handles=validate_handles)
+        for i in checks
+    ]
+    llm = [
+        normalize_issue(i, frame_index=index, source="llm",
+                        known_handles=known, validate_handles=validate_handles)
+        for i in (conclusion or {}).get("issues", [])
+    ]
+    issues = det + llm
+    return {
+        "frame": index,
+        "sheet": (evidence.get("frame") or {}).get("sheet_size"),
+        "summary": (conclusion or {}).get("summary", ""),
+        "issues": issues,
+        "counts": {
+            "deterministic": len(det),
+            "llm": len(llm),
+            "unverified": sum(1 for i in issues if not i["verified"]),
+        },
+    }
+
+
+def flatten_issues(findings):
+    """拍平所有图框的 issues，供 annotator / save_report 使用。"""
+    out = []
+    for finding in findings or []:
+        out.extend(finding.get("issues") or [])
+    return out

dwg_review/phases/report.py

@@ -1,10 +1,37 @@
-"""阶段 4：汇总报告 + 结构化问题抽取。"""
+"""阶段 4：汇总报告 + 落盘。
+
+审查结论在阶段 3 已是结构化对象（`{summary, issues[]}`），
+因此这里**不再**需要第二次 LLM 调用来抽取结构。
+"""
 
 import json
 from datetime import datetime
 from pathlib import Path
 
 from .. import config
+from ..findings import flatten_issues
+
+_SUMMARY_SYSTEM = (
+    "你是图纸审查负责人。请把下面各图框的审查意见汇总为一份结构化的审查报告："
+    "先给整体结论，再按图框列出问题，最后给出整改建议。语言简洁、条目化。"
+)
+
+
+def format_finding(finding):
+    """把一个 findings 元素渲染成给汇总 LLM 的文本块。"""
+    lines = [f"【图框 {finding.get('frame')}（图幅 {finding.get('sheet')}）】"]
+    summary = finding.get("summary")
+    if summary:
+        lines.append(str(summary))
+    for issue in finding.get("issues") or []:
+        tag = "确定性" if issue.get("source") == "deterministic" else "LLM"
+        flag = "" if issue.get("verified", True) else "（未在证据中定位，仅供参考）"
+        handle = issue.get("handle")
+        where = f" handle={handle}" if handle else ""
+        lines.append(
+            f"- [{issue.get('severity', '?')}][{tag}] {issue.get('message', '')}{where}{flag}"
+        )
+    return "\n".join(lines)
 
 
 def summarize(ctx, findings, log=print):
@@ -12,54 +39,22 @@ def summarize(ctx, findings, log=print):
     if not findings:
         return "未识别到可审查的图框，或所有图框均无有效包围盒。"
 
-    report_input = "\n\n".join(
-        f"【图框 {f['frame']}（图幅 {f.get('sheet')}）】\n{f['finding']}" for f in findings
-    )
-    system = (
-        "你是图纸审查负责人。请把下面各图框的审查意见汇总为一份结构化的审查报告："
-        "先给整体结论，再按图框列出问题，最后给出整改建议。语言简洁、条目化。"
-    )
+    report_input = "\n\n".join(format_finding(f) for f in findings)
     try:
-        return ctx.llm.chat_text(system, report_input)
+        return ctx.llm.chat_text(_SUMMARY_SYSTEM, report_input)
     except Exception as exc:  # noqa: BLE001
         log(f"[汇总失败] {exc}")
         return report_input
 
 
-_EXTRACT_SYSTEM = (
-    "你是结构化信息抽取器。请把下面的图纸审查意见抽取为 JSON 数组，"
-    "每个元素形如："
-    '{"frame": <图框序号>, "handle": "<实体句柄，没有则省略>", '
-    '"category": "<图层/文字/标注/几何/图面整体>", '
-    '"severity": "<high|medium|low>", "message": "<问题描述>"}。'
-    "只输出 JSON 数组，不要输出任何解释、不要用 markdown 代码块包裹。"
-    "若某条意见无法定位到具体实体，省略 handle 字段。"
-)
-
-
-def extract_issues(ctx, findings, log=print):
-    """把自由文本的审查意见抽取为结构化问题列表（用于批注回写）。"""
-    if not findings:
-        return []
-    source = "\n\n".join(
-        f"图框 {f['frame']}：\n{f['finding']}" for f in findings
-    )
-    try:
-        text = ctx.llm.chat_text(_EXTRACT_SYSTEM, source)
-        text = text.strip()
-        if text.startswith("```"):
-            text = text.strip("`")
-            text = text.split("\n", 1)[1] if "\n" in text else text
-        data = json.loads(text)
-        if isinstance(data, list):
-            return data
-    except Exception as exc:  # noqa: BLE001
-        log(f"[结构化抽取失败] {exc}")
-    return []
-
-
 def save_report(markdown, findings, plan, issues=None, log=print):
-    """落盘 Markdown + JSON 报告，返回两个路径。"""
+    """落盘 Markdown + JSON 报告，返回两个路径。
+
+    `issues` 省略时从 findings 里拍平派生（供批注回写）。
+    """
+    if issues is None:
+        issues = flatten_issues(findings)
+
     out_dir = Path(config.REPORT_DIR)
     out_dir.mkdir(parents=True, exist_ok=True)
     stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -72,7 +67,7 @@ def save_report(markdown, findings, plan, issues=None, log=print):
             {
                 "generated_at": stamp,
                 "findings": findings,
-                "issues": issues or [],
+                "issues": issues,
                 "frames": plan.get("frames"),
             },
             ensure_ascii=False,

dwg_review/phases/review.py

@@ -3,12 +3,16 @@
 流程：
   1. 用清单生成系统提示，喂入结构化证据 + 截图。
   2. LLM 若请求只读取证工具 → 执行后把观测回灌，再问（有界循环）。
-  3. LLM 不再请求工具时，其输出即为该图框的审查结论。
+  3. LLM 不再请求工具时，其输出（受约束的 JSON）即为该图框的审查结论。
+
+结论是**结构化对象** `{"summary": str, "issues": [dict], "raw": str}`，
+不再依赖第二次 LLM 调用来抽取结构（见 `findings.parse_conclusion`）。
 """
 
 import json
 
 from ..cad_tools import READ_ONLY_TOOLS, is_write_tool
+from ..findings import parse_conclusion
 from ..llm_client import LlmClient, downsample_image_b64
 from ..mcp_client import McpError, McpToolError, result_images, result_texts
 
@@ -59,7 +63,7 @@ def _run_readonly_tool(ctx, name, args, image_budget):
 
 
 def review_frame(ctx, evidence, system_prompt, log=print):
-    """审查单个图框，返回审查意见文本。"""
+    """审查单个图框，返回结构化结论 `{"summary", "issues", "raw"}`。"""
     payload = evidence_payload(evidence)
     user_content = [
         {
@@ -86,8 +90,9 @@ def review_frame(ctx, evidence, system_prompt, log=print):
         msg = resp.choices[0].message
         messages.append(msg)
 
+        # 仅在「不再请求工具」的回合解析结论；中间工具回合的 content 永不解析
         if not msg.tool_calls:
-            return msg.content or ""
+            return parse_conclusion(msg.content or "", log=log)
 
         for tool_call in msg.tool_calls:
             name = tool_call.function.name
@@ -101,12 +106,12 @@ def review_frame(ctx, evidence, system_prompt, log=print):
                 {"role": "tool", "tool_call_id": tool_call.id, "content": content}
             )
 
-    # 达到回合上限：强制收口
+    # 达到回合上限：强制收口（不传 tools，保证不再请求工具，输出必为终稿 JSON）
     messages.append(
         {
             "role": "user",
-            "content": "已达到取证回合上限，请基于现有信息直接给出该图框的审查结论。",
+            "content": "已达到取证回合上限，请基于现有信息直接给出该图框的审查结论（JSON）。",
         }
     )
     resp = ctx.llm.chat(messages)
-    return resp.choices[0].message.content or ""
+    return parse_conclusion(resp.choices[0].message.content or "", log=log)

tests/fakes.py

@@ -101,7 +101,7 @@ class FakeLlmClient:
 
     - `script`      ：`chat()` 依次弹出的响应队列；弹空后回落到 `text`。
     - `text`        ：`chat()` / `chat_text()` 的兜底文本。
-    - `text_script` ：`chat_text()` 专用队列（便于同时喂 summarize 与 extract_issues）。
+    - `text_script` ：`chat_text()` 专用队列（便于连续喂多次 `chat_text` 调用）。
     """
 
     def __init__(self, script=None, text="", text_script=None, model="fake-model"):

tests/test_annotator.py

@@ -120,8 +120,8 @@ def test_annotate_handles_multiple_created_entities():
     assert created == ["N1", "N2"]
 
 
-def test_annotate_created_not_deduped_known_gap():
-    """已知瑕疵：同一 handle 出现两次会创建两条批注（created 无去重，仅 highlighted 去重）。"""
+def test_annotate_dedupes_repeated_handle():
+    """同一 handle 出现两次只批注一次（created 与 highlighted 均去重）。"""
     ctx = make_ctx(client=build_client())
     created, highlighted = annotate(
         ctx,
@@ -131,8 +131,9 @@ def test_annotate_created_not_deduped_known_gap():
         ],
         log=lambda *a: None,
     )
-    assert created == ["NEW1", "NEW1"]  # 重复批注（期望去重）
-    assert highlighted == ["A"]  # 高亮已去重
+    assert created == ["NEW1"]
+    assert highlighted == ["A"]
+    assert len(ctx.client.called("add_leader_annotation")) == 1
 
 
 def test_annotate_propagates_protocol_error_on_detail():

tests/test_checklist.py

@@ -65,6 +65,22 @@ def test_build_system_prompt_tool_hint_toggle():
     assert "只读" not in without
 
 
+def test_build_system_prompt_requires_structured_json_output():
+    # allow_tools=False 以隔离「工具提示块」里的「只读」字样
+    prompt = build_system_prompt(DEFAULT_CHECKLIST, allow_tools=False)
+    assert "JSON" in prompt
+    assert '"summary"' in prompt
+    assert '"issues"' in prompt
+    assert '"severity"' in prompt and '"category"' in prompt and '"handle"' in prompt
+    # 结构化契约本身不得含「只读」，否则会破坏 allow_tools 开关的判别
+    assert "只读" not in prompt
+
+
+def test_build_system_prompt_forbids_fabricated_handle():
+    prompt = build_system_prompt(DEFAULT_CHECKLIST)
+    assert "不要编造" in prompt
+
+
 def test_checklist_summary_mentions_name_and_count():
     summary = checklist_summary(DEFAULT_CHECKLIST)
     assert DEFAULT_CHECKLIST["name"] in summary

tests/test_checks.py

@@ -0,0 +1,212 @@
+"""`dwg_review.checks` 单元测试：确定性检查前置。"""
+
+from dwg_review import checks
+from dwg_review.checks import (
+    check_duplicate_entities,
+    check_zero_length,
+    derive_from_evidence,
+    make_issue,
+    run_deterministic_checks,
+)
+from dwg_review.mcp_client import McpError, McpToolError
+from tests.fakes import FakeMcpClient, make_ctx, make_evidence, ok_text
+
+QUIET = lambda *a: None  # noqa: E731
+WINDOW = {"min_x": 0.0, "min_y": 0.0, "max_x": 10.0, "max_y": 10.0}
+
+
+# ---------------------------------------------------------------------------
+#  make_issue
+# ---------------------------------------------------------------------------
+def test_make_issue_minimal():
+    issue = make_issue("几何", "medium", "零长度")
+    assert issue == {"category": "几何", "severity": "medium",
+                     "message": "零长度", "source": "deterministic"}
+
+
+def test_make_issue_with_handles():
+    issue = make_issue("几何", "low", "重复", handle="A", handles=["A", "B"])
+    assert issue["handle"] == "A"
+    assert issue["handles"] == ["A", "B"]
+
+
+def test_make_issue_omits_empty_handle():
+    assert "handle" not in make_issue("文字", "low", "空文字", handle=None)
+
+
+# ---------------------------------------------------------------------------
+#  check_zero_length
+# ---------------------------------------------------------------------------
+def test_zero_length_turns_items_into_issues():
+    client = FakeMcpClient(responses={"find_zero_length": ok_text({
+        "total": 2, "returned": 2, "offset": 0,
+        "items": [
+            {"handle": "A1", "type": "LINE", "layer": "0", "length": 0.0},
+            {"handle": "A2", "type": "ARC", "layer": "0", "length": 0.0001},
+        ],
+    })})
+    ctx = make_ctx(client=client)
+    issues = check_zero_length(ctx, WINDOW, tolerance=0.001, limit=200, log=QUIET)
+    assert [i["handle"] for i in issues] == ["A1", "A2"]
+    assert all(i["category"] == "几何" and i["severity"] == "medium" for i in issues)
+
+
+def test_zero_length_skips_items_without_handle():
+    client = FakeMcpClient(responses={"find_zero_length": ok_text(
+        {"items": [{"type": "LINE"}, {"handle": "B1"}]}
+    )})
+    ctx = make_ctx(client=client)
+    assert [i["handle"] for i in check_zero_length(ctx, WINDOW, tolerance=0.001,
+                                                   limit=200, log=QUIET)] == ["B1"]
+
+
+def test_zero_length_passes_window_and_params():
+    client = FakeMcpClient(responses={"find_zero_length": ok_text({"items": []})})
+    ctx = make_ctx(client=client)
+    check_zero_length(ctx, WINDOW, tolerance=0.5, limit=77, log=QUIET)
+    assert client.arguments_of("find_zero_length") == {
+        "window": WINDOW, "tolerance": 0.5, "limit": 77,
+    }
+
+
+def test_zero_length_swallows_tool_error():
+    logs = []
+    client = FakeMcpClient(responses={"find_zero_length": McpToolError("炸了")})
+    ctx = make_ctx(client=client)
+    assert check_zero_length(ctx, WINDOW, tolerance=0.001, limit=200, log=logs.append) == []
+    assert any("find_zero_length 失败" in line for line in logs)
+
+
+def test_zero_length_tolerates_non_dict_envelope():
+    client = FakeMcpClient(responses={"find_zero_length": ok_text("oops")})
+    ctx = make_ctx(client=client)
+    assert check_zero_length(ctx, WINDOW, tolerance=0.001, limit=200, log=QUIET) == []
+
+
+# ---------------------------------------------------------------------------
+#  check_duplicate_entities
+# ---------------------------------------------------------------------------
+def test_duplicate_group_becomes_single_issue_with_all_handles():
+    client = FakeMcpClient(responses={"find_duplicate_entities": ok_text({
+        "count": 1,
+        "groups": [{"reason": "identical_geometry", "type": "LINE",
+                    "handles": ["B1", "B2", "B3"]}],
+    })})
+    ctx = make_ctx(client=client)
+    issues = check_duplicate_entities(ctx, WINDOW, tolerance=0.001, group_limit=50, log=QUIET)
+    assert len(issues) == 1  # 一组 = 一个问题
+    assert issues[0]["handle"] == "B1"
+    assert issues[0]["handles"] == ["B1", "B2", "B3"]
+    assert issues[0]["severity"] == "low"
+
+
+def test_duplicate_skips_empty_groups():
+    client = FakeMcpClient(responses={"find_duplicate_entities": ok_text({
+        "groups": [{"handles": []}, {"handles": None}, {"handles": ["C1", "C2"]}],
+    })})
+    ctx = make_ctx(client=client)
+    issues = check_duplicate_entities(ctx, WINDOW, tolerance=0.001, group_limit=50, log=QUIET)
+    assert [i["handle"] for i in issues] == ["C1"]
+
+
+def test_duplicate_passes_group_limit():
+    client = FakeMcpClient(responses={"find_duplicate_entities": ok_text({"groups": []})})
+    ctx = make_ctx(client=client)
+    check_duplicate_entities(ctx, WINDOW, tolerance=0.2, group_limit=9, log=QUIET)
+    assert client.arguments_of("find_duplicate_entities") == {
+        "window": WINDOW, "tolerance": 0.2, "limit": 9,
+    }
+
+
+def test_duplicate_swallows_tool_error():
+    logs = []
+    client = FakeMcpClient(responses={"find_duplicate_entities": McpError(-32603, "x")})
+    ctx = make_ctx(client=client)
+    assert check_duplicate_entities(ctx, WINDOW, tolerance=0.001, group_limit=50,
+                                    log=logs.append) == []
+    assert any("find_duplicate_entities 失败" in line for line in logs)
+
+
+# ---------------------------------------------------------------------------
+#  derive_from_evidence
+# ---------------------------------------------------------------------------
+def test_derive_flags_empty_and_whitespace_texts():
+    ev = make_evidence(texts={"items": [
+        {"handle": "T1", "content": ""},
+        {"handle": "T2", "content": "   "},
+        {"handle": "T3", "content": None},
+        {"handle": "T4", "content": "正常文字"},
+    ]})
+    issues = derive_from_evidence(ev)
+    assert [i["handle"] for i in issues] == ["T1", "T2", "T3"]
+    assert all(i["category"] == "文字" for i in issues)
+
+
+def test_derive_flags_unclosed_polyline():
+    ev = make_evidence(polylines={"items": [
+        {"handle": "P1", "vertex_count": 4, "closed": False},
+        {"handle": "P2", "vertex_count": 4, "closed": True},
+        {"handle": "P3", "vertex_count": 2, "closed": False},
+        {"handle": "P4", "closed": False},
+    ]})
+    issues = derive_from_evidence(ev)
+    assert [i["handle"] for i in issues] == ["P1"]
+
+
+def test_derive_can_disable_unclosed_check():
+    ev = make_evidence(polylines={"items": [{"handle": "P1", "vertex_count": 4, "closed": False}]})
+    assert derive_from_evidence(ev, check_unclosed=False) == []
+
+
+def test_derive_tolerates_missing_sources():
+    assert derive_from_evidence({}) == []
+    assert derive_from_evidence(make_evidence()) == []
+
+
+# ---------------------------------------------------------------------------
+#  run_deterministic_checks
+# ---------------------------------------------------------------------------
+def build_ctx():
+    client = FakeMcpClient(responses={
+        "find_zero_length": ok_text({"items": [{"handle": "Z1", "type": "LINE", "length": 0.0}]}),
+        "find_duplicate_entities": ok_text({"groups": [
+            {"type": "LINE", "handles": ["D1", "D2"]}]}),
+    })
+    return make_ctx(client=client)
+
+
+def test_run_combines_derived_and_mcp():
+    ctx = build_ctx()
+    ev = make_evidence(
+        texts={"items": [{"handle": "T1", "content": ""}]},
+        polylines={"items": [{"handle": "P1", "vertex_count": 3, "closed": False}]},
+    )
+    issues = run_deterministic_checks(ctx, ev, log=QUIET)
+    by_handle = {i.get("handle"): i for i in issues}
+    assert set(by_handle) == {"T1", "P1", "Z1", "D1"}
+    assert by_handle["D1"]["handles"] == ["D1", "D2"]
+
+
+def test_run_passes_window_to_mcp_tools():
+    ctx = build_ctx()
+    ev = make_evidence(bbox=(0.0, 0.0, 10.0, 10.0))
+    run_deterministic_checks(ctx, ev, log=QUIET)
+    assert ctx.client.arguments_of("find_zero_length")["window"] == ev["window"]
+    assert ctx.client.arguments_of("find_duplicate_entities")["window"] == ev["window"]
+
+
+def test_run_without_window_skips_mcp_but_keeps_derived():
+    ctx = build_ctx()
+    ev = make_evidence()
+    ev["window"] = None
+    ev["texts"] = {"items": [{"handle": "T1", "content": ""}]}
+    issues = run_deterministic_checks(ctx, ev, log=QUIET)
+    assert [i["handle"] for i in issues] == ["T1"]
+    assert ctx.client.calls == []
+
+
+def test_run_respects_unclosed_switch(monkeypatch):
+    monkeypatch.setattr(checks.config, "CHECK_UNCLOSED_POLYLINES", False)
+    ctx = build_ctx()
+    ev = make_evidence(polylines={"items": [{"handle": "P1", "vertex_count": 4, "closed": False}]})
+    assert [i["handle"] for i in run_deterministic_checks(ctx, ev, log=QUIET)] == ["Z1", "D1"]

tests/test_config.py

@@ -27,6 +27,26 @@ def test_read_env_missing_returns_empty(monkeypatch):
     assert config._read_env("CAD_TEST_DEFINITELY_MISSING_XYZ") == ""
 
 
+def test_read_bool_defaults_when_unset(monkeypatch):
+    monkeypatch.delenv("CAD_TEST_BOOL_UNSET", raising=False)
+    assert config._read_bool("CAD_TEST_BOOL_UNSET", True) is True
+    assert config._read_bool("CAD_TEST_BOOL_UNSET", False) is False
+
+
+def test_read_bool_empty_string_uses_default(monkeypatch):
+    monkeypatch.setenv("CAD_TEST_BOOL_EMPTY", "  ")
+    assert config._read_bool("CAD_TEST_BOOL_EMPTY", True) is True
+
+
+def test_read_bool_truthy_and_falsy(monkeypatch):
+    for raw in ("1", "true", "TRUE", "Yes", "on"):
+        monkeypatch.setenv("CAD_TEST_BOOL", raw)
+        assert config._read_bool("CAD_TEST_BOOL", False) is True
+    for raw in ("0", "false", "no", "off", "nope"):
+        monkeypatch.setenv("CAD_TEST_BOOL", raw)
+        assert config._read_bool("CAD_TEST_BOOL", True) is False
+
+
 def test_has_api_key_reflects_module_value(monkeypatch):
     monkeypatch.setattr(config, "API_KEY", "")
     assert config.has_api_key() is False
@@ -50,6 +70,22 @@ def test_constants_have_expected_types():
     assert isinstance(config.CHECKLIST_PATH, type(config.PROJECT_ROOT))
 
 
+def test_credibility_constants_have_expected_types():
+    assert isinstance(config.DETERMINISTIC_CHECKS, bool)
+    assert isinstance(config.VALIDATE_HANDLES, bool)
+    assert isinstance(config.CHECK_UNCLOSED_POLYLINES, bool)
+    assert isinstance(config.ZERO_LENGTH_TOLERANCE, float)
+    assert isinstance(config.DUP_TOLERANCE, float)
+    assert isinstance(config.DUP_GROUP_LIMIT, int)
+
+
+def test_credibility_defaults_are_on():
+    # 默认开启，保证默认拿到可信度收益
+    assert config.DETERMINISTIC_CHECKS is True
+    assert config.VALIDATE_HANDLES is True
+    assert config.CHECK_UNCLOSED_POLYLINES is True
+
+
 def test_checklist_path_resolves_relative_to_root():
     assert config.CHECKLIST_PATH.name == "review_checklist.json"
     assert config.CHECKLIST_PATH.parent == config.PROJECT_ROOT

tests/test_findings.py

@@ -0,0 +1,241 @@
+"""`dwg_review.findings` 单元测试：结构化解析 + handle 交叉校验 + 装配。"""
+
+from dwg_review.findings import (
+    build_finding,
+    collect_known_handles,
+    extract_json_block,
+    flatten_issues,
+    normalize_issue,
+    parse_conclusion,
+)
+from tests.fakes import make_evidence
+
+QUIET = lambda *a: None  # noqa: E731
+
+
+# ---------------------------------------------------------------------------
+#  extract_json_block
+# ---------------------------------------------------------------------------
+def test_extract_bare_object():
+    assert extract_json_block('{"a": 1}') == '{"a": 1}'
+
+
+def test_extract_bare_array():
+    assert extract_json_block('[{"a": 1}]') == '[{"a": 1}]'
+
+
+def test_extract_fenced_object():
+    assert extract_json_block('```json\n{"a": 1}\n```') == '{"a": 1}'
+
+
+def test_extract_single_line_fence():
+    # 旧 extract_issues 的已知瑕疵（单行围栏剥离失败）在此不再复现
+    assert extract_json_block('```json{"a": 1}```') == '{"a": 1}'
+
+
+def test_extract_from_surrounding_prose():
+    text = '好的，结论如下：\n{"summary": "ok", "issues": []}\n以上。'
+    assert extract_json_block(text) == '{"summary": "ok", "issues": []}'
+
+
+def test_extract_handles_braces_inside_strings():
+    text = 'prefix {"message": "包含 } 和 { 的文字", "n": 1} suffix'
+    assert extract_json_block(text) == '{"message": "包含 } 和 { 的文字", "n": 1}'
+
+
+def test_extract_returns_none_for_garbage():
+    assert extract_json_block("未发现明显问题") is None
+    assert extract_json_block("") is None
+    assert extract_json_block(None) is None
+    assert extract_json_block("{ not json") is None
+
+
+# ---------------------------------------------------------------------------
+#  parse_conclusion
+# ---------------------------------------------------------------------------
+def test_parse_object_with_issues():
+    text = '{"summary": "整体尚可", "issues": [{"category": "几何", "message": "x"}]}'
+    out = parse_conclusion(text, log=QUIET)
+    assert out["summary"] == "整体尚可"
+    assert out["issues"] == [{"category": "几何", "message": "x"}]
+    assert out["raw"] == text
+
+
+def test_parse_object_without_issues_key():
+    out = parse_conclusion('{"summary": "无问题"}', log=QUIET)
+    assert out["summary"] == "无问题"
+    assert out["issues"] == []
+
+
+def test_parse_bare_array_becomes_issues():
+    out = parse_conclusion('[{"message": "a"}, {"message": "b"}]', log=QUIET)
+    assert out["summary"] == ""
+    assert [i["message"] for i in out["issues"]] == ["a", "b"]
+
+
+def test_parse_filters_non_dict_issues():
+    out = parse_conclusion('{"summary": "s", "issues": [{"message": "ok"}, "junk", 3]}', log=QUIET)
+    assert out["issues"] == [{"message": "ok"}]
+
+
+def test_parse_degrades_prose_to_summary():
+    logs = []
+    out = parse_conclusion("未发现明显问题。", log=logs.append)
+    assert out["summary"] == "未发现明显问题。"
+    assert out["issues"] == []
+    assert any("降级" in line for line in logs)
+
+
+def test_parse_degrades_empty_text():
+    out = parse_conclusion("", log=QUIET)
+    assert out["summary"] == "" and out["issues"] == []
+
+
+# ---------------------------------------------------------------------------
+#  collect_known_handles
+# ---------------------------------------------------------------------------
+def test_collect_known_handles_from_all_sources():
+    ev = make_evidence(
+        entities={"total": 1, "items": [{"handle": "AA"}]},
+        texts={"total": 1, "items": [{"handle": "BB"}]},
+        dimensions={"total": 1, "items": [{"handle": "CC"}]},
+        blocks={"total": 1, "items": [{"handle": "DD"}]},
+        polylines={"total": 1, "items": [{"handle": "EE"}]},
+    )
+    assert collect_known_handles(ev) == {"AA", "BB", "CC", "DD", "EE"}
+
+
+def test_collect_known_handles_uppercases():
+    ev = make_evidence(entities={"items": [{"handle": "2f3a"}]})
+    assert collect_known_handles(ev) == {"2F3A"}
+
+
+def test_collect_known_handles_includes_extra():
+    ev = make_evidence()
+    assert collect_known_handles(ev, extra_handles=["Ff", None, ""]) == {"FF"}
+
+
+def test_collect_known_handles_tolerates_non_dict_envelope():
+    # result_json 解析失败会返回原始字符串
+    assert collect_known_handles({"entities": "oops", "texts": None}) == set()
+
+
+def test_collect_known_handles_skips_items_without_handle():
+    ev = make_evidence(entities={"items": [{"type": "LINE"}, {"handle": ""}]})
+    assert collect_known_handles(ev) == set()
+
+
+# ---------------------------------------------------------------------------
+#  normalize_issue
+# ---------------------------------------------------------------------------
+def test_normalize_llm_issue_with_known_handle_is_verified():
+    out = normalize_issue(
+        {"category": "文字", "severity": "high", "message": "重叠", "handle": "2f3a"},
+        frame_index=1, source="llm", known_handles={"2F3A"}, validate_handles=True,
+    )
+    assert out["handle"] == "2f3a"  # 保留原样（大小写归一仅用于比对）
+    assert out["verified"] is True
+    assert out["source"] == "llm"
+    assert out["frame"] == 1
+
+
+def test_normalize_llm_issue_with_unknown_handle_strips_it():
+    out = normalize_issue(
+        {"message": "疑似问题", "handle": "DEAD"},
+        frame_index=2, source="llm", known_handles={"2F3A"}, validate_handles=True,
+    )
+    assert out["handle"] is None
+    assert out["verified"] is False
+    assert out["message"] == "疑似问题"  # 问题保留
+
+
+def test_normalize_deterministic_never_stripped():
+    out = normalize_issue(
+        {"message": "零长度", "handle": "ZZZZ"},
+        frame_index=1, source="deterministic", known_handles=set(), validate_handles=True,
+    )
+    assert out["handle"] == "ZZZZ" and out["verified"] is True
+
+
+def test_normalize_validation_disabled_keeps_handle():
+    out = normalize_issue(
+        {"message": "x", "handle": "DEAD"},
+        frame_index=1, source="llm", known_handles=set(), validate_handles=False,
+    )
+    assert out["handle"] == "DEAD" and out["verified"] is True
+
+
+def test_normalize_llm_issue_without_handle_is_verified():
+    out = normalize_issue({"message": "整体偏空"}, frame_index=1, source="llm",
+                          known_handles=set(), validate_handles=True)
+    assert out["handle"] is None and out["verified"] is True
+
+
+def test_normalize_falls_back_on_invalid_severity_and_category():
+    out = normalize_issue({"severity": "critical", "message": "x"},
+                          frame_index=1, source="llm", known_handles=set(),
+                          validate_handles=True)
+    assert out["severity"] == "low"
+    assert out["category"] == "图面整体"
+
+
+def test_normalize_preserves_handles_list():
+    out = normalize_issue({"message": "重复组", "handle": "A", "handles": ["A", "B"]},
+                          frame_index=1, source="deterministic", known_handles=set(),
+                          validate_handles=True)
+    assert out["handles"] == ["A", "B"]
+
+
+# ---------------------------------------------------------------------------
+#  build_finding
+# ---------------------------------------------------------------------------
+def test_build_finding_merges_deterministic_then_llm():
+    ev = make_evidence(index=3, entities={"items": [{"handle": "AA"}]})
+    checks = [{"category": "几何", "severity": "medium", "message": "零长度", "handle": "ZZ"}]
+    conclusion = {
+        "summary": "有 2 处问题",
+        "issues": [
+            {"category": "文字", "severity": "high", "message": "重叠", "handle": "AA"},
+            {"category": "几何", "severity": "low", "message": "臆测", "handle": "NOPE"},
+        ],
+    }
+    finding = build_finding(ev, conclusion, checks, validate_handles=True, log=QUIET)
+    assert finding["frame"] == 3
+    assert finding["sheet"] == "A1"
+    assert finding["summary"] == "有 2 处问题"
+    assert [i["source"] for i in finding["issues"]] == ["deterministic", "llm", "llm"]
+    assert finding["counts"] == {"deterministic": 1, "llm": 2, "unverified": 1}
+    # 被剥掉 handle 的那条仍在
+    assert finding["issues"][2]["verified"] is False
+
+
+def test_build_finding_deterministic_handle_counts_as_known():
+    # 确定性检查命中的 handle 应进入已知集合，供同框 LLM 引用
+    ev = make_evidence(entities={"items": []})
+    checks = [{"message": "零长度", "handle": "ZZ"}]
+    conclusion = {"summary": "", "issues": [{"message": "同一条", "handle": "zz"}]}
+    finding = build_finding(ev, conclusion, checks, validate_handles=True, log=QUIET)
+    assert finding["issues"][1]["verified"] is True
+
+
+def test_build_finding_without_checks_or_conclusion():
+    finding = build_finding(make_evidence(), {}, None, log=QUIET)
+    assert finding["issues"] == []
+    assert finding["counts"] == {"deterministic": 0, "llm": 0, "unverified": 0}
+
+
+# ---------------------------------------------------------------------------
+#  flatten_issues
+# ---------------------------------------------------------------------------
+def test_flatten_issues_concatenates():
+    findings = [
+        {"issues": [{"message": "a"}, {"message": "b"}]},
+        {"issues": [{"message": "c"}]},
+        {},
+    ]
+    assert [i["message"] for i in flatten_issues(findings)] == ["a", "b", "c"]
+
+
+def test_flatten_issues_empty():
+    assert flatten_issues([]) == []
+    assert flatten_issues(None) == []

tests/test_report.py

@@ -2,19 +2,64 @@
 
 import json
 
-import pytest
-
-from dwg_review.phases.report import extract_issues, save_report, summarize
+from dwg_review.phases.report import format_finding, save_report, summarize
 from tests.fakes import FakeLlmClient, make_ctx
 
 
+def issue(**over):
+    base = {
+        "frame": 1, "category": "文字", "severity": "high", "message": "文字重叠",
+        "handle": "A1", "source": "llm", "verified": True,
+    }
+    base.update(over)
+    return base
+
+
 def findings_fixture():
     return [
-        {"frame": 1, "sheet": "A1", "finding": "图框1：文字重叠"},
-        {"frame": 2, "sheet": "A3", "finding": "图框2：未发现明显问题"},
+        {
+            "frame": 1, "sheet": "A1", "summary": "该图框存在文字重叠",
+            "issues": [issue()],
+            "counts": {"deterministic": 0, "llm": 1, "unverified": 0},
+        },
+        {
+            "frame": 2, "sheet": "A3", "summary": "未发现明显问题",
+            "issues": [],
+            "counts": {"deterministic": 0, "llm": 0, "unverified": 0},
+        },
     ]
 
 
+# ---------------------------------------------------------------------------
+#  format_finding
+# ---------------------------------------------------------------------------
+def test_format_finding_renders_summary_and_issues():
+    text = format_finding(findings_fixture()[0])
+    assert "图框 1" in text and "图幅 A1" in text
+    assert "该图框存在文字重叠" in text
+    assert "[high][LLM] 文字重叠" in text
+    assert "handle=A1" in text
+
+
+def test_format_finding_marks_deterministic_source():
+    finding = {"frame": 1, "sheet": "A1", "summary": "s",
+               "issues": [issue(source="deterministic", handle="Z1")]}
+    assert "[确定性]" in format_finding(finding)
+
+
+def test_format_finding_marks_unverified():
+    finding = {"frame": 1, "sheet": "A1", "summary": "s",
+               "issues": [issue(handle=None, verified=False)]}
+    text = format_finding(finding)
+    assert "未在证据中定位" in text
+    assert "handle=" not in text
+
+
+def test_format_finding_tolerates_missing_keys():
+    text = format_finding({})
+    assert "图框 None" in text  # 不再 KeyError（修复既有硬下标 gap）
+
+
 # ---------------------------------------------------------------------------
 #  summarize
 # ---------------------------------------------------------------------------
@@ -32,6 +77,15 @@ def test_summarize_uses_chat_text():
     assert summarize(ctx, findings_fixture(), log=lambda *a: None) == "## 总报告"
     assert len(llm.calls) == 1
     assert "图框 1" in llm.calls[0]["user"]
+    assert "图框 2" in llm.calls[0]["user"]
+
+
+def test_summarize_input_includes_issue_lines():
+    llm = FakeLlmClient(text="x")
+    ctx = make_ctx(llm=llm)
+    summarize(ctx, findings_fixture(), log=lambda *a: None)
+    user = llm.calls[0]["user"]
+    assert "[high][LLM] 文字重叠" in user
 
 
 def test_summarize_falls_back_to_raw_input_on_error():
@@ -46,66 +100,12 @@ def test_summarize_falls_back_to_raw_input_on_error():
     assert any("汇总失败" in line for line in logs)
 
 
-def test_summarize_missing_frame_key_known_gap():
-    """已知瑕疵：summarize 用硬下标 f['frame']，findings 缺该键直接 KeyError。"""
-    ctx = make_ctx(llm=FakeLlmClient(text="x"))
-    with pytest.raises(KeyError):
-        summarize(ctx, [{"finding": "缺 frame 键"}], log=lambda *a: None)
-
-
-# ---------------------------------------------------------------------------
-#  extract_issues
-# ---------------------------------------------------------------------------
-def test_extract_issues_empty_returns_empty_list():
-    llm = FakeLlmClient(text_script=["[]"])
+def test_summarize_tolerates_finding_without_frame_key():
+    # 修复既有硬下标 gap：缺 frame 键不再 KeyError
+    llm = FakeLlmClient(text="ok")
     ctx = make_ctx(llm=llm)
-    assert extract_issues(ctx, [], log=lambda *a: None) == []
-    assert llm.calls == []
-
-
-def test_extract_issues_parses_json_array():
-    payload = '[{"frame": 1, "handle": "A", "severity": "high", "message": "x"}]'
-    llm = FakeLlmClient(text_script=[payload])
-    ctx = make_ctx(llm=llm)
-    issues = extract_issues(ctx, findings_fixture(), log=lambda *a: None)
-    assert issues == [{"frame": 1, "handle": "A", "severity": "high", "message": "x"}]
-
-
-def test_extract_issues_strips_code_fence():
-    fenced = '```json\n[{"frame": 1, "message": "y"}]\n```'
-    llm = FakeLlmClient(text_script=[fenced])
-    ctx = make_ctx(llm=llm)
-    assert extract_issues(ctx, findings_fixture(), log=lambda *a: None) == [
-        {"frame": 1, "message": "y"}
-    ]
-
-
-def test_extract_issues_returns_empty_on_invalid_json():
-    logs = []
-    llm = FakeLlmClient(text_script=["这不是 JSON"])
-    ctx = make_ctx(llm=llm)
-    assert extract_issues(ctx, findings_fixture(), log=logs.append) == []
-    assert any("结构化抽取失败" in line for line in logs)
-
-
-def test_extract_issues_returns_empty_when_not_a_list():
-    llm = FakeLlmClient(text_script=['{"frame": 1}'])
-    ctx = make_ctx(llm=llm)
-    assert extract_issues(ctx, findings_fixture(), log=lambda *a: None) == []
-
-
-def test_extract_issues_single_line_fence_known_gap():
-    """已知瑕疵：围栏剥离依赖换行，单行 ```json[...]``` 会残留 json 前缀 → 解析失败。"""
-    llm = FakeLlmClient(text_script=['```json[{"frame": 1, "message": "z"}]```'])
-    ctx = make_ctx(llm=llm)
-    assert extract_issues(ctx, findings_fixture(), log=lambda *a: None) == []
-
-
-def test_extract_issues_missing_frame_key_known_gap():
-    """已知瑕疵：extract_issues 同样用硬下标 f['frame']，findings 缺该键直接 KeyError。"""
-    ctx = make_ctx(llm=FakeLlmClient(text="[]"))
-    with pytest.raises(KeyError):
-        extract_issues(ctx, [{"finding": "缺 frame 键"}], log=lambda *a: None)
+    out = summarize(ctx, [{"summary": "无 frame 键", "issues": []}], log=lambda *a: None)
+    assert out == "ok"
 
 
 # ---------------------------------------------------------------------------
@@ -114,7 +114,7 @@ def test_extract_issues_missing_frame_key_known_gap():
 def test_save_report_writes_md_and_json(tmp_reports):
     findings = findings_fixture()
     plan = {"frames": [{"bbox": [0, 0, 10, 10]}]}
-    issues = [{"frame": 1, "handle": "A", "message": "x"}]
+    issues = [issue()]
 
     md_path, json_path = save_report("# 报告", findings, plan, issues, log=lambda *a: None)
 
@@ -127,9 +127,11 @@ def test_save_report_writes_md_and_json(tmp_reports):
     assert "generated_at" in data
 
 
-def test_save_report_issues_default_to_empty(tmp_reports):
-    _, json_path = save_report("md", findings_fixture(), {"frames": []}, log=lambda *a: None)
-    assert json.loads(json_path.read_text(encoding="utf-8"))["issues"] == []
+def test_save_report_derives_issues_from_findings(tmp_reports):
+    findings = findings_fixture()
+    _, json_path = save_report("md", findings, {"frames": []}, log=lambda *a: None)
+    data = json.loads(json_path.read_text(encoding="utf-8"))
+    assert data["issues"] == [issue()]  # 拍平自 findings
 
 
 def test_save_report_creates_missing_directory(tmp_path, monkeypatch):

tests/test_review.py

@@ -111,11 +111,12 @@ def test_readonly_empty_response_marker():
 # ---------------------------------------------------------------------------
 #  review_frame
 # ---------------------------------------------------------------------------
-def test_review_frame_returns_text_when_no_tools():
+def test_review_frame_returns_summary_when_no_tools():
     llm = FakeLlmClient(script=[llm_text("结论A")])
     ctx = make_ctx(client=FakeMcpClient(), llm=llm)
     out = review_frame(ctx, make_evidence(), "系统提示", log=lambda *a: None)
-    assert out == "结论A"
+    assert out["summary"] == "结论A"  # 非 JSON → 降级为纯文本 summary
+    assert out["issues"] == []
     assert len(llm.calls) == 1
     assert llm.calls[0]["tools"] is not None  # 传了只读工具 schema
 
@@ -147,7 +148,7 @@ def test_review_frame_runs_tool_then_concludes():
     client = FakeMcpClient(responses={"query_entities": ok_text({"total": 1})})
     ctx = make_ctx(client=client, llm=llm)
     out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
-    assert out == "结论B"
+    assert out["summary"] == "结论B"
     assert len(llm.calls) == 2
     assert client.called_names() == ["query_entities"]
 
@@ -179,7 +180,7 @@ def test_review_frame_force_closes_at_iteration_limit():
     )
     ctx = make_ctx(client=FakeMcpClient(), llm=llm, max_iters=1)
     out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
-    assert out == "收口结论"
+    assert out["summary"] == "收口结论"
     last_msg = llm.calls[-1]["messages"][-1]
     assert last_msg["role"] == "user"
     assert "上限" in last_msg["content"]
@@ -189,7 +190,7 @@ def test_review_frame_with_zero_iters_forces_immediate_close():
     llm = FakeLlmClient(script=[llm_text("直接结论")])
     ctx = make_ctx(client=FakeMcpClient(), llm=llm, max_iters=0)
     out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
-    assert out == "直接结论"
+    assert out["summary"] == "直接结论"
     assert len(llm.calls) == 1
 
 
@@ -207,3 +208,38 @@ def test_review_frame_payload_is_json_serializable():
     review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
     text = llm.calls[0]["messages"][1]["content"][0]["text"]
     json.loads(text.split("：\n", 1)[1])  # 不抛异常即通过
+
+
+def test_review_frame_parses_structured_json_conclusion():
+    text = (
+        '{"summary": "整体尚可", "issues": ['
+        '{"category": "几何", "severity": "medium", "message": "零长度", "handle": "A1"}]}'
+    )
+    llm = FakeLlmClient(script=[llm_text(text)])
+    ctx = make_ctx(client=FakeMcpClient(), llm=llm)
+    out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
+    assert out["summary"] == "整体尚可"
+    assert out["issues"] == [
+        {"category": "几何", "severity": "medium", "message": "零长度", "handle": "A1"}
+    ]
+
+
+def test_review_frame_parses_fenced_json():
+    llm = FakeLlmClient(script=[llm_text('```json\n{"summary": "s", "issues": []}\n```')])
+    ctx = make_ctx(client=FakeMcpClient(), llm=llm)
+    out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
+    assert out["summary"] == "s"
+    assert out["issues"] == []
+
+
+def test_review_frame_intermediate_tool_turn_json_not_parsed():
+    # 中间工具回合的 content 里即使塞了 JSON，也不能被当成最终结论
+    intermediate = '{"summary": "这不是最终结论", "issues": [{"message": "x"}]}'
+    llm = FakeLlmClient(script=[
+        llm_calls([("query_entities", {})], content=intermediate),
+        llm_text('{"summary": "最终结论", "issues": []}'),
+    ])
+    ctx = make_ctx(client=FakeMcpClient(), llm=llm)
+    out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
+    assert out["summary"] == "最终结论"
+    assert out["issues"] == []