gapfill G7: make _merge_task status commit and success audit atomic (same connection, audit_on_conn, one commit; audit failure rolls back to completed instead of merged-plus-500; ledger-head IntegrityError retries whole segment 5x mirroring AuthStore.audit); success semantics unchanged

cloud/server.py

@@ -68,6 +68,7 @@ from cloud.quota import quota_for_owner as _quota_for_owner
 from cloud.timeouts import STALE_SWEEP_SECONDS as _STALE_SWEEP
 from cloud.timeouts import approval_default_seconds as _approval_default
 from cloud.timeouts import validate_server_timeout as _validate_timeout
+from cloud.auth import audit_on_conn as _audit_on_conn
 from cloud.db import connect as _db_connect
 from cloud.db import create_database as _create_database
 from cloud.db import init_postgres_schema as _init_postgres_schema
@@ -6578,15 +6579,47 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au
                 "merge_gate": gate,
             })
             return
+        # G7（A4 时序残留）:状态提交与成功审计必须同一事务。
+        # 旧写法先 commit 状态、再另开 AuthStore 连接写审计：
+        # 审计失败时状态已是 merged、响应却是 500——“合并了却
+        # 报失败”，调用方无法靠状态判断，也无法安全重试
+        # （重复合并 409）。新写法用 audit_on_conn 把审计对
+        # 写进同一连接、一次 commit：审计失败则整段回滚，状态
+        # 仍是 completed、响应 500（_dispatch_safe 兜底），调用
+        # 方可原样重试合并。链头唯一冲突的重试口径与
+        # AuthStore.audit 完全一致（5 次、只重试
+        # IntegrityError，其它错误直接抛）。成功语义不变：
+        # 200 体、审计内容（merge_task/task_id）、门禁响应全同。
+        # 前提：tasks 与 audit_log/audit_ledger 在同一库
+        # （_db 与 _auth_store 都连 server.db_path；PG 模式下
+        # _connect 同样分发到同一 PG 库，? 占位由适配层转 %s）。
         now = time.time()
         conn = self._db()
         try:
