fix(labeler): kill-switch on in-flight runs, durable dead-letter retry, pointer tie-break (held review findings)

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];

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,

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));

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;

apps/labeler/src/console-mutation-api.ts

@@ -163,9 +163,10 @@ export interface ConsoleMutationDeps {
 	afterCommit: (actionId: string) => Promise<void>;
 	/** Schedules `afterCommit` without blocking the response (workerd `waitUntil`). */
 	defer: (work: Promise<unknown>) => 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<void>;
 	/** 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. */

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);
+	});
+});

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);