feat(labeler): production Workflow shell over AssessmentOrchestrator (#1978)

apps/labeler/calibration/tsconfig.json

@@ -5,5 +5,5 @@
 		"verbatimModuleSyntax": true,
 		"noEmit": true
 	},
-	"include": ["**/*.ts", "../worker-configuration.d.ts"]
+	"include": ["**/*.ts", "../worker-configuration.d.ts", "../src/vite-env.d.ts"]
 }

apps/labeler/src/assessment-dispatch.ts

@@ -0,0 +1,91 @@
+/**
+ * Dispatch seam between a created assessment run and its Workflow.
+ *
+ * Each run executes as one Cloudflare Workflow instance whose id IS the run's
+ * `runKey` — the deterministic per-run identity already computed by the
+ * discovery consumer / rerun path (`computeRunKey`: SHA-256 over uri, cid,
+ * policy, model, prompt, scanner-set, and triggerId). It is a 64-char lowercase
+ * hex string, within the 100-char instance-id limit, so it is used verbatim as
+ * the id — no second hash, no second formula.
+ *
+ * Keying on the runKey (not the subject) gives the right dedup granularity:
+ *
+ *   - Redelivery of the SAME discovery event recomputes the SAME runKey → same
+ *     instance id → `create` collides → idempotent no-op, no duplicate run.
+ *   - A re-assessment (operator rerun, intel re-trigger) mints a run with a new
+ *     triggerId → different runKey → different instance id → it dispatches its
+ *     own instance. A subject-only id would instead collide with the RETAINED
+ *     id of the prior completed/failed run (Cloudflare keeps instance ids ~30
+ *     days) and strand the re-assessment `pending` forever.
+ *
+ * This is per-run serialization, not per-subject: two DIFFERENT triggers for one
+ * subject can run concurrently. That is acceptable — the orchestrator's currency
+ * re-check + supersede and idempotency-keyed label issuance keep the outcome
+ * correct (at worst recomputed wastefully); the instance id is not the arbiter
+ * there.
+ *
+ * This module holds no reference to `AssessmentOrchestrator` or
+ * `cloudflare:workers`: the discovery consumer imports only this, so the
+ * orchestrator reaches production solely through `assessment-workflow.ts`.
+ */
+
+/** Payload the Workflow instance is triggered with. */
+export interface AssessmentWorkflowParams {
+	/** The pending assessment run the Workflow will drive to finalization. */
+	assessmentId: string;
+}
+
+/**
+ * Structural subset of the generated `Workflow` binding this module needs. A
+ * real `Workflow<AssessmentWorkflowParams>` satisfies it; tests supply an
+ * in-memory fake that enforces instance-id uniqueness the same way.
+ */
+export interface AssessmentWorkflowBinding {
+	create(options: { id: string; params: AssessmentWorkflowParams }): Promise<{ id: string }>;
+	get(id: string): Promise<{ id: string }>;
+}
+
+/**
+ * Raised when the Workflow instance could not be created for an infrastructure
+ * reason (not an already-exists collision). The discovery consumer retries the
+ * message; every upstream step is idempotent, so redelivery re-dispatches.
+ */
+export class AssessmentDispatchError extends Error {
+	override readonly name = "AssessmentDispatchError";
+}
+
+export interface DispatchAssessmentInput {
+	/** The run's `runKey`, used verbatim as the Workflow instance id. */
+	runKey: string;
+	assessmentId: string;
+}
+
+/**
+ * Create the run's Workflow instance, or converge if its id already exists.
+ * Returns `"created"` on a fresh dispatch and `"exists"` when the runKey-derived
+ * id was already taken (redelivery of the same run). A `create` failure with no
+ * surviving instance is a real infrastructure error and throws
+ * `AssessmentDispatchError`.
+ */
+export async function dispatchAssessmentWorkflow(
+	workflow: AssessmentWorkflowBinding,
+	input: DispatchAssessmentInput,
+): Promise<"created" | "exists"> {
+	const id = input.runKey;
+	const params: AssessmentWorkflowParams = { assessmentId: input.assessmentId };
+	try {
+		await workflow.create({ id, params });
+		return "created";
+	} catch (err) {
+		// `create` throws on an already-taken id (same-run redelivery) and on
+		// transient infra failures alike. Disambiguate by probing for the
+		// instance: present means the collision is a redelivery (idempotent
+		// no-op); absent means the create genuinely failed and the caller retries.
+		const existing = await workflow.get(id).catch(() => null);
+		if (existing) return "exists";
+		throw new AssessmentDispatchError(
+			`failed to dispatch assessment Workflow for run ${input.assessmentId}`,
+			{ cause: err },
+		);
+	}
+}

