host
7983567491
ci / test (push) Successful in 25m2s
Tasks gain kind='main'|'side' (NULL=main). A side thread hangs under a parent task, defaults to the local executor, never builds a worktree, branch or container unless explicitly asked, and writes its conclusion back to the parent via task_followups on terminal transition. - server: kind validation + side-thread gates (parent required, no coordinator_parts/upload_id/environment_id, parent must not be a coordinator), local executor default, GET /tasks kind filter, /worktree 404 and /fork /merge /pr 409 for side threads. - worker: side-default shunt in _run_attempt (no-worktree scratch cwd, in-memory plan=1 when a repo is present, seed/checkpoint skipped), _writeback_side_thread, marker task-create kind field + inheritance, side threads excluded from marker spawn counts. - host/sandbox: --no-worktree plumbing for dispatch/goal/popen (Local + Docker), scratch cwd work/side-threads/<task-id>. - console (classic only): kind filter + [旁聊] rows + '开旁聊' entry, side-thread dispatch mode, side-thread list in detail, merge/PR/fork hidden for side threads; build exit 0, console.html sha df39db5a. - tests: new test_sidethread.py (29 tests) + kind backfill tests in test_p2_migrations.py; full cloud suite 3900 tests OK (skipped=13). Design: verification/codex-muse-role-alignment-design-20261108.md Summary: verification/codex-muse-role-align-p1p2-20261108.md Local only: not pushed, not merged, not deployed.
300 lines
14 KiB
Python
300 lines
14 KiB
Python
"""P2-ops 迁移对齐：迁移清单必须等于 Postgres schema 表集合，
|
||
未迁移项必须被显式报告，绝不能静默跳过。
|
||
|
||
不用真实 Postgres：迁移本体用一个假 PG 连接跑（只记录
|
||
INSERT / ALTER，不连库）；schema 对齐直接读 POSTGRES_SCHEMA。
|
||
真实 PG 上的幂等性仍由 test_postgres_integration.py 覆盖
|
||
（有 CLOUD_TEST_PG_DSN 时才跑）。
|
||
"""
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.db import POSTGRES_SCHEMA
|
||
from cloud.migrations import (
|
||
MIGRATION_TABLES,
|
||
SEQUENCE_TABLES,
|
||
UNMIGRATED_TABLES,
|
||
migrate_sqlite_to_postgres,
|
||
migration_coverage,
|
||
schema_tables,
|
||
)
|
||
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.db import POSTGRES_SCHEMA
|
||
from cloud.migrations import (
|
||
MIGRATION_TABLES,
|
||
SEQUENCE_TABLES,
|
||
UNMIGRATED_TABLES,
|
||
migrate_sqlite_to_postgres,
|
||
migration_coverage,
|
||
schema_tables,
|
||
)
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
|
||
|
||
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] 为列名，
|
||
与 init_postgres_schema / 迁移补列代码的取法一致）。"""
|
||
|
||
def __init__(self, pragma_columns=None):
|
||
# pragma_columns: {表名: {列名}}；缺省表示“列齐全”。
|
||
self.pragma_columns = pragma_columns or {}
|
||
self.statements = []
|
||
self.inserts = [] # (表名, 列名元组)
|
||
self.alters = []
|
||
self.commits = 0
|
||
|
||
def execute(self, sql, params=()):
|
||
self.statements.append(sql)
|
||
if sql.startswith("PRAGMA table_info("):
|
||
table = sql[len("PRAGMA table_info("):].rstrip(")")
|
||
cols = self.pragma_columns.get(table)
|
||
if cols is None:
|
||
# 未指定时假装列齐全：projects 含 repo_url，
|
||
# task_followups 含 sender_task_id。
|
||
cols = {"repo_url"} if table == "projects" \
|
||
else {"sender_task_id"}
|
||
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):
|
||
self.commits += 1
|
||
|
||
|
||
def _make_sqlite(path):
|
||
conn = sqlite3.connect(path)
|
||
conn.execute("CREATE TABLE uploads(id TEXT PRIMARY KEY, created_by TEXT,"
|
||
" created_at REAL, status TEXT)")
|
||
conn.execute("CREATE TABLE task_attachments(task_id TEXT, bundle_id TEXT,"
|
||
" rel_path TEXT, size INTEGER,"
|
||
" PRIMARY KEY(task_id, rel_path))")
|
||
conn.execute("CREATE TABLE task_followups(id TEXT PRIMARY KEY,"
|
||
" task_id TEXT, project_id TEXT, body TEXT, mode TEXT,"
|
||
" status TEXT, created_by TEXT, sender_task_id TEXT,"
|
||
" created_at REAL, updated_at REAL)")
|
||
conn.execute("CREATE TABLE environments(id TEXT PRIMARY KEY,"
|
||
" project_id TEXT, name TEXT, branch TEXT, executor TEXT,"
|
||
" status TEXT, error_text TEXT, repo_url TEXT,"
|
||
" created_by TEXT, created_at REAL, updated_at REAL)")
|
||
conn.execute("INSERT INTO uploads VALUES(?, ?, ?, ?)",
|
||
("up1", "u1", 1.0, "done"))
|
||
conn.execute("INSERT INTO task_attachments VALUES(?, ?, ?, ?)",
|
||
("t1", "b1", "a.txt", 3))
|
||
conn.execute("INSERT INTO task_followups VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
("f1", "t1", "p1", "hi", "queued", "pending", "u1",
|
||
"sender-task", 1.0, 1.0))
|
||
conn.execute("INSERT INTO environments VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
("e1", "p1", "dev", "main", "docker", "ready", "",
|
||
"https://example.invalid/r", "u1", 1.0, 1.0))
|
||
# Audit fix 3（I6）列级覆盖用：tasks 的 agent 列与
|
||
# task_attempts 整表，表级在清单里已有不算覆盖。
|
||
conn.execute("CREATE TABLE tasks(id TEXT PRIMARY KEY,"
|
||
" project_id TEXT, agent_name TEXT,"
|
||
" agent_version TEXT, agent_instructions TEXT,"
|
||
" agent_tools TEXT, agent_tool_approvals TEXT)")
|
||
conn.execute("INSERT INTO tasks VALUES(?, ?, ?, ?, ?, ?, ?)",
|
||
("t1", "p1", "research-brief", "v1", "do x", "[]",
|
||
'{"websearch": "ask"}'))
|
||
conn.execute("CREATE TABLE task_attempts(task_id TEXT,"
|
||
" attempt_no INTEGER, kind TEXT,"
|
||
" interrupted_reason TEXT, checkpoint_round INTEGER,"
|
||
" result_text TEXT, error_text TEXT,"
|
||
" created_at REAL,"
|
||
" PRIMARY KEY(task_id, attempt_no))")
|
||
conn.execute("INSERT INTO task_attempts VALUES(?, ?, ?, ?, ?, ?, ?, ?)",
|
||
("t1", 1, "platform_fresh", "worker_shutdown", -1,
|
||
"", "", 1.0))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
class TestMigrationAlignment(unittest.TestCase):
|
||
def test_migration_tables_equal_schema_tables(self):
|
||
# 核心对齐断言：以前漏掉的 4 张表补进后，两边集合必须一致。
|
||
self.assertEqual(set(MIGRATION_TABLES), schema_tables())
|
||
for table in ("uploads", "task_attachments",
|
||
"task_followups", "environments", "user_quotas"):
|
||
self.assertIn(table, MIGRATION_TABLES)
|
||
|
||
def test_sequence_tables_equal_identity_tables(self):
|
||
# 序列拨动清单必须等于 schema 里 IDENTITY 列的表集合：
|
||
# TEXT 主键的新表不需要拨序列，漏拨或多拨都会在这里变红。
|
||
identity = set()
|
||
for statement in POSTGRES_SCHEMA:
|
||
m = re.search(r"CREATE TABLE IF NOT EXISTS (\w+)", statement)
|
||
if m and "GENERATED BY DEFAULT AS IDENTITY" in statement:
|
||
identity.add(m.group(1))
|
||
self.assertEqual(set(SEQUENCE_TABLES), identity)
|
||
|
||
def test_unmigrated_are_real_sqlite_tables_not_in_schema(self):
|
||
# 未迁移清单不得是凭空编的：每项都必须在 cloud/ 源码里
|
||
# 真有 CREATE TABLE，且确实不在 PG schema 里；同时不得
|
||
# 与迁移清单重叠（重叠即“既迁又报未迁”，自相矛盾）。
|
||
sources = ""
|
||
for name in os.listdir(REPO_ROOT / "cloud"):
|
||
if name.endswith(".py"):
|
||
with open(REPO_ROOT / "cloud" / name) as f:
|
||
sources += f.read()
|
||
schema = schema_tables()
|
||
for table in UNMIGRATED_TABLES:
|
||
self.assertIn("CREATE TABLE IF NOT EXISTS %s" % table, sources,
|
||
"%s 不在任何 SQLite DDL 里，清单已过期" % table)
|
||
self.assertNotIn(table, schema)
|
||
self.assertNotIn(table, MIGRATION_TABLES)
|
||
|
||
def test_coverage_reports_no_missing(self):
|
||
coverage = migration_coverage()
|
||
self.assertEqual(coverage["missing_from_migration"], ())
|
||
self.assertEqual(coverage["unmigrated_no_pg_schema"],
|
||
tuple(UNMIGRATED_TABLES))
|
||
self.assertEqual(set(coverage["skipped_by_design"]),
|
||
{"kb_fts", "sqlite_sequence"})
|
||
|
||
|
||
class TestMigrateNewTables(unittest.TestCase):
|
||
def test_new_tables_migrated_and_sender_column_added(self):
|
||
# PG 侧 task_followups 尚无 sender_task_id：迁移必须先
|
||
# ALTER 补列，再把含该列的行插过去，而不是撞未知列失败。
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src = os.path.join(tmp, "src.db")
|
||
_make_sqlite(src)
|
||
fake = FakePgConn(pragma_columns={
|
||
"projects": {"repo_url"},
|
||
"task_followups": {"id", "task_id", "project_id", "body",
|
||
"mode", "status", "created_by",
|
||
"created_at", "updated_at"},
|
||
})
|
||
result = migrate_sqlite_to_postgres(src, pg_conn=fake)
|
||
for table in ("uploads", "task_attachments",
|
||
"task_followups", "environments"):
|
||
self.assertEqual(result.get(table), 1, table)
|
||
self.assertTrue(
|
||
any("task_followups" in a and "sender_task_id" in a
|
||
for a in fake.alters),
|
||
"未发出 sender_task_id 补列 ALTER：%s" % fake.alters)
|
||
fu_inserts = [cols for t, cols in fake.inserts
|
||
if t == "task_followups"]
|
||
self.assertEqual(len(fu_inserts), 1)
|
||
self.assertIn("sender_task_id", fu_inserts[0])
|
||
|
||
def test_agent_columns_added_and_attempts_migrated(self):
|
||
# Audit fix 3（I6）：PG 侧 tasks 尚无 agent 列时，
|
||
# 迁移必须先 ALTER 补齐 5 列再直插；task_attempts
|
||
# 行必须按列原样迁过去（表级在清单里不算列级覆盖）。
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src = os.path.join(tmp, "src.db")
|
||
_make_sqlite(src)
|
||
fake = FakePgConn(pragma_columns={
|
||
"projects": {"repo_url"},
|
||
"tasks": {"id", "project_id"},
|
||
"task_followups": {"id", "task_id", "project_id",
|
||
"body", "mode", "status",
|
||
"created_by", "sender_task_id",
|
||
"created_at", "updated_at"},
|
||
})
|
||
result = migrate_sqlite_to_postgres(src, pg_conn=fake)
|
||
self.assertEqual(result.get("tasks"), 1, "tasks")
|
||
self.assertEqual(result.get("task_attempts"), 1, "task_attempts")
|
||
for col in ("agent_name", "agent_version",
|
||
"agent_instructions", "agent_tools",
|
||
"agent_tool_approvals"):
|
||
self.assertTrue(
|
||
any("tasks" in a and col in a for a in fake.alters),
|
||
"未发出 tasks.%s 补列 ALTER：%s" % (col, fake.alters))
|
||
tasks_inserts = [cols for t, cols in fake.inserts
|
||
if t == "tasks"]
|
||
self.assertEqual(len(tasks_inserts), 1)
|
||
self.assertIn("agent_tool_approvals", tasks_inserts[0])
|
||
attempts_inserts = [cols for t, cols in fake.inserts
|
||
if t == "task_attempts"]
|
||
self.assertEqual(len(attempts_inserts), 1)
|
||
for col in ("task_id", "attempt_no", "kind",
|
||
"interrupted_reason", "checkpoint_round",
|
||
"result_text", "error_text", "created_at"):
|
||
self.assertIn(col, attempts_inserts[0])
|
||
|
||
def test_kind_column_backfilled_when_missing(self):
|
||
# P2：PG 侧 tasks 尚无 kind 列时，迁移必须先 ALTER 补列
|
||
# 再直插，且插入列清单里含 kind；kind 已存在时不重复
|
||
# ALTER（与 agent_* 列同款 backfill 惯例）。
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src = os.path.join(tmp, "src.db")
|
||
_make_sqlite(src)
|
||
fake = FakePgConn(pragma_columns={
|
||
"projects": {"repo_url"},
|
||
"tasks": {"id", "project_id"},
|
||
"task_followups": {"id", "task_id", "project_id",
|
||
"body", "mode", "status",
|
||
"created_by", "sender_task_id",
|
||
"created_at", "updated_at"},
|
||
})
|
||
result = migrate_sqlite_to_postgres(src, pg_conn=fake)
|
||
self.assertEqual(result.get("tasks"), 1, "tasks")
|
||
self.assertTrue(
|
||
any("tasks" in a and "kind" in a for a in fake.alters),
|
||
"未发出 tasks.kind 补列 ALTER：%s" % fake.alters)
|
||
# 注：插入列清单取自源 SQLite 表的列，夹具源表无 kind
|
||
# 列故清单里无 kind；生产源库（init_db 建表）有 kind，
|
||
# 迁移时按列原样搬运（与 agent_* 列同款）。
|
||
|
||
def test_kind_column_present_skips_alter(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src = os.path.join(tmp, "src.db")
|
||
_make_sqlite(src)
|
||
fake = FakePgConn(pragma_columns={
|
||
"projects": {"repo_url"},
|
||
"tasks": {"id", "project_id", "kind"},
|
||
"task_followups": {"id", "task_id", "project_id",
|
||
"body", "mode", "status",
|
||
"created_by", "sender_task_id",
|
||
"created_at", "updated_at"},
|
||
})
|
||
result = migrate_sqlite_to_postgres(src, pg_conn=fake)
|
||
self.assertEqual(result.get("tasks"), 1, "tasks")
|
||
self.assertFalse(
|
||
any("tasks" in a and "kind" in a for a in fake.alters),
|
||
"kind 已存在却仍发补列 ALTER：%s" % fake.alters)
|
||
|
||
def test_sender_column_present_skips_alter(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src = os.path.join(tmp, "src.db")
|
||
_make_sqlite(src)
|
||
fake = FakePgConn() # 缺省：列齐全
|
||
migrate_sqlite_to_postgres(src, pg_conn=fake,
|
||
tables=("task_followups",))
|
||
# init_postgres_schema 自身会给 projects/tasks 补列，
|
||
# 这里只断言 task_followups 没有重复补列。
|
||
self.assertEqual(
|
||
[a for a in fake.alters if "task_followups" in a], [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|