Add real artifact rollback and phase reconciliation tests

Why: Stub journals do not prove that a real RDM binary rolls back after a switched fault, a failed build, or a candidate that exits on serve.
Evidence: SELF unittest of the three new LifecycleRealTests methods exited 0. Staging leaves the old pid and release in place; the next start clears that journal. Switched and an immediate serve exit both restore the previous real binary.
Impact: The crash candidate delegates version and identity to the real binary so the gate still sees a capable artifact. Login and rdm.ping stay NOT_RUN.
Refs: tests/integration/lifecycle/test_lifecycle_real.py
Co-authored-by: Cursor <cursoragent@cursor.com>

tests/integration/lifecycle/test_lifecycle_real.py

@@ -3,6 +3,7 @@
 由 make test-lifecycle 调用。未设置 SVC_TEST_REAL 时整组跳过，普通 make test
 不因此失败。登录与 rdm.ping 不在这里实现：RDM 没有 user/apikey。
 """
+import hashlib
 import importlib.util
 import json
 import os
@@ -249,6 +250,122 @@ class LifecycleRealTests(tsr.SvcAutoRecoverTests):
         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")
+
 
 if __name__ == "__main__":
     unittest.main()