diff --git a/cloud/audit_loop.py b/cloud/audit_loop.py new file mode 100644 index 0000000..69a5ae2 --- /dev/null +++ b/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": [], + } diff --git a/cloud/db.py b/cloud/db.py index 8080819..4eaee4d 100644 --- a/cloud/db.py +++ b/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, " diff --git a/cloud/migrations.py b/cloud/migrations.py index 53d002d..e720bd7 100644 --- a/cloud/migrations.py +++ b/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"}) diff --git a/cloud/server.py b/cloud/server.py index 69154a3..2e7e9d5 100644 --- a/cloud/server.py +++ b/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， diff --git a/cloud/tests/test_audit_loop.py b/cloud/tests/test_audit_loop.py new file mode 100644 index 0000000..2c90749 --- /dev/null +++ b/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() diff --git a/cloud/worker.py b/cloud/worker.py index 49ad58c..0988a1d 100644 --- a/cloud/worker.py +++ b/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，绝不因通知失败影响任务状态。