feat(deliver): 代码空壳硬门禁+产出文件机械核验——write_file实写的.py必须ast可解析且有真实可执行语句(纯注释/pass/docstring占位/语法错误拒绝,__init__.py豁免),.json必须可解析;deliver files参数先验后写;引擎自动计算每文件字节/行数/语句数回填交付件正文给QC当证据底座(2026-09-16 pbls M1a verify_gate.py声称17141字符实测1225字节纯注释事故根治)

pipeline_service/agent_loop.py

@@ -2396,6 +2396,169 @@ def _validate_stub_docs(space_dir, written_files):
               "若因信息缺失或上游故障无法产出正文，用 ask_question 如实冒泡等待回答，禁止交付占位文档。")
 
 
+_CODE_STUB_EXEMPT_BASENAMES = {'__init__.py', 'conftest.py'}
+
+
+def _py_code_stats(txt):
+    """Python 源码机械核验：返回 (ok, info)。
+
+    ok=False → info 为拒绝原因（解析失败 / 无有效语句的空壳）。
+    ok=True  → info 为「py_compile=OK 有效语句=N」。
+    有效语句 = ast 语句节点，剔除 docstring、pass、...（Ellipsis）——
+    纯注释/pass 占位文件语句数为 0，判空壳。
+    """
+    import ast as _ast
+    try:
+        tree = _ast.parse(txt)
+    except SyntaxError as e:
+        return False, f"py_compile 失败: {e.msg} (line {e.lineno})"
+    except Exception as e:
+        return False, f"py 解析失败: {type(e).__name__}: {str(e)[:80]}"
+    n = 0
+    for node in _ast.walk(tree):
+        if not isinstance(node, _ast.stmt) or isinstance(node, _ast.Module):
+            continue
+        if isinstance(node, _ast.Pass):
+            continue
+        if isinstance(node, _ast.Expr) and isinstance(getattr(node, 'value', None), _ast.Constant):
+            continue  # docstring / ... 字面量
+        if isinstance(node, (_ast.ClassDef, _ast.FunctionDef, _ast.AsyncFunctionDef)):
+            # 函数体只剩 pass/docstring 的类/函数定义 = 占位骨架，不算有效语句
+            if all(isinstance(s, _ast.Pass)
+                   or (isinstance(s, _ast.Expr)
+                       and isinstance(getattr(s, 'value', None), _ast.Constant))
+                   for s in node.body):
+                continue
+        n += 1
+    if n == 0:
+        return False, "全文无可执行语句（纯注释/pass/…占位）"
+    return True, f"py_compile=OK 有效语句={n}"
+
+
+def _check_code_content(path, content):
+    """单个产出文件内容核验。返回 None=通过；str=拒绝原因。"""
+    base = os.path.basename(path or '')
+    if (path or '').endswith('.py'):
+        if not (content or '').strip():
+            if base in _CODE_STUB_EXEMPT_BASENAMES:
+                return None
+            return "空文件（0 字节代码）"
+        ok, info = _py_code_stats(content)
+        if not ok:
+            if base in _CODE_STUB_EXEMPT_BASENAMES and info.startswith("全文无可执行语句"):
+                return None  # __init__.py 允许纯 docstring/空
+            return info
+        return None
+    if (path or '').endswith('.json'):
+        try:
+            json.loads(content or '')
+        except Exception as e:
+            return f"json 解析失败: {str(e)[:100]}"
+        return None
+    return None
+
+
+def _validate_stub_code(space_dir, written_files):
+    """代码空壳硬门禁（2026-09-16 pbls M1a verify_gate.py 事故根治）。
+
+    事故形状：develop 交付的核心门禁脚本 verify_gate.py 声称「已用 write_file 落盘
+    17141 字符，G1~G6 严格退出码门禁」——磁盘实测 9 行纯注释、1225 字节、零代码语句
+    （LLM 把「对文件的描述」当文件内容写了进去）。deliver 入口既有的三个门禁
+    （占位 .md / 配图形态 / 交付类型）全不覆盖代码文件 → 空壳脚本一路放行到 QC，
+    QC 抓到时代价已是 4 轮退回达上限 → fault → 项目 paused。
+
+    处置（确定性，对齐 _validate_stub_docs 模式）：本任务 write_file 实写的
+    .py 必须语法可解析且有真实可执行语句（__init__.py/conftest.py 豁免语句数），
+    .json 必须可解析——否则拒绝 deliver，回填可行动 FAIL（写真内容或 ask_question
+    冒泡），agent 当轮即可重写，不烧 QC 轮次。
+    返回 None=通过；字符串=拒绝原因。
+    """
+    bad = []
+    for f in dict.fromkeys(written_files or []):
+        if not f.endswith(('.py', '.json')):
+            continue
+        try:
+            with open(f, encoding='utf-8', errors='ignore') as fh:
+                txt = fh.read()
+        except Exception:
+            continue  # 读不到=已删/权限问题，交给 git 收口与 QC 核验
+        rel = os.path.relpath(f, space_dir) if f.startswith(space_dir) else f
+        reason = _check_code_content(f, txt)
+        if reason:
+            bad.append(f"{rel}（{reason}）")
+    if not bad:
+        return None
+    return ("FAIL: 交付含空壳/损坏代码文件——「对文件的描述」不是文件内容，声称的字符数与"
+            "实测不符会被 QC 按造假退回。以下文件没有真实可执行内容："
+            + "；".join(bad[:10])
+            + (f"等共 {len(bad)} 个" if len(bad) > 10 else "")
+            + "。请用 write_file 重写完整正文（真实 def/断言/逻辑，不是注释大纲）后重新 "
+              "deliver；若因上下文放不下大文件，拆成多个小文件分批写入；确实无法产出时"
+              "用 ask_question 如实冒泡，禁止交付空壳。")
+
+
+def _validate_code_files_params(files):
+    """deliver params.files（JSON 数组 [{path, content}]）落盘前内容核验。
+
+    与 _validate_stub_code 同一把尺子：先验后写，避免空壳内容落盘后被引擎
+    git 收口「代为提交」变成既成事实。files 为 str 时先尝试 json.loads。
+    返回 None=通过；字符串=拒绝原因。
+    """
+    if isinstance(files, str):
+        try:
+            files = json.loads(files)
+        except Exception:
+            files = []
+    if not isinstance(files, list):
+        return None
+    bad = []
+    for f in files:
+        if not (isinstance(f, dict) and f.get("path") and f.get("content")):
+            continue
+        reason = _check_code_content(f["path"], f["content"])
+        if reason:
+            bad.append(f"{f['path']}（{reason}）")
+    if not bad:
+        return None
+    return ("FAIL: deliver files 含空壳/损坏代码——以下文件没有真实可执行内容："
+            + "；".join(bad[:10])
+            + "。请补全完整正文后重新 deliver（大文件拆小分批 write_file），"
+              "无法产出时 ask_question 冒泡，禁止用描述性注释大纲冒充代码。")
+
+
+def _code_closure_report(space_dir, files):
+    """产出文件机械核验报告（引擎自动计算，非 agent 声明）。
+
+    对齐 git 收口核验模式：把每个产出代码文件的 实测字节数/行数/语法/语句数
+    回填交付件正文，QC/PM 拿引擎级证据底座——「声称 17141 字符实测 1225 字节」
+    这类造假直接可见，不再依赖 QC 自己 run_shell 逐个取证（取证失败=盲审烧轮次）。
+    返回报告字符串（空=无代码产出文件）。
+    """
+    lines = []
+    for f in dict.fromkeys(files or []):
+        if not f.endswith(('.py', '.json', '.sql', '.dspy', '.ui')):
+            continue
+        try:
+            st = os.stat(f)
+            with open(f, encoding='utf-8', errors='ignore') as fh:
+                txt = fh.read()
+        except Exception:
+            continue
+        rel = os.path.relpath(f, space_dir) if f.startswith(space_dir) else f
+        info = f"- {rel}: {st.st_size}B / {txt.count(chr(10)) + 1}行"
+        if f.endswith('.py'):
+            ok, why = _py_code_stats(txt)
+            info += f" | {why}" if ok else f" | ⚠️ {why}"
+        elif f.endswith('.json'):
+            try:
+                json.loads(txt)
+                info += " | json=OK"
+            except Exception as e:
+                info += f" | ⚠️ json 解析失败: {str(e)[:60]}"
+        lines.append(info)
+    return "\n".join(lines)
+
+
 async def _enforce_git_closure(space_dir, written_files, do_commit=True):
     """git 收口硬门禁（2026-09-16 pbls M1a「编造 git 证据」事故根治）。
 
@@ -2867,6 +3030,14 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
                             msgs.append({"role": "tool", "tool_call_id": tc.get("id", ""), "content": _st_err})
                             logger.info(f"role_agent deliver stub-docs rejected: task={task_id} {_st_err[:160]}")
                             continue
+                        # 代码空壳门禁（2026-09-16 pbls M1a verify_gate.py 空壳事故）：
+                        # write_file 实写的 .py/.json 先验语法与真实语句，files 参数先验后写
+                        _sc_err = _validate_stub_code(space_dir, written_files) \
+                            or _validate_code_files_params(params.get("files"))
+                        if _sc_err:
+                            msgs.append({"role": "tool", "tool_call_id": tc.get("id", ""), "content": _sc_err})
+                            logger.info(f"role_agent deliver stub-code rejected: task={task_id} {_sc_err[:160]}")
+                            continue
                         deliverable = {"action": "deliver", **params}
                         break
                     if tool == "ask_question":
@@ -2930,6 +3101,14 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
                     msgs.append({"role": "user", "content": _st_err})
                     logger.info(f"role_agent deliver stub-docs rejected(text): task={task_id} {_st_err[:160]}")
                     continue
+                # 代码空壳门禁（文本兜底路径同款，2026-09-16）
+                _sc_err = _validate_stub_code(space_dir, written_files) \
+                    or _validate_code_files_params(act.get("files"))
+                if _sc_err:
+                    msgs.append({"role": "assistant", "content": raw})
+                    msgs.append({"role": "user", "content": _sc_err})
+                    logger.info(f"role_agent deliver stub-code rejected(text): task={task_id} {_sc_err[:160]}")
+                    continue
                 deliverable = act
                 break
             elif act.get('action') == 'ask':
@@ -3016,6 +3195,16 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
             logger.info(f"git closure enforced: task={task_id} ok={_gc_ok} "
                         f"report={_gc_report[:200]}")
 
+        # 产出文件机械核验（2026-09-16 pbls M1a 空壳事故配套）：每个代码产出文件的
+        # 实测字节/行数/语法/语句数由引擎计算回填，QC 拿引擎级证据底座，不再依赖
+        # 自己 run_shell 逐个取证（取证失败=盲审烧轮次），声称与实测不符直接可见。
+        _cc_report = _code_closure_report(space_dir, list(written_files) + files_written)
+        if _cc_report:
+            result_text += ("\n\n---\n## 产出文件机械核验（引擎自动计算，非 agent 声明）\n"
+                            + _cc_report)
+            logger.info(f"code closure report: task={task_id} files="
+                        f"{len(_cc_report.splitlines())}")
+
         # 写交付件文档（result 摘要快照，评审留痕用；交付件本体是 docs/ 编号树 + files 清单）
         file_path = ''
         if result_text: