feat(deliver): import闭包硬门禁——跨文件符号引用断裂当轮拦(2026-09-16 pbls pbl_common半迁移重写事故根治:develop重写公共内核删errors/tenant既有符号面,但内部context/api与6+依赖模块仍import旧名→整包ImportError对外契约瘫痪;设计文档明文向后兼容不删既有签名,而既有门禁全是单文件尺子py_compile空壳/字节数/git收口,量不出跨文件断裂)。新模块import_closure(stdlib ast静态分析不执行被检代码):核验范围=本任务写入.py所属包+全空间引用这些包的文件,from-import符号级核验;逃逸阀=__getattr__动态模块/ast解析失败/外部库/核验器自身异常放行;未触碰包存量断裂不拦防死锁。挂载:native+文本兜底两条deliver路径硬拦(FAIL给逐条文件:行:符号+处置二选一:补兼容符号或同轮改齐引用方)+机械核验段回填闭包报告给QC当证据。离线单测13例全绿(事故形状精确复刻+逃逸阀+相对导入/别名/星号)

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:

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)

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)