host
6a70775ad9
User decision 2026-11-05: unified schema, rejecting the rehearsal runtime column-union workaround and a ledger rename. One definition (cloud.attempt_ledger.TASK_ATTEMPTS_DDL, 12 columns) shared by cloud.attempts and cloud.db.POSTGRES_SCHEMA; ensure builds the unified table fresh and migrates old resume, gap-3 ledger, and rehearsal shapes in place with rows intact. aggregate counts ledger rows only (outcome NULL = resume row, never counted). SQLite verified by 8 new tests; Postgres real-DB upgrade path explicitly NOT verified (see batch summary). Local commit only: no merge, no push, no deploy.
237 lines
9.7 KiB
Python
237 lines
9.7 KiB
Python
"""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,
|
|
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_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()
|