From cdac8039b3a68696c142bbe84c1f95114ed5110c Mon Sep 17 00:00:00 2001 From: host Date: Mon, 2 Nov 2026 16:29:20 +0000 Subject: [PATCH] Audit batch 4: I-6 whitespace audit format, I-7 explicit non-sub2api capability gate, I-8 coord gate attribution, I-10 checked gates, I-11 gate-without-finding escalate, I-13 structured-first classification --- cloud/audit_loop.py | 29 ++ cloud/models.py | 65 ++- cloud/routing_shadow.py | 21 +- cloud/tests/test_audit_batch4.py | 673 +++++++++++++++++++++++++++++++ cloud/worker.py | 218 ++++++++-- 5 files changed, 959 insertions(+), 47 deletions(-) create mode 100644 cloud/tests/test_audit_batch4.py diff --git a/cloud/audit_loop.py b/cloud/audit_loop.py index c2570f9..40c17fe 100644 --- a/cloud/audit_loop.py +++ b/cloud/audit_loop.py @@ -231,9 +231,16 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None): 本轮全部门的 {门名: 是否通过}，必须含全部已跑过的门；findings 为本轮确定性/主观发现；fixed 为 {finding_id: 已修复证据}，调用方 以此证明上一轮 finding 已修复（无证据不得丢标准）。 + 注意：fixed 仅供测试/未来调用方——生产两条接线（单任务 + _audit_loop_wire、协调员 _coord_audit_loop_merge）均不传 + fixed，只认重报与门通过；不要顺手给生产接线加 fixed + 语义，那是另一项功能。 返回决策 dict：decision 为 pass / pushback / escalate / needs_review 之一；packet 只在 pushback 时有值；new_state 只在 pass / pushback / escalate 时有值（needs_review 不写状态）。 + 有失败门但无 finding 时返回 escalate（reason 为 + gate_failed_without_finding），不再走到 PushbackPacket 抛 + ValueError——调用方拿到的是决策，不是异常。 """ findings = list(findings or []) fixed = dict(fixed or {}) @@ -329,6 +336,28 @@ def evaluate(state, task_id, gate_results, findings=None, fixed=None): # 有 finding 但门全过、finding 为空的矛盾态：门说通过、发现说有 # 问题，以 finding 为准（finding 是审计的直接证据）。 + # 审计批 4（I-11）：有失败门但无 finding 时，旧代码走到 + # 下面 PushbackPacket 抛 ValueError（findings 为空）—— + # 调用方拿到的是异常而不是决策。显式升级：门说失败、 + # 发现说没问题，无法构成推回包，reason 点名缺 finding； + # 状态正常保存为 escalate（不再像异常路径那样回滚）。 + # 生产两条接线实际走不到这支（单任务接线在终局非 + # completed 时必有 execution_failed finding，协调员合流 + # 为失败子任务必有 finding），它是直接调用方的契约修复。 + # 注意 wire 层原来看到的是 evaluate_failed 升级（异常 + # 转 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": [], + } + if prev_round >= MAX_PUSHBACKS: new_state = LoopState( task_id, round=prev_round, passed_gates=sorted(passed_now), diff --git a/cloud/models.py b/cloud/models.py index 095048a..1a92b5f 100644 --- a/cloud/models.py +++ b/cloud/models.py @@ -155,7 +155,11 @@ def parse_model_capabilities(raw): return out for model, entry in raw.items(): model = _clean_model_name(model) - if not model or not isinstance(entry, dict): + # 审计批 4（I-6）：能力目录键含空白时整条丢弃—— + # 含空白的键在空白分隔的审计 detail 里无法被消费点 + # 稳健识别，保留它只会让能力检查静默走 unknown。 + if not model or _model_name_has_whitespace(model) \ + or not isinstance(entry, dict): continue cleaned = {} valid = True @@ -432,17 +436,28 @@ def classify_task(task) -> str: def _clean_model_name(value): - # Audit fix 3（M7，文档级）：只 strip、不校验字符集。 - # 含空白的模型名仍会原样进审计 detail，而影子审计 - # 格式是空白分隔，整行解析失败即无声丢样本（不报错、 - # 不计数）。这里不加字符集校验——那会改变模型名接受 - # 语义、可能拦住合法模型名；风险以文档记录，修复 - # 需另起一轨（审计格式改分隔或模型名白名单）。 + """把模型名规范成可比较的字符串：仅 strip。 + + 注意：这里刻意不做空白白名单。decide_routing 里清洗后 + 的 None 会被当成“无显式模型”而改走策略/worker 默认路由， + 全局白名单会把“显式模型含空白（目前原样派发、provider + 侧 loud fail）”悄悄改成“静默替换用户选型”，比现状更差。 + 空白模型名的处理因此分两层（审计批 4，I-6）： + (a) 策略写入路径白名单——能力目录键、档模型、 + fallback_models 含空白的一律丢弃；(b) 运行时消费点 + 稳健解析——影子 chosen 贪到行尾、_is_persisted_model + 按键提取 model 值，不再依赖空白分隔的位置。 + """ if isinstance(value, str) and value.strip(): return value.strip() return None +def _model_name_has_whitespace(name): + """清洗后的模型名是否含空白（审计批 4，I-6 白名单判定）。""" + return isinstance(name, str) and any(ch.isspace() for ch in name) + + def _clean_tiers(raw): """Return {tier: model} keeping only valid non-empty entries.""" tiers = {} @@ -450,13 +465,17 @@ def _clean_tiers(raw): return tiers for tier in ROUTING_TIERS: model = _clean_model_name(raw.get(tier)) - if isinstance(model, str): + # 审计批 4（I-6）：档模型含空白时整档丢弃（parse 层 + # 另行记入 dropped_tiers + 警告），不让含空白的模型 + # 名进空白分隔的审计 detail。 + if isinstance(model, str) \ + and not _model_name_has_whitespace(model): # A tier entry may itself be a dict {"model": ...}; accept # that shape too so policy authors are not punished for it. tiers[tier] = model elif isinstance(raw.get(tier), dict): model = _clean_model_name(raw[tier].get("model")) - if model: + if model and not _model_name_has_whitespace(model): tiers[tier] = model return tiers @@ -501,7 +520,12 @@ class RoutingPolicy: self.version = version if isinstance(version, str) \ and version.strip() else None self.mode = mode if mode in ROUTING_MODES else "off" - self.tiers = dict(tiers or {}) + # 审计批 4（I-6）：直接构造的策略同样过白名单——含 + # 空白的档模型进不了空白分隔的审计 detail，整档丢弃 + # （与 parse 层口径一致；parse 层另行记 dropped_tiers）。 + self.tiers = {tier: model for tier, model in (tiers or {}).items() + if not (isinstance(model, str) + and _model_name_has_whitespace(model))} self.tier_modes = {} for tier, tier_mode in (tier_modes or {}).items(): if tier in self.tiers and tier_mode in ROUTING_MODES: @@ -517,7 +541,9 @@ class RoutingPolicy: self.fallback_models = [] for item in fallback_models or []: model = _clean_model_name(item) - if model and model not in self.fallback_models: + # 审计批 4（I-6）：fallback 模型名同款白名单。 + if model and not _model_name_has_whitespace(model) \ + and model not in self.fallback_models: self.fallback_models.append(model) self.model_capabilities = parse_model_capabilities( model_capabilities) @@ -598,6 +624,23 @@ def parse_routing_policy(data): # for an empty model) is not a second drop and is not logged: # the warning must name a tier this pop really removed. dropped_tiers = [] + # 审计批 4（I-6）：_clean_tiers 已把含空白模型名的档 + # 丢掉，这里只记名——与无效模式/无效 requires 的 pop + # 同款可见性。只认 ROUTING_TIERS 内的档（未知档名本 + # 来就被 _clean_tiers 忽略，不是“被丢弃”）；档已被 + # _clean_tiers 滤掉时，后面的 mode/requires pop 因 + # `tier in tiers` 为假不会重复记，不会双计。 + for tier, entry in raw_tiers.items(): + if tier not in ROUTING_TIERS or tier in tiers: + continue + raw_model = entry if isinstance(entry, str) else \ + entry.get("model") if isinstance(entry, dict) else None + cleaned = _clean_model_name(raw_model) + if cleaned and _model_name_has_whitespace(cleaned): + dropped_tiers.append(tier) + _logger.warning( + "routing policy tier %r dropped: model name contains" + " whitespace %r", tier, cleaned) tier_modes = {} for tier, entry in raw_tiers.items(): if not isinstance(entry, dict) or "mode" not in entry: diff --git a/cloud/routing_shadow.py b/cloud/routing_shadow.py index 8e7ed1e..67f67a9 100644 --- a/cloud/routing_shadow.py +++ b/cloud/routing_shadow.py @@ -60,7 +60,16 @@ _DETAIL_RE = re.compile( r"^(?P\S+)\s+tier=(?P\S+)\s+source=(?P\S+)" r"\s+applied=(?P\S+)" r"(?:\s+suggested=(?P\S+))?" - r"(?:\s+chosen=(?P\S+))?\s*$") + # Audit batch 4 (I-6): chosen is the LAST field, so it may + # greedily run to end of line — a legacy chosen model name + # containing whitespace ("my model") now parses instead of + # dropping the whole row. suggested stays \S+ (it is a middle + # field); new policy models cannot contain whitespace (the + # batch-4 policy whitelist drops them), so new rows always + # parse, and legacy suggested-with-whitespace rows still land + # in the unparseable bucket (counted, batch 2) rather than + # being silently lost. + r"(?:\s+chosen=(?P.+?))?\s*$") def parse_worker_routing_detail(detail): @@ -70,9 +79,11 @@ def parse_worker_routing_detail(detail): A ``-`` placeholder also means None, never a model named "-". Audit batch 2 (I-1): 解析语义本身不变（失败仍返回 None）， 但 evaluate_shadow 不再静默跳过：不可解析的行计入 - dropped_rows["unparseable"]，报告里可见（例如模型名含 - 空白会使整行匹配失败，见 M7——行不再无声消失，只是 - 不计为样本）。 + dropped_rows["unparseable"]，报告里可见。 + Audit batch 4 (I-6): chosen 是行尾字段，含空白的 + legacy chosen 模型名现在能解析（见 _DETAIL_RE 注释）； + suggested 仍要求无空白，legacy suggested 含空白的行 + 仍进 unparseable 桶计数，不静默消失。 """ if not isinstance(detail, str): return None @@ -89,7 +100,7 @@ def parse_worker_routing_detail(detail): "source": match.group("source"), "applied": applied, "suggested_model": None if suggested in (None, "-") else suggested, - "chosen_model": None if chosen in (None, "-") else chosen, + "chosen_model": None if chosen in (None, "-") else chosen.strip(), } diff --git a/cloud/tests/test_audit_batch4.py b/cloud/tests/test_audit_batch4.py new file mode 100644 index 0000000..99deaaf --- /dev/null +++ b/cloud/tests/test_audit_batch4.py @@ -0,0 +1,673 @@ +"""审计修复批 4 测试：I-6、I-7、I-8、I-10、I-11、I-13。 + +只用注入的假 runner 与 init_db 的本地库，不跑真实模型回合、 +不联网、不登录任何服务；env 里的值只是占位用的假名。 + +行为变更锁定（本批有意改语义，测试即契约）： +- I-6：策略写入路径（能力目录键、档模型、fallback_models） + 含空白的一律丢弃；影子行 chosen（行尾字段）含空白可解析； + persisted 行模型含空白按鍵提取、复核不再默认显式 bypass。 +- I-7：显式 + 非 sub2api 网关在 active 下已知不满足时从 + “直接派发”变为“拦”（同款审计 action、error 文案、 + blocked 标志）；unknown-only 与 shadow 仍只审计。 +- I-8：缺 key / 仓库门审计 detail 带任务 id 前缀，补进门拒绝 + 查询集合；仓库解析失败子任务归 gate_rejected、父走升级。 +- I-10：gate_results 只列本轮实际跑过的门；显式与政策任务 + 两轮往返均不误触发 gate_missing。 +- I-11：失败门 + 空 findings 时 evaluate 返回 escalate + （gate_failed_without_finding），不再抛 ValueError。 +- I-13：分类优先读结构化审计行，子串只作最后回落且回落 + 命中时记 worker_coord_gate_classified_by_marker 审计行。 +""" + +import json +import logging +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.audit_loop import Finding, evaluate + from cloud.auth import AuthStore + from cloud.models import ( + RoutingPolicy, evaluate_capability, parse_model_capabilities, + parse_routing_policy) + from cloud.routing_shadow import ( + evaluate_shadow, parse_worker_routing_detail) + from cloud.server import init_db + from cloud.worker import Worker, _GATE_ERROR_MARKERS +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.audit_loop import Finding, evaluate + from cloud.auth import AuthStore + from cloud.models import ( + RoutingPolicy, evaluate_capability, parse_model_capabilities, + parse_routing_policy) + from cloud.routing_shadow import ( + evaluate_shadow, parse_worker_routing_detail) + from cloud.server import init_db + from cloud.worker import Worker, _GATE_ERROR_MARKERS + + +def _insert_task(db_path, task_id, model=None, status="queued", + project_id="p1", audit_loop=0, parent=None, part=None, + error=None, result=None): + now = time.time() + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat, audit_loop," + " parent_task_id, coordinator_part, result_text, error_text," + " model)" + " VALUES(?, ?, 'do work', ?, NULL, '[]', 0, 'tester'," + " ?, ?, NULL, ?, ?, ?, ?, ?, ?)", + (task_id, project_id, status, now, now, audit_loop, + parent, part, result, error, model), + ) + conn.commit() + finally: + conn.close() + + +def _set_started_at(db_path, task_id, started_at): + conn = sqlite3.connect(db_path) + try: + conn.execute( + "UPDATE tasks SET started_at = ? WHERE id = ?", + (started_at, task_id)) + conn.commit() + finally: + conn.close() + + +def _insert_audit(db_path, project_id, action, detail, created_at=None): + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO audit_log(user_id, project_id, action, detail," + " created_at) VALUES(NULL, ?, ?, ?, ?)", + (project_id, action, detail, + created_at if created_at is not None else time.time())) + conn.commit() + finally: + conn.close() + + +def _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 _completed_runner(cmd): + name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t" + return SimpleNamespace( + returncode=0, + stdout=json.dumps({"task": name, "status": "completed", + "text": "done"}), + stderr="") + + +def _failing_runner(cmd): + name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t" + return SimpleNamespace( + returncode=0, + stdout=json.dumps({"task": name, "status": "failed", + "text": "boom"}), + stderr="") + + +def _worker(db_path, policy_doc=None, env=None, gateway=None, + runner=None): + calls = [] + + def recording(cmd): + calls.append(list(cmd)) + return (runner or _completed_runner)(cmd) + + worker = Worker(db_path, "/fake/host.py", runner=recording, + env=env or {"XCM_API_KEY": "test-placeholder"}, + routing_policy=policy_doc) + if gateway is not None: + worker._ensure_gateway = lambda task: gateway + worker._sub2api_default_model = lambda e: "gateway-model" + worker._calls = calls + return worker + + +def _state_row(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + return conn.execute( + "SELECT round, passed_gates, last_decision" + " FROM task_audit_loop_state WHERE task_id = ?", + (task_id,)).fetchone() + finally: + conn.close() + + +def _task_row(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + return conn.execute( + "SELECT status, error_text FROM tasks WHERE id = ?", + (task_id,)).fetchone() + finally: + conn.close() + + +class WhitespacePolicyTestCase(unittest.TestCase): + """I-6：策略写入路径白名单。""" + + def test_whitespace_tier_model_dropped_and_named(self): + with self.assertLogs("cloud.models", level="WARNING") as logs: + policy = parse_routing_policy({ + "mode": "active", + "tiers": {"plan": "my model", "default": "ok-model"}, + }) + self.assertIsNotNone(policy) + self.assertEqual(policy.tiers, {"default": "ok-model"}) + self.assertEqual(policy.dropped_tiers, ["plan"]) + self.assertTrue(any("plan" in line for line in logs.output)) + + def test_whitespace_capability_key_dropped_is_unknown(self): + caps = parse_model_capabilities({ + "my model": {"tool_calling": True}, + "ok": {"tool_calling": True}, + }) + self.assertEqual(caps, {"ok": {"tool_calling": True}}) + verdict = evaluate_capability(caps, "my model", + {"tool_calling": True}) + self.assertFalse(verdict.ok) + self.assertTrue(verdict.unknown) + self.assertEqual(verdict.unknown_missing, ["tool_calling"]) + self.assertEqual(verdict.known_missing, []) + + def test_whitespace_fallback_and_direct_tiers_dropped(self): + policy = RoutingPolicy( + tiers={"default": "my model", "plan": "ok"}, + fallback_models=["my model", "ok"]) + self.assertEqual(policy.tiers, {"plan": "ok"}) + self.assertEqual(policy.fallback_models, ["ok"]) + + def test_shadow_row_with_whitespace_chosen_parses(self): + parsed = parse_worker_routing_detail( + "T9 tier=default source=policy applied=true" + " suggested=- chosen=my model") + self.assertIsNotNone(parsed) + self.assertEqual(parsed["chosen_model"], "my model") + # 进样本计数，不再整行丢弃。 + report = evaluate_shadow([ + {"detail": "T9 tier=default source=policy applied=true" + " suggested=- chosen=my model", + "created_at": 1000.0}], {}, {}) + self.assertEqual(report["routed_task_n"], 1) + self.assertEqual(report["dropped_rows"]["unparseable"], 0) + + def test_legacy_suggested_whitespace_still_counted_not_lost(self): + # suggested 是中间字段，仍要求无空白：legacy 含空白 + # suggested 的行解析失败，但按批 2 计入丢行桶，不静默消失。 + parsed = parse_worker_routing_detail( + "T9 tier=default source=policy applied=true" + " suggested=my model chosen=ok") + self.assertIsNone(parsed) + report = evaluate_shadow([ + {"detail": "T9 tier=default source=policy applied=true" + " suggested=my model chosen=ok", + "created_at": 1000.0}], {}, {}) + self.assertEqual(report["routed_task_n"], 0) + self.assertEqual(report["dropped_rows"]["unparseable"], 1) + + +class PersistedWhitespaceModelTestCase(unittest.TestCase): + """I-6：含空白写回模型的复核不再被 bypass。""" + + def setUp(self): + self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + store = AuthStore(self.db_path) + store.conn.close() + + def tearDown(self): + if self._old is None: + os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) + else: + os.environ[CODEX_PUSH_ALLOW_PRIVATE_TARGETS] = self._old + self.tmp.cleanup() + + def _policy(self): + return { + "mode": "active", + "tiers": {"default": {"model": "policy-model", + "requires": {"tool_calling": True}}}, + "model_capabilities": { + "my model": {"tool_calling": False}, + }, + } + + def test_persisted_whitespace_model_matches_and_is_rechecked(self): + _insert_audit(self.db_path, "p1", "worker_model_persisted", + "T1 tier=default model=my model") + worker = _worker(self.db_path, self._policy()) + audits = [] + worker._audit = lambda p, a, d="": audits.append((p, a, d)) + # 旧 split()[2] 定位置取在这里永远匹配不上（model + # 含空白被切成两段）；新按键提取必须认出写回模型。 + self.assertTrue(worker._is_persisted_model( + "T1", "p1", "my model")) + # 复核必须真正跑起来、不再默认显式 bypass：含空白的 + # 模型名进不了能力目录（I-6 白名单已丢弃该键），故 + # 复核结果是 unknown——按 fix5 分档只审计不拦；断言 + # 点是复核确实跑了（有 recheck_unknown 审计行）， + # 而不是悄悄按显式放行（那时连审计行都没有）。 + # active 下 known 非空才拦，空白模型永远到不了 + # known（目录里没有它的条目），这是白名单的直接 + # 推论，不是复核 bypass。 + error = worker._recheck_persisted_model({ + "id": "T1", "project_id": "p1", "model": "my model"}) + self.assertIsNone(error) + self.assertTrue([a for _p, a, _d in audits + if a == "worker_capability_recheck_unknown"]) + self.assertTrue(any("my model" in d for _p, a, d in audits + if a == "worker_capability_recheck_unknown")) + + def test_persisted_whitespace_model_mismatch_is_not_persisted(self): + _insert_audit(self.db_path, "p1", "worker_model_persisted", + "T1 tier=default model=my model") + worker = _worker(self.db_path, self._policy()) + self.assertFalse(worker._is_persisted_model( + "T1", "p1", "other model")) + self.assertFalse(worker._is_persisted_model( + "T2", "p1", "my model")) + + +class ExplicitNonSub2apiGateTestCase(unittest.TestCase): + """I-7：显式 + 非 sub2api 网关的能力门。""" + + def setUp(self): + self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + + def tearDown(self): + if self._old is None: + os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) + else: + os.environ[CODEX_PUSH_ALLOW_PRIVATE_TARGETS] = self._old + self.tmp.cleanup() + + 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_non_sub2api_is_blocked(self): + # 行为变更：旧行为是直接派发（能力门只在 sub2api + # 时跑），新行为是 active 下已知不满足即拦。 + _insert_task(self.db_path, "E1", model="explicit-model") + worker = _worker(self.db_path, self._policy("active", { + "explicit-model": {"tool_calling": False}, + }), gateway=("other-gw", {})) + result = worker.run_one() + self.assertEqual(result["status"], "failed") + self.assertFalse(worker._calls) + self.assertIn("lacks capability", 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_non_sub2api_is_audited_not_blocked(self): + _insert_task(self.db_path, "E1", model="explicit-model") + worker = _worker(self.db_path, self._policy("active", {}), + gateway=("other-gw", {})) + 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_shadow_explicit_non_sub2api_is_audited_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=("other-gw", {})) + 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]) + + def test_nonexplicit_non_sub2api_has_no_gateway_gate(self): + # 非显式 + 非 sub2api 根本不进网关能力块：模型为空， + # 后续走 decide_routing，不存在独立门误拦。 + _insert_task(self.db_path, "E1") + worker = _worker(self.db_path, self._policy("active", { + "policy-model": {"tool_calling": True}, + }), gateway=("other-gw", {})) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + actions = _audit_actions(self.db_path) + self.assertFalse([a for a, _ in actions + if a == "worker_capability_gateway_explicit"]) + + def test_explicit_no_gateway_keeps_explicit_semantics(self): + # 边界：显式 + 无网关（provider 为 None）不进网关 + # 能力块——它根本没走网关，显式语义不变：已知不满足 + # 也不在这里拦（写回模型仍走后续复核，用户显式仍 + # 不被复核，fixes5 契约）。无网关显式任务被网关块 + # 拦住会是语义倒退，不是 I-7 的意图。 + _insert_task(self.db_path, "E1", model="explicit-model") + worker = _worker(self.db_path, self._policy("active", { + "explicit-model": {"tool_calling": False}, + }), gateway=(None, {})) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + self.assertTrue(worker._calls) + actions = _audit_actions(self.db_path) + self.assertFalse([a for a, _ in actions + if a == "worker_capability_gateway_explicit"]) + self.assertFalse([a for a, _ in actions + if a == "worker_capability_gate_rejected"]) + + def test_explicit_sub2api_path_unchanged(self): + # sub2api 显式路径与原来逐字一致：同一审计 action、 + # 同一 error 文案、同一 blocked 标志。 + _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("lacks capability", 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]) + + +class CoordGateAttributionTestCase(unittest.TestCase): + """I-8 + I-13：协调员门拒绝归因与分类回落可见性。""" + + def setUp(self): + self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + store = AuthStore(self.db_path) + store.conn.close() + self.audits = [] + self.worker = Worker(self.db_path, "/fake/host.py", + runner=_completed_runner, + env={"XCM_API_KEY": "test-placeholder"}, + max_retries=0, retry_backoff_seconds=0) + self.worker._audit = lambda p, a, d="": self.audits.append( + (p, a, d)) + conn = sqlite3.connect(self.db_path) + try: + self.worker._ensure_result_columns(conn) + conn.commit() + finally: + conn.close() + + def tearDown(self): + if self._old is None: + os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) + else: + os.environ[CODEX_PUSH_ALLOW_PRIVATE_TARGETS] = self._old + self.tmp.cleanup() + + def _merge(self, child_error, audit_rows=()): + now = time.time() + _insert_task(self.db_path, "P1", status="running", audit_loop=1, + parent=None) + _insert_task(self.db_path, "C1", status="failed", parent="P1", + part="a", error=child_error) + _set_started_at(self.db_path, "C1", now - 100) + for action, detail in audit_rows: + _insert_audit(self.db_path, "p1", action, detail, + created_at=now) + out = self.worker._coord_audit_loop_merge( + {"id": "P1", "project_id": "p1", "audit_loop": 1}, + [{"name": "a", "prompt": "do a"}], + [{"id": "C1", "name": "a"}]) + return out + + def test_repo_failure_child_is_gate_rejected_not_execution(self): + out = self._merge( + "仓库解析失败: bad repo", + audit_rows=[("worker_repo_failed", + "C1: 仓库解析失败: bad repo")]) + self.assertEqual(out["decision"], "escalate") + self.assertIn("a", out.get("gate_rejected", [])) + self.assertEqual(_task_row(self.db_path, "C1")[0], "failed") + # 结构化证据已命中，不走子串回落、无回落审计行。 + self.assertFalse([a for _p, a, _d in self.audits + if a == "worker_coord_gate_classified_by_marker"]) + + def test_missing_key_child_is_gate_rejected_not_execution(self): + out = self._merge( + "Missing environment variable: OPENKUN_API_KEY", + audit_rows=[("worker_missing_key", + "C1: Missing environment variable:" + " OPENKUN_API_KEY")]) + self.assertEqual(out["decision"], "escalate") + self.assertIn("a", out.get("gate_rejected", [])) + self.assertEqual(_task_row(self.db_path, "C1")[0], "failed") + + def test_recheck_rejected_child_is_gate_rejected(self): + out = self._merge( + "default tier persisted model m lacks capability", + audit_rows=[("worker_capability_recheck_rejected", + "C1: default tier persisted model m lacks" + " capability")]) + self.assertEqual(out["decision"], "escalate") + self.assertIn("a", out.get("gate_rejected", [])) + + def test_structured_evidence_wins_over_reworded_error(self): + # error_text 措辞改掉（不含任何 marker 子串），结构化 + # 审计行不变时分类仍为 gate_rejected：分类不再依赖文案。 + out = self._merge( + "repo exploded unexpectedly", + audit_rows=[("worker_repo_failed", + "C1: 仓库解析失败: bad repo")]) + self.assertEqual(out["decision"], "escalate") + self.assertIn("a", out.get("gate_rejected", [])) + self.assertFalse([a for _p, a, _d in self.audits + if a == "worker_coord_gate_classified_by_marker"]) + + def test_marker_fallback_is_audited(self): + # 删结构化审计行、仅留 error_text 时分类走回落， + # 且合流里有回落审计行点名 marker。 + out = self._merge("必需工具缺失: playwright") + self.assertEqual(out["decision"], "escalate") + self.assertIn("a", out.get("gate_rejected", [])) + marker_rows = [d for _p, a, d in self.audits + if a == "worker_coord_gate_classified_by_marker"] + self.assertTrue(marker_rows) + self.assertIn("C1", marker_rows[0]) + self.assertIn("marker=必需工具缺失", marker_rows[0]) + + def test_gate_lookup_failure_still_escalates(self): + # I-5 回归：审计表缺失导致查询失败时合流走 escalate， + # 不改回空集推回（与批 2 测试同款口径，本夹具先删表）。 + _insert_task(self.db_path, "P1", status="running", audit_loop=1) + _insert_task(self.db_path, "C1", status="failed", parent="P1", + part="a", error="boom") + conn = sqlite3.connect(self.db_path) + try: + conn.execute("DROP TABLE IF EXISTS audit_log") + conn.commit() + finally: + conn.close() + out = self.worker._coord_audit_loop_merge( + {"id": "P1", "project_id": "p1", "audit_loop": 1}, + [{"name": "a", "prompt": "do a"}], + [{"id": "C1", "name": "a"}]) + self.assertEqual(out["decision"], "escalate") + self.assertTrue(out.get("gate_lookup_failed")) + self.assertEqual(_task_row(self.db_path, "C1")[0], "failed") + + +class GateResultsCheckedTestCase(unittest.TestCase): + """I-10：gate_results 只列实际跑过的门 + 两轮不误升级。""" + + def setUp(self): + self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + + def tearDown(self): + if self._old is None: + os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) + else: + os.environ[CODEX_PUSH_ALLOW_PRIVATE_TARGETS] = self._old + self.tmp.cleanup() + + def test_direct_call_checked_subset_and_legacy(self): + worker = _worker(self.db_path) + task = {"id": "T1", "project_id": "p1"} + gates, _ = worker._audit_loop_gate_results( + task, "failed", {"error": "boom"}, gates_checked=["gateway"]) + self.assertEqual(gates, {"execution": False, "gateway": True}) + gates, _ = worker._audit_loop_gate_results( + task, "failed", {"error": "boom"}, gates_checked=[]) + self.assertEqual(gates, {"execution": False}) + # 未传 gates_checked 的旧直接调用保持旧行为。 + gates, _ = worker._audit_loop_gate_results( + task, "failed", {"error": "boom"}) + self.assertEqual(gates["capability"], True) + self.assertEqual(gates["required_tools"], True) + self.assertEqual(gates["gateway"], True) + + def test_policy_task_two_rounds_pushback_without_gate_missing(self): + _insert_task(self.db_path, "T2", audit_loop=1) + worker = _worker(self.db_path, { + "mode": "active", + "tiers": {"default": {"model": "policy-model", + "requires": {"tool_calling": True}}}, + "model_capabilities": { + "policy-model": {"tool_calling": True}, + }, + }, gateway=(None, {}), runner=_failing_runner) + for _ in range(2): + out = worker.run_one() + self.assertEqual(out["status"], "queued") + row = _state_row(self.db_path, "T2") + self.assertIsNotNone(row) + self.assertEqual(row[0], 2) + # 政策任务本轮实际跑过的门只有 capability（网关桩 + # 回 None、无 agent_tools 故工具门未跑），状态里 + # passed_gates 不含没跑过的门。 + self.assertEqual(json.loads(row[1]), ["capability"]) + self.assertEqual(row[2], "pushback") + + def test_explicit_task_two_rounds_pushback_without_gate_missing(self): + _insert_task(self.db_path, "T3", model="explicit-model", + audit_loop=1) + worker = _worker(self.db_path, gateway=(None, {}), + runner=_failing_runner) + for _ in range(2): + out = worker.run_one() + self.assertEqual(out["status"], "queued") + row = _state_row(self.db_path, "T3") + self.assertIsNotNone(row) + self.assertEqual(row[0], 2) + self.assertEqual(row[2], "pushback") + + +class EvaluateNoFindingTestCase(unittest.TestCase): + """I-11：失败门 + 空 findings 返回 escalate，不抛异常。""" + + def test_failed_gate_without_finding_escalates(self): + out = evaluate(None, "T1", {"execution": False}, findings=[]) + self.assertEqual(out["decision"], "escalate") + self.assertEqual(out["reason"], "gate_failed_without_finding") + self.assertEqual(out["round"], 0) + self.assertIsNone(out["packet"]) + self.assertEqual(out["new_state"].round, 0) + self.assertEqual(out["new_state"].last_decision, "escalate") + + def test_failed_gate_without_finding_keeps_round(self): + from cloud.audit_loop import LoopState + state = LoopState("T1", round=2, passed_gates=["execution"], + criteria={}, last_decision="pushback") + # 注意：上一轮通过的门本轮变失败会生成 regression + # finding（all_findings 非空），走不到本支；所以 + # 本例让失败的是上一轮未通过过的门（capability）， + # 且上一轮通过的 execution 本轮仍通过——无回归、 + # 无缺门、无弱化、无 finding，只有失败门。 + out = evaluate(state, "T1", + {"execution": True, "capability": False}, + findings=[]) + self.assertEqual(out["decision"], "escalate") + self.assertEqual(out["reason"], "gate_failed_without_finding") + self.assertEqual(out["new_state"].round, 2) + + def test_finding_still_pushbacks_and_max_rounds_unchanged(self): + finding = Finding("execution_failed", "high", "boom", "re-run", + "task completes", kind="deterministic", + gate="execution") + out = evaluate(None, "T1", {"execution": False}, + findings=[finding]) + self.assertEqual(out["decision"], "pushback") + self.assertEqual(out["round"], 1) + from cloud.audit_loop import LoopState + state = LoopState("T1", round=3, passed_gates=[], + criteria={}, last_decision="pushback") + out = evaluate(state, "T1", {"execution": False}, + findings=[finding]) + self.assertEqual(out["decision"], "escalate") + self.assertEqual(out["reason"], "max_rounds_exceeded") + + +class GateErrorMarkersTestCase(unittest.TestCase): + """I-13：分类子串表抽成模块常量。""" + + def test_marker_constant_covers_known_markers(self): + self.assertIn("Missing environment variable:", + _GATE_ERROR_MARKERS) + self.assertIn("网关准备失败", _GATE_ERROR_MARKERS) + self.assertIn("必需工具缺失", _GATE_ERROR_MARKERS) + self.assertIn("工具审批声明矛盾", _GATE_ERROR_MARKERS) + self.assertIn("refusing to dispatch an incapable or" + " unverifiable model", _GATE_ERROR_MARKERS) + self.assertIn("fails capability", _GATE_ERROR_MARKERS) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 7b5d656..5556724 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -482,6 +482,45 @@ class _CoordGateLookupFailed(Exception): """ +def _mark_gate_checked(task, gate): + """在任务 dict 上记一笔“某门本轮确实跑过”（审计批 4，I-10）。 + + 用 task["_gates_checked"] 列表承载，键名带下划线、私有、 + 内存态，与 _gateway_provider/_gateway_env 同款约定： + 不进库、不进审计 detail、不随 prompt 落库。列表而非 + 集合，保证 task dict 仍是 JSON 安全的；run_one 内 task + 经多次 dict(task) 浅拷贝，拷贝共享同一列表对象，因此 + 无论哪份拷贝打标记，终局块读到的都是同一份集合。 + 终局时 _audit_loop_gate_results 只列入被标记过的门， + 不再把没跑过的门硬编码成通过。 + """ + if not isinstance(task, dict): + return + checked = task.get("_gates_checked") + if not isinstance(checked, list): + checked = [] + task["_gates_checked"] = checked + if gate not in checked: + checked.append(gate) + + +# 审计批 4（I-13）：合流分类用的门拒绝子串标记表。 +# 分类优先读结构化证据（_coord_gate_rejected_child_ids 的 +# 审计行集合），子串只作最后回落——回落命中时合流记一条 +# worker_coord_gate_classified_by_marker 审计行点名 marker， +# 让回落可见。生产方 error 文案本身不改措辞、不做全仓 +# 常量化重构（那是另一批的事）；分类侧的这份拷贝抽成 +# 常量，文案改动时分类侧只有这一处要改。 +_GATE_ERROR_MARKERS = ( + "Missing environment variable:", + "网关准备失败", + "必需工具缺失", + "工具审批声明矛盾", + "refusing to dispatch an incapable or unverifiable model", + "fails capability", +) + + class Worker: """单机 worker：一轮认领一个 queued 任务并执行到底。 @@ -1924,7 +1963,8 @@ class Worker: result["new_state"] = result["new_state"].to_dict() return result - def _audit_loop_gate_results(self, task, final, result): + def _audit_loop_gate_results(self, task, final, result, + gates_checked=None): """本轮已产生的确定性门结果（只取已有结果，不捏造判定）。 门的取值口径（核实结论）： @@ -1938,8 +1978,25 @@ class Worker: - capability / required_tools / gateway：本方法只在终局 块被调用，而终局块的前提是尝试跑到了 runner——此前 的前置门若拒绝，run_one 早已在 runner 之前提前返回， - 根本到不了这里；所以能到这里的尝试，这三门要么已 - 通过、要么不适用，记 True 是事实陈述，不是复查。 + 根本到不了这里；所以能到这里的尝试，已跑过的门 + 要么已通过、要么不适用。审计批 4（I-10）之前这里 + 把三门无条件记 True，显式任务（尤其 I-7 修复前的 + 非 sub2api 组合）能力门根本没跑过却同样记入 + passed_gates 并持久化——“通过”一词在状态表里失真。 + 现在：调用方传 gates_checked（run_one 派发点用 + _mark_gate_checked 在 task["_gates_checked"] 里记的 + 本轮实际跑过的门）时，只列入集合内的门；未传 + （直接调用的旧调用方）时保持旧行为、三门全记 True。 + gate_missing 语义核查：引擎只对“上一轮 passed_gates + 里有、本轮缺失”的门升级；同一任务的跑过门集合在 + 两轮间是稳定的（显式任务无策略时两轮都不记 + capability，政策任务两轮都记；门集合由任务/策略/ + 执行器决定，不由轮次决定），所以显式任务两轮不 + 会因“not_checked”误触发升级——两轮往返测试锁定。 + 唯一可预见的边界是部署时已在循环中的任务：旧状态 + passed_gates 里含硬编码 True 的三门，新代码首轮 + 可能缺省其中几门 → gate_missing 升级一次，方向是 + fail-closed 不是放行。 前置门拒绝的路径不接审计循环（见 _audit_loop_wire 的文档）：那类任务在 agent 跑任何一轮之前就已失败， 推回包没有可投递的 agent 回合，且重排会确定性地 @@ -1970,9 +2027,15 @@ class Worker: "re-run goal verification", "goal verification passes", kind="deterministic", gate="goal_verify")) - gate_results["capability"] = True - gate_results["required_tools"] = True - gate_results["gateway"] = True + if gates_checked is None: + gate_results["capability"] = True + gate_results["required_tools"] = True + gate_results["gateway"] = True + else: + checked = set(gates_checked) + for gate in ("capability", "required_tools", "gateway"): + if gate in checked: + gate_results[gate] = True return gate_results, findings def _audit_loop_wire(self, task, final, result, error): @@ -2007,7 +2070,11 @@ class Worker: if task.get("coordinator_parts"): return final, error, "skipped" gate_results, findings = self._audit_loop_gate_results( - task, final, result) + # 审计批 4（I-10）：把派发点记的本轮实际跑过 + # 的门透传进去；没记过的门不列入 gate_results， + # 不再硬编码成通过。未传（旧直接调用）时保持 + # 旧行为，见 _audit_loop_gate_results 文档。 + task, final, result, task.get("_gates_checked")) try: decision = self.audit_loop_evaluate( task, gate_results, findings=findings) @@ -6129,11 +6196,19 @@ class Worker: attempt_started_at=None): """前置门拒绝的子任务 id 集（合流分类用）。 - 依据只有审计行：能力门、工具门、审批门的拒绝审计 detail - 是 "<子任务 id>: <错误>" 格式，可按 id 前缀归属；缺 key - 门（worker_missing_key）与仓库门（worker_repo_failed） - 的 detail 只有错误文本、没有任务 id，审计行无法归属， - 故不在此列——那两类只走下面的 error_text 强标记回落。 + 依据只有审计行：下列 6 个拒绝审计的 detail 均是 + "<子任务 id>: <错误>" 格式，可按 id 前缀归属—— + 能力门、工具门、审批门（旧有 3 门），以及审计批 4 + （I-8）补进的缺 key 门（worker_missing_key）、仓库门 + （worker_repo_failed）、写回复核拒绝门 + （worker_capability_recheck_rejected，其 detail 本就 + 是 ": <错误>" 格式）。不补进集合的同类审计行： + worker_capability_gateway_explicit（含 blocked=false + 的纯审计行，补进会把未被拦的显式任务误判为门拒绝）、 + worker_capability_gateway_shadow（shadow 永不拦）、 + worker_capability_fallback（回落不是拒绝）； + worker_gateway_failed（网关准备失败）已有 error_text + 标记覆盖，不补进集合。 审计是 best-effort，查不到不等于没拒绝，调用方不得把 “无审计行”当成“没被门拒绝”的证据（回落标记见分类）。 @@ -6175,7 +6250,10 @@ class Worker: """ actions = ("worker_capability_gate_rejected", "worker_tool_gate_rejected", - "worker_tool_approval_gate_rejected") + "worker_tool_approval_gate_rejected", + "worker_missing_key", + "worker_repo_failed", + "worker_capability_recheck_rejected") rows = None try: conn = _db_connect(self.db_path) @@ -6183,17 +6261,21 @@ class Worker: try: rows = conn.execute( "SELECT detail, created_at FROM audit_log" - " WHERE project_id = ? AND action IN (?, ?, ?)" + " WHERE project_id = ? AND action IN" + " (?, ?, ?, ?, ?, ?)" " ORDER BY created_at DESC, id DESC LIMIT 5000", (project_id, actions[0], actions[1], - actions[2])).fetchall() + actions[2], actions[3], actions[4], + actions[5])).fetchall() except Exception: rows = conn.execute( "SELECT detail, created_at FROM audit_log" - " WHERE project_id = ? AND action IN (?, ?, ?)" + " WHERE project_id = ? AND action IN" + " (?, ?, ?, ?, ?, ?)" " ORDER BY created_at DESC LIMIT 5000", (project_id, actions[0], actions[1], - actions[2])).fetchall() + actions[2], actions[3], actions[4], + actions[5])).fetchall() finally: conn.close() except Exception as exc: @@ -6259,13 +6341,7 @@ class Worker: if child_id in (gate_rejected_ids or set()): return "gate_rejected", "gate_audit" err = (row.get("error_text") or "") - for marker in ("Missing environment variable:", - "网关准备失败", - "必需工具缺失", - "工具审批声明矛盾", - "refusing to dispatch an incapable or" - " unverifiable model", - "fails capability"): + for marker in _GATE_ERROR_MARKERS: if marker in err: return "gate_rejected", "gate_error_marker" if status == "failed" or status == "interrupted": @@ -6392,6 +6468,19 @@ class Worker: excluded_notes.append("%s:%s" % (name, reason)) elif kind == "gate_rejected": gate_rejected.append(name) + if reason == "gate_error_marker": + # 审计批 4（I-13）：子串回落命中必须可见， + # 不能静默——记一条点名 marker 的审计行， + # 让“分类依赖 error_text 回落”在审计里 + # 可查；文案改动导致回落失效时，这里 + # 的缺失本身就是信号。 + fired = next( + (m for m in _GATE_ERROR_MARKERS + if m in (row.get("error_text") or "")), + "?") + self._audit(project_id, + "worker_coord_gate_classified_by_marker", + "%s marker=%s" % (cid, fired)) elif kind == "execution": gate_results["child:" + name] = False findings.append(_AuditFinding( @@ -6910,6 +6999,11 @@ class Worker: return [] if not isinstance(self.executor, DockerExecutor): return [] + # 审计批 4（I-10）：走到这里说明工具门真正查了租户 + # 配置（读失败也算跑过：返回 [] 的放行按 I-5 记审计， + # 且该任务随后在租户准备失败处提前返回、到不了 + # 终局块，标记本身无实际影响）。 + _mark_gate_checked(task, "required_tools") try: home = self._tenant_dirs( task.get("project_id"))["codex_home"] @@ -6994,12 +7088,22 @@ class Worker: "worker_persisted_model_lookup_failed", "%s error=%s" % (task_id, exc)) return False + # 审计批 4（I-6）：persisted 行格式固定为 + # "%s tier=%s model=%s"，model 是最后一个字段， + # 按键提取而不是 split()[2] 定位置取。旧代码在 + # 模型名含空白时永远匹配不上，于是写回模型被 + # 默认当成“用户显式”而跳过复核——两处同时失效 + # 的另一半（影子解析是 routing_shadow 侧）。 + # 格式完全不符的行仍视为无证据：worker 写出的 + # persisted 行格式固定，格式不符不可能是写回行， + # fail-closed 备选（按写回对待）在此不适用。 + persisted_re = re.compile( + r"^%s\s+tier=\S+\s+model=(.*)$" % re.escape(task_id)) for row in rows: if row[0] != "worker_model_persisted": continue - parts = (row[1] or "").split() - if len(parts) >= 3 and parts[0] == task_id \ - and parts[2] == "model=%s" % model: + match = persisted_re.match((row[1] or "").strip()) + if match and match.group(1).strip() == model: return True for row in rows: if row[0] != "worker_routing": @@ -7033,6 +7137,9 @@ class Worker: model = task.get("model") verdict = evaluate_capability( self.routing_policy.model_capabilities, model, requires) + # 审计批 4（I-10）：走到这里说明复核真正算了 + # verdict（无策略/无 requires/off 档已在上面返回）。 + _mark_gate_checked(task, "capability") if verdict.ok: return None known = verdict.known_missing @@ -7206,7 +7313,12 @@ class Worker: # 缺 key 门提前返回：同样要清暂存，否则与仓库解析 # 失败一路同款残留（见 _cleanup_task_seed）。 self._cleanup_task_seed(task) - self._audit(task["project_id"], "worker_missing_key", error) + # 审计批 4（I-8）：缺 key 门审计 detail 改为 + # "<任务 id>: <错误>"，与能力/工具/审批门同款 + # 可归属格式，协调员合流才能按 id 前缀认领； + # error_text 子串标记仍能命中（子串匹配）。 + self._audit(task["project_id"], "worker_missing_key", + "%s: %s" % (task_id, error)) log_event(self.logger, "worker_missing_key", task_id=task_id, project_id=task["project_id"], status="failed") self._notify_task_finished( @@ -7234,7 +7346,11 @@ class Worker: # 仓库解析失败提前返回：够不到正常终局后的清理， # 这里必须先清暂存，否则 bundle 目录永久残留。 self._cleanup_task_seed(task) - self._audit(task["project_id"], "worker_repo_failed", repo_error) + # 审计批 4（I-8）：仓库门审计 detail 改为 + # "<任务 id>: <错误>"，与 3 门同款可归属格式； + # error_text 子串标记仍能命中（子串匹配）。 + self._audit(task["project_id"], "worker_repo_failed", + "%s: %s" % (task_id, repo_error)) log_event(self.logger, "worker_repo_failed", task_id=task_id, project_id=task["project_id"], status="failed") self._notify_task_finished( @@ -7419,6 +7535,12 @@ class Worker: "stdout": "", "error": error, "events": [], "cost_record_id": None} + if gateway_provider: + # 审计批 4（I-10）：网关准备实际跑过且成功 + # （失败已在上面提前返回），终局时 gateway + # 门可列入 gate_results。 + _mark_gate_checked(task, "gateway") + # Audit fix：网关能力检查——此前路由决策只在模型 # 为空时跑，sub2api 网关先填默认模型后决策直接跳 # 过，能力门在生产路径上完全不生效。补齐：网关填 @@ -7435,8 +7557,29 @@ class Worker: # 不满足才拦，unknown 只审计（见下），因为显式 # 优先保护的是用户选了哪个模型，不是选的模型 # 一定能跑；目录陈旧不应惩罚正确的显式选择。 - if gateway_provider == "sub2api" \ - and self.routing_policy is not None: + # 审计批 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 {} @@ -7446,6 +7589,9 @@ class Worker: _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" % ( @@ -7584,6 +7730,16 @@ class Worker: # 断言 events[0] 就是 fallback 事件、无事件时为空）， # 而旧行为本身已由 task["model"] / actual_model 体现， # 不需要再用一条事件重复说明。 + # 审计批 4（I-10）：decide_routing 对档模型算过 + # capability 的决策才记 capability——policy/shadow + # 决策算过、capability_blocked 决策算过、 + # capability_fallback 决策算过；worker_default 经 + # 无档模型/off/无策略、explicit/worker_explicit + # 提前返回时均未算，不记。 + if decision.source in ("policy", "shadow", + "capability_blocked") \ + or getattr(decision, "capability_fallback", False): + _mark_gate_checked(task, "capability") if decision.source in ("policy", "shadow"): record_routing({"id": task_id, "events": events}, decision) # 影子评估（v0b）需要建议模型：旧格式只有