.changeset/openworkflow-upgrade.md

@@ -0,0 +1,8 @@
+---
+"@luzzle/web": patch
+"@luzzle/web.worker": patch
+---
+
+Upgrade OpenWorkflow to 0.11.0 across web, worker and shared jobs. Use its native
+logger option and close worker-owned resources in order. Stop deleting engine job
+history; preview files and application logs remain application-owned cleanup.

apps/web/package.json

@@ -90,6 +90,6 @@
     "eventsource-parser": "4.1.1",
     "kysely": "0.28.17",
     "micromark": "4.0.2",
-    "openworkflow": "0.9.2"
+    "openworkflow": "0.11.0"
   }
 }

packages/web/jobs/package.json

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

packages/web/jobs/src/index.ts

@@ -1,12 +1,13 @@
 export {
 	initOpenWorkflow,
 	getOpenWorkflow,
+	getOpenWorkflowBackend,
+	closeOpenWorkflow,
 	getLatestWorkflowRun,
 	getWorkflowRun,
 	getStepAttempts,
-	purgeExpiredWorkflowRuns,
 } from './openworkflow.js'
-export type { WorkflowRunRow, StepAttemptRow } from './openworkflow.js'
+export type { InitOpenWorkflowOptions, WorkflowRunRow, StepAttemptRow } from './openworkflow.js'
 export {
 	jobProgressPurgeSpec,
 	previewSpec,

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

@@ -1,8 +1,8 @@
 import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'
 import { BackendSqlite } from 'openworkflow/sqlite'
+import type { Logger } from 'openworkflow'
+import type * as WorkflowClient from './openworkflow.js'
 import {
-	initOpenWorkflow,
-	getOpenWorkflow,
 	getLatestWorkflowRun,
 	getWorkflowRun,
 	getStepAttempts,
@@ -12,15 +12,88 @@ import {
 } from './index.js'
 
 describe('openworkflow client initialization', () => {
-	test('throws before initialization', () => {
-		expect(() => getOpenWorkflow()).toThrow(/has not been initialized/)
+	let lifecycle: typeof WorkflowClient
+
+	beforeEach(async () => {
+		vi.resetModules()
+		lifecycle = await import('./openworkflow.js')
 	})
 
-	test('initializes and returns singleton instance', () => {
-		const client = initOpenWorkflow({ dbPath: ':memory:' })
-		expect(client).toBeDefined()
-		expect(getOpenWorkflow()).toBe(client)
-		expect(initOpenWorkflow({ dbPath: ':memory:' })).toBe(client)
+	afterEach(async () => {
+		await lifecycle.closeOpenWorkflow().catch(() => {})
+		vi.restoreAllMocks()
+	})
+
+	test('throws before initialization', () => {
+		expect(() => lifecycle.getOpenWorkflow()).toThrow(/has not been initialized/)
+		expect(() => lifecycle.getOpenWorkflowBackend()).toThrow(/has not been initialized/)
+	})
+
+	test('owns one backend in the SDK default namespace and returns the singleton', async () => {
+		const client = lifecycle.initOpenWorkflow({ dbPath: ':memory:' })
+		const backend = lifecycle.getOpenWorkflowBackend()
+		expect(lifecycle.getOpenWorkflow()).toBe(client)
+		expect(lifecycle.initOpenWorkflow({ dbPath: 'not-opened.sqlite' })).toBe(client)
+		expect(lifecycle.getOpenWorkflowBackend()).toBe(backend)
+		const handle = await client.runWorkflow(publishSpec, {})
+		expect(handle.workflowRun.namespaceId).toBe('default')
+		expect((await backend.getWorkflowRun({ workflowRunId: handle.workflowRun.id }))?.id).toBe(
+			handle.workflowRun.id
+		)
+	})
+
+	test('passes the native logger to workers without adapting its arguments', async () => {
+		const logger: Logger = { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }
+		const client = lifecycle.initOpenWorkflow({ dbPath: ':memory:', logger })
+		const backend = lifecycle.getOpenWorkflowBackend()
+		const failure = new Error('poll failed')
+		vi.spyOn(backend, 'claimWorkflowRun').mockRejectedValue(failure)
+		const worker = client.newWorker({ concurrency: 1 })
+		try {
+			await worker.start()
+			await vi.waitFor(() => {
+				expect(logger.error).toHaveBeenCalledWith(expect.any(String), failure)
+			})
+		} finally {
+			await worker.stop()
+		}
+	})
+
+	test('shares one terminal close and never reopens the backend', async () => {
+		lifecycle.initOpenWorkflow({ dbPath: ':memory:' })
+		const backend = lifecycle.getOpenWorkflowBackend()
+		const stop = vi.spyOn(backend, 'stop')
+		const closing = lifecycle.closeOpenWorkflow()
+		expect(lifecycle.closeOpenWorkflow()).toBe(closing)
+		expect(() => lifecycle.getOpenWorkflow()).toThrow(/closed/)
+		expect(() => lifecycle.getOpenWorkflowBackend()).toThrow(/closed/)
+		expect(() => lifecycle.initOpenWorkflow({ dbPath: ':memory:' })).toThrow(/closed/)
+		await closing
+		expect(stop).toHaveBeenCalledTimes(1)
+		expect(lifecycle.closeOpenWorkflow()).toBe(closing)
+	})
+
+	test('preserves a close failure without retrying or reporting success', async () => {
+		lifecycle.initOpenWorkflow({ dbPath: ':memory:' })
+		const backend = lifecycle.getOpenWorkflowBackend()
+		const actualStop = backend.stop.bind(backend)
+		const failure = new Error('close failed')
+		const stop = vi.spyOn(backend, 'stop').mockRejectedValue(failure)
+		try {
+			const closing = lifecycle.closeOpenWorkflow()
+			await expect(closing).rejects.toBe(failure)
+			expect(lifecycle.closeOpenWorkflow()).toBe(closing)
+			expect(stop).toHaveBeenCalledTimes(1)
+			expect(() => lifecycle.initOpenWorkflow({ dbPath: ':memory:' })).toThrow(/closed/)
+		} finally {
+			await actualStop()
+		}
+	})
+
+	test('closing before initialization is harmless', async () => {
+		await lifecycle.closeOpenWorkflow()
+		expect(() => lifecycle.getOpenWorkflow()).toThrow(/has not been initialized/)
+		expect(lifecycle.initOpenWorkflow({ dbPath: ':memory:' })).toBeDefined()
 	})
 })
 
@@ -137,7 +210,7 @@ describe('SDK read helpers', () => {
 		{ name: 'PublishError', message: 'upload failed', stack: 'stack', detail: { code: 503 } },
 		'legacy failure',
 		'',
-	])('preserves workflow error JSON encoding: %j', async error => {
+	])('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'))
@@ -206,20 +279,26 @@ describe('SDK read helpers', () => {
 		await claimRun(run.id)
 		await createStep(run.id, 'wait')
 		const original = backend.listStepAttempts.bind(backend)
-		vi.spyOn(backend, 'listStepAttempts').mockImplementation(async params => {
+		vi.spyOn(backend, 'listStepAttempts').mockImplementation(async (params) => {
 			const page = await original(params)
-			return { ...page, data: page.data.map(attempt => ({ ...attempt, startedAt: null })) }
+			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,
-		}])
+		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 => {
+	])('preserves step error JSON as message: %j', async (error) => {
 		const run = await createRun()
 		await claimRun(run.id)
 		const step = await createStep(run.id, 'transform')
@@ -231,13 +310,15 @@ describe('SDK read helpers', () => {
 			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),
-		}])
+		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 () => {
@@ -248,24 +329,27 @@ describe('SDK read helpers', () => {
 			// 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,
-			}))
+			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 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 => ({
+			.map((attempt) => ({
 				phase: attempt.stepName,
 				status: 'completed',
 				started_at: attempt.startedAt!.toISOString(),

packages/web/jobs/src/openworkflow.ts

@@ -1,22 +1,45 @@
 import { OpenWorkflow } from 'openworkflow'
+import type { Logger } from 'openworkflow'
 import { BackendSqlite } from 'openworkflow/sqlite'
-import type { DatabaseSync } from 'node:sqlite'
 
 let openWorkflowInstance: OpenWorkflow | null = null
+let openWorkflowBackend: BackendSqlite | null = null
+let closePromise: Promise<void> | null = null
 
 export interface InitOpenWorkflowOptions {
 	dbPath: string
+	logger?: Logger
+}
+
+function assertOpen(): void {
+	if (closePromise) {
+		throw new Error('OpenWorkflow has been closed and cannot be reinitialized.')
+	}
 }
 
 export function initOpenWorkflow(opts: InitOpenWorkflowOptions): OpenWorkflow {
+	assertOpen()
 	if (!openWorkflowInstance) {
-		const backend = BackendSqlite.connect(opts.dbPath)
-		openWorkflowInstance = new OpenWorkflow({ backend })
+		openWorkflowBackend = BackendSqlite.connect(opts.dbPath)
+		openWorkflowInstance = new OpenWorkflow({ backend: openWorkflowBackend, logger: opts.logger })
 	}
 	return openWorkflowInstance
 }
 
+/** The caller must stop workers before closing their shared backend. */
+export function closeOpenWorkflow(): Promise<void> {
+	if (!openWorkflowBackend) return Promise.resolve()
+	closePromise ??= openWorkflowBackend.stop()
+	return closePromise
+}
+
+export function getOpenWorkflowBackend(): BackendSqlite {
+	getOpenWorkflow()
+	return openWorkflowBackend!
+}
+
 export function getOpenWorkflow(): OpenWorkflow {
+	assertOpen()
 	if (!openWorkflowInstance) {
 		throw new Error(
 			'OpenWorkflow client has not been initialized. Call initOpenWorkflow({ dbPath }) first.'
@@ -96,58 +119,17 @@ export async function getStepAttempts(
 	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),
-		})))
+		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
 }
-
-/**
- * Purges old workflow runs and step attempts from the database in batches of 100.
- * Note: OpenWorkflow will add cron and cleanup in a future release,
- * so we will need to make adjustments later if/when they do.
- */
-export function purgeExpiredWorkflowRuns(
-	db: DatabaseSync,
-	retentionDays: number,
-	limit: number = 100
-): number {
-	const cutoffIso = new Date(Date.now() - retentionDays * 24 * 60 * 60 * 1000).toISOString()
-	let totalPurged = 0
-	let batch: { id: string }[] = []
-
-	do {
-		const stmtSelect = db.prepare(`
-			SELECT id FROM workflow_runs 
-			WHERE created_at < ? 
-			LIMIT ?
-		`)
-		batch = stmtSelect.all(cutoffIso, limit) as { id: string }[]
-
-		if (batch.length > 0) {
-			const ids = batch.map(r => r.id)
-			const placeholders = ids.map(() => '?').join(',')
-
-			db.prepare(`
-				DELETE FROM step_attempts 
-				WHERE workflow_run_id IN (${placeholders})
-			`).run(...ids)
-
-			db.prepare(`
-				DELETE FROM workflow_runs 
-				WHERE id IN (${placeholders})
-			`).run(...ids)
-
-			totalPurged += batch.length
-		}
-	} while (batch.length === limit)
-
-	return totalPurged
-}

packages/web/worker/package.json

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

packages/web/worker/src/index.test.ts

@@ -0,0 +1,117 @@
+import { afterEach, expect, test, vi } from 'vitest'
+import { promises as fs } from 'node:fs'
+import { tmpdir } from 'node:os'
+import path from 'node:path'
+import { createConnection, type AddressInfo, type Socket } from 'node:net'
+import { once } from 'node:events'
+import { defineWorkflowSpec } from 'openworkflow'
+import { BackendSqlite } from 'openworkflow/sqlite'
+import { validateConfig } from '@luzzle/web.config'
+import { configSchema } from '@luzzle/web.templates/config'
+import type { main as startWorker } from './index.js'
+import type * as Workflows from './workflows/index.js'
+
+const probe = defineWorkflowSpec<void, string>({ name: 'ShutdownProbe' })
+
+afterEach(() => {
+	vi.doUnmock('./workflows/index.js')
+	vi.resetModules()
+})
+
+test('starts the native worker and health server, then lets active work finish before closing databases', async () => {
+	vi.resetModules()
+	const directory = await fs.mkdtemp(path.join(tmpdir(), 'luzzle-worker-'))
+	const queuePath = path.join(directory, 'queue.sqlite')
+	const config = validateConfig(
+		{
+			storage: { root: directory },
+			paths: {
+				config: path.join(directory, 'config.yaml'),
+				database: path.join(directory, 'app.sqlite'),
+				assets: path.join(directory, 'assets/pieces'),
+			},
+			worker: { queue: { path: queuePath } },
+		},
+		{},
+		{ schema: configSchema }
+	)
+	const jobs = await import('@luzzle/web.jobs')
+	let release!: () => void
+	const gate = new Promise<void>((resolve) => {
+		release = resolve
+	})
+	let entered = false
+	vi.doMock('./workflows/index.js', async () => {
+		const actual = await vi.importActual<typeof Workflows>('./workflows/index.js')
+		return {
+			registerWorkflows() {
+				actual.registerWorkflows()
+				jobs.getOpenWorkflow().implementWorkflow(probe, async ({ step }) =>
+					step.run({ name: 'active-work' }, async () => {
+						entered = true
+						await gate
+						return 'finished'
+					})
+				)
+			},
+		}
+	})
+	let running: Awaited<ReturnType<typeof startWorker>> | undefined
+	let stalledHealth: Socket | undefined
+	try {
+		const { main } = await import('./index.js')
+		running = await main(config, 0)
+		const { port } = running.server.address() as AddressInfo
+		expect((await fetch(`http://127.0.0.1:${port}/health`)).status).toBe(200)
+		const backend = jobs.getOpenWorkflowBackend()
+		await expect
+			.poll(
+				async () =>
+					(await backend.listWorkflowRuns({ workflowName: 'JobProgressPurge' })).data[0]?.status
+			)
+			.toBe('completed')
+		const handle = await jobs.getOpenWorkflow().runWorkflow(probe)
+		await expect.poll(() => entered).toBe(true)
+
+		const accepted = once(running.server, 'connection')
+		stalledHealth = createConnection(port, '127.0.0.1')
+		stalledHealth.on('error', () => {})
+		await once(stalledHealth, 'connect')
+		await accepted
+		stalledHealth.write('GET /health HTTP/1.1\r\nHost: localhost\r\n')
+		let healthConnectionClosed = false
+		stalledHealth.once('close', () => {
+			healthConnectionClosed = true
+		})
+
+		const stopping = running.stop()
+		expect(running.stop()).toBe(stopping)
+		await expect.poll(() => running!.server.listening).toBe(false)
+		await expect.poll(() => healthConnectionClosed).toBe(true)
+		const { getWorkerContext } = await import('./services/context.js')
+		expect(await getWorkerContext().db.selectFrom('job_progress').selectAll().execute()).toEqual([])
+		expect(await backend.getWorkflowRun({ workflowRunId: handle.workflowRun.id })).toMatchObject({
+			status: 'running',
+		})
+
+		release()
+		await stopping
+		expect(() => jobs.getOpenWorkflow()).toThrow(/closed/)
+		const reopened = BackendSqlite.connect(queuePath)
+		try {
+			expect(await reopened.getWorkflowRun({ workflowRunId: handle.workflowRun.id })).toMatchObject(
+				{ status: 'completed', output: 'finished' }
+			)
+			expect(
+				(await reopened.listWorkflowRuns({ workflowName: 'JobProgressPurge' })).data
+			).toHaveLength(1)
+		} finally {
+			await reopened.stop()
+		}
+	} finally {
+		release()
+		stalledHealth?.destroy()
+		await running?.stop()
+		await fs.rm(directory, { recursive: true, force: true })
+	}
+}, 15_000)

packages/web/worker/src/index.ts

@@ -1,83 +1,131 @@
-import { getDatabaseClient, migrate } from '@luzzle/core'
-import { loadConfig } from '@luzzle/web.templates/config'
-import { resolveDbPath, resolveOpenWorkflowDbPath } from './services/db.js'
+import { asCoreDatabase, migrate } from '@luzzle/core'
+import { loadConfig, type TemplateConfig } from '@luzzle/web.templates/config'
+import { createAppDb, resolveDbPath, resolveOpenWorkflowDbPath } from './services/db.js'
 import { runWebMigrations } from '@luzzle/web.db'
 import { createHealthServer } from './services/health.js'
-import { log } from './services/logger.js'
-import { initOpenWorkflow } from '@luzzle/web.jobs'
+import { createLogger, createOpenWorkflowLogger, log } from './services/logger.js'
+import { createWorkerContext, setWorkerContext } from './services/context.js'
+import { closeOpenWorkflow, initOpenWorkflow } from '@luzzle/web.jobs'
 import { jobProgressPurgeSpec } from '@luzzle/web.jobs/specs'
 import { registerWorkflows } from './workflows/index.js'
+import { once } from 'node:events'
+import type { Server } from 'node:http'
+import type { Worker } from 'openworkflow'
 
 const DEFAULT_PORT = 9000
 const PURGE_RETENTION_DAYS = 7
 
-async function main() {
-	const config = loadConfig('./config.yaml')
-
-	const db = getDatabaseClient(resolveDbPath(config))
-
-	const coreResult = await migrate(db)
-	if (coreResult.error) {
-		throw new Error(`luzzle core migration failed: ${coreResult.error}`)
-	}
-	log('info', 'luzzle core migrations applied')
-
-	const webResult = await runWebMigrations(db)
-	if (webResult.error) {
-		throw new Error(`web migration failed: ${webResult.error}`)
-	}
-	log('info', 'web migrations applied')
-
-	const openWorkflowDb = resolveOpenWorkflowDbPath(config)
-	const openWorkflow = initOpenWorkflow({ dbPath: openWorkflowDb })
-	log('info', 'openworkflow client configured', { openWorkflowDb })
-
-	registerWorkflows()
-
-	const openWorkflowWorker = openWorkflow.newWorker()
-	await openWorkflowWorker.start()
-	log('info', 'openworkflow worker started')
-
-	const runPurge = async () => {
-		try {
-			log('info', 'triggering openworkflow job progress purge...')
-			await openWorkflow.runWorkflow(jobProgressPurgeSpec, { retentionDays: PURGE_RETENTION_DAYS })
-			log('info', 'openworkflow job progress purge run completed successfully')
-		} catch (err) {
-			log('error', 'openworkflow job progress purge run failed', {
-				error: err instanceof Error ? err.message : String(err),
-			})
-		}
-	}
-
-	// Trigger immediately on start (non-blocking)
-	runPurge()
-
-	// Schedule to run every 24 hours
-	const purgeInterval = setInterval(runPurge, 24 * 60 * 60 * 1000)
-
-	const shutdown = async () => {
-		log('info', 'shutting down openworkflow worker...')
-		clearInterval(purgeInterval)
-		await openWorkflowWorker.stop()
-		log('info', 'openworkflow worker stopped')
-		process.exit(0)
-	}
-	process.on('SIGTERM', shutdown)
-	process.on('SIGINT', shutdown)
-
-	const port = Number(process.env.PORT) || DEFAULT_PORT
-	const server = createHealthServer()
-	server.listen(port, () => {
-		log('info', 'health server listening', { port })
+async function closeHealthServer(server: Server): Promise<void> {
+	if (!server.listening) return
+	await new Promise<void>((resolve, reject) => {
+		server.close((error) => {
+			if (error) reject(error)
+			else resolve()
+		})
+		server.closeAllConnections()
 	})
 }
 
+export async function main(
+	config: TemplateConfig = loadConfig('./config.yaml'),
+	port: number = Number(process.env.PORT) || DEFAULT_PORT
+) {
+	const logger = createLogger()
+	const db = createAppDb(resolveDbPath(config))
+	const server = createHealthServer()
+	let worker: Worker | undefined
+	let purgeInterval: ReturnType<typeof setInterval> | undefined
+	let stopPromise: Promise<void> | undefined
+
+	async function stopResources(): Promise<void> {
+		clearInterval(purgeInterval)
+		await closeHealthServer(server)
+		await worker?.stop()
+		const errors: unknown[] = []
+		try {
+			await closeOpenWorkflow()
+		} catch (error) {
+			errors.push(error)
+		}
+		try {
+			await db.destroy()
+		} catch (error) {
+			errors.push(error)
+		}
+		if (errors.length) throw new AggregateError(errors, 'Worker database cleanup failed')
+		logger.info('openworkflow worker stopped')
+	}
+
+	function stop(): Promise<void> {
+		stopPromise ??= stopResources()
+		return stopPromise
+	}
+
+	try {
+		const coreResult = await migrate(asCoreDatabase(db))
+		if (coreResult.error)
+			throw new Error('Luzzle core migration failed', { cause: coreResult.error })
+		logger.info('luzzle core migrations applied')
+		const webResult = await runWebMigrations(db)
+		if (webResult.error) throw new Error('Web migration failed', { cause: webResult.error })
+		logger.info('web migrations applied')
+
+		setWorkerContext(createWorkerContext(config, db, logger))
+		const openWorkflowDb = resolveOpenWorkflowDbPath(config)
+		const openWorkflow = initOpenWorkflow({
+			dbPath: openWorkflowDb,
+			logger: createOpenWorkflowLogger(logger),
+		})
+		logger.info('openworkflow client configured', { openWorkflowDb })
+		registerWorkflows()
+		worker = openWorkflow.newWorker()
+		await worker.start()
+		logger.info('openworkflow worker started')
+
+		async function enqueuePurge(): Promise<void> {
+			try {
+				const run = await openWorkflow.runWorkflow(jobProgressPurgeSpec, {
+					retentionDays: PURGE_RETENTION_DAYS,
+				})
+				logger.info('openworkflow job progress purge enqueued', { jobId: run.workflowRun.id })
+			} catch {
+				logger.error('failed to enqueue openworkflow job progress purge')
+			}
+		}
+
+		void enqueuePurge()
+		purgeInterval = setInterval(enqueuePurge, 24 * 60 * 60 * 1000)
+		const listening = once(server, 'listening')
+		server.listen(port)
+		await listening
+		logger.info('health server listening', { port })
+		return { server, stop }
+	} catch (error) {
+		try {
+			await stop()
+		} catch (cleanupError) {
+			throw new AggregateError([error, cleanupError], 'Worker startup and cleanup failed')
+		}
+		throw error
+	}
+}
+
 if (import.meta.url === `file://${process.argv[1]}`) {
-	main().catch((err) => {
-		log('error', 'worker failed to start', {
-			error: err instanceof Error ? err.message : String(err),
+	main()
+		.then(({ stop }) => {
+			const shutdown = () => {
+				void stop().catch(() => {
+					log('error', 'worker shutdown failed; resources may still be active')
+					process.exitCode = 1
+				})
+			}
+			process.once('SIGTERM', shutdown)
+			process.once('SIGINT', shutdown)
+		})
+		.catch((error) => {
+			log('error', 'worker failed to start', {
+				error: error instanceof Error ? error.message : String(error),
+			})
+			process.exitCode = 1
 		})
-		process.exit(1)
-	})
 }

packages/web/worker/src/services/context.test.ts

@@ -1,7 +1,7 @@
 import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'
 import { EventEmitter } from 'node:events'
 import { spawn } from 'child_process'
-import { createPhaseContext, type WorkerContext } from './context.js'
+import { createPhaseContext, createWorkerContext, type WorkerContext } from './context.js'
 import { PhaseLogger } from '../core/phase-logger.js'
 import type { AppDatabase } from './db.js'
 import { RcloneClient } from './rclone.js'
@@ -38,6 +38,14 @@ describe('createPhaseContext', () => {
 		await teardownDatabase(ctx.db)
 	})
 
+	test('creates a context around the application database supplied by startup', () => {
+		const created = createWorkerContext(ctx.config, ctx.db, baseLogger)
+		expect(created.db).toBe(ctx.db)
+		expect(created.config).toBe(ctx.config)
+		expect(created.logger).toBeInstanceOf(PhaseLogger)
+		expect(created.rclone).toBeInstanceOf(RcloneClient)
+	})
+
 	test('leaves contexts with a plain logger unchanged', () => {
 		const plain = { ...ctx, logger: baseLogger }
 		expect(createPhaseContext(plain, 'job1', 'sync')).toBe(plain)

packages/web/worker/src/services/context.ts

@@ -24,6 +24,15 @@ export function createPhaseContext(
 	return { ...ctx, logger, rclone: ctx.rclone.withLogger(logger) }
 }
 
+export function createWorkerContext(
+	config: TemplateConfig,
+	db: Kysely<AppDatabase>,
+	baseLogger: Logger = createLogger()
+): WorkerContext {
+	const logger = new PhaseLogger(baseLogger, db)
+	return { config, db, logger, rclone: new RcloneClient(logger) }
+}
+
 let workerContext: WorkerContext | null = null
 
 export function setWorkerContext(ctx: WorkerContext): void {
@@ -33,14 +42,6 @@ export function setWorkerContext(ctx: WorkerContext): void {
 export function getWorkerContext(): WorkerContext {
 	if (workerContext) return workerContext
 	const config = loadConfig('./config.yaml')
-	const baseLogger: Logger = createLogger()
-	const db = createAppDb(resolveDbPath(config))
-	const logger = new PhaseLogger(baseLogger, db)
-	workerContext = {
-		config,
-		logger,
-		rclone: new RcloneClient(logger),
-		db,
-	}
+	workerContext = createWorkerContext(config, createAppDb(resolveDbPath(config)))
 	return workerContext
 }

packages/web/worker/src/services/logger.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, test, vi, afterEach } from 'vitest'
-import { log, createLogger } from './logger.js'
+import { log, createLogger, createOpenWorkflowLogger } from './logger.js'
 
 describe('log', () => {
 	afterEach(() => {
@@ -14,7 +14,7 @@ describe('log', () => {
 		})
 		return {
 			getLine: () => line,
-			spy
+			spy,
 		}
 	}
 
@@ -54,6 +54,52 @@ describe('log', () => {
 	})
 })
 
+describe('createOpenWorkflowLogger', () => {
+	function logger() {
+		return {
+			debug: vi.fn(),
+			info: vi.fn(),
+			warn: vi.fn(),
+			error: vi.fn(),
+			stdout: vi.fn(),
+			stderr: vi.fn(),
+		}
+	}
+
+	test('routes native levels without forwarding engine messages or arguments', () => {
+		const base = logger()
+		const engine = createOpenWorkflowLogger(base)
+		for (const level of ['debug', 'info', 'warn', 'error'] as const) {
+			engine[level]('private payload', new Error('private credential'), { input: 'secret' })
+			expect(base[level]).toHaveBeenCalledWith('OpenWorkflow operational log')
+		}
+	})
+
+	test('keeps known operational categories without error details', () => {
+		const base = logger()
+		const engine = createOpenWorkflowLogger(base)
+		engine.error('Worker tick failed:', new Error('private path'))
+		engine.warn('Worker schedule tick failed:', { secret: 'private' })
+		engine.error('Heartbeat failed:', 'private error')
+		engine.error('Critical error during workflow execution for run private-value:', 'private error')
+		expect(base.error.mock.calls).toEqual([
+			['OpenWorkflow worker tick failed'],
+			['OpenWorkflow heartbeat failed'],
+			['OpenWorkflow execution failed'],
+		])
+		expect(base.warn).toHaveBeenCalledWith('OpenWorkflow schedule tick failed')
+	})
+
+	test('accepts console-style calls with no message or non-text values', () => {
+		const base = logger()
+		const engine = createOpenWorkflowLogger(base)
+		engine.error()
+		engine.info({ secret: 'private' })
+		expect(base.error).toHaveBeenCalledWith('OpenWorkflow operational log')
+		expect(base.info).toHaveBeenCalledWith('OpenWorkflow operational log')
+	})
+})
+
 describe('createLogger', () => {
 	afterEach(() => {
 		vi.restoreAllMocks()

packages/web/worker/src/services/logger.ts

@@ -1,3 +1,5 @@
+import type { Logger as OpenWorkflowLogger } from 'openworkflow'
+
 export type LogLevel = 'debug' | 'info' | 'warn' | 'error' | 'stdout' | 'stderr'
 
 export interface Logger {
@@ -9,27 +11,58 @@ export interface Logger {
 	stderr(message: string, fields?: Record<string, unknown>): void
 }
 
-export function log(
-	level: LogLevel,
-	message: string,
-	fields: Record<string, unknown> = {}
-): void {
+export function log(level: LogLevel, message: string, fields: Record<string, unknown> = {}): void {
 	const line = JSON.stringify({
 		ts: new Date().toISOString(),
 		level,
 		message,
-		...fields
+		...fields,
 	})
 	process.stdout.write(line + '\n')
 }
 
-export function createLogger(): Logger {
+const engineMessages = new Map([
+	['Worker tick failed:', 'OpenWorkflow worker tick failed'],
+	['Worker schedule tick failed:', 'OpenWorkflow schedule tick failed'],
+	['Heartbeat failed:', 'OpenWorkflow heartbeat failed'],
+])
+
+function openWorkflowMessage(message: unknown): string {
+	if (typeof message !== 'string') return 'OpenWorkflow operational log'
+	if (message.startsWith('Critical error during workflow execution for run ')) {
+		return 'OpenWorkflow execution failed'
+	}
+	return engineMessages.get(message) ?? 'OpenWorkflow operational log'
+}
+
+export function createOpenWorkflowLogger(logger: Logger): OpenWorkflowLogger {
 	return {
-		debug(message, fields) { log('debug', message, fields) },
-		info(message, fields) { log('info', message, fields) },
-		warn(message, fields) { log('warn', message, fields) },
-		error(message, fields) { log('error', message, fields) },
-		stdout(message, fields) { log('stdout', message, fields) },
-		stderr(message, fields) { log('stderr', message, fields) },
+		debug: (message) => logger.debug(openWorkflowMessage(message)),
+		info: (message) => logger.info(openWorkflowMessage(message)),
+		warn: (message) => logger.warn(openWorkflowMessage(message)),
+		error: (message) => logger.error(openWorkflowMessage(message)),
+	}
+}
+
+export function createLogger(): Logger {
+	return {
+		debug(message, fields) {
+			log('debug', message, fields)
+		},
+		info(message, fields) {
+			log('info', message, fields)
+		},
+		warn(message, fields) {
+			log('warn', message, fields)
+		},
+		error(message, fields) {
+			log('error', message, fields)
+		},
+		stdout(message, fields) {
+			log('stdout', message, fields)
+		},
+		stderr(message, fields) {
+			log('stderr', message, fields)
+		},
 	}
 }

packages/web/worker/src/workflows/job-progress-purge.ts

@@ -1,11 +1,9 @@
-import { getOpenWorkflow, purgeExpiredWorkflowRuns } from '@luzzle/web.jobs'
+import { getOpenWorkflow } from '@luzzle/web.jobs'
 import { jobProgressPurgeSpec } from '@luzzle/web.jobs/specs'
 import { getWorkerContext } from '../services/context.js'
 import { JobProgress } from '../core/job-progress.js'
 import { promises as fs } from 'node:fs'
 import path from 'node:path'
-import { resolveOpenWorkflowDbPath } from '../services/db.js'
-import { DatabaseSync } from 'node:sqlite'
 
 export function registerJobProgressPurgeWorkflow(): void {
 	const openWorkflow = getOpenWorkflow()
@@ -15,12 +13,10 @@ export function registerJobProgressPurgeWorkflow(): void {
 		const { db, logger, config } = ctx
 		const retentionDays = input.retentionDays ?? 7
 
-		// 1. Purge old entries from database
 		const purgedJobIds = await step.run({ name: 'purge-db-entries' }, async () => {
 			return await new JobProgress(db, retentionDays).purgeExpired()
 		})
 
-		// 2. Clean up preview directories on disk
 		await step.run({ name: 'purge-disk-directories' }, async () => {
 			for (const jobId of purgedJobIds) {
 				const previewDir = path.join(
@@ -40,21 +36,6 @@ export function registerJobProgressPurgeWorkflow(): void {
 			}
 		})
 
-		// 3. Clean up OpenWorkflow database entries
-		await step.run({ name: 'purge-openworkflow-entries' }, async () => {
-			try {
-				const openWorkflowDbPath = resolveOpenWorkflowDbPath(config)
-				const openWorkflowDb = new DatabaseSync(openWorkflowDbPath)
-				openWorkflowDb.exec('PRAGMA journal_mode = WAL;')
-				openWorkflowDb.exec('PRAGMA synchronous = NORMAL;')
-				openWorkflowDb.exec('PRAGMA busy_timeout = 5000;')
-				const count = purgeExpiredWorkflowRuns(openWorkflowDb, retentionDays)
-				logger.info('purged openworkflow runs', { count })
-			} catch (err) {
-				logger.error('failed to purge openworkflow runs', { error: String(err) })
-			}
-		})
-
 		logger.info('openworkflow purge complete', {
 			retentionDays,
 			purgedCount: purgedJobIds.length,

pnpm-lock.yaml

@@ -291,8 +291,8 @@ importers:
         specifier: 4.0.2
         version: 4.0.2
       openworkflow:
-        specifier: 0.9.2
-        version: 0.9.2
+        specifier: 0.11.0
+        version: 0.11.0
     devDependencies:
       '@auth/core':
         specifier: 0.41.1
@@ -552,8 +552,8 @@ importers:
         specifier: workspace:*
         version: link:../../../apps/core
       openworkflow:
-        specifier: 0.9.2
-        version: 0.9.2
+        specifier: 0.11.0
+        version: 0.11.0
     devDependencies:
       '@eslint/js':
         specifier: 'catalog:'
@@ -865,8 +865,8 @@ importers:
         specifier: 3.0.2
         version: 3.0.2
       openworkflow:
-        specifier: 0.9.2
-        version: 0.9.2
+        specifier: 0.11.0
+        version: 0.11.0
       rehype-autolink-headings:
         specifier: 7.1.0
         version: 7.1.0
@@ -2635,6 +2635,10 @@ packages:
   crelt@1.0.6:
     resolution: {integrity: sha512-VQ2MBenTq1fWZUH9DJNGti7kKv6EeAuYr3cLwxUWhIu1baTaXh4Ib5W2CqHVqib4/MqbYGJqiL3Zb8GJZr3l4g==}
 
+  cron-parser@5.10.1:
+    resolution: {integrity: sha512-pKRrRagItwk9rGIStcUyMxFX34x56zoX3KDf9HJ4ttwkXIK1qxK63EvXvEDmlxI9AdPJ+Y7TEKRftRlW5eSS7A==}
+    engines: {node: '>=18'}
+
   cross-spawn@7.0.6:
     resolution: {integrity: sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==}
     engines: {node: '>= 8'}
@@ -3314,6 +3318,10 @@ packages:
   lru-cache@10.4.3:
     resolution: {integrity: sha512-JNAzZcXrCt42VGLuYz0zfAzDfAvJWW6AfYlDBQyDV5DClI2m5sAmK+OIO7s59XfsRsWHp02jAJrRadPRGTt6SQ==}
 
+  luxon@3.7.2:
+    resolution: {integrity: sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==}
+    engines: {node: '>=12'}
+
   lz-string@1.5.0:
     resolution: {integrity: sha512-h5bgJWpxJNswbU7qCrV0tIKQCaS3blPDrqKWx+QxzuzL1zGUzij9XCWLrSLsJPu5t+eWA/ycetzYAO5IOMcWAQ==}
     hasBin: true
@@ -3580,12 +3588,15 @@ packages:
   oniguruma-to-es@4.3.6:
     resolution: {integrity: sha512-csuQ9x3Yr0cEIs/Zgx/OEt9iBw9vqIunAPQkx19R/fiMq2oGVTgcMqO/V3Ybqefr1TBvosI6jU539ksaBULJyA==}
 
-  openworkflow@0.9.2:
-    resolution: {integrity: sha512-NluYomuVLf+rAMJ9ASKqw8nbPAAD+C/HSTZoJacZdelLfTpndzHvPcP8SCANg9me9zek6mao0uUOGGbrWUWhGw==}
-    engines: {node: '>=20'}
+  openworkflow@0.11.0:
+    resolution: {integrity: sha512-fdDfRjuNMLV+p3svZShrV5i/ZPbBS8GruWLlthk8Ezt3ueYPwVKk0mcD+knAmQ4J9RASBsY5JqiatiLtEcwgVw==}
+    engines: {node: '>=22.22.0'}
     peerDependencies:
+      '@opentelemetry/api': '1'
       postgres: ^3.4.9
     peerDependenciesMeta:
+      '@opentelemetry/api':
+        optional: true
       postgres:
         optional: true
 
@@ -6273,6 +6284,10 @@ snapshots:
 
   crelt@1.0.6: {}
 
+  cron-parser@5.10.1:
+    dependencies:
+      luxon: 3.7.2
+
   cross-spawn@7.0.6:
     dependencies:
       path-key: 3.1.1
@@ -7001,6 +7016,8 @@ snapshots:
 
   lru-cache@10.4.3: {}
 
+  luxon@3.7.2: {}
+
   lz-string@1.5.0: {}
 
   magic-string@0.30.21:
@@ -7503,7 +7520,9 @@ snapshots:
       regex: 6.1.0
       regex-recursion: 6.0.2
 
-  openworkflow@0.9.2: {}
+  openworkflow@0.11.0:
+    dependencies:
+      cron-parser: 5.10.1
 
   optionator@0.9.4:
     dependencies: