- 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).
229 lines
9.9 KiB
Python
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
|