host
37ba6876bf
ci / test (push) Successful in 55m32s
- test_column_notnull_probe_sees_gap3_outcome now branches on the post-ensure probe: when outcome is relaxed (PG; SQLite 3.53+) the resume write must land and read back NULL; on older SQLite the ALTER is swallowed and INSERT OR IGNORE drops the row (version-dependent boundary, not a migration failure). - attempt_ledger.py comments updated: ALTER COLUMN DROP NOT NULL works on Postgres and SQLite 3.53.0+, swallowed before that. - PG proof summary section 3.9/8 corrected to the same boundary.
579 lines
26 KiB
Python
579 lines
26 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. 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
|