diff --git a/cloud/models.py b/cloud/models.py index dbef83c..6820b02 100644 --- a/cloud/models.py +++ b/cloud/models.py @@ -379,7 +379,16 @@ class RoutingPolicy: def __init__(self, version=None, mode="shadow", tiers=None, fallback_models=None, tier_modes=None, - model_capabilities=None, tier_requirements=None): + model_capabilities=None, tier_requirements=None, + dropped_tiers=None, partial_failures=None): + # Audit batch 2 (I-2): tiers the parser dropped for an + # invalid mode/requires, and source-level failures that + # still left a usable policy. Both default to [] so + # directly-constructed policies never raise AttributeError + # on these fields; parse/load fill them in. They are + # visibility only: they never change routing decisions. + self.dropped_tiers = list(dropped_tiers or []) + self.partial_failures = list(partial_failures or []) self.version = version if isinstance(version, str) \ and version.strip() else None self.mode = mode if mode in ROUTING_MODES else "off" @@ -474,6 +483,12 @@ def parse_routing_policy(data): # accidentally activate, and never kills the other tiers. raw_tiers = data.get("tiers") if isinstance(data.get("tiers"), dict) else {} + # Audit batch 2 (I-2): record every tier the fail-closed pops + # actually remove, with a warning naming the tier. A pop that + # finds the tier already absent (e.g. _clean_tiers dropped it + # 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 = [] tier_modes = {} for tier, entry in raw_tiers.items(): if not isinstance(entry, dict) or "mode" not in entry: @@ -481,8 +496,12 @@ def parse_routing_policy(data): tier_mode = entry.get("mode") if tier_mode in ROUTING_MODES: tier_modes[tier] = tier_mode - else: + elif tier in tiers: tiers.pop(tier, None) + dropped_tiers.append(tier) + _logger.warning( + "routing policy tier %r dropped: invalid mode %r", + tier, tier_mode) if not tiers: return None # Per-tier requirements (v8): a tier entry dict may carry @@ -496,7 +515,12 @@ def parse_routing_policy(data): continue cleaned = parse_tier_requires(entry.get("requires")) if cleaned is None: - tiers.pop(tier, None) + if tier in tiers: + tiers.pop(tier, None) + dropped_tiers.append(tier) + _logger.warning( + "routing policy tier %r dropped: invalid requires", + tier) elif cleaned: tier_requirements[tier] = cleaned if not tiers: @@ -510,7 +534,8 @@ def parse_routing_policy(data): tier_modes=tier_modes, model_capabilities=data.get( "model_capabilities"), - tier_requirements=tier_requirements) + tier_requirements=tier_requirements, + dropped_tiers=dropped_tiers) def load_routing_policy(env=None): @@ -573,17 +598,14 @@ def load_routing_policy(env=None): " worker default", "; ".join(failures)) return None if failures: - # 9.5-code fail-open 清单：部分失败是 fail-open—— + # Audit batch 2 (I-2): 部分失败仍是 fail-open—— # 可用来源继续生效、失败来源被丢弃，派发不炸。 - # 可见性只有这条警告日志：load_routing_policy 仍 - # 返回一个 policy，worker 的 - # routing_policy_load_failed 标志只在全失败时 - # 置位，所以部分失败不会产生任务级审计；运营方 - # 只能从 worker 日志看到它，不能从审计里查到。 - # 这是已知缺口（非静默，但不如全失败可见）， - # 本轨不改返回类型：生产调用点只有 2 处（server - # 影子端点经 _load_routing_policy、worker 派发）， - # 真正的成本是同步改调用方与测试，而不是“牵动全部”。 + # 但可见性不再只有这条警告日志：失败清单挂在返回 + # 的 policy 上（partial_failures），worker 派发时 + # 据此记任务级审计（与全失败同款 action），影子 + # 报告里也附指纹与丢弃档；返回类型本身不变， + # 仍是 RoutingPolicy 或 None。 + policy.partial_failures = list(failures) _logger.warning( "routing policy partially failed (%s); continuing with" " the usable source", "; ".join(failures)) diff --git a/cloud/routing_shadow.py b/cloud/routing_shadow.py index 74f41c3..8e7ed1e 100644 --- a/cloud/routing_shadow.py +++ b/cloud/routing_shadow.py @@ -68,13 +68,11 @@ def parse_worker_routing_detail(detail): Old format (no suggested/chosen) parses with both models None. A ``-`` placeholder also means None, never a model named "-". - 9.5-code fail-open 清单：解析失败的行返回 None 后在 - evaluate_shadow 里直接跳过、不计数、不报错——样本无 - 声消失（例如模型名含空白会使整行匹配失败，见 M7）。 - 这是已知缺口：改返回类型只需同步评估函数的唯一生产 - 调用方（server 影子端点）与测试，本轨只在此处写明， - 不改语义；运营方无法从报告里看到丢了多少行，只能靠 - 审计行总数与样本数对不上来发现。 + Audit batch 2 (I-1): 解析语义本身不变（失败仍返回 None）， + 但 evaluate_shadow 不再静默跳过：不可解析的行计入 + dropped_rows["unparseable"]，报告里可见（例如模型名含 + 空白会使整行匹配失败，见 M7——行不再无声消失，只是 + 不计为样本）。 """ if not isinstance(detail, str): return None @@ -144,10 +142,23 @@ def evaluate_shadow(audit_rows, tasks_by_id, costs_by_task, "unpriced_tokens": int}. ``explicit_n``: number of tasks with a creation-time model in the same project (counted separately, never as routed samples). + Audit batch 2 (I-1): every input row lands in exactly one + bucket — a routed sample, or ``dropped_rows``. Dropped + buckets are the real skip reasons in the loop below: + ``unparseable`` (detail does not match the format), + ``unknown_tier`` (tier is not a known tier), ``explicit`` + (source=explicit rows, also counted per-tier as explicit_n, + never as samples), ``duplicate`` (a later row for an + already-counted task). Identity: ``audit_rows_total == + routed_task_n + dropped_total`` — the caller can reconcile + the report against the audit rows it fetched instead of + guessing whether samples were lost. """ tiers = {tier: _empty_tier() for tier in TIERS} seen = set() routed_task_ids = set() + dropped_rows = {"unparseable": 0, "unknown_tier": 0, + "explicit": 0, "duplicate": 0} rows = list(audit_rows or []) if any(isinstance(r, dict) and r.get("created_at") is not None for r in rows): @@ -169,6 +180,7 @@ def evaluate_shadow(audit_rows, tasks_by_id, costs_by_task, parsed = parse_worker_routing_detail( (row or {}).get("detail")) if parsed is None: + dropped_rows["unparseable"] += 1 continue task_id = parsed["task_id"] tier = parsed["tier"] @@ -178,14 +190,23 @@ def evaluate_shadow(audit_rows, tasks_by_id, costs_by_task, # occupy the dedup slot (Audit fix 3, M1): a task # whose first row is an unknown tier must not lose # its later known-tier row to dedup. + # Audit batch 2 (I-1): the row is counted as dropped, + # not silently skipped. + dropped_rows["unknown_tier"] += 1 continue if parsed["source"] == "explicit": # Audit fix 3（I8）：显式行只计 per-tier # explicit_n：不计 routed_n、不占 seen（同一任务 # 日后若出现路由行仍可计为样本），不破坏去重。 + # Audit batch 2 (I-1): explicit rows are also + # counted in dropped_rows so the row identity + # reconciles; they are not "lost" — they are + # counted separately by design. tiers[tier]["explicit_n"] += 1 + dropped_rows["explicit"] += 1 continue if task_id in seen: + dropped_rows["duplicate"] += 1 continue seen.add(task_id) bucket = tiers[tier] @@ -250,6 +271,12 @@ def evaluate_shadow(audit_rows, tasks_by_id, costs_by_task, "tiers": tiers, "explicit_n": int(explicit_n or 0), "routed_task_n": len(routed_task_ids), + # Audit batch 2 (I-1): row reconciliation. Every input + # row is either a routed sample or in dropped_rows, so + # audit_rows_total == routed_task_n + dropped_total. + "audit_rows_total": len(rows), + "dropped_rows": dropped_rows, + "dropped_total": sum(dropped_rows.values()), "counterfactual": "not_measured", "counterfactual_note": ( "Shadow suggested models never ran, so this report cannot " diff --git a/cloud/server.py b/cloud/server.py index 5b57450..6996b3c 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -4212,7 +4212,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): conn = self._db() try: audit_rows = conn.execute( - "SELECT detail FROM audit_log" + "SELECT detail, created_at, id FROM audit_log" " WHERE project_id = ? AND action = 'worker_routing'" " ORDER BY created_at, id LIMIT 10001", (project_id,), @@ -4262,8 +4262,16 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): entry["unpriced_tokens"] += int(row["total_tokens"] or 0) else: entry["usd_micros"] += int(row["usd_micros"]) + # Audit batch 2 (I-1): pass created_at/id through instead + # of stripping them — evaluate_shadow sorts by them for + # dedup, and the fetched rows were already ordered this + # way; stripping forced the sort to rely on input order. + # The report now carries audit_rows_total (from evaluate) + # next to routed_task_n, reconciled by dropped_rows. report = _evaluate_shadow( - [{"detail": row["detail"]} for row in audit_rows], + [{"detail": row["detail"], + "created_at": row["created_at"], + "id": row["id"]} for row in audit_rows], tasks_by_id, costs_by_task, explicit_n=explicit_n) report["project_id"] = project_id report["truncated"] = truncated @@ -4360,6 +4368,27 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): report["recommendation"] += ( " Note: audit rows were truncated to 10000, so" " this report is based on a partial sample.") + # Audit batch 2 (I-2): the currently loaded policy, as the + # report sees it — fingerprint plus any tiers dropped by + # fail-closed parsing and any source-level partial + # failures. This endpoint has no task context, so it + # cannot emit a task-level audit; that visibility lives + # on the worker dispatch path (see worker.py run_one). + # When no usable policy loads, status is "unavailable" + # and dropped tiers are unknown (not empty-by-claim). + if _policy is None: + report["routing_policy"] = { + "status": "unavailable", "fingerprint": None, + "dropped_tiers": [], "partial_failures": []} + else: + _dropped = list(getattr(_policy, "dropped_tiers", []) or []) + _partial = list(getattr(_policy, "partial_failures", []) + or []) + report["routing_policy"] = { + "status": "partial" if (_dropped or _partial) else "ok", + "fingerprint": _policy.content_fingerprint(), + "dropped_tiers": _dropped, + "partial_failures": _partial} self._send_json(200, report) def _task_cost(self, user_id, task_id): diff --git a/cloud/tests/test_audit_batch2.py b/cloud/tests/test_audit_batch2.py new file mode 100644 index 0000000..2c781fc --- /dev/null +++ b/cloud/tests/test_audit_batch2.py @@ -0,0 +1,422 @@ +"""审计第 2 批（I-1、I-2、I-5）单测：可见性三件。 + +运行（在仓库根目录下）： + python3 -m unittest discover -s cloud/tests -p "test_audit_batch2.py" -v + +说明：只断言本批新增的可见性行为，不碰 I-3、I-4、I-6～I-14、 +N-1～N-12。I-1 断在报告层（不只断解析返回 None）；I-2 断在 +任务级审计与影子报告字段层；I-5 四处各一例失败注入。 +不跑真实模型回合、不联网。 +""" + +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 + +try: + from cloud.auth import AuthStore + from cloud.models import load_routing_policy + from cloud.models import parse_routing_policy + from cloud.routing_shadow import evaluate_shadow + from cloud.sandbox import DockerExecutor + from cloud.server import create_server + from cloud.server import init_db + from cloud.worker import Worker + from cloud.worker import _CoordGateLookupFailed +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.models import load_routing_policy + from cloud.models import parse_routing_policy + from cloud.routing_shadow import evaluate_shadow + from cloud.sandbox import DockerExecutor + from cloud.server import create_server + from cloud.server import init_db + from cloud.worker import Worker + from cloud.worker import _CoordGateLookupFailed + + +def _completed_runner(cmd): + return SimpleNamespace(returncode=0, stdout="{}", stderr="") + + +def _insert_task(db_path, task_id, status="queued", parent=None, + part=None, error=None, audit_loop=0, parts=None, + project_id="p1"): + 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," + " paused_by_user, interrupted_reason, coordinator_parts," + " started_at)" + " VALUES(?, ?, 'do', ?, NULL, '[]', 0, 'tester'," + " ?, ?, NULL, ?, ?, ?, NULL, ?, 0, NULL, ?, ?)", + (task_id, project_id, status, now, now, audit_loop, + parent, part, error, + json.dumps(parts) if parts is not None else None, now)) + conn.commit() + finally: + conn.close() + + +def _audit_rows(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 _task_row(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + row = conn.execute( + "SELECT status, audit_loop_packet FROM tasks WHERE id = ?", + (task_id,)).fetchone() + return row + finally: + conn.close() + + +class DroppedRowsTestCase(unittest.TestCase): + def test_buckets_and_identity_at_report_level(self): + # 6 行输入：2 个有效样本 + 4 种丢行原因各 1 行。 + rows = [ + {"detail": "t1 tier=plan source=shadow applied=False", + "created_at": 1.0, "id": 1}, + {"detail": "t2 tier=goal source=policy applied=True", + "created_at": 2.0, "id": 2}, + {"detail": "this line does not parse at all", + "created_at": 3.0, "id": 3}, + {"detail": "t3 tier=weird source=shadow applied=True", + "created_at": 4.0, "id": 4}, + {"detail": "t4 tier=plan source=explicit applied=False", + "created_at": 5.0, "id": 5}, + {"detail": "t1 tier=plan source=shadow applied=False", + "created_at": 6.0, "id": 6}, + ] + tasks = {"t1": {"status": "completed"}, + "t2": {"status": "completed"}} + report = evaluate_shadow(rows, tasks, {}) + self.assertEqual(report["dropped_rows"], { + "unparseable": 1, "unknown_tier": 1, + "explicit": 1, "duplicate": 1}) + self.assertEqual(report["dropped_total"], 4) + self.assertEqual(report["audit_rows_total"], 6) + self.assertEqual(report["routed_task_n"], 2) + # 行恒等式：总行数 = 样本数 + 丢行总数。 + self.assertEqual( + report["audit_rows_total"], + report["routed_task_n"] + report["dropped_total"]) + # 显式行同时计 per-tier explicit_n（不是丢失）。 + self.assertEqual(report["tiers"]["plan"]["explicit_n"], 1) + self.assertEqual(report["tiers"]["plan"]["routed_n"], 1) + + +class ShadowEndpointPolicyTestCase(unittest.TestCase): + def setUp(self): + self._old_json = os.environ.get("CLOUD_ROUTING_POLICY_JSON") + self._old_file = os.environ.get("CLOUD_ROUTING_POLICY_FILE") + self._old_mode = os.environ.get("CLOUD_ROUTING_MODE") + # 策略含一个无效档模式的档（plan）与一个有效档（goal）： + # 端点报告里必须出现丢弃档 plan，而不是只剩 goal。 + os.environ["CLOUD_ROUTING_POLICY_JSON"] = json.dumps({ + "mode": "shadow", + "tiers": {"plan": {"model": "m-plan", "mode": "bogus"}, + "goal": "m-goal"}}) + os.environ.pop("CLOUD_ROUTING_POLICY_FILE", None) + os.environ.pop("CLOUD_ROUTING_MODE", None) + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.auth = AuthStore(self.db_path) + _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, self.project) + + 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() + for key, old in (("CLOUD_ROUTING_POLICY_JSON", self._old_json), + ("CLOUD_ROUTING_POLICY_FILE", self._old_file), + ("CLOUD_ROUTING_MODE", self._old_mode)): + if old is None: + os.environ.pop(key, None) + else: + os.environ[key] = old + + 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 test_endpoint_reports_dropped_tiers_and_fingerprint(self): + status, report = self._req( + "GET", "/routing-shadow?project_id=" + self.project["id"], + key=self.owner_key) + self.assertEqual(status, 200, report) + policy = report["routing_policy"] + self.assertEqual(policy["status"], "partial") + self.assertEqual(policy["dropped_tiers"], ["plan"]) + self.assertEqual(len(policy["fingerprint"]), 64) + # 无样本时丢行计数仍在报告里（不是只有样本才有）。 + self.assertEqual(report["audit_rows_total"], 0) + self.assertEqual(report["dropped_total"], 0) + + +class PolicyDropWarningTestCase(unittest.TestCase): + def test_invalid_mode_warns_with_tier_name_and_keeps_others(self): + with self.assertLogs("cloud.models", level="WARNING") as logs: + policy = parse_routing_policy({ + "mode": "shadow", + "tiers": {"plan": {"model": "m-plan", + "mode": "bogus"}, + "goal": "m-goal"}}) + self.assertIsNotNone(policy) + self.assertNotIn("plan", policy.tiers) + self.assertIn("goal", policy.tiers) + self.assertEqual(policy.dropped_tiers, ["plan"]) + self.assertTrue(any("plan" in m for m in logs.output)) + + def test_invalid_requires_drops_tier_with_warning(self): + with self.assertLogs("cloud.models", level="WARNING") as logs: + policy = parse_routing_policy({ + "mode": "shadow", + "tiers": {"plan": {"model": "m-plan", + "requires": {"bogus_key": True}}, + "goal": "m-goal"}}) + self.assertIsNotNone(policy) + self.assertNotIn("plan", policy.tiers) + self.assertIn("goal", policy.tiers) + self.assertEqual(policy.dropped_tiers, ["plan"]) + self.assertTrue(any("plan" in m for m in logs.output)) + + +class PartialFailureTaskAuditTestCase(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) + _insert_task(self.db_path, "P1") + + def tearDown(self): + self.tmp.cleanup() + + def test_partial_failure_writes_task_audit(self): + # 文件来源失败 + JSON 来源可用 = 部分失败：策略可用， + # 但必须在任务级审计里可见（与全失败同款 action）。 + env = {"XCM_API_KEY": "test-placeholder", + "CLOUD_ROUTING_POLICY_FILE": "/nonexistent/policy.json", + "CLOUD_ROUTING_POLICY_JSON": json.dumps({ + "mode": "shadow", "tiers": {"default": "m1"}})} + worker = Worker(self.db_path, "/fake/host.py", + runner=_completed_runner, env=env, + max_retries=0, retry_backoff_seconds=0) + self.assertFalse(worker.routing_policy_load_failed) + self.assertTrue(worker.routing_policy_partial_failure) + self.assertTrue(worker.routing_policy.partial_failures) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + actions = _audit_rows(self.db_path) + partial = [d for a, d in actions + if a == "worker_routing_policy_load_failed"] + self.assertTrue(partial) + self.assertIn("failure=partial", partial[0]) + self.assertIn("policy.json", partial[0]) + + +class CoordGateLookupFailureTestCase(unittest.TestCase): + def setUp(self): + self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.env = {"XCM_API_KEY": "test-placeholder"} + self.audits = [] + self.worker = Worker(self.db_path, "/fake/host.py", + runner=_completed_runner, env=self.env, + 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 test_direct_lookup_raises_instead_of_empty_set(self): + # init_db 不建 audit_log 表（生产里由 AuthStore 建）： + # 查询失败时旧代码返回空集，新代码必须抛错——空集 + # 会被合流读成“没有门拒绝”而把门拒绝推回。 + with self.assertRaises(_CoordGateLookupFailed): + self.worker._coord_gate_rejected_child_ids("p1", ["C1"]) + + def test_merge_escalates_without_requeue_or_round(self): + # 父任务审计开启、子任务执行失败；门拒绝审计表缺失 + # 导致查询失败时，合流必须升级、不重排、不进推回。 + # 注意：这里故意不建 audit_log 表——缺表即查询失败。 + _insert_task(self.db_path, "P1", status="running", + audit_loop=1, + parts=[{"name": "a", "prompt": "do a"}]) + _insert_task(self.db_path, "C1", status="failed", + parent="P1", part="a", error="boom") + 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.assertIn("查询失败", out["error"]) + self.assertTrue(out.get("gate_lookup_failed")) + # 子任务未被重排回 queued，仍是 failed。 + self.assertEqual(_task_row(self.db_path, "C1")[0], "failed") + # 升级审计里点名了查询失败。 + self.assertTrue(any( + a == "worker_coord_audit_loop_escalated" + and "gate_lookup_failed" in d + for _p, a, d in self.audits)) + + +class PersistedModelLookupFailureTestCase(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_lookup_failure_audits_and_still_dispatches(self): + # _db_connect 桩成抛错：复核查不到依据时返回 False + # （放行），但必须在审计里点名依据缺失，不能只有日志。 + # _audit 走 AuthStore 独立连接，不受该桩影响。 + import cloud.worker as worker_mod + real_connect = worker_mod._db_connect + + def _boom(_path): + raise RuntimeError("db down for lookup") + + worker_mod._db_connect = _boom + try: + 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.assertFalse(worker._is_persisted_model( + "T1", "p1", "m1")) + finally: + worker_mod._db_connect = real_connect + actions = _audit_rows(self.db_path) + found = [d for a, d in actions + if a == "worker_persisted_model_lookup_failed"] + self.assertTrue(found) + self.assertIn("T1", found[0]) + + +class RequiredToolsCheckFailureTestCase(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_check_failure_audits_and_still_allows(self): + # 租户目录读取失败时返回 [] 放行不变，但必须在审计 + # 里点名依据缺失，不能只有 warning 日志。 + worker = Worker(self.db_path, "/fake/host.py", + runner=_completed_runner, + env={"XCM_API_KEY": "test-placeholder"}, + max_retries=0, retry_backoff_seconds=0) + worker.tenant_isolation = True + worker.executor = DockerExecutor.__new__(DockerExecutor) + + def _boom(_project_id): + raise RuntimeError("tenant home unreadable") + + worker._tenant_dirs = _boom + result = worker._missing_required_tools( + {"id": "T1", "project_id": "p1", + "agent_tools": ["websearch"]}) + self.assertEqual(result, []) + actions = _audit_rows(self.db_path) + found = [d for a, d in actions + if a == "worker_required_tools_check_failed"] + self.assertTrue(found) + self.assertIn("T1", found[0]) + + +class AuditLoopEvaluateFailureTestCase(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_evaluate_failure_still_escalates(self): + # 回归断言：审计循环评估失败仍 fail-closed 到 escalate + # （I-5 策略表里评估失败→升级这一路本来就有，不动）。 + # 非法门结果值让纯函数 evaluate 抛 ValueError，经 + # audit_loop_evaluate 捕获后应为 escalate。 + worker = Worker(self.db_path, "/fake/host.py", + runner=_completed_runner, + env={"XCM_API_KEY": "test-placeholder"}, + max_retries=0, retry_backoff_seconds=0) + out = worker.audit_loop_evaluate( + {"id": "T1", "project_id": "p1", "audit_loop": 1}, + {"child:a": "not-a-bool"}) + self.assertEqual(out["decision"], "escalate") + self.assertEqual(out["reason"], "evaluate_failed") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_audit_loop_coord.py b/cloud/tests/test_audit_loop_coord.py index 453a48a..94e2544 100644 --- a/cloud/tests/test_audit_loop_coord.py +++ b/cloud/tests/test_audit_loop_coord.py @@ -28,11 +28,13 @@ from types import SimpleNamespace try: from cloud.audit_loop import Finding + from cloud.auth import AuthStore from cloud.server import init_db from cloud.worker import Worker except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.audit_loop import Finding + from cloud.auth import AuthStore from cloud.server import init_db from cloud.worker import Worker @@ -100,6 +102,14 @@ class CoordAuditTestCase(unittest.TestCase): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) + # Audit batch 2 (I-5): 合流的门拒绝审计查询失败时走 + # escalate，不再回落空集。生产里 audit_log 由首次 + # 真实 _audit 经 AuthStore 建好，合流时表一定存在； + # 本文件把 _audit 桩成列表，表永远不会出现，所以 + # 在夹具里按生产实际补建表——否则所有合流测试都会 + # 因缺表改走 escalate，与被测行为无关。 + _store = AuthStore(self.db_path) + _store.conn.close() self.env = {"XCM_API_KEY": "test-placeholder"} self.audits = [] self.worker = self._worker() @@ -399,8 +409,13 @@ class CoordAuditTestCase(unittest.TestCase): "CREATE TABLE IF NOT EXISTS audit_log(user_id TEXT," " project_id TEXT, action TEXT, detail TEXT," " created_at DOUBLE PRECISION)") + # Audit batch 2 (I-5)：setUp 已按生产实际经 AuthStore + # 建好含 id 列的 audit_log 表，INSERT 必须显式列名， + # 否则在 6 列的生产表上会因 5 个值报错；显式列名在 + # 旧 5 列手工表上同样成立。 conn.execute( - "INSERT INTO audit_log VALUES('system', 'p1'," + "INSERT INTO audit_log(user_id, project_id, action," + " detail, created_at) VALUES('system', 'p1'," " 'worker_tool_gate_rejected'," " 'C1: agent 声明必需工具缺失：websearch', ?)", (time.time(),)) diff --git a/cloud/tests/test_audit_loop_coord_fix.py b/cloud/tests/test_audit_loop_coord_fix.py index b67dffb..9a877ee 100644 --- a/cloud/tests/test_audit_loop_coord_fix.py +++ b/cloud/tests/test_audit_loop_coord_fix.py @@ -25,11 +25,13 @@ from pathlib import Path from types import SimpleNamespace try: + from cloud.auth import AuthStore from cloud.coordinator import build_report, evaluate_outcome from cloud.server import init_db from cloud.worker import Worker except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore from cloud.coordinator import build_report, evaluate_outcome from cloud.server import init_db from cloud.worker import Worker @@ -111,6 +113,13 @@ class CoordFixTestCase(unittest.TestCase): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) + # Audit batch 2 (I-5)，与 test_audit_loop_coord.py 同款 + # 理由：生产里 audit_log 由首次真实 _audit 经 AuthStore + # 建好，合流时表一定存在；本文件把 _audit 桩成列表， + # 表永远不会出现，夹具里按生产实际补建表，否则合流 + # 测试会因缺表改走 escalate，与被测行为无关。 + _store = AuthStore(self.db_path) + _store.conn.close() self.env = {"XCM_API_KEY": "test-placeholder"} self.audits = [] self.worker = Worker(self.db_path, "/fake/host.py", @@ -200,10 +209,14 @@ class CoordFixTestCase(unittest.TestCase): # 最旧 id 的子任务必须不在，且集合恰好 5000 条。 conn = sqlite3.connect(self.db_path) try: + # Audit batch 2 (I-5)：setUp 已按生产实际经 AuthStore + # 建好含 id 自增列的 audit_log 表，建表改 IF NOT + # EXISTS——生产表与本测试要的表在 id 语义上一致， + # tie-break 断言不受影响。 conn.execute( - "CREATE TABLE audit_log(id INTEGER PRIMARY KEY" - " AUTOINCREMENT, user_id TEXT, project_id TEXT," - " action TEXT, detail TEXT," + "CREATE TABLE IF NOT EXISTS audit_log(id INTEGER" + " PRIMARY KEY AUTOINCREMENT, user_id TEXT," + " project_id TEXT, action TEXT, detail TEXT," " created_at DOUBLE PRECISION)") now = time.time() conn.executemany( diff --git a/cloud/worker.py b/cloud/worker.py index 94d9898..5fe3650 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -472,6 +472,17 @@ def _task_from_row(row): } +class _CoordGateLookupFailed(Exception): + """协调员门拒绝审计查询失败（Audit batch 2，I-5）。 + + 由 _coord_gate_rejected_child_ids 在两次查询都失败时抛出， + 调用方（_coord_audit_loop_merge）捕获后转 escalate——分类 + 依据缺失时绝不能回落到“空集 → 执行失败 → 推回”旧路。 + 抛错而不是返回哨兵值：哨兵会被任何忘记检查的调用方 + 当成集合误用，异常则无法被静默忽略。 + """ + + class Worker: """单机 worker：一轮认领一个 queued 任务并执行到底。 @@ -555,6 +566,12 @@ class Worker: # 审计，让“策略已死”在运行期可见；models 侧的警告 # 日志见 load_routing_policy。 self.routing_policy_load_failed = False + # Audit batch 2 (I-2): 部分失败标志——策略可用，但有 + # 来源失败或档被 fail-closed 弹出（见 models.py 的 + # partial_failures / dropped_tiers）。run_one 里与 + # 全失败同款 action 记任务级审计，detail 里区分 + # partial 与 full，不混为一谈。 + self.routing_policy_partial_failure = False if isinstance(routing_policy, RoutingPolicy): self.routing_policy = routing_policy elif isinstance(routing_policy, dict): @@ -573,6 +590,11 @@ class Worker: self.routing_policy_load_failed = True else: raise ValueError("routing_policy 必须是 RoutingPolicy 或 dict") + if self.routing_policy is not None and ( + getattr(self.routing_policy, "partial_failures", None) + or getattr(self.routing_policy, "dropped_tiers", + None)): + self.routing_policy_partial_failure = True # 真实 fallback 换模型列表：默认空 = 关闭，只记事件不换模型 # （老行为，生产 worker 未配该参数时绝不突然多花模型钱）。 # 归一化：去空白、去重保序、去掉与主模型相同的项（换成同一 @@ -6139,8 +6161,15 @@ class Worker: IDENTITY——此前注释称“没有 id 列”是错的，真正无 id 的只是测试手工建的表）。排序用 created_at 倒序、id 作 tie-break；缺 id 列的库（如手工建表）会炸掉第一 - 条查询，此时回落到仅按 created_at 排序再查一次， - 仍失败才记警告日志返回空集，绝不静默吞掉。 + 条查询，此时回落到仅按 created_at 排序再查一次。 + Audit batch 2（I-5）：两次查询都失败时不再返回空集 + ——空集会被合流读成“没有门拒绝”，把门拒绝子任务 + 重分类为执行失败推进回循环。此时抛 + _CoordGateLookupFailed，由合流捕获后转 escalate + （分类依据缺失→升级而非推回）；警告日志仍记。 + 注意：生产里 audit_log 由 AuthStore 在首次审计时建 + 好，合流时表一定存在；缺表只会出现在 _audit 被桩 + 掉的测试里，那些夹具须按生产实际补建表（见测试）。 查询取 5000 行封顶：超限时回落标记仍可兜底，绝不 全表扫；封顶漏行只会导致执行失败走推回而非升级， 是 fail-open 向推回，不是向放行。 @@ -6174,7 +6203,7 @@ class Worker: "coord gate audit lookup failed: %s", exc) except Exception: pass - return set() + raise _CoordGateLookupFailed(str(exc)) from exc wanted = set(child_ids or []) started = attempt_started_at or {} found = set() @@ -6327,10 +6356,26 @@ class Worker: for child in children or []: child_names[child["id"]] = child["name"] rows = self._coord_child_rows(child_ids) - gate_rejected_ids = self._coord_gate_rejected_child_ids( - project_id, child_ids, - {cid: (rows.get(cid) or {}).get("started_at") - for cid in child_ids}) + # Audit batch 2 (I-5): 门拒绝审计查不到时不猜—— + # 分类依据缺失→升级而非推回。查询失败时清包、 + # 记 escalated 审计、返回升级决策（父任务由调用 + # 方置 failed，不重排、不耗轮次），与门拒绝升级 + # 同款终局，只是原因不同。 + try: + gate_rejected_ids = self._coord_gate_rejected_child_ids( + project_id, child_ids, + {cid: (rows.get(cid) or {}).get("started_at") + for cid in child_ids}) + except _CoordGateLookupFailed as exc: + self._coord_clear_packets(task, child_ids) + self._audit(project_id, + "worker_coord_audit_loop_escalated", + "%s gate_lookup_failed=%s" % (task_id, exc)) + return { + "decision": "escalate", + "error": "审计循环升级（门拒绝审计查询失败，分类依据" + "缺失，不推回）", + "gate_lookup_failed": True} gate_results = {} findings = [] gate_rejected = [] @@ -6856,6 +6901,8 @@ class Worker: 一个注定缺工具的任务。租户目录准备失败时不由本 方法定罪（返回 []，让 _docker_task_context 的 租户准备失败路径处理），避免同一失败被两处各说一遍。 + Audit batch 2（I-5）：返回 [] 放行时同时记一条审计行 + （worker_required_tools_check_failed），放行可见。 """ tools = task.get("agent_tools") or [] if not tools: @@ -6873,11 +6920,20 @@ class Worker: except Exception as exc: # 不由本方法定罪，但绝不静默吞掉：记日志，真正 # 的租户准备失败由 _docker_task_context 报错。 + # Audit batch 2（I-5）：放行仍是放行（返回 [] 不变）， + # 但同时记一条点名依据缺失的审计行，让“工具门 + # 因读不到租户配置而放行”在审计里可见；审计 + # 本身 best-effort，写失败不改放行决策。 try: self.logger.warning( "required tools check skipped: %s", exc) except Exception: pass + if task.get("project_id"): + self._audit(task["project_id"], + "worker_required_tools_check_failed", + "%s error=%s" % ( + task.get("id") or "-", exc)) return [] return _missing_tools(tools, config_text) @@ -6901,7 +6957,11 @@ class Worker: policy 行只可能来自该任务曾经模型为空的那次派发， 也就是写回的来源。 查找失败（DB 异常）时返回 False 并记警告日志：复核查 - 不到依据就跳过，而不是拦任务。 + 不到依据就跳过，而不是拦任务。Audit batch 2（I-5）： + 跳过不再静默——同时记一条点名依据缺失的审计行 + （worker_persisted_model_lookup_failed），让“复核 + 因查不到依据而放行”在审计里可见；审计本身 + best-effort，写失败不改放行决策。 残留缺口措辞（9.5-code，M1）：fixes3 时期由 shadow / worker_default 决策写回的旧模型既无 persisted 行又 @@ -6928,6 +6988,12 @@ class Worker: task_id=task_id, project_id=project_id, status="warning", message="persisted model lookup failed: %s" % exc) + # Audit batch 2 (I-5): 依据缺失也要在审计里点名， + # 不能只有日志——日志与审计是两条可见性通道。 + # _audit 本身 best-effort，写失败不改返回 False。 + self._audit(project_id, + "worker_persisted_model_lookup_failed", + "%s error=%s" % (task_id, exc)) return False for row in rows: if row[0] != "worker_model_persisted": @@ -7100,14 +7166,28 @@ class Worker: task = self._apply_audit_loop_packet(task) self._audit(task["project_id"], "worker_claim", "%s executor=%s" % (task_id, self._executor_name())) - # Audit fix 3（I3）：策略已死时每任务记一条审计， - # 否则运营方只能靠日志猜“策略为什么没生效”。 - # _audit 本身 best-effort，不影响派发。 + # Audit fix 3（I3）+ batch 2（I-2）：策略已死或部分 + # 失败时每任务记一条审计，否则运营方只能靠日志猜 + # “策略为什么没生效”。全失败与部分失败用同款 + # action、detail 里区分 failure=full/partial 并附 + # 丢弃档与来源失败清单；_audit 本身 best-effort， + # 不影响派发。 if getattr(self, "routing_policy_load_failed", False): self._audit(task["project_id"], "worker_routing_policy_load_failed", - "%s project=%s" % (task_id, - task["project_id"])) + "%s project=%s failure=full" % ( + task_id, task["project_id"])) + elif getattr(self, "routing_policy_partial_failure", False): + _policy = self.routing_policy + self._audit(task["project_id"], + "worker_routing_policy_load_failed", + "%s project=%s failure=partial dropped_tiers=%s" + " partial_failures=%s" % ( + task_id, task["project_id"], + ",".join(getattr( + _policy, "dropped_tiers", []) or []), + "; ".join(getattr( + _policy, "partial_failures", []) or []))) # log_event 只接受 task_id/project_id/status/message 四个 # 字段，executor 只进审计 detail，不进日志调用（否则 # TypeError 会炸掉整个 run_one）。