"""Lifecycle hook policy for the task-attempt ledger (audit gap 3 follow-up: "which dispatch counts as an attempt, and who judges ``is_defect``"). This module is the CANONICAL policy text. The same policy is restated (shorter) in ``docs/arch-conformance-metrics.md`` and pinned by ``cloud/tests/test_attempt_hook.py``; if the three ever disagree, this module wins and the other two must be fixed. The problem it solves: audit gap 3 built the ledger but deliberately left the writer unconnected, because a naive writer -- one row per worker execution -- misclassifies exactly twice: it counts a platform auto-resume as an attempt, and it counts a transient retry as a defect. This module fixes the writer's trigger, not the ledger's math: ``aggregate_attempts`` is untouched by the hook (its only change in this track is the correction-row rule in ``attempt_ledger.py``), and the cost dual-column code (``cloud.cost``) is untouched entirely. POLICY (each rule is enforced in code below, not just prose): 1. One counted dispatch = one row, at most. A dispatch that the hook counts writes EXACTLY ONE ledger row, when its outcome is known (terminal ``completed`` / ``failed`` / ``interrupted``, or the goal-conversion case in ``worker.run_one``), with ``is_defect=False`` always (rule 6). A dispatch that the hook does not count writes NO row at all -- not even an ``interrupted`` row -- so an uncounted dispatch can never create the ledger table as a side effect. 2. What counts as one attempt -- a logical agent round, not a worker execution: - First dispatch of a task: counts (``first_or_human_retry``). It is the task's first agent round. - Human ``/retry`` of a failed/interrupted/merge-failed task: counts (same kind). A retry is a human-initiated new round: new model work, new result, new cost. The server endpoint leaves no marker behind, so the worker cannot distinguish a retry from a first dispatch -- and does not need to, because both count. - Human ``/continue`` with a new message: counts (``human_continue``). The message starts a new round on the same session thread. Non-goal tasks and goal tasks opening a NEW goal loop (``_resume_kind == "new-goal"``) are both here. - Human breakpoint resume (``/goal-resume``, or ``/continue`` on an old-verify goal task, i.e. ``_resume_kind == "breakpoint"``): counts (``breakpoint_resume``) -- but as the SAME logical round's row, not as an extra attempt. The fragment that was interrupted wrote only an excluded ``interrupted`` row (gap 3 rule), so the resume's terminal row is the round's single counted row. A resume that ends interrupted again writes only another excluded row: the round is still pending, and still contributes zero counted attempts. Rationale: one goal loop = one attempt, however many executions the loop took; the interruption history stays visible in the ``interrupted`` rows and in ``auto_resume_count`` / audit / task_steps, not in the attempt count. EXCEPTION (ordering, rule 4): when the resume was itself platform-triggered, the platform marker wins and the dispatch does NOT count -- a platform goal auto-resume and a human goal-resume are the same loop continuation, but only the human one is attributed to the agent. That asymmetry is deliberate and uniform with the platform rule below; flipping it is the listed user decision, not a silent default. - Coordinator parent dispatch: does NOT count (``coordinator_orchestration``). The parent runs no model round (``cost_record_id`` is None) and its terminal status is an aggregate of its children; each child records its own row through its own dispatch. Counting the parent would double-count orchestration as an agent attempt. - Platform auto-resume dispatch (worker ``auto_resume_interrupted`` requeue, goal or non-goal): does NOT count (``platform_auto_resume``). This is the default the gap-3 track demanded, and it is UNIFORM: every platform requeue is excluded, whether it continues a goal loop or fresh-runs a non-goal task. Rationale: the platform killed the previous execution and re-ran it without any human or coordinator decision; attributing that execution to the agent would punish platform instability in an agent-quality rate. TRADE-OFF, STATED NOT HIDDEN: a task whose terminal outcome came ONLY from platform dispatches stays unknown in the ledger (no counted row), so it is excluded from rate denominators -- the rates describe tasks whose terminal outcome came from a human/first round. The platform reruns themselves are NOT lost: they are recorded in ``tasks.auto_resume_count``, the ``worker_auto_resumed`` audit events, and the ``task_steps`` ``auto_resume`` rows. The rejected alternative was a separately marked ledger row that is excluded from denominators: same denominator outcome, but a second row kind, a new column, and aggregate exclusion logic, for co-location only. If a future policy wants platform completions counted as loop completions, the flip is one line in ``classify_dispatch`` (platform + breakpoint -> counted) plus tests -- it is listed as a user decision in the track summary, not silently decided here. - Pre-execution gate failures: do NOT count. A dispatch that never reached the model -- missing key, repository resolution failure, quota gate, goal-rounds cap, gateway preparation failure -- is an admission rejection, not an attempt. It writes no row (and no cost record either, which is the existing behaviour); it stays visible in audit (``worker_missing_key`` / ``worker_repo_failed`` / ``worker_quota_stop`` / ``worker_goal_cap_rejected`` / ``worker_gateway_failed``). - Dispatches that never ran (budget gate, no ready task, claim lost to another worker): do NOT count and are not even classified -- ``run_one`` returns before the hook runs, and no marker is consumed (see rule 4). 3. Transient retries inside one dispatch are NOT new attempts. The worker's ``max_retries`` loop, its one real fallback re-run, its goal "already exists" breakpoint fallback, and its follow-up drain all run inside a single ``run_one`` dispatch, share one cost aggregate, and produce ONE ledger row: the dispatch's terminal outcome. An inner failure that is later retried successfully leaves no failure row behind; an inner failure that exhausts retries leaves one failure row, not N. This is enforced by placement (the hook is called once per dispatch, at ``run_one``'s terminal points) rather than by counting executions. 4. Platform identification -- the marker mechanism. The worker that requeues a task is not necessarily the worker that later claims it, so the origin must survive in the database, not in memory. ``auto_resume_interrupted`` writes one marker row (``task_attempt_origin_markers``) in the SAME transaction as the requeue, with ``queued_at`` equal to the ``updated_at`` the requeue writes. ``run_one`` captures the queued task's ``updated_at`` BEFORE claiming (claim overwrites it), and after a successful claim consumes the marker ONLY if ``queued_at`` still matches. Any intervening human action (cancel, retry, continue) rewrites ``updated_at``, so a stale marker can never misclassify a later human round: on mismatch the marker is deleted as stale and the dispatch is classified by its own fields. A claim that loses the race consumes nothing, so the winning worker still sees its marker. Marker writes and reads happen ONLY when the hook is enabled; with the hook off, ``auto_resume_interrupted`` writes no marker and ``run_one`` reads none, so off behaviour is byte-identical to gap 3 (no new table, no new row). Marker-table creation is a write-path-only step, like the ledger table itself; the read path (``GET /metrics-report``) never touches it. 5. Switch. ``Worker(attempt_ledger_hook=False)`` (CLI ``--attempt-ledger-hook``), default OFF. OFF means: no ledger row, no marker row, no marker-table creation, no ledger-table creation from the worker -- exactly gap 3's "explicit calls only" behaviour, including old tasks still returning ``None`` (unknown, not 0) and the read endpoint never creating a table. ON means rules 1-4 apply. Enabling the hook mid-flight has one documented edge: a task requeued by auto-resume WHILE the hook was off carries no marker, so its next dispatch counts as a human/first round (one extra counted row for that in-flight task, once). That is the fail-safe direction (a human round is the common case), and it self-heals: every later requeue writes a marker. 6. ``is_defect`` -- who judges, and how a wrong judgement is fixed. The worker hook NEVER judges a defect: it always records ``is_defect=False``. A failed dispatch is not a defect (transient failures, quota stops, and cancelled rounds all fail without being defects), and the worker cannot tell a transient failure from a defect round at runtime -- so it must not guess. The ONLY defect judgements in the ledger are explicit ones, made at record time by an explicit recorder: a human maintainer (or a coordinator review) calling ``record_attempt`` / ``record_correction`` directly with ``is_defect=True`` and a category. The bar for that authority is maintainer-level (``approve_merge``, the same bar as ``/retry``: both spend model money and move decision-grade rates). A defect judgement made in error is corrected ONLY by appending a correction row (``record_correction``): the old row is never updated or deleted, corrections apply in attempt order with last-wins, and a correction can only change the defect judgement, never the outcome. There is NO HTTP endpoint for marking or correcting defects in this track -- the function is the interface, and an endpoint (with the ``approve_merge`` permission check) is follow-up work, listed as unconnected in the track summary. Until it exists, defect marks come only from explicit, reviewed calls, never from automation. 7. Failure discipline. Every hook write is best-effort: a ledger or marker failure is logged (``worker_attempt_ledger_failed`` / ``worker_attempt_marker_failed``) and NEVER fails the task, blocks a requeue, or changes a task's status. The one exception is validation inside ``record_attempt`` / ``record_correction`` (bad outcome, defect on an interrupted row, correcting a nonexistent attempt): those raise ``ValueError`` BEFORE writing, by design, so ambiguous data is never stored -- but the worker hook only ever passes validated values (a terminal outcome, ``is_defect=False``), so it cannot hit that path in normal operation. 8. Postgres. The marker table follows the ledger table's manual discipline: it is NOT created by ``init_postgres_schema``, and ``PG_MARKER_DDL`` below is the exact statement to run by hand on production Postgres (and verify) before enabling the hook against a Postgres-backed control plane. Until that step runs, marker writes fail (best-effort, rule 7) and platform dispatches degrade to counted -- the same fail-safe direction as rule 5's mid-flight edge, and the reason the PG manual step is a precondition for enabling the hook in production, not an afterthought. Real Postgres verification was NOT done on this track (no local Postgres): PG evidence here is adapter-level only (``?`` placeholders translate to ``%s`` through ``cloud.db``, DDL uses only Postgres-valid types, no autoincrement/identity column), exactly like gap 3's PG evidence, and labelled as such. Integration v2 out-of-box closure (see ``cloud/ledger_onboarding.py``): the DDL checklist, the explicit enable path and the read-only ``verify_ledger_ready`` / ``ledger_setup_report`` probe close the "hook off by default + manual PG DDL" gap WITHOUT changing the default (still OFF) or auto-running DDL on production Postgres -- the manual PG step remains an operator action, now checkable instead of assumed. This module imports ``cloud.attempt_ledger`` only, opens no database connection itself (callers pass a connection in), and contains no ``sqlite3.connect`` / ``psycopg*`` import, so the ``arch_conformance`` layer rules are untouched. """ from cloud.attempt_ledger import record_attempt MARKER_TABLE = "task_attempt_origin_markers" # One DDL serves both backends: TEXT / DOUBLE PRECISION / composite # primary key, no AUTOINCREMENT or identity column -- the same # discipline as the ledger DDL. ``queued_at`` is the ``updated_at`` # the requeue wrote; it is the generation guard, not a timestamp # for display. MARKER_DDL = ( "CREATE TABLE IF NOT EXISTS task_attempt_origin_markers(" "task_id TEXT NOT NULL PRIMARY KEY, " "origin TEXT NOT NULL, " "queued_at DOUBLE PRECISION NOT NULL)" ) # Production Postgres manual step (see rule 8). Intentionally NOT # part of POSTGRES_SCHEMA, same reason as the ledger PG_DDL. PG_MARKER_DDL = MARKER_DDL PLATFORM_ORIGIN = "platform_auto_resume" # Decision kinds returned by classify_dispatch. ``counted`` is the # only field the worker acts on; ``kind`` is the audit label. KINDS = frozenset({ "coordinator_orchestration", "platform_auto_resume", "breakpoint_resume", "human_continue", "first_or_human_retry", }) def classify_dispatch(task, platform_marker): """Pure policy: task fields + marker -> decision dict. ``task`` is the dispatch's task dict (post-claim, post breakpoint resolution: ``_goal_resume_breakpoint`` is True only when a breakpoint resume actually applies, False when resolution fell back to a fresh goal run). ``platform_marker`` is True when this dispatch consumed a platform auto-resume marker for the queued generation (rule 4), else False. Returns ``{"counted": bool, "kind": str}``. Order matters and is part of the policy: coordinator fan-out never counts even with a marker; a platform marker beats every other signal (uniform platform exclusion, rule 2); a breakpoint resume beats a continue message (a stale resume_message on a breakpoint task is not a new round); anything else with a resume message is a human continue; the rest is a first dispatch or a human retry, which count identically. """ task = task or {} if task.get("coordinator_parts"): return {"counted": False, "kind": "coordinator_orchestration"} if platform_marker: return {"counted": False, "kind": "platform_auto_resume"} if task.get("_goal_resume_breakpoint") is True or ( task.get("_resume_kind") == "breakpoint"): return {"counted": True, "kind": "breakpoint_resume"} if task.get("resume_message"): return {"counted": True, "kind": "human_continue"} return {"counted": True, "kind": "first_or_human_retry"} def ensure_marker_table(conn): """Create the marker table if absent. Idempotent; the caller owns the commit (``mark_platform_rerun`` commits via the caller's requeue transaction in the worker; a bare ensure inside a larger transaction does not commit).""" conn.execute(MARKER_DDL) def mark_platform_rerun(conn, task_id, queued_at): """Record that ``task_id`` was requeued by platform auto-resume at ``queued_at`` (the requeue's ``updated_at``). Overwrites any marker for the task: a requeue supersedes any older generation. Callers must call this inside the requeue transaction (the worker does: same connection, same commit as the status change), so a marker can never exist without its queued state, or vice versa. Best-effort at the call site: on failure the caller logs and proceeds with the requeue (rule 7) -- a missing marker degrades to counted, never to a blocked requeue.""" if not task_id: raise ValueError("task_id is required") ensure_marker_table(conn) conn.execute( "DELETE FROM task_attempt_origin_markers WHERE task_id = ?", (task_id,), ) conn.execute( "INSERT INTO task_attempt_origin_markers(task_id, origin, " "queued_at) VALUES(?, ?, ?)", (task_id, PLATFORM_ORIGIN, queued_at), ) def consume_platform_marker(conn, task_id, queued_at): """Consume the platform marker for ``task_id`` iff its ``queued_at`` still matches this dispatch's queued ``updated_at``. Returns True on a match (this dispatch IS the platform rerun). On mismatch the marker belongs to an older generation (a human action rewrote ``updated_at`` since) and is deleted as stale; returns False. When ``queued_at`` is None the generation cannot be verified, so nothing is consumed or deleted and False is returned (the dispatch then counts -- the rule-5 fail-safe direction). A missing marker table raises the backend's OperationalError (loud, not silent): the worker call site treats that as "no marker" and logs it, it never fails the dispatch.""" if not task_id or queued_at is None: return False cur = conn.execute( "DELETE FROM task_attempt_origin_markers " "WHERE task_id = ? AND queued_at = ?", (task_id, queued_at), ) if cur.rowcount == 1: return True conn.execute( "DELETE FROM task_attempt_origin_markers WHERE task_id = ?", (task_id,), ) return False def record_dispatch_result(conn, task_id, outcome): """Write the ledger row for one COUNTED dispatch. The worker calls this only when ``classify_dispatch`` said counted, and only with a terminal outcome (``success`` / ``failure`` / ``interrupted``). ``is_defect`` is always False here (rule 6): the worker never judges a defect. Returns the new ``attempt_no``. The counted filter lives in the caller (``classify_dispatch``); this function takes the outcome only, so a miscounted call site cannot smuggle a defect judgement in through here either.""" return record_attempt(conn, task_id, outcome, is_defect=False)