"""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()