From 6a70775ad98edb7f83884e91c5ff8fe2251ed186 Mon Sep 17 00:00:00 2001 From: host Date: Thu, 5 Nov 2026 02:22:47 +0000 Subject: [PATCH] FLAG-1 unified task_attempts schema (single canonical DDL + 3-shape migration) 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. --- cloud/attempt_ledger.py | 245 ++++++++++++------ cloud/attempts.py | 32 ++- cloud/db.py | 23 +- cloud/ledger_onboarding.py | 29 ++- .../test_task_attempts_unified_schema.py | 236 +++++++++++++++++ cloud/tests/test_v2cov_gaps_a.py | 15 +- 6 files changed, 467 insertions(+), 113 deletions(-) create mode 100644 cloud/tests/test_task_attempts_unified_schema.py diff --git a/cloud/attempt_ledger.py b/cloud/attempt_ledger.py index 5df2b94..330026e 100644 --- a/cloud/attempt_ledger.py +++ b/cloud/attempt_ledger.py @@ -67,10 +67,50 @@ Semantics (pinned by tests, not left to each caller): Persistence notes: +- One table, one definition. ``task_attempts`` is defined ONCE + below (``TASK_ATTEMPTS_DDL``, aliased as ``SQLITE_DDL`` / + ``PG_DDL``); ``cloud.attempts`` (the resume-history writer) and + ``cloud.db.POSTGRES_SCHEMA`` (the fresh-Postgres schema) both + import this same string instead of carrying their own copy. + The two historical shapes -- main's resume DDL (``kind`` / + ``interrupted_reason`` / ``checkpoint_round`` / ``result_text`` + / ``error_text``) and gap 3's ledger DDL (``outcome`` / + ``is_defect`` / ``defect_category`` / ``corrects_attempt_no``) + -- are superseded by this unified definition, which carries + both column groups. The shared columns (``task_id``, + ``attempt_no``, ``created_at``) had identical meaning and type + on both sides, so there is no semantic conflict to arbitrate: + the union of the column sets IS the unified schema, with one + deliberate relaxation (see below). +- Row kinds in one table. A row is a ledger row iff ``outcome`` + is non-NULL (written by this module, outcome validated in + Python before the insert); a row with ``outcome`` NULL is a + resume-history row (written by ``cloud.attempts``). The two + writers insert disjoint column groups and read by column name, + so they share the table without overwriting each other's + fields. ``aggregate_attempts`` counts ONLY ledger rows: a + resume row (outcome NULL) is skipped, never counted as a + defect-free first pass -- otherwise a task that was merely + auto-resumed would silently gain a counted attempt. +- The one deliberate relaxation: ``outcome`` is nullable in the + unified DDL even though the ledger writer always supplies it. + A ``NOT NULL`` on ``outcome`` would make the resume writer's + insert (which has no outcome to report) fail on a fresh table, + so the constraint lives in Python validation here, not in the + DDL. ``is_defect`` keeps ``NOT NULL DEFAULT 0`` so a resume + row that omits it still reads as "not a defect judgement". - SQLite: ``ensure_task_attempts_table`` runs - ``CREATE TABLE IF NOT EXISTS`` -- idempotent, safe to call on - every write, and a no-op on a database that already has the - table (including one that already has rows). + ``CREATE TABLE IF NOT EXISTS`` with the unified DDL, then + migrates a pre-existing legacy table (old resume shape, gap-3 + ledger shape, or the rehearsal-era 12-column shape) by adding + whichever unified columns it lacks. Added columns are always + nullable or defaulted, so existing rows are never blocked; + SQLite ``ALTER TABLE`` cannot tighten an existing column's + nullability, so a migrated table keeps its looser column + types -- the column SET and the data are unified, the + ``NOT NULL`` spelling on fresh tables is not retrofitted. + This is a migration to the single canonical definition, not a + second definition assembled at runtime. - Postgres: the production table is NOT created by ``init_postgres_schema`` (deliberately: that schema is covered by the migrations test asserting ``MIGRATION_TABLES`` equals the @@ -87,7 +127,10 @@ Persistence notes: ``test_attempt_ledger.py``). Until that manual step runs, the endpoint treats a missing table exactly like "no ledger rows": unknown rates, not an error -- the same fail-soft shape already - used for missing ``cost_records``. + used for missing ``cost_records``. The Postgres upgrade path + for a legacy-shaped table (fresh ``PG_DDL`` vs. adding the + missing unified columns) has NOT been verified against a real + Postgres instance in this batch -- see the batch summary. - The read path (``GET /metrics-report``) never creates the table: a read-only endpoint must not mutate schema. Table creation happens only on the write path, inside ``record_attempt`` / @@ -104,18 +147,15 @@ Persistence notes: unknown until the hook is deliberately enabled. The worker hook always records ``is_defect=False``: it never judges a defect, so enabling it can never turn a transient retry into a defect. -- Schema evolution note (pre-production, this track): - ``corrects_attempt_no`` was added to the DDL after audit gap 3 - but BEFORE either DDL reached production (gap 3 is still an - OPEN, unmerged PR), so no production table needs migration: - fresh tables get the column from the DDL below. A database - created from gap 3's DDL alone (temporary/test databases only) - gains the column with one statement -- - ``ALTER TABLE task_attempts ADD COLUMN corrects_attempt_no - INTEGER`` -- which is also the documented step if gap 3's DDL - was ever run on production Postgres before this track lands. - The read path uses ``SELECT *`` deliberately so a table without - the column still reads (its rows aggregate as plain attempts). +- Schema evolution note (unified schema, this batch): + the old resume DDL, the gap-3 ledger DDL, and the rehearsal + 12-column shape are all legacy shapes of the SAME table. + ``ensure_task_attempts_table`` migrates any of them to the + unified column set in place (add missing columns only, never + drop or rewrite rows); the read path uses ``SELECT *`` + deliberately so rows of any shape still read, and + ``aggregate_attempts`` treats a row without an outcome as a + resume row, not as a ledger attempt (see above). This module imports no ``cloud.*`` module and opens no database connection itself: callers pass a connection in, so the layer rules @@ -130,38 +170,92 @@ TABLE = "task_attempts" OUTCOMES = frozenset({"success", "failure", "interrupted"}) -# One DDL serves both backends: every type and the composite key are -# valid in SQLite and in Postgres, and there is no AUTOINCREMENT or -# identity column for either backend's sequence handling to chase. -# ``corrects_attempt_no`` (NULL = a real attempt row; non-NULL = a -# correction row for that attempt_no, never counted as an attempt) -# was added by the lifecycle-hook track; it SUPERSEDES gap 3's DDL -# for any production manual step (gap 3 never merged, so no -# production table exists yet -- see the module docstring). -SQLITE_DDL = ( +# THE single canonical definition of task_attempts (unified +# schema, FLAG-1 decision 2026-11-05: "unified schema", rejecting +# both the rehearsal-era runtime column-union workaround and a +# ledger rename). One DDL serves both backends: every type and +# the composite key are valid in SQLite and in Postgres, and +# there is no AUTOINCREMENT or identity column for either +# backend's sequence handling to chase. +# +# Column groups, in order (each group's internal order preserves +# its historical DDL order): +# - shared: task_id, attempt_no (composite PK), created_at last; +# - ledger (gap 3 + lifecycle hook): outcome, is_defect, +# defect_category, corrects_attempt_no (NULL = a real attempt +# row; non-NULL = a correction row for that attempt_no, never +# counted as an attempt); +# - resume (main v6): kind, interrupted_reason, +# checkpoint_round, result_text, error_text. +# ``outcome`` is deliberately nullable (see the module docstring: +# the resume writer has no outcome to report, so a NOT NULL here +# would break its insert; the ledger writer validates outcome in +# Python instead). ``cloud.attempts.CREATE_TABLE_SQL`` and the +# ``task_attempts`` entry of ``cloud.db.POSTGRES_SCHEMA`` are +# this same string object, imported -- never a second copy. +TASK_ATTEMPTS_DDL = ( "CREATE TABLE IF NOT EXISTS task_attempts(" "task_id TEXT NOT NULL, " "attempt_no INTEGER NOT NULL, " - "outcome TEXT NOT NULL, " + "outcome TEXT, " "is_defect INTEGER NOT NULL DEFAULT 0, " "defect_category TEXT, " - "created_at DOUBLE PRECISION, " "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))" ) +SQLITE_DDL = TASK_ATTEMPTS_DDL + # Production Postgres: run this verbatim by hand and verify the # table exists before expecting real rates from a Postgres-backed # endpoint. It is intentionally NOT part of POSTGRES_SCHEMA (see # the module docstring: migrations coverage must not widen silently). -# This is the NEW DDL (with ``corrects_attempt_no``): if gap 3's -# older DDL was ever run first, follow it with -# ``ALTER TABLE task_attempts ADD COLUMN corrects_attempt_no -# INTEGER`` and verify before expecting corrections to work. -PG_DDL = SQLITE_DDL +# For a legacy-shaped table already on Postgres, add whichever +# unified columns are missing (see UNIFIED_COLUMN_TYPES); +# that Postgres upgrade path is NOT yet verified against a real +# Postgres instance -- see the batch summary. +PG_DDL = TASK_ATTEMPTS_DDL -# One-line upgrade for a table created from gap 3's DDL (no -# ``corrects_attempt_no`` column yet). Valid in both backends. +# Unified column set, in DDL order. The migration in +# ``ensure_task_attempts_table`` adds exactly these columns when +# missing; the ALTER types below are the migration spellings +# (nullable/defaulted, so existing rows are never blocked -- +# SQLite ALTER cannot add a NOT NULL column without a default +# to a table that already has rows, and cannot tighten an +# existing column at all). +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", +) + +UNIFIED_COLUMN_TYPES = { + "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", +} + +# Legacy one-line upgrade for a table created from gap 3's DDL +# (no ``corrects_attempt_no`` column yet). Retained for +# backward compatibility with existing callers/tests; it is a +# single step of the general migration above, not a second +# schema definition. Valid in both backends. OLD_TABLE_ALTER = ( "ALTER TABLE task_attempts " "ADD COLUMN corrects_attempt_no INTEGER" @@ -179,39 +273,34 @@ def _existing_columns(conn, table): def ensure_task_attempts_table(conn): - """Create the ledger table if absent. Idempotent; does not - touch existing rows. The caller owns the commit (``record_attempt`` + """Create the unified table if absent; migrate a legacy + table to it if present. Idempotent; does not touch existing + rows. The caller owns the commit (``record_attempt`` commits; a bare ensure inside a larger transaction does not). - 演练注 FLAG（合入语义取舍点，主结论见演练报告）： - main 侧 init_db/worker 用 cloud.attempts.py 的旧 DDL 建同名 - task_attempts 表（kind/interrupted_reason/checkpoint_round/ - result_text/error_text，无 outcome 系列）；本模块的 CREATE - TABLE IF NOT EXISTS 在旧表已存在时是空操作，之后按台账 - 列写行即报 no column named outcome。本解法把表升级为两 - 组列的并集：旧表先建时补台账 4 列，台账表先建时补旧 5 - 列（新增列一律可空/带默认值，绝不对已有行加 NOT NULL， - 否则旧行会挡住 ALTER）。两边读路径均按列名取数，互不 - 干扰；但这只是演练期保守解法——真合前须用户定：并集表、 - 台账改名、或统一 schema 三选一，不可悄悄定为并集表。 + Unified schema (FLAG-1, user decision 2026-11-05): there is + exactly one ``task_attempts`` definition + (``TASK_ATTEMPTS_DDL``). A fresh database gets it verbatim + from ``CREATE TABLE IF NOT EXISTS``. A database whose table + was built from any legacy shape -- the old resume DDL, the + gap-3 ledger DDL, or the rehearsal-era 12-column shape -- + is migrated in place by adding whichever unified columns it + lacks (``UNIFIED_COLUMN_TYPES``); added columns are always + nullable or defaulted, so existing rows are never blocked, + and no row is dropped or rewritten. This replaces the + rehearsal-era runtime column-completion workaround: the + target here is the single canonical definition, and + ``cloud.attempts`` / ``cloud.db`` share that same string + instead of each carrying a DDL of their own. """ - conn.execute(SQLITE_DDL) + conn.execute(TASK_ATTEMPTS_DDL) existing = _existing_columns(conn, "task_attempts") - for col, ddl in ( - ("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")): + for col in UNIFIED_COLUMNS: if col not in existing: try: conn.execute( "ALTER TABLE task_attempts ADD COLUMN %s %s" - % (col, ddl)) + % (col, UNIFIED_COLUMN_TYPES[col])) except Exception: pass @@ -272,11 +361,11 @@ def record_correction(conn, task_id, attempt_no, is_defect, correction row, a target whose outcome is ``interrupted`` (interrupted rows are never defects, so there is no defect judgement to correct), and a defect category without a defect - flag all raise ``ValueError`` and write nothing. On a table - created from gap 3's DDL (no ``corrects_attempt_no`` column), - the INSERT raises the backend's OperationalError loudly -- - run ``OLD_TABLE_ALTER`` first (see the module docstring); - a correction must never silently degrade into a plain row. + flag all raise ``ValueError`` and write nothing. A table + created from a legacy DDL is migrated by ``ensure`` first + (above), so the ``corrects_attempt_no`` column is present + before the INSERT runs; a correction must never silently + degrade into a plain row. Returns the correction row's own ``attempt_no``. """ if not task_id: @@ -325,12 +414,13 @@ def read_attempt_rows(conn): OperationalError when the table does not exist yet; the endpoint caller treats that as "no ledger", not as a failure. - ``SELECT *`` is deliberate: a table created from gap 3's DDL - has no ``corrects_attempt_no`` column, and naming that column - here would turn those readable rows into an endpoint error. + ``SELECT *`` is deliberate: a legacy-shaped table may lack + columns the unified DDL has, and naming a missing column + here would turn readable rows into an endpoint error. With ``*`` the rows still read, and ``aggregate_attempts`` treats a missing ``corrects_attempt_no`` as a plain attempt - row (see ``_field_optional``). + row (see ``_field_optional``) and a missing/NULL ``outcome`` + as a resume row, not a ledger attempt. """ return conn.execute( "SELECT * FROM task_attempts " @@ -364,10 +454,14 @@ def aggregate_attempts(rows): Interrupted rows are excluded from both counts, and a task with only interrupted rows is omitted entirely (no counted - attempt => its rates stay unknown, never 0). Correction rows - (``corrects_attempt_no`` set) are excluded from the attempt - count and instead OVERWRITE the defect judgement of their - target attempt, in attempt order, last-wins: a task's + attempt => its rates stay unknown, never 0). Resume rows + (``outcome`` NULL -- written by ``cloud.attempts``, which + records an interrupted attempt's history, not a ledger + outcome) are likewise excluded: they are not ledger attempts + and must never inflate a task's counted attempts. Correction + rows (``corrects_attempt_no`` set) are excluded from the + attempt count and instead OVERWRITE the defect judgement of + their target attempt, in attempt order, last-wins: a task's ``defects`` is the number of its counted attempts whose judgement AFTER corrections is a defect, and its ``defect_category`` is the category of its FIRST defect @@ -392,7 +486,12 @@ def aggregate_attempts(rows): bool(_field(row, "is_defect")), _field(row, "defect_category")] continue - if _field(row, "outcome") == "interrupted": + outcome = _field_optional(row, "outcome") + if outcome is None: + # Resume-history row (cloud.attempts): no ledger + # outcome, so not a counted ledger attempt. + continue + if outcome == "interrupted": continue entries = judged.setdefault(task_id, {}) entries[_field(row, "attempt_no")] = [ diff --git a/cloud/attempts.py b/cloud/attempts.py index 850fb03..6826184 100644 --- a/cloud/attempts.py +++ b/cloud/attempts.py @@ -31,7 +31,13 @@ host.py、不改 codex 线程恢复： result_text / error_text 各截断 2000 字。记账与回 queued 在同一事务里，记账失败不单独炸恢复循环（外层已兜底）， 但同一 (task_id, attempt_no) 重复记时 INSERT OR IGNORE - 保留第一次，不覆盖。 + 保留第一次，不覆盖。表定义是统一 schema（FLAG-1， + 2026-11-05）：本模块不自带一份 DDL，CREATE_TABLE_SQL + 与建表/迁移均取自 ``cloud.attempt_ledger`` 的 + ``TASK_ATTEMPTS_DDL`` / ``ensure_task_attempts_table``， + 与台账写路径共用同一定义；本模块写的行 outcome 为 + NULL（恢复历史行，非台账计数行，见 attempt_ledger 对 + aggregate 的口径）。 ② 重跑前言：非 goal 平台全新重排的任务在认领时，worker 在内存副本的 prompt 前贴一段上次尝试说明（原 prompt 不落库，与项目/待办/记忆注入同款只改内存副本），内容 @@ -63,25 +69,23 @@ host.py、不改 codex 线程恢复： import time +from cloud.attempt_ledger import TASK_ATTEMPTS_DDL +from cloud.attempt_ledger import ensure_task_attempts_table as _ensure_unified + ATTEMPT_TEXT_CHARS = 2000 PREAMBLE_TEXT_CHARS = 1000 -CREATE_TABLE_SQL = ( - "CREATE TABLE IF NOT EXISTS 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))") +# 统一 schema：与台账共用同一定义，绝不在此另写一份 DDL +# （历史教训：两份 DDL 同名异构，CREATE TABLE IF NOT EXISTS +# 使后建的一方成空操作、写行报缺列）。别名保留，供旧调用 +# 方按名取定义；建表与老表迁移一律走台账的 ensure。 +CREATE_TABLE_SQL = TASK_ATTEMPTS_DDL def ensure_attempts_table(conn): - """幂等建尝试账本表（server/worker 两支共用同一 DDL）。""" - conn.execute(CREATE_TABLE_SQL) + """幂等建尝试账本表（server/worker 两支共用同一 DDL）： + 新库建统一表，老库迁移到统一列集（见 attempt_ledger）。""" + _ensure_unified(conn) def _trunc(text, limit): diff --git a/cloud/db.py b/cloud/db.py index 91e4b16..aa6a359 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -65,6 +65,12 @@ import time from abc import ABC, abstractmethod from contextlib import contextmanager +# 统一 schema：task_attempts 的 PG 新库定义与台账定义是同一 +# 字符串（cloud.attempt_ledger.TASK_ATTEMPTS_DDL），不在此另 +# 写一份。attempt_ledger 只依赖标准库 time，与本模块无循环 +# 导入；层规则上它也不属于 orchestration 模块集合。 +from cloud.attempt_ledger import TASK_ATTEMPTS_DDL as _TASK_ATTEMPTS_DDL + # 与 server.py 旧 _connect 的 timeout=30s 保持一致的连接级等待； # PRAGMA busy_timeout 是语句级等待，两者不冲突，后者以毫秒为单位 # 在每个连接上单独设置（不随库文件持久化，见 connect 的注释）。 @@ -876,18 +882,11 @@ POSTGRES_SCHEMA = ( "created_at DOUBLE PRECISION)", "CREATE INDEX IF NOT EXISTS idx_task_steps_task" " ON task_steps(task_id, step_no)", - # 尝试账本（v6，见 cloud/attempts.py）：(task_id, attempt_no) - # 唯一；PG 新库一次建齐，老库走 server/worker ensure。 - "CREATE TABLE IF NOT EXISTS 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))", + # 尝试账本（统一 schema，见 cloud/attempt_ledger.py）： + # (task_id, attempt_no) 唯一；PG 新库一次建齐，老库走 + # server/worker ensure 迁移到同一列集。此处是导入的统 + # 一定义本身，不是第二份 DDL。 + _TASK_ATTEMPTS_DDL, # 审计循环状态（audit loop）：task_id 单行主键；PG 新库一次 # 建齐，老库走 server/worker ensure（见 cloud/audit_loop.py）。 "CREATE TABLE IF NOT EXISTS task_audit_loop_state(" diff --git a/cloud/ledger_onboarding.py b/cloud/ledger_onboarding.py index c80ada6..dd36b47 100644 --- a/cloud/ledger_onboarding.py +++ b/cloud/ledger_onboarding.py @@ -21,9 +21,10 @@ 报告：表齐但钩子关 ≠ 就绪，钩子开但表不齐 ≠ 就绪。 3. DDL 清单不另造：清单直接引用 attempt_ledger.PG_DDL 与 attempt_hook.PG_MARKER_DDL 两个常量（本模块不 - 复制 DDL 文本，复制就会漂移）。另含 gap 3 旧表升级句 - attempt_ledger.OLD_TABLE_ALTER（仅当旧 DDL 建过表时 - 需要；gap 3 未合并，生产尚无旧表，属防御项）。 + 复制 DDL 文本，复制就会漂移）。老形态表升级走 + attempt_ledger.ensure_task_attempts_table 的统一列补齐 + （仅当老形态建过表时需要；属防御项，PG 真实库升级 + 路径未实证）。 4. 校验函数 verify_ledger_ready(conn)：只读探针——查两 张表的存在与必需列，不建表、不写行（读路径绝不改 schema，与 metrics-report 端点的只读纪律一致）。 @@ -42,11 +43,14 @@ from cloud.attempt_hook import PG_MARKER_DDL as _PG_MARKER_DDL from cloud.attempt_ledger import OLD_TABLE_ALTER as _OLD_TABLE_ALTER from cloud.attempt_ledger import PG_DDL as _LEDGER_PG_DDL -# 必需列：缺任一列即视为表不齐（旧 DDL 建的表缺 -# corrects_attempt_no 时走 OLD_TABLE_ALTER 补列）。 +# 必需列：统一 schema 的全部 12 列；缺任一列即视为表 +# 不齐（老形态建的表由 ensure_task_attempts_table 补齐 +# 缺失列后复核，不再只补 corrects_attempt_no 一列）。 LEDGER_REQUIRED_COLUMNS = ( "task_id", "attempt_no", "outcome", "is_defect", - "defect_category", "created_at", "corrects_attempt_no", + "defect_category", "corrects_attempt_no", "kind", + "interrupted_reason", "checkpoint_round", "result_text", + "error_text", "created_at", ) MARKER_REQUIRED_COLUMNS = ("task_id", "origin", "queued_at") @@ -72,12 +76,13 @@ LEDGER_DDL_CHECKLIST = ( }, { "step": 3, - "action": "仅当表由 gap 3 旧 DDL 建过时：执行" - " attempt_ledger.OLD_TABLE_ALTER 补" - " corrects_attempt_no 列（gap 3 未合并，" - "生产正常路径跳过此步）", - "ddl_source": "cloud.attempt_ledger.OLD_TABLE_ALTER", - "verify": "task_attempts 含 corrects_attempt_no 列", + "action": "仅当表已由老形态（旧恢复 DDL、gap 3 台账 DDL" + " 或演练形态）建过时：经 attempt_ledger." + "ensure_task_attempts_table 补齐缺失的统一列" + "（SQLite 已实测；PG 真实库升级路径未实证，" + "生产 PG 上执行前须先在真实 PG 上验证）", + "ddl_source": "cloud.attempt_ledger.TASK_ATTEMPTS_DDL", + "verify": "task_attempts 含统一 schema 全部 12 列", }, { "step": 4, diff --git a/cloud/tests/test_task_attempts_unified_schema.py b/cloud/tests/test_task_attempts_unified_schema.py new file mode 100644 index 0000000..bc664cc --- /dev/null +++ b/cloud/tests/test_task_attempts_unified_schema.py @@ -0,0 +1,236 @@ +"""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() diff --git a/cloud/tests/test_v2cov_gaps_a.py b/cloud/tests/test_v2cov_gaps_a.py index d6316d2..a3568ab 100644 --- a/cloud/tests/test_v2cov_gaps_a.py +++ b/cloud/tests/test_v2cov_gaps_a.py @@ -223,9 +223,14 @@ class LedgerOnboardingGapTestCase(unittest.TestCase): finally: conn.close() self.assertFalse(out["tables_ready"]) + # Unified schema: the required set is all 12 columns, so a + # gap-3-shaped table is missing corrects_attempt_no AND + # the five resume columns, not just the one column. self.assertEqual( out["missing"], - ["task_attempts missing columns: corrects_attempt_no"]) + ["task_attempts missing columns: corrects_attempt_no, " + "kind, interrupted_reason, checkpoint_round, " + "result_text, error_text"]) def test_verify_marker_missing_column(self): conn = self._sqlite( @@ -237,9 +242,15 @@ class LedgerOnboardingGapTestCase(unittest.TestCase): finally: conn.close() self.assertFalse(out["tables_ready"]) + # Unified schema: the gap-3-shaped ledger table is also + # missing the five resume columns, reported before the + # marker table's own missing column. self.assertEqual( out["missing"], - ["task_attempt_origin_markers missing columns: queued_at"]) + ["task_attempts missing columns: kind, " + "interrupted_reason, checkpoint_round, result_text, " + "error_text", + "task_attempt_origin_markers missing columns: queued_at"]) class LedgerOutboxGapTestCase(unittest.TestCase):