"""svc.sh 数据库配置可恢复性与启动门禁回归（2026-10-11 修复）。

对 scripts/svc.sh 的数据库配置统一解析做端到端验证：连续重启 / 部署 / 更新保绑定、
绑定失效拒绝启动（绝不静默回退内嵌 SQLite）、认证预检与环境变量 DSN 对齐、显式
解绑阻止环境残留重新选库、凭据不进命令行与日志。全部使用临时目录 + 桩二进制
（bash + python3 http.server 模拟 agent-system），不触碰真实 runtime/ 与真实数据库。
真实 PostgreSQL + 真实产物的验证在隔离环境另行执行并记录于任务卡。
"""
import os
import shutil
import subprocess
import socket
import tempfile
import unittest
import urllib.error
import urllib.request
from pathlib import Path

SOURCE = Path(__file__).resolve().parents[1]
SVC_SCRIPT = SOURCE / "scripts" / "svc.sh"

# 桩 agent-system：version / config explain（环境层 DSN 感知，供能力探测）/ serve
# （记录参数与 AGENT_SYSTEM_* 环境后用 http.server 应答健康探测）。TERM 转发给子
# 进程保证 stop 干净；bash 自身不 exec——pidcheck（二进制路径）须留在 cmdline。
# SVC_STUB_NO_ENV_DSN=1 模拟不支持 AGENT_SYSTEM_DSN 的旧产物。
STUB = r"""#!/usr/bin/env bash
set -u
trace_dir="${SVC_STUB_TRACE:-$PWD/.svc-stub-trace}"
mkdir -p "$trace_dir"
case "${1:-}" in
  version)
    if [ "${2:-}" = "--json" ]; then
      printf '%s\n' '{"protocol":1,"product":"agent-system","version":"stub","identity":{"ops":["ensure","verify"],"open":"existing-no-migrate","dsn":"env"},"cli":{"unknownSubcommand":"reject"}}'
      exit 0
    fi
    echo "stub-agent-system 1.0"
    exit 0 ;;
  config)
    if [ "${2:-}" = "explain" ]; then
      echo "config file: (none)"
      echo "field            value                                                    source"
      if [ -n "${AGENT_SYSTEM_DSN:-}" ] && [ -z "${SVC_STUB_NO_ENV_DSN:-}" ]; then
        printf 'dsn              postgresql://%s   env\n' "$AGENT_SYSTEM_DSN"
      else
        echo "dsn              -                                                        default"
      fi
      exit 0
    fi
    exit 0 ;;
  serve)
    shift
    case "${1:-}" in
      -h|--help)
        echo "Usage: agent-system serve -addr H:P -data DIR -plugins DIR -auth MODE -config FILE"
        exit 0 ;;
    esac
    {
      printf 'args: %s\n' "$*"
      [ -n "${AGENT_SYSTEM_DSN:-}" ] && printf 'AGENT_SYSTEM_DSN=%s\n' "$AGENT_SYSTEM_DSN"
      [ -n "${AGENT_SYSTEM_AUTH:-}" ] && printf 'AGENT_SYSTEM_AUTH=%s\n' "$AGENT_SYSTEM_AUTH"
    } > "$trace_dir/serve.txt"
    addr="" data=""
    while [ $# -gt 0 ]; do
      case "$1" in
        --addr) addr="$2"; shift 2 ;;
        --data) data="$2"; shift 2 ;;
        *) shift ;;
      esac
    done
    port="${addr##*:}"; host="${addr%:*}"
    trap 'kill "$child" 2>/dev/null; exit 0' TERM INT
    # GET/HEAD 一律 200（健康探测语义与 http.server 等价）；PUT /api/* 捕获头与
    # 请求体到 trace/put.txt 供播种断言（settings 播种验证，2026-10-11）。
    # HTTP 子进程轮询 ppid：父进程被 SIGKILL 后自行退出，避免与 recover 套件
    # 共用 1880x 端口时泄漏占用（与 test_svc_recover 桩同规）。
    python3 - "$port" "$host" "$trace_dir" "$data" <<'PYSRV' >/dev/null 2>&1 &
import json, os, sqlite3, sys, threading, time
from http.server import BaseHTTPRequestHandler, HTTPServer
port, host, trace, data = int(sys.argv[1]), sys.argv[2], sys.argv[3], sys.argv[4]
class H(BaseHTTPRequestHandler):
    def _ok(self, body=b'{"ok":true}'):
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)
    def do_GET(self):
        self._ok()
    def do_HEAD(self):
        self._ok(b'')
    def do_PUT(self):
        n = int(self.headers.get('Content-Length') or 0)
        body = self.rfile.read(n).decode('utf-8', 'replace')
        with open(trace + '/put.txt', 'a') as f:
            f.write('AUTH=%s\nBODY=%s\n--\n' % (self.headers.get('Authorization') or '-', body))
        self._adopt(body)
        self._ok()
    def _adopt(self, body):
        # 模拟当前后端：系统设置里的非掩码主模型 key 成为系统行生效密钥。
        if os.environ.get('SVC_STUB_KEEP_OLD_SECRET') == '1':
            return
        db = os.path.join(data, 'agent-system.db')
        if not db or not os.path.exists(db):
            return
        try:
            incoming = json.loads(body)
            chat = incoming.get('models', {}).get('chat', {}).get('apiKey')
        except Exception:
            return
        if not chat:
            return
        con = sqlite3.connect(db, timeout=5)
        try:
            tables = {row[0] for row in con.execute("SELECT name FROM sqlite_master WHERE type='table'")}
            if 'secret_keys' not in tables:
                return
            con.execute("UPDATE secret_keys SET active=0 WHERE owner='' AND service='models.chat'")
            con.execute(
                "INSERT INTO secret_keys(id,owner,service,label,value,active,created_at) VALUES(?,?,?,?,?,1,?)",
                ('key_seeded', '', 'models.chat', '播种', chat, '2026-10-15T00:00:00Z'),
            )
            con.commit()
        finally:
            con.close()
    def log_message(self, *args):
        pass
ppid = os.getppid()
def watch():
    while True:
        try:
            if os.getppid() != ppid:
                os._exit(0)
        except Exception:
            os._exit(0)
        time.sleep(0.3)
threading.Thread(target=watch, daemon=True).start()
try:
    HTTPServer((host, port), H).serve_forever()
except Exception:
    os._exit(1)
PYSRV
    child=$!
    wait "$child"
    ;;
  user)
    printf 'user %s\n' "$*" >> "$trace_dir/cli.txt"
    case "${2:-}" in
      list) echo '[{"owner":"op","name":"op","isAdmin":true,"mustChangePassword":false},{"owner":"plain","name":"plain","isAdmin":false,"mustChangePassword":false}]' ;;
    esac
    exit 0 ;;
  apikey)
    printf 'apikey %s\n' "$*" >> "$trace_dir/cli.txt"
    case "${2:-}" in
      create) echo "ask_stub_seed_key" ;;
      revoke) echo "revoked" ;;
    esac
    exit 0 ;;
  *)
    exit 0 ;;
esac
"""


