.changeset/publish-admission-guard.md

@@ -0,0 +1,5 @@
+---
+"@luzzle/web": patch
+---
+
+Upgrade OpenWorkflow to 0.9.2 and serialize publish and audit check-and-enqueue requests within one web process. Check the latest five runs per workflow through the SDK, retain active-run conflict responses and audit checks, and release admission on enqueue failure. Older active runs outside that window are not detected.

.changeset/sdk-workflow-reads.md

@@ -0,0 +1,5 @@
+---
+"@luzzle/web": patch
+---
+
+Use OpenWorkflow SDK APIs for workflow metadata reads in publish/preview pages, dashboard, connectivity checks and progress streams. Preserve existing payload shapes and step-attempt ordering, remove the unused numeric job-ID lookup and the web's raw queue connection, and keep retention cleanup SQL unchanged because the SDK has no purge API.

apps/web/package.json

@@ -90,6 +90,6 @@
     "@replit/codemirror-vim": "6.4.0",
     "eventsource-parser": "4.1.1",
     "kysely": "0.28.17",
-    "openworkflow": "0.9.0"
+    "openworkflow": "0.9.2"
   }
 }

apps/web/src/lib/server/health/connectivity.test.ts

@@ -1,11 +1,11 @@
 import { describe, test, expect, vi, beforeEach } from 'vitest'
