"""审计修复批 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()