diff --git a/apps/labeler/src/assessment-orchestrator.ts b/apps/labeler/src/assessment-orchestrator.ts index f577d1f146..1384622274 100644 --- a/apps/labeler/src/assessment-orchestrator.ts +++ b/apps/labeler/src/assessment-orchestrator.ts @@ -294,6 +294,10 @@ export class AssessmentOrchestrator { // subject after the currency re-check below no-ops this CAS at commit // time, so no outcome/block label commits for a deleted subject. guardSubjectNotDeleted: true, + // Halt an in-flight run whose automation kill-switch was paused after + // dispatch: the CAS (and every label gated on `toState`) no-ops, leaving + // the run `running` for the Workflow retry to re-run after resume. + guardAutomationNotPaused: true, }); const statements = [...finalization.statements]; diff --git a/apps/labeler/src/assessment-store.ts b/apps/labeler/src/assessment-store.ts index 6b55b6d553..6b28b705bb 100644 --- a/apps/labeler/src/assessment-store.ts +++ b/apps/labeler/src/assessment-store.ts @@ -621,6 +621,16 @@ export interface FinalizationInput { * finalization retries and stales out instead of labelling a deleted subject. */ guardSubjectNotDeleted?: boolean; + /** + * Gate the CAS on the automation kill-switch reading `paused = 0` at commit + * time. An admin pausing automation after a run was dispatched otherwise lets + * the in-flight run commit its automated labels; folding the predicate into the + * CAS (and, transitively, every label gated on `toState`) no-ops finalization + * under a pause, leaving the run `running` for the Workflow retry to re-run + * after resume. Fails closed like `isAutomationPaused`: a missing singleton row + * fails the `paused = 0` read and halts finalization. + */ + guardAutomationNotPaused?: boolean; } export interface FinalizationStatements { @@ -682,6 +692,12 @@ export function buildFinalizationStatements( ? `\n\t\t\t\t AND EXISTS (SELECT 1 FROM subjects WHERE uri = ? AND cid = ? AND deleted_at IS NULL)` : ""; if (input.guardSubjectNotDeleted) casBinds.push(input.uri, input.cid); + // Fail-closed automation kill-switch: `EXISTS (… paused = 0)` no-ops the CAS + // (and every label gated on `toState`) when automation is paused or the + // singleton row is missing. + const automationGuardSql = input.guardAutomationNotPaused + ? `\n\t\t\t\t AND EXISTS (SELECT 1 FROM automation_state WHERE id = 1 AND paused = 0)` + : ""; const statements: D1PreparedStatement[] = [ db .prepare( @@ -690,7 +706,7 @@ export function buildFinalizationStatements( public_summary = COALESCE(?, public_summary), coverage_json = COALESCE(?, coverage_json), supersedes_assessment_id = COALESCE(?, supersedes_assessment_id) - WHERE id = ? AND state = ?${signingGuardSql}${subjectGuardSql}`, + WHERE id = ? AND state = ?${signingGuardSql}${subjectGuardSql}${automationGuardSql}`, ) .bind(...casBinds), ]; @@ -706,8 +722,15 @@ export function buildFinalizationStatements( ON CONFLICT(src, uri, cid) DO UPDATE SET assessment_id = excluded.assessment_id, updated_at = excluded.updated_at WHERE EXISTS (SELECT 1 FROM assessments WHERE id = ? AND state = ?) - AND (SELECT created_at_epoch_ms FROM assessments WHERE id = excluded.assessment_id) - >= (SELECT created_at_epoch_ms FROM assessments WHERE id = current_assessments.assessment_id)`, + AND ( + (SELECT created_at_epoch_ms FROM assessments WHERE id = excluded.assessment_id) + > (SELECT created_at_epoch_ms FROM assessments WHERE id = current_assessments.assessment_id) + OR ( + (SELECT created_at_epoch_ms FROM assessments WHERE id = excluded.assessment_id) + = (SELECT created_at_epoch_ms FROM assessments WHERE id = current_assessments.assessment_id) + AND excluded.assessment_id >= current_assessments.assessment_id + ) + )`, ) .bind( input.src, diff --git a/apps/labeler/src/assessment-workflow.ts b/apps/labeler/src/assessment-workflow.ts index f0d91fef1c..dda09d10ff 100644 --- a/apps/labeler/src/assessment-workflow.ts +++ b/apps/labeler/src/assessment-workflow.ts @@ -44,6 +44,7 @@ import { type CoverageAccumulator, } from "./assessment-stages.js"; import { getAssessment, type Assessment } from "./assessment-store.js"; +import { AutomationPausedError, isAutomationPaused } from "./automation-state.js"; import type { AiBinding } from "./code-ai-adapter.js"; import { getLabelerIdentityConfig, type LabelerConfig } from "./config.js"; import type { PublisherVerificationReader } from "./history-context.js"; @@ -94,6 +95,14 @@ export async function executeAssessmentInstance( return existing.state; } + // Re-read the kill-switch on entry: an admin pausing automation after this run + // was dispatched halts it here, before it spends AI/network work. Throwing + // leaves the row untouched for the Workflow-step retry to resume once + // automation is unpaused; finalization carries the same guard as the backstop. + // `isAutomationPaused` fails closed — an unreadable switch throws and retries. + if (await isAutomationPaused(env.DB)) + throw new AutomationPausedError(`assessment ${assessmentId} halted: automation is paused`); + assertRequiredBindings(env); const config = await getLabelerIdentityConfig(env); const versioned = await createRuntimeSigner(config, getRuntimeSigningSecret(env)); diff --git a/apps/labeler/src/automation-state.ts b/apps/labeler/src/automation-state.ts index c57438e2b3..beb49b752e 100644 --- a/apps/labeler/src/automation-state.ts +++ b/apps/labeler/src/automation-state.ts @@ -14,6 +14,15 @@ export class AutomationStateUnavailableError extends Error { override readonly name = "AutomationStateUnavailableError"; } +/** + * Raised when an in-flight assessment run re-reads the kill-switch on Workflow + * entry and finds automation paused. Halts the run before it spends AI/network + * work; the Workflow step retries, resuming once automation is unpaused. + */ +export class AutomationPausedError extends Error { + override readonly name = "AutomationPausedError"; +} + export interface AutomationPauseUpdate { paused: boolean; reason: string | null; diff --git a/apps/labeler/src/console-mutation-api.ts b/apps/labeler/src/console-mutation-api.ts index b0f55613a4..077bd870d2 100644 --- a/apps/labeler/src/console-mutation-api.ts +++ b/apps/labeler/src/console-mutation-api.ts @@ -163,9 +163,10 @@ export interface ConsoleMutationDeps { afterCommit: (actionId: string) => Promise; /** Schedules `afterCommit` without blocking the response (workerd `waitUntil`). */ defer: (work: Promise) => void; - /** Re-enqueues a dead letter's discovery job on retry (design §6). A queue op - * cannot join the D1 batch, so it runs in the deferred tail; the discovery - * consumer's `runKey` dedup absorbs a duplicate re-drive. */ + /** Re-enqueues a dead letter's discovery job on retry. A queue op cannot join + * the D1 batch, so it is awaited before `retried` is committed — `retried` is a + * durable claim only once the job is accepted; the discovery consumer's `runKey` + * dedup absorbs a duplicate re-drive. */ sendDiscoveryJob: (job: DiscoveryJob) => Promise; /** Dispatches a rerun's fresh assessment run to its Workflow instance (the same * binding discovery uses). The instance id is the run's `runKey`, so a @@ -247,6 +248,8 @@ export async function handleConsoleMutation( error instanceof ReadGuardError || error instanceof LabelMutationError || error instanceof DeadLetterResolvedError || + error instanceof DeadLetterUndeliverableError || + error instanceof DeadLetterReenqueueError || error instanceof NoActiveLabelError || error instanceof ReconsiderationStateError ) @@ -1307,6 +1310,47 @@ class DeadLetterResolvedError extends Error { } } +/** A dead letter whose stored payload no longer decodes to a discovery job, so a + * retry could never re-drive it. A 422 that leaves the row `new` (the operator + * quarantines it) rather than stranding it `retried` with nothing delivered. */ +class DeadLetterUndeliverableError extends Error { + override readonly name = "DeadLetterUndeliverableError"; + readonly code = "DEAD_LETTER_UNDELIVERABLE"; + readonly status = 422; + + constructor() { + super( + "Dead letter payload cannot be reconstructed into a discovery job; quarantine it instead", + ); + } + + toResponse(): Response { + return Response.json( + { error: { code: this.code, message: this.message } }, + { status: this.status }, + ); + } +} + +/** The re-drive enqueue failed, so `retried` was not committed. A retryable 503: + * the row stays `new` and a later retry re-attempts the enqueue. */ +class DeadLetterReenqueueError extends Error { + override readonly name = "DeadLetterReenqueueError"; + readonly code = "DEAD_LETTER_REENQUEUE_FAILED"; + readonly status = 503; + + constructor(options?: { cause?: unknown }) { + super("Dead letter re-drive could not be enqueued; retry.", options); + } + + toResponse(): Response { + return Response.json( + { error: { code: this.code, message: this.message } }, + { status: this.status }, + ); + } +} + /** The id from the path, folded into the parsed body so it joins the request * fingerprint — a replayed key must target the same dead letter. */ interface DeadLetterActionBody { @@ -1326,9 +1370,10 @@ interface DeadLetterActionDescriptor { * operational controls (design §6). Neither issues a label: the effect is a * `status` UPDATE guarded by `status = 'new'` plus an ungated `info` operational * event, both committed in one atomic `commitMutation` batch with the audit row. - * Retry additionally re-enqueues the dead letter's discovery job in the deferred - * tail. A row that is absent (404) or already resolved (409) is rejected before - * any commit, so a late request commits nothing and enqueues nothing (T6). + * Retry first durably enqueues the discovery job, then commits `retried`, so the + * status flips only once the re-drive is accepted. A row that is absent (404), + * already resolved (409), undeliverable (422), or whose enqueue fails (503) is + * rejected before the commit, leaving the row `new` and actionable (T6). */ async function runDeadLetterAction( request: Request, @@ -1353,6 +1398,22 @@ async function runDeadLetterAction( if (!letter) throw new ReadGuardError("NOT_FOUND"); if (letter.status !== "new") throw new DeadLetterResolvedError(); + // Durably enqueue the re-drive BEFORE committing `retried`, so the status can + // only flip once the discovery consumer has accepted the job. An undecodable + // payload or a queue failure leaves the row `new` — still counted by health and + // retryable — rather than stranding it `retried` with nothing delivered. A + // duplicate enqueue under a concurrent double-retry is absorbed by the + // consumer's `runKey` dedup. + if (status === "retried") { + const job = await readDeadLetterJob(deps.db, id); + if (!job) throw new DeadLetterUndeliverableError(); + try { + await deps.sendDiscoveryJob(job); + } catch (error) { + throw new DeadLetterReenqueueError({ cause: error }); + } + } + const subjectUri = `at://${letter.did}/${letter.collection}/${letter.rkey}`; const update = buildDeadLetterResolveUpdate(deps.db, { id, @@ -1388,42 +1449,9 @@ async function runDeadLetterAction( descriptor, ); - if (status === "retried") deferDeadLetterReenqueue(deps, id, ctx.actionId); return jsonData(returned); } -/** - * Re-enqueues a retried dead letter's discovery job off the response path, but - * only when this action won the `status = 'new'` race: a concurrent retry that - * committed its audit row yet matched zero rows in the UPDATE reads back a - * `resolved_by_action_id` that is not ours and skips, so the job enqueues once - * even under a double retry. The consumer's `runKey` dedup is the backstop; a - * lost enqueue is recoverable by re-driving `status = 'retried'` rows. - */ -function deferDeadLetterReenqueue(deps: ConsoleMutationDeps, id: number, actionId: string): void { - deps.defer( - (async () => { - try { - const letter = await getDeadLetter(deps.db, id); - if (!letter || letter.resolvedByActionId !== actionId) return; - const job = await readDeadLetterJob(deps.db, id); - if (!job) { - // The row flipped to 'retried' but its payload didn't decode to a - // valid job (e.g. a row written before the full-job payload shape), - // so nothing re-drives it and a fresh retry is barred by status. - console.error("[console-mutation] dead-letter payload undecodable, re-enqueue skipped", { - id, - }); - return; - } - await deps.sendDiscoveryJob(job); - } catch (error) { - console.error("[console-mutation] dead-letter re-enqueue failed", error); - } - })(), - ); -} - /** A non-numeric or non-positive path segment names no dead letter — a 404, * matching the absent-row case. Runs after the guard's role gate, so a non-admin * is rejected before this. */ diff --git a/apps/labeler/test/assessment-orchestrator.test.ts b/apps/labeler/test/assessment-orchestrator.test.ts index f4c2634590..8b8cdff7c5 100644 --- a/apps/labeler/test/assessment-orchestrator.test.ts +++ b/apps/labeler/test/assessment-orchestrator.test.ts @@ -1387,3 +1387,60 @@ describe("AssessmentOrchestrator: delete racing finalization (Blocker 1)", () => expect(labelsAfter?.n).toBe(0); }); }); + +describe("AssessmentOrchestrator: automation pause between prep and commit", () => { + it("leaves the run running and issues no labels when automation pauses before the batch commits", async () => { + const run = await pendingRun({ + name: "automation-pause", + cidValue: await cid("automation-pause"), + }); + // Pause automation in the seam between finalization prep (statements built + // while unpaused) and the batch commit. + const db = pauseBeforeFirstBatch(testEnv.DB, async () => { + await testEnv.DB.prepare(`UPDATE automation_state SET paused = 1 WHERE id = 1`).run(); + }); + const orchestrator = new AssessmentOrchestrator({ + db, + config, + signer: await signer(), + policy: MODERATION_POLICY, + stages: stubStages, + sleep: () => Promise.resolve(), + }); + + // The CAS carries the automation guard, so the whole batch no-ops and the + // finalization conflict is raised rather than terminal state with labels. + await expect(orchestrator.runAssessment(run.id)).rejects.toBeInstanceOf( + AssessmentFinalizationConflictError, + ); + + expect((await getAssessment(testEnv.DB, run.id))?.state).toBe("running"); + const labels = await testEnv.DB.prepare( + `SELECT COUNT(*) AS n FROM issued_labels l JOIN issuance_actions a ON a.id = l.action_id + WHERE a.assessment_id = ?`, + ) + .bind(run.id) + .first<{ n: number }>(); + expect(labels?.n).toBe(0); + + // Resume automation and re-run (as the Workflow retry does): finalization now + // completes and issues the labels the paused attempt withheld. + await testEnv.DB.prepare(`UPDATE automation_state SET paused = 0 WHERE id = 1`).run(); + const resumed = new AssessmentOrchestrator({ + db: testEnv.DB, + config, + signer: await signer(), + policy: MODERATION_POLICY, + stages: stubStages, + sleep: () => Promise.resolve(), + }); + const finalized = await resumed.runAssessment(run.id); + expect(finalized.state).toBe("passed"); + const pendingNeg = await testEnv.DB.prepare( + `SELECT neg FROM issued_labels WHERE uri = ? AND cid = ? AND val = 'assessment-pending'`, + ) + .bind(run.uri, run.cid) + .first<{ neg: number }>(); + expect(pendingNeg?.neg).toBe(1); + }); +}); diff --git a/apps/labeler/test/assessment-store.test.ts b/apps/labeler/test/assessment-store.test.ts index 11d452d407..561f27c999 100644 --- a/apps/labeler/test/assessment-store.test.ts +++ b/apps/labeler/test/assessment-store.test.ts @@ -9,6 +9,7 @@ import { operatorTriggerId, } from "../src/assessment-lifecycle.js"; import { + buildAssessmentRunStatement, buildFinalizationStatements, createAssessmentRun, createSubject, @@ -575,6 +576,67 @@ describe("supersession", () => { expect(pointer?.assessmentId).toBe(newerId); }); + it("a lower-ID run at an equal epoch never displaces a higher-ID current", async () => { + const subject = await observedSubject(); + const createdAt = new Date("2026-07-12T00:00:00.000Z"); + + async function runningRun(id: string, triggerSuffix: string) { + const triggerId = operatorTriggerId(triggerSuffix); + const runKey = await computeRunKey({ + uri: subject.uri, + cid: subject.cid, + policyVersion: "v1", + modelId: "m", + promptHash: "p", + scannerSetVersion: "v1", + triggerId, + }); + await buildAssessmentRunStatement(testEnv.DB, { + id, + runKey, + uri: subject.uri, + cid: subject.cid, + trigger: "operator", + triggerId, + policyVersion: "v1", + coverageJson: "{}", + now: createdAt, + }).run(); + for (const [from, to] of [ + ["observed", "verifying"], + ["verifying", "pending"], + ["pending", "running"], + ] as const) { + await transitionAssessmentState(testEnv.DB, { id, from, to }); + } + } + + const higherId = "asmt_ZZZZZZZZZZZZZZZZZZZZZZZZZZ"; + const lowerId = "asmt_00000000000000000000000000"; + await runningRun(higherId, "higher"); + await runningRun(lowerId, "lower"); + + const finalize = (id: string) => + buildFinalizationStatements(testEnv.DB, { + assessmentId: id, + fromState: "running", + toState: "passed", + src: LABELER_DID, + uri: subject.uri, + cid: subject.cid, + }).statements; + + await testEnv.DB.batch(finalize(higherId)); + await testEnv.DB.batch(finalize(lowerId)); + + const pointer = await getCurrentAssessment(testEnv.DB, { + src: LABELER_DID, + uri: subject.uri, + cid: subject.cid, + }); + expect(pointer?.assessmentId).toBe(higherId); + }); + it("a pending newer run never moves the current pointer", async () => { const subject = await observedSubject(); const first = await runningAssessment(subject); diff --git a/apps/labeler/test/assessment-workflow.test.ts b/apps/labeler/test/assessment-workflow.test.ts index f89162cfd4..e92b2209b0 100644 --- a/apps/labeler/test/assessment-workflow.test.ts +++ b/apps/labeler/test/assessment-workflow.test.ts @@ -31,9 +31,11 @@ import { serializeCoverage, type CoverageAccumulator } from "../src/assessment-s import { createAssessmentRun, createSubject, + getAssessment, transitionAssessmentState, } from "../src/assessment-store.js"; import { buildStages, executeAssessmentInstance } from "../src/assessment-workflow.js"; +import { AutomationPausedError } from "../src/automation-state.js"; import type { AiBinding } from "../src/code-ai-adapter.js"; import type { PublisherVerificationReader } from "../src/history-context.js"; import type { ImageAiBinding } from "../src/image-ai-adapter.js"; @@ -352,4 +354,19 @@ describe("executeAssessmentInstance: shell", () => { /missing required bindings/, ); }); + + it("halts on entry, before any stage work, when automation is paused", async () => { + const run = await pendingRun("wf-paused-entry"); + await testEnv.DB.prepare(`UPDATE automation_state SET paused = 1 WHERE id = 1`).run(); + + // The paused switch throws before `assertRequiredBindings`, so the error is + // the kill-switch halt — not the missing-binding failure minimalEnv would + // otherwise raise once it reached stage assembly. + await expect(executeAssessmentInstance(minimalEnv, run.id)).rejects.toBeInstanceOf( + AutomationPausedError, + ); + + const assessment = await getAssessment(testEnv.DB, run.id); + expect(assessment?.state).toBe("pending"); + }); }); diff --git a/apps/labeler/test/console-mutation-api.test.ts b/apps/labeler/test/console-mutation-api.test.ts index 4f8f0ce98c..2883e198b0 100644 --- a/apps/labeler/test/console-mutation-api.test.ts +++ b/apps/labeler/test/console-mutation-api.test.ts @@ -1965,6 +1965,52 @@ describe("console mutation: dead-letter retry (admin)", () => { }); }); +describe("console mutation: dead-letter retry durability", () => { + it("leaves the row new (not retried) when the queue enqueue fails", async () => { + const { id } = await seedDeadLetter(); + const deps = mutationDeps({ + sendDiscoveryJob: async () => { + throw new Error("queue unavailable"); + }, + }); + const response = await handleConsoleMutation( + deadLetterReq(`/admin/api/dead-letters/${id}/retry`), + deps, + ); + // A failed re-drive is retryable, not a phantom success: the row is untouched + // (still new, unresolved) so a later retry works and health still counts it. + expect(response.status).toBe(503); + const row = await deadLetterRow(id); + expect(row?.status).toBe("new"); + expect(row?.resolved_at).toBeNull(); + expect(row?.resolved_by_action_id).toBeNull(); + }); + + it("leaves the row new (not retried) when the stored payload cannot be decoded", async () => { + const garbage = new TextEncoder().encode(JSON.stringify({ not: "a discovery job" })); + const result = await testEnv.DB.prepare( + `INSERT INTO dead_letters (did, collection, rkey, reason, detail, payload, received_at, status) + VALUES (?, 'com.emdashcms.experimental.package.release', 'dl-garbage', 'verify-failed', 'detail', ?, datetime('now'), 'new')`, + ) + .bind(PUBLISHER_DID, garbage) + .run(); + const id = Number(result.meta.last_row_id); + const { deps, sent, settle } = captureReenqueue(); + + const response = await handleConsoleMutation( + deadLetterReq(`/admin/api/dead-letters/${id}/retry`), + deps, + ); + + // An undecodable payload can never be delivered, so the retry is refused and + // the row stays new (the operator quarantines it) rather than stranding retried. + expect(response.status).toBe(422); + expect((await deadLetterRow(id))?.status).toBe("new"); + await settle(); + expect(sent).toHaveLength(0); + }); +}); + describe("console mutation: dead-letter quarantine (admin)", () => { it("flips new→quarantined, emits one info event, enqueues nothing", async () => { const { id } = await seedDeadLetter();