12f8384d49
根因1(逃逸阀缺口): agent用工具脚本生成db.py compat块(globals()[_name]=_val逐符号注入)修import闭包断裂,
运行时实测全通(pbl_blueprint.db/api/blueprint_crud/subobjects/templates import OK,PblError/require_tenant/
get_env全True),但ast静态量不出Subscript下标赋值→门禁恒报20处断裂→QC按'声称修复vs引擎实测断裂直接矛盾'
判造假证据→4轮退回达上限→fault→项目paused。与PEP562 __getattr__同类:静态不可判定即保守放行。
识别三形态: globals()[k]=v(含For/Try嵌套walk_top可达) / globals().update({...}) / _g=globals()别名。
函数体内globals()不算模块级注入(walk_top只走模块级stmts,不误放行)。
根因2(责任范围过宽): 依赖方向拉进来的外部文件(pbl_agent_runtime/api.py因import pbl_blueprint入范围)的
断裂源在pbl_common(半迁移存量欠账114处,别的任务的账)——违背门禁自己声明的'任务未触碰的包存量断裂不拦,
防死锁'原则,M1b实测26处里6处是pbl_common欠账,小任务永远修不完全空间账。
修: problems过滤 source.split('.')[0] in owned_pkgs(只拦涉及仓库自有包的契约断裂)。
原pbl_common事故形状防护不受影响: 重写pbl_common的任务owned_pkgs含pbl_common,断裂全保留。
修后M1b剩余真实断裂仅1处: m1b_common.py:114 from .errors import PblError而errors.py只有
PblBlueprintError——门禁拦它是对的(补一行别名即闭环),FAIL消息可行动。
本地走查6例通过(注入3形态识别/普通模块不误伤/函数内globals不误放行/真实compat块形状)。
411 lines
18 KiB
Python
411 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""import 闭包静态核验（deliver 硬门禁 + 交付件机械核验段共用，2026-09-16）。
|
||
|
||
事故（pbls pbl_common 半迁移重写）：develop 重写公共内核时删除 errors.py/
|
||
tenant.py 既有符号面（TenantMissingError/ErrorCode/PblError 类族 → 换
|
||
PBLError/err/fail 函数族），但 context.py/api.py/crud_factory.py 等内部文件
|
||
与 6+ 依赖模块仍 import 旧名 → `from pbl_common.api import ...` 全部
|
||
ImportError，公共内核对外契约瘫痪。设计文档明文「接口变更需向后兼容
|
||
（新增可选参数，不删既有签名）」，但引擎既有门禁全是【单文件】尺子
|
||
（py_compile 空壳/字节数/git 收口），量不出【跨文件符号引用断裂】——
|
||
半迁移状态照样过闸，事故到运行期才暴露。
|
||
|
||
机制（确定性，stdlib-only ast 静态分析，不执行被检代码）：
|
||
1. 扫描工作空间 modules/*/ 与 apps/*/ 的 Python 包，建 {包名: 包根} 映射。
|
||
2. 对每个 .py 收集模块级名字绑定（def/class/赋值/import 别名，含 If/Try
|
||
分支），对每条 from-import 核验符号在目标模块有定义（含再导出/子模块）。
|
||
3. 逃逸阀（防误报死锁）：目标模块定义 module-level __getattr__（PEP 562
|
||
动态属性）或 module-level `globals()[...] = ...` 运行时符号注入（2026-09-17
|
||
pbls M1b 死循环根治：agent 用工具脚本生成 compat 注入块修断裂，运行时
|
||
import 全通但 ast 量不出下标赋值 → 门禁恒报断裂 → QC 判「声称修复 vs
|
||
实测断裂」矛盾 → 无限退回；与 PEP 562 同类，静态不可判定即保守放行）
|
||
→ 跳过该模块符号级核验；ast 解析失败跳过（语法归空壳门禁）；外部库
|
||
（顶层名不在包映射）不核验；本核验器自身异常 → 返回 None 放行（记日志）。
|
||
|
||
责任范围过滤（2026-09-17 同日修正）：依赖方向拉进来的外部文件，只拦
|
||
「引用涉及包」的断裂；它引用其他包（如半迁移欠账 pbl_common）的存量断裂
|
||
不是本任务责任，不进门禁清单——否则小任务背上全空间欠账，永远修不完
|
||
（pbls M1b 实测：26 处里 6 处是 pbl_agent_runtime→pbl_common 存量断裂）。
|
||
|
||
门禁范围（check_closure_for_delivery）：
|
||
- 本任务写入 .py 所属的包 = 涉及包；核验对象 = 涉及包内全部文件 + 全空间
|
||
import 涉及包的文件。改公共包断掉依赖方（事故形状）会在 deliver 当轮被拦，
|
||
FAIL 消息给出逐条 文件:行:符号 → agent 要么补兼容符号、要么同轮改齐引用
|
||
（用户裁定：保持兼容或一并改造，二选一，禁止半迁移交付）。
|
||
- 任务未触碰的包即使有存量断裂也不拦（不是本任务责任，防死锁）。
|
||
"""
|
||
import ast
|
||
import logging
|
||
import os
|
||
|
||
logger = logging.getLogger("pipeline.import_closure")
|
||
|
||
_SKIP_DIRS = {"__pycache__", ".git", "node_modules", ".venv", "venv"}
|
||
_NON_PKG_DIRS = {"scripts", "tests", "wwwroot", "docs", "json", "models",
|
||
"init", "skill", "backups"}
|
||
|
||
|
||
def _add_target(t, names):
|
||
if isinstance(t, ast.Name):
|
||
names.add(t.id)
|
||
elif isinstance(t, (ast.Tuple, ast.List)):
|
||
for e in t.elts:
|
||
_add_target(e, names)
|
||
|
||
|
||
def _collect_defs(tree):
|
||
"""模块级名字绑定集合 + 是否定义 __getattr__/globals()注入 + 星号导入列表。"""
|
||
names = set()
|
||
dynamic = False
|
||
stars = []
|
||
|
||
def _is_ns_call(node):
|
||
"""globals()/vars()/locals() 调用形态（运行时符号注入，静态不可判定）。"""
|
||
return (isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
|
||
and node.func.id in ("globals", "vars", "locals"))
|
||
|
||
def walk_top(stmts):
|
||
nonlocal dynamic
|
||
for s in stmts:
|
||
if isinstance(s, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
||
names.add(s.name)
|
||
if (isinstance(s, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||
and s.name == "__getattr__"):
|
||
dynamic = True
|
||
elif isinstance(s, ast.Assign):
|
||
for t in s.targets:
|
||
_add_target(t, names)
|
||
# globals()[name] = val / vars()[...] = ... 运行时注入：
|
||
# 符号面静态不可判定 → 整模块按动态放行（同 PEP 562 逃逸阀）
|
||
if (isinstance(t, ast.Subscript) and _is_ns_call(t.value)):
|
||
dynamic = True
|
||
if _is_ns_call(s.value):
|
||
# X = globals() 之后再 X[...] = ...：保守也视为注入形态
|
||
dynamic = True
|
||
elif isinstance(s, ast.Expr) and isinstance(s.value, ast.Call):
|
||
# globals().update({...}) 形态
|
||
fn = s.value.func
|
||
if (isinstance(fn, ast.Attribute) and fn.attr == "update"
|
||
and _is_ns_call(fn.value)):
|
||
dynamic = True
|
||
elif isinstance(s, ast.AnnAssign):
|
||
if s.target is not None:
|
||
_add_target(s.target, names)
|
||
elif isinstance(s, ast.Import):
|
||
for a in s.names:
|
||
names.add(a.asname or a.name.split(".")[0])
|
||
elif isinstance(s, ast.ImportFrom):
|
||
for a in s.names:
|
||
if a.name == "*":
|
||
stars.append((s.module or "", s.level))
|
||
continue
|
||
names.add(a.asname or a.name)
|
||
elif isinstance(s, (ast.If, ast.Try, ast.While, ast.For,
|
||
ast.With, ast.AsyncFor, ast.AsyncWith)):
|
||
for field in ("body", "orelse", "finalbody"):
|
||
walk_top(getattr(s, field, []) or [])
|
||
for h in getattr(s, "handlers", []) or []:
|
||
if h.name:
|
||
names.add(h.name)
|
||
walk_top(h.body or [])
|
||
|
||
walk_top(tree.body)
|
||
return names, dynamic, stars
|
||
|
||
|
||
def build_package_map(space_dir):
|
||
"""扫描 modules/*/ 与 apps/*/，返回 {包名: 包父目录}（含 __init__.py 的目录）。"""
|
||
pkgs = {}
|
||
for group in ("modules", "apps"):
|
||
base = os.path.join(space_dir, group)
|
||
if not os.path.isdir(base):
|
||
continue
|
||
for repo in sorted(os.listdir(base)):
|
||
repo_dir = os.path.join(base, repo)
|
||
if not os.path.isdir(repo_dir) or repo.startswith("."):
|
||
continue
|
||
if os.path.isfile(os.path.join(repo_dir, "__init__.py")):
|
||
pkgs.setdefault(repo, os.path.dirname(repo_dir))
|
||
for sub in sorted(os.listdir(repo_dir)):
|
||
sub_dir = os.path.join(repo_dir, sub)
|
||
if (os.path.isdir(sub_dir) and not sub.startswith((".", "_"))
|
||
and sub not in _NON_PKG_DIRS
|
||
and os.path.isfile(os.path.join(sub_dir, "__init__.py"))):
|
||
pkgs.setdefault(sub, repo_dir)
|
||
return pkgs
|
||
|
||
|
||
def build_repo_map(space_dir):
|
||
"""modules/<repo> 与 apps/<repo> 的仓库根映射：[(仓库绝对路径, 仓库名)]。
|
||
|
||
仓库 ≠ 包：modules/pbls/ 只有 app/ scripts/ 没有包目录，但入口文件
|
||
modules/pbls/app/pbls.py 的 import 断裂同样是真实事故面（pbls 入口
|
||
from world.api import 实测断裂）——门禁范围必须按仓库归属算，不能只按包。
|
||
"""
|
||
repos = []
|
||
for group in ("modules", "apps"):
|
||
base = os.path.join(space_dir, group)
|
||
if not os.path.isdir(base):
|
||
continue
|
||
for repo in sorted(os.listdir(base)):
|
||
repo_dir = os.path.join(base, repo)
|
||
if os.path.isdir(repo_dir) and not repo.startswith("."):
|
||
repos.append((os.path.abspath(repo_dir), repo))
|
||
return repos
|
||
|
||
|
||
def _resolve_module_file(mod, pkgs):
|
||
"""绝对模块名 → 文件路径；'EXT'=外部包；None=包内但文件不存在。"""
|
||
top = mod.split(".")[0]
|
||
if top not in pkgs:
|
||
return "EXT"
|
||
root = pkgs[top]
|
||
parts = mod.split(".")
|
||
path = os.path.join(root, *parts) + ".py"
|
||
if os.path.isfile(path):
|
||
return path
|
||
init = os.path.join(root, *parts, "__init__.py")
|
||
if os.path.isfile(init):
|
||
return init
|
||
if os.path.isdir(os.path.join(root, *parts)):
|
||
return os.path.join(root, *parts) # 命名空间目录：保守按动态放行
|
||
return None
|
||
|
||
|
||
class _ClosureChecker:
|
||
def __init__(self, space_dir):
|
||
self.space_dir = space_dir
|
||
self.pkgs = build_package_map(space_dir)
|
||
self._mod_cache = {}
|
||
|
||
def _pkg_of_file(self, f):
|
||
best = None
|
||
ap = os.path.abspath(f)
|
||
for name, root in self.pkgs.items():
|
||
pr = os.path.abspath(os.path.join(root, name))
|
||
if ap.startswith(pr + os.sep):
|
||
if best is None or len(pr) > len(best[1]):
|
||
best = (name, pr)
|
||
return best
|
||
|
||
def _module_defs(self, mod):
|
||
if mod in self._mod_cache:
|
||
return self._mod_cache[mod]
|
||
path = _resolve_module_file(mod, self.pkgs)
|
||
if path in (None, "EXT"):
|
||
self._mod_cache[mod] = None
|
||
return None
|
||
if os.path.isdir(path):
|
||
self._mod_cache[mod] = (set(), True)
|
||
return self._mod_cache[mod]
|
||
try:
|
||
with open(path, encoding="utf-8", errors="ignore") as fh:
|
||
tree = ast.parse(fh.read())
|
||
except SyntaxError:
|
||
self._mod_cache[mod] = (set(), True) # 语法坏→动态放行（空壳门禁管）
|
||
return self._mod_cache[mod]
|
||
except Exception:
|
||
self._mod_cache[mod] = None
|
||
return None
|
||
names, dynamic, stars = _collect_defs(tree)
|
||
for s_mod, s_level in stars:
|
||
if s_level > 0 or not s_mod or s_mod.split(".")[0] not in self.pkgs:
|
||
continue
|
||
sub = self._module_defs(s_mod)
|
||
if sub:
|
||
names |= sub[0]
|
||
dynamic = dynamic or sub[1]
|
||
self._mod_cache[mod] = (names, dynamic)
|
||
return self._mod_cache[mod]
|
||
|
||
def _abs_mod(self, f, node):
|
||
if node.level == 0:
|
||
return node.module or ""
|
||
po = self._pkg_of_file(f)
|
||
if not po:
|
||
return None
|
||
pkg_name, pkg_root = po
|
||
rel_dir = os.path.dirname(os.path.abspath(f))
|
||
base_parts = [pkg_name]
|
||
pkg_abs = os.path.join(os.path.abspath(pkg_root), pkg_name)
|
||
if os.path.abspath(rel_dir) != os.path.abspath(pkg_abs):
|
||
sub = os.path.relpath(rel_dir, pkg_abs)
|
||
if sub and sub != ".":
|
||
base_parts += sub.split(os.sep)
|
||
up = node.level - 1
|
||
if up > 0:
|
||
if up >= len(base_parts):
|
||
return None
|
||
base_parts = base_parts[: len(base_parts) - up]
|
||
if node.module:
|
||
base_parts += node.module.split(".")
|
||
return ".".join(base_parts)
|
||
|
||
def iter_imports(self, f):
|
||
"""产出 (node, 绝对模块名)；外部/无法归一的不产出。"""
|
||
try:
|
||
with open(f, encoding="utf-8", errors="ignore") as fh:
|
||
tree = ast.parse(fh.read())
|
||
except Exception:
|
||
return
|
||
for node in ast.walk(tree):
|
||
if not isinstance(node, ast.ImportFrom):
|
||
continue
|
||
mod = self._abs_mod(f, node)
|
||
if not mod or mod.split(".")[0] not in self.pkgs:
|
||
continue
|
||
yield node, mod
|
||
|
||
def check_file(self, f):
|
||
"""单文件核验 → problem dict 列表。"""
|
||
problems = []
|
||
rel = os.path.relpath(f, self.space_dir)
|
||
for node, mod in self.iter_imports(f):
|
||
path = _resolve_module_file(mod, self.pkgs)
|
||
if path == "EXT":
|
||
continue
|
||
if path is None:
|
||
parent = mod.rsplit(".", 1)[0] if "." in mod else mod
|
||
p_path = _resolve_module_file(parent, self.pkgs) if parent != mod else None
|
||
if p_path and p_path != "EXT":
|
||
problems.append({
|
||
"file": rel, "line": node.lineno,
|
||
"kind": "missing-module", "symbol": mod,
|
||
"source": parent})
|
||
continue
|
||
defs = self._module_defs(mod)
|
||
if defs is None:
|
||
continue
|
||
names, dynamic = defs
|
||
if dynamic:
|
||
continue
|
||
pkg_dir = path if os.path.isdir(path) else os.path.dirname(path)
|
||
for a in node.names:
|
||
if a.name == "*" or a.name in names:
|
||
continue
|
||
if (os.path.isfile(os.path.join(pkg_dir, a.name + ".py"))
|
||
or os.path.isfile(os.path.join(pkg_dir, a.name,
|
||
"__init__.py"))):
|
||
continue # from pkg import submod 形态
|
||
problems.append({
|
||
"file": rel, "line": node.lineno,
|
||
"kind": "missing-symbol", "symbol": a.name,
|
||
"source": mod})
|
||
return problems
|
||
|
||
def all_py_files(self):
|
||
for group in ("modules", "apps"):
|
||
base = os.path.join(self.space_dir, group)
|
||
if not os.path.isdir(base):
|
||
continue
|
||
for repo in sorted(os.listdir(base)):
|
||
repo_dir = os.path.join(base, repo)
|
||
if not os.path.isdir(repo_dir):
|
||
continue
|
||
for root, dirs, files in os.walk(repo_dir):
|
||
dirs[:] = [d for d in dirs if d not in _SKIP_DIRS]
|
||
for fn in sorted(files):
|
||
if fn.endswith(".py"):
|
||
yield os.path.join(root, fn)
|
||
|
||
|
||
def _repo_of(path, repos):
|
||
"""文件归属仓库名（最长前缀匹配）。"""
|
||
ap = os.path.abspath(path)
|
||
best = None
|
||
for repo_dir, name in repos:
|
||
if ap.startswith(repo_dir + os.sep):
|
||
if best is None or len(repo_dir) > len(best[0]):
|
||
best = (repo_dir, name)
|
||
return best[1] if best else None
|
||
|
||
|
||
def check_closure_for_delivery(space_dir, written_files, max_problems=30):
|
||
"""deliver 门禁用：核验范围=本任务写入 .py 所属【仓库】+ 全空间引用这些仓库的文件。
|
||
|
||
仓库归属而非包归属（2026-09-16 修正）：modules/pbls 只有 app/ scripts/
|
||
没有包目录，但入口 modules/pbls/app/pbls.py 的 `from world.api import`
|
||
断裂同样是真实事故面——只按包算会漏掉。跨仓库方向：写入文件所属仓库
|
||
拥有的包（包根在仓库目录下）被别处 import 断裂也拦（pbl_common 事故形状）。
|
||
|
||
返回 (problems, involved_names)；核验器自身异常返回 (None, set())（逃逸阀，
|
||
调用方放行并记日志——门禁故障不阻断交付，但必须在日志可见）。
|
||
"""
|
||
try:
|
||
ck = _ClosureChecker(space_dir)
|
||
repos = build_repo_map(space_dir)
|
||
if not repos:
|
||
return [], set()
|
||
# 本任务写入 .py 所属仓库
|
||
involved = set()
|
||
for f in dict.fromkeys(written_files or []):
|
||
if not f.endswith(".py"):
|
||
continue
|
||
rn = _repo_of(f, repos)
|
||
if rn:
|
||
involved.add(rn)
|
||
if not involved:
|
||
return [], set()
|
||
# 涉及仓库拥有的包（跨仓库引用方向）
|
||
repo_dirs = {name: rd for rd, name in repos if name in involved}
|
||
owned_pkgs = set()
|
||
for pkg_name, pkg_root in ck.pkgs.items():
|
||
pkg_abs = os.path.abspath(os.path.join(pkg_root, pkg_name))
|
||
for rn, rd in repo_dirs.items():
|
||
if pkg_abs.startswith(rd + os.sep):
|
||
owned_pkgs.add(pkg_name)
|
||
break
|
||
# 核验对象：涉及仓库内全部文件 + 全空间 import 涉及包的文件
|
||
targets = set()
|
||
for py in ck.all_py_files():
|
||
rn = _repo_of(py, repos)
|
||
if rn in involved:
|
||
targets.add(py)
|
||
continue
|
||
for _node, mod in ck.iter_imports(py):
|
||
if mod.split(".")[0] in owned_pkgs:
|
||
targets.add(py)
|
||
break
|
||
problems = []
|
||
for f in sorted(targets):
|
||
problems.extend(ck.check_file(f))
|
||
# 责任范围过滤（2026-09-17 pbls M1b 死循环修正）：只保留「断裂源在
|
||
# 涉及仓库自有包」的问题——即本任务改动面自己的契约断裂（含依赖方
|
||
# import 涉及包断裂 = pbl_common 事故形状）。断裂源在其他包（如
|
||
# pbl_common 半迁移存量欠账）的不是本任务责任：小任务修不完全空间
|
||
# 欠账，硬拦只会无限退回（实测 26 处里 6 处是别的包的存量账）。
|
||
problems = [p for p in problems
|
||
if p.get("source", "").split(".")[0] in owned_pkgs]
|
||
# 去重
|
||
seen, uniq = set(), []
|
||
for p in problems:
|
||
k = (p["file"], p["line"], p["symbol"], p["source"])
|
||
if k not in seen:
|
||
seen.add(k)
|
||
uniq.append(p)
|
||
return uniq[:max_problems], involved | owned_pkgs
|
||
except Exception as e:
|
||
logger.warning("import closure check failed (escape, deliver 放行): %r", e)
|
||
return None, set()
|
||
|
||
|
||
def closure_report(space_dir, written_files, max_lines=20):
|
||
"""交付件机械核验段用：断裂清单文本（空串=无断裂/不涉及包）。"""
|
||
problems, involved = check_closure_for_delivery(space_dir, written_files)
|
||
if problems is None:
|
||
return "⚠️ import 闭包核验器执行失败（已放行，见服务端日志）"
|
||
if not involved:
|
||
return ""
|
||
if not problems:
|
||
return ("import 闭包核验：涉及包 %s，未发现跨文件符号引用断裂。"
|
||
% "/".join(sorted(involved)))
|
||
lines = ["⚠️ import 闭包断裂 %d 处（涉及包 %s）——以下引用在目标模块无定义，"
|
||
"import 即 ImportError：" % (len(problems), "/".join(sorted(involved)))]
|
||
for p in problems[:max_lines]:
|
||
lines.append("- %s:%s %s `%s`（目标 %s 无此%s）" % (
|
||
p["file"], p["line"],
|
||
"引用模块" if p["kind"] == "missing-module" else "引用符号",
|
||
p["symbol"], p["source"],
|
||
"模块" if p["kind"] == "missing-module" else "符号"))
|
||
return "\n".join(lines)
|