audit: wire loop driver to server endpoints and release gate (track 2)

Server-side execution for the track-1 pure driver (#260), no driver
semantics changed: audit verdict / fix completion / correction /
rebuild confirmation endpoints persist LoopState in audit_loop_state,
write every step through the #241/#247 ledger path (ledger first,
state second, fail-closed), derive fix/rebuild subtasks, and block
merge/release via the existing #247 chain reader. POST
/tasks/{id}/release follows the #259 contract (global-admin write,
fresh hard-gate evidence, merged status, multi-source 409 with no
audit write on block); deploy.py stays outside task gates.

cloud/server.py

@@ -106,6 +106,16 @@ from cloud.marker_tools import parse_spawn_defaults as _parse_spawn_defaults
 from cloud.merge_gate import evaluate_merge_gate as _merge_gate_evaluate
 from cloud.merge_gate import protection_from_env as _merge_gate_protection
 from cloud.audit_wiring import pushback_block_for_task as _pushback_block
+from cloud.audit_loop_wiring import build_quota_check as _loop_quota_check
+from cloud.audit_loop_wiring import cost_threshold_from_env as _loop_cost_threshold
+from cloud.audit_loop_wiring import evaluate_release_gate as _release_gate_evaluate
+from cloud.audit_loop_wiring import process_audit_verdict as _loop_verdict
+from cloud.audit_loop_wiring import process_correction as _loop_correction
+from cloud.audit_loop_wiring import process_fix_completion as _loop_fix
+from cloud.audit_loop_wiring import process_rebuild_confirmation as _loop_confirm
+from cloud.audit_loop_wiring import state_to_dict as _loop_state_to_dict
+from cloud.audit_loop_wiring import load_loop_state as _loop_load_state
+from cloud.ledger import LedgerStore as _LedgerStore
 from cloud.merge_gate_hard import blocked_audit_detail as _gate_blocked_detail
 from cloud.merge_gate_hard import evaluate_hard_merge_gate as _merge_gate_hard
 from cloud.merge_gate_hard import hard_mode_from_env as _merge_gate_hard_mode
@@ -2781,6 +2791,22 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             self._pause_task(user_id, rest[: -len("/pause")])
         elif rest.endswith("/merge") and method == "POST":
             self._merge_task(user_id, rest[: -len("/merge")])
+        elif rest.endswith("/release") and method == "POST":
+            self._release_task(user_id, rest[: -len("/release")])
+        elif rest.endswith("/audit-verdict") and method == "POST":
+            self._audit_loop_verdict(user_id,
+                                     rest[: -len("/audit-verdict")])
+        elif rest.endswith("/audit-fix") and method == "POST":
+            self._audit_loop_fix(user_id, rest[: -len("/audit-fix")])
+        elif rest.endswith("/audit-correction") and method == "POST":
+            self._audit_loop_correction(
+                user_id, rest[: -len("/audit-correction")])
+        elif rest.endswith("/audit-rebuild-confirm") and method == "POST":
+            self._audit_loop_confirm(
+                user_id, rest[: -len("/audit-rebuild-confirm")])
+        elif rest.endswith("/audit-loop") and method == "GET":
+            self._audit_loop_get(user_id, rest[: -len("/audit-loop")],
+                                 query)
         elif rest.endswith("/pr") and method == "POST":
             self._task_pr(user_id, rest[: -len("/pr")])
         elif rest.endswith("/retry") and method == "POST":
@@ -8734,6 +8760,504 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         task["merge_gate"] = gate
         self._send_json(200, task)
 
