.changeset/selective-publishing-backend.md

@@ -0,0 +1,9 @@
+---
+"@luzzle/web": minor
+"@luzzle/web.jobs": minor
+"@luzzle/web.worker": minor
+---
+
+Accept audited selections of pending additions and updates. Keep unselected work
+pending, apply global removals and schema changes, and preserve selected plans
+and hashes across worker retries. Omitted selections retain publish-all behavior.

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

@@ -20,7 +20,7 @@ export async function closeWorkflowQueue(queue: WorkflowQueue): Promise<void> {
 export function createPendingRun(
 	backend: BackendSqlite,
 	workflowName: string,
-	input: Record<string, boolean> = {}
+	input: Parameters<BackendSqlite['createWorkflowRun']>[0]['input'] = {}
 ) {
 	return backend.createWorkflowRun({
 		workflowName,
@@ -45,18 +45,25 @@ export async function claimNextRun(backend: BackendSqlite) {
 	return run
 }
 
-export async function completeNextRun(backend: BackendSqlite) {
+export async function completeNextRun(
+	backend: BackendSqlite,
+	output: Parameters<BackendSqlite['completeWorkflowRun']>[0]['output'] = null
+) {
 	const run = await claimNextRun(backend)
 	return backend.completeWorkflowRun({
 		workflowRunId: run.id,
 		workerId: 'fixture-worker',
-		output: null
+		output
 	})
 }
 
-export async function createCompletedRun(backend: BackendSqlite, workflowName: string) {
+export async function createCompletedRun(
+	backend: BackendSqlite,
+	workflowName: string,
+	output: Parameters<BackendSqlite['completeWorkflowRun']>[0]['output'] = null
+) {
 	const run = await createPendingRun(backend, workflowName)
-	const completed = await completeNextRun(backend)
+	const completed = await completeNextRun(backend, output)
 	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

@@ -232,6 +232,71 @@ describe('validateAuditForPublish', () => {
 		expect(await validateAuditForPublish(backend, 'audit-1')).toEqual({ ok: true })
 	})
 
+	test.each([
+		{ filePaths: [] },
+		{ filePaths: ['added.book.md', 'updated.book.md', 'added.book.md'] }
+	])(
+		'validates and deduplicates explicit selection $filePaths against audit upserts',
+		async ({ filePaths }) => {
+			const run = makeRun({
+				output: JSON.stringify({
+					schemas: { added: ['book'], updated: [], pruned: [] },
+					pieces: {
+						added: ['added.book.md'],
+						updated: ['updated.book.md'],
+						pruned: ['removed.book.md']
+					}
+				})
+			})
+			mocks.getWorkflowRun.mockResolvedValue(run)
+			mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) =>
+				name === 'PublishAudit' ? run : null
+			)
+			expect(await validateAuditForPublish(backend, run.id, filePaths)).toEqual({
+				ok: true,
+				filePaths: [...new Set(filePaths)]
+			})
+		}
+	)
+
+	test.each([null, 'not JSON', '"ok"', '{}', '{"pieces":{"added":[]}}'])(
+		'requires a valid audit report even for an empty selection: %s',
+		async (output) => {
+			const run = makeRun({ output })
+			mocks.getWorkflowRun.mockResolvedValue(run)
+			mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) =>
+				name === 'PublishAudit' ? run : null
+			)
+			expect(await validateAuditForPublish(backend, run.id, [])).toEqual({
+				ok: false,
+				reason: 'audit has no valid changes report; re-check before publishing'
+			})
+			// Omitted selection keeps accepting legacy completed audits.
+			expect(await validateAuditForPublish(backend, run.id)).toEqual({ ok: true })
+		}
+	)
+
+	test.each(['removed.book.md', 'book', './added.book.md', 'unknown.book.md'])(
+		'rejects a path outside exact audit additions/updates: %s',
+		async (filePath) => {
+			const run = makeRun({
+				output: JSON.stringify({
+					schemas: { added: ['book'], updated: [], pruned: [] },
+					pieces: { added: ['added.book.md'], updated: [], pruned: ['removed.book.md'] }
+				})
+			})
+			mocks.getWorkflowRun.mockResolvedValue(run)
+			mocks.getLatestWorkflowRun.mockImplementation(async (_backend, name) =>
+				name === 'PublishAudit' ? run : null
+			)
+			expect(await validateAuditForPublish(backend, run.id, [filePath])).toEqual({
+				ok: false,
+				reason: 'selected file paths must be pending additions or updates from this audit',
+				status: 400
+			})
+		}
+	)
+
 	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' })

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

@@ -30,11 +30,14 @@ export async function findInFlightPublishRun(
 	return null
 }
 
-export type AuditGuard = { ok: true } | { ok: false; reason: string }
+export type AuditGuard =
+	| { ok: true; filePaths?: string[] }
+	| { ok: false; reason: string; status?: 400 }
 
 export async function validateAuditForPublish(
 	backend: Pick<BackendSqlite, 'getWorkflowRun' | 'listWorkflowRuns'>,
-	auditRunId: unknown
+	auditRunId: unknown,
+	filePaths?: string[]
 ): Promise<AuditGuard> {
 	if (typeof auditRunId !== 'string' || auditRunId.length === 0) {
 		return { ok: false, reason: 'no audit run provided; check for changes before publishing' }
@@ -62,6 +65,22 @@ export async function validateAuditForPublish(
 		}
 	}
 
+	if (filePaths !== undefined) {
+		const diff = parsePiecesDiff(run.output)
+		if (!diff) {
+			return { ok: false, reason: 'audit has no valid changes report; re-check before publishing' }
+		}
+		const pending = new Set([...diff.pieces.added, ...diff.pieces.updated])
+		if (filePaths.some((filePath) => !pending.has(filePath))) {
+			return {
+				ok: false,
+				reason: 'selected file paths must be pending additions or updates from this audit',
+				status: 400
+			}
+		}
+		return { ok: true, filePaths: [...new Set(filePaths)] }
+	}
+
 	return { ok: true }
 }
 

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

@@ -7,6 +7,7 @@ import {
 	withPublishAdmission
 } from '$lib/server/workflow/publish.js'
 import { publishSpec } from '@luzzle/web.jobs/specs'
+import type { PublishPayload } from '@luzzle/web.jobs'
 
 export const POST: RequestHandler = async ({ request }) => {
 	try {
@@ -18,19 +19,34 @@ export const POST: RequestHandler = async ({ request }) => {
 			}
 
 			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 selectionProvided = Object.hasOwn(body ?? {}, 'filePaths')
+			let filePaths: string[] | undefined
+			if (selectionProvided) {
+				const requestedPaths = body.filePaths
+				if (!Array.isArray(requestedPaths)) {
+					return json({ message: 'filePaths must be an array of strings' }, { status: 400 })
 				}
+				if (!requestedPaths.every((filePath: unknown) => typeof filePath === 'string')) {
+					return json({ message: 'filePaths must be an array of strings' }, { status: 400 })
+				}
+				filePaths = requestedPaths
 			}
 
+			if (auditRunId || selectionProvided) {
+				const guard = await validateAuditForPublish(getOpenWorkflowBackend(), auditRunId, filePaths)
+				if (!guard.ok) {
+					return json({ message: guard.reason }, { status: guard.status ?? 412 })
+				}
+				filePaths = guard.filePaths
+			}
+
+			let bisync = body?.bisync === true
+			if (auditRunId) bisync = false
+			const input: PublishPayload = { bisync }
+			if (filePaths !== undefined) input.filePaths = filePaths
+
 			const openWorkflow = getOpenWorkflow()
-			const handle = await openWorkflow.runWorkflow(publishSpec, {
-				bisync: auditRunId ? false : bisync
-			})
+			const handle = await openWorkflow.runWorkflow(publishSpec, input)
 			return json({ jobId: handle.workflowRun.id })
 		})
 	} catch (error) {

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

@@ -18,6 +18,15 @@ vi.mock('$lib/server/workflow/index.js', () => ({
 	getOpenWorkflowBackend: vi.fn()
 }))
 
+const auditDiff = {
+	schemas: { added: ['book'], updated: [], pruned: [] },
+	pieces: {
+		added: ['added.book.md', '書 shelves/été?draft=1&tag=#%.book.md'],
+		updated: ['updated.book.md'],
+		pruned: ['removed.book.md']
+	}
+}
+
 function event(body: unknown = {}) {
 	return {
 		request: new Request('http://localhost/api/admin/publish', {
@@ -30,7 +39,7 @@ function event(body: unknown = {}) {
 describe('publish/audit admission', () => {
 	let queue: WorkflowQueue
 
-	async function enqueue(spec: { name: string }, input: { bisync: boolean }) {
+	async function enqueue(spec: { name: string }, input: { bisync: boolean; filePaths?: string[] }) {
 		vi.setSystemTime(new Date('2026-06-21T00:00:00Z'))
 		const workflowRun = await createPendingRun(queue.backend, spec.name, input)
 		return { workflowRun }
@@ -220,6 +229,201 @@ describe('publish/audit admission', () => {
 		expect(runWorkflow).toHaveBeenCalledTimes(1)
 	})
 
+	test.each([false, true])(
+		'persists omitted selection as publish-all with bisync=%s',
+		async (bisync) => {
+			const response = await publish(event({ bisync }))
+			expect(response.status).toBe(200)
+			const { jobId } = await response.json()
+			const run = await queue.backend.getWorkflowRun({ workflowRunId: jobId })
+			expect(run?.input).toEqual({ bisync })
+			expect(run?.input).not.toHaveProperty('filePaths')
+		}
+	)
+
+	test('keeps selection omitted for a checked publish-all with a legacy audit report', async () => {
+		const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', 'ok')
+		const response = await publish(event({ auditRunId: auditRun.id, bisync: true }))
+		expect(response.status).toBe(200)
+		const { jobId } = await response.json()
+		expect((await queue.backend.getWorkflowRun({ workflowRunId: jobId }))?.input).toEqual({
+			bisync: false
+		})
+	})
+
+	test.each([
+		{ label: 'pending upserts', diff: auditDiff },
+		{
+			label: 'removals only',
+			diff: {
+				schemas: { added: [], updated: [], pruned: [] },
+				pieces: { added: [], updated: [], pruned: ['removed.book.md'] }
+			}
+		},
+		{
+			label: 'schemas only',
+			diff: {
+				schemas: { added: [], updated: ['book'], pruned: [] },
+				pieces: { added: [], updated: [], pruned: [] }
+			}
+		},
+		{
+			label: 'no changes',
+			diff: {
+				schemas: { added: [], updated: [], pruned: [] },
+				pieces: { added: [], updated: [], pruned: [] }
+			}
+		}
+	])('preserves explicit [] with $label for global processing', async ({ diff }) => {
+		const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', diff)
+		const response = await publish(event({ auditRunId: auditRun.id, bisync: true, filePaths: [] }))
+		expect(response.status).toBe(200)
+		const { jobId } = await response.json()
+		expect((await queue.backend.getWorkflowRun({ workflowRunId: jobId }))?.input).toEqual({
+			bisync: false,
+			filePaths: []
+		})
+		expect(runWorkflow).toHaveBeenCalledWith(expect.objectContaining({ name: 'Publish' }), {
+			bisync: false,
+			filePaths: []
+		})
+	})
+
+	test('persists deduplicated exact added/updated paths without decoding special characters', async () => {
+		const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff)
+		const filePaths = [auditDiff.pieces.added[1], 'updated.book.md', 'added.book.md']
+		const response = await publish(
+			event({ auditRunId: auditRun.id, filePaths: [...filePaths, filePaths[0]] })
+		)
+		expect(response.status).toBe(200)
+		const { jobId } = await response.json()
+		expect((await queue.backend.getWorkflowRun({ workflowRunId: jobId }))?.input).toEqual({
+			bisync: false,
+			filePaths
+		})
+	})
+
+	test.each([
+		{ filePaths: null },
+		{ filePaths: 'added.book.md' },
+		{ filePaths: {} },
+		{ filePaths: true },
+		{ filePaths: 1 },
+		{ filePaths: [null] },
+		{ filePaths: [1] },
+		{ filePaths: [['added.book.md']] },
+		{ filePaths: ['added.book.md', {}] }
+	])('rejects malformed selection $filePaths before enqueue', async ({ filePaths }) => {
+		const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff)
+		const response = await publish(event({ auditRunId: auditRun.id, filePaths }))
+		expect(response.status).toBe(400)
+		expect(await response.json()).toEqual({ message: 'filePaths must be an array of strings' })
+		expect(runWorkflow).not.toHaveBeenCalled()
+		expect((await queue.backend.countWorkflowRuns()).pending).toBe(0)
+	})
+
+	test.each([
+		'',
+		'unknown.book.md',
+		'removed.book.md',
+		'book',
+		'/tmp/added.book.md',
+		'../added.book.md',
+		'./added.book.md',
+		'folder/../added.book.md',
+		encodeURIComponent(auditDiff.pieces.added[1])
+	])('rejects paths outside exact audit upsert membership: %s', async (filePath) => {
+		const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff)
+		const response = await publish(
+			event({ auditRunId: auditRun.id, filePaths: ['added.book.md', filePath] })
+		)
+		expect(response.status).toBe(400)
+		expect(await response.json()).toEqual({
+			message: 'selected file paths must be pending additions or updates from this audit'
+		})
+		expect(runWorkflow).not.toHaveBeenCalled()
+	})
+
+	test.each([{ filePaths: [] }, { filePaths: ['added.book.md'] }])(
+		'requires an audit for explicit selection $filePaths',
+		async ({ filePaths }) => {
+			const response = await publish(event({ filePaths, bisync: true }))
+			expect(response.status).toBe(412)
+			expect(await response.json()).toEqual({
+				message: 'no audit run provided; check for changes before publishing'
+			})
+			expect(runWorkflow).not.toHaveBeenCalled()
+		}
+	)
+
+	test.each([null, '', 42, 'missing-run'])(
+		'rejects invalid audit identity for explicit []: %s',
+		async (auditRunId) => {
+			expect((await publish(event({ auditRunId, filePaths: [] }))).status).toBe(412)
+			expect(runWorkflow).not.toHaveBeenCalled()
+		}
+	)
+
+	const invalidAuditOutputs: Parameters<typeof createCompletedRun>[2][] = [
+		null,
+		'ok',
+		{},
+		{ pieces: auditDiff.pieces },
+		{ ...auditDiff, pieces: { ...auditDiff.pieces, updated: [42] } }
+	]
+
+	test.each(invalidAuditOutputs)(
+		'rejects absent/legacy/malformed completed audit output for []: %j',
+		async (output) => {
+			const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', output)
+			const response = await publish(event({ auditRunId: auditRun.id, filePaths: [] }))
+			expect(response.status).toBe(412)
+			expect(await response.json()).toEqual({
+				message: 'audit has no valid changes report; re-check before publishing'
+			})
+			expect(runWorkflow).not.toHaveBeenCalled()
+		}
+	)
+
+	test.each([{ filePaths: [] }, { filePaths: ['added.book.md'] }])(
+		'rejects stale audit selection $filePaths and releases admission',
+		async ({ filePaths }) => {
+			const oldAudit = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff)
+			vi.setSystemTime(new Date('2026-06-20T01:00:00Z'))
+			const latestAudit = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff)
+			const rejected = await publish(event({ auditRunId: oldAudit.id, filePaths }))
+			expect(rejected.status).toBe(412)
+			expect(runWorkflow).not.toHaveBeenCalled()
+			expect((await publish(event({ auditRunId: latestAudit.id, filePaths }))).status).toBe(200)
+		}
+	)
+
+	test.each([{ filePaths: [] }, { filePaths: ['added.book.md'] }])(
+		'consumes the audit after selected publish $filePaths completes',
+		async ({ filePaths }) => {
+			const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff)
+			const response = await publish(event({ auditRunId: auditRun.id, filePaths }))
+			expect(response.status).toBe(200)
+			await completeNextRun(queue.backend)
+			const rejected = await publish(event({ auditRunId: auditRun.id, filePaths }))
+			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 active-run conflict precedence for an explicit selection', async () => {
+		const auditRun = await createCompletedRun(queue.backend, 'PublishAudit', auditDiff)
+		const accepted = await publish(event({ auditRunId: auditRun.id, filePaths: [] }))
+		const rejected = await publish(event({ auditRunId: auditRun.id, filePaths: ['unknown'] }))
+		expect(accepted.status).toBe(200)
+		expect(rejected.status).toBe(409)
+		expect(await rejected.json()).toEqual(await accepted.json())
+		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)

packages/web/jobs/src/types.ts

@@ -12,6 +12,8 @@ export interface AssetRecord {
 
 export interface PublishPayload {
 	bisync?: boolean
+	/** Omitted publishes all pending pieces; an empty list selects no upserts. */
+	filePaths?: string[]
 }
 
 export interface PublishPieceFailure {

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

@@ -162,7 +162,7 @@ describe('publication', () => {
 		const summary = emptyPiecesDiff()
 		summary.pieces.added.push('new.books.md')
 
-		const prepared = await publishPrepareStep.run(summary, ctx)
+		const prepared = await publishPrepareStep.run({ summary }, ctx)
 
 		expect(prepared).toEqual({
 			status: 'completed',
@@ -180,6 +180,57 @@ describe('publication', () => {
 		expect(summary.pieces.updated).toEqual([])
 	})
 
+	test.each([
+		{ filePaths: undefined, added: ['new.books.md'], updated: ['changed.books.md'] },
+		{ filePaths: [], added: [], updated: [] },
+		{
+			filePaths: ['changed.books.md', 'changed.books.md', 'unchanged.books.md', 'missing.books.md'],
+			added: [],
+			updated: ['changed.books.md'],
+		},
+	])('prepare intersects fresh pending work with selection $filePaths, retaining global work', async ({
+		filePaths,
+		added,
+		updated,
+	}) => {
+		await indexPiece('new.books.md', 'new-hash')
+		await indexPiece('changed.books.md', 'current-hash')
+		await indexPiece('unchanged.books.md', 'same-hash')
+		await indexPiece('gone.books.md', 'gone-hash')
+		await publishMetadata('changed.books.md', 'old-hash', 123)
+		await publishMetadata('unchanged.books.md', 'same-hash', 456)
+		await publishMetadata('gone.books.md', 'gone-hash', 789)
+		await db.deleteFrom('pieces_items').where('file_path', '=', 'gone.books.md').execute()
+		const summary = {
+			schemas: { added: ['books'], updated: ['films'], pruned: ['old'] },
+			pieces: {
+				added: ['changed.books.md', 'missing.books.md'],
+				updated: ['unchanged.books.md', 'new.books.md'],
+				pruned: ['already-pruned.books.md', 'gone.books.md'],
+			},
+		}
+		const before = await db.selectFrom('web_pieces').selectAll().orderBy('file_path').execute()
+
+		const result = await publishPrepareStep.run({ summary, filePaths }, ctx)
+
+		expect(result.status).toBe('completed')
+		if (result.status !== 'completed') throw new Error('Expected a completed plan')
+		expect(result.value).toEqual({
+			pieces: [
+				...(updated.length ? [{ filePath: 'changed.books.md', contentHash: 'current-hash' }] : []),
+				...(added.length ? [{ filePath: 'new.books.md', contentHash: 'new-hash' }] : []),
+			],
+			summary: {
+				schemas: summary.schemas,
+				pieces: { added, updated, pruned: ['already-pruned.books.md', 'gone.books.md'] },
+			},
+		})
+		expect(await db.selectFrom('web_pieces').selectAll().orderBy('file_path').execute()).toEqual(
+			before
+		)
+		expect(summary.pieces.added).toEqual(['changed.books.md', 'missing.books.md'])
+	})
+
 	test('complete advances healthy pieces only, using their captured hashes', async () => {
 		await indexPiece('healthy.books.md', 'even-newer-hash')
 		await indexPiece('failed.books.md', 'failed-hash')
@@ -338,7 +389,7 @@ describe('publication', () => {
 	})
 
 	test('complete and prepare accept an empty archive', async () => {
-		expect(await publishPrepareStep.run(emptyPiecesDiff(), ctx)).toEqual({
+		expect(await publishPrepareStep.run({ summary: emptyPiecesDiff() }, ctx)).toEqual({
 			status: 'completed',
 			value: { pieces: [], summary: emptyPiecesDiff() },
 		})

packages/web/worker/src/steps/publish-prepare.ts

@@ -8,14 +8,35 @@ export interface PublishPlan {
 	summary: PiecesDiff
 }
 
-export const publishPrepareStep: Step<PiecesDiff, PublishPlan> = {
+export interface PublishPrepareInput {
+	summary: PiecesDiff
+	filePaths?: string[]
+}
+
+export const publishPrepareStep: Step<PublishPrepareInput, PublishPlan> = {
 	name: 'publish.prepare',
-	async run(summary, { db, config, logger }) {
+	async run({ summary: indexSummary, filePaths }, { db, config, logger }) {
 		const pending = await getPendingPublication(db, config)
-		logger.info('publish.prepare complete', { count: pending.pieces.length })
-		return completed({
-			pieces: pending.pieces,
-			summary: includePendingPieces(summary, pending.diff),
-		})
+		let pieces = pending.pieces
+		if (filePaths !== undefined) {
+			const selectedPaths = new Set(filePaths)
+			pieces = pending.pieces.filter((piece) => selectedPaths.has(piece.filePath))
+		}
+		const targetPaths = new Set(pieces.map((piece) => piece.filePath))
+
+		// Core sync is global; only authoritative pending targets count as upserts.
+		const globalChanges: PiecesDiff = {
+			schemas: indexSummary.schemas,
+			pieces: { added: [], updated: [], pruned: indexSummary.pieces.pruned },
+		}
+		const selectedChanges: PiecesDiff['pieces'] = {
+			added: pending.diff.added.filter((filePath) => targetPaths.has(filePath)),
+			updated: pending.diff.updated.filter((filePath) => targetPaths.has(filePath)),
+			pruned: pending.diff.pruned,
+		}
+		const publicationSummary = includePendingPieces(globalChanges, selectedChanges)
+
+		logger.info('publish.prepare complete', { count: pieces.length })
+		return completed({ pieces, summary: publicationSummary })
 	},
 }

packages/web/worker/src/workflows/publish-recovery.test.ts

@@ -7,14 +7,16 @@ import { validateConfig } from '@luzzle/web.config'
 import { createAppDb, runWebMigrations } from '@luzzle/web.db'
 import type { PublishPayload, PublishResult } from '@luzzle/web.jobs'
 import { publishSpec } from '@luzzle/web.jobs/specs'
-import type { Workflow } from 'openworkflow'
+import { OpenWorkflow, type Workflow } from 'openworkflow'
+import { BackendSqlite } from 'openworkflow/sqlite'
 import { completed, type StepResult } from '../core/step.js'
 import { RcloneClient } from '../services/rclone.js'
+import { createOpenWorkflowLogger } from '../services/logger.js'
 import { setWorkerContext, type WorkerContext } from '../services/context.js'
 import { getPendingPublication } from '../services/publication.js'
 import { luzzleSyncStep } from '../steps/luzzle-sync.js'
 import { luzzleAuditStep } from '../steps/luzzle-audit.js'
-import { publishPrepareStep } from '../steps/publish-prepare.js'
+import { publishPrepareStep, type PublishPlan } from '../steps/publish-prepare.js'
 import { webSyncStep } from '../steps/web-sync/index.js'
 import type { AssetsGenerateInput, AssetsGenerateResult } from '../steps/assets-generate.js'
 import { registerPublishWorkflow } from './publish.js'
@@ -160,6 +162,77 @@ describe('publish hash recovery with a real archive and SQLite', () => {
 		return { result, phaseResults }
 	}
 
+	// Real SDK queue/checkpoint storage; only external delivery and transforms are stubbed.
+	async function withQueue(
+		run: (
+			client: OpenWorkflow,
+			worker: ReturnType<OpenWorkflow['newWorker']>,
+			backend: BackendSqlite
+		) => Promise<void>
+	) {
+		const backend = BackendSqlite.connect(path.join(directory, 'queue.sqlite'))
+		const client = new OpenWorkflow({ backend, logger: createOpenWorkflowLogger(ctx.logger) })
+		client.implementWorkflow(publishSpec, workflow)
+		const worker = client.newWorker({ concurrency: 1 })
+		try {
+			await run(client, worker, backend)
+		} finally {
+			await worker.stop()
+			await backend.stop()
+		}
+	}
+
+	async function finishQueued(
+		worker: ReturnType<OpenWorkflow['newWorker']>,
+		backend: BackendSqlite,
+		id: string
+	) {
+		await vi.waitFor(
+			async () => {
+				await worker.tick()
+				expect((await backend.getWorkflowRun({ workflowRunId: id }))?.status).toBe('completed')
+			},
+			{ timeout: 5000 }
+		)
+		const run = await backend.getWorkflowRun({ workflowRunId: id })
+		return run!.output as unknown as PublishResult
+	}
+
+	async function publicRows(filePath: string) {
+		return {
+			pieces: await ctx.db
+				.selectFrom('web_pieces')
+				.selectAll()
+				.where('file_path', '=', filePath)
+				.execute(),
+			assets: await ctx.db
+				.selectFrom('web_pieces_assets')
+				.selectAll()
+				.where('piece_file_path', '=', filePath)
+				.execute(),
+		}
+	}
+
+	async function seedAsset(filePath: string) {
+		const piece = await ctx.db
+			.selectFrom('web_pieces')
+			.select('key')
+			.where('file_path', '=', filePath)
+			.executeTakeFirstOrThrow()
+		await ctx.db
+			.insertInto('web_pieces_assets')
+			.values({
+				piece_file_path: filePath,
+				piece_key: piece.key,
+				piece_asset_path: '.assets/cover.jpg',
+				asset_key: `${piece.key}-cover`,
+				transformation: 'image',
+				asset_path: `${piece.key}/cover.webp`,
+				mime_type: 'image/webp',
+			})
+			.execute()
+	}
+
 	async function hashes(filePath: string) {
 		const indexed = await ctx.db
 			.selectFrom('pieces_cache')
@@ -183,6 +256,219 @@ describe('publish hash recovery with a real archive and SQLite', () => {
 		return valueOf(await luzzleAuditStep.run(undefined, ctx))
 	}
 
+	test.each(['added', 'updated'] as const)(
+		'queued selection publishes only the pending %s target, preserving unselected rows and pruning globally',
+		async (action) => {
+			const removed = 'removed.books.md'
+			const unchanged = 'unchanged.books.md'
+			if (action === 'updated') await writePiece(healthy, 'Original selected')
+			await writePiece(broken, 'Original unselected')
+			await writePiece(removed, 'Removed globally')
+			await writePiece(unchanged, 'Unchanged')
+			await withQueue(async (client, worker, backend) => {
+				const baseline = await client.runWorkflow(publishSpec, {})
+				await finishQueued(worker, backend, baseline.workflowRun.id)
+				await seedAsset(broken)
+				await seedAsset(removed)
+				const unselectedBefore = await publicRows(broken)
+				const unchangedBefore = await publicRows(unchanged)
+				await writePiece(healthy, 'Selected current')
+				await writePiece(broken, 'Unselected current')
+				await writePiece('new-unselected.books.md', 'New unselected')
+				await rm(path.join(ctx.config.storage.root, removed))
+				mocks.assets.mockClear()
+				mocks.cdn.mockClear()
+				mocks.cache.mockClear()
+
+				const selected = await client.runWorkflow(publishSpec, {
+					filePaths: [healthy, healthy, unchanged, removed, 'missing.books.md'],
+				})
+				const result = await finishQueued(worker, backend, selected.workflowRun.id)
+
+				expect(result).toEqual({
+					schemas: emptyPiecesDiff().schemas,
+					pieces: {
+						added: action === 'added' ? [healthy] : [],
+						updated: action === 'updated' ? [healthy] : [],
+						pruned: [removed],
+					},
+					failedPieces: [],
+				})
+				expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: [healthy] }, ctx)
+				expect(mocks.cdn).toHaveBeenCalledOnce()
+				expect(mocks.cache).toHaveBeenCalledOnce()
+				expect(await publicRows(broken)).toEqual(unselectedBefore)
+				expect(await publicRows(unchanged)).toEqual(unchangedBefore)
+				expect(await publicRows(removed)).toEqual({ pieces: [], assets: [] })
+				expect(await publicRows('new-unselected.books.md')).toEqual({ pieces: [], assets: [] })
+				const state = await hashes(healthy)
+				expect(state.published).toBe(state.indexed)
+				expect(state.title).toBe('Selected current')
+				expect((await audit()).pieces).toEqual({
+					added: ['new-unselected.books.md'],
+					updated: [broken],
+					pruned: [],
+				})
+			})
+		}
+	)
+
+	test.each([false, true])(
+		'queued explicit empty selection keeps schema housekeeping and global removals (%s)',
+		async (removePiece) => {
+			const removed = 'removed.books.md'
+			for (const filePath of [healthy, broken, removed]) {
+				await writePiece(filePath, `Original ${filePath}`)
+			}
+			await publish('baseline-empty-selection')
+			await seedAsset(broken)
+			await seedAsset(removed)
+			const healthyBefore = await publicRows(healthy)
+			const brokenBefore = await publicRows(broken)
+			const removedBefore = await publicRows(removed)
+			await writePiece(healthy, 'Pending selected none')
+			await writePiece(broken, 'Pending unselected')
+			if (removePiece) await rm(path.join(ctx.config.storage.root, removed))
+			await writeFile(
+				path.join(ctx.config.storage.root, '.luzzle', 'schemas', 'books.json'),
+				JSON.stringify({
+					type: 'object',
+					title: 'books',
+					description: 'Updated globally',
+					properties: { title: { type: 'string' } },
+					required: ['title'],
+					additionalProperties: true,
+				})
+			)
+			mocks.assets.mockClear()
+			mocks.cdn.mockClear()
+			mocks.cache.mockClear()
+
+			await withQueue(async (client, worker, backend) => {
+				const empty = await client.runWorkflow(publishSpec, { filePaths: [] })
+				const result = await finishQueued(worker, backend, empty.workflowRun.id)
+				expect(result).toEqual({
+					schemas: { added: [], updated: ['books'], pruned: [] },
+					pieces: { added: [], updated: [], pruned: removePiece ? [removed] : [] },
+					failedPieces: [],
+				})
+				expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: [] }, ctx)
+				expect(mocks.cdn).toHaveBeenCalledOnce()
+				expect(mocks.cache).toHaveBeenCalledOnce()
+				expect(await publicRows(healthy)).toEqual(healthyBefore)
+				expect(await publicRows(broken)).toEqual(brokenBefore)
+				expect(await publicRows(removed)).toEqual(
+					removePiece ? { pieces: [], assets: [] } : removedBefore
+				)
+				expect((await audit()).pieces).toEqual({ added: [], updated: batch, pruned: [] })
+			})
+		}
+	)
+
+	test('queued interrupted delivery replays the captured selected plan without widening membership or hashes', async () => {
+		const late = 'late.books.md'
+		await writePiece(healthy, 'Original selected')
+		await writePiece(broken, 'Original unselected')
+		await publish('baseline-interruption')
+		await seedAsset(broken)
+		const unselectedBefore = await publicRows(broken)
+		await writePiece(healthy, 'Selected revision two')
+		await writePiece(broken, 'Unselected revision two')
+		mocks.assets.mockClear()
+		mocks.cache.mockRejectedValueOnce(new Error('interrupted delivery'))
+
+		await withQueue(async (client, worker, backend) => {
+			const handle = await client.runWorkflow(publishSpec, { filePaths: [healthy, late] })
+			const id = handle.workflowRun.id
+			await worker.tick()
+			await vi.waitFor(
+				async () => {
+					const run = await backend.getWorkflowRun({ workflowRunId: id })
+					expect(run?.status).toBe('pending')
+					expect(run?.attempts).toBe(1)
+				},
+				{ timeout: 5000 }
+			)
+			const attempts = (await backend.listStepAttempts({ workflowRunId: id })).data
+			const plan = attempts.find((attempt) => attempt.stepName === 'publish.prepare')!
+				.output as unknown as PublishPlan
+			expect(plan.pieces.map((piece) => piece.filePath)).toEqual([healthy])
+			const capturedHash = plan.pieces[0]!.contentHash
+			await writePiece(healthy, 'Selected revision three')
+			await writePiece(late, 'Selected but not pending when prepared')
+			await luzzleSyncStep.run(undefined, ctx)
+
+			const result = await finishQueued(worker, backend, id)
+
+			expect(result).toEqual({ ...plan.summary, failedPieces: [] })
+			expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: [healthy] }, ctx)
+			expect(await publicRows(broken)).toEqual(unselectedBefore)
+			expect(await publicRows(late)).toEqual({ pieces: [], assets: [] })
+			const state = await hashes(healthy)
+			expect(state.published).toBe(capturedHash)
+			expect(state.indexed).not.toBe(capturedHash)
+			expect(state.title).toBe('Selected revision two')
+			const retried = (await backend.listStepAttempts({ workflowRunId: id })).data
+			expect(retried.filter((attempt) => attempt.stepName === 'publish.prepare')).toHaveLength(1)
+			expect(retried.find((attempt) => attempt.stepName === 'publish.prepare')!.output).toEqual(plan)
+			expect(retried.filter((attempt) => attempt.stepName === 'cache.purge')).toHaveLength(2)
+			expect((await audit()).pieces).toEqual({ added: [late], updated: batch, pruned: [] })
+		})
+	})
+
+	test('queued partial-failure rerun retains selected hashes and leaves unselected pending work untouched', async () => {
+		const unselected = 'unselected.books.md'
+		const late = 'late.books.md'
+		for (const filePath of [...batch, unselected]) await writePiece(filePath, `Original ${filePath}`)
+		await publish('baseline-partial-selection')
+		await seedAsset(unselected)
+		const unselectedBefore = await publicRows(unselected)
+		for (const filePath of [...batch, unselected]) await writePiece(filePath, `Changed ${filePath}`)
+		const failedPieces = [{ filePath: broken, message: 'selected transform failed' }]
+		mocks.assets.mockClear()
+		mocks.assets.mockResolvedValueOnce(completed({ failedPieces }))
+
+		await withQueue(async (client, worker, backend) => {
+			const partial = await client.runWorkflow(publishSpec, { filePaths: [...batch, late] })
+			const partialResult = await finishQueued(worker, backend, partial.workflowRun.id)
+			expect(partialResult.failedPieces).toEqual(failedPieces)
+			expect((await hashes(healthy)).published).toBe((await hashes(healthy)).indexed)
+			expect((await hashes(broken)).published).toBeNull()
+			expect(await publicRows(unselected)).toEqual(unselectedBefore)
+			const attempts = (
+				await backend.listStepAttempts({ workflowRunId: partial.workflowRun.id })
+			).data
+			const plan = attempts.find((attempt) => attempt.stepName === 'publish.prepare')!
+				.output as unknown as PublishPlan
+			expect(plan.pieces.map((piece) => piece.filePath)).toEqual(batch)
+			await writePiece(healthy, 'Newer than captured selected hash')
+			await writePiece(late, 'Late selected addition')
+			await luzzleSyncStep.run(undefined, ctx)
+			mocks.assets.mockClear()
+
+			const retry = await client.rerunWorkflowRun(partial.workflowRun.id, {
+				fromStep: 'assets.generate',
+			})
+			const result = await finishQueued(worker, backend, retry.id)
+
+			expect(result).toEqual({ ...plan.summary, failedPieces: [] })
+			expect(mocks.assets).toHaveBeenCalledExactlyOnceWith({ filePaths: batch }, ctx)
+			for (const piece of plan.pieces) {
+				expect((await hashes(piece.filePath)).published).toBe(piece.contentHash)
+			}
+			expect((await hashes(healthy)).indexed).not.toBe((await hashes(healthy)).published)
+			expect(await publicRows(unselected)).toEqual(unselectedBefore)
+			expect(await publicRows(late)).toEqual({ pieces: [], assets: [] })
+			const retried = (await backend.listStepAttempts({ workflowRunId: retry.id })).data
+			expect(retried.find((attempt) => attempt.stepName === 'publish.prepare')!.output).toEqual(plan)
+			expect((await audit()).pieces).toEqual({
+				added: [late],
+				updated: [healthy, unselected],
+				pruned: [],
+			})
+		})
+	})
+
 	test('a fresh publish rediscovers new and updated pieces after the sync result is discarded', async () => {
 		await writePiece(healthy, 'Original')
 		await publish('baseline')
@@ -235,7 +521,7 @@ describe('publish hash recovery with a real archive and SQLite', () => {
 	test('metadata sync alone never records publication success for either new or existing rows', async () => {
 		await writePiece(healthy, 'Original')
 		const initialDiff = valueOf(await luzzleSyncStep.run(undefined, ctx))
-		const initialPlan = valueOf(await publishPrepareStep.run(initialDiff, ctx))
+		const initialPlan = valueOf(await publishPrepareStep.run({ summary: initialDiff }, ctx))
 		await webSyncStep.run({ filePaths: initialPlan.pieces.map((piece) => piece.filePath) }, ctx)
 		expect(await hashes(healthy)).toMatchObject({ published: null, lastPublishedAt: null })
 		expect((await getPendingPublication(ctx.db, ctx.config)).pieces).toEqual(initialPlan.pieces)
@@ -244,7 +530,7 @@ describe('publish hash recovery with a real archive and SQLite', () => {
 		const original = await hashes(healthy)
 		await writePiece(healthy, 'Metadata changed')
 		const diff = valueOf(await luzzleSyncStep.run(undefined, ctx))
-		const plan = valueOf(await publishPrepareStep.run(diff, ctx))
+		const plan = valueOf(await publishPrepareStep.run({ summary: diff }, ctx))
 		await webSyncStep.run({ filePaths: plan.pieces.map((piece) => piece.filePath) }, ctx)
 		const state = await hashes(healthy)
 		expect(state.title).toBe('Metadata changed')

packages/web/worker/src/workflows/publish.test.ts

@@ -186,6 +186,57 @@ describe('workflows/publish', () => {
 		)
 	})
 
+	test.each([
+		{ filePaths: undefined },
+		{ filePaths: [] },
+		{ filePaths: ['healthy.books.md'] },
+	])(
+		'passes selection $filePaths to prepare without narrowing global sync or delivery',
+		async ({ filePaths }) => {
+			const { result } = await runPublish({ filePaths })
+
+			expect(mocks.luzzleSync.run).toHaveBeenCalledExactlyOnceWith(undefined, expect.anything())
+			expect(mocks.publishPrepare.run).toHaveBeenCalledExactlyOnceWith(
+				{ summary, filePaths },
+				expect.anything()
+			)
+			expect(mocks.cdnSync.run).toHaveBeenCalledExactlyOnceWith(undefined, expect.anything())
+			expect(mocks.cachePurge.run).toHaveBeenCalledExactlyOnceWith(undefined, expect.anything())
+			expect(result).toEqual({ ...summary, failedPieces: [] })
+		}
+	)
+
+	test('uses only the saved selected plan and hashes when retrying, not the input or global diff', async () => {
+		const plan = {
+			pieces: [{ filePath: 'healthy.books.md', contentHash: 'captured-hash' }],
+			summary: {
+				schemas: summary.schemas,
+				pieces: { added: ['healthy.books.md'], updated: [], pruned: summary.pieces.pruned },
+			},
+		}
+		const { result } = await runPublish(
+			{ filePaths: ['healthy.books.md', 'broken.books.md'] },
+			new Map<string, unknown>([['luzzle.sync', summary], ['publish.prepare', plan]])
+		)
+
+		expect(mocks.luzzleSync.run).not.toHaveBeenCalled()
+		expect(mocks.publishPrepare.run).not.toHaveBeenCalled()
+		expect(mocks.webSync.run).toHaveBeenCalledExactlyOnceWith(
+			{ filePaths: ['healthy.books.md'] },
+			expect.anything()
+		)
+		expect(mocks.assetsGenerate.run).toHaveBeenCalledExactlyOnceWith(
+			{ filePaths: ['healthy.books.md'] },
+			expect.anything()
+		)
+		expect(mocks.publishComplete.run).toHaveBeenCalledExactlyOnceWith(
+			{ pieces: plan.pieces, failedPieces: [] },
+			expect.anything()
+		)
+		expect(result).toEqual({ ...plan.summary, failedPieces: [] })
+		expectHealthyPhasesComplete()
+	})
+
 	test.each(['cdn.sync', 'cache.purge'])(
 		'keeps a global %s rejection as a real publish failure',
 		async (phase) => {

packages/web/worker/src/workflows/publish.ts

@@ -36,7 +36,7 @@ export function registerPublishWorkflow(): void {
 			jobId,
 			progress,
 			publishPrepareStep,
-			summary ?? emptyPiecesDiff()
+			{ summary: summary ?? emptyPiecesDiff(), filePaths: input.filePaths }
 		)
 		if (!plan) throw new Error('publish.prepare did not return a plan')
 		const changedPaths = plan.pieces.map((piece) => piece.filePath)