fix: preserve verdict state and skip cancelled preview gates

.github/workflows/pr-preview.yml

@@ -257,7 +257,12 @@ jobs:
       - prepare-preview-database
       - deploy-preview-backend
       - update-vercel-preview
-    if: always() && github.event.action != 'closed'
+    # `!cancelled()` rather than `always()`: the gate must still run when
+    # upstream jobs are SKIPPED (fork and promotion paths), but a run that
+    # cancel-in-progress superseded must not publish a failing check and a
+    # failing deployment status for a commit whose replacement run is still
+    # building.
+    if: "!cancelled() && github.event.action != 'closed'"
     runs-on: ubuntu-latest
     # No job-level `environment:` key here: GitHub attributes that deployment
     # record to the person who triggered the run, so pull requests showed a

AGENTS.md

@@ -160,7 +160,12 @@ High-level flow:
    handles; a task is failed only when no other live trial remains. If quota
    cancellation interrupts a replacement QA pass, the last successful verdict
    is restored through `cancel_verdict`; a terminal QA failure instead clears
-   that preserved payload through `fail_verdict`.
+   that preserved payload through `fail_verdict`. All task verdict-column
+   mutations go through `oddish.core.verdict_state`: a published payload may
+   coexist with QUEUED/RUNNING while its replacement is active, but it must
+   return to SUCCESS if that pass is abandoned. The
+   `ck_tasks_published_verdict_status` database constraint rejects a published
+   payload with a missing or FAILED status.
 6. Trial completion persists queryable execution metrics on the trial row:
    input/cache/output tokens, total trajectory steps, native runtime cost when
    reported, phase timing, trajectory availability, arbitrary verifier
@@ -221,6 +226,9 @@ block. Editing a prompt is a code change that ships with a deploy.
   `TagProjectJobHandler`, plus the legacy `AnalysisJobHandler`)
 - the task-level QA job (`run_task_qa_job`): classify every live trial via
   the shared `classify_trial_and_store`, then synthesize the task verdict
+- the verdict state machine (`oddish.core.verdict_state`), which is the only
+  writer for `tasks.verdict*` lifecycle columns and preserves the last
+  published result until a replacement succeeds or terminally fails
 - post-trial classification runs through `AnalyzerBlock`. It reads two
   already-downloaded directories and executes nothing, so `resolve_substrate`
   keeps it on the worker-local Claude Code client (`CLAUDE_CLI`) everywhere;

CHANGELOG.md

@@ -38,7 +38,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Fixed
 
-- Quota cancellation no longer turns a preserved accepted verdict into a failed verdict, or leaves a failed task paired with that accepted payload. Cancelling a replacement QA pass restores the previous successful verdict; a genuine terminal QA failure clears the superseded payload.
+- Quota cancellation, retry, and append reconciliation no longer hide a preserved accepted verdict by leaving its payload paired with a missing status. Verdict lifecycle changes now use one state-transition module: replacement QA retains the published payload while queued/running, cancellation or a no-op restores it to `SUCCESS`, and only terminal QA failure discards it. A database constraint repairs and prevents invalid payload/status pairs.
 - Worker heartbeats used to stop as soon as the agent finished, but the worker still had to upload and save the results. When that took over 15 minutes, the cleanup sweep marked the trial "Worker heartbeat stalled for over 15 minutes", threw away the finished result, and re-ran the whole trial. The heartbeat now runs until the results are saved and settled.
 
 ---

oddish/alembic/versions/verdict_state_001_enforce_published_verdict.py

