Merge origin/main into the follow-up branch

.gitea/workflows/ci.yaml

@@ -387,7 +387,7 @@ jobs:
           # 凭据纪律与 Checkout 块同源：经 GIT_CONFIG_* 环境注入 URL 级 token 头，
           # 不落文件/命令行；allowlist 扩项 = Agent-System-RDM 及其 pins（同一
           # 组件集合），由本 PR 评审背书。SSH pins URL 以 insteadOf 改写到受信
-          # https 基址后再取。
+          # https 基址后再取。只追加 pin 的三个 40 位 hex，不 source pin 文件。
           SERVER="$(printf '%s' "$GITHUB_SERVER_URL" | sed 's:/*$::')"
           scheme="${GITHUB_SERVER_URL%%://*}"
           export GIT_CONFIG_COUNT=4
@@ -400,15 +400,45 @@ jobs:
           export GIT_CONFIG_KEY_3="core.hooksPath"
           export GIT_CONFIG_VALUE_3="/dev/null"
           git clone --depth=1 "$SERVER/weichen/Agent-System-RDM.git" rdm
-          # 凭据环境须保持到 pins 子模块取完（insteadOf + token 头覆盖嵌套 SSH URL），
-          # 之后再 unset。
-          git -C rdm submodule update --init --recursive
+          mapfile -t pins < <(python3 - <<'PY'
+          import pathlib
+          got = {}
+          for line in pathlib.Path("tests/integration/lifecycle/rdm.pin").read_text().splitlines():
+              line = line.strip()
+              if not line or line.startswith("#") or "=" not in line:
+                  continue
+              key, value = line.split("=", 1)
+              if key not in ("sha", "prev_sha", "legacy_sha") or len(value) != 40:
+                  raise SystemExit("rdm.pin rejected")
+              if any(c not in "0123456789abcdef" for c in value):
+                  raise SystemExit("rdm.pin rejected")
+              got[key] = value
+          if set(got) != {"sha", "prev_sha", "legacy_sha"}:
+              raise SystemExit("rdm.pin rejected")
+          print(got["prev_sha"])
+          print(got["legacy_sha"])
+          print(got["sha"])
+          PY
+          )
+          [ "${#pins[@]}" -eq 3 ]
+          # depth=1 会把三个 pin 变成互不相关的浅根，update 会把 prev→sha 误判成分叉。
+          # 当前 first-parent 距离是 39，200 留余量，仍不拉全史。
+          git -C rdm fetch --depth=200 origin "${pins[@]}"
+          for rev in "${pins[@]}"; do
+            git -C rdm checkout --detach "$rev"
+            git -C rdm submodule update --init --recursive
+          done
           echo "RDM commit: $(git -C rdm rev-parse HEAD)"
           unset GIT_CONFIG_COUNT GIT_CONFIG_KEY_0 GIT_CONFIG_VALUE_0 \
                 GIT_CONFIG_KEY_1 GIT_CONFIG_VALUE_1 GIT_CONFIG_KEY_2 \
                 GIT_CONFIG_VALUE_2 GIT_CONFIG_KEY_3 GIT_CONFIG_VALUE_3
           unset SUBMODULE_TOKEN
-          make -C rdm build
+          : > rdm/pin-build-digests.txt
+          for rev in "${pins[@]}"; do
+            git -C rdm checkout --detach "$rev"
+            make -C rdm build
+            sha256sum rdm/bin/agent-system-rdm | awk -v rev="$rev" '{print rev, $1}' >> rdm/pin-build-digests.txt
+          done
       - name: Install Node 22
         run: |
           set -euo pipefail
@@ -419,14 +449,20 @@ jobs:
       - name: Real-combination lifecycle gate (make test-lifecycle, required)
         env:
           AGENT_PG_TEST_DSN: postgres://ci_agent:ci_job_local_only@127.0.0.1:5432/agent_test?sslmode=disable
+          SUBMODULE_TOKEN: ${{ secrets.SUBMODULE_TOKEN }}
         run: |
           set -euo pipefail
           # 路径变量在脚本内解析（YAML env: 不做 shell 展开）。
           # 不设 SVC_RDM_ORIGIN：全轨 git 部署由 svc-realcombo 作业覆盖。
+          # pins 物化没有 SSH known_hosts。token 只经 SVC_CI_GIT_TOKEN 进
+          # ci_git_env 的 URL 级头，不写文件。
           export SVC_TEST_REAL=1 SVC_TEST_PG=1
           export SVC_REAL_BIN="$PWD/bin/agent-system"
           export SVC_REAL_RDM_BIN="$PWD/rdm/bin/agent-system-rdm"
           export SVC_RDM_SRC_DIR="$PWD/rdm"
