Agent-System-Dev/tests/integration/lifecycle/test_lifecycle_real.py

"""真实 RDM pin 的部署场景。

由 make test-lifecycle 调用。未设置 SVC_TEST_REAL 时整组跳过，普通 make test
不因此失败。登录与 rdm.ping 不在这里实现：RDM 没有 user/apikey。
"""
import hashlib
import importlib.util
import json
import os
import shutil
import socket
import sqlite3
import subprocess
import time
import unittest
import urllib.error
import urllib.request
from urllib.parse import urlparse, urlunparse
from pathlib import Path

from tests.test_svc_realcombo import ci_git_env
from urllib.parse import urlparse, urlunparse

from tests.test_svc_realcombo import ci_git_env

SOURCE = Path(__file__).resolve().parents[3]
PIN = Path(__file__).resolve().parent / "rdm.pin"

_spec = importlib.util.spec_from_file_location(
    "tsr", SOURCE / "tests" / "test_svc_recover.py")
tsr = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(tsr)

REAL = os.environ.get("SVC_TEST_REAL") == "1"
SRC = os.environ.get("SVC_RDM_SRC_DIR", "")
REAL_BIN = os.environ.get("SVC_REAL_RDM_BIN", "")


def load_pin():
    out = {}
    for line in PIN.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        out[key] = value
    return out


@unittest.skipUnless(REAL and SRC and REAL_BIN,
                     "真实 pin 场景 NOT_RUN：需 SVC_TEST_REAL=1、SVC_RDM_SRC_DIR、SVC_REAL_RDM_BIN")
