test(pbls): 新增 scripts/dspy_shell_scan.py —— .dspy 空壳/无return/return在async with内 静态扫描(QC 退回 #5 自检口径落地)

QC #5 指出「纯注释文件同样能通过 ast 解析,ast OK 不构成修复在位证据」。
本脚本把自检口径改为可机械核验三项:非注释可执行语句数>0(ast stmt 计数)、
显式 return 数>0 且不在 async with 块内、ast 包装语法通过。
实测:apps/pbls+modules/pbls 6 个 .dspy violations=0;
扩扫 15 个 pbl_* 自研模块 + 9 个复用模块共 156 个 .dspy,violations=2
(modules/world 两处 return 在 async with 内,属 referenced_modules 写保护范围,
 且两文件均有块外 return result 兜底不会返回 None,仅记录不修改)。
This commit is contained in:
agent.develop 2026-09-25 12:35:22 +08:00
parent 6176b41ac9
commit ebf6d79ef3

View File

@ -0,0 +1,92 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
.dspy 空壳/语法/return 静态扫描(QC 退回 #5 要求的自检口径)
背景:Bug WIu0Rkn0KwB0XTjYZz0qk 的 QC 退回意见 #5 指出——「纯注释文件同样能通过
ast 解析,ast OK 不构成『修复在位』证据」。本脚本把自检口径改为可机械核验的三项:
1) 非注释可执行语句数 > 0(ast 统计 stmt 节点,排除包装函数本身)
2) 显式 return 数 > 0(ast.Return 计数),且 return 不落在 async with 块内
3) ast 包装(async def 包裹)语法解析通过
用法:
python3 scripts/dspy_shell_scan.py [root ...]
缺省 root = apps modules(相对机构工作空间根)。退出码:0=全通过,1=存在违规。
"""
import ast
import os
import sys
SKIP_PARTS = {".git", "venv", ".pylibs", "node_modules", "__pycache__", "wheelhouse"}
def scan_file(path):
"""返回 (ok, reason, metrics)"""
try:
with open(path, encoding="utf-8", errors="replace") as fh:
src = fh.read()
except Exception as exc: # pragma: no cover
return False, "READ_FAIL %s" % exc, {}
wrapped = "async def _h():\n" + "".join(" " + ln + "\n" for ln in src.split("\n"))
try:
tree = ast.parse(wrapped)
except SyntaxError as exc:
return False, "AST_FAIL %s" % exc, {"bytes": len(src.encode("utf-8"))}
stmts = 0
returns = 0
for node in ast.walk(tree):
if isinstance(node, ast.stmt) and not isinstance(
node, (ast.FunctionDef, ast.AsyncFunctionDef)
):
stmts += 1
if isinstance(node, ast.Return):
returns += 1
ret_in_asyncwith = []
for node in ast.walk(tree):
if isinstance(node, ast.AsyncWith):
for sub in ast.walk(node):
if isinstance(sub, ast.Return):
ret_in_asyncwith.append(sub.lineno)
metrics = {
"bytes": len(src.encode("utf-8")),
"ast_stmts": stmts,
"ast_return": returns,
"return_in_asyncwith": ret_in_asyncwith,
}
if stmts == 0:
return False, "SHELL_ONLY: 0 executable statements", metrics
if returns == 0:
return False, "NO_RETURN: stmts=%d returns=0" % stmts, metrics
if ret_in_asyncwith:
return False, "RETURN_INSIDE_ASYNC_WITH lines=%s" % ret_in_asyncwith, metrics
return True, "OK", metrics
def main(argv):
roots = argv[1:] or ["apps", "modules"]
total = 0
bad = []
for root in roots:
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = [d for d in dirnames if d not in SKIP_PARTS]
for name in filenames:
if not name.endswith(".dspy"):
continue
path = os.path.join(dirpath, name)
total += 1
ok, reason, metrics = scan_file(path)
if not ok:
bad.append((path, reason, metrics))
print("scanned .dspy files: %d" % total)
print("violations: %d" % len(bad))
for path, reason, metrics in bad:
print(" %s -> %s | %s" % (path, reason, metrics))
return 1 if bad else 0
if __name__ == "__main__":
sys.exit(main(sys.argv))