diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 1786e58..5f52b4a 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -2523,6 +2523,40 @@ def _validate_code_files_params(files): "无法产出时 ask_question 冒泡，禁止用描述性注释大纲冒充代码。") +def _validate_import_closure(space_dir, written_files): + """import 闭包硬门禁（2026-09-16 pbls pbl_common 半迁移重写事故根治）。 + + 事故形状：develop 重写公共内核 pbl_common，删除 errors.py/tenant.py 既有 + 符号面（TenantMissingError/PblError 类族），但 context.py/api.py 等内部 + 文件与 6+ 依赖模块仍 import 旧名 → 整包 ImportError 对外契约瘫痪。设计 + 文档明文「接口变更需向后兼容（不删既有签名）」，而既有门禁全是单文件 + 尺子（py_compile 空壳/字节数/git 收口），量不出跨文件符号断裂。 + + 机制（确定性 ast 静态分析，不执行被检代码）：核验范围 = 本任务写入 .py + 所属包 + 全空间引用这些包的文件；符号级核验 from-import 在目标模块有定义。 + 逃逸阀：目标模块 __getattr__（PEP 562）/ast 解析失败/外部库/核验器自身 + 异常 → 放行（记日志）。任务未触碰的包存量断裂不拦（非本任务责任，防死锁）。 + 返回 None=通过；字符串=拒绝原因（逐条 文件:行:符号，可行动）。 + """ + from .import_closure import check_closure_for_delivery + problems, involved = check_closure_for_delivery(space_dir, written_files) + if problems is None or not problems: + return None + lines = [] + for p in problems[:15]: + lines.append("- %s:%s 从 `%s` 引用 `%s`，目标模块无此%s" % ( + p["file"], p["line"], p["source"], p["symbol"], + "模块" if p["kind"] == "missing-module" else "符号")) + more = ("\n（另有 %d 处未列出）" % (len(problems) - 15)) if len(problems) > 15 else "" + return ("FAIL: import 闭包断裂 %d 处（涉及包 %s）——你改动/删除的符号仍被下列文件" + "引用，交付后 import 即 ImportError（公共包对外契约必须向后兼容：新增可以，" + "删除/改名必须同轮把全部引用方一并改造）：\n%s%s\n处置二选一：" + "①在目标模块补回兼容符号（别名/包装，推荐——依赖面大时）；" + "②把上述引用方文件同轮改到新符号并一起 deliver。" + "改完后本门禁自动复检；无法当轮闭环时 ask_question 如实冒泡，禁止交付半迁移状态。" + % (len(problems), "/".join(sorted(involved)), "\n".join(lines), more)) + + def _code_closure_report(space_dir, files): """产出文件机械核验报告（引擎自动计算，非 agent 声明）。 @@ -3035,6 +3069,13 @@ 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": _sc_err}) logger.info(f"role_agent deliver stub-code rejected: task={task_id} {_sc_err[:160]}") continue + # import 闭包门禁（2026-09-16 pbls pbl_common 半迁移重写事故）： + # 公共包重写删既有符号面 → 依赖模块 import 断裂必须当轮拦 + _ic_err = _validate_import_closure(space_dir, written_files) + if _ic_err: + msgs.append({"role": "tool", "tool_call_id": tc.get("id", ""), "content": _ic_err}) + logger.info(f"role_agent deliver import-closure rejected: task={task_id} {_ic_err[:160]}") + continue deliverable = {"action": "deliver", **params} break if tool == "ask_question": @@ -3106,6 +3147,13 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): msgs.append({"role": "user", "content": _sc_err}) logger.info(f"role_agent deliver stub-code rejected(text): task={task_id} {_sc_err[:160]}") continue + # import 闭包门禁（文本兜底路径同款，2026-09-16） + _ic_err = _validate_import_closure(space_dir, written_files) + if _ic_err: + msgs.append({"role": "assistant", "content": raw}) + msgs.append({"role": "user", "content": _ic_err}) + logger.info(f"role_agent deliver import-closure rejected(text): task={task_id} {_ic_err[:160]}") + continue deliverable = act break elif act.get('action') == 'ask': @@ -3202,6 +3250,15 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): logger.info(f"code closure report: task={task_id} files=" f"{len(_cc_report.splitlines())}") + # import 闭包报告（机械核验段第二款，2026-09-16）：deliver 门禁按 written_files + # 拦在先，此处兜底 files 参数落盘后的断裂（含依赖方向），QC 拿引擎级证据。 + from .import_closure import closure_report + _ic_report = closure_report(space_dir, list(written_files) + files_written) + if _ic_report: + result_text += ("\n\n## import 闭包核验（引擎自动计算，非 agent 声明）\n" + + _ic_report) + logger.info(f"import closure report: task={task_id} len={len(_ic_report)}") + # 写交付件文档（result 摘要快照，评审留痕用；交付件本体是 docs/ 编号树 + files 清单） file_path = '' if result_text: diff --git a/pipeline_service/import_closure.py b/pipeline_service/import_closure.py new file mode 100644 index 0000000..607a19d --- /dev/null +++ b/pipeline_service/import_closure.py @@ -0,0 +1,331 @@ +# -*- 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 + 动态属性）→ 跳过符号级核验；ast 解析失败跳过（语法归空壳门禁）；外部库 + （顶层名不在包映射）不核验；本核验器自身异常 → 返回 None 放行（记日志）。 + +门禁范围（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__ + 星号导入 (module, level) 列表。""" + names = set() + dynamic = False + stars = [] + + 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) + 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 _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 check_closure_for_delivery(space_dir, written_files, max_problems=30): + """deliver 门禁用：核验范围=本任务写入 .py 所属包 + 全空间引用这些包的文件。 + + 返回 (problems, involved_pkgs)；核验器自身异常返回 (None, set())（逃逸阀， + 调用方放行并记日志——门禁故障不阻断交付，但必须在日志可见）。 + """ + try: + ck = _ClosureChecker(space_dir) + if not ck.pkgs: + return [], set() + # 本任务写入 .py 所属包 + involved = set() + for f in dict.fromkeys(written_files or []): + if not f.endswith(".py"): + continue + po = ck._pkg_of_file(f) + if po: + involved.add(po[0]) + if not involved: + return [], set() + # 核验对象：涉及包内全部文件 + 全空间 import 涉及包的文件 + targets = set() + for py in ck.all_py_files(): + po = ck._pkg_of_file(py) + if po and po[0] in involved: + targets.add(py) + continue + for _node, mod in ck.iter_imports(py): + if mod.split(".")[0] in involved: + targets.add(py) + break + problems = [] + for f in sorted(targets): + problems.extend(ck.check_file(f)) + # 去重 + 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 + 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) diff --git a/test_import_closure.py b/test_import_closure.py new file mode 100644 index 0000000..38c51ab --- /dev/null +++ b/test_import_closure.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""import_closure 门禁离线单测（复刻 pbls pbl_common 半迁移事故形状）。""" +import os +import shutil +import sys + +# 本地无 ahserver 等平台依赖 → 绕开包 __init__，按文件直接加载被测模块 +import importlib.util as _ilu +_spec = _ilu.spec_from_file_location( + "import_closure", + os.path.join(os.path.dirname(os.path.abspath(__file__)), + "pipeline_service", "import_closure.py")) +_ic = _ilu.module_from_spec(_spec) +_spec.loader.exec_module(_ic) +check_closure_for_delivery = _ic.check_closure_for_delivery +closure_report = _ic.closure_report + +ROOT = "/tmp/ic_test" +FAILS = [] + + +def check(name, cond, detail=""): + print(("PASS" if cond else "FAIL"), name, detail[:160]) + if not cond: + FAILS.append(name) + + +def w(path, content): + full = os.path.join(ROOT, path) + os.makedirs(os.path.dirname(full), exist_ok=True) + with open(full, "w", encoding="utf-8") as f: + f.write(content) + + +# ── 用例 1：事故形状复刻——重写 errors 删旧符号，内部+依赖仍 import 旧名 ── +shutil.rmtree(ROOT, ignore_errors=True) +w("modules/pbl_common/pbl_common/__init__.py", "from .kernel import PblError\n") +w("modules/pbl_common/pbl_common/errors.py", + "class PBLError(Exception):\n pass\n\n\ndef err(cat, key):\n return {}\n\n\ndef fail(cat, key):\n return {}\n") +w("modules/pbl_common/pbl_common/kernel.py", "class PblError(Exception):\n code = 'X'\n") +w("modules/pbl_common/pbl_common/context.py", + "from pbl_common.errors import TenantMissingError, ErrorCode\n") +w("modules/pbl_validation/pbl_validation/__init__.py", "") +w("modules/pbl_validation/pbl_validation/api.py", + "from pbl_common.errors import fail\nfrom pbl_common.audit import write_audit\n") + +# 本任务写了 pbl_common 的 errors.py（重写）→ 门禁应抓到 context.py 断裂 +written = [os.path.join(ROOT, "modules/pbl_common/pbl_common/errors.py")] +probs, involved = check_closure_for_delivery(ROOT, written) +check("事故形状被拦截", probs is not None and len(probs) >= 2, + "problems=%s involved=%s" % (probs and len(probs), involved)) +check("涉及包=pbl_common", involved == {"pbl_common"}, str(involved)) +syms = {(p["file"], p["symbol"]) for p in (probs or [])} +check("抓到 context.py TenantMissingError", + ("modules/pbl_common/pbl_common/context.py", "TenantMissingError") in syms, str(syms)) +check("抓到依赖方 pbl_validation missing-module", + any(p["kind"] == "missing-module" and p["symbol"] == "pbl_common.audit" + for p in (probs or [])), str(probs)) +rep = closure_report(ROOT, written) +check("closure_report 非空含断裂", "断裂" in rep and "ImportError" in rep, rep[:80]) + +# ── 用例 2：未触碰的包存量断裂不拦（防死锁）── +w("modules/pbl_unrelated/pbl_unrelated/__init__.py", "") +w("modules/pbl_unrelated/pbl_unrelated/api.py", "from pbl_common.nothere import X\n") +probs2, involved2 = check_closure_for_delivery(ROOT, written) +check("未触碰包不在 involved", "pbl_unrelated" not in involved2, str(involved2)) +# 但它 import pbl_common（涉及包）→ 应作为引用方被核验 +check("引用涉及包的文件被核验", + any(p["file"].startswith("modules/pbl_unrelated") for p in (probs2 or [])), + str(probs2)) + +# ── 用例 3：修复后（补兼容符号）门禁放行 ── +w("modules/pbl_common/pbl_common/errors.py", + "class PBLError(Exception):\n pass\n\n\ndef err(cat, key):\n return {}\n\n\ndef fail(cat, key):\n return {}\n\n\n" + "# 兼容层（设计文档§5 向后兼容）\n" + "class PblError(PBLError):\n code = 'PBL_E'\n\n\nclass TenantMissingError(PblError):\n pass\n\n\nclass TenantInvalidError(PblError):\n pass\n\n\nclass ErrorCode(object):\n TENANT_MISSING = 'PBL-TENANT-0001'\n") +w("modules/pbl_common/pbl_common/audit.py", "def write_audit(*a, **kw):\n return ''\n") +# context.py 引用修复 + unrelated 的 nothere 仍在 → 只应剩 unrelated 断裂 +probs3, _ = check_closure_for_delivery(ROOT, written) +remain = [p for p in (probs3 or []) if not p["file"].startswith("modules/pbl_unrelated")] +check("补兼容符号后 pbl_common 闭包干净", not remain, str(remain)) + +# ── 用例 4：逃逸阀——__getattr__ 动态模块不误伤 ── +w("modules/pbl_dyn/pbl_dyn/__init__.py", "def __getattr__(name):\n return None\n") +w("modules/pbl_dyn/pbl_dyn/sub.py", "X = 1\n") +w("modules/pbl_validation/pbl_validation/dyn_user.py", "from pbl_dyn import WHATEVER\n") +probs4, _ = check_closure_for_delivery( + ROOT, [os.path.join(ROOT, "modules/pbl_dyn/pbl_dyn/sub.py")]) +check("动态模块逃逸阀", not any("dyn" in p["file"] for p in (probs4 or [])), str(probs4)) + +# ── 用例 5：语法坏文件跳过（归空壳门禁）不崩 ── +w("modules/pbl_validation/pbl_validation/broken.py", "def broken(:\n") +probs5, _ = check_closure_for_delivery( + ROOT, [os.path.join(ROOT, "modules/pbl_validation/pbl_validation/broken.py")]) +check("语法坏文件不崩", probs5 is not None, str(probs5)) + +# ── 用例 6：无 .py 写入（纯文档任务）零成本放行 ── +probs6, involved6 = check_closure_for_delivery(ROOT, [os.path.join(ROOT, "docs/x.md")]) +check("纯文档任务不拦", probs6 == [] and not involved6, str((probs6, involved6))) + +# ── 用例 7：相对导入 + 别名 + 星号 ── +w("modules/pbl_rel/pbl_rel/__init__.py", "from .core import helper\n__all__ = ['helper']\n") +w("modules/pbl_rel/pbl_rel/core.py", "def helper():\n return 1\n") +w("modules/pbl_rel/pbl_rel/use.py", "from . import helper\nfrom .core import helper as h2\nfrom pbl_rel.core import *\n") +probs7, _ = check_closure_for_delivery(ROOT, [os.path.join(ROOT, "modules/pbl_rel/pbl_rel/use.py")]) +check("相对导入/别名/星号解析", not any("pbl_rel" in p["file"] for p in (probs7 or [])), str(probs7)) +w("modules/pbl_rel/pbl_rel/bad.py", "from . import NOT_THERE\n") +probs8, _ = check_closure_for_delivery(ROOT, [os.path.join(ROOT, "modules/pbl_rel/pbl_rel/bad.py")]) +check("相对导入断裂也抓", any(p["symbol"] == "NOT_THERE" for p in (probs8 or [])), str(probs8)) + +print("\nFAILS:", FAILS or "NONE") +sys.exit(1 if FAILS else 0)