Files
codex-app-server/cloud/attempt_ledger.py
T
host 0db51f8145 audit: task-attempt ledger feeds real first-pass/defect/rework rates (gap 3)
- cloud/attempt_ledger.py: separate task_attempts table (not tasks
  columns) with per-attempt rows, MAX+1 attempt_no, append-only
  intent; interrupted rows kept but never counted; is_defect is
  explicit, never inferred from failure; first defect category
  wins; no-ledger tasks stay unknown (None), never backfilled.
  SQLite DDL is CREATE TABLE IF NOT EXISTS (idempotent); PG_DDL
  is the verbatim manual step for production Postgres, kept out
  of POSTGRES_SCHEMA on purpose (migrations coverage untouched).
- cloud/server.py GET /metrics-report: aggregates the ledger
  into compute_metrics records; missing ledger table degrades to
  unknown rates, not an error. Cost dual-column code untouched.
- cloud/metrics.py: docstring only (ledger is now the source);
  compute_metrics logic and all #238 semantics unchanged.
- Tests: 16 ledger tests (store/aggregate/metrics/PG adapter
  evidence) + 3 endpoint tests (real rates, old task beside
  ledger task, interrupted status non-terminal).
2026-11-03 02:49:25 +00:00

229 lines
9.9 KiB
Python

"""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.
- 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:
- 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).
- 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 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``.
- No lifecycle hook writes to this ledger yet. Hooking a writer
into the worker's retry/resume loops would risk exactly the
misclassification this module refuses to make (platform
auto-resume counted as an attempt, transient retry counted as a
defect). The writer -- which lifecycle event counts as an
attempt, and who judges ``is_defect`` -- is a separate decision,
listed as an open item; until it lands, the ledger fills only
through explicit ``record_attempt`` calls, and production rates
stay honestly unknown rather than auto-filled.
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"})
# 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.
SQLITE_DDL = (
"CREATE TABLE IF NOT EXISTS 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))"
)
# 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).
PG_DDL = SQLITE_DDL
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``
commits; a bare ensure inside a larger transaction does not)."""
conn.execute(SQLITE_DDL)
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) "
"VALUES(?, ?, ?, ?, ?, ?)",
(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 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."""
return conn.execute(
"SELECT task_id, attempt_no, outcome, is_defect, "
"defect_category 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 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).
"""
per_task = {}
ordered = sorted(
rows,
key=lambda r: (_field(r, "task_id"), _field(r, "attempt_no")))
for row in ordered:
task_id = _field(row, "task_id")
if _field(row, "outcome") == "interrupted":
continue
entry = per_task.setdefault(
task_id, {"attempts": 0, "defects": 0,
"defect_category": None})
entry["attempts"] += 1
if _field(row, "is_defect"):
entry["defects"] += 1
if entry["defect_category"] is None:
entry["defect_category"] = _field(row, "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