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).

cloud/attempt_ledger.py

@@ -0,0 +1,228 @@
+"""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

cloud/metrics.py

@@ -4,8 +4,8 @@ Zero third-party dependencies; standard library only. This module
 imports no ``cloud.*`` module: it computes over plain record dicts so
 the collection logic is testable without a database, and so the
 report shape is identical whether the records came from the control
-plane (``server._metrics_report``), a test fixture, or a future
-attempt ledger.
+plane (``server._metrics_report`` reading the attempt ledger), a
+test fixture, or a hand-built record list.
 
 Definitions (pinned here, not left to each caller):
 
@@ -17,12 +17,13 @@ Definitions (pinned here, not left to each caller):
 - First-pass green: terminal task whose ``attempts`` == 1 and whose
   status is ``completed`` / ``merged``.
 - Defect: a task with ``defects`` > 0, where ``defects`` is the count
-  of failed CI/audit rounds or follow-up rounds for that task. The
-  current tasks schema has NO attempts/defects column, so a caller
-  that cannot supply them must pass ``None`` -- and the rates then
-  come back as ``None`` with ``*_known`` False. An unknown rate is
-  never reported as 0: 0 would claim "no defects", which is a
-  decision-grade lie.
+  of failed CI/audit rounds or follow-up rounds for that task. Since
+  audit gap 3 these numbers come from the ``task_attempts`` ledger
+  (``cloud/attempt_ledger.py``: one row per real attempt, aggregated
+  by the endpoint); a task with no ledger rows must still pass
+  ``None`` -- and the rates then come back as ``None`` with
+  ``*_known`` False. An unknown rate is never reported as 0: 0
+  would claim "no defects", which is a decision-grade lie.
 - Rework: terminal task with ``attempts`` > 1. Same None discipline.
 - Per-task cost reuses the #212 dual-column discipline from
   ``cloud.cost``: priced USD micros and unpriced actual tokens are

cloud/server.py

@@ -169,6 +169,9 @@ from cloud.approvals import ensure_approvals_table as _appr_ensure
 from cloud.approvals import expire_overdue as _appr_expire
 from cloud.approvals import get_approval as _appr_get
 from cloud.approvals import list_approvals as _appr_list
+from cloud.attempt_ledger import aggregate_attempts as _aggregate_attempts
+from cloud.attempt_ledger import apply_to_records as _apply_attempts
+from cloud.attempt_ledger import read_attempt_rows as _read_attempt_rows
 from cloud.auto_memory import AutoMemoryStore
 from cloud.auto_memory import process_due_tasks as _auto_process_due
 from cloud.identity import IdentityStore
@@ -10129,12 +10132,14 @@ pid 须为整数、address 须为字符串，否则 400（注册表里的
         """指标闭环报告：首过绿灯率 / 缺陷率 / 返工率 / 每任务成本。
 
         权限与 /metrics 同款：只统计调用者是成员的项目。
