"""Audit batch 1：阻塞项 B-A 与 B-C 的回归测试。 B-A：迁移工具回填 tasks 列时 audit_loop 必须用 INTEGER（与 db.py 回填清单、server/worker ensure 清单一致），否则 PG 读回 的 "0" 为真值，worker 的 `or 0` 兜底拦不住，audit_loop=0 的 历史任务会被悄悄当成启用。 B-C：网关准备失败早返回必须清认领占位，否则容量槽位漏到 worker 退出（容量默认 1 时下一次认领直接被堵住）。 只用假 PG 连接、假网关与假 runner，不跑真实模型回合、不联网、 不登录任何服务；env 里的值只是占位用的假名。 """ import json import os import re import sqlite3 import sys import tempfile import time import unittest from pathlib import Path from types import SimpleNamespace try: from cloud.migrations import migrate_sqlite_to_postgres from cloud.server import init_db from cloud.worker import Worker, _task_from_row except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.migrations import migrate_sqlite_to_postgres from cloud.server import init_db from cloud.worker import Worker, _task_from_row class _FakeCursor: def __init__(self, rows=None, rowcount=1): self._rows = rows or [] self.rowcount = rowcount def fetchall(self): return self._rows class _FakePgConn: """只记录 SQL 的假 PG 连接：init_postgres_schema 的 DDL 直接忽略；PRAGMA 按构造时给的列集合回答（row[1] 为列名）。""" def __init__(self, pragma_columns=None): self.pragma_columns = pragma_columns or {} self.alters = [] self.inserts = [] def execute(self, sql, params=()): if sql.startswith("PRAGMA table_info("): table = sql[len("PRAGMA table_info("):].rstrip(")") cols = self.pragma_columns.get(table) or set() rows = [(i, c, None) for i, c in enumerate(sorted(cols))] return _FakeCursor(rows=rows) m = re.search(r'INSERT INTO "(\w+)" \(([^)]*)\)', sql) if m: cols = tuple(c.strip().strip('"') for c in m.group(2).split(",")) self.inserts.append((m.group(1), cols)) return _FakeCursor(rowcount=1) if sql.startswith("ALTER TABLE"): self.alters.append(sql) return _FakeCursor(rowcount=0) return _FakeCursor(rowcount=0) def commit(self): pass def _make_migration_source(path): conn = sqlite3.connect(path) # 旧 tasks 表：缺 7 列回填目标列，只有一行历史任务。 conn.execute("CREATE TABLE tasks(id TEXT PRIMARY KEY," " project_id TEXT)") conn.execute("INSERT INTO tasks VALUES(?, ?)", ("old-1", "p1")) conn.commit() conn.close() class AuditBatch1MigrationTestCase(unittest.TestCase): def test_audit_loop_backfilled_as_integer_others_as_text(self): with tempfile.TemporaryDirectory() as tmp: src = os.path.join(tmp, "src.db") _make_migration_source(src) fake = _FakePgConn(pragma_columns={ "projects": {"repo_url"}, "tasks": {"id", "project_id"}, }) result = migrate_sqlite_to_postgres( src, pg_conn=fake, tables=("tasks",)) self.assertEqual(result.get("tasks"), 1, "tasks") tasks_alters = [a for a in fake.alters if "tasks" in a] # 注意：假连接是无状态的（ALTER 后 PRAGMA 回答不变），所以 # init_postgres_schema 的回填会先发一遍 ALTER，迁移循环 # 随后又发一遍；真实适配连接上 init 已补的列循环会跳过。 # 因此断言取每列最后一次 ALTER——循环在 init 之后跑， # 最后一次即循环本身发出的语句；循环改回 TEXT 时必败。 last_by_col = {} for stmt in tasks_alters: m = re.search(r"ADD COLUMN (\w+) (\w+)", stmt) self.assertIsNotNone(m, stmt) last_by_col[m.group(1)] = m.group(2) # 关键断言：audit_loop 必须是 INTEGER，其余 6 列仍是 TEXT。 self.assertEqual(last_by_col.get("audit_loop"), "INTEGER", last_by_col) for col in ("agent_name", "agent_version", "agent_instructions", "agent_tools", "agent_tool_approvals", "audit_loop_packet"): self.assertEqual(last_by_col.get(col), "TEXT", (col, last_by_col)) def test_audit_loop_zero_readback_not_enabled(self): # 真实 SQLite 类型语义 + 真实 worker 行映射： # INTEGER 列的 0 读回为 int 0，`or 0` 后仍为 0（关闭）； # 而 TEXT 列的 "0" 读回为非空字符串，`or 0` 拦不住（启用）。 # 本测试断言 INTEGER 一侧的行为，并以 TEXT 一侧作对照， # 证明类型本身就是开关语义的分界线。 conn = sqlite3.connect(":memory:") conn.row_factory = sqlite3.Row try: conn.execute("CREATE TABLE tasks(id TEXT PRIMARY KEY," " project_id TEXT, prompt TEXT, status TEXT," " idempotency_key TEXT, depends_on TEXT," " priority INTEGER, created_by TEXT," " created_at REAL, updated_at REAL," " last_heartbeat REAL, audit_loop INTEGER)") conn.execute( "INSERT INTO tasks VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ("t-int", "p1", "do", "queued", None, "[]", 0, "tester", 1.0, 1.0, None, 0)) row = conn.execute( "SELECT * FROM tasks WHERE id = ?", ("t-int",)).fetchone() task = _task_from_row(row) self.assertEqual(task["audit_loop"], 0) # worker 启用判断的语义：`if not task.get("audit_loop")` # 为真才关闭；INTEGER 0 必须走关闭分支。 self.assertTrue(not task.get("audit_loop")) # 对照：TEXT 列的 "0" 会走启用分支（bug 机制本身）。 conn.execute("CREATE TABLE tasks_text(id TEXT PRIMARY KEY," " project_id TEXT, prompt TEXT, status TEXT," " idempotency_key TEXT, depends_on TEXT," " priority INTEGER, created_by TEXT," " created_at REAL, updated_at REAL," " last_heartbeat REAL, audit_loop TEXT)") conn.execute( "INSERT INTO tasks_text VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ("t-text", "p1", "do", "queued", None, "[]", 0, "tester", 1.0, 1.0, None, "0")) row_text = conn.execute( "SELECT * FROM tasks_text WHERE id = ?", ("t-text",)).fetchone() task_text = _task_from_row(row_text) self.assertFalse(not task_text.get("audit_loop"), "TEXT '0' 读回为真值，会被当成启用：" "这正是 B-A 要求类型为 INTEGER 的原因") finally: 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 _insert_task(db_path, task_id): now = time.time() conn = sqlite3.connect(db_path) try: conn.execute( "INSERT INTO tasks(id, project_id, prompt, status," " idempotency_key, depends_on, priority, created_by," " created_at, updated_at, last_heartbeat, audit_loop," " goal_mode)" " VALUES(?, ?, ?, ?, NULL, '[]', 0, 'tester', ?, ?, NULL," " ?, ?)", (task_id, "p1", "do work", "queued", now, now, 0, 0), ) conn.commit() finally: conn.close() def _status(db_path, task_id): conn = sqlite3.connect(db_path) try: row = conn.execute( "SELECT status FROM tasks WHERE id = ?", (task_id,)).fetchone() finally: conn.close() return row[0] if row else None class AuditBatch1GatewaySlotTestCase(unittest.TestCase): def setUp(self): self._old_push_allow = 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) _insert_task(self.db_path, "G1") _insert_task(self.db_path, "G2") def tearDown(self): if getattr(self, "_old_push_allow", None) is None: os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) else: os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = ( self._old_push_allow) self.tmp.cleanup() def test_gateway_failure_clears_running_slot(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) worker._audit = lambda p, a, d="": None # 网关准备失败：_ensure_gateway 抛错，走真实 run_one 的 # 网关失败早返回分支（容量默认 1，槽位漏了就堵死下一次认领）。 worker._ensure_gateway = lambda task: (_ for _ in ()).throw( RuntimeError("gateway boom")) result = worker.run_one() self.assertEqual(result["status"], "failed") self.assertIn("网关准备失败", result["error"]) self.assertEqual(_status(self.db_path, "G1"), "failed") # 关键断言：失败后占位已清，否则容量 1 下 G2 认领不到。 self.assertEqual(worker._running_snapshot(), {}) # 容量可再认领：去掉网关桩后，同一 worker 能认领 G2 并完成。 del worker._ensure_gateway result2 = worker.run_one() self.assertTrue(result2["dispatched"]) self.assertEqual(result2["status"], "completed") self.assertEqual(_status(self.db_path, "G2"), "completed") if __name__ == "__main__": unittest.main()