diff --git a/cloud/attempt_ledger.py b/cloud/attempt_ledger.py new file mode 100644 index 0000000..6de95ec --- /dev/null +++ b/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 diff --git a/cloud/metrics.py b/cloud/metrics.py index 2ff2cdf..2639694 100644 --- a/cloud/metrics.py +++ b/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 diff --git a/cloud/server.py b/cloud/server.py index 94025cc..82c4266 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -154,6 +154,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 @@ -9638,12 +9641,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。 """ @@ -9662,6 +9667,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," diff --git a/cloud/tests/test_attempt_ledger.py b/cloud/tests/test_attempt_ledger.py new file mode 100644 index 0000000..a8ae6c9 --- /dev/null +++ b/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() diff --git a/cloud/tests/test_metrics_report_endpoint.py b/cloud/tests/test_metrics_report_endpoint.py index 32e7991..35eabf4 100644 --- a/cloud/tests/test_metrics_report_endpoint.py +++ b/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() diff --git a/docs/arch-conformance-metrics.md b/docs/arch-conformance-metrics.md index 426f829..b994c6f 100644 --- a/docs/arch-conformance-metrics.md +++ b/docs/arch-conformance-metrics.md @@ -40,10 +40,55 @@ workflow——第 2 轨拥有 lint/类型/安扫,本轨不碰。 `priced_usd_micros_partial` 出现——无价永不记 0、永不折进总额。 - 未知率回 `None` + `*_known: false`,绝不报 0。 -当前诚实边界(端点实测):tasks 表无 attempts/defects 列,端点 -传 `None`,首过/缺陷/返工率恒为 `None`;金任务集未经人工审定, -报告恒为 `pending_human_approval`、`decision_ready: false`。接真实 -输入需两项前置(见待审定清单),本轨不编造。 +当前诚实边界(端点实测,审计加固补缺口第 3 项之后):tasks 表 +仍无 attempts/defects 列——台账改为独立表 +`task_attempts`(`cloud/attempt_ledger.py`),不改主表结构。 +端点 `GET /metrics-report` 聚合台账后传给 `compute_metrics`: +有台账行的任务算出真实首过/缺陷/返工率;无台账行的老任务 +仍回 `None` + `*_known: false`,绝不被算成首过或无缺陷; +台账表尚不存在(老库、或生产 PG 未手动执行 `PG_DDL`)时 +端点与无台账行同等处理,不报错。金任务集未经人工审定, +报告恒为 `pending_human_approval`、`decision_ready: false`。 + +## 任务尝试台账(审计加固补缺口第 3 项,选型与口径) + +选型:独立台账表,不改主表。列方案查询简单(一行一任务、 +无关联),但把每次尝试的历史压成两个计数器——哪次尝试失败、 +属哪类缺陷、顺序如何,事后无法恢复;且在生产 Postgres 上拓 +宽高频 `tasks` 表比重建一张空表更重。台账表多一次聚合查询, +换来完整尝试历史与不碰主表的迁移,取舍写在 +`cloud/attempt_ledger.py` 模块 docstring 里。 + +表 `task_attempts(task_id, attempt_no, outcome, is_defect, +defect_category, created_at)`,主键 `(task_id, attempt_no)`, +`attempt_no` 按任务 `MAX + 1` 递增、只追加不改写。口径: + +- `outcome` ∈ success / failure / interrupted;`interrupted` + 行只留档,不计入 attempts 也不计入 defects(平台中断自动 + 重跑是平台事件,不是 agent 失败;只有中断行的任务在聚合 + 里直接缺席,仍回未知而非 0)。 +- `is_defect` 由记录方显式判定,绝不从 `failure` 自动推导 + (失败重试未必是缺陷轮);`defect_category` 须配 + `is_defect=True`,中断行永不能是缺陷,违例写时即 + `ValueError`,不存歧义数据。任务的单类别标签取首个缺陷 + 行的类别(attempt_no 顺序,先到先得,确定性)。 +- 迁移幂等:SQLite 侧 `CREATE TABLE IF NOT EXISTS`,重跑 + 不动已有行;Postgres 侧 `PG_DDL`(与 SQLite DDL 同文,只用 + Postgres 合法类型与复合主键、无自增/identity 列,故无需 + 序列处理)**须在生产 PG 上手动执行并验证**(`\d + task_attempts` 或 information_schema 等价查询),且有意不 + 进 `POSTGRES_SCHEMA`(不静默拓宽迁移覆盖集);手动步骤 + 未做之前,Postgres 端点只会回未知率,不会报错。本机无 + PG,真实 PG 验证未做;PG 证据目前只有适配层级别:`?` 占位 + 经 `cloud.db` 适配翻译为 `%s`、DDL 无需序列处理(测试 + `TestPostgresAdapterEvidence` 钉住)。 +- 未做项(有意不做,原因见模块 docstring):生命周期写 + 入钩子未接——worker 重试/续跑循环里哪一次算一次尝试、谁 + 判定 `is_defect`,接错会把平台自动续跑算成尝试、把瞬时 + 重试算成缺陷,正好违反本轨钉住的规则;在此之前台账只 + 经显式 `record_attempt` 调用填充,生产率在写钩子落地前 + 保持诚实未知,不自动填数。读端点不建表(只读端点不改 + schema),建表只发生在写路径。 ## 回流与审计标准迭代(可测试规则,非口号) @@ -93,3 +138,7 @@ workflow——第 2 轨拥有 lint/类型/安扫,本轨不碰。 待人工项(单独列):金任务集审定清单的四项人工确认;抽查 seed 的记录与保管;生产抽查比例拍板(默认值 0.10/0.50 只是测试初值)。 +台账待人工项(补缺口第 3 项之后):生产 PG 上手动执行 +`task_attempts` 的 `PG_DDL` 并验证;生命周期写入钩子拍板—— +哪次执行算一次尝试、谁判定 `is_defect`(在拍板前台账只经 +显式调用填充,生产率保持未知)。