.changeset/native-worker-maintenance.md

@@ -0,0 +1,8 @@
+---
+"@luzzle/web.worker": patch
+---
+
+Schedule preview-data cleanup with OpenWorkflow cron and durable jitter near
+02:00 UTC instead of a startup run and process-local timer. Remove expired preview
+files before their application progress/log rows, retain active jobs and engine
+history, and remove the unused duplicate cleanup step.

docs/web/deployment.md

@@ -67,6 +67,13 @@ To update, back up your archive and deployment data, change `LUZZLE_VERSION` if
 pinned, then run `docker compose pull` and `docker compose up -d`.
 [Republish affected pieces](templates.md#apply-your-choices) after OG changes.
 
+## Preview cleanup
+
+The worker cleans expired preview files and app logs daily, near **02:00 UTC
+±12 minutes**. It waits through OpenWorkflow, so other jobs can run during the
+random delay. There is no extra cleanup run at startup; downtime or a busy worker
+can delay it. Finished-job data is kept for seven days. Workflow history stays.
+
 ## Optional remote sync
 
 Put your rclone configuration in `rclone/rclone.conf`. Set `sync.archive` for

packages/web/jobs/src/specs.ts

@@ -17,6 +17,14 @@ export const jobProgressPurgeSpec = defineWorkflowSpec<
 	JobProgressPurgeResult
 >({
 	name: 'JobProgressPurge',
+	schedules: [
+		{
+			id: 'luzzle:job-progress-purge',
+			cron: '48 1 * * *',
+			timezone: 'UTC',
+			input: { retentionDays: 7 },
+		},
+	],
 })
 
 export const previewSpec = defineWorkflowSpec<PreviewPayload, PreviewResult>({

packages/web/worker/src/core/job-progress.test.ts

@@ -1,119 +1,113 @@
-/* eslint-disable @typescript-eslint/no-explicit-any */
 import { describe, it, expect, beforeEach, afterEach } from 'vitest'
-import { setupDatabase, teardownDatabase } from '../../test/db.js'
+import { createAppDb, runWebMigrations } from '@luzzle/web.db'
+import { sql } from 'kysely'
 import { JobProgress } from './job-progress.js'
-import type { Kysely } from 'kysely'
-import type { WebDatabase } from '../services/db.js'
 
 describe('JobProgress', () => {
-	let testDb: Kysely<WebDatabase>
+	let testDb: ReturnType<typeof createAppDb>
 	let jobProgress: JobProgress
 
 	beforeEach(async () => {
-		testDb = await setupDatabase()
-		jobProgress = new JobProgress(testDb, 2)
+		testDb = createAppDb(':memory:')
+		expect((await runWebMigrations(testDb)).error).toBeUndefined()
+		jobProgress = new JobProgress(testDb)
 	})
 
 	afterEach(async () => {
-		await teardownDatabase(testDb)
+		await testDb.destroy()
 	})
 
-	it('should start a phase', async () => {
+	it('starts and restarts a phase', async () => {
 		await jobProgress.start('42', 'archive.sync')
-
-		const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute()
+		await jobProgress.complete('42', 'archive.sync', 'done')
+		await jobProgress.start('42', 'archive.sync')
+		const rows = await testDb.selectFrom('job_progress').selectAll().execute()
 		expect(rows).toHaveLength(1)
 		expect(rows[0]).toMatchObject({
 			job_id: '42',
 			phase: 'archive.sync',
 			status: 'running',
 			message: null,
-			finished_at: null
+			finished_at: null,
 		})
 		expect(rows[0].started_at).toBeGreaterThan(0)
 	})
 
-	it('should complete a phase', async () => {
+	it('completes a phase', async () => {
 		await jobProgress.start('42', 'archive.sync')
 		await jobProgress.complete('42', 'archive.sync', 'completed successfully')
-
-		const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute()
-		expect(rows).toHaveLength(1)
-		expect(rows[0]).toMatchObject({
-			job_id: '42',
-			phase: 'archive.sync',
-			status: 'completed',
-			message: 'completed successfully'
-		})
-		expect(rows[0].finished_at).toBeGreaterThan(0)
+		const row = await testDb.selectFrom('job_progress').selectAll().executeTakeFirstOrThrow()
+		expect(row).toMatchObject({ status: 'completed', message: 'completed successfully' })
+		expect(row.finished_at).toBeGreaterThan(0)
 	})
 
-	it('should skip a phase', async () => {
+	it('skips a phase', async () => {
 		await jobProgress.start('42', 'archive.sync')
 		await jobProgress.skip('42', 'archive.sync', 'no remote configured')
-
-		const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute()
-		expect(rows).toHaveLength(1)
-		expect(rows[0]).toMatchObject({
-			job_id: '42',
-			phase: 'archive.sync',
-			status: 'skipped',
-			message: 'no remote configured'
-		})
-		expect(rows[0].finished_at).toBeGreaterThan(0)
+		const row = await testDb.selectFrom('job_progress').selectAll().executeTakeFirstOrThrow()
+		expect(row).toMatchObject({ status: 'skipped', message: 'no remote configured' })
+		expect(row.finished_at).toBeGreaterThan(0)
 	})
 
-	it('should fail a phase with an Error', async () => {
+	it.each([new Error('network error'), 'network error'])('fails a phase: %s', async (error) => {
 		await jobProgress.start('42', 'archive.sync')
-		await jobProgress.fail('42', 'archive.sync', new Error('network error'))
-
-		const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute()
-		expect(rows).toHaveLength(1)
-		expect(rows[0]).toMatchObject({
-			job_id: '42',
-			phase: 'archive.sync',
-			status: 'failed',
-			message: 'network error'
-		})
-		expect(rows[0].finished_at).toBeGreaterThan(0)
+		await jobProgress.fail('42', 'archive.sync', error)
+		const row = await testDb.selectFrom('job_progress').selectAll().executeTakeFirstOrThrow()
+		expect(row).toMatchObject({ status: 'failed', message: 'network error' })
+		expect(row.finished_at).toBeGreaterThan(0)
 	})
 
-	it('should fail a phase with a string', async () => {
-		await jobProgress.start('42', 'archive.sync')
-		await jobProgress.fail('42', 'archive.sync', 'string error')
+	async function log(jobId: string) {
+		await testDb
+			.insertInto('job_progress_logs')
+			.values({
+				job_id: jobId,
+				phase: 'render',
+				line_number: 1,
+				ts: Date.now(),
+				level: 'info',
+				message: 'log',
+			})
+			.execute()
+	}
 
-		const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute()
-		expect(rows).toHaveLength(1)
-		expect(rows[0]).toMatchObject({
-			job_id: '42',
-			phase: 'archive.sync',
-			status: 'failed',
-			message: 'string error'
-		})
-		expect(rows[0].finished_at).toBeGreaterThan(0)
+	it('lists the distinct union of progress and log IDs, including logs without progress', async () => {
+		await jobProgress.start('both', 'one')
+		await jobProgress.start('both', 'two')
+		await jobProgress.start('progress-only', 'one')
+		await log('both')
+		await log('logs-only')
+		expect((await jobProgress.listJobIds()).sort()).toEqual(['both', 'logs-only', 'progress-only'])
 	})
 
-	it('should purge old records based on started_at', async () => {
-		// Insert an old record
-		await testDb.insertInto('job_progress' as any).values({
-			job_id: '1',
-			phase: 'old.phase',
-			status: 'completed',
-			started_at: Date.now() - 3 * 24 * 60 * 60 * 1000, // 3 days old
-		}).execute()
-
-		// Insert a recent record
-		await testDb.insertInto('job_progress' as any).values({
-			job_id: '2',
-			phase: 'new.phase',
-			status: 'completed',
-			started_at: Date.now() - 1 * 24 * 60 * 60 * 1000, // 1 day old
-		}).execute()
-
-		await jobProgress.purgeExpired()
-
-		const rows = await testDb.selectFrom('job_progress' as any).selectAll().execute()
-		expect(rows).toHaveLength(1)
-		expect(rows[0].job_id).toBe('2')
+	it('purges only selected IDs in bounded chunks, idempotently', async () => {
+		const ids = Array.from({ length: 205 }, (_, index) => `job-${index}`)
+		for (const id of [...ids, 'keep']) {
+			await jobProgress.start(id, 'render')
+			await log(id)
+		}
+		await jobProgress.purgeJobs(ids)
+		await jobProgress.purgeJobs(ids)
+		await jobProgress.purgeJobs([])
+		expect(await jobProgress.listJobIds()).toEqual(['keep'])
+		expect(await testDb.selectFrom('job_progress').select('job_id').execute()).toEqual([
+			{ job_id: 'keep' },
+		])
+		expect(await testDb.selectFrom('job_progress_logs').select('job_id').execute()).toEqual([
+			{ job_id: 'keep' },
+		])
 	})
-})
+
+	it('rolls back logs and earlier chunks and propagates database failures', async () => {
+		const ids = Array.from({ length: 101 }, (_, index) => `job-${index}`)
+		for (const id of ids) {
+			await jobProgress.start(id, 'render')
+			await log(id)
+		}
+		await sql`CREATE TRIGGER prevent_purge BEFORE DELETE ON job_progress
+			WHEN OLD.job_id = 'job-100' BEGIN SELECT RAISE(FAIL, 'delete failed'); END`.execute(testDb)
+		await expect(jobProgress.purgeJobs(ids)).rejects.toThrow('delete failed')
+		expect(await testDb.selectFrom('job_progress').selectAll().execute()).toHaveLength(101)
+		expect(await testDb.selectFrom('job_progress_logs').selectAll().execute()).toHaveLength(101)
+	})
+})

packages/web/worker/src/core/job-progress.ts

@@ -3,47 +3,29 @@ import type { AppDatabase } from '../services/db.js'
 
 export class JobProgress {
 	private readonly db: Kysely<AppDatabase>
-	private readonly retentionDays: number
 
-	constructor(db: Kysely<AppDatabase>, retentionDays: number = 7) {
+	constructor(db: Kysely<AppDatabase>) {
 		this.db = db
-		this.retentionDays = retentionDays
 	}
 
-	async purgeExpiredBatch(limit: number): Promise<string[]> {
-		try {
-			const cutoffMs = Date.now() - this.retentionDays * 24 * 60 * 60 * 1000
+	async listJobIds(): Promise<string[]> {
+		const rows = await this.db
+			.selectFrom('job_progress')
+			.select('job_id')
+			.union(this.db.selectFrom('job_progress_logs').select('job_id'))
+			.execute()
+		return rows.map((row) => row.job_id)
+	}
 
-			const oldJobs = await this.db
-				.selectFrom('job_progress')
-				.select('job_id')
-				.distinct()
-				.where('started_at', '<', cutoffMs)
-				.limit(limit)
-				.execute()
-
-			const oldJobIds = oldJobs.map((row) => row.job_id)
-
-			if (oldJobIds.length > 0) {
-				await this.db.deleteFrom('job_progress_logs').where('job_id', 'in', oldJobIds).execute()
-				await this.db.deleteFrom('job_progress').where('job_id', 'in', oldJobIds).execute()
+	async purgeJobs(jobIds: string[]): Promise<void> {
+		if (jobIds.length === 0) return
+		await this.db.transaction().execute(async (trx) => {
+			for (let offset = 0; offset < jobIds.length; offset += 100) {
+				const batch = jobIds.slice(offset, offset + 100)
+				await trx.deleteFrom('job_progress_logs').where('job_id', 'in', batch).execute()
+				await trx.deleteFrom('job_progress').where('job_id', 'in', batch).execute()
 			}
-
-			return oldJobIds
-		} catch (err) {
-			console.error('JobProgress purgeExpiredBatch failed:', err)
-			return []
-		}
-	}
-
-	async purgeExpired(): Promise<string[]> {
-		const allPurged: string[] = []
-		let batch: string[] = []
-		do {
-			batch = await this.purgeExpiredBatch(100)
-			allPurged.push(...batch)
-		} while (batch.length === 100)
-		return allPurged
+		})
 	}
 
 	async start(jobId: string, phase: string): Promise<void> {

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

@@ -14,12 +14,15 @@ import type * as Workflows from './workflows/index.js'
 const probe = defineWorkflowSpec<void, string>({ name: 'ShutdownProbe' })
 
 afterEach(() => {
+	vi.useRealTimers()
 	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()
+	vi.useFakeTimers({ toFake: ['Date'] })
+	vi.setSystemTime(new Date('2026-06-02T12:00:00.000Z'))
 	const directory = await fs.mkdtemp(path.join(tmpdir(), 'luzzle-worker-'))
 	const queuePath = path.join(directory, 'queue.sqlite')
 	const config = validateConfig(
@@ -64,12 +67,12 @@ test('starts the native worker and health server, then lets active work finish b
 		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')
+		expect(await jobs.getOpenWorkflow().schedules.get('luzzle:job-progress-purge')).toMatchObject({
+			cron: '0 48 1 * * *',
+			timezone: 'UTC',
+			nextRunAt: new Date('2026-06-03T01:48:00.000Z'),
+		})
+		expect((await backend.listWorkflowRuns({ workflowName: 'JobProgressPurge' })).data).toEqual([])
 		const handle = await jobs.getOpenWorkflow().runWorkflow(probe)
 		await expect.poll(() => entered).toBe(true)
 
@@ -104,7 +107,7 @@ test('starts the native worker and health server, then lets active work finish b
 			)
 			expect(
 				(await reopened.listWorkflowRuns({ workflowName: 'JobProgressPurge' })).data
-			).toHaveLength(1)
+			).toHaveLength(0)
 		} finally {
 			await reopened.stop()
 		}

packages/web/worker/src/index.ts

@@ -6,14 +6,12 @@ import { createHealthServer } from './services/health.js'
 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 closeHealthServer(server: Server): Promise<void> {
 	if (!server.listening) return
@@ -34,11 +32,9 @@ export async function main(
 	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[] = []
@@ -82,19 +78,6 @@ export async function main(
 		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

packages/web/worker/src/steps/job-progress-purge.test.ts

@@ -1,57 +0,0 @@
-import { describe, test, expect, vi } from 'vitest'
-import { jobProgressPurgeStep } from './job-progress-purge.js'
-import { JobProgress } from '../core/job-progress.js'
-import type { WorkerContext } from '../services/context.js'
-import { promises as fs } from 'node:fs'
-
-const mockPurgeExpired = vi.fn().mockResolvedValue([])
-
-vi.mock('../core/job-progress.js', () => ({
-	JobProgress: vi.fn().mockImplementation(() => ({
-		purgeExpired: mockPurgeExpired,
-	})),
-}))
-
-function makeCtx(): WorkerContext {
-	return {
-		config: { paths: { assets: '/assets/pieces' } } as unknown as WorkerContext['config'],
-		logger: {
-			debug: vi.fn(),
-			info: vi.fn(),
-			warn: vi.fn(),
-			error: vi.fn(),
-			stdout: vi.fn(),
-			stderr: vi.fn(),
-		},
-		rclone: {} as WorkerContext['rclone'],
-		db: {} as WorkerContext['db'],
-	}
-}
-
-describe('jobProgressPurgeStep', () => {
-	test('uses default retentionDays=7', async () => {
-		mockPurgeExpired.mockResolvedValueOnce([])
-		const result = await jobProgressPurgeStep.run({}, makeCtx())
-		expect(result.status).toBe('completed')
-		expect(JobProgress).toHaveBeenCalledWith(expect.anything(), 7)
-	})
-
-	test('honors explicit retentionDays', async () => {
-		mockPurgeExpired.mockResolvedValueOnce([])
-		await jobProgressPurgeStep.run({ retentionDays: 14 }, makeCtx())
-		expect(JobProgress).toHaveBeenCalledWith(expect.anything(), 14)
-	})
-
-	test('deletes cached preview folders for purged job IDs', async () => {
-		mockPurgeExpired.mockResolvedValueOnce([101, 102])
-		const rmSpy = vi.spyOn(fs, 'rm').mockResolvedValue(undefined)
-
-		await jobProgressPurgeStep.run({}, makeCtx())
-
-		expect(rmSpy).toHaveBeenCalledTimes(2)
-		expect(rmSpy).toHaveBeenNthCalledWith(1, '/assets/previews/101', { recursive: true, force: true })
-		expect(rmSpy).toHaveBeenNthCalledWith(2, '/assets/previews/102', { recursive: true, force: true })
-
-		rmSpy.mockRestore()
-	})
-})

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

@@ -1,30 +0,0 @@
-import { completed, type Step, type StepResult } from '../core/step.js'
-import { JobProgress } from '../core/job-progress.js'
-import { promises as fs } from 'node:fs'
-import path from 'node:path'
-
-export interface JobProgressPurgeInput {
-	retentionDays?: number
-}
-
-export const jobProgressPurgeStep: Step<JobProgressPurgeInput, void> = {
-	name: 'job_progress.purge',
-	async run(input, ctx): Promise<StepResult<void>> {
-		const { db, logger, config } = ctx
-		const retentionDays = input.retentionDays ?? 7
-		const purgedJobIds = await new JobProgress(db, retentionDays).purgeExpired()
-
-		for (const jobId of purgedJobIds) {
-			const previewDir = path.join(path.dirname(config.paths.assets), 'previews', jobId.toString())
-			try {
-				await fs.rm(previewDir, { recursive: true, force: true })
-				logger.info('purged preview assets directory', { jobId, previewDir })
-			} catch (err) {
-				logger.warn('failed to purge preview assets directory', { jobId, error: String(err) })
-			}
-		}
-
-		logger.info('job_progress purge complete', { retentionDays, purgedCount: purgedJobIds.length })
-		return completed(undefined)
-	},
-}

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

@@ -1,27 +1,335 @@
-import { describe, test, expect, beforeEach } from 'vitest'
-import { registerJobProgressPurgeWorkflow } from './job-progress-purge.js'
-import { initOpenWorkflow, getOpenWorkflow } from '@luzzle/web.jobs'
+import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'
+import { OpenWorkflow, defineWorkflowSpec, type Worker } from 'openworkflow'
+import { BackendSqlite } from 'openworkflow/sqlite'
+import { createAppDb, runWebMigrations } from '@luzzle/web.db'
+import { getOpenWorkflow, getOpenWorkflowBackend } from '@luzzle/web.jobs'
 import { jobProgressPurgeSpec } from '@luzzle/web.jobs/specs'
+import { promises as fs } from 'node:fs'
+import type * as Crypto from 'node:crypto'
+import { tmpdir } from 'node:os'
+import path from 'node:path'
+import { validateConfig } from '@luzzle/web.config'
+import { getWorkerContext, type WorkerContext } from '../services/context.js'
+import { JobProgress } from '../core/job-progress.js'
+import { makeLogger } from '../../test/logger.js'
+import { registerJobProgressPurgeWorkflow } from './job-progress-purge.js'
 
-describe('workflows/job-progress-purge', () => {
-	beforeEach(() => {
-		initOpenWorkflow({ dbPath: ':memory:' })
+const { jitter } = vi.hoisted(() => ({ jitter: vi.fn(() => 0) }))
+vi.mock('node:crypto', async (original) => ({
+	...(await original<typeof Crypto>()),
+	randomInt: jitter,
+}))
+vi.mock('@luzzle/web.jobs', () => ({ getOpenWorkflow: vi.fn(), getOpenWorkflowBackend: vi.fn() }))
+vi.mock('../services/context.js', () => ({ getWorkerContext: vi.fn() }))
+const day = 86_400_000
+const now = new Date('2026-06-02T12:00:00Z')
+const candidateSpec = defineWorkflowSpec({ name: 'PreviewFixture' })
+
+describe('native preview maintenance', () => {
+	let directory: string, previews: string
+	let backend: BackendSqlite, client: OpenWorkflow, worker: Worker
+	let ctx: WorkerContext, progress: JobProgress
+	const getRun = (workflowRunId: string) => backend.getWorkflowRun({ workflowRunId })
+	const steps = async (workflowRunId: string) =>
+		(await backend.listStepAttempts({ workflowRunId })).data
+	const preview = (id: string) => path.join(previews, id)
+
+	function connect() {
+		backend = BackendSqlite.connect(path.join(directory, 'workflows.sqlite'))
+		client = new OpenWorkflow({ backend, logger: makeLogger() })
+		vi.mocked(getOpenWorkflow).mockReturnValue(client)
+		vi.mocked(getOpenWorkflowBackend).mockReturnValue(backend)
+		registerJobProgressPurgeWorkflow()
+		worker = client.newWorker({ concurrency: 1 })
+	}
+	beforeEach(async () => {
+		vi.useFakeTimers({ toFake: ['Date'] })
+		vi.setSystemTime(now)
+		jitter.mockReset().mockReturnValue(0)
+		directory = await fs.mkdtemp(path.join(tmpdir(), 'luzzle-maintenance-'))
+		previews = path.join(directory, 'previews')
+		const db = createAppDb(path.join(directory, 'app.sqlite'))
+		expect((await runWebMigrations(db)).error).toBeUndefined()
+		const config = validateConfig(
+			{
+				storage: { root: directory },
+				paths: { assets: path.join(directory, 'pieces') },
+			},
+			{}
+		)
+		ctx = { db, config, logger: makeLogger(), rclone: {} as WorkerContext['rclone'] }
+		vi.mocked(getWorkerContext).mockReturnValue(ctx)
+		progress = new JobProgress(db)
+		connect()
+	})
+	afterEach(async () => {
+		await worker.stop()
+		vi.restoreAllMocks()
+		await backend.stop()
+		await ctx.db.destroy()
+		await fs.rm(directory, { recursive: true, force: true })
+		vi.useRealTimers()
+	})
+	async function tick() {
+		expect(await worker.tick()).toBe(1)
+		await worker.stop()
+	}
+	async function seedPreviewData(id: string, rows: 'both' | 'logs' | 'none' = 'both') {
+		if (rows === 'both') await progress.start(id, 'render')
+		if (rows !== 'none')
+			await ctx.db
+				.insertInto('job_progress_logs')
+				.values({
+					job_id: id,
+					phase: 'render',
+					line_number: 1,
+					ts: Date.now(),
+					level: 'info',
+					message: 'rendered',
+				})
+				.execute()
+		await fs.mkdir(preview(id), { recursive: true })
+		await fs.writeFile(path.join(preview(id), 'preview.bin'), 'preview output')
+	}
+	async function createCandidateRun(
+		status = 'completed',
+		finishedAgeMs = 8 * day,
+		rows: 'both' | 'logs' | 'none' = 'both'
+	) {
+		vi.setSystemTime(now.getTime() - 20 * day)
+		const options = status === 'pending' ? { availableAt: new Date(now.getTime() + day) } : {}
+		const { workflowRun: run } = await client.runWorkflow(candidateSpec, {}, options)
+		const owned = { workflowRunId: run.id, workerId: 'fixture' }
+		if (status !== 'pending') {
+			const claim = await backend.claimWorkflowRun({
+				workerId: 'fixture',
+				leaseDurationMs: 40 * day,
+			})
+			expect(claim?.id).toBe(run.id)
+			const step = await backend.createStepAttempt({
+				...owned,
+				stepName: 'render',
+				kind: 'function',
+				config: {},
+				context: null,
+			})
+			await backend.completeStepAttempt({ ...owned, stepAttemptId: step.id, output: 'rendered' })
+			vi.setSystemTime(now.getTime() - finishedAgeMs)
+			if (status === 'completed') await backend.completeWorkflowRun({ ...owned, output: 'ok' })
+			if (status === 'failed')
+				await backend.failWorkflowRun({
+					...owned,
+					error: { message: 'failed' },
+					retryPolicy: {
+						maximumAttempts: 1,
+						initialInterval: '1s',
+						maximumInterval: '1s',
+						backoffCoefficient: 1,
+					},
+				})
+			if (status === 'canceled') await backend.cancelWorkflowRun({ workflowRunId: run.id })
+		}
+		vi.setSystemTime(now)
+		await seedPreviewData(run.id, rows)
+		return run.id
+	}
+	async function purge() {
+		const handle = await client.runWorkflow(jobProgressPurgeSpec, {})
+		await tick()
+		return handle.workflowRun.id
+	}
+
+	test('removes aged terminal app data and directories without deleting SDK history', async () => {
+		const expired = [
+			await createCandidateRun(),
+			await createCandidateRun('failed', 8 * day, 'logs'),
+			await createCandidateRun('canceled', 8 * day, 'none'),
+		]
+		const kept = [
+			await createCandidateRun('completed', day),
+			await createCandidateRun('completed', 7 * day),
+			await createCandidateRun('running'),
+			await createCandidateRun('pending'),
+			'unknown',
+		]
+		await seedPreviewData('unknown')
+		await progress.start('../outside', 'render')
+		await fs.writeFile(path.join(directory, 'outside'), 'keep')
+		const history = await Promise.all(
+			expired.map(async (id) => [await getRun(id), await steps(id)])
+		)
+		await purge()
+		for (let index = 0; index < expired.length; index++) {
+			const id = expired[index]
+			await expect(fs.stat(preview(id))).rejects.toMatchObject({ code: 'ENOENT' })
+			expect(await progress.listJobIds()).not.toContain(id)
+			expect([await getRun(id), await steps(id)]).toEqual(history[index])
+		}
+		for (const id of kept) {
+			expect(await fs.readFile(path.join(preview(id), 'preview.bin'), 'utf8')).toBe(
+				'preview output'
+			)
+			expect(await progress.listJobIds()).toContain(id)
+		}
+		expect(await fs.readFile(path.join(directory, 'outside'), 'utf8')).toBe('keep')
 	})
 
-	test('registers workflow implementation successfully', () => {
-		registerJobProgressPurgeWorkflow()
-		const ow = getOpenWorkflow()
+	test('honors an explicit retention window and rejects a non-positive window without deleting data', async () => {
+		const id = await createCandidateRun()
+		const extended = await client.runWorkflow(jobProgressPurgeSpec, { retentionDays: 14 })
+		await tick()
+		expect((await getRun(extended.workflowRun.id))?.status).toBe('completed')
+		expect(await fs.stat(preview(id))).toBeDefined()
+		expect(await progress.listJobIds()).toContain(id)
+		const invalid = await client.runWorkflow(jobProgressPurgeSpec, { retentionDays: 0 })
+		await tick()
+		expect((await getRun(invalid.workflowRun.id))?.status).toBe('failed')
+		expect(await fs.stat(preview(id))).toBeDefined()
+		expect(await progress.listJobIds()).toContain(id)
+	})
 
-		// Verify the workflow is correctly registered under its name
-		const registered = (
-			ow as unknown as {
-				registry: {
-					get: (name: string) => { spec: { name: string }; fn: (...args: unknown[]) => unknown }
-				}
-			}
-		).registry.get(jobProgressPurgeSpec.name)
-		expect(registered).toBeDefined()
-		expect(registered.spec.name).toBe(jobProgressPurgeSpec.name)
-		expect(typeof registered.fn).toBe('function')
+	test('file failure preserves rows and retries saved paths and IDs, even after deleting files', async () => {
+		const id = await createCandidateRun()
+		const recent = await createCandidateRun('completed', day)
+		const list = vi.spyOn(JobProgress.prototype, 'listJobIds')
+		const rm = fs.rm
+		vi.spyOn(fs, 'rm').mockImplementationOnce(async (target, options) => {
+			await rm(target, options)
+			throw new Error('file deletion interrupted')
+		})
+		const purgeId = await purge()
+		await expect(fs.stat(preview(id))).rejects.toMatchObject({ code: 'ENOENT' })
+		expect(
+			await ctx.db
+				.selectFrom('job_progress_logs')
+				.select('job_id')
+				.where('job_id', '=', id)
+				.execute()
+		).toHaveLength(1)
+		expect(
+			await ctx.db.selectFrom('job_progress').select('job_id').where('job_id', '=', id).execute()
+		).toHaveLength(1)
+		const attempts = await steps(purgeId)
+		expect(attempts.find((step) => step.stepName === 'plan-cleanup')?.output).toEqual({
+			cutoffMs: now.getTime() - 7 * day,
+			jobIds: [id],
+			previewDirs: [preview(id)],
+		})
+		expect(attempts.find((step) => step.stepName === 'purge-preview-directories')?.status).toBe(
+			'failed'
+		)
+		ctx.config.paths.assets = path.join(directory, 'different-root', 'pieces')
+		vi.setSystemTime(now.getTime() + 10 * day)
+		await tick()
+		expect(list).toHaveBeenCalledTimes(1)
+		expect(jitter).toHaveBeenCalledTimes(1)
+		expect(await progress.listJobIds()).toEqual([recent])
+		expect(await fs.stat(preview(recent))).toBeDefined()
+		expect((await getRun(purgeId))?.status).toBe('completed')
+	})
+
+	test.each(['before', 'after'])(
+		'retries app deletion %s commit after files are gone',
+		async (boundary) => {
+			const id = await createCandidateRun()
+			const original = JobProgress.prototype.purgeJobs
+			const remove = vi
+				.spyOn(JobProgress.prototype, 'purgeJobs')
+				.mockImplementationOnce(async function (this: JobProgress, ids) {
+					if (boundary === 'after') await original.call(this, ids)
+					throw new Error('app deletion interrupted')
+				})
+			const files = vi.spyOn(fs, 'rm')
+			const list = vi.spyOn(JobProgress.prototype, 'listJobIds')
+			const purgeId = await purge()
+			await expect(fs.stat(preview(id))).rejects.toMatchObject({ code: 'ENOENT' })
+			expect(await progress.listJobIds()).toEqual(boundary === 'before' ? [id] : [])
+			list.mockClear()
+			vi.setSystemTime(now.getTime() + day)
+			await tick()
+			expect(list).not.toHaveBeenCalled()
+			expect(remove.mock.calls).toEqual([[[id]], [[id]]])
+			expect(files).toHaveBeenCalledTimes(1)
+			expect(await progress.listJobIds()).toEqual([])
+			expect((await getRun(purgeId))?.status).toBe('completed')
+		}
+	)
+
+	test('propagates discovery errors but treats a missing previews root as empty', async () => {
+		const id = await createCandidateRun()
+		await fs.rm(previews, { recursive: true })
+		const list = vi.spyOn(fs, 'readdir').mockRejectedValueOnce(new Error('directory unavailable'))
+		const purgeId = await purge()
+		expect(await progress.listJobIds()).toEqual([id])
+		vi.setSystemTime(now.getTime() + day)
+		await tick()
+		expect(list).toHaveBeenCalledTimes(2)
+		expect(await progress.listJobIds()).toEqual([])
+		expect((await getRun(purgeId))?.status).toBe('completed')
+	})
+
+	test('persists jitter, releases the only slot during native sleep, and resumes the saved choice', async () => {
+		jitter.mockReturnValueOnce(1440).mockReturnValueOnce(0)
+		const first = await purge()
+		expect(await getRun(first)).toMatchObject({
+			status: 'running',
+			workerId: null,
+			availableAt: new Date(now.getTime() + 1440_000),
+		})
+		const attempts = await steps(first)
+		expect(attempts.find((step) => step.stepName === 'choose-cleanup-jitter')).toMatchObject({
+			status: 'completed',
+			output: 1440,
+		})
+		expect(attempts.find((step) => step.stepName === 'cleanup-jitter')).toMatchObject({
+			kind: 'sleep',
+		})
+		await backend.stop()
+		connect()
+		const second = await purge()
+		expect((await getRun(second))?.status).toBe('completed')
+		vi.setSystemTime(now.getTime() + 1440_000)
+		await tick()
+		expect((await getRun(first))?.status).toBe('completed')
+		expect(jitter.mock.calls).toEqual([
+			[0, 1441],
+			[0, 1441],
+		])
+	})
+
+	test('persists cron and pause across restart and coalesces missed occurrences', async () => {
+		const scheduleId = 'luzzle:job-progress-purge'
+		await worker.start()
+		await worker.stop()
+		const saved = await client.schedules.get(scheduleId)
+		expect(saved).toMatchObject({
+			cron: '0 48 1 * * *',
+			timezone: 'UTC',
+			input: { retentionDays: 7 },
+			nextRunAt: new Date('2026-06-03T01:48:00Z'),
+		})
+		await backend.stop()
+		connect()
+		await worker.start()
+		await worker.stop()
+		expect(await client.schedules.get(scheduleId)).toEqual(saved)
+		await client.schedules.pause(scheduleId)
+		await backend.stop()
+		connect()
+		await worker.start()
+		await worker.stop()
+		expect((await client.schedules.get(scheduleId))?.nextRunAt).toBeNull()
+		await client.schedules.resume(scheduleId)
+		vi.setSystemTime(new Date('2026-06-06T12:00:00Z'))
+		expect(await backend.enqueueScheduledRuns({ before: new Date(), limit: 100 })).toBe(1)
+		expect(await backend.enqueueScheduledRuns({ before: new Date(), limit: 100 })).toBe(0)
+		const runs = (await backend.listWorkflowRuns({ workflowName: jobProgressPurgeSpec.name })).data
+		expect(runs).toHaveLength(1)
+		expect(runs[0].workflowScheduleId).toBe(scheduleId)
+		await tick()
+		expect((await getRun(runs[0].id))?.status).toBe('completed')
+		expect((await client.schedules.get(scheduleId))?.nextRunAt).toEqual(
+			new Date('2026-06-07T01:48:00Z')
+		)
 	})
 })

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

@@ -1,46 +1,67 @@
-import { getOpenWorkflow } from '@luzzle/web.jobs'
+import { getOpenWorkflow, getOpenWorkflowBackend } 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 { randomInt } from 'node:crypto'
 import { promises as fs } from 'node:fs'
 import path from 'node:path'
+import { getWorkerContext } from '../services/context.js'
+import { JobProgress } from '../core/job-progress.js'
 
 export function registerJobProgressPurgeWorkflow(): void {
 	const openWorkflow = getOpenWorkflow()
 
 	openWorkflow.implementWorkflow(jobProgressPurgeSpec, async ({ input, step }) => {
-		const ctx = getWorkerContext()
-		const { db, logger, config } = ctx
 		const retentionDays = input.retentionDays ?? 7
+		if (!Number.isFinite(retentionDays) || retentionDays <= 0) {
+			throw new Error('Cleanup retentionDays must be positive and finite')
+		}
+		const delaySeconds = await step.run({ name: 'choose-cleanup-jitter' }, () => randomInt(0, 1441))
+		if (delaySeconds > 0) await step.sleep('cleanup-jitter', `${delaySeconds}s`)
 
-		const purgedJobIds = await step.run({ name: 'purge-db-entries' }, async () => {
-			return await new JobProgress(db, retentionDays).purgeExpired()
+		const { db, logger, config } = getWorkerContext()
+		const progress = new JobProgress(db)
+		const plan = await step.run({ name: 'plan-cleanup' }, async () => {
+			const cutoffMs = Date.now() - retentionDays * 24 * 60 * 60 * 1000
+			const previewsRoot = path.resolve(path.dirname(config.paths.assets), 'previews')
+			let previewNames: string[]
+			try {
+				previewNames = (await fs.readdir(previewsRoot, { withFileTypes: true }))
+					.filter((entry) => entry.isDirectory())
+					.map((entry) => entry.name)
+			} catch (error) {
+				if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error
+				previewNames = []
+			}
+			const directories = new Set(previewNames)
+			const candidates = new Set([...(await progress.listJobIds()), ...previewNames])
+			const jobIds: string[] = []
+			const previewDirs: string[] = []
+			const backend = getOpenWorkflowBackend()
+			for (const jobId of candidates) {
+				if (!/^[a-zA-Z0-9_-]+$/.test(jobId)) continue
+				const previewDir = path.resolve(previewsRoot, jobId)
+				if (path.dirname(previewDir) !== previewsRoot) continue
+				const run = await backend.getWorkflowRun({ workflowRunId: jobId })
+				if (!run) continue
+				if (!['completed', 'failed', 'canceled', 'succeeded'].includes(run.status)) continue
+				if (!run.finishedAt) continue
+				const isExpired = run.finishedAt.getTime() < cutoffMs
+				if (!isExpired) continue
+				jobIds.push(jobId)
+				if (directories.has(jobId)) previewDirs.push(previewDir)
+			}
+			return { cutoffMs, jobIds, previewDirs }
 		})
 
-		await step.run({ name: 'purge-disk-directories' }, async () => {
-			for (const jobId of purgedJobIds) {
-				const previewDir = path.join(
-					path.dirname(config.paths.assets),
-					'previews',
-					jobId.toString()
-				)
-				try {
-					await fs.rm(previewDir, { recursive: true, force: true })
-					logger.info('purged preview assets directory', { jobId, previewDir })
-				} catch (err) {
-					logger.warn('failed to purge preview assets directory', {
-						jobId,
-						error: String(err),
-					})
-				}
+		await step.run({ name: 'purge-preview-directories' }, async () => {
+			for (const previewDir of plan.previewDirs) {
+				await fs.rm(previewDir, { recursive: true, force: true })
 			}
 		})
-
-		logger.info('openworkflow purge complete', {
-			retentionDays,
-			purgedCount: purgedJobIds.length,
+		await step.run({ name: 'purge-app-rows' }, async () => {
+			await progress.purgeJobs(plan.jobIds)
 		})
 
+		logger.info('openworkflow purge complete', { purgedCount: plan.jobIds.length })
 		return 'ok'
 	})
 }