"""Unified task_attempts schema (FLAG-1, user decision 2026-11-05). Pins the single-definition rule and the three-shape migration: - one DDL: attempt_ledger.TASK_ATTEMPTS_DDL is the same string used by cloud.attempts.CREATE_TABLE_SQL and by the task_attempts entry of cloud.db.POSTGRES_SCHEMA; - fresh database: ensure builds exactly the 12 unified columns; - old resume shape, gap-3 ledger shape, and the rehearsal-era 12-column shape all migrate in place with rows intact, and both writers (resume + ledger) work on the migrated table; - aggregation counts ledger rows only: a resume row (outcome NULL) never becomes a counted, defect-free attempt. SQLite only, run live. The Postgres upgrade path is NOT proven here (adapter-level DDL text check only, same as the other ledger tests); the batch summary states that explicitly. """ import os import sqlite3 import sys import unittest sys.path.insert(0, os.path.dirname(os.path.dirname( os.path.dirname(os.path.abspath(__file__))))) from cloud.attempt_ledger import ( # noqa: E402 PG_DDL, SQLITE_DDL, TASK_ATTEMPTS_DDL, UNIFIED_COLUMNS, _column_notnull, aggregate_attempts, ensure_task_attempts_table, read_attempt_rows, record_attempt as ledger_record, record_correction as ledger_correct, ) from cloud.attempts import ( # noqa: E402 CREATE_TABLE_SQL, ensure_attempts_table, latest_attempt, record_attempt as resume_record, ) from cloud.db import POSTGRES_SCHEMA # noqa: E402 from cloud.db import _translate_sql # noqa: E402 OLD_DDL = ( "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 DOUBLE PRECISION, " "PRIMARY KEY(task_id, attempt_no))") GAP3_DDL = ( "CREATE TABLE task_attempts(" "task_id TEXT NOT NULL, attempt_no INTEGER NOT NULL, " "outcome TEXT NOT NULL, is_defect INTEGER NOT NULL DEFAULT 0, " "defect_category TEXT, created_at DOUBLE PRECISION, " "PRIMARY KEY (task_id, attempt_no))") # Rehearsal-era shape: same 12 column names as unified, but the # looser rehearsal types (outcome nullable, is_defect without # NOT NULL, PK columns without NOT NULL spelling). REHEARSAL_DDL = ( "CREATE TABLE task_attempts(" "task_id TEXT, attempt_no INTEGER, outcome TEXT, " "is_defect INTEGER DEFAULT 0, defect_category TEXT, " "corrects_attempt_no INTEGER, kind TEXT, " "interrupted_reason TEXT, checkpoint_round INTEGER, " "result_text TEXT, error_text TEXT, " "created_at DOUBLE PRECISION, " "PRIMARY KEY(task_id, attempt_no))") def _columns(conn): return [row[1] for row in conn.execute( "PRAGMA table_info(task_attempts)").fetchall()] def _row_by_key(conn, task_id, attempt_no): row = conn.execute( "SELECT * FROM task_attempts WHERE task_id = ? " "AND attempt_no = ?", (task_id, attempt_no)).fetchone() cols = [d[0] for d in conn.execute( "SELECT * FROM task_attempts LIMIT 0").description] return dict(zip(cols, row)) if row is not None else None class TestSingleDefinition(unittest.TestCase): def test_one_ddl_string_everywhere(self): self.assertEqual(SQLITE_DDL, TASK_ATTEMPTS_DDL) self.assertEqual(PG_DDL, TASK_ATTEMPTS_DDL) self.assertEqual(CREATE_TABLE_SQL, TASK_ATTEMPTS_DDL) schema_statements = [s for s in POSTGRES_SCHEMA if "task_attempts" in s] self.assertEqual(len(schema_statements), 1) self.assertEqual(schema_statements[0], TASK_ATTEMPTS_DDL) def test_unified_column_set(self): self.assertEqual(tuple(UNIFIED_COLUMNS), ( "task_id", "attempt_no", "outcome", "is_defect", "defect_category", "corrects_attempt_no", "kind", "interrupted_reason", "checkpoint_round", "result_text", "error_text", "created_at")) def test_pg_ddl_text_is_static_only(self): translated, _kind = _translate_sql(PG_DDL) for col in UNIFIED_COLUMNS: self.assertIn(col, translated) self.assertIn("PRIMARY KEY (task_id, attempt_no)", translated) self.assertNotIn("AUTOINCREMENT", translated) self.assertNotIn("GENERATED", translated) class TestFreshAndMigrations(unittest.TestCase): def setUp(self): self.conn = sqlite3.connect(":memory:") self.conn.row_factory = sqlite3.Row def tearDown(self): self.conn.close() def test_fresh_build_has_exact_unified_columns(self): ensure_task_attempts_table(self.conn) self.assertEqual(_columns(self.conn), list(UNIFIED_COLUMNS)) # Idempotent, rows preserved. ledger_record(self.conn, "t-1", "success") ensure_task_attempts_table(self.conn) ensure_attempts_table(self.conn) self.assertEqual(_columns(self.conn), list(UNIFIED_COLUMNS)) self.assertEqual(self.conn.execute( "SELECT COUNT(*) FROM task_attempts").fetchone()[0], 1) def test_column_notnull_probe(self): # PG 实证（2026-11-06）：ensure 的 outcome 放宽依赖 # PRAGMA row[3] 的 notnull 语义；SQLite 侧钉住探针本身。 ensure_task_attempts_table(self.conn) flags = _column_notnull(self.conn, "task_attempts") self.assertFalse(flags["outcome"]) self.assertTrue(flags["task_id"]) self.assertTrue(flags["attempt_no"]) self.assertTrue(flags["is_defect"]) def test_column_notnull_probe_sees_gap3_outcome(self): # gap-3 形态的 outcome 是 NOT NULL；探针必须报 True， # 否则 PG 上的放宽不会触发，恢复写路仍会炸。 self.conn.execute(GAP3_DDL) flags = _column_notnull(self.conn, "task_attempts") self.assertTrue(flags["outcome"]) # 迁移本身不抛错、列集仍齐；真正的不变量是 ensure # 之后恢复写路与实际的 schema 状态一致，而不是 # outcome 的 NOT NULL 位：outcome 放宽了（PG 上 # ALTER 生效；SQLite 3.53.0 起也支持 ALTER COLUMN # DROP NOT NULL），resume 风格写行必须能写入并读回 # outcome 为 NULL 的行；outcome 仍 NOT NULL 时（更 # 早的 SQLite：ALTER 不合法、被 try/except 吞掉）， # 该行会被 INSERT OR IGNORE 静默丢弃——版本相关 # 边界，不是迁移失败。 ensure_task_attempts_table(self.conn) self.assertEqual(set(_columns(self.conn)), set(UNIFIED_COLUMNS)) relaxed = not _column_notnull( self.conn, "task_attempts")["outcome"] resume_record(self.conn, "t-gap", 2, "platform_fresh", "worker_shutdown", -1, "second", None) self.conn.commit() row = _row_by_key(self.conn, "t-gap", 2) if relaxed: self.assertIsNotNone(row) self.assertIsNone(row["outcome"]) latest = latest_attempt(self.conn, "t-gap") self.assertEqual(latest["attempt_no"], 2) else: self.assertIsNone(row) def test_old_shape_migrates_with_rows_intact(self): self.conn.execute(OLD_DDL) self.conn.execute( "INSERT INTO task_attempts(task_id, attempt_no, kind, " "interrupted_reason, checkpoint_round, result_text, " "error_text, created_at) VALUES(?, ?, ?, ?, ?, ?, ?, ?)", ("t-old", 1, "platform_fresh", "worker_shutdown", -1, "partial output", "killed", 1.0)) self.conn.commit() before = _row_by_key(self.conn, "t-old", 1) ensure_task_attempts_table(self.conn) # Migrated tables keep ALTER-append order, not fresh DDL # order; every read path is by column name, so the set # is the invariant, not the order (fresh tables pin order). self.assertEqual(set(_columns(self.conn)), set(UNIFIED_COLUMNS)) after = _row_by_key(self.conn, "t-old", 1) for col in ("task_id", "attempt_no", "kind", "interrupted_reason", "checkpoint_round", "result_text", "error_text", "created_at"): self.assertEqual(after[col], before[col], col) self.assertIsNone(after["outcome"]) # Both writers now work on the migrated table. resume_record(self.conn, "t-old", 2, "platform_fresh", "worker_shutdown", -1, "second", None) ledger_record(self.conn, "t-led", "failure", is_defect=True, defect_category="contract") self.conn.commit() latest = latest_attempt(self.conn, "t-old") self.assertEqual(latest["attempt_no"], 2) self.assertEqual(latest["kind"], "platform_fresh") # Aggregation counts the ledger row only, never the # resume rows (outcome NULL). agg = aggregate_attempts(read_attempt_rows(self.conn)) self.assertEqual(agg, {"t-led": { "attempts": 1, "defects": 1, "defect_category": "contract"}}) def test_gap3_shape_migrates_and_corrections_work(self): self.conn.execute(GAP3_DDL) self.conn.execute( "INSERT INTO task_attempts(task_id, attempt_no, outcome, " "is_defect, defect_category, created_at) " "VALUES('t-gap', 1, 'success', 0, NULL, 1.0)") self.conn.commit() ensure_task_attempts_table(self.conn) self.assertEqual(set(_columns(self.conn)), set(UNIFIED_COLUMNS)) row = _row_by_key(self.conn, "t-gap", 1) self.assertEqual(row["outcome"], "success") ledger_correct(self.conn, "t-gap", 1, is_defect=True, defect_category="contract") agg = aggregate_attempts(read_attempt_rows(self.conn)) self.assertEqual(agg["t-gap"]["defects"], 1) def test_rehearsal_shape_migrates_with_rows_intact(self): self.conn.execute(REHEARSAL_DDL) self.conn.execute( "INSERT INTO task_attempts(task_id, attempt_no, outcome, " "is_defect, defect_category, corrects_attempt_no, kind, " "interrupted_reason, checkpoint_round, result_text, " "error_text, created_at) VALUES(?, ?, ?, ?, ?, ?, ?, ?, " "?, ?, ?, ?)", ("t-reh", 1, "failure", 1, "contract", None, "platform_fresh", "worker_shutdown", -1, "out", "err", 2.0)) self.conn.commit() before = _row_by_key(self.conn, "t-reh", 1) ensure_task_attempts_table(self.conn) self.assertEqual(_columns(self.conn), list(UNIFIED_COLUMNS)) after = _row_by_key(self.conn, "t-reh", 1) self.assertEqual(after, before) # Both writers still work after the migration. resume_record(self.conn, "t-reh", 2, "goal_breakpoint", "paused", 3, "r", "e") ledger_record(self.conn, "t-reh2", "success") agg = aggregate_attempts(read_attempt_rows(self.conn)) self.assertEqual(agg["t-reh"]["defects"], 1) self.assertEqual(agg["t-reh2"]["attempts"], 1) def test_resume_rows_never_count_in_aggregation(self): # Pure aggregation: a NULL-outcome row is a resume row. agg = aggregate_attempts([ {"task_id": "t-1", "attempt_no": 1, "outcome": None, "is_defect": 0, "defect_category": None, "corrects_attempt_no": None}, {"task_id": "t-1", "attempt_no": 2, "outcome": "success", "is_defect": 0, "defect_category": None, "corrects_attempt_no": None}, ]) self.assertEqual(agg["t-1"]["attempts"], 1) agg_only_resume = aggregate_attempts([ {"task_id": "t-2", "attempt_no": 1, "outcome": None, "is_defect": 0, "defect_category": None, "corrects_attempt_no": None}, ]) self.assertNotIn("t-2", agg_only_resume) if __name__ == "__main__": unittest.main()