feat: configurable merge gate (branch protection + directory isolation) and skill body/appendix checks

- cloud/merge_gate.py: pure gate; directory check reuses the v0.3
  git diff --name-only merge-base command; protection modes
  record (default, explicit confirm required when unprotected),
  enforce (always blocks when unprotected), off (explicit only).
  Confirm covers protection risk only, never directory violations.
  Protection status is caller/env-declared only; unknown is a
  recorded risk, not a claim of protection.
- server POST /tasks/{id}/merge: applies the gate only when gate
  evidence is supplied (or env declares protection); no-evidence
  calls behave as before plus a merge_gate field in the response.
  Blocked merges return 409 without changing status; confirmed
  unprotected merges write a merge_gate_risk audit entry.
- skill_seed.py: whole-directory seeding evaluated and rejected
  (cannot preserve tenant deletions; no benefit while the pack
  holds only SKILL.md); SKILL.md-only behaviour unchanged.
- software-engineering SKILL.md: gate paragraph added to the body
  and to the coordinator + client appendix sections (v0.2 text on
  this base; port to v0.3 text on rebase after #199 merges).
- tests: 16 new tests covering gate modes, directory reuse,
  body/appendix bidirectional assertions naming the missing side,
  SKILL.md-only seeding and tenant-edit preservation.

cloud/tests/test_merge_gate_protectskill.py

@@ -0,0 +1,318 @@
+"""主分支保护 + 目录隔离门禁（cloud/merge_gate）与技能工程双向断言。
+
+门禁口径：
+- 目录隔离复用 v0.3 技能文本的 git diff --name-only 检查，不另造；
+  服务端 POST /tasks/{id}/merge 在此之前只改状态、不查目录、不查保护。
+- 保护默认 record：未保护 + 无确认阻断，未保护 + 确认放行并记风险；
+  enforce 下未保护一律阻断；unknown 记风险、不阻断（代码无法核实
+  GitHub 保护，unknown 不等于已保护，也不等于未保护）。
+- 显式确认只覆盖保护风险，绝不覆盖目录违规。
+技能工程：
+- 正文与附录双向断言：每个新关键词在正文、附录都要出现，失败
+  信息指明缺失的一侧（正文缺失 / 附录缺失），不是笼统的 assertIn。
+- 播种仍只复制 SKILL.md：包内多余文件不进租户 home，租户已改
+  的 SKILL.md 绝不被覆盖；整目录复制明确不做（见 skill_seed.py）。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.merge_gate import (
+        check_directory_isolation, directory_diff_command,
+        evaluate_merge_gate, gate_from_request, gate_mode_from_env,
+        protection_from_env)
+    from cloud.server import create_server
+    from cloud.skill_seed import ensure_codex_home_skills
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.merge_gate import (
+        check_directory_isolation, directory_diff_command,
+        evaluate_merge_gate, gate_from_request, gate_mode_from_env,
+        protection_from_env)
+    from cloud.server import create_server
+    from cloud.skill_seed import ensure_codex_home_skills
+
+SKILL = Path(__file__).resolve().parents[1] / "skills" \
+    / "software-engineering" / "SKILL.md"
+BOTH_KEYWORDS = ("合并门禁", "CLOUD_MERGE_GATE_MODE", "显式确认",
+                  "git diff --name-only", "不做整目录复制")
+
+
+class MergeGateUnitTestCase(unittest.TestCase):
+    def test_record_unprotected_needs_confirm(self):
+        blocked = evaluate_merge_gate(protection="unprotected")
+        self.assertFalse(blocked["allowed"])
+        self.assertIn("branch_protection_risk_unconfirmed",
+                      blocked["blocks"])
+        ok = evaluate_merge_gate(protection="unprotected",
+                                 explicit_confirm=True)
+        self.assertTrue(ok["allowed"])
+        self.assertIn("branch protection is not enabled",
+                      " ".join(ok["risks"]))
+
+    def test_enforce_blocks_even_with_confirm(self):
+        gate = evaluate_merge_gate(protection="unprotected", mode="enforce",
+                                   explicit_confirm=True)
+        self.assertFalse(gate["allowed"])
+        self.assertIn("branch_protection_enforce", gate["blocks"])
+
+    def test_off_allows_but_records_risk(self):
+        gate = evaluate_merge_gate(protection="unprotected", mode="off")
+        self.assertTrue(gate["allowed"])
+        self.assertTrue(gate["risks"])
+
+    def test_unknown_is_risk_not_block(self):
+        gate = evaluate_merge_gate(protection="unknown")
+        self.assertTrue(gate["allowed"])
+        self.assertTrue(gate["risks"])
+        self.assertEqual(gate["directory"], None)
+
+    def test_protected_allows(self):
+        self.assertTrue(evaluate_merge_gate(
+            protection="protected")["allowed"])
+
+    def test_invalid_mode_falls_back_to_record_not_off(self):
+        self.assertEqual(gate_mode_from_env({"CLOUD_MERGE_GATE_MODE": "x"}),
+                         "record")
+        self.assertEqual(gate_mode_from_env({}), "record")
+        self.assertEqual(protection_from_env({}), "unknown")
+        gate = gate_from_request({}, env={"CLOUD_MERGE_GATE_MODE": "bogus",
+                                          "CLOUD_BRANCH_PROTECTION":
+                                          "unprotected"})
+        self.assertFalse(gate["allowed"])
+
+    def test_directory_reuses_v03_rule(self):
+        self.assertEqual(
+            directory_diff_command(),
+            "git diff --name-only $(git merge-base origin/main HEAD)...HEAD")
+        ok = check_directory_isolation(
+            ["apps/web/a.py", "apps/web/sub/b.py"], "apps/web")
+        self.assertTrue(ok["passed"])
+        self.assertEqual(ok["violations"], [])
+        bad = check_directory_isolation(
+            ["apps/web/a.py", "apps/ios/b.py", "server/c.py"], "apps/web")
+        self.assertFalse(bad["passed"])
+        self.assertEqual(sorted(bad["violations"]),
+                         ["apps/ios/b.py", "server/c.py"])
+        # 目录前缀边界：apps/web2 不是 apps/web 的子目录。
+        edge = check_directory_isolation(["apps/web2/x.py"], "apps/web")
+        self.assertEqual(edge["violations"], ["apps/web2/x.py"])
+        # 越界路径与契约变更批准：前者违规，后者放行但要记入 used。
+        esc = check_directory_isolation(["../secret.py", "/abs.py"],
+                                        "apps/web")
+        self.assertEqual(sorted(esc["violations"]),
+                         ["../secret.py", "/abs.py"])
+        exc = check_directory_isolation(
+            ["apps/web/a.py", "contracts/api.md"], "apps/web",
+            approved_exceptions=["contracts/api.md"])
+        self.assertTrue(exc["passed"])
+        self.assertEqual(exc["exceptions_used"], ["contracts/api.md"])
+
+    def test_confirm_never_overrides_directory_violation(self):
+        gate = evaluate_merge_gate(
+            files_changed=["apps/ios/b.py"], owned_dir="apps/web",
+            protection="unprotected", explicit_confirm=True)
+        self.assertFalse(gate["allowed"])
+        self.assertIn("directory_isolation_violation", gate["blocks"])
+
+    def test_files_without_owned_dir_is_inconclusive_and_blocks(self):
+        gate = evaluate_merge_gate(files_changed=["apps/web/a.py"],
+                                   owned_dir=None)
+        self.assertFalse(gate["allowed"])
+        self.assertIn("directory_owned_dir_unknown", gate["blocks"])
+        no_evidence = evaluate_merge_gate()
+        self.assertTrue(no_evidence["allowed"])
+        self.assertIsNone(no_evidence["directory"])
+
+
+class SkillBidirectionalTestCase(unittest.TestCase):
+    def test_protectskill_keywords_in_body_and_appendix(self):
+        text = SKILL.read_text(encoding="utf-8")
+        main, sep, appendix = text.partition("## 附录：角色模板")
+        self.assertTrue(sep, "appendix marker missing")
+        for kw in BOTH_KEYWORDS:
+            self.assertIn(kw, main, "正文缺失：" + kw)
+            self.assertIn(kw, appendix, "附录缺失：" + kw)
+
+    def test_gate_in_coordinator_and_client_appendix_sections(self):
+        text = SKILL.read_text(encoding="utf-8")
+        coord = text.split("## 附录角色：00-coordinator", 1)[1].split(
+            "## 附录角色：01-", 1)[0]
+        client = text.split("## 附录角色：04-client-developer", 1)[1].split(
+            "## 附录角色：05-", 1)[0]
+        self.assertIn("合并门禁", coord, "附录协调员章节缺失：合并门禁")
+        self.assertIn("合并门禁", client, "附录客户端章节缺失：合并门禁")
+
+
+class SkillSeedNoWholeDirTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _pack(self):
+        src = os.path.join(self.tmp.name, "pack", "demo-skill")
+        os.makedirs(src, exist_ok=True)
+        open(os.path.join(src, "SKILL.md"), "w").write("pack version\n")
+        # 整目录复制才会带走的多余文件：现行行为必须忽略它们。
+        open(os.path.join(src, "extra.txt"), "w").write("extra\n")
+        os.makedirs(os.path.join(src, "templates"), exist_ok=True)
+        open(os.path.join(src, "templates", "x.md"), "w").write("x\n")
+        return os.path.join(self.tmp.name, "pack")
+
+    def test_only_skill_md_seeded_and_tenant_edit_never_overwritten(self):
+        home = os.path.join(self.tmp.name, "home")
+        out = ensure_codex_home_skills(home, self._pack())
+        self.assertEqual(out["skills_copied"], ["demo-skill"])
+        dest_dir = os.path.join(home, "skills", "demo-skill")
+        self.assertTrue(os.path.isfile(
+            os.path.join(dest_dir, "SKILL.md")))
+        self.assertFalse(os.path.exists(
+            os.path.join(dest_dir, "extra.txt")))
+        self.assertFalse(os.path.exists(
+            os.path.join(dest_dir, "templates")))
+        dest = os.path.join(dest_dir, "SKILL.md")
+        open(dest, "w").write("tenant edited version\n")
+        out2 = ensure_codex_home_skills(home, self._pack())
+        self.assertEqual(out2["skills_skipped_differ"], ["demo-skill"])
+        self.assertEqual(open(dest).read(), "tenant edited version\n")
+
+
+class MergeEndpointGateTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+        self._saved_env = {k: os.environ.get(k) for k in
+                           ("CLOUD_MERGE_GATE_MODE",
+                            "CLOUD_BRANCH_PROTECTION")}
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        for key, value in self._saved_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _completed_task(self):
+        status, task = self._req(
+            "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "x"},
+            self.owner_key)
+        self.assertEqual(status, 201)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE tasks SET status = 'completed' WHERE id = ?",
+                (task["id"],))
+            conn.commit()
+        finally:
+            conn.close()
+        return task
+
+    def test_no_gate_fields_behaves_like_old_endpoint(self):
+        task = self._completed_task()
+        status, merged = self._req(
+            "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(merged["status"], "merged")
+        self.assertIn("merge_gate", merged)
+
+    def test_directory_violation_blocks_without_changing_status(self):
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            {"files_changed": ["apps/web/a.py", "apps/ios/b.py"],
+             "owned_dir": "apps/web"},
+            self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertIn("directory_isolation_violation",
+                      body["merge_gate"]["blocks"])
+        _s, got = self._req("GET", "/tasks/%s" % task["id"],
+                            key=self.owner_key)
+        self.assertEqual(got["status"], "completed")
+
+    def test_unprotected_confirm_allows_and_audits_risk(self):
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            {"branch_protection": "unprotected"},
+            self.owner_key)
+        self.assertEqual(status, 409)
+        task2 = self._completed_task()
+        status, merged = self._req(
+            "POST", "/tasks/%s/merge" % task2["id"],
+            {"branch_protection": "unprotected",
+             "merge_gate_confirm": True},
+            self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(merged["merge_gate"]["protection"], "unprotected")
+        audit = self.auth.list_audit(self.project["id"])
+        self.assertTrue(any(a["action"] == "merge_gate_risk"
+                            for a in audit))
+
+    def test_enforce_env_blocks_even_with_confirm(self):
+        os.environ["CLOUD_MERGE_GATE_MODE"] = "enforce"
+        os.environ["CLOUD_BRANCH_PROTECTION"] = "unprotected"
+        task = self._completed_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/merge" % task["id"],
+            {"merge_gate_confirm": True},
+            self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertIn("branch_protection_enforce",
+                      body["merge_gate"]["blocks"])
+
+
+if __name__ == "__main__":
+    unittest.main()