"""svc.sh 实例身份回归（U1 设计 §7 T1–T6 + I3 DB marker，manifest 治理）。

覆盖：连续裸启停身份稳定（T1）、只改端口锚点不动（T2）、数据根漂移拒绝（T3）、
迁移窗口/损坏阻塞不猜测（T4）、marker 缺失/外宾/manifest 丢失均为事件（T5）、
cwd 无关性（T6）、SQLite DB marker ensure/错 id 阻塞（I3 桩路径）、原子写崩溃窗口。
桩二进制与夹具复用 tests/test_svc_recover.py。真实 PG 身份专项（跨库）在
SVC_TEST_PG=1 且具备环境时执行；否则 test_Z_real_pg_not_run 显式 NOT_RUN。
"""
import importlib.util
import json
import os
import shutil
import sqlite3
import subprocess
import tempfile
import unittest
from pathlib import Path

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

_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)


class IdentityTests(tsr.SvcAutoRecoverTests):
    def setUp(self):
        super().setUp()
        shutil.copy2(SOURCE / "scripts" / "svc_manifest.py",
                     self.root / "scripts" / "svc_manifest.py")

    # ---- 夹具 ----

    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 marker(self, name):
        m = self.manifest(name)
        if not m:
            return None
        p = Path(m["identity"]["data_root"]) / ".agentx" / "data-marker.json"
        return json.loads(p.read_text(encoding="utf-8")) if p.exists() else None

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

    # ---- T1 连续裸启停：同一身份与数据根 ----

    def test_T1_bare_restart_stable_identity(self):
        self.seed_agent("id1", 9431)
        self.start_ok("id1")
        m1 = self.manifest("id1")
        self.assertIsNotNone(m1, "start 应完成 manifest 迁移")
        iid = m1["instance_id"]
        self.assertTrue(iid.startswith("svc-id1-"), iid)
        mk = self.marker("id1")
        self.assertEqual(mk["instance_id"], iid)
        data_before = m1["identity"]["data_root"]
        exec_before = m1["kv"]["exec"]
        self.svc("id1", "stop")
        self.start_ok("id1")
        m2 = self.manifest("id1")
        self.assertEqual(m2["instance_id"], iid, "裸启停不得换身份")
        self.assertEqual(m2["identity"]["data_root"], data_before)
        self.assertEqual(m2["kv"]["exec"], exec_before, "裸启停不得漂移 exec（数据/插件根）")
        self.assertGreaterEqual(m2["revision"], m1["revision"], "整份提交 revision 单调")
        self.assertTrue(self.running("id1"))
        self.svc("id1", "stop")

    # ---- T2 只改端口：数据/DB 锚点不动 ----

    def test_T2_port_change_keeps_anchors(self):
        self.seed_agent("id2", 9432)
        self.start_ok("id2")
        anchor = self.manifest("id2")["identity"]
        self.svc("id2", "stop")
        result = self.svc("id2", "start", "--port", "9433")
        self.assertIn("已启动", result.stdout)
        m2 = self.manifest("id2")
        self.assertEqual(m2["identity"]["data_root"], anchor["data_root"], "改端口不得动数据根")
        self.assertEqual(m2["identity"]["db_mode"], anchor["db_mode"])
        self.assertEqual(m2["identity"]["db_ref"], anchor["db_ref"])
        self.assertIn("9433", m2["kv"]["addr"])
        self.svc("id2", "stop")

    # ---- T3 数据根漂移：拒绝启动，不新建空系统 ----

    def test_T3_data_drift_rejected(self):
        self.seed_agent("id3", 9434)
        self.start_ok("id3")
        self.svc("id3", "stop")
        other = self.root / "other-data"
        other.mkdir()
        result = self.svc("id3", "start", "--data", str(other), check=False)
        self.assertNotEqual(result.returncode, 0, "数据根漂移必须拒绝")
        self.assertTrue(any(k in result.stdout for k in ("身份校验失败", "数据根漂移")),
                        result.stdout)
        self.assertFalse(self.running("id3"), "拒绝后不得有服务进程")
        self.assertEqual(self.manifest("id3")["identity"]["data_root"],
                         str(self.runtime("id3") / "data"), "锚点不得被污染")

    # ---- T4 迁移窗口与损坏：不猜测 ----

    def test_T4a_leftover_new_file_ignored(self):
        self.seed_agent("id4", 9435)
        # 模拟 rename 前崩溃残留：.new 半成品必须被忽略/覆盖，不影响迁移结果
        (self.runtime("id4") / "manifest.json.new").write_text("{ truncated")
        self.start_ok("id4")
        m = self.manifest("id4")
        self.assertIsNotNone(m)
        self.assertEqual(m["schema"], 1)
        self.svc("id4", "stop")

    def test_T4b_corrupt_manifest_blocks(self):
        self.seed_agent("id5", 9436)
        self.start_ok("id5")
        self.svc("id5", "stop")
        (self.runtime("id5") / "manifest.json").write_text("{ not json")
        result = self.svc("id5", "start", check=False)
        self.assertNotEqual(result.returncode, 0, "损坏 manifest 必须阻塞而非猜测重建")
        self.assertIn("损坏", result.stdout)
        self.assertFalse(self.running("id5"))

    # ---- T5 marker/manifest 事件：缺一即阻塞，不自动新建 ----

    def test_T5a_missing_marker_blocks(self):
        self.seed_agent("id6", 9437)
        self.start_ok("id6")
        self.svc("id6", "stop")
        m = self.manifest("id6")
        mk = Path(m["identity"]["data_root"]) / ".agentx" / "data-marker.json"
        mk.unlink()
        result = self.svc("id6", "start", check=False)
        self.assertNotEqual(result.returncode, 0, "marker 缺失即事件")
        self.assertIn("marker", result.stdout)
        self.assertFalse(self.running("id6"))

    def test_T5b_foreign_marker_blocks(self):
        self.seed_agent("id7", 9438)
        self.start_ok("id7")
        self.svc("id7", "stop")
        m = self.manifest("id7")
        mk = Path(m["identity"]["data_root"]) / ".agentx" / "data-marker.json"
        data = json.loads(mk.read_text(encoding="utf-8"))
        data["instance_id"] = "svc-other-zzzzzzzzzzzz"
        mk.write_text(json.dumps(data), encoding="utf-8")
        result = self.svc("id7", "start", check=False)
        self.assertNotEqual(result.returncode, 0, "marker 身份不符即事件")
        self.assertFalse(self.running("id7"))

    def test_T5c_lost_manifest_blocks_not_rebuild(self):
        self.seed_agent("id8", 9439)
        self.start_ok("id8")
        m0 = self.manifest("id8")
        iid = m0["instance_id"]
        marker_path = Path(m0["identity"]["data_root"]) / ".agentx" / "data-marker.json"
        self.svc("id8", "stop")
        (self.runtime("id8") / "manifest.json").unlink()
        result = self.svc("id8", "start", check=False)
        # manifest 丢失后 marker 仍在：重建会撞归属冲突 → 中止（I4），不得 legacy 复活
        self.assertNotEqual(result.returncode, 0, "丢 manifest 不得静默重建/legacy 复活")
        self.assertTrue(any(k in result.stdout for k in ("归属冲突", "身份治理冲突")),
                        result.stdout)
        self.assertFalse(self.running("id8"))
        # marker 保持原身份，未被新随机身份覆盖
        mk = json.loads(marker_path.read_text(encoding="utf-8"))
        self.assertEqual(mk["instance_id"], iid)

    # ---- T6 cwd 无关：从别处调用身份不变 ----

    def test_T6_cwd_independent_identity(self):
        self.seed_agent("id9", 9440)
        self.start_ok("id9")
        iid = self.manifest("id9")["instance_id"]
        self.svc("id9", "stop")
        elsewhere = tempfile.mkdtemp(prefix="svc-elsewhere-")
        self.addCleanup(shutil.rmtree, elsewhere, True)
        proc = subprocess.run(
            ["bash", str(self.root / "scripts" / "svc.sh"), "id9", "start"],
            cwd=elsewhere, env=self.env, text=True,
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=120)
        self.assertEqual(proc.returncode, 0, proc.stdout)
        self.assertIn("已启动", proc.stdout)
        self.assertEqual(self.manifest("id9")["instance_id"], iid, "换 cwd 不得换身份")
        self.svc("id9", "stop")

    # ---- I3 DB marker（SQLite 桩路径；与真实 PG 跨库专项分离） ----

    def db_identity_row(self, name):
        m = self.manifest(name)
        if not m:
            return None
        db = Path(m["identity"]["data_root"]) / "agent-system.db"
        if not db.exists():
            return None
        con = sqlite3.connect(str(db))
        try:
            rows = list(con.execute(
                "SELECT instance_id, dbref FROM agent_system_identity"))
        finally:
            con.close()
        return rows

    # ---- U1 批量隔离、候选门禁与启动语义（2026-10-12 rdm1/rdmcand 事故回归） ----

    def noid_stub(self):
        """自包含「无 identity」桩：复用 recover STUB，把 identity 分支换成 2026-10-12
        RDM 组合产物形态——identity 子命令落入服务路径、被身份门禁拒绝，报错文本
        恰含「identity ensure」字样（旧探测 grep ensure 的误报源）。不 exec 转发：
        serve 进程 cmdline 必须保留实例 bin 路径（pidcheck 契约与真实产物一致）。"""
        p = self.root / "stub-agent-system-noid"
        if not p.exists():
            stub = tsr.STUB
            head, sep, tail = stub.partition("  identity)")
            assert sep, "STUB 缺 identity 分支锚点"
            _, sep2, tail2 = tail.partition("    exit $? ;;")
            assert sep2, "STUB 缺 identity 分支结尾锚点"
            noid_case = (
                "  identity)\n"
                "    echo \"system identity: system identity missing: agent_system_identity 无行"
                "（先 svc init / identity ensure，禁止静默建空身份）\" >&2\n"
                "    exit 1 ;;")
            body = head + noid_case + tail2
            body = body.replace(
                '\'{"protocol":1,"product":"agent-system","version":"stub","identity":{"ops":["ensure","verify"],"open":"existing-no-migrate","dsn":"env"},"cli":{"unknownSubcommand":"reject"}}\'',
                '"not-json"',
            )
            p.write_text(body)
            p.chmod(0o755)
        return p

    def install_noid_bin(self, name):
        """把实例磁盘产物换成无 identity 形态（rdm1/rdmcand 事故现场的制品状态）。"""
        shutil.copy2(self.noid_stub(), self.runtime(name) / "bin" / "agent-system")

    def set_kv(self, name, key, value):
        f = self.runtime(name) / "svc.env"
        lines = f.read_text().splitlines()
        out, hit = [], False
        for l in lines:
            if l.startswith(key + "="):
                out.append("%s=%s" % (key, value)); hit = True
            else:
                out.append(l)
        if not hit:
            out.append("%s=%s" % (key, value))
        f.write_text("\n".join(out) + "\n")

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

    def clear_serve_record(self):
        (self.trace / "serve.txt").unlink(missing_ok=True)

    def git_ok(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)

    def git_sha(self, cwd, rev):
        proc = subprocess.run(["git", "rev-parse", "--short", rev], cwd=cwd,
                              env=self.env, text=True,
                              stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=30)
        return proc.stdout.strip()

    def _make_track_repo(self, artifact, target, with_web=False):
        """双提交源仓夹具（旧/新两版）：Makefile 产物 = artifact 复制为 target。"""
        remote, work = self.root / "trackremote.git", self.root / "trackwork"
        for p in (remote, work):
            if p.exists():
                shutil.rmtree(p)
        self.git_ok(self.root, "init", "--bare", "--initial-branch=main", str(remote))
        self.git_ok(self.root, "init", "--initial-branch=main", str(work))
        (work / "Makefile").write_text(
            "build:\n\tmkdir -p bin\n\tcp %s bin/%s\n" % (artifact, target))
        if with_web:
            (work / "web").mkdir()
            (work / "web" / "index.html").write_text("<html>rdm</html>\n")
        self.git_ok(work, "add", ".")
        self.git_ok(work, "commit", "-m", "track v1")
        self.git_ok(work, "remote", "add", "origin", str(remote))
        self.git_ok(work, "push", "-u", "origin", "main")
        (work / "v2.marker").write_text("v2\n")
        self.git_ok(work, "add", ".")
        self.git_ok(work, "commit", "-m", "track v2")
        self.git_ok(work, "push")
        return remote, self.git_sha(work, "HEAD~1"), self.git_sha(work, "HEAD")

    def make_src_repo(self):
        """dev 轨道：产物 = 可用 identity 桩；预克隆进 SVC_SRC_DIR（srccache）。"""
        remote, old, new = self._make_track_repo(self.stub, "agent-system")
        cache = self.root / "srccache"
        if cache.exists():
            shutil.rmtree(cache)
        self.git_ok(self.root, "clone", str(remote), str(cache))
        return old, new

    def make_rdm_repo(self, artifact):
        """rdm 轨道：产物 = 指定桩（可注入无 identity 形态）。"""
        return self._make_track_repo(artifact, "agent-system-rdm", with_web=True)

    def rdm_env(self, remote):
        env = dict(self.env)
        env.update({"SVC_RDM_SRC_DIR": str(self.root / "rdmcache"),
                    "SVC_RDM_ORIGIN": str(remote)})
        return env

    def test_U1_update_refuses_identityless_candidate_keeps_old_state(self):
        """manifest 治理实例 × 无 identity 候选（RDM 旧形态）：停服前拒绝——
        旧服务继续运行、旧制品与登记不变（2026-10-12 rdm1/rdmcand 反例：停服→
        覆盖→启动即拒，只剩混合状态）。"""
        remote, old_sha, new_sha = self.make_rdm_repo(self.noid_stub())
        self.seed_agent("rdx", 9451)
        self.start_ok("rdx")               # 已建立身份，有能力制品在运行
        self.set_kv("rdx", "source", "rdm@%s" % old_sha)
        pid = self.pid_of("rdx")
        bin_before = (self.runtime("rdx") / "bin" / "agent-system").read_bytes()

        result = self.svc("update", "rdx", check=False, env=self.rdm_env(remote))
        self.assertNotEqual(result.returncode, 0, result.stdout)
        self.assertIn("候选产物不支持 identity 管理", result.stdout)
        # 拒绝发生在停旧服务之前：进程、制品、登记全部保持
        self.assertTrue(self.running("rdx"), "拒绝不得停掉可用旧服务")
        self.assertEqual(self.pid_of("rdx"), pid)
        self.assertEqual((self.runtime("rdx") / "bin" / "agent-system").read_bytes(),
                         bin_before, "旧制品不得被覆盖")
        self.assertEqual(self.kv("rdx", "source"), "rdm@%s" % old_sha)
        hist = (self.runtime("rdx") / "history.jsonl").read_text()
        self.assertIn('"event":"update"', hist)
        self.assertIn('"result":"fail"', hist)
        self.svc("rdx", "stop")

    def test_U1_update_ensures_missing_row_before_stop(self):
        """已建立身份后删行：update 在停服前拒绝，不自动补行，旧进程保持。
        补行只走 adopt --confirm。"""
        old_sha, new_sha = self.make_src_repo()
        self.seed_agent("rowx", 9452)
        self.start_ok("rowx")
        self.set_kv("rowx", "source", "dev@%s" % old_sha)
        db = Path(self.manifest("rowx")["identity"]["data_root"]) / "agent-system.db"
        con = sqlite3.connect(str(db))
        con.execute("DELETE FROM agent_system_identity")
        con.commit()
        con.close()

        self.clear_serve_record()
        result = self.svc("update", "rowx", check=False)
        self.assertNotEqual(result.returncode, 0, result.stdout)
        self.assertIn("候选无法验证已建立的 DB 身份", result.stdout)
        self.assertTrue(self.running("rowx"), "缺行必须在停旧服务前拒绝")
        self.assertEqual(self.db_identity_row("rowx"), [], "established 不得自动补行")
        self.svc("rowx", "stop")

    def test_U1_bare_start_missing_row_is_event_init_recovers(self):
        """普通启动只验证不补行：DB 行缺失 → 拒绝且不建空行（缺行即事件）；
        显式 init 补齐后恢复。"""
        self.seed_agent("rowm", 9453)
        self.start_ok("rowm")
        iid = self.manifest("rowm")["instance_id"]
        self.svc("rowm", "stop")
        db = Path(self.manifest("rowm")["identity"]["data_root"]) / "agent-system.db"
        con = sqlite3.connect(str(db))
        con.execute("DELETE FROM agent_system_identity")
        con.commit()
        con.close()

        result = self.svc("rowm", "start", check=False)
        self.assertNotEqual(result.returncode, 0, result.stdout)
        self.assertIn("DB 身份校验失败", result.stdout)
        self.assertFalse(self.running("rowm"))
        self.assertEqual(self.db_identity_row("rowm"), [], "缺行不得被 start 静默补建")

        refused = self.svc("rowm", "init", check=False)
        self.assertNotEqual(refused.returncode, 0, refused.stdout)
        self.assertIn("adopt", refused.stdout)
        self.svc("rowm", "adopt", "--confirm", iid)
        self.assertEqual(self.db_identity_row("rowm"), [(iid, "sqlite")])
        self.start_ok("rowm")
        self.svc("rowm", "stop")

    def test_U1_capability_probe_no_false_positive_no_hang(self):
        """能力探测无副作用：含 ensure 字样的服务态门禁报错不算能力（旧实现
        误报点）；identity 即挂死的产物探测有界返回无能力。"""
        def probe(binpath):
            # svc.sh 顶层启用 errexit：source 后裸调用失败函数会杀掉探针 shell，
            # 用 if 条件上下文（errexit 豁免）取回真实返回码。
            return subprocess.run(
                ["bash", "-c",
                 'source "%s" >/dev/null 2>&1; ENV_BIN="%s"; '
                 'if svc_bin_has_identity; then echo rc=0; else echo rc=$?; fi'
                 % (self.root / "scripts" / "svc.sh", binpath)],
                cwd=self.root, env=self.env, text=True,
                stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                timeout=30).stdout.strip()

        self.assertIn("rc=0", probe(self.stub), "identity 桩应判有能力")
        self.assertIn("rc=1", probe(self.noid_stub()),
                      "报错文本含 ensure 不得误报为有能力")
        hang = self.root / "stub-hang"
        hang.write_text(
            '#!/usr/bin/env bash\ncase "${1:-}" in version) sleep 30 ;; '
            '*) exec "%s" "$@" ;; esac\n' % self.stub)
        hang.chmod(0o755)
        self.assertIn("rc=1", probe(hang), "探测挂死须有界返回无能力")

    def test_I3_start_ensures_sqlite_identity(self):
        self.seed_agent("idb1", 9441)
        self.start_ok("idb1")
        iid = self.manifest("idb1")["instance_id"]
        rows = self.db_identity_row("idb1")
        self.assertEqual(rows, [(iid, "sqlite")], rows)
        self.svc("idb1", "stop")
        self.start_ok("idb1")  # ensure 幂等 + verify
        self.assertEqual(self.db_identity_row("idb1"), [(iid, "sqlite")])
        self.svc("idb1", "stop")

    def test_I3_wrong_db_identity_blocks_start(self):
        self.seed_agent("idb2", 9442)
        self.start_ok("idb2")
        iid = self.manifest("idb2")["instance_id"]
        self.svc("idb2", "stop")
        db = Path(self.manifest("idb2")["identity"]["data_root"]) / "agent-system.db"
        con = sqlite3.connect(str(db))
        con.execute("DELETE FROM agent_system_identity")
        con.execute(
            "INSERT INTO agent_system_identity(instance_id,dbref,schema_version,created_at)"
            " VALUES(?,?,?,?)",
            ("svc-foreign-deadbeef", "sqlite", 26, "2026-01-01T00:00:00Z"))
        con.commit()
        con.close()
        result = self.svc("idb2", "start", check=False)
        self.assertNotEqual(result.returncode, 0, result.stdout)
        self.assertTrue(any(k in result.stdout for k in ("DB 身份", "ensure", "mismatch", "校验失败")),
                        result.stdout)
        self.assertFalse(self.running("idb2"))
        # 外宾行不得被 start 静默改写
        self.assertEqual(self.db_identity_row("idb2"),
                         [("svc-foreign-deadbeef", "sqlite")])
        self.assertNotEqual(iid, "svc-foreign-deadbeef")

    # ---- 真实 PG 专项：跨库身份；缺环境显式 NOT_RUN ----

    def test_Z_real_pg_not_run(self):
        if os.environ.get("SVC_TEST_PG") == "1" and os.environ.get("AGENT_PG_TEST_DSN"):
            self.skipTest("真 PG 专项应由隔离套件执行（本桩文件不连真实 PG）")
        self.skipTest("真实 PG 身份专项 NOT_RUN：需真实实例（SVC_TEST_PG=1 + AGENT_PG_TEST_DSN）")


class AtomicWriteTests(unittest.TestCase):
    """崩溃窗口单元级：os.replace 失败时原文件必须完好（I3 半写不可能）。"""

    def test_replace_failure_keeps_original(self):
        spec = importlib.util.spec_from_file_location(
            "sm", SOURCE / "scripts" / "svc_manifest.py")
        sm = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(sm)
        with tempfile.TemporaryDirectory() as d:
            path = os.path.join(d, "m.json")
            sm.atomic_write(path, '{"v": 1}\n')
            real_replace = os.replace
            calls = {"n": 0}

            def boom(a, b):
                calls["n"] += 1
                raise OSError("simulated crash before rename completes")

            os.replace = boom
            try:
                try:
                    sm.atomic_write(path, '{"v": 2}\n')
                except OSError:
                    pass
            finally:
                os.replace = real_replace
            self.assertEqual(calls["n"], 1)
            with open(path, encoding="utf-8") as f:
                self.assertEqual(f.read(), '{"v": 1}\n', "rename 前崩溃必须保留完整旧版")


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