+    # -- 项目级审计循环接线（#260 驱动的服务端执行层） --------------
+    # 决策一律来自 cloud/audit_loop_driver，本节只做执行：持久化
+    # 循环状态、经 #241/#247 写链、派生修复/重做子任务、拦合并与
+    # 发布。合并端点本身不改：它继续用 #247 reader + 五项硬门禁
+    # 做 AND，本节不给合并造第二套 reader。原任务状态在本节的
+    # 任何一步都不被回退：fail 只阻塞 + 派生子任务，completed
+    # 保持 completed，直到合并端点按门禁把它标 merged。
+    def _loop_task_for_body(self, task, body):
+        """把请求体里的模块/合同声明并入任务 dict，供接线层按
+        (module_id, contract_version) 解析循环键。任务行本身没
+        有 module 列，所以模块归属必须由提交方在每次调用里显
+        式声明（或沿用任务 id 回落）；接线层绝不悄悄猜模块。"""
+        merged = dict(task)
+        if isinstance(body.get("module_id"), str) \
+                and body.get("module_id").strip():
+            merged["module_id"] = body["module_id"].strip()
+        if isinstance(body.get("contract_version"), str) \
+                and body.get("contract_version").strip():
+            merged["contract_version"] = body["contract_version"].strip()
+        return merged
+
+    def _loop_quota_check(self, task, caller_id):
+        """服务端配额 -> 驱动 quota_check：配额归属取原任务
+        创建人（成本归属口径与子任务一致），无创建人时取调用
+        人。_quota_status 抛错时闭包同样抛错，驱动 fail-closed
+        为 quota_blocked，绝不把查不到配额当成放行。"""
+        quota_user = task.get("created_by") or caller_id
+
+        def _check(estimate):
+            quota, used, _month = self._quota_status(quota_user)
+            return _loop_quota_check(quota, used)(estimate)
+        return _check
+
+    def _loop_root_task_id(self, task):
+        """循环根任务 id：沿 forked_from 回溯到链头（最多 20
+        步防环）。链行必须归属根任务，否则 #247 的按任务读链
+        在修复子任务上写下的通过结论永远解不开父任务的合并
+        ——这是本节最容易踩错的归属点，单列一处 helper。"""
+        current = task
+        for _step in range(20):
+            parent_id = current.get("forked_from")
+            if not parent_id:
+                return current["id"]
+            parent_row = self._fetch_task(parent_id)
+            if parent_row is None:
+                return current["id"]
+            current = _task_to_dict(parent_row)
+        return current["id"]
+
+    def _loop_ledger_store(self):
+        """#241/#247 写路径的 store：LedgerStore 自带连接与
+        ensure_schema，每条 record_* 调用内提交。调用方用完
+        必须 close（各 handler 的 finally 里做）。"""
+        return _LedgerStore(self.server.db_path)
+
+    def _loop_create_subtask(self, descriptor, parent_task, caller_id):
+        """把接线层返回的子任务描述符落成一条 queued 任务行：
+        同项目、prompt 写明模块/合同/失败项与完整重审标准、
+        created_by 取原任务创建人（成本归属）、forked_from 记
+        父任务。返回新任务 dict；INSERT 失败时抛错，由调用方
+        按 503 回执（循环已推进、描述符在响应里，调用方可按
+        描述符经 POST /tasks 手工补建，绝不悄悄重跑决策）。"""
+        package = descriptor.get("pushback_package") or {}
+        failures = package.get("failures") or []
+        criteria = package.get("reaudit_criteria") or []
+        lines = [
+            "Audit %s subtask for module %s@%s (parent task %s)."
+            % (descriptor.get("kind"), descriptor.get("module_id"),
+               descriptor.get("contract_version"),
+               descriptor.get("parent_task_id")),
+            "Scope: %s." % descriptor.get("scope"),
+            "Failures: %s" % json.dumps(failures, ensure_ascii=False)[:2000],
+            "Full re-audit criteria (all must pass, not only the "
+            "failed items): %s" % ", ".join(str(c) for c in criteria),
+        ]
+        if descriptor.get("targeted_items"):
+            lines.append("Targeted items only: %s"
+                         % json.dumps(descriptor["targeted_items"],
+                                      ensure_ascii=False)[:1000])
+        if package.get("retained_assets"):
+            lines.append("Retained assets (do not rebuild these): %s"
+                         % json.dumps(package["retained_assets"],
+                                      ensure_ascii=False)[:2000])
+        prompt = "\n".join(lines)
+        subtask_id = uuid.uuid4().hex
+        now = time.time()
+        conn = self._db()
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " created_by, created_at, updated_at, forked_from)"
+                " VALUES(?, ?, ?, 'queued', ?, ?, ?, ?)",
+                (subtask_id, parent_task["project_id"], prompt,
+                 parent_task.get("created_by") or caller_id, now, now,
+                 parent_task["id"]))
+            conn.commit()
+            row = conn.execute(
+                "SELECT * FROM tasks WHERE id = ?", (subtask_id,)
+            ).fetchone()
+            return _task_to_dict(row)
+        finally:
+            conn.close()
+
+    def _loop_run(self, user_id, task_id, body, kind):
+        """循环端点的公共骨架：404 / approve_merge 403 / 体校验 /
+        接线调用 / 子任务落行。ValueError/TypeError（驱动拒绝、
+        映射缺失、写链预校验）回 400；其它异常（账本写失败等）
+        回 503 且状态未推进（接线层先写链后存状态的顺序保证
+        了这一点）；子任务落行失败同样 503，但循环已推进的事
+        实在响应里明说，不谎称“未推进”。"""
+        row = self._fetch_task(task_id)
+        if row is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        task = _task_to_dict(row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, task["project_id"], "approve_merge"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        loop_task = self._loop_task_for_body(task, body)
+        store = self._loop_ledger_store()
+        loop_conn = self._db()
+        try:
+            if kind == "verdict":
+                verdict = body.get("verdict")
+                if not isinstance(verdict, dict) \
+                        or not verdict.get("verdict"):
+                    self._send_json(400, {
+                        "error": "verdict dict with a verdict "
+                                 "field is required"})
+                    return
+                result = _loop_verdict(
+                    loop_conn, store, loop_task, verdict,
+                    quota_check=self._loop_quota_check(task, user_id),
+                    cost_threshold=_loop_cost_threshold(),
+                    old_impl_ref=body.get("old_impl_ref"),
+                    user_id=user_id,
+                    ledger_task_id=self._loop_root_task_id(task))
+            elif kind == "fix":
+                fix_result = body.get("fix_result")
+                if not isinstance(fix_result, dict):
+                    self._send_json(400, {
+                        "error": "fix_result dict is required"})
+                    return
+                result = _loop_fix(
+                    loop_conn, store, loop_task, fix_result,
+                    automated=body.get("automated"),
+                    quota_check=self._loop_quota_check(task, user_id),
+                    cost_threshold=_loop_cost_threshold(),
+                    old_impl_ref=body.get("old_impl_ref"),
+                    user_id=user_id,
+                    ledger_task_id=self._loop_root_task_id(task))
+            elif kind == "correction":
+                correction = body.get("correction")
+                if not isinstance(correction, dict):
+                    self._send_json(400, {
+                        "error": "correction dict is required"})
+                    return
+                if isinstance(body.get("contract_version"), str):
+                    correction = dict(
+                        correction,
+                        contract_version=body["contract_version"])
+                result = _loop_correction(
+                    loop_conn, store, loop_task, correction,
+                    user_id=user_id,
+                    ledger_task_id=self._loop_root_task_id(task))
+            else:
+                raise ValueError("unknown loop kind: %r" % (kind,))
+        except (ValueError, TypeError) as exc:
+            self._send_json(400, {"error": str(exc)})
+            return
+        except Exception as exc:
+            self._send_json(503, {
+                "error": "audit_loop_unavailable",
+                "detail": "ledger or loop state write failed; "
+                          "loop state was not advanced: %s" % exc})
+            return
+        finally:
+            try:
+                store.close()
+            except Exception:
+                pass
+            try:
+                loop_conn.close()
+            except Exception:
+                pass
+        subtask_task = None
+        subtask_error = None
+        descriptor = result.get("subtask")
+        if descriptor is not None and descriptor.get("kind") != "reaudit":
+            try:
+                subtask_task = self._loop_create_subtask(
+                    descriptor, task, user_id)
+            except Exception as exc:
+                subtask_error = str(exc)
+        payload = {
+            "task": task,
+            "loop_state": _loop_state_to_dict(result["next_state"]),
+            "action_kind": result["action_kind"],
+            "reason": result["reason"],
+            "merge_blocked": result["merge_blocked"],
+            "release_blocked": result["release_blocked"],
+            "requires_reaudit": result["requires_reaudit"],
+            "escalation_grants_user_acceptance":
+                result["escalation_grants_user_acceptance"],
+            "subtask": descriptor,
+            "subtask_task": subtask_task,
+            "notification": result["notification"],
+        }
+        if "automated_checks" in result:
+            payload["automated_checks"] = result["automated_checks"]
+        if "correction_action" in result:
+            payload["correction_action"] = result["correction_action"]
+        if subtask_error is not None:
+            payload["subtask_error"] = subtask_error
+            payload["subtask_note"] = (
+                "loop state advanced and ledger written, but the "
+                "subtask row was not created; create it from the "
+                "subtask descriptor via POST /tasks, do not "
+                "re-submit the verdict")
+            self._send_json(503, payload)
+            return
+        self._send_json(200, payload)
+
+    def _audit_loop_verdict(self, user_id, task_id):
+        """POST /tasks/{id}/audit-verdict：提交一次独立审计或
+        验收结论。权限：项目 approve_merge（审计提交方须是
+        maintainer 以上，与合并权同级；普通 contributor 不能
+        凭空给循环喂结论）。体：{module_id, contract_version,
+        verdict: {verdict, source, failures, ...},
+        old_impl_ref?}。原任务状态不动；fail 自动派生修复子
+        任务并经链阻塞合并/发布。"""
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        self._loop_run(user_id, task_id, body, "verdict")
+
+    def _audit_loop_fix(self, user_id, task_id):
+        """POST /tasks/{id}/audit-fix：修复完成事件。体：
+        {module_id, contract_version, fix_result:
+        {reaudit_verdict: {...}}, automated?: {hard_gate: {...},
+        extra_checks: {...}}, old_impl_ref?}。注意“自动完整
+        重审”的准确含义：本端点把修复完成推进到重审（跑自动
+        化检查 + 要求完整重审结论），而不是自动判通过——
+        reaudit_verdict 仍须审计方提交，缺失或 diff_only 由
+        驱动以 400 拒绝；自动化检查全过也绝不等于审计通过。
+        本端点调用的任务应为修复子任务（其 module 与父任务
+        相同，体里显式声明），原任务与修复子任务的状态本端
+        点都不改（修复子任务的 completed 由 worker/调用方写
+        回，本端点只推进循环）。"""
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        self._loop_run(user_id, task_id, body, "fix")
+
+    def _audit_loop_correction(self, user_id, task_id):
+        """POST /tasks/{id}/audit-correction：审计的审计纠错。
+        体：{module_id, contract_version, correction:
+        {original_verdict, corrected_verdict, reviewer,
+        original_entry_id, ...}}。纠正为 fail 立即重拦；纠正
+        为 pass 不解封、只安排重审（原链行永不改动，只追加
+        meta-review 行）。"""
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        self._loop_run(user_id, task_id, body, "correction")
+
+    def _audit_loop_confirm(self, user_id, task_id):
+        """POST /tasks/{id}/audit-rebuild-confirm：验收触发推
+        倒的用户确认。体必须显式带 user_confirmed（true/false）
+        与 confirmed_by，且 confirmed_by 必须等于认证调用方
+        本人：协调员或系统没有任何路径可以代用户确认（本端
+        点之外没有任何代码调用确认流程；pending 状态下不建
+        重做任务、不耗推倒次数）。确认后才进 rebuilding，拒
+        绝则 escalate 且不耗预算；确认不等于 G6 验收。"""
+        row = self._fetch_task(task_id)
+        if row is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        task = _task_to_dict(row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, task["project_id"], "approve_merge"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        if "user_confirmed" not in body \
+                or not isinstance(body.get("user_confirmed"), bool):
+            self._send_json(400, {
+                "error": "user_confirmed boolean is required; "
+                         "confirmation must be explicit"})
+            return
+        if body.get("confirmed_by") != user_id:
+            self._send_json(403, {
+                "error": "confirmed_by must equal the authenticated "
+                         "caller; no coordinator or system may "
+                         "confirm on a user's behalf"})
+            return
+        loop_task = self._loop_task_for_body(task, body)
+        store = self._loop_ledger_store()
+        loop_conn = self._db()
+        try:
+            result = _loop_confirm(
+                loop_conn, store, loop_task,
+                user_confirmed=body["user_confirmed"],
+                confirmed_by=body["confirmed_by"],
+                quota_check=self._loop_quota_check(task, user_id),
+                cost_threshold=_loop_cost_threshold(),
+                cost_estimate=body.get("cost_estimate"),
+                old_impl_ref=body.get("old_impl_ref"),
+                user_id=user_id,
+                ledger_task_id=self._loop_root_task_id(task))
+        except (ValueError, TypeError) as exc:
+            self._send_json(400, {"error": str(exc)})
+            return
+        except Exception as exc:
+            self._send_json(503, {
+                "error": "audit_loop_unavailable",
+                "detail": "ledger or loop state write failed; "
+                          "loop state was not advanced: %s" % exc})
+            return
+        finally:
+            try:
+                store.close()
+            except Exception:
+                pass
+            try:
+                loop_conn.close()
+            except Exception:
+                pass
+        subtask_task = None
+        subtask_error = None
+        descriptor = result.get("subtask")
+        if descriptor is not None:
+            try:
+                subtask_task = self._loop_create_subtask(
+                    descriptor, task, user_id)
+            except Exception as exc:
+                subtask_error = str(exc)
+        payload = {
+            "task": task,
+            "loop_state": _loop_state_to_dict(result["next_state"]),
+            "action_kind": result["action_kind"],
+            "reason": result["reason"],
+            "merge_blocked": result["merge_blocked"],
+            "release_blocked": result["release_blocked"],
+            "subtask": descriptor,
+            "subtask_task": subtask_task,
+            "notification": result["notification"],
+        }
+        if subtask_error is not None:
+            payload["subtask_error"] = subtask_error
+            self._send_json(503, payload)
+            return
+        self._send_json(200, payload)
+
+    def _audit_loop_get(self, user_id, task_id, query):
+        """GET /tasks/{id}/audit-loop?module_id=&contract_version=：
+        读循环持久化状态 + 同一 #247 reader 的链阻塞视图。
+        权限：项目 read。读链失败（表不存在除外）回 503、不
+        谎称无阻塞。"""
+        row = self._fetch_task(task_id)
+        if row is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        task = _task_to_dict(row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, task["project_id"], "read"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        module_id = (query.get("module_id") or [None])[0]
+        contract_version = (query.get("contract_version") or [None])[0]
+        if not module_id or not contract_version:
+            self._send_json(400, {
+                "error": "module_id and contract_version query "
+                         "parameters are required"})
+            return
+        conn = self._db()
+        try:
+            state = _loop_load_state(conn, module_id, contract_version)
+        finally:
+            conn.close()
+        if state is None:
+            self._send_json(404, {"error": "no audit loop state"})
+            return
+        try:
+            conn = self._db()
+            try:
+                pushback = _pushback_block(conn, task_id)
+            finally:
+                conn.close()
+        except Exception:
+            self._send_json(503, {
+                "error": "audit_pushback_state_unavailable"})
+            return
+        self._send_json(200, {
+            "task": task,
+            "loop_state": _loop_state_to_dict(state),
+            "audit_pushback": pushback,
+        })
+
+    def _release_task(self, user_id, task_id):
+        """POST /tasks/{id}/release：发布门禁 + 发布授权（#259
+        契约正式建路由）。权限：只认 CODEX_GLOBAL_ADMINS 的
+        全局管理员写（与 verify/export 同身份来源，但 verify/
+        export 是只读、发布是写；项目 approve_merge 不能放行
+        发布）。三者 AND：① 发布时刻重读 #247 同一 reader 的
+        链；② 发布时刻调用方新提交的硬门禁证据重算五项（合
+        并时的 200 永远不作输入；硬门禁一律按 hard 算，模式环
+        境变量不能把发布降级）；③ 任务状态须为 merged。
+        阻塞源区分见 release_gate：blocked_sources 列出全部，
+        code/blocked_by 取求值顺序第一项。409 阻塞不写任何审
+        计（检查写链会移动链头、使锚点失效）；403 被拒沿用
+        _audit_admin 惯例记审计。200 不改任务状态（本站没有
+        released 状态）：200 是“发布已授权”，真正的部署动作
+        在平台部署通道（deploy.py）里，它部署平台本身、无任
+        务 id、不走服务端 HTTP 鉴权，因此不在任务门禁内——
+        把它绑到任务链上会在最需要部署修复与回滚时反锁部署。
+        200 时写一条显式的 release_task 审计作为发布后动作记
+        录（检查本身是只读的，这条是检查完成后的显式动作，
+        不是检查的副产品；锚点须在这条之后重取）。"""
+        row = self._fetch_task(task_id)
+        if row is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        task = _task_to_dict(row)
+        if not self._is_global_admin(user_id):
+            self._audit_admin(user_id, "release_denied", task_id)
+            self._send_json(403, {"error": "forbidden"})
+            return
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        if body.get("files_changed") is not None and not isinstance(
+                body.get("files_changed"), list):
+            self._send_json(400, {"error": "files_changed must be a list"})
+            return
+        hard_err = _validate_hard_gate_body(body)
+        if hard_err:
+            self._send_json(400, {"error": hard_err})
+            return
+        # 发布时刻的新证据：体里没给任何硬门禁字段时，hard_gate
+        # 为 None -> 证据缺失 409，而不是用合并时的旧结论补。
+        has_evidence = any(
+            body.get(field) is not None
+            for field in ("files_changed", "ci_status", "ci_checks",
+                          "required_checks", "contract_version",
+                          "contract_frozen", "coordinator_approved",
+                          "unresolved_review_comments",
+                          "branch_protection"))
+        hard_gate = None
+        if has_evidence:
+            protection = body.get("branch_protection") \
+                or _merge_gate_protection()
+            hard_gate = _merge_gate_hard(
+                files_changed=body.get("files_changed"),
+                owned_dir=body.get("owned_dir"),
+                approved_exceptions=body.get("approved_exceptions") or (),
+                protection=protection,
+                ci_status=body.get("ci_status"),
+                ci_checks=body.get("ci_checks"),
+                required_checks=body.get("required_checks"),
+                contract_version=body.get("contract_version"),
+                contract_frozen=body.get("contract_frozen", False),
+                contract_expected_version=body.get(
+                    "contract_expected_version"),
+                coordinator_approved=body.get("coordinator_approved"),
+                unresolved_review_comments=body.get(
+                    "unresolved_review_comments"))
+        conn = self._db()
+        try:
+            decision = _release_gate_evaluate(
+                task_id, conn, task_status=task["status"],
+                hard_gate=hard_gate)
+        finally:
+            conn.close()
+        if not decision["allowed"]:
+            # 409 不写审计：见本方法 docstring。
+            self._send_json(409, {
+                "error": "release blocked",
+                "release_gate": decision,
+            })
+            return
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, task["project_id"], "release_task",
+                        task_id)
+        task["release_gate"] = decision
+        self._send_json(200, task)
+
     # -- 运行时审批队列 ------------------------------------------------
     def _list_approvals(self, user_id, task_id):
         """列出某任务的审批项（含已决定的，回执一并带出）。