feat: D2落库核验硬门禁——approved任务产出未落库抛人工（防只写文件不写库静默卡死）

pipeline_bidding/bid_flow.py

@@ -88,6 +88,14 @@ ANALYSIS_DIMS = (
 DIM_QC_TYPES = {d: ts for d, _, ts in ANALYSIS_DIMS}
 QC_TYPE_TO_DIM = {q: d for d, _, ts in ANALYSIS_DIMS for q in ts}
 
+# 维度 → 落库表（落库核验用；reqs_outline 维度同时写两张表）
+DIM_TABLES = {
+    "scoring": (QC_OUTPUT_TABLE[QC_TYPE_SCORING],),
+    "quals": (QC_OUTPUT_TABLE[QC_TYPE_QUALS],),
+    "reqs_outline": (QC_OUTPUT_TABLE[QC_TYPE_REQS], QC_OUTPUT_TABLE[QC_TYPE_OUTLINE]),
+    "cost_benefit": (QC_OUTPUT_TABLE[QC_TYPE_COST],),
+}
+
 # 章节级上游依赖矩阵（2026-09-03）：每种章节 section 只依赖自己必需的分析产出
 # 通过 QC——某一维度 QC 卡住只阻塞依赖它的章节，其余章节照常流转。
 # 公共事实：章节骨架（bid_chapters）本身就是章节，OUTLINE 过了章节才可信；
@@ -710,6 +718,61 @@ async def reconcile_project(sor, project_id, project_name=""):
     if created_r:
         await sor.sqlExe("COMMIT", {})
 
+    # ── D2. 落库核验（2026-09-03，机制硬门禁）──
+    # 实测根因：修正任务 approved 但产出只写文件没落库（bid_chapters 停在旧时间），
+    # 机制对旧数据不复审、流程静默卡死。核验：approved 的分析/修正任务，
+    # 对应产出表 max 更新时间必须 ≥ 任务创建时间，否则抛人工补落库/重跑。
+    verify_bad = []
+    _vtasks = await sor.sqlExe(
+        "SELECT id, title, params, created_at FROM pipeline_tasks WHERE tenant_id=${p}$ "
+        "AND role=${r}$ AND state='approved'", {"p": project_id, "r": R_ANALYST}) or []
+    for t in _vtasks:
+        try:
+            p = json.loads(getattr(t, 'params', '') or '{}')
+        except Exception:
+            p = {}
+        dim = (p.get('analysis_dim') or '').strip()
+        title = getattr(t, 'title', '') or ''
+        is_manual = ('修正' in title or '补录' in title) and not p.get('qc_redo')
+        if not dim and not is_manual:
+            continue
+        if dim:
+            tbls = DIM_TABLES.get(dim) or (QC_OUTPUT_TABLE[QC_TYPE_REQS], QC_OUTPUT_TABLE[QC_TYPE_OUTLINE])
+        else:
+            # 人工修正任务按标题关键词定目标表，避免跨维度误判
+            if 'chapter_outline' in title or '骨架' in title:
+                tbls = (QC_OUTPUT_TABLE[QC_TYPE_OUTLINE],)
+            elif 'doc_requirements' in title or '文件要求' in title:
+                tbls = (QC_OUTPUT_TABLE[QC_TYPE_REQS],)
+            else:
+                tbls = (QC_OUTPUT_TABLE[QC_TYPE_REQS], QC_OUTPUT_TABLE[QC_TYPE_OUTLINE])
+        t_created = str(getattr(t, 'created_at', '') or '')
+        fresh = False
+        for tbl in tbls:
+            rec = await sor.sqlExe(
+                "SELECT MAX(GREATEST(created_at, IFNULL(updated_at, created_at))) AS mt FROM " + tbl +
+                " WHERE project_id=${p}$", {"p": project_id})
+            mt = str(getattr(rec[0], 'mt', '') or '') if rec else ''
+            if mt and mt >= t_created:
+                fresh = True
+                break
+        if not fresh:
+            verify_bad.append((getattr(t, 'id', ''), title[:50], dim or 'reqs_outline'))
+    if verify_bad:
+        ht = await find_human_task(sor, project_id, "output_verify", status="pending")
+        if not ht:
+            await create_human_task(
+                sor, project_id, "output_verify",
+                "分析/修正任务声称完成但产出未落库",
+                ("以下任务已 approved，但对应产出表的最大更新时间早于任务创建时间"
+                 "（只写了文件或根本没写库）：\n%s\n\n"
+                 "请处理：在对应管理页补录落库，或重跑该任务；落库后流程自动复审。"
+                 "处理完完成本任务。"
+                 % "\n".join("- %s [%s] %s" % (tid, dm, ti) for tid, ti, dm in verify_bad)),
+                assignee_role="owner.superuser")
+            await sor.sqlExe("COMMIT", {})
+        acts.append("escalated: 产出落库核验不通过（%d 个任务），抛人工" % len(verify_bad))
+
     if created_w or created_r:
         return acts