-        诚实边界：当前 tasks 表没有 attempts/defects 列，所以
-        attempts/defects 一律传 None，首过/缺陷/返工率回 None
-        （未知，不报 0）；金任务集未经人工审定，报告恒为
-        pending_human_approval、decision_ready False。等任务尝试
-        台账落库、金任务人工审定后，这两处再接真实输入——
-        在此之前绝不把样本集或未知率包装成可决策数字。
+        台账（audit gap 3，见 ``cloud/attempt_ledger.py``）：任务的
+        attempts/defects 来自 ``task_attempts`` 台账的聚合，不再
+        恒为 None；台账里只有中断行的任务、以及台账里根本没有
+        行的老任务，仍传 None——未知率回 None（未知，不报 0），
+        老任务绝不被算成首过或无缺陷。台账表尚不存在（老库、
+        或生产 PG 未手动执行 ``PG_DDL``）时与无台账行同等处理，
+        不因此让端点报错；金任务集未经人工审定，报告恒为
+        pending_human_approval、decision_ready False。
         成本走 #212 双列口径：有价合计与无价 actual token 分开，
         总额在成本不完整时为 None，不把无价折成 0。
         """
@@ -10153,6 +10158,14 @@ pid 须为整数、address 须为字符串，否则 400（注册表里的
                         "attempts": None, "defects": None}
                        for row in rows
                        if row["project_id"] in member_ids]
+            try:
+                aggregates = _aggregate_attempts(
+                    _read_attempt_rows(conn))
+            except (sqlite3.OperationalError, _PgOperationalError):
+                # No ledger table yet (old DB, or production PG
+                # before the manual PG_DDL step): same as no rows.
+                aggregates = {}
+            records = _apply_attempts(records, aggregates)
             try:
                 cost_rows = conn.execute(
                     "SELECT task_id, project_id, source, total_tokens,"

cloud/tests/test_attempt_ledger.py

@@ -0,0 +1,245 @@
+"""Tests for the task-attempt ledger (audit gap 3).
+
+Only the standard library; a temporary SQLite DB for the store
+tests, plain dicts for the pure aggregation. No real model, no
+network, no secrets, no real Postgres (the Postgres claim is
+checked at the adapter level only, and labelled as such).
+
+Pinned boundaries:
+- interrupted rows never inflate attempts/defects, and a task with
+  only interrupted rows stays unknown (omitted, not 0);
+- old tasks with no ledger rows stay unknown (None + *_known False),
+  never counted as first-pass or defect-free;
+- interrupted task STATUS is never terminal, even with ledger rows;
+- the dual-column cost discipline is unchanged by the ledger.
+"""
+
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(
+    os.path.dirname(os.path.abspath(__file__)))))
+
+from cloud.attempt_ledger import (  # noqa: E402
+    PG_DDL,
+    aggregate_attempts,
+    apply_to_records,
+    ensure_task_attempts_table,
+    read_attempt_rows,
+    record_attempt,
+)
+from cloud.db import _convert_placeholders, _translate_sql  # noqa: E402
+from cloud.metrics import compute_metrics  # noqa: E402
+
+
+def _sqlite_conn(path):
+    from cloud.db import connect as db_connect
+    return db_connect(path)
+
+
+class TestLedgerStore(unittest.TestCase):
+    def setUp(self):
+        import tempfile
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.conn = _sqlite_conn(self.db_path)
+
+    def tearDown(self):
+        self.conn.close()
+        self.tmp.cleanup()
+
+    def _count(self):
+        # Validation in record_attempt runs before ensure, so a
+        # rejected write may leave the table uncreated: that is
+        # still "writes nothing", counted as 0 here.
+        try:
+            return self.conn.execute(
+                "SELECT COUNT(*) FROM task_attempts").fetchone()[0]
+        except Exception:  # noqa: BLE001 -- missing table == 0 rows
+            return 0
+
+    def test_ensure_is_idempotent_and_preserves_rows(self):
+        ensure_task_attempts_table(self.conn)
+        record_attempt(self.conn, "t-1", "failure", is_defect=True,
+                       defect_category="contract")
+        ensure_task_attempts_table(self.conn)
+        ensure_task_attempts_table(self.conn)
+        self.assertEqual(self._count(), 1)
+
+    def test_attempt_numbers_are_per_task_and_sequential(self):
+        self.assertEqual(
+            record_attempt(self.conn, "t-1", "failure", is_defect=True,
+                           defect_category="contract"), 1)
+        self.assertEqual(
+            record_attempt(self.conn, "t-1", "success"), 2)
+        self.assertEqual(
+            record_attempt(self.conn, "t-2", "success"), 1)
+        rows = {(r["task_id"], r["attempt_no"])
+                for r in read_attempt_rows(self.conn)}
+        self.assertEqual(rows, {("t-1", 1), ("t-1", 2), ("t-2", 1)})
+
+    def test_unknown_outcome_rejected_and_writes_nothing(self):
+        with self.assertRaises(ValueError):
+            record_attempt(self.conn, "t-1", "maybe")
+        # The table was ensured but no row was written.
+        self.assertEqual(self._count(), 0)
+
+    def test_interrupted_row_can_never_be_a_defect(self):
+        with self.assertRaises(ValueError):
+            record_attempt(self.conn, "t-1", "interrupted",
+                           is_defect=True)
+        self.assertEqual(self._count(), 0)
+
+    def test_defect_category_without_defect_flag_rejected(self):
+        with self.assertRaises(ValueError):
+            record_attempt(self.conn, "t-1", "failure",
+                           defect_category="contract")
+        self.assertEqual(self._count(), 0)
+
+
+class TestAggregate(unittest.TestCase):
+    def _row(self, task_id, attempt_no, outcome, is_defect=0,
+             category=None):
+        return {"task_id": task_id, "attempt_no": attempt_no,
+                "outcome": outcome, "is_defect": is_defect,
+                "defect_category": category}
+
+    def test_interrupted_rows_excluded_from_both_counts(self):
+        agg = aggregate_attempts([
+            self._row("t-1", 1, "interrupted"),
+            self._row("t-1", 2, "failure", 1, "contract"),
+            self._row("t-1", 3, "success"),
+        ])
+        self.assertEqual(agg["t-1"]["attempts"], 2)
+        self.assertEqual(agg["t-1"]["defects"], 1)
+        self.assertEqual(agg["t-1"]["defect_category"], "contract")
+
+    def test_only_interrupted_task_is_omitted_not_zero(self):
+        agg = aggregate_attempts([
+            self._row("t-1", 1, "interrupted"),
+        ])
+        self.assertNotIn("t-1", agg)
+
+    def test_first_defect_category_wins_regardless_of_row_order(self):
+        agg = aggregate_attempts([
+            self._row("t-1", 2, "failure", 1, "directory"),
+            self._row("t-1", 1, "failure", 1, "contract"),
+        ])
+        self.assertEqual(agg["t-1"]["defects"], 2)
+        self.assertEqual(agg["t-1"]["defect_category"], "contract")
+
+    def test_failed_attempt_is_not_automatically_a_defect(self):
+        agg = aggregate_attempts([
+            self._row("t-1", 1, "failure"),
+            self._row("t-1", 2, "success"),
+        ])
+        self.assertEqual(agg["t-1"]["attempts"], 2)
+        self.assertEqual(agg["t-1"]["defects"], 0)
+        self.assertIsNone(agg["t-1"]["defect_category"])
+
+    def test_apply_does_not_mutate_input_and_keeps_old_unknown(self):
+        records = [{"id": "old", "status": "completed",
+                    "attempts": None, "defects": None},
+                   {"id": "new", "status": "completed",
+                    "attempts": None, "defects": None}]
+        merged = apply_to_records(
+            records, {"new": {"attempts": 1, "defects": 0,
+                              "defect_category": None}})
+        self.assertIsNone(records[1]["attempts"])  # input untouched
+        self.assertIsNone(merged[0]["attempts"])  # old stays unknown
+        self.assertEqual(merged[1]["attempts"], 1)
+        self.assertEqual(merged[1]["defects"], 0)
+
+
+class TestMetricsFromLedger(unittest.TestCase):
+    def test_real_rates_from_ledger(self):
+        records = apply_to_records([
+            {"id": "a", "status": "completed",
+             "attempts": None, "defects": None},
+            {"id": "b", "status": "merged",
+             "attempts": None, "defects": None},
+            {"id": "c", "status": "failed",
+             "attempts": None, "defects": None},
+        ], {
+            "a": {"attempts": 1, "defects": 0,
+                  "defect_category": None},
+            "b": {"attempts": 3, "defects": 2,
+                  "defect_category": "contract"},
+            "c": {"attempts": 2, "defects": 1,
+                  "defect_category": "contract"},
+        })
+        m = compute_metrics(records)
+        self.assertEqual(m["terminal_tasks"], 3)
+        self.assertAlmostEqual(m["first_pass_rate"], 1 / 3)
+        self.assertAlmostEqual(m["rework_rate"], 2 / 3)
+        self.assertAlmostEqual(m["defect_rate"], 2 / 3)
+        self.assertEqual(m["defect_by_category"], {"contract": 2})
+        self.assertTrue(m["first_pass_known"])
+        self.assertTrue(m["defect_known"])
+
+    def test_old_tasks_without_rows_stay_unknown_not_zero(self):
+        records = [{"id": "old", "status": "completed",
+                    "attempts": None, "defects": None}]
+        m = compute_metrics(records)
+        self.assertEqual(m["terminal_tasks"], 1)
+        self.assertIsNone(m["first_pass_rate"])
+        self.assertIsNone(m["rework_rate"])
+        self.assertIsNone(m["defect_rate"])
+        self.assertFalse(m["first_pass_known"])
+        self.assertFalse(m["defect_known"])
+        self.assertEqual(m["defect_tasks"], 0)  # 0 known defective,
+        # not a claim of "no defects": defect_known is False above.
+
+    def test_interrupted_status_never_terminal_even_with_rows(self):
+        records = apply_to_records([
+            {"id": "i", "status": "interrupted",
+             "attempts": None, "defects": None},
+        ], {"i": {"attempts": 1, "defects": 0,
+                   "defect_category": None}})
+        m = compute_metrics(records)
+        self.assertEqual(m["terminal_tasks"], 0)
+        self.assertIsNone(m["first_pass_rate"])
+        self.assertFalse(m["first_pass_known"])
+
+    def test_ledger_does_not_change_dual_column_cost(self):
+        records = apply_to_records([
+            {"id": "a", "status": "completed",
+             "attempts": None, "defects": None},
+        ], {"a": {"attempts": 1, "defects": 0,
+                   "defect_category": None}})
+        costs = {"a": {"priced_usd_micros": 5000,
+                        "unpriced_actual_tokens": 700,
+                        "has_rows": True}}
+        m = compute_metrics(records, costs)
+        self.assertAlmostEqual(m["first_pass_rate"], 1.0)
+        self.assertIsNone(m["total_usd_micros"])
+        self.assertEqual(m["priced_usd_micros_partial"], 5000)
+        self.assertEqual(
+            m["per_task_cost"][0]["unpriced_actual_tokens"], 700)
+
+
+class TestPostgresAdapterEvidence(unittest.TestCase):
+    """Adapter-level evidence only -- no real Postgres here (see
+    the module docstring: production PG still needs the manual
+    PG_DDL step and a live verification)."""
+
+    def test_pg_ddl_needs_no_sequence_or_identity_handling(self):
+        translated, _kind = _translate_sql(PG_DDL)
+        self.assertIn("CREATE TABLE IF NOT EXISTS task_attempts",
+                      translated)
+        self.assertNotIn("AUTOINCREMENT", translated)
+        self.assertNotIn("GENERATED", translated)
+        self.assertIn("PRIMARY KEY (task_id, attempt_no)", translated)
+
+    def test_insert_placeholders_translate_to_percent_s(self):
+        sql = ("INSERT INTO task_attempts(task_id, attempt_no, "
+               "outcome, is_defect, defect_category, created_at) "
+               "VALUES(?, ?, ?, ?, ?, ?)")
+        translated = _convert_placeholders(sql)
+        self.assertEqual(translated.count("%s"), 6)
+        self.assertNotIn("?", translated)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_metrics_report_endpoint.py

@@ -4,8 +4,11 @@ Only the standard library http.client; a temporary SQLite DB and a
 temporary port; no real model, no network, no secrets.
 
 Covers: unauthenticated -> 401, empty project -> report shape with
-unknown rates (None, not 0) and pending golden set, and a completed
-task with a priced cost row -> terminal count and dual-column cost.
+unknown rates (None, not 0) and pending golden set, a completed
+task with a priced cost row -> terminal count and dual-column cost,
+and (audit gap 3) ledger-backed real rates, old tasks staying
+unknown beside ledger tasks, and interrupted status staying
+non-terminal even with ledger rows.
 """
 
 import http.client
@@ -109,12 +112,90 @@ class MetricsReportEndpointTestCase(unittest.TestCase):
             "GET", "/metrics-report", key=self.owner_key)
         self.assertEqual(status, 200)
         self.assertEqual(report["metrics"]["terminal_tasks"], 1)
-        # Attempts are not in the schema: first-pass stays unknown,
-        # it must not be fabricated as 0 or 1 from status alone.
+        # No ledger rows for this task (and no ledger table yet):
+        # first-pass stays unknown, it must not be fabricated as
+        # 0 or 1 from status alone.
         self.assertIsNone(report["metrics"]["first_pass_rate"])
         self.assertEqual(report["metrics"]["total_usd_micros"], 5000)
         self.assertTrue(report["metrics"]["cost_complete"])
 
+    def test_ledger_task_gives_real_rates(self):
+        from cloud.attempt_ledger import record_attempt
+        from cloud.db import connect as db_connect
+        conn = db_connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, status) VALUES(?, ?, ?)",
+                ("t-led", self.project["id"], "completed"))
+            conn.commit()
+            record_attempt(conn, "t-led", "success")
+        finally:
+            conn.close()
+        status, report = self._req(
+            "GET", "/metrics-report", key=self.owner_key)
+        self.assertEqual(status, 200)
+        metrics = report["metrics"]
+        self.assertEqual(metrics["terminal_tasks"], 1)
+        # A ledger row exists, so 0 defects here is a real 0, not
+        # a fabricated one: defect_known is True alongside 0.0.
+        self.assertAlmostEqual(metrics["first_pass_rate"], 1.0)
+        self.assertTrue(metrics["first_pass_known"])
+        self.assertAlmostEqual(metrics["defect_rate"], 0.0)
+        self.assertTrue(metrics["defect_known"])
+        self.assertAlmostEqual(metrics["rework_rate"], 0.0)
+
+    def test_old_task_without_ledger_does_not_dilute_real_rate(self):
+        from cloud.attempt_ledger import record_attempt
+        from cloud.db import connect as db_connect
+        conn = db_connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, status) VALUES(?, ?, ?)",
+                ("t-old", self.project["id"], "completed"))
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, status) VALUES(?, ?, ?)",
+                ("t-new", self.project["id"], "merged"))
+            conn.commit()
+            record_attempt(conn, "t-new", "failure", is_defect=True,
+                           defect_category="contract")
+            record_attempt(conn, "t-new", "success")
+        finally:
+            conn.close()
+        status, report = self._req(
+            "GET", "/metrics-report", key=self.owner_key)
+        self.assertEqual(status, 200)
+        metrics = report["metrics"]
+        self.assertEqual(metrics["terminal_tasks"], 2)
+        # The old task is excluded from the denominator (known-only
+        # denominator, as in compute_metrics); it is neither counted
+        # as first-pass nor as defect-free.
+        self.assertAlmostEqual(metrics["first_pass_rate"], 0.0)
+        self.assertAlmostEqual(metrics["rework_rate"], 1.0)
+        self.assertAlmostEqual(metrics["defect_rate"], 1.0)
+        self.assertEqual(metrics["defect_by_category"],
+                         {"contract": 1})
+
+    def test_interrupted_status_excluded_even_with_ledger_rows(self):
+        from cloud.attempt_ledger import record_attempt
+        from cloud.db import connect as db_connect
+        conn = db_connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, status) VALUES(?, ?, ?)",
+                ("t-int", self.project["id"], "interrupted"))
+            conn.commit()
+            record_attempt(conn, "t-int", "failure", is_defect=True,
+                           defect_category="contract")
+        finally:
+            conn.close()
+        status, report = self._req(
+            "GET", "/metrics-report", key=self.owner_key)
+        self.assertEqual(status, 200)
+        metrics = report["metrics"]
+        self.assertEqual(metrics["terminal_tasks"], 0)
+        self.assertIsNone(metrics["first_pass_rate"])
+        self.assertFalse(metrics["first_pass_known"])
+
 
 if __name__ == "__main__":
     unittest.main()