fix: bounded audit loop (deterministic gates only, 3 pushbacks, escalate)

Audit loop engine (cloud/audit_loop.py): Finding/PushbackPacket/LoopState pure functions; deterministic gates only, subjective -> needs_review; packet completeness enforced; standards-weakening -> escalate; regression and missing-gate checks enforce full re-run; MAX_PUSHBACKS=3, round>=3 -> escalate. Persistence in new single-row task_audit_loop_state (attempts DDL pattern: module CREATE_TABLE_SQL + db.py POSTGRES_SCHEMA + migrations list + server init + worker ensure); task_attempts NOT reused (attempt_no PK collision) and task_followups NOT reused (user messages). Worker.audit_loop_evaluate is the explicit opt-in hook (task audit_loop field only, no env switch, never called by run_one); escalate audits worker_audit_loop_escalated; corrupted/missing count fail-closes to escalate.

cloud/audit_loop.py

@@ -0,0 +1,353 @@
+"""有界自动审计循环（audit loop），只对确定性门做自动循环。
+
+设计前提（先核实再动手，见模块末尾的“复用清单”）：
+生产代码里没有“审计员 agent”，所以这里的“审”只能是确定性
+检查的结果——测试/校验命令、能力门、干跑阻塞、迁移列、格式
+等可客观判定对错的门。主观审计（架构、设计、代码美观）绝不
+触发自动循环，只返回 needs_review，走人工。
+
+有界方案（用户已定，一字不差地落地）：
+1. 只对确定性门循环：Finding.kind 只能是 deterministic 或
+   subjective；subjective 一律 needs_review，不推回、不计轮。
+2. 推回包写死：每条 Finding 必须带 id、evidence、reproduction、
+   acceptance，四者缺一即 ValueError（调用方 bug，不进循环）；
+   PushbackPacket 带 round、findings、criteria（本轮结束后仍未
+   解决的验收标准，finding_id -> gate_or_None）。
+3. 上限 3 轮：round = 已发出的推回数；round < 3 且有 finding 才
+   推回，round >= 3 且仍有 finding 即升级（escalate），不再循环。
+4. 重审审全量：每轮必须重跑全部门；上一轮通过的门本轮未跑
+   即 gate_missing 升级；上一轮通过的门本轮变失败即 regression
+   finding，同样不得放行。
+5. 标准不得弱化：上一轮的 finding 只有在三种情况下才算解决——
+   本轮仍被报告（仍在解决中）、调用方以 fixed 传入已修复证据、
+   其关联 gate 本轮通过；否则该 finding 被丢掉即 standards_weakened
+   升级，绝不默许把标准越放越宽。
+6. 默认关闭：引擎只由 Worker.audit_loop_evaluate 显式调用，且仅当
+   任务字典含 audit_loop 真值时才启用；没有 env 开关——全局 env
+   会让所有任务悄悄启用循环，这是本设计刻意拒绝的。开关关闭时
+   run_one、auto_resume、协调员、goal 断点行为逐字节不变（本模块
+   不改 run_one，见 Worker 方法的文档）。
+7. 轮次持久化：状态存在 task_audit_loop_state 单行表（task_id 主键），
+   跨 worker 重启不丢；计数缺失或损坏时 fail-closed 到升级，绝不
+   默认循环。审计行（audit_log）是 best-effort，写失败会丢计数，
+   所以轮次绝不以审计行计数为准——这是不用审计行做持久化的原因。
+
+复用清单（核实结论）：
+- goal 的 verify_command：goal 任务已有的确定性校验门，本引擎
+  不重复造校验，只消费“门的结果”（gate_results）。
+- task_attempts 尝试账本：本引擎不复用它做轮次持久化，原因
+  是它的主键是 (task_id, attempt_no)，与 goal/platform 尝试共用
+  attempt_no 序号——audit_loop 若也用 attempt_no 记轮次，序号会
+  与已有尝试撞主键、被 INSERT OR IGNORE 静默丢掉，轮次就丢了。
+  因此按 attempts 的 DDL 手法（模块内 CREATE_TABLE_SQL +
+  ensure + db.py POSTGRES_SCHEMA + migrations 表清单 + server
+  初始化调用）另起单行表，序号冲突问题不存在。
+- task_followups：特意不复用——它是用户消息（用户含任务间互发），
+  把机器推回包写进 followups 会把机器包暴露给用户、混淆用户消息。
+  推回包由调用方持有（返回的 packet dict），不落 followups。
+- 协调员重排（UPDATE tasks SET status='queued' 带状态 CAS）：
+  调用方若要重排，用同一手法；本引擎本身不改任务状态、不重排、
+  不调 runner——状态变更属于调用方，引擎只给决策。
+- worker 的 fail-fast 门（能力门、工具门等）：本引擎不替代它们，
+  只在它们已经判定出的结果之上做“循环还是升级”的决策。
+"""
+
+import json
+import time
+
+MAX_PUSHBACKS = 3
+SEVERITIES = ("low", "medium", "high", "critical")
+KINDS = ("deterministic", "subjective")
+
+CREATE_TABLE_SQL = (
+    "CREATE TABLE IF NOT EXISTS task_audit_loop_state("
+    "task_id TEXT PRIMARY KEY, "
+    "round INTEGER, "
+    "passed_gates TEXT, "
+    "criteria TEXT, "
+    "last_decision TEXT, "
+    "updated_at DOUBLE PRECISION)")
+
+
+def ensure_audit_loop_table(conn):
+    """幂等建审计循环状态表（与 attempts.ensure_attempts_table 同款）。"""
+    conn.execute(CREATE_TABLE_SQL)
+
+
+def _require_text(value, name):
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError("%s must be a non-empty string" % name)
+    return value
+
+
+class Finding:
+    """一条确定性（或主观）发现；四要素缺一即调用方 bug。"""
+
+    def __init__(self, id, severity, evidence, reproduction, acceptance,
+                 kind="deterministic", gate=None):
+        self.id = _require_text(id, "finding id")
+        if severity not in SEVERITIES:
+            raise ValueError("severity must be one of %s" % (SEVERITIES,))
+        if kind not in KINDS:
+            raise ValueError("kind must be one of %s" % (KINDS,))
+        self.severity = severity
+        self.evidence = _require_text(evidence, "evidence")
+        self.reproduction = _require_text(reproduction, "reproduction")
+        self.acceptance = _require_text(acceptance, "acceptance")
+        self.kind = kind
+        # gate：该发现来自哪个确定性门；None 表示无关联门，
+        # 此时“门通过即解决”不适用，只能靠重报或 fixed 解决。
+        self.gate = gate
+
+    def to_dict(self):
+        return {
+            "id": self.id, "severity": self.severity,
+            "evidence": self.evidence, "reproduction": self.reproduction,
+            "acceptance": self.acceptance, "kind": self.kind,
+            "gate": self.gate,
+        }
+
+    @classmethod
+    def from_dict(cls, raw):
+        if not isinstance(raw, dict):
+            raise ValueError("finding must be a dict")
+        return cls(
+            raw.get("id"), raw.get("severity"), raw.get("evidence"),
+            raw.get("reproduction"), raw.get("acceptance"),
+            kind=raw.get("kind", "deterministic"),
+            gate=raw.get("gate"))
+
+
+class PushbackPacket:
+    """一轮推回包：round、findings、criteria 缺一不可。"""
+
+    def __init__(self, task_id, round, findings, criteria):
+        self.task_id = task_id
+        self.round = round
+        self.findings = list(findings)
+        if not self.findings:
+            raise ValueError("pushback packet requires at least one finding")
+        for finding in self.findings:
+            if not isinstance(finding, Finding):
+                raise ValueError("packet findings must be Finding objects")
+        # criteria：本轮结束后仍未解决的验收标准，
+        # {finding_id: gate_or_None}，下一轮不得弱化它。
+        self.criteria = dict(criteria)
+        if not self.criteria:
+            raise ValueError("pushback packet requires non-empty criteria")
+
+    def to_dict(self):
+        return {
+            "task_id": self.task_id, "round": self.round,
+            "findings": [f.to_dict() for f in self.findings],
+            "criteria": self.criteria,
+        }
+
+
+class LoopState:
+    """一任务的循环状态；round = 已发出的推回数。"""
+
+    def __init__(self, task_id, round=0, passed_gates=None, criteria=None,
+                 last_decision=None):
+        self.task_id = task_id
+        if not isinstance(round, int) or isinstance(round, bool) or round < 0:
+            raise ValueError("round must be a non-negative int")
+        self.round = round
+        self.passed_gates = sorted(passed_gates or [])
+        if not all(isinstance(g, str) for g in self.passed_gates):
+            raise ValueError("passed_gates must be strings")
+        self.criteria = dict(criteria or {})
+        if not all(isinstance(k, str) for k in self.criteria):
+            raise ValueError("criteria keys must be strings")
+        self.last_decision = last_decision
+
+    def to_dict(self):
+        return {
+            "task_id": self.task_id, "round": self.round,
+            "passed_gates": self.passed_gates, "criteria": self.criteria,
+            "last_decision": self.last_decision,
+        }
+
+    @classmethod
+    def from_dict(cls, raw):
+        if not isinstance(raw, dict):
+            raise ValueError("loop state must be a dict")
+        return cls(
+            raw.get("task_id"), round=raw.get("round", 0),
+            passed_gates=raw.get("passed_gates"),
+            criteria=raw.get("criteria"),
+            last_decision=raw.get("last_decision"))
+
+
+def load_loop_state(conn, task_id):
+    """读状态；返回 (state_or_None, corrupted_bool, error_or_None)。
+
+    表不存在或无行时返回 (None, False, None)；行 JSON 损坏或校验
+    失败时返回 (None, True, err)——调用方必须 fail-closed 到升级。
+    """
+    try:
+        row = conn.execute(
+            "SELECT round, passed_gates, criteria, last_decision"
+            " FROM task_audit_loop_state WHERE task_id = ?",
+            (task_id,)).fetchone()
+    except Exception as exc:
+        return None, True, "audit loop state unreadable: %s" % exc
+    if row is None:
+        return None, False, None
+    try:
+        state = LoopState(
+            task_id, round=row[0],
+            passed_gates=json.loads(row[1]) if row[1] else [],
+            criteria=json.loads(row[2]) if row[2] else {},
+            last_decision=row[3])
+    except (ValueError, TypeError, json.JSONDecodeError) as exc:
+        return None, True, "audit loop state corrupted: %s" % exc
+    return state, False, None
+
+
+def save_loop_state(conn, state):
+    """写回状态（单行，INSERT OR REPLACE）；调用方负责提交。"""
+    conn.execute(
+        "INSERT OR REPLACE INTO task_audit_loop_state"
+        "(task_id, round, passed_gates, criteria, last_decision,"
+        " updated_at) VALUES(?, ?, ?, ?, ?, ?)",
+        (state.task_id, state.round,
+         json.dumps(state.passed_gates), json.dumps(state.criteria),
+         state.last_decision, time.time()))
+
+
+def _validate_gate_results(gate_results):
+    if not isinstance(gate_results, dict):
+        raise ValueError("gate_results must be a dict")
+    for name, value in gate_results.items():
+        if not isinstance(name, str) or not isinstance(value, bool):
+            raise ValueError("gate_results must map gate name to bool")
+
+
+def evaluate(state, task_id, gate_results, findings=None, fixed=None):
+    """纯函数：一轮审计的决策。
+
+    参数：state 为上一轮 LoopState 或 None（首轮）；gate_results 为
+    本轮全部门的 {门名: 是否通过}，必须含全部已跑过的门；findings
+    为本轮确定性/主观发现；fixed 为 {finding_id: 已修复证据}，调用方
+    以此证明上一轮 finding 已修复（无证据不得丢标准）。
+    返回决策 dict：decision 为 pass / pushback / escalate /
+    needs_review 之一；packet 只在 pushback 时有值；new_state 只在
+    pass / pushback / escalate 时有值（needs_review 不写状态）。
+    """
+    findings = list(findings or [])
+    fixed = dict(fixed or {})
+    _validate_gate_results(gate_results)
+    if not gate_results and not findings:
+        raise ValueError(
+            "at least one deterministic gate or finding is required")
+    for finding in findings:
+        if not isinstance(finding, Finding):
+            raise ValueError("findings must be Finding objects")
+
+    # 主观发现绝不触发自动循环：只标记 needs_review，不计轮、不写状态。
+    subjective = [f for f in findings if f.kind != "deterministic"]
+    if subjective:
+        return {
+            "decision": "needs_review", "round": state.round if state else 0,
+            "packet": None, "new_state": None, "reason": "subjective",
+            "subjective": [f.to_dict() for f in subjective],
+            "regression": [], "missing_gates": [], "weakened": [],
+        }
+
+    prev_round = state.round if state else 0
+    prev_passed = set(state.passed_gates) if state else set()
+    prev_criteria = dict(state.criteria) if state else {}
+
+    # 全量回归：上一轮通过的门本轮必须重跑，缺跑即升级。
+    missing_gates = sorted(g for g in prev_passed if g not in gate_results)
+    if missing_gates:
+        new_state = LoopState(
+            task_id, round=prev_round,
+            passed_gates=sorted(
+                g for g, v in gate_results.items() if v),
+            criteria=prev_criteria, last_decision="escalate")
+        return {
+            "decision": "escalate", "round": prev_round, "packet": None,
+            "new_state": new_state, "reason": "gate_missing",
+            "regression": [], "missing_gates": missing_gates,
+            "weakened": [],
+        }
+
+    passed_now = {g for g, v in gate_results.items() if v}
+    failed_now = {g for g, v in gate_results.items() if not v}
+
+    # 回归：上一轮通过的门本轮变失败，自动生成一条 finding。
+    regression = []
+    for gate in sorted(prev_passed & failed_now):
+        regression.append(Finding(
+            "regression:%s" % gate, "high",
+            "previously passed gate %s now fails" % gate,
+            "re-run gate %s" % gate,
+            "gate %s passes" % gate,
+            kind="deterministic", gate=gate))
+    all_findings = list(findings) + regression
+
+    # 标准弱化：上一轮 finding 只有三种解决方式——本轮仍被报告、
+    # 调用方以 fixed 给出已修复证据、其关联 gate 本轮通过；否则丢标准。
+    reported_ids = {f.id for f in all_findings}
+    weakened = []
+    new_criteria = {}
+    for fid, gate in prev_criteria.items():
+        if fid in reported_ids:
+            new_criteria[fid] = gate
+        elif fid in fixed:
+            continue  # 已修复证据，解决，不弱化
+        elif gate is not None and gate in passed_now:
+            continue  # 关联门通过，解决，不弱化
+        else:
+            weakened.append(fid)
+    if weakened:
+        new_state = LoopState(
+            task_id, round=prev_round, passed_gates=sorted(passed_now),
+            criteria=prev_criteria, last_decision="escalate")
+        return {
+            "decision": "escalate", "round": prev_round, "packet": None,
+            "new_state": new_state, "reason": "standards_weakened",
+            "regression": [f.to_dict() for f in regression],
+            "missing_gates": [], "weakened": sorted(weakened),
+        }
+
+    # 新标准 = 未解决的旧标准 + 本轮新 finding（含回归）。
+    for finding in all_findings:
+        new_criteria[finding.id] = finding.gate
+
+    if not all_findings and not failed_now:
+        new_state = LoopState(
+            task_id, round=prev_round, passed_gates=sorted(passed_now),
+            criteria=new_criteria, last_decision="pass")
+        return {
+            "decision": "pass", "round": prev_round, "packet": None,
+            "new_state": new_state, "reason": None,
+            "regression": [], "missing_gates": [], "weakened": [],
+        }
+
+    # 有 finding 但门全过、finding 为空的矛盾态：门说通过、发现说有
+    # 问题，以 finding 为准（finding 是审计的直接证据）。
+    if prev_round >= MAX_PUSHBACKS:
+        new_state = LoopState(
+            task_id, round=prev_round, passed_gates=sorted(passed_now),
+            criteria=new_criteria, last_decision="escalate")
+        return {
+            "decision": "escalate", "round": prev_round, "packet": None,
+            "new_state": new_state, "reason": "max_rounds_exceeded",
+            "regression": [f.to_dict() for f in regression],
+            "missing_gates": [], "weakened": [],
+        }
+
+    new_round = prev_round + 1
+    packet = PushbackPacket(task_id, new_round, all_findings, new_criteria)
+    new_state = LoopState(
+        task_id, round=new_round, passed_gates=sorted(passed_now),
+        criteria=new_criteria, last_decision="pushback")
+    return {
+        "decision": "pushback", "round": new_round, "packet": packet,
+        "new_state": new_state, "reason": None,
+        "regression": [f.to_dict() for f in regression],
+        "missing_gates": [], "weakened": [],
+    }