class LifecycleRealTests(tsr.SvcAutoRecoverTests):
    """候选与切换都走 scripts/svc.sh。已装制品用 SVC_REAL_RDM_BIN，候选用 rdm@<pin>。"""

    def setUp(self):
        super().setUp()
        shutil.copy2(SOURCE / "scripts" / "svc_manifest.py",
                     self.root / "scripts" / "svc_manifest.py")
        self.real_bin = Path(REAL_BIN).resolve()
        self.assertTrue(self.real_bin.is_file() and os.access(self.real_bin, os.X_OK))
        self.pin = load_pin()
        self.env["SVC_RDM_SRC_DIR"] = SRC
        self.env["RDM_DISABLE_EXEC_TOOLS"] = "1"
        self.env.update(ci_git_env())
        self.env.pop("AGENT_SYSTEM_DSN", None)
        self.env.pop("AGENT_SYSTEM_DATA", None)

    def seed_established(self, name, port):
        """用已具备 identity 的制品把实例跑起来，源种类写成 rdm，避免换源门禁抢在身份门禁前。"""
        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.real_bin, d / "bin" / "agent-system")
        (d / "svc.env").write_text("".join("%s=%s\n" % kv for kv in [
            ("kind", "agent"), ("env", name), ("pid", ""),
            ("addr", "127.0.0.1:%d" % port), ("tags", ""),
            ("source", "rdm@seed"), ("version", "real"), ("started", ""),
            ("pidcheck", str(d / "bin" / "agent-system")),
            ("exec", "--addr 127.0.0.1:%d --data %s --plugins %s --auth auto"
                     % (port, d / "data", d / "plugins")),
        ]))
        self.initialize_store(d / "data")
        self.svc(name, "start")

    def initialize_store(self, data, env=None):
        init_env = dict(self.env if env is None else env)
        init_env["RDM_DISABLE_EXEC_TOOLS"] = "1"
        if env is None:
            init_env.pop("AGENT_SYSTEM_DSN", None)
        with socket.socket() as sock:
            sock.bind(("127.0.0.1", 0))
            port = sock.getsockname()[1]
        log_path = self.root / ("initialize-%s.log" % data.name)
        with log_path.open("wb") as log:
            proc = subprocess.Popen(
                [str(self.real_bin), "serve", "--addr", "127.0.0.1:%d" % port,
                 "--data", str(data), "--auth", "on"],
                cwd=self.root, env=init_env, stdout=log, stderr=log)
            try:
                for _ in range(200):
                    self.assertIsNone(proc.poll(), "fixture startup exited; see %s" % log_path)
                    try:
                        urllib.request.urlopen(
                            "http://127.0.0.1:%d/api/health" % port, timeout=1).close()
                        break
                    except urllib.error.HTTPError as response:
                        status = response.code
                        response.close()
                        if status == 401:
                            break
                        self.fail("fixture health returned %d; see %s" % (status, log_path))
                    except OSError:
                        time.sleep(0.1)
                else:
                    self.fail("fixture startup timed out; see %s" % log_path)
            finally:
                proc.terminate()
                try:
                    proc.wait(timeout=10)
                except subprocess.TimeoutExpired:
                    proc.kill()
                    proc.wait(timeout=10)

    def manifest(self, name):
        p = self.runtime(name) / "manifest.json"
        return json.loads(p.read_text(encoding="utf-8")) if p.exists() else None

    def identity_row(self, name):
        con = sqlite3.connect(str(self.db_path(name)))
        try:
            return list(con.execute(
                "SELECT instance_id, dbref FROM agent_system_identity"))
        finally:
            con.close()

    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=900)
        if check:
            self.assertEqual(proc.returncode, 0, proc.stdout)
        return proc

    def db_path(self, name):
        return self.runtime(name) / "data" / "agent-system.db"

    def db_inode(self, name):
        return self.db_path(name).stat().st_ino

    def release_id(self, name):
        return os.path.basename(os.readlink(self.runtime(name) / "current"))

    def alive(self, name):
        pid = self.pid_of(name)
        if not pid:
            return False
        try:
            os.kill(pid, 0)
        except OSError:
            return False
        return True

    def free_port(self):
        with socket.socket() as sock:
            sock.bind(("127.0.0.1", 0))
            return sock.getsockname()[1]

    def deploy(self, name, sha, port, check=True):
        return self.svc(name, "deploy", "--from", "rdm@" + sha, "--port", str(port),
                        check=check)

    def test_legacy_candidate_refused_before_stop(self):
        """无 identity 的 legacy_sha 在停旧服务前被拒绝，且不改已有库。"""
        self.seed_established("leg", self.free_port())
        pid = self.pid_of("leg")
        inode = self.db_inode("leg")
        before = sorted(p.name for p in (self.runtime("leg") / "data").iterdir())
        result = self.deploy("leg", self.pin["legacy_sha"], 19641, check=False)
        self.assertNotEqual(result.returncode, 0, result.stdout)
        self.assertIn("不支持 identity", result.stdout)
        self.assertEqual(self.pid_of("leg"), pid)
        self.assertTrue(self.alive("leg"))
        self.assertEqual(self.db_inode("leg"), inode)
        after = sorted(p.name for p in (self.runtime("leg") / "data").iterdir())
        self.assertEqual(after, before)
        self.svc("leg", "stop")

    def test_missing_row_refused_before_stop(self):
        port = self.free_port()
        self.seed_established("miss", port)
        pid = self.pid_of("miss")
        db = self.db_path("miss")
        con = sqlite3.connect(str(db))
        try:
            con.execute("DELETE FROM agent_system_identity")
            con.commit()
        finally:
            con.close()
        result = self.deploy("miss", self.pin["sha"], port, check=False)
        self.assertNotEqual(result.returncode, 0, result.stdout)
        self.assertIn("无法验证", result.stdout)
        self.assertEqual(self.pid_of("miss"), pid)
        con = sqlite3.connect(str(db))
        try:
            count = con.execute("SELECT count(*) FROM agent_system_identity").fetchone()[0]
        finally:
            con.close()
        self.assertEqual(count, 0)
        self.svc("miss", "stop")

    def test_mismatch_refused_before_stop(self):
        port = self.free_port()
        self.seed_established("mis", port)
        pid = self.pid_of("mis")
        db = self.db_path("mis")
        con = sqlite3.connect(str(db))
        try:
            con.execute("UPDATE agent_system_identity SET instance_id='svc-foreign-mismatch'")
            con.commit()
        finally:
            con.close()
        result = self.deploy("mis", self.pin["sha"], port, check=False)
        self.assertNotEqual(result.returncode, 0, result.stdout)
        self.assertIn("无法验证", result.stdout)
        self.assertEqual(self.pid_of("mis"), pid)
        con = sqlite3.connect(str(db))
        try:
            row = con.execute("SELECT instance_id FROM agent_system_identity").fetchone()
        finally:
            con.close()
        self.assertEqual(row[0], "svc-foreign-mismatch")
        self.svc("mis", "stop")

    def test_prev_and_current_pin_switch_and_back(self):
        """prev_sha 与 sha 来回部署，身份行保持，release 目录跟着源 SHA 走。"""
        port = self.free_port()
        result = self.deploy("sw", self.pin["prev_sha"], port, check=False)
        self.assertEqual(result.returncode, 0, result.stdout)
        first = self.release_id("sw")
        self.assertIn(self.pin["prev_sha"][:7], first)
        iid = self.manifest("sw")["instance_id"]
        row = self.identity_row("sw")
        result = self.deploy("sw", self.pin["sha"], port, check=False)
        self.assertEqual(result.returncode, 0, result.stdout)
        second = self.release_id("sw")
        self.assertIn(self.pin["sha"][:7], second)
        self.assertNotEqual(second, first)
        self.assertEqual(self.identity_row("sw"), row)
        result = self.deploy("sw", self.pin["prev_sha"], port, check=False)
        self.assertEqual(result.returncode, 0, result.stdout)
        self.assertIn(self.pin["prev_sha"][:7], self.release_id("sw"))
        self.assertEqual(self.manifest("sw")["instance_id"], iid)
        self.assertEqual(self.identity_row("sw"), row)
        self.svc("sw", "stop")
        web = self.runtime("swweb")
        if (web / "svc.pid").exists():
            self.svc("swweb", "stop")

    def file_sha256(self, path):
        return hashlib.sha256(Path(path).read_bytes()).hexdigest()

    def journal(self, name):
        path = self.runtime(name) / "deploy.journal"
        return json.loads(path.read_text()) if path.exists() else None

    def set_source(self, name, source):
        path = self.runtime(name) / "svc.env"
        lines = []
        for line in path.read_text().splitlines():
            if line.startswith("source="):
                lines.append("source=" + source)
            else:
                lines.append(line)
        path.write_text("\n".join(lines) + "\n")

    def ensure_git(self):
        subprocess.run(["git", "init", "--initial-branch=main"], cwd=self.root,
                       env=self.env, check=True, stdout=subprocess.DEVNULL)
        subprocess.run(["git", "add", "."], cwd=self.root, env=self.env, check=True,
                       stdout=subprocess.DEVNULL)
        subprocess.run(["git", "commit", "-m", "fixture"], cwd=self.root, env=self.env,
                       check=True, stdout=subprocess.DEVNULL)

    def deploy_fault(self, name, sha, port, phase):
        env = dict(self.env)
        env["SVC_TEST_FAULTS"] = "1"
        env["SVC_FAULT_AT"] = phase
        return self.svc(name, "deploy", "--from", "rdm@" + sha, "--port", str(port),
                        check=False, env=env)

    def test_phase_interrupt_reconciles_on_real_artifacts(self):
        """staging 在停旧前退出，旧进程与指针不动；switched 在指针已切后退出，下次 start 回退。"""
        port = self.free_port()
        result = self.deploy("ph", self.pin["prev_sha"], port, check=False)
        self.assertEqual(result.returncode, 0, result.stdout)
        pid = self.pid_of("ph")
        first = self.release_id("ph")
        digest = self.file_sha256(self.runtime("ph") / "current" / "bin" / "agent-system")
        row = self.identity_row("ph")

        staged = self.deploy_fault("ph", self.pin["sha"], port, "staging")
        self.assertEqual(staged.returncode, 86, staged.stdout)
        self.assertIn("故障注入于 staging", staged.stdout)
        self.assertEqual(self.pid_of("ph"), pid)
        self.assertTrue(self.alive("ph"))
        self.assertEqual(self.release_id("ph"), first)
        self.assertEqual(self.journal("ph")["phase"], "staging")
        resumed = self.svc("ph", "start")
        self.assertIn("未切换", resumed.stdout)
        self.assertIsNone(self.journal("ph"))
        self.assertEqual(self.pid_of("ph"), pid)
        self.assertEqual(self.release_id("ph"), first)

        switched = self.deploy_fault("ph", self.pin["sha"], port, "switched")
        self.assertEqual(switched.returncode, 86, switched.stdout)
        self.assertIn("故障注入于 switched", switched.stdout)
        self.assertIn(self.pin["sha"][:7], self.release_id("ph"))
        self.assertFalse(self.alive("ph"))
        self.assertEqual(self.journal("ph")["phase"], "switched")
        rolled = self.svc("ph", "start")
        self.assertIn("已回退上一版本", rolled.stdout)
        self.assertEqual(self.release_id("ph"), first)
        self.assertEqual(
            self.file_sha256(self.runtime("ph") / "current" / "bin" / "agent-system"),
            digest)
        self.assertEqual(self.identity_row("ph"), row)
        self.assertTrue(self.alive("ph"))
        self.assertIsNone(self.journal("ph"))
        self.svc("ph", "stop")

    def test_build_failure_keeps_real_process(self):
        self.ensure_git()
        port = self.free_port()
        self.seed_established("bld", port)
        self.set_source("bld", "local@seed")
        pid = self.pid_of("bld")
        before = self.file_sha256("/proc/%d/exe" % pid)
        (self.root / "Makefile").write_text("build:\n\texit 1\n")
        result = self.svc("bld", "deploy", "--from", "local", check=False)
        self.assertNotEqual(result.returncode, 0, result.stdout)
        self.assertIn("构建失败", result.stdout)
        self.assertEqual(self.pid_of("bld"), pid)
        self.assertTrue(self.alive("bld"))
        self.assertEqual(self.file_sha256("/proc/%d/exe" % pid), before)
        self.svc("bld", "stop")

    def test_start_exit_rolls_back_to_real_binary(self):
        """serve 立即退出的包装会回退。version 与 identity 仍转给真实制品，门禁才看得到能力。"""
        self.ensure_git()
        port = self.free_port()
        self.seed_established("die", port)
        self.set_source("die", "local@seed")
        real = self.file_sha256(self.real_bin)
        shim = self.root / "crash-shim"
        shim.write_text(
            "#!/bin/bash\n"
            "real=%s\n"
            "if [ \"${1:-}\" = serve ]; then\n"
            "  echo 'crash-shim: serve exits immediately' >&2\n"
            "  exit 1\n"
            "fi\n"
            "exec \"$real\" \"$@\"\n" % self.real_bin)
        shim.chmod(0o755)
        (self.root / "Makefile").write_text(
            "build:\n\tmkdir -p bin\n\tcp %s bin/agent-system\n" % shim)
        result = self.svc("die", "deploy", "--from", "local", check=False)
        self.assertNotEqual(result.returncode, 0, result.stdout)
        self.assertIn("已回退上一版本", result.stdout)
        current = self.runtime("die") / "current" / "bin" / "agent-system"
        self.assertEqual(self.file_sha256(current), real)
        self.assertTrue(self.alive("die"))
        self.assertIsNone(self.journal("die"))
        self.svc("die", "stop")

    def git(self, repo, *args):
        subprocess.run(["git", "-C", str(repo), *args], cwd=self.root, env=self.env,
                       check=True, stdout=subprocess.DEVNULL)

    def make_copy_repo(self):
        """两笔提交的部署源。build 复制真实制品，frontend 只写 dist，避免整仓前端构建。"""
        src = self.root / "devsrc"
        (src / "modules" / "frontend" / "dist").mkdir(parents=True)
        (src / "Makefile").write_text(
            "build:\n\tmkdir -p bin\n\tcp %s bin/agent-system\n"
            "frontend:\n\tmkdir -p modules/frontend/dist\n"
            "\tcp web/index.html modules/frontend/dist/index.html\n" % self.real_bin)
        (src / "web").mkdir()
        (src / "web" / "index.html").write_text("old\n")
        self.git(src, "init", "--initial-branch=main")
        self.git(src, "add", ".")
        self.git(src, "commit", "-m", "old")
        old = subprocess.check_output(
            ["git", "-C", str(src), "rev-parse", "HEAD"], text=True).strip()
        (src / "web" / "index.html").write_text("new\n")
        self.git(src, "add", ".")
        self.git(src, "commit", "-m", "new")
        new = subprocess.check_output(
            ["git", "-C", str(src), "rev-parse", "HEAD"], text=True).strip()
        self.git(src, "update-ref", "refs/remotes/origin/main", "HEAD")
        return src, old, new

    def move_ref(self, repo, sha):
        ref = "refs/remotes/origin/main"
        prev = subprocess.check_output(
            ["git", "-C", str(repo), "rev-parse", ref], text=True).strip()

        def restore():
            subprocess.run(["git", "-C", str(repo), "update-ref", ref, prev],
                           check=True, stdout=subprocess.DEVNULL)

        self.addCleanup(restore)
        subprocess.run(["git", "-C", str(repo), "update-ref", ref, sha],
                       check=True, stdout=subprocess.DEVNULL)

    def test_mixed_batch_updates_forward_and_deploy_reverses(self):
        """dev、genui 走复制型源仓；rdm1、rdmcand 走 pin。update 只向前，回退必须显式 deploy。"""
        src, old, new = self.make_copy_repo()
        self.env["SVC_SRC_DIR"] = str(src)
        ports = {name: self.free_port() for name in ("dev", "genui", "rdm1", "rdmcand")}
        self.svc("dev", "deploy", "--from", "dev@" + old, "--port", str(ports["dev"]))
        self.svc("genui", "deploy", "--kind", "static", "--from", "dev@" + old,
                 "--port", str(ports["genui"]))
        self.deploy("rdm1", self.pin["prev_sha"], ports["rdm1"])
        self.deploy("rdmcand", self.pin["prev_sha"], ports["rdmcand"])
        # deploy 会 fetch 并把 origin/main 拉回远端。更新前再钉回 pin，且 update --no-fetch。
        self.move_ref(SRC, self.pin["sha"])
        old_src = {
            "dev": self.kv("dev", "source"),
            "genui": self.kv("genui", "source"),
            "rdm1": self.kv("rdm1", "source"),
            "rdmcand": self.kv("rdmcand", "source"),
        }
        self.assertIn(old[:7], old_src["dev"])
        self.assertIn(self.pin["prev_sha"][:7], old_src["rdm1"])

        for index, name in enumerate(("dev", "genui", "rdm1", "rdmcand")):
            if name in ("rdm1", "rdmcand"):
                # 上一服务的 deploy 子进程会 fetch，把 origin/main 拉回远端。
                subprocess.run(
                    ["git", "-C", SRC, "update-ref", "refs/remotes/origin/main",
                     self.pin["sha"]],
                    check=True, stdout=subprocess.DEVNULL)
            result = self.svc("update", name, "--no-fetch")
            self.assertIn("已更新", result.stdout)
            for later in ("dev", "genui", "rdm1", "rdmcand")[index + 1:]:
                self.assertEqual(self.kv(later, "source"), old_src[later])
        self.assertIn(new[:7], self.kv("dev", "source"))
        self.assertIn(new[:7], self.kv("genui", "source"))
        self.assertIn(self.pin["sha"][:7], self.kv("rdm1", "source"))
        self.assertIn(self.pin["sha"][:7], self.kv("rdmcand", "source"))
        self.assertEqual((self.runtime("genui") / "htdocs" / "index.html").read_text(), "new\n")
        self.assertEqual(len(self.identity_row("rdm1")), 1)
        self.assertEqual(len(self.identity_row("rdmcand")), 1)

        self.move_ref(src, old)
        self.move_ref(SRC, self.pin["prev_sha"])
        ahead = self.svc("update", "--check", "--no-fetch", "dev", "genui", "rdm1", "rdmcand")
        self.assertIn("领先", ahead.stdout)
        self.assertNotIn("已更新", ahead.stdout)
        self.assertIn(new[:7], self.kv("dev", "source"))
        self.assertIn(self.pin["sha"][:7], self.kv("rdmcand", "source"))

        newest = {
            "dev": self.kv("dev", "source"),
            "genui": self.kv("genui", "source"),
            "rdm1": self.kv("rdm1", "source"),
            "rdmcand": self.kv("rdmcand", "source"),
        }
        reverse = (
            ("rdmcand", ("rdmcand", "deploy", "--from", "rdm@" + self.pin["prev_sha"],
                         "--port", str(ports["rdmcand"]))),
            ("rdm1", ("rdm1", "deploy", "--from", "rdm@" + self.pin["prev_sha"],
                      "--port", str(ports["rdm1"]))),
            ("genui", ("genui", "deploy", "--kind", "static", "--from", "dev@" + old,
                       "--port", str(ports["genui"]))),
            ("dev", ("dev", "deploy", "--from", "dev@" + old, "--port", str(ports["dev"]))),
        )
        for index, (name, args) in enumerate(reverse):
            self.svc(*args)
            self.assertNotEqual(self.kv(name, "source"), newest[name])
            for later, _args in reverse[index + 1:]:
                self.assertEqual(self.kv(later, "source"), newest[later])
        self.assertEqual((self.runtime("genui") / "htdocs" / "index.html").read_text(), "old\n")
        for name in ("dev", "genui", "rdm1", "rdmcand", "rdm1web", "rdmcandweb"):
            if (self.runtime(name) / "svc.pid").exists():
                self.svc(name, "stop")

    def marker_id(self, name):
        path = self.runtime(name) / "data" / ".agentx" / "data-marker.json"
        return json.loads(path.read_text(encoding="utf-8"))["instance_id"]

    def test_missing_manifest_refuses_the_same_way(self):
        """manifest 丢失后 deploy、update、recover 都停在归属冲突，不新建身份。"""
        src, _old, _new = self.make_copy_repo()
        self.env["SVC_SRC_DIR"] = str(src)
        self.move_ref(SRC, self.pin["sha"])
        names = ("nmdep", "nmupd", "nmrec")
        rows = {}
        for name in names:
            self.seed_established(name, self.free_port())
            self.svc(name, "stop")
            rows[name] = (self.identity_row(name), self.marker_id(name))
            (self.runtime(name) / "manifest.json").unlink()
        self.set_source("nmupd", "rdm@" + self.pin["prev_sha"])
        envf = self.runtime("nmrec") / "svc.env"
        envf.write_text(envf.read_text() + "autostart=1\n")

        deployed = self.svc("nmdep", "deploy", "--from", "rdm@" + self.pin["sha"],
                            "--port", str(self.free_port()), check=False)
        updated = self.svc("update", "nmupd", "--no-fetch", check=False)
        recovered = self.svc("recover", "nmrec", check=False)
        for result in (deployed, updated, recovered):
            self.assertNotEqual(result.returncode, 0, result.stdout)
            self.assertIn("身份治理冲突", result.stdout)
        for name in names:
            self.assertIsNone(self.manifest(name))
            self.assertEqual((self.identity_row(name), self.marker_id(name)), rows[name])
            self.assertFalse(self.alive(name))

    def test_pg_cross_instance_verify_refuses(self):
        dsn = os.environ.get("AGENT_PG_TEST_DSN", "")
        if not dsn:
            self.skipTest("PG 跨实例 NOT_RUN：未设置 AGENT_PG_TEST_DSN")
        dsn = self.private_pg_dsn(dsn)
        port = self.free_port()
        self.seed_established("pga", port)
        self.svc("pga", "stop")
        owner = self.manifest("pga")["instance_id"]
        data = self.runtime("pga") / "data"
        env = dict(self.env)
        env["AGENT_SYSTEM_DSN"] = dsn
        # 身份命令不建库。先用真实制品按环境 DSN 启动一次，让受控迁移落表。
        self.initialize_store(data, env)
        ensured = subprocess.run(
            [str(self.real_bin), "identity", "ensure", "--instance-id", owner,
             "--dbref", "pg", "--data", str(data)],
            env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
            timeout=60)
        self.assertEqual(ensured.returncode, 0, ensured.stdout.replace(dsn, "<dsn>"))
        self.seed_established("pgb", self.free_port())
        self.svc("pgb", "stop")
        other = self.manifest("pgb")["instance_id"]
        self.assertNotEqual(other, owner)
        envf = self.runtime("pgb") / "svc.env"
        envf.write_text(envf.read_text() + "db=pg\ndsn=%s\n" % dsn)
        manifest = self.manifest("pgb")
        manifest["identity"]["db_mode"] = "pg"
        manifest["identity"]["db_ref"] = "pg"
        manifest["identity"]["db_marker"] = "established"
        (self.runtime("pgb") / "manifest.json").write_text(
            json.dumps(manifest, ensure_ascii=False, indent=2) + "\n")
        result = self.svc("pgb", "start", check=False)
        redacted = result.stdout.replace(dsn, "<dsn>")
        self.assertNotEqual(result.returncode, 0, redacted)
        self.assertFalse(self.alive("pgb"))
        still = subprocess.run(
            [str(self.real_bin), "identity", "verify", "--instance-id", owner,
             "--dbref", "pg", "--data", str(data)],
            env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
            timeout=60)
        self.assertEqual(still.returncode, 0, still.stdout.replace(dsn, "<dsn>"))
        self.svc("pga", "stop", check=False)

    def test_caller_dsn_stays_out_of_process_and_logs(self):
        sentinel = "credscan-sentinel-7c1e"
        self.seed_established("cred", self.free_port())
        self.svc("cred", "stop")
        env = dict(self.env)
        env["AGENT_SYSTEM_DSN"] = "postgres://scanner:%s@127.0.0.1/none" % sentinel
        self.svc("cred", "start", env=env)
        pid = self.pid_of("cred")
        cmdline = Path("/proc/%d/cmdline" % pid).read_bytes()
        environ = Path("/proc/%d/environ" % pid).read_bytes()
        self.assertNotIn(sentinel.encode(), cmdline)
        self.assertNotIn(sentinel.encode(), environ)
        for path in ("svc.log", "history.jsonl", "svc.env"):
            text = (self.runtime("cred") / path).read_text(errors="replace")
            self.assertNotIn(sentinel, text, path)
        self.svc("cred", "stop")

    def private_pg_dsn(self, dsn):
        """共享的 AGENT_PG_TEST_DSN 已被其他用例写入身份。本用例单独建库。"""
        parsed = urlparse(dsn)
        name = "d3pg_" + hashlib.sha256(os.urandom(8)).hexdigest()[:12]
        created = subprocess.run(
            ["psql", dsn, "-v", "ON_ERROR_STOP=1", "-c",
             'CREATE DATABASE "%s"' % name],
            text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=30)
        self.assertEqual(created.returncode, 0, created.stdout.replace(dsn, "<dsn>"))
        self.addCleanup(self.drop_private_pg, dsn, name)
        return urlunparse(parsed._replace(path="/" + name))

    def drop_private_pg(self, dsn, name):
        subprocess.run(
            ["psql", dsn, "-c", 'DROP DATABASE IF EXISTS "%s" WITH (FORCE)' % name],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=30)

    def event_count(self, name):
        db = self.runtime(name) / "data" / "rdm.db"
        con = sqlite3.connect("file:%s?mode=ro" % db, uri=True)
        try:
            return con.execute("SELECT count(*) FROM events").fetchone()[0]
        finally:
            con.close()

    def test_bootstrap_domain_survives_upgrade(self):
        port = self.free_port()
        self.seed_established("boot", port)
        self.svc("boot", "stop")
        docs = self.root / "bootstrap-docs"
        docs.mkdir()
        (docs / "note.md").write_text("# 领域笔记\n\n升级后应仍在。\n")
        data = self.runtime("boot") / "data"
        ingested = subprocess.run(
            [str(self.real_bin), "-bootstrap", str(docs), "-data", str(data)],
            env=self.env, text=True, stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT, timeout=60)
        self.assertEqual(ingested.returncode, 0, ingested.stdout)
        before = self.event_count("boot")
        self.assertGreater(before, 0)
        result = self.deploy("boot", self.pin["sha"], port, check=False)
        self.assertEqual(result.returncode, 0, result.stdout)
        self.assertEqual(self.event_count("boot"), before)
        self.svc("boot", "stop")
        if (self.runtime("bootweb") / "svc.pid").exists():
            self.svc("bootweb", "stop")


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