apps/labeler/src/assessment-orchestrator.ts

@@ -3,13 +3,17 @@
  * through `running`, the analysis stages, and atomic finalization (spec
  * §9.9, §10).
  *
- * BINDING DECISION: production wiring stops at `assessment-pending`
- * (discovery-consumer.ts). Nothing in a production code path constructs or
- * calls `AssessmentOrchestrator` — it ships as code, exercised only by
- * tests, until W7/W8 supply real stage adapters (acquire, deterministic
- * validation, dependency/SBOM, code/metadata AI, image AI). `stubStages` in
- * this module exists for that same reason: test fixtures, not production
- * defaults.
+ * Driven in production by `AssessmentWorkflow` (assessment-workflow.ts): each
+ * run executes as one Workflow instance whose id is the run's runKey, so a
+ * redelivered discovery event dedups onto the same instance rather than starting
+ * a duplicate (see assessment-dispatch.ts). The Workflow constructs this
+ * orchestrator per run and calls `runAssessment`. Until W7/W8 supply the real stage adapters
+ * (acquire, deterministic validation, dependency/SBOM, code/metadata AI, image
+ * AI), that wiring runs `stubStages` — every stage resolves with no findings, so
+ * `resolvePolicyOutcome` returns `passed` and finalization issues a real signed
+ * `assessment-passed` label for EVERY subject (not merely clearing its own
+ * `assessment-pending`). That is a hard deploy gate — see the DEPLOY GATE note
+ * in assessment-workflow.ts.
  */
 
 import type { LabelSigner } from "@emdash-cms/registry-moderation";
@@ -48,6 +52,25 @@ export class StageTransientError extends Error {
 	override readonly name = "StageTransientError";
 }
 
+/**
+ * Thrown when the finalization batch's CAS out of `running` changed no row — a
+ * cancel or delete raced finalization. The in-batch state guard means no labels
+ * were issued; this signals the caller (the Workflow step) to retry, where the
+ * now-terminal row short-circuits.
+ */
+export class AssessmentFinalizationConflictError extends Error {
+	override readonly name = "AssessmentFinalizationConflictError";
+	constructor(
+		readonly assessmentId: string,
+		readonly expectedState: AssessmentState,
+		readonly actualState: AssessmentState | null,
+	) {
+		super(
+			`assessment ${assessmentId} finalization lost the race: expected ${expectedState}, found ${actualState ?? "missing"}`,
+		);
+	}
+}
+
 export interface StageContext {
 	readonly assessment: Assessment;
 }