class SvcDbRecoveryTests(unittest.TestCase):
    def setUp(self):
        tmp = tempfile.mkdtemp(prefix="svc-db-regress-")
        self.addCleanup(shutil.rmtree, tmp, True)
        self.root = Path(tmp)
        (self.root / "scripts").mkdir(parents=True)
        shutil.copy2(SVC_SCRIPT, self.root / "scripts" / "svc.sh")
        self.trace = self.root / "trace"
        self.stub = self.root / "stub-agent-system"
        self.stub.write_text(STUB)
        self.stub.chmod(0o755)
        (self.root / "plugins").mkdir()
        (self.root / "Makefile").write_text(
            "build:\n\tmkdir -p bin\n\tcp %s bin/agent-system\n" % self.stub)
        self.env = os.environ.copy()
        self.env.update({
            "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null",
            "GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0",
            "GIT_AUTHOR_NAME": "svc regression", "GIT_COMMITTER_NAME": "svc regression",
            "GIT_AUTHOR_EMAIL": "svc-test@example.invalid",
            "GIT_COMMITTER_EMAIL": "svc-test@example.invalid",
            "SVC_STUB_TRACE": str(self.trace),
            "SVC_SRC_DIR": str(self.root / "srccache"),
        })
        for key in list(self.env):
            if key.startswith("SVC_") and key not in ("SVC_STUB_TRACE", "SVC_SRC_DIR"):
                self.env.pop(key)
        for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE"):
            self.env.pop(key, None)
        # 根目录须为 git 仓（deploy --from local 取 HEAD 作 source 标签）。
        self.git(self.root, "init", "--initial-branch=main", ".")
        self.git(self.root, "add", ".")
        self.git(self.root, "commit", "-m", "fixture")

    def git(self, cwd, *args):
        proc = subprocess.run(["git", *args], cwd=cwd, env=self.env, text=True,
                              stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=60)
        self.assertEqual(proc.returncode, 0, proc.stdout)
        return proc

    def tearDown(self):
        # 兜底清理：停掉本测试遗留的服务进程，避免端口跨用例泄漏。
        runtime = self.root / "runtime"
        if runtime.is_dir():
            for d in sorted(runtime.iterdir()):
                if (d / "svc.pid").exists():
                    subprocess.run(
                        ["bash", str(self.root / "scripts" / "svc.sh"), d.name, "stop"],
                        cwd=self.root, env=self.env, timeout=60,
                        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    # ---- 夹具 ----

    def svc(self, *args, check=True, env=None):
        proc = subprocess.run(
            ["bash", str(self.root / "scripts" / "svc.sh"), *map(str, args)],
            cwd=self.root, env=env if env is not None else self.env, text=True,
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=120)
        if check:
            self.assertEqual(proc.returncode, 0, proc.stdout)
        return proc

    def runtime(self, name):
        return self.root / "runtime" / name

    def kv(self, name, key):
        envfile = self.runtime(name) / "svc.env"
        if not envfile.exists():
            return None
        for line in envfile.read_text().splitlines():
            if line.startswith(key + "="):
                return line[len(key) + 1:]
        return None

    def register_pg(self, name, port, password="pgpw-A1", user="agent",
                    dbname="agent", complete=True):
        d = self.runtime(name)
        d.mkdir(parents=True, exist_ok=True)
        rows = [("kind", "postgres"), ("env", name), ("addr", "127.0.0.1:%d" % port),
                ("tags", ""), ("source", "postgres@local"), ("version", "16"),
                ("user", user)]
        if complete:
            rows.append(("password", password))
        rows += [("dbname", dbname), ("authmode", "scram-sha-256"),
                 ("pgdata", str(d / "pgdata"))]
        (d / "svc.env").write_text("".join("%s=%s\n" % kv for kv in rows))

    def seed_agent(self, name, port, exec_args, extra_rows=(), addr=None):
        d = self.runtime(name)
        (d / "bin").mkdir(parents=True, exist_ok=True)
        (d / "data").mkdir(exist_ok=True)
        (d / "plugins").mkdir(exist_ok=True)
        shutil.copy2(self.stub, d / "bin" / "agent-system")
        rows = [("kind", "agent"), ("env", name), ("pid", ""),
                ("addr", addr or ("127.0.0.1:%d" % port)), ("tags", ""),
                ("source", "local@deadbee"), ("version", "stub"), ("started", ""),
                ("pidcheck", str(d / "bin" / "agent-system")),
                ("exec", exec_args)]
        merged = dict(rows)
        merged.update(dict(extra_rows))            # extra 覆盖基础键（如 source/version）
        (d / "svc.env").write_text(
            "".join("%s=%s\n" % kv for kv in merged.items()))

    def serve_record(self):
        f = self.trace / "serve.txt"
        return f.read_text() if f.exists() else ""

    def clear_serve_record(self):
        f = self.trace / "serve.txt"
        if f.exists():
            f.unlink()

    def start_ok(self, name, *args):
        self.clear_serve_record()
        result = self.svc(name, "start", *args)
        self.assertIn("已启动", result.stdout)
        return result

    def pid_alive(self, name):
        pidf = self.runtime(name) / "svc.pid"
        if not pidf.exists():
            return False
        try:
            os.kill(int(pidf.read_text().strip()), 0)
            return True
        except (ValueError, ProcessLookupError, PermissionError):
            return False

    # ---- 1. 存量 exec= DSN：连续启动两次均恢复，进程参数无秘密 ----

    def test_legacy_exec_dsn_survives_consecutive_restarts(self):
        self.register_pg("pg1", 18890, password="legacy-pw-1")
        dsn = "postgresql://agent:legacy-pw-1@127.0.0.1:18890/agent"
        d = self.runtime("dev")
        self.seed_agent("dev", 18791,
                        "--addr 127.0.0.1:18791 --data %s --plugins %s --auth auto --dsn %s"
                        % (d / "data", d / "plugins", dsn))
        for _ in range(2):
            self.clear_serve_record()
            self.svc("dev", "restart")
            record = self.serve_record()
            self.assertIn("AGENT_SYSTEM_DSN=%s" % dsn, record)
            self.assertNotIn("--dsn", record.splitlines()[0])
            self.assertNotIn("legacy-pw-1", record.splitlines()[0])
            self.assertEqual(self.kv("dev", "dbref"), "pg1/agent")
            self.assertEqual(self.kv("dev", "db"), "pg")
            self.assertNotIn("--dsn", self.kv("dev", "exec"))
            self.assertIsNone(self.kv("dev", "dsn"))
        # 停止后的第三次裸 start 同样恢复同一数据库配置。
        self.svc("dev", "stop")
        self.start_ok("dev")
        self.assertIn("AGENT_SYSTEM_DSN=%s" % dsn, self.serve_record())
        self.assertEqual(self.kv("dev", "dbref"), "pg1/agent")

    # ---- 2. 仅 dbref：裸 restart 与重新部署均保留绑定 ----

    def test_dbref_binding_survives_bare_restart_and_redeploy(self):
        self.register_pg("pg1", 18891, password="bind-pw-2")
        dsn = "postgresql://agent:bind-pw-2@127.0.0.1:18891/agent"
        d = self.runtime("dev")
        self.seed_agent("dev", 18792,
                        "--addr 127.0.0.1:18792 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"),
                        extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
        self.clear_serve_record()
        self.svc("dev", "restart")
        self.assertIn("AGENT_SYSTEM_DSN=%s" % dsn, self.serve_record())
        self.assertEqual(self.kv("dev", "dbref"), "pg1/agent")
        # 重新部署（换产物）不改变数据库选择。
        self.clear_serve_record()
        self.svc("dev", "deploy", "--from", "local")
        self.assertIn("AGENT_SYSTEM_DSN=%s" % dsn, self.serve_record())
        self.assertEqual(self.kv("dev", "dbref"), "pg1/agent")
        self.assertEqual(self.kv("dev", "db"), "pg")
        # 查看视图展示托管绑定。
        view = self.svc("dev", "db")
        self.assertIn("pg1/agent", view.stdout)

    # ---- 3. 改地址不改数据库与认证 ----

    def test_addr_change_keeps_database_and_auth(self):
        self.register_pg("pg1", 18892, password="addr-pw-3")
        dsn = "postgresql://agent:addr-pw-3@127.0.0.1:18892/agent"
        d = self.runtime("dev")
        self.seed_agent("dev", 18793,
                        "--addr 127.0.0.1:18793 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"),
                        extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
        self.clear_serve_record()
        self.svc("dev", "restart", "--addr", "127.0.0.1:18794")
        record = self.serve_record()
        self.assertIn("AGENT_SYSTEM_DSN=%s" % dsn, record)
        self.assertIn("--auth auto", record)
        self.assertEqual(self.kv("dev", "dbref"), "pg1/agent")
        self.assertEqual(self.kv("dev", "addr"), "127.0.0.1:18794")
        self.assertIn("--addr 127.0.0.1:18794", self.kv("dev", "exec"))
        self.assertEqual(self.kv("dev", "db"), "pg")

    # ---- 4. 外部直连与配置文件来源：后续无参数启动仍恢复 ----

    def test_external_dsn_and_conffile_recover_on_bare_start(self):
        d = self.runtime("dev")
        with socket.socket() as reservation:
            reservation.bind(("127.0.0.1", 0))
            port = reservation.getsockname()[1]
        self.seed_agent("dev", port,
                        "--addr 127.0.0.1:%d --data %s --plugins %s --auth auto"
                        % (port, d / "data", d / "plugins"))
        external = "postgresql://extu:ext-pw-4@10.9.8.7:5432/extdb"
        self.clear_serve_record()
        self.svc("dev", "deploy", "--from", "local", "--dsn", external)
        self.assertEqual(self.kv("dev", "dsn"), external)
        self.assertEqual(self.kv("dev", "db"), "pg")
        self.assertIsNone(self.kv("dev", "dbref"))
        self.assertIn("AGENT_SYSTEM_DSN=%s" % external, self.serve_record())
        self.svc("dev", "stop")
        self.start_ok("dev")
        self.assertIn("AGENT_SYSTEM_DSN=%s" % external, self.serve_record())
        # 显式重新绑定 = 切换（托管绑定替换直连，dsn= 清除，不视为冲突）。
        self.register_pg("pgother", 18893, password="other-pw")
        self.svc("dev", "db", "pgother", "--no-restart")
        self.assertEqual(self.kv("dev", "dbref"), "pgother/agent")
        self.assertIsNone(self.kv("dev", "dsn"))

        # conffile 来源：DSN 由文件提供，不经环境重复传递；db=pg 标记保留。
        conf = self.root / "conf-db.json"
        conf.write_text('{"dsn": "postgresql://cfu:cf-pw-5@127.0.0.1:18894/cfdb",'
                        ' "auth": "auto"}')
        self.svc("dev", "db", "--clear", "--no-restart")
        self.svc("dev", "config", conf, "--no-restart")
        self.assertEqual(self.kv("dev", "conffile"), str(conf))
        self.svc("dev", "stop")
        self.start_ok("dev")
        self.assertNotIn("AGENT_SYSTEM_DSN=", self.serve_record())
        self.assertEqual(self.kv("dev", "db"), "pg")
        self.assertEqual(self.kv("dev", "conffile"), str(conf))
        self.svc("dev", "stop")
        self.start_ok("dev")
        self.assertNotIn("AGENT_SYSTEM_DSN=", self.serve_record())
        self.assertEqual(self.kv("dev", "conffile"), str(conf))

    # ---- 5. 拒启场景：不降级 SQLite、不启动进程、登记可恢复、停机前发现 ----

    def test_refusals_keep_registration_without_sqlite_fallback(self):
        d = self.runtime("a1")
        self.seed_agent("a1", 18796,
                        "--addr 127.0.0.1:18796 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"),
                        extra_rows=[("dbref", "pgmissing/agent"), ("db", "pg")])
        self.clear_serve_record()
        result = self.svc("a1", "restart", check=False)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("拒绝", result.stdout)
        self.assertEqual(self.serve_record(), "")          # 未启动任何进程
        self.assertEqual(self.kv("a1", "dbref"), "pgmissing/agent")  # 可恢复配置保留

        # 实例登记不全（缺 password）→ 拒启。
        self.register_pg("pginc", 18895, complete=False)
        d = self.runtime("a2")
        self.seed_agent("a2", 18797,
                        "--addr 127.0.0.1:18797 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"),
                        extra_rows=[("dbref", "pginc/agent"), ("db", "pg")])
        result = self.svc("a2", "restart", check=False)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("拒绝", result.stdout)

        # dbref 与 dsn= 指向不同库 → 配置冲突拒启。
        self.register_pg("pg1", 18896, password="clash-pw")
        d = self.runtime("a3")
        self.seed_agent("a3", 18798,
                        "--addr 127.0.0.1:18798 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"),
                        extra_rows=[("dbref", "pg1/agent"), ("db", "pg"),
                                    ("dsn", "postgresql://u:p@127.0.0.1:18999/otherdb")])
        result = self.svc("a3", "restart", check=False)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("拒绝", result.stdout)

        # db=pg 但无可恢复 DSN → 拒启。
        d = self.runtime("a4")
        self.seed_agent("a4", 18799,
                        "--addr 127.0.0.1:18799 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"),
                        extra_rows=[("db", "pg")])
        result = self.svc("a4", "restart", check=False)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("拒绝", result.stdout)
        self.assertEqual(self.kv("a4", "db"), "pg")

    def test_restart_refusal_happens_before_stopping_running_service(self):
        self.register_pg("pg1", 18897, password="run-pw")
        d = self.runtime("a5")
        self.seed_agent("a5", 18800,
                        "--addr 127.0.0.1:18800 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"),
                        extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
        self.start_ok("a5")
        self.assertTrue(self.pid_alive("a5"))
        # 破坏实例登记（去掉 password）后 restart：预检拒绝发生在停止旧进程之前。
        self.register_pg("pg1", 18897, password="run-pw", complete=False)
        self.clear_serve_record()
        result = self.svc("a5", "restart", check=False)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("拒绝", result.stdout)
        self.assertTrue(self.pid_alive("a5"))              # 服务未被停止
        self.assertEqual(self.serve_record(), "")          # 也没有重启新进程
        self.assertEqual(self.kv("a5", "dbref"), "pg1/agent")

    # ---- 6. 认证预检与后端口径一致（环境变量 DSN 覆盖 auto）----

    def test_auth_preflight_aligns_with_effective_env_dsn(self):
        # 非回环 + auto + 无任何 DSN → 预检快速拒绝，不启动进程。
        d = self.runtime("nl1")
        self.seed_agent("nl1", 18801,
                        "--addr 10.255.255.1:18801 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"))
        self.clear_serve_record()
        result = self.svc("nl1", "start", check=False)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("拒绝启动", result.stdout)
        self.assertEqual(self.serve_record(), "")

        # 环境变量提供 DSN → auto 判为已开认证：预检放行（桩绑定非回环地址失败即退，
        # 区别于预检拒绝——证明进入启动路径且 DSN 经环境交付）。
        d = self.runtime("nl2")
        self.seed_agent("nl2", 18802,
                        "--addr 10.255.255.1:18802 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"), addr="10.255.255.1:18802")
        env_dsn = "postgresql://envu:env-pw@127.0.0.1:18898/agent"
        env2 = dict(self.env, SVC_NL2_DSN=env_dsn)
        self.clear_serve_record()
        result = self.svc("nl2", "start", check=False, env=env2)
        self.assertNotEqual(result.returncode, 0)
        self.assertNotIn("拒绝启动", result.stdout)
        self.assertIn("AGENT_SYSTEM_DSN=%s" % env_dsn, self.serve_record())

        # 显式 auth=off 覆盖环境 DSN → 仍拒绝。
        d = self.runtime("nl3")
        self.seed_agent("nl3", 18803,
                        "--addr 10.255.255.1:18803 --data %s --plugins %s --auth off"
                        % (d / "data", d / "plugins"), addr="10.255.255.1:18803")
        self.clear_serve_record()
        result = self.svc("nl3", "start", check=False,
                          env=dict(self.env, SVC_NL3_DSN=env_dsn))
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("拒绝启动", result.stdout)
        self.assertEqual(self.serve_record(), "")

    # ---- 7. 显式解绑：环境残留不再悄悄重新选库 ----

    def test_explicit_unbind_blocks_residual_env_dsn(self):
        self.register_pg("pg1", 18899, password="unbind-pw")
        d = self.runtime("dev")
        self.seed_agent("dev", 18804,
                        "--addr 127.0.0.1:18804 --data %s --plugins %s"
                        % (d / "data", d / "plugins"))
        self.svc("dev", "db", "pg1", "--no-restart")
        self.assertEqual(self.kv("dev", "dbref"), "pg1/agent")
        self.assertEqual(self.kv("dev", "db"), "pg")
        self.assertIsNone(self.kv("dev", "dsn"))
        self.assertNotIn("--dsn", self.kv("dev", "exec") or "")

        residual = "postgresql://other:other-pw@127.0.0.1:18901/other"
        env_residual = dict(self.env, SVC_DEV_DSN=residual)
        # 绑定状态下环境残留指向别库 → 冲突拒绝，不静默选其一。
        result = self.svc("dev", "restart", check=False, env=env_residual)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("拒绝", result.stdout)

        # 显式解绑后：同一环境残留被忽略，按内嵌档启动。
        self.svc("dev", "db", "--clear", "--no-restart")
        self.assertIsNone(self.kv("dev", "dbref"))
        self.assertIsNone(self.kv("dev", "dsn"))
        self.assertEqual(self.kv("dev", "db"), "sqlite")
        self.clear_serve_record()
        self.svc("dev", "restart", env=env_residual)
        self.assertNotIn("AGENT_SYSTEM_DSN=", self.serve_record())
        self.assertEqual(self.kv("dev", "db"), "sqlite")

    # ---- 8. mock 冲突与旧产物：行为明确，不隐式降级 ----

    def test_mock_conflict_and_legacy_binary_refuse(self):
        self.register_pg("pg1", 18902, password="mock-pw")
        d = self.runtime("mk1")
        self.seed_agent("mk1", 18805,
                        "--addr 127.0.0.1:18805 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"),
                        extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
        result = self.svc("mk1", "deploy", "--from", "local", "--mock", check=False)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("mock", result.stdout)
        self.assertIn("拒绝", result.stdout)
        self.assertEqual(self.kv("mk1", "dbref"), "pg1/agent")

        # 旧产物（不消费环境变量 DSN）绑定服务 restart → 拒绝，提示可操作出路。
        d = self.runtime("lb1")
        self.seed_agent("lb1", 18806,
                        "--addr 127.0.0.1:18806 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"),
                        extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
        env_legacy = dict(self.env, SVC_STUB_NO_ENV_DSN="1")
        result = self.svc("lb1", "restart", check=False, env=env_legacy)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("环境变量 DSN", result.stdout)
        self.assertEqual(self.kv("lb1", "dbref"), "pg1/agent")

    # ---- 9. 多实例顺序更新不串用 DSN；凭据不进日志与历史 ----

    def test_sequential_restarts_no_cross_dsn_and_no_secret_leak(self):
        self.register_pg("pg1", 18903, password="iso-pw-one")
        self.register_pg("pg2", 18904, password="iso-pw-two")
        d1, d2 = self.runtime("dev"), self.runtime("prod")
        self.seed_agent("dev", 18807,
                        "--addr 127.0.0.1:18807 --data %s --plugins %s --auth auto"
                        % (d1 / "data", d1 / "plugins"),
                        extra_rows=[("dbref", "pg1/agent"), ("db", "pg")])
        self.seed_agent("prod", 18808,
                        "--addr 127.0.0.1:18808 --data %s --plugins %s --auth auto"
                        % (d2 / "data", d2 / "plugins"),
                        extra_rows=[("dbref", "pg2/agent"), ("db", "pg")])
        dsn1 = "postgresql://agent:iso-pw-one@127.0.0.1:18903/agent"
        dsn2 = "postgresql://agent:iso-pw-two@127.0.0.1:18904/agent"
        for name, dsn in (("dev", dsn1), ("prod", dsn2)):
            self.clear_serve_record()
            self.svc(name, "restart")
            self.assertIn("AGENT_SYSTEM_DSN=%s" % dsn, self.serve_record())
            self.assertEqual(self.kv(name, "dbref"),
                             "pg1/agent" if name == "dev" else "pg2/agent")
        # 交叉重启仍正确（验证解析态不跨实例串用）。
        self.clear_serve_record()
        self.svc("prod", "restart")
        self.assertIn("AGENT_SYSTEM_DSN=%s" % dsn2, self.serve_record())

        # 凭据不进 svc.log、history.jsonl 与 exec=（命令行断言见各启动用例的 serve 记录）。
        for name in ("dev", "prod"):
            log = self.runtime(name) / "svc.log"
            hist = self.runtime(name) / "history.jsonl"
            blob = (log.read_text() if log.exists() else "") \
                 + (hist.read_text() if hist.exists() else "") \
                 + (self.kv(name, "exec") or "")
            self.assertNotIn("iso-pw-one", blob)
            self.assertNotIn("iso-pw-two", blob)
        # 最后一次 serve 的命令行参数行同样无秘密。
        args_line = self.serve_record().splitlines()[0] if self.serve_record() else ""
        self.assertNotIn("iso-pw-two", args_line)
        self.assertNotIn("--dsn", args_line)
        # svc.env 权限 600。
        mode = (self.runtime("dev") / "svc.env").stat().st_mode & 0o777
        self.assertEqual(mode, 0o600)

    # ---- 10. update（dev 源）保留绑定 ----

    def make_src_repo(self):
        remote = self.root / "srcremote.git"
        work = self.root / "srcwork"
        src = self.root / "srccache"
        run = lambda cwd, *args: subprocess.run(
            ["git", *args], cwd=cwd, env=self.env, text=True,
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=60,
            check=True)
        run(self.root, "init", "--bare", "--initial-branch=main", str(remote))
        run(self.root, "init", "--initial-branch=main", str(work))
        makefile = "build:\n\tmkdir -p bin\n\tcp %s bin/agent-system\n" % self.stub
        (work / "Makefile").write_text(makefile)
        run(work, "add", ".")
        run(work, "commit", "-m", "v1")
        (work / "marker").write_text("v2\n")
        run(work, "add", ".")
        run(work, "commit", "-m", "v2")
        run(work, "remote", "add", "origin", str(remote))
        run(work, "push", "-u", "origin", "main")
        run(self.root, "clone", str(remote), str(src))
        shas = run(src, "log", "--format=%h", "origin/main").stdout.split()
        return shas[1], shas[0]  # (旧, 新)

    def test_update_preserves_binding(self):
        old_sha, new_sha = self.make_src_repo()
        self.register_pg("pg1", 18905, password="upd-pw")
        dsn = "postgresql://agent:upd-pw@127.0.0.1:18905/agent"
        d = self.runtime("dev")
        self.seed_agent("dev", 18809,
                        "--addr 127.0.0.1:18809 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"),
                        extra_rows=[("dbref", "pg1/agent"), ("db", "pg"),
                                    ("source", "dev@%s" % old_sha)])
        self.clear_serve_record()
        self.svc("update", "dev")
        self.assertEqual(self.kv("dev", "source"), "dev@%s" % new_sha)
        self.assertEqual(self.kv("dev", "dbref"), "pg1/agent")
        self.assertEqual(self.kv("dev", "db"), "pg")
        self.assertIn("AGENT_SYSTEM_DSN=%s" % dsn, self.serve_record())
        # 更新后裸 restart 仍保绑定。
        self.clear_serve_record()
        self.svc("dev", "restart")
        self.assertIn("AGENT_SYSTEM_DSN=%s" % dsn, self.serve_record())

    def test_batch_update_stops_second_service_despite_prior_pidcheck(self):
        """批量 update 不得用上一服务的 SVC_PIDCHECK 误判下一服务进程。

        复现 2026-10-12：update --all 先成功部署 a，内存 SVC_PIDCHECK 残留 a 的
        bin 路径；do_stop_quiet / reclaim_own_port 对 b 匹配失败 → 误清 pidfile、
        旧进程占口 → 报「非本服务」。env_config 须按实例清空 SVC_PIDCHECK 等。
        """
        old_sha, new_sha = self.make_src_repo()
        ports = {"a": 18841, "b": 18842}
        old_pids = {}
        for name, port in ports.items():
            d = self.runtime(name)
            self.seed_agent(
                name, port,
                "--addr 127.0.0.1:%d --data %s --plugins %s --auth auto"
                % (port, d / "data", d / "plugins"),
                extra_rows=[("source", "dev@%s" % old_sha)])
            self.start_ok(name)
            pidf = d / "svc.pid"
            self.assertTrue(pidf.exists(), name)
            old_pids[name] = int(pidf.read_text().strip())
            self.assertTrue(self.pid_alive(name), name)

        result = self.svc("update", "a", "b")
        self.assertNotIn("非本服务", result.stdout, result.stdout)
        self.assertNotIn("更新失败", result.stdout, result.stdout)
        for name in ("a", "b"):
            self.assertEqual(self.kv(name, "source"), "dev@%s" % new_sha, name)
            self.assertTrue(self.pid_alive(name), "%s 应在运行" % name)
            new_pid = int((self.runtime(name) / "svc.pid").read_text().strip())
            self.assertNotEqual(new_pid, old_pids[name],
                                "%s 应已换新进程" % name)
            # 与 svc.sh / recover 夹具一致：僵尸按死处理（CI 容器 PID 1 不收割）。
            old_still = False
            try:
                os.kill(old_pids[name], 0)
                try:
                    state = Path("/proc/%d/stat" % old_pids[name]).read_text() \
                        .rsplit(")", 1)[1].split()[0]
                    old_still = state != "Z"
                except (OSError, IndexError):
                    old_still = True
            except (ProcessLookupError, PermissionError):
                old_still = False
            self.assertFalse(old_still, "%s 旧进程应已停止" % name)
            hist = (self.runtime(name) / "history.jsonl").read_text()
            self.assertIn('"event":"stop"', hist, hist)
            self.assertIn('"result":"ok"', hist)
            self.assertIn('"event":"update"', hist)


    # ---- 12. 播种认证（2026-10-11 修复）：认证开启自动铸临时 key，用后回收 ----

    def put_record(self):
        f = self.trace / "put.txt"
        return f.read_text() if f.exists() else ""

    def cli_record(self):
        f = self.trace / "cli.txt"
        return f.read_text() if f.exists() else ""

    def model_env(self, **extra):
        env = dict(self.env)
        env.update({
            "SVC_MODEL_BASE": "https://stub.example/v1",
            "SVC_MODEL_KEY": "sk-stub-model",
            "SVC_MODEL_NAME": "stub-model",
        })
        env.update(extra)
        return env

    def test_seed_puts_with_temporary_bearer_key_when_auth_on(self):
        self.register_pg("pg1", 18831, password="seed-pw-1")
        dsn = "postgresql://agent:seed-pw-1@127.0.0.1:18831/agent"
        d = self.runtime("dev")
        self.seed_agent("dev", 18832,
                        "--addr 127.0.0.1:18832 --data %s --plugins %s --auth auto --dsn %s"
                        % (d / "data", d / "plugins", dsn))
        result = self.svc("dev", "start", env=self.model_env())
        self.assertIn("已播种设置：主模型", result.stdout)
        put = self.put_record()
        self.assertIn("AUTH=Bearer ask_stub_seed_key", put)
        body_part = put.split("BODY=", 1)[1]
        self.assertIn('"model": "stub-model"', body_part)
        self.assertNotIn("ask_stub_seed_key", body_part)  # 临时 key 不进请求体
        cli = self.cli_record()
        # owner 解析：user list 首个管理员（op）→ 铸 key → 播种后回收。
        self.assertIn("user list --dsn %s" % dsn, cli)
        self.assertIn("apikey create --dsn %s --owner op --name svc-seed" % dsn, cli)
        self.assertIn("apikey revoke --dsn %s --owner op --name svc-seed" % dsn, cli)

    def test_seed_writes_fallbacks_and_vision(self):
        self.register_pg("pg1", 18951, password="seed-pw-v")
        dsn = "postgresql://agent:seed-pw-v@127.0.0.1:18951/agent"
        d = self.runtime("dev")
        self.seed_agent("dev", 18952,
                        "--addr 127.0.0.1:18952 --data %s --plugins %s --auth auto --dsn %s"
                        % (d / "data", d / "plugins", dsn))
        result = self.svc("dev", "start", env=self.model_env(
            SVC_MODEL_FALLBACKS="ZHIPU/GLM-5.3-FlashX,ZHIPU/GLM-5.3-Flash",
            SVC_VISION_MODEL="qwen3.8-flash",
            SVC_VISION_CAPABLE="deepseek-v4.1-flash,qwen3.8-flash",
        ))
        self.assertIn("已播种设置：主模型", result.stdout)
        self.assertIn("回退链", result.stdout)
        self.assertIn("视觉模型", result.stdout)
        body_part = self.put_record().split("BODY=", 1)[1]
        self.assertIn('"fallbacks": "ZHIPU/GLM-5.3-FlashX,ZHIPU/GLM-5.3-Flash"', body_part)
        self.assertIn('"model": "qwen3.8-flash"', body_part)
        self.assertIn('"capable": "deepseek-v4.1-flash,qwen3.8-flash"', body_part)

    def test_seed_owner_override_skips_account_lookup(self):
        self.register_pg("pg1", 18833, password="seed-pw-2")
        dsn = "postgresql://agent:seed-pw-2@127.0.0.1:18833/agent"
        d = self.runtime("dev")
        self.seed_agent("dev", 18834,
                        "--addr 127.0.0.1:18834 --data %s --plugins %s --auth auto --dsn %s"
                        % (d / "data", d / "plugins", dsn))
        result = self.svc("dev", "start", env=self.model_env(SVC_SEED_OWNER="boss"))
        self.assertIn("已播种设置：主模型", result.stdout)
        self.assertIn("AUTH=Bearer ask_stub_seed_key", self.put_record())
        cli = self.cli_record()
        self.assertIn("--owner boss --name svc-seed", cli)
        self.assertNotIn("user list", cli)  # 显式指定即跳过账号解析

    def test_seed_puts_without_credentials_when_auth_off(self):
        d = self.runtime("dev")
        self.seed_agent("dev", 18835,
                        "--addr 127.0.0.1:18835 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"))
        result = self.svc("dev", "start", env=self.model_env())
        self.assertIn("已播种设置：主模型", result.stdout)
        put = self.put_record()
        self.assertIn("AUTH=-", put)
        self.assertIn('"model": "stub-model"', put.split("BODY=", 1)[1])
        self.assertNotIn("apikey", self.cli_record())  # 认证关闭不铸 key

    def plant_old_chat_secret(self, name, value):
        import sqlite3
        db = self.runtime(name) / "data" / "agent-system.db"
        con = sqlite3.connect(db)
        con.execute(
            "CREATE TABLE secret_keys(id TEXT, owner TEXT, service TEXT, label TEXT, value TEXT, active INTEGER, created_at TEXT)")
        con.execute(
            "INSERT INTO secret_keys VALUES('key_old','','models.chat','迁入',?,1,'2026-10-11T00:00:00Z')",
            (value,))
        con.commit()
        con.close()
        return db

    def test_seed_replaces_old_system_secret(self):
        d = self.runtime("dev")
        self.seed_agent("dev", 18836,
                        "--addr 127.0.0.1:18836 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"))
        db = self.plant_old_chat_secret("dev", "sk-old-secret")
        result = self.svc("dev", "start", env=self.model_env())
        self.assertIn("已播种设置：主模型", result.stdout)
        self.assertNotIn("生效密钥与环境变量不一致", result.stdout)
        self.assertNotIn("sk-old-secret", result.stdout)
        self.assertNotIn("sk-stub-model", result.stdout)
        import sqlite3
        active = sqlite3.connect(db).execute(
            "SELECT value FROM secret_keys WHERE owner='' AND service='models.chat' AND active=1"
        ).fetchone()
        self.assertEqual(active[0], "sk-stub-model")

    def test_seed_warns_when_old_secret_stays_active(self):
        d = self.runtime("dev")
        self.seed_agent("dev", 18837,
                        "--addr 127.0.0.1:18837 --data %s --plugins %s --auth auto"
                        % (d / "data", d / "plugins"))
        self.plant_old_chat_secret("dev", "sk-old-secret")
        result = self.svc("dev", "start", env=self.model_env(SVC_STUB_KEEP_OLD_SECRET="1"))
        self.assertIn("已播种设置：主模型", result.stdout)
        self.assertIn("生效密钥与环境变量不一致", result.stdout)
        self.assertIn("models.chat fp=", result.stdout)
        self.assertNotIn("sk-old-secret", result.stdout)
        self.assertNotIn("sk-stub-model", result.stdout)


    # ---- 13. rdm 部署源（下游组合仓 Agent-System-RDM）：物化/产物/工作台/播种跳过 ----

    def make_rdm_repo(self, with_web=True):
        """构造下游组合仓夹具：Makefile build 产出桩 agent-system-rdm + 可选 web/。"""
        remote = self.root / "rdmremote.git"
        work = self.root / "rdmwork"
        if remote.exists():
            shutil.rmtree(remote)
        if work.exists():
            shutil.rmtree(work)
        run = lambda cwd, *args: subprocess.run(
            ["git", *args], cwd=cwd, env=self.env, text=True,
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=60,
            check=True)
        run(self.root, "init", "--bare", "--initial-branch=main", str(remote))
        run(self.root, "init", "--initial-branch=main", str(work))
        (work / "Makefile").write_text(
            "build:\n\tmkdir -p bin\n\tcp %s bin/agent-system-rdm\n" % self.stub)
        if with_web:
            (work / "web").mkdir()
            (work / "web" / "index.html").write_text("<html>rdm workbench</html>\n")
        run(work, "add", ".")
        run(work, "commit", "-m", "rdm v1")
        run(work, "remote", "add", "origin", str(remote))
        run(work, "push", "-u", "origin", "main")
        return remote, work

    def rdm_env(self, remote):
        env = dict(self.env)
        env.update({
            "SVC_RDM_SRC_DIR": str(self.root / "rdmcache"),
            "SVC_RDM_ORIGIN": str(remote),
            "SVC_MODEL_BASE": "https://stub.example/v1",
            "SVC_MODEL_KEY": "sk-stub-model",
            "SVC_MODEL_NAME": "stub-model",
        })
        return env

    def http_get(self, url, timeout=2):
        try:
            with urllib.request.urlopen(url, timeout=timeout) as resp:
                return resp.status, resp.read().decode("utf-8", "replace")
        except urllib.error.HTTPError as exc:
            return exc.code, ""
        except OSError:
            return 0, ""

    def companion_url(self, agent_name):
        addr = self.kv(agent_name + "web", "addr")
        self.assertIsNotNone(addr, "companion %sweb not registered" % agent_name)
        host, port = addr.rsplit(":", 1)
        if host in ("0.0.0.0", "::"):
            host = "127.0.0.1"
        return "http://%s:%s/" % (host, port)

    def test_rdm_source_deploy_end_to_end(self):
        remote, work = self.make_rdm_repo()
        env = self.rdm_env(remote)
        result = self.svc("rdm1", "deploy", "--from", "rdm", "--port", "18843", env=env)
        d = self.runtime("rdm1")
        # U2：组合产物入不可变 release，经 current 指针解析；插件空集、工作台随 release。
        active = d / "current"
        self.assertTrue(active.is_symlink(), "current 指针须就位")
        self.assertTrue((active / "bin" / "agent-system").exists())
        self.assertTrue((active / "bin" / "agent-system").stat().st_mode & 0o111)
        self.assertEqual(self.kv("rdm1", "kind"), "agent")
        self.assertTrue(self.kv("rdm1", "source").startswith("rdm@"))
        self.assertEqual(list((active / "plugins").iterdir()), [])
        self.assertIn("rdm workbench", (d / "htdocs" / "index.html").read_text())
        self.assertIn("args:", self.serve_record())  # 健康通过即登记启动参数
        # 伴生 static 自动部署并探活 200。
        self.assertEqual(self.kv("rdm1web", "kind"), "static")
        self.assertTrue(self.kv("rdm1web", "source").startswith("dir@"))
        code, body = self.http_get(self.companion_url("rdm1"))
        self.assertEqual(code, 200)
        self.assertIn("rdm workbench", body)
        self.assertIn("UI 就绪", result.stdout)
        # rdm 形态跳过设置播种（无 user/apikey 子命令）：给了模型键也不产生 PUT/CLI。
        self.assertFalse((self.trace / "put.txt").exists())
        self.assertNotIn("user list", self.cli_record())

    def test_rdm_companion_missing_web_fails_deploy(self):
        remote, work = self.make_rdm_repo(with_web=False)
        env = self.rdm_env(remote)
        result = self.svc("rdm2", "deploy", "--from", "rdm", "--port", "18843",
                          env=env, check=False)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("ui-missing", result.stdout)
        hist = (self.runtime("rdm2") / "history.jsonl").read_text()
        self.assertIn("ui-missing", hist)
        self.assertIsNone(self.kv("rdm2web", "kind"))

    def test_source_switch_requires_flag(self):
        remote, work = self.make_rdm_repo()
        env = self.rdm_env(remote)
        # 先以 local 源部署（内嵌 UI 桩一律 200）。
        self.svc("sw1", "deploy", "--from", "local", "--port", "18844")
        old = self.kv("sw1", "source")
        self.assertTrue(old.startswith("local@"))
        # 无旗标换到 rdm → 拒绝，source 不变。
        denied = self.svc("sw1", "deploy", "--from", "rdm", "--port", "18844",
                          env=env, check=False)
        self.assertNotEqual(denied.returncode, 0)
        self.assertIn("allow-source-switch", denied.stdout)
        self.assertEqual(self.kv("sw1", "source"), old)
        # 带旗标允许换源，伴生就位。
        ok = self.svc("sw1", "deploy", "--from", "rdm", "--port", "18844",
                      "--allow-source-switch", env=env)
        self.assertTrue(self.kv("sw1", "source").startswith("rdm@"))
        self.assertIn("UI 就绪", ok.stdout)
        code, body = self.http_get(self.companion_url("sw1"))
        self.assertEqual(code, 200)
        self.assertIn("rdm workbench", body)

    def test_rdm_source_commits_and_update_track(self):
        remote, work = self.make_rdm_repo()
        env = self.rdm_env(remote)
        # commits rdm 列出来自独立 RDM 源仓的可部署提交。
        result = self.svc("commits", "rdm", env=env)
        self.assertIn("RDM 仓提交", result.stdout)
        self.assertIn("rdm v1", result.stdout)
        # 部署 v1 → 远端推进 v2 → update 检测可更新并真更新（source 保持 rdm@ 轨道）。
        self.svc("rdm1", "deploy", "--from", "rdm", "--port", "18842", env=env)
        old = self.kv("rdm1", "source")
        (work / "marker").write_text("v2\n")
        subprocess.run(["git", "add", "."], cwd=work, env=self.env, timeout=60, check=True)
        subprocess.run(["git", "commit", "-m", "rdm v2"], cwd=work, env=self.env,
                       timeout=60, check=True)
        subprocess.run(["git", "push"], cwd=work, env=self.env, timeout=60, check=True)
        self.svc("update", "rdm1", "--check", env=env)
        self.svc("update", "rdm1", env=env)
        new = self.kv("rdm1", "source")
        self.assertTrue(new.startswith("rdm@"))
        self.assertNotEqual(new, old)
        self.assertIn("args:", self.serve_record())  # 更新后健康在跑
        # 同轨 update 不需 --allow-source-switch；伴生仍 200。
        code, body = self.http_get(self.companion_url("rdm1"))
        self.assertEqual(code, 200)
        self.assertIn("rdm workbench", body)

    # ---- static 全新部署：首启健康循环不得误判「进程即退」 ----

    def test_static_fresh_deploy_registers_pidcheck_and_starts(self):
        """全新 static --dir：登记须含 pidcheck；首启探活不得把 http.server 判死。"""
        htdocs = self.root / "static-site"
        htdocs.mkdir()
        (htdocs / "index.html").write_text("<html>static-ok</html>\n")
        with socket.socket() as reservation:
            reservation.bind(("127.0.0.1", 0))
            port = reservation.getsockname()[1]
        result = self.svc(
            "uiqa", "deploy", "--kind", "static",
            "--dir", str(htdocs), "--addr", "127.0.0.1:%d" % port)
        self.assertIn("已启动", result.stdout)
        self.assertNotIn("进程即退", result.stdout)
        self.assertEqual(self.kv("uiqa", "kind"), "static")
        pidcheck = self.kv("uiqa", "pidcheck")
        self.assertIsNotNone(pidcheck)
        self.assertIn("--directory", pidcheck)
        # write_static_reg 用 cd 后的绝对路径；与 resolve 后的 htdocs 对齐。
        self.assertIn(str(htdocs.resolve()), pidcheck)
        self.assertTrue(self.pid_alive("uiqa"))
        with urllib.request.urlopen("http://127.0.0.1:%d/" % port, timeout=3) as resp:
            self.assertEqual(resp.status, 200)
            self.assertIn(b"static-ok", resp.read())


if __name__ == "__main__":
    unittest.main()