From f9a77577c51a51bf0d96ecb49ec79b1c69d60a50 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 10:00:20 +0800 Subject: [PATCH] fix: audit findings (fallback dead code, silent fallback, registry re-parse, shadow sort, gateway capability) - RoutingPolicy.fallback_models: move the cleaning loop into __init__ (it was unreachable after tier_mode's return, so the attribute never existed); still parsed but not applied. - Capability fallback: RoutingDecision gains capability_fallback; the worker audits worker_capability_fallback when it fires, so a tier-model rejection followed by a worker-default fallback is no longer invisible (source stays worker_default, no model_routing event, so shadow samples are unaffected). - Registry: resolve_registry_agent checks skills directly against the loaded definition instead of re-parsing a hand-built dict, so future fields cannot be silently dropped the way tool_approvals was in v9. - Shadow: evaluate_shadow sorts rows earliest-first by created_at (id tiebreak) when timestamps are present, so dedup no longer depends on caller row order; the server path (which strips created_at) is unchanged. - Gateway: when sub2api fills the default model, the worker now checks it against the tier's capability requirements with the same mode semantics as v8 -- shadow only audits, active fails fast with worker_capability_gate_rejected and never calls the runner. No worker-default fallback here: the gateway model is the task's gateway model, falling back to a worker default would run against the wrong gateway. --- cloud/agent_def.py | 33 ++--- cloud/models.py | 31 ++++- cloud/routing_shadow.py | 41 ++++-- cloud/tests/test_audit_fixes.py | 233 ++++++++++++++++++++++++++++++++ cloud/worker.py | 88 ++++++++++++ 5 files changed, 392 insertions(+), 34 deletions(-) create mode 100644 cloud/tests/test_audit_fixes.py diff --git a/cloud/agent_def.py b/cloud/agent_def.py index 3b0a766..0a6f6c9 100644 --- a/cloud/agent_def.py +++ b/cloud/agent_def.py @@ -460,25 +460,20 @@ def resolve_registry_agent(name, version=None, agents_dir=None, "registry file %s says version %r, not requested %r" % (filename, definition.version, version)) if known_skills is not None: - # Re-parse for skill existence only (load already validated - # format); the re-parsed definition is the one returned, so - # callers never see a definition that failed skill checks. - definition = parse_agent_definition( - { - "name": definition.name, - "version": definition.version, - "description": definition.description, - "instructions": definition.instructions, - "skills": definition.skills, - "tools": definition.tools, - # v9: the re-parse must carry tool_approvals, or - # a registry definition's declaration would be - # silently dropped on the server path (inline - # definitions never re-parse, so they were fine). - "tool_approvals": definition.tool_approvals, - "defaults": definition.defaults, - }, - known_skills=known_skills) + # Audit fix: skill existence is now checked directly + # against the loaded definition, instead of re-parsing + # a hand-built dict. The old re-parse silently dropped + # any field not copied into that dict (tool_approvals + # was lost exactly this way in v9, and the next new + # field would have been lost the same way). The + # original definition is returned unchanged, so future + # fields survive automatically. The error message is + # kept identical to parse_agent_definition's. + known = set(known_skills) + missing = [s for s in definition.skills if s not in known] + if missing: + raise ValueError("agent.skills unknown skill(s): %s" + % ", ".join(missing)) return definition diff --git a/cloud/models.py b/cloud/models.py index 20f4b53..124543a 100644 --- a/cloud/models.py +++ b/cloud/models.py @@ -336,6 +336,19 @@ class RoutingPolicy: for tier, tier_mode in (tier_modes or {}).items(): if tier in self.tiers and tier_mode in ROUTING_MODES: self.tier_modes[tier] = tier_mode + # Audit fix: fallback_models used to live after the + # return in tier_mode (unreachable dead code), so a + # RoutingPolicy never actually had the attribute and + # any future reader would get AttributeError. It is + # stored here, cleaned and deduplicated, and still + # deliberately NOT applied by the router (see class + # docstring): only worker CLI --fallback-models enables + # a real fallback retry. + self.fallback_models = [] + for item in fallback_models or []: + model = _clean_model_name(item) + if model and model not in self.fallback_models: + self.fallback_models.append(model) self.model_capabilities = parse_model_capabilities( model_capabilities) self.tier_requirements = {} @@ -348,11 +361,6 @@ class RoutingPolicy: def tier_mode(self, tier): """Effective mode for one tier: explicit tier mode, else global.""" return self.tier_modes.get(tier, self.mode) - self.fallback_models = [] - for item in fallback_models or []: - model = _clean_model_name(item) - if model and model not in self.fallback_models: - self.fallback_models.append(model) def parse_routing_policy(data): @@ -466,7 +474,8 @@ class RoutingDecision: """One pre-dispatch model decision, kept auditable by construction.""" def __init__(self, task_id, tier, chosen_model, suggested_model, - source, policy_version, applied, reason, signals): + source, policy_version, applied, reason, signals, + capability_fallback=False): self.task_id = task_id self.tier = tier self.chosen_model = chosen_model @@ -476,6 +485,13 @@ class RoutingDecision: self.applied = applied self.reason = reason self.signals = dict(signals or {}) + # Audit fix: True only when an active tier model failed + # its capability check and the worker default was used + # instead. The source stays "worker_default" (existing + # callers rely on it), so the worker must check this + # flag to audit the fallback -- otherwise the fallback + # is invisible in the audit trail. + self.capability_fallback = bool(capability_fallback) def to_event(self) -> dict: return { @@ -488,6 +504,7 @@ class RoutingDecision: "policy_version": self.policy_version, "applied": self.applied, "reason": self.reason, + "capability_fallback": self.capability_fallback, } @@ -614,7 +631,7 @@ def decide_routing(task, worker_model, policy=None, policy_version=policy.version, applied=True, reason="%s tier model %s%s, worker default satisfies" " requirements, falling back" % (tier, tier_model, cap_note), - signals=signals) + signals=signals, capability_fallback=True) return RoutingDecision( task_id=task.get("id"), tier=tier, chosen_model=None, suggested_model=tier_model, source="capability_blocked", diff --git a/cloud/routing_shadow.py b/cloud/routing_shadow.py index 961e7a2..8fb0c71 100644 --- a/cloud/routing_shadow.py +++ b/cloud/routing_shadow.py @@ -111,18 +111,43 @@ def evaluate_shadow(audit_rows, tasks_by_id, costs_by_task, explicit_n=0, min_samples=MIN_SAMPLES): """Evaluate shadow evidence from already-fetched rows. - ``audit_rows``: dicts with at least ``detail`` (and optionally - ``created_at`` for earliest-first dedup; without it, first-seen - wins). ``tasks_by_id``: task_id -> {"status": str}. - ``costs_by_task``: task_id -> {"total_tokens": int, - "usd_micros": int|None, "unpriced_tokens": int}. - ``explicit_n``: number of tasks with a creation-time model in - the same project (counted separately, never as routed samples). + ``audit_rows``: dicts with at least ``detail``. When rows + carry ``created_at`` (and optionally ``id`` as a tiebreak), + they are sorted earliest-first inside this function before + dedup, so the first-seen sample for a task is its earliest + routing audit regardless of the caller's row order. Rows + without ``created_at`` keep their given order (stable sort). + The server endpoint passes rows already ordered, and it + strips ``created_at`` before calling -- that path is + unaffected, but a future caller passing raw rows no longer + silently changes dedup semantics. ``tasks_by_id``: task_id + -> {"status": str}. ``costs_by_task``: task_id -> + {"total_tokens": int, "usd_micros": int|None, + "unpriced_tokens": int}. ``explicit_n``: number of tasks + with a creation-time model in the same project (counted + separately, never as routed samples). """ tiers = {tier: _empty_tier() for tier in TIERS} seen = set() routed_task_ids = set() - for row in audit_rows or []: + rows = list(audit_rows or []) + if any(isinstance(r, dict) and r.get("created_at") is not None + for r in rows): + def _sort_key(row): + created = row.get("created_at") if isinstance(row, dict) \ + else None + row_id = row.get("id") if isinstance(row, dict) else None + # Rows without a timestamp sort first only if they + # were first already: use a sentinel that keeps + # stable order among themselves, but never jumps + # ahead of a timestamped row -- so put them last. + return (created is None, + created if isinstance(created, (int, float)) + else 0, + row_id if isinstance(row_id, (int, float)) + else 0) + rows = sorted(rows, key=_sort_key) + for row in rows: parsed = parse_worker_routing_detail( (row or {}).get("detail")) if parsed is None: diff --git a/cloud/tests/test_audit_fixes.py b/cloud/tests/test_audit_fixes.py new file mode 100644 index 0000000..e060f57 --- /dev/null +++ b/cloud/tests/test_audit_fixes.py @@ -0,0 +1,233 @@ +"""Audit fixes: fallback_models storage, capability fallback audit, +registry skill check without re-parse, shadow dedup sorting, and +gateway-model capability check. + +Run: python3 -m unittest discover -s cloud/tests -p "test_audit_fixes.py" -v +""" + +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from types import SimpleNamespace + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) + +from cloud.agent_def import default_agents_dir, resolve_registry_agent +from cloud.models import decide_routing, parse_routing_policy +from cloud.routing_shadow import evaluate_shadow +from cloud.server import init_db +from cloud.worker import Worker + + +class FallbackModelsTestCase(unittest.TestCase): + def test_policy_stores_fallback_models(self): + policy = parse_routing_policy({ + "mode": "shadow", + "tiers": {"default": "m1"}, + "fallback_models": ["m2", "m2", " ", "m3"], + }) + # Previously unreachable dead code: the attribute did + # not exist at all on a RoutingPolicy. + self.assertEqual(policy.fallback_models, ["m2", "m3"]) + + +class CapabilityFallbackAuditTestCase(unittest.TestCase): + def test_decision_flags_capability_fallback(self): + policy = parse_routing_policy({ + "mode": "active", + "tiers": {"plan": {"model": "no-tools", + "requires": {"tool_calling": True}}}, + "model_capabilities": { + "no-tools": {"tool_calling": False}, + "worker-model": {"tool_calling": True}, + }, + }) + decision = decide_routing( + {"id": "t1", "plan": True}, "worker-model", policy=policy) + self.assertEqual(decision.source, "worker_default") + self.assertTrue(decision.capability_fallback) + self.assertEqual(decision.chosen_model, "worker-model") + + def test_normal_worker_default_is_not_flagged(self): + policy = parse_routing_policy({ + "mode": "active", "tiers": {"plan": "plan-model"}}) + decision = decide_routing( + {"id": "t1", "plan": True}, "worker-model", policy=policy) + self.assertFalse(decision.capability_fallback) + + +class CapabilityFallbackWorkerAuditTestCase(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.calls = [] + + def tearDown(self): + self.tmp.cleanup() + + def test_fallback_is_audited(self): + _insert_task(self.db_path, "F1") + + def recording(cmd): + self.calls.append(list(cmd)) + return _completed_runner(cmd) + + worker = Worker( + self.db_path, "/fake/host.py", runner=recording, + env={"XCM_API_KEY": "test-placeholder"}, + model="worker-model", model_explicit=False, + routing_policy={ + "mode": "active", + "tiers": {"default": {"model": "no-tools", + "requires": {"tool_calling": True}}}, + "model_capabilities": { + "no-tools": {"tool_calling": False}, + "worker-model": {"tool_calling": True}, + }, + }) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + from cloud.auth import AuthStore + store = AuthStore(self.db_path) + try: + entries = [e for e in store.list_audit("p1") + if e["action"] == "worker_capability_fallback"] + finally: + store.conn.close() + self.assertTrue(entries) + self.assertIn("no-tools", entries[0]["detail"]) + + +class RegistrySkillCheckTestCase(unittest.TestCase): + def test_registry_definition_keeps_all_fields(self): + # The old re-parse dropped any field not copied into + # its hand-built dict; the direct check must return + # the loaded definition unchanged, including + # tool_approvals, and still reject unknown skills. + definition = resolve_registry_agent( + "research-brief-approval", agents_dir=default_agents_dir(), + known_skills=["live-info"]) + self.assertEqual(definition.tool_approvals, + {"websearch": "ask"}) + self.assertEqual(definition.tools, ["websearch"]) + with self.assertRaises(ValueError): + resolve_registry_agent( + "research-brief-approval", + agents_dir=default_agents_dir(), + known_skills=["other-skill"]) + + +class ShadowSortTestCase(unittest.TestCase): + def test_dedup_uses_earliest_row_regardless_of_input_order(self): + rows = [ + {"detail": "t1 tier=default source=shadow applied=False" + " suggested=new-model chosen=worker-model", + "created_at": 200, "id": 2}, + {"detail": "t1 tier=default source=shadow applied=False" + " suggested=old-model chosen=worker-model", + "created_at": 100, "id": 1}, + ] + report = evaluate_shadow( + rows, {"t1": {"status": "completed"}}, {}) + self.assertEqual( + report["tiers"]["default"]["suggested_models"], + {"old-model": 1}) + + +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='{"task": "%s", "status": "completed", "text": "done"}' % name, + stderr="") + + +def _insert_task(db_path, task_id): + 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)" + " VALUES(?, 'p1', 'do work', 'queued', NULL, '[]', 0," + " 'tester', ?, ?, NULL)", (task_id, now, now)) + conn.commit() + finally: + conn.close() + + +class GatewayCapabilityTestCase(unittest.TestCase): + """Gateway check is exercised through the worker's routing + policy path by stubbing _ensure_gateway: the worker's + run_one gateway step is the only place the sub2api + default model is filled, so the check must live there.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.calls = [] + + def tearDown(self): + self.tmp.cleanup() + + def _worker(self, policy_doc, gateway_model): + def recording(cmd): + self.calls.append(list(cmd)) + return _completed_runner(cmd) + + worker = Worker(self.db_path, "/fake/host.py", + runner=recording, + env={"XCM_API_KEY": "test-placeholder"}, + routing_policy=policy_doc) + # Stub the gateway to behave like sub2api filling a + # default model, without touching real gateway code. + worker._ensure_gateway = lambda task: ("sub2api", {}) + worker._sub2api_default_model = lambda env: gateway_model + return worker + + def test_active_gateway_incapable_blocks_before_runner(self): + _insert_task(self.db_path, "G1") + worker = self._worker({ + "mode": "active", + "tiers": {"default": {"model": "policy-model", + "requires": {"tool_calling": True}}}, + "model_capabilities": { + "gateway-model": {"tool_calling": False}, + }, + }, gateway_model="gateway-model") + result = worker.run_one() + self.assertTrue(result["dispatched"]) + self.assertEqual(result["status"], "failed") + self.assertIn("tool_calling", result["error"]) + self.assertEqual(self.calls, []) + + def test_shadow_gateway_incapable_only_audits(self): + _insert_task(self.db_path, "G2") + worker = self._worker({ + "mode": "shadow", + "tiers": {"default": {"model": "policy-model", + "requires": {"tool_calling": True}}}, + "model_capabilities": { + "gateway-model": {"tool_calling": False}, + }, + }, gateway_model="gateway-model") + result = worker.run_one() + self.assertEqual(result["status"], "completed") + from cloud.auth import AuthStore + store = AuthStore(self.db_path) + try: + entries = [e for e in store.list_audit("p1") + if e["action"] + == "worker_capability_gateway_shadow"] + finally: + store.conn.close() + self.assertTrue(entries) diff --git a/cloud/worker.py b/cloud/worker.py index 0d7b52b..b0d1aa7 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -156,6 +156,7 @@ from cloud.db import connect as _db_connect from cloud.goal_command import parse_goal_command as _parse_goal_command 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 from cloud.auto_memory import AutoMemoryStore @@ -6019,10 +6020,12 @@ class Worker: # xcmapi 模型名去撞 sub2api）。xcmapi 下 _ensure_gateway # 直接返回 (None, None)，不做任何事。注意：这里改的是 # 原 task dict，随后 task = dict(task) 会原样拷走。 + _gateway_filled = False try: gateway_provider, gateway_env = self._ensure_gateway(task) if not task.get("model") and gateway_provider == "sub2api": task["model"] = self._sub2api_default_model(gateway_env) + _gateway_filled = True except Exception as exc: error = "网关准备失败: %s" % exc self._set_status(task_id, "failed") @@ -6044,6 +6047,78 @@ class Worker: "stdout": "", "error": error, "events": [], "cost_record_id": None} + # Audit fix：网关能力检查——此前路由决策只在模型 + # 为空时跑，sub2api 网关先填默认模型后决策直接跳 + # 过，能力门在生产路径上完全不生效。补齐：网关填 + # 的模型同样按档需求检查，档模式与 v8 一致：shadow + # 只审计不拦，active 不满足则 fail-fast（同款工具 + # 门形状，清占位、不调 runner）。没有策略、没有档 + # 需求、档模式 off 时一律不查，旧行为不变。注意： + # 这里没有 worker 默认可回落（网关模型就是该任务 + # 的网关内模型，换成 worker 默认会跑错网关），所以 + # active 下不满足即拦，不存在“先试 worker 默认” + # 的中间态；运营方须先把网关模型写进能力目录， + # 否则 unknown 同样拦（fail-closed，与 v8 一致）。 + if _gateway_filled and self.routing_policy 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_ok, _gw_missing, _gw_unknown = \ + check_model_capabilities( + self.routing_policy.model_capabilities, + task.get("model"), _gw_requires) + if not _gw_ok: + _gw_note = ("gateway model %s %s capability:" + " %s" % ( + task.get("model"), + "cannot verify" + if _gw_unknown else "lacks", + ", ".join(_gw_missing))) + if _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_error = ( + "%s tier gateway model %s; refusing" + " to dispatch an incapable or" + " unverifiable model" + % (_gw_tier, _gw_note)) + 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") + 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} + task = dict(task) if not task.get("model"): # 规则路由 v0（只在模型仍为空时跑一次，整个派发只算一次， @@ -6103,6 +6178,19 @@ class Worker: decision.applied, decision.suggested_model or "-", decision.chosen_model or "-")) + # Audit fix：能力回落以前不记任何审计（source 是 + # worker_default，不在 policy/shadow 之列），运营 + # 方看不到档模型被能力门否掉过。现在用 decision + # 上的 capability_fallback 标志单独记一条审计， + # 不改 source、不记 model_routing 事件（影子样本 + # 只算 policy/shadow 决策，回落不是一次路由决策）。 + if getattr(decision, "capability_fallback", False): + self._audit(task["project_id"], + "worker_capability_fallback", + "%s tier=%s chosen=%s reason=%s" % ( + task_id, decision.tier, + decision.chosen_model or "-", + decision.reason)) task["model"] = decision.chosen_model or self.model # agent 指令通道（v2）：local/docker 由 sandbox 的 # --developer-instructions 派发，prompt 保持干净；通用