+          export SVC_CI_GIT_SERVER="$(printf '%s' "$GITHUB_SERVER_URL" | sed 's:/*$::')"
+          export SVC_CI_GIT_TOKEN="$SUBMODULE_TOKEN"
+          unset SUBMODULE_TOKEN
           make test-lifecycle
 
   # U3（统一部署方案）：真实 Dev/RDM 制品的 svc.sh 生命周期门禁——桩套件

scripts/svc.sh

@@ -1435,8 +1435,9 @@ svc_launch_serve() {
 }
 
 # svc_lifecycle_child ACTION [extra args...]：用清空的环境拉起本脚本的
-# __lifecycle。白名单：PATH/HOME/USER/LOGNAME/LANG/LC_*/TZ/TMPDIR、代理、SVC_*。
-# 从不传入 AGENT_SYSTEM_*。
+# __lifecycle。白名单：PATH/HOME/USER/LOGNAME/LANG/LC_*/TZ/TMPDIR、代理、SVC_*、
+# GIT_CONFIG_*（运行器把 SSH pin 改写成受信 HTTPS，token 只在环境里）。
+# 从不传入 AGENT_SYSTEM_*。serve 仍是 env -i，不继承这些变量。
 svc_lifecycle_child() {
   local -a args=(env -i "PATH=${PATH:-/usr/bin:/bin}" "HOME=${HOME:-/tmp}"
     "USER=${USER:-}" "LOGNAME=${LOGNAME:-${USER:-}}" "LANG=${LANG:-C.UTF-8}"
@@ -1450,7 +1451,7 @@ svc_lifecycle_child() {
   while IFS= read -r k; do
     [ -n "$k" ] || continue
     args+=("$k=${!k}")
-  done < <(compgen -e | grep -E '^(SVC_|LC_)' || true)
+  done < <(compgen -e | grep -E '^(SVC_|LC_|GIT_CONFIG_)' || true)
   if [ -n "${RDM_DISABLE_EXEC_TOOLS:-}" ]; then
     args+=("RDM_DISABLE_EXEC_TOOLS=$RDM_DISABLE_EXEC_TOOLS")
   fi
@@ -3358,7 +3359,6 @@ static_do_deploy() {
       wt="$(materialize_dev_commit "$SRC_FULL" "$SRC_SHA" frontend)" \
         || { hist_event deploy fail "$(read_pid)" "物化/构建失败（$SRC_LABEL）"; return 1; }
       src_dist="$wt/modules/frontend/dist"
-      recycle_worktree "$SRC_SHA"
       ;;
     release)
       c_err "[$ENV_NAME] release 制品内嵌前端（无独立 dist）；static 请用 --from dev|local 或 --dir"
@@ -3372,10 +3372,20 @@ static_do_deploy() {
       return 1
       ;;
   esac
-  [ -d "$src_dist" ] || { c_err "[$ENV_NAME] 前端产物缺失：$src_dist"; hist_event deploy fail "$(read_pid)" "dist 缺失"; return 1; }
+  if [ ! -d "$src_dist" ]; then
+    [ "$SRC_KINDSRC" = dev ] && recycle_worktree "$SRC_SHA"
+    c_err "[$ENV_NAME] 前端产物缺失：$src_dist"
+    hist_event deploy fail "$(read_pid)" "dist 缺失"
+    return 1
+  fi
   mkdir -p "$ENV_DIR"
   rm -rf "${ENV_DIR:?}/htdocs"
-  cp -r "$src_dist" "$ENV_DIR/htdocs" || { hist_event deploy fail "$(read_pid)" "dist 复制失败"; return 1; }
+  cp -r "$src_dist" "$ENV_DIR/htdocs" || {
+    [ "$SRC_KINDSRC" = dev ] && recycle_worktree "$SRC_SHA"
+    hist_event deploy fail "$(read_pid)" "dist 复制失败"
+    return 1
+  }
+  [ "$SRC_KINDSRC" = dev ] && recycle_worktree "$SRC_SHA"
   write_static_reg "$ENV_DIR/htdocs"
   c_ok "[$ENV_NAME] 前端产物就位：$ENV_DIR/htdocs（源 $SRC_LABEL）"
   hist_event deploy ok "$(read_pid)" "from=$SRC_LABEL"

tests/integration/lifecycle/lifecycle_suite.py

@@ -1,18 +1,77 @@
 """真实 Dev/RDM 组合门禁。由 make test-lifecycle 调用。
 
 缺 PostgreSQL、RDM 源码或制品时直接失败，不静默跳过。
-登录依赖账户 CLI 公共化（账户方案 U7/U5），单独标 NOT_RUN，
+登录与 rdm.ping 依赖账户 CLI 公共化，单独标 NOT_RUN，
 不因此把已经跑过的真实组合场景记成失败。
 """
+import hashlib
 import os
 import subprocess
 import sys
 from pathlib import Path
+from urllib.parse import urlparse
 
 ROOT = Path(__file__).resolve().parents[3]
 PIN = Path(__file__).resolve().parent / "rdm.pin"
 
 
+def sha256_file(path):
+    digest = hashlib.sha256()
+    with open(path, "rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def print_inventory(src, dev, rdm):
+    print("组合清单")
+    head = subprocess.run(["git", "-C", str(ROOT), "rev-parse", "HEAD"],
+                          capture_output=True, text=True)
+    print("dev_sha", head.stdout.strip() if head.returncode == 0 else "NOT_RUN")
+    modules = subprocess.run(
+        ["git", "-C", str(ROOT), "submodule", "status", "--recursive"],
+        capture_output=True, text=True)
+    if modules.returncode == 0:
+        for line in modules.stdout.splitlines():
+            print("submodule", " ".join(line.split()[:2]))
+    else:
+        print("submodule NOT_RUN")
+    print("pin")
+    for line in PIN.read_text().splitlines():
+        if line.strip() and not line.startswith("#"):
+            print(" ", line.strip())
+    print("dev_bin", dev, sha256_file(dev))
+    print("rdm_bin", rdm, sha256_file(rdm))
+    digests = Path(src) / "pin-build-digests.txt"
+    if digests.is_file():
+        for line in digests.read_text().splitlines():
+            print("pin_build", line)
+    else:
+        print("pin_build NOT_RUN：本进程没有预构建三个 pin")
+    print("pg_version", pg_version())
+    print("登录 NOT_RUN：RDM 二进制没有 user/apikey，工作台登录依赖账户方案 U7/U5")
+    print("rdm.ping NOT_RUN：需要已登录会话，本轮不公共化账户 CLI")
+
+
+def pg_version():
+    parsed = urlparse(os.environ.get("AGENT_PG_TEST_DSN", ""))
+    if not parsed.hostname:
+        return "NOT_RUN"
+    env = os.environ.copy()
+    env.update({
+        "PGHOST": parsed.hostname,
+        "PGPORT": str(parsed.port or 5432),
+        "PGUSER": parsed.username or "",
+        "PGPASSWORD": parsed.password or "",
+        "PGDATABASE": parsed.path.lstrip("/"),
+    })
+    proc = subprocess.run(["psql", "-tAc", "SHOW server_version"],
+                          env=env, capture_output=True, text=True)
+    if proc.returncode != 0:
+        return "NOT_RUN"
+    return proc.stdout.strip() or "NOT_RUN"
+
+
 def main():
     missing = []
     if not os.environ.get("AGENT_PG_TEST_DSN"):
@@ -30,29 +89,27 @@ def main():
         print("test-lifecycle 失败。缺：%s" % ", ".join(missing))
         return 1
 
-    print("组合清单")
-    print("pin", PIN.read_text().strip())
-    print("rdm_src", src)
-    print("dev_bin", dev)
-    print("rdm_bin", rdm)
     print("将运行 tests.test_svc_realcombo：SQLite 身份、RDM 二进制生命周期、PostgreSQL 身份环境交付")
+    print("将运行 tests.integration.lifecycle.test_lifecycle_real：pin 上的真实候选与切换")
     if os.environ.get("SVC_RDM_ORIGIN"):
         print("将运行全轨部署：SVC_RDM_ORIGIN 已设置")
     else:
         print("全轨部署 NOT_RUN：未设置 SVC_RDM_ORIGIN（由 svc-realcombo 作业覆盖）")
-    print("登录 NOT_RUN：RDM 二进制没有 user/apikey，工作台登录依赖账户方案 U7/U5")
     env = os.environ.copy()
     env["SVC_TEST_REAL"] = "1"
     env["SVC_TEST_PG"] = "1"
     env["SVC_REAL_BIN"] = dev
     env["SVC_REAL_RDM_BIN"] = rdm
     proc = subprocess.run(
-        [sys.executable, "-m", "unittest", "tests.test_svc_realcombo", "-q"],
+        [sys.executable, "-m", "unittest",
+         "tests.test_svc_realcombo",
+         "tests.integration.lifecycle.test_lifecycle_real", "-q"],
         cwd=str(ROOT), env=env)
+    print_inventory(src, dev, rdm)
     if proc.returncode != 0:
         print("test-lifecycle 失败：真实组合用例未通过")
         return proc.returncode
-    print("test-lifecycle 可跑场景通过；登录与未设 origin 的全轨部署保持 NOT_RUN")
+    print("test-lifecycle 可跑场景通过；登录与 rdm.ping 保持 NOT_RUN")
     return 0
 
 

tests/integration/lifecycle/rdm.pin

@@ -1,4 +1,6 @@
 # RDM 制品钉扎。三行都是 40 位 hex。升级走 PR，不在 CI 里执行候选文件。
-sha=68de2aa41022c8a77206ea30c2164bec33632ee3
-prev_sha=00ae1494f5b45ae0b4ed955ee82debdfe830a1c3
+# legacy 没有 version --json。prev 是身份能力合入（#291）。sha 与 prev 钉同一
+# backend（0f192a4），schema 兼容，但产品树不同，切换能在 release 名上看到。
+sha=c6304224f995747ea3ffcb8d0e81f3ad46a96546
+prev_sha=3484ec225b0087a9065466ed436ea97a768691ef
 legacy_sha=72bb1fec79d64c2af5e49416640712320816c3b4

tests/integration/lifecycle/test_lifecycle_real.py

@@ -0,0 +1,632 @@
+"""真实 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()

tests/test_ci_credentials.py

@@ -25,6 +25,14 @@ def checkout_script():
     return textwrap.dedent("\n".join(lines[start:end]))
 
 
+def rdm_clone_script():
+    lines = WORKFLOW.read_text().splitlines()
+    start = next(i for i, line in enumerate(lines) if "Clone Agent-System-RDM" in line)
+    start = next(i for i in range(start, len(lines)) if lines[i].strip() == "run: |") + 1
+    end = next(i for i in range(start, len(lines)) if lines[i].startswith("      - name:"))
+    return textwrap.dedent("\n".join(lines[start:end]))
+
+
 class CheckoutTests(unittest.TestCase):
     def setUp(self):
         self.temp = tempfile.TemporaryDirectory()
@@ -236,6 +244,54 @@ class CheckoutTests(unittest.TestCase):
         self.assertNotEqual(result.returncode, 0)
         self.assertFalse(any("untrusted" in path for path, auth in self.requests))
 
+    def test_rdm_pin_fetch_is_three_shas_on_the_allowlist(self):
+        source = self.base / "Agent-System-RDM"
+        source.mkdir()
+        self.git(source, "init", "-q", "-b", "main")
+        shas = []
+        for index in range(3):
+            (source / "Makefile").write_text(
+                "build:\n\tmkdir -p bin\n\tprintf 'rev%d\\n' > bin/agent-system-rdm\n" % index)
+            self.git(source, "add", ".")
+            self.git(source, "-c", "user.name=Fixture",
+                     "-c", "user.email=fixture@example.invalid",
+                     "commit", "-q", "-m", "rev %d" % index)
+            shas.append(self.git(source, "rev-parse", "HEAD"))
+        target = self.published / "weichen" / "Agent-System-RDM.git"
+        target.parent.mkdir(parents=True, exist_ok=True)
+        self.git(self.base, "clone", "-q", "--bare", str(source), str(target))
+        self.git(target, "config", "uploadpack.allowReachableSHA1InWant", "true")
+        self.git(target, "config", "uploadpack.allowAnySHA1InWant", "true")
+        work = self.base / "rdm-job"
+        pin = work / "tests" / "integration" / "lifecycle"
+        pin.mkdir(parents=True)
+        (pin / "rdm.pin").write_text(
+            "sha=%s\nprev_sha=%s\nlegacy_sha=%s\n" % (shas[2], shas[1], shas[0]))
+        script = rdm_clone_script()
+        self.assertNotRegex(script, r"(?m)^\s*source\s")
+        self.requests.clear()
+        result = subprocess.run(
+            ["bash", "-c", script], cwd=work,
+            env=self.env | {"SUBMODULE_TOKEN": SUB_TOKEN, "GITHUB_SERVER_URL": self.url},
+            capture_output=True, text=True, timeout=60)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertNotIn(SUB_TOKEN, result.stdout + result.stderr)
+        self.assertTrue(self.requests)
+        for path, auth in self.requests:
+            self.assertRegex(path, r"^/weichen/Agent-System-RDM\.git/")
+            self.assertEqual(auth, "token " + SUB_TOKEN)
+        for sha in shas:
+            self.assertEqual(self.git(work / "rdm", "cat-file", "-t", sha), "commit")
+        self.assertEqual(self.git(work / "rdm", "rev-parse", "HEAD"), shas[2])
+        digests = (work / "rdm" / "pin-build-digests.txt").read_text()
+        for sha in shas:
+            self.assertIn(sha, digests)
+        gate = WORKFLOW.read_text().split(
+            "Real-combination lifecycle gate", 1)[1].split("\n  svc-realcombo:", 1)[0]
+        self.assertIn('export SVC_CI_GIT_TOKEN="$SUBMODULE_TOKEN"', gate)
+        self.assertNotRegex(gate, r"(?m)^\s*source\s")
+        self.assertNotIn(">>", gate)
+
 
 if __name__ == "__main__":
     unittest.main()