From fe63352fb3253df43b460e5eb8984fcdf86c22b3 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Thu, 16 Jul 2026 12:11:05 +0100 Subject: [PATCH 1/3] feat(labeler): assessment Workflow shell with per-subject lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the AssessmentOrchestrator to production as a Cloudflare Workflow: each verified subject runs as one Workflow instance whose id is a deterministic SHA-256 of (uri, cid). Creating an instance with an existing id throws, so a duplicate subject cannot start a second concurrent run — the instance-per- subject serialization IS the §14.1 per-subject lock, closing the duplicate-run race #1978 deferred. - assessment-workflow.ts: AssessmentWorkflow (WorkflowEntrypoint) runs the orchestrator inside a single durable step over executeAssessmentInstance, which loads the pending run, composes the orchestrator (stub stages for now), and finalizes. Idempotent resume: an already-finalized run returns its terminal state instead of re-entering the pending-guarded orchestrator. - assessment-dispatch.ts: instance-id derivation + dispatchAssessmentWorkflow, which treats an already-exists collision as an idempotent no-op and surfaces infra failures as AssessmentDispatchError. - discovery-consumer.ts narrows to "verified discovery → dispatch"; a dispatch failure retries the message (upstream steps are idempotent). - ASSESSMENT_WORKFLOW binding in wrangler.jsonc + worker entry export. Stages are stubStages until W7/W8 land the real acquire/deterministic/ dependency/AI/history adapters; the per-run AcquisitionHolder they need is constructed at the per-run buildStages() seam. The orchestrator's finalization race comments are corrected: the lock closes duplicate runs, not a delete or cancel racing an in-flight run — that still needs an in-batch state guard. --- apps/labeler/src/assessment-dispatch.ts | 88 +++++++++++ apps/labeler/src/assessment-orchestrator.ts | 49 ++++--- apps/labeler/src/assessment-workflow.ts | 94 ++++++++++++ apps/labeler/src/discovery-consumer.ts | 42 +++++- apps/labeler/src/index.ts | 1 + apps/labeler/test/assessment-dispatch.test.ts | 118 +++++++++++++++ apps/labeler/test/assessment-workflow.test.ts | 137 ++++++++++++++++++ .../labeler/test/console-mutation-api.test.ts | 4 + apps/labeler/test/discovery-consumer.test.ts | 124 ++++++++++++++++ apps/labeler/test/production-boundary.test.ts | 23 +-- apps/labeler/worker-configuration.d.ts | 3 +- apps/labeler/wrangler.jsonc | 12 ++ 12 files changed, 654 insertions(+), 41 deletions(-) create mode 100644 apps/labeler/src/assessment-dispatch.ts create mode 100644 apps/labeler/src/assessment-workflow.ts create mode 100644 apps/labeler/test/assessment-dispatch.test.ts create mode 100644 apps/labeler/test/assessment-workflow.test.ts diff --git a/apps/labeler/src/assessment-dispatch.ts b/apps/labeler/src/assessment-dispatch.ts new file mode 100644 index 000000000..f269ec72b --- /dev/null +++ b/apps/labeler/src/assessment-dispatch.ts @@ -0,0 +1,88 @@ +/** + * Dispatch seam between verified discovery and the assessment Workflow. + * + * Each subject runs as one Cloudflare Workflow instance whose id is a + * deterministic function of (uri, cid). That id IS the spec §14.1 per-subject + * lock: `create` throws when an instance with the id already exists, so a + * duplicate subject cannot start a second concurrent run — resolving the + * queue-vs-Workflow serialization question deferred in #1978. + * `dispatchAssessmentWorkflow` treats that collision as an idempotent no-op and + * surfaces genuine infrastructure failures as `AssessmentDispatchError` for the + * caller to retry. + * + * 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` 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"; +} + +/** + * Deterministic Workflow instance id for a subject: SHA-256 hex of (uri, cid). + * Derived from the subject alone — not the run key — so any run for the same + * subject maps to the same instance and is serialized by it. 64 hex chars, well + * within the 100-char instance-id limit. + */ +export async function assessmentWorkflowInstanceId(uri: string, cid: string): Promise { + const material = `${uri}\n${cid}`; + const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(material)); + return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""); +} + +export interface DispatchAssessmentInput { + uri: string; + cid: string; + assessmentId: string; +} + +/** + * Create the subject's Workflow instance, or converge if it already exists. + * Returns `"created"` on a fresh dispatch and `"exists"` when the deterministic + * id was already taken (the lock held — no second 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 = await assessmentWorkflowInstanceId(input.uri, input.cid); + const params: AssessmentWorkflowParams = { assessmentId: input.assessmentId }; + try { + await workflow.create({ id, params }); + return "created"; + } catch (err) { + // `create` throws on an already-taken id (the lock) and on transient + // infra failures alike. Disambiguate by probing for the instance: present + // means the collision is the lock (idempotent no-op); absent means the + // create genuinely failed and the caller must retry. + const existing = await workflow.get(id).catch(() => null); + if (existing) return "exists"; + throw new AssessmentDispatchError(`failed to dispatch assessment Workflow for ${input.uri}`, { + cause: err, + }); + } +} diff --git a/apps/labeler/src/assessment-orchestrator.ts b/apps/labeler/src/assessment-orchestrator.ts index 3cfc23d66..5cf17048d 100644 --- a/apps/labeler/src/assessment-orchestrator.ts +++ b/apps/labeler/src/assessment-orchestrator.ts @@ -3,13 +3,15 @@ * 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 + * verified subject runs as one Workflow instance whose id serializes duplicate + * runs of that subject (the §14.1 per-subject lock — 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 a real subject finalizes as `passed`, clearing only its own + * `assessment-pending`. */ import type { LabelSigner } from "@emdash-cms/registry-moderation"; @@ -164,11 +166,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 instance-per-subject Workflow lock + // (assessment-dispatch.ts) serializes duplicate runs of the same subject + // but 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 }); @@ -293,20 +295,19 @@ export class AssessmentOrchestrator { // 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. + // every label's signing round-trip above. Full closure still needs an + // in-batch assessment-state guard (see the re-check before this method); + // this shrinks the exposure in the interim. // - // 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. + // That same guard is what a signing-state flip during the batch needs: + // 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. // - // 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: + // The instance-per-subject Workflow lock closes concurrent duplicate runs + // of a subject, but NOT a delete, operator cancel, or signing flip racing + // an in-flight run, so three related gaps in this method remain open until + // that in-batch state guard lands (tracked with the real-stage wiring): // - a stale (CID-superseded) run returns without negating its own // assessment-pending, so a superseded subject keeps advertising an // in-progress assessment; diff --git a/apps/labeler/src/assessment-workflow.ts b/apps/labeler/src/assessment-workflow.ts new file mode 100644 index 000000000..a0248e95b --- /dev/null +++ b/apps/labeler/src/assessment-workflow.ts @@ -0,0 +1,94 @@ +/** + * Production assessment execution: one Cloudflare Workflow instance per + * subject, a thin durable shell over `AssessmentOrchestrator`. The instance id + * (see `assessment-dispatch.ts`) serializes runs per subject; `run` loads the + * pending assessment and drives it through the orchestrator's stages and atomic + * finalization inside a single durable step. + * + * 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 from `pending`, and because `runAssessment` guards + * on `pending`, a row already advanced to `running` cannot be resumed here; the + * reconciliation pass (reconciliation.ts) surfaces such stuck runs. 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 { + override async run( + event: Readonly>, + 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 { + const existing = await getAssessment(env.DB, assessmentId); + if (!existing) throw new Error(`assessment ${assessmentId} not found`); + // Idempotent resume: a previous execution already finalized this run (an + // eviction after the finalization batch committed but before the step result + // was durable). Return the terminal state rather than re-entering the + // orchestrator, whose pending-guard would throw on a non-pending row. + 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 { + return stubStages; +} diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index 9318e761d..ebf40ea1d 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/apps/labeler/src/discovery-consumer.ts @@ -1,11 +1,14 @@ /** * Discovery queue consumer. Replaces the Jetstream-observed event with a - * verified subject, an idempotent assessment run, and (once verified) an - * `assessment-pending` label — spec §9.1 steps 5-7. Per binding decision, - * production wiring stops here: nothing in this file advances a run past - * `pending`. `assessment-orchestrator.ts` drives `pending → running → - * finalization` and is exercised only by tests until W7/W8 land real stage - * adapters. + * verified subject, an idempotent assessment run, an `assessment-pending` + * label, and a dispatched assessment Workflow instance — spec §9.1 steps 5-7. + * This file's job ends at dispatch: it never constructs + * `AssessmentOrchestrator` itself. The instance id derives from the subject + * (assessment-dispatch.ts), so a duplicate subject cannot start a second + * concurrent run — that collision is the spec §14.1 per-subject lock. + * `AssessmentWorkflow` (assessment-workflow.ts) then drives `pending → running + * → finalization` through the orchestrator (still stub stages until W7/W8 land + * real stage adapters). * * Error policy mirrors the aggregator's records-consumer with one * difference (spec §9.1): a forged or unverifiable event is ALWAYS a dead @@ -30,6 +33,11 @@ import { } from "@atcute/identity-resolver"; import type { LabelSigner } from "@emdash-cms/registry-moderation"; +import { + AssessmentDispatchError, + dispatchAssessmentWorkflow, + type AssessmentWorkflowBinding, +} from "./assessment-dispatch.js"; import { AssessmentTransitionConflictError, automatedIdempotencyKey, @@ -80,6 +88,10 @@ export interface DiscoveryConsumerDeps { config: LabelerConfig; signer: LabelSigner; didDocumentResolver: DidDocumentResolverLike; + /** The assessment Workflow binding. Once a verified subject reaches + * `pending`, the consumer dispatches one instance per subject; the + * deterministic instance id is the per-subject lock (assessment-dispatch.ts). */ + assessmentWorkflow: AssessmentWorkflowBinding; fetch?: typeof fetch; now?: () => Date; /** @@ -272,6 +284,12 @@ async function classifyDiscoveryError( controller.retry(); return; } + if (err instanceof AssessmentDispatchError) { + // Subject verified and pending; only the Workflow dispatch failed. Retry + // so redelivery re-dispatches — every upstream step is idempotent. + controller.retry(); + return; + } if (err instanceof PdsVerificationError) { if (isTransient(err.reason, err.status)) { controller.retry(); @@ -381,6 +399,17 @@ async function verifyAndCreateRun( { uri, cid: job.cid, val: "assessment-pending" }, now, ); + + // Hand the run to its Workflow instance. The instance id derives from the + // subject, so a redelivered event converges on the same instance rather than + // starting a second run (the §14.1 lock). A dispatch infra failure throws + // AssessmentDispatchError → the message retries; upstream steps are all + // idempotent, so redelivery re-dispatches. + await dispatchAssessmentWorkflow(deps.assessmentWorkflow, { + uri, + cid: job.cid, + assessmentId: assessment.id, + }); } /** @@ -513,6 +542,7 @@ async function createProductionDiscoveryDeps(env: Env): Promise(); + readonly created: Recorded[] = []; + /** When set, `create` rejects with this error. */ + failCreate: Error | undefined; + /** When true, a failing `create` still leaves the instance behind (models a + * create that ran but whose acknowledgement was lost). */ + persistOnFailure = false; + + create(options: { id: string; params: AssessmentWorkflowParams }): Promise<{ id: string }> { + if (this.failCreate) { + if (this.persistOnFailure) + this.instances.set(options.id, { id: options.id, params: options.params }); + return Promise.reject(this.failCreate); + } + if (this.instances.has(options.id)) + return Promise.reject(new Error(`instance ${options.id} already exists`)); + const recorded = { id: options.id, params: options.params }; + this.instances.set(options.id, recorded); + this.created.push(recorded); + return Promise.resolve({ id: options.id }); + } + + get(id: string): Promise<{ id: string }> { + const instance = this.instances.get(id); + if (!instance) return Promise.reject(new Error(`instance ${id} not found`)); + return Promise.resolve({ id }); + } +} + +describe("assessmentWorkflowInstanceId", () => { + it("is deterministic and 64 lowercase hex chars", async () => { + const a = await assessmentWorkflowInstanceId(URI, CID); + const b = await assessmentWorkflowInstanceId(URI, CID); + expect(a).toBe(b); + expect(a).toMatch(/^[0-9a-f]{64}$/); + }); + + it("derives from both uri and cid — either changing changes the id", async () => { + const base = await assessmentWorkflowInstanceId(URI, CID); + expect(await assessmentWorkflowInstanceId(`${URI}x`, CID)).not.toBe(base); + expect(await assessmentWorkflowInstanceId(URI, `${CID}x`)).not.toBe(base); + }); + + it("does not collide across a uri/cid boundary shift", async () => { + // `a\nb` vs `a` + `\nb` must not hash the same: guards the delimiter. + expect(await assessmentWorkflowInstanceId("a", "b")).not.toBe( + await assessmentWorkflowInstanceId("a\nb", ""), + ); + }); +}); + +describe("dispatchAssessmentWorkflow", () => { + it("creates the instance with the derived id and returns 'created'", async () => { + const workflow = new FakeWorkflow(); + const outcome = await dispatchAssessmentWorkflow(workflow, { + uri: URI, + cid: CID, + assessmentId: "asmt_1", + }); + expect(outcome).toBe("created"); + const expectedId = await assessmentWorkflowInstanceId(URI, CID); + expect(workflow.created).toEqual([{ id: expectedId, params: { assessmentId: "asmt_1" } }]); + }); + + it("returns 'exists' when the id is already taken — the lock held", async () => { + const workflow = new FakeWorkflow(); + await dispatchAssessmentWorkflow(workflow, { uri: URI, cid: CID, assessmentId: "asmt_1" }); + const outcome = await dispatchAssessmentWorkflow(workflow, { + uri: URI, + cid: CID, + assessmentId: "asmt_1", + }); + expect(outcome).toBe("exists"); + expect(workflow.created).toHaveLength(1); + }); + + it("returns 'exists' when create fails but the instance survives", async () => { + const workflow = new FakeWorkflow(); + workflow.failCreate = new Error("ack lost"); + workflow.persistOnFailure = true; + const outcome = await dispatchAssessmentWorkflow(workflow, { + uri: URI, + cid: CID, + assessmentId: "asmt_1", + }); + expect(outcome).toBe("exists"); + }); + + it("throws AssessmentDispatchError when create fails and no instance survives", async () => { + const workflow = new FakeWorkflow(); + workflow.failCreate = new Error("backend down"); + await expect( + dispatchAssessmentWorkflow(workflow, { uri: URI, cid: CID, assessmentId: "asmt_1" }), + ).rejects.toBeInstanceOf(AssessmentDispatchError); + }); +}); diff --git a/apps/labeler/test/assessment-workflow.test.ts b/apps/labeler/test/assessment-workflow.test.ts new file mode 100644 index 000000000..09555add9 --- /dev/null +++ b/apps/labeler/test/assessment-workflow.test.ts @@ -0,0 +1,137 @@ +/** + * The assessment Workflow is a thin durable shell over `AssessmentOrchestrator` + * (whose stage execution and finalization `assessment-orchestrator.test.ts` + * already covers). Cloudflare's Workflow runtime has no local test harness and + * the entrypoint class cannot be constructed in the workers pool, so these + * tests exercise `executeAssessmentInstance` — the shell body `run` wraps in a + * durable step — directly: it must load the pending run, compose the + * orchestrator, and finalize, plus resume idempotently on a re-run. + */ + +import { CODEC_RAW, create as createCid, toString as cidToString } from "@atcute/cid"; +import { applyD1Migrations, env } from "cloudflare:test"; +import { beforeAll, describe, expect, it } from "vitest"; + +import { computeRunKey, initialTriggerId } from "../src/assessment-lifecycle.js"; +import { + createAssessmentRun, + createSubject, + getAssessment, + transitionAssessmentState, +} from "../src/assessment-store.js"; +import { executeAssessmentInstance } from "../src/assessment-workflow.js"; +import { MODERATION_POLICY } from "../src/policy.js"; +import { initializeSigningState } from "../src/signing-rotation.js"; + +interface TestEnv { + DB: D1Database; + TEST_MIGRATIONS: Parameters[1]; +} + +const testEnv = env as unknown as TestEnv; +const LABELER_DID = "did:web:labels.emdashcms.com"; +const PUBLISHER_DID = "did:plc:publisher000000000000000000"; +const MULTIKEY = "zDnaepsL7AXenJkVYdkh5KuKsSU7Ykh7kyXaLLU7auN9FWSiZ"; + +beforeAll(async () => { + await applyD1Migrations(testEnv.DB, testEnv.TEST_MIGRATIONS); + await initializeSigningState(testEnv.DB, { + issuerDid: LABELER_DID, + keyVersion: "v1", + publicKeyMultibase: MULTIKEY, + }); +}); + +async function cid(seed: string): Promise { + const bytes = new TextEncoder().encode(seed); + const buffer = new ArrayBuffer(bytes.byteLength); + new Uint8Array(buffer).set(bytes); + const value = await createCid(CODEC_RAW, new Uint8Array(buffer)); + return cidToString(value); +} + +function releaseUri(name: string): string { + return `at://${PUBLISHER_DID}/com.emdashcms.experimental.package.release/${name}:1.0.0`; +} + +/** Creates a verified subject and a `pending` assessment run. */ +async function pendingRun(name: string): Promise<{ id: string; uri: string; cid: string }> { + const uri = releaseUri(name); + const cidValue = await cid(name); + await createSubject(testEnv.DB, { + uri, + cid: cidValue, + did: PUBLISHER_DID, + collection: "com.emdashcms.experimental.package.release", + rkey: `${name}:1.0.0`, + }); + const triggerId = initialTriggerId(cidValue); + const runKey = await computeRunKey({ + uri, + cid: cidValue, + policyVersion: MODERATION_POLICY.policyVersion, + modelId: "unassigned", + promptHash: "unassigned", + scannerSetVersion: "unassigned", + triggerId, + }); + const { assessment } = await createAssessmentRun(testEnv.DB, { + runKey, + uri, + cid: cidValue, + trigger: "initial", + triggerId, + policyVersion: MODERATION_POLICY.policyVersion, + coverageJson: "{}", + }); + await transitionAssessmentState(testEnv.DB, { + id: assessment.id, + from: "observed", + to: "verifying", + }); + await transitionAssessmentState(testEnv.DB, { + id: assessment.id, + from: "verifying", + to: "pending", + }); + return { id: assessment.id, uri, cid: cidValue }; +} + +const workflowEnv = env as unknown as Env; + +describe("executeAssessmentInstance", () => { + it("drives a pending run through the orchestrator to finalization (passed, stub stages)", async () => { + const run = await pendingRun("wf-happy"); + + const state = await executeAssessmentInstance(workflowEnv, run.id); + + expect(state).toBe("passed"); + const finalized = await getAssessment(testEnv.DB, run.id); + expect(finalized?.state).toBe("passed"); + + // The run's own assessment-pending is negated on finalization. + const pending = await testEnv.DB.prepare( + `SELECT l.neg FROM issued_labels l JOIN issuance_actions a ON a.id = l.action_id + WHERE l.uri = ? AND l.cid = ? AND l.val = 'assessment-pending'`, + ) + .bind(run.uri, run.cid) + .first<{ neg: number }>(); + expect(pending?.neg).toBe(1); + }); + + it("resumes idempotently: a re-run of an already-finalized assessment returns its state without re-entering the orchestrator", async () => { + const run = await pendingRun("wf-resume"); + + expect(await executeAssessmentInstance(workflowEnv, run.id)).toBe("passed"); + // A second execution (as a durable retry would do) must not throw on the + // now-terminal row — the orchestrator's pending-guard would otherwise + // reject it. + expect(await executeAssessmentInstance(workflowEnv, run.id)).toBe("passed"); + }); + + it("throws when the assessment does not exist", async () => { + await expect( + executeAssessmentInstance(workflowEnv, "asmt_00000000000000000000000000"), + ).rejects.toThrow(/not found/); + }); +}); diff --git a/apps/labeler/test/console-mutation-api.test.ts b/apps/labeler/test/console-mutation-api.test.ts index 25b2c7e17..3992b11c8 100644 --- a/apps/labeler/test/console-mutation-api.test.ts +++ b/apps/labeler/test/console-mutation-api.test.ts @@ -1770,6 +1770,10 @@ describe("console mutation: pause endpoint drives the discovery consumer gate (e config: CONFIG, signer: await testSigner(), didDocumentResolver: new KillSwitchStubResolver(), + assessmentWorkflow: { + create: (options) => Promise.resolve({ id: options.id }), + get: (id) => Promise.resolve({ id }), + }, verify: () => Promise.resolve({ cid: CID, diff --git a/apps/labeler/test/discovery-consumer.test.ts b/apps/labeler/test/discovery-consumer.test.ts index e0e140e7d..057262255 100644 --- a/apps/labeler/test/discovery-consumer.test.ts +++ b/apps/labeler/test/discovery-consumer.test.ts @@ -8,6 +8,11 @@ import { import { applyD1Migrations, env } from "cloudflare:test"; import { afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + assessmentWorkflowInstanceId, + type AssessmentWorkflowBinding, + type AssessmentWorkflowParams, +} from "../src/assessment-dispatch.js"; import { automatedIdempotencyKey, computeRunKey, @@ -119,12 +124,44 @@ class FakeMessage implements MessageController { } } +interface FakeInstance { + id: string; + params: AssessmentWorkflowParams; +} + +/** In-memory stand-in for the assessment Workflow binding that enforces + * instance-id uniqueness the way the real one does: `create` throws when the id + * is already taken (the per-subject lock), `get` resolves it. `createError` + * simulates an infrastructure failure. */ +class FakeAssessmentWorkflow implements AssessmentWorkflowBinding { + readonly instances = new Map(); + readonly created: FakeInstance[] = []; + createError: Error | undefined; + + create(options: { id: string; params: AssessmentWorkflowParams }): Promise<{ id: string }> { + if (this.createError) return Promise.reject(this.createError); + if (this.instances.has(options.id)) + return Promise.reject(new Error(`instance ${options.id} already exists`)); + const instance = { id: options.id, params: options.params }; + this.instances.set(options.id, instance); + this.created.push(instance); + return Promise.resolve({ id: options.id }); + } + + get(id: string): Promise<{ id: string }> { + const instance = this.instances.get(id); + if (!instance) return Promise.reject(new Error(`instance ${id} not found`)); + return Promise.resolve({ id }); + } +} + async function buildDeps(): Promise { return { db: testEnv.DB, config, signer: await signer(), didDocumentResolver: new StubResolver(), + assessmentWorkflow: new FakeAssessmentWorkflow(), }; } @@ -259,6 +296,93 @@ describe("processDiscoveryMessage: verified create", () => { }); }); +describe("processDiscoveryMessage: Workflow dispatch and per-subject lock", () => { + it("dispatches one Workflow instance per verified subject, id derived from (uri, cid)", async () => { + const job = await jobFor({ rkey: rkey() }); + const workflow = new FakeAssessmentWorkflow(); + const deps = { ...(await buildDeps()), assessmentWorkflow: workflow }; + const msg = new FakeMessage(); + + await processDiscoveryMessage(job, msg, { ...deps, verify: verifiedFor(job) }); + + expect(msg.acked).toBe(1); + const assessment = await getAssessmentByRunKey(testEnv.DB, await runKeyFor(job)); + const expectedId = await assessmentWorkflowInstanceId(uriFor(job), job.cid); + expect(workflow.created).toHaveLength(1); + expect(workflow.created[0]?.id).toBe(expectedId); + expect(workflow.created[0]?.params.assessmentId).toBe(assessment!.id); + }); + + it("redelivery does not start a second run — the instance id is the per-subject lock", async () => { + const job = await jobFor({ rkey: rkey() }); + const workflow = new FakeAssessmentWorkflow(); + const deps = { ...(await buildDeps()), assessmentWorkflow: workflow }; + + const first = new FakeMessage(); + const second = new FakeMessage(); + await processDiscoveryMessage(job, first, { ...deps, verify: verifiedFor(job) }); + await processDiscoveryMessage(job, second, { ...deps, verify: verifiedFor(job) }); + + // Second create collides on the deterministic id; dispatch converges + // rather than starting a duplicate run, and the message still acks. + expect(workflow.created).toHaveLength(1); + expect(first.acked).toBe(1); + expect(second.acked).toBe(1); + expect(second.retried).toBe(0); + }); + + it("distinct subjects (a new CID) get distinct Workflow instances", async () => { + const name = rkey(); + const workflow = new FakeAssessmentWorkflow(); + const deps = { ...(await buildDeps()), assessmentWorkflow: workflow }; + const jobV1 = await jobFor({ rkey: name, cid: await cid(`${name}-wf-v1`) }); + const jobV2 = { ...jobV1, operation: "update" as const, cid: await cid(`${name}-wf-v2`) }; + + await processDiscoveryMessage(jobV1, new FakeMessage(), { + ...deps, + verify: verifiedFor(jobV1), + }); + await processDiscoveryMessage(jobV2, new FakeMessage(), { + ...deps, + verify: verifiedFor(jobV2), + }); + + expect(workflow.created).toHaveLength(2); + expect(workflow.created[0]?.id).not.toBe(workflow.created[1]?.id); + }); + + it("retries (no dead letter) when dispatch fails, then re-dispatches on redelivery", async () => { + const job = await jobFor({ rkey: rkey() }); + const workflow = new FakeAssessmentWorkflow(); + workflow.createError = new Error("workflows backend unavailable"); + const deps = { ...(await buildDeps()), assessmentWorkflow: workflow }; + + const first = new FakeMessage(); + await processDiscoveryMessage(job, first, { ...deps, verify: verifiedFor(job) }); + + // Dispatch failed with no surviving instance → retry, not dead-letter. + expect(first.retried).toBe(1); + expect(first.acked).toBe(0); + expect(workflow.created).toHaveLength(0); + const dl = await testEnv.DB.prepare(`SELECT COUNT(*) AS n FROM dead_letters WHERE rkey = ?`) + .bind(job.rkey) + .first<{ n: number }>(); + expect(dl?.n).toBe(0); + // The run still reached pending before the dispatch failure. + const pending = await getAssessmentByRunKey(testEnv.DB, await runKeyFor(job)); + expect(pending?.state).toBe("pending"); + + // Redelivery with the backend recovered converges: same run, one instance. + workflow.createError = undefined; + const second = new FakeMessage(); + await processDiscoveryMessage(job, second, { ...deps, verify: verifiedFor(job) }); + expect(second.acked).toBe(1); + expect(workflow.created).toHaveLength(1); + const expectedId = await assessmentWorkflowInstanceId(uriFor(job), job.cid); + expect(workflow.created[0]?.id).toBe(expectedId); + }); +}); + describe("processDiscoveryMessage: verification failures", () => { it("permanent verify failure (INVALID_PROOF) dead-letters and acks — nothing else happens", async () => { const job = await jobFor({ rkey: rkey() }); diff --git a/apps/labeler/test/production-boundary.test.ts b/apps/labeler/test/production-boundary.test.ts index a350df8a9..7a6482117 100644 --- a/apps/labeler/test/production-boundary.test.ts +++ b/apps/labeler/test/production-boundary.test.ts @@ -1,13 +1,16 @@ /** - * Binding decision: production wiring stops at `assessment-pending`. Nothing - * in a production code path may construct or call `AssessmentOrchestrator` - * until W7/W8 land real stage adapters — it exists only to be exercised by - * `assessment-orchestrator.test.ts`. + * Boundary invariant: `AssessmentOrchestrator` reaches production through + * exactly one door — `AssessmentWorkflow` (assessment-workflow.ts), which + * constructs it per run. The queue consumer, discovery DO, Jetstream ingestor, + * reconciliation pass, and worker entry must never construct or call it + * directly; their job ends at dispatching a Workflow instance. This keeps the + * per-subject serialization in one place (the Workflow instance id) rather than + * letting an ad-hoc call site drive a run outside the lock. * - * `?raw` pulls each production entry point's source as a string (a Vite - * import query, supported under `@cloudflare/vitest-pool-workers`'s Vite - * pipeline) so this check runs without any filesystem access at test time — - * a static grep for the module specifier, not a runtime behavioural test. + * `?raw` pulls each entry point's source as a string (a Vite import query, + * supported under `@cloudflare/vitest-pool-workers`'s Vite pipeline) so this + * check runs without any filesystem access at test time — a static grep for the + * module specifier, not a runtime behavioural test. */ import { describe, expect, it } from "vitest"; @@ -39,9 +42,9 @@ const PRODUCTION_ENTRY_POINTS: Record = { const IMPORTS_ORCHESTRATOR = /(?:from\s+["'][^"']*assessment-orchestrator[^"']*["']|import\(\s*["'][^"']*assessment-orchestrator|import\s+["'][^"']*assessment-orchestrator)/; -describe("production boundary: AssessmentOrchestrator is test-only", () => { +describe("production boundary: AssessmentOrchestrator is reached only via the Workflow", () => { for (const [name, source] of Object.entries(PRODUCTION_ENTRY_POINTS)) { - it(`${name} never imports assessment-orchestrator`, () => { + it(`${name} never imports assessment-orchestrator directly`, () => { expect(source).not.toMatch(IMPORTS_ORCHESTRATOR); }); } diff --git a/apps/labeler/worker-configuration.d.ts b/apps/labeler/worker-configuration.d.ts index 5d654f933..fbd732422 100644 --- a/apps/labeler/worker-configuration.d.ts +++ b/apps/labeler/worker-configuration.d.ts @@ -1,5 +1,5 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types` (hash: e21c438219f0e3a8c154bb09b450524b) +// Generated by Wrangler by running `wrangler types` (hash: 5adbc65d092da13a1fa972a4426062c0) // Runtime types generated with workerd@1.20260708.1 2026-02-24 nodejs_compat interface __BaseEnv_Env { EVIDENCE: R2Bucket; @@ -18,6 +18,7 @@ interface __BaseEnv_Env { LABEL_SUBSCRIPTION: DurableObjectNamespace; LABELER_DISCOVERY_DO: DurableObjectNamespace; AGGREGATOR: Fetcher /* emdash-aggregator */; + ASSESSMENT_WORKFLOW: Workflow[0]['payload']>; } declare namespace Cloudflare { interface GlobalProps { diff --git a/apps/labeler/wrangler.jsonc b/apps/labeler/wrangler.jsonc index 924b5aeed..d8fe8a471 100644 --- a/apps/labeler/wrangler.jsonc +++ b/apps/labeler/wrangler.jsonc @@ -52,6 +52,18 @@ "service": "emdash-aggregator", }, ], + // One Workflow instance per subject drives an assessment run to + // finalization (a durable shell over AssessmentOrchestrator). The instance + // id derives from the subject (uri, cid), so a duplicate subject cannot run + // concurrently — that is the spec §14.1 per-subject lock. See + // src/assessment-workflow.ts and src/assessment-dispatch.ts. + "workflows": [ + { + "name": "emdash-labeler-assessment", + "binding": "ASSESSMENT_WORKFLOW", + "class_name": "AssessmentWorkflow", + }, + ], // Operator console SPA (apps/labeler/console), built to ./dist/console by // `console:build`. `run_worker_first: true` keeps the Worker the sole // router: the SPA fallback only fires where index.ts explicitly calls From e3f426bfedab83c07cbc01722dd58ef7711044ae Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Thu, 16 Jul 2026 12:45:11 +0100 Subject: [PATCH 2/3] fix(labeler): runKey-derived Workflow id + resumable-from-running MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the adversary pass on the Workflow shell. Fix B (Matt's ruling — instance id from runKey, not subject): a subject-only id (SHA-256 of uri+cid) is permanent (Cloudflare retains instance ids ~30 days for completed AND failed runs), so it blocked all re-assessment — operator rerun / intel re-triggers mint a new run (new runKey, which already includes triggerId) for the same subject and would collide with the retained id and strand the run pending forever. The instance id is now the run's runKey itself (already a 64-char SHA-256 hex from computeRunKey, reused verbatim — no second formula). Same-trigger discovery redelivery → same runKey → same id → dedup; distinct triggers → distinct runKeys → distinct instances → re-assessment works. Two different triggers for one subject can now run concurrently — that falls back to the currency/supersede + idempotency-keyed issuance path, not the instance lock. Fix A (resumable-from-running): any failure after the pending→running CAS (reachable even with stub stages) left the row `running`; the step retry re-entered runAssessment, whose pending-guard threw again → run stranded `running` forever. runAssessment now accepts a `running` row and resumes it to finalization (stages recomputed, findings held in memory not persisted mid-run, label issuance idempotency-keyed). Minimal orchestrator change: broaden the entry guard + skip the CAS when already `running`. Fix C (deploy gate): with stub stages, a run finalizes `passed` AND issues a real signed assessment-passed label for every subject. Corrected the orchestrator header to say so and added a prominent DEPLOY GATE note in the workflow docblock — the shell must not reach an enforcing/consumed deployment until real stages land. Also: corrected the stale console-mutation rerun comment (discovery now dispatches) and reconciled the orchestrator's finalize race comments to the per-run (runKey) dedup semantics. Track D (delete/cancel/signing-flip vs finalize) remains gated on the in-batch assessment-state guard. Tests: runKey-id dispatch semantics (dedup on same runKey, distinct runKey → own instance); shell resume from a running row; orchestrator resume after a post-transition failure then success. --- apps/labeler/src/assessment-dispatch.ts | 73 ++++++++++--------- apps/labeler/src/assessment-orchestrator.ts | 60 +++++++++------ apps/labeler/src/assessment-workflow.ts | 36 +++++---- apps/labeler/src/console-mutation-api.ts | 8 +- apps/labeler/src/discovery-consumer.ts | 26 +++---- apps/labeler/test/assessment-dispatch.test.ts | 62 ++++++---------- .../test/assessment-orchestrator.test.ts | 33 +++++++++ apps/labeler/test/assessment-workflow.test.ts | 12 +++ apps/labeler/test/discovery-consumer.test.ts | 10 +-- apps/labeler/wrangler.jsonc | 10 +-- 10 files changed, 191 insertions(+), 139 deletions(-) diff --git a/apps/labeler/src/assessment-dispatch.ts b/apps/labeler/src/assessment-dispatch.ts index f269ec72b..c708c7c70 100644 --- a/apps/labeler/src/assessment-dispatch.ts +++ b/apps/labeler/src/assessment-dispatch.ts @@ -1,14 +1,28 @@ /** - * Dispatch seam between verified discovery and the assessment Workflow. + * Dispatch seam between a created assessment run and its Workflow. * - * Each subject runs as one Cloudflare Workflow instance whose id is a - * deterministic function of (uri, cid). That id IS the spec §14.1 per-subject - * lock: `create` throws when an instance with the id already exists, so a - * duplicate subject cannot start a second concurrent run — resolving the - * queue-vs-Workflow serialization question deferred in #1978. - * `dispatchAssessmentWorkflow` treats that collision as an idempotent no-op and - * surfaces genuine infrastructure failures as `AssessmentDispatchError` for the - * caller to retry. + * 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 @@ -40,49 +54,38 @@ export class AssessmentDispatchError extends Error { override readonly name = "AssessmentDispatchError"; } -/** - * Deterministic Workflow instance id for a subject: SHA-256 hex of (uri, cid). - * Derived from the subject alone — not the run key — so any run for the same - * subject maps to the same instance and is serialized by it. 64 hex chars, well - * within the 100-char instance-id limit. - */ -export async function assessmentWorkflowInstanceId(uri: string, cid: string): Promise { - const material = `${uri}\n${cid}`; - const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(material)); - return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""); -} - export interface DispatchAssessmentInput { - uri: string; - cid: string; + /** The run's `runKey`, used verbatim as the Workflow instance id. */ + runKey: string; assessmentId: string; } /** - * Create the subject's Workflow instance, or converge if it already exists. - * Returns `"created"` on a fresh dispatch and `"exists"` when the deterministic - * id was already taken (the lock held — no second run). A `create` failure with - * no surviving instance is a real infrastructure error and throws + * 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 = await assessmentWorkflowInstanceId(input.uri, input.cid); + 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 (the lock) and on transient - // infra failures alike. Disambiguate by probing for the instance: present - // means the collision is the lock (idempotent no-op); absent means the - // create genuinely failed and the caller must retry. + // `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 ${input.uri}`, { - cause: err, - }); + throw new AssessmentDispatchError( + `failed to dispatch assessment Workflow for run ${input.assessmentId}`, + { cause: err }, + ); } } diff --git a/apps/labeler/src/assessment-orchestrator.ts b/apps/labeler/src/assessment-orchestrator.ts index 5cf17048d..fb033086f 100644 --- a/apps/labeler/src/assessment-orchestrator.ts +++ b/apps/labeler/src/assessment-orchestrator.ts @@ -4,14 +4,16 @@ * §9.9, §10). * * Driven in production by `AssessmentWorkflow` (assessment-workflow.ts): each - * verified subject runs as one Workflow instance whose id serializes duplicate - * runs of that subject (the §14.1 per-subject lock — see - * assessment-dispatch.ts). The Workflow constructs this orchestrator per run - * and calls `runAssessment`. Until W7/W8 supply the real stage adapters + * 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 a real subject finalizes as `passed`, clearing only its own - * `assessment-pending`. + * 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"; @@ -121,14 +123,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; @@ -166,11 +178,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. The instance-per-subject Workflow lock - // (assessment-dispatch.ts) serializes duplicate runs of the same subject - // but 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. + // 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 }); @@ -304,10 +316,10 @@ export class AssessmentOrchestrator { // flips, but the state CAS below is not, so a flip after this point could // commit the terminal state without its labels. // - // The instance-per-subject Workflow lock closes concurrent duplicate runs - // of a subject, but NOT a delete, operator cancel, or signing flip racing - // an in-flight run, so three related gaps in this method remain open until - // that in-batch state guard lands (tracked with the real-stage wiring): + // The Workflow instance id (the runKey) dedups only redelivery of the same + // run, NOT a delete, operator cancel, or signing flip racing an in-flight + // run, so three related gaps in this method remain open until that in-batch + // state guard lands (tracked with the real-stage wiring): // - a stale (CID-superseded) run returns without negating its own // assessment-pending, so a superseded subject keeps advertising an // in-progress assessment; diff --git a/apps/labeler/src/assessment-workflow.ts b/apps/labeler/src/assessment-workflow.ts index a0248e95b..c5ba54695 100644 --- a/apps/labeler/src/assessment-workflow.ts +++ b/apps/labeler/src/assessment-workflow.ts @@ -1,20 +1,26 @@ /** - * Production assessment execution: one Cloudflare Workflow instance per - * subject, a thin durable shell over `AssessmentOrchestrator`. The instance id - * (see `assessment-dispatch.ts`) serializes runs per subject; `run` loads the - * pending assessment and drives it through the orchestrator's stages and atomic - * finalization inside a single durable step. + * 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 from `pending`, and because `runAssessment` guards - * on `pending`, a row already advanced to `running` cannot be resumed here; the - * reconciliation pass (reconciliation.ts) surfaces such stuck runs. Finer, - * per-stage durable resume lands with the real-stage wiring. + * 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"; @@ -61,10 +67,10 @@ export async function executeAssessmentInstance( ): Promise { const existing = await getAssessment(env.DB, assessmentId); if (!existing) throw new Error(`assessment ${assessmentId} not found`); - // Idempotent resume: a previous execution already finalized this run (an - // eviction after the finalization batch committed but before the step result - // was durable). Return the terminal state rather than re-entering the - // orchestrator, whose pending-guard would throw on a non-pending row. + // 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); diff --git a/apps/labeler/src/console-mutation-api.ts b/apps/labeler/src/console-mutation-api.ts index a47b916b3..85c1cec51 100644 --- a/apps/labeler/src/console-mutation-api.ts +++ b/apps/labeler/src/console-mutation-api.ts @@ -545,9 +545,11 @@ function deferPublishAll(deps: ConsoleMutationDeps, issuanceKeys: readonly strin * `POST /admin/api/assessments/:id/rerun` — mints the immutable operator trigger * (`operator:`), creates a fresh run for the assessment's exact URI+CID * anchored to that trigger, and re-issues `assessment-pending`, all in one atomic - * batch with the audit row (spec §10/§11.2). Production wiring stops at pending - * (as initial discovery does today), so the run sits at `pending` until W7/W8 - * supply stage adapters. + * batch with the audit row (spec §10/§11.2). Initial discovery now dispatches an + * assessment Workflow after `pending`; this rerun path still stops at `pending` + * (its own Workflow dispatch is a follow-on). The operator trigger yields a + * distinct `runKey`, so the rerun maps to its own Workflow instance id rather + * than colliding with the prior run's — the re-assessment is not stranded. */ async function runRerun( request: Request, diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index ebf40ea1d..f109804cc 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/apps/labeler/src/discovery-consumer.ts @@ -3,12 +3,12 @@ * verified subject, an idempotent assessment run, an `assessment-pending` * label, and a dispatched assessment Workflow instance — spec §9.1 steps 5-7. * This file's job ends at dispatch: it never constructs - * `AssessmentOrchestrator` itself. The instance id derives from the subject - * (assessment-dispatch.ts), so a duplicate subject cannot start a second - * concurrent run — that collision is the spec §14.1 per-subject lock. - * `AssessmentWorkflow` (assessment-workflow.ts) then drives `pending → running - * → finalization` through the orchestrator (still stub stages until W7/W8 land - * real stage adapters). + * `AssessmentOrchestrator` itself. The instance id is the run's runKey + * (assessment-dispatch.ts), so a redelivered discovery event dedups onto the + * same instance while a later re-assessment (distinct trigger → distinct + * runKey) gets its own. `AssessmentWorkflow` (assessment-workflow.ts) then + * drives `pending → running → finalization` through the orchestrator (still + * stub stages until W7/W8 land real stage adapters). * * Error policy mirrors the aggregator's records-consumer with one * difference (spec §9.1): a forged or unverifiable event is ALWAYS a dead @@ -89,8 +89,9 @@ export interface DiscoveryConsumerDeps { signer: LabelSigner; didDocumentResolver: DidDocumentResolverLike; /** The assessment Workflow binding. Once a verified subject reaches - * `pending`, the consumer dispatches one instance per subject; the - * deterministic instance id is the per-subject lock (assessment-dispatch.ts). */ + * `pending`, the consumer dispatches its run; the instance id is the run's + * runKey, so a redelivered event dedups onto the same instance + * (assessment-dispatch.ts). */ assessmentWorkflow: AssessmentWorkflowBinding; fetch?: typeof fetch; now?: () => Date; @@ -400,14 +401,13 @@ async function verifyAndCreateRun( now, ); - // Hand the run to its Workflow instance. The instance id derives from the - // subject, so a redelivered event converges on the same instance rather than - // starting a second run (the §14.1 lock). A dispatch infra failure throws + // Hand the run to its Workflow instance. The instance id is the run's runKey, + // so a redelivered event (same runKey) converges on the same instance rather + // than starting a second run. A dispatch infra failure throws // AssessmentDispatchError → the message retries; upstream steps are all // idempotent, so redelivery re-dispatches. await dispatchAssessmentWorkflow(deps.assessmentWorkflow, { - uri, - cid: job.cid, + runKey, assessmentId: assessment.id, }); } diff --git a/apps/labeler/test/assessment-dispatch.test.ts b/apps/labeler/test/assessment-dispatch.test.ts index ac212b3ef..f9d593525 100644 --- a/apps/labeler/test/assessment-dispatch.test.ts +++ b/apps/labeler/test/assessment-dispatch.test.ts @@ -2,15 +2,15 @@ import { describe, expect, it } from "vitest"; import { AssessmentDispatchError, - assessmentWorkflowInstanceId, dispatchAssessmentWorkflow, type AssessmentWorkflowBinding, type AssessmentWorkflowParams, } from "../src/assessment-dispatch.js"; -const URI = - "at://did:plc:publisher000000000000000000/com.emdashcms.experimental.package.release/pkg:1.0.0"; -const CID = "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"; +// Stand-in runKeys (real ones are 64-char SHA-256 hex from computeRunKey). Two +// distinct values model same-subject-different-trigger re-assessment. +const RUNKEY_A = "a".repeat(64); +const RUNKEY_B = "b".repeat(64); interface Recorded { id: string; @@ -49,60 +49,46 @@ class FakeWorkflow implements AssessmentWorkflowBinding { } } -describe("assessmentWorkflowInstanceId", () => { - it("is deterministic and 64 lowercase hex chars", async () => { - const a = await assessmentWorkflowInstanceId(URI, CID); - const b = await assessmentWorkflowInstanceId(URI, CID); - expect(a).toBe(b); - expect(a).toMatch(/^[0-9a-f]{64}$/); - }); - - it("derives from both uri and cid — either changing changes the id", async () => { - const base = await assessmentWorkflowInstanceId(URI, CID); - expect(await assessmentWorkflowInstanceId(`${URI}x`, CID)).not.toBe(base); - expect(await assessmentWorkflowInstanceId(URI, `${CID}x`)).not.toBe(base); - }); - - it("does not collide across a uri/cid boundary shift", async () => { - // `a\nb` vs `a` + `\nb` must not hash the same: guards the delimiter. - expect(await assessmentWorkflowInstanceId("a", "b")).not.toBe( - await assessmentWorkflowInstanceId("a\nb", ""), - ); - }); -}); - describe("dispatchAssessmentWorkflow", () => { - it("creates the instance with the derived id and returns 'created'", async () => { + it("creates the instance with the runKey as its id and returns 'created'", async () => { const workflow = new FakeWorkflow(); const outcome = await dispatchAssessmentWorkflow(workflow, { - uri: URI, - cid: CID, + runKey: RUNKEY_A, assessmentId: "asmt_1", }); expect(outcome).toBe("created"); - const expectedId = await assessmentWorkflowInstanceId(URI, CID); - expect(workflow.created).toEqual([{ id: expectedId, params: { assessmentId: "asmt_1" } }]); + expect(workflow.created).toEqual([{ id: RUNKEY_A, params: { assessmentId: "asmt_1" } }]); }); - it("returns 'exists' when the id is already taken — the lock held", async () => { + it("returns 'exists' when the same runKey is dispatched again — redelivery dedup", async () => { const workflow = new FakeWorkflow(); - await dispatchAssessmentWorkflow(workflow, { uri: URI, cid: CID, assessmentId: "asmt_1" }); + await dispatchAssessmentWorkflow(workflow, { runKey: RUNKEY_A, assessmentId: "asmt_1" }); const outcome = await dispatchAssessmentWorkflow(workflow, { - uri: URI, - cid: CID, + runKey: RUNKEY_A, assessmentId: "asmt_1", }); expect(outcome).toBe("exists"); expect(workflow.created).toHaveLength(1); }); + it("a distinct runKey (re-assessment's new trigger) dispatches its own instance", async () => { + const workflow = new FakeWorkflow(); + // Same subject, different trigger → different runKey → must not collide. + await dispatchAssessmentWorkflow(workflow, { runKey: RUNKEY_A, assessmentId: "asmt_1" }); + const outcome = await dispatchAssessmentWorkflow(workflow, { + runKey: RUNKEY_B, + assessmentId: "asmt_2", + }); + expect(outcome).toBe("created"); + expect(workflow.created.map((c) => c.id)).toEqual([RUNKEY_A, RUNKEY_B]); + }); + it("returns 'exists' when create fails but the instance survives", async () => { const workflow = new FakeWorkflow(); workflow.failCreate = new Error("ack lost"); workflow.persistOnFailure = true; const outcome = await dispatchAssessmentWorkflow(workflow, { - uri: URI, - cid: CID, + runKey: RUNKEY_A, assessmentId: "asmt_1", }); expect(outcome).toBe("exists"); @@ -112,7 +98,7 @@ describe("dispatchAssessmentWorkflow", () => { const workflow = new FakeWorkflow(); workflow.failCreate = new Error("backend down"); await expect( - dispatchAssessmentWorkflow(workflow, { uri: URI, cid: CID, assessmentId: "asmt_1" }), + dispatchAssessmentWorkflow(workflow, { runKey: RUNKEY_A, assessmentId: "asmt_1" }), ).rejects.toBeInstanceOf(AssessmentDispatchError); }); }); diff --git a/apps/labeler/test/assessment-orchestrator.test.ts b/apps/labeler/test/assessment-orchestrator.test.ts index c38d2c25d..ac33d44a8 100644 --- a/apps/labeler/test/assessment-orchestrator.test.ts +++ b/apps/labeler/test/assessment-orchestrator.test.ts @@ -707,6 +707,39 @@ describe("AssessmentOrchestrator: invalid findings", () => { }); }); +describe("AssessmentOrchestrator: resume from running", () => { + it("resumes a run left `running` by a crashed attempt and finalizes on the next call", async () => { + const run = await pendingRun({ name: "resume-running", cidValue: await cid("resume-running") }); + let attempts = 0; + const flaky: StageAdapter = () => { + attempts += 1; + // Attempt 1: a non-transient failure after the pending→running CAS + // (models a crash in a later stage or in finalize). Later attempts pass. + if (attempts === 1) throw new Error("stage crashed post-transition"); + return Promise.resolve([]); + }; + const orchestrator = await buildOrchestrator({ ...stubStages, deterministic: flaky }); + + await expect(orchestrator.runAssessment(run.id)).rejects.toThrow( + "stage crashed post-transition", + ); + expect((await getAssessment(testEnv.DB, run.id))?.state).toBe("running"); + + // Re-invoking (as the durable step retry does) resumes the `running` row + // and finalizes passed — no pending-guard rejection, no duplicate labels. + const finalized = await orchestrator.runAssessment(run.id); + expect(finalized.state).toBe("passed"); + + const pending = await testEnv.DB.prepare( + `SELECT l.neg FROM issued_labels l JOIN issuance_actions a ON a.id = l.action_id + WHERE l.uri = ? AND l.cid = ? AND l.val = 'assessment-pending'`, + ) + .bind(run.uri, run.cid) + .first<{ neg: number }>(); + expect(pending?.neg).toBe(1); + }); +}); + describe("AssessmentOrchestrator: history stage never auto-labels (W8.4)", () => { it("runs the history stage, whose finding surfaces but is never turned into an issued label", async () => { const run = await pendingRun({ diff --git a/apps/labeler/test/assessment-workflow.test.ts b/apps/labeler/test/assessment-workflow.test.ts index 09555add9..27a51c1e4 100644 --- a/apps/labeler/test/assessment-workflow.test.ts +++ b/apps/labeler/test/assessment-workflow.test.ts @@ -129,6 +129,18 @@ describe("executeAssessmentInstance", () => { expect(await executeAssessmentInstance(workflowEnv, run.id)).toBe("passed"); }); + it("resumes a run left `running` by a crashed attempt and finalizes it", async () => { + const run = await pendingRun("wf-running-resume"); + // Simulate a prior attempt that made the pending→running CAS then crashed + // (e.g. a durable step evicted mid-finalize) before finalizing. + await transitionAssessmentState(testEnv.DB, { id: run.id, from: "pending", to: "running" }); + + const state = await executeAssessmentInstance(workflowEnv, run.id); + + expect(state).toBe("passed"); + expect((await getAssessment(testEnv.DB, run.id))?.state).toBe("passed"); + }); + it("throws when the assessment does not exist", async () => { await expect( executeAssessmentInstance(workflowEnv, "asmt_00000000000000000000000000"), diff --git a/apps/labeler/test/discovery-consumer.test.ts b/apps/labeler/test/discovery-consumer.test.ts index 057262255..e76529aa8 100644 --- a/apps/labeler/test/discovery-consumer.test.ts +++ b/apps/labeler/test/discovery-consumer.test.ts @@ -9,7 +9,6 @@ import { applyD1Migrations, env } from "cloudflare:test"; import { afterEach, beforeAll, describe, expect, it } from "vitest"; import { - assessmentWorkflowInstanceId, type AssessmentWorkflowBinding, type AssessmentWorkflowParams, } from "../src/assessment-dispatch.js"; @@ -306,10 +305,10 @@ describe("processDiscoveryMessage: Workflow dispatch and per-subject lock", () = await processDiscoveryMessage(job, msg, { ...deps, verify: verifiedFor(job) }); expect(msg.acked).toBe(1); - const assessment = await getAssessmentByRunKey(testEnv.DB, await runKeyFor(job)); - const expectedId = await assessmentWorkflowInstanceId(uriFor(job), job.cid); + const runKey = await runKeyFor(job); + const assessment = await getAssessmentByRunKey(testEnv.DB, runKey); expect(workflow.created).toHaveLength(1); - expect(workflow.created[0]?.id).toBe(expectedId); + expect(workflow.created[0]?.id).toBe(runKey); expect(workflow.created[0]?.params.assessmentId).toBe(assessment!.id); }); @@ -378,8 +377,7 @@ describe("processDiscoveryMessage: Workflow dispatch and per-subject lock", () = await processDiscoveryMessage(job, second, { ...deps, verify: verifiedFor(job) }); expect(second.acked).toBe(1); expect(workflow.created).toHaveLength(1); - const expectedId = await assessmentWorkflowInstanceId(uriFor(job), job.cid); - expect(workflow.created[0]?.id).toBe(expectedId); + expect(workflow.created[0]?.id).toBe(await runKeyFor(job)); }); }); diff --git a/apps/labeler/wrangler.jsonc b/apps/labeler/wrangler.jsonc index d8fe8a471..6aeeeb753 100644 --- a/apps/labeler/wrangler.jsonc +++ b/apps/labeler/wrangler.jsonc @@ -52,11 +52,11 @@ "service": "emdash-aggregator", }, ], - // One Workflow instance per subject drives an assessment run to - // finalization (a durable shell over AssessmentOrchestrator). The instance - // id derives from the subject (uri, cid), so a duplicate subject cannot run - // concurrently — that is the spec §14.1 per-subject lock. See - // src/assessment-workflow.ts and src/assessment-dispatch.ts. + // One Workflow instance per assessment run drives it to finalization (a + // durable shell over AssessmentOrchestrator). The instance id is the run's + // runKey, so a redelivered discovery event dedups onto the same instance + // while a re-assessment (distinct trigger → distinct runKey) gets its own. + // See src/assessment-workflow.ts and src/assessment-dispatch.ts. "workflows": [ { "name": "emdash-labeler-assessment", From c46a9ccfe3ab69744e7e6e9f049ece587c521455 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Thu, 16 Jul 2026 13:47:58 +0100 Subject: [PATCH 3/3] fix(labeler): close finalization label-leak race + mechanical deploy gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address emdashbot review findings on #2072. Finding 2 (finalization CAS race): the run→toState CAS and the label inserts shared one db.batch, but a 0-row CAS (concurrent cancel/delete moved the run out of `running`) did not roll back the inserts — signed assessment-passed labels leaked for a cancelled subject, and a post-hoc re-read could not un-leak them. Now every finalization issuance action is gated in-batch on the assessment reaching `toState` (buildIssuanceStatements gains an optional requireAssessmentState; the label insert selects from the action, so gating the action gates the label). The batch is all-or-nothing against the CAS: a cancel/delete that no-ops the CAS also no-ops every label. The CAS row count is checked after commit — 0 means the race was lost, so the postCommits (which would otherwise mis-diagnose the absent label as a signing failure and throw) are skipped and AssessmentFinalizationConflictError is raised; the Workflow step retries and short-circuits on the now-terminal row. The existing second isSubjectCurrent check only covers delete/supersede down to the check-to-commit gap (the delete path tombstones the subject); it does not cover an operator cancel that leaves the subject current. The in-batch state guard closes both. Finding 1 (deploy gate): buildStages() now throws when import.meta.env.PROD, so a production build with only stub stages fails loudly instead of signing 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); a minimal ambient declaration types it without pulling in vite/client browser globals. Test: a stage that cancels the run mid-flight proves finalization issues no labels and throws the conflict. --- apps/labeler/calibration/tsconfig.json | 2 +- apps/labeler/src/assessment-orchestrator.ts | 74 +++++++++++++------ apps/labeler/src/assessment-workflow.ts | 9 +++ apps/labeler/src/service.ts | 67 ++++++++++++----- apps/labeler/src/vite-env.d.ts | 15 ++++ .../test/assessment-orchestrator.test.ts | 37 ++++++++++ 6 files changed, 161 insertions(+), 43 deletions(-) create mode 100644 apps/labeler/src/vite-env.d.ts diff --git a/apps/labeler/calibration/tsconfig.json b/apps/labeler/calibration/tsconfig.json index 61162e2ca..2c915190f 100644 --- a/apps/labeler/calibration/tsconfig.json +++ b/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"] } diff --git a/apps/labeler/src/assessment-orchestrator.ts b/apps/labeler/src/assessment-orchestrator.ts index fb033086f..a2d8e7f21 100644 --- a/apps/labeler/src/assessment-orchestrator.ts +++ b/apps/labeler/src/assessment-orchestrator.ts @@ -52,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; } @@ -271,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); @@ -305,29 +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 an - // in-batch assessment-state guard (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. // - // That same guard is what a signing-state flip during the batch needs: - // 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. + // 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. // - // The Workflow instance id (the runKey) dedups only redelivery of the same - // run, NOT a delete, operator cancel, or signing flip racing an in-flight - // run, so three related gaps in this method remain open until that in-batch - // state guard lands (tracked with the real-stage wiring): - // - 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, @@ -342,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); diff --git a/apps/labeler/src/assessment-workflow.ts b/apps/labeler/src/assessment-workflow.ts index c5ba54695..de108a1cf 100644 --- a/apps/labeler/src/assessment-workflow.ts +++ b/apps/labeler/src/assessment-workflow.ts @@ -96,5 +96,14 @@ export async function executeAssessmentInstance( * 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; } diff --git a/apps/labeler/src/service.ts b/apps/labeler/src/service.ts index 19ac4e066..1256fcd6f 100644 --- a/apps/labeler/src/service.ts +++ b/apps/labeler/src/service.ts @@ -1,6 +1,6 @@ import type { LabelSigner, SignedLabel } from "@emdash-cms/registry-moderation"; -import { ASSESSMENT_ID } from "./assessment-lifecycle.js"; +import { ASSESSMENT_ID, type AssessmentState } from "./assessment-lifecycle.js"; import { getNegatableAutomatedLabels } from "./assessment-store.js"; import type { LabelerConfig } from "./config.js"; import type { FindingSeverity } from "./evidence.js"; @@ -110,6 +110,18 @@ export interface IssuanceStatements { postCommit: () => Promise; } +export interface BuildIssuanceOptions { + /** + * Gate the action insert (and thus its label) on the assessment row being in + * this state at commit time. Finalization passes its `toState` so the whole + * batch — the run→toState CAS and every label — is all-or-nothing against that + * transition: a concurrent cancel/delete that no-ops the CAS also no-ops every + * label, so no label leaks for a run that is no longer `running`. Requires an + * automated-assessment action (it carries the assessmentId to check). + */ + requireAssessmentState?: AssessmentState; +} + /** * Builds the two INSERT statements that record an issuance action and its * signed label, sharing the same signing/rotation guards for both the @@ -127,6 +139,7 @@ export async function buildIssuanceStatements( proposal: IssuanceProposal, now: Date, publicationPending: boolean, + options: BuildIssuanceOptions = {}, ): Promise { const signingStatus = await getSigningStatusIfInitialized(db); if (signingStatus?.phase === "paused") { @@ -190,6 +203,37 @@ export async function buildIssuanceStatements( // selects from an action that was never written. const isAutomatedNegation = action.type === "automated-assessment" && proposal.neg === true; + // Optional finalization guard: gate the action insert on the assessment still + // being in the required state at commit time. The label insert selects from + // this action, so gating the action gates the label too — a no-op action + // leaves no orphan label (same pattern as the §10 negation guard above). + const requireState = options.requireAssessmentState; + if (requireState !== undefined && action.type !== "automated-assessment") + throw new TypeError("requireAssessmentState requires an automated-assessment action"); + const stateGuardSql = + requireState === undefined + ? "" + : `\n\t\t\t\t AND EXISTS (SELECT 1 FROM assessments WHERE id = ? AND state = ?)`; + const actionBinds: unknown[] = [ + action.actor, + action.type, + action.reason, + action.idempotencyKey, + assessmentId, + now.toISOString(), + isPrebootstrap ? 1 : 0, + isPrebootstrap ? 1 : 0, + config.signingKeyVersion, + isAutomatedNegation ? 1 : 0, + signer.issuerDid, + proposal.uri, + proposal.val, + signer.issuerDid, + proposal.uri, + proposal.val, + ]; + if (requireState !== undefined) actionBinds.push(assessmentId, requireState); + const statements: D1PreparedStatement[] = [ db .prepare( @@ -213,27 +257,10 @@ export async function buildIssuanceStatements( ) AND l2.neg = 0 AND a2.type <> 'automated-assessment' ) - ) + )${stateGuardSql} ON CONFLICT(idempotency_key) DO NOTHING`, ) - .bind( - action.actor, - action.type, - action.reason, - action.idempotencyKey, - assessmentId, - now.toISOString(), - isPrebootstrap ? 1 : 0, - isPrebootstrap ? 1 : 0, - config.signingKeyVersion, - isAutomatedNegation ? 1 : 0, - signer.issuerDid, - proposal.uri, - proposal.val, - signer.issuerDid, - proposal.uri, - proposal.val, - ), + .bind(...actionBinds), db .prepare( `INSERT INTO issued_labels diff --git a/apps/labeler/src/vite-env.d.ts b/apps/labeler/src/vite-env.d.ts new file mode 100644 index 000000000..3e5510e7f --- /dev/null +++ b/apps/labeler/src/vite-env.d.ts @@ -0,0 +1,15 @@ +/** + * Minimal typing for the Vite compile-time env constants this Worker reads. + * The full `vite/client` types pull in browser globals the Worker tsconfig does + * not want, and these are the only members used. Vite statically replaces + * `import.meta.env.PROD` at build time (true in `vite build`, false in dev and + * the vitest pool), so it must be written as that literal to be replaced. + */ +interface ImportMetaEnv { + readonly PROD: boolean; + readonly DEV: boolean; +} + +interface ImportMeta { + readonly env: ImportMetaEnv; +} diff --git a/apps/labeler/test/assessment-orchestrator.test.ts b/apps/labeler/test/assessment-orchestrator.test.ts index ac33d44a8..8e07bb486 100644 --- a/apps/labeler/test/assessment-orchestrator.test.ts +++ b/apps/labeler/test/assessment-orchestrator.test.ts @@ -14,6 +14,7 @@ import { } from "../src/artifact-acquisition.js"; import { computeRunKey, initialTriggerId, operatorTriggerId } from "../src/assessment-lifecycle.js"; import { + AssessmentFinalizationConflictError, AssessmentOrchestrator, StageTransientError, stubStages, @@ -740,6 +741,42 @@ describe("AssessmentOrchestrator: resume from running", () => { }); }); +describe("AssessmentOrchestrator: cancel racing finalization", () => { + it("issues no labels and throws when a cancel moves the run out of running before commit", async () => { + const run = await pendingRun({ name: "cancel-race", cidValue: await cid("cancel-race") }); + // A stage that cancels the run mid-flight simulates a delete/operator cancel + // landing after the currency check but before the finalization batch commits. + const cancelDuringStage: StageAdapter = async () => { + await transitionAssessmentState(testEnv.DB, { + id: run.id, + from: "running", + to: "cancelled", + }); + return []; + }; + const orchestrator = await buildOrchestrator({ + ...stubStages, + deterministic: cancelDuringStage, + }); + + await expect(orchestrator.runAssessment(run.id)).rejects.toBeInstanceOf( + AssessmentFinalizationConflictError, + ); + + // The run stays cancelled and NO label was issued — not the positive + // outcome label, not even the assessment-pending negation. + expect((await getAssessment(testEnv.DB, run.id))?.state).toBe("cancelled"); + 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); + }); +}); + describe("AssessmentOrchestrator: history stage never auto-labels (W8.4)", () => { it("runs the history stage, whose finding surfaces but is never turned into an issued label", async () => { const run = await pendingRun({