From bc19f09dff64d8d7d86c527afc966551bc2a88c3 Mon Sep 17 00:00:00 2001 From: host Date: Wed, 4 Nov 2026 15:57:48 +0000 Subject: [PATCH] Audit batch 6: B1 zero-gate sentinel, I1 known-to-lack marker, I2 wire error returns row status, M1 max-rounds label, M2 chosen-last convention, M5 escalate write retry, M6/M7 clear/pass CAS tests --- cloud/audit_loop.py | 19 +- cloud/routing_shadow.py | 7 + cloud/tests/test_audit_batch5.py | 7 +- cloud/tests/test_audit_batch6.py | 536 +++++++++++++++++++++++++++++++ cloud/worker.py | 140 ++++++-- 5 files changed, 683 insertions(+), 26 deletions(-) create mode 100644 cloud/tests/test_audit_batch6.py diff --git a/cloud/audit_loop.py b/cloud/audit_loop.py index 2909151..a823da0 100644 --- a/cloud/audit_loop.py +++ b/cloud/audit_loop.py @@ -345,6 +345,19 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None): "regression": [], "missing_gates": [], "weakened": [], } + # 审计批 6（M1）：轮次耗尽的标签优先——max_rounds 分支 + # 必须排在 gate_failed_without_finding 分支之前，否则 + # 轮次耗尽 + 无 finding 时 reason 会被后者掩盖（决策同 + # 为 escalate、轮次不变，只是原因标签变化；交换顺序不 + # 改变任何决策，只改标签）。本支不要求 findings 非空， + # 但注意它在 pass 分支之后：无 finding、无失败门时已在 + # 上面返回 pass，不会误升级。 + if prev_round >= MAX_PUSHBACKS: + return _escalate( + task_id, prev_round, passed_now, new_criteria, + "max_rounds_exceeded", + regression=[f.to_dict() for f in regression]) + # 有 finding 但门全过、finding 为空的矛盾态：门说通过、发现说有 # 问题，以 finding 为准（finding 是审计的直接证据）。 # 审计批 4（I-11）：有失败门但无 finding 时，旧代码走到 @@ -364,12 +377,6 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None): "gate_failed_without_finding", regression=[f.to_dict() for f in regression]) - if prev_round >= MAX_PUSHBACKS: - return _escalate( - task_id, prev_round, passed_now, new_criteria, - "max_rounds_exceeded", - regression=[f.to_dict() for f in regression]) - new_round = prev_round + 1 packet = PushbackPacket(task_id, new_round, all_findings, new_criteria) new_state = LoopState( diff --git a/cloud/routing_shadow.py b/cloud/routing_shadow.py index 67f67a9..33de27c 100644 --- a/cloud/routing_shadow.py +++ b/cloud/routing_shadow.py @@ -69,6 +69,13 @@ _DETAIL_RE = re.compile( # parse, and legacy suggested-with-whitespace rows still land # in the unparseable bucket (counted, batch 2) rather than # being silently lost. + # Audit batch 6 (M2): the "chosen is last" placement is a + # standing convention, not an accident — (.+?) runs to end + # of line, so any field appended after chosen would be + # swallowed into chosen and misparsed. Writers (the two + # worker_routing audit sites in cloud/worker.py) must keep + # chosen as the final field; a test in test_audit_batch6.py + # locks both sites to that shape. r"(?:\s+chosen=(?P.+?))?\s*$") diff --git a/cloud/tests/test_audit_batch5.py b/cloud/tests/test_audit_batch5.py index 88f000b..37e90e3 100644 --- a/cloud/tests/test_audit_batch5.py +++ b/cloud/tests/test_audit_batch5.py @@ -231,6 +231,11 @@ class WireTxTestCase(_WorkerCase): def test_escalate_cas_miss_audits_skipped(self): # N-2：升级 CAS 未命中（状态已被改动）时记 # escalate_skipped 审计、不谎称已升级，任务与状态原样不动。 + # 审计批 6（I2）：接线 error 时返回的不再是输入终局， + # 而是回滚后读回的行实际状态（queued）；run_one 终局块 + # 据此抑制终态审计与通知。本用例断言意图随之改写： + # 旧断言 final == "failed"（输入终局）已是 I2 要消除的 + # 矛盾，新断言 final == 行状态 "queued"。 _insert_task(self.db_path, "TC", status="failed", audit_loop=1) _seed_state(self.db_path, "TC", 3, "pushback") _set_status(self.db_path, "TC", "queued") @@ -239,7 +244,7 @@ class WireTxTestCase(_WorkerCase): final, _, decision = worker._audit_loop_wire( task, "failed", {"stdout": ""}, "boom") self.assertEqual(decision, "error") - self.assertEqual(final, "failed") + self.assertEqual(final, "queued") self.assertEqual(_row(self.db_path, "TC")[0], "queued") self.assertEqual(_state_row(self.db_path, "TC")[0], 3) self.assertIn("worker_audit_loop_escalate_skipped", diff --git a/cloud/tests/test_audit_batch6.py b/cloud/tests/test_audit_batch6.py new file mode 100644 index 0000000..95a81e5 --- /dev/null +++ b/cloud/tests/test_audit_batch6.py @@ -0,0 +1,536 @@ +"""审计修复批 6 测试：B1、I1、I2、M1、M2、M5、M6、M7。 + +只用注入的假 runner 与 init_db 的本地库，不跑真实模型回合、 +不联网。按任务书：M3（报告措辞）、M4（观察项）、M8/M9 +（已知边界）不需要代码，本批只在报告里写明不做；N-12 的 +goal 包前言续跑缺口不在本批范围。 + +行为变更锁定（本批有意改语义，测试即契约）： +- B1：run_one 入口初始化 _gates_checked=[]，零门任务两轮后 + passed_gates 不再含硬编码三门；政策任务标记不受影响。 +- I1：_GATE_ERROR_MARKERS 补全长措辞 + "known to lack a required capability"，无结构化行时显式拦 + 与复核拒绝仍归 gate_rejected 且有回落审计行。 +- I2：接线 error 时返回行实际状态（不再是输入终局）， + run_one 终局块在行已不是终态时抑制终态审计与通知； + 批 5 的 test_escalate_cas_miss_audits_skipped 断言意图 + 随之改写（final == 行状态 queued，不再是输入终局 failed）。 +- M1：轮次耗尽 + 无 finding 时 reason 为 max_rounds_exceeded + （不再被 gate_failed_without_finding 掩盖），决策不变。 +- M2：两处 worker_routing 写点均以 chosen 结尾（行尾约定）。 +- M5：升级结果写回有界重试一次（仍带 only_if_status 守卫）。 +- M6：_coord_clear_packets 在清包前即失败时包列原样不动。 +- M7：pass 清包 CAS 未命中按 I2 新行为断言。 +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +try: + from cloud.audit_loop import LoopState, evaluate + from cloud.auth import AuthStore + from cloud.routing_shadow import parse_worker_routing_detail + from cloud.server import init_db + from cloud.worker import Worker, _GATE_ERROR_MARKERS + import cloud.worker as worker_module +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.audit_loop import LoopState, evaluate + from cloud.auth import AuthStore + from cloud.routing_shadow import parse_worker_routing_detail + from cloud.server import init_db + from cloud.worker import Worker, _GATE_ERROR_MARKERS + import cloud.worker as worker_module + + +def _insert_task(db_path, task_id, model=None, status="queued", + project_id="p1", audit_loop=0, parent=None, part=None, + error=None, result=None, packet=None): + now = time.time() + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat, audit_loop," + " parent_task_id, coordinator_part, result_text, error_text," + " model, audit_loop_packet)" + " VALUES(?, ?, 'do work', ?, NULL, '[]', 0, 'tester'," + " ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?)", + (task_id, project_id, status, now, now, audit_loop, + parent, part, result, error, model, packet)) + conn.commit() + finally: + conn.close() + + +def _set_started_at(db_path, task_id, started_at): + conn = sqlite3.connect(db_path) + try: + conn.execute( + "UPDATE tasks SET started_at = ? WHERE id = ?", + (started_at, task_id)) + conn.commit() + finally: + conn.close() + + +def _insert_audit(db_path, project_id, action, detail, created_at=None): + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO audit_log(user_id, project_id, action, detail," + " created_at) VALUES(NULL, ?, ?, ?, ?)", + (project_id, action, detail, + created_at if created_at is not None else time.time())) + conn.commit() + finally: + conn.close() + + +def _task_row(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + return conn.execute( + "SELECT status, audit_loop_packet FROM tasks WHERE id = ?", + (task_id,)).fetchone() + finally: + conn.close() + + +def _state_row(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + return conn.execute( + "SELECT round, passed_gates, last_decision" + " FROM task_audit_loop_state WHERE task_id = ?", + (task_id,)).fetchone() + finally: + conn.close() + + +def _seed_state(db_path, task_id, round_no, last_decision, + passed='["execution"]'): + conn = sqlite3.connect(db_path) + try: + conn.execute( + "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)") + conn.execute( + "INSERT OR REPLACE INTO task_audit_loop_state" + "(task_id, round, passed_gates, criteria, last_decision," + " updated_at) VALUES(?, ?, ?, ?, ?, ?)", + (task_id, round_no, passed, '{}', last_decision, + time.time())) + conn.commit() + finally: + conn.close() + + +def _routing_details(db_path): + conn = sqlite3.connect(db_path) + try: + return [r[0] for r in conn.execute( + "SELECT detail FROM audit_log WHERE action = 'worker_routing'" + " ORDER BY id").fetchall()] + finally: + conn.close() + + +def _completed_runner(cmd): + name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t" + return SimpleNamespace( + returncode=0, + stdout=json.dumps({"task": name, "status": "completed", + "text": "done"}), + stderr="") + + +def _failing_runner(cmd): + name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t" + return SimpleNamespace( + returncode=0, + stdout=json.dumps({"task": name, "status": "failed", + "text": "boom"}), + stderr="") + + +def _worker(db_path, policy_doc=None, env=None, gateway=None, + runner=None): + calls = [] + + def recording(cmd): + calls.append(list(cmd)) + return (runner or _completed_runner)(cmd) + + worker = Worker(db_path, "/fake/host.py", runner=recording, + env=env or {"XCM_API_KEY": "test-placeholder"}, + routing_policy=policy_doc) + if gateway is not None: + worker._ensure_gateway = lambda task: gateway + worker._sub2api_default_model = lambda e: "gateway-model" + worker._calls = calls + return worker + + +class _RunOneCase(unittest.TestCase): + """run_one 级夹具（批 4 GateResultsCheckedTestCase 同款口径）。""" + + def setUp(self): + self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + + def tearDown(self): + if self._old is None: + os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) + else: + os.environ[CODEX_PUSH_ALLOW_PRIVATE_TARGETS] = self._old + self.tmp.cleanup() + + +class B1ZeroGateTestCase(_RunOneCase): + """B1：run_one 入口初始化 _gates_checked，零门不再硬编码三门。""" + + def test_explicit_zero_gate_two_rounds_passed_gates_empty(self): + # 显式任务、无网关、无策略、无工具、开审计循环： + # 旧代码两轮后 passed_gates 是三门硬编码（漏网的直接 + # 原因正是本例缺 passed_gates 断言）；新代码应为空集。 + _insert_task(self.db_path, "T3", model="explicit-model", + audit_loop=1) + worker = _worker(self.db_path, gateway=(None, {}), + runner=_failing_runner) + for _ in range(2): + out = worker.run_one() + self.assertEqual(out["status"], "queued") + row = _state_row(self.db_path, "T3") + self.assertIsNotNone(row) + self.assertEqual(row[0], 2) + self.assertEqual(json.loads(row[1]), []) + self.assertEqual(row[2], "pushback") + + def test_policy_two_rounds_marks_survive_init(self): + # 对照：政策任务本轮实际跑过的 capability 标记不能被 + # 入口初始化冲掉，两轮后 passed_gates 仍为 ["capability"]。 + _insert_task(self.db_path, "T2", audit_loop=1) + worker = _worker(self.db_path, { + "mode": "active", + "tiers": {"default": {"model": "policy-model", + "requires": {"tool_calling": True}}}, + "model_capabilities": { + "policy-model": {"tool_calling": True}, + }, + }, gateway=(None, {}), runner=_failing_runner) + for _ in range(2): + out = worker.run_one() + self.assertEqual(out["status"], "queued") + row = _state_row(self.db_path, "T2") + self.assertIsNotNone(row) + self.assertEqual(row[0], 2) + self.assertEqual(json.loads(row[1]), ["capability"]) + self.assertEqual(row[2], "pushback") + + +class I1MarkerTestCase(unittest.TestCase): + """I1：无结构化行时显式拦与复核拒绝仍归 gate_rejected。""" + + def setUp(self): + self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + store = AuthStore(self.db_path) + store.conn.close() + self.audits = [] + self.worker = Worker(self.db_path, "/fake/host.py", + runner=_completed_runner, + env={"XCM_API_KEY": "test-placeholder"}, + max_retries=0, retry_backoff_seconds=0) + self.worker._audit = lambda p, a, d="": self.audits.append( + (p, a, d)) + conn = sqlite3.connect(self.db_path) + try: + self.worker._ensure_result_columns(conn) + conn.commit() + finally: + conn.close() + + def tearDown(self): + if self._old is None: + os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) + else: + os.environ[CODEX_PUSH_ALLOW_PRIVATE_TARGETS] = self._old + self.tmp.cleanup() + + def _merge(self, child_error, audit_rows=()): + now = time.time() + _insert_task(self.db_path, "P1", status="running", audit_loop=1, + parent=None) + _insert_task(self.db_path, "C1", status="failed", parent="P1", + part="a", error=child_error) + _set_started_at(self.db_path, "C1", now - 100) + for action, detail in audit_rows: + _insert_audit(self.db_path, "p1", action, detail, + created_at=now) + return self.worker._coord_audit_loop_merge( + {"id": "P1", "project_id": "p1", "audit_loop": 1}, + [{"name": "a", "prompt": "do a"}], + [{"id": "C1", "name": "a"}]) + + def test_explicit_block_error_only_is_gate_rejected(self): + # 显式网关拦的生产方 error 文案（worker.py 网关块）， + # 无 gate_rejected 行、无 gateway_explicit 行时仍须 + # 走子串回落归 gate_rejected（旧代码误归 execution + # 而推回）。回落审计行须点名新补的 marker。 + out = self._merge( + "default tier explicit model m lacks capability:" + " tool_calling; refusing to dispatch a model known to" + " lack a required capability") + self.assertEqual(out["decision"], "escalate") + self.assertIn("a", out.get("gate_rejected", [])) + self.assertEqual(_task_row(self.db_path, "C1")[0], "failed") + marker_rows = [d for _p, a, d in self.audits + if a == "worker_coord_gate_classified_by_marker"] + self.assertTrue(marker_rows) + self.assertIn("C1", marker_rows[0]) + self.assertIn("marker=known to lack a required capability", + marker_rows[0]) + + def test_recheck_reject_error_only_is_gate_rejected(self): + # 写回复核拒绝的生产方 error 文案（worker.py 复核块）， + # 同款无结构化行情形。 + out = self._merge( + "default tier persisted model m lacks capability:" + " tool_calling; refusing to dispatch a model known to" + " lack a required capability") + self.assertEqual(out["decision"], "escalate") + self.assertIn("a", out.get("gate_rejected", [])) + self.assertEqual(_task_row(self.db_path, "C1")[0], "failed") + marker_rows = [d for _p, a, d in self.audits + if a == "worker_coord_gate_classified_by_marker"] + self.assertTrue(marker_rows) + self.assertIn("C1", marker_rows[0]) + self.assertIn("marker=known to lack a required capability", + marker_rows[0]) + + +class _WireCase(unittest.TestCase): + """接线级夹具（批 5 _WorkerCase 同款口径，另加 run_one 级 + 测试所需的 CODEX_PUSH_ALLOW_PRIVATE_TARGETS 环境）。""" + + def setUp(self): + self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.audits = [] + + def tearDown(self): + if self._old is None: + os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) + else: + os.environ[CODEX_PUSH_ALLOW_PRIVATE_TARGETS] = self._old + self.tmp.cleanup() + + def _worker(self, **kwargs): + kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"}) + kwargs.setdefault("max_retries", 0) + kwargs.setdefault("retry_backoff_seconds", 0) + kwargs.setdefault("runner", _completed_runner) + worker = Worker(self.db_path, "/fake/host.py", **kwargs) + worker._audit = lambda p, a, d="": self.audits.append((p, a, d)) + return worker + + def _actions(self): + return [a for _, a, _ in self.audits] + + +class I2WireTestCase(_WireCase): + """I2 + M7：接线 error 返回行实际状态；pass CAS 未命中同款。""" + + def test_pass_cas_miss_returns_row_status_and_keeps_packet(self): + # M7：pass 清包 CAS 未命中——行已被并发重排回 queued， + # 接线返回 error + 行状态 queued（不再是输入终局 + # completed），旧包原样不动（事务回滚），状态行不动。 + _insert_task(self.db_path, "TP2", status="queued", audit_loop=1, + packet='{"round": 1}') + _seed_state(self.db_path, "TP2", 1, "pushback") + worker = self._worker() + task = worker.get_task("TP2") + final, _, decision = worker._audit_loop_wire( + task, "completed", {"stdout": ""}, "boom") + self.assertEqual(decision, "error") + self.assertEqual(final, "queued") + row = _task_row(self.db_path, "TP2") + self.assertEqual(row[0], "queued") + self.assertEqual(row[1], '{"round": 1}') + state = _state_row(self.db_path, "TP2") + self.assertEqual(state[0], 1) + self.assertEqual(state[2], "pushback") + self.assertNotIn("worker_audit_loop_passed", self._actions()) + + def test_run_one_suppresses_terminal_on_error_nonterminal(self): + # I2 终局块：接线 error 且行已不是终态时，抑制终态审计 + # 与 failed 通知，返回体状态与行一致（queued）。 + _insert_task(self.db_path, "TS", status="queued", audit_loop=1) + worker = self._worker(runner=_failing_runner) + notifies = [] + worker._notify_task_finished = lambda *a, **k: notifies.append( + (a, k)) + worker._audit_loop_wire = lambda t, f, r, e: ( + "queued", "boom", "error") + out = worker.run_one() + self.assertEqual(out["status"], "queued") + self.assertNotIn("worker_failed", self._actions()) + self.assertNotIn("worker_completed", self._actions()) + self.assertFalse([n for n in notifies if n[0][2] == "failed"]) + + def test_run_one_terminal_still_runs_on_error_terminal(self): + # 对照：接线 error 但行仍是终态 failed 时，终态审计与 + # failed 通知照常跑——抑制只针对“行已不是终态”情形， + # 不是见 error 就全抑。 + _insert_task(self.db_path, "TF", status="queued", audit_loop=1) + worker = self._worker(runner=_failing_runner) + notifies = [] + worker._notify_task_finished = lambda *a, **k: notifies.append( + (a, k)) + worker._audit_loop_wire = lambda t, f, r, e: ( + "failed", "boom", "error") + out = worker.run_one() + self.assertEqual(out["status"], "failed") + self.assertIn("worker_failed", self._actions()) + self.assertTrue([n for n in notifies if n[0][2] == "failed"]) + + +class M1MaxRoundsLabelTestCase(unittest.TestCase): + """M1：轮次耗尽 + 无 finding 时 reason 为 max_rounds_exceeded。""" + + def test_max_rounds_label_wins_over_no_finding(self): + # 上一轮未通过过的门本轮失败、无 finding、无回归、无缺门： + # 旧代码 reason 是 gate_failed_without_finding；新代码 + # max_rounds 分支在前，reason 为 max_rounds_exceeded， + # 决策同为 escalate、轮次不变。 + state = LoopState("T1", round=3, passed_gates=[], + criteria={}, last_decision="pushback") + out = evaluate(state, "T1", {"execution": False}, findings=[]) + self.assertEqual(out["decision"], "escalate") + self.assertEqual(out["reason"], "max_rounds_exceeded") + self.assertEqual(out["new_state"].round, 3) + self.assertEqual(out["new_state"].last_decision, "escalate") + + +class M2ChosenLastTestCase(_RunOneCase): + """M2：两处 worker_routing 写点均以 chosen 结尾（行尾约定）。""" + + def _assert_chosen_last(self, details): + self.assertTrue(details) + for detail in details: + parsed = parse_worker_routing_detail(detail) + self.assertIsNotNone(parsed, detail) + last_token = detail.strip().rsplit(" ", 1)[-1] + self.assertTrue(last_token.startswith("chosen="), detail) + + def test_policy_routing_row_ends_with_chosen(self): + # 策略/shadow 决策路径的 worker_routing 行。 + _insert_task(self.db_path, "M1", audit_loop=0) + worker = _worker(self.db_path, { + "mode": "shadow", + "tiers": {"default": "policy-model"}, + }, gateway=(None, {}), runner=_failing_runner) + worker.run_one() + details = _routing_details(self.db_path) + self.assertTrue([d for d in details if "source=shadow" in d]) + self._assert_chosen_last( + [d for d in details if "source=shadow" in d]) + + def test_explicit_routing_row_ends_with_chosen(self): + # 显式任务路径的 worker_routing 行（source=explicit）。 + _insert_task(self.db_path, "M2", model="explicit-model", + audit_loop=0) + worker = _worker(self.db_path, gateway=(None, {}), + runner=_failing_runner) + worker.run_one() + details = _routing_details(self.db_path) + self.assertTrue([d for d in details if "source=explicit" in d]) + self._assert_chosen_last( + [d for d in details if "source=explicit" in d]) + + +class M5EscalateRetryTestCase(_WireCase): + """M5：升级结果写回有界重试一次（仍带 failed 守卫）。""" + + def _escalating_worker(self): + _insert_task(self.db_path, "TE", status="failed", audit_loop=1) + _seed_state(self.db_path, "TE", 3, "pushback") + return self._worker() + + def test_escalate_result_write_retried_once(self): + # 第一次写失败、第二次成功：接线仍返回 escalate， + # 写回恰好调两次（重试生效，且不是无界重试）。 + worker = self._escalating_worker() + calls = [] + real = worker._write_execution_result + + def flaky(*a, **k): + calls.append(1) + if len(calls) == 1: + raise RuntimeError("transient write failure") + return real(*a, **k) + + worker._write_execution_result = flaky + task = worker.get_task("TE") + final, _, decision = worker._audit_loop_wire( + task, "failed", {"stdout": ""}, "boom") + self.assertEqual(decision, "escalate") + self.assertEqual(final, "failed") + self.assertEqual(len(calls), 2) + + def test_escalate_result_write_bounded_on_repeated_failure(self): + # 两次都失败：仍返回 escalate（不谎报成 error）， + # 写回恰好调两次，不会无界重试。 + worker = self._escalating_worker() + worker._write_execution_result = mock.Mock( + side_effect=RuntimeError("persistent write failure")) + task = worker.get_task("TE") + final, _, decision = worker._audit_loop_wire( + task, "failed", {"stdout": ""}, "boom") + self.assertEqual(decision, "escalate") + self.assertEqual(final, "failed") + self.assertEqual(worker._write_execution_result.call_count, 2) + + +class M6ClearPacketsEarlyFailureTestCase(_WireCase): + """M6：清包前即失败时包列原样不动（旧测试只在清干净后注错）。""" + + def test_clear_packets_failure_before_clear_keeps_packets(self): + _insert_task(self.db_path, "PP", status="failed", audit_loop=1, + packet='{"round": 1}') + _insert_task(self.db_path, "CC", status="failed", audit_loop=0, + packet='{"round": 1}') + worker = self._worker() + + def _boom(path): + raise RuntimeError("injected connect failure") + + with mock.patch.object(worker_module, "_db_connect", _boom): + # 不抛错，且包列原样不动——失败不改已定结局。 + worker._coord_clear_packets( + {"id": "PP", "project_id": "p1"}, ["CC"]) + self.assertEqual(_task_row(self.db_path, "PP")[1], + '{"round": 1}') + self.assertEqual(_task_row(self.db_path, "CC")[1], + '{"round": 1}') diff --git a/cloud/worker.py b/cloud/worker.py index 425416f..ab8c869 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -508,6 +508,9 @@ def _mark_gate_checked(task, gate): 无论哪份拷贝打标记，终局块读到的都是同一份集合。 终局时 _audit_loop_gate_results 只列入被标记过的门， 不再把没跑过的门硬编码成通过。 + 审计批 6（B1）：run_one 入口已初始化该列表为空——零门 + 任务的 [] 与旧直接调用的 None 是两个不同哨兵，终局块 + 传 [] 时零门不列入，传 None（旧直接调用）时才走旧行为。 """ if not isinstance(task, dict): return @@ -533,6 +536,14 @@ _GATE_ERROR_MARKERS = ( "工具审批声明矛盾", "refusing to dispatch an incapable or unverifiable model", "fails capability", + # 审计批 6（I1）：显式网关拦与写回复核拒绝共用的措辞 + # "…refusing to dispatch a model known to lack a required + # capability"。选全长措辞而不用较短的 "known to lack"： + # 全长措辞在全库只出现在上述两处生产方 error 文案里， + # 误命中面最小；较短形式可能被执行失败的 error_text + # （例如模型输出复述）偶然命中。误命中方向是 escalate + # （安全侧、丢一次重试），不是放行，但精度仍优先。 + "known to lack a required capability", ) @@ -2047,8 +2058,10 @@ class Worker: passed_gates 并持久化——“通过”一词在状态表里失真。 现在：调用方传 gates_checked（run_one 派发点用 _mark_gate_checked 在 task["_gates_checked"] 里记的 - 本轮实际跑过的门）时，只列入集合内的门；未传 - （直接调用的旧调用方）时保持旧行为、三门全记 True。 + 本轮实际跑过的门；审计批 6（B1）起 run_one 入口 + 已初始化该列表为空，零门任务传 []、不再传 None） + 时，只列入集合内的门；未传（直接调用的旧调用方） + 时保持旧行为、三门全记 True。 gate_missing 语义核查：引擎只对“上一轮 passed_gates 里有、本轮缺失”的门升级；同一任务的跑过门集合在 两轮间是稳定的（显式任务无策略时两轮都不记 @@ -2100,6 +2113,27 @@ class Worker: gate_results[gate] = True return gate_results, findings + def _wire_error_final(self, conn, task_id, final): + """CAS 未命中/存库失败后读回行实际状态（审计批 6，I2）。 + + 接线 error 时调用方（run_one 终局块）不能再拿输入终局 + 当行状态：CAS 未命中意味着行已被第三方改动（例如重排 + 回 queued），继续按输入终局记终态审计、发 failed 通知 + 会与行状态自相矛盾。同一连接在回滚后仍可读，读到即 + 以行为准；读不到（连接已坏、行已删）时回落输入终局， + 不比旧行为更差。连接都连不上时调用方无连接可用， + 不走本方法（行未被接线改动，输入终局即行状态）。 + """ + try: + row = conn.execute( + "SELECT status FROM tasks WHERE id = ?", + (task_id,)).fetchone() + except Exception: + return final + if row and row[0]: + return row[0] + return final + def _audit_loop_wire(self, task, final, result, error): """审计循环调用方接线：把引擎决策变成真实的任务结局。 @@ -2142,6 +2176,15 @@ class Worker: result，自带连接）仍是提交后的独立写：它失败时只记 警告、不再把已 failed 的任务谎报成 error（旧版写回 失败会返回 error，尽管状态已是 failed）。 + 审计批 6（I2）：CAS 未命中/存库失败返回 error 时， + final 不再是输入终局，而是回滚后读回的行实际状态 + （例如行已被重排回 queued 时返回 queued）；run_one + 终局块据此抑制终态审计与通知，不再按输入终局谎报。 + 连接都连不上时无连接可读，仍返回输入终局（行未被 + 接线改动，输入终局即行状态）。 + 审计批 6（M5）：升级成功后的结果写回有界重试一次 + （仍带 only_if_status="failed" 守卫，幂等），两次 + 都失败时仍只记警告、不谎报成 error。 goal 任务（审计批 5，N-12 定性，只文档化、不改行为）： 推回只改 tasks.status 与包列（+ 轮次状态表），不碰 goal / verify_command / goal_max_rounds 列，也不碰宿主 @@ -2221,7 +2264,10 @@ class Worker: message="task status changed before packet clear") except Exception: pass - return final, error, "error" + # 审计批 6（I2）：回滚后读回行实际状态，不再 + # 拿输入终局当行状态返回（行可能已被重排）。 + return self._wire_error_final(conn, task_id, final), \ + error, "error" conn.commit() self._audit(project_id, "worker_audit_loop_passed", "%s round=%s" % (task_id, decision.get("round"))) @@ -2253,7 +2299,9 @@ class Worker: message="task status changed before pushback") except Exception: pass - return final, error, "error" + # 审计批 6（I2）：回滚后读回行实际状态。 + return self._wire_error_final(conn, task_id, final), \ + error, "error" conn.commit() self._audit(project_id, "worker_audit_loop_pushback", "%s round=%s findings=%s" % ( @@ -2289,7 +2337,9 @@ class Worker: message="task status changed before escalate") except Exception: pass - return final, error, "error" + # 审计批 6（I2）：回滚后读回行实际状态。 + return self._wire_error_final(conn, task_id, final), \ + error, "error" conn.commit() # 升级审计在提交后记（与旧接线“引擎已记”的行同款， # 只是时序从评估提交后挪到接线提交后，绝不出现 @@ -2310,13 +2360,27 @@ class Worker: "%s reason=%s round=%s" % (task_id, decision.get("reason"), decision.get("round"))) - try: - self._write_execution_result( - task_id, result_text=(result or {}).get("stdout", ""), - error_text=esc_error, actual_model=None, - started_at=None, finished_at=time.time(), - only_if_status="failed") - except Exception as exc: + # 审计批 6（M5）：提交后的结果写回仍是独立写， + # 但瞬时失败不再直接认命——有界单次重试（仍带 + # only_if_status="failed" 守卫，重试是幂等的： + # 行被并发重排后守卫挡住，不会覆盖新结果）。 + # 两次都失败时保持旧行为：只记警告、不把已 + # failed 的任务谎报成 error，DB error_text + # 停留旧文本的残留仍在，见报告。 + _write_exc = None + for _attempt in range(2): + try: + self._write_execution_result( + task_id, + result_text=(result or {}).get("stdout", ""), + error_text=esc_error, actual_model=None, + started_at=None, finished_at=time.time(), + only_if_status="failed") + _write_exc = None + break + except Exception as exc: + _write_exc = exc + if _write_exc is not None: # 结果文本写回是提交后的独立写：失败只记警告， # 任务已是 failed，不再谎报成 error。 try: @@ -2325,12 +2389,15 @@ class Worker: "worker_audit_loop_escalate_result_failed", task_id=task_id, project_id=project_id, status="warning", - message="escalate result write failed: %s" % exc) + message="escalate result write failed: %s" + % _write_exc) except Exception: pass return "failed", esc_error, "escalate" conn.rollback() - return final, error, "error" + # 审计批 6（I2）：未知决策同样回滚后读回行实际状态。 + return self._wire_error_final(conn, task_id, final), \ + error, "error" except Exception as exc: try: conn.rollback() @@ -2343,7 +2410,10 @@ class Worker: message="audit loop wire store failed: %s" % exc) except Exception: pass - return final, error, "error" + # 审计批 6（I2）：异常后回滚，行未被接线改动时读回 + # 即输入终局本身；行已被第三方改动时以行为准。 + return self._wire_error_final(conn, task_id, final), \ + error, "error" finally: try: conn.close() @@ -7583,6 +7653,17 @@ class Worker: except Exception: pass task = self.get_task(task_id) or task + # 审计批 6（B1，批 4 I-10 的 None 哨兵坍塌）：run_one + # 入口立即初始化 _gates_checked 为空列表，使“本轮零 + # 门跑过”（[]）与“旧直接调用”（None）可区分——否则 + # 零门任务在终局块传 None 时 _audit_loop_gate_results + # 走旧行为分支，把没跑过的三门硬编码成通过（批 4 + # 实测：显式零门任务两轮后 passed_gates 仍是三门）。 + # 列表对象经后续 dict(task) 浅拷贝共享，标记点与终局 + # 块读到同一份集合；协调员父任务不走单任务接线， + # 初始化对它无害（内存态，不进库、不进审计）。 + if not isinstance(task.get("_gates_checked"), list): + task["_gates_checked"] = [] # G10：goal 断点续跑标记——认领后立即清掉库里标记（本轮 # 已认领为 running，不会重复触发；本轮若再被杀回 # interrupted，用户需重新调 /goal-resume 才会再置标记， @@ -8005,6 +8086,9 @@ class Worker: # tier/source/applied，评估时建议模型只能记 # unknown；新格式补 suggested/chosen，旧解析 # 器仍能读（见 cloud/routing_shadow.py）。 + # 审计批 6（M2）：chosen 必须是行尾字段——影子 + # 解析器按行尾锚定 chosen，后续写点不得在其 + # 后追加字段，否则会被吞进 chosen 而误解析。 self._audit(task["project_id"], "worker_routing", "%s tier=%s source=%s applied=%s" " suggested=%s chosen=%s" % ( @@ -8061,6 +8145,8 @@ class Worker: # 行：评估函数只计 per-tier explicit_n，不计 # routed_n、不占 seen（见 routing_shadow.py）， # 父模型语义不变，档模型绝不覆盖它。 + # 审计批 6（M2）：同款行尾约定——chosen 必须是 + # 行尾字段，后续写点不得在其后追加字段。 self._audit(task["project_id"], "worker_routing", "%s tier=%s source=explicit applied=false" " suggested=- chosen=%s" % ( @@ -8588,6 +8674,7 @@ class Worker: # 本轮花过的钱照记，推回不等于没花钱。 audit_loop_decision = None audit_loop_pushback = False + audit_loop_suppress = False if (final_written and not quota_stop and final in ("completed", "failed") and task.get("audit_loop") @@ -8596,6 +8683,17 @@ class Worker: final, error, audit_loop_decision = \ self._audit_loop_wire(task, final, result, error) audit_loop_pushback = (audit_loop_decision == "pushback") + # 审计批 6（I2）：接线 error 且行实际状态已不是 + # 终态（例如 CAS 未命中后行回 queued）时，终态 + # 审计/时间线/通知必须随之抑制——按输入终局记 + # worker_failed 并发 failed 通知会与行状态自相 + # 矛盾。行仍是终态（回滚情形：failed/completed） + # 或已被取消（interrupted，按 interrupted 终局 + # 处理、不发 failed 通知）时不抑制。 + audit_loop_suppress = ( + audit_loop_decision == "error" + and final not in ("completed", "failed", + "interrupted")) except Exception as exc: try: log_event(self.logger, "worker_audit_loop_wire_failed", @@ -8615,7 +8713,8 @@ class Worker: # 持锁写，否则两个线程共用同一 sqlite 连接会直接报错。 cost_id = self._record_usage_cost( task, actual_model, usage_total, stdout, fallback_count) - if final_written and not audit_loop_pushback: + if final_written and not audit_loop_pushback \ + and not audit_loop_suppress: # 审计名须与终态一致：interrupted 任务记 worker_failed # 会把“可续跑的中断”误读成失败，误导后续排查。 # 终态审计细节（2026-10-29 任务 A 教训）：interrupted / @@ -8668,11 +8767,14 @@ class Worker: elif final in ("completed", "failed"): self._notify_task_finished( task_id, task["project_id"], final, error=error) - elif audit_loop_pushback: + elif audit_loop_pushback or audit_loop_suppress: # 推回已由接线处理（状态 queued、包列已写、推回审计 # 已记）：这里不跑终态审计、不改 final、不发通知。 - # 注意不能落入下面的 else——else 会把 final 改成 - # interrupted，与库里的 queued 自相矛盾。 + # 审计批 6（I2）：接线 error 且行已不是终态时同款 + # 抑制——这里同样不跑终态审计、不改 final、不发 + # 通知，final 已是接线读回的行实际状态，返回体 + # 与行一致。注意不能落入下面的 else——else 会把 + # final 改成 interrupted，与库里的 queued 自相矛盾。 pass else: if goal_converted: