diff --git a/cloud/agent_def.py b/cloud/agent_def.py index e0e3019..30e8a2c 100644 --- a/cloud/agent_def.py +++ b/cloud/agent_def.py @@ -216,6 +216,17 @@ def parse_agent_tools_column(raw): parse logic therefore changes no observable behaviour on either side; the old server/worker defaults (None vs []) were only different spellings of the same dispatch outcome. + Audit batch 5 (N-9, deliberate choice: document, do not filter): + the parser guarantees list shape only, NOT item types — a dirty + row like ["a", 1] is returned as-is. Downstream tolerates + non-string items, which is why filtering here would be churn, + not a fix: the worker required-tools gate only formats items + with "%s" (missing_required_tools) and compares them with == + (deny check), the server only JSON-dumps the list, and no use + site calls a str method on an item. Filtering at parse time + would instead change serialization of old dirty rows (the + server today echoes them exactly), so the loose contract stays + and is locked by test. """ if raw is None: return None diff --git a/cloud/audit_loop.py b/cloud/audit_loop.py index 40c17fe..2909151 100644 --- a/cloud/audit_loop.py +++ b/cloud/audit_loop.py @@ -224,6 +224,28 @@ def _validate_gate_results(gate_results): raise ValueError("gate_results must map gate name to bool") +def _escalate(task_id, prev_round, passed_gates, criteria, reason, + regression=(), missing_gates=(), weakened=()): + """升级决策 dict（审计批 5，N-8：纯结构整理，行为不变）。 + + evaluate() 内四处 escalate 返回原来各写一遍同样的 dict + 骨架（new_state + decision/round/packet/reason + 三个清单）， + 只有 passed_gates / criteria / reason / 清单内容不同；抽成 + 助手后各处只传不同项。new_state 的 round 恒为 prev_round + （升级不推进轮次）、last_decision 恒为 escalate。 + """ + new_state = LoopState( + task_id, round=prev_round, passed_gates=sorted(passed_gates), + criteria=criteria, last_decision="escalate") + return { + "decision": "escalate", "round": prev_round, "packet": None, + "new_state": new_state, "reason": reason, + "regression": list(regression), + "missing_gates": list(missing_gates), + "weakened": list(weakened), + } + + def evaluate(state, task_id, gate_results, findings=None, fixed=None): """纯函数：一轮审计的决策。 @@ -269,17 +291,10 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None): # 全量回归：上一轮通过的门本轮必须重跑，缺跑即升级。 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": [], - } + return _escalate( + task_id, prev_round, + (g for g, v in gate_results.items() if v), + prev_criteria, "gate_missing", missing_gates=missing_gates) 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} @@ -310,15 +325,11 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None): 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), - } + return _escalate( + task_id, prev_round, passed_now, prev_criteria, + "standards_weakened", + regression=[f.to_dict() for f in regression], + weakened=sorted(weakened)) # 新标准 = 未解决的旧标准 + 本轮新 finding（含回归）。 for finding in all_findings: @@ -348,26 +359,16 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None): # 转 escalate），新行为后看到的是 gate_failed_without_ # finding 升级——原因更精确，不是行为倒退。 if failed_now and not all_findings: - 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": "gate_failed_without_finding", - "regression": [f.to_dict() for f in regression], - "missing_gates": [], "weakened": [], - } + return _escalate( + task_id, prev_round, passed_now, new_criteria, + "gate_failed_without_finding", + regression=[f.to_dict() for f in regression]) 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": [], - } + 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) diff --git a/cloud/routing_dryrun.py b/cloud/routing_dryrun.py index cc792b8..c1fb60e 100644 --- a/cloud/routing_dryrun.py +++ b/cloud/routing_dryrun.py @@ -98,6 +98,25 @@ def _tier_dropped_reason(raw_entry): return "unrecognised tier entry (expected string or dict)" +def _empty_report(warnings, tiers, dropped, unknown, gateway): + """无效策略的空报告（审计批 5，N-10：纯结构整理，行为不变）。 + + dry_run_policy_document 的两处早期返回（非对象文档、解析 + 出空策略）原来各写一遍同样的 dict 骨架，只有 warnings 不同； + 抽成助手后各处只传不同项。传入的列表按引用进报告（与原来 + 直接写字面量时同一语义），不拷贝。 + """ + return { + "valid": False, "blocked": True, "version": None, + "content_fingerprint": None, + "global_mode": None, "global_mode_source": None, + "tiers": tiers, "dropped_tiers": dropped, + "unknown_tiers": unknown, "warnings": warnings, + "would_activate": [], "would_shadow": [], "would_off": [], + "gateway": gateway, + } + + def dry_run_policy_document(data, env_mode_override=None, gateway_model=None): """Dry-run one merged policy document (pure, no IO). @@ -118,17 +137,10 @@ def dry_run_policy_document(data, env_mode_override=None, if gateway_model else "env_unset"), "tiers": {}} if not isinstance(data, dict): - return { - "valid": False, "blocked": True, "version": None, - "content_fingerprint": None, - "global_mode": None, "global_mode_source": None, - "tiers": tiers_report, "dropped_tiers": dropped, - "unknown_tiers": unknown, - "warnings": ["policy document is not an object; " - "the loader would ignore it entirely"], - "would_activate": [], "would_shadow": [], "would_off": [], - "gateway": _gw, - } + return _empty_report( + ["policy document is not an object; " + "the loader would ignore it entirely"], + tiers_report, dropped, unknown, _gw) raw_tiers = data.get("tiers") if isinstance(raw_tiers, dict): for name in sorted(raw_tiers): @@ -142,15 +154,8 @@ def dry_run_policy_document(data, env_mode_override=None, "policy is unusable (no usable tiers, or global mode " "outside off/shadow/active); the loader would ignore it " "entirely and workers would keep the worker default") - return { - "valid": False, "blocked": True, "version": None, - "content_fingerprint": None, - "global_mode": None, "global_mode_source": None, - "tiers": tiers_report, "dropped_tiers": dropped, - "unknown_tiers": unknown, "warnings": warnings, - "would_activate": [], "would_shadow": [], "would_off": [], - "gateway": _gw, - } + return _empty_report( + warnings, tiers_report, dropped, unknown, _gw) global_mode = policy.mode global_source = "policy" if env_mode_override: diff --git a/cloud/server.py b/cloud/server.py index b38e007..978b798 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -4200,7 +4200,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): # cost_records 表只在 CostStore 首次初始化时建：新鲜 # 库上直接查会报 no such table，报告端点不应因此 500。 # 先建表（幂等），再读；建表失败时成本按空算、报告 - # 里 token/usd 为 0，不假装成本已计。 + # 里 token/usd 为 0，不假装成本已计。审计批 5（N-1）： + # 下面成本 SELECT 同样加 except 兜底（此前只有 finally、 + # 没有 except，建表失败会直接抛到 handler 层），失败时 + # 成本按空算并在报告里标 cost_unavailable。 try: _cost_store = CostStore(self.server.db_path) try: @@ -4238,11 +4241,19 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): " WHERE project_id = ?", (project_id,), ).fetchall() - cost_rows = conn.execute( - "SELECT task_id, total_tokens, usd_micros" - " FROM cost_records WHERE project_id = ?", - (project_id,), - ).fetchall() + cost_unavailable = False + try: + cost_rows = conn.execute( + "SELECT task_id, total_tokens, usd_micros" + " FROM cost_records WHERE project_id = ?", + (project_id,), + ).fetchall() + except Exception: + # 审计批 5（N-1）：成本查询失败（通常是建表失败后 + # 表仍不存在）时成本按空算，不抛到 handler 层； + # 报告里的 cost_unavailable 标明成本不可用。 + cost_rows = [] + cost_unavailable = True finally: conn.close() tasks_by_id = {} @@ -4275,6 +4286,18 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): tasks_by_id, costs_by_task, explicit_n=explicit_n) report["project_id"] = project_id report["truncated"] = truncated + report["cost_unavailable"] = cost_unavailable + # 审计批 5（N-5）：截断取最早 10000 行（ORDER BY + # created_at ASC），与协调员门查询取最新 5000 行方向 + # 相反——调用方须知样本是哪一段：同时返回样本时间窗 + # （保留样本的最早/最晚 created_at；无样本时为 None）。 + if audit_rows: + report["sample_window"] = { + "earliest_created_at": audit_rows[0]["created_at"], + "latest_created_at": audit_rows[-1]["created_at"]} + else: + report["sample_window"] = { + "earliest_created_at": None, "latest_created_at": None} # 显式失败清单：解析 worker_capability_gateway_explicit # 行（格式：task_id model=.. tier=.. mode=.. missing=.. # unknown=.. blocked=..；fix4 之前的行无 blocked 项， @@ -4570,6 +4593,57 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): for path, size in saved], }) + @staticmethod + def _resolve_create_agent_definition(body): + """建任务的 agent 定义解析（审计批 5，I-9③抽取，行为逐字不变）。 + + 原 _create_task 内 agent/agent_name 解析段平移：返回 + (definition, error)——error 非空时调用方按原样发 400 且 + 不建任务；definition 为 None 且 error 为空表示未给定义。 + 副作用：无库写、无响应发送（400 由调用方发，与原地一致）。 + """ + if body.get("agent") is not None \ + and body.get("agent_name") is not None: + return None, "agent and agent_name cannot both be used" + if body.get("agent") is not None \ + and body.get("agent_version") is not None: + return None, ("agent_version is only for agent_name;" + " an inline agent carries its own version") + definition = None + if body.get("agent") is not None: + if not isinstance(body.get("agent"), dict): + return None, "agent must be an object" + try: + definition = _parse_agent_definition( + body.get("agent"), + known_skills=_agent_known_skills()) + except ValueError as exc: + return None, str(exc) + if body.get("agent_name") is not None: + try: + definition = _resolve_agent( + body.get("agent_name"), + version=body.get("agent_version"), + agents_dir=_agent_default_dir(), + known_skills=_agent_known_skills()) + except ValueError as exc: + return None, str(exc) + return definition, None + + @staticmethod + def _validate_create_audit_loop(body): + """建任务的 audit_loop 开关校验（审计批 5，I-9③抽取，行为逐字不变）。 + + 原 _create_task 内审计循环接线校验段平移：只认 True/1 + （其它真值一律错，不猜）。返回 (audit_loop_val, error)—— + error 非空时调用方按原样发 400 且不建任务。 + """ + if body.get("audit_loop") is None: + return 0, None + if body.get("audit_loop") not in (True, 1): + return None, "audit_loop must be true or 1" + return 1, None + def _create_task(self, user_id): body, err = self._read_json() if err: @@ -4606,39 +4680,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): # 优先于定义；解析出的字段落进已有列，定义身份与 # 指令落进 agent_name/agent_version/agent_instructions # 列（v2），prompt 不再带头。 - definition = None - if body.get("agent") is not None \ - and body.get("agent_name") is not None: - self._send_json(400, { - "error": "agent and agent_name cannot both be used"}) + definition, _agent_err = \ + self._resolve_create_agent_definition(body) + if _agent_err: + self._send_json(400, {"error": _agent_err}) return - if body.get("agent") is not None \ - and body.get("agent_version") is not None: - self._send_json(400, { - "error": "agent_version is only for agent_name;" - " an inline agent carries its own version"}) - return - if body.get("agent") is not None: - if not isinstance(body.get("agent"), dict): - self._send_json(400, {"error": "agent must be an object"}) - return - try: - definition = _parse_agent_definition( - body.get("agent"), - known_skills=_agent_known_skills()) - except ValueError as exc: - self._send_json(400, {"error": str(exc)}) - return - if body.get("agent_name") is not None: - try: - definition = _resolve_agent( - body.get("agent_name"), - version=body.get("agent_version"), - agents_dir=_agent_default_dir(), - known_skills=_agent_known_skills()) - except ValueError as exc: - self._send_json(400, {"error": str(exc)}) - return depends_on = body.get("depends_on") or [] if isinstance(depends_on, str): depends_on = [depends_on] @@ -4976,13 +5022,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): # 就不写 audit_loop 列（默认 0），父任务的循环只记在 # 父任务上，子任务包只是前言载体。goal 任务允许（goal # 校验门是单任务接线要消费的确定性门之一）。 - audit_loop_val = 0 - if body.get("audit_loop") is not None: - if body.get("audit_loop") not in (True, 1): - self._send_json(400, { - "error": "audit_loop must be true or 1"}) - return - audit_loop_val = 1 + audit_loop_val, _al_err = self._validate_create_audit_loop(body) + if _al_err: + self._send_json(400, {"error": _al_err}) + return agent_name_val = definition.name if definition else None agent_version_val = definition.version if definition else None agent_instructions_val = ( diff --git a/cloud/tests/test_audit_batch5.py b/cloud/tests/test_audit_batch5.py new file mode 100644 index 0000000..88f000b --- /dev/null +++ b/cloud/tests/test_audit_batch5.py @@ -0,0 +1,593 @@ +"""审计批 5（I-9、I-14、N-1～N-12、I-12、事务分离）测试。 + +只用注入的假 runner 与 init_db 的本地库，不跑真实模型回合、 +不联网。按任务书：新测试 + 相关旧套件确认无回归；本文件 +只覆盖本批的新行为与新契约，旧行为由旧套件锁定。 +""" + +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +try: + from cloud.agent_def import ( + missing_required_tools, + parse_agent_tools_column, + ) + from cloud.audit_loop import Finding + from cloud.auth import AuthStore + from cloud.models import parse_routing_policy + from cloud.routing_dryrun import dry_run_policy_document + from cloud.server import CloudControlPlaneHandler, create_server, init_db + from cloud.worker import Worker + import cloud.worker as worker_module + import cloud.server as server_module +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.agent_def import ( + missing_required_tools, + parse_agent_tools_column, + ) + from cloud.audit_loop import Finding + from cloud.auth import AuthStore + from cloud.models import parse_routing_policy + from cloud.routing_dryrun import dry_run_policy_document + from cloud.server import CloudControlPlaneHandler, create_server, init_db + from cloud.worker import Worker + import cloud.worker as worker_module + import cloud.server as server_module + + +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 _insert_task(db_path, task_id, status="queued", audit_loop=0, + goal_mode=0, goal=None, verify_command=None, + goal_max_rounds=None, model=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," + " goal_mode, goal, verify_command, goal_max_rounds, model," + " audit_loop_packet)" + " VALUES(?, 'p1', 'do work', ?, NULL, '[]', 0, 'tester'," + " ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?)", + (task_id, status, now, now, audit_loop, goal_mode, goal, + verify_command, goal_max_rounds, model, packet)) + conn.commit() + finally: + conn.close() + + +def _set_status(db_path, task_id, status): + conn = sqlite3.connect(db_path) + try: + conn.execute("UPDATE tasks SET status = ? WHERE id = ?", + (status, task_id)) + conn.commit() + finally: + conn.close() + + +def _row(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + return conn.execute( + "SELECT status, audit_loop_packet, goal, verify_command," + " goal_max_rounds, model 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, 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): + 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, '["execution"]', '{}', last_decision, + time.time())) + conn.commit() + finally: + conn.close() + + +class _FailingExecuteConn: + """注入失败的连接包装：指定 SQL 片段一律抛错，其余透传。""" + + def __init__(self, real, needle): + self._real = real + self._needle = needle + + def execute(self, sql, params=()): + if self._needle in sql: + raise RuntimeError("injected store failure") + return self._real.execute(sql, params) + + def __getattr__(self, name): + return getattr(self._real, name) + + +class _WorkerCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.audits = [] + + def tearDown(self): + 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 FingerprintTestCase(unittest.TestCase): + def test_fallback_models_change_fingerprint(self): + # I-14①：指纹宣称含 fallback_models，变化必须敏感。 + base = {"mode": "shadow", "tiers": {"default": "m1"}, + "model_capabilities": {"m1": {"tool_calling": True}}} + a = parse_routing_policy(base) + b = parse_routing_policy( + dict(base, fallback_models=["m2"])) + self.assertNotEqual(a.content_fingerprint(), + b.content_fingerprint()) + c = parse_routing_policy(dict(base, fallback_models=["m2"])) + self.assertEqual(b.content_fingerprint(), + c.content_fingerprint()) + + +class WireTxTestCase(_WorkerCase): + def _patched_worker(self, needle): + real_connect = worker_module._db_connect + + def _connect(path): + return _FailingExecuteConn(real_connect(path), needle) + + patcher = mock.patch.object(worker_module, "_db_connect", + _connect) + patcher.start() + self.addCleanup(patcher.stop) + return self._worker() + + def test_pushback_store_failure_rolls_back_state(self): + # 事务分离：推回 UPDATE 失败时状态保存一并回滚，任务不 + # 以“已推回状态 + failed 任务行”的半截态落地。 + _insert_task(self.db_path, "TP", status="failed", audit_loop=1) + worker = self._patched_worker("SET status = 'queued'") + task = worker.get_task("TP") + final, _, decision = worker._audit_loop_wire( + task, "failed", {"stdout": ""}, "boom") + self.assertEqual(decision, "error") + self.assertEqual(final, "failed") + row = _row(self.db_path, "TP") + self.assertEqual(row[0], "failed") + self.assertIsNone(row[1]) + self.assertIsNone(_state_row(self.db_path, "TP")) + self.assertNotIn("worker_audit_loop_pushback", self._actions()) + + def test_escalate_store_failure_claims_nothing(self): + # I-14②（改动后新行为）：升级 UPDATE 失败时接线返回 + # error、任务保持原终局、状态行不动，且不记“已升级” + # 审计——升级审计只在提交后记，绝不谎称已升级。 + _insert_task(self.db_path, "TE", status="failed", audit_loop=1) + _seed_state(self.db_path, "TE", 3, "pushback") + worker = self._patched_worker("SET status = 'failed'") + task = worker.get_task("TE") + final, _, decision = worker._audit_loop_wire( + task, "failed", {"stdout": ""}, "boom") + self.assertEqual(decision, "error") + self.assertEqual(final, "failed") + self.assertEqual(_row(self.db_path, "TE")[0], "failed") + state = _state_row(self.db_path, "TE") + self.assertEqual(state[0], 3) + self.assertEqual(state[1], "pushback") + self.assertNotIn("worker_audit_loop_escalated", self._actions()) + + def test_escalate_cas_miss_audits_skipped(self): + # N-2：升级 CAS 未命中（状态已被改动）时记 + # escalate_skipped 审计、不谎称已升级，任务与状态原样不动。 + _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") + worker = self._worker() + task = worker.get_task("TC") + final, _, decision = worker._audit_loop_wire( + task, "failed", {"stdout": ""}, "boom") + self.assertEqual(decision, "error") + self.assertEqual(final, "failed") + 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", + self._actions()) + self.assertNotIn("worker_audit_loop_escalated", self._actions()) + + +class CoordClearTestCase(_WorkerCase): + def test_clear_packets_failure_is_best_effort(self): + # I-14③：清包失败只记警告、不炸调用方、不改已定结局。 + _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() + worker._coord_clear_packets( + {"id": "PP", "project_id": "p1"}, ["CC"]) + self.assertIsNone(_row(self.db_path, "PP")[1]) + self.assertIsNone(_row(self.db_path, "CC")[1]) + + 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"]) + + +class ClassifyAndSplitTestCase(_WorkerCase): + def test_merge_failed_is_gate_rejected(self): + # N-3：merge-failed 归 gate_rejected，不再是 + # excluded/non_terminal；容忍声明仍优先。 + worker = self._worker() + kind, reason = worker._coord_classify_child( + None, {"id": "C1", "status": "merge-failed"}, set()) + self.assertEqual((kind, reason), ("gate_rejected", "merge_failed")) + kind, _ = worker._coord_classify_child( + {"on_failure": "tolerate"}, + {"id": "C2", "status": "merge-failed"}, set()) + self.assertEqual(kind, "excluded") + kind, _ = worker._coord_classify_child( + None, {"id": "C3", "status": "merged"}, set()) + self.assertEqual(kind, "completed") + + def test_wire_splits_subjective_before_evaluate(self): + # N-4：单任务接线先拆分再评估——确定性照常推回，主观 + # 只记审计，不吞推回。 + _insert_task(self.db_path, "TS", status="failed", audit_loop=1) + worker = self._worker() + det = Finding("execution_failed", "high", "boom", "re-run", + "task completes", kind="deterministic", + gate="execution") + subj = Finding("design_note", "low", "looks odd", "review", + "human ok", kind="subjective", gate=None) + worker._audit_loop_gate_results = ( + lambda task, final, result, gates_checked=None: ( + {"execution": False}, [det, subj])) + task = worker.get_task("TS") + final, _, decision = worker._audit_loop_wire( + task, "failed", {"stdout": ""}, "boom") + self.assertEqual(decision, "pushback") + self.assertEqual(final, "queued") + needs = [d for _, a, d in self.audits + if a == "worker_audit_loop_needs_review"] + self.assertTrue(any("design_note" in d for d in needs)) + + +class TerminalSqlTestCase(unittest.TestCase): + def test_terminal_sql_matches_old_literal_set(self): + # N-6：生成的终态串集合与旧手写字面量逐字等价。 + from cloud.worker import _terminal_status_sql + sql = _terminal_status_sql() + inner = sql.strip()[1:-1] + got = {part.strip().strip("'") for part in inner.split(",")} + self.assertEqual( + got, + {"completed", "failed", "interrupted", "merged", + "merge-failed"}) + + +class GoalWireTestCase(_WorkerCase): + def test_goal_pushback_leaves_goal_columns_and_bounds_rounds(self): + # N-12 定性：推回不改 goal 列；包贴进内存 goal 与 prompt； + # 审计轮次 3 轮后必升级（无无界循环）。 + _insert_task(self.db_path, "TG", status="failed", audit_loop=1, + goal_mode=1, goal="reach the goal", + verify_command="true", goal_max_rounds=3) + worker = self._worker() + task = worker.get_task("TG") + final, _, decision = worker._audit_loop_wire( + task, "failed", {"goal_met": False, "goal_error": "nope"}, + "goal unmet") + self.assertEqual(decision, "pushback") + self.assertEqual(final, "queued") + row = _row(self.db_path, "TG") + self.assertEqual(row[2], "reach the goal") + self.assertEqual(row[3], "true") + self.assertEqual(row[4], 3) + packet = row[1] + self.assertIsNotNone(packet) + injected = worker._apply_audit_loop_packet({ + "id": "TG", "project_id": "p1", "prompt": "do work", + "goal": "reach the goal", "goal_mode": 1, + "audit_loop_packet": packet}) + self.assertTrue(injected["prompt"].startswith("[审计推回包")) + self.assertTrue(injected["goal"].startswith("[审计推回包")) + # 库里的原 prompt/goal 不动。 + self.assertEqual(_row(self.db_path, "TG")[2], "reach the goal") + # 继续推回直至升级：轮次有界。 + decisions = [] + for _ in range(3): + _set_status(self.db_path, "TG", "failed") + task = worker.get_task("TG") + _, _, decision = worker._audit_loop_wire( + task, "failed", + {"goal_met": False, "goal_error": "nope"}, "goal unmet") + decisions.append(decision) + if decision == "escalate": + break + self.assertEqual(decisions[-1], "escalate") + self.assertEqual(_row(self.db_path, "TG")[0], "failed") + self.assertIsNone(_row(self.db_path, "TG")[1]) + + +class ToolApprovalEffectiveTestCase(_WorkerCase): + def _insert_approval_task(self, task_id, approvals): + _insert_task(self.db_path, task_id, status="queued") + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET agent_tools = ?," + " agent_tool_approvals = ? WHERE id = ?", + (json.dumps(["websearch"]), json.dumps(approvals), + task_id)) + conn.commit() + finally: + conn.close() + + def test_ask_is_audited_as_decline_under_host_deny(self): + # I-12：派发时真的调 effective，ask 在行里折算成 + # decline，ask_honoured 恒为 False。 + self._insert_approval_task("TA", {"websearch": "ask"}) + worker = self._worker() + out = worker.run_one() + self.assertEqual(out["status"], "completed") + details = [d for _, a, d in self.audits + if a == "worker_tool_approvals_declared"] + self.assertTrue(details) + self.assertIn('"websearch"', details[0]) + self.assertIn('"declared": "ask"', details[0]) + self.assertIn('"effective": "decline"', details[0]) + self.assertIn('"ask_honoured": false', details[0]) + + def test_auto_accepts_only_when_host_config_is_auto(self): + # host approval 取自传给 host 的 --config 文件：auto 时 + # auto 声明 effective 为 accept；默认 deny 时为 decline。 + self._insert_approval_task("TB", {"websearch": "auto"}) + cfg = os.path.join(self.tmp.name, "host-config.json") + with open(cfg, "w", encoding="utf-8") as fh: + json.dump({"approval": "auto"}, fh) + worker = self._worker(config_path=cfg) + out = worker.run_one() + self.assertEqual(out["status"], "completed") + details = [d for _, a, d in self.audits + if a == "worker_tool_approvals_declared"] + self.assertIn('"effective": "accept"', details[0]) + + +class GatewayExtractTestCase(_WorkerCase): + def _policy(self, capable): + return {"mode": "active", + "tiers": {"default": { + "model": "m1", + "requires": {"tool_calling": True}}}, + "model_capabilities": {"m1": {"tool_calling": capable}}} + + def test_blocked_and_pass_through(self): + # I-9①：抽取后行为逐字不变——被拦时早返回 dict + + # 门审计 + 标记点；满足时返回 None 继续派发。 + _insert_task(self.db_path, "TG1", status="queued") + worker = self._worker(routing_policy=self._policy(False)) + task = {"id": "TG1", "project_id": "p1", "model": "m1"} + blocked = worker._gateway_capability_check( + task, "sub2api", False, time.time()) + self.assertIsNotNone(blocked) + self.assertEqual(blocked["status"], "failed") + self.assertIn("worker_capability_gate_rejected", self._actions()) + self.assertIn("capability", task.get("_gates_checked", [])) + + self.audits.clear() + _insert_task(self.db_path, "TG2", status="queued") + worker2 = self._worker(routing_policy=self._policy(True)) + task2 = {"id": "TG2", "project_id": "p1", "model": "m1"} + self.assertIsNone(worker2._gateway_capability_check( + task2, "sub2api", False, time.time())) + self.assertNotIn("worker_capability_gate_rejected", + self._actions()) + + +class PersistExtractTestCase(_WorkerCase): + def test_writeback_only_for_applied_policy(self): + # I-9②：只持久化 source=policy 且 applied 的决策。 + _insert_task(self.db_path, "TW", status="queued", model=None) + worker = self._worker() + task = {"id": "TW", "project_id": "p1"} + worker._persist_policy_model( + task, SimpleNamespace(chosen_model="m9", source="policy", + applied=True, tier="default")) + self.assertEqual(_row(self.db_path, "TW")[5], "m9") + self.assertIn("worker_model_persisted", self._actions()) + self.audits.clear() + _insert_task(self.db_path, "TW2", status="queued", model=None) + worker._persist_policy_model( + {"id": "TW2", "project_id": "p1"}, + SimpleNamespace(chosen_model="m9", source="shadow", + applied=False, tier="default")) + self.assertIsNone(_row(self.db_path, "TW2")[5]) + self.assertNotIn("worker_model_persisted", self._actions()) + + +class ServerHelperTestCase(unittest.TestCase): + def test_agent_definition_and_audit_loop_helpers(self): + # I-9③：两个小函数与原地行为一致（错误文案逐字）。 + resolve = CloudControlPlaneHandler._resolve_create_agent_definition + _, err = resolve({"agent": {"a": 1}, "agent_name": "x"}) + self.assertEqual(err, "agent and agent_name cannot both be used") + _, err = resolve({"agent": "not-a-dict"}) + self.assertEqual(err, "agent must be an object") + validate = CloudControlPlaneHandler._validate_create_audit_loop + self.assertEqual(validate({}), (0, None)) + self.assertEqual(validate({"audit_loop": True}), (1, None)) + self.assertEqual(validate({"audit_loop": 1}), (1, None)) + _, err = validate({"audit_loop": 2}) + self.assertEqual(err, "audit_loop must be true or 1") + + +class ParseToleranceTestCase(unittest.TestCase): + def test_non_string_items_survive_and_are_tolerated(self): + # N-9（文档选项）：解析层只保 list 形状，下游容忍非 + # 字符串项——不过滤、不改旧脏行序列化。 + self.assertEqual(parse_agent_tools_column('["a", 1]'), + ["a", 1]) + self.assertEqual(missing_required_tools(["a", 1], ""), + ["a", 1]) + + +class DryRunEmptyTestCase(unittest.TestCase): + def test_early_returns_share_shape(self): + # N-10：两处早期返回经 _empty_report 归一，键集合一致。 + a = dry_run_policy_document("not-a-dict") + b = dry_run_policy_document({"mode": "bogus"}) + self.assertFalse(a["valid"]) + self.assertFalse(b["valid"]) + self.assertEqual(set(a.keys()), set(b.keys())) + + +class ShadowEndpointTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + status, self.project = self._req( + "POST", "/projects", {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201) + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection("127.0.0.1", self.port, + timeout=10) + headers = {"Content-Type": "application/json"} + if key: + headers["X-API-Key"] = key + conn.request(method, path, + json.dumps(body) if body is not None else None, + headers) + resp = conn.getresponse() + raw = resp.read() + conn.close() + try: + data = json.loads(raw) + except ValueError: + data = None + return resp.status, data + + def _shadow(self): + status, report = self._req( + "GET", "/routing-shadow?project_id=" + self.project["id"], + key=self.owner_key) + self.assertEqual(status, 200, report) + return report + + def test_cost_query_failure_flagged_not_500(self): + # N-1 + I-14④：成本查询失败时报告仍 200、成本按空算、 + # cost_unavailable 标明，不抛到 handler 层。 + conn = sqlite3.connect(self.db_path) + try: + conn.execute("DROP TABLE IF EXISTS cost_records") + conn.commit() + finally: + conn.close() + + class _BoomCostStore: + def __init__(self, *args, **kwargs): + raise RuntimeError("injected cost store failure") + + with mock.patch.object(server_module, "CostStore", + _BoomCostStore): + report = self._shadow() + self.assertTrue(report["cost_unavailable"]) + + def test_sample_window_and_normal_cost_flag(self): + # N-5：截断样本（最早 10000 行）附带最早/最晚时间窗； + # 正常路径 cost_unavailable 为 False。 + conn = sqlite3.connect(self.db_path) + try: + for task_id, created in (("t1", 10.0), ("t2", 20.0)): + conn.execute( + "INSERT INTO audit_log(user_id, project_id, action," + " detail, created_at) VALUES(?, ?, ?, ?, ?)", + ("system", self.project["id"], "worker_routing", + "%s tier=plan source=shadow applied=False" + " suggested=mx" % task_id, created)) + conn.commit() + finally: + conn.close() + report = self._shadow() + self.assertFalse(report["cost_unavailable"]) + self.assertEqual(report["sample_window"], + {"earliest_created_at": 10.0, + "latest_created_at": 20.0}) + self.assertFalse(report["truncated"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_audit_loop_wire.py b/cloud/tests/test_audit_loop_wire.py index eed2810..b91bca0 100644 --- a/cloud/tests/test_audit_loop_wire.py +++ b/cloud/tests/test_audit_loop_wire.py @@ -232,9 +232,13 @@ class WireTestCase(unittest.TestCase): def test_needs_review_leaves_outcome_untouched(self): _insert_task(self.db_path, "T6", status="failed", audit_loop=1) worker = self._worker() - worker.audit_loop_evaluate = lambda t, g, findings=None, fixed=None: { - "enabled": True, "decision": "needs_review", "packet": None, - "new_state": None, "reason": "subjective"} + # 审计批 5：接线改走 _audit_loop_evaluate_on_conn（同连接 + # 单事务），桩点随之移到新内核；意图不变——防御分支仍在， + # needs_review 时终局、包列一律不动、只记审计。 + worker._audit_loop_evaluate_on_conn = ( + lambda conn, t, g, findings=None, fixed=None: { + "enabled": True, "decision": "needs_review", "packet": None, + "new_state": None, "reason": "subjective"}) task = worker.get_task("T6") final, _, decision = worker._audit_loop_wire( task, "failed", {"stdout": ""}, "boom") diff --git a/cloud/worker.py b/cloud/worker.py index 5556724..425416f 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -135,6 +135,7 @@ import uuid from types import SimpleNamespace from cloud.agent_def import AgentDefinition as _AgentDefinition +from cloud.agent_def import effective_tool_approvals as _effective_tool_approvals from cloud.agent_def import format_instructions_header as _agent_header from cloud.agent_def import missing_required_tools as _missing_tools from cloud.agent_def import parse_agent_tool_approvals_column as _parse_tool_approvals_col @@ -227,6 +228,20 @@ from cloud.workers import heartbeat_worker, register_worker # 与 server.TERMINAL_STATUSES 对齐：这些状态的任务不再被认领或取消。 TERMINAL_STATUSES = frozenset( {"completed", "failed", "interrupted", "merged", "merge-failed"}) + + +def _terminal_status_sql(): + """终态集合的 SQL IN 串（审计批 5，N-6）。 + + worker 内多处“不在终态才改状态”的 SQL 原先各自手写同一串 + 字面量，改一处漏一处就会口径分裂；现在统一从 + TERMINAL_STATUSES 生成。排序只为输出稳定，SQL 的 IN 是 + 集合语义，顺序不影响结果；测试断言生成串解析出的集合与 + 旧字面量集合逐字等价。其它模块（server、coordinator）各 + 自的 TERMINAL_STATUSES 常量本批不动，只统一 worker 内 SQL。 + """ + return "(%s)" % ", ".join( + "'%s'" % status for status in sorted(TERMINAL_STATUSES)) COMPLETED_STATUSES = frozenset({"completed", "merged"}) DEFAULT_ENV_KEY = "XCM_API_KEY" # 默认模型（2026-10-27 起为 gpt-5.6-terra）：此前默认的 @@ -962,8 +977,9 @@ class Worker: （已被别的 worker 抢走）时，只清掉本次新建的占位，绝 不动已在跑任务的条目。占位条目在执行路径的 finally 里清（本地 _execute_via_process / Docker 同步路径）， - 认领后提前返回的三道门（缺 key / 仓库 / 配额）在 - run_one 里显式清占位；CAS 语义本身一句未改。 + 认领后提前返回的门（缺 key、仓库、配额、goal 轮次上限、 + 网关准备、网关能力、路由能力、工具审批、必需工具、 + 写回复核等）在 run_one 里显式清占位；CAS 语义本身一句未改。 P1 签名门：占位之前先比对多文件签名，签名缺失或已变 时直接返回 False（fail-closed，不抢任务、不占槽）， 并记日志；不要只在 run_loop 轮次边界查——并行池线程 @@ -1868,11 +1884,55 @@ class Worker: except Exception: pass + def _audit_loop_evaluate_on_conn(self, conn, task, gate_results, + findings=None, fixed=None): + """在调用方给定的连接上跑评估并写状态（不提交、不关连接）。 + + 审计批 5（事务分离）：本方法是 audit_loop_evaluate 与 + _audit_loop_wire 共用的内核——前者自带连接、写完提交， + 后者用同一连接把“状态保存、包写、终局更新”一次提交， + 不再出现“状态已提交为 escalate、终局更新却没落库”的 + 自相矛盾。返回的决策 dict 与引擎原样一致（packet / + new_state 仍为对象，未转 dict）；失败类决策另带内部键 + “_error”（损坏/评估失败的错误文本），调用方在提交后 + 用它写升级审计，公开返回前必须 strip 掉（见 + audit_loop_evaluate）。评估域内的异常一律收敛为 + evaluate_failed 升级决策（与旧 audit_loop_evaluate 的 + except 分支同款），不向调用方抛；连接级异常（提交失败 + 等）仍由调用方处理。 + """ + task_id = task.get("id") + if not task.get("audit_loop"): + return {"enabled": False, "decision": "off", "packet": None, + "new_state": None, "reason": None} + try: + _ensure_audit_loop(conn) + state, corrupted, error = _audit_loop_load(conn, task_id) + if corrupted: + return {"enabled": True, "decision": "escalate", + "packet": None, "new_state": None, + "reason": "count_corrupted", "_error": error} + 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"]) + return decision + except Exception as exc: + # 评估或写状态失败：轮次不可信，fail-closed 到升级。 + try: + conn.rollback() + except Exception: + pass + return {"enabled": True, "decision": "escalate", + "packet": None, "new_state": None, + "reason": "evaluate_failed", "_error": str(exc)} + def audit_loop_evaluate(self, task, gate_results, findings=None, fixed=None): """有界审计循环的显式入口（默认关闭，绝不由 run_one 自动调用）。 - 启用条件二选一中只选其一：任务字典含 audit_loop 真值。 + 启用条件只有一个：任务字典含 audit_loop 真值。 没有 env 开关——全局 env 会让所有任务悄悄启用循环，本方法 刻意拒绝它。开关关闭时返回 enabled=False 且不碰数据库， run_one、auto_resume、协调员、goal 断点行为逐字节不变。 @@ -1889,6 +1949,11 @@ class Worker: fail-closed 到 escalate，不进循环。 gate_results 必须含本轮全部已跑的门；上一轮通过的门本轮 缺跑或变失败时，引擎按 gate_missing / regression 升级或推回。 + 调用方契约（审计批 5，N-4）：主观发现绝不直接进本方法—— + 调用方须先用 _coord_split_findings 拆分，确定性进循环， + 主观只记审计；协调员合流与单任务接线均已如此。 + 内部实现走 _audit_loop_evaluate_on_conn（自带连接、写完 + 提交），与单任务接线共用同一内核。 """ task = task if isinstance(task, dict) else {} if not task.get("audit_loop"): @@ -1909,45 +1974,41 @@ class Worker: 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) + decision = self._audit_loop_evaluate_on_conn( + conn, task, 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"} + try: + conn.commit() + except Exception as exc: + try: + conn.rollback() + except Exception: + pass + decision = {"enabled": True, "decision": "escalate", + "packet": None, "new_state": None, + "reason": "evaluate_failed", + "_error": str(exc)} finally: try: conn.close() except Exception: pass - if decision["decision"] == "escalate": + if decision["reason"] == "count_corrupted": + try: + self._audit( + project_id, "worker_audit_loop_escalated", + "%s reason=count_corrupted error=%s" + % (task_id, decision.get("_error"))) + except Exception: + pass + elif decision["reason"] == "evaluate_failed": + try: + self._audit(project_id, "worker_audit_loop_escalated", + "%s reason=evaluate_failed error=%s" + % (task_id, decision.get("_error"))) + except Exception: + pass + elif decision["decision"] == "escalate": try: self._audit( project_id, "worker_audit_loop_escalated", @@ -1955,7 +2016,8 @@ class Worker: % (task_id, decision["reason"], decision["round"])) except Exception: pass - result = dict(decision) + result = {k: v for k, v in decision.items() + if not k.startswith("_")} result["enabled"] = True if result.get("packet") is not None: result["packet"] = result["packet"].to_dict() @@ -2054,14 +2116,48 @@ class Worker: - escalate：清掉推回包列，状态置 failed（合法终态； VALID_STATUSES 里没有 needs_review，绝不发明它）， error_text 写明“审计循环升级”+ 轮次 + 原因 + 上一轮 - 错误摘要，之后不再自动重排。引擎已记 - worker_audit_loop_escalated 审计，这里不重复记。 - - needs_review：防御分支（接线从不产生主观发现，正常 - 流程到不了这里）：只记 worker_audit_loop_needs_review + 错误摘要，之后不再自动重排。升级审计在事务提交后记， + 与旧接线“引擎已记、这里不重复记”的审计行一模一样， + 只是时序从“评估提交后”挪到“接线提交后”。 + - needs_review：防御分支（接线只产生确定性发现，正常 + 流程到不了这里；审计批 5（N-4）起接线先拆分主观发现 + 只记审计再评确定性，所以 needs_review 更到不了）： + 只记 worker_audit_loop_needs_review 审计，终局、包列一律不动，主观审计永不自动循环。 - off / skipped：未启用或协调员父任务，不碰库。 - 包列写失败、CAS 失败时 fail-closed 回原终局并记警告， - 绝不谎称已推回或已升级。 + 审计批 5（事务分离 + N-2）：状态保存、包写、终局更新在 + 同一连接、同一事务里一次提交——评估内核走 + _audit_loop_evaluate_on_conn（只写不提交），分支写完 + 才 commit；事务内任何一步失败（包括升级/推回的 CAS + 未命中）即整体回滚，任务不以“已升级状态 + completed + 任务行”的矛盾状态落地。pass 的清包与 escalate 的置 + failed 同样带 AND status=<终局> CAS（旧版 pass 清包与 + escalate 更新无 CAS）；CAS 未命中时记警告/审计（升级 + 未命中另记 worker_audit_loop_escalate_skipped 审计）， + fail-closed 回原终局，绝不谎称已推回或已升级。 + 例外：接线连库都连不上时直接返回 error、任务保持原 + 终局（旧版评估连库失败会返回 state_unreadable 升级、 + 接线另开连接置 failed；单事务下无连接即无决策可提交， + 不谎称升级）。升级成功后结果文本写回（_write_execution_ + result，自带连接）仍是提交后的独立写：它失败时只记 + 警告、不再把已 failed 的任务谎报成 error（旧版写回 + 失败会返回 error，尽管状态已是 failed）。 + goal 任务（审计批 5，N-12 定性，只文档化、不改行为）： + 推回只改 tasks.status 与包列（+ 轮次状态表），不碰 + goal / verify_command / goal_max_rounds 列，也不碰宿主 + goal 状态文件——goal_rounds_done 根本不在任务表里， + 接线既无权重置它、也不由接线延续它。下一轮正常认领 + 时包前言贴进内存 prompt 与 goal（_apply_audit_loop_ + packet 对 goal 任务同步贴两者），但 goal 任务下一轮 + 实际会撞宿主任务名已存在、经 run_one 的已存在回落走 + 断点续跑（goal-resume 不传 --goal-text，宿主用存储的 + goal 与 goal_rounds_done+1 继续），所以包前言贴了却 + 不进续跑循环——这是审计推回对 goal 任务的已知效果 + 缺口，不是轮次会计错误，本批不改行为、留待后续批 + 定夺。轮次会计本身有界：审计轮次硬上限 3（MAX_ + PUSHBACKS），3 轮后必升级；goal 旧形态总回合数受 + goal_max_rounds 约束（续跑不重置计数），新形态受配额 + 约束；两组计数器独立、不叠加成无界循环。 """ task_id = task.get("id") project_id = task.get("project_id") @@ -2075,124 +2171,184 @@ class Worker: # 不再硬编码成通过。未传（旧直接调用）时保持 # 旧行为，见 _audit_loop_gate_results 文档。 task, final, result, task.get("_gates_checked")) + # 审计批 5（N-4）：先拆分再评估，与协调员合流对称。 + # 单任务接线目前只产生确定性发现，拆分在生产路径上 + # 不改变决策；主观发现只记审计，不进循环、不吞推回。 + deterministic, subjective = self._coord_split_findings(findings) + if subjective: + self._audit(project_id, "worker_audit_loop_needs_review", + "%s subjective=%s" % ( + task_id, + ",".join(f.id for f in subjective))) try: - decision = self.audit_loop_evaluate( - task, gate_results, findings=findings) + conn = _db_connect(self.db_path) except Exception as exc: try: log_event(self.logger, "worker_audit_loop_wire_failed", task_id=task_id, project_id=project_id, status="warning", - message="audit loop evaluate failed: %s" % exc) + message="audit loop wire connect failed: %s" % exc) except Exception: pass return final, error, "error" - kind = decision.get("decision") - if kind == "off": - return final, error, "off" - if kind == "needs_review": - self._audit(project_id, "worker_audit_loop_needs_review", - "%s" % task_id) - return final, error, "needs_review" - if kind == "pass": + try: + decision = self._audit_loop_evaluate_on_conn( + conn, task, gate_results, findings=deterministic) + kind = decision.get("decision") + if kind == "off": + return final, error, "off" + if kind == "needs_review": + self._audit(project_id, "worker_audit_loop_needs_review", + "%s" % task_id) + return final, error, "needs_review" + if kind == "pass": + self._ensure_result_columns(conn) + cur = conn.execute( + "UPDATE tasks SET audit_loop_packet = NULL," + " updated_at = ? WHERE id = ? AND status = ?", + (time.time(), task_id, final)) + if cur.rowcount != 1: + # 审计批 5（N-2）：清包同样带终局 CAS——状态 + # 已被改动时不清包（可能已是新一轮的包）， + # 整个事务回滚（状态保存亦不落库），不谎称 pass。 + conn.rollback() + try: + log_event( + self.logger, + "worker_audit_loop_packet_clear_skipped", + task_id=task_id, project_id=project_id, + status="warning", + message="task status changed before packet clear") + except Exception: + pass + return final, error, "error" + conn.commit() + self._audit(project_id, "worker_audit_loop_passed", + "%s round=%s" % (task_id, decision.get("round"))) + return final, error, "pass" + if kind == "pushback": + packet = decision.get("packet") + if packet is None: + packet_dict = {} + elif hasattr(packet, "to_dict"): + packet_dict = packet.to_dict() + else: + packet_dict = packet + self._ensure_result_columns(conn) + cur = conn.execute( + "UPDATE tasks SET status = 'queued'," + " audit_loop_packet = ?, updated_at = ?" + " WHERE id = ? AND status = ?", + (json.dumps(packet_dict), time.time(), task_id, final)) + if cur.rowcount != 1: + # CAS 未命中即整个事务回滚：状态保存亦不落库， + # 不出现“状态已推回、任务仍是终局”的半截态。 + conn.rollback() + try: + log_event( + self.logger, + "worker_audit_loop_pushback_skipped", + task_id=task_id, project_id=project_id, + status="warning", + message="task status changed before pushback") + except Exception: + pass + return final, error, "error" + conn.commit() + self._audit(project_id, "worker_audit_loop_pushback", + "%s round=%s findings=%s" % ( + task_id, decision.get("round"), + ",".join( + f["id"] + for f in packet_dict.get("findings", [])))) + return "queued", error, "pushback" + if kind == "escalate": + esc_error = ( + "审计循环升级（第 %s 轮，原因 %s）：%s" + % (decision.get("round"), decision.get("reason"), + (error or "未通过确定性门")[:500])) + self._ensure_result_columns(conn) + cur = conn.execute( + "UPDATE tasks SET status = 'failed'," + " audit_loop_packet = NULL, updated_at = ?" + " WHERE id = ? AND status = ?", + (time.time(), task_id, final)) + if cur.rowcount != 1: + # 审计批 5（N-2）：升级同样带终局 CAS；未命中 + # 时记审计、不谎称已升级，整个事务回滚。 + conn.rollback() + self._audit( + project_id, "worker_audit_loop_escalate_skipped", + "%s status changed before escalate" % task_id) + try: + log_event( + self.logger, + "worker_audit_loop_escalate_skipped", + task_id=task_id, project_id=project_id, + status="warning", + message="task status changed before escalate") + except Exception: + pass + return final, error, "error" + conn.commit() + # 升级审计在提交后记（与旧接线“引擎已记”的行同款， + # 只是时序从评估提交后挪到接线提交后，绝不出现 + # “审计写着已升级、任务行却仍是终局”的矛盾）。 + if decision.get("reason") == "count_corrupted": + self._audit( + project_id, "worker_audit_loop_escalated", + "%s reason=count_corrupted error=%s" + % (task_id, decision.get("_error"))) + elif decision.get("reason") == "evaluate_failed": + self._audit( + project_id, "worker_audit_loop_escalated", + "%s reason=evaluate_failed error=%s" + % (task_id, decision.get("_error"))) + else: + self._audit( + project_id, "worker_audit_loop_escalated", + "%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: + # 结果文本写回是提交后的独立写：失败只记警告， + # 任务已是 failed，不再谎报成 error。 + try: + log_event( + self.logger, + "worker_audit_loop_escalate_result_failed", + task_id=task_id, project_id=project_id, + status="warning", + message="escalate result write failed: %s" % exc) + except Exception: + pass + return "failed", esc_error, "escalate" + conn.rollback() + return final, error, "error" + except Exception as exc: try: - conn = _db_connect(self.db_path) - try: - self._ensure_result_columns(conn) - conn.execute( - "UPDATE tasks SET audit_loop_packet = NULL," - " updated_at = ? WHERE id = ?", - (time.time(), task_id)) - conn.commit() - finally: - conn.close() - except Exception as exc: - try: - log_event(self.logger, - "worker_audit_loop_packet_clear_failed", - task_id=task_id, project_id=project_id, - status="warning", - message="clear packet failed: %s" % exc) - except Exception: - pass - self._audit(project_id, "worker_audit_loop_passed", - "%s round=%s" % (task_id, decision.get("round"))) - return final, error, "pass" - if kind == "pushback": - packet = decision.get("packet") or {} + conn.rollback() + except Exception: + pass try: - conn = _db_connect(self.db_path) - try: - self._ensure_result_columns(conn) - cur = conn.execute( - "UPDATE tasks SET status = 'queued'," - " audit_loop_packet = ?, updated_at = ?" - " WHERE id = ? AND status = ?", - (json.dumps(packet), time.time(), task_id, final)) - conn.commit() - moved = cur.rowcount == 1 - finally: - conn.close() - except Exception as exc: - try: - log_event(self.logger, - "worker_audit_loop_pushback_failed", - task_id=task_id, project_id=project_id, - status="warning", - message="pushback store failed: %s" % exc) - except Exception: - pass - return final, error, "error" - if not moved: - try: - log_event(self.logger, - "worker_audit_loop_pushback_skipped", - task_id=task_id, project_id=project_id, - status="warning", - message="task status changed before pushback") - except Exception: - pass - return final, error, "error" - self._audit(project_id, "worker_audit_loop_pushback", - "%s round=%s findings=%s" % ( - task_id, decision.get("round"), - ",".join( - f["id"] - for f in packet.get("findings", [])))) - return "queued", error, "pushback" - if kind == "escalate": - esc_error = ( - "审计循环升级（第 %s 轮，原因 %s）：%s" - % (decision.get("round"), decision.get("reason"), - (error or "未通过确定性门")[:500])) + log_event(self.logger, "worker_audit_loop_wire_failed", + task_id=task_id, project_id=project_id, + status="warning", + message="audit loop wire store failed: %s" % exc) + except Exception: + pass + return final, error, "error" + finally: try: - conn = _db_connect(self.db_path) - try: - self._ensure_result_columns(conn) - conn.execute( - "UPDATE tasks SET status = 'failed'," - " audit_loop_packet = NULL, updated_at = ?" - " WHERE id = ?", (time.time(), task_id)) - conn.commit() - finally: - conn.close() - 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: - try: - log_event(self.logger, - "worker_audit_loop_escalate_failed", - task_id=task_id, project_id=project_id, - status="warning", - message="escalate store failed: %s" % exc) - except Exception: - pass - return final, error, "error" - return "failed", esc_error, "escalate" - return final, error, "error" + conn.close() + except Exception: + pass def _notify_task_finished(self, task_id, project_id, event_type, error=None): @@ -2939,9 +3095,8 @@ class Worker: try: cur = conn.execute( "UPDATE tasks SET status = 'interrupted', updated_at = ?" - " WHERE id = ? AND status NOT IN" - " ('completed', 'failed', 'interrupted', 'merged'," - " 'merge-failed')", + " WHERE id = ? AND status NOT IN " + + _terminal_status_sql(), (time.time(), task_id), ) conn.commit() @@ -5946,9 +6101,8 @@ class Worker: cur = conn.execute( "UPDATE tasks SET status = 'interrupted'," " interrupted_reason = 'coordinator_orphan'," - " updated_at = ? WHERE id = ? AND status NOT IN" - " ('completed', 'failed', 'interrupted', 'merged'," - " 'merge-failed')", + " updated_at = ? WHERE id = ? AND status NOT IN " + + _terminal_status_sql(), (time.time(), ch["id"])) if cur.rowcount == 1: marked.append(ch["id"]) @@ -6096,9 +6250,8 @@ class Worker: try: rows = conn.execute( "SELECT id, parent_task_id, project_id FROM tasks" - " WHERE parent_task_id IS NOT NULL AND status NOT IN" - " ('completed', 'failed', 'interrupted', 'merged'," - " 'merge-failed')").fetchall() + " WHERE parent_task_id IS NOT NULL AND status NOT IN " + + _terminal_status_sql()).fetchall() parents = {} for row in rows: try: @@ -6128,9 +6281,8 @@ class Worker: cur = conn.execute( "UPDATE tasks SET status = 'interrupted'," " interrupted_reason = 'coordinator_orphan'," - " updated_at = ? WHERE id = ? AND status NOT IN" - " ('completed', 'failed', 'interrupted', 'merged'," - " 'merge-failed')", + " updated_at = ? WHERE id = ? AND status NOT IN " + + _terminal_status_sql(), (time.time(), cid)) if cur.rowcount == 1: marked.append((pid, cid)) @@ -6155,6 +6307,10 @@ class Worker: 再调 evaluate：确定性进循环，主观只记审计。返回 (deterministic, subjective)。本方法本身不产生发现， 合流代码产生的发现全是确定性的，拆分是防御层。 + 审计批 5（N-4）：单任务接线 _audit_loop_wire 同样先拆分 + 再调 evaluate（对称；单任务接线目前只产生确定性发现， + 拆分在生产路径上不改变决策，只防未来某处混入主观发现 + 时把确定性推回吞掉）。 """ deterministic = [] subjective = [] @@ -6326,6 +6482,8 @@ class Worker: completed：用户取消等 excluded 情形下，审计 pass 只表示“无可重做的确定性失败”，父任务终局仍由 overall 定（取消的子任务仍会使 overall 为 failed）。 + 审计批 5（N-3）：merge-failed 归 gate_rejected（见 + 分类体内的注释），不再落入 excluded/non_terminal。 """ status = (row or {}).get("status") if status == "completed" or status == "merged": @@ -6337,6 +6495,21 @@ class Worker: return "excluded", "user_cancelled" if row.get("interrupted_reason") == "coordinator_orphan": return "excluded", "orphaned" + # 审计批 5（N-3）：merge-failed 明确归 gate_rejected， + # 不再是 excluded/non_terminal。理由：①它是确定性终态—— + # agent 的工作已做完，卡在合入环节（冲突/目标分支状态）， + # 重跑 agent 解决不了合入问题，修复方在运营方（手动解 + # 冲突、手动 /pr 重试——手动重排 SQL 本就另接受 + # merge-failed，与审计推回是两条路）；②审计推回的重排 + # SQL 只接受 ('failed', 'interrupted')，即使归 execution + # 也会被重排步静默跳过（rowcount 0 → pushback_skipped）， + # 父任务却已回 queued，等于一次无子任务重排的空推回； + # 归 gate_rejected 则父任务直接升级，不烧审计轮次。 + # 位置在门拒绝审计/标记检查之前：merge-failed 本身已是 + # 足够强的确定性证据，不必再靠审计行或 error 文案佐证； + # 容忍/取消/孤儿仍优先（运营方已声明的处置意图更强）。 + if status == "merge-failed": + return "gate_rejected", "merge_failed" child_id = row.get("id") if child_id in (gate_rejected_ids or set()): return "gate_rejected", "gate_audit" @@ -7160,6 +7333,227 @@ class Worker: " required capability" % (tier, model, ", ".join(known))) + def _gateway_capability_check(self, task, gateway_provider, + _explicit_model, started_at): + """网关能力检查门（审计批 5，I-9①抽取，行为逐字不变）。 + + 原 run_one 内整段平移：判定本身（evaluate_capability） + 已是纯函数，本方法是 worker 内唯一的一处接线调用—— + 这就是立约的新门形状：纯判定函数 + 一处接线调用； + coord_store 抽取是中长期项，本批只在注释里写明不做。 + 返回：被拦时返回 run_one 同款早返回 dict（dispatched + True、status failed、error 为门错误文本），否则返回 + None（调用方继续派发）。副作用点逐项核对（与原代码 + 一致，抽取后无增减）： + - 审计：显式模型记 worker_capability_gateway_explicit + （含 blocked 标志）；shadow 档记 + worker_capability_gateway_shadow；被拦时记 + worker_capability_gate_rejected（detail 为 + 任务 id 加错误格式）。 + - 标记：verdict 实际算出时 _mark_gate_checked(task, + “capability”)（审计批 4，I-10 标记点，位置不变）。 + - 状态：被拦时 _set_status(task_id, “failed”)。 + - 结果写回：被拦时 _write_execution_result（ + only_if_status 为“failed”）。 + - 清理：被拦时 _cleanup_task_seed + _running_clear。 + - 不改 task 的 model，不碰事件列表；无策略、无档需求、 + 档模式 off 时一律不查（旧行为不变）。 + """ + task_id = task.get("id") + # Audit fix：网关能力检查——此前路由决策只在模型 + # 为空时跑，sub2api 网关先填默认模型后决策直接跳 + # 过，能力门在生产路径上完全不生效。补齐：网关填 + # 的模型同样按档需求检查，档模式与 v8 一致：shadow + # 只审计不拦，active 不满足则 fail-fast（同款工具 + # 门形状，清占位、不调 runner）。没有策略、没有档 + # 需求、档模式 off 时一律不查，旧行为不变。注意： + # 这里没有 worker 默认可回落（网关模型就是该任务 + # 的网关内模型，换成 worker 默认会跑错网关），所以 + # active 下不满足即拦，不存在“先试 worker 默认” + # 的中间态；运营方须先把网关模型写进能力目录， + # 否则 unknown 同样拦（fail-closed，与 v8 一致）。 + # 例外：任务级显式模型按 Audit fix 4 分档——已知 + # 不满足才拦，unknown 只审计（见下），因为显式 + # 优先保护的是用户选了哪个模型，不是选的模型 + # 一定能跑；目录陈旧不应惩罚正确的显式选择。 + # 审计批 4（I-7）：显式模型的 known/unknown 分档检查 + # 与网关无关——网关分支只决定“待查模型是哪一个” + # （网关填充模型或显式模型），分档判定走第 3 批统一 + # 的 evaluate_capability。进入条件因此是“有策略且 + # （走了 sub2api 网关，或显式任务走了任一真实网关）”： + # 非显式 + 非 sub2api 的任务根本不进块（它们模型 + # 为空、后续走 decide_routing，不存在误拦）；显式 + # + 无网关（provider 为 None）的任务也不进块——它 + # 根本没走网关，显式语义不变（写回模型仍走后续 + # 复核，用户显式仍不被复核，见 fixes5 契约）； + # 显式 + sub2api 路径与原来逐字一致（同一审计 + # action、同一 error 文案、同一 blocked 标志）， + # 不会重复审计或重复拦；显式 + 非 sub2api 网关 + # 在 active 下已知不满足时从“直接派发”变为“拦”， + # 这是有意的行为变更。 + # 附带：已持久化的显式任务在非 sub2api 网关下会 + # 同时走本门与后续写回复核（本门不拦时复核再按 + # 分档审计/拦），两者审计 action 不同 + # （gateway_explicit 与 recheck_*），不是重复拦。 + if self.routing_policy is not None \ + and (gateway_provider == "sub2api" + or (_explicit_model + and gateway_provider is not None)): + _gw_tier = classify_task(task) + _gw_requires = self.routing_policy.tier_requirements.get( + _gw_tier) or {} + if _gw_requires: + _gw_mode = self.routing_policy.tier_mode(_gw_tier) + if _gw_mode in ("active", "shadow"): + _gw_verdict = evaluate_capability( + self.routing_policy.model_capabilities, + task.get("model"), _gw_requires) + # 审计批 4（I-10）：走到这里说明网关 + # 能力检查真正算了 verdict。 + _mark_gate_checked(task, "capability") + if not _gw_verdict.ok: + _gw_note = ("gateway model %s %s capability:" + " %s" % ( + task.get("model"), + "cannot verify" + if _gw_verdict.unknown + else "lacks", + ", ".join(_gw_verdict.missing))) + # Audit fix 4（第三遍审计分档建议）： + # 显式模型按 known/unknown 分档。known + # 是目录明确写了不满足（布尔 False、 + # 或 context 低于需求），与网关默认 + # 模型同款 fail-fast；unknown 是目录 + # 无该模型、或条目缺字段，目录陈旧 + # 不应惩罚用户的正确选择，只审计不拦。 + # shadow 档两种情况都只审计。 + _gw_known = _gw_verdict.known_missing + _gw_block = False + _gw_error = None + if _explicit_model: + _gw_blocked = ( + _gw_mode == "active" and bool(_gw_known)) + # 显式检查结果独立审计，含 blocked + # 标志：被拦的已知不满足也要进影子 + # 报告的显式失败清单，否则报告会 + # 漏掉恰恰最该被看到的一类。 + self._audit( + task["project_id"], + "worker_capability_gateway_explicit", + "%s model=%s tier=%s mode=%s" + " missing=%s unknown=%s blocked=%s" + % (task_id, task.get("model"), + _gw_tier, _gw_mode, + ",".join(_gw_verdict.missing) or "-", + _gw_verdict.unknown, + _gw_blocked)) + if _gw_blocked: + _gw_block = True + _gw_error = ( + "%s tier explicit model %s lacks" + " capability: %s; refusing to" + " dispatch a model known to lack" + " a required capability" + % (_gw_tier, task.get("model"), + ", ".join(_gw_known))) + elif _gw_mode == "shadow": + self._audit( + task["project_id"], + "worker_capability_gateway_shadow", + "%s tier=%s %s (shadow, not blocked)" + % (task_id, _gw_tier, _gw_note)) + else: + _gw_block = True + _gw_error = ( + "%s tier gateway model %s; refusing" + " to dispatch an incapable or" + " unverifiable model" + % (_gw_tier, _gw_note)) + if _gw_block: + self._set_status(task_id, "failed") + self._write_execution_result( + task_id, error_text=_gw_error, + actual_model=self.model, + started_at=started_at, + finished_at=time.time(), + only_if_status="failed") + self._audit( + task["project_id"], + "worker_capability_gate_rejected", + "%s: %s" % (task_id, _gw_error)) + log_event( + self.logger, + "worker_capability_gate_rejected", + task_id=task_id, + project_id=task["project_id"], + status="failed") + # Audit fix 3（M5）：网关门拒绝 + # 早返回也要清暂存，同缺 key 门。 + self._cleanup_task_seed(task) + self._running_clear(task_id) + return {"dispatched": True, + "task_id": task_id, + "host_name": + self.executor._host_name(task) + if hasattr(self.executor, + "_host_name") + else task_id, + "status": "failed", + "executor": self._executor_name(), + "stdout": "", "error": _gw_error, + "events": [], "cost_record_id": None} + return None + + def _persist_policy_model(self, task, decision): + """策略写回（审计批 5，I-9②抽取，行为逐字不变）。 + + 原 run_one 内 decide_routing 后的写回段平移：只持久化 + source=policy 且 applied 的决策（shadow / worker_default / + worker_explicit 一律不写回，见原注释）。副作用点逐项核对 + （与原代码一致，抽取后无增减）： + - 库写：CAS 式 UPDATE tasks SET model（仅当库模型仍为 + NULL/空时写，绝不覆盖显式模型），自带连接、写完提交。 + - 日志：写失败时记 worker_model_writeback_failed 警告， + 不炸派发（本轮内存模型照常生效）。 + - 审计：仅写成功时记 worker_model_persisted（身份标记， + 后续尝试据此区分写回模型与用户显式模型）。 + - 不改任务状态、不清占位、不记 gates_checked 标记、不碰 + task["model"]（内存模型的赋值仍在调用方原地）。 + 新门形状规则同 _gateway_capability_check：纯判定 + （decide_routing）+ worker 内一处接线调用。 + """ + task_id = task.get("id") + _persisted = False + if decision.chosen_model and decision.source == "policy" \ + and decision.applied: + try: + _wb_conn = _db_connect(self.db_path) + try: + _wb_conn.execute( + "UPDATE tasks SET model = ? WHERE id = ?" + " AND (model IS NULL OR model = '')", + (decision.chosen_model, task_id)) + _wb_conn.commit() + finally: + _wb_conn.close() + _persisted = True + except Exception as exc: + log_event(self.logger, "worker_model_writeback_failed", + task_id=task_id, + project_id=task["project_id"], + status="warning", + message="model writeback failed: %s" % exc) + if _persisted: + # 写回模型的身份标记：后续尝试据此区分 + # 写回模型与用户显式模型（见 + # _is_persisted_model）。审计本身 best-effort， + # 丢了也不影响写回，旧模型靠 policy 行兜底。 + self._audit(task["project_id"], + "worker_model_persisted", + "%s tier=%s model=%s" % ( + task_id, decision.tier, + decision.chosen_model)) + def run_one(self, task=None): """认领并执行一个任务，返回结果 dict（见模块 docstring）。 @@ -7541,148 +7935,13 @@ class Worker: # 门可列入 gate_results。 _mark_gate_checked(task, "gateway") - # Audit fix：网关能力检查——此前路由决策只在模型 - # 为空时跑，sub2api 网关先填默认模型后决策直接跳 - # 过，能力门在生产路径上完全不生效。补齐：网关填 - # 的模型同样按档需求检查，档模式与 v8 一致：shadow - # 只审计不拦，active 不满足则 fail-fast（同款工具 - # 门形状，清占位、不调 runner）。没有策略、没有档 - # 需求、档模式 off 时一律不查，旧行为不变。注意： - # 这里没有 worker 默认可回落（网关模型就是该任务 - # 的网关内模型，换成 worker 默认会跑错网关），所以 - # active 下不满足即拦，不存在“先试 worker 默认” - # 的中间态；运营方须先把网关模型写进能力目录， - # 否则 unknown 同样拦（fail-closed，与 v8 一致）。 - # 例外：任务级显式模型按 Audit fix 4 分档——已知 - # 不满足才拦，unknown 只审计（见下），因为显式 - # 优先保护的是用户选了哪个模型，不是选的模型 - # 一定能跑；目录陈旧不应惩罚正确的显式选择。 - # 审计批 4（I-7）：显式模型的 known/unknown 分档检查 - # 与网关无关——网关分支只决定“待查模型是哪一个” - # （网关填充模型或显式模型），分档判定走第 3 批统一 - # 的 evaluate_capability。进入条件因此是“有策略且 - # （走了 sub2api 网关，或显式任务走了任一真实网关）”： - # 非显式 + 非 sub2api 的任务根本不进块（它们模型 - # 为空、后续走 decide_routing，不存在误拦）；显式 - # + 无网关（provider 为 None）的任务也不进块——它 - # 根本没走网关，显式语义不变（写回模型仍走后续 - # 复核，用户显式仍不被复核，见 fixes5 契约）； - # 显式 + sub2api 路径与原来逐字一致（同一审计 - # action、同一 error 文案、同一 blocked 标志）， - # 不会重复审计或重复拦；显式 + 非 sub2api 网关 - # 在 active 下已知不满足时从“直接派发”变为“拦”， - # 这是有意的行为变更。 - # 附带：已持久化的显式任务在非 sub2api 网关下会 - # 同时走本门与后续写回复核（本门不拦时复核再按 - # 分档审计/拦），两者审计 action 不同 - # （gateway_explicit 与 recheck_*），不是重复拦。 - if self.routing_policy is not None \ - and (gateway_provider == "sub2api" - or (_explicit_model - and gateway_provider is not None)): - _gw_tier = classify_task(task) - _gw_requires = self.routing_policy.tier_requirements.get( - _gw_tier) or {} - if _gw_requires: - _gw_mode = self.routing_policy.tier_mode(_gw_tier) - if _gw_mode in ("active", "shadow"): - _gw_verdict = evaluate_capability( - self.routing_policy.model_capabilities, - task.get("model"), _gw_requires) - # 审计批 4（I-10）：走到这里说明网关 - # 能力检查真正算了 verdict。 - _mark_gate_checked(task, "capability") - if not _gw_verdict.ok: - _gw_note = ("gateway model %s %s capability:" - " %s" % ( - task.get("model"), - "cannot verify" - if _gw_verdict.unknown - else "lacks", - ", ".join(_gw_verdict.missing))) - # Audit fix 4（第三遍审计分档建议）： - # 显式模型按 known/unknown 分档。known - # 是目录明确写了不满足（布尔 False、 - # 或 context 低于需求），与网关默认 - # 模型同款 fail-fast；unknown 是目录 - # 无该模型、或条目缺字段，目录陈旧 - # 不应惩罚用户的正确选择，只审计不拦。 - # shadow 档两种情况都只审计。 - _gw_known = _gw_verdict.known_missing - _gw_block = False - _gw_error = None - if _explicit_model: - _gw_blocked = ( - _gw_mode == "active" and bool(_gw_known)) - # 显式检查结果独立审计，含 blocked - # 标志：被拦的已知不满足也要进影子 - # 报告的显式失败清单，否则报告会 - # 漏掉恰恰最该被看到的一类。 - self._audit( - task["project_id"], - "worker_capability_gateway_explicit", - "%s model=%s tier=%s mode=%s" - " missing=%s unknown=%s blocked=%s" - % (task_id, task.get("model"), - _gw_tier, _gw_mode, - ",".join(_gw_verdict.missing) or "-", - _gw_verdict.unknown, - _gw_blocked)) - if _gw_blocked: - _gw_block = True - _gw_error = ( - "%s tier explicit model %s lacks" - " capability: %s; refusing to" - " dispatch a model known to lack" - " a required capability" - % (_gw_tier, task.get("model"), - ", ".join(_gw_known))) - elif _gw_mode == "shadow": - self._audit( - task["project_id"], - "worker_capability_gateway_shadow", - "%s tier=%s %s (shadow, not blocked)" - % (task_id, _gw_tier, _gw_note)) - else: - _gw_block = True - _gw_error = ( - "%s tier gateway model %s; refusing" - " to dispatch an incapable or" - " unverifiable model" - % (_gw_tier, _gw_note)) - if _gw_block: - self._set_status(task_id, "failed") - self._write_execution_result( - task_id, error_text=_gw_error, - actual_model=self.model, - started_at=started_at, - finished_at=time.time(), - only_if_status="failed") - self._audit( - task["project_id"], - "worker_capability_gate_rejected", - "%s: %s" % (task_id, _gw_error)) - log_event( - self.logger, - "worker_capability_gate_rejected", - task_id=task_id, - project_id=task["project_id"], - status="failed") - # Audit fix 3（M5）：网关门拒绝 - # 早返回也要清暂存，同缺 key 门。 - self._cleanup_task_seed(task) - self._running_clear(task_id) - return {"dispatched": True, - "task_id": task_id, - "host_name": - self.executor._host_name(task) - if hasattr(self.executor, - "_host_name") - else task_id, - "status": "failed", - "executor": self._executor_name(), - "stdout": "", "error": _gw_error, - "events": [], "cost_record_id": None} + # 审计批 5（I-9①）：网关能力检查整段已抽成 + # _gateway_capability_check（纯平移，行为逐字不变， + # 副作用清单见该方法文档串）。 + _gw_blocked = self._gateway_capability_check( + task, gateway_provider, _explicit_model, started_at) + if _gw_blocked is not None: + return _gw_blocked task = dict(task) if not task.get("model"): @@ -7793,36 +8052,7 @@ class Worker: # 网关路径不受影响：网关填模型的任务跳过 # decide_routing 与写回，每次尝试都重新填网关 # 模型并重新接受网关能力检查。 - _persisted = False - if decision.chosen_model and decision.source == "policy" \ - and decision.applied: - try: - _wb_conn = _db_connect(self.db_path) - try: - _wb_conn.execute( - "UPDATE tasks SET model = ? WHERE id = ?" - " AND (model IS NULL OR model = '')", - (decision.chosen_model, task_id)) - _wb_conn.commit() - finally: - _wb_conn.close() - _persisted = True - except Exception as exc: - log_event(self.logger, "worker_model_writeback_failed", - task_id=task_id, - project_id=task["project_id"], - status="warning", - message="model writeback failed: %s" % exc) - if _persisted: - # 写回模型的身份标记：后续尝试据此区分 - # 写回模型与用户显式模型（见 - # _is_persisted_model）。审计本身 best-effort， - # 丢了也不影响写回，旧模型靠 policy 行兜底。 - self._audit(task["project_id"], - "worker_model_persisted", - "%s tier=%s model=%s" % ( - task_id, decision.tier, - decision.chosen_model)) + self._persist_policy_model(task, decision) task["model"] = decision.chosen_model or self.model elif not _gateway_filled: # Audit fix 3（I8）：显式模型任务（含协调员子任务 @@ -7896,6 +8126,13 @@ class Worker: # 宣称运行时按工具拦截：host 的 elicitation 回答是 # 全局开关（auto 接受、其余拒绝），没有按工具分流， # ask 在今天没有询问通道，见 agent_def.py 的 v9 边界。 + # 审计批 5（I-12）：派发时真的调 effective_tool_approvals + # 并把 effective 映射写进同一审计行——此前模块文档称 + # “派发时审计其有效结果”，实际只记了声明本身。host 的 + # 全局开关取自传给 host 的 --config 文件里的 approval + # （读不到/未配置时按 host 自身的默认 deny 计，与 + # host.py 的 args.approval 回落同款）；ask_honoured 恒为 + # False，ask 在行里明确折算成 decline，不谎称“会问”。 _tool_approvals = task.get("agent_tool_approvals") or {} if _tool_approvals: _denied_required = [ @@ -7928,13 +8165,28 @@ class Worker: "executor": self._executor_name(), "stdout": "", "error": _appr_error, "events": [], "cost_record_id": None} + _host_approval = "deny" + _cfg_path = getattr(self.executor, "config_path", None) + if _cfg_path: + try: + with open(_cfg_path, encoding="utf-8") as _fh: + _host_approval = ( + json.load(_fh).get("approval") or "deny") + except Exception: + _host_approval = "deny" + _effective_approvals = _effective_tool_approvals( + _tool_approvals, _host_approval) self._audit(task["project_id"], "worker_tool_approvals_declared", "%s: %s (per-tool interception is not enforced" " at runtime; host answers elicitation globally)" + " effective=%s" % (task_id, json.dumps(_tool_approvals, ensure_ascii=False, - sort_keys=True))) + sort_keys=True), + json.dumps(_effective_approvals, + ensure_ascii=False, + sort_keys=True))) # agent 必需工具门（v3）：协调员父任务已在前面返回， # 这里只剩真正要跑模型的任务（普通/goal/fork 首轮都 # 走后面的尝试循环，门放在循环前，一次定罪不重试）。