diff --git a/cloud/migrations.py b/cloud/migrations.py index 014f618..5fad3c4 100644 --- a/cloud/migrations.py +++ b/cloud/migrations.py @@ -221,15 +221,24 @@ def migrate_sqlite_to_postgres(sqlite_path, pg_conn=None, tables=None): "PRAGMA table_info(tasks)").fetchall()} except Exception: _existing = set() - for _col in ("agent_name", "agent_version", - "agent_instructions", "agent_tools", - "agent_tool_approvals", "audit_loop", - "audit_loop_packet"): + for _col, _coltype in ( + ("agent_name", "TEXT"), + ("agent_version", "TEXT"), + ("agent_instructions", "TEXT"), + ("agent_tools", "TEXT"), + ("agent_tool_approvals", "TEXT"), + # 审计循环接线：开关列必须与 db.py 回填清单、 + # server/worker ensure 清单一致用 INTEGER； + # 若建成 TEXT，PG 读回的 "0" 为真值，worker + # 的 `or 0` 兜底拦不住，audit_loop=0 的 + # 历史任务会被悄悄当成启用（Audit batch 1）。 + ("audit_loop", "INTEGER"), + ("audit_loop_packet", "TEXT")): if _col not in _existing: try: pg_conn.execute( - "ALTER TABLE tasks ADD COLUMN %s TEXT" - % _col) + "ALTER TABLE tasks ADD COLUMN %s %s" + % (_col, _coltype)) except Exception: pass if table == "task_followups" and "sender_task_id" in columns: diff --git a/cloud/tests/test_audit_batch1.py b/cloud/tests/test_audit_batch1.py new file mode 100644 index 0000000..b739a09 --- /dev/null +++ b/cloud/tests/test_audit_batch1.py @@ -0,0 +1,247 @@ +"""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() diff --git a/cloud/worker.py b/cloud/worker.py index e006d91..94d9898 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -7330,6 +7330,9 @@ class Worker: # Audit fix 3（M5）：网关失败早返回也要清暂存， # 与缺 key 门、仓库门同款，否则 bundle 目录残留。 self._cleanup_task_seed(task) + # Audit batch 1（B-C）：网关失败早返回也要清认领占位， + # 与缺 key 门、仓库门同款，否则容量槽位漏到 worker 退出。 + self._running_clear(task_id) return {"dispatched": True, "task_id": task_id, "host_name": self.executor._host_name(task) if hasattr(self.executor, "_host_name") else task_id,