@@ -119,14 +142,24 @@ export class AssessmentOrchestrator {
 		const now = this.now();
 		const loaded = await getAssessment(this.db, runId);
 		if (!loaded) throw new Error(`assessment ${runId} not found`);
-		if (loaded.state !== "pending")
-			throw new Error(`assessment ${runId} is not pending (state=${loaded.state})`);
-		const assessment = await transitionAssessmentState(this.db, {
-			id: runId,
-			from: "pending",
-			to: "running",
-			now,
-		});
+		// Accept a `running` row left by a crashed prior attempt and resume it: the
+		// production driver is a Workflow step that retries `runAssessment`, and any
+		// failure after the pending→running CAS (e.g. a transient D1 error in
+		// `finalize`) leaves the row `running`. Re-running is safe — stages are
+		// recomputed (findings are held in memory, not persisted mid-run) and every
+		// label issuance is idempotency-keyed. A `pending` row is transitioned to
+		// `running` first; a terminal row is the caller's to short-circuit.
+		if (loaded.state !== "pending" && loaded.state !== "running")
+			throw new Error(`assessment ${runId} is not resumable (state=${loaded.state})`);
+		const assessment =
+			loaded.state === "pending"
+				? await transitionAssessmentState(this.db, {
+						id: runId,
+						from: "pending",
+						to: "running",
+						now,
+					})
+				: loaded;
 
 		const findings: StageFinding[] = [];
 		let transientExhausted = false;
@@ -164,11 +197,11 @@ export class AssessmentOrchestrator {
 		// label (an async round-trip) between here and `db.batch`, so a delete or
 		// a cancel landing in that window still commits labels — the finalization
 		// CAS guards its own row, but the issuance statements carry no
-		// assessment-state guard. Closing it requires either the per-subject
-		// workflow lock the spec mandates (§14.1) or an in-batch state guard on
-		// every issuance statement. That belongs with wiring the orchestrator to
-		// production in W7/W8; today the production-boundary test guarantees no
-		// production path reaches this method.
+		// assessment-state guard. The Workflow instance id (the runKey) dedups only
+		// redelivery of the same run (assessment-dispatch.ts) — not a delete or
+		// operator cancel against an in-flight run — so closing this window still
+		// needs an in-batch assessment-state guard on every issuance statement,
+		// tracked with the real-stage wiring.
 		const current = await isSubjectCurrent(this.db, { uri: assessment.uri, cid: assessment.cid });
 		if (!current) {
 			return transitionAssessmentState(this.db, { id: runId, from: "running", to: "stale", now });
@@ -257,6 +290,9 @@ export class AssessmentOrchestrator {
 				},
 				now,
 				false,
+				// Gate every finalization label on the run reaching `toState`, so a
+				// concurrent cancel/delete that no-ops the CAS also no-ops the labels.
+				{ requireAssessmentState: toState },
 			);
 			statements.push(...built.statements);
 			postCommits.push(built.postCommit);
@@ -291,30 +327,26 @@ export class AssessmentOrchestrator {
 			await issue(prior.val, true);
 		}
 
-		// Final currency re-check with no signing between it and the commit, so
-		// the delete/cancel window is two adjacent D1 ops rather than spanning
-		// every label's signing round-trip above. Full closure still needs the
-		// workflow lock (see the re-check before this method); this shrinks the
-		// exposure in the interim.
+		// Final currency re-check with no signing between it and the commit, so the
+		// delete/cancel window is two adjacent D1 ops rather than spanning every
+		// label's signing round-trip above.
 		//
-		// The same lock also closes a signing-state flip during the batch: the
-		// label inserts are guarded on active signing state and no-op if it
-		// flips, but the state CAS below is not, so a flip after this point
-		// could commit the terminal state without its labels. Deferred to the
-		// same W7/W8 production wiring as the delete race above.
+		// The batch is all-or-nothing against the run→toState CAS: every issuance
+		// action is gated on the assessment reaching `toState`
+		// (buildIssuanceStatements' requireAssessmentState), and the CAS's row count
+		// is checked after commit (below). A cancel or delete that moves the run out
+		// of `running` in this gap therefore no-ops the CAS AND every label — nothing
+		// leaks — and the lost race raises AssessmentFinalizationConflictError.
 		//
-		// Three related concurrency gaps in this method are owned by that same
-		// W7/W8 workflow lock (§14.1), which serialises finalization per subject
-		// and makes them all unreachable — the production-boundary test proves
-		// nothing wires this method live until then:
-		//   - a stale (CID-superseded) run returns without negating its own
-		//     assessment-pending, so a superseded subject keeps advertising an
-		//     in-progress assessment;
-		//   - the finalization CAS result below is not checked, so if a
-		//     concurrent cancel moved the run out of `running`, the CAS no-ops
-		//     while the label inserts commit and this returns a run that never
-		//     exited `running`;
-		//   - `transitionAssessmentState` here and in the stale branch can throw
+		// Narrower gaps remain, tracked with the real-stage wiring:
+		//   - a signing-state flip mid-batch: the label inserts are guarded on active
+		//     signing state and no-op if it flips, but the CAS is not, so a flip could
+		//     commit the terminal state with its labels suppressed (the CAS still
+		//     changed a row, so the postCommit below surfaces it as a signing error);
+		//   - a CID supersession landing in this gap does not move the run out of
+		//     `running`, so the CAS succeeds and this run finalizes labels for its own
+		//     CID (the pointer upsert is guarded on created-at ordering);
+		//   - `transitionAssessmentState` in the stale branch below can throw
 		//     `AssessmentTransitionConflictError` under a racing delete.
 		const stillCurrent = await isSubjectCurrent(this.db, {
 			uri: assessment.uri,
@@ -329,7 +361,18 @@ export class AssessmentOrchestrator {
 			});
 		}
 
-		await this.db.batch(statements);
+		const results = await this.db.batch(statements);
+		// If the finalization CAS changed no row, a cancel/delete moved the run out
+		// of `running` between the currency re-check and commit. The in-batch state
+		// guard on every issuance statement means NO labels were written, so skip
+		// the postCommits (each would otherwise mis-diagnose the absent label as a
+		// signing failure and throw) and surface the lost race loudly — the Workflow
+		// step retries and short-circuits on the now-terminal row.
+		const casChanged = results[finalization.assessmentUpdateIndex]?.meta.changes ?? 0;
+		if (casChanged !== 1) {
+			const raced = await getAssessment(this.db, assessment.id);
+			throw new AssessmentFinalizationConflictError(assessment.id, toState, raced?.state ?? null);
+		}
 		for (const postCommit of postCommits) await postCommit();
 
 		const finalised = await getAssessment(this.db, assessment.id);

apps/labeler/src/assessment-workflow.ts

@@ -0,0 +1,109 @@
+/**
+ * Production assessment execution: one Cloudflare Workflow instance per run, a
+ * thin durable shell over `AssessmentOrchestrator`. The instance id is the run's
+ * runKey (see `assessment-dispatch.ts`); `run` loads the assessment and drives
+ * it through the orchestrator's stages and atomic finalization inside a single
+ * durable step.
+ *
+ * DEPLOY GATE: with `stubStages`, a run finalizes `passed` and issues a real
+ * signed `assessment-passed` label for EVERY subject — an unconditional "this is
+ * safe" attestation over unscanned content. This shell must NOT reach an
+ * enforcing or label-consuming production deployment until the real analysis
+ * stages (W7/W8) land; shipping it live before then would vouch for everything.
+ *
+ * The whole run executes in one `step.do`, not one step per stage: the
+ * orchestrator accumulates stage findings in memory and the acquire stage
+ * publishes the acquired artifact to an in-process `AcquisitionHolder` that
+ * downstream stages read, so a per-stage step boundary would drop that shared
+ * state across a durable resume. Running the orchestrator whole keeps its atomic
+ * finalization intact. The tradeoff is coarse resume granularity: a mid-run
+ * eviction re-runs the whole step. `executeAssessmentInstance` makes that
+ * idempotent — a terminal row short-circuits, and `runAssessment` resumes a row
+ * left `running` by a crashed attempt. Finer, per-stage durable resume lands
+ * with the real-stage wiring.
+ */
+
+import { WorkflowEntrypoint } from "cloudflare:workers";
+import type { WorkflowEvent, WorkflowStep } from "cloudflare:workers";
+
+import type { AssessmentWorkflowParams } from "./assessment-dispatch.js";
+import { TERMINAL_STATES, type AssessmentState } from "./assessment-lifecycle.js";
+import {
+	AssessmentOrchestrator,
+	stubStages,
+	type OrchestratorStages,
+} from "./assessment-orchestrator.js";
+import { getAssessment } from "./assessment-store.js";
+import { getLabelerIdentityConfig } from "./config.js";
+import { MODERATION_POLICY } from "./policy.js";
+import { createRuntimeSigner, getRuntimeSigningSecret } from "./signing-runtime.js";
+
+const RUN_STEP_CONFIG = {
+	retries: { limit: 3, delay: "10 seconds" as const, backoff: "exponential" as const },
+};
+
+export class AssessmentWorkflow extends WorkflowEntrypoint<Env, AssessmentWorkflowParams> {
+	override async run(
+		event: Readonly<WorkflowEvent<AssessmentWorkflowParams>>,
+		step: WorkflowStep,
+	): Promise<{ assessmentId: string; state: AssessmentState }> {
+		const { assessmentId } = event.payload;
+		const state = await step.do("assess-subject", RUN_STEP_CONFIG, () =>
+			executeAssessmentInstance(this.env, assessmentId),
+		);
+		return { assessmentId, state };
+	}
+}
+
+/**
+ * The shell body: load the pending run, drive it through the orchestrator, and
+ * return the finalized state. Separate from the entrypoint class so it is
+ * testable without Cloudflare's Workflow runtime (which has no local harness) —
+ * `run` above is the thin durable-step wrapper over it.
+ */
+export async function executeAssessmentInstance(
+	env: Env,
+	assessmentId: string,
+): Promise<AssessmentState> {
+	const existing = await getAssessment(env.DB, assessmentId);
+	if (!existing) throw new Error(`assessment ${assessmentId} not found`);
+	// Idempotent resume for a durable-step retry: a terminal row means a prior
+	// attempt already finalized (its batch committed but the step result was
+	// lost) — return that state. A `pending` or crash-left `running` row falls
+	// through to `runAssessment`, which drives (or resumes) it to finalization.
+	if (TERMINAL_STATES.has(existing.state)) return existing.state;
+
+	const config = await getLabelerIdentityConfig(env);
+	const versioned = await createRuntimeSigner(config, getRuntimeSigningSecret(env));
+	const orchestrator = new AssessmentOrchestrator({
+		db: env.DB,
+		config,
+		signer: versioned.signer,
+		policy: MODERATION_POLICY,
+		stages: buildStages(),
+	});
+	const finalized = await orchestrator.runAssessment(assessmentId);
+	return finalized.state;
+}
+
+/**
+ * The orchestrator's stage adapters, built per instance execution. Real
+ * adapters attach here in the W7/W8 follow-on — acquire (via the aggregator
+ * client), deterministic/dependency/AI scanning, and publisher history; today
+ * the run executes the exported stub stages. Building them per execution rather
+ * than sharing module-scope state keeps per-run state — notably the acquire
+ * stage's `AcquisitionHolder` — isolated to this instance, never shared across
+ * concurrent subjects.
+ */
+function buildStages(): OrchestratorStages {
+	// Mechanical enforcement of the DEPLOY GATE above: a production build must not
+	// run stub stages (which would sign `assessment-passed` for every unscanned
+	// subject). `import.meta.env.PROD` is a Vite compile-time constant — true in
+	// `vite build`, false in dev and the vitest pool — so this cannot be spoofed
+	// at runtime. Remove once real stages are wired here.
+	if (import.meta.env.PROD)
+		throw new Error(
+			"AssessmentWorkflow has only stub stages in a production build — refusing to issue assessment-passed for unscanned subjects. Wire the real analysis stages (W7/W8) before deploying.",
+		);
+	return stubStages;
+}