diff --git a/package.json b/package.json
index 07bb3d6f..595d2206 100644
--- a/package.json
+++ b/package.json
@@ -111,7 +111,8 @@
     "test:redaction": "tsx tests/redaction.test.ts",
     "test:browser-preview-routing": "tsx --test tests/browser-preview-routing.test.ts",
     "test:browser-routed-command-security": "tsx --test tests/browser-routed-command-security.test.ts",
-    "test:cloudflare-runtime": "node --test tests/cloudflare-d1-provisioner.test.mjs && tsx tests/cloudflare-site-context.test.ts && tsx tests/cloudflare-coordinator-site-partitioning.test.ts && tsx tests/cloudflare-d1-operation-repository.test.ts && tsx tests/cloudflare-allocation-lifecycle.test.ts && tsx tests/cloudflare-provisioning-api.test.ts && tsx tests/cloudflare-runtime.test.ts && tsx tests/cloudflare-public-reader.test.ts && node ./node_modules/typescript/bin/tsc -p packages/runtime-cloudflare --noEmit",
+    "test:cloudflare-runtime": "node --test tests/cloudflare-d1-provisioner.test.mjs && tsx tests/cloudflare-site-context.test.ts && tsx tests/cloudflare-coordinator-site-partitioning.test.ts && tsx tests/cloudflare-d1-operation-repository.test.ts && tsx tests/cloudflare-allocation-lifecycle.test.ts && tsx tests/cloudflare-provisioning-api.test.ts && tsx tests/cloudflare-queue-batch.test.ts && tsx tests/cloudflare-runtime.test.ts && tsx tests/cloudflare-public-reader.test.ts && node ./node_modules/typescript/bin/tsc -p packages/runtime-cloudflare --noEmit",
+    "test:cloudflare-queue": "tsx tests/cloudflare-queue-batch.test.ts && tsx tests/cloudflare-d1-operation-repository.test.ts && node ./node_modules/typescript/bin/tsc -p packages/runtime-cloudflare --noEmit",
     "test:cloudflare-administrator-claim": "tsx tests/cloudflare-provisioning-api.test.ts",
     "test:cloudflare-wordpress-auth": "tsx tests/cloudflare-wordpress-auth.test.ts",
     "test:cloudflare-wordpress-archive-corpus": "tsx tests/cloudflare-wordpress-archive-corpus.test.ts",
diff --git a/packages/runtime-cloudflare/src/allocation-lifecycle.ts b/packages/runtime-cloudflare/src/allocation-lifecycle.ts
index 47d82af8..f7215de2 100644
--- a/packages/runtime-cloudflare/src/allocation-lifecycle.ts
+++ b/packages/runtime-cloudflare/src/allocation-lifecycle.ts
@@ -102,6 +102,7 @@ export class CloudflareAllocationLifecycle {
       ])
     }
     if (await tableExists(this.database, "wp_codebox_api_admin_claims")) await this.database.prepare("UPDATE wp_codebox_api_admin_claims SET state = 'expired', updated_at = ? WHERE site_id = ? AND state = 'pending'").bind(now, identity.siteId).run()
