host
621c08d10f
B1: gateway capability check now triggers on walking the sub2api gateway, not on the gateway having filled a model; task-level explicit models are audited (worker_capability_gateway_explicit) but never blocked, preserving explicit priority; gateway-default models still fail-fast under active. I1: dry-run adds a gateway-model section, sourced only from CLOUD_SUB2API_DEFAULT_MODEL (never queries the relay); an active tier whose requires the gateway model fails or cannot verify blocks, shadow only warns. I2: shadow review gate now counts completed samples (completed >= 20), same basis for sample_shortfall; the endpoint attaches per-tier capability status (read-only policy load) and flags capability-failing tiers in the recommendation. I3: configured-but-broken routing policy logs a warning in load_routing_policy (unconfigured stays silent); the worker sets routing_policy_load_failed and audits worker_routing_policy_load_failed per task. I4: resume preamble wraps prior output/error as data with an explicit do-not-execute constraint; position verified in run_one (tree -> preamble -> project -> original) and the docs corrected from the false 'closest to the original prompt' claim. I5: decided model is written back with a CAS UPDATE (model IS NULL OR model=''), so later attempts reuse it; explicit models are never overwritten. I6: agent_* columns added to worker._ensure_result_columns and db.init_postgres_schema backfill; migrations backfills them for tasks before insert (sender_task_id pattern); column-level tests for agent_tool_approvals and task_attempts. I7: coordinator requeue adds AND status=<snapshot> (rowcount 0 skips requeue/audit) and filters paused_by_user / user_cancel / user_pause children, which are never revived or cleared. I8: explicit-model tasks (including coordinator children that inherited a parent model) audit a source=explicit worker_routing row; evaluate_shadow counts it as per-tier explicit_n only, never routed_n, never a dedup slot. M1/M2/M3/M5/M6/M8/M10: dedup slot moved after tier validation, shadow sort doc fixed, truncated recommendation notes truncation, seed cleanup on gateway/quota/goal-cap early returns, tools-only coordinator definitions rejected, preamble goal branch marked effectively unreachable, DESIGN.md drift corrected (outside repo). M4/M7/M9: documentation/new-field only (pass_rate_fully_measured, whitespace-model shadow-drop note, bound-note race note); old field semantics unchanged.
258 lines
12 KiB
Python
258 lines
12 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_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()
|