-import { getOpenWorkflow, getOpenWorkflowDb } from '$lib/server/workflow/index.js'
+import { getOpenWorkflow, getOpenWorkflowBackend } from '$lib/server/workflow/index.js'
 import { getWorkflowRun } from '@luzzle/web.jobs'
 import { testConnectivity } from './connectivity.js'
 
 vi.mock('$lib/server/workflow/index.js', () => ({
 	getOpenWorkflow: vi.fn(),
-	getOpenWorkflowDb: vi.fn()
+	getOpenWorkflowBackend: vi.fn()
 }))
 
 vi.mock('@luzzle/web.jobs', () => ({
@@ -18,7 +18,7 @@ vi.mock('@luzzle/web.jobs/specs', () => ({
 
 const mocks = {
 	getOpenWorkflow: vi.mocked(getOpenWorkflow),
-	getOpenWorkflowDb: vi.mocked(getOpenWorkflowDb),
+	getOpenWorkflowBackend: vi.mocked(getOpenWorkflowBackend),
 	getWorkflowRun: vi.mocked(getWorkflowRun)
 }
 
@@ -27,16 +27,18 @@ const runWorkflow = vi.fn()
 beforeEach(() => {
 	vi.clearAllMocks()
 	mocks.getOpenWorkflow.mockReturnValue({ runWorkflow } as never)
-	mocks.getOpenWorkflowDb.mockReturnValue({} as never)
+	mocks.getOpenWorkflowBackend.mockReturnValue({} as never)
 	runWorkflow.mockResolvedValue({ workflowRun: { id: 'run-1' } })
 })
 
 describe('testConnectivity', () => {
 	test('triggers the workflow and polls until it completes', async () => {
-		mocks.getWorkflowRun.mockReturnValueOnce({ status: 'running' } as never).mockReturnValueOnce({
-			status: 'completed',
-			output: JSON.stringify({ ok: true })
-		} as never)
+		mocks.getWorkflowRun
+			.mockResolvedValueOnce({ status: 'running' } as never)
+			.mockResolvedValueOnce({
+				status: 'completed',
+				output: JSON.stringify({ ok: true })
+			} as never)
 
 		const result = await testConnectivity('archive', { pollIntervalMs: 1 })
 
@@ -48,7 +50,7 @@ describe('testConnectivity', () => {
 	})
 
 	test('returns not-ok when the run fails', async () => {
-		mocks.getWorkflowRun.mockReturnValue({ status: 'failed', error: 'boom' } as never)
+		mocks.getWorkflowRun.mockResolvedValue({ status: 'failed', error: 'boom' } as never)
 
 		const result = await testConnectivity('cdn', { pollIntervalMs: 1 })
 
@@ -56,7 +58,7 @@ describe('testConnectivity', () => {
 	})
 
 	test('returns a generic reason when a failed run has no error message', async () => {
-		mocks.getWorkflowRun.mockReturnValue({ status: 'canceled', error: null } as never)
+		mocks.getWorkflowRun.mockResolvedValue({ status: 'canceled', error: null } as never)
 
 		const result = await testConnectivity('cdn', { pollIntervalMs: 1 })
 
@@ -64,7 +66,7 @@ describe('testConnectivity', () => {
 	})
 
 	test('returns not-ok when the output cannot be parsed', async () => {
-		mocks.getWorkflowRun.mockReturnValue({ status: 'completed', output: 'not json' } as never)
+		mocks.getWorkflowRun.mockResolvedValue({ status: 'completed', output: 'not json' } as never)
 
 		const result = await testConnectivity('archive', { pollIntervalMs: 1 })
 
@@ -72,7 +74,7 @@ describe('testConnectivity', () => {
 	})
 
 	test('times out waiting for the worker', async () => {
-		mocks.getWorkflowRun.mockReturnValue({ status: 'running' } as never)
+		mocks.getWorkflowRun.mockResolvedValue({ status: 'running' } as never)
 
 		const result = await testConnectivity('archive', { pollIntervalMs: 5, timeoutMs: 20 })
 

apps/web/src/lib/server/health/connectivity.ts

@@ -1,4 +1,4 @@
-import { getOpenWorkflow, getOpenWorkflowDb } from '$lib/server/workflow/index.js'
+import { getOpenWorkflow, getOpenWorkflowBackend } from '$lib/server/workflow/index.js'
 import { getWorkflowRun, type TestConnectivityResult } from '@luzzle/web.jobs'
 import { testConnectivitySpec } from '@luzzle/web.jobs/specs'
 
@@ -20,11 +20,11 @@ export async function testConnectivity(
 	const openWorkflow = getOpenWorkflow()
 	const handle = await openWorkflow.runWorkflow(testConnectivitySpec, { target })
 	const runId = handle.workflowRun.id
-	const openWorkflowDb = getOpenWorkflowDb()
+	const backend = getOpenWorkflowBackend()
 
 	const deadline = Date.now() + timeoutMs
 	while (Date.now() < deadline) {
-		const run = getWorkflowRun(openWorkflowDb, runId)
+		const run = await getWorkflowRun(backend, runId)
 		if (run && TERMINAL_STATES.has(run.status)) {
 			if (run.status === 'failed' || run.status === 'canceled') {
 				return { ok: false, reason: run.error ?? `workflow ${run.status}` }

apps/web/src/lib/server/load/dashboard.test.ts

@@ -2,7 +2,7 @@ import { describe, test, expect, vi, beforeEach } from 'vitest'
 import { config } from '$lib/server/config'
 import { getPieceStats, getAssetStats } from '../dashboardStats.js'
 import { getRecentlyEditedPieces } from '../pieces.js'
-import { getOpenWorkflowDb } from '../workflow/index.js'
+import { getOpenWorkflowBackend } from '../workflow/index.js'
 import { getLatestWorkflowRun, type WorkflowRunRow } from '@luzzle/web.jobs'
 import { makeConfig, credentialsAuth, oidcAuth } from '../config.fixture'
 import { loadDashboardPage } from './dashboard.js'
@@ -36,7 +36,7 @@ vi.mock('../pieces.js', () => ({
 }))
 
 vi.mock('../workflow/index.js', () => ({
-	getOpenWorkflowDb: vi.fn()
+	getOpenWorkflowBackend: vi.fn()
 }))
 
 vi.mock('@luzzle/web.jobs', () => ({
@@ -47,7 +47,7 @@ const mocks = {
 	getPieceStats: vi.mocked(getPieceStats),
 	getAssetStats: vi.mocked(getAssetStats),
 	getRecentlyEditedPieces: vi.mocked(getRecentlyEditedPieces),
-	getOpenWorkflowDb: vi.mocked(getOpenWorkflowDb),
+	getOpenWorkflowBackend: vi.mocked(getOpenWorkflowBackend),
 	getLatestWorkflowRun: vi.mocked(getLatestWorkflowRun)
 }
 
@@ -70,8 +70,8 @@ beforeEach(() => {
 	mocks.getPieceStats.mockResolvedValue({ total: 3, byType: [{ type: 'article', count: 3 }] })
 	mocks.getAssetStats.mockResolvedValue({ total: 2 })
 	mocks.getRecentlyEditedPieces.mockResolvedValue([])
-	mocks.getOpenWorkflowDb.mockReturnValue({} as never)
-	mocks.getLatestWorkflowRun.mockReturnValue(null)
+	mocks.getOpenWorkflowBackend.mockReturnValue({} as never)
+	mocks.getLatestWorkflowRun.mockResolvedValue(null)
 	config.auth = oidcAuth
 	const defaults = makeConfig()
 	config.sync = {
@@ -82,7 +82,7 @@ beforeEach(() => {
 
 describe('loadDashboardPage', () => {
 	test('assembles stats, setup config, and publish state into the page shape', async () => {
-		mocks.getLatestWorkflowRun.mockReturnValue(makeRun())
+		mocks.getLatestWorkflowRun.mockResolvedValue(makeRun())
 
 		const result = await loadDashboardPage()
 
@@ -103,7 +103,7 @@ describe('loadDashboardPage', () => {
 	})
 
 	test('reports the latest publish attempt even if it failed', async () => {
-		mocks.getLatestWorkflowRun.mockReturnValue(makeRun({ status: 'failed' }))
+		mocks.getLatestWorkflowRun.mockResolvedValue(makeRun({ status: 'failed' }))
 
 		const result = await loadDashboardPage()
 
@@ -127,7 +127,7 @@ describe('loadDashboardPage', () => {
 	})
 
 	test('falls back to null publish state when OpenWorkflow queries fail', async () => {
-		mocks.getOpenWorkflowDb.mockImplementation(() => {
+		mocks.getOpenWorkflowBackend.mockImplementation(() => {
 			throw new Error('db unavailable')
 		})
 

apps/web/src/lib/server/load/dashboard.ts

@@ -2,7 +2,7 @@ import { db } from '$lib/server/database'
 import { config } from '$lib/server/config'
 import { getPieceStats, getAssetStats, type PieceStats } from '../dashboardStats.js'
 import { getRecentlyEditedPieces } from '../pieces.js'
-import { getOpenWorkflowDb } from '../workflow/index.js'
+import { getOpenWorkflowBackend } from '../workflow/index.js'
 import { getLatestWorkflowRun, type WorkflowRunRow } from '@luzzle/web.jobs'
 import type { WebPieces } from '@luzzle/web.db'
 
@@ -33,8 +33,7 @@ export async function loadDashboardPage(): Promise<DashboardView> {
 	let lastPublish: WorkflowRunRow | null = null
 
 	try {
-		const openWorkflowDb = getOpenWorkflowDb()
-		lastPublish = getLatestWorkflowRun(openWorkflowDb, 'Publish')
+		lastPublish = await getLatestWorkflowRun(getOpenWorkflowBackend(), 'Publish')
 	} catch (err) {
 		console.error('Failed to query OpenWorkflow runs in dashboard loader:', err)
 	}

apps/web/src/lib/server/workflow/index.test.ts

@@ -0,0 +1,31 @@
+import { expect, test, vi } from 'vitest'
+import { publishSpec } from '@luzzle/web.jobs/specs'
+import { config } from '$lib/server/config.js'
+import { getOpenWorkflow, getOpenWorkflowBackend } from './index.js'
+import { createWorkflowQueue, closeWorkflowQueue } from './publish.fixture.js'
+
+vi.mock('$lib/server/config.js', async () => {
+	const { makeConfig } = await import('$lib/server/config.fixture.js')
+	return { config: makeConfig() }
+})
+
+test('exposes the singleton SDK backend used by the web workflow client', async () => {
+	const queue = createWorkflowQueue()
+	config.worker.queue.path = `${queue.directory}/queue.sqlite`
+	const backend = getOpenWorkflowBackend()
+	try {
+		expect(getOpenWorkflowBackend()).toBe(backend)
+		const client = getOpenWorkflow()
+		expect(getOpenWorkflow()).toBe(client)
+		const handle = await client.runWorkflow(publishSpec, { bisync: false })
+		const runs = await backend.listWorkflowRuns({
+			workflowName: 'Publish',
+			status: 'pending',
+			limit: 1
+		})
+		expect(runs.data.map((run) => run.id)).toEqual([handle.workflowRun.id])
+	} finally {
+		await backend.stop()
+		await closeWorkflowQueue(queue)
+	}
+})

apps/web/src/lib/server/workflow/index.ts

@@ -1,11 +1,10 @@
 import path from 'node:path'
-import { DatabaseSync } from 'node:sqlite'
 import { OpenWorkflow } from 'openworkflow'
 import { BackendSqlite } from 'openworkflow/sqlite'
 import { config } from '$lib/server/config.js'
 
 let openWorkflowInstance: OpenWorkflow | null = null
-let openWorkflowDbInstance: DatabaseSync | null = null
+let openWorkflowBackendInstance: BackendSqlite | null = null
 
 function resolveOpenWorkflowDbPath(): string {
 	const queuePath = config.worker?.queue?.path || './data/sidequest.sqlite'
@@ -13,22 +12,16 @@ function resolveOpenWorkflowDbPath(): string {
 	return path.resolve(process.cwd(), openWorkflowPath)
 }
 
+export function getOpenWorkflowBackend(): BackendSqlite {
+	if (!openWorkflowBackendInstance) {
+		openWorkflowBackendInstance = BackendSqlite.connect(resolveOpenWorkflowDbPath())
+	}
+	return openWorkflowBackendInstance
+}
+
 export function getOpenWorkflow(): OpenWorkflow {
 	if (!openWorkflowInstance) {
-		const dbPath = resolveOpenWorkflowDbPath()
-		const backend = BackendSqlite.connect(dbPath)
-		openWorkflowInstance = new OpenWorkflow({ backend })
+		openWorkflowInstance = new OpenWorkflow({ backend: getOpenWorkflowBackend() })
 	}
 	return openWorkflowInstance
 }
-
-export function getOpenWorkflowDb(): DatabaseSync {
-	if (!openWorkflowDbInstance) {
-		const db = new DatabaseSync(resolveOpenWorkflowDbPath())
-		db.exec('PRAGMA journal_mode = WAL;')
-		db.exec('PRAGMA synchronous = NORMAL;')
-		db.exec('PRAGMA busy_timeout = 5000;')
-		openWorkflowDbInstance = db
-	}
-	return openWorkflowDbInstance
-}

apps/web/src/lib/server/workflow/publish.fixture.ts

@@ -0,0 +1,62 @@
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import path from 'node:path'
+import { BackendSqlite } from 'openworkflow/sqlite'
+
+export function createWorkflowQueue() {
+	const directory = mkdtempSync(path.join(tmpdir(), 'luzzle-workflow-'))
+	const filename = path.join(directory, 'queue.sqlite')
+	const backend = BackendSqlite.connect(filename)
+	return { directory, backend }
+}
+
+export type WorkflowQueue = ReturnType<typeof createWorkflowQueue>
+
+export async function closeWorkflowQueue(queue: WorkflowQueue): Promise<void> {
+	await queue.backend.stop()
+	rmSync(queue.directory, { recursive: true, force: true })
+}
+
+export function createPendingRun(
+	backend: BackendSqlite,
+	workflowName: string,
+	input: Record<string, boolean> = {}
+) {
+	return backend.createWorkflowRun({
+		workflowName,
+		version: null,
+		idempotencyKey: null,
+		config: {},
+		context: null,
+		input,
+		parentStepAttemptNamespaceId: null,
+		parentStepAttemptId: null,
+		availableAt: null,
+		deadlineAt: null
+	})
+}
+
+export async function claimNextRun(backend: BackendSqlite) {
+	const run = await backend.claimWorkflowRun({
+		workerId: 'fixture-worker',
+		leaseDurationMs: 60_000
+	})
+	if (!run) throw new Error('Expected a pending workflow run')
+	return run
+}
+
+export async function completeNextRun(backend: BackendSqlite) {
+	const run = await claimNextRun(backend)
+	return backend.completeWorkflowRun({
+		workflowRunId: run.id,
+		workerId: 'fixture-worker',
+		output: null
+	})
+}
+
+export async function createCompletedRun(backend: BackendSqlite, workflowName: string) {
+	const run = await createPendingRun(backend, workflowName)
+	const completed = await completeNextRun(backend)
+	if (completed.id !== run.id) throw new Error('Fixture completed a different pending run')
+	return completed
+}

apps/web/src/lib/server/workflow/publish.test.ts

@@ -1,4 +1,12 @@
-import { describe, test, expect, vi, beforeEach } from 'vitest'
+import { describe, test, expect, vi, beforeEach, afterEach } from 'vitest'
+import {
+	createWorkflowQueue,
+	closeWorkflowQueue,
+	createPendingRun,
+	createCompletedRun,
+	claimNextRun,
+	type WorkflowQueue
+} from './publish.fixture.js'
 import { getLatestWorkflowRun, getWorkflowRun, type WorkflowRunRow } from '@luzzle/web.jobs'
 import { findInFlightPublishRun, validateAuditForPublish, parsePiecesDiff } from './publish.js'
 
@@ -12,7 +20,7 @@ const mocks = {
 	getWorkflowRun: vi.mocked(getWorkflowRun)
 }
 
-const db = {} as never
+const backend = {} as never
 
 function makeRun(overrides: Partial<WorkflowRunRow> = {}): WorkflowRunRow {
 	return {
@@ -33,104 +41,202 @@ beforeEach(() => {
 })
 
 describe('findInFlightPublishRun', () => {
-	test('returns the in-flight Publish run', () => {
-		mocks.getLatestWorkflowRun.mockImplementation((_db, name) =>
-			name === 'Publish'
-				? makeRun({ id: 'pub-1', workflow_name: 'Publish', status: 'running' })
-				: null
-		)
-		expect(findInFlightPublishRun(db)?.id).toBe('pub-1')
+	let queue: WorkflowQueue
+
+	beforeEach(() => {
+		vi.setSystemTime(new Date('2026-06-20T00:00:00Z'))
+		queue = createWorkflowQueue()
 	})
 
-	test('returns the in-flight PublishAudit run', () => {
-		mocks.getLatestWorkflowRun.mockImplementation((_db, name) =>
-			name === 'PublishAudit' ? makeRun({ id: 'aud-1', status: 'pending' }) : null
-		)
-		expect(findInFlightPublishRun(db)?.id).toBe('aud-1')
+	afterEach(async () => {
+		vi.useRealTimers()
+		await closeWorkflowQueue(queue)
 	})
 
-	test('returns null when neither is in-flight', () => {
-		mocks.getLatestWorkflowRun.mockImplementation((_db, name) =>
-			makeRun({ workflow_name: name, status: 'completed' })
-		)
-		expect(findInFlightPublishRun(db)).toBeNull()
+	test.each([
+		{ name: 'Publish', status: 'pending' },
+		{ name: 'Publish', status: 'running' },
+		{ name: 'PublishAudit', status: 'pending' },
+		{ name: 'PublishAudit', status: 'running' }
+	])('finds an older $status $name behind a newer completed run', async ({ name, status }) => {
+		const old = await createPendingRun(queue.backend, name)
+		await claimNextRun(queue.backend)
+		vi.setSystemTime(new Date('2026-06-20T01:00:00Z'))
+		await createCompletedRun(queue.backend, name)
+		if (status === 'pending') {
+			await queue.backend.rescheduleWorkflowRunAfterFailedStepAttempt({
+				workflowRunId: old.id,
+				workerId: 'fixture-worker',
+				error: { message: 'retry' },
+				availableAt: new Date()
+			})
+		}
+		expect(await findInFlightPublishRun(queue.backend)).toEqual({ id: old.id })
 	})
 
-	test('returns null when there are no runs', () => {
-		mocks.getLatestWorkflowRun.mockReturnValue(null)
-		expect(findInFlightPublishRun(db)).toBeNull()
+	test('returns an active run when both workflows have pending work', async () => {
+		const audit = await createPendingRun(queue.backend, 'PublishAudit')
+		const publish = await createPendingRun(queue.backend, 'Publish')
+		const active = await findInFlightPublishRun(queue.backend)
+		expect([audit.id, publish.id]).toContain(active?.id)
+	})
+
+	test('ignores finished runs and unrelated active workflows', async () => {
+		await createCompletedRun(queue.backend, 'PublishAudit')
+		await createPendingRun(queue.backend, 'Publish')
+		const failed = await claimNextRun(queue.backend)
+		await queue.backend.failWorkflowRun({
+			workflowRunId: failed.id,
+			workerId: 'fixture-worker',
+			error: { message: 'failed' },
+			retryPolicy: {
+				maximumAttempts: 1,
+				initialInterval: '1s',
+				backoffCoefficient: 2,
+				maximumInterval: '1s'
+			}
+		})
+		await createPendingRun(queue.backend, 'Preview')
+		expect(await findInFlightPublishRun(queue.backend)).toBeNull()
+	})
+
+	test.each(['Publish', 'PublishAudit'])(
+		'finds an active %s in the fifth-newest position',
+		async (name) => {
+			const old = await createPendingRun(queue.backend, name)
+			await claimNextRun(queue.backend)
+			vi.setSystemTime(new Date('2026-06-20T01:00:00Z'))
+			for (let i = 0; i < 4; i++) await createCompletedRun(queue.backend, name)
+			expect(await findInFlightPublishRun(queue.backend)).toEqual({ id: old.id })
+		}
+	)
+
+	test.each(['Publish', 'PublishAudit'])(
+		'accepts the known limit: a sixth-newest active %s is not detected',
+		async (name) => {
+			const old = await createPendingRun(queue.backend, name)
+			await claimNextRun(queue.backend)
+			vi.setSystemTime(new Date('2026-06-20T01:00:00Z'))
+			for (let i = 0; i < 5; i++) await createCompletedRun(queue.backend, name)
+			expect(await findInFlightPublishRun(queue.backend)).toBeNull()
+			expect((await queue.backend.getWorkflowRun({ workflowRunId: old.id }))?.status).toBe(
+				'running'
+			)
+		}
+	)
+
+	test('checks only the latest five per workflow without paging, status filters or counts', async () => {
+		for (let i = 0; i < 5; i++) {
+			await createCompletedRun(queue.backend, 'Publish')
+			await createCompletedRun(queue.backend, 'PublishAudit')
+		}
+		await createPendingRun(queue.backend, 'Preview')
+		const list = vi.spyOn(queue.backend, 'listWorkflowRuns')
+		const count = vi.spyOn(queue.backend, 'countWorkflowRuns')
+		expect(await findInFlightPublishRun(queue.backend)).toBeNull()
+		expect(list.mock.calls.map(([params]) => params)).toEqual([
+			{ workflowName: 'Publish', limit: 5 },
+			{ workflowName: 'PublishAudit', limit: 5 }
+		])
+		expect(count).not.toHaveBeenCalled()
+	})
+
+	test('does not miss a running run rescheduled to pending during lookup', async () => {
+		const run = await createPendingRun(queue.backend, 'Publish')
+		await claimNextRun(queue.backend)
+		const original = queue.backend.listWorkflowRuns.bind(queue.backend)
+		let rescheduled = false
+		vi.spyOn(queue.backend, 'listWorkflowRuns').mockImplementation(async (params) => {
+			const page = await original(params)
+			if (params.workflowName === 'Publish' && !rescheduled) {
+				await queue.backend.rescheduleWorkflowRunAfterFailedStepAttempt({
+					workflowRunId: run.id,
+					workerId: 'fixture-worker',
+					error: { message: 'retry' },
+					availableAt: new Date()
+				})
+				rescheduled = true
+			}
+			return page
+		})
+		expect(await findInFlightPublishRun(queue.backend)).toEqual({ id: run.id })
+		expect((await queue.backend.getWorkflowRun({ workflowRunId: run.id }))?.status).toBe('pending')
+	})
+
+	test('returns null when there are no runs', async () => {
+		expect(await findInFlightPublishRun(queue.backend)).toBeNull()
 	})
 })
 
 describe('validateAuditForPublish', () => {
-	test('rejects a missing or non-string id', () => {
-		expect(validateAuditForPublish(db, undefined).ok).toBe(false)
-		expect(validateAuditForPublish(db, '').ok).toBe(false)
+	test('rejects a missing or non-string id', async () => {
+		expect((await validateAuditForPublish(backend, undefined)).ok).toBe(false)
+		expect((await validateAuditForPublish(backend, '')).ok).toBe(false)
 	})
 
-	test('rejects when the run is not found', () => {
-		mocks.getWorkflowRun.mockReturnValue(null)
-		expect(validateAuditForPublish(db, 'audit-1')).toEqual({
+	test('rejects when the run is not found', async () => {
+		mocks.getWorkflowRun.mockResolvedValue(null)
+		expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({
 			ok: false,
 			reason: 'audit run not found'
 		})
 	})
 
-	test('rejects when the run is not a PublishAudit', () => {
-		mocks.getWorkflowRun.mockReturnValue(makeRun({ workflow_name: 'Publish' }))
-		expect(validateAuditForPublish(db, 'audit-1').ok).toBe(false)
+	test('rejects when the run is not a PublishAudit', async () => {
+		mocks.getWorkflowRun.mockResolvedValue(makeRun({ workflow_name: 'Publish' }))
+		expect((await validateAuditForPublish(backend, 'audit-1')).ok).toBe(false)
 	})
 
-	test('rejects when the audit has not completed', () => {
-		mocks.getWorkflowRun.mockReturnValue(makeRun({ status: 'running' }))
-		expect(validateAuditForPublish(db, 'audit-1')).toEqual({
+	test('rejects when the audit has not completed', async () => {
+		mocks.getWorkflowRun.mockResolvedValue(makeRun({ status: 'running' }))
+		expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({
 			ok: false,
 			reason: 'audit has not completed'
 		})
 	})
 
-	test('rejects when a newer audit has run since', () => {
-		mocks.getWorkflowRun.mockReturnValue(makeRun({ id: 'audit-1' }))
-		mocks.getLatestWorkflowRun.mockReturnValue(makeRun({ id: 'audit-2' }))
-		expect(validateAuditForPublish(db, 'audit-1')).toEqual({
+	test('rejects when a newer audit has run since', async () => {
+		mocks.getWorkflowRun.mockResolvedValue(makeRun({ id: 'audit-1' }))
+		mocks.getLatestWorkflowRun.mockResolvedValue(makeRun({ id: 'audit-2' }))
+		expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({
 			ok: false,
 			reason: 'a newer audit has run; re-check before publishing'
 		})
 	})
 
-	test('rejects when a publish ran after the audit', () => {
-		mocks.getWorkflowRun.mockReturnValue(
+	test('rejects when a publish ran after the audit', async () => {
+		mocks.getWorkflowRun.mockResolvedValue(
 			makeRun({ id: 'audit-1', created_at: '2026-06-20T00:00:00Z' })
 		)
-		mocks.getLatestWorkflowRun.mockImplementation((_db, name) =>
+		mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) =>
 			name === 'PublishAudit'
 				? makeRun({ id: 'audit-1', created_at: '2026-06-20T00:00:00Z' })
 				: makeRun({ id: 'pub-1', workflow_name: 'Publish', created_at: '2026-06-20T01:00:00Z' })
 		)
-		expect(validateAuditForPublish(db, 'audit-1')).toEqual({
+		expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({
 			ok: false,
 			reason: 'changes were published after this check; re-check before publishing'
 		})
 	})
 
-	test('accepts a completed audit that is the latest with no later publish', () => {
-		mocks.getWorkflowRun.mockReturnValue(makeRun({ id: 'audit-1' }))
-		mocks.getLatestWorkflowRun.mockImplementation((_db, name) =>
+	test('accepts a completed audit that is the latest with no later publish', async () => {
+		mocks.getWorkflowRun.mockResolvedValue(makeRun({ id: 'audit-1' }))
+		mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) =>
 			name === 'PublishAudit' ? makeRun({ id: 'audit-1' }) : null
 		)
-		expect(validateAuditForPublish(db, 'audit-1')).toEqual({ ok: true })
+		expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({ ok: true })
 	})
 
-	test('accepts a succeeded audit newer than the last publish', () => {
-		mocks.getWorkflowRun.mockReturnValue(
+	test('accepts a succeeded audit newer than the last publish', async () => {
+		mocks.getWorkflowRun.mockResolvedValue(
 			makeRun({ id: 'audit-1', status: 'succeeded', created_at: '2026-06-20T02:00:00Z' })
 		)
-		mocks.getLatestWorkflowRun.mockImplementation((_db, name) =>
+		mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) =>
 			name === 'PublishAudit'
 				? makeRun({ id: 'audit-1', status: 'succeeded', created_at: '2026-06-20T02:00:00Z' })
 				: makeRun({ id: 'pub-1', workflow_name: 'Publish', created_at: '2026-06-20T01:00:00Z' })
 		)
-		expect(validateAuditForPublish(db, 'audit-1')).toEqual({ ok: true })
+		expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({ ok: true })
 	})
 })
 

apps/web/src/lib/server/workflow/publish.ts

@@ -1,30 +1,46 @@
-import type { DatabaseSync } from 'node:sqlite'
-import { getLatestWorkflowRun, getWorkflowRun, type WorkflowRunRow } from '@luzzle/web.jobs'
+import type { BackendSqlite } from 'openworkflow/sqlite'
+import { getLatestWorkflowRun, getWorkflowRun } from '@luzzle/web.jobs'
 import type { PiecesDiff } from '@luzzle/core'
 
-const PUBLISH_WORKFLOWS = ['Publish', 'PublishAudit'] as const
-const IN_FLIGHT_STATES = new Set(['pending', 'running'])
 const COMPLETED_STATES = new Set(['completed', 'succeeded'])
+let publishAdmission: Promise<void> = Promise.resolve()
 
-// Publish and PublishAudit are mutually exclusive: an audit's bisync mutates the archive a publish reads.
-export function findInFlightPublishRun(db: DatabaseSync): WorkflowRunRow | null {
-	for (const name of PUBLISH_WORKFLOWS) {
-		const run = getLatestWorkflowRun(db, name)
-		if (run && IN_FLIGHT_STATES.has(run.status)) {
-			return run
-		}
+export async function withPublishAdmission<T>(operation: () => Promise<T>): Promise<T> {
+	const previous = publishAdmission
+	let release!: () => void
+	publishAdmission = new Promise<void>((resolve) => {
+		release = resolve
+	})
+	await previous
+	try {
+		return await operation()
+	} finally {
+		release()
+	}
+}
+
+export async function findInFlightPublishRun(
+	backend: Pick<BackendSqlite, 'listWorkflowRuns'>
+): Promise<{ id: string } | null> {
+	for (const workflowName of ['Publish', 'PublishAudit']) {
+		const { data } = await backend.listWorkflowRuns({ workflowName, limit: 5 })
+		const active = data.find((run) => run.status === 'pending' || run.status === 'running')
+		if (active) return { id: active.id }
 	}
 	return null
 }
 
 export type AuditGuard = { ok: true } | { ok: false; reason: string }
 
-export function validateAuditForPublish(db: DatabaseSync, auditRunId: unknown): AuditGuard {
+export async function validateAuditForPublish(
+	backend: Pick<BackendSqlite, 'getWorkflowRun' | 'listWorkflowRuns'>,
+	auditRunId: unknown
+): Promise<AuditGuard> {
 	if (typeof auditRunId !== 'string' || auditRunId.length === 0) {
 		return { ok: false, reason: 'no audit run provided; check for changes before publishing' }
 	}
 
-	const run = getWorkflowRun(db, auditRunId)
+	const run = await getWorkflowRun(backend, auditRunId)
 	if (!run || run.workflow_name !== 'PublishAudit') {
 		return { ok: false, reason: 'audit run not found' }
 	}
@@ -32,13 +48,13 @@ export function validateAuditForPublish(db: DatabaseSync, auditRunId: unknown):
 		return { ok: false, reason: 'audit has not completed' }
 	}
 
-	const latest = getLatestWorkflowRun(db, 'PublishAudit')
+	const latest = await getLatestWorkflowRun(backend, 'PublishAudit')
 	if (!latest || latest.id !== auditRunId) {
 		return { ok: false, reason: 'a newer audit has run; re-check before publishing' }
 	}
 
 	// a publish consumes its audit; created_at is ISO so string compare is chronological
-	const lastPublish = getLatestWorkflowRun(db, 'Publish')
+	const lastPublish = await getLatestWorkflowRun(backend, 'Publish')
 	if (lastPublish && lastPublish.created_at > run.created_at) {
 		return {
 			ok: false,

apps/web/src/lib/server/workflow/stream.test.ts

@@ -12,7 +12,7 @@ const mocks = vi.hoisted(() => {
 	return {
 		query,
 		selectFrom: vi.fn(() => query),
-		getOpenWorkflowDb: vi.fn(() => ({})),
+		getOpenWorkflowBackend: vi.fn(() => ({})),
 		getOpenWorkflow: vi.fn(),
 		getWorkflowRun: vi.fn(),
 		getStepAttempts: vi.fn()
@@ -23,7 +23,7 @@ vi.mock('$lib/server/database/index.js', () => ({
 	db: { selectFrom: mocks.selectFrom }
 }))
 vi.mock('./index.js', () => ({
-	getOpenWorkflowDb: mocks.getOpenWorkflowDb,
+	getOpenWorkflowBackend: mocks.getOpenWorkflowBackend,
 	getOpenWorkflow: mocks.getOpenWorkflow
 }))
 vi.mock('@luzzle/web.jobs', () => ({
@@ -100,9 +100,9 @@ beforeEach(() => {
 	mocks.query.selectAll.mockReturnThis()
 	mocks.query.where.mockReturnThis()
 	mocks.query.orderBy.mockReturnThis()
-	mocks.getOpenWorkflowDb.mockReturnValue({})
-	mocks.getWorkflowRun.mockReturnValue(run)
-	mocks.getStepAttempts.mockReturnValue([step])
+	mocks.getOpenWorkflowBackend.mockReturnValue({})
+	mocks.getWorkflowRun.mockResolvedValue(run)
+	mocks.getStepAttempts.mockResolvedValue([step])
 	mocks.query.execute.mockResolvedValue([])
 })
 
@@ -342,6 +342,39 @@ describe('streamJobProgress', () => {
 		}
 	)
 
+	test.each([
+		{ read: 'run', disconnect: 'request abort' },
+		{ read: 'run', disconnect: 'reader cancel' },
+		{ read: 'steps', disconnect: 'request abort' },
+		{ read: 'steps', disconnect: 'reader cancel' }
+	])(
+		'$disconnect during an SDK $read read does not start later queries',
+		async ({ read, disconnect }) => {
+			vi.useFakeTimers()
+			let finishRead!: () => void
+			const pending = new Promise<unknown>((resolve) => {
+				finishRead = () => resolve(read === 'run' ? run : [step])
+			})
+			if (read === 'run') mocks.getWorkflowRun.mockReturnValueOnce(pending)
+			else mocks.getStepAttempts.mockReturnValueOnce(pending)
+			const abort = new AbortController()
+			const reader = openStream({ signal: abort.signal }).body!.getReader()
+			if (read === 'steps') await readFrame(reader)
+			expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(1)
+			if (disconnect === 'request abort') abort.abort()
+			else await reader.cancel()
+			expect(await reader.read()).toEqual({ done: true, value: undefined })
+			expect(vi.getTimerCount()).toBe(0)
+
+			finishRead()
+			await vi.advanceTimersByTimeAsync(0)
+			expect(mocks.getStepAttempts).toHaveBeenCalledTimes(read === 'run' ? 0 : 1)
+			expect(mocks.query.execute).not.toHaveBeenCalled()
+			expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(1)
+			expect(vi.getTimerCount()).toBe(0)
+		}
+	)
+
 	test('an already-aborted request never polls, emits, or installs timers', async () => {
 		vi.useFakeTimers()
 		const abort = new AbortController()
@@ -350,7 +383,7 @@ describe('streamJobProgress', () => {
 		expect(await readChunks(openStream({ signal: abort.signal }))).toEqual([])
 		expect(vi.getTimerCount()).toBe(0)
 		await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 2)
-		expect(mocks.getOpenWorkflowDb).not.toHaveBeenCalled()
+		expect(mocks.getOpenWorkflowBackend).not.toHaveBeenCalled()
 		expect(mocks.getWorkflowRun).not.toHaveBeenCalled()
 		expect(mocks.getStepAttempts).not.toHaveBeenCalled()
 		expect(mocks.query.execute).not.toHaveBeenCalled()

apps/web/src/lib/server/workflow/stream.ts

@@ -1,7 +1,7 @@
 import { db, type JobProgressRow } from '$lib/server/database/index.js'
 import { createEventStream } from '../sse.js'
 import { WORKFLOW_POLL_INTERVAL_MS } from '../constants.js'
-import { getOpenWorkflowDb } from './index.js'
+import { getOpenWorkflowBackend } from './index.js'
 import { getWorkflowRun, getStepAttempts } from '@luzzle/web.jobs'
 
 const TERMINAL_STATES = new Set(['completed', 'failed', 'canceled'])
@@ -56,7 +56,8 @@ async function pollOnce(
 	jobId: string,
 	jobClass: string | string[],
 	cursors: Cursors,
-	emit: Emit
+	emit: Emit,
+	signal: AbortSignal
 ): Promise<boolean> {
 	try {
 		let job: { class: string; state: string; result: unknown; errors: unknown } | null = null
@@ -64,8 +65,8 @@ async function pollOnce(
 
 		// Query OpenWorkflow
 		try {
-			const openWorkflowDb = getOpenWorkflowDb()
-			const run = getWorkflowRun(openWorkflowDb, jobId)
+			const run = await getWorkflowRun(getOpenWorkflowBackend(), jobId)
+			if (signal.aborted) return true
 			if (run) {
 				let state = 'waiting'
 				if (run.status === 'running') state = 'running'
@@ -100,9 +101,10 @@ async function pollOnce(
 
 		let phases: JobProgressRow[] = []
 		if (runId) {
+			if (signal.aborted) return true
 			try {
-				const openWorkflowDb = getOpenWorkflowDb()
-				const rows = getStepAttempts(openWorkflowDb, runId)
+				const rows = await getStepAttempts(getOpenWorkflowBackend(), runId)
+				if (signal.aborted) return true
 				phases = rows.map((r) => {
 					let status = 'waiting'
 					if (r.status === 'running') status = 'running'
@@ -129,7 +131,9 @@ async function pollOnce(
 
 		let hasNewLogs = false
 		for (const phase of phases) {
+			if (signal.aborted) return true
 			const newLogs = await fetchNewLogs(jobId, phase.phase, cursors[phase.phase] ?? 0)
+			if (signal.aborted) return true
 			if (newLogs.length > 0) {
 				hasNewLogs = true
 				emit('log', newLogs)
@@ -167,7 +171,7 @@ export function streamJobProgress({
 	async function poll() {
 		try {
 			while (!events.signal.aborted) {
-				if (await pollOnce(jobId, jobClass, cursors, events.emit)) break
+				if (await pollOnce(jobId, jobClass, cursors, events.emit, events.signal)) break
 				await waitForNextPoll(events.signal)
 			}
 		} catch (error) {

apps/web/src/routes/admin/piece/[...path]/preview/[jobId]/+layout.server.ts

@@ -1,7 +1,7 @@
 import { error } from '@sveltejs/kit'
 import { config } from '$lib/server/config'
 import { assemblePreview, type PreviewWorkerResult } from '$lib/pieces/preview/assemble.server.js'
-import { getOpenWorkflowDb } from '$lib/server/workflow/index.js'
+import { getOpenWorkflowBackend } from '$lib/server/workflow/index.js'
 import { getWorkflowRun, getStepAttempts } from '@luzzle/web.jobs'
 import { db, type JobProgressLogsRow, type JobProgressRow } from '$lib/server/database/index.js'
 import type { LayoutServerLoad } from './$types'
@@ -9,11 +9,10 @@ import type { LayoutServerLoad } from './$types'
 export const load: LayoutServerLoad = async ({ params }) => {
 	const runId = params.jobId
 
-	let run: ReturnType<typeof getWorkflowRun> | undefined
+	let run: Awaited<ReturnType<typeof getWorkflowRun>> | undefined
 
 	try {
-		const openWorkflowDb = getOpenWorkflowDb()
-		run = getWorkflowRun(openWorkflowDb, runId)
+		run = await getWorkflowRun(getOpenWorkflowBackend(), runId)
 	} catch (err) {
 		console.error('Failed to query OpenWorkflow preview run:', err)
 	}
@@ -30,8 +29,7 @@ export const load: LayoutServerLoad = async ({ params }) => {
 
 	let phases: JobProgressRow[] = []
 	try {
-		const openWorkflowDb = getOpenWorkflowDb()
-		const rows = getStepAttempts(openWorkflowDb, runId)
+		const rows = await getStepAttempts(getOpenWorkflowBackend(), runId)
 		phases = rows.map((r) => {
 			let status = 'waiting'
 			if (r.status === 'running') status = 'running'

apps/web/src/routes/admin/piece/[...path]/preview/[jobId]/layout.server.test.ts

@@ -0,0 +1,91 @@
+import { afterEach, beforeEach, expect, test, vi } from 'vitest'
+import { getWorkflowRun, getStepAttempts, type WorkflowRunRow } from '@luzzle/web.jobs'
+import { load } from './+layout.server.js'
+
+const { assemblePreview, query } = vi.hoisted(() => ({
+	assemblePreview: vi.fn(() => ({ html: 'rendered-preview' })),
+	query: {
+		selectAll: vi.fn().mockReturnThis(),
+		where: vi.fn().mockReturnThis(),
+		orderBy: vi.fn().mockReturnThis(),
+		execute: vi.fn()
+	}
+}))
+vi.mock('$lib/server/database/index.js', () => ({ db: { selectFrom: () => query } }))
+vi.mock('$lib/server/config', async () => {
+	const { makeConfig } = await import('$lib/server/config.fixture.js')
+	return {
+		config: makeConfig({
+			pieces: [{ type: 'book', fields: { title: 'title', date_consumed: 'read_on' } }]
+		})
+	}
+})
+vi.mock('$lib/server/workflow/index.js', () => ({ getOpenWorkflowBackend: () => ({}) }))
+vi.mock('$lib/pieces/preview/assemble.server.js', () => ({ assemblePreview }))
+vi.mock('@luzzle/web.jobs', () => ({ getWorkflowRun: vi.fn(), getStepAttempts: vi.fn() }))
+
+const preview = {
+	filePath: 'example.book.md',
+	type: 'book',
+	slug: 'example',
+	pieceKey: 'piece-key',
+	sanitizedFrontmatter: { title: 'Draft' },
+	note: 'note',
+	pathToKey: {},
+	transforms: []
+}
+function run(overrides: Partial<WorkflowRunRow> = {}): WorkflowRunRow {
+	return {
+		id: 'preview-id',
+		workflow_name: 'Preview',
+		status: 'completed',
+		error: null,
+		input: '{}',
+		output: JSON.stringify(preview),
+		finished_at: '2026-06-21T00:00:00Z',
+		created_at: '2026-06-20T00:00:00Z',
+		...overrides
+	}
+}
+function event() {
+	return { params: { jobId: 'preview-id', path: 'example.book.md' } } as Parameters<typeof load>[0]
+}
+
+beforeEach(() => {
+	vi.clearAllMocks()
+	vi.setSystemTime(new Date('2026-06-22T00:00:00Z'))
+	query.execute.mockResolvedValue([])
+	vi.mocked(getWorkflowRun).mockResolvedValue(run())
+	vi.mocked(getStepAttempts).mockResolvedValue([])
+})
+afterEach(() => vi.useRealTimers())
+
+test('awaits SDK-backed metadata and preserves completed preview assembly', async () => {
+	expect(await load(event())).toMatchObject({
+		file: 'example.book.md',
+		status: 'completed',
+		jobId: 'preview-id',
+		phases: [],
+		logs: [],
+		html: 'rendered-preview'
+	})
+	expect(assemblePreview).toHaveBeenCalledWith(preview, expect.objectContaining({ type: 'book' }))
+})
+
+test('preserves failed preview error data', async () => {
+	vi.mocked(getWorkflowRun).mockResolvedValue(run({ status: 'failed', error: 'preview failed' }))
+	expect(await load(event())).toMatchObject({ status: 'failed', errorMessage: 'preview failed' })
+	expect(assemblePreview).not.toHaveBeenCalled()
+})
+
+test('preserves expiration based on the serialized SDK finish time', async () => {
+	vi.mocked(getWorkflowRun).mockResolvedValue(run({ finished_at: '2026-06-19T00:00:00Z' }))
+	expect(await load(event())).toMatchObject({ status: 'expired', jobId: 'preview-id' })
+	expect(assemblePreview).not.toHaveBeenCalled()
+})
+
+test('keeps missing SDK runs as a 404', async () => {
+	vi.mocked(getWorkflowRun).mockResolvedValue(null)
+	await expect(load(event())).rejects.toMatchObject({ status: 404 })
+	expect(getStepAttempts).not.toHaveBeenCalled()
+})

apps/web/src/routes/admin/publish/+page.server.ts

@@ -1,6 +1,6 @@
 import { config } from '$lib/server/config'
 import { db, type JobProgressRow, type JobProgressLogsRow } from '$lib/server/database/index.js'
-import { getOpenWorkflowDb } from '$lib/server/workflow/index.js'
+import { getOpenWorkflowBackend } from '$lib/server/workflow/index.js'
 import { parsePiecesDiff } from '$lib/server/workflow/publish.js'
 import { getLatestWorkflowRun, getStepAttempts, type WorkflowRunRow } from '@luzzle/web.jobs'
 import type { PiecesDiff } from '@luzzle/core'
@@ -24,8 +24,8 @@ function mapState(status: string): string {
 	return 'waiting'
 }
 
-function mapPhases(jobId: string): JobProgressRow[] {
-	const rows = getStepAttempts(getOpenWorkflowDb(), jobId)
+async function mapPhases(jobId: string): Promise<JobProgressRow[]> {
+	const rows = await getStepAttempts(getOpenWorkflowBackend(), jobId)
 	return rows.map((r) => ({
 		job_id: jobId,
 		phase: r.phase,
@@ -51,7 +51,7 @@ async function buildRunView(run: WorkflowRunRow | null): Promise<RunView | null>
 	let phases: JobProgressRow[] = []
 	let logs: JobProgressLogsRow[] = []
 	try {
-		phases = mapPhases(run.id)
+		phases = await mapPhases(run.id)
 		logs = await loadLogs(run.id)
 	} catch (err) {
 		console.error('Failed to load run progress in publish loader:', err)
@@ -74,9 +74,9 @@ export const load: PageServerLoad = async () => {
 	let publish: RunView | null = null
 
 	try {
-		const openWorkflowDb = getOpenWorkflowDb()
-		audit = await buildRunView(getLatestWorkflowRun(openWorkflowDb, 'PublishAudit'))
-		publish = await buildRunView(getLatestWorkflowRun(openWorkflowDb, 'Publish'))
+		const backend = getOpenWorkflowBackend()
+		audit = await buildRunView(await getLatestWorkflowRun(backend, 'PublishAudit'))
+		publish = await buildRunView(await getLatestWorkflowRun(backend, 'Publish'))
 	} catch (err) {
 		console.error('Failed to query OpenWorkflow runs in publish loader:', err)
 	}

apps/web/src/routes/admin/publish/page.server.test.ts

@@ -0,0 +1,89 @@
+import { beforeEach, expect, test, vi } from 'vitest'
+import { getLatestWorkflowRun, getStepAttempts, type WorkflowRunRow } from '@luzzle/web.jobs'
+import { load } from './+page.server.js'
+
+const { query } = vi.hoisted(() => ({
+	query: {
+		selectAll: vi.fn().mockReturnThis(),
+		where: vi.fn().mockReturnThis(),
+		orderBy: vi.fn().mockReturnThis(),
+		execute: vi.fn()
+	}
+}))
+vi.mock('$lib/server/database/index.js', () => ({ db: { selectFrom: () => query } }))
+vi.mock('$lib/server/config', async () => {
+	const { makeConfig } = await import('$lib/server/config.fixture.js')
+	return { config: makeConfig() }
+})
+vi.mock('$lib/server/workflow/index.js', () => ({ getOpenWorkflowBackend: () => ({}) }))
+vi.mock('@luzzle/web.jobs', () => ({ getLatestWorkflowRun: vi.fn(), getStepAttempts: vi.fn() }))
+
+const diff = {
+	schemas: { added: [], updated: [], pruned: [] },
+	pieces: { added: ['new.book.md'], updated: [], pruned: [] }
+}
+function run(name: string): WorkflowRunRow {
+	return {
+		id: `${name}-id`,
+		workflow_name: name,
+		status: 'completed',
+		error: null,
+		input: '{}',
+		output: JSON.stringify(diff),
+		finished_at: '2026-06-20T00:01:00Z',
+		created_at: '2026-06-20T00:00:00Z'
+	}
+}
+
+beforeEach(() => {
+	vi.clearAllMocks()
+	query.execute.mockResolvedValue([])
+	vi.mocked(getLatestWorkflowRun).mockImplementation(async (_backend, name) => run(name))
+	vi.mocked(getStepAttempts).mockResolvedValue([
+		{
+			phase: 'sync',
+			status: 'completed',
+			started_at: '2026-06-20T00:00:00Z',
+			finished_at: '2026-06-20T00:01:00Z',
+			message: null
+		}
+	])
+})
+
+test('awaits SDK-backed run and step reads while preserving publish page data', async () => {
+	const result = await load({} as Parameters<typeof load>[0])
+	if (!result) throw new Error('Expected publish page data')
+	for (const [view, name] of [
+		[result.audit, 'PublishAudit'],
+		[result.publish, 'Publish']
+	] as const) {
+		expect(view).toEqual({
+			jobId: `${name}-id`,
+			state: 'completed',
+			errors: null,
+			logs: [],
+			diff,
+			phases: [
+				{
+					job_id: `${name}-id`,
+					phase: 'sync',
+					status: 'completed',
+					started_at: Date.parse('2026-06-20T00:00:00Z'),
+					finished_at: Date.parse('2026-06-20T00:01:00Z'),
+					message: null
+				}
+			]
+		})
+	}
+	expect(query.where).toHaveBeenCalledWith('job_id', '=', 'PublishAudit-id')
+	expect(query.where).toHaveBeenCalledWith('job_id', '=', 'Publish-id')
+})
+
+test('keeps absent run views null', async () => {
+	vi.mocked(getLatestWorkflowRun).mockResolvedValue(null)
+	const result = await load({} as Parameters<typeof load>[0])
+	if (!result) throw new Error('Expected publish page data')
+	expect(result.audit).toBeNull()
+	expect(result.publish).toBeNull()
+	expect(getStepAttempts).not.toHaveBeenCalled()
+})

apps/web/src/routes/api/admin/publish/+server.ts

@@ -1,34 +1,38 @@
 import { json } from '@sveltejs/kit'
 import type { RequestHandler } from './$types'
-import { getOpenWorkflow, getOpenWorkflowDb } from '$lib/server/workflow/index.js'
-import { findInFlightPublishRun, validateAuditForPublish } from '$lib/server/workflow/publish.js'
+import { getOpenWorkflow, getOpenWorkflowBackend } from '$lib/server/workflow/index.js'
+import {
+	findInFlightPublishRun,
+	validateAuditForPublish,
+	withPublishAdmission
+} from '$lib/server/workflow/publish.js'
 import { publishSpec } from '@luzzle/web.jobs/specs'
 
 export const POST: RequestHandler = async ({ request }) => {
 	try {
-		const db = getOpenWorkflowDb()
-
-		const inFlight = findInFlightPublishRun(db)
-		if (inFlight) {
-			return json({ jobId: inFlight.id }, { status: 409 })
-		}
-
 		const body = await request.json().catch(() => ({}))
-		const auditRunId = body?.auditRunId
-		const bisync = body?.bisync === true
-
-		if (auditRunId) {
-			const guard = validateAuditForPublish(db, auditRunId)
-			if (!guard.ok) {
-				return json({ message: guard.reason }, { status: 412 })
+		return await withPublishAdmission(async () => {
+			const inFlight = await findInFlightPublishRun(getOpenWorkflowBackend())
+			if (inFlight) {
+				return json({ jobId: inFlight.id }, { status: 409 })
 			}
-		}
 
-		const openWorkflow = getOpenWorkflow()
-		const handle = await openWorkflow.runWorkflow(publishSpec, {
-			bisync: auditRunId ? false : bisync
+			const auditRunId = body?.auditRunId
+			const bisync = body?.bisync === true
+
+			if (auditRunId) {
+				const guard = await validateAuditForPublish(getOpenWorkflowBackend(), auditRunId)
+				if (!guard.ok) {
+					return json({ message: guard.reason }, { status: 412 })
+				}
+			}
+
+			const openWorkflow = getOpenWorkflow()
+			const handle = await openWorkflow.runWorkflow(publishSpec, {
+				bisync: auditRunId ? false : bisync
+			})
+			return json({ jobId: handle.workflowRun.id })
 		})
-		return json({ jobId: handle.workflowRun.id })
 	} catch (error) {
 		const message = error instanceof Error ? error.message : String(error)
 		return new Response(`Internal server error: ${message}`, { status: 500 })

apps/web/src/routes/api/admin/publish/admission.test.ts

@@ -0,0 +1,240 @@
+import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'
+import { getOpenWorkflowBackend } from '$lib/server/workflow/index.js'
+import {
+	createWorkflowQueue,
+	closeWorkflowQueue,
+	createPendingRun,
+	createCompletedRun,
+	claimNextRun,
+	completeNextRun,
+	type WorkflowQueue
+} from '$lib/server/workflow/publish.fixture.js'
+import { POST as publish } from './+server.js'
+import { POST as audit } from './audit/+server.js'
+
+const { runWorkflow } = vi.hoisted(() => ({ runWorkflow: vi.fn() }))
+vi.mock('$lib/server/workflow/index.js', () => ({
+	getOpenWorkflow: () => ({ runWorkflow }),
+	getOpenWorkflowBackend: vi.fn()
+}))
+
+function event(body: unknown = {}) {
+	return {
+		request: new Request('http://localhost/api/admin/publish', {
+			method: 'POST',
+			body: JSON.stringify(body)
+		})
+	} as Parameters<typeof publish>[0] & Parameters<typeof audit>[0]
+}
+
+describe('publish/audit admission', () => {
+	let queue: WorkflowQueue
+
+	async function enqueue(spec: { name: string }, input: { bisync: boolean }) {
+		vi.setSystemTime(new Date('2026-06-21T00:00:00Z'))
+		const workflowRun = await createPendingRun(queue.backend, spec.name, input)
+		return { workflowRun }
+	}
+
+	beforeEach(() => {
+		vi.resetAllMocks()
+		vi.setSystemTime(new Date('2026-06-20T00:00:00Z'))
+		queue = createWorkflowQueue()
+		vi.mocked(getOpenWorkflowBackend).mockReturnValue(queue.backend)
+		runWorkflow.mockImplementation(async (spec, input) => enqueue(spec, input))
+	})
+
+	afterEach(async () => {
+		vi.useRealTimers()
+		await closeWorkflowQueue(queue)
+	})
+
+	test.each([
+		{ firstName: 'Publish', firstRoute: publish, otherName: 'Publish', otherRoute: publish },
+		{ firstName: 'Publish', firstRoute: publish, otherName: 'PublishAudit', otherRoute: audit },
+		{ firstName: 'PublishAudit', firstRoute: audit, otherName: 'Publish', otherRoute: publish },
+		{ firstName: 'PublishAudit', firstRoute: audit, otherName: 'PublishAudit', otherRoute: audit }
+	])(
+		'admits only one overlapping $firstName/$otherName request',
+		async ({ firstRoute, otherRoute }) => {
+			let release!: () => void
+			let entered!: () => void
+			const hold = new Promise<void>((resolve) => {
+				release = resolve
+			})
+			const started = new Promise<void>((resolve) => {
+				entered = resolve
+			})
+			runWorkflow.mockImplementationOnce(async (spec, input) => {
+				entered()
+				await hold
+				return enqueue(spec, input)
+			})
+
+			const first = Promise.resolve(firstRoute(event()))
+			try {
+				await Promise.race([
+					started,
+					first.then(() => {
+						throw new Error('First request finished before enqueue')
+					})
+				])
+				const other = otherRoute(event())
+				release()
+				const [accepted, rejected] = await Promise.all([first, other])
+				expect(accepted.status).toBe(200)
+				expect(rejected.status).toBe(409)
+				expect(await rejected.json()).toEqual(await accepted.json())
+				expect(runWorkflow).toHaveBeenCalledTimes(1)
+				expect((await queue.backend.countWorkflowRuns()).pending).toBe(1)
+			} finally {
+				release()
+				await first
+			}
+		}
+	)
+
+	test.each([
+		{ slowName: 'Publish', slowRoute: publish, otherRoute: audit },
+		{ slowName: 'PublishAudit', slowRoute: audit, otherRoute: publish }
+	])(
+		'does not let a slow $slowName body block other admissions',
+		async ({ slowRoute, otherRoute }) => {
+			let releaseBody!: () => void
+			const body = new Promise<object>((resolve) => {
+				releaseBody = () => resolve({})
+			})
+			const request = { json: vi.fn(() => body) } as unknown as Request
+			const slow = Promise.resolve(slowRoute({ ...event(), request }))
+			const other = Promise.resolve(otherRoute(event()))
+			try {
+				await expect.poll(() => runWorkflow.mock.calls.length).toBe(1)
+				expect((await other).status).toBe(200)
+			} finally {
+				releaseBody()
+				await Promise.all([slow, other])
+			}
+			expect((await slow).status).toBe(409)
+			expect(await (await slow).json()).toEqual(await (await other).json())
+			expect(runWorkflow).toHaveBeenCalledTimes(1)
+		}
+	)
+
+	test('releases admission after enqueue failure so a waiting request can succeed', async () => {
+		let release!: () => void
+		let entered!: () => void
+		const hold = new Promise<void>((resolve) => {
+			release = resolve
+		})
+		const started = new Promise<void>((resolve) => {
+			entered = resolve
+		})
+		runWorkflow.mockImplementationOnce(async () => {
+			entered()
+			await hold
+			throw new Error('enqueue failed')
+		})
+
+		const first = Promise.resolve(publish(event()))
+		try {
+			await Promise.race([
+				started,
+				first.then(() => {
+					throw new Error('First request finished before enqueue')
+				})
+			])
+			const other = audit(event({ bisync: true }))
+			release()
+			const [failed, accepted] = await Promise.all([first, other])
+			expect(failed.status).toBe(500)
+			expect(await failed.text()).toContain('enqueue failed')
+			expect(accepted.status).toBe(200)
+			expect(runWorkflow).toHaveBeenCalledTimes(2)
+			expect(runWorkflow).toHaveBeenLastCalledWith(
+				expect.objectContaining({ name: 'PublishAudit' }),
+				{ bisync: true }
+			)
+		} finally {
+			release()
+			await first
+		}
+	})
+
+	test('keeps detecting persisted work when enqueue writes the run before rejecting', async () => {
+		let persistedRunId: string | undefined
+		runWorkflow.mockImplementationOnce(async (spec, input) => {
+			persistedRunId = (await enqueue(spec, input)).workflowRun.id
+			throw new Error('enqueue response failed')
+		})
+		expect((await publish(event())).status).toBe(500)
+		const rejected = await audit(event())
+		expect(rejected.status).toBe(409)
+		expect(await rejected.json()).toEqual({ jobId: persistedRunId })
+		expect(runWorkflow).toHaveBeenCalledTimes(1)
+	})
+
+	test.each([publish, audit])(
+		'uses persisted active runs even without a local submission',
+		async (route) => {
+			const active = await createPendingRun(queue.backend, 'Publish')
+			await claimNextRun(queue.backend)
+			vi.setSystemTime(new Date('2026-06-20T01:00:00Z'))
+			await createCompletedRun(queue.backend, 'Publish')
+			const response = await route(event())
+			expect(response.status).toBe(409)
+			expect(await response.json()).toEqual({ jobId: active.id })
+			expect(runWorkflow).not.toHaveBeenCalled()
+		}
+	)
+
+	test('retains audit freshness checks and releases admission after rejecting an audit', async () => {
+		const oldAudit = await createCompletedRun(queue.backend, 'PublishAudit')
+		vi.setSystemTime(new Date('2026-06-20T01:00:00Z'))
+		const newAudit = await createCompletedRun(queue.backend, 'PublishAudit')
+		const rejected = await publish(event({ auditRunId: oldAudit.id }))
+		expect(rejected.status).toBe(412)
+		expect(await rejected.json()).toEqual({
+			message: 'a newer audit has run; re-check before publishing'
+		})
+		expect(runWorkflow).not.toHaveBeenCalled()
+
+		const accepted = await publish(event({ auditRunId: newAudit.id, bisync: true }))
+		expect(accepted.status).toBe(200)
+		expect(runWorkflow).toHaveBeenCalledWith(expect.objectContaining({ name: 'Publish' }), {
+			bisync: false
+		})
+	})
+
+	test('does not reuse a consumed audit after the admitted publish finishes', async () => {
+		const auditRun = await createCompletedRun(queue.backend, 'PublishAudit')
+		const accepted = await publish(event({ auditRunId: auditRun.id }))
+		expect(accepted.status).toBe(200)
+		const { jobId } = await accepted.json()
+		expect((await completeNextRun(queue.backend)).id).toBe(jobId)
+
+		const rejected = await publish(event({ auditRunId: auditRun.id }))
+		expect(rejected.status).toBe(412)
+		expect(await rejected.json()).toEqual({
+			message: 'changes were published after this check; re-check before publishing'
+		})
+		expect(runWorkflow).toHaveBeenCalledTimes(1)
+	})
+
+	test('retains unchecked publish input and tolerates malformed request JSON', async () => {
+		const accepted = await publish(event({ bisync: true }))
+		expect(accepted.status).toBe(200)
+		expect(runWorkflow).toHaveBeenCalledWith(expect.objectContaining({ name: 'Publish' }), {
+			bisync: true
+		})
+		await completeNextRun(queue.backend)
+		const request = new Request('http://localhost/api/admin/publish/audit', {
+			method: 'POST',
+			body: '{'
+		})
+		expect((await audit({ request } as Parameters<typeof audit>[0])).status).toBe(200)
+		expect(runWorkflow).toHaveBeenLastCalledWith(
+			expect.objectContaining({ name: 'PublishAudit' }),
+			{ bisync: false }
+		)
+	})
+})

apps/web/src/routes/api/admin/publish/audit/+server.ts

@@ -1,22 +1,24 @@
 import { json } from '@sveltejs/kit'
 import type { RequestHandler } from './$types'
-import { getOpenWorkflow, getOpenWorkflowDb } from '$lib/server/workflow/index.js'
-import { findInFlightPublishRun } from '$lib/server/workflow/publish.js'
+import { getOpenWorkflow, getOpenWorkflowBackend } from '$lib/server/workflow/index.js'
+import { findInFlightPublishRun, withPublishAdmission } from '$lib/server/workflow/publish.js'
 import { publishAuditSpec } from '@luzzle/web.jobs/specs'
 
 export const POST: RequestHandler = async ({ request }) => {
 	try {
-		const inFlight = findInFlightPublishRun(getOpenWorkflowDb())
-		if (inFlight) {
-			return json({ jobId: inFlight.id }, { status: 409 })
-		}
-
 		const body = await request.json().catch(() => ({}))
-		const bisync = body?.bisync === true
+		return await withPublishAdmission(async () => {
+			const inFlight = await findInFlightPublishRun(getOpenWorkflowBackend())
+			if (inFlight) {
+				return json({ jobId: inFlight.id }, { status: 409 })
+			}
 
-		const openWorkflow = getOpenWorkflow()
-		const handle = await openWorkflow.runWorkflow(publishAuditSpec, { bisync })
-		return json({ jobId: handle.workflowRun.id })
+			const bisync = body?.bisync === true
+
+			const openWorkflow = getOpenWorkflow()
+			const handle = await openWorkflow.runWorkflow(publishAuditSpec, { bisync })
+			return json({ jobId: handle.workflowRun.id })
+		})
 	} catch (error) {
 		const message = error instanceof Error ? error.message : String(error)
 		return new Response(`Internal server error: ${message}`, { status: 500 })

packages/web/jobs/README.md

@@ -6,7 +6,7 @@ job queue instantiations for Luzzle.
 ## Purpose 🧠
 
 This library defines the schemas, payload interfaces, and results for tasks
-processed by the Sidequest worker queue:
+processed by the OpenWorkflow worker queue:
 
 - **`publish`** — Triggers CDN asset generation, cache purging, and remote cloud
 synchronization.

packages/web/jobs/package.json

@@ -49,7 +49,7 @@
 	},
 	"dependencies": {
 		"@luzzle/core": "workspace:*",
-		"openworkflow": "0.9.0"
+		"openworkflow": "0.9.2"
 	},
 	"type": "module"
 }

packages/web/jobs/src/index.ts

@@ -2,7 +2,6 @@ export {
 	initOpenWorkflow,
 	getOpenWorkflow,
 	getLatestWorkflowRun,
-	getWorkflowRunByJobId,
 	getWorkflowRun,
 	getStepAttempts,
 	purgeExpiredWorkflowRuns,

packages/web/jobs/src/openworkflow.test.ts

@@ -1,10 +1,9 @@
-import { describe, expect, test } from 'vitest'
-import { DatabaseSync } from 'node:sqlite'
+import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'
+import { BackendSqlite } from 'openworkflow/sqlite'
 import {
 	initOpenWorkflow,
 	getOpenWorkflow,
 	getLatestWorkflowRun,
-	getWorkflowRunByJobId,
 	getWorkflowRun,
 	getStepAttempts,
 	jobProgressPurgeSpec,
@@ -21,103 +20,269 @@ describe('openworkflow client initialization', () => {
 		const client = initOpenWorkflow({ dbPath: ':memory:' })
 		expect(client).toBeDefined()
 		expect(getOpenWorkflow()).toBe(client)
+		expect(initOpenWorkflow({ dbPath: ':memory:' })).toBe(client)
 	})
 })
 
-describe('database helpers', () => {
-	test('queries workflow runs and step attempts', () => {
-		const db = new DatabaseSync(':memory:')
+type CreateRunParams = Parameters<BackendSqlite['createWorkflowRun']>[0]
+type RunError = Parameters<BackendSqlite['failWorkflowRun']>[0]['error']
+type StepAttempt = Awaited<ReturnType<BackendSqlite['createStepAttempt']>>
 
-		// Create tables
-		db.exec(`
-			CREATE TABLE workflow_runs (
-				id TEXT PRIMARY KEY,
-				workflow_name TEXT NOT NULL,
-				status TEXT NOT NULL,
-				error TEXT,
-				input TEXT NOT NULL,
-				output TEXT,
-				finished_at TEXT,
-				created_at TEXT NOT NULL
-			)
-		`)
+const workerId = 'read-helper-test'
+const initialTime = new Date('2026-06-02T05:00:00.000Z')
 
-		db.exec(`
-			CREATE TABLE step_attempts (
-				id TEXT PRIMARY KEY,
-				workflow_run_id TEXT NOT NULL,
-				step_name TEXT NOT NULL,
-				status TEXT NOT NULL,
-				started_at TEXT,
-				finished_at TEXT,
-				error TEXT,
-				created_at TEXT NOT NULL
-			)
-		`)
+describe('SDK read helpers', () => {
+	let backend: BackendSqlite
 
-		// 1. Test getLatestWorkflowRun when no runs exist
-		expect(getLatestWorkflowRun(db, 'Publish')).toBeNull()
+	beforeEach(() => {
+		vi.useFakeTimers({ toFake: ['Date'] })
+		vi.setSystemTime(initialTime)
+		backend = BackendSqlite.connect(':memory:')
+	})
 
-		// Insert dummy runs
-		db.prepare(`
-			INSERT INTO workflow_runs (id, workflow_name, status, error, input, output, finished_at, created_at)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-		`).run('run-1', 'Publish', 'completed', null, '{}', 'ok', null, '2026-06-02T05:00:00Z')
+	afterEach(async () => {
+		vi.restoreAllMocks()
+		vi.useRealTimers()
+		await backend.stop()
+	})
 
-		db.prepare(`
-			INSERT INTO workflow_runs (id, workflow_name, status, error, input, output, finished_at, created_at)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-		`).run('run-2', 'Publish', 'failed', 'some error', '{"jobId":123}', null, null, '2026-06-02T05:10:00Z')
+	function createRun(overrides: Partial<CreateRunParams> = {}) {
+		return backend.createWorkflowRun({
+			workflowName: 'Publish',
+			version: null,
+			idempotencyKey: null,
+			config: {},
+			context: null,
+			input: { jobId: 123 },
+			parentStepAttemptNamespaceId: null,
+			parentStepAttemptId: null,
+			availableAt: null,
+			deadlineAt: null,
+			...overrides,
+		})
+	}
 
-		// Test getLatestWorkflowRun returns the latest by created_at DESC
-		const latest = getLatestWorkflowRun(db, 'Publish')
-		expect(latest).not.toBeNull()
-		expect(latest!.id).toBe('run-2')
-		expect(latest!.error).toBe('some error')
+	async function claimRun(id: string) {
+		const claimed = await backend.claimWorkflowRun({ workerId, leaseDurationMs: 60_000 })
+		expect(claimed?.id).toBe(id)
+	}
 
-		// 2. Test getWorkflowRunByJobId
-		const runByJobId = getWorkflowRunByJobId(db, 123)
-		expect(runByJobId).not.toBeNull()
-		expect(runByJobId!.id).toBe('run-2')
+	function createStep(workflowRunId: string, stepName: string) {
+		return backend.createStepAttempt({
+			workflowRunId,
+			workerId,
+			stepName,
+			kind: 'function',
+			config: {},
+			context: null,
+		})
+	}
 
-		const runByNonexistentJobId = getWorkflowRunByJobId(db, 999)
-		expect(runByNonexistentJobId).toBeNull()
+	test('returns null or an empty list for missing runs', async () => {
+		expect(await getLatestWorkflowRun(backend, 'Publish')).toBeNull()
+		expect(await getWorkflowRun(backend, 'missing')).toBeNull()
+		expect(await getStepAttempts(backend, 'missing')).toEqual([])
+	})
 
-		// 3. Test getWorkflowRun
-		const run = getWorkflowRun(db, 'run-1')
-		expect(run).not.toBeNull()
-		expect(run!.id).toBe('run-1')
+	test('returns the newest matching workflow as the existing DTO, not an SDK run', async () => {
+		const first = await createRun()
+		vi.setSystemTime(new Date('2026-06-02T05:10:00.000Z'))
+		const latest = await createRun({ input: { jobId: 456, assets: ['cover'] } })
+		vi.setSystemTime(new Date('2026-06-02T05:20:00.000Z'))
+		await createRun({ workflowName: 'Preview' })
 
-		const nonexistentRun = getWorkflowRun(db, 'nonexistent')
-		expect(nonexistentRun).toBeNull()
+		expect(await getLatestWorkflowRun(backend, 'Publish')).toEqual({
+			id: latest.id,
+			workflow_name: 'Publish',
+			status: 'pending',
+			error: null,
+			input: '{"jobId":456,"assets":["cover"]}',
+			output: null,
+			finished_at: null,
+			created_at: '2026-06-02T05:10:00.000Z',
+		})
+		expect(await getWorkflowRun(backend, latest.id)).toEqual(
+			await getLatestWorkflowRun(backend, 'Publish')
+		)
+		expect((await getWorkflowRun(backend, first.id))?.id).toBe(first.id)
+		expect(await getLatestWorkflowRun(backend, 'Unknown')).toBeNull()
+	})
 
-		// 4. Test getStepAttempts
-		db.prepare(`
-			INSERT INTO step_attempts (id, workflow_run_id, step_name, status, started_at, finished_at, error, created_at)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-		`).run('step-1', 'run-1', 'step-parse', 'completed', '2026-06-02T05:01:00Z', '2026-06-02T05:02:00Z', null, '2026-06-02T05:01:00Z')
+	test.each([
+		{ input: { jobId: 123 }, output: { published: ['cover'], count: 1 } },
+		{ input: 'input string', output: 'ok' },
+		{ input: '', output: '' },
+		{ input: false, output: false },
+		{ input: 0, output: 0 },
+		{ input: null, output: null },
+	])('serializes input/output JSON and completed ISO timestamps: %j', async ({ input, output }) => {
+		const run = await createRun({ input })
+		await claimRun(run.id)
+		vi.setSystemTime(new Date('2026-06-02T05:01:00.000Z'))
+		await backend.completeWorkflowRun({ workflowRunId: run.id, workerId, output })
 
-		db.prepare(`
-			INSERT INTO step_attempts (id, workflow_run_id, step_name, status, started_at, finished_at, error, created_at)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-		`).run('step-2', 'run-1', 'step-transform', 'failed', '2026-06-02T05:03:00Z', '2026-06-02T05:04:00Z', 'transform error', '2026-06-02T05:03:00Z')
-
-		const steps = getStepAttempts(db, 'run-1')
-		expect(steps).toHaveLength(2)
-		expect(steps[0]).toEqual({
-			phase: 'step-parse',
+		expect(await getWorkflowRun(backend, run.id)).toEqual({
+			id: run.id,
+			workflow_name: 'Publish',
 			status: 'completed',
-			started_at: '2026-06-02T05:01:00Z',
-			finished_at: '2026-06-02T05:02:00Z',
-			message: null,
+			error: null,
+			input: JSON.stringify(input),
+			output: output === null ? null : JSON.stringify(output),
+			finished_at: '2026-06-02T05:01:00.000Z',
+			created_at: initialTime.toISOString(),
 		})
-		expect(steps[1]).toEqual({
-			phase: 'step-transform',
+	})
+
+	test.each([
+		{ name: 'PublishError', message: 'upload failed', stack: 'stack', detail: { code: 503 } },
+		'legacy failure',
+		'',
+	])('preserves workflow error JSON encoding: %j', async error => {
+		const run = await createRun()
+		await claimRun(run.id)
+		vi.setSystemTime(new Date('2026-06-02T05:02:00.000Z'))
+		await backend.failWorkflowRun({
+			workflowRunId: run.id,
+			workerId,
+			// The public mutation type requires an object, but SQLite also stores
+			// string JSON errors. Exercise that compatibility through the real API.
+			error: error as unknown as RunError,
+			retryPolicy: {
+				maximumAttempts: 1,
+				initialInterval: '1s',
+				maximumInterval: '1s',
+				backoffCoefficient: 1,
+			},
+		})
+
+		const expected = {
+			id: run.id,
+			workflow_name: 'Publish',
 			status: 'failed',
-			started_at: '2026-06-02T05:03:00Z',
-			finished_at: '2026-06-02T05:04:00Z',
-			message: 'transform error',
+			error: JSON.stringify(error),
+			input: '{"jobId":123}',
+			output: null,
+			finished_at: '2026-06-02T05:02:00.000Z',
+			created_at: initialTime.toISOString(),
+		}
+		expect(await getWorkflowRun(backend, run.id)).toEqual(expected)
+		expect(await getLatestWorkflowRun(backend, 'Publish')).toEqual(expected)
+	})
+
+	test('maps completed and running step attempts, omitting SDK internals', async () => {
+		const run = await createRun()
+		await claimRun(run.id)
+		expect(await getStepAttempts(backend, run.id)).toEqual([])
+		const completed = await createStep(run.id, 'parse')
+		vi.setSystemTime(new Date('2026-06-02T05:01:00.000Z'))
+		await backend.completeStepAttempt({
+			workflowRunId: run.id,
+			stepAttemptId: completed.id,
+			workerId,
+			output: { internalStepResult: true },
 		})
+		await createStep(run.id, 'transform')
+
+		expect(await getStepAttempts(backend, run.id)).toEqual([
+			{
+				phase: 'parse',
+				status: 'completed',
+				started_at: initialTime.toISOString(),
+				finished_at: '2026-06-02T05:01:00.000Z',
+				message: null,
+			},
+			{
+				phase: 'transform',
+				status: 'running',
+				started_at: '2026-06-02T05:01:00.000Z',
+				finished_at: null,
+				message: null,
+			},
+		])
+	})
+
+	test('maps an absent SDK step start time to the existing nullable DTO field', async () => {
+		const run = await createRun()
+		await claimRun(run.id)
+		await createStep(run.id, 'wait')
+		const original = backend.listStepAttempts.bind(backend)
+		vi.spyOn(backend, 'listStepAttempts').mockImplementation(async params => {
+			const page = await original(params)
+			return { ...page, data: page.data.map(attempt => ({ ...attempt, startedAt: null })) }
+		})
+		expect(await getStepAttempts(backend, run.id)).toEqual([{
+			phase: 'wait', status: 'running', started_at: null, finished_at: null, message: null,
+		}])
+	})
+
+	test.each([
+		{ name: 'TransformError', message: 'transform failed', detail: ['cover'] },
+		'legacy step failure',
+		'',
+	])('preserves step error JSON as message: %j', async error => {
+		const run = await createRun()
+		await claimRun(run.id)
+		const step = await createStep(run.id, 'transform')
+		vi.setSystemTime(new Date('2026-06-02T05:03:00.000Z'))
+		await backend.failStepAttempt({
+			workflowRunId: run.id,
+			stepAttemptId: step.id,
+			workerId,
+			error: error as unknown as RunError,
+		})
+
+		expect(await getStepAttempts(backend, run.id)).toEqual([{
+			phase: 'transform',
+			status: 'failed',
+			started_at: initialTime.toISOString(),
+			finished_at: '2026-06-02T05:03:00.000Z',
+			message: JSON.stringify(error),
+		}])
+	})
+
+	test('follows every SDK page in chronological order, including timestamp ties and retries', async () => {
+		const run = await createRun()
+		await claimRun(run.id)
+		const attempts: StepAttempt[] = []
+		for (let i = 0; i < 7; i++) {
+			// Shared timestamps also exercise the SDK cursor ID tie-breaker.
+			vi.setSystemTime(new Date(initialTime.getTime() + Math.floor(i / 3) * 1000))
+			const step = await createStep(run.id, `phase-${i % 3}`)
+			attempts.push(await backend.completeStepAttempt({
+				workflowRunId: run.id,
+				stepAttemptId: step.id,
+				workerId,
+				output: null,
+			}))
+		}
+		const other = await createRun({ workflowName: 'Preview' })
+		await claimRun(other.id)
+		await createStep(other.id, 'not-this-run')
+
+		const original = backend.listStepAttempts.bind(backend)
+		const list = vi.spyOn(backend, 'listStepAttempts')
+			.mockImplementation(params => original({ ...params, limit: 2 }))
+		const rows = await getStepAttempts(backend, run.id)
+		const expected = attempts
+			.sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime() || a.id.localeCompare(b.id))
+			.map(attempt => ({
+				phase: attempt.stepName,
+				status: 'completed',
+				started_at: attempt.startedAt!.toISOString(),
+				finished_at: attempt.finishedAt!.toISOString(),
+				message: null,
+			}))
+
+		expect(rows).toHaveLength(7)
+		expect(rows).toEqual(expected)
+		expect(list).toHaveBeenCalledTimes(4)
+		for (let i = 1; i < list.mock.calls.length; i++) {
+			const previousPage = await list.mock.results[i - 1].value
+			expect(list.mock.calls[i][0]).toEqual({
+				workflowRunId: run.id,
+				after: previousPage.pagination.next,
+			})
+		}
 	})
 })
 

packages/web/jobs/src/openworkflow.ts

@@ -44,64 +44,69 @@ export interface StepAttemptRow {
 	message: string | null
 }
 
-/**
- * Finds the latest workflow run for a given workflow name.
- */
-export function getLatestWorkflowRun(
-	db: DatabaseSync,
-	workflowName: string
-): WorkflowRunRow | null {
-	const stmt = db.prepare(`
-		SELECT id, workflow_name, status, error, input, output, finished_at, created_at FROM workflow_runs
-		WHERE workflow_name = ?
-		ORDER BY created_at DESC
-		LIMIT 1
-	`)
-	const row = stmt.get(workflowName)
-	return row ? (row as unknown as WorkflowRunRow) : null
+type WorkflowRun = NonNullable<Awaited<ReturnType<BackendSqlite['getWorkflowRun']>>>
+
+function toWorkflowRunRow(run: WorkflowRun): WorkflowRunRow {
+	return {
+		id: run.id,
+		workflow_name: run.workflowName,
+		status: run.status,
+		error: run.error === null ? null : JSON.stringify(run.error),
+		// Keep the existing non-null string DTO even for SDK runs with no input.
+		input: JSON.stringify(run.input),
+		output: run.output === null ? null : JSON.stringify(run.output),
+		finished_at: run.finishedAt?.toISOString() ?? null,
+		created_at: run.createdAt.toISOString(),
+	}
 }
 
 /**
- * Finds a workflow run by the jobId nested inside its JSON input.
+ * Finds the latest workflow run for a given workflow name.
  */
-export function getWorkflowRunByJobId(db: DatabaseSync, jobId: number): WorkflowRunRow | null {
-	const stmt = db.prepare(`
-		SELECT id, workflow_name, status, error, input, output, finished_at, created_at FROM workflow_runs
-		WHERE json_extract(input, '$.jobId') = ?
-		LIMIT 1
-	`)
-	const row = stmt.get(jobId)
-	return row ? (row as unknown as WorkflowRunRow) : null
+export async function getLatestWorkflowRun(
+	backend: Pick<BackendSqlite, 'listWorkflowRuns'>,
+	workflowName: string
+): Promise<WorkflowRunRow | null> {
+	// The SDK lists workflow runs newest first.
+	const { data } = await backend.listWorkflowRuns({ workflowName, limit: 1 })
+	return data[0] ? toWorkflowRunRow(data[0]) : null
 }
 
 /**
  * Finds a workflow run by its unique ID.
  */
-export function getWorkflowRun(
-	db: DatabaseSync,
-	id: string,
-): WorkflowRunRow | null {
-	const stmt = db.prepare(`
-		SELECT id, workflow_name, status, error, input, output, finished_at, created_at FROM workflow_runs
-		WHERE id = ?
-		LIMIT 1
-	`);
-	const row = stmt.get(id);
-	return row ? (row as unknown as WorkflowRunRow) : null;
+export async function getWorkflowRun(
+	backend: Pick<BackendSqlite, 'getWorkflowRun'>,
+	id: string
+): Promise<WorkflowRunRow | null> {
+	const run = await backend.getWorkflowRun({ workflowRunId: id })
+	return run ? toWorkflowRunRow(run) : null
 }
 
 /**
- * Lists all step attempts for a given workflow run ID.
+ * Lists all step attempts for a given workflow run ID, oldest first.
  */
-export function getStepAttempts(db: DatabaseSync, workflowRunId: string): StepAttemptRow[] {
-	const stmt = db.prepare(`
-		SELECT step_name as phase, status, started_at, finished_at, error as message
-		FROM step_attempts
-		WHERE workflow_run_id = ?
-		ORDER BY created_at ASC
-	`)
-	const rows = stmt.all(workflowRunId)
-	return rows as unknown as StepAttemptRow[]
+export async function getStepAttempts(
+	backend: Pick<BackendSqlite, 'listStepAttempts'>,
+	workflowRunId: string
+): Promise<StepAttemptRow[]> {
+	const rows: StepAttemptRow[] = []
+	let after: string | undefined
+
+	do {
+		// Forward SDK pagination preserves chronological order across pages.
+		const page = await backend.listStepAttempts({ workflowRunId, after })
+		rows.push(...page.data.map(attempt => ({
+			phase: attempt.stepName,
+			status: attempt.status,
+			started_at: attempt.startedAt?.toISOString() ?? null,
+			finished_at: attempt.finishedAt?.toISOString() ?? null,
+			message: attempt.error === null ? null : JSON.stringify(attempt.error),
+		})))
+		after = page.pagination.next ?? undefined
+	} while (after)
+
+	return rows
 }
 
 /**

packages/web/worker/package.json

@@ -70,7 +70,7 @@
 		"micromark-util-symbol": "2.0.1",
 		"micromark-util-types": "2.0.2",
 		"mime-types": "3.0.2",
-		"openworkflow": "0.9.0",
+		"openworkflow": "0.9.2",
 		"rehype-autolink-headings": "7.1.0",
 		"rehype-pretty-code": "0.14.1",
 		"rehype-raw": "7.0.0",

pnpm-lock.yaml

@@ -282,8 +282,8 @@ importers:
         specifier: 0.28.17
         version: 0.28.17
       openworkflow:
-        specifier: 0.9.0
-        version: 0.9.0
+        specifier: 0.9.2
+        version: 0.9.2
     devDependencies:
       '@auth/core':
         specifier: 0.41.1
@@ -546,8 +546,8 @@ importers:
         specifier: workspace:*
         version: link:../../../apps/core
       openworkflow:
-        specifier: 0.9.0
-        version: 0.9.0
+        specifier: 0.9.2
+        version: 0.9.2
     devDependencies:
       '@eslint/js':
         specifier: 'catalog:'
@@ -816,8 +816,8 @@ importers:
         specifier: 3.0.2
         version: 3.0.2
       openworkflow:
-        specifier: 0.9.0
-        version: 0.9.0
+        specifier: 0.9.2
+        version: 0.9.2
       rehype-autolink-headings:
         specifier: 7.1.0
         version: 7.1.0
@@ -3535,8 +3535,8 @@ packages:
   oniguruma-to-es@4.3.6:
     resolution: {integrity: sha512-csuQ9x3Yr0cEIs/Zgx/OEt9iBw9vqIunAPQkx19R/fiMq2oGVTgcMqO/V3Ybqefr1TBvosI6jU539ksaBULJyA==}
 
-  openworkflow@0.9.0:
-    resolution: {integrity: sha512-Ii6qkAh4ozbTHH5w7PdjjNYGF7u8xV2rsvTaBleo21rRlS8sEzrvDdxWsHawiY1VLfI1nfuRf8I3NDerlrugtg==}
+  openworkflow@0.9.2:
+    resolution: {integrity: sha512-NluYomuVLf+rAMJ9ASKqw8nbPAAD+C/HSTZoJacZdelLfTpndzHvPcP8SCANg9me9zek6mao0uUOGGbrWUWhGw==}
     engines: {node: '>=20'}
     peerDependencies:
       postgres: ^3.4.9
@@ -5939,7 +5939,7 @@ snapshots:
       std-env: 3.10.0
       test-exclude: 7.0.2
       tinyrainbow: 2.0.0
-      vitest: 3.2.4(@types/debug@4.1.13)(@types/node@24.12.4)(@vitest/ui@3.2.4)(yaml@2.7.0)
+      vitest: 3.2.4(@types/debug@4.1.13)(@types/node@24.12.4)(@vitest/ui@3.2.4)(yaml@2.9.0)
     transitivePeerDependencies:
       - supports-color
 
@@ -7464,7 +7464,7 @@ snapshots:
       regex: 6.1.0
       regex-recursion: 6.0.2
 
-  openworkflow@0.9.0: {}
+  openworkflow@0.9.2: {}
 
   optionator@0.9.4:
     dependencies: