codex-app-server/cloud/tests/test_attempt_hook.py

"""Tests for the task-attempt lifecycle hook (audit gap 3
follow-up).

Only the standard library; temporary SQLite databases, a fake
CLI runner, and a temporary HTTP server for the endpoint test.
No real model, no network, no secrets, no real Postgres (the
Postgres claims are adapter-level only, labelled as such).

Pinned policy (see ``cloud/attempt_hook.py`` -- the module is
canonical, these tests pin its observable behaviour):

- classify: coordinator never counts; a platform marker beats
  every other signal; a breakpoint resume counts as the same
  logical round's row (not an extra attempt); a continue message
  counts as a new round; first dispatch / human retry count.
- markers: consume only matches the queued generation; a stale
  marker is deleted, never misapplied; an unverifiable
  generation consumes nothing.
- corrections: append-only, last-wins, never counted as an
  attempt, cannot change an outcome, cannot target interrupted
  rows / correction rows / nonexistent attempts.
- worker: hook OFF writes nothing and creates no table (gap-3
  behaviour); hook ON writes exactly one row per counted
  dispatch (a transient retry inside one dispatch is still one
  row); a platform rerun writes no row; a gate failure writes
  no row; a ledger write failure never fails the task.
- endpoint: with ledger tasks + old tasks + a platform-completed
  task + an interrupted task side by side, terminal denominators
  contain only known tasks, interrupted status is non-terminal,
  interrupted rows count for nothing, and unknown rates are
  None + *_known False, never 0; the cost dual-column discipline
  is unchanged (priced partial shown, total None when incomplete,
  unpriced never folded to 0).
"""

import http.client
import json
import os
import sqlite3
import sys
import tempfile
import threading
import time
import unittest
from pathlib import Path
from types import SimpleNamespace

try:
    from cloud.attempt_hook import (
        MARKER_DDL,
        PG_MARKER_DDL,
        classify_dispatch,
        consume_platform_marker,
        ensure_marker_table,
        mark_platform_rerun,
    )
    from cloud.attempt_ledger import (
        OLD_TABLE_ALTER,
        PG_DDL,
        aggregate_attempts,
        apply_to_records,
        read_attempt_rows,
        record_attempt,
        record_correction,
    )
    from cloud.db import _convert_placeholders, _translate_sql
    from cloud.metrics import compute_metrics
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.attempt_hook import (
        MARKER_DDL,
        PG_MARKER_DDL,
        classify_dispatch,
        consume_platform_marker,
        ensure_marker_table,
        mark_platform_rerun,
    )
    from cloud.attempt_ledger import (
        OLD_TABLE_ALTER,
        PG_DDL,
        aggregate_attempts,
        apply_to_records,
        read_attempt_rows,
        record_attempt,
        record_correction,
    )
    from cloud.db import _convert_placeholders, _translate_sql
    from cloud.metrics import compute_metrics


def _sqlite_conn(path):
    from cloud.db import connect as db_connect
    return db_connect(path)


def _insert_task(db_path, task_id, prompt="do work", status="queued",
                 project_id="p1", **extra):
    now = time.time()
    cols = ["id", "project_id", "prompt", "status",
            "idempotency_key", "depends_on", "priority", "created_by",
            "created_at", "updated_at", "last_heartbeat"]
    vals = [task_id, project_id, prompt, status, None, "[]", 0,
            "tester", now, now, None]
    for key, value in extra.items():
        cols.append(key)
        vals.append(value)
    conn = sqlite3.connect(db_path)
    try:
        conn.execute(
            "INSERT INTO tasks(%s) VALUES(%s)"
            % (", ".join(cols), ", ".join("?" * len(cols))),
            vals)
        conn.commit()
    finally:
        conn.close()


def _task_status(db_path, task_id):
    conn = sqlite3.connect(db_path)
    try:
        row = conn.execute(
            "SELECT status FROM tasks WHERE id = ?",
            (task_id,)).fetchone()
    finally:
        conn.close()
    return row[0] if row else None


def _ledger_rows(db_path):
    conn = sqlite3.connect(db_path)
    try:
        rows = conn.execute(
            "SELECT task_id, attempt_no, outcome, is_defect, "
            "defect_category, corrects_attempt_no FROM task_attempts "
            "ORDER BY task_id, attempt_no").fetchall()
    except sqlite3.OperationalError:
        return []
    finally:
        conn.close()
    return [tuple(r) for r in rows]


def _table_exists(db_path, table):
    conn = sqlite3.connect(db_path)
    try:
        row = conn.execute(
            "SELECT COUNT(*) FROM sqlite_master WHERE type='table' "
            "AND name=?", (table,)).fetchone()
    finally:
        conn.close()
    return row[0] == 1


def _completed_runner(cmd):
    """Fake CLI: dispatch/run always succeed, host.py-style JSON."""
    if "dispatch" in cmd:
        name = cmd[cmd.index("--name") + 1]
    else:
        name = cmd[cmd.index("--task") + 1]
    return SimpleNamespace(
        returncode=0,
        stdout=json.dumps({"task": name, "status": "completed",
                           "text": "done"}),
        stderr="")


class TestClassifyDispatch(unittest.TestCase):
    def test_coordinator_beats_marker(self):
        decision = classify_dispatch(
            {"coordinator_parts": [{"name": "a"}]}, True)
        self.assertFalse(decision["counted"])
        self.assertEqual(decision["kind"], "coordinator_orchestration")

    def test_platform_marker_beats_breakpoint(self):
        # A platform goal auto-resume is a breakpoint continuation
        # AND a platform rerun: uniform platform exclusion wins.
        decision = classify_dispatch(
            {"_goal_resume_breakpoint": True}, True)
        self.assertFalse(decision["counted"])
        self.assertEqual(decision["kind"], "platform_auto_resume")

    def test_human_breakpoint_resume_counts(self):
        decision = classify_dispatch(
            {"_goal_resume_breakpoint": True}, False)
        self.assertTrue(decision["counted"])
        self.assertEqual(decision["kind"], "breakpoint_resume")

    def test_resume_kind_breakpoint_counts_despite_message(self):
        # Old-verify goal + /continue: the message is NOT a new
        # goal, the loop continues from its breakpoint.
        decision = classify_dispatch(
            {"resume_message": "keep going",
             "_resume_kind": "breakpoint"}, False)
        self.assertTrue(decision["counted"])
        self.assertEqual(decision["kind"], "breakpoint_resume")

    def test_breakpoint_fallback_is_not_a_resume(self):
        # Resolution fell back to a fresh goal run: the pending
        # flag alone must not classify it as a resume.
        decision = classify_dispatch(
            {"goal_resume_pending": 1,
             "_goal_resume_breakpoint": False}, False)
        self.assertTrue(decision["counted"])
        self.assertEqual(decision["kind"], "first_or_human_retry")

    def test_new_goal_continue_counts_as_new_round(self):
        decision = classify_dispatch(
            {"resume_message": "new goal",
             "_resume_kind": "new-goal"}, False)
        self.assertTrue(decision["counted"])
        self.assertEqual(decision["kind"], "human_continue")

    def test_plain_continue_counts(self):
        decision = classify_dispatch(
            {"resume_message": "hello"}, False)
        self.assertTrue(decision["counted"])
        self.assertEqual(decision["kind"], "human_continue")

    def test_first_dispatch_counts(self):
        decision = classify_dispatch({}, False)
        self.assertTrue(decision["counted"])
        self.assertEqual(decision["kind"], "first_or_human_retry")


class TestMarkerStore(unittest.TestCase):
    def setUp(self):
        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 _markers(self):
        try:
            return self.conn.execute(
                "SELECT task_id, origin, queued_at FROM "
                "task_attempt_origin_markers").fetchall()
        except Exception:
            return []

    def test_mark_then_consume_matching(self):
        mark_platform_rerun(self.conn, "t-1", 1234.5)
        self.conn.commit()
        self.assertTrue(consume_platform_marker(self.conn, "t-1", 1234.5))
        self.conn.commit()
        self.assertEqual(self._markers(), [])
        # Consuming twice never matches twice.
        self.assertFalse(
            consume_platform_marker(self.conn, "t-1", 1234.5))

    def test_mismatched_marker_is_stale_and_deleted(self):
        mark_platform_rerun(self.conn, "t-1", 111.0)
        self.conn.commit()
        self.assertFalse(
            consume_platform_marker(self.conn, "t-1", 222.0))
        self.conn.commit()
        self.assertEqual(self._markers(), [])

    def test_none_generation_consumes_nothing(self):
        mark_platform_rerun(self.conn, "t-1", 111.0)
        self.conn.commit()
        self.assertFalse(
            consume_platform_marker(self.conn, "t-1", None))
        # The marker survives for a later verifiable consume.
        self.assertTrue(
            consume_platform_marker(self.conn, "t-1", 111.0))

    def test_requeue_overwrites_older_marker(self):
        mark_platform_rerun(self.conn, "t-1", 111.0)
        mark_platform_rerun(self.conn, "t-1", 222.0)
        self.conn.commit()
        self.assertTrue(
            consume_platform_marker(self.conn, "t-1", 222.0))

    def test_mark_requires_task_id(self):
        with self.assertRaises(ValueError):
            mark_platform_rerun(self.conn, "", 1.0)

    def test_marker_pg_adapter_evidence(self):
        # Adapter-level only: no real Postgres on this machine.
        self.assertEqual(PG_MARKER_DDL, MARKER_DDL)
        translated, _kind = _translate_sql(MARKER_DDL)
        self.assertIn(
            "CREATE TABLE IF NOT EXISTS task_attempt_origin_markers",
            translated)
        self.assertNotIn("AUTOINCREMENT", translated)
        self.assertNotIn("GENERATED", translated)
        self.assertIn("PRIMARY KEY", translated)
        insert = ("INSERT INTO task_attempt_origin_markers(task_id, "
                  "origin, queued_at) VALUES(?, ?, ?)")
        self.assertEqual(_convert_placeholders(insert).count("%s"), 3)
        self.assertNotIn("?", _convert_placeholders(insert))


class TestCorrectionRows(unittest.TestCase):
    def setUp(self):
        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 _agg(self, task_id):
        rows = read_attempt_rows(self.conn)
        agg = aggregate_attempts(rows)
        return agg.get(task_id)

    def test_correction_unmarks_a_defect(self):
        record_attempt(self.conn, "t-1", "failure", is_defect=True,
                       defect_category="contract")
        record_correction(self.conn, "t-1", 1, is_defect=False)
        agg = self._agg("t-1")
        self.assertEqual(agg["attempts"], 1)
        self.assertEqual(agg["defects"], 0)
        self.assertIsNone(agg["defect_category"])

    def test_correction_marks_a_missed_defect(self):
        record_attempt(self.conn, "t-1", "failure")
        record_correction(self.conn, "t-1", 1, is_defect=True,
                          defect_category="directory")
        agg = self._agg("t-1")
        self.assertEqual(agg["attempts"], 1)
        self.assertEqual(agg["defects"], 1)
        self.assertEqual(agg["defect_category"], "directory")

    def test_correction_is_never_counted_as_attempt(self):
        record_attempt(self.conn, "t-1", "failure", is_defect=True,
                       defect_category="contract")
        record_attempt(self.conn, "t-1", "success")
        record_correction(self.conn, "t-1", 1, is_defect=False)
        agg = self._agg("t-1")
        self.assertEqual(agg["attempts"], 2)
        self.assertEqual(agg["defects"], 0)

    def test_last_correction_wins(self):
        record_attempt(self.conn, "t-1", "failure", is_defect=True,
                       defect_category="contract")
        record_correction(self.conn, "t-1", 1, is_defect=False)
        record_correction(self.conn, "t-1", 1, is_defect=True,
                          defect_category="directory")
        agg = self._agg("t-1")
        self.assertEqual(agg["defects"], 1)
        self.assertEqual(agg["defect_category"], "directory")

    def test_first_defect_category_reevaluated_after_correction(self):
        record_attempt(self.conn, "t-1", "failure", is_defect=True,
                       defect_category="contract")
        record_attempt(self.conn, "t-1", "failure", is_defect=True,
                       defect_category="directory")
        record_correction(self.conn, "t-1", 1, is_defect=False)
        agg = self._agg("t-1")
        self.assertEqual(agg["defects"], 1)
        self.assertEqual(agg["defect_category"], "directory")

    def test_success_row_can_be_judged_defective_on_review(self):
        # Defect judgement is independent of outcome: a shipped
        # success can still be judged defective after review.
        record_attempt(self.conn, "t-1", "success")
        record_correction(self.conn, "t-1", 1, is_defect=True,
                          defect_category="contract")
        agg = self._agg("t-1")
        self.assertEqual(agg["attempts"], 1)
        self.assertEqual(agg["defects"], 1)

    def test_correction_targets_rejected(self):
        record_attempt(self.conn, "t-1", "failure")
        record_attempt(self.conn, "t-1", "interrupted")
        with self.assertRaises(ValueError):
            record_correction(self.conn, "t-1", 99, is_defect=True)
        with self.assertRaises(ValueError):
            # Attempt 2 is the interrupted row: no defect
            # judgement exists to correct.
            record_correction(self.conn, "t-1", 2, is_defect=True)
        with self.assertRaises(ValueError):
            # Category without a defect flag.
            record_correction(self.conn, "t-1", 1, is_defect=False,
                              defect_category="contract")
        # The two rejected writes above left exactly the two
        # original rows; the category rejection wrote nothing.
        self.assertEqual(len(_ledger_rows_for(self.conn)), 2)
        corr = record_correction(self.conn, "t-1", 1, is_defect=True,
                                 defect_category="contract")
        with self.assertRaises(ValueError):
            # A correction row is not itself correctable.
            record_correction(self.conn, "t-1", corr, is_defect=False)

    def test_old_gap3_table_reads_and_upgrades(self):
        # A table built from gap 3's DDL (no corrects column):
        # old rows still read as plain attempts...
        gap3_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))")
        self.conn.execute(gap3_ddl)
        self.conn.execute(
            "INSERT INTO task_attempts(task_id, attempt_no, outcome, "
            "is_defect, defect_category, created_at) "
            "VALUES('t-old', 1, 'success', 0, NULL, 1.0)")
        self.conn.commit()
        agg = aggregate_attempts(read_attempt_rows(self.conn))
        self.assertEqual(agg["t-old"]["attempts"], 1)
        # ...and the documented one-line ALTER enables
        # corrections on that table.
        self.conn.execute(OLD_TABLE_ALTER)
        self.conn.commit()
        record_correction(self.conn, "t-old", 1, is_defect=True,
                          defect_category="contract")
        self.assertEqual(self._agg("t-old")["defects"], 1)

    def test_correction_pg_adapter_evidence(self):
        translated, _kind = _translate_sql(PG_DDL)
        self.assertIn("corrects_attempt_no INTEGER", translated)
        self.assertIn("PRIMARY KEY (task_id, attempt_no)", translated)
        self.assertNotIn("AUTOINCREMENT", translated)
        self.assertNotIn("GENERATED", translated)
        insert = (
            "INSERT INTO task_attempts(task_id, attempt_no, outcome, "
            "is_defect, defect_category, created_at, "
            "corrects_attempt_no) VALUES(?, ?, ?, ?, ?, ?, ?)")
        self.assertEqual(_convert_placeholders(insert).count("%s"), 7)
        self.assertNotIn("?", _convert_placeholders(insert))


def _ledger_rows_for(conn):
    return conn.execute(
        "SELECT * FROM task_attempts ORDER BY task_id, attempt_no"
    ).fetchall()


class TestWorkerHook(unittest.TestCase):
    def setUp(self):
        self._old_push_allow = os.environ.get(
            "CODEX_PUSH_ALLOW_PRIVATE_TARGETS")
        os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1"
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        from cloud.server import init_db
        init_db(self.db_path)
        self.env = {"XCM_API_KEY": "test-placeholder"}

    def tearDown(self):
        if getattr(self, "_old_push_allow", None) is None:
            os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None)
        else:
            os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = (
                self._old_push_allow)
        self.tmp.cleanup()

    def _worker(self, runner=None, **kwargs):
        from cloud.worker import Worker
        kwargs.setdefault("env", self.env)
        kwargs.setdefault("retry_backoff_seconds", 0)
        return Worker(self.db_path, "/fake/host.py",
                      runner=runner or _completed_runner, **kwargs)

    def test_hook_off_writes_nothing_and_creates_no_table(self):
        _insert_task(self.db_path, "A")
        worker = self._worker()
        self.assertFalse(worker.attempt_ledger_hook)
        result = worker.run_one()
        self.assertEqual(result["status"], "completed")
        self.assertFalse(_table_exists(self.db_path, "task_attempts"))
        self.assertFalse(
            _table_exists(self.db_path, "task_attempt_origin_markers"))

    def test_hook_on_first_run_writes_one_success_row(self):
        _insert_task(self.db_path, "A")
        worker = self._worker(attempt_ledger_hook=True)
        result = worker.run_one()
        self.assertEqual(result["status"], "completed")
        rows = _ledger_rows(self.db_path)
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0][:5], ("A", 1, "success", 0, None))
        self.assertIsNone(rows[0][5])

    def test_hook_on_transient_retry_is_still_one_row(self):
        calls = {"n": 0}

        def flaky(cmd):
            calls["n"] += 1
            if calls["n"] == 1:
                return SimpleNamespace(returncode=1, stdout="boom",
                                       stderr="")
            return _completed_runner(cmd)

        _insert_task(self.db_path, "A")
        worker = self._worker(runner=flaky, attempt_ledger_hook=True,
                              max_retries=1)
        result = worker.run_one()
        self.assertEqual(result["status"], "completed")
        rows = _ledger_rows(self.db_path)
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0][2], "success")

    def test_hook_on_human_retry_writes_second_row(self):
        _insert_task(self.db_path, "A")
        worker = self._worker(attempt_ledger_hook=True)
        self.assertEqual(worker.run_one()["status"], "completed")
        # Simulate the server /retry endpoint: back to queued,
        # results cleared, NO platform marker written.
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "UPDATE tasks SET status='queued', result_text=NULL, "
                "error_text=NULL, actual_model=NULL WHERE id='A'")
            conn.commit()
        finally:
            conn.close()
        worker2 = self._worker(attempt_ledger_hook=True)
        self.assertEqual(worker2.run_one()["status"], "completed")
        rows = _ledger_rows(self.db_path)
        self.assertEqual(len(rows), 2)
        self.assertEqual([r[1] for r in rows], [1, 2])
        self.assertEqual([r[2] for r in rows],
                         ["success", "success"])

    def test_hook_on_platform_rerun_writes_no_row(self):
        from cloud.attempt_ledger import record_attempt
        from cloud.db import connect as db_connect
        # Previous dispatch ended interrupted (its row is
        # count-neutral by the gap-3 rule).
        _insert_task(self.db_path, "A", status="interrupted",
                     interrupted_reason="worker_shutdown",
                     paused_by_user=0, auto_resume_count=0)
        conn = db_connect(self.db_path)
        try:
            record_attempt(conn, "A", "interrupted")
        finally:
            conn.close()
        worker = self._worker(attempt_ledger_hook=True)
        resumed = worker.auto_resume_interrupted()
        self.assertEqual(resumed, ["A"])
        self.assertEqual(_task_status(self.db_path, "A"), "queued")
        result = worker.run_one()
        self.assertEqual(result["status"], "completed")
        rows = _ledger_rows(self.db_path)
        # Still exactly the one interrupted row: the platform
        # rerun added no counted row, and its marker is gone.
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0][2], "interrupted")
        conn = sqlite3.connect(self.db_path)
        try:
            markers = conn.execute(
                "SELECT COUNT(*) FROM task_attempt_origin_markers"
            ).fetchone()[0]
        finally:
            conn.close()
        self.assertEqual(markers, 0)

    def test_hook_on_gate_failure_writes_no_row(self):
        _insert_task(self.db_path, "A")
        worker = self._worker(attempt_ledger_hook=True, env={})
        result = worker.run_one()
        self.assertEqual(result["status"], "failed")
        self.assertFalse(_table_exists(self.db_path, "task_attempts"))

    def test_hook_on_cancelled_dispatch_writes_interrupted_row(self):
        # Established pattern (test_worker
        # test_cancel_during_dispatch_keeps_interrupted): the fake
        # runner cancels the task mid-dispatch, then reports
        # completed; run_one must keep interrupted, and the hook
        # must write exactly one interrupted row for it.
        worker_ref = {}

        def cancelling(cmd):
            worker_ref["w"].cancel("A")
            return _completed_runner(cmd)

        _insert_task(self.db_path, "A")
        worker = self._worker(runner=cancelling,
                              attempt_ledger_hook=True)
        worker_ref["w"] = worker
        result = worker.run_one()
        self.assertEqual(result["status"], "interrupted")
        self.assertEqual(_task_status(self.db_path, "A"), "interrupted")
        rows = _ledger_rows(self.db_path)
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0][2], "interrupted")
        # Interrupted rows never enter the aggregate.
        records = apply_to_records(
            [{"id": "A", "status": "interrupted",
              "attempts": None, "defects": None}],
            aggregate_attempts(read_attempt_rows(
                _sqlite_conn(self.db_path))))
        metrics = compute_metrics(records)
        self.assertEqual(metrics["terminal_tasks"], 0)
        self.assertIsNone(metrics["first_pass_rate"])
        self.assertFalse(metrics["first_pass_known"])

    def test_hook_write_failure_never_fails_task(self):
        import cloud.worker as worker_mod
        _insert_task(self.db_path, "A")
        worker = self._worker(attempt_ledger_hook=True)
        original = worker_mod._attempt_record

        def boom(conn, task_id, outcome):
            raise RuntimeError("ledger down")

        worker_mod._attempt_record = boom
        try:
            result = worker.run_one()
        finally:
            worker_mod._attempt_record = original
        self.assertEqual(result["status"], "completed")
        self.assertEqual(_task_status(self.db_path, "A"), "completed")


class MetricsHookEndpointTestCase(unittest.TestCase):
    """Mixed ledger/old/platform/interrupted tasks on the real
    endpoint: terminal denominators contain only known tasks."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        from cloud.auth import AuthStore
        from cloud.cost import CostStore
        from cloud.server import create_server
        self.auth = AuthStore(self.db_path)
        self.owner_id, self.owner_key = self.auth.create_user("owner")
        self.server = create_server(self.db_path, "127.0.0.1", 0)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(
            target=self.server.serve_forever, daemon=True)
        self.thread.start()
        status, self.project = self._req(
            "POST", "/projects", {"name": "demo"}, self.owner_key)
        self.assertEqual(status, 201)
        self.costs = CostStore(self.db_path)

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)
        try:
            self.auth.conn.close()
        except Exception:
            pass
        self.tmp.cleanup()

    def _req(self, method, path, body=None, key=None):
        conn = http.client.HTTPConnection("127.0.0.1", self.port,
                                          timeout=10)
        hdrs = {}
        payload = None
        if body is not None:
            payload = json.dumps(body).encode("utf-8")
            hdrs["Content-Type"] = "application/json"
        if key is not None:
            hdrs["X-API-Key"] = key
        try:
            conn.request(method, path, body=payload, headers=hdrs)
            resp = conn.getresponse()
            data = resp.read()
        finally:
            conn.close()
        try:
            parsed = json.loads(data.decode("utf-8")) if data else None
        except ValueError:
            parsed = None
        return resp.status, parsed

    def test_mixed_tasks_known_only_denominators(self):
        from cloud.db import connect as db_connect
        pid = self.project["id"]
        conn = db_connect(self.db_path)
        try:
            for tid, status in (
                    ("t-a", "completed"), ("t-b", "merged"),
                    ("t-old", "completed"), ("t-plat", "completed"),
                    ("t-int", "interrupted")):
                conn.execute(
                    "INSERT INTO tasks(id, project_id, status) "
                    "VALUES(?, ?, ?)", (tid, pid, status))
            conn.commit()
            # t-a: one clean round -> first-pass candidate.
            record_attempt(conn, "t-a", "success")
            # t-b: defect round then success -> rework + defect.
            record_attempt(conn, "t-b", "failure", is_defect=True,
                           defect_category="contract")
            record_attempt(conn, "t-b", "success")
            # t-plat: completed ONLY via a platform rerun: its
            # ledger holds an interrupted row and nothing else,
            # so it stays unknown (never 0, never first-pass).
            record_attempt(conn, "t-plat", "interrupted")
            # t-int: interrupted STATUS with a failure row: the
            # status alone keeps it out of terminal tasks.
            record_attempt(conn, "t-int", "failure", is_defect=True,
                           defect_category="contract")
        finally:
            conn.close()
        # Priced cost only for t-a: the dual-column discipline
        # must survive the ledger (total None when incomplete,
        # priced partial still shown, unpriced never folded to 0).
        self.costs.record("t-a", pid, "gpt-6-sol",
                          input_tokens=1000, output_tokens=100,
                          usd_micros=5000, source="actual")
        status, report = self._req(
            "GET", "/metrics-report", key=self.owner_key)
        self.assertEqual(status, 200)
        metrics = report["metrics"]
        # Terminal tasks: t-a, t-b, t-old, t-plat (t-int is not
        # terminal despite its ledger row).
        self.assertEqual(metrics["terminal_tasks"], 4)
        # Known-only denominators: t-a + t-b (2 tasks). If the
        # old or platform task were wrongly counted as
        # first-pass / defect-free, these would be 3/4 and 1/4.
        self.assertAlmostEqual(metrics["first_pass_rate"], 0.5)
        self.assertTrue(metrics["first_pass_known"])
        self.assertAlmostEqual(metrics["rework_rate"], 0.5)
        self.assertAlmostEqual(metrics["defect_rate"], 0.5)
        self.assertTrue(metrics["defect_known"])
        self.assertEqual(metrics["defect_by_category"],
                         {"contract": 1})
        self.assertIsNone(metrics["total_usd_micros"])
        self.assertEqual(metrics["priced_usd_micros_partial"], 5000)
        self.assertFalse(metrics["cost_complete"])


if __name__ == "__main__":
    unittest.main()