From 18e3f1c82d675d6e722cd92848c96bcf6296eef4 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 04:44:27 +0000 Subject: [PATCH] fix: third-audit gate findings (explicit split, shadow explicit failures, N2-N5, N1 note) Explicit gateway capability now splits known/unknown: known-missing explicit models fail-fast under active (same shape as the gateway default gate), unknown explicit models stay audit-only, shadow never blocks; explicit priority now protects which model the user chose, not that the chosen model must run. Shadow report reads worker_capability_gateway_explicit rows as explicit_capability_failures with a field note distinguishing the three explicit counts. N2: coordinator requeue CAS rowcount=0 branch covered via a proxy connection test (no refactor needed). N3: shadow endpoint test on the create_server + real HTTP skeleton (capability, truncated, and explicit-failure fields). N4: creation-layer rejection tests for tools/instructions/tool_approvals on goal/coordinator tasks. N5: model writeback failure logs a warning (best-effort kept). N1: writeback capability-recheck trade-off documented in comment only; recheck itself is a later batch. --- cloud/models.py | 43 +++ cloud/server.py | 59 ++++ cloud/tests/test_audit_fixes3.py | 18 +- cloud/tests/test_audit_fixes4.py | 481 +++++++++++++++++++++++++++++++ cloud/worker.py | 80 ++++- 5 files changed, 660 insertions(+), 21 deletions(-) create mode 100644 cloud/tests/test_audit_fixes4.py diff --git a/cloud/models.py b/cloud/models.py index d41e25e..f9af66e 100644 --- a/cloud/models.py +++ b/cloud/models.py @@ -246,6 +246,49 @@ def check_model_capabilities(capabilities, model, requires): return (not missing), sorted(missing), unknown +def split_model_capability_missing(capabilities, model, requires): + """把 check_model_capabilities 的缺项拆成已知 / 未知两档。 + + Audit fix 4（第三遍审计分档建议）：显式模型的网关检查不能 + 只用 (missing, unknown) 的总口径——unknown 为真时可能是 + “目录里根本没有该模型”，也可能是“条目里某项能力没填”， + 两者决策不同。返回 (known_missing, unknown_missing)： + + - 目录里没有该模型：全部需求都算 unknown，known 为空； + - 布尔能力明确为 False、或 context_tokens 明确低于需求： + known（目录明确写了不满足）； + - 布尔能力未填（None）、或 context_tokens 未填：unknown。 + + 与 check_model_capabilities 的口径一致：requires 里非布尔/ + 非整数键被忽略，不计入任何一档。 + """ + requires = requires or {} + model = _clean_model_name(model) + entry = (capabilities or {}).get(model) if model else None + if entry is None: + return [], sorted( + k for k in requires + if k in REQUIREMENT_BOOL_KEYS or k in REQUIREMENT_INT_KEYS) + known = [] + unknown = [] + for key, value in requires.items(): + if key in REQUIREMENT_BOOL_KEYS: + flag = entry.get(key) + if flag is True: + continue + if flag is None: + unknown.append(key) + else: + known.append(key) + elif key in REQUIREMENT_INT_KEYS: + context = entry.get("context_tokens") + if context is None: + unknown.append(key) + elif context < value: + known.append(key) + return sorted(known), sorted(unknown) + + def prompt_len_bucket(prompt) -> str: """Bucket a prompt by length for logging only, never for routing. diff --git a/cloud/server.py b/cloud/server.py index 40d46b5..59c3cbe 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -4195,6 +4195,20 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): ).fetchall() truncated = len(audit_rows) > 10000 audit_rows = audit_rows[:10000] + # Audit fix 4（第三遍审计显式闭环）：显式能力检查 + # 结果只记 worker_capability_gateway_explicit，影子 + # 端点此前只读 worker_routing，运营方根本看不到它。 + # 在此之外加读显式行，附为独立字段，不与已有两个 + # explicit_n 口径混在一起（见报告里的字段说明）。 + explicit_rows = conn.execute( + "SELECT detail FROM audit_log" + " WHERE project_id = ?" + " AND action = 'worker_capability_gateway_explicit'" + " ORDER BY created_at, id LIMIT 10001", + (project_id,), + ).fetchall() + explicit_truncated = len(explicit_rows) > 10000 + explicit_rows = explicit_rows[:10000] task_rows = conn.execute( "SELECT id, status, model FROM tasks" " WHERE project_id = ?", @@ -4229,6 +4243,51 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): tasks_by_id, costs_by_task, explicit_n=explicit_n) report["project_id"] = project_id report["truncated"] = truncated + # 显式失败清单：解析 worker_capability_gateway_explicit + # 行（格式：task_id model=.. tier=.. mode=.. missing=.. + # unknown=.. blocked=..；fix4 之前的行无 blocked 项， + # 按 False 计）。这与报告里两个 explicit_n 不同义： + # 顶层 explicit_n 是任务表里带模型字段的任务数； + # per-tier explicit_n 是 worker_routing source=explicit + # 审计行数；本字段只计网关显式能力检查的结果行， + # 含被拦的已知不满足与只审计的 unknown，blocked + # 字段区分两者。解析失败的行直接跳过、不谎报。 + _explicit_failures = [] + for _row in explicit_rows: + _detail = _row["detail"] or "" + _tokens = _detail.split() + if not _tokens: + continue + _kv = {} + for _tok in _tokens[1:]: + if "=" in _tok: + _k, _v = _tok.split("=", 1) + _kv[_k] = _v + if "model" not in _kv or "tier" not in _kv: + continue + _missing = _kv.get("missing") or "-" + _explicit_failures.append({ + "task_id": _tokens[0], + "model": _kv["model"], + "tier": _kv["tier"], + "mode": _kv.get("mode") or "-", + "missing": [] if _missing == "-" else _missing.split(","), + "unknown": _kv.get("unknown") == "True", + "blocked": _kv.get("blocked") == "True", + }) + report["explicit_capability_failures"] = _explicit_failures + report["explicit_capability_failures_truncated"] = \ + explicit_truncated + report["explicit_counts_note"] = ( + "explicit_n (top level) counts tasks whose model field is" + " set in the tasks table; per-tier explicit_n counts" + " worker_routing rows with source=explicit; " + "explicit_capability_failures lists" + " worker_capability_gateway_explicit rows only, i.e." + " gateway capability check outcomes for explicit models," + " with blocked=True for known-missing models that were" + " refused and blocked=False for unknown models that were" + " audited but not blocked.") # Audit fix 3（I2，端点侧）：只读加载当前策略， # 给每档附能力状态（档模型 vs requires）。加载 # 失败/未配置时状态为 policy_unavailable，不拦、 diff --git a/cloud/tests/test_audit_fixes3.py b/cloud/tests/test_audit_fixes3.py index f7b31d3..ebd7b87 100644 --- a/cloud/tests/test_audit_fixes3.py +++ b/cloud/tests/test_audit_fixes3.py @@ -113,7 +113,11 @@ class B1ExplicitGatewayTestCase(unittest.TestCase): def tearDown(self): self.tmp.cleanup() - def test_explicit_model_is_audited_not_blocked(self): + def test_explicit_model_known_missing_is_blocked(self): + # Audit fix 4: known-missing explicit models are now blocked + # (split from unknown, which stays audit-only). The old + # audit-only expectation for tool_calling=False moved to + # test_audit_fixes4's unknown case. _insert_task(self.db_path, "E1", model="explicit-model") worker = _worker(self.db_path, { "mode": "active", @@ -124,16 +128,16 @@ class B1ExplicitGatewayTestCase(unittest.TestCase): }, }, gateway=("sub2api", {})) result = worker.run_one() - self.assertEqual(result["status"], "completed") - self.assertTrue(worker._calls) + self.assertEqual(result["status"], "failed") + self.assertFalse(worker._calls) + self.assertIn("tool_calling", result["error"]) actions = _audit_actions(self.db_path) explicit = [d for a, d in actions if a == "worker_capability_gateway_explicit"] self.assertTrue(explicit) - self.assertIn("explicit-model", explicit[0]) - self.assertIn("tool_calling", explicit[0]) - self.assertFalse([a for a, _ in actions - if a == "worker_capability_gate_rejected"]) + self.assertIn("blocked=True", explicit[0]) + self.assertTrue([a for a, _ in actions + if a == "worker_capability_gate_rejected"]) def test_gateway_default_active_still_blocks(self): _insert_task(self.db_path, "G1") diff --git a/cloud/tests/test_audit_fixes4.py b/cloud/tests/test_audit_fixes4.py new file mode 100644 index 0000000..c9b7b0f --- /dev/null +++ b/cloud/tests/test_audit_fixes4.py @@ -0,0 +1,481 @@ +"""Audit fix 4: third-audit gate findings. + +- Explicit gateway capability split: known-missing blocks (active), + unknown audits only; shadow never blocks. +- Explicit failures surfaced in the shadow report as an independent + field, distinct from the two explicit_n counts. +- N2: coordinator requeue CAS rowcount=0 branch, via a proxy + connection (the auditor's low-cost approach, no refactor). +- N3: shadow endpoint test on the create_server + real HTTP skeleton. +- N4: creation-layer rejections for tools / instructions / + tool_approvals on goal/coordinator tasks, in one pass. +- N5: model writeback failure logs a warning (best-effort kept). + +Run: python3 -m unittest discover -s cloud/tests -p "test_audit_fixes4.py" -v +No real models, no network, no real gateway: gateway steps are +stubbed exactly like test_audit_fixes.py / test_audit_fixes3.py. +""" + +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import time +import unittest +from types import SimpleNamespace +from unittest import mock + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) + +import cloud.worker as worker_module +from cloud.auth import AuthStore +from cloud.models import split_model_capability_missing +from cloud.server import create_server, init_db +from cloud.worker import Worker + + +def _completed_runner(cmd): + if "dispatch" in cmd: + name = cmd[cmd.index("--name") + 1] + else: + name = cmd[cmd.index("--task") + 1] + return SimpleNamespace( + returncode=0, + stdout='{"status": "completed", "result": "done",' + ' "task_id": "%s", "thread_id": "th1"}' % name, + stderr="") + + +def _insert_task(db_path, task_id, model=None, status="queued", + project_id="p1", coordinator_parts=None, + parent_task_id=None, coordinator_part=None, + result_text=None, error_text=None, paused=0, + interrupted_reason=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, coordinator_parts," + " parent_task_id, coordinator_part, result_text, error_text," + " model, paused_by_user, interrupted_reason)" + " VALUES(?, ?, 'do work', ?, NULL, '[]', 0, 'tester'," + " ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?)", + (task_id, project_id, status, now, now, + json.dumps(coordinator_parts) if coordinator_parts else None, + parent_task_id, coordinator_part, result_text, error_text, + model, paused, interrupted_reason), + ) + conn.commit() + finally: + conn.close() + + +def _row(db_path, task_id): + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + try: + row = conn.execute("SELECT * FROM tasks WHERE id = ?", + (task_id,)).fetchone() + return dict(row) if row else None + finally: + conn.close() + + +def _audit_actions(db_path, project_id="p1"): + store = AuthStore(db_path) + try: + return [(e["action"], e["detail"]) for e in store.list_audit( + project_id)] + finally: + store.conn.close() + + +def _worker(db_path, policy_doc=None, env=None, gateway=None): + calls = [] + + def recording(cmd): + calls.append(list(cmd)) + return _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 SplitMissingTestCase(unittest.TestCase): + def test_absent_model_is_all_unknown(self): + known, unknown = split_model_capability_missing( + {}, "no-such-model", {"tool_calling": True, + "min_context_tokens": 1000}) + self.assertEqual(known, []) + self.assertEqual(unknown, ["min_context_tokens", "tool_calling"]) + + def test_explicit_false_is_known(self): + known, unknown = split_model_capability_missing( + {"m": {"tool_calling": False}}, + "m", {"tool_calling": True}) + self.assertEqual(known, ["tool_calling"]) + self.assertEqual(unknown, []) + + def test_unfilled_field_is_unknown(self): + known, unknown = split_model_capability_missing( + {"m": {}}, "m", {"tool_calling": True}) + self.assertEqual(known, []) + self.assertEqual(unknown, ["tool_calling"]) + + def test_short_context_is_known_and_missing_context_is_unknown(self): + known, unknown = split_model_capability_missing( + {"m": {"context_tokens": 100}}, + "m", {"min_context_tokens": 1000}) + self.assertEqual(known, ["min_context_tokens"]) + known2, unknown2 = split_model_capability_missing( + {"m": {}}, "m", {"min_context_tokens": 1000}) + self.assertEqual(known2, []) + self.assertEqual(unknown2, ["min_context_tokens"]) + + def test_mixed_known_and_unknown_keeps_known(self): + known, unknown = split_model_capability_missing( + {"m": {"tool_calling": False}}, + "m", {"tool_calling": True, "vision": True}) + self.assertEqual(known, ["tool_calling"]) + self.assertEqual(unknown, ["vision"]) + + +class ExplicitGatewaySplitTestCase(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) + + def tearDown(self): + self.tmp.cleanup() + + def _policy(self, mode, capabilities): + return { + "mode": mode, + "tiers": {"default": {"model": "policy-model", + "requires": {"tool_calling": True}}}, + "model_capabilities": capabilities, + } + + def test_known_missing_explicit_is_blocked(self): + _insert_task(self.db_path, "E1", model="explicit-model") + worker = _worker(self.db_path, self._policy("active", { + "explicit-model": {"tool_calling": False}, + }), gateway=("sub2api", {})) + result = worker.run_one() + self.assertEqual(result["status"], "failed") + self.assertFalse(worker._calls) + self.assertIn("tool_calling", result["error"]) + actions = _audit_actions(self.db_path) + explicit = [d for a, d in actions + if a == "worker_capability_gateway_explicit"] + self.assertTrue(explicit) + self.assertIn("blocked=True", explicit[0]) + self.assertTrue([a for a, _ in actions + if a == "worker_capability_gate_rejected"]) + + def test_unknown_explicit_is_audited_not_blocked(self): + _insert_task(self.db_path, "E1", model="explicit-model") + worker = _worker(self.db_path, self._policy("active", {}), + gateway=("sub2api", {})) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + self.assertTrue(worker._calls) + actions = _audit_actions(self.db_path) + explicit = [d for a, d in actions + if a == "worker_capability_gateway_explicit"] + self.assertTrue(explicit) + self.assertIn("blocked=False", explicit[0]) + self.assertFalse([a for a, _ in actions + if a == "worker_capability_gate_rejected"]) + + def test_mixed_known_and_unknown_explicit_is_blocked(self): + # One known-missing capability is enough to block: the task + # cannot succeed regardless of the unknown field. + _insert_task(self.db_path, "E1", model="explicit-model") + policy = self._policy("active", { + "explicit-model": {"tool_calling": False}, + }) + policy["tiers"]["default"]["requires"] = { + "tool_calling": True, "vision": True} + worker = _worker(self.db_path, policy, gateway=("sub2api", {})) + result = worker.run_one() + self.assertEqual(result["status"], "failed") + self.assertFalse(worker._calls) + + def test_shadow_known_missing_explicit_is_not_blocked(self): + _insert_task(self.db_path, "E1", model="explicit-model") + worker = _worker(self.db_path, self._policy("shadow", { + "explicit-model": {"tool_calling": False}, + }), gateway=("sub2api", {})) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + self.assertTrue(worker._calls) + actions = _audit_actions(self.db_path) + explicit = [d for a, d in actions + if a == "worker_capability_gateway_explicit"] + self.assertTrue(explicit) + self.assertIn("blocked=False", explicit[0]) + self.assertFalse([a for a, _ in actions + if a == "worker_capability_gate_rejected"]) + + +class I7RowcountZeroTestCase(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) + + def tearDown(self): + self.tmp.cleanup() + + def test_rowcount_zero_skips_requeue_and_audit(self): + parts = [{"name": "后端", "prompt": "做后端"}] + _insert_task(self.db_path, "coord", coordinator_parts=parts) + _insert_task(self.db_path, "c-old", status="failed", + parent_task_id="coord", coordinator_part="后端", + result_text="旧输出", error_text="旧错") + worker = _worker(self.db_path, gateway=(None, None)) + task = worker.get_task("coord") + orig_connect = worker_module._db_connect + db_path = self.db_path + + class _FlippingConn: + """Proxy: flip the child to completed just before the + requeue UPDATE executes, so its rowcount is 0. The + snapshot connection is wrapped too, but only the + requeue UPDATE triggers the flip.""" + + def __init__(self, real): + self._real = real + self._flipped = False + + def execute(self, sql, *args): + if not self._flipped and \ + "UPDATE tasks SET status = 'queued'" in sql: + flip = sqlite3.connect(db_path) + try: + flip.execute( + "UPDATE tasks SET status = 'completed'" + " WHERE id = 'c-old'") + flip.commit() + finally: + flip.close() + self._flipped = True + return self._real.execute(sql, *args) + + def commit(self): + return self._real.commit() + + def close(self): + return self._real.close() + + def __getattr__(self, name): + return getattr(self._real, name) + + with mock.patch.object( + worker_module, "_db_connect", + side_effect=lambda p: _FlippingConn(orig_connect(p))): + children = worker._coord_sync_children( + task, task["coordinator_parts"]) + self.assertEqual(children, [{"name": "后端", "id": "c-old"}]) + old = _row(self.db_path, "c-old") + self.assertEqual(old["status"], "completed") + self.assertEqual(old["result_text"], "旧输出") + actions = _audit_actions(self.db_path) + self.assertFalse([a for a, _ in actions + if a == "coordinator_child_requeued"]) + + +class _HttpServerCase(unittest.TestCase): + def setUp(self): + self._old_policy_json = os.environ.get( + "CLOUD_ROUTING_POLICY_JSON") + self._old_policy_file = os.environ.get( + "CLOUD_ROUTING_POLICY_FILE") + os.environ.pop("CLOUD_ROUTING_POLICY_FILE", None) + 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() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + if self._old_policy_json is None: + os.environ.pop("CLOUD_ROUTING_POLICY_JSON", None) + else: + os.environ["CLOUD_ROUTING_POLICY_JSON"] = \ + self._old_policy_json + if self._old_policy_file is None: + os.environ.pop("CLOUD_ROUTING_POLICY_FILE", None) + else: + os.environ["CLOUD_ROUTING_POLICY_FILE"] = \ + self._old_policy_file + 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) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def _project(self, name="p"): + status, project = self._req( + "POST", "/projects", {"name": name, "auto_pr": 1}, + self.owner_key) + self.assertEqual(status, 201, project) + return project["id"] + + def _insert_audit(self, project_id, action, detail): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO audit_log(user_id, project_id, action," + " detail, created_at) VALUES(?, ?, ?, ?, ?)", + ("tester", project_id, action, detail, time.time())) + conn.commit() + finally: + conn.close() + + +class ShadowEndpointTestCase(_HttpServerCase): + def test_report_carries_capability_and_explicit_failures(self): + os.environ["CLOUD_ROUTING_POLICY_JSON"] = json.dumps({ + "mode": "active", + "tiers": {"default": {"model": "policy-model", + "requires": {"tool_calling": True}}}, + "model_capabilities": { + "policy-model": {"tool_calling": False}, + }, + }) + pid = self._project() + _insert_task(self.db_path, "T1", model="policy-model", + status="completed", project_id=pid) + self._insert_audit( + pid, "worker_routing", + "T1 tier=default source=policy applied=true" + " suggested=policy-model chosen=policy-model") + self._insert_audit( + pid, "worker_capability_gateway_explicit", + "E1 model=explicit-model tier=default mode=active" + " missing=tool_calling unknown=False blocked=True") + status, report = self._req( + "GET", "/routing-shadow?project_id=" + pid, + key=self.owner_key) + self.assertEqual(status, 200, report) + self.assertEqual( + report["tiers"]["default"]["capability"]["status"], "fail") + self.assertEqual( + report["tiers"]["default"]["capability"]["missing"], + ["tool_calling"]) + self.assertFalse(report["truncated"]) + failures = report["explicit_capability_failures"] + self.assertEqual(len(failures), 1) + self.assertEqual(failures[0], { + "task_id": "E1", "model": "explicit-model", + "tier": "default", "mode": "active", + "missing": ["tool_calling"], "unknown": False, + "blocked": True}) + self.assertFalse(report["explicit_capability_failures_truncated"]) + self.assertIn("top level", report["explicit_counts_note"]) + self.assertIn("per-tier", report["explicit_counts_note"]) + self.assertIn("worker_capability_gateway_explicit", + report["explicit_counts_note"]) + + def test_report_truncated_flag(self): + pid = self._project() + _insert_task(self.db_path, "T1", model="policy-model", + status="completed", project_id=pid) + conn = sqlite3.connect(self.db_path) + try: + conn.executemany( + "INSERT INTO audit_log(user_id, project_id, action," + " detail, created_at) VALUES(?, ?, ?, ?, ?)", + [("tester", pid, "worker_routing", + "T1 tier=default source=policy applied=true" + " suggested=policy-model chosen=policy-model", + time.time())] * 10001) + conn.commit() + finally: + conn.close() + status, report = self._req( + "GET", "/routing-shadow?project_id=" + pid, + key=self.owner_key) + self.assertEqual(status, 200, report) + self.assertTrue(report["truncated"]) + self.assertIn("truncated", report["recommendation"]) + + +class CreateRejectionTestCase(_HttpServerCase): + def _create(self, extra): + pid = self._project() + body = {"project_id": pid, "prompt": "协调干活"} + body.update(extra) + return self._req("POST", "/tasks", body, self.owner_key) + + def test_coordinator_rejects_tools_instructions_approvals(self): + parts = [{"name": "后端", "prompt": "做后端"}] + agents = [ + {"name": "x-agent", "tools": ["websearch"]}, + {"name": "x-agent", "instructions": "Do it."}, + {"name": "x-agent", "tool_approvals": {"websearch": "ask"}}, + ] + for agent in agents: + status, data = self._create({ + "coordinator_parts": parts, "agent": agent}) + self.assertEqual(status, 400, (agent, data)) + self.assertIn("cannot be used with", data["error"]) + + def test_goal_rejects_tools_instructions_approvals(self): + agents = [ + {"name": "x-agent", "tools": ["websearch"]}, + {"name": "x-agent", "instructions": "Do it."}, + {"name": "x-agent", "tool_approvals": {"websearch": "ask"}}, + ] + for agent in agents: + status, data = self._create({ + "goal_mode": 1, "agent": agent}) + self.assertEqual(status, 400, (agent, data)) + self.assertIn("cannot be used with", data["error"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 49fe73a..8828c85 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -158,7 +158,8 @@ from cloud.logging_config import get_logger, log_event, setup_logging from cloud.models import ModelRouter, RoutingPolicy, decide_routing, \ check_model_capabilities, classify_task, \ is_fallbackable, load_routing_policy, parse_routing_policy, \ - record_fallback, record_routing + record_fallback, record_routing, \ + split_model_capability_missing from cloud.auto_memory import AutoMemoryStore from cloud.auto_memory import build_injection_prefix as _auto_prefix from cloud.auto_memory import process_due_tasks as _auto_process_due @@ -6121,8 +6122,10 @@ class Worker: # Audit fix 3（B1）：触发条件是“走了 sub2api 网关”， # 不是“网关填了模型”。任务级显式模型此前因 # _gateway_filled=False 全程跳过能力检查，是生产 - # 覆盖的一半缺口；显式优先原则不变——显式模型 - # 不满足/unknown 时只审计、不拦任务（见下）。 + # 覆盖的一半缺口。Audit fix 4（第三遍审计分档建议）： + # 显式优先保护的是“用户选了哪个模型”，不是“选的 + # 模型一定能跑”——显式模型已知不满足时同样拦， + # 只有无法验证时才只审计不拦（见下）。 _explicit_model = bool(task.get("model")) _gateway_filled = False gateway_provider = None @@ -6167,6 +6170,10 @@ class Worker: # active 下不满足即拦，不存在“先试 worker 默认” # 的中间态；运营方须先把网关模型写进能力目录， # 否则 unknown 同样拦（fail-closed，与 v8 一致）。 + # 例外：任务级显式模型按 Audit fix 4 分档——已知 + # 不满足才拦，unknown 只审计（见下），因为显式 + # 优先保护的是用户选了哪个模型，不是选的模型 + # 一定能跑；目录陈旧不应惩罚正确的显式选择。 if gateway_provider == "sub2api" \ and self.routing_policy is not None: _gw_tier = classify_task(task) @@ -6186,19 +6193,46 @@ class Worker: "cannot verify" if _gw_unknown else "lacks", ", ".join(_gw_missing))) + # Audit fix 4（第三遍审计分档建议）： + # 显式模型按 known/unknown 分档。known + # 是目录明确写了不满足（布尔 False、 + # 或 context 低于需求），与网关默认 + # 模型同款 fail-fast；unknown 是目录 + # 无该模型、或条目缺字段，目录陈旧 + # 不应惩罚用户的正确选择，只审计不拦。 + # shadow 档两种情况都只审计。 + _gw_known, _gw_unk = \ + split_model_capability_missing( + self.routing_policy.model_capabilities, + task.get("model"), _gw_requires) + _gw_block = False + _gw_error = None if _explicit_model: - # 显式优先：不拦任务，但绝不能无声 - # 豁免——记独立审计，含缺项/unknown， - # 影子解析器不认此 action，不污染样本。 + _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" % ( - task_id, task.get("model"), - _gw_tier, _gw_mode, - ",".join(_gw_missing) or "-", - _gw_unknown)) + " missing=%s unknown=%s blocked=%s" + % (task_id, task.get("model"), + _gw_tier, _gw_mode, + ",".join(_gw_missing) or "-", + _gw_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"], @@ -6206,11 +6240,13 @@ class Worker: "%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, @@ -6322,7 +6358,19 @@ class Worker: # 策略一改同一任务中途就换模型。CAS 式写回： # 只在库模型仍为空时写，绝不覆盖显式模型； # 写失败 best-effort，不炸派发（本轮内存模型 - # 照常生效，只是下次派发可能再次决策）。 + # 照常生效，只是下次派发可能再次决策），但 + # 必须记警告日志，不能静默吞掉（Audit fix 4 + # N5：否则 I5 的保证悄悄失效还看不出来）。 + # Audit fix 4（N1 trade-off，只写注释、不改 + # 语义）：写回是一把双刃剑——非网关路径下， + # 写回的模型使后续尝试走显式分支，decide_routing + # 的能力检查不再跑；两次尝试之间档 requires + # 一收紧，续跑任务将带着旧模型、不经复核直 + # 接派发。网关路径不受影响：网关填模型的任务 + # 跳过 decide_routing 与写回，每次尝试都重新 + # 填网关模型并重新接受网关能力检查。复核方案 + # （写回时记校验版本、或续跑时重查当前 + # requires）是下一批的事，本轨不改写回语义。 if decision.chosen_model: try: _wb_conn = _db_connect(self.db_path) @@ -6334,8 +6382,12 @@ class Worker: _wb_conn.commit() finally: _wb_conn.close() - except Exception: - pass + 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) task["model"] = decision.chosen_model or self.model elif not _gateway_filled: # Audit fix 3（I8）：显式模型任务（含协调员子任务