codex-app-server/cloud/attempt_ledger.py

"""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 (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`` /
  ``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);
# that Postgres upgrade path is NOT yet verified against a real
# Postgres instance -- see the batch summary.
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 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. 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


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