-            conn.execute(
-                "UPDATE tasks SET status = 'merged', updated_at = ?"
-                " WHERE id = ?",
-                (now, task_id),
-            )
-            conn.commit()
+            last_exc = None
+            for _attempt in range(5):
+                try:
+                    conn.execute(
+                        "UPDATE tasks SET status = 'merged',"
+                        " updated_at = ? WHERE id = ?",
+                        (now, task_id),
+                    )
+                    _audit_on_conn(conn, user_id, task["project_id"],
+                                   "merge_task", task_id)
+                    conn.commit()
+                    break
+                except Exception as exc:
+                    last_exc = exc
+                    try:
+                        conn.rollback()
+                    except Exception:
+                        pass
+                    if not isinstance(
+                            exc, (sqlite3.IntegrityError,
+                                  _PgIntegrityError)):
+                        raise
+            else:
+                raise last_exc
             task = _task_to_dict(
                 conn.execute(
                     "SELECT * FROM tasks WHERE id = ?", (task_id,)
@@ -6594,8 +6627,6 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au
             )
         finally:
             conn.close()
-        with _auth_store(self.server.db_path) as store:
-            store.audit(user_id, task["project_id"], "merge_task", task_id)
         task["merge_gate"] = gate
         self._send_json(200, task)
 

cloud/tests/test_gapfill_g7_merge_atomic.py

@@ -0,0 +1,184 @@
+"""gapfill G7：_merge_task 状态提交与成功审计同一事务。
+
+旧写法（A4 时序残留）：先 commit 状态为 merged、再另开
+AuthStore 连接写 merge_task 审计；审计失败时状态已是
+merged、响应却是 500——“合并了却报失败”，调用方无法靠
+状态判断，也无法安全重试（重复合并 409）。
+
+新写法：audit_on_conn 把审计对（audit_log 行 + 链条目）
+写进同一连接、一次 commit；审计失败则整段回滚，状态仍是
+completed、异常上抛后由 _dispatch_safe 转 500 JSON，调用方
+可原样重试合并。链头唯一冲突的重试口径与 AuthStore.audit
+完全一致（5 次、只重试 IntegrityError，其它错误直接抛）。
+成功语义不变：200 体、审计内容、门禁响应全同；本文件第
+一项钉住成功路径与旧行为一致，后三项钉住失败路径的新行为。
+
+真实模型 0 轮，不联网；不跑真实 HTTP（handler 直调 +
+真实 AuthStore/DB，审计行直接查库验证）。
+"""
+
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+import cloud.server as server_mod  # noqa: E402
+from cloud.auth import AuthStore, audit_on_conn  # noqa: E402
+from cloud.server import CloudControlPlaneHandler, init_db  # noqa: E402
+
+NO_BLOCK = {"merge_blocked": False, "release_blocked": False,
+            "blocked_by": None, "round": None, "failed_checks": []}
+
+HARD_BODY = {
+    "files_changed": ["apps/web/a.py"],
+    "owned_dir": "apps/web",
+    "branch_protection": "protected",
+    "ci_status": "success",
+    "ci_checks": [{"name": "ci", "status": "success"}],
+    "required_checks": ["ci"],
+    "contract_version": "v2",
+    "contract_frozen": True,
+    "coordinator_approved": True,
+    "unresolved_review_comments": 0,
+}
+
+
+class MergeAtomicTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, _key = self.auth.create_user("owner")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO projects(id, name, owner_id, created_at,"
+                " repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, 0)",
+                ("p1", "P", self.owner_id, time.time(), None))
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " created_by, created_at, updated_at)"
+                " VALUES(?, ?, ?, ?, ?, ?, ?)",
+                ("t1", "p1", "do work", "completed", self.owner_id,
+                 time.time(), time.time()))
+            conn.commit()
+        finally:
+            conn.close()
+        self.auth.create_project("p1", self.owner_id)
+
+    def tearDown(self):
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _handler(self):
+        h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler)
+        h.server = SimpleNamespace(db_path=self.db_path)
+        sent = []
+        h._send_json = lambda status, obj, extra_headers=None: sent.append(
+            (status, obj))
+        h._read_json = lambda: (dict(HARD_BODY), None)
+        h.sent = sent
+        return h
+
+    def _merge(self, h):
+        with mock.patch.object(server_mod, "_pushback_block",
+                               return_value=dict(NO_BLOCK)):
+            h._merge_task(self.owner_id, "t1")
+
+    def _status(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            return conn.execute(
+                "SELECT status FROM tasks WHERE id = 't1'").fetchone()[0]
+        finally:
+            conn.close()
+
+    def _merge_audits(self):
+        return [a for a in self.auth.list_audit("p1")
+                if a["action"] == "merge_task"]
+
+    def test_success_unchanged(self):
+        # 成功语义与旧写法一致：200、merged、审计有行、门禁在体里。
+        h = self._handler()
+        self._merge(h)
+        status, obj = h.sent[-1]
+        self.assertEqual(status, 200)
+        self.assertEqual(obj["status"], "merged")
+        self.assertIn("merge_gate", obj)
+        self.assertEqual(self._status(), "merged")
+        self.assertEqual(len(self._merge_audits()), 1)
+        self.assertEqual(self._merge_audits()[0]["detail"], "t1")
+
+    def test_audit_failure_rolls_back_status(self):
+        # 旧行为是 merged + 500；新行为是 completed + 抛错
+        # （_dispatch_safe 转 500 JSON），调用方可原样重试。
+        h = self._handler()
+        with mock.patch.object(
+                server_mod, "_audit_on_conn",
+                side_effect=RuntimeError("audit down")):
+            with self.assertRaises(RuntimeError):
+                self._merge(h)
+        self.assertEqual(h.sent, [])
+        self.assertEqual(self._status(), "completed")
+        self.assertEqual(self._merge_audits(), [])
+        # 原样重试（审计恢复后）必须成功：状态仍是 completed，
+        # 不是卡在 merged 导致重复合并 409。
+        h2 = self._handler()
+        self._merge(h2)
+        self.assertEqual(h2.sent[-1][0], 200)
+        self.assertEqual(self._status(), "merged")
+
+    def test_ledger_conflict_retries_then_succeeds(self):
+        # 链头唯一冲突重试整段（与 AuthStore.audit 同口径）：
+        # 前两次冲突、第三次成功，整体仍是一次合并、一条审计。
+        calls = {"n": 0}
+
+        def flaky(conn, user_id, project_id, action, detail="",
+                  created_at=None):
+            calls["n"] += 1
+            if calls["n"] <= 2:
+                raise sqlite3.IntegrityError("chain head race")
+            return audit_on_conn(conn, user_id, project_id, action,
+                                 detail=detail, created_at=created_at)
+
+        h = self._handler()
+        with mock.patch.object(server_mod, "_audit_on_conn",
+                               side_effect=flaky):
+            self._merge(h)
+        self.assertEqual(h.sent[-1][0], 200)
+        self.assertEqual(calls["n"], 3)
+        self.assertEqual(self._status(), "merged")
+        self.assertEqual(len(self._merge_audits()), 1)
+
+    def test_ledger_conflict_exhausted_rolls_back(self):
+        # 5 次冲突耗尽：抛错、状态回滚、审计无行、不留半截合并。
+        calls = {"n": 0}
+
+        def always_conflict(*args, **kwargs):
+            calls["n"] += 1
+            raise sqlite3.IntegrityError("chain head race")
+
+        h = self._handler()
+        with mock.patch.object(server_mod, "_audit_on_conn",
+                               side_effect=always_conflict):
+            with self.assertRaises(sqlite3.IntegrityError):
+                self._merge(h)
+        self.assertEqual(calls["n"], 5)
+        self.assertEqual(self._status(), "completed")
+        self.assertEqual(self._merge_audits(), [])
+
+
+if __name__ == "__main__":
+    unittest.main()