diff --git a/scripts/svc.sh b/scripts/svc.sh
index 70c5d196..6bd6a309 100755
--- a/scripts/svc.sh
+++ b/scripts/svc.sh
@@ -3186,7 +3186,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"
@@ -3200,10 +3199,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"
diff --git a/tests/integration/lifecycle/test_lifecycle_real.py b/tests/integration/lifecycle/test_lifecycle_real.py
index ec465b45..98eeeabb 100644
--- a/tests/integration/lifecycle/test_lifecycle_real.py
+++ b/tests/integration/lifecycle/test_lifecycle_real.py
@@ -76,10 +76,11 @@ class LifecycleRealTests(tsr.SvcAutoRecoverTests):
         self.initialize_store(d / "data")
         self.svc(name, "start")
 
-    def initialize_store(self, data):
-        init_env = dict(self.env)
+    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"
-        init_env.pop("AGENT_SYSTEM_DSN", None)
+        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]
@@ -366,6 +367,194 @@ class LifecycleRealTests(tsr.SvcAutoRecoverTests):
         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")
+        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)
+
 
 if __name__ == "__main__":
     unittest.main()