cloud/db.py

@@ -843,6 +843,15 @@ POSTGRES_SCHEMA = (
     "error_text TEXT, "
     "created_at DOUBLE PRECISION, "
     "PRIMARY KEY(task_id, attempt_no))",
+    # 审计循环状态（audit loop）：task_id 单行主键；PG 新库一次
+    # 建齐，老库走 server/worker ensure（见 cloud/audit_loop.py）。
+    "CREATE TABLE IF NOT EXISTS task_audit_loop_state("
+    "task_id TEXT PRIMARY KEY, "
+    "round INTEGER, "
+    "passed_gates TEXT, "
+    "criteria TEXT, "
+    "last_decision TEXT, "
+    "updated_at DOUBLE PRECISION)",
     "CREATE TABLE IF NOT EXISTS projects("
     "id TEXT PRIMARY KEY, "
     "name TEXT, "

cloud/migrations.py

@@ -89,6 +89,8 @@ MIGRATION_TABLES = (
     "resident_polls",
     # 尝试账本（v6）：同款复合主键、无身份列。
     "task_attempts",
+    # 审计循环状态（audit loop）：task_id 单行主键、无身份列。
+    "task_audit_loop_state",
 )
 
 _SKIP_TABLES = frozenset({"kb_fts", "sqlite_sequence"})

cloud/server.py

@@ -64,6 +64,7 @@ from cloud.agent_def import parse_agent_tool_approvals_column as _parse_tool_app
 from cloud.agent_def import parse_agent_tools_column as _parse_tools_col
 from cloud.agent_def import resolve_registry_agent as _resolve_agent
 from cloud.attempts import ensure_attempts_table as _ensure_attempts
+from cloud.audit_loop import ensure_audit_loop_table as _ensure_audit_loop
 from cloud.auth import ROLES as _MEMBER_ROLES
 from cloud.auth import AuthStore
 from cloud.coordinator import parse_parts as _coord_parse_parts
@@ -1725,6 +1726,7 @@ def init_db(db_path):
             _ensure_task_checkpoints_table(conn)
             _ensure_task_steps_table(conn)
             _ensure_attempts(conn)
+            _ensure_audit_loop(conn)
             conn.commit()
             return
         conn.execute(
@@ -1762,6 +1764,7 @@ def init_db(db_path):
         _ensure_task_checkpoints_table(conn)
         _ensure_task_steps_table(conn)
         _ensure_attempts(conn)
+        _ensure_audit_loop(conn)
         ensure_kill_requests_table(conn)
         # 常驻监听表（P0 S3 的 SQLite 侧）：启动时建一次，
         # 热路径（claim/run_due/批量）不再逐条 ensure，

cloud/tests/test_audit_loop.py

@@ -0,0 +1,209 @@
+"""有界自动审计循环测试：纯状态机 + worker 显式入口。
+
+只测本轨：确定性门循环、推回包完整性、标准不弱化、上限升级、
+全量回归、主观不循环、计数损坏 fail-closed、默认关闭。
+"""
+
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from cloud import audit_loop
+from cloud.audit_loop import Finding, LoopState, evaluate
+from cloud.worker import Worker
+
+
+def det(fid, gate=None):
+    return Finding(fid, "high", "evidence for %s" % fid,
+                   "re-run %s" % (gate or fid),
+                   "%s passes" % (gate or fid), gate=gate)
+
+
+class PureEngineTests(unittest.TestCase):
+    def test_pass_all_gates(self):
+        out = evaluate(None, "t1", {"tests": True, "capability": True})
+        self.assertEqual(out["decision"], "pass")
+        self.assertEqual(out["round"], 0)
+        self.assertIsNone(out["packet"])
+
+    def test_pushback_packet_complete(self):
+        out = evaluate(None, "t1", {"tests": False}, [det("f1", "tests")])
+        self.assertEqual(out["decision"], "pushback")
+        self.assertEqual(out["round"], 1)
+        packet = out["packet"]
+        self.assertEqual(packet.findings[0].evidence, "evidence for f1")
+        self.assertEqual(packet.findings[0].reproduction, "re-run tests")
+        self.assertEqual(packet.findings[0].acceptance, "tests passes")
+        self.assertIn("f1", packet.criteria)
+
+    def test_invalid_finding_is_caller_bug(self):
+        with self.assertRaises(ValueError):
+            Finding("f1", "high", "", "repro", "accept")
+        with self.assertRaises(ValueError):
+            evaluate(None, "t1", {}, [])
+
+    def test_escalate_at_third_round(self):
+        state = LoopState("t1", round=3, passed_gates=[],
+                          criteria={"f1": "tests"})
+        out = evaluate(state, "t1", {"tests": False}, [det("f1", "tests")])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertEqual(out["reason"], "max_rounds_exceeded")
+        self.assertEqual(out["round"], 3)
+
+    def test_weakening_is_escalate(self):
+        # 上一轮 finding f1（无关联门）本轮既不重报、又无 fixed 证据。
+        state = LoopState("t1", round=1, passed_gates=[],
+                          criteria={"f1": None})
+        out = evaluate(state, "t1", {"other": True}, [])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertEqual(out["reason"], "standards_weakened")
+        self.assertIn("f1", out["weakened"])
+
+    def test_fixed_via_gate_pass_is_not_weakening(self):
+        state = LoopState("t1", round=1, passed_gates=[],
+                          criteria={"f1": "tests"})
+        out = evaluate(state, "t1", {"tests": True}, [])
+        self.assertEqual(out["decision"], "pass")
+
+    def test_regression_blocks_pass(self):
+        state = LoopState("t1", round=1, passed_gates=["tests"],
+                          criteria={})
+        out = evaluate(state, "t1", {"tests": False}, [])
+        self.assertEqual(out["decision"], "pushback")
+        self.assertTrue(any(f.id == "regression:tests"
+                            for f in out["packet"].findings))
+
+    def test_missing_gate_escalates(self):
+        state = LoopState("t1", round=1, passed_gates=["tests"],
+                          criteria={})
+        out = evaluate(state, "t1", {"other": True}, [])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertEqual(out["reason"], "gate_missing")
+        self.assertIn("tests", out["missing_gates"])
+
+    def test_subjective_never_loops(self):
+        sub = Finding("design", "medium", "looks off", "read code",
+                      "reviewer approves", kind="subjective")
+        out = evaluate(None, "t1", {"tests": True}, [sub])
+        self.assertEqual(out["decision"], "needs_review")
+        self.assertIsNone(out["new_state"])
+        self.assertEqual(out["round"], 0)
+
+    def test_corrupted_state_rejected(self):
+        with self.assertRaises(ValueError):
+            LoopState.from_dict({"task_id": "t1", "round": -1})
+        with self.assertRaises(ValueError):
+            LoopState.from_dict({"task_id": "t1", "round": "x"})
+
+
+class WorkerEntryTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db = os.path.join(self.tmp.name, "test.db")
+        conn = sqlite3.connect(self.db)
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS task_attempts("
+            "task_id TEXT, attempt_no INTEGER, kind TEXT,"
+            " interrupted_reason TEXT, checkpoint_round INTEGER,"
+            " result_text TEXT, error_text TEXT,"
+            " created_at DOUBLE PRECISION,"
+            " PRIMARY KEY(task_id, attempt_no))")
+        conn.commit()
+        conn.close()
+        self.worker = Worker.__new__(Worker)
+        self.worker.db_path = self.db
+        self.worker.audits = []
+        self.worker._audit = lambda p, a, d="": self.worker.audits.append(
+            (p, a, d))
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def task(self, loop=True):
+        task = {"id": "task-1", "project_id": "proj-1"}
+        if loop:
+            task["audit_loop"] = True
+        return task
+
+    def test_off_by_default_touches_nothing(self):
+        out = self.worker.audit_loop_evaluate(
+            self.task(loop=False), {"tests": False}, [det("f1", "tests")])
+        self.assertEqual(out["decision"], "off")
+        conn = sqlite3.connect(self.db)
+        rows = conn.execute(
+            "SELECT name FROM sqlite_master WHERE type='table' AND"
+            " name='task_audit_loop_state'").fetchall()
+        conn.close()
+        self.assertEqual(rows, [])
+
+    def test_fail_fix_fail_fix_pass_path(self):
+        # 轮 1：fail -> 推回；轮 2：fail -> 推回；轮 3：pass。
+        out = self.worker.audit_loop_evaluate(
+            self.task(), {"tests": False}, [det("f1", "tests")])
+        self.assertEqual(out["decision"], "pushback")
+        self.assertEqual(out["round"], 1)
+        self.assertEqual(out["packet"]["findings"][0]["id"], "f1")
+        out = self.worker.audit_loop_evaluate(
+            self.task(), {"tests": False}, [det("f1", "tests")])
+        self.assertEqual(out["decision"], "pushback")
+        self.assertEqual(out["round"], 2)
+        out = self.worker.audit_loop_evaluate(
+            self.task(), {"tests": True}, [])
+        self.assertEqual(out["decision"], "pass")
+
+    def test_three_rounds_then_escalate(self):
+        for expected in (1, 2, 3):
+            out = self.worker.audit_loop_evaluate(
+                self.task(), {"tests": False}, [det("f1", "tests")])
+            self.assertEqual(out["decision"], "pushback")
+            self.assertEqual(out["round"], expected)
+        out = self.worker.audit_loop_evaluate(
+            self.task(), {"tests": False}, [det("f1", "tests")])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertEqual(out["reason"], "max_rounds_exceeded")
+        self.assertTrue(any(a == "worker_audit_loop_escalated"
+                            for _, a, _ in self.worker.audits))
+
+    def test_corrupted_count_never_loops(self):
+        self.worker.audit_loop_evaluate(
+            self.task(), {"tests": False}, [det("f1", "tests")])
+        conn = sqlite3.connect(self.db)
+        conn.execute(
+            "UPDATE task_audit_loop_state SET round='not-a-number'"
+            " WHERE task_id='task-1'")
+        conn.commit()
+        conn.close()
+        out = self.worker.audit_loop_evaluate(
+            self.task(), {"tests": False}, [det("f1", "tests")])
+        self.assertEqual(out["decision"], "escalate")
+        self.assertEqual(out["reason"], "count_corrupted")
+        self.assertTrue(any("count_corrupted" in d
+                            for _, _, d in self.worker.audits))
+
+    def test_state_survives_worker_restart(self):
+        self.worker.audit_loop_evaluate(
+            self.task(), {"tests": False}, [det("f1", "tests")])
+        fresh = Worker.__new__(Worker)
+        fresh.db_path = self.db
+        fresh.audits = []
+        fresh._audit = lambda p, a, d="": fresh.audits.append((p, a, d))
+        out = fresh.audit_loop_evaluate(
+            self.task(), {"tests": False}, [det("f1", "tests")])
+        self.assertEqual(out["decision"], "pushback")
+        self.assertEqual(out["round"], 2)
+
+    def test_subjective_via_worker_needs_review(self):
+        sub = Finding("design", "medium", "looks off", "read code",
+                      "reviewer approves", kind="subjective")
+        out = self.worker.audit_loop_evaluate(
+            self.task(), {"tests": True}, [sub])
+        self.assertEqual(out["decision"], "needs_review")
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -143,6 +143,10 @@ from cloud.attempts import build_resume_preamble as _resume_preamble
 from cloud.attempts import ensure_attempts_table as _ensure_attempts
 from cloud.attempts import latest_attempt as _latest_attempt
 from cloud.attempts import record_attempt as _record_attempt
+from cloud.audit_loop import ensure_audit_loop_table as _ensure_audit_loop
+from cloud.audit_loop import evaluate as _audit_loop_evaluate
+from cloud.audit_loop import load_loop_state as _audit_loop_load
+from cloud.audit_loop import save_loop_state as _audit_loop_save
 from cloud.auth import AuthStore
 from cloud.coordinator import build_report as _coord_build_report
 from cloud.coordinator import DEFAULT_POLL_INTERVAL as _COORD_POLL_INTERVAL
@@ -1792,6 +1796,101 @@ class Worker:
             except Exception:
                 pass
 
+    def audit_loop_evaluate(self, task, gate_results, findings=None,
+                            fixed=None):
+        """有界审计循环的显式入口（默认关闭，绝不由 run_one 自动调用）。
+
+        启用条件二选一中只选其一：任务字典含 audit_loop 真值。
+        没有 env 开关——全局 env 会让所有任务悄悄启用循环，本方法
+        刻意拒绝它。开关关闭时返回 enabled=False 且不碰数据库，
+        run_one、auto_resume、协调员、goal 断点行为逐字节不变。
+
+        本方法不改任务状态、不重排、不调 runner、不重新决策：
+        它只给决策——pass（放行）、pushback（调用方持 packet 重排
+        任务，packet 里有失败项、证据、复现、验收标准）、escalate
+        （调用方置任务 failed/needs_review，不再自动循环）、
+        needs_review（主观发现，走人工）、off（未启用）。
+        升级时记 worker_audit_loop_escalated 审计（best-effort，
+        审计失败不改决策）。轮次存在 task_audit_loop_state 单行表，
+        计数损坏时 fail-closed 到 escalate，绝不默认循环。
+        调用参数非法（空证据、空门集、非法门结果）时同样
+        fail-closed 到 escalate，不进循环。
+        gate_results 必须含本轮全部已跑的门；上一轮通过的门本轮
+        缺跑或变失败时，引擎按 gate_missing / regression 升级或推回。
+        """
+        task = task if isinstance(task, dict) else {}
+        if not task.get("audit_loop"):
+            return {"enabled": False, "decision": "off", "packet": None,
+                    "new_state": None, "reason": None}
+        task_id = task.get("id")
+        project_id = task.get("project_id")
+        try:
+            conn = _db_connect(self.db_path)
+        except Exception as exc:
+            # 连库都连不上：无法持久化轮次，fail-closed 到升级。
+            try:
+                self._audit(project_id, "worker_audit_loop_escalated",
+                            "%s reason=state_unreadable error=%s"
+                            % (task_id, exc))
+            except Exception:
+                pass
+            return {"enabled": True, "decision": "escalate", "packet": None,
+                    "new_state": None, "reason": "state_unreadable"}
+        try:
+            _ensure_audit_loop(conn)
+            state, corrupted, error = _audit_loop_load(conn, task_id)
+            if corrupted:
+                try:
+                    self._audit(
+                        project_id, "worker_audit_loop_escalated",
+                        "%s reason=count_corrupted error=%s"
+                        % (task_id, error))
+                except Exception:
+                    pass
+                return {"enabled": True, "decision": "escalate",
+                        "packet": None, "new_state": None,
+                        "reason": "count_corrupted"}
+            decision = _audit_loop_evaluate(
+                state, task_id, gate_results, findings=findings,
+                fixed=fixed)
+            if decision["new_state"] is not None:
+                _audit_loop_save(conn, decision["new_state"])
+            conn.commit()
+        except Exception as exc:
+            # 评估或写状态失败：轮次不可信，fail-closed 到升级。
+            try:
+                conn.rollback()
+            except Exception:
+                pass
+            try:
+                self._audit(project_id, "worker_audit_loop_escalated",
+                            "%s reason=evaluate_failed error=%s"
+                            % (task_id, exc))
+            except Exception:
+                pass
+            return {"enabled": True, "decision": "escalate", "packet": None,
+                    "new_state": None, "reason": "evaluate_failed"}
+        finally:
+            try:
+                conn.close()
+            except Exception:
+                pass
+        if decision["decision"] == "escalate":
+            try:
+                self._audit(
+                    project_id, "worker_audit_loop_escalated",
+                    "%s reason=%s round=%s"
+                    % (task_id, decision["reason"], decision["round"]))
+            except Exception:
+                pass
+        result = dict(decision)
+        result["enabled"] = True
+        if result.get("packet") is not None:
+            result["packet"] = result["packet"].to_dict()
+        if result.get("new_state") is not None:
+            result["new_state"] = result["new_state"].to_dict()
+        return result
+
     def _notify_task_finished(self, task_id, project_id, event_type,
                               error=None):
         """任务终态推送：best-effort，绝不因通知失败影响任务状态。