diff --git a/cloud/attempt_hook.py b/cloud/attempt_hook.py new file mode 100644 index 0000000..a19cbc8 --- /dev/null +++ b/cloud/attempt_hook.py @@ -0,0 +1,358 @@ +"""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. + +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) diff --git a/cloud/attempt_ledger.py b/cloud/attempt_ledger.py index 6de95ec..3b5fc9d 100644 --- a/cloud/attempt_ledger.py +++ b/cloud/attempt_ledger.py @@ -41,7 +41,22 @@ Semantics (pinned by tests, not left to each caller): ``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. + is deterministic. Corrections (below) re-evaluate this label: + the first defect row AFTER corrections is applied wins. +- A defect judgement can be CORRECTED, never rewritten: + ``record_correction`` appends a new row whose + ``corrects_attempt_no`` points at the attempt it corrects; the + old row stays untouched, and aggregation applies corrections in + attempt order with last-wins (so a wrongly marked defect can be + unmarked, and a defect missed at record time can be added after + review). A correction row is never itself counted as an + attempt, cannot change an attempt's outcome (only its defect + judgement), and cannot target an interrupted row, another + correction row, or a nonexistent attempt -- all raise + ``ValueError`` and write nothing. Who may correct is a policy + question answered in ``cloud/attempt_hook.py`` (explicit + recorder only, maintainer bar); the worker hook never writes + corrections and never marks a defect. - 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`` @@ -75,16 +90,32 @@ Persistence notes: 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. + happens only on the write path, inside ``record_attempt`` / + ``record_correction``. +- Lifecycle hook (this track, see ``cloud/attempt_hook.py``): + the worker writes exactly one row per COUNTED dispatch (see + that module's policy for which dispatches count: human/first + rounds and human breakpoint resumes count; platform auto-resume + reruns, coordinator fan-out, and pre-execution gate failures do + not). The hook is OFF by default (``attempt_ledger_hook=False``): + with it off, this module's behaviour is exactly audit gap 3 -- + the ledger fills only through explicit ``record_attempt`` / + ``record_correction`` calls, and production rates stay honestly + unknown until the hook is deliberately enabled. The worker hook + always records ``is_defect=False``: it never judges a defect, + so enabling it can never turn a transient retry into a defect. +- Schema evolution note (pre-production, this track): + ``corrects_attempt_no`` was added to the DDL after audit gap 3 + but BEFORE either DDL reached production (gap 3 is still an + OPEN, unmerged PR), so no production table needs migration: + fresh tables get the column from the DDL below. A database + created from gap 3's DDL alone (temporary/test databases only) + gains the column with one statement -- + ``ALTER TABLE task_attempts ADD COLUMN corrects_attempt_no + INTEGER`` -- which is also the documented step if gap 3's DDL + was ever run on production Postgres before this track lands. + The read path uses ``SELECT *`` deliberately so a table without + the column still reads (its rows aggregate as plain attempts). This module imports no ``cloud.*`` module and opens no database connection itself: callers pass a connection in, so the layer rules @@ -102,6 +133,11 @@ 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. +# ``corrects_attempt_no`` (NULL = a real attempt row; non-NULL = a +# correction row for that attempt_no, never counted as an attempt) +# was added by the lifecycle-hook track; it SUPERSEDES gap 3's DDL +# for any production manual step (gap 3 never merged, so no +# production table exists yet -- see the module docstring). SQLITE_DDL = ( "CREATE TABLE IF NOT EXISTS task_attempts(" "task_id TEXT NOT NULL, " @@ -110,6 +146,7 @@ SQLITE_DDL = ( "is_defect INTEGER NOT NULL DEFAULT 0, " "defect_category TEXT, " "created_at DOUBLE PRECISION, " + "corrects_attempt_no INTEGER, " "PRIMARY KEY (task_id, attempt_no))" ) @@ -117,8 +154,19 @@ SQLITE_DDL = ( # 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). +# This is the NEW DDL (with ``corrects_attempt_no``): if gap 3's +# older DDL was ever run first, follow it with +# ``ALTER TABLE task_attempts ADD COLUMN corrects_attempt_no +# INTEGER`` and verify before expecting corrections to work. PG_DDL = SQLITE_DDL +# One-line upgrade for a table created from gap 3's DDL (no +# ``corrects_attempt_no`` column yet). Valid in both backends. +OLD_TABLE_ALTER = ( + "ALTER TABLE task_attempts " + "ADD COLUMN corrects_attempt_no INTEGER" +) + def ensure_task_attempts_table(conn): """Create the ledger table if absent. Idempotent; does not @@ -153,8 +201,9 @@ def record_attempt(conn, task_id, outcome, is_defect=False, attempt_no = int(row[0]) + 1 conn.execute( "INSERT INTO task_attempts(task_id, attempt_no, outcome, " - "is_defect, defect_category, created_at) " - "VALUES(?, ?, ?, ?, ?, ?)", + "is_defect, defect_category, created_at, " + "corrects_attempt_no) " + "VALUES(?, ?, ?, ?, ?, ?, NULL)", (task_id, attempt_no, outcome, 1 if is_defect else 0, defect_category, time.time() if created_at is None else created_at), @@ -163,13 +212,87 @@ def record_attempt(conn, task_id, outcome, is_defect=False, return attempt_no +def record_correction(conn, task_id, attempt_no, is_defect, + defect_category=None, created_at=None): + """Append a CORRECTION row for an already-recorded attempt. + + The correction does not rewrite history: the target row stays + exactly as recorded, and this new row (its own ``MAX + 1`` + ``attempt_no``) carries the corrected defect judgement in + ``is_defect`` / ``defect_category`` plus + ``corrects_attempt_no = attempt_no``. Aggregation applies + corrections in attempt order, last-wins, and never counts a + correction row as an attempt. + + Only the defect judgement is correctable: the correction row + copies the target's outcome, so it cannot turn a success into + a failure or vice versa. Validates before writing: missing + ``task_id``, a nonexistent target, a target that is itself a + correction row, a target whose outcome is ``interrupted`` + (interrupted rows are never defects, so there is no defect + judgement to correct), and a defect category without a defect + flag all raise ``ValueError`` and write nothing. On a table + created from gap 3's DDL (no ``corrects_attempt_no`` column), + the INSERT raises the backend's OperationalError loudly -- + run ``OLD_TABLE_ALTER`` first (see the module docstring); + a correction must never silently degrade into a plain row. + Returns the correction row's own ``attempt_no``. + """ + if not task_id: + raise ValueError("task_id is required") + if defect_category is not None and not is_defect: + raise ValueError("defect_category requires is_defect=True") + ensure_task_attempts_table(conn) + target = conn.execute( + "SELECT outcome, corrects_attempt_no FROM task_attempts " + "WHERE task_id = ? AND attempt_no = ?", + (task_id, attempt_no), + ).fetchone() + if target is None: + raise ValueError( + "no such attempt to correct: %s #%s" % (task_id, attempt_no)) + if target["corrects_attempt_no"] is not None: + raise ValueError( + "cannot correct a correction row: %s #%s" + % (task_id, attempt_no)) + if target["outcome"] == "interrupted": + raise ValueError( + "an interrupted attempt has no defect judgement to " + "correct: %s #%s" % (task_id, attempt_no)) + row = conn.execute( + "SELECT COALESCE(MAX(attempt_no), 0) FROM task_attempts " + "WHERE task_id = ?", + (task_id,), + ).fetchone() + correction_no = int(row[0]) + 1 + conn.execute( + "INSERT INTO task_attempts(task_id, attempt_no, outcome, " + "is_defect, defect_category, created_at, " + "corrects_attempt_no) " + "VALUES(?, ?, ?, ?, ?, ?, ?)", + (task_id, correction_no, target["outcome"], + 1 if is_defect else 0, defect_category, + time.time() if created_at is None else created_at, + attempt_no), + ) + conn.commit() + return correction_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.""" + endpoint caller treats that as "no ledger", not as a failure. + + ``SELECT *`` is deliberate: a table created from gap 3's DDL + has no ``corrects_attempt_no`` column, and naming that column + here would turn those readable rows into an endpoint error. + With ``*`` the rows still read, and ``aggregate_attempts`` + treats a missing ``corrects_attempt_no`` as a plain attempt + row (see ``_field_optional``). + """ return conn.execute( - "SELECT task_id, attempt_no, outcome, is_defect, " - "defect_category FROM task_attempts " + "SELECT * FROM task_attempts " "ORDER BY task_id, attempt_no").fetchall() @@ -180,6 +303,19 @@ def _field(row, name): return row[name] +def _field_optional(row, name): + """Like ``_field`` but missing keys mean "absent", not an + error. Used ONLY for ``corrects_attempt_no``: rows built by + gap-3-era readers (plain dicts, or tables without the column) + carry no such key, and they are plain attempt rows by + definition -- failing loudly there would break every gap-3 + aggregate test for no benefit.""" + try: + return row[name] + except (KeyError, IndexError, TypeError): + return None + + def aggregate_attempts(rows): """Pure aggregation: ledger rows -> ``{task_id: record}`` where ``record`` is the ``attempts``/``defects``/``defect_category`` @@ -187,24 +323,55 @@ def aggregate_attempts(rows): 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). + attempt => its rates stay unknown, never 0). Correction rows + (``corrects_attempt_no`` set) are excluded from the attempt + count and instead OVERWRITE the defect judgement of their + target attempt, in attempt order, last-wins: a task's + ``defects`` is the number of its counted attempts whose + judgement AFTER corrections is a defect, and its + ``defect_category`` is the category of its FIRST defect + attempt in attempt order AFTER corrections. """ per_task = {} ordered = sorted( rows, key=lambda r: (_field(r, "task_id"), _field(r, "attempt_no"))) + # Per task: attempt_no -> [is_defect, defect_category], in + # attempt order (insertion order = attempt_no order, since the + # input was sorted). Corrections mutate the target entry; they + # never create an entry of their own. + judged = {} for row in ordered: task_id = _field(row, "task_id") + corrects = _field_optional(row, "corrects_attempt_no") + if corrects is not None: + entries = judged.get(task_id) + if entries is not None and corrects in entries: + entries[corrects] = [ + bool(_field(row, "is_defect")), + _field(row, "defect_category")] + continue 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") + entries = judged.setdefault(task_id, {}) + entries[_field(row, "attempt_no")] = [ + bool(_field(row, "is_defect")), + _field(row, "defect_category")] + for task_id, entries in judged.items(): + attempts = len(entries) + if attempts == 0: + continue + defects = 0 + defect_category = None + for _no in sorted(entries): + is_defect, category = entries[_no] + if is_defect: + defects += 1 + if defect_category is None: + defect_category = category + per_task[task_id] = { + "attempts": attempts, "defects": defects, + "defect_category": defect_category} return per_task diff --git a/cloud/tests/test_attempt_hook.py b/cloud/tests/test_attempt_hook.py new file mode 100644 index 0000000..4aea02e --- /dev/null +++ b/cloud/tests/test_attempt_hook.py @@ -0,0 +1,729 @@ +"""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() diff --git a/cloud/worker.py b/cloud/worker.py index a88c4a5..0b1f2a2 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -116,6 +116,19 @@ Docker 下 repos_dir 必须放在 extra-mounts 已挂载的目录下， 推送到远端不在本轮范围：任务在本地检出上执行、本地 worktree 合并仍按 host.py 现有流程，git push 需要仓库写凭证，控制面 不存、不推。 + +任务尝试台账写入钩子（见 cloud/attempt_hook.py 的完整政策， +默认关，Worker 参数 attempt_ledger_hook / CLI +--attempt-ledger-hook）：开时 run_one 给每个计数的派发写且 +只写一行台账（终态 completed -> success、failed -> failure、 +interrupted -> interrupted，is_defect 恒 False）；同一派发内 +的 max_retries 重试、fallback 换模型重跑、goal 已存在断点回 +落、后续消息排空都不另起行。平台自动续跑派发（认领时消费 +到 auto_resume_interrupted 在同一事务里写的标记）不写行； +协调员父派发不写行（子任务各自写行）；缺 key / 仓库 / 配额 / +轮次上限 / 网关等执行前门失败不写行（准入拒绝不是尝试）。 +关时 worker 不写任何台账行、不写标记、不建表，与审计加固 +补缺口第 3 项的显式填充行为完全一致。 """ import argparse @@ -134,6 +147,10 @@ import time import uuid from types import SimpleNamespace +from cloud.attempt_hook import classify_dispatch as _attempt_classify +from cloud.attempt_hook import consume_platform_marker as _attempt_consume +from cloud.attempt_hook import mark_platform_rerun as _attempt_mark +from cloud.attempt_hook import record_dispatch_result as _attempt_record from cloud.auth import AuthStore from cloud.coordinator import build_report as _coord_build_report from cloud.coordinator import DEFAULT_POLL_INTERVAL as _COORD_POLL_INTERVAL @@ -478,6 +495,7 @@ class Worker: outbox_max_attempts=3, quota_poll_interval_seconds=30, auto_resume_max=50, + attempt_ledger_hook=False, ignore_host_budget=True, worker_address=None): self.db_path = str(db_path) @@ -583,6 +601,13 @@ class Worker: # 防确定性失败（每次续跑立刻又中断）无限烧额度； # 正常部署重启/偶发中断远达不到 50 次。 self.auto_resume_max = auto_resume_max + # 任务尝试台账写入钩子（见 cloud/attempt_hook.py 的完整 + # 政策）：默认关。关时 worker 不写任何台账行、不写平台 + # 标记、不建台账/标记表，行为与审计加固补缺口第 3 项 + # （显式 record_attempt 填充）完全一致；开时 run_one + # 按政策给计数的派发写且只写一行（is_defect 恒 False， + # worker 永不判定缺陷），平台自动续跑派发不写行。 + self.attempt_ledger_hook = bool(attempt_ledger_hook) # 项目仓库检出根目录：CLI --repos-dir 可调（默认 ./repos）； # Docker 执行器下它必须放在 extra-mounts 已挂载的目录下， # 否则容器内 task["repo"] 路径不存在——见 cloud/README.md。 @@ -1124,6 +1149,23 @@ class Worker: resumed.append(task_id) self._audit(project_id, "worker_auto_resumed", task_id) + # 台账钩子（仅开时）：同一事务里写平台 + # 标记，queued_at 取本次重排队写的 + # updated_at（即 now），下次认领时凭它 + # 识别这就是平台续跑派发（见 + # attempt_hook.py 规则 4）。标记写失败 + # 绝不挡重排队：只记日志、本次重排队照 + # 常提交，派发退化为计数（ fail-safe + # 方向，见规则 5 的 mid-flight 说明）。 + if self.attempt_ledger_hook: + try: + _attempt_mark(conn, task_id, now) + except Exception as exc: + log_event( + self.logger, + "worker_attempt_marker_failed", + message="attempt marker failed: %s" + % exc) if is_goal: summary = ("自动续跑（无进展计数 %d，" "下次退避 %d 秒）" @@ -5608,6 +5650,26 @@ class Worker: self.usage_cost_usd += micros / 1000000.0 return cost_id + def _record_ledger_attempt(self, task_id, outcome, counted): + """台账钩子唯一写行入口（best-effort，见 attempt_hook.py + 规则 7）：关时或不计数时直接返回，不碰数据库、不建 + 表；开且计数时写一行（is_defect 恒 False）。写失败 + 只记日志，绝不影响任务状态与返回值。调用方须只在 + run_one 的终局点（取消分支、正常终局块）调用，门失败 + 与协调员分支故意不调。""" + if not self.attempt_ledger_hook or not counted: + return None + try: + conn = _db_connect(self.db_path) + try: + return _attempt_record(conn, task_id, outcome) + finally: + conn.close() + except Exception as exc: + log_event(self.logger, "worker_attempt_ledger_failed", + message="attempt ledger write failed: %s" % exc) + return None + def run_one(self, task=None): """认领并执行一个任务，返回结果 dict（见模块 docstring）。 @@ -5621,6 +5683,25 @@ class Worker: if task is None: return {"dispatched": False, "reason": "no_ready_task"} task_id = task["id"] + # 台账钩子：认领前抓排队态的 updated_at（认领会改写 + # 它）。仅开时用：凭它在认领后消费平台标记（见 + # attempt_hook.py 规则 4）；任务字典里没有时（调用方 + # 自建字典）认领前直读一次库，读不到则为 None（标记 + # 无法验证 → 派发按计数处理，fail-safe 方向）。 + _ledger_queued_at = task.get("updated_at") + if self.attempt_ledger_hook and _ledger_queued_at is None: + try: + _qconn = _db_connect(self.db_path) + try: + _qrow = _qconn.execute( + "SELECT updated_at FROM tasks WHERE id = ?", + (task_id,)).fetchone() + if _qrow is not None: + _ledger_queued_at = _qrow[0] + finally: + _qconn.close() + except Exception: + _ledger_queued_at = None if not self._claim(task_id, executor=self._executor_name()): return {"dispatched": False, "reason": "already_claimed", "task_id": task_id} @@ -5637,6 +5718,28 @@ class Worker: except Exception: pass task = self.get_task(task_id) or task + # 台账钩子：认领成功后立即消费平台标记（仅开时）。 + # 认领已赢才消费：认领失败的分支在上面已返回，标记 + # 留给赢的 worker；queued_at 不匹配时消费函数删掉 + # 陈旧标记并返回 False（见 attempt_hook.py 规则 4）。 + # 标记表不存在等错误只记日志、不挡派发（派发退化为 + # 计数，fail-safe 方向）。 + _ledger_platform = False + if self.attempt_ledger_hook: + try: + _mconn = _db_connect(self.db_path) + try: + _ledger_platform = _attempt_consume( + _mconn, task_id, _ledger_queued_at) + _mconn.commit() + finally: + _mconn.close() + except Exception as exc: + log_event(self.logger, + "worker_attempt_marker_failed", + message="attempt marker consume failed: %s" + % exc) + _ledger_platform = False # G10：goal 断点续跑标记——认领后立即清掉库里标记（本轮 # 已认领为 running，不会重复触发；本轮若再被杀回 # interrupted，用户需重新调 /goal-resume 才会再置标记， @@ -5698,6 +5801,19 @@ class Worker: else: task["prompt"] = resume_message task["_resume"] = True + # 台账钩子：分类必须放在断点解析与续跑消息处理之后， + # 因为分类读的是解析后的字段：_goal_resume_breakpoint + # 为 False 表示断点解析已回落成全新 goal（那就是一轮 + # 新尝试，不是断点续跑）；_resume_kind == "breakpoint" + # 表示旧验收 goal 的续跑消息沿用原循环（同轮续跑）。 + # 关时 _ledger_counted 恒 False，后面所有写行点都 + # 是空操作；门失败提前返回（缺 key / 仓库 / 配额 / + # 轮次上限 / 网关）故意不写行（见 attempt_hook.py + # 规则 2：准入拒绝不是尝试），协调员分支同理。 + _ledger_decision = _attempt_classify( + task, _ledger_platform) if self.attempt_ledger_hook \ + else {"counted": False, "kind": "hook_disabled"} + _ledger_counted = bool(_ledger_decision["counted"]) # 自动记忆注入（官方 use_memories）：首轮才注入，见 # _apply_auto_memory_injection 的跳过规则；只改内存副本。 task = self._apply_auto_memory_injection(task) @@ -5872,6 +5988,11 @@ class Worker: # 认领不了（2026-10-28 40 人与 18 人旧轮的第 4 槽/C-6 # 未认领即此漏）。finally 清槽，异常路径同样不漏。 if task.get("coordinator_parts"): + # 台账钩子：协调员父派发不写行（attempt_hook.py + # 规则 2：父任务只做扇出聚合、无模型回合，子任务 + # 各自经自己的 run_one 写行）。这里直接返回，刻 + # 意绕过下面的终局写行点；_ledger_counted 对协调 + # 员恒 False， defensive 双保险。 try: return self._run_coordinator(task, started_at) finally: @@ -6208,6 +6329,10 @@ class Worker: _cancel_cost_id = None log_event(self.logger, "worker_cancelled", task_id=task_id, project_id=task["project_id"], status="interrupted") + # 台账钩子：取消派发写 interrupted 行（不计分母， + # 只留档）；不计数派发（平台/协调员）时为空操作。 + self._record_ledger_attempt(task_id, "interrupted", + _ledger_counted) return {"dispatched": True, "task_id": task_id, "host_name": self.executor._host_name(task) if hasattr(self.executor, "_host_name") else task_id, @@ -6360,6 +6485,21 @@ class Worker: # 补一条 worker_completed/failed 审计误导排查。 final = "interrupted" self._audit(task["project_id"], "worker_interrupted", task_id) + # 台账钩子：本派发唯一正常写行点（取消分支已单独 + # 写，协调员与门失败分支不写）。终态映射：completed + # -> success，failed -> failure，interrupted -> + # interrupted；/goal 转换时 final 已被改写成 queued， + # 但主轮已完成，记 success；final 落在三个终态之外 + # 时不写（不猜测 outcome，宁缺勿错）。 + if goal_converted: + self._record_ledger_attempt(task_id, "success", + _ledger_counted) + elif final in ("completed", "failed", "interrupted"): + self._record_ledger_attempt( + task_id, + {"completed": "success", "failed": "failure", + "interrupted": "interrupted"}[final], + _ledger_counted) if resume_message and not quota_stop: # 本轮已跑完（无论成败）：续跑消息已消费，清掉；进程 # 中途被杀时清不到，下次认领会重跑同一条消息——可接受 @@ -7130,6 +7270,14 @@ def main(argv=None): help="长任务自动恢复上限（次数，非时间，默认 50）：" "同一 goal 任务被系统中断后自动续跑超过该次数" "即停住等人工，防确定性失败无限烧额度") + parser.add_argument("--attempt-ledger-hook", action="store_true", + help="任务尝试台账写入钩子（默认关）：开时 worker" + " 按 cloud/attempt_hook.py 的政策给计数的派发" + " 写且只写一行台账（is_defect 恒 False），平台" + " 自动续跑派发不写行；关时行为与审计加固补缺口" + " 第 3 项完全一致（只经显式 record_attempt 填充）。" + " 生产 PG 上启用前须先手动执行台账 PG_DDL 与" + " 标记表 PG_MARKER_DDL 并验证") parser.add_argument("--loop", action="store_true", help="常驻循环，每 --interval 秒跑一轮") parser.add_argument("--interval", type=int, default=300) @@ -7164,6 +7312,7 @@ def main(argv=None): max_retries=args.max_retries, retry_backoff_seconds=args.retry_backoff_seconds, auto_resume_max=args.auto_resume_max, + attempt_ledger_hook=args.attempt_ledger_hook, repos_dir=args.repos_dir, heartbeat_file=args.heartbeat_file, worker_id=args.worker_id, diff --git a/docs/arch-conformance-metrics.md b/docs/arch-conformance-metrics.md index b994c6f..83ecbf8 100644 --- a/docs/arch-conformance-metrics.md +++ b/docs/arch-conformance-metrics.md @@ -60,8 +60,10 @@ workflow——第 2 轨拥有 lint/类型/安扫，本轨不碰。 `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` 递增、只追加不改写。口径： +defect_category, created_at, corrects_attempt_no)`，主键 +`(task_id, attempt_no)`，`attempt_no` 按任务 `MAX + 1` +递增、只追加不改写（`corrects_attempt_no` 为 NULL 时是真 +尝试行，非 NULL 时是纠正行，见钩子一节）。口径： - `outcome` ∈ success / failure / interrupted；`interrupted` 行只留档，不计入 attempts 也不计入 defects（平台中断自动 @@ -82,13 +84,81 @@ defect_category, created_at)`，主键 `(task_id, attempt_no)`， PG，真实 PG 验证未做；PG 证据目前只有适配层级别：`?` 占位 经 `cloud.db` 适配翻译为 `%s`、DDL 无需序列处理（测试 `TestPostgresAdapterEvidence` 钉住）。 -- 未做项（有意不做，原因见模块 docstring）：生命周期写 - 入钩子未接——worker 重试/续跑循环里哪一次算一次尝试、谁 - 判定 `is_defect`，接错会把平台自动续跑算成尝试、把瞬时 - 重试算成缺陷，正好违反本轨钉住的规则；在此之前台账只 - 经显式 `record_attempt` 调用填充，生产率在写钩子落地前 - 保持诚实未知，不自动填数。读端点不建表（只读端点不改 - schema），建表只发生在写路径。 +- 生命周期写入钩子已接（9+ 第 2 项，见下一节）：worker + 按政策给计数的派发写且只写一行，开关默认关；关时台账 + 仍只经显式 `record_attempt` / `record_correction` 调用填 + 充，生产率保持诚实未知。读端点不建表（只读端点不改 + schema），建表只发生在写路径；标记表同理，只在钩子开 + 且平台自动续跑发生时写。 + +## 任务尝试生命周期写入钩子（9+ 第 2 项，政策与开关） + +权威政策在 `cloud/attempt_hook.py` 模块 docstring（代码、 +本文档、测试三处一致，冲突时以模块为准）。要点： + +- 计为一次尝试的（每派发恰好一行，`is_defect` 恒 + `False`）：任务首派发；人工 `/retry`（与首派发无法区分、 + 也不必区分，两者都算）；人工 `/continue` 新消息轮次 + （含开新目标循环的 goal 任务）；人工断点续跑 + （`/goal-resume`、旧验收 goal 的 `/continue` 断点语义） + ——断点续跑算同一逻辑轮次的行，不是额外尝试：被中断 + 的片段只写了不计数的 `interrupted` 行，续跑终局行就是 + 该轮次唯一计数行；续跑再中断则仍只写 `interrupted` + 行，该轮次继续保持未知。 +- 不计为尝试的（不写行）：平台自动续跑派发（goal 断点续 + 跑与非 goal 全新跑一律不计——统一口径，平台杀掉又重跑 + 的执行不罚 agent 质量分）；协调员父派发（只做扇出聚 + 合、无模型回合，子任务各自写行）；执行前门失败（缺 + key / 仓库解析 / 配额门 / 轮次上限 / 网关准备，准入拒绝 + 不是尝试）；未派发（预算门、无就绪任务、认领输给别的 + worker，连分类都不跑）。 +- 同一派发内的瞬时重试不是新尝试：`max_retries` 循环、 + 一次真实 fallback 换模型重跑、goal“已存在”断点回落、 + 后续消息排空都在一次 `run_one` 里，共享成本汇总、只 + 产一行（终局 outcome）。内层失败后重试成功不留失败 + 行；重试耗尽只留一行失败行。 +- 平台识别靠标记表 `task_attempt_origin_markers`：自动续 + 跑在同一事务里写标记（`queued_at` = 重排队写的 + `updated_at`），认领方在认领前抓排队态 `updated_at`、 + 认领成功后按 `queued_at` 匹配消费；人为动作（取消/ + 重试/继续）改写 `updated_at` 后陈旧标记自动作废，绝不 + 误伤后一轮人工派发；认领输了不消费，标记留给赢家。 + 标记写失败绝不挡重排队（退化为计数，见模块规则 5）。 +- 开关：`Worker(attempt_ledger_hook=False)` / CLI + `--attempt-ledger-hook`，默认关。关时 worker 不写任何 + 台账行、不写标记、不建台账/标记表，与补缺口第 3 项行 + 为完全一致；开时才适用本节政策。开前中途（钩子关时已 + 重排队的任务无标记）有一次性的退化：该派发按人工/首 + 轮计数，之后每次重排队都有标记，即自愈。 +- `is_defect` 判定主体：只有显式记录方（人工维护者或协 + 调员复核）在记录时判定，worker 钩子永不判定（失败 ≠ + 缺陷，瞬时失败尤其不是）。标错只追加纠正行 + （`record_correction`：新行 `corrects_attempt_no` 指 + 向原尝试，旧行不动；聚合按 attempt_no 顺序后赢者通 + 吃；纠正行本身不计尝试、不能改 outcome、不能指中断 + 行/纠正行/不存在的尝试，违例写时 `ValueError`）。 + 维护者门槛与 `/retry` 同款（`approve_merge`），但本轨 + 不建标记/纠正的 HTTP 端点（函数即接口，端点是后续项， + 见未做项）。 +- 取舍（明写）：仅经平台派发终局的任务在台账里保持未 + 知（无计数行），分母只含人工/首轮终局的任务——率值描 + 述的是“人工/首轮终局任务”，不是全部终局任务；平台重 + 跑本身未丢失（`auto_resume_count`、审计 + `worker_auto_resumed`、时间线 `auto_resume` 行）。被拒 + 的替代方案是“单独标记平台行但不进分母”：分母结果相 + 同，但多一种行、多一列、多一套聚合排除逻辑，只换来 + 台账内同址可查——本轨不取。另：平台 goal 续跑与人工 + goal 续跑同属一轮循环，但前者不计、后者计（标记优于 + 断点），这是有意的统一平台排除；若用户要把平台 goal + 续跑也算作轮次完成，改 `classify_dispatch` 一行 + 测 + 试即可，属用户拍板项（见待人工项）。 +- Postgres：标记表与台账表同款手动纪律——不进 + `POSTGRES_SCHEMA`，`PG_MARKER_DDL`（与台账新 `PG_DDL` + 含 `corrects_attempt_no` 列）须在生产 PG 上手动执行并 + 验证后，才可在 PG 控制面启用钩子；未执行时标记写失 + 败（best-effort），平台派发退化为计数。本机无 PG，真 + 实 PG 验证未做；PG 证据只有适配层（`?`→`%s` 翻译、 + DDL 无自增/identity 列，测试钉住）。 ## 回流与审计标准迭代（可测试规则，非口号） @@ -138,7 +208,18 @@ defect_category, created_at)`，主键 `(task_id, attempt_no)`， 待人工项（单独列）：金任务集审定清单的四项人工确认；抽查 seed 的记录与保管；生产抽查比例拍板（默认值 0.10/0.50 只是测试初值）。 -台账待人工项（补缺口第 3 项之后）：生产 PG 上手动执行 -`task_attempts` 的 `PG_DDL` 并验证；生命周期写入钩子拍板—— -哪次执行算一次尝试、谁判定 `is_defect`（在拍板前台账只经 -显式调用填充，生产率保持未知）。 +台账待人工项（钩子落地之后）：生产 PG 上手动执行台账新 +`PG_DDL`（含 `corrects_attempt_no` 列；若曾跑过补缺口第 3 +项的旧 DDL，先跑 `OLD_TABLE_ALTER`）与标记表 +`PG_MARKER_DDL` 并验证（`\d task_attempts` / +`task_attempt_origin_markers` 或 information_schema 等价查询）， +验证前不在 PG 控制面启用钩子；钩子政策默认值拍板——平台 +goal 续跑不计（本轨默认）还是计为轮次完成、平台非 goal +全新跑不计（本轨默认）是否接受、缺陷标记/纠正的 HTTP +端点（`approve_merge` 门）是否立项；金任务集审定后报告才 +可决策（现仍恒为 `pending_human_approval`）。 +未做项（本轨有意不做）：缺陷标记/纠正的 HTTP 端点（函数即 +接口，权限检查随端点做）；worker 之外的写行点（服务端 +`/retry` / `/continue` / `/goal-resume` 本身不写行——写行 +只发生在 worker 派发终局，服务端重排队不直接产生尝试）； +真实 PG 验证（本机无 PG）；生产部署（只到 PR OPEN）。