fix: audit batch 1 (B-A migration audit_loop INTEGER, B-C gateway failure clears running slot)

- B-A: migrations.py tasks backfill loop now types audit_loop as
  INTEGER (other 6 columns stay TEXT), matching db.py backfill,
  server/worker ensure lists; TEXT readback '0' is truthy and
  would silently enable the audit loop for audit_loop=0 rows.
- B-C: gateway preparation failure early-return now calls
  _running_clear(task_id) next to _cleanup_task_seed, matching
  missing-key/repo/capability/tool gates; previously the capacity
  slot leaked until worker exit (default capacity 1 blocks reclaim).
- Tests: cloud/tests/test_audit_batch1.py (3 tests): migration
  ALTER types, INTEGER 0 readback not enabled via real worker
  row mapping, gateway failure clears slot and capacity reclaim.
  _claim doc 'three gates' list left unchanged (full early-return
  gate enumeration not verified this batch; residual wording).
- No other audit items touched (I-1..I-14, N-1..N-12).

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:

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()

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,