+    if (await tableExists(this.database, "wp_codebox_runtime_dispatches")) await this.database.prepare("UPDATE wp_codebox_runtime_dispatches SET state = 'done', last_error = 'Allocation deletion fenced this dispatch.', completed_at = ?, updated_at = ? WHERE site_id = ? AND state NOT IN ('done','dead-letter')").bind(now, now, identity.siteId).run()
   }
   private async cleanupWork(siteId: string, limit: number): Promise<{ deleted: number; unresolved: number }> {
     let deleted = 0
@@ -114,7 +115,10 @@ export class CloudflareAllocationLifecycle {
     if (await tableExists(this.database, "wp_codebox_api_admin_claims")) {
       const result = await this.database.prepare("DELETE FROM wp_codebox_api_admin_claims WHERE site_id = ?").bind(siteId).run(); deleted += result.meta.changes
     }
-    const tables = ["wp_codebox_operation_attempts", "wp_codebox_operations", "wp_codebox_api_admin_claims"]
+    for (const table of ["wp_codebox_runtime_dead_letters", "wp_codebox_runtime_dispatches", "wp_codebox_runtime_fairness"]) if (await tableExists(this.database, table)) {
+      const result = await this.database.prepare(`DELETE FROM ${table} WHERE site_id = ?`).bind(siteId).run(); deleted += result.meta.changes
+    }
+    const tables = ["wp_codebox_operation_attempts", "wp_codebox_operations", "wp_codebox_api_admin_claims", "wp_codebox_runtime_dead_letters", "wp_codebox_runtime_dispatches", "wp_codebox_runtime_fairness"]
     let unresolved = 0
     for (const table of tables) if (await tableExists(this.database, table)) unresolved += (await this.database.prepare(`SELECT COUNT(*) AS count FROM ${table} WHERE site_id = ?`).bind(siteId).first<{ count: number }>())?.count ?? 0
     return { deleted, unresolved }
diff --git a/packages/runtime-cloudflare/src/d1-operation-repository.ts b/packages/runtime-cloudflare/src/d1-operation-repository.ts
index f1d41587..5f2fa932 100644
--- a/packages/runtime-cloudflare/src/d1-operation-repository.ts
+++ b/packages/runtime-cloudflare/src/d1-operation-repository.ts
@@ -1,5 +1,6 @@
 import type { MarkdownPointer } from "./revision-coordinator.js"
 import type { SiteContext } from "./site-context.js"
+import { runtimeQueueMessage, type RuntimeQueueMessage, type RuntimeQueuePolicy } from "./queue-dispatch.js"
 
 export const STATIC_ARTIFACT_OPERATION_SCHEMA = "wp-codebox/cloudflare-static-artifact-operation/v1"
 export type StaticArtifactOperationState = "queued" | "running" | "retryable" | "publication-pending" | "succeeded" | "failed"
@@ -24,6 +25,7 @@ export interface StaticArtifactOperation {
 }
 
 export class OperationConflict extends Error {}
+export class DispatchBackpressure extends Error {}
 const schemaReady = new WeakMap<object, Promise<void>>()
 
 export function shouldRecoverPreparedCommit(prepared: StaticArtifactOperation["prepared"], committed: MarkdownPointer | null): boolean {
@@ -77,6 +79,83 @@ export class D1OperationRepository {
     return rows.results.map((row) => ({ id: row.site_id, hostname: row.hostname, origin: row.origin }))
   }
 
+  async stageDispatch(site: SiteContext, kind: RuntimeQueueMessage["kind"], identity: string, principal: string): Promise<RuntimeQueueMessage> {
+    await ensureSchema(this.database)
+    if (!/^[A-Za-z0-9._:@-]{1,128}$/.test(principal)) throw new OperationConflict("Runtime dispatch principal is invalid.")
+    const row = await this.database.prepare("SELECT generation FROM wp_codebox_sites WHERE site_id = ? AND state = 'active'").bind(site.id).first<{ generation: number }>()
+    if (!row) throw new OperationConflict("The site is not active for queue dispatch.")
+    const message = runtimeQueueMessage(site, row.generation, kind, identity)
+    const now = Date.now()
+    await this.database.prepare("INSERT OR IGNORE INTO wp_codebox_runtime_dispatches (site_id, generation, kind, identity, principal, state, attempts, created_at, updated_at) VALUES (?, ?, ?, ?, ?, 'backpressured', 0, ?, ?)").bind(site.id, row.generation, kind, identity, principal, now, now).run()
+    const stored = await this.database.prepare("SELECT principal FROM wp_codebox_runtime_dispatches WHERE site_id = ? AND generation = ? AND kind = ? AND identity = ?").bind(site.id, row.generation, kind, identity).first<{ principal: string }>()
+    if (stored?.principal !== principal) throw new OperationConflict("Runtime dispatch identity is bound to another principal.")
+    return message
+  }
+
+  async admitDispatch(message: RuntimeQueueMessage, policy: RuntimeQueuePolicy): Promise<boolean> {
+    await ensureSchema(this.database)
+    const now = Date.now()
+    const result = await this.database.prepare(`UPDATE wp_codebox_runtime_dispatches SET state = 'pending', retry_at = NULL, updated_at = ? WHERE site_id = ? AND generation = ? AND kind = ? AND identity = ? AND state = 'backpressured' AND (SELECT COUNT(*) FROM wp_codebox_runtime_dispatches WHERE state IN ('pending','sent','processing','retryable')) < ? AND (SELECT COUNT(*) FROM wp_codebox_runtime_dispatches active WHERE active.principal = wp_codebox_runtime_dispatches.principal AND active.state IN ('pending','sent','processing','retryable')) < ?`).bind(now, message.siteId, message.generation, message.kind, message.identity, policy.maxActive, policy.maxActivePerPrincipal).run()
+    if (result.meta.changes === 1) return true
+    const state = await this.dispatchState(message)
+    return state !== null && state !== "backpressured" && state !== "dead-letter"
+  }
+
+  async dispatchState(message: RuntimeQueueMessage): Promise<string | null> { await ensureSchema(this.database); const row = await this.database.prepare("SELECT state FROM wp_codebox_runtime_dispatches WHERE site_id = ? AND generation = ? AND kind = ? AND identity = ?").bind(message.siteId, message.generation, message.kind, message.identity).first<{ state: string }>(); return row?.state ?? null }
+  async dispatchPrincipal(message: RuntimeQueueMessage): Promise<string | null> { await ensureSchema(this.database); const row = await this.database.prepare("SELECT principal FROM wp_codebox_runtime_dispatches WHERE site_id = ? AND generation = ? AND kind = ? AND identity = ?").bind(message.siteId, message.generation, message.kind, message.identity).first<{ principal: string }>(); return row?.principal ?? null }
+  async repairMissingDispatches(limit = 32, now = Date.now()): Promise<void> {
+    await ensureSchema(this.database)
+    const principal = "COALESCE(l.principal, 'system:' || o.site_id)"
+    await this.database.prepare(`INSERT OR IGNORE INTO wp_codebox_runtime_dispatches (site_id, generation, kind, identity, principal, state, attempts, created_at, updated_at) SELECT o.site_id, s.generation, 'operation', o.operation_id, ${principal}, 'backpressured', 0, ?, ? FROM wp_codebox_operations o JOIN wp_codebox_sites s ON s.site_id = o.site_id LEFT JOIN wp_codebox_site_lifecycles l ON l.site_id = s.site_id AND l.generation = s.generation WHERE (o.state IN ('queued','retryable') OR (o.state = 'running' AND o.claim_expires_at <= ?)) ORDER BY o.updated_at, o.site_id LIMIT ?`).bind(now, now, now, limit).run()
+    await this.database.prepare(`INSERT OR IGNORE INTO wp_codebox_runtime_dispatches (site_id, generation, kind, identity, principal, state, attempts, created_at, updated_at) SELECT o.site_id, s.generation, 'publication', o.prepared_publication_job, ${principal}, 'backpressured', 0, ?, ? FROM wp_codebox_operations o JOIN wp_codebox_sites s ON s.site_id = o.site_id LEFT JOIN wp_codebox_site_lifecycles l ON l.site_id = s.site_id AND l.generation = s.generation WHERE o.state = 'publication-pending' AND o.prepared_publication_job IS NOT NULL ORDER BY o.updated_at, o.site_id LIMIT ?`).bind(now, now, limit).run()
+  }
+
+  async matchesGeneration(siteId: string, generation: number): Promise<boolean> { await ensureSchema(this.database); const row = await this.database.prepare("SELECT generation FROM wp_codebox_sites WHERE site_id = ? AND state = 'active'").bind(siteId).first<{ generation: number }>(); return row?.generation === generation }
+
+  async deliveredDispatch(message: RuntimeQueueMessage): Promise<void> { await ensureSchema(this.database); await this.database.prepare("UPDATE wp_codebox_runtime_dispatches SET state = 'sent', attempts = attempts + 1, sent_at = ?, last_error = NULL, updated_at = ? WHERE site_id = ? AND generation = ? AND kind = ? AND identity = ? AND state IN ('pending','retryable')").bind(Date.now(), Date.now(), message.siteId, message.generation, message.kind, message.identity).run() }
+  async processingDispatch(message: RuntimeQueueMessage, now = Date.now()): Promise<boolean> { await ensureSchema(this.database); const result = await this.database.prepare("UPDATE wp_codebox_runtime_dispatches SET state = 'processing', processing_at = ?, updated_at = ? WHERE site_id = ? AND generation = ? AND kind = ? AND identity = ? AND (state IN ('pending','sent') OR (state = 'retryable' AND retry_at <= ?))").bind(now, now, message.siteId, message.generation, message.kind, message.identity, now).run(); return result.meta.changes === 1 }
+  async retryDispatch(message: RuntimeQueueMessage, error: unknown): Promise<void> { await ensureSchema(this.database); const now = Date.now(); await this.database.prepare("UPDATE wp_codebox_runtime_dispatches SET state = 'retryable', attempts = attempts + 1, last_error = ?, retry_at = ?, updated_at = ? WHERE site_id = ? AND generation = ? AND kind = ? AND identity = ? AND state != 'dead-letter'").bind(sanitizeError(error).message, now + 30_000, now, message.siteId, message.generation, message.kind, message.identity).run() }
+  async failedDispatch(message: RuntimeQueueMessage, error: unknown): Promise<void> { await ensureSchema(this.database); const now = Date.now(); await this.database.prepare("UPDATE wp_codebox_runtime_dispatches SET state = 'backpressured', attempts = attempts + 1, last_error = ?, retry_at = NULL, updated_at = ? WHERE site_id = ? AND generation = ? AND kind = ? AND identity = ? AND state NOT IN ('done','dead-letter')").bind(sanitizeError(error).message, now, message.siteId, message.generation, message.kind, message.identity).run() }
+  async completeDispatch(message: RuntimeQueueMessage): Promise<void> { await ensureSchema(this.database); await this.database.prepare("UPDATE wp_codebox_runtime_dispatches SET state = 'done', completed_at = ?, updated_at = ? WHERE site_id = ? AND generation = ? AND kind = ? AND identity = ? AND state != 'dead-letter'").bind(Date.now(), Date.now(), message.siteId, message.generation, message.kind, message.identity).run() }
+  async deadDispatch(message: RuntimeQueueMessage, attempts: number, error: unknown): Promise<void> {
+    await ensureSchema(this.database)
+    const failure = sanitizeError(error); const now = Date.now()
+    const input = message.kind === "operation" ? await this.database.prepare("SELECT fingerprint FROM wp_codebox_operations WHERE site_id = ? AND operation_id = ?").bind(message.siteId, message.identity).first<{ fingerprint: string }>() : null
+    const digest = input?.fingerprint ?? message.identity.split("-").at(-1)?.replace(/\.json$/, "") ?? message.identity
+    const command = recoveryCommand(message)
+    const [evidence, dispatch] = await this.database.batch([
+      this.database.prepare("INSERT OR IGNORE INTO wp_codebox_runtime_dead_letters (site_id, generation, kind, identity, attempts, input_digest, last_error, recovery_command, recorded_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)").bind(message.siteId, message.generation, message.kind, message.identity, attempts, digest, failure.message, command, now),
+      this.database.prepare("UPDATE wp_codebox_runtime_dispatches SET state = 'dead-letter', attempts = ?, last_error = ?, completed_at = ?, updated_at = ? WHERE site_id = ? AND generation = ? AND kind = ? AND identity = ? AND state != 'done'").bind(attempts, failure.message, now, now, message.siteId, message.generation, message.kind, message.identity),
+    ])
+    if (evidence.meta.changes !== 1 || dispatch.meta.changes !== 1) throw new OperationConflict("Runtime dead-letter transition did not converge.")
+  }
+  async pendingDispatches(policy: RuntimeQueuePolicy, limit = 32, now = Date.now()): Promise<RuntimeQueueMessage[]> {
+    await ensureSchema(this.database)
+    await this.database.prepare("UPDATE wp_codebox_runtime_dispatches SET state = 'backpressured', updated_at = ? WHERE state IN ('sent','processing') AND updated_at <= ?").bind(now, now - this.claimMs).run()
+    const rows = await this.database.prepare("SELECT d.site_id, d.generation, d.kind, d.identity, d.state FROM wp_codebox_runtime_dispatches d JOIN wp_codebox_sites s ON s.site_id = d.site_id AND s.generation = d.generation WHERE d.state IN ('backpressured','pending','retryable') AND s.state = 'active' AND (d.retry_at IS NULL OR d.retry_at <= ?) ORDER BY d.updated_at, d.site_id, d.kind, d.identity LIMIT ?").bind(now, limit).all<{ site_id: string; generation: number; kind: RuntimeQueueMessage["kind"]; identity: string; state: string }>()
+    const admitted: RuntimeQueueMessage[] = []
+    for (const row of rows.results) { const message = runtimeQueueMessage({ id: row.site_id }, row.generation, row.kind, row.identity); if (row.state !== "backpressured" || await this.admitDispatch(message, policy)) admitted.push(message) }
+    return admitted
+  }
+  async deadLetter(message: RuntimeQueueMessage, attempts: number, error: unknown): Promise<void> {
+    await ensureSchema(this.database)
+    const failure = sanitizeError(error)
+    const input = message.kind === "operation" ? await this.database.prepare("SELECT fingerprint FROM wp_codebox_operations WHERE site_id = ? AND operation_id = ?").bind(message.siteId, message.identity).first<{ fingerprint: string }>() : null
+    const digest = input?.fingerprint ?? message.identity.split("-").at(-1) ?? message.identity
+    const command = recoveryCommand(message)
+    await this.database.prepare("INSERT INTO wp_codebox_runtime_dead_letters (site_id, generation, kind, identity, attempts, input_digest, last_error, recovery_command, recorded_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)").bind(message.siteId, message.generation, message.kind, message.identity, attempts, digest, failure.message, command, Date.now()).run()
+  }
+  async deadLetterEvidence(message: RuntimeQueueMessage): Promise<{ attempts: number; inputDigest: string; lastError: string; recoveryCommand: string } | null> { await ensureSchema(this.database); const row = await this.database.prepare("SELECT attempts, input_digest, last_error, recovery_command FROM wp_codebox_runtime_dead_letters WHERE site_id = ? AND generation = ? AND kind = ? AND identity = ? ORDER BY attempts DESC LIMIT 1").bind(message.siteId, message.generation, message.kind, message.identity).first<{ attempts: number; input_digest: string; last_error: string; recovery_command: string }>(); return row ? { attempts: row.attempts, inputDigest: row.input_digest, lastError: row.last_error, recoveryCommand: row.recovery_command } : null }
+
+  async selectLaneDispatch(siteId: string, messages: RuntimeQueueMessage[]): Promise<RuntimeQueueMessage | null> {
+    await ensureSchema(this.database)
+    const state = await this.database.prepare("SELECT next_kind FROM wp_codebox_runtime_fairness WHERE site_id = ?").bind(siteId).first<{ next_kind: RuntimeQueueMessage["kind"] }>()
+    const preferred = state?.next_kind ?? "operation"
+    const selected = messages.find((message) => message.kind === preferred) ?? messages[0] ?? null
+    if (selected) await this.database.prepare("INSERT INTO wp_codebox_runtime_fairness (site_id, next_kind, updated_at) VALUES (?, ?, ?) ON CONFLICT(site_id) DO UPDATE SET next_kind = excluded.next_kind, updated_at = excluded.updated_at").bind(siteId, selected.kind === "operation" ? "publication" : "operation", Date.now()).run()
+    return selected
+  }
+
   async claimNext(siteId: string, now = Date.now()): Promise<(StaticArtifactOperation & { claimToken: string }) | null> {
     await ensureSchema(this.database)
     // A dynamic allocation may be fenced between queue deliveries. Configured
@@ -96,6 +175,27 @@ export class D1OperationRepository {
     return { ...operation, claimToken: token }
   }
 
+  async claimOperation(siteId: string, operationId: string, now = Date.now()): Promise<(StaticArtifactOperation & { claimToken: string }) | null> {
+    await ensureSchema(this.database)
+    const candidate = await this.database.prepare("SELECT operation_id FROM wp_codebox_operations WHERE site_id = ? AND operation_id = ? AND (state = 'queued' OR (state = 'retryable' AND retry_at <= ?) OR (state = 'running' AND claim_expires_at <= ?))").bind(siteId, operationId, now, now).first<{ operation_id: string }>()
+    if (!candidate) return null
+    // claimNext owns the atomic claim implementation; temporarily target its one candidate.
+    return this.claimExact(siteId, operationId, now)
+  }
+
+  private async claimExact(siteId: string, operationId: string, now: number): Promise<(StaticArtifactOperation & { claimToken: string }) | null> {
+    const token = crypto.randomUUID(); const expiresAt = now + this.claimMs
+    const [update, attempt] = await this.database.batch([
+      this.database.prepare(`UPDATE wp_codebox_operations SET state = 'running', stage = 'claimed', claim_token = ?, claim_expires_at = ?, attempts = attempts + 1, retry_at = NULL, error_code = NULL, error_message = NULL, updated_at = ? WHERE site_id = ? AND operation_id = ? AND (state = 'queued' OR (state = 'retryable' AND retry_at <= ?) OR (state = 'running' AND claim_expires_at <= ?)) AND EXISTS (SELECT 1 FROM wp_codebox_sites s LEFT JOIN wp_codebox_site_lifecycles l ON l.site_id = s.site_id AND l.generation = s.generation WHERE s.site_id = wp_codebox_operations.site_id AND (l.site_id IS NULL OR (l.state = 'active' AND l.expires_at > ? AND l.operation_fence = wp_codebox_operations.allocation_fence)))`).bind(token, expiresAt, now, siteId, operationId, now, now, now),
+      this.database.prepare("INSERT INTO wp_codebox_operation_attempts (site_id, operation_id, attempt_number, claim_token, started_at, state, stage) SELECT site_id, operation_id, attempts, ?, ?, 'running', 'claimed' FROM wp_codebox_operations WHERE site_id = ? AND operation_id = ? AND state = 'running' AND claim_token = ? AND claim_expires_at = ?").bind(token, new Date(now).toISOString(), siteId, operationId, token, expiresAt),
+    ])
+    if (update.meta.changes !== 1) return null
+    if (attempt.meta.changes !== 1) throw new OperationConflict("Operation attempt creation did not match its claim.")
+    const operation = await this.get(siteId, operationId)
+    if (!operation) throw new Error("Claimed operation is unavailable.")
+    return { ...operation, claimToken: token }
+  }
+
   async renew(siteId: string, operationId: string, token: string): Promise<void> {
     await this.ownedUpdate(siteId, operationId, token, `UPDATE wp_codebox_operations SET claim_expires_at = ?, updated_at = ?`, [Date.now() + this.claimMs, Date.now()])
   }
@@ -188,6 +288,7 @@ function operationFromRow(row: Row, attempts: AttemptRow[]): StaticArtifactOpera
   return { schema: STATIC_ARTIFACT_OPERATION_SCHEMA, site: { id: row.site_id, hostname: row.hostname, origin: row.origin }, operationId: row.operation_id, state: row.state, input, attempts: row.attempts, attemptHistory: attempts.map((attempt) => ({ number: attempt.attempt_number, startedAt: attempt.started_at, completedAt: attempt.completed_at, state: attempt.state, stage: attempt.stage, error: attempt.error_code ? { code: attempt.error_code, message: attempt.error_message ?? "Operation failed." } : null })), stage: row.stage, progress: row.progress, retryAt: row.retry_at, claimExpiresAt: row.claim_expires_at, prepared, error: row.error_code ? { code: row.error_code, message: row.error_message ?? "Operation failed." } : null, receipt: row.receipt_json ? JSON.parse(row.receipt_json) as OperationReceipt : null }
 }
 function sanitizeError(error: unknown): { code: string; message: string } { const message = error instanceof Error ? error.message : "Operation failed."; return { code: error instanceof OperationConflict ? "conflict" : "execution_failed", message: message.replace(/[\r\n\t]/g, " ").slice(0, 500) } }
+function recoveryCommand(message: RuntimeQueueMessage): string { return `npx wrangler d1 execute wp-codebox-runtime-state --remote --command "UPDATE wp_codebox_runtime_dispatches SET state='backpressured', retry_at=NULL, completed_at=NULL, updated_at=unixepoch()*1000 WHERE site_id='${message.siteId}' AND generation=${message.generation} AND kind='${message.kind}' AND identity='${message.identity}' AND state='dead-letter'"` }
 async function ensureSchema(database: D1Database): Promise<void> {
   const existing = schemaReady.get(database as object)
   if (!existing) {
@@ -204,6 +305,10 @@ async function ensureSchema(database: D1Database): Promise<void> {
       await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_operation_attempts (site_id TEXT NOT NULL, operation_id TEXT NOT NULL, attempt_number INTEGER NOT NULL, claim_token TEXT NOT NULL, started_at TEXT NOT NULL, completed_at TEXT, state TEXT NOT NULL, stage TEXT NOT NULL, error_code TEXT, error_message TEXT, PRIMARY KEY (site_id, operation_id, attempt_number), FOREIGN KEY (site_id, operation_id) REFERENCES wp_codebox_operations(site_id, operation_id))`).run()
       await database.prepare(`CREATE UNIQUE INDEX IF NOT EXISTS wp_codebox_one_active_operation ON wp_codebox_operations(site_id) WHERE state IN ('queued','running','retryable')`).run()
       await database.prepare(`CREATE INDEX IF NOT EXISTS wp_codebox_operation_ready ON wp_codebox_operations(site_id, state, retry_at, claim_expires_at, created_at)`).run()
+      await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_runtime_dispatches (site_id TEXT NOT NULL, generation INTEGER NOT NULL, kind TEXT NOT NULL CHECK (kind IN ('operation','publication')), identity TEXT NOT NULL, principal TEXT NOT NULL, state TEXT NOT NULL CHECK (state IN ('backpressured','pending','sent','processing','retryable','done','dead-letter')), attempts INTEGER NOT NULL DEFAULT 0, last_error TEXT, retry_at INTEGER, sent_at INTEGER, processing_at INTEGER, completed_at INTEGER, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (site_id, generation, kind, identity))`).run()
+      await database.prepare(`CREATE INDEX IF NOT EXISTS wp_codebox_runtime_dispatch_pending ON wp_codebox_runtime_dispatches(state, updated_at)`).run()
+      await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_runtime_dead_letters (site_id TEXT NOT NULL, generation INTEGER NOT NULL, kind TEXT NOT NULL, identity TEXT NOT NULL, attempts INTEGER NOT NULL, input_digest TEXT NOT NULL, last_error TEXT NOT NULL, recovery_command TEXT NOT NULL, recorded_at INTEGER NOT NULL, PRIMARY KEY (site_id, generation, kind, identity, attempts))`).run()
+      await database.prepare(`CREATE TABLE IF NOT EXISTS wp_codebox_runtime_fairness (site_id TEXT PRIMARY KEY, next_kind TEXT NOT NULL CHECK (next_kind IN ('operation','publication')), updated_at INTEGER NOT NULL)`).run()
     })()
     schemaReady.set(database as object, pending)
     pending.catch(() => schemaReady.delete(database as object))
diff --git a/packages/runtime-cloudflare/src/provisioning-api.ts b/packages/runtime-cloudflare/src/provisioning-api.ts
index 36134a19..8d5be16d 100644
--- a/packages/runtime-cloudflare/src/provisioning-api.ts
+++ b/packages/runtime-cloudflare/src/provisioning-api.ts
@@ -1,4 +1,5 @@
 import { D1OperationRepository, OperationConflict, type StaticArtifactOperation, type StaticArtifactOperationInput } from "./d1-operation-repository.js"
+import { parseRuntimeQueuePolicy, runtimeQueueMessage, type RuntimeQueue } from "./queue-dispatch.js"
 import { MAX_STATIC_ARTIFACT_BYTES, readBoundedRequestBytes, readStaticArtifactImport, StaticArtifactImportError, validateStaticArtifact } from "./static-artifact-import.js"
 import { allocatePreviewSiteContext, parseSiteContexts, previewDomain, siteStorageKeys, type SiteContext } from "./site-context.js"
 import { deriveSiteCredential } from "./wordpress-auth.js"
@@ -17,7 +18,7 @@ export interface ProvisioningAllocation {
   artifactSha256: string; artifactSize: number; options: StaticArtifactOperationInput["options"]
 }
 interface CreateInput { key: string; fingerprint: string; artifactSha256: string; artifactSize: number; options: StaticArtifactOperationInput["options"] }
-export interface ProvisioningEnv { WORDPRESS_STATE_DATABASE: D1Database; WORDPRESS_STATE_BUCKET: R2Bucket; WORDPRESS_SITE_CONTEXTS?: string; WORDPRESS_PREVIEW_DOMAIN?: string; WORDPRESS_PREVIEW_HOST_SECRET?: string; WORDPRESS_API_TOKENS?: string; WORDPRESS_ADMIN_CLAIM_SECRET?: string; WORDPRESS_ADMIN_PASSWORD?: string }
+export interface ProvisioningEnv { WORDPRESS_STATE_DATABASE: D1Database; WORDPRESS_STATE_BUCKET: R2Bucket; WORDPRESS_RUNTIME_QUEUE?: RuntimeQueue; WORDPRESS_QUEUE_POLICY?: string; WORDPRESS_SITE_CONTEXTS?: string; WORDPRESS_PREVIEW_DOMAIN?: string; WORDPRESS_PREVIEW_HOST_SECRET?: string; WORDPRESS_API_TOKENS?: string; WORDPRESS_ADMIN_CLAIM_SECRET?: string; WORDPRESS_ADMIN_PASSWORD?: string }
 
 export async function routeProvisioningApi(request: Request, env: ProvisioningEnv, operations: D1OperationRepository): Promise<Response> {
   const parts = new URL(request.url).pathname.split("/").filter(Boolean)
@@ -38,6 +39,12 @@ export async function routeProvisioningApi(request: Request, env: ProvisioningEn
   return notFound()
 }
 
+async function dispatchOperation(env: ProvisioningEnv, operations: D1OperationRepository, site: SiteContext, operation: StaticArtifactOperation, principal: string): Promise<"queued" | "backpressured"> {
+  const message = await operations.stageDispatch(site, "operation", operation.operationId, principal)
+  if (!await operations.admitDispatch(message, parseRuntimeQueuePolicy(env.WORDPRESS_QUEUE_POLICY))) return "backpressured"
+  try { if (!env.WORDPRESS_RUNTIME_QUEUE) throw new Error("Runtime queue binding is unavailable."); await env.WORDPRESS_RUNTIME_QUEUE.send(message); await operations.deliveredDispatch(message); return "queued" } catch (error) { await operations.failedDispatch(message, error); console.error("Runtime queue producer is backpressured; reconciliation will retry.", error); return "backpressured" }
+}
+
 async function lifecycleStatus(request: Request, env: ProvisioningEnv, siteId: string): Promise<Response> {
   const token = await authenticate(request, env, "sites:read"); if (token instanceof Response) return token
   const allocation = await new AllocationStore(env.WORDPRESS_STATE_DATABASE).bySite(siteId)
@@ -110,7 +117,9 @@ async function create(request: Request, env: ProvisioningEnv, operations: D1Oper
   try {
     const operation = await resumeProvisioningAllocation(env, site, operations)
     const claim = await new AdministratorClaimStore(env.WORDPRESS_STATE_DATABASE).issue(allocation, env)
-    return siteResource(site, operation, 202, claim)
+    const response = siteResource(site, operation, 202, claim)
+    if (operation) response.headers.set("x-wp-codebox-dispatch", dispatchHeader(await operations.dispatchState(runtimeQueueMessage(site, allocationIdentity(site.id).generation, "operation", operation.operationId))))
+    return response
   } catch (error) { if (error instanceof OperationConflict) return apiError(409, "idempotency_conflict", error.message); if (error instanceof AdministratorClaimError) return apiError(409, "administrator_claim_unavailable", "The administrator claim cannot continue provisioning."); throw error }
 }
 
@@ -153,8 +162,11 @@ async function importSite(request: Request, env: ProvisioningEnv, operations: D1
     const input = await readStaticArtifactImport(request, env.WORDPRESS_STATE_BUCKET, site)
     if (input.idempotencyKey !== key) return apiError(409, "idempotency_conflict", "Idempotency-Key must match the import request.")
     const result = await operations.createOrConverge(site, { ...input, artifact: input.artifactReference })
+    const dispatch = await dispatchOperation(env, operations, site, result.operation, token.principal)
     await store.linkOperation(token.principal, siteId, result.operation.operationId, "import", key)
-    return operationResource(siteId, result.operation, 202)
+    const response = operationResource(siteId, await operations.get(siteId, result.operation.operationId) ?? result.operation, 202)
+    response.headers.set("x-wp-codebox-dispatch", dispatch)
+    return response
   } catch (error) { return error instanceof OperationConflict ? apiError(409, "operation_conflict", error.message) : importError(error) }
 }
 
@@ -187,9 +199,12 @@ export async function resumeProvisioningAllocation(env: ProvisioningEnv, site: S
   const result = await operations.createOrConverge(site, input)
   await store.bindOperation(allocation, result.operation.operationId)
   await store.linkOperation(allocation.principal, site.id, result.operation.operationId, "provision", allocation.key)
-  return result.operation
+  await dispatchOperation(env, operations, site, result.operation, allocation.principal)
+  return await operations.get(site.id, result.operation.operationId) ?? result.operation
 }
 
+function dispatchHeader(state: string | null): "queued" | "backpressured" { return state && !["backpressured", "dead-letter"].includes(state) ? "queued" : "backpressured" }
+
 async function readCreate(request: Request, bucket: R2Bucket, key: string): Promise<CreateInput> {
   const bytes = await readBoundedRequestBytes(request)
   let body: Record<string, unknown>; try { body = JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(bytes)) } catch { throw new StaticArtifactImportError("Provisioning request must be valid UTF-8 JSON.", 400) }
diff --git a/packages/runtime-cloudflare/src/queue-batch.ts b/packages/runtime-cloudflare/src/queue-batch.ts
new file mode 100644
index 00000000..dff2a6b9
--- /dev/null
+++ b/packages/runtime-cloudflare/src/queue-batch.ts
@@ -0,0 +1,33 @@
+import type { RuntimeQueueMessage } from "./queue-dispatch.js"
+
+export interface QueueDelivery { body: unknown; attempts: number; ack(): void; retry(options?: { delaySeconds?: number }): void }
+export interface ParsedQueueDelivery { raw: QueueDelivery; value: RuntimeQueueMessage }
+export type QueueExecutionResult = "ack" | "retry" | { retryAfterSeconds: number }
+
+/** One heavyweight turn per site keeps a delivery batch fair without cross-site serialization. */
+export async function dispatchQueueBatch(
+  deliveries: QueueDelivery[],
+  parse: (body: unknown) => RuntimeQueueMessage | null,
+  select: (siteId: string, deliveries: ParsedQueueDelivery[]) => Promise<ParsedQueueDelivery | null>,
+  execute: (delivery: ParsedQueueDelivery) => Promise<QueueExecutionResult>,
+): Promise<void> {
+  const lanes = new Map<string, ParsedQueueDelivery[]>()
+  for (const raw of deliveries) {
+    const value = parse(raw.body)
+    if (!value) { raw.ack(); continue }
+    const lane = lanes.get(value.siteId) ?? []
+    lane.push({ raw, value })
+    lanes.set(value.siteId, lane)
+  }
+  await Promise.all([...lanes.entries()].map(async ([siteId, lane]) => {
+    const selected = await select(siteId, lane)
+    for (const delivery of lane) {
+      if (delivery !== selected) delivery.raw.retry()
+    }
+    if (!selected) return
+    const result = await execute(selected)
+    if (result === "ack") selected.raw.ack()
+    else if (result === "retry") selected.raw.retry()
+    else selected.raw.retry({ delaySeconds: result.retryAfterSeconds })
+  }))
+}
diff --git a/packages/runtime-cloudflare/src/queue-dispatch.ts b/packages/runtime-cloudflare/src/queue-dispatch.ts
new file mode 100644
index 00000000..d49e8f3c
--- /dev/null
+++ b/packages/runtime-cloudflare/src/queue-dispatch.ts
@@ -0,0 +1,42 @@
+import type { SiteContext } from "./site-context.js"
+
+export const RUNTIME_QUEUE_MESSAGE_SCHEMA = "wp-codebox/runtime-dispatch/v1"
+export const RUNTIME_QUEUE_MAX_ATTEMPTS = 3
+export const DEFAULT_RUNTIME_QUEUE_POLICY: RuntimeQueuePolicy = { maxActive: 100, maxActivePerPrincipal: 10 }
+
+/** Queue messages wake durable work; they never own canonical state. */
+export type RuntimeQueueMessage = { schema: typeof RUNTIME_QUEUE_MESSAGE_SCHEMA; siteId: string; generation: number; kind: "operation" | "publication"; identity: string }
+export interface RuntimeQueuePolicy { maxActive: number; maxActivePerPrincipal: number }
+
+const operationIdentity = /^[0-9a-f]{8}-(?:[0-9a-f]{4}-){3}[0-9a-f]{12}$/
+const publicationIdentity = /^sites\/([a-z0-9][a-z0-9-]{0,127})\/publications\/jobs\/[0-9]{20}-[a-f0-9-]{36}\.json$/
+
+export function validRuntimeQueueIdentity(siteId: string, kind: RuntimeQueueMessage["kind"], identity: string): boolean {
+  if (kind === "operation") return operationIdentity.test(identity)
+  return publicationIdentity.test(identity) && publicationIdentity.exec(identity)?.[1] === siteId
+}
+
+export function runtimeQueueMessage(site: Pick<SiteContext, "id">, generation: number, kind: RuntimeQueueMessage["kind"], identity: string): RuntimeQueueMessage {
+  if (!/^[a-z0-9][a-z0-9-]{0,127}$/.test(site.id) || !Number.isSafeInteger(generation) || generation < 1 || !validRuntimeQueueIdentity(site.id, kind, identity)) throw new Error("Runtime queue dispatch identity is invalid.")
+  return { schema: RUNTIME_QUEUE_MESSAGE_SCHEMA, siteId: site.id, generation, kind, identity }
+}
+
+export function parseRuntimeQueueMessage(value: unknown): RuntimeQueueMessage | null {
+  if (!value || typeof value !== "object") return null
+  const message = value as Partial<RuntimeQueueMessage>
+  if (message.schema !== RUNTIME_QUEUE_MESSAGE_SCHEMA || (message.kind !== "operation" && message.kind !== "publication") || typeof message.siteId !== "string" || !/^[a-z0-9][a-z0-9-]{0,127}$/.test(message.siteId) || !Number.isSafeInteger(message.generation) || message.generation! < 1 || typeof message.identity !== "string" || !validRuntimeQueueIdentity(message.siteId, message.kind, message.identity)) return null
+  return message as RuntimeQueueMessage
+}
+
+export interface RuntimeQueue { send(message: RuntimeQueueMessage): Promise<void> }
+
+export function parseRuntimeQueuePolicy(value: string | undefined): RuntimeQueuePolicy {
+  if (!value) return DEFAULT_RUNTIME_QUEUE_POLICY
+  let policy: Partial<RuntimeQueuePolicy>
+  try { policy = JSON.parse(value) as Partial<RuntimeQueuePolicy> } catch { throw new Error("Runtime queue policy is invalid.") }
+  if (!Number.isSafeInteger(policy.maxActive) || !Number.isSafeInteger(policy.maxActivePerPrincipal)
+    || policy.maxActive! < 1 || policy.maxActive! > 10_000 || policy.maxActivePerPrincipal! < 1 || policy.maxActivePerPrincipal! > policy.maxActive!) {
+    throw new Error("Runtime queue policy is invalid.")
+  }
+  return policy as RuntimeQueuePolicy
+}
diff --git a/packages/runtime-cloudflare/src/worker.ts b/packages/runtime-cloudflare/src/worker.ts
index 4867f04b..7f0cf4a5 100644
--- a/packages/runtime-cloudflare/src/worker.ts
+++ b/packages/runtime-cloudflare/src/worker.ts
@@ -14,6 +14,8 @@ import { RevisionConflict, type MarkdownPointer, type MutationFence, type Revisi
 import { routeWorkerRequest } from "./request-routing.js"
 import { readStaticArtifactImport, STATIC_ARTIFACT_IMPORT_RESULT_SCHEMA, StaticArtifactImportError, type StaticArtifactImport } from "./static-artifact-import.js"
 import { D1OperationRepository, OperationConflict, STATIC_ARTIFACT_OPERATION_SCHEMA, shouldRecoverPreparedCommit } from "./d1-operation-repository.js"
+import { parseRuntimeQueueMessage, parseRuntimeQueuePolicy, type RuntimeQueue, type RuntimeQueueMessage } from "./queue-dispatch.js"
+import { dispatchQueueBatch, type ParsedQueueDelivery, type QueueDelivery } from "./queue-batch.js"
 import { resumeProvisioningAllocation, routeProvisioningApi } from "./provisioning-api.js"
 import { allocationIdentity, CloudflareAllocationLifecycle } from "./allocation-lifecycle.js"
 import { toFetchResponse, toPHPRequest } from "./request-translation.js"
@@ -216,6 +218,8 @@ export interface RuntimeEnv {
   WORDPRESS_OPERATOR_TOKEN?: string
   WORDPRESS_API_TOKENS?: string
   WORDPRESS_STATE_DATABASE?: D1Database
+  WORDPRESS_RUNTIME_QUEUE?: RuntimeQueue
+  WORDPRESS_QUEUE_POLICY?: string
 }
 
 export function createCloudflareRuntime<Env extends RuntimeEnv>(
@@ -293,33 +297,55 @@ export function createCloudflareRuntime<Env extends RuntimeEnv>(
         const deleting = (await lifecycle.pendingDeletions(1, controller.scheduledTime))[0]
         if (deleting) await lifecycle.reclaim(env.WORDPRESS_STATE_BUCKET, deleting.identity, deleting.operationFence, 100, controller.scheduledTime)
       }
+      const operations = resolveOperations?.(env)
+      // Cron repairs non-transactional producer sends and runs bounded lifecycle work;
+      // normal mutations and publications are only entered by Queue delivery.
+      if (operations && env.WORDPRESS_RUNTIME_QUEUE) {
+        await operations.repairMissingDispatches(32, controller.scheduledTime)
+        for (const message of await operations.pendingDispatches(parseRuntimeQueuePolicy(env.WORDPRESS_QUEUE_POLICY), 32)) {
+        try { await env.WORDPRESS_RUNTIME_QUEUE.send(message); await operations.deliveredDispatch(message) } catch (error) { await operations.failedDispatch(message, error); console.error("Runtime queue reconciliation is backpressured.", error) }
+        }
+      }
+      // WordPress due events remain a bounded cron responsibility. They do not
+      // drain provisioning or publication work, which is Queue-only above.
       const configured = parseSiteContexts(env.WORDPRESS_SITE_CONTEXTS)
-      const registered = env.WORDPRESS_STATE_DATABASE && resolveOperations?.(env) ? await resolveOperations(env)!.activeSites() : []
+      const registered = operations ? await operations.activeSites() : []
       const sites = [...new Map([...configured, ...registered].map((site) => [site.id, site])).values()].sort((left, right) => left.id.localeCompare(right.id))
-      if (!sites.length) return
-      const site = sites[Math.floor(controller.scheduledTime / 60_000) % sites.length]
-      const coordinator = resolveCoordinator(env, site)
+      if (sites.length) {
+        const site = sites[Math.floor(controller.scheduledTime / 60_000) % sites.length]
+        console.log(JSON.stringify({ siteId: site.id, ...await runScheduledWordPressCron(env, resolveCoordinator(env, site), site, controller.scheduledTime) }))
+      }
+    },
+    async queue(batch: { messages: QueueDelivery[] }, env: Env): Promise<void> {
       const operations = resolveOperations?.(env)
-      let provisioningReady = true
-      if (operations && env.WORDPRESS_STATE_DATABASE) {
+      if (!operations) { for (const message of batch.messages) message.ack(); return }
+      await dispatchQueueBatch(batch.messages, parseRuntimeQueueMessage, async (siteId, messages) => {
+        const selected = await operations.selectLaneDispatch(siteId, messages.map(({ value }) => value))
+        return selected ? messages.find(({ value }) => value.kind === selected.kind && value.identity === selected.identity) ?? null : null
+      }, async ({ raw, value }: ParsedQueueDelivery) => {
+        const siteId = value.siteId
+        const configured = parseSiteContexts(env.WORDPRESS_SITE_CONTEXTS)
+        const registered = await operations.activeSites()
+        const site = [...configured, ...registered].find((candidate) => candidate.id === siteId)
         try {
-          await resumeProvisioningAllocation(env as Env & { WORDPRESS_STATE_DATABASE: D1Database }, site, operations)
+          if (!site || !await operations.matchesGeneration(value.siteId, value.generation)) return "ack"
+          if (!await operations.processingDispatch(value)) return "ack"
+          const coordinator = resolveCoordinator(env, site)
+          const principal = await operations.dispatchPrincipal(value)
+          if (!principal) return "ack"
+          const result = value.kind === "operation" ? await runStaticArtifactOperation(env, coordinator, site, operations, value.identity, principal) : await drainPublicationJob(env, coordinator, site, operations, value.identity)
+          if (result && (result.status === "retryable" || result.status === "pending" || result.status === "rendered")) {
+            await operations.retryDispatch(value, new Error(`Dispatch ${result.status}.`))
+            return result.status === "rendered" ? "retry" : { retryAfterSeconds: 30 }
+          }
+          await operations.completeDispatch(value)
+          return "ack"
         } catch (error) {
-          provisioningReady = false
-          console.error("Provisioning allocation recovery failed.", error)
+          if (raw.attempts >= 3) { await operations.deadDispatch(value, raw.attempts, error); return "retry" }
+          await operations.retryDispatch(value, error)
+          return { retryAfterSeconds: 30 }
         }
-      }
-      const fence = await coordinator.fenceStatus()
-      const siteCycle = Math.floor(controller.scheduledTime / (60_000 * sites.length))
-      const operationFirst = siteCycle % 2 === 0
-      // Alternate priority while allowing only one heavyweight runtime per invocation.
-      let operation = operationFirst && provisioningReady && !fence.active ? await runNextStaticArtifactOperation(env, coordinator, site, operations) : null
-      const publication = operation ? null : await drainNextPublicationJob(env, coordinator, site, operations)
-      if (!operationFirst && provisioningReady && !publication && !fence.active) operation = await runNextStaticArtifactOperation(env, coordinator, site, operations)
-      const evidence = operation ?? publication ?? (fence.active
-        ? { schema: "wp-codebox/cloudflare-publication/v1" as const, status: "fenced" as const, jobKey: "coordinator" }
-        : await runScheduledWordPressCron(env, coordinator, site, controller.scheduledTime))
-      console.log(JSON.stringify({ siteId: site.id, ...evidence }))
+      })
     },
   }
 }
@@ -582,7 +608,8 @@ async function importCanonicalStaticArtifact(request: Request, env: RuntimeEnv,
   if (operations) {
     try {
       const created = await operations.createOrConverge(site, { ...input, artifact: input.artifactReference })
-      return Response.json(created.operation, { status: 202, headers: { location: `?phase=operator-static-artifact-operation&operationId=${created.operation.operationId}` } })
+      const dispatch = await dispatchRuntimeWork(env, operations, site, "operation", created.operation.operationId, `operator:${site.id}`)
+      return Response.json(await operations.get(site.id, created.operation.operationId) ?? created.operation, { status: 202, headers: { location: `?phase=operator-static-artifact-operation&operationId=${created.operation.operationId}`, "x-wp-codebox-dispatch": dispatch } })
     } catch (error) {
       if (error instanceof OperationConflict) return Response.json({ schema: STATIC_ARTIFACT_IMPORT_RESULT_SCHEMA, status: "conflict", error: error.message }, { status: 409 })
       throw error
@@ -606,10 +633,12 @@ async function readStaticArtifactOperation(request: Request, env: RuntimeEnv, si
   return operation ? Response.json(operation) : new Response("Static artifact operation is unavailable.", { status: 404 })
 }
 
-async function runNextStaticArtifactOperation(env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, operations?: D1OperationRepository): Promise<{ schema: string; status: string; operationId: string } | null> {
-  if (!operations) return null
-  const claimed = await operations.claimNext(site.id)
-  if (!claimed) return null
+async function runStaticArtifactOperation(env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, operations: D1OperationRepository, operationId: string, principal: string): Promise<{ schema: string; status: string; operationId: string } | null> {
+  const claimed = await operations.claimOperation(site.id, operationId)
+  if (!claimed) {
+    const operation = await operations.get(site.id, operationId)
+    return operation && ["queued", "running", "retryable"].includes(operation.state) ? { schema: STATIC_ARTIFACT_OPERATION_SCHEMA, status: "pending", operationId } : null
+  }
   try {
     if (claimed.prepared) {
       const receipt = await coordinator.committed(claimed.prepared.version)
@@ -628,7 +657,10 @@ async function runNextStaticArtifactOperation(env: RuntimeEnv, coordinator: Revi
     const response = await runCoordinatedWordPressRequest(new Request("https://scheduled.invalid/?phase=operator-static-artifact-import", { method: "POST" }), env, coordinator, site, "static-artifact-import", input, async () => operations.recordCommit(site.id, claimed.operationId, claimed.claimToken), async (prepared) => operations.prepareCommit(site.id, claimed.operationId, claimed.claimToken, prepared.version, prepared.pointer, prepared.ssiResult, prepared.publicationJobKey), heartbeat)
     const result = await response.json()
     if (!result || typeof result !== "object" || (result as { status?: unknown }).status !== "imported") throw new Error("Static artifact import did not produce an import receipt.")
-    await operations.complete(site.id, claimed.operationId, claimed.claimToken, result, response.headers.get("x-wp-codebox-publication-job") ?? undefined, site.origin)
+    const publicationJob = response.headers.get("x-wp-codebox-publication-job") ?? undefined
+    await operations.complete(site.id, claimed.operationId, claimed.claimToken, result, publicationJob, site.origin)
+    // The operation receipt and coordinator commit are durable before the queue wake-up.
+    if (publicationJob) await dispatchRuntimeWork(env, operations, site, "publication", publicationJob, principal)
     return { schema: STATIC_ARTIFACT_OPERATION_SCHEMA, status: "completed", operationId: claimed.operationId }
   } catch (error) {
     try {
@@ -645,6 +677,22 @@ async function runNextStaticArtifactOperation(env: RuntimeEnv, coordinator: Revi
   }
 }
 
+async function dispatchRuntimeWork(env: RuntimeEnv, operations: D1OperationRepository, site: SiteContext, kind: RuntimeQueueMessage["kind"], identity: string, principal: string): Promise<"queued" | "backpressured"> {
+  let message: RuntimeQueueMessage | undefined
+  try {
+    message = await operations.stageDispatch(site, kind, identity, principal)
+    if (!await operations.admitDispatch(message, parseRuntimeQueuePolicy(env.WORDPRESS_QUEUE_POLICY))) return "backpressured"
+    if (!env.WORDPRESS_RUNTIME_QUEUE) throw new Error("Runtime queue binding is unavailable.")
+    await env.WORDPRESS_RUNTIME_QUEUE.send(message)
+    await operations.deliveredDispatch(message)
+    return "queued"
+  } catch (error) {
+    if (message) await operations.failedDispatch(message, error)
+    console.error("Runtime queue producer is backpressured; reconciliation will retry.", error)
+    return "backpressured"
+  }
+}
+
 function isCanonicalRestorePointer(pointer: unknown, site: SiteContext): pointer is MarkdownPointer {
   if (!pointer || typeof pointer !== "object") return false
   const candidate = pointer as Partial<MarkdownPointer>
@@ -915,7 +963,7 @@ async function releasePublicationClaim(bucket: R2Bucket, job: PublicationJob, si
   if (claim.etag) await bucket.put(publicationClaimObjectKey(job, site), JSON.stringify({ expiresAt: 0 }), { onlyIf: { etagMatches: claim.etag }, httpMetadata: { contentType: "application/json" } })
 }
 
-async function drainNextPublicationJob(env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, operations?: D1OperationRepository): Promise<PublicationInvocationEvidence | null> {
+async function drainPublicationJob(env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, operations: D1OperationRepository, jobKey: string): Promise<PublicationInvocationEvidence | null> {
   await reconcilePublicationReceipts(env.WORDPRESS_STATE_BUCKET, site, operations)
   let publicationLease: RevisionLease
   try {
@@ -926,7 +974,7 @@ async function drainNextPublicationJob(env: RuntimeEnv, coordinator: RevisionCoo
     return { schema: "wp-codebox/cloudflare-publication/v1", status: fence.active ? "fenced" : "pending", jobKey: "coordinator" }
   }
   try {
-    return await drainNextPublicationJobWhileLeased(env, coordinator, site, async () => {
+    return await drainPublicationJobWhileLeased(env, coordinator, site, jobKey, async () => {
       publicationLease = await coordinator.renew(publicationLease)
     }, operations)
   } finally {
@@ -951,28 +999,25 @@ async function reconcilePublicationReceipts(bucket: R2Bucket, site: SiteContext,
   }
 }
 
-async function drainNextPublicationJobWhileLeased(env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, renewLease: () => Promise<void>, operations?: D1OperationRepository): Promise<PublicationInvocationEvidence | null> {
+async function drainPublicationJobWhileLeased(env: RuntimeEnv, coordinator: RevisionCoordinator, site: SiteContext, jobKey: string, renewLease: () => Promise<void>, operations?: D1OperationRepository): Promise<PublicationInvocationEvidence | null> {
   await assertActiveAllocation(env, site)
-  const listed = await env.WORDPRESS_STATE_BUCKET.list({ prefix: `${siteStorageKeys(site).publicationJobPrefix}/`, limit: 16 })
-  if (!listed.objects.length) return null
+  // A Queue body is an authority-bearing immutable job key, never permission to drain another job.
+  let exact: PublicationJob
+  try { exact = await readPublicationJob(env.WORDPRESS_STATE_BUCKET, jobKey, site) } catch { return null }
   const state = await coordinator.state()
-  let job: PublicationJob | undefined
-  for (const object of listed.objects) {
-    const candidate = await readPublicationJob(env.WORDPRESS_STATE_BUCKET, object.key, site)
-    const committed = await coordinator.committed(candidate.coordinatorVersion)
-    if (committed?.revision === candidate.canonical.revision) {
-      job = candidate
-      break
-    }
-    if (state.version > candidate.coordinatorVersion || Date.now() - Date.parse(candidate.createdAt) > PUBLICATION_CLAIM_MS) {
+  const committed = await coordinator.committed(exact.coordinatorVersion)
+  if (committed?.revision !== exact.canonical.revision || committed.manifestKey !== exact.canonical.manifestKey || committed.persistedAt !== exact.canonical.persistedAt) {
+    if (state.version > exact.coordinatorVersion || Date.now() - Date.parse(exact.createdAt) > PUBLICATION_CLAIM_MS) {
       await renewLease()
-      await putImmutableJson(env.WORDPRESS_STATE_BUCKET, publicationReceiptObjectKey(candidate, site), JSON.stringify({ status: "orphaned", job: candidate.key, recordedAt: new Date().toISOString() }))
-      await operations?.reconcilePublication(site.id, candidate.key, "orphaned")
+      await putImmutableJson(env.WORDPRESS_STATE_BUCKET, publicationReceiptObjectKey(exact, site), JSON.stringify({ status: "orphaned", job: exact.key, recordedAt: new Date().toISOString() }))
+      await operations?.reconcilePublication(site.id, exact.key, "orphaned")
       await renewLease()
-      await env.WORDPRESS_STATE_BUCKET.delete(candidate.key)
+      await env.WORDPRESS_STATE_BUCKET.delete(exact.key)
+      return { schema: "wp-codebox/cloudflare-publication/v1", status: "stale", jobKey: exact.key }
     }
+    return { schema: "wp-codebox/cloudflare-publication/v1", status: "pending", jobKey: exact.key }
   }
-  if (!job) return listed.truncated ? { schema: "wp-codebox/cloudflare-publication/v1", status: "pending", jobKey: "scan" } : null
+  const job = exact
   const claim = await claimPublicationJob(env.WORDPRESS_STATE_BUCKET, job, site, renewLease)
   if (!claim) return { schema: "wp-codebox/cloudflare-publication/v1", status: "pending", jobKey: job.key }
   let runtime: Runtime | undefined
@@ -1049,7 +1094,7 @@ async function drainNextPublicationJobWhileLeased(env: RuntimeEnv, coordinator:
     return { schema: "wp-codebox/cloudflare-publication/v1", status: "promoted", jobKey: job.key }
   } catch (error) {
     console.error("Publication job failed.", error)
-    return { schema: "wp-codebox/cloudflare-publication/v1", status: "failed", jobKey: job.key }
+    throw error
   } finally {
     if (runtime) await discardRuntime(runtime, site)
     try {
@@ -1216,6 +1261,11 @@ async function runCoordinatedWordPressRequest(request: Request, env: RuntimeEnv,
       }
       const committed = await commitLease(coordinator, request.url, lease, next)
       await onCommitted?.(committed)
+      // A pre-commit R2 job is inert; only the observable coordinator receipt may wake it.
+      if (publicationJob && env.WORDPRESS_STATE_DATABASE) {
+        const operations = new D1OperationRepository(env.WORDPRESS_STATE_DATABASE)
+        await dispatchRuntimeWork(env, operations, site, "publication", publicationJob.key, `system:${site.id}`)
+      }
       response.headers.set("x-wp-codebox-canonical-revision", committed.pointer.revision)
       response.headers.set("x-wp-codebox-canonical-version", String(committed.version))
       logMutationPhase(diagnosticsStartedAt, "commit", retained, { publication: publicationJob ? "queued" : "unchanged" })
@@ -1596,6 +1646,7 @@ async function runScheduledWordPressCron(env: RuntimeEnv, coordinator: RevisionC
       runtime = undefined
       const publicationJob = await enqueuePublicationJob(env.WORDPRESS_STATE_BUCKET, lease, next, currentPublication, event.publicationChanges, site)
       await commitLease(coordinator, requestUrl, lease, next)
+      if (publicationJob && env.WORDPRESS_STATE_DATABASE) await dispatchRuntimeWork(env, new D1OperationRepository(env.WORDPRESS_STATE_DATABASE), site, "publication", publicationJob.key, `system:${site.id}`)
       finalized = true
       evidence.events.push({ hook: event.hook, timestamp: event.timestamp, revision: next.revision, publication: publicationJob ? "queued" : "unchanged" })
     } catch (error) {
diff --git a/packages/runtime-cloudflare/wrangler.d1.jsonc b/packages/runtime-cloudflare/wrangler.d1.jsonc
index aff62efe..6647bfce 100644
--- a/packages/runtime-cloudflare/wrangler.d1.jsonc
+++ b/packages/runtime-cloudflare/wrangler.d1.jsonc
@@ -6,6 +6,10 @@
   "compatibility_flags": ["nodejs_compat"],
   "limits": { "cpu_ms": 300000 },
 	"triggers": { "crons": ["* * * * *"] },
+	"queues": {
+		"producers": [{ "binding": "WORDPRESS_RUNTIME_QUEUE", "queue": "wp-codebox-runtime-dispatch" }],
+		"consumers": [{ "queue": "wp-codebox-runtime-dispatch", "max_batch_size": 10, "max_batch_timeout": 5, "max_retries": 3, "max_concurrency": 10, "dead_letter_queue": "wp-codebox-runtime-dispatch-dlq" }]
+	},
 	"r2_buckets": [
 		{ "binding": "WORDPRESS_STATE_BUCKET", "bucket_name": "wp-codebox-runtime-chubes" }
 	],
diff --git a/tests/cloudflare-d1-operation-repository.test.ts b/tests/cloudflare-d1-operation-repository.test.ts
index 00c92610..eb9d4466 100644
--- a/tests/cloudflare-d1-operation-repository.test.ts
+++ b/tests/cloudflare-d1-operation-repository.test.ts
@@ -3,6 +3,7 @@ import { DatabaseSync } from "node:sqlite"
 import test from "node:test"
 import { D1OperationRepository, OperationConflict, shouldRecoverPreparedCommit } from "../packages/runtime-cloudflare/src/d1-operation-repository.js"
 import { CloudflareAllocationLifecycle } from "../packages/runtime-cloudflare/src/allocation-lifecycle.js"
+import { DEFAULT_RUNTIME_QUEUE_POLICY, parseRuntimeQueueMessage, RUNTIME_QUEUE_MESSAGE_SCHEMA, runtimeQueueMessage } from "../packages/runtime-cloudflare/src/queue-dispatch.js"
 
 function database(): D1Database {
   const sqlite = new DatabaseSync(":memory:")
@@ -36,6 +37,14 @@ test("D1 enforces one active mutation per site while independent sites claim con
   assert.notEqual((alphaClaim ?? racedAlphaClaim)!.operationId, betaClaim.operationId)
 })
 
+test("an operation wake-up can claim only its exact durable operation identity", async () => {
+  const repository = new D1OperationRepository(database())
+  const created = await repository.createOrConverge(site, input("target"))
+  assert.equal(await repository.claimOperation(site.id, "00000000-0000-4000-8000-000000000000"), null)
+  const claim = await repository.claimOperation(site.id, created.operation.operationId)
+  assert.equal(claim?.operationId, created.operation.operationId)
+})
+
 test("expired claims recover, retries retain attempt history, and terminal receipts are immutable", async () => {
   const repository = new D1OperationRepository(database())
   const created = await repository.createOrConverge(site, input())
@@ -108,3 +117,54 @@ test("deletion fences an in-flight claim and a publication callback", async () =
   assert.equal((await repository.get(dynamic.id, created.operation.operationId))?.state, "failed")
   assert.ok(fence > active.operationFence)
 })
+
+test("queue dispatches are versioned wake-ups with durable repair and dead-letter evidence", async () => {
+  const repository = new D1OperationRepository(database())
+  const created = await repository.createOrConverge(site, input())
+  const message = await repository.stageDispatch(site, "operation", created.operation.operationId, "principal:a")
+  assert.deepEqual(message, { schema: RUNTIME_QUEUE_MESSAGE_SCHEMA, siteId: site.id, generation: 1, kind: "operation", identity: created.operation.operationId })
+  assert.deepEqual(await repository.pendingDispatches(DEFAULT_RUNTIME_QUEUE_POLICY), [message])
+  await repository.failedDispatch(message, new Error("queue unavailable\nsecret"))
+  assert.deepEqual(await repository.pendingDispatches(DEFAULT_RUNTIME_QUEUE_POLICY), [message], "a producer failure leaves durable repair work")
+  await repository.deadDispatch(message, 3, new Error("isolate evicted\nsecret"))
+  const evidence = await repository.deadLetterEvidence(message)
+  assert.equal(evidence?.attempts, 3)
+  assert.equal(evidence?.inputDigest, created.operation.input.fingerprint)
+  assert.equal(evidence?.lastError, "isolate evicted secret")
+  assert.match(evidence?.recoveryCommand ?? "", /^npx wrangler d1 execute wp-codebox-runtime-state --remote --command/)
+  await repository.completeDispatch(message)
+  assert.equal(await repository.dispatchState(message), "dead-letter")
+  assert.deepEqual(await repository.pendingDispatches(DEFAULT_RUNTIME_QUEUE_POLICY), [])
+  assert.equal(parseRuntimeQueueMessage({ ...message, schema: "foreign" }), null)
+  assert.equal(parseRuntimeQueueMessage({ ...message, generation: 0 }), null)
+  assert.equal(parseRuntimeQueueMessage({ ...message, identity: "x".repeat(1024) }), null)
+  assert.throws(() => runtimeQueueMessage({ id: "alpha" }, 1, "operation", ""), /invalid/)
+})
+
+test("queue admission enforces global and explicit principal backpressure", async () => {
+  const repository = new D1OperationRepository(database())
+  const betaSite = { id: "beta", hostname: "beta.example", origin: "https://beta.example" }
+  const alphaOperation = await repository.createOrConverge(site, input("alpha-dispatch"))
+  const betaOperation = await repository.createOrConverge(betaSite, input("beta-dispatch"))
+  const alpha = await repository.stageDispatch(site, "operation", alphaOperation.operation.operationId, "principal:a")
+  const betaMessage = await repository.stageDispatch(betaSite, "operation", betaOperation.operation.operationId, "principal:a")
+  assert.equal(await repository.admitDispatch(alpha, { maxActive: 2, maxActivePerPrincipal: 1 }), true)
+  assert.equal(await repository.admitDispatch(betaMessage, { maxActive: 2, maxActivePerPrincipal: 1 }), false)
+  await repository.completeDispatch(alpha)
+  assert.equal(await repository.admitDispatch(betaMessage, { maxActive: 2, maxActivePerPrincipal: 1 }), true)
+})
+
+test("queue processing ownership survives producer replay and honors retry delay", async () => {
+  const repository = new D1OperationRepository(database())
+  const created = await repository.createOrConverge(site, input("dispatch-ownership"))
+  const message = await repository.stageDispatch(site, "operation", created.operation.operationId, "principal:a")
+  assert.equal(await repository.admitDispatch(message, DEFAULT_RUNTIME_QUEUE_POLICY), true)
+  await repository.deliveredDispatch(message)
+  assert.equal(await repository.processingDispatch(message), true)
+  await repository.deliveredDispatch(message)
+  assert.equal(await repository.dispatchState(message), "processing", "a replayed producer send cannot clear processing ownership")
+  assert.equal(await repository.processingDispatch(message), false)
+  await repository.retryDispatch(message, new Error("retry later"))
+  assert.equal(await repository.processingDispatch(message), false, "redelivery cannot bypass durable retry_at")
+  assert.equal(await repository.processingDispatch(message, Date.now() + 31_000), true)
+})
diff --git a/tests/cloudflare-provisioning-api.test.ts b/tests/cloudflare-provisioning-api.test.ts
index 359a9329..b820587a 100644
--- a/tests/cloudflare-provisioning-api.test.ts
+++ b/tests/cloudflare-provisioning-api.test.ts
@@ -77,7 +77,8 @@ test("artifact staging verifies the winner of a conditional R2 race", async () =
   assert.equal((await routeProvisioningApi(stageRequest(), conflicting.env, conflicting.operations)).status, 409)
 })
 test("registered active contexts are excluded", async () => { const r = runtime(); await r.operations.createOrConverge({ id: "alpha", hostname: "alpha.example", origin: "https://alpha.example" }, { idempotencyKey: "legacy", fingerprint: "x", artifact: { r2Key: "x", sha256: digest, size: artifact.byteLength }, options: { slug: "x", name: "x", siteTitle: "x" } }); const body = await (await create(r)).json() as { site: { id: string } }; assert.equal(body.site.id, "beta") })
-test("exact and concurrent same-key creates converge", async () => { const r = runtime(); const [one, two] = await Promise.all([create(r), create(r)]); assert.equal(one.status, 202); assert.deepEqual(await one.json(), await two.json()); assert.equal(count(r.db, "wp_codebox_api_sites"), 1) })
+test("exact and concurrent same-key creates converge", async () => { const r = runtime(); const [one, two] = await Promise.all([create(r), create(r)]); assert.equal(one.status, 202); assert.equal(one.headers.get("x-wp-codebox-dispatch"), "backpressured"); assert.deepEqual(await one.json(), await two.json()); assert.equal(count(r.db, "wp_codebox_api_sites"), 1) })
+test("accepted provisioning reports successful queue admission", async () => { const r = runtime(); r.env.WORDPRESS_RUNTIME_QUEUE = { async send() {} }; const response = await create(r); assert.equal(response.status, 202); assert.equal(response.headers.get("x-wp-codebox-dispatch"), "queued") })
 test("changed fingerprints conflict without another allocation", async () => { const r = runtime(); await create(r); const response = await routeProvisioningApi(createRequest("create-1", { title: "Changed" }), r.env, r.operations); assert.equal(response.status, 409); assert.equal(count(r.db, "wp_codebox_api_sites"), 1) })
 test("create replay honors current site restrictions", async () => { const r = runtime(); await create(r); const records = JSON.parse(r.env.WORDPRESS_API_TOKENS) as Array<Record<string, unknown>>; records[0].sites = ["beta"]; r.env.WORDPRESS_API_TOKENS = JSON.stringify(records); assert.equal((await create(r)).status, 404); assert.equal(count(r.db, "wp_codebox_api_sites"), 1) })
 test("concurrent distinct keys allocate unique contexts", async () => { const r = runtime(); await Promise.all([create(r, "one"), create(r, "two")]); const ids = r.db.sqlite.prepare("SELECT site_id FROM wp_codebox_api_sites ORDER BY site_id").all() as Array<{ site_id: string }>; assert.deepEqual(ids.map((row) => row.site_id), ["alpha", "beta"]) })
diff --git a/tests/cloudflare-queue-batch.test.ts b/tests/cloudflare-queue-batch.test.ts
new file mode 100644
index 00000000..2c2dd7c0
--- /dev/null
+++ b/tests/cloudflare-queue-batch.test.ts
@@ -0,0 +1,48 @@
+import assert from "node:assert/strict"
+import test from "node:test"
+import { dispatchQueueBatch, type QueueDelivery } from "../packages/runtime-cloudflare/src/queue-batch.js"
+import { DEFAULT_RUNTIME_QUEUE_POLICY, parseRuntimeQueuePolicy, runtimeQueueMessage } from "../packages/runtime-cloudflare/src/queue-dispatch.js"
+
+function delivery(siteId: string, kind: "operation" | "publication", identity: string): QueueDelivery & { result?: string; delaySeconds?: number } {
+  return { body: runtimeQueueMessage({ id: siteId }, 1, kind, identity), attempts: 1, ack() { this.result = "ack" }, retry(options) { this.result = "retry"; this.delaySeconds = options?.delaySeconds } }
+}
+const id = (n: number) => `00000000-0000-4000-8000-${String(n).padStart(12, "0")}`
+
+test("queue batch overlaps unrelated lanes but never overlaps a site lane", async () => {
+  const messages = Array.from({ length: 10 }, (_, index) => delivery(`site-${index}`, "operation", id(index)))
+  messages.push(delivery("site-0", "operation", id(99)))
+  let active = 0; let peak = 0; const siteActive = new Map<string, number>()
+  await dispatchQueueBatch(messages, (body) => body as ReturnType<typeof runtimeQueueMessage>, async (_site, lane) => lane[0], async ({ value }) => {
+    const current = (siteActive.get(value.siteId) ?? 0) + 1; siteActive.set(value.siteId, current); assert.equal(current, 1)
+    active++; peak = Math.max(peak, active)
+    await new Promise((resolve) => setTimeout(resolve, 15))
+    active--; siteActive.set(value.siteId, current - 1)
+    return "ack"
+  })
+  assert.equal(peak, 10)
+  assert.equal(messages.at(-1)?.result, "retry")
+})
+
+test("persisted lane preference can choose either ready kind without starvation", async () => {
+  for (const first of ["operation", "publication"] as const) {
+    const operation = delivery("alpha", "operation", id(1))
+    const publication = delivery("alpha", "publication", "sites/alpha/publications/jobs/00000000000000000001-00000000-0000-4000-8000-000000000002.json")
+    const executed: string[] = []
+    await dispatchQueueBatch([operation, publication], (body) => body as ReturnType<typeof runtimeQueueMessage>, async (_site, lane) => lane.find(({ value }) => value.kind === first)!, async ({ value }) => { executed.push(value.kind); return "ack" })
+    assert.deepEqual(executed, [first])
+    assert.equal(first === "operation" ? publication.result : operation.result, "retry")
+  }
+})
+
+test("queue policy is explicit and bounded", () => {
+  assert.deepEqual(parseRuntimeQueuePolicy(undefined), DEFAULT_RUNTIME_QUEUE_POLICY)
+  assert.deepEqual(parseRuntimeQueuePolicy('{"maxActive":20,"maxActivePerPrincipal":4}'), { maxActive: 20, maxActivePerPrincipal: 4 })
+  assert.throws(() => parseRuntimeQueuePolicy('{"maxActive":1,"maxActivePerPrincipal":2}'), /invalid/)
+})
+
+test("queue execution can request delayed redelivery", async () => {
+  const message = delivery("alpha", "operation", id(1))
+  await dispatchQueueBatch([message], (body) => body as ReturnType<typeof runtimeQueueMessage>, async (_site, lane) => lane[0], async () => ({ retryAfterSeconds: 30 }))
+  assert.equal(message.result, "retry")
+  assert.equal(message.delaySeconds, 30)
+})
diff --git a/tests/cloudflare-runtime.test.ts b/tests/cloudflare-runtime.test.ts
index bd0a6ed5..5eeb940f 100644
--- a/tests/cloudflare-runtime.test.ts
+++ b/tests/cloudflare-runtime.test.ts
@@ -226,6 +226,7 @@ test("Cloudflare runtime declares bounded CPU and scheduled execution", async ()
     limits?: { cpu_ms?: number }
     triggers?: { crons?: string[] }
     d1_databases?: Array<{ binding?: string; database_name?: string }>
+    queues?: { producers?: Array<{ binding?: string; queue?: string }>; consumers?: Array<{ queue?: string; max_batch_size?: number; max_retries?: number; max_concurrency?: number; dead_letter_queue?: string }> }
     durable_objects?: unknown
     migrations?: unknown
   }
@@ -238,6 +239,8 @@ test("Cloudflare runtime declares bounded CPU and scheduled execution", async ()
   assert.deepEqual(d1Config.d1_databases?.map(({ binding, database_name }) => ({ binding, database_name })), [{ binding: "WORDPRESS_STATE_DATABASE", database_name: "wp-codebox-runtime-state" }])
   assert.deepEqual((d1Config.durable_objects as { bindings?: Array<{ name?: string; class_name?: string }> } | undefined)?.bindings, [{ name: "WORDPRESS_STATE", class_name: "WordPressStateCoordinator" }])
   assert.deepEqual(d1Config.migrations, [{ tag: "v1", new_sqlite_classes: ["WordPressStateCoordinator"] }])
+  assert.deepEqual(d1Config.queues?.producers, [{ binding: "WORDPRESS_RUNTIME_QUEUE", queue: "wp-codebox-runtime-dispatch" }])
+  assert.deepEqual(d1Config.queues?.consumers, [{ queue: "wp-codebox-runtime-dispatch", max_batch_size: 10, max_batch_timeout: 5, max_retries: 3, max_concurrency: 10, dead_letter_queue: "wp-codebox-runtime-dispatch-dlq" }])
 })
 
 test("Cloudflare lease contention honors Retry-After without exceeding the acquisition deadline", () => {
@@ -444,14 +447,13 @@ test("Cloudflare runtime injects composable coordinators without moving PHP out
   assert.match(worker, /PUBLICATION_JOB_SCHEMA/)
   assert.match(worker, /publicationClaimPrefix/)
   assert.match(worker, /publicationReceiptPrefix/)
-  assert.match(worker, /async function drainNextPublicationJob/)
-  assert.match(worker, /coordinator\.committed\(candidate\.coordinatorVersion\)/)
-  assert.match(worker, /\.list\(\{ prefix: `\$\{siteStorageKeys\(site\)\.publicationJobPrefix\}\/`, limit: 16 \}\)/)
+  assert.match(worker, /async function drainPublicationJob/)
+  assert.match(worker, /coordinator\.committed\(exact\.coordinatorVersion\)/)
   assert.match(worker, /canonicalVersion: job\.coordinatorVersion/)
   assert.match(worker, /sourceJob: job\.key/)
   assert.match(worker, /readWordPressPageSnapshot\(env\.WORDPRESS_STATE_BUCKET, objectKey, job\.canonical\.revision, route\)/)
   assert.match(worker, /if \(!await readWordPressPageSnapshot\(env\.WORDPRESS_STATE_BUCKET, objectKey, job\.canonical\.revision, route\)\) throw error/)
-  const publicationDrain = worker.slice(worker.indexOf("async function drainNextPublicationJob"), worker.indexOf("async function compilePublicationRoutes"))
+  const publicationDrain = worker.slice(worker.indexOf("async function drainPublicationJob"), worker.indexOf("async function compilePublicationRoutes"))
   assert.match(publicationDrain, /const recoveredPromotion = current\?\.publication\.canonicalVersion === job\.coordinatorVersion[\s\S]*current\.publication\.canonicalRevision === job\.canonical\.revision[\s\S]*current\.publication\.sourceJob === job\.key/)
   assert.match(publicationDrain, /if \(recoveredPromotion\)[\s\S]*status: "promoted"[\s\S]*operations\?\.reconcilePublication\(site\.id, job\.key, "promoted", current\.publication\.revision\)/)
   assert.match(publicationDrain, /bootRuntime\(env\.WORDPRESS_STATE_BUCKET, job\.canonical, site\.origin/)
@@ -485,10 +487,14 @@ test("Cloudflare runtime injects composable coordinators without moving PHP out
   assert.match(worker, /validateWpContentDeletedPaths\(manifest\.wpContentDeleted \?\? \[\]\)/)
   assert.match(worker, /materializeWpContentTombstones\(php, wpContentDeleted\)/)
   assert.match(worker, /async scheduled\(controller: ScheduledController, env: Env\)/)
-  assert.match(worker, /const publication = operation \? null : await drainNextPublicationJob\(env, coordinator, site, operations\)/)
-  assert.match(worker, /const fence = await coordinator\.fenceStatus\(\)/)
-  assert.match(worker, /const operationFirst = siteCycle % 2 === 0/)
-  assert.match(worker, /: await runScheduledWordPressCron\(env, coordinator, site, controller\.scheduledTime\)/)
+  const scheduledHandler = worker.slice(worker.indexOf("async scheduled(controller"), worker.indexOf("async queue(batch"))
+  assert.match(scheduledHandler, /pendingDispatches\(parseRuntimeQueuePolicy\(env\.WORDPRESS_QUEUE_POLICY\), 32\)/)
+  assert.match(scheduledHandler, /runScheduledWordPressCron/)
+  assert.doesNotMatch(scheduledHandler, /runStaticArtifactOperation|drainPublicationJob/)
+  assert.match(worker, /async queue\(batch:/)
+  assert.match(worker, /dispatchQueueBatch/)
+  assert.match(worker, /matchesGeneration/)
+  assert.doesNotMatch(worker, /siteCycle % 2/)
   assert.match(worker, /pathname === "\/wp-cron\.php"/)
   assert.match(worker, /MAX_CRON_EVENTS_PER_INVOCATION = 5/)
   assert.match(worker, /while \(evidence\.events\.length < MAX_CRON_EVENTS_PER_INVOCATION/)