diff --git a/test/test_build_flags.py b/test/test_build_flags.py
new file mode 100644
index 0000000..9c8ad38
--- /dev/null
+++ b/test/test_build_flags.py
@@ -0,0 +1,214 @@
+"""
+build 命令开关（--encrypt / --license）冒烟测试
+
+两个开关默认都是关的（不加参数 = 明文模块 + 不带授权的内部调试产物），
+这里验证"默认关"与"显式打开"在构建产物上都真的生效：
+  1. 参数解析：默认 encrypt/license 全关，--encrypt / --license 打开对应开关；
+     --only 里 license_guard 是合法步骤（文档里一直写着，之前被漏在校验表外）
+  2. 默认（不加密）：.mbank 存明文字节码（encrypted=false）、不写 .key（并清掉旧
+     的）、无 packer_key 记录，且**不需要密钥即可导入**；加 --encrypt 那条路仍然
+     加密、仍然写 .key，且用 .key 里的密钥能正常导入
+  3. 默认（不授权）：prepare 不要求/不注入授权公钥；加 --license 时缺公钥构建失败
+
+运行: uv run python test/test_build_flags.py
+"""
+import importlib
+import os
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+
+SRC = Path(__file__).parent
+sys.path.insert(0, str(SRC.parent / "utils" / "cli" / "src"))
+
+from module_bank import PythonToSQLite, layout  # noqa: E402
+from sqlite_vfs.core import SQLiteVFS  # noqa: E402
+
+from chongming_cli import build as b  # noqa: E402
+
+failures = []
+
+
+def check(name, cond, detail=""):
+    status = "PASS" if cond else "FAIL"
+    print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
+    if not cond:
+        failures.append(name)
+
+
+tmp = Path(tempfile.mkdtemp(prefix="build_flags_test_"))
+
+# ========== 1. 参数解析 ==========
+_argv, sys.argv = sys.argv, ["build"]
+mode, steps, options = b._parse_args()
+check(
+    "默认参数：pyinstaller + 全流程 + 不加密 + 不授权",
+    mode == "pyinstaller" and not steps and not options.encrypt and not options.license,
+)
+
+sys.argv = ["build", "--encrypt", "--license", "nuitka"]
+mode, steps, options = b._parse_args()
+check(
+    "--encrypt/--license 打开对应开关",
+    mode == "nuitka" and options.encrypt and options.license,
+    f"mode={mode} encrypt={options.encrypt} license={options.license}",
+)
+
+sys.argv = ["build", "--only", "mbank,license_guard"]
+mode, steps, options = b._parse_args()
+check("--only 接受 license_guard", steps == {"mbank", "license_guard"}, str(steps))
+
+sys.argv = ["build", "--only", "mbank", "--encrypt", "pyinstaller"]
+mode, steps, options = b._parse_args()
+check(
+    "位置参数与开关可混用",
+    mode == "pyinstaller" and steps == {"mbank"} and options.encrypt,
+)
+
+sys.argv = ["build", "--unknown-flag"]
+try:
+    b._parse_args()
+    check("未知参数应退出", False)
+except SystemExit:
+    check("未知参数应退出", True)
+sys.argv = _argv
+
+
+# ========== 2. mbank 加密 / 明文字节码 ==========
+def make_source(root: Path) -> Path:
+    """造一个最小可导入的包：demo/__init__.py"""
+    pkg = root / "demo"
+    pkg.mkdir(parents=True)
+    (pkg / "__init__.py").write_text("VALUE = 42\n", encoding="utf-8")
+    return pkg
+
+
+def bank_meta(bank: Path, module: str) -> dict:
+    vfs = SQLiteVFS(str(bank), compress=False, readonly=True)
+    try:
+        # 包名 → 包内路径由 module_bank 定，别自己拼；明文未混淆的模块不写
+        # meta.json（默认值即"什么都不用记"），用 read_meta 读缺省
+        return layout.read_meta(vfs, layout.module_paths(module, True)["meta"])
+    finally:
+        vfs.close()
+
+
+def bank_has_key(bank: Path) -> bool:
+    vfs = SQLiteVFS(str(bank), compress=False, readonly=True)
+    try:
+        return vfs.is_file("packer_key")
+    finally:
+        vfs.close()
+
+
+def import_from_bank(bank: Path, module: str, key):
+    """像 utils/launch.py 那样装载：install_importer → import。返回 (模块, 错误)。"""
+    packer = PythonToSQLite(str(bank))
+    finder = packer.install_importer(key=key)
+    try:
+        mod = importlib.import_module(module)
+        return mod, None
+    except Exception as exc:  # noqa: BLE001 — 测试就是要区分"能导/不能导"
+        return None, exc
+    finally:
+        if finder in sys.meta_path:
+            sys.meta_path.remove(finder)
+        sys.modules.pop(module, None)
+
+
+plain_dir = make_source(tmp / "plain_src")
+plain_bank = tmp / "plain.mbank"
+plain_size = b._pack_mbank(
+    str(plain_bank), str(plain_dir), "demo", obfuscate=False, encrypt=False
+)
+check("明文打包产出 .mbank", plain_bank.exists(), plain_size)
+check("明文打包不写 .key", not plain_bank.with_suffix(".key").exists())
+check("明文打包 encrypted=false", bank_meta(plain_bank, "demo")["encrypted"] is False)
+check("明文打包无 packer_key 记录", not bank_has_key(plain_bank))
+
+mod, err = import_from_bank(plain_bank, "demo", key=None)
+check("明文包无密钥即可导入", mod is not None and getattr(mod, "VALUE", None) == 42, str(err))
+
+enc_dir = make_source(tmp / "enc_src")
+enc_bank = tmp / "enc.mbank"
+enc_key_path = enc_bank.with_suffix(".key")
+enc_key_path.write_text("stale-key-from-previous-build", encoding="utf-8")  # 模拟上一次加密构建
+enc_size = b._pack_mbank(
+    str(enc_bank), str(enc_dir), "demo", obfuscate=False, encrypt=False
+)
+check("默认（不加密）清掉上一次加密构建留下的 .key", not enc_key_path.exists(), enc_size)
+
+encrypted_bank = tmp / "encrypted.mbank"
+key = b.Encryption.generate_key()
+_orig_gen = b.Encryption.generate_key
+b.Encryption.generate_key = staticmethod(lambda: key)  # 固定密钥，便于用 .key 验证导入
+try:
+    b._pack_mbank(str(encrypted_bank), str(enc_dir), "demo", obfuscate=False, encrypt=True)
+finally:
+    b.Encryption.generate_key = _orig_gen
+
+check("--encrypt 打包写 .key", encrypted_bank.with_suffix(".key").exists())
+check("--encrypt 打包 encrypted=true", bank_meta(encrypted_bank, "demo")["encrypted"] is True)
+
+mod, err = import_from_bank(encrypted_bank, "demo", key=encrypted_bank.with_suffix(".key").read_text())
+check("加密包用 .key 里的密钥可导入", mod is not None, str(err))
+
+mod, err = import_from_bank(encrypted_bank, "demo", key=None)
+check("加密包缺密钥无法导入（说明 .key 必须随包提供给加载端）", mod is None and isinstance(err, ImportError), str(err))
+
+
+# ========== 3. 授权公钥注入 / 默认不注入 ==========
+def make_project(root: Path) -> Path:
+    """造一个最小构建现场：public/{main.py,server.py,config.toml}"""
+    (root / "public" / "utils").mkdir(parents=True)
+    (root / "public" / "utils" / "launch.py").write_text("", encoding="utf-8")
+    (root / "public" / "main.py").write_text("", encoding="utf-8")
+    (root / "public" / "server.py").write_text("", encoding="utf-8")
+    (root / "public" / "config.toml").write_text(
+        '[security]\nsecret_key = "your-secret-key-change-in-production"\n\n'
+        '[license]\n# 机器码授权公钥（构建时注入）\npublic_key = ""\n',
+        encoding="utf-8",
+    )
+    return root
+
+
+env_keys = (b.LICENSE_PUBKEY_ENV, b.LICENSE_PUBKEY_FILE_ENV, b.ALLOW_NO_KEY_ENV)
+saved_env = {k: os.environ.pop(k, None) for k in env_keys}  # 环境变量不该影响测试结论
+
+cwd = Path.cwd()
+proj = make_project(tmp / "proj_default")
+os.chdir(proj)
+try:
+    msg = b._stage_prepare_build_dir(None, "pyinstaller", b.BuildOptions())
+    conf = (proj / "build" / "config.toml").read_text(encoding="utf-8")
+    check("默认（不授权）的 prepare 不报错", msg == "资源准备完成", msg)
+    check("默认不注入公钥（模板 public_key 保持空）", 'public_key = ""' in conf)
+    check("JWT 密钥仍会注入", "your-secret-key-change-in-production" not in conf)
+
+    try:
+        b._stage_prepare_build_dir(None, "pyinstaller", b.BuildOptions(license=True))
+        check("--license 缺授权公钥应失败", False)
+    except RuntimeError as exc:
+        check("--license 缺授权公钥应失败", "缺少厂商授权公钥" in str(exc))
+
+    os.environ[b.LICENSE_PUBKEY_ENV] = "test-pubkey-b64"
+    try:
+        b._stage_prepare_build_dir(None, "pyinstaller", b.BuildOptions(license=True))
+        conf = (proj / "build" / "config.toml").read_text(encoding="utf-8")
+        check("--license + 提供公钥后注入 [license] public_key", 'public_key = "test-pubkey-b64"' in conf)
+    finally:
+        os.environ.pop(b.LICENSE_PUBKEY_ENV, None)
+finally:
+    os.chdir(cwd)
+    for k, v in saved_env.items():
+        if v is not None:
+            os.environ[k] = v
+
+shutil.rmtree(tmp, ignore_errors=True)
+
+print("\n" + ("=" * 40))
+if failures:
+    print(f"❌ {len(failures)} 项未通过: {failures}")
+    sys.exit(1)
+print("✅ 全部通过")
diff --git a/test/test_logging_fork.py b/test/test_logging_fork.py
new file mode 100644
index 0000000..1884339
--- /dev/null
+++ b/test/test_logging_fork.py
@@ -0,0 +1,100 @@
+"""
+fork 安全：文件日志流在子进程重开（reentrant call 修复）测试
+
+背景：gateway 的日志是"队列 handler + 监听线程"结构，文件 handler 的流在父进程
+打开。granian/gunicorn 的 worker 由 master **fork** 而来（实测 worker 与 master 的
+fd 表完全一致）——子进程会继承父进程的 BufferedWriter，其内部锁/owner 状态在
+fork 瞬间可能正被父进程监听线程占用（跨 fork 复制不可靠）。症状：冻结版（Nuitka）
+里每个 worker 的**每条文件日志**都报
+  RuntimeError: reentrant call inside <_io.BufferedWriter ...>
+（控制台 handler 正常，文件日志整体失效）。
+
+修复：_after_fork_child 里把文件 handler 的流在子进程重开（新文件描述符）。
+本测试验证：
+  1. fork 后子进程的 handler.stream 不再是继承的那个对象（结构保证）
+  2. 子进程能真正把记录写进日志文件（行为保证——旧实现下这里写不进去）
+  3. 控制台等非文件 handler 不被动（无 _open，跳过）
+
+运行: uv run python test_logging_fork.py
+"""
+import importlib.util
+import logging
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+tmp = tempfile.TemporaryDirectory()
+log_dir = Path(tmp.name)
+
+# 与 test_logging_async.py 相同的加载方式：独立加载 logger 模块并把日志目录重定向
+_logger_path = (
+    Path(__file__).resolve().parent.parent / "gateway" / "src" / "chongming" / "app" / "core" / "logger.py"
+)
+_spec = importlib.util.spec_from_file_location("chongming.app.core.logger", _logger_path)
+logger_mod = importlib.util.module_from_spec(_spec)
+sys.modules["chongming.app.core.logger"] = logger_mod
+_spec.loader.exec_module(logger_mod)  # type: ignore[union-attr]
+logger_mod.log_config["file"] = str(log_dir)
+
+failures = []
+
+
+def check(name, cond, detail=""):
+    status = "PASS" if cond else "FAIL"
+    print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
+    if not cond:
+        failures.append(name)
+
+
+check("os.register_at_fork 已注册 after_in_child", hasattr(os, "register_at_fork"))
+
+logger = logging.getLogger("forktest")
+logger_mod.setup_logging(logger, "forktest")
+
+file_handler = next(
+    h for h in logger_mod._listener_handlers if isinstance(h, logger_mod.DailyRotatingFileHandler)
+)
+# 同步直写一条（不经队列），保持监听线程与流都活着，便于接下来 fork
+parent_record = logging.LogRecord("forktest", logging.INFO, __file__, 0, "parent-line", (), None)
+file_handler.handle(parent_record)
+file_handler.flush()
+
+check("父进程日志已落盘", "parent-line" in next(iter(log_dir.glob("forktest-*.log"))).read_text(encoding="utf-8"))
+
+inherited_stream = file_handler.stream
+check("父进程流已打开", inherited_stream is not None)
+
+# ---- fork：子进程应换掉继承的流，并能独立写日志 ----
+pid = os.fork()
+if pid == 0:  # 子进程
+    try:
+        reopened = file_handler.stream is not inherited_stream
+        record = logging.LogRecord(
+            "forktest", logging.INFO, __file__, 0, "child-line", (), None
+        )
+        # 直接同步走 handler（不经队列），写入结果立即可断言
+        file_handler.handle(record)
+        os._exit(0 if reopened else 3)
+    except BaseException:  # noqa: BLE001 - 子进程异常用退出码传出
+        os._exit(4)
+
+_, status = os.waitpid(pid, 0)
+exit_code = os.waitstatus_to_exitcode(status)
+check("fork 后子进程的流已换新（不再是继承对象）", exit_code == 0, f"exit={exit_code}")
+check(
+    "子进程可直接写入日志文件（旧实现下 reentrant 失败）",
+    "child-line" in next(iter(log_dir.glob("forktest-*.log"))).read_text(encoding="utf-8"),
+)
+check(
+    "控制台 handler 不受影响（仍指向 stderr）",
+    any(isinstance(h, logging.StreamHandler) and not isinstance(h, logger_mod.DailyRotatingFileHandler) for h in logger_mod._listener_handlers),
+)
+
+tmp.cleanup()
+
+print("\n" + ("=" * 40))
+if failures:
+    print(f"❌ {len(failures)} 项未通过: {failures}")
+    sys.exit(1)
+print("✅ 全部通过")
diff --git a/test/test_migrate_plugin.py b/test/test_migrate_plugin.py
new file mode 100644
index 0000000..802beaa
--- /dev/null
+++ b/test/test_migrate_plugin.py
@@ -0,0 +1,280 @@
+"""
+migrate_plugin（内置插件 → 进程插件 一键迁移）测试
+
+覆盖:
+  1. 前置拦截: 自定义 FastAPI 路由（APIRouter 实例化 / 顶层 router 绑定 /
+     star-import 再导出）拒绝迁移；--force 可跳过
+  2. 前置拦截: 网关包导入（import app / from chongming...）拒绝迁移
+  3. 放行: `from . import router`（子模块，非 APIRouter）不误伤（ip_sound_column 写法）
+  4. 迁移产物: 项目骨架（pyproject/start.sh/config.toml/README/zenoh 配置）+ 代码/配置
+     迁移 + [workers] 改写（module=workers.<name>、load_mode=external）+ 保留其余配置
+  5. 依赖迁移: plugins_deps.toml [<name>] 节 → 新项目 pyproject（按实际 import 过滤、
+     保留根 pyproject 的 extras 写法、共用节带入）；[<name>] 节注释掉、其它节不动
+  6. --dry-run 不落盘；目标已存在 / 插件不存在时拒绝
+
+运行: uv run python test/test_migrate_plugin.py
+"""
+import contextlib
+import io
+import os
+import shutil
+import sys
+import tempfile
+import tomllib
+from pathlib import Path
+
+ROOT = Path(__file__).parent.parent
+sys.path.insert(0, str(ROOT / "utils" / "cli" / "src"))
+
+os.environ.pop("CHONGMING_ROOT", None)  # 根目录定位不应受外部环境变量影响
+
+from chongming_cli import migrate_plugin as mp  # noqa: E402
+
+failures = []
+
+
+def check(name, cond, detail=""):
+    status = "PASS" if cond else "FAIL"
+    print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
+    if not cond:
+        failures.append(name)
+
+
+DEPS_TOML = """\
+imports = ["httpx"]
+
+[foo]
+imports = [
+  "volcenginesdkarkruntime",
+  "chongming_cache",
+  "orjson",
+  "totally_missing_pkg",
+]
+
+[bar]
+imports = ["openai"]
+"""
+
+PLUGIN_INIT = '''"""fixture 插件"""
+import orjson
+import httpx
+from chongming_cache import Cache  # noqa: F401
+from chongming_worker.plugin import Plugin
+from volcenginesdkarkruntime import AsyncArk  # noqa: F401
+from totally_missing_pkg import thing  # noqa: F401
+
+from . import router  # noqa: F401  触发 @rpc_method 注册
+{extra}
+
+class FooPlugin(Plugin):
+    type = "foo"
+    db_name = None
+
+    async def on_init(self):
+        pass
+
+
+plugin = FooPlugin()
+create_workers = plugin.create_workers
+cleanup = plugin.cleanup
+'''
+
+CONFIG_TOML = """\
+# foo Worker 插件配置
+[workers]
+type = "foo"
+module = "plugins.foo"
+factory = "create_workers"
+cleanup = "cleanup"
+count = 2
+
+[worker]
+name = "foo"
+version = "0.1.0"
+description = "fixture 插件"
+config_version = "1.0"
+defaults = { method = "POST", timeout = 30.0 }
+
+[worker.registration]
+router_prefix = "/foo"
+
+[workers.config]
+data_dir = "./data"
+api_key = "PLACEHOLDER"
+"""
+
+
+def make_root(tmp: Path) -> Path:
+    """最小仓库现场：根 pyproject（含 extras 依赖）/ config/zenoh_config.json5 副本 / plugins_deps.toml。"""
+    tmp.mkdir(parents=True, exist_ok=True)
+    (tmp / "pyproject.toml").write_text(
+        "[project]\nname = 'fixture'\ndependencies = [\"volcengine-python-sdk[ark]\"]\n",
+        encoding="utf-8",
+    )
+    (tmp / "config").mkdir(parents=True, exist_ok=True)
+    (tmp / "config" / "zenoh_config.json5").write_text("{}\n", encoding="utf-8")
+    (tmp / "plugins_deps.toml").write_text(DEPS_TOML, encoding="utf-8")
+    return tmp
+
+
+def make_plugin(tmp: Path, name: str = "foo", init_extra: str = "", files: dict | None = None):
+    """造内置插件现场：src/plugins/<name>/ + config/workers/<name>/。"""
+    code = tmp / "src" / "plugins" / name
+    (code / "router").mkdir(parents=True, exist_ok=True)
+    (code / "models").mkdir()
+    (code / "models" / "__init__.py").write_text("", encoding="utf-8")
+    (code / "__init__.py").write_text(PLUGIN_INIT.format(extra=init_extra), encoding="utf-8")
+    (code / "router" / "__init__.py").write_text("from . import hello  # noqa: F401\n", encoding="utf-8")
+    (code / "router" / "hello.py").write_text("async def hello():\n    return {}\n", encoding="utf-8")
+    conf = tmp / "config" / "workers" / name
+    conf.mkdir(parents=True, exist_ok=True)
+    (conf / "config.toml").write_text(CONFIG_TOML, encoding="utf-8")
+    (conf / "hello.toml").write_text("[POST]\nsummary = \"hi\"\n", encoding="utf-8")
+    for rel, text in (files or {}).items():
+        path = code / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(text, encoding="utf-8")
+    return tmp
+
+
+def run(tmp: Path, *args, name: str = "foo") -> tuple[int, str]:
+    """chdir 到现场执行迁移，捕获 stdout。"""
+    buf = io.StringIO()
+    cwd = Path.cwd()
+    os.chdir(tmp)
+    try:
+        with contextlib.redirect_stdout(buf):
+            rc = mp.migrate_plugin_package(name, **kwargs_for(args))
+    finally:
+        os.chdir(cwd)
+    return rc, buf.getvalue()
+
+
+def kwargs_for(args):
+    kwargs = {}
+    for a in args:
+        kwargs[a] = True
+    return kwargs
+
+
+ROUTER_FILES = {
+    "routes.py": (
+        "from fastapi import APIRouter\n\n"
+        "router = APIRouter(prefix=\"/foo\")\n"
+    ),
+}
+
+tmp_root = Path(tempfile.mkdtemp(prefix="migrate_plugin_test_"))
+
+# ========== 1. 自定义路由拦截（aibox 写法） ==========
+case = make_root(tmp_root / "case_router")
+make_plugin(case, init_extra="from .routes import router  # noqa: F401", files=ROUTER_FILES)
+rc, out = run(case)
+check("路由挂载（from .routes import router + APIRouter）拒绝迁移", rc == 1, f"rc={rc}")
+check("拒绝时给出路由提示", "APIRouter" in out and "拒绝迁移" in out)
+check("拒绝时不动任何文件", (case / "src" / "plugins" / "foo").is_dir() and not (case / "workers").exists())
+
+# ========== 2. star-import 再导出的 APIRouter 同样拦截 ==========
+case = make_root(tmp_root / "case_star")
+make_plugin(
+    case,
+    init_extra="from .router import *  # noqa: F401,F403",
+    files={},  # router/ 由 make_plugin 建好，下面覆盖 __init__.py 为 APIRouter 定义
+)
+(case / "src" / "plugins" / "foo" / "router" / "__init__.py").write_text(
+    "from fastapi import APIRouter\n\nrouter = APIRouter()\n", encoding="utf-8"
+)
+rc, out = run(case)
+check("star-import 链上的 APIRouter 拒绝迁移", rc == 1 and "APIRouter" in out, f"rc={rc}")
+
+# ========== 3. --force 跳过拦截 ==========
+rc, out = run(case, "force")
+check("--force 跳过后可迁移", rc == 0 and (case / "workers" / "foo" / "pyproject.toml").is_file(), f"rc={rc}")
+check("--force 输出回显跳过", "跳过上述拦截" in out)
+check(
+    "配置缺 module/load_mode 时自动补全",
+    'module = "workers.foo"' in (case / "workers" / "foo" / "config" / "workers" / "foo" / "config.toml").read_text(),
+)
+
+# ========== 4. 网关包导入拦截 ==========
+case = make_root(tmp_root / "case_gw")
+make_plugin(case, init_extra="from app.core.logger import get_logger  # noqa: F401")
+rc, out = run(case)
+check("网关包导入拒绝迁移", rc == 1 and "网关" in out, f"rc={rc}")
+
+# ========== 5. dry-run 不落盘 ==========
+case = make_root(tmp_root / "case_dry")
+make_plugin(case)
+before = (case / "plugins_deps.toml").read_text(encoding="utf-8")
+rc, out = run(case, "dry_run")
+check("--dry-run 返回成功且不落盘", rc == 0 and not (case / "workers").exists())
+check("--dry-run 不改 plugins_deps.toml", (case / "plugins_deps.toml").read_text(encoding="utf-8") == before)
+check("--dry-run 打印演练提示", "未改动任何文件" in out)
+
+# ========== 6. 目标已存在 / 插件不存在 ==========
+case = make_root(tmp_root / "case_exists")
+make_plugin(case)
+(case / "workers" / "foo").mkdir(parents=True)
+rc, out = run(case)
+check("目标已存在拒绝迁移", rc == 1 and "目标已存在" in out)
+case = make_root(tmp_root / "case_missing")
+rc, out = run(case)
+check("插件不存在拒绝迁移", rc == 1 and "未找到内置插件" in out)
+
+# ========== 7. 完整迁移（happy path） ==========
+case = make_root(tmp_root / "case_full")
+make_plugin(case)
+rc, out = run(case)
+project = case / "workers" / "foo"
+check("完整迁移成功", rc == 0 and project.is_dir(), f"rc={rc}")
+
+code_dst = project / "workers" / "foo"
+conf_dst = project / "config" / "workers" / "foo"
+check(
+    "代码迁移（含 router/ 与 models/）",
+    (code_dst / "__init__.py").is_file()
+    and (code_dst / "router" / "hello.py").is_file()
+    and (code_dst / "models" / "__init__.py").is_file(),
+)
+check("路由配置迁移", (conf_dst / "hello.toml").is_file())
+conf_text = (conf_dst / "config.toml").read_text(encoding="utf-8")
+check(
+    "[workers] 改写: module=workers.foo / load_mode=external / 其余保留",
+    'module = "workers.foo"' in conf_text
+    and 'load_mode = "external"' in conf_text
+    and 'count = 2' in conf_text
+    and 'api_key = "PLACEHOLDER"' in conf_text,
+)
+check("项目骨架生成", all((project / f).is_file() for f in ("config.toml", "pyproject.toml", "README.md", "start.sh")))
+check("zenoh 配置复制", (project / "config" / "zenoh_config.json5").is_file())
+check("start.sh 可执行且入口正确", (project / "start.sh").stat().st_mode & 0o111 and "chongming-worker --type foo" in (project / "start.sh").read_text())
+check("README 标注迁移来源", "migrate_plugin foo" in (project / "README.md").read_text())
+check("源目录已删除", not (case / "src" / "plugins" / "foo").exists() and not (case / "config" / "workers" / "foo").exists())
+
+pyproject = tomllib.loads((project / "pyproject.toml").read_text(encoding="utf-8"))
+deps = pyproject["project"]["dependencies"]
+check("新项目依赖: chongming-worker + 迁移带入", "chongming-worker" in deps and "httpx" in deps and "orjson" in deps and "chongming-cache" in deps, str(deps))
+check("extras 写法沿用根 pyproject", "volcengine-python-sdk[ark]" in deps, str(deps))
+check("未解析/未使用的依赖不进新项目", not any("totally_missing_pkg" in d for d in deps), str(deps))
+check("未解析依赖有告警", "totally_missing_pkg" in out)
+
+deps_data = tomllib.loads((case / "plugins_deps.toml").read_text(encoding="utf-8"))
+check(
+    "plugins_deps.toml: [foo] 节注释掉、共用节与其它节不动",
+    "foo" not in deps_data and deps_data.get("imports") == ["httpx"] and deps_data.get("bar", {}).get("imports") == ["openai"],
+    str(deps_data),
+)
+
+# ========== 8. ip_sound_column 写法（from . import router 子模块）不误伤 ==========
+case = make_root(tmp_root / "case_submod")
+make_plugin(case, init_extra="from . import router  # noqa: F401  触发 @rpc_method 注册")
+rc, out = run(case, "dry_run")
+check("from . import router（子模块）不误伤", rc == 0, f"rc={rc}")
+
+shutil.rmtree(tmp_root, ignore_errors=True)
+
+print("\n" + ("=" * 40))
+if failures:
+    print(f"❌ {len(failures)} 项未通过: {failures}")
+    sys.exit(1)
+print("✅ 全部通过")
diff --git a/test/test_nuitka_pack_flags.py b/test/test_nuitka_pack_flags.py
new file mode 100644
index 0000000..46d751d
--- /dev/null
+++ b/test/test_nuitka_pack_flags.py
@@ -0,0 +1,123 @@
+"""
+Nuitka 打包资源档位 / 链接器注入 测试（不触发真实编译）
+
+覆盖 build.py 里 nuitka 打包的三处防坑逻辑：
+  1. 链接器：装了 ld.lld 时自动经 LDFLAGS 注入 -fuse-ld=lld（默认 GNU ld 链接大工程
+     峰值数 GB，小内存机器/WSL 会被 OOM 杀掉："clang: unable to execute command: Killed"）；
+     已有 LDFLAGS 时不覆盖（用户在 CHONGMING_NUITKA_PGO 之外自行指定链接器优先）
+  2. 内存档位：可用内存 ≥ NUITKA_LTO_MIN_GB 完整优化（LTO）；低于则去 LTO（LTO 链接
+     峰值 3 GB+，实测 ld/lld 都被 OOM 杀过）；更低（或 CHONGMING_NUITKA_LOW_MEM=1）
+     进一步 --low-memory + 降并发；CHONGMING_NUITKA_FULL=1 强制完整优化
+  3. PGO 缺省关闭：--pgo-c 与 standalone/onefile 不兼容且 profiling 运行会等程序退出
+     （常驻服务卡死），只能由 CHONGMING_NUITKA_PGO=1 显式打开——
+     这里用源码断言兜住"默认关"这条（防止回潮）
+
+运行: uv run python test/test_nuitka_pack_flags.py
+"""
+import io
+import contextlib
+import os
+import shutil
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).parent.parent
+sys.path.insert(0, str(ROOT / "utils" / "cli" / "src"))
+
+from chongming_cli import build as b  # noqa: E402
+
+failures = []
+
+
+def check(name, cond, detail=""):
+    status = "PASS" if cond else "FAIL"
+    print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
+    if not cond:
+        failures.append(name)
+
+
+def quiet_note():
+    """调 _nuitka_env_and_mem_note 时屏蔽它打印的降档提示。"""
+    with contextlib.redirect_stdout(io.StringIO()):
+        return b._nuitka_env_and_mem_note()
+
+
+saved = {k: os.environ.pop(k, None) for k in
+         (b.NUITKA_LOW_MEM_ENV, b.NUITKA_FULL_ENV, b.NUITKA_PGO_ENV)}
+
+try:
+    # ========== 1. 链接器注入 ==========
+    saved_ldflags = os.environ.pop("LDFLAGS", None)
+    env, note, profile = quiet_note()
+    if sys.platform.startswith("linux") and shutil.which("ld.lld"):
+        check("装了 lld 自动注入 -fuse-ld=lld", env.get("LDFLAGS", "").endswith("-fuse-ld=lld"), env.get("LDFLAGS"))
+        check("成功信息回显链接器", "lld" in note, note)
+    else:
+        check("无 lld 时不注入（本机无 lld，跳过）", "-fuse-ld" not in (env.get("LDFLAGS") or ""))
+
+    os.environ["LDFLAGS"] = "-Wl,--as-needed"
+    env, _, _ = quiet_note()
+    check("已有 LDFLAGS 时追加而不覆盖", env["LDFLAGS"].startswith("-Wl,--as-needed"), env["LDFLAGS"])
+    if saved_ldflags is None:
+        os.environ.pop("LDFLAGS", None)
+    else:
+        os.environ["LDFLAGS"] = saved_ldflags
+
+    # ========== 2. 内存档位 ==========
+    mem = b._mem_available_gb()
+    check("_mem_available_gb 可读（Linux）", mem > 0 or not sys.platform.startswith("linux"), f"{mem:.1f} GB")
+
+    env, note, profile = quiet_note()
+    if mem >= 0:
+        expect = (
+            "squeeze" if mem < b.NUITKA_SQUEEZE_GB
+            else "no-lto" if mem < b.NUITKA_LTO_MIN_GB
+            else "full"
+        )
+        check(
+            "默认档位与内存阈值一致",
+            profile == expect,
+            f"{mem:.1f} GB → {profile}（阈值 {b.NUITKA_SQUEEZE_GB}/{b.NUITKA_LTO_MIN_GB} GB）",
+        )
+        check(
+            "非 full 档在提示里说明原因",
+            (profile == "full" and "LTO" not in note) or (
+                profile == "no-lto" and "未启用 LTO" in note
+            ) or (profile == "squeeze" and "--low-memory" in note),
+            note,
+        )
+
+    os.environ[b.NUITKA_LOW_MEM_ENV] = "1"
+    env, note, profile = quiet_note()
+    check(f"{b.NUITKA_LOW_MEM_ENV}=1 强制 squeeze 档", profile == "squeeze" and "--low-memory" in note, note)
+
+    os.environ.pop(b.NUITKA_LOW_MEM_ENV)
+    os.environ[b.NUITKA_FULL_ENV] = "1"
+    env, note, profile = quiet_note()
+    check(f"{b.NUITKA_FULL_ENV}=1 强制完整优化（覆盖内存自动判断）", profile == "full" and "--low-memory" not in note, note)
+    os.environ.pop(b.NUITKA_FULL_ENV)
+
+    # ========== 3. PGO 缺省关闭（源码兜底，防回潮） ==========
+    src = (ROOT / "utils" / "cli" / "src" / "chongming_cli" / "build.py").read_text(encoding="utf-8")
+    check(
+        "PGO 参数只在 NUITKA_PGO_ENV 开关下出现",
+        f'os.environ.get(NUITKA_PGO_ENV) == "1"' in src and b.NUITKA_PGO_ENV == "CHONGMING_NUITKA_PGO",
+    )
+    check(
+        "Nuitka 缺 patchelf 时提前失败（链接后才检查，白跑一整轮）",
+        "patchelf" in src and "缺少 patchelf" in src,
+    )
+    check(
+        "Nuitka 失败会抛错而不是静默返回（构建要非零退出）",
+        'raise RuntimeError("Nuitka 编译失败' in src,
+    )
+finally:
+    for k, v in saved.items():
+        if v is not None:
+            os.environ[k] = v
+
+print("\n" + ("=" * 40))
+if failures:
+    print(f"❌ {len(failures)} 项未通过: {failures}")
+    sys.exit(1)
+print("✅ 全部通过")
diff --git a/test/test_parent_death.py b/test/test_parent_death.py
new file mode 100644
index 0000000..0e4933c
--- /dev/null
+++ b/test/test_parent_death.py
@@ -0,0 +1,147 @@
+"""
+父进程死亡防护（PR_SET_PDEATHSIG）测试
+
+背景与语义见 gateway/src/chongming/app/core/parent_death.py：进程树 onefile 引导 →
+granian master → worker（fork）。上游被 SIGKILL 时 worker 会成孤儿（继续监听
+端口、挂在服务发现里，其它网关派发到"半死副本"每次白吃 2 秒超时）。守卫让内核
+在父进程死亡时给本进程发 SIGTERM（走既有优雅停机摘除注册）。
+
+验证（三层 fork 复刻真实进程树：本测试(引导) → 中间进程(master) → 孙进程(worker)）：
+  1. 关闭开关 CHONGMING_NO_PARENT_GUARD=1 → 不生效（返回 False）
+  2. 中间进程 SIGKILL 自己（模拟 master 被 -9，内核不可捕获）→ 已设守卫的孙进程
+     收到 SIGTERM 并退出
+  3. 对照：同样结构但不设守卫 → 孙进程收不到信号（证明信号确实来自 PDEATHSIG，
+     而不是别的东西清理孤儿）
+  4. 幂等：重复调用不重复 prctl
+
+运行: uv run python test/test_parent_death.py
+"""
+import importlib.util
+import os
+import signal
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).parent.parent
+_PD_PATH = ROOT / "gateway" / "src" / "chongming" / "app" / "core" / "parent_death.py"
+
+# 独立加载（parent_death 无包内依赖，避免连带导入整个 app 包）
+_spec = importlib.util.spec_from_file_location("chongming.app.core.parent_death", _PD_PATH)
+pd = importlib.util.module_from_spec(_spec)
+sys.modules["chongming.app.core.parent_death"] = pd
+_spec.loader.exec_module(pd)  # type: ignore[union-attr]
+
+failures = []
+
+
+def check(name, cond, detail=""):
+    status = "PASS" if cond else "FAIL"
+    print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
+    if not cond:
+        failures.append(name)
+
+
+# ========== 1. 关闭开关 ==========
+os.environ[pd.GUARD_OFF_ENV] = "1"
+check("关闭开关打开时不生效", pd.guard_parent_death() is False)
+os.environ.pop(pd.GUARD_OFF_ENV, None)
+
+# ========== 2/3. 三层 fork：守卫组 / 对照组 ==========
+tmp = Path(tempfile.mkdtemp(prefix="parent_death_test_"))
+
+
+def scenario(name: str, use_guard: bool) -> dict:
+    """中间进程 SIGKILL 自己，观察孙进程是否收到 SIGTERM。"""
+    d = tmp / name
+    d.mkdir()
+    ready, dead, pidf = d / "ready", d / "dead", d / "gc.pid"
+
+    middle = os.fork()
+    if middle == 0:  # 中间进程（master 角色）
+        gc = os.fork()
+        if gc == 0:  # 孙进程（worker 角色）
+            def _on_term(*_):
+                dead.write_text("term", encoding="utf-8")
+                os._exit(0)
+
+            signal.signal(signal.SIGTERM, _on_term)
+            if use_guard:
+                pd.guard_parent_death()
+            pidf.write_text(str(os.getpid()), encoding="utf-8")
+            ready.write_text("1", encoding="utf-8")
+            time.sleep(15)  # 等信号；超时说明没收到
+            os._exit(1)
+        for _ in range(250):  # 等孙进程就绪，再硬杀自己
+            if ready.exists():
+                break
+            time.sleep(0.02)
+        os.kill(os.getpid(), signal.SIGKILL)
+        os._exit(1)  # 不会走到
+
+    os.waitpid(middle, 0)  # 中间进程必死于 SIGKILL
+    gc_pid = int(pidf.read_text(encoding="utf-8"))
+    for _ in range(100):  # 最多等 5s
+        if dead.exists():
+            break
+        time.sleep(0.05)
+    got = dead.exists()
+    try:  # 清理（对照组孙进程仍活着）
+        os.kill(gc_pid, signal.SIGKILL)
+    except ProcessLookupError:
+        pass
+    return {"got_term": got, "gc_pid": gc_pid}
+
+
+guarded = scenario("guarded", use_guard=True)
+check(
+    "master(SIGKILL) 死后，已设守卫的 worker 收到 SIGTERM 退出",
+    guarded["got_term"],
+    f"gc={guarded['gc_pid']}",
+)
+
+control = scenario("control", use_guard=False)
+check(
+    "对照组（未设守卫）收不到信号 —— 证明退出确实来自 PDEATHSIG",
+    not control["got_term"],
+    f"gc={control['gc_pid']}",
+)
+
+# ========== 4. 幂等（在子进程里做：避免给测试进程本身设 PDEATHSIG） ==========
+pid = os.fork()
+if pid == 0:
+    a = pd.guard_parent_death()
+    b = pd.guard_parent_death()
+    os._exit(0 if (a is True and b is True) else 1)
+_, status = os.waitpid(pid, 0)
+check("重复调用幂等（不重复 prctl）", os.waitstatus_to_exitcode(status) == 0)
+
+# ========== 5. fork 继承回归（master 先设 → worker 仍必须自己设上） ==========
+# 背景：若用模块级"已设置"标志位做幂等，fork 的子进程会继承 True 而短路——
+# 但内核在 fork 时恰好清零 PDEATHSIG，于是 worker 其实没设上（本轮实测到过）。
+# 这里父进程先真实设置，再 fork，断言子进程调用守卫时**内核状态确实被设上**。
+libc5 = pd._load_libc()
+pd.guard_parent_death()  # 父进程（本测试进程）真实设置一次
+pid = os.fork()
+if pid == 0:
+    cleared = pd._current_pdeathsig(libc5) == 0  # fork 后内核已清零（不继承）
+    ok = pd.guard_parent_death() is True  # 回归点：不能因父进程设过而短路
+    set_now = pd._current_pdeathsig(libc5) == int(signal.SIGTERM)
+    os._exit(0 if (cleared and ok and set_now) else 1)
+_, status = os.waitpid(pid, 0)
+libc5.prctl(pd._PR_SET_PDEATHSIG, 0)  # 清除测试进程自己的设置，避免副作用
+check(
+    "父进程设过后，fork 的子进程仍会真实设置 PDEATHSIG",
+    os.waitstatus_to_exitcode(status) == 0,
+)
+
+import shutil
+
+shutil.rmtree(tmp, ignore_errors=True)
+
+print("\n" + ("=" * 40))
+if failures:
+    print(f"❌ {len(failures)} 项未通过: {failures}")
+    sys.exit(1)
+print("✅ 全部通过")
diff --git a/test/test_plugin_deps.py b/test/test_plugin_deps.py
new file mode 100644
index 0000000..083011c
--- /dev/null
+++ b/test/test_plugin_deps.py
@@ -0,0 +1,226 @@
+"""
+插件专属依赖清单（plugins_deps.toml）测试
+
+"插件专属依赖跟插件走、不混进网关构建配置"这条约定的守卫：
+  1. 仓库清单可解析：doubao 专属（openai / volcengine / bytedance）与共用（httpx）
+     都能被 build 的读取/合并逻辑读出来
+  2. 收集清单 = config.toml [pyinstaller_build_import] + plugins_deps.toml，
+     去重且网关本体依赖在前；文件缺失时不报错、不追加
+  3. 拆分不回潮：插件专属依赖不再出现在 config.toml 里（防止又手工加回网关清单）
+  4. 配置写错（imports 非字符串列表 / 顶层多余键 / 插件节缺 imports）直接抛错，
+     不静默漏收依赖（漏了要到运行期加载插件才 ImportError）
+  5. venv 里 chongming-cli 版本低于本地源码时告警：打包用的是 venv 里的 CLI，
+     不先发布升级，plugins_deps.toml 不会生效
+
+运行: uv run python test/test_plugin_deps.py
+"""
+import importlib.metadata
+import os
+import shutil
+import sys
+import tempfile
+import tomllib
+from pathlib import Path
+
+ROOT = Path(__file__).parent.parent
+sys.path.insert(0, str(ROOT / "utils" / "cli" / "src"))
+
+DOUBAO_DEPS = {
+    "openai",
+    "volcenginesdkarkruntime",
+    "bytedance",
+    "bytedance.volcengine_aicc_sdk",
+    "bytedance.volcengine_aicc_sdk.secure_http_client",
+}
+
+# build 的根目录定位支持 CHONGMING_ROOT 环境变量，测试结论不应受它影响
+os.environ.pop("CHONGMING_ROOT", None)
+
+failures = []
+
+
+def check(name, cond, detail=""):
+    status = "PASS" if cond else "FAIL"
+    print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
+    if not cond:
+        failures.append(name)
+
+
+try:
+    from chongming_cli import build as b  # noqa: E402
+
+    HAVE_CLI = True
+except ImportError as exc:  # utils/cli 子仓库未就位（干净克隆）时只跑清单一致性检查
+    HAVE_CLI = False
+    print(f"[SKIP] chongming_cli 源码不可用（{exc}）：仅跑清单一致性检查")
+
+cwd = Path.cwd()
+
+
+def load_repo_deps() -> dict:
+    """读仓库的 plugins_deps.toml（不依赖 chongming_cli，一致性检查也可用）。"""
+    with open(ROOT / "plugins_deps.toml", "rb") as f:
+        return tomllib.load(f)
+
+
+def flat_modules(data: dict) -> set[str]:
+    """plugins_deps.toml 里的全部模块（共用 imports + 各插件节）。"""
+    mods = set(data.get("imports", []))
+    for value in data.values():
+        if isinstance(value, dict):
+            mods |= set(value.get("imports", []))
+    return mods
+
+
+# ========== 1. 仓库清单可解析（build 的读取逻辑直测） ==========
+if HAVE_CLI:
+    os.chdir(ROOT)
+    try:
+        modules, summary = b._load_plugin_imports()
+    finally:
+        os.chdir(cwd)
+
+    check(
+        "仓库 plugins_deps.toml 可解析且含 doubao 专属依赖",
+        DOUBAO_DEPS <= set(modules),
+        f"缺失: {sorted(DOUBAO_DEPS - set(modules))}",
+    )
+    check("共用节（httpx）并入清单", "httpx" in modules)
+    check("摘要含 doubao 节名", "doubao" in summary, summary)
+
+    # ========== 2. 合并：网关本体在前、去重保序 ==========
+    fake_config = {
+        "pyinstaller_build_import": {
+            "base": ["fastapi", "openai"],  # openai 故意与插件清单重复，验证去重
+            "linux": ["uvloop"],
+            "windows": ["win_only"],
+        }
+    }
+    os.chdir(ROOT)
+    try:
+        merged, _ = b._collect_build_imports(fake_config)
+    finally:
+        os.chdir(cwd)
+
+    check(
+        "合并后网关本体依赖在前且 openai 只出现一次",
+        merged[0] == "fastapi" and merged[1] == "openai" and merged.count("openai") == 1,
+        str(merged[:6]),
+    )
+    check(
+        "插件专属依赖追加在网关清单之后",
+        "httpx" in merged and merged.index("httpx") > merged.index("openai"),
+    )
+
+# ========== 3. 拆分不回潮：config.toml 里不应再有插件专属依赖 ==========
+repo_modules = flat_modules(load_repo_deps())
+with open(ROOT / "config.toml", "rb") as f:
+    cfg = tomllib.load(f)
+bcfg = cfg["pyinstaller_build_import"]
+gateway_imports = set(bcfg["base"]) | set(bcfg["linux"]) | set(bcfg["windows"])
+
+check(
+    "doubao 专属依赖已从 config.toml 移出（只留 plugins_deps.toml）",
+    not (DOUBAO_DEPS & gateway_imports),
+    f"回潮: {sorted(DOUBAO_DEPS & gateway_imports)}",
+)
+check(
+    "两份清单不重叠（插件专属 vs 网关本体）",
+    not (repo_modules & gateway_imports),
+    f"重叠: {sorted(repo_modules & gateway_imports)}",
+)
+
+# ========== 4. 文件缺失 / 写错的行为 ==========
+tmp = Path(tempfile.mkdtemp(prefix="plugin_deps_test_"))
+(tmp / "pyproject.toml").write_text("[project]\nname = 'x'\n", encoding="utf-8")
+
+
+if HAVE_CLI:
+    (tmp / "plugins_deps.toml").unlink(missing_ok=True)
+    cfg_only = {"pyinstaller_build_import": {"base": ["fastapi"], "linux": [], "windows": []}}
+    os.chdir(tmp)
+    try:
+        mods, summary = b._load_plugin_imports()
+        merged, _ = b._collect_build_imports(cfg_only)
+    finally:
+        os.chdir(cwd)
+    check(
+        "无 plugins_deps.toml：不报错、不追加",
+        mods == [] and summary == "" and merged == ["fastapi"],
+        f"{mods} / {merged}",
+    )
+
+    def expect_error(name: str, text: str) -> None:
+        (tmp / "plugins_deps.toml").write_text(text, encoding="utf-8")
+        os.chdir(tmp)
+        try:
+            try:
+                b._load_plugin_imports()
+                check(name, False, "未抛错")
+            except ValueError as exc:
+                check(name, True, str(exc))
+        finally:
+            os.chdir(cwd)
+
+    expect_error("imports 非列表 → 报错", 'imports = "openai"\n')
+    expect_error("imports 含非字符串 → 报错", "[doubao]\nimports = [1, 2]\n")
+    expect_error("插件节缺 imports（如写成 import）→ 报错", "[doubao]\nimport = ['openai']\n")
+    expect_error("顶层多余标量键 → 报错", "foo = 1\n")
+
+    # 合法多节：共用 + 逐插件（去重发生在合并阶段，读取阶段保持原样）
+    (tmp / "plugins_deps.toml").write_text(
+        'imports = ["httpx"]\n\n[alpha]\nimports = ["pkg_a"]\n\n'
+        '[beta]\nimports = ["pkg_b", "pkg_a"]\n',
+        encoding="utf-8",
+    )
+    os.chdir(tmp)
+    try:
+        mods, summary = b._load_plugin_imports()
+    finally:
+        os.chdir(cwd)
+    check(
+        "多节解析：共用 + 逐插件均并入，摘要含各节",
+        mods == ["httpx", "pkg_a", "pkg_b", "pkg_a"]
+        and "alpha:1" in summary
+        and "beta:2" in summary,
+        f"{mods} / {summary}",
+    )
+
+shutil.rmtree(tmp, ignore_errors=True)
+
+# ========== 5. venv 里的 CLI 版本（打包装箱用的是它，不是本文件旁边的源码） ==========
+local_ver = tomllib.loads((ROOT / "utils" / "cli" / "pyproject.toml").read_text())[
+    "project"
+]["version"]
+
+
+def as_tuple(v: str):
+    return tuple(int(x) for x in v.split("."))
+
+
+try:
+    installed_ver = importlib.metadata.version("chongming-cli")
+except importlib.metadata.PackageNotFoundError:
+    installed_ver = None
+
+if installed_ver is None:
+    print("[WARN] venv 未安装 chongming-cli：先 uv sync 或发布后再构建")
+else:
+    try:
+        outdated = as_tuple(installed_ver) < as_tuple(local_ver)
+    except ValueError:  # 非常规版本号（如 rc/后缀），不猜，直接提示
+        outdated = True
+    if outdated:
+        print(
+            f"[WARN] venv 里 chongming-cli {installed_ver} < 源码 {local_ver}："
+            "plugins_deps.toml 需新版 build 才生效 —— 发布 chongming-cli 后执行 "
+            "uv lock --upgrade-package chongming-cli && uv sync"
+        )
+    else:
+        check(f"venv chongming-cli {installed_ver} ≥ 源码 {local_ver}", True)
+
+print("\n" + ("=" * 40))
+if failures:
+    print(f"❌ {len(failures)} 项未通过: {failures}")
+    sys.exit(1)
+print("✅ 全部通过")
diff --git a/test/test_worker_suspect.py b/test/test_worker_suspect.py
new file mode 100644
index 0000000..4cc5a54
--- /dev/null
+++ b/test/test_worker_suspect.py
@@ -0,0 +1,91 @@
+"""
+worker 超时冷静期（派发负反馈）测试
+
+背景：半死副本（被 SIGKILL 的实例留下的孤儿 worker 等）只要还在服务发现里，
+每次被随机选中都会白吃一整轮超时（默认 2s）；worker_manager 只做"本次请求换副本
+重试"，没有跨请求记忆，于是 2 秒惩罚反复发生（实测一轮 31 条超时日志）。
+
+修复：WorkerManager.mark_worker_suspect 给超时/无回复的副本记一段冷静期
+（_suspect_until，TTL 30s），三条选副本路径（get_worker / _pick_from_instance /
+_pick_other_worker）都优先跳过它，全在冷静期时退回全集（宁慢不缺）。
+
+验证：
+  1. 标记后的副本跨请求不再被选中
+  2. 全部候选都处于冷静期时退回全集（不饿死）
+  3. TTL 到期自动恢复
+  4. 过期项会被清理（防长期运行膨胀）
+  5. 换副本路径同样避开冷静期
+  6. 源码兜底：超时/无回复分支确实调用了标记
+
+运行: uv run python test/test_worker_suspect.py
+"""
+import asyncio
+import sys
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).parent.parent
+sys.path.insert(0, str(ROOT / "gateway" / "src"))
+
+from chongming.app.core.worker_manager import WorkerManager  # noqa: E402
+
+failures = []
+
+
+def check(name, cond, detail=""):
+    status = "PASS" if cond else "FAIL"
+    print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
+    if not cond:
+        failures.append(name)
+
+
+# 跳过 __init__（依赖 zenoh 会话/缓存），只装配选副本所需的字段
+m = WorkerManager.__new__(WorkerManager)
+m._suspect_until = {}
+m._workers_cache = {"demo": ["w1", "w2", "w3"]}
+m._local_workers = {}
+
+
+def sample(fn, n=40):
+    return {asyncio.run(fn()) for _ in range(n)}
+
+
+# ========== 1. 标记后跨请求跳过 ==========
+m.mark_worker_suspect("w1")
+m.mark_worker_suspect("w2")
+picked = sample(lambda: m.get_worker("demo"))
+check("冷静期内的副本不再被选中", picked == {"w3"}, str(picked))
+
+# ========== 2. 全在冷静期 → 退回全集 ==========
+m.mark_worker_suspect("w3")
+picked = sample(lambda: m.get_worker("demo"))
+check("全部处于冷静期时退回全集（不饿死）", picked == {"w1", "w2", "w3"}, str(picked))
+
+# ========== 3. TTL 到期恢复 ==========
+m._suspect_until = {"w1": time.monotonic() - 0.1}
+check("冷静期过期自动恢复", m._filter_suspects(["w1"]) == ["w1"])
+
+# ========== 4. 过期项清理 ==========
+m._suspect_until = {}
+for i in range(40):
+    m.mark_worker_suspect(f"x{i}", ttl=-1.0)  # 立即过期
+check("过期项会被清理（防膨胀）", len(m._suspect_until) <= 32, str(len(m._suspect_until)))
+
+# ========== 5. 换副本路径同样避开 ==========
+m._suspect_until = {}
+m.mark_worker_suspect("w1")
+picked = sample(lambda: m._pick_other_worker("demo", []))
+check("换副本路径同样避开冷静期", picked == {"w2", "w3"}, str(picked))
+
+# ========== 6. 源码兜底：失败分支确实标记 ==========
+src = (ROOT / "gateway" / "src" / "chongming" / "app" / "core" / "worker_manager.py").read_text(
+    encoding="utf-8"
+)
+n = src.count("self.mark_worker_suspect(worker_expr)")
+check("超时/无回复分支都调用了标记", n >= 2, f"{n} 处")
+
+print("\n" + ("=" * 40))
+if failures:
+    print(f"❌ {len(failures)} 项未通过: {failures}")
+    sys.exit(1)
+print("✅ 全部通过")