@@ -0,0 +1,77 @@
+"""Enforce the published-verdict state invariant.
+
+``tasks.verdict`` stores the last successfully published result even while a
+replacement QA pass is queued or running. A payload without a status makes the
+result disappear from status-based readers; a payload paired with FAILED keeps
+a result that the failed replacement should have invalidated.
+
+Revision ID: verdict_state_001
+Revises: drop_prompt_registry_001
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "verdict_state_001"
+down_revision: Union[str, Sequence[str], None] = "drop_prompt_registry_001"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+_CONSTRAINT = "ck_tasks_published_verdict_status"
+
+
+def _constraint_exists(bind: sa.engine.Connection) -> bool:
+    return _CONSTRAINT in {
+        item["name"] for item in sa.inspect(bind).get_check_constraints("tasks")
+    }
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if _constraint_exists(bind):
+        return
+
+    # NOT VALID starts protecting concurrent writes without first scanning the
+    # table. Repair historical rows under that protection, then validate.
+    # SQLAlchemy's JSONB type writes Python None as JSON null, so both null
+    # representations are unpublished and valid.
+    op.execute(
+        f"""
+        ALTER TABLE tasks
+        ADD CONSTRAINT {_CONSTRAINT}
+        CHECK (
+            verdict IS NULL OR verdict = 'null'::jsonb OR
+            (verdict_status IS NOT NULL AND verdict_status <> 'FAILED')
+        ) NOT VALID
+        """
+    )
+    op.execute(
+        """
+        UPDATE tasks
+        SET verdict_status = 'SUCCESS',
+            verdict_error = NULL,
+            verdict_started_at = NULL
+        WHERE verdict IS NOT NULL
+          AND verdict <> 'null'::jsonb
+          AND verdict_status IS NULL
+        """
+    )
+    op.execute(
+        """
+        UPDATE tasks
+        SET verdict = NULL
+        WHERE verdict IS NOT NULL
+          AND verdict <> 'null'::jsonb
+          AND verdict_status = 'FAILED'
+        """
+    )
+    op.execute(f"ALTER TABLE tasks VALIDATE CONSTRAINT {_CONSTRAINT}")
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if _constraint_exists(bind):
+        op.drop_constraint(_CONSTRAINT, "tasks", type_="check")

oddish/src/oddish/core/endpoints/_common.py

@@ -5,6 +5,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
+from oddish.core.verdict_state import reset_verdict
 from oddish.db import ExperimentModel, TaskModel, TrialModel
 
 USER_CANCELLED_MESSAGE = "Cancelled by user"
@@ -86,9 +87,5 @@ async def get_trial_for_org_core(
 
 
 def _reset_task_verdict(task: TaskModel) -> None:
-    """Clear cached verdict state before re-running analysis or verdict."""
-    task.verdict = None
-    task.verdict_status = None
-    task.verdict_error = None
-    task.verdict_started_at = None
-    task.verdict_finished_at = None
+    """Discard the published verdict and all of its lifecycle metadata."""
+    reset_verdict(task)

oddish/src/oddish/core/endpoints/qa.py

@@ -10,9 +10,8 @@
 from oddish.core.endpoints._common import (
     USER_CANCELLED_MESSAGE,
     _ACTIVE_WORKER_JOB_STATUSES_SQL,
-    _reset_task_verdict,
 )
-from oddish.core.verdict_sync import cancel_verdict
+from oddish.core.verdict_state import cancel_verdict, queue_verdict
 from oddish.db import (
     AnalysisStatus,
     TaskModel,
@@ -148,9 +147,7 @@ async def cancel_task_qa_core(
     # these). Left alive, one would flip the cancelled analysis back to
     # QUEUED on claim and overwrite the cancelled state.
     live_trial_ids = [
-        trial.id
-        for trial in task.trials or []
-        if trial.superseded_by_trial_id is None
+        trial.id for trial in task.trials or [] if trial.superseded_by_trial_id is None
     ]
     analysis_rows = await _cancel_worker_jobs_for_kind(
         session,
@@ -209,9 +206,7 @@ async def cancel_task_qa_core(
         if pinned:
             version_ids.add(str(pinned))
     for version_id in version_ids:
-        version = await session.get(
-            TaskVersionModel, version_id, with_for_update=True
-        )
+        version = await session.get(TaskVersionModel, version_id, with_for_update=True)
         if version is not None and version.pre_trial_status in (
             VerdictStatus.PENDING,
             VerdictStatus.QUEUED,
@@ -268,9 +263,9 @@ async def rerun_task_qa_core(
 ) -> dict[str, str | int]:
     """(Re)run the single task-level QA job for a finished task.
 
-    Resets every live trial's classification and the task verdict, then
-    enqueues one QA job that re-classifies all live trials and synthesizes a
-    fresh verdict.
+    Resets every live trial's classification, then enqueues one QA job that
+    re-classifies all live trials and synthesizes a fresh verdict. The current
+    published verdict remains visible until that job replaces it.
     """
     return await backfill_task_analysis_core(
         session,
@@ -293,9 +288,8 @@ async def backfill_task_analysis_core(
 ) -> dict[str, str | int]:
     """(Re)run task-level QA to backfill trial analysis.
 
-    Resets the task verdict (so the QA job runs instead of short-circuiting
-    on a terminal verdict) and enqueues one QA job. The QA job is idempotent
-    at trial granularity, so:
+    Queues a replacement verdict without withdrawing the published result.
+    The QA job is idempotent at trial granularity, so:
 
     * ``force=False`` resets no trial analyses -> only genuinely-missing
       trials are (re)classified;
@@ -439,12 +433,11 @@ def _analysis_in_progress(trial: TrialModel) -> bool:
             _reset_trial_analysis(trial)
             reset_count += 1
 
