"""Task-attempt ledger for the step-5 metrics loop (audit gap 3). Choice made here, with the trade-off stated rather than hidden: - This module adds a SEPARATE ``task_attempts`` table; it does NOT add ``attempts``/``defects`` columns to ``tasks``. Columns would be simpler to query (one row per task, no join), but they collapse history into two counters: a task's per-attempt story (which attempt failed, with what defect category, in what order) would be unrecoverable, and widening the hot ``tasks`` table on production Postgres is a heavier migration than adding a new, empty table. The ledger costs one aggregate query at report time, and that query is exactly what ``aggregate_attempts`` exists to pin down. - The ledger is append-only in intent: ``record_attempt`` only ever inserts a new ``(task_id, attempt_no)`` row (``attempt_no`` is ``MAX + 1`` per task, never reused). There is no update path, so a recorded attempt cannot be silently rewritten into a first-pass success after the fact. Semantics (pinned by tests, not left to each caller): - One row = one real attempt at the task. ``outcome`` is one of ``success`` / ``failure`` / ``interrupted``. - ``interrupted`` rows are kept for audit but NEVER count toward ``attempts`` or ``defects``: a platform interruption that auto-resumes as a fresh run is a platform event, not an agent failure, and the step-5 rule that ``interrupted`` is not a terminal task status would be meaningless if the interruption still inflated the attempt count. A task whose ledger holds only interrupted rows has NO counted attempt, so it is omitted from the aggregate entirely -- the endpoint then still reports its rates as unknown (``None``), never as 0. - ``is_defect`` is set explicitly by the recorder, never inferred from ``outcome``: a failed attempt is not automatically a defect (a retry can fail for a transient reason without being a defect round), and a defect round is a judgement the recording caller makes. ``defect_category`` requires ``is_defect``; an interrupted row can never be a defect. Both contradictions raise ``ValueError`` instead of being stored as ambiguous data. - ``defect_category`` for a task (the single category that ``compute_metrics`` carries per task) is the category of its FIRST defect row in attempt order. Later defect rows still count toward ``defects``; only the category label is first-wins, so it is deterministic. Corrections (below) re-evaluate this label: the first defect row AFTER corrections is applied wins. - A defect judgement can be CORRECTED, never rewritten: ``record_correction`` appends a new row whose ``corrects_attempt_no`` points at the attempt it corrects; the old row stays untouched, and aggregation applies corrections in attempt order with last-wins (so a wrongly marked defect can be unmarked, and a defect missed at record time can be added after review). A correction row is never itself counted as an attempt, cannot change an attempt's outcome (only its defect judgement), and cannot target an interrupted row, another correction row, or a nonexistent attempt -- all raise ``ValueError`` and write nothing. Who may correct is a policy question answered in ``cloud/attempt_hook.py`` (explicit recorder only, maintainer bar); the worker hook never writes corrections and never marks a defect. - Old tasks have no ledger rows, and that is NOT backfilled: the aggregate simply has no entry for them, the endpoint leaves their ``attempts``/``defects`` as ``None``, and ``compute_metrics`` excludes them from rate denominators (they are neither counted as first-pass nor as defect-free). Fabricating a row for an old task (e.g. ``attempts=1, defects=0`` from "it completed") would be inventing history. 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`` 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 schema set, and this ledger must not silently widen that set). ``PG_DDL`` below is the exact statement to run manually on production Postgres and verify (``\\d task_attempts`` or the ``information_schema`` equivalent) before any Postgres-backed endpoint is expected to return real rates. The DDL uses only Postgres-valid types (``TEXT``/``INTEGER``/``DOUBLE PRECISION``), a composite primary key, and no ``AUTOINCREMENT``/identity column, so it needs no sequence handling; the parameterized ``INSERT`` uses ``?`` placeholders, which the ``cloud.db`` Postgres adapter translates to ``%s`` (see the adapter test in ``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``. The Postgres upgrade path for a legacy-shaped table was verified against a real Postgres 16.15 instance on 2026-11-06 (fresh ``PG_DDL``, three legacy shapes migrated in place, both writers, PK, idempotent ensure; see ``verification/codex-task-attempts-pg-proof-20261106.md``). One gap-3-only step relaxes a gap-3 table's ``outcome`` from NOT NULL with ``ALTER COLUMN outcome DROP NOT NULL`` (see ``ensure_task_attempts_table``); the statement works on Postgres and on SQLite 3.53.0+, and is swallowed by ``try/except`` on older SQLite, where the column stays NOT NULL. - 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`` / ``record_correction``. - Lifecycle hook (this track, see ``cloud/attempt_hook.py``): the worker writes exactly one row per COUNTED dispatch (see that module's policy for which dispatches count: human/first rounds and human breakpoint resumes count; platform auto-resume reruns, coordinator fan-out, and pre-execution gate failures do not). The hook is OFF by default (``attempt_ledger_hook=False``): with it off, this module's behaviour is exactly audit gap 3 -- the ledger fills only through explicit ``record_attempt`` / ``record_correction`` calls, and production rates stay honestly 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 (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 (``arch_conformance``: no direct driver connection outside ``cloud.db``) are untouched -- there is no ``sqlite3.connect`` and no ``psycopg*`` import here at all. """ import time TABLE = "task_attempts" OUTCOMES = frozenset({"success", "failure", "interrupted"}) # 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, " "is_defect INTEGER NOT NULL 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))" ) 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). # For a legacy-shaped table already on Postgres, add whichever # unified columns are missing (see UNIFIED_COLUMN_TYPES) and # relax a NOT NULL outcome if the legacy shape declared one # (see ensure_task_attempts_table); that Postgres upgrade path # was verified against a real Postgres 16.15 instance on # 2026-11-06 -- see verification/ # codex-task-attempts-pg-proof-20261106.md. PG_DDL = TASK_ATTEMPTS_DDL # 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" ) def _existing_columns(conn, table): """已有列集合（automations.py 同款口径：SQLite 走 PRAGMA， 其它后端异常时回空集、后续 ALTER 逐条 try/except 兜底）。""" try: return {row[1] for row in conn.execute( "PRAGMA table_info(%s)" % table).fetchall()} except Exception: return set() def _column_notnull(conn, table): """列名 -> 是否 NOT NULL（PRAGMA table_info 的 row[3]； SQLite 与 PG 适配层含义一致：1 = NOT NULL，0 = 可空； 探针失败时回空 dict，调用方按“无需放宽”处理）。""" try: return {row[1]: bool(row[3]) for row in conn.execute( "PRAGMA table_info(%s)" % table).fetchall()} except Exception: return {} def ensure_task_attempts_table(conn): """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). 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. A gap-3 table's ``outcome`` was created NOT NULL; ensure also relaxes it to nullable (``ALTER COLUMN outcome DROP NOT NULL`` -- works on Postgres and on SQLite 3.53.0+, swallowed on older SQLite), because the resume writer inserts outcome NULL -- without the relax every resume write on such a table fails. 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(TASK_ATTEMPTS_DDL) existing = _existing_columns(conn, "task_attempts") for col in UNIFIED_COLUMNS: if col not in existing: try: conn.execute( "ALTER TABLE task_attempts ADD COLUMN %s %s" % (col, UNIFIED_COLUMN_TYPES[col])) except Exception: pass # gap-3 形态的 outcome 是 NOT NULL，而统一 DDL 要求它可空 # （恢复写路不报 outcome，NOT NULL 会使它在 PG 上每次写 # 行都报 NotNullViolation）。只在探针报 NOT NULL 时放宽： # ALTER COLUMN outcome DROP NOT NULL 在 Postgres 上生效； # SQLite 3.53.0 起也支持该语法，在此之前不合法、被 # try/except 吞掉，gap-3 表仍保留 NOT NULL（恢复写路 # 会被 INSERT OR IGNORE 静默丢弃——版本相关边界，见 # test_column_notnull_probe_sees_gap3_outcome）。 if _column_notnull(conn, "task_attempts").get("outcome"): try: conn.execute( "ALTER TABLE task_attempts " "ALTER COLUMN outcome DROP NOT NULL") except Exception: pass def record_attempt(conn, task_id, outcome, is_defect=False, defect_category=None, created_at=None): """Append one attempt row for ``task_id``. Returns ``attempt_no``. Validates before writing: unknown outcome, a defect flag on an interrupted row, and a defect category without a defect flag all raise ``ValueError`` and write nothing. """ if not task_id: raise ValueError("task_id is required") if outcome not in OUTCOMES: raise ValueError("outcome must be one of %s, got %r" % (sorted(OUTCOMES), outcome)) if outcome == "interrupted" and is_defect: raise ValueError("an interrupted attempt is never a defect") if defect_category is not None and not is_defect: raise ValueError("defect_category requires is_defect=True") ensure_task_attempts_table(conn) row = conn.execute( "SELECT COALESCE(MAX(attempt_no), 0) FROM task_attempts " "WHERE task_id = ?", (task_id,), ).fetchone() attempt_no = int(row[0]) + 1 conn.execute( "INSERT INTO task_attempts(task_id, attempt_no, outcome, " "is_defect, defect_category, created_at, " "corrects_attempt_no) " "VALUES(?, ?, ?, ?, ?, ?, NULL)", (task_id, attempt_no, outcome, 1 if is_defect else 0, defect_category, time.time() if created_at is None else created_at), ) conn.commit() return attempt_no def record_correction(conn, task_id, attempt_no, is_defect, defect_category=None, created_at=None): """Append a CORRECTION row for an already-recorded attempt. The correction does not rewrite history: the target row stays exactly as recorded, and this new row (its own ``MAX + 1`` ``attempt_no``) carries the corrected defect judgement in ``is_defect`` / ``defect_category`` plus ``corrects_attempt_no = attempt_no``. Aggregation applies corrections in attempt order, last-wins, and never counts a correction row as an attempt. Only the defect judgement is correctable: the correction row copies the target's outcome, so it cannot turn a success into a failure or vice versa. Validates before writing: missing ``task_id``, a nonexistent target, a target that is itself a 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. 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: raise ValueError("task_id is required") if defect_category is not None and not is_defect: raise ValueError("defect_category requires is_defect=True") ensure_task_attempts_table(conn) target = conn.execute( "SELECT outcome, corrects_attempt_no FROM task_attempts " "WHERE task_id = ? AND attempt_no = ?", (task_id, attempt_no), ).fetchone() if target is None: raise ValueError( "no such attempt to correct: %s #%s" % (task_id, attempt_no)) if target["corrects_attempt_no"] is not None: raise ValueError( "cannot correct a correction row: %s #%s" % (task_id, attempt_no)) if target["outcome"] == "interrupted": raise ValueError( "an interrupted attempt has no defect judgement to " "correct: %s #%s" % (task_id, attempt_no)) row = conn.execute( "SELECT COALESCE(MAX(attempt_no), 0) FROM task_attempts " "WHERE task_id = ?", (task_id,), ).fetchone() correction_no = int(row[0]) + 1 conn.execute( "INSERT INTO task_attempts(task_id, attempt_no, outcome, " "is_defect, defect_category, created_at, " "corrects_attempt_no) " "VALUES(?, ?, ?, ?, ?, ?, ?)", (task_id, correction_no, target["outcome"], 1 if is_defect else 0, defect_category, time.time() if created_at is None else created_at, attempt_no), ) conn.commit() return correction_no def read_attempt_rows(conn): """All ledger rows in attempt order. Raises the backend's OperationalError when the table does not exist yet; the endpoint caller treats that as "no ledger", not as a failure. ``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``) and a missing/NULL ``outcome`` as a resume row, not a ledger attempt. """ return conn.execute( "SELECT * FROM task_attempts " "ORDER BY task_id, attempt_no").fetchall() def _field(row, name): # sqlite3.Row, dict, and the Postgres adapter's rows all support # key access; a plain tuple row is not accepted (callers pass # rows, not raw tuples, so a wrong shape fails loudly here). return row[name] def _field_optional(row, name): """Like ``_field`` but missing keys mean "absent", not an error. Used ONLY for ``corrects_attempt_no``: rows built by gap-3-era readers (plain dicts, or tables without the column) carry no such key, and they are plain attempt rows by definition -- failing loudly there would break every gap-3 aggregate test for no benefit.""" try: return row[name] except (KeyError, IndexError, TypeError): return None def aggregate_attempts(rows): """Pure aggregation: ledger rows -> ``{task_id: record}`` where ``record`` is the ``attempts``/``defects``/``defect_category`` dict that ``compute_metrics`` consumes. 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). 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 attempt in attempt order AFTER corrections. """ per_task = {} ordered = sorted( rows, key=lambda r: (_field(r, "task_id"), _field(r, "attempt_no"))) # Per task: attempt_no -> [is_defect, defect_category], in # attempt order (insertion order = attempt_no order, since the # input was sorted). Corrections mutate the target entry; they # never create an entry of their own. judged = {} for row in ordered: task_id = _field(row, "task_id") corrects = _field_optional(row, "corrects_attempt_no") if corrects is not None: entries = judged.get(task_id) if entries is not None and corrects in entries: entries[corrects] = [ bool(_field(row, "is_defect")), _field(row, "defect_category")] continue 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")] = [ bool(_field(row, "is_defect")), _field(row, "defect_category")] for task_id, entries in judged.items(): attempts = len(entries) if attempts == 0: continue defects = 0 defect_category = None for _no in sorted(entries): is_defect, category = entries[_no] if is_defect: defects += 1 if defect_category is None: defect_category = category per_task[task_id] = { "attempts": attempts, "defects": defects, "defect_category": defect_category} return per_task def apply_to_records(records, aggregates): """Return a new record list with ledger values merged in. Records for tasks with no aggregate entry keep whatever ``attempts``/``defects`` they already carried (the endpoint passes ``None`` there, i.e. old tasks stay unknown). Input records are not modified. """ merged = [] for record in records: out = dict(record) agg = aggregates.get(record.get("id")) if agg is not None: out["attempts"] = agg["attempts"] out["defects"] = agg["defects"] out["defect_category"] = agg["defect_category"] merged.append(out) return merged