-    _reset_task_verdict(task)
     if enable_analysis:
         task.run_analysis = True
     task.status = TaskStatus.VERDICT_PENDING
     task.finished_at = None
-    task.verdict_status = VerdictStatus.QUEUED
+    queue_verdict(task)
 
     from oddish.queue import enqueue_qa_worker_job
 

oddish/src/oddish/core/endpoints/trials.py

@@ -9,7 +9,6 @@
 from oddish.core.endpoints._common import (
     get_trial_for_org_core,
 )
-from oddish.core.verdict_sync import clear_inflight_verdict
 from oddish.core.endpoints.qa_cost import get_trial_qa_costs
 from oddish.core.helpers import (
     build_trial_response,
@@ -22,6 +21,7 @@
     read_trial_result,
     read_trial_trajectory,
 )
+from oddish.core.verdict_state import abandon_verdict
 from oddish.db import (
     AnalysisStatus,
     TaskModel,
@@ -133,9 +133,7 @@ async def rerun_trial_analysis_core(
         raise HTTPException(status_code=404, detail=f"Trial {trial_id} not found")
     trial = (
         await session.execute(
-            select(TrialModel)
-            .where(TrialModel.id == trial_id)
-            .with_for_update()
+            select(TrialModel).where(TrialModel.id == trial_id).with_for_update()
         )
     ).scalar_one_or_none()
     if trial is None:
@@ -166,7 +164,8 @@ async def rerun_trial_analysis_core(
     # A failed analysis can leave its job in RETRYING. Enqueuing another job
     # then would run the analysis twice. The existing job serves the re-run.
     active_job = await session.scalar(
-        select(WorkerJobModel.id).where(
+        select(WorkerJobModel.id)
+        .where(
             WorkerJobModel.kind == WorkerJobKind.ANALYSIS,
             WorkerJobModel.subject_table == "trials",
             WorkerJobModel.subject_id == trial_id,
@@ -191,13 +190,13 @@ async def rerun_trial_analysis_core(
     # A QUEUED full QA job is fine: it classifies this trial itself when it
     # starts, and the per-trial claim keeps the two jobs from colliding.
     running_task_qa = await session.scalar(
-        select(WorkerJobModel.id).where(
+        select(WorkerJobModel.id)
+        .where(
             WorkerJobModel.kind == WorkerJobKind.QA,
             WorkerJobModel.subject_table == "tasks",
             WorkerJobModel.subject_id == trial.task_id,
             WorkerJobModel.status == WorkerJobStatus.RUNNING,
-            func.coalesce(WorkerJobModel.payload["mode"].astext, "full")
-            != "pre_trial",
+            func.coalesce(WorkerJobModel.payload["mode"].astext, "full") != "pre_trial",
         )
         .limit(1)
     )
@@ -275,9 +274,7 @@ async def _waiting_job(kind: WorkerJobKind, subject_table: str, subject_id: str)
                     WorkerJobModel.queue_key,
                     WorkerJobModel.priority,
                     WorkerJobModel.created_at,
-                    (WorkerJobModel.available_after > func.now()).label(
-                        "in_backoff"
-                    ),
+                    (WorkerJobModel.available_after > func.now()).label("in_backoff"),
                 )
                 .where(
                     WorkerJobModel.kind == kind,
@@ -596,7 +593,7 @@ async def retry_trial_core(
     ):
         task.status = TaskStatus.RUNNING
         task.finished_at = None
-    clear_inflight_verdict(task)
+    abandon_verdict(task)
     await session.execute(
         text(
             """

oddish/src/oddish/core/quota_enforcement.py

@@ -25,7 +25,11 @@
     sum_org_cost_usd,
     try_acquire_quota_locks,
 )
-from oddish.core.verdict_sync import cancel_verdict
+from oddish.core.verdict_state import (
+    cancel_verdict,
+    has_active_verdict,
+    has_published_verdict,
+)
 from oddish.db import (
     AnalysisStatus,
     TaskModel,
@@ -342,11 +346,7 @@ async def _reconcile_cancelled_tasks(
     for task in tasks:
         if task.id not in exhausted:
             continue
-        if task.verdict or task.verdict_status in (
-            VerdictStatus.PENDING,
-            VerdictStatus.QUEUED,
-            VerdictStatus.RUNNING,
-        ):
+        if has_published_verdict(task) or has_active_verdict(task):
             cancel_verdict(task, error=QUOTA_CANCELLED_MESSAGE, now=now)
         if task.status in _ACTIVE_TASK_STATUSES:
             restored = task.verdict_status == VerdictStatus.SUCCESS