.changeset/native-maintenance.md

@@ -0,0 +1,9 @@
+---
+'@when/web': patch
+'@when/worker': patch
+'@when/cli': patch
+---
+
+Run refresh, pending expiry and calendar repair as awaited native UTC cron workflows, with independent startup passes and pause-preserving schedule persistence. Remove application maintenance timers and coalescing; one execution slot serializes passes, so slow maintenance can delay other jobs. Stop the worker before closing queue and application databases.
+
+Document attempt-level email outcomes, retained personal data after purge failure, sensitive queue backups, and the drained grouped rollout with paired SQLite backup/restore checks. Deploy web, worker and CLI together only after all old work drains; rollback requires proven compatibility or explicit approval for paired restore and loss of later writes.

apps/cli/src/utils/worker.test.ts

@@ -0,0 +1,77 @@
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { DatabaseSync } from 'node:sqlite';
+import { afterEach, expect, test, vi } from 'vitest';
+import type { WhenConfiguration } from '@when/config';
+import * as jobs from '@when/jobs';
+import { runInWorker } from './worker.ts';
+
+afterEach(() => vi.restoreAllMocks());
+
+test('CLI initializes only a producer and gives each probe its own random enqueue key', async () => {
+	const directory = await mkdtemp(join(tmpdir(), 'when-cli-producer-'));
+	const dbPath = join(directory, 'queue.sqlite');
+	const config = {
+		database: { queue: dbPath },
+		url: { worker: 'http://worker.example.com:9000/' }
+	} as WhenConfiguration;
+	const client = jobs.initOpenWorkflow({ dbPath });
+	const init = vi.spyOn(jobs, 'initOpenWorkflow');
+	const newWorker = vi.spyOn(client, 'newWorker');
+	const implement = vi.spyOn(client, 'implementWorkflow');
+	const schedule = vi.spyOn(client, 'scheduleWorkflow');
+	const result = vi.fn().mockResolvedValue('authenticated');
+	const enqueue = client.runWorkflow.bind(client);
+	const run = vi.spyOn(client, 'runWorkflow').mockImplementation(async (...args) => {
+		const handle = await enqueue(...args);
+		vi.spyOn(handle, 'result').mockImplementation(result);
+		return handle;
+	});
+	const fetch = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(null));
+	const uuid = vi.spyOn(crypto, 'randomUUID');
+	uuid.mockReturnValueOnce('00000000-0000-4000-8000-000000000001');
+	uuid.mockReturnValueOnce('00000000-0000-4000-8000-000000000002');
+	try {
+		for (let i = 1; i <= 2; i++) {
+			const probeResult = await runInWorker(config, jobs.testProvider, { name: 'work' });
+			expect(probeResult).toEqual({
+				ok: true,
+				value: 'authenticated'
+			});
+			expect(run).toHaveBeenNthCalledWith(
+				i,
+				jobs.testProvider,
+				{ name: 'work' },
+				{ idempotencyKey: `00000000-0000-4000-8000-00000000000${i}` }
+			);
+		}
+		expect(jobs.testProvider.name).toBe('test-provider');
+		expect(init).toHaveBeenCalledTimes(2);
+		expect(init).toHaveBeenCalledWith({ dbPath });
+		expect(fetch).toHaveBeenCalledWith('http://worker.example.com:9000/healthz');
+		expect(result).toHaveBeenCalledTimes(2);
+		expect(result).toHaveBeenCalledWith({ timeoutMs: 30_000 });
+		expect(newWorker).not.toHaveBeenCalled();
+		expect(implement).not.toHaveBeenCalled();
+		expect(schedule).not.toHaveBeenCalled();
+		const raw = new DatabaseSync(dbPath, { readOnly: true });
+		try {
+			expect(raw.prepare('SELECT COUNT(*) AS count FROM workflow_schedules').get()).toEqual({
+				count: 0
+			});
+			expect(raw.prepare('SELECT workflow_name, status, input FROM workflow_runs').all()).toEqual(
+				Array.from({ length: 2 }, () => ({
+					workflow_name: 'test-provider',
+					status: 'pending',
+					input: '{"name":"work"}'
+				}))
+			);
+		} finally {
+			raw.close();
+		}
+	} finally {
+		await jobs.closeOpenWorkflow();
+		await rm(directory, { recursive: true, force: true });
+	}
+});

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

@@ -1,15 +1,29 @@
 import { beforeEach, describe, expect, test, vi } from 'vitest';
 import { openDb, runMigrations } from '@when/db';
+import { OpenWorkflow } from 'openworkflow';
+import { BackendSqlite } from 'openworkflow/sqlite';
 
-const runWorkflow = vi.fn();
-
-vi.mock('@when/jobs', () => ({
-	getOpenWorkflow: () => ({ runWorkflow }),
-	reconcileAppointment: { spec: { name: 'reconcile-appointment' } }
+const { runWorkflow, getOpenWorkflow } = vi.hoisted(() => ({
+	runWorkflow: vi.fn(),
+	getOpenWorkflow: vi.fn()
 }));
 
-import { enqueueAppointmentReconciliation } from './index';
-import { reconcileAppointment } from '@when/jobs';
+vi.mock('@when/jobs', async (importOriginal) => {
+	const actual = await importOriginal<typeof import('@when/jobs')>();
+	return { ...actual, getOpenWorkflow };
+});
+
+import {
+	enqueueAppointmentReconciliation,
+	enqueueCalendarSync,
+	enqueuePurgeAppointment
+} from './index';
+import { reconcileAppointment, syncCalendars, purgeAppointment } from '@when/jobs';
+
+beforeEach(() => {
+	runWorkflow.mockReset();
+	getOpenWorkflow.mockReturnValue({ runWorkflow });
+});
 
 async function seed() {
 	const db = openDb(':memory:');
@@ -33,9 +47,7 @@ async function seed() {
 }
 
 describe('enqueueAppointmentReconciliation', () => {
-	beforeEach(() => runWorkflow.mockReset());
-
-	test('snapshots the appointment and runs the workflow', async () => {
+	test('returns the appointment and enqueues the public workflow contract', async () => {
 		const db = await seed();
 
 		const result = await enqueueAppointmentReconciliation(db, 'appt-1', 'confirmed');
@@ -48,6 +60,7 @@ describe('enqueueAppointmentReconciliation', () => {
 			},
 			{ idempotencyKey: 'appt-1:confirmed:0' }
 		);
+		expect(reconcileAppointment.name).toBe('reconcile-appointment');
 		expect(result.id).toBe('appt-1');
 
 		await db.destroy();
@@ -63,10 +76,133 @@ describe('enqueueAppointmentReconciliation', () => {
 
 		await enqueueAppointmentReconciliation(db, 'appt-1', 'rescheduled-by-guest');
 
-		expect(runWorkflow).toHaveBeenCalledWith(reconcileAppointment, expect.anything(), {
-			idempotencyKey: 'appt-1:rescheduled-by-guest:2'
-		});
+		expect(runWorkflow).toHaveBeenCalledWith(
+			reconcileAppointment,
+			{ appointmentId: 'appt-1', emailKind: 'rescheduled-by-guest' },
+			{ idempotencyKey: 'appt-1:rescheduled-by-guest:2' }
+		);
 
 		await db.destroy();
 	});
 });
+
+describe('unchanged producer inputs and keys', () => {
+	test('forwards the cleanup URL and keys email-free reconciliation by sequence', async () => {
+		const db = await seed();
+		await db.updateTable('appointments').set({ ics_sequence: 3 }).execute();
+
+		await enqueueAppointmentReconciliation(
+			db,
+			'appt-1',
+			undefined,
+			'https://video.example.com/obsolete-room'
+		);
+
+		expect(runWorkflow).toHaveBeenCalledWith(
+			reconcileAppointment,
+			{
+				appointmentId: 'appt-1',
+				emailKind: undefined,
+				cleanupVideoChatUrl: 'https://video.example.com/obsolete-room'
+			},
+			{ idempotencyKey: 'appt-1:sync:3' }
+		);
+		await db.destroy();
+	});
+
+	test('sorts purge IDs for the key without changing row order or remote identifiers', async () => {
+		const rows = [
+			{ id: 'z', externalEventId: 'remote-z', externalCalendarId: 'work' },
+			{ id: 'a', externalEventId: null, externalCalendarId: null }
+		];
+
+		await enqueuePurgeAppointment(rows);
+		await enqueuePurgeAppointment([...rows].reverse());
+
+		expect(purgeAppointment.name).toBe('purge-appointment');
+		expect(runWorkflow).toHaveBeenNthCalledWith(
+			1,
+			purgeAppointment,
+			{ rows },
+			{ idempotencyKey: 'purge:a,z' }
+		);
+		expect(runWorkflow).toHaveBeenNthCalledWith(
+			2,
+			purgeAppointment,
+			{ rows: [...rows].reverse() },
+			{ idempotencyKey: 'purge:a,z' }
+		);
+		expect(rows.map((row) => row.id)).toEqual(['z', 'a']);
+	});
+
+	test('sync enqueues an empty input with a fresh random key and does not wait for the scan', async () => {
+		const result = vi.fn();
+		runWorkflow.mockResolvedValue({ result });
+		const uuid = vi.spyOn(crypto, 'randomUUID');
+		uuid.mockReturnValueOnce('00000000-0000-4000-8000-000000000001');
+		uuid.mockReturnValueOnce('00000000-0000-4000-8000-000000000002');
+		try {
+			const firstEnqueueResult = await enqueueCalendarSync();
+			expect(firstEnqueueResult).toBeUndefined();
+			const secondEnqueueResult = await enqueueCalendarSync();
+			expect(secondEnqueueResult).toBeUndefined();
+			expect(syncCalendars.name).toBe('sync-calendars');
+			for (let i = 1; i <= 2; i++) {
+				expect(runWorkflow).toHaveBeenNthCalledWith(
+					i,
+					syncCalendars,
+					{},
+					{
+						idempotencyKey: `sync-calendars:00000000-0000-4000-8000-00000000000${i}`
+					}
+				);
+			}
+			expect(result).not.toHaveBeenCalled();
+		} finally {
+			uuid.mockRestore();
+		}
+	});
+
+	test('an identical enqueue reuses terminal history within the engine dedup window', async () => {
+		const db = await seed();
+		const backend = BackendSqlite.connect(':memory:');
+		const client = new OpenWorkflow({ backend });
+		const previousHandler = vi.fn(() => 'reconciled' as const);
+		const changedHandler = vi.fn(() => 'reconciled' as const);
+		client.implementWorkflow(reconcileAppointment, previousHandler);
+		getOpenWorkflow.mockReturnValue(client);
+		const enqueue = vi.spyOn(client, 'runWorkflow');
+		const worker = client.newWorker({ concurrency: 1 });
+		try {
+			await enqueueAppointmentReconciliation(db, 'appt-1', 'confirmed');
+			const first = await enqueue.mock.results[0].value;
+			const processedRuns = await worker.tick();
+			expect(processedRuns).toBe(1);
+			await worker.stop();
+			const reconciliationResult = await first.result();
+			expect(reconciliationResult).toBe('reconciled');
+
+			const restartedClient = new OpenWorkflow({ backend });
+			restartedClient.implementWorkflow(reconcileAppointment, changedHandler);
+			getOpenWorkflow.mockReturnValue(restartedClient);
+			const repeatedEnqueue = vi.spyOn(restartedClient, 'runWorkflow');
+			await enqueueAppointmentReconciliation(db, 'appt-1', 'confirmed');
+			const repeated = await repeatedEnqueue.mock.results[0].value;
+			expect(repeated.workflowRun.id).toBe(first.workflowRun.id);
+			expect(repeated.workflowRun.status).toBe('completed');
+			const restartedWorker = restartedClient.newWorker({ concurrency: 1 });
+			try {
+				const restartedProcessedRuns = await restartedWorker.tick();
+				expect(restartedProcessedRuns).toBe(0);
+			} finally {
+				await restartedWorker.stop();
+			}
+			expect(previousHandler).toHaveBeenCalledTimes(1);
+			expect(changedHandler).not.toHaveBeenCalled();
+		} finally {
+			await worker.stop();
+			await backend.stop();
+			await db.destroy();
+		}
+	});
+});

apps/worker/README.md

@@ -10,7 +10,9 @@ coordinates calendar/video updates, renders emails, sends them over SMTP, and
 records outcomes in the appointment action log and service status. It also handles
 purge, calendar-repair requests, video-link generation, and service probes. On boot
 it loads config, migrates the app DB, connects the queue, registers workflows,
-starts polling with one execution slot, and serves `/healthz`.
+starts polling with one execution slot and a 30-second lease, and serves `/healthz`.
+Native `worker.start()` persists declared schedules before startup enqueues all three
+maintenance passes with `{}` and no run options.
 
 ## Run it
 
@@ -36,14 +38,14 @@ database:
   queue: ./data/openworkflow.sqlite # default
 ```
 
-Relative paths resolve against the config file's directory, so web and worker
+Relative paths resolve against the config directory's parent (the deployment root), so web and worker
 (sharing one `when.yaml`) open the same files.
 
 ## Environment
 
 - `CONFIG_PATH` — path to `when.yaml` (default `/app/config/when.yaml` in prod,
   `<cwd>/config/when.yaml` otherwise).
-- `DATABASE_PATH` / `QUEUE_DB_PATH` — optional escape hatches that override the
+- `WHEN_DATABASE_PATH` / `WHEN_QUEUE_DB_PATH` — optional escape hatches that override the
   config's `database.app` / `database.queue`.
 - `PORT` — health server port (default `9000`).
 
@@ -63,25 +65,24 @@ attempts (30-second initial delay, doubling up to ten minutes). A successful sen
 is checkpointed separately from its status write. Host and guest keep independent
 retry budgets, even when they share an email address.
 
-An email `failed` action means a provider attempt failed and retries may remain.
-The parent records `done` only when every intended email was sent and the appointment
-still exists and is not purged. Ineligible deliveries are skipped. Exhaustion fails
-the engine run rather than returning a successful summary of failed delivery.
-
 Purge deletes each remote event with up to thirty native attempts (one-minute
-initial delay, doubling up to one hour). Success or an already-missing event permits
-video cleanup and local deletion. Exhaustion fails the job, retaining that row and
-all later rows in the chain. These rows are already marked `purged` and hidden from
-normal scheduling; another purge click does not automatically recover them. Review
-failed jobs before deciding how to finish cleanup.
+initial delay, doubling up to one hour). Exhaustion fails the job and retains the
+current and later hidden `purged` rows, including local personal data. Clicking
+purge again is not promised recovery; cleanup requires explicit operator review.
+
+Email action logs describe attempts, not terminal job outcomes. See
+[interpreting outcomes](../../docs/deployment.md#interpreting-outcomes) for email
+log/service-status meanings and purge recovery guidance.
 
 Prepared envelopes include personal data and guest capability links: treat queue
-backups like application backups. Checkpoints do not guarantee exactly-once SMTP;
-a crash after remote acceptance but before saving success can still cause a duplicate.
+backups like application backups. Effects are at-least-once: a crash after remote
+acceptance but before saving success can still duplicate effects or action-log entries.
+Metrics and logs are not exactly-once ledgers.
 
-Before deploying these changed workflows, stop producers, let the old worker drain
-unfinished jobs, then stop it and back up both databases before a grouped restart.
-Do not replay unfinished old workflow histories through the changed step graph.
+Deploy web, worker and CLI together only after draining old work. Follow the
+[grouped rollout runbook](../../docs/deployment.md#grouped-workflow-rollout) for read-only
+inspection, paired SQLite backups, restore verification and rollback restrictions.
+Do not replay unfinished old workflow histories through changed step graphs.
 
 ## Engine integration
 
@@ -94,6 +95,36 @@ escaped handler execution may mean durable suspension rather than a terminal job
 failure. Email metrics count physical SMTP attempt outcomes, not an exactly-once
 delivery ledger.
 
-OpenWorkflow `0.11.0` supports persisted cron schedules. Refresh, expiry, and repair
-still use the existing process timers; migrating those tasks to awaited cron
-workflows is a separate application change.
+## Maintenance and shutdown
+
+| Workflow                      | UTC cron       | Stable schedule ID                 | Result      |
+| ----------------------------- | -------------- | ---------------------------------- | ----------- |
+| `refresh-calendars`           | `*/5 * * * *`  | `when:refresh-calendars`           | `refreshed` |
+| `expire-pending-appointments` | `0 * * * *`    | `when:expire-pending-appointments` | `expired`   |
+| `sync-calendars`              | `*/10 * * * *` | `when:sync-calendars`              | `requested` |
+
+All inputs are `{}`. Each handler awaits the actual pass in durable steps; `requested`
+means the scan finished, not that an in-memory request was accepted. Refresh checks
+per-calendar due intervals, keeps stale mirrors on provider fetch failure, and runs
+before conflict checks. Expiry SQL is idempotent. Subsequent callbacks read live config.
+Expected provider failures record status for later passes; local errors escape to native
+retries. Completion is not a promise that every provider succeeded.
+
+Cron follows UTC wall-clock boundaries, not time since the last pass finished. After
+downtime it dispatches one overdue run per schedule. Native startup upserts preserve
+administrative pause state, but independent startup runs still execute while paused;
+startup and overdue runs can coexist. See [schedule management](../../packages/jobs/README.md#schedule-management)
+for the public list/pause/resume/delete API; removing declarations does not retire stored schedules.
+
+No application maintenance timers or coalescers remain. Every cron/manual request gets
+its own pass. One slot serializes active work: slow calls block appointment jobs and
+probes, and repeated scans can build a backlog. Native retry delays park and release
+the slot. Ordinary function steps use the engine default of ten attempts, one-second
+initial delay, doubling to a 100-second cap. Workflow backstops are not extra provider
+budgets; claim counts include resumes, and native exhaustion is terminal. Bookkeeping
+failures can consume step attempts too: these limits are not exact physical-call counts.
+
+Signals and config restarts share shutdown: stop config watching and HTTP admission,
+await `worker.stop()`, finish HTTP closure, close the queue, then close the app DB.
+Stop waits for active passes to finish or park, not for pending work/future retries;
+it is not queue draining. Force-kill recovery relies on saved history and lease expiry.

apps/worker/src/calendar/refresh.test.ts

@@ -1,10 +1,18 @@
 import { expect, test, vi, beforeEach } from 'vitest';
 import pino from 'pino';
+import { sql } from 'kysely';
 import type { ResolvedCalendar, WhenConfiguration } from '@when/config';
 import { openDb, runMigrations, replaceCalendarBusy, recordServiceOutcome } from '@when/db';
 import type { Logger } from '../services/logger.js';
 import type { WorkerContext } from '../services/context.js';
-import { busyCalendarIds, refreshCalendar, refreshCalendars, refreshWindow } from './refresh.js';
+import { calendarRefreshTotal } from '../services/metrics.js';
+import {
+	busyCalendarIds,
+	refreshCalendar,
+	refreshCalendars,
+	refreshCycle,
+	refreshWindow
+} from './refresh.js';
 
 const inst = (s: string) => Temporal.Instant.from(s);
 const window = { start: inst('2026-04-01T00:00:00Z'), end: inst('2026-05-01T00:00:00Z') };
@@ -45,6 +53,26 @@ END:VCALENDAR</C:calendar-data>
   </response>
 </multistatus>`;
 
+const workMeetings = {
+	chat: {
+		booking_calendar: 'work',
+		additional_busy_calendars: [],
+		booking_window_days: 60,
+		schedule: 'standard'
+	}
+} as unknown as WhenConfiguration['meetings'];
+
+const appointment = {
+	id: '1',
+	event_type_id: 'chat',
+	start_time: '2026-04-15T14:00:00Z',
+	end_time: '2026-04-15T15:00:00Z',
+	guest_name: 'A',
+	guest_email: 'a@example.com',
+	status: 'confirmed' as const,
+	cancel_token: 't'
+};
+
 const defaultTestConfig = {
 	url: { app: 'https://when.example.com' },
 	user: { name: 'Jane', email: 'jane@example.com' },
@@ -78,7 +106,14 @@ test('refreshCalendar populates the mirror and records success', async () => {
 		vi.spyOn(globalThis, 'fetch').mockResolvedValue(
 			new Response(oneEvent('remote-1@x'), { status: 207 })
 		);
-		await refreshCalendar(ctx, workCal, window, { now: window.start });
+		const metric = vi.spyOn(calendarRefreshTotal, 'inc');
+		const result = await refreshCalendar(ctx, workCal, window, { now: window.start });
+		expect(result).toEqual({ ok: true, busyCount: 1 });
+		expect(metric).toHaveBeenCalledExactlyOnceWith({
+			calendar_id: 'work',
+			provider_type: 'caldav',
+			status: 'success'
+		});
 		const busy = await ctx.db.selectFrom('external_calendar_busy').selectAll().execute();
 		expect(busy).toHaveLength(1);
 		expect(busy[0].start_time).toBe('2026-04-15T14:00:00Z');
@@ -101,10 +136,23 @@ test('refreshCalendar keeps stale data and records the error on provider failure
 		await replaceCalendarBusy(ctx.db, 'work', [
 			{ start: '2026-04-10T09:00:00Z', end: '2026-04-10T09:30:00Z' }
 		]);
+		const lastOk = window.start.subtract({ minutes: 30 }).toString();
+		await recordServiceOutcome(
+			ctx.db,
+			{ kind: 'calendar', name: 'work' },
+			{ at: lastOk, via: 'refresh' }
+		);
 		vi.spyOn(globalThis, 'fetch').mockResolvedValue(
 			new Response('boom', { status: 500, statusText: 'err' })
 		);
-		await refreshCalendar(ctx, workCal, window, { now: window.start });
+		const metric = vi.spyOn(calendarRefreshTotal, 'inc');
+		const result = await refreshCalendar(ctx, workCal, window, { now: window.start, via: 'test' });
+		expect(result).toEqual({ ok: false, error: expect.stringContaining('500') });
+		expect(metric).toHaveBeenCalledExactlyOnceWith({
+			calendar_id: 'work',
+			provider_type: 'caldav',
+			status: 'failure'
+		});
 		const busy = await ctx.db.selectFrom('external_calendar_busy').selectAll().execute();
 		expect(busy).toHaveLength(1);
 		expect(busy[0].start_time).toBe('2026-04-10T09:00:00Z');
@@ -115,7 +163,8 @@ test('refreshCalendar keeps stale data and records the error on provider failure
 			.where('name', '=', 'work')
 			.executeTakeFirstOrThrow();
 		expect(status.error).toContain('500');
-		expect(status.last_ok_at).toBeNull();
+		expect(status.last_ok_at).toBe(lastOk);
+		expect(status.via).toBe('test');
 	} finally {
 		await ctx.db.destroy();
 	}
@@ -170,15 +219,8 @@ test('refreshCalendar drops our own published event', async () => {
 		await ctx.db
 			.insertInto('appointments')
 			.values({
-				id: '1',
-				event_type_id: 'chat',
-				start_time: '2026-04-15T14:00:00Z',
-				end_time: '2026-04-15T15:00:00Z',
-				guest_name: 'A',
-				guest_email: 'a@example.com',
+				...appointment,
 				location: null,
-				status: 'confirmed',
-				cancel_token: 't',
 				external_event_id: 'remote-1@x',
 				external_calendar_id: 'work',
 				calendar_revision: 0,
@@ -381,10 +423,286 @@ test('refreshCalendars skips a calendar refreshed within its interval, refreshes
 		await refreshCalendars(ctx, { now: window.start.add({ minutes: 10 }) });
 		expect(fetched).toBe(false);
 
-		// past the interval → due, refreshes
-		await refreshCalendars(ctx, { now: window.start.add({ minutes: 31 }) });
+		await refreshCalendars(ctx, { now: window.start.add({ minutes: 30 }) });
 		expect(fetched).toBe(true);
 	} finally {
 		await db.destroy();
 	}
 });
+
+test('refreshCalendar lets the own-event DB read fail before fetching or recording an outcome', async () => {
+	const ctx = await ctxWithDb();
+	try {
+		await ctx.db.schema.dropTable('appointments').execute();
+		const fetch = vi.spyOn(globalThis, 'fetch');
+		const log = vi.spyOn(ctx.logger, 'error');
+		const metric = vi.spyOn(calendarRefreshTotal, 'inc');
+
+		await expect(refreshCalendar(ctx, workCal, window)).rejects.toThrow('no such table');
+
+		expect(fetch).not.toHaveBeenCalled();
+		expect(log).not.toHaveBeenCalled();
+		expect(metric).not.toHaveBeenCalled();
+		const statuses = await ctx.db.selectFrom('service_status').selectAll().execute();
+		expect(statuses).toEqual([]);
+	} finally {
+		await ctx.db.destroy();
+	}
+});
+
+test('refreshCalendar rejects a mirror replacement failure without recording provider failure', async () => {
+	const ctx = await ctxWithDb();
+	try {
+		const stale = { start: '2026-04-10T09:00:00Z', end: '2026-04-10T09:30:00Z' };
+		await replaceCalendarBusy(ctx.db, 'work', [stale]);
+		await sql`create trigger fail_mirror before insert on external_calendar_busy
+			begin select raise(abort, 'mirror unavailable'); end`.execute(ctx.db);
+		vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(oneEvent('r1'), { status: 207 }));
+		const log = vi.spyOn(ctx.logger, 'error');
+		const metric = vi.spyOn(calendarRefreshTotal, 'inc');
+
+		await expect(refreshCalendar(ctx, workCal, window)).rejects.toThrow('mirror unavailable');
+
+		expect(log).not.toHaveBeenCalled();
+		expect(metric).not.toHaveBeenCalled();
+		const statuses = await ctx.db.selectFrom('service_status').selectAll().execute();
+		expect(statuses).toEqual([]);
+		const busy = await ctx.db
+			.selectFrom('external_calendar_busy')
+			.select(['start_time', 'end_time'])
+			.execute();
+		expect(busy).toEqual([{ start_time: stale.start, end_time: stale.end }]);
+	} finally {
+		await ctx.db.destroy();
+	}
+});
+
+test.each(['calendar', 'provider'] as const)(
+	'refreshCalendar does not reclassify a successful fetch when the %s status write fails',
+	async (kind) => {
+		const ctx = await ctxWithDb();
+		try {
+			await sql`create trigger fail_status before insert on service_status
+				when new.kind = ${sql.lit(kind)}
+				begin select raise(abort, 'status unavailable'); end`.execute(ctx.db);
+			vi.spyOn(globalThis, 'fetch').mockResolvedValue(
+				new Response(oneEvent('r1'), { status: 207 })
+			);
+			const log = vi.spyOn(ctx.logger, 'error');
+			const metric = vi.spyOn(calendarRefreshTotal, 'inc');
+
+			await expect(refreshCalendar(ctx, workCal, window)).rejects.toThrow('status unavailable');
+
+			expect(log).not.toHaveBeenCalled();
+			expect(metric).not.toHaveBeenCalled();
+			const statuses = await ctx.db.selectFrom('service_status').selectAll().execute();
+			expect(statuses).toHaveLength(kind === 'provider' ? 1 : 0);
+			for (const status of statuses) {
+				expect(status.error).toBeNull();
+				expect(status.last_ok_at).toBe(window.start.toString());
+			}
+			const busy = await ctx.db.selectFrom('external_calendar_busy').selectAll().execute();
+			expect(busy).toHaveLength(1);
+		} finally {
+			await ctx.db.destroy();
+		}
+	}
+);
+
+test('refreshCalendar lets a success metric failure escape without recording provider failure', async () => {
+	const ctx = await ctxWithDb();
+	try {
+		vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(oneEvent('r1'), { status: 207 }));
+		const error = new Error('metric unavailable');
+		const metric = vi.spyOn(calendarRefreshTotal, 'inc').mockImplementationOnce(() => {
+			throw error;
+		});
+		const log = vi.spyOn(ctx.logger, 'error');
+
+		await expect(refreshCalendar(ctx, workCal, window)).rejects.toBe(error);
+
+		expect(log).not.toHaveBeenCalled();
+		expect(metric).toHaveBeenCalledTimes(1);
+		const statuses = await ctx.db.selectFrom('service_status').selectAll().execute();
+		expect(statuses).toHaveLength(2);
+		for (const status of statuses) expect(status.error).toBeNull();
+	} finally {
+		await ctx.db.destroy();
+	}
+});
+
+test('refreshCalendar rejects if recording an expected provider failure fails locally', async () => {
+	const ctx = await ctxWithDb();
+	try {
+		await sql`create trigger fail_status before insert on service_status
+			begin select raise(abort, 'status unavailable'); end`.execute(ctx.db);
+		vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('boom', { status: 500 }));
+		const metric = vi.spyOn(calendarRefreshTotal, 'inc');
+
+		await expect(refreshCalendar(ctx, workCal, window)).rejects.toThrow('status unavailable');
+
+		expect(metric).not.toHaveBeenCalled();
+	} finally {
+		await ctx.db.destroy();
+	}
+});
+
+test('refreshCycle flags conflicts using the newly refreshed mirror', async () => {
+	const ctx = await ctxWithDb({ meetings: workMeetings });
+	try {
+		await ctx.db.insertInto('appointments').values(appointment).execute();
+		await replaceCalendarBusy(ctx.db, 'work', [
+			{ start: '2026-04-10T09:00:00Z', end: '2026-04-10T09:30:00Z' }
+		]);
+		vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(oneEvent('r1'), { status: 207 }));
+
+		await refreshCycle(ctx, { now: window.start });
+
+		const row = await ctx.db
+			.selectFrom('appointments')
+			.select('has_possible_conflict')
+			.executeTakeFirstOrThrow();
+		expect(row.has_possible_conflict).toBe(1);
+		const statuses = await ctx.db.selectFrom('service_status').selectAll().execute();
+		expect(statuses).toHaveLength(2);
+		for (const status of statuses) {
+			expect(status.last_attempt_at).toBe(window.start.toString());
+			expect(status.error).toBeNull();
+		}
+	} finally {
+		await ctx.db.destroy();
+	}
+});
+
+test('refreshCycle flags conflicts using the stale mirror after a provider failure', async () => {
+	const ctx = await ctxWithDb({ meetings: workMeetings });
+	try {
+		await ctx.db.insertInto('appointments').values(appointment).execute();
+		await replaceCalendarBusy(ctx.db, 'work', [
+			{ start: appointment.start_time, end: appointment.end_time }
+		]);
+		vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('boom', { status: 500 }));
+
+		await refreshCycle(ctx, { now: window.start });
+
+		const row = await ctx.db
+			.selectFrom('appointments')
+			.select('has_possible_conflict')
+			.executeTakeFirstOrThrow();
+		expect(row.has_possible_conflict).toBe(1);
+		const statuses = await ctx.db.selectFrom('service_status').selectAll().execute();
+		expect(statuses).toHaveLength(2);
+		for (const status of statuses) {
+			expect(status.last_attempt_at).toBe(window.start.toString());
+			expect(status.error).toEqual(expect.stringContaining('500'));
+		}
+	} finally {
+		await ctx.db.destroy();
+	}
+});
+
+test('refreshCycle lets conflict writes fail without marking a successful provider refresh as failed', async () => {
+	const ctx = await ctxWithDb({ meetings: workMeetings });
+	try {
+		await ctx.db.insertInto('appointments').values(appointment).execute();
+		await sql`create trigger fail_conflict before update of has_possible_conflict on appointments
+			begin select raise(abort, 'conflict unavailable'); end`.execute(ctx.db);
+		vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(oneEvent('r1'), { status: 207 }));
+		const log = vi.spyOn(ctx.logger, 'error');
+		const metric = vi.spyOn(calendarRefreshTotal, 'inc');
+
+		await expect(refreshCycle(ctx, { now: window.start })).rejects.toThrow('conflict unavailable');
+
+		expect(log).not.toHaveBeenCalled();
+		expect(metric).toHaveBeenCalledExactlyOnceWith({
+			calendar_id: 'work',
+			provider_type: 'caldav',
+			status: 'success'
+		});
+		const statuses = await ctx.db.selectFrom('service_status').selectAll().execute();
+		expect(statuses).toHaveLength(2);
+		for (const status of statuses) {
+			expect(status.error).toBeNull();
+			expect(status.last_ok_at).toBe(window.start.toString());
+		}
+	} finally {
+		await ctx.db.destroy();
+	}
+});
+
+test('refreshCalendars keeps independent due intervals and retries a stale failure on later passes', async () => {
+	const ctx = await ctxWithDb({
+		providers: {
+			'work-dav': {
+				...workDav,
+				calendars: {
+					...workDav.calendars,
+					other: { href: 'https://cal.example.com/other/', sync: { refresh_every_minutes: 30 } }
+				}
+			}
+		},
+		meetings: {
+			chat: { ...workMeetings.chat, additional_busy_calendars: ['other'] }
+		}
+	});
+	try {
+		for (const name of ['work', 'other']) {
+			await recordServiceOutcome(
+				ctx.db,
+				{ kind: 'calendar', name },
+				{ at: window.start.toString(), via: 'refresh' }
+			);
+		}
+		const fetch = vi.spyOn(globalThis, 'fetch').mockImplementation(async () => {
+			return new Response('boom', { status: 500 });
+		});
+
+		await refreshCalendars(ctx, { now: window.start.add({ minutes: 9 }) });
+		expect(fetch).not.toHaveBeenCalled();
+		await refreshCalendars(ctx, { now: window.start.add({ minutes: 10 }) });
+		await refreshCalendars(ctx, { now: window.start.add({ minutes: 11 }) });
+		expect(fetch.mock.calls.map(([url]) => String(url))).toEqual([
+			'https://cal.example.com/work/',
+			'https://cal.example.com/work/'
+		]);
+
+		await refreshCalendars(ctx, { now: window.start.add({ minutes: 30 }) });
+		expect(fetch.mock.calls.slice(2).map(([url]) => String(url))).toEqual([
+			'https://cal.example.com/work/',
+			'https://cal.example.com/other/'
+		]);
+	} finally {
+		await ctx.db.destroy();
+	}
+});
+
+test('refreshCalendars uses a reloaded calendar interval on its next pass', async () => {
+	const ctx = await ctxWithDb({ meetings: workMeetings });
+	try {
+		await recordServiceOutcome(
+			ctx.db,
+			{ kind: 'calendar', name: 'work' },
+			{ at: window.start.toString(), via: 'refresh' }
+		);
+		const fetch = vi
+			.spyOn(globalThis, 'fetch')
+			.mockResolvedValue(new Response(oneEvent('r1'), { status: 207 }));
+		const now = window.start.add({ minutes: 5 });
+		await refreshCalendars(ctx, { now });
+		expect(fetch).not.toHaveBeenCalled();
+
+		ctx.config = {
+			...ctx.config,
+			providers: {
+				'work-dav': {
+					...workDav,
+					calendars: { work: { ...workDav.calendars.work, sync: { refresh_every_minutes: 5 } } }
+				}
+			}
+		};
+		await refreshCalendars(ctx, { now });
+		expect(fetch).toHaveBeenCalledTimes(1);
+	} finally {
+		await ctx.db.destroy();
+	}
+});

apps/worker/src/calendar/refresh.ts

@@ -1,5 +1,5 @@
 import { findCalendar, type ResolvedCalendar, type WhenConfiguration } from '@when/config';
-import type { ExpandWindow } from '@when/calendar';
+import type { ExpandWindow, Interval } from '@when/calendar';
 import { busyCalendarsFor, fetchBusyIntervals } from '@when/calendar';
 import {
 	listOwnEventIds,
@@ -52,21 +52,11 @@ export async function refreshCalendar(
 ): Promise<RefreshResult> {
 	const at = (opts.now ?? window.start).toString();
 	const via = opts.via ?? 'refresh';
+	const ownEventIds = await listOwnEventIds(ctx.db, cal.name);
+	const excludeUids = new Set(ownEventIds);
+	let intervals: Interval[];
 	try {
-		const excludeUids = new Set(await listOwnEventIds(ctx.db, cal.name));
-		const intervals = await fetchBusyIntervals(cal, window, { excludeUids });
-		await replaceCalendarBusy(
-			ctx.db,
-			cal.name,
-			intervals.map((i) => ({ start: i.start.toString(), end: i.end.toString() }))
-		);
-		await recordRefreshOutcome(ctx, cal, { at, via });
-		calendarRefreshTotal.inc({
-			calendar_id: cal.name,
-			provider_type: cal.provider.type,
-			status: 'success'
-		});
-		return { ok: true, busyCount: intervals.length };
+		intervals = await fetchBusyIntervals(cal, window, { excludeUids });
 	} catch (err) {
 		const error = err instanceof Error ? err.message : String(err);
 		ctx.logger.error(
@@ -84,6 +74,18 @@ export async function refreshCalendar(
 		});
 		return { ok: false, error };
 	}
+	await replaceCalendarBusy(
+		ctx.db,
+		cal.name,
+		intervals.map((i) => ({ start: i.start.toString(), end: i.end.toString() }))
+	);
+	await recordRefreshOutcome(ctx, cal, { at, via });
+	calendarRefreshTotal.inc({
+		calendar_id: cal.name,
+		provider_type: cal.provider.type,
+		status: 'success'
+	});
+	return { ok: true, busyCount: intervals.length };
 }
 
 // A provider backing no calendars has no other source of status.
@@ -109,8 +111,6 @@ export async function refreshCalendars(
 			ctx.logger.warn({ calendarId: id }, 'busy calendar id not found in calendars; skipping');
 			continue;
 		}
-		// Each calendar refreshes on its own interval; a never-synced or failing one
-		// (no recent success) is always due, so failures retry every tick.
 		if (!isDue(lastSuccess.get(id) ?? null, cal, now)) continue;
 		await refreshCalendar(ctx, cal, refreshWindow(ctx.config, id, now), { ...opts, now });
 	}

apps/worker/src/calendar/scheduler.test.ts

@@ -1,25 +0,0 @@
-import { afterEach, expect, test, vi } from 'vitest';
-import { createRefreshScheduler } from './scheduler.js';
-
-afterEach(() => vi.useRealTimers());
-
-test('scheduler runs on start, reschedules each interval, and stops', async () => {
-	vi.useFakeTimers();
-	let calls = 0;
-	const scheduler = createRefreshScheduler(async () => {
-		calls++;
-	}, 1000);
-
-	scheduler.start();
-	expect(calls).toBe(1);
-
-	await vi.advanceTimersByTimeAsync(1000);
-	expect(calls).toBe(2);
-
-	await vi.advanceTimersByTimeAsync(1000);
-	expect(calls).toBe(3);
-
-	scheduler.stop();
-	await vi.advanceTimersByTimeAsync(5000);
-	expect(calls).toBe(3);
-});

apps/worker/src/calendar/scheduler.ts

@@ -1,32 +0,0 @@
-export interface RefreshScheduler {
-	start(): void;
-	stop(): void;
-}
-
-export function createRefreshScheduler(
-	run: () => Promise<void>,
-	intervalMs: number
-): RefreshScheduler {
-	let timer: ReturnType<typeof setTimeout> | null = null;
-	let stopped = false;
-
-	async function tick(): Promise<void> {
-		try {
-			await run();
-		} finally {
-			if (!stopped) timer = setTimeout(() => void tick(), intervalMs);
-		}
-	}
-
-	return {
-		start() {
-			stopped = false;
-			void tick();
-		},
-		stop() {
-			stopped = true;
-			if (timer) clearTimeout(timer);
-			timer = null;
-		}
-	};
-}

apps/worker/src/calendar/sync-scanner.test.ts

@@ -1,66 +0,0 @@
-import { afterEach, expect, test, vi } from 'vitest';
-import { createCalendarSyncScanner } from './sync-scanner.js';
-
-afterEach(() => vi.useRealTimers());
-
-const flush = () => new Promise((r) => setTimeout(r, 0));
-
-test('runs on request and reschedules on the floor', async () => {
-	vi.useFakeTimers();
-	let calls = 0;
-	const scanner = createCalendarSyncScanner(async () => {
-		calls++;
-	}, 1000);
-
-	scanner.requestScan();
-	await vi.advanceTimersByTimeAsync(0);
-	expect(calls).toBe(1);
-
-	await vi.advanceTimersByTimeAsync(1000);
-	expect(calls).toBe(2);
-
-	scanner.stop();
-	await vi.advanceTimersByTimeAsync(5000);
-	expect(calls).toBe(2);
-});
-
-test('coalesces requests that arrive during a scan into one rerun', async () => {
-	let calls = 0;
-	let release: () => void = () => {};
-	const run = () => {
-		calls++;
-		return new Promise<void>((r) => {
-			release = r;
-		});
-	};
-	const scanner = createCalendarSyncScanner(run, 10_000);
-
-	scanner.requestScan();
-	await flush();
-	expect(calls).toBe(1);
-
-	scanner.requestScan();
-	scanner.requestScan();
-	expect(calls).toBe(1);
-
-	release();
-	await flush();
-	expect(calls).toBe(2);
-
-	release();
-	await flush();
-	expect(calls).toBe(2);
-
-	scanner.stop();
-});
-
-test('stop prevents new requests from running', async () => {
-	let calls = 0;
-	const scanner = createCalendarSyncScanner(async () => {
-		calls++;
-	}, 10_000);
-	scanner.stop();
-	scanner.requestScan();
-	await flush();
-	expect(calls).toBe(0);
-});

apps/worker/src/calendar/sync-scanner.ts

@@ -1,54 +0,0 @@
-export interface CalendarSyncScanner {
-	requestScan(): void;
-	stop(): void;
-}
-
-// One scan at a time (overlap guard via `scanning`); a request that arrives
-// mid-scan is remembered and runs once more after, so bursts collapse into a
-// single extra pass. Each run reschedules a floor timer, so an idle worker still
-// scans every `floorMs`.
-export function createCalendarSyncScanner(
-	run: () => Promise<void>,
-	floorMs: number
-): CalendarSyncScanner {
-	let scanning = false;
-	let rescanRequested = false;
-	let floorTimer: ReturnType<typeof setTimeout> | null = null;
-	let stopped = false;
-
-	function requestScan(): void {
-		if (stopped) return;
-		if (scanning) {
-			rescanRequested = true;
-			return;
-		}
-		scanning = true;
-		void loop();
-	}
-
-	async function loop(): Promise<void> {
-		try {
-			do {
-				rescanRequested = false;
-				await run();
-			} while (rescanRequested && !stopped);
-		} catch {
-			// A scan threw; the floor timer below retries.
-		} finally {
-			scanning = false;
-			if (!stopped) {
-				if (floorTimer) clearTimeout(floorTimer);
-				floorTimer = setTimeout(requestScan, floorMs);
-			}
-		}
-	}
-
-	return {
-		requestScan,
-		stop() {
-			stopped = true;
-			if (floorTimer) clearTimeout(floorTimer);
-			floorTimer = null;
-		}
-	};
-}

apps/worker/src/index.test.ts

@@ -0,0 +1,388 @@
+import { EventEmitter } from 'node:events';
+import { afterEach, beforeEach, expect, test, vi } from 'vitest';
+import type { ReloadResult, WhenConfiguration } from '@when/config';
+import { expirePendingAppointments, refreshCalendars, syncCalendars } from '@when/jobs';
+
+const mocks = vi.hoisted(() => ({
+	loadConfig: vi.fn(),
+	watchConfig: vi.fn(),
+	stopWatch: vi.fn(),
+	openDb: vi.fn(),
+	runMigrations: vi.fn(),
+	destroy: vi.fn(),
+	initOpenWorkflow: vi.fn(),
+	closeOpenWorkflow: vi.fn(),
+	newWorker: vi.fn(),
+	start: vi.fn(),
+	stop: vi.fn(),
+	runWorkflow: vi.fn(),
+	result: vi.fn(),
+	setLogger: vi.fn(),
+	createMailer: vi.fn(),
+	setWorkerContext: vi.fn(),
+	createHealthServer: vi.fn(),
+	createEngineLogger: vi.fn(),
+	registerWorkflows: vi.fn(),
+	logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }
+}));
+
+vi.mock('@when/config', async (importOriginal) => {
+	const actual = await importOriginal<typeof import('@when/config')>();
+	return { ...actual, loadConfig: mocks.loadConfig, watchConfig: mocks.watchConfig };
+});
+vi.mock('@when/db', () => ({ openDb: mocks.openDb, runMigrations: mocks.runMigrations }));
+vi.mock('@when/jobs', async (importOriginal) => {
+	const actual = await importOriginal<typeof import('@when/jobs')>();
+	return {
+		...actual,
+		initOpenWorkflow: mocks.initOpenWorkflow,
+		closeOpenWorkflow: mocks.closeOpenWorkflow
+	};
+});
+vi.mock('@when/calendar', () => ({ setLogger: mocks.setLogger }));
+vi.mock('./email/smtp.js', () => ({ createMailer: mocks.createMailer }));
+vi.mock('./services/context.js', () => ({ setWorkerContext: mocks.setWorkerContext }));
+vi.mock('./services/health.js', () => ({ createHealthServer: mocks.createHealthServer }));
+vi.mock('./services/logger.js', () => ({ logger: mocks.logger }));
+vi.mock('./services/engine-logger.js', () => ({ createEngineLogger: mocks.createEngineLogger }));
+vi.mock('./workflows/index.js', () => ({ registerWorkflows: mocks.registerWorkflows }));
+
+const config = {
+	user: { name: 'Owner', email: 'owner@example.test', timezone: 'UTC' },
+	database: { app: '/fixture/app.sqlite', queue: '/fixture/queue.sqlite' },
+	smtp: { host: 'smtp.example.test', port: 587, username: 'owner', password: 'test' }
+} as WhenConfiguration;
+const db = { destroy: mocks.destroy };
+const mailer = {};
+const engineLogger = {};
+const worker = { start: mocks.start, stop: mocks.stop };
+const client = { newWorker: mocks.newWorker, runWorkflow: mocks.runWorkflow };
+let events: string[];
+let signals: Map<string, () => void>;
+let reload: (result: ReloadResult) => void;
+let server: EventEmitter & {
+	listening: boolean;
+	listen: ReturnType<typeof vi.fn>;
+	close: ReturnType<typeof vi.fn>;
+};
+
+function finishListening() {
+	server.listening = true;
+	server.emit('listening');
+}
+
+async function boot() {
+	await import('./index.js');
+	await vi.waitFor(() => {
+		expect(mocks.watchConfig).toHaveBeenCalledOnce();
+		expect(signals.has('SIGTERM')).toBe(true);
+		expect(signals.has('SIGINT')).toBe(true);
+	});
+}
+
+beforeEach(() => {
+	vi.resetModules();
+	vi.resetAllMocks();
+	vi.stubEnv('PORT', '19099');
+	events = [];
+	signals = new Map();
+	vi.spyOn(process, 'on').mockImplementation((event, listener) => {
+		if (event === 'SIGTERM' || event === 'SIGINT') signals.set(event, listener);
+		return process;
+	});
+	vi.spyOn(process, 'exit').mockImplementation(() => undefined as never);
+	mocks.loadConfig.mockResolvedValue(config);
+	mocks.openDb.mockReturnValue(db);
+	mocks.runMigrations.mockResolvedValue([]);
+	mocks.createMailer.mockReturnValue(mailer);
+	mocks.createEngineLogger.mockReturnValue(engineLogger);
+	mocks.initOpenWorkflow.mockReturnValue(client);
+	mocks.newWorker.mockReturnValue(worker);
+	mocks.start.mockImplementation(async () => {
+		events.push('worker:start');
+	});
+	mocks.runWorkflow.mockImplementation(async (spec: { name: string }) => {
+		events.push(`enqueue:${spec.name}`);
+		return { result: mocks.result };
+	});
+	mocks.stopWatch.mockImplementation(() => {
+		events.push('watch:stop');
+	});
+	mocks.watchConfig.mockImplementation((callback: typeof reload) => {
+		reload = callback;
+		return mocks.stopWatch;
+	});
+	mocks.stop.mockImplementation(async () => {
+		events.push('worker:stop');
+	});
+	mocks.closeOpenWorkflow.mockImplementation(async () => {
+		events.push('queue:close');
+	});
+	mocks.destroy.mockImplementation(async () => {
+		events.push('db:close');
+	});
+	server = Object.assign(new EventEmitter(), {
+		listening: false,
+		listen: vi.fn(() => {
+			events.push('server:listen');
+			queueMicrotask(finishListening);
+			return server;
+		}),
+		close: vi.fn((callback: (error?: Error) => void) => {
+			events.push('server:close');
+			server.listening = false;
+			queueMicrotask(() => {
+				events.push('server:closed');
+				server.emit('close');
+				callback();
+			});
+			return server;
+		})
+	});
+	mocks.createHealthServer.mockReturnValue(server);
+});
+
+afterEach(() => {
+	vi.restoreAllMocks();
+	vi.unstubAllEnvs();
+});
+
+test('registers workflows before starting one default worker, then enqueues all maintenance', async () => {
+	const started = Promise.withResolvers<void>();
+	mocks.start.mockImplementation(async () => {
+		events.push('worker:start');
+		await started.promise;
+		events.push('worker:started');
+	});
+	await import('./index.js');
+	await vi.waitFor(() => expect(mocks.start).toHaveBeenCalledOnce());
+	expect(mocks.runWorkflow).not.toHaveBeenCalled();
+	expect(mocks.createHealthServer).not.toHaveBeenCalled();
+	expect(mocks.loadConfig).toHaveBeenCalledWith();
+	expect(mocks.openDb).toHaveBeenCalledWith(config.database.app);
+	expect(mocks.runMigrations).toHaveBeenCalledWith(db);
+	expect(mocks.createMailer).toHaveBeenCalledWith(config, mocks.logger);
+	expect(mocks.setWorkerContext).toHaveBeenCalledWith({ config, db, mailer, logger: mocks.logger });
+	expect(mocks.setLogger).toHaveBeenCalledWith(mocks.logger);
+	expect(mocks.createEngineLogger).toHaveBeenCalledWith(mocks.logger);
+	expect(mocks.initOpenWorkflow).toHaveBeenCalledWith({
+		dbPath: config.database.queue,
+		logger: engineLogger
+	});
+	expect(mocks.registerWorkflows).toHaveBeenCalledOnce();
+	expect(mocks.initOpenWorkflow).toHaveBeenCalledBefore(mocks.registerWorkflows);
+	expect(mocks.registerWorkflows).toHaveBeenCalledBefore(mocks.newWorker);
+	expect(mocks.newWorker).toHaveBeenCalledWith();
+
+	started.resolve();
+	await vi.waitFor(() => expect(mocks.watchConfig).toHaveBeenCalledOnce());
+	expect(mocks.runWorkflow.mock.calls).toEqual([
+		[refreshCalendars, {}],
+		[expirePendingAppointments, {}],
+		[syncCalendars, {}]
+	]);
+	expect(mocks.runWorkflow.mock.calls.map(([spec]) => spec.name)).toEqual([
+		'refresh-calendars',
+		'expire-pending-appointments',
+		'sync-calendars'
+	]);
+	expect(mocks.result).not.toHaveBeenCalled();
+	expect(events).toEqual([
+		'worker:start',
+		'worker:started',
+		'enqueue:refresh-calendars',
+		'enqueue:expire-pending-appointments',
+		'enqueue:sync-calendars',
+		'server:listen'
+	]);
+	expect(server.listen.mock.calls[0][0]).toBe(19099);
+	const getConfig = mocks.createHealthServer.mock.calls[0][0] as () => WhenConfiguration;
+	expect(getConfig()).toBe(config);
+	expect(process.exit).not.toHaveBeenCalled();
+});
+
+test.each([0, 1, 2])(
+	'awaits startup enqueue %i and HTTP listening before watching config or installing signals',
+	async (index) => {
+		const enqueued = Promise.withResolvers<void>();
+		for (let preceding = 0; preceding < index; preceding++) {
+			mocks.runWorkflow.mockResolvedValueOnce({ result: mocks.result });
+		}
+		mocks.runWorkflow.mockImplementationOnce(() => enqueued.promise);
+		server.listen.mockImplementation(() => server);
+		await import('./index.js');
+		await vi.waitFor(() => expect(mocks.runWorkflow).toHaveBeenCalledTimes(index + 1));
+		expect(mocks.createHealthServer).not.toHaveBeenCalled();
+		enqueued.resolve();
+		await vi.waitFor(() => expect(server.listen).toHaveBeenCalledOnce());
+		expect(mocks.runWorkflow).toHaveBeenCalledTimes(3);
+		expect(mocks.watchConfig).not.toHaveBeenCalled();
+		expect(signals.size).toBe(0);
+		finishListening();
+		await vi.waitFor(() => expect(mocks.watchConfig).toHaveBeenCalledOnce());
+		expect(signals.size).toBe(2);
+	}
+);
+
+test('signals and restart-required config changes share one safe resource shutdown', async () => {
+	await boot();
+	const stopped = Promise.withResolvers<void>();
+	const httpClosed = Promise.withResolvers<void>();
+	const queueClosed = Promise.withResolvers<void>();
+	mocks.closeOpenWorkflow.mockImplementation(async () => {
+		events.push('queue:close');
+		await queueClosed.promise;
+	});
+	mocks.stop.mockImplementation(async () => {
+		events.push('worker:stop');
+		await stopped.promise;
+		events.push('worker:stopped');
+	});
+	server.close.mockImplementation((callback: () => void) => {
+		events.push('server:close');
+		void httpClosed.promise.then(() => {
+			events.push('server:closed');
+			callback();
+		});
+		return server;
+	});
+	events = [];
+	signals.get('SIGTERM')!();
+	signals.get('SIGINT')!();
+	reload({ ok: true, config: { ...config, database: { ...config.database, app: '/new.sqlite' } } });
+	expect(events).toEqual(['watch:stop', 'server:close', 'worker:stop']);
+	expect(mocks.closeOpenWorkflow).not.toHaveBeenCalled();
+	expect(mocks.destroy).not.toHaveBeenCalled();
+	expect(process.exit).not.toHaveBeenCalled();
+	stopped.resolve();
+	await vi.waitFor(() => expect(events).toContain('worker:stopped'));
+	expect(mocks.closeOpenWorkflow).not.toHaveBeenCalled();
+	httpClosed.resolve();
+	await vi.waitFor(() => expect(mocks.closeOpenWorkflow).toHaveBeenCalledOnce());
+	expect(mocks.destroy).not.toHaveBeenCalled();
+	expect(process.exit).not.toHaveBeenCalled();
+	queueClosed.resolve();
+	await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(0));
+	expect(events).toEqual([
+		'watch:stop',
+		'server:close',
+		'worker:stop',
+		'worker:stopped',
+		'server:closed',
+		'queue:close',
+		'db:close'
+	]);
+	for (const close of [
+		mocks.stopWatch,
+		server.close,
+		mocks.stop,
+		mocks.closeOpenWorkflow,
+		mocks.destroy
+	]) {
+		expect(close).toHaveBeenCalledOnce();
+	}
+});
+
+test.each(['migration', 'start', 'enqueue', 'listen'] as const)(
+	'%s failure awaits cleanup of partially created resources before exiting',
+	async (stage) => {
+		const failure = new Error(`${stage} failed`);
+		if (stage === 'migration') mocks.runMigrations.mockRejectedValue(failure);
+		if (stage === 'start') mocks.start.mockRejectedValue(failure);
+		if (stage === 'enqueue') mocks.runWorkflow.mockRejectedValueOnce(failure);
+		if (stage === 'listen') {
+			server.listen.mockImplementation(() => {
+				queueMicrotask(() => server.emit('error', failure));
+				return server;
+			});
+		}
+		const databaseClosed = Promise.withResolvers<void>();
+		mocks.destroy.mockImplementation(async () => {
+			events.push('db:close');
+			await databaseClosed.promise;
+		});
+		await import('./index.js');
+		await vi.waitFor(() => expect(mocks.destroy).toHaveBeenCalledOnce());
+		expect(process.exit).not.toHaveBeenCalled();
+		expect(mocks.watchConfig).not.toHaveBeenCalled();
+		expect(signals.size).toBe(0);
+		if (stage === 'migration') {
+			expect(mocks.initOpenWorkflow).not.toHaveBeenCalled();
+			expect(mocks.stop).not.toHaveBeenCalled();
+		} else {
+			expect(mocks.stop).toHaveBeenCalledOnce();
+			expect(mocks.stop).toHaveBeenCalledBefore(mocks.closeOpenWorkflow);
+			expect(mocks.closeOpenWorkflow).toHaveBeenCalledBefore(mocks.destroy);
+		}
+		if (stage !== 'listen') expect(mocks.createHealthServer).not.toHaveBeenCalled();
+		databaseClosed.resolve();
+		await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(1));
+		expect(mocks.logger.error).toHaveBeenCalledWith(
+			{ error: failure.message },
+			'worker failed to start'
+		);
+	}
+);
+
+test.each(['SIGTERM', 'SIGINT', 'config-change'])(
+	'%s handles rejected worker stop without closing live connections',
+	async (trigger) => {
+		await boot();
+		mocks.stop.mockRejectedValue(new Error('active work did not stop'));
+		if (trigger === 'config-change') {
+			reload({
+				ok: true,
+				config: { ...config, smtp: { ...config.smtp, host: 'new.example.test' } }
+			});
+		} else {
+			signals.get(trigger)!();
+		}
+		await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(1));
+		expect(mocks.logger.error).toHaveBeenCalled();
+		expect(process.exit).not.toHaveBeenCalledWith(0);
+		expect(mocks.stopWatch).toHaveBeenCalledOnce();
+		expect(server.close).toHaveBeenCalledOnce();
+		expect(mocks.closeOpenWorkflow).not.toHaveBeenCalled();
+		expect(mocks.destroy).not.toHaveBeenCalled();
+	}
+);
+
+test('queue close rejection still closes the app database and exits unsuccessfully', async () => {
+	await boot();
+	mocks.closeOpenWorkflow.mockRejectedValue(new Error('queue close failed'));
+	signals.get('SIGTERM')!();
+	await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(1));
+	expect(mocks.stop).toHaveBeenCalledBefore(mocks.closeOpenWorkflow);
+	expect(mocks.closeOpenWorkflow).toHaveBeenCalledBefore(mocks.destroy);
+	expect(mocks.destroy).toHaveBeenCalledOnce();
+	expect(mocks.logger.error).toHaveBeenCalled();
+	expect(process.exit).not.toHaveBeenCalledWith(0);
+});
+
+test('startup cleanup handles a failed stop without closing connections or losing the boot error', async () => {
+	mocks.start.mockRejectedValue(new Error('start failed'));
+	mocks.stop.mockRejectedValue(new Error('stop failed'));
+	await import('./index.js');
+	await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(1));
+	expect(mocks.stop).toHaveBeenCalledOnce();
+	expect(mocks.closeOpenWorkflow).not.toHaveBeenCalled();
+	expect(mocks.destroy).not.toHaveBeenCalled();
+	expect(mocks.logger.error.mock.calls).toEqual([
+		[{ error: 'stop failed' }, 'worker startup cleanup failed'],
+		[{ error: 'start failed' }, 'worker failed to start']
+	]);
+});
+
+test('startup cleanup closes the app database even if closing the queue fails', async () => {
+	mocks.runWorkflow.mockRejectedValueOnce(new Error('enqueue failed'));
+	mocks.closeOpenWorkflow.mockRejectedValue(new Error('queue close failed'));
+	await import('./index.js');
+	await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(1));
+	expect(mocks.stop).toHaveBeenCalledBefore(mocks.closeOpenWorkflow);
+	expect(mocks.closeOpenWorkflow).toHaveBeenCalledBefore(mocks.destroy);
+	expect(mocks.destroy).toHaveBeenCalledOnce();
+	expect(mocks.logger.error.mock.calls).toEqual([
+		[{ error: 'queue close failed' }, 'worker startup cleanup failed'],
+		[{ error: 'enqueue failed' }, 'worker failed to start']
+	]);
+});

apps/worker/src/index.ts

@@ -1,109 +1,44 @@
-import { loadConfig, watchConfig } from '@when/config';
-import { expireStalePending, openDb, runMigrations } from '@when/db';
-import { initOpenWorkflow } from '@when/jobs';
-import { setLogger } from '@when/calendar';
-import { createMailer } from './email/smtp.js';
-import { applyConfig } from './services/config-reload.js';
-import { setWorkerContext, type WorkerContext } from './services/context.js';
-import { createHealthServer } from './services/health.js';
+import { loadConfig } from '@when/config';
+import { openDb } from '@when/db';
+import { closeOpenWorkflow } from '@when/jobs';
 import { logger } from './services/logger.js';
-import { createEngineLogger } from './services/engine-logger.js';
-import { registerWorkflows } from './workflows/index.js';
-import { registerSyncCalendarsWorkflow } from './workflows/sync-calendars.js';
-import { refreshCycle } from './calendar/refresh.js';
-import { createRefreshScheduler } from './calendar/scheduler.js';
-import { scanOnce } from './calendar/sync.js';
-import { createCalendarSyncScanner } from './calendar/sync-scanner.js';
-import { calendarSyncTotal } from './services/metrics.js';
-
-const DEFAULT_PORT = 9000;
-const CALENDAR_SYNC_FLOOR_MS = 10 * 60_000;
-const REFRESH_TICK_MINUTES = 5;
-const EXPIRE_TICK_MINUTES = 60;
+import { createShutdown, type ShutdownResources } from './services/lifecycle.js';
+import {
+	initializeWorkerContext,
+	startWorkflowWorker,
+	enqueueStartupMaintenance,
+	startHealthServer,
+	watchConfigAndSignals
+} from './services/startup.js';
 
 async function main(): Promise<void> {
 	const config = await loadConfig();
 	logger.info({ user: config.user.email }, 'config loaded');
 
 	const db = openDb(config.database.app);
-	const applied = await runMigrations(db);
-	if (applied.length > 0) logger.info({ migrations: applied }, 'migrations applied');
-
-	const ctx: WorkerContext = { config, logger, db, mailer: createMailer(config, logger) };
-	setWorkerContext(ctx);
-
-	setLogger(logger);
-
-	const calendarSync = createCalendarSyncScanner(async () => {
-		try {
-			await scanOnce(ctx);
-			calendarSyncTotal.inc({ status: 'success' });
-		} catch (err) {
-			calendarSyncTotal.inc({ status: 'failure' });
-			logger.error(
-				{
-					error: err instanceof Error ? err.message : String(err)
-				},
-				'calendar sync failed'
-			);
-		}
-	}, CALENDAR_SYNC_FLOOR_MS);
-
-	const client = initOpenWorkflow({
-		dbPath: config.database.queue,
-		logger: createEngineLogger(logger)
-	});
-	registerWorkflows();
-	registerSyncCalendarsWorkflow(calendarSync);
-	const worker = client.newWorker();
-	await worker.start();
-	logger.info('worker started');
-
-	calendarSync.requestScan();
-
-	const refresh = createRefreshScheduler(() => refreshCycle(ctx), REFRESH_TICK_MINUTES * 60_000);
-	refresh.start();
-	logger.info({ tickMinutes: REFRESH_TICK_MINUTES }, 'calendar refresh scheduled');
-
-	const expireSweep = createRefreshScheduler(async () => {
-		try {
-			const expired = await expireStalePending(ctx.db, Temporal.Now.instant().toString());
-			if (expired > 0) logger.info({ count: expired }, 'expired stale pending requests');
-		} catch (err) {
-			logger.error(
-				{
-					error: err instanceof Error ? err.message : String(err)
-				},
-				'expire sweep failed'
-			);
-		}
-	}, EXPIRE_TICK_MINUTES * 60_000);
-	expireSweep.start();
-	logger.info({ tickMinutes: EXPIRE_TICK_MINUTES }, 'expiry sweep scheduled');
-
-	const port = Number(process.env.PORT) || DEFAULT_PORT;
-	const server = createHealthServer(() => ctx.config);
-	server.listen(port, () => logger.info({ port }, 'health server listening'));
-
-	let stopWatch: () => void = () => {};
-
-	const shutdown = async (signal: string): Promise<void> => {
-		logger.info({ signal }, 'worker shutting down');
-		stopWatch();
-		refresh.stop();
-		expireSweep.stop();
-		calendarSync.stop();
-		server.close();
-		await worker.stop();
-		process.exit(0);
+	const resources: ShutdownResources = {
+		closeQueue: closeOpenWorkflow,
+		closeDatabase: () => db.destroy()
 	};
+	const shutdown = createShutdown(resources);
 
-	stopWatch = watchConfig((result) =>
-		applyConfig(result, ctx, () => void shutdown('config-change'))
-	);
-
-	process.on('SIGTERM', () => void shutdown('SIGTERM'));
-	process.on('SIGINT', () => void shutdown('SIGINT'));
+	try {
+		const ctx = await initializeWorkerContext(config, db);
+		const client = await startWorkflowWorker(ctx, resources);
+		await enqueueStartupMaintenance(client);
+		await startHealthServer(ctx, resources);
+		watchConfigAndSignals(ctx, resources, shutdown);
+	} catch (error) {
+		try {
+			await shutdown();
+		} catch (shutdownError) {
+			logger.error(
+				{ error: shutdownError instanceof Error ? shutdownError.message : String(shutdownError) },
+				'worker startup cleanup failed'
+			);
+		}
+		throw error;
+	}
 }
 
 main().catch((err) => {

apps/worker/src/services/lifecycle.integration.test.ts

@@ -0,0 +1,157 @@
+import { expect, test } from 'vitest';
+import { OpenWorkflow, defineWorkflowSpec } from 'openworkflow';
+import { BackendSqlite } from 'openworkflow/sqlite';
+import { openDb, runMigrations } from '@when/db';
+import { sampleAppointment } from '../email/__fixtures__/appointment.js';
+import { createShutdown } from './lifecycle.js';
+
+test('shutdown waits for an active native step before closing either database', async () => {
+	const db = openDb(':memory:');
+	await runMigrations(db);
+	await db.insertInto('appointments').values(sampleAppointment).execute();
+	const backend = BackendSqlite.connect(':memory:');
+	const client = new OpenWorkflow({ backend });
+	const started = Promise.withResolvers<void>();
+	const release = Promise.withResolvers<void>();
+	const events: string[] = [];
+	const spec = defineWorkflowSpec<Record<string, never>, 'done'>({ name: 'active-maintenance' });
+	client.implementWorkflow(spec, async ({ step }) => {
+		await step.run({ name: 'active-pass' }, async () => {
+			started.resolve();
+			await release.promise;
+			await db.updateTable('appointments').set({ note: 'finished' }).execute();
+			events.push('pass:finished');
+		});
+		return 'done';
+	});
+	const worker = client.newWorker({ concurrency: 1 });
+	const job = await client.runWorkflow(spec, {});
+	let status: string | undefined;
+	let note: string | null | undefined;
+	const shutdown = createShutdown({
+		worker,
+		closeQueue: async () => {
+			status = (await backend.getWorkflowRun({ workflowRunId: job.workflowRun.id }))?.status;
+			events.push('queue:close');
+			await backend.stop();
+		},
+		closeDatabase: async () => {
+			note = (await db.selectFrom('appointments').select('note').executeTakeFirst())?.note;
+			events.push('database:close');
+			await db.destroy();
+		}
+	});
+	try {
+		const claimedCount = await worker.tick();
+		expect(claimedCount).toBe(1);
+		await started.promise;
+		const stopping = shutdown();
+		expect(shutdown()).toBe(stopping);
+		expect(events).toEqual([]);
+		release.resolve();
+		await stopping;
+		expect(events).toEqual(['pass:finished', 'queue:close', 'database:close']);
+		expect(status).toBe('completed');
+		expect(note).toBe('finished');
+	} finally {
+		release.resolve();
+		await shutdown();
+	}
+});
+
+test('shutdown closes a native queue without waiting for a future retry', async () => {
+	const backend = BackendSqlite.connect(':memory:');
+	const client = new OpenWorkflow({ backend });
+	const spec = defineWorkflowSpec<Record<string, never>, 'done'>({ name: 'parked-retry' });
+	let calls = 0;
+	client.implementWorkflow(spec, async ({ step }) => {
+		await step.run(
+			{
+				name: 'retrying-pass',
+				retryPolicy: { maximumAttempts: 5, initialInterval: '1h', maximumInterval: '1h' }
+			},
+			() => {
+				calls += 1;
+				throw new Error('temporary provider failure');
+			}
+		);
+		return 'done';
+	});
+	const worker = client.newWorker({ concurrency: 1 });
+	const job = await client.runWorkflow(spec, {});
+	let status: string | undefined;
+	let availableAt: Date | null | undefined;
+	let databaseClosed = false;
+	const shutdown = createShutdown({
+		worker,
+		closeQueue: async () => {
+			const run = await backend.getWorkflowRun({ workflowRunId: job.workflowRun.id });
+			status = run?.status;
+			availableAt = run?.availableAt;
+			await backend.stop();
+		},
+		closeDatabase: async () => {
+			databaseClosed = true;
+		}
+	});
+	try {
+		const claimedCount = await worker.tick();
+		expect(claimedCount).toBe(1);
+		await worker.stop();
+		await shutdown();
+		expect(status).toBe('pending');
+		expect(availableAt?.getTime()).toBeGreaterThan(Date.now());
+		expect(calls).toBe(1);
+		expect(databaseClosed).toBe(true);
+	} finally {
+		await shutdown();
+	}
+});
+
+test('shutdown does not drain pending children of a parked native parent', async () => {
+	const backend = BackendSqlite.connect(':memory:');
+	const client = new OpenWorkflow({ backend });
+	const child = defineWorkflowSpec<Record<string, never>, 'sent'>({ name: 'parked-child' });
+	const parent = defineWorkflowSpec<Record<string, never>, 'done'>({ name: 'parked-parent' });
+	let childCalls = 0;
+	client.implementWorkflow(child, () => {
+		childCalls += 1;
+		return 'sent';
+	});
+	client.implementWorkflow(parent, async ({ step }) => {
+		await step.runWorkflow(child, {}, { name: 'delivery' });
+		return 'done';
+	});
+	const worker = client.newWorker({ concurrency: 1 });
+	await client.runWorkflow(parent, {});
+	let states: { workflowName: string; status: string; workerId: string | null }[] = [];
+	const shutdown = createShutdown({
+		worker,
+		closeQueue: async () => {
+			const page = await backend.listWorkflowRuns({ limit: 10 });
+			states = page.data.map(({ workflowName, status, workerId }) => ({
+				workflowName,
+				status,
+				workerId
+			}));
+			await backend.stop();
+		},
+		closeDatabase: async () => {}
+	});
+	try {
+		const claimedCount = await worker.tick();
+		expect(claimedCount).toBe(1);
+		await worker.stop();
+		await shutdown();
+		expect(states).toEqual(
+			expect.arrayContaining([
+				{ workflowName: child.name, status: 'pending', workerId: null },
+				{ workflowName: parent.name, status: 'running', workerId: null }
+			])
+		);
+		expect(states).toHaveLength(2);
+		expect(childCalls).toBe(0);
+	} finally {
+		await shutdown();
+	}
+});

apps/worker/src/services/lifecycle.test.ts

@@ -0,0 +1,124 @@
+import { createServer } from 'node:http';
+import { afterEach, expect, test, vi } from 'vitest';
+import { createShutdown } from './lifecycle.js';
+
+afterEach(() => vi.restoreAllMocks());
+
+test('repeated shutdown requests share a promise and close resources in order', async () => {
+	const events: string[] = [];
+	const activePass = Promise.withResolvers<void>();
+	const server = createServer();
+	vi.spyOn(server, 'listening', 'get').mockReturnValue(true);
+	let finishHttp = () => {};
+	vi.spyOn(server, 'close').mockImplementation((callback) => {
+		events.push('http:stop');
+		finishHttp = () => {
+			events.push('http:closed');
+			callback?.();
+		};
+		return server;
+	});
+	const stopWorker = vi.fn(async () => {
+		events.push('worker:stop');
+		await activePass.promise;
+		events.push('worker:stopped');
+	});
+	const shutdown = createShutdown({
+		server,
+		worker: { stop: stopWorker },
+		stopWatch: () => events.push('watch:stop'),
+		closeQueue: async () => {
+			events.push('queue:close');
+		},
+		closeDatabase: async () => {
+			events.push('database:close');
+		}
+	});
+
+	const first = shutdown();
+	expect(shutdown()).toBe(first);
+	expect(events).toEqual(['watch:stop', 'http:stop', 'worker:stop']);
+	activePass.resolve();
+	await vi.waitFor(() => expect(events).toContain('worker:stopped'));
+	expect(events).not.toContain('queue:close');
+	finishHttp();
+	await first;
+	expect(events).toEqual([
+		'watch:stop',
+		'http:stop',
+		'worker:stop',
+		'worker:stopped',
+		'http:closed',
+		'queue:close',
+		'database:close'
+	]);
+	expect(shutdown()).toBe(first);
+	expect(stopWorker).toHaveBeenCalledOnce();
+});
+
+test('partial startup closes the queue and database without a worker or server', async () => {
+	const closeQueue = vi.fn(async () => {});
+	const closeDatabase = vi.fn(async () => {});
+	const shutdown = createShutdown({ closeQueue, closeDatabase });
+	await shutdown();
+	await shutdown();
+	expect(closeQueue).toHaveBeenCalledOnce();
+	expect(closeDatabase).toHaveBeenCalledOnce();
+	expect(closeQueue.mock.invocationCallOrder[0]).toBeLessThan(
+		closeDatabase.mock.invocationCallOrder[0]
+	);
+});
+
+test('a failed worker stop leaves its database connections open', async () => {
+	const error = new Error('worker stop failed');
+	const closeQueue = vi.fn(async () => {});
+	const closeDatabase = vi.fn(async () => {});
+	const stop = vi.fn().mockRejectedValue(error);
+	const shutdown = createShutdown({ worker: { stop }, closeQueue, closeDatabase });
+	const pending = shutdown();
+	await expect(pending).rejects.toBe(error);
+	expect(shutdown()).toBe(pending);
+	expect(stop).toHaveBeenCalledOnce();
+	expect(closeQueue).not.toHaveBeenCalled();
+	expect(closeDatabase).not.toHaveBeenCalled();
+});
+
+test('a queue close failure still closes the application database', async () => {
+	const error = new Error('queue close failed');
+	const closeDatabase = vi.fn(async () => {});
+	const shutdown = createShutdown({
+		closeQueue: vi.fn().mockRejectedValue(error),
+		closeDatabase
+	});
+	await expect(shutdown()).rejects.toBe(error);
+	expect(closeDatabase).toHaveBeenCalledOnce();
+});
+
+test.each(['callback', 'throw'])(
+	'an HTTP close %s error is reported after database cleanup',
+	async (mode) => {
+		const error = new Error('HTTP close failed');
+		const server = createServer();
+		vi.spyOn(server, 'listening', 'get').mockReturnValue(true);
+		vi.spyOn(server, 'close').mockImplementation((callback) => {
+			if (mode === 'throw') throw error;
+			callback?.(error);
+			return server;
+		});
+		const closeQueue = vi.fn(async () => {});
+		const closeDatabase = vi.fn(async () => {});
+		const shutdown = createShutdown({ server, closeQueue, closeDatabase });
+		await expect(shutdown()).rejects.toBe(error);
+		expect(closeQueue).toHaveBeenCalledOnce();
+		expect(closeDatabase).toHaveBeenCalledOnce();
+	}
+);
+
+test('a database close failure is propagated and not retried by another signal', async () => {
+	const error = new Error('database close failed');
+	const closeDatabase = vi.fn().mockRejectedValue(error);
+	const shutdown = createShutdown({ closeQueue: async () => {}, closeDatabase });
+	await expect(shutdown()).rejects.toBe(error);
+	await expect(shutdown()).rejects.toBe(error);
+	expect(closeDatabase).toHaveBeenCalledOnce();
+});

apps/worker/src/services/lifecycle.ts

@@ -0,0 +1,38 @@
+import type { Server } from 'node:http';
+import type { Worker } from 'openworkflow';
+
+export interface ShutdownResources {
+	worker?: Pick<Worker, 'stop'>;
+	server?: Server;
+	stopWatch?: () => void;
+	closeQueue(): Promise<void>;
+	closeDatabase(): Promise<void>;
+}
+
+export function createShutdown(resources: ShutdownResources): () => Promise<void> {
+	let shutdownPromise: Promise<void> | undefined;
+	return () => {
+		shutdownPromise ??= stopResources(resources);
+		return shutdownPromise;
+	};
+}
+
+async function stopResources(resources: ShutdownResources): Promise<void> {
+	resources.stopWatch?.();
+	const serverClosed = new Promise<unknown>((resolve) => {
+		try {
+			if (resources.server?.listening) resources.server.close(resolve);
+			else resolve(undefined);
+		} catch (error) {
+			resolve(error);
+		}
+	});
+	await resources.worker?.stop();
+	const serverError = await serverClosed;
+	try {
+		await resources.closeQueue();
+	} finally {
+		await resources.closeDatabase();
+	}
+	if (serverError) throw serverError;
+}

apps/worker/src/services/startup.ts

@@ -0,0 +1,94 @@
+import { once } from 'node:events';
+import { watchConfig, type WhenConfiguration } from '@when/config';
+import { runMigrations } from '@when/db';
+import {
+	expirePendingAppointments,
+	initOpenWorkflow,
+	refreshCalendars,
+	syncCalendars
+} from '@when/jobs';
+import { setLogger } from '@when/calendar';
+import type { OpenWorkflow } from 'openworkflow';
+import { createMailer } from '../email/smtp.js';
+import { registerWorkflows } from '../workflows/index.js';
+import { applyConfig } from './config-reload.js';
+import { setWorkerContext, type WorkerContext } from './context.js';
+import { createEngineLogger } from './engine-logger.js';
+import { createHealthServer } from './health.js';
+import type { ShutdownResources } from './lifecycle.js';
+import { logger } from './logger.js';
+
+const DEFAULT_PORT = 9000;
+
+export async function initializeWorkerContext(
+	config: WhenConfiguration,
+	db: WorkerContext['db']
+): Promise<WorkerContext> {
+	const applied = await runMigrations(db);
+	if (applied.length > 0) logger.info({ migrations: applied }, 'migrations applied');
+
+	const ctx: WorkerContext = { config, logger, db, mailer: createMailer(config, logger) };
+	setWorkerContext(ctx);
+	setLogger(logger);
+	return ctx;
+}
+
+export async function startWorkflowWorker(
+	ctx: WorkerContext,
+	resources: ShutdownResources
+): Promise<OpenWorkflow> {
+	const client = initOpenWorkflow({
+		dbPath: ctx.config.database.queue,
+		logger: createEngineLogger(ctx.logger)
+	});
+	registerWorkflows();
+	const worker = client.newWorker();
+	resources.worker = worker;
+	await worker.start();
+	ctx.logger.info('worker started');
+	return client;
+}
+
+export async function enqueueStartupMaintenance(client: OpenWorkflow): Promise<void> {
+	await client.runWorkflow(refreshCalendars, {});
+	await client.runWorkflow(expirePendingAppointments, {});
+	await client.runWorkflow(syncCalendars, {});
+}
+
+export async function startHealthServer(
+	ctx: WorkerContext,
+	resources: ShutdownResources
+): Promise<void> {
+	const port = Number(process.env.PORT) || DEFAULT_PORT;
+	const server = createHealthServer(() => ctx.config);
+	resources.server = server;
+	server.listen(port);
+	await once(server, 'listening');
+	ctx.logger.info({ port }, 'health server listening');
+}
+
+export function watchConfigAndSignals(
+	ctx: WorkerContext,
+	resources: ShutdownResources,
+	shutdown: () => Promise<void>
+): void {
+	const requestShutdown = (signal: string): void => {
+		ctx.logger.info({ signal }, 'worker shutting down');
+		void shutdown().then(
+			() => process.exit(0),
+			(error) => {
+				ctx.logger.error(
+					{ error: error instanceof Error ? error.message : String(error) },
+					'worker shutdown failed'
+				);
+				process.exit(1);
+			}
+		);
+	};
+
+	resources.stopWatch = watchConfig((result) =>
+		applyConfig(result, ctx, () => requestShutdown('config-change'))
+	);
+	process.on('SIGTERM', () => requestShutdown('SIGTERM'));
+	process.on('SIGINT', () => requestShutdown('SIGINT'));
+}

apps/worker/src/workflows/expire-pending-appointments.ts

@@ -0,0 +1,27 @@
+import { expireStalePending } from '@when/db';
+import { expirePendingAppointments, type ExpirePendingAppointmentsResult } from '@when/jobs';
+import { systemClock, type Clock } from '../email/clock.js';
+import { getWorkerContext } from '../services/context.js';
+import { implementObservedWorkflow } from '../services/metrics.js';
+
+interface ExpiryStep {
+	run<T>(config: { name: string }, fn: () => Promise<T>): Promise<T>;
+}
+
+export async function runExpirePendingAppointments(
+	step: ExpiryStep,
+	clock: Clock = systemClock
+): Promise<ExpirePendingAppointmentsResult> {
+	await step.run({ name: 'expire-pending-appointments' }, async () => {
+		const ctx = getWorkerContext();
+		const count = await expireStalePending(ctx.db, clock.now().toISOString());
+		if (count > 0) ctx.logger.info({ count }, 'expired stale pending requests');
+	});
+	return 'expired';
+}
+
+export function registerExpirePendingAppointmentsWorkflow(): void {
+	implementObservedWorkflow(expirePendingAppointments, ({ step }) =>
+		runExpirePendingAppointments(step)
+	);
+}

apps/worker/src/workflows/index.ts

@@ -5,6 +5,9 @@ import { registerTestEmailWorkflow } from './test-email.js';
 import { registerProbeProviderWorkflows } from './probe-provider.js';
 import { registerProbeCalendarWorkflow } from './probe-calendar.js';
 import { registerGenerateVideoChatWorkflow } from './generate-video-chat.js';
+import { registerSyncCalendarsWorkflow } from './sync-calendars.js';
+import { registerRefreshCalendarsWorkflow } from './refresh-calendars.js';
+import { registerExpirePendingAppointmentsWorkflow } from './expire-pending-appointments.js';
 
 /**
  * Register every workflow implementation on the openworkflow singleton. Called
@@ -18,4 +21,7 @@ export function registerWorkflows(): void {
 	registerProbeProviderWorkflows();
 	registerProbeCalendarWorkflow();
 	registerGenerateVideoChatWorkflow();
+	registerSyncCalendarsWorkflow();
+	registerRefreshCalendarsWorkflow();
+	registerExpirePendingAppointmentsWorkflow();
 }

apps/worker/src/workflows/maintenance.integration.test.ts

@@ -0,0 +1,376 @@
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { afterEach, beforeEach, expect, test, vi } from 'vitest';
+import { OpenWorkflow, type Worker } from 'openworkflow';
+import { BackendSqlite } from 'openworkflow/sqlite';
+import { openDb, runMigrations } from '@when/db';
+import {
+	refreshCalendars,
+	expirePendingAppointments,
+	syncCalendars,
+	testCalendar
+} from '@when/jobs';
+import { fetchBusyIntervals, pushAppointment } from '@when/calendar';
+import { sampleAppointment, sampleConfig } from '../email/__fixtures__/appointment.js';
+import { setWorkerContext, type WorkerContext } from '../services/context.js';
+import { createLogger } from '../services/logger.js';
+import { registerWorkflows } from './index.js';
+
+const engine = vi.hoisted(() => ({ client: null as OpenWorkflow | null }));
+vi.mock('@when/jobs', async (importOriginal) => {
+	const actual = await importOriginal<typeof import('@when/jobs')>();
+	return { ...actual, getOpenWorkflow: () => engine.client! };
+});
+vi.mock('@when/calendar', async (importOriginal) => {
+	const actual = await importOriginal<typeof import('@when/calendar')>();
+	return { ...actual, fetchBusyIntervals: vi.fn(), pushAppointment: vi.fn() };
+});
+
+const specs = [refreshCalendars, expirePendingAppointments, syncCalendars];
+const at = '2026-04-01T07:03:19.000Z';
+let directory: string;
+let backend: BackendSqlite;
+let client: OpenWorkflow;
+let worker: Worker;
+let ctx: WorkerContext;
+
+beforeEach(async () => {
+	vi.useFakeTimers({ toFake: ['Date'] });
+	vi.setSystemTime(new Date(at));
+	vi.mocked(fetchBusyIntervals).mockReset().mockResolvedValue([]);
+	vi.mocked(pushAppointment).mockReset().mockResolvedValue({ ok: false, reason: 'offline' });
+	directory = await mkdtemp(join(tmpdir(), 'when-maintenance-'));
+	const db = openDb(join(directory, 'app.sqlite'));
+	await runMigrations(db);
+	ctx = {
+		db,
+		config: { ...sampleConfig, providers: {} },
+		logger: createLogger(),
+		mailer: { send: async () => ({ ok: true }) }
+	};
+	setWorkerContext(ctx);
+	backend = BackendSqlite.connect(join(directory, 'queue.sqlite'));
+	client = new OpenWorkflow({ backend });
+	engine.client = client;
+	registerWorkflows();
+	worker = client.newWorker({ concurrency: 1 });
+});
+
+afterEach(async () => {
+	try {
+		await worker.stop();
+		await backend.stop();
+		await ctx.db.destroy();
+	} finally {
+		vi.useRealTimers();
+		await rm(directory, { recursive: true, force: true });
+	}
+});
+
+async function schedules() {
+	const page = await client.schedules.list({ limit: 100 });
+	expect(page.pagination.next).toBeNull();
+	return page.data;
+}
+
+async function runs() {
+	const page = await backend.listWorkflowRuns({ limit: 100 });
+	expect(page.pagination.next).toBeNull();
+	return page.data;
+}
+
+async function drain(count: number) {
+	for (let i = 0; i < count; i++) {
+		const claimedCount = await worker.tick();
+		expect(claimedCount).toBe(1);
+		await worker.stop();
+	}
+}
+
+function useCalendar() {
+	ctx.config = {
+		...ctx.config,
+		providers: {
+			dav: {
+				type: 'caldav',
+				url: 'https://calendar.example.com',
+				username: 'owner',
+				password: 'secret',
+				calendars: {
+					work: { href: '/work/', sync: { refresh_every_minutes: 10 } }
+				}
+			}
+		},
+		meetings: {
+			chat: {
+				title: 'Chat',
+				booking_calendar: 'work',
+				additional_busy_calendars: [],
+				booking_window_days: 60
+			}
+		} as unknown as WorkerContext['config']['meetings']
+	};
+}
+
+test('producer-only enqueue and implementation registration persist no schedules', async () => {
+	const run = await client.runWorkflow(syncCalendars, {});
+	const schedulesBeforeDrain = await schedules();
+	expect(schedulesBeforeDrain).toEqual([]);
+	const pendingRuns = await runs();
+	expect(pendingRuns).toMatchObject([
+		{ id: run.workflowRun.id, workflowName: 'sync-calendars', input: {}, status: 'pending' }
+	]);
+	await drain(1);
+	const syncResult = await run.result({ timeoutMs: 500 });
+	expect(syncResult).toBe('requested');
+	const schedulesAfterDrain = await schedules();
+	expect(schedulesAfterDrain).toEqual([]);
+});
+
+test('native start upserts declarations and preserves pause across a file-backed restart', async () => {
+	await worker.start();
+	await worker.stop();
+	const initialSchedules = await schedules();
+	expect(initialSchedules).toHaveLength(3);
+	await client.scheduleWorkflow(
+		refreshCalendars,
+		{},
+		{
+			id: 'when:refresh-calendars',
+			cron: '*/7 * * * *',
+			timezone: 'UTC'
+		}
+	);
+	await client.schedules.pause('when:sync-calendars');
+	await backend.stop();
+	backend = BackendSqlite.connect(join(directory, 'queue.sqlite'));
+	client = new OpenWorkflow({ backend });
+	engine.client = client;
+	registerWorkflows();
+	worker = client.newWorker({ concurrency: 1 });
+	await worker.start();
+	await worker.stop();
+	const persisted = await schedules();
+	expect(persisted).toHaveLength(3);
+	for (const spec of specs) {
+		const declaration = spec.schedules![0]!;
+		expect(persisted.find((schedule) => schedule.id === declaration.id)).toMatchObject({
+			...declaration,
+			cron: `0 ${declaration.cron}`,
+			workflowName: spec.name
+		});
+	}
+	const pausedSchedule = await client.schedules.get('when:sync-calendars');
+	expect(pausedSchedule).toMatchObject({ nextRunAt: null });
+	const resumedSchedule = await client.schedules.resume('when:sync-calendars');
+	expect(resumedSchedule).toMatchObject({
+		nextRunAt: new Date('2026-04-01T07:10:00Z')
+	});
+});
+
+test('UTC cron aligns to wall-clock boundaries rather than startup time', async () => {
+	await worker.start();
+	await worker.stop();
+	const persisted = await schedules();
+	const expected = new Map([
+		['when:refresh-calendars', '2026-04-01T07:05:00.000Z'],
+		['when:expire-pending-appointments', '2026-04-01T08:00:00.000Z'],
+		['when:sync-calendars', '2026-04-01T07:10:00.000Z']
+	]);
+	for (const schedule of persisted) {
+		expect(schedule.timezone).toBe('UTC');
+		expect(schedule.nextRunAt?.toISOString()).toBe(expected.get(schedule.id));
+	}
+	const enqueuedBeforeFirstBoundary = await backend.enqueueScheduledRuns({
+		before: new Date('2026-04-01T07:04:59Z'),
+		limit: 100
+	});
+	expect(enqueuedBeforeFirstBoundary).toBe(0);
+	const enqueuedAtFirstBoundary = await backend.enqueueScheduledRuns({
+		before: new Date('2026-04-01T07:05:00Z'),
+		limit: 100
+	});
+	expect(enqueuedAtFirstBoundary).toBe(1);
+	const enqueuedBeforeSecondBoundary = await backend.enqueueScheduledRuns({
+		before: new Date('2026-04-01T07:09:59Z'),
+		limit: 100
+	});
+	expect(enqueuedBeforeSecondBoundary).toBe(0);
+	const enqueuedAtSecondBoundary = await backend.enqueueScheduledRuns({
+		before: new Date('2026-04-01T07:10:00Z'),
+		limit: 100
+	});
+	expect(enqueuedAtSecondBoundary).toBe(2);
+	vi.setSystemTime(new Date('2026-04-01T07:10:00Z'));
+	await drain(3);
+	const completedRuns = await runs();
+	expect(completedRuns.every((run) => run.status === 'completed')).toBe(true);
+});
+
+test('downtime catches up once per schedule and explicit startup passes coexist with cron', async () => {
+	await worker.start();
+	await worker.stop();
+	vi.setSystemTime(new Date('2026-04-03T15:37:00Z'));
+	const before = new Date();
+	const catchupCount = await backend.enqueueScheduledRuns({ before, limit: 100 });
+	expect(catchupCount).toBe(3);
+	const repeatedCatchupCount = await backend.enqueueScheduledRuns({ before, limit: 100 });
+	expect(repeatedCatchupCount).toBe(0);
+	for (const spec of specs) await client.runWorkflow(spec, {});
+	await drain(6);
+	const completed = await runs();
+	expect(completed).toHaveLength(6);
+	for (const spec of specs) {
+		const matching = completed.filter((run) => run.workflowName === spec.name);
+		expect(matching).toHaveLength(2);
+		expect(matching.map((run) => run.workflowScheduleId)).toEqual(
+			expect.arrayContaining([null, spec.schedules![0]!.id])
+		);
+		for (const run of matching) {
+			expect(run).toMatchObject({ status: 'completed', input: {}, idempotencyKey: null });
+			const attempts = await backend.listStepAttempts({ workflowRunId: run.id, limit: 100 });
+			expect(attempts.data).toMatchObject([{ status: 'completed', kind: 'function' }]);
+		}
+	}
+	for (const schedule of await schedules()) {
+		expect(schedule.nextRunAt!.getTime()).toBeGreaterThan(before.getTime());
+	}
+});
+
+test('paused cron survives startup but does not suppress explicit startup work', async () => {
+	await worker.start();
+	await worker.stop();
+	for (const schedule of await schedules()) await client.schedules.pause(schedule.id);
+	vi.setSystemTime(new Date('2026-04-03T15:37:00Z'));
+	await worker.start();
+	await worker.stop();
+	const pausedSchedules = await schedules();
+	expect(pausedSchedules.every((schedule) => schedule.nextRunAt === null)).toBe(true);
+	const enqueuedCount = await backend.enqueueScheduledRuns({ before: new Date(), limit: 100 });
+	expect(enqueuedCount).toBe(0);
+	for (const spec of specs) await client.runWorkflow(spec, {});
+	await drain(3);
+	const completedRuns = await runs();
+	expect(completedRuns).toMatchObject([
+		{ status: 'completed', workflowScheduleId: null },
+		{ status: 'completed', workflowScheduleId: null },
+		{ status: 'completed', workflowScheduleId: null }
+	]);
+});
+
+test('repeated sync requests serialize without coalescing and stop awaits the active pass', async () => {
+	useCalendar();
+	await ctx.db
+		.insertInto('appointments')
+		.values({ ...sampleAppointment, event_type_id: 'chat' })
+		.execute();
+	const release = Promise.withResolvers<void>();
+	let entered = false;
+	let active = 0;
+	let maximumActive = 0;
+	vi.mocked(pushAppointment).mockImplementation(async () => {
+		active += 1;
+		maximumActive = Math.max(active, maximumActive);
+		entered = true;
+		await release.promise;
+		active -= 1;
+		return { ok: false, reason: 'offline' };
+	});
+	const first = await client.runWorkflow(syncCalendars, {});
+	try {
+		const claimedCount = await worker.tick();
+		expect(claimedCount).toBe(1);
+		await vi.waitUntil(() => entered, { timeout: 1000, interval: 5 });
+		await client.runWorkflow(syncCalendars, {});
+		await client.runWorkflow(syncCalendars, {});
+		const claimedWhileActive = await worker.tick();
+		expect(claimedWhileActive).toBe(0);
+		let stopped = false;
+		const stopping = worker.stop().then(() => {
+			stopped = true;
+		});
+		await Promise.resolve();
+		expect(stopped).toBe(false);
+		release.resolve();
+		await stopping;
+		const syncResult = await first.result({ timeoutMs: 500 });
+		expect(syncResult).toBe('requested');
+		const runsAfterStop = await runs();
+		expect(runsAfterStop.filter((run) => run.status === 'pending')).toHaveLength(2);
+		await drain(2);
+		const completedRuns = await runs();
+		expect(completedRuns.every((run) => run.status === 'completed')).toBe(true);
+		expect(pushAppointment).toHaveBeenCalledTimes(3);
+		expect(maximumActive).toBe(1);
+		const runsForUniqueIds = await runs();
+		expect(new Set(runsForUniqueIds.map((run) => run.id)).size).toBe(3);
+	} finally {
+		release.resolve();
+	}
+});
+
+test('slow refresh blocks a queued probe at concurrency one until the real provider call finishes', async () => {
+	useCalendar();
+	const release = Promise.withResolvers<void>();
+	let entered = false;
+	vi.mocked(fetchBusyIntervals).mockImplementationOnce(async () => {
+		entered = true;
+		await release.promise;
+		return [];
+	});
+	const refresh = await client.runWorkflow(refreshCalendars, {});
+	try {
+		const claimedCount = await worker.tick();
+		expect(claimedCount).toBe(1);
+		await vi.waitUntil(() => entered, { timeout: 1000, interval: 5 });
+		const probe = await client.runWorkflow(testCalendar, { name: 'work' });
+		const claimedWhileActive = await worker.tick();
+		expect(claimedWhileActive).toBe(0);
+		release.resolve();
+		await worker.stop();
+		const refreshResult = await refresh.result({ timeoutMs: 500 });
+		expect(refreshResult).toBe('refreshed');
+		const pendingProbe = await backend.getWorkflowRun({ workflowRunId: probe.workflowRun.id });
+		expect(pendingProbe).toMatchObject({
+			status: 'pending'
+		});
+		expect(fetchBusyIntervals).toHaveBeenCalledTimes(1);
+		await drain(1);
+		const probeResult = await probe.result({ timeoutMs: 500 });
+		expect(probeResult).toEqual({ busyCount: 0, days: 60 });
+		expect(fetchBusyIntervals).toHaveBeenCalledTimes(2);
+	} finally {
+		release.resolve();
+	}
+});
+
+test.each(specs)(
+	'$name local failure uses ten native attempts with 1s ×2 backoff capped at 100s',
+	async (spec) => {
+		await ctx.db.schema.dropTable('appointments').execute();
+		const handle = await client.runWorkflow(spec, {});
+		const delays = [1000, 2000, 4000, 8000, 16000, 32000, 64000, 100000, 100000];
+		for (let i = 0; i < 10; i++) {
+			const started = Date.now();
+			const claimedCount = await worker.tick();
+			expect(claimedCount).toBe(1);
+			await worker.stop();
+			const run = await backend.getWorkflowRun({ workflowRunId: handle.workflowRun.id });
+			const attempts = await backend.listStepAttempts({
+				workflowRunId: handle.workflowRun.id,
+				limit: 100
+			});
+			expect(attempts.data).toHaveLength(i + 1);
+			expect(attempts.data.every((attempt) => attempt.status === 'failed')).toBe(true);
+			if (i < 9) {
+				expect(run).toMatchObject({ status: 'pending', workerId: null });
+				expect(run!.availableAt!.getTime() - started).toBe(delays[i]);
+				vi.setSystemTime(run!.availableAt!);
+			} else {
+				expect(run).toMatchObject({ status: 'failed', output: null, workerId: null });
+			}
+		}
+		await expect(handle.result({ timeoutMs: 500 })).rejects.toThrow();
+	}
+);

apps/worker/src/workflows/maintenance.test.ts

@@ -0,0 +1,183 @@
+import { afterEach, beforeEach, expect, test, vi } from 'vitest';
+import { openDb, parseActionLog, runMigrations } from '@when/db';
+import { scanOnce } from '../calendar/sync.js';
+import { refreshCycle } from '../calendar/refresh.js';
+import { sampleAppointment, sampleConfig } from '../email/__fixtures__/appointment.js';
+import { setWorkerContext, type WorkerContext } from '../services/context.js';
+import { createLogger } from '../services/logger.js';
+import { calendarSyncTotal } from '../services/metrics.js';
+import { runSyncCalendars } from './sync-calendars.js';
+import { runRefreshCalendars } from './refresh-calendars.js';
+import { runExpirePendingAppointments } from './expire-pending-appointments.js';
+
+vi.mock('../calendar/sync.js', () => ({ scanOnce: vi.fn() }));
+vi.mock('../calendar/refresh.js', () => ({ refreshCycle: vi.fn() }));
+
+const at = '2026-04-01T12:00:00.000Z';
+const clock = { now: () => new Date(at), nowMs: () => Date.parse(at) };
+let ctx: WorkerContext;
+
+function makeStep() {
+	const configs: { name: string }[] = [];
+	return {
+		configs,
+		async run<T>(config: { name: string }, fn: () => Promise<T>): Promise<T> {
+			configs.push(config);
+			const result = await fn();
+			return result;
+		}
+	};
+}
+
+beforeEach(async () => {
+	vi.resetAllMocks();
+	calendarSyncTotal.reset();
+	const db = openDb(':memory:');
+	await runMigrations(db);
+	ctx = {
+		db,
+		config: sampleConfig,
+		logger: createLogger(),
+		mailer: { send: async () => ({ ok: true }) }
+	};
+	vi.spyOn(ctx.logger, 'info');
+	setWorkerContext(ctx);
+});
+
+afterEach(async () => {
+	vi.restoreAllMocks();
+	await ctx.db.destroy();
+});
+
+test('sync awaits the real callback and counts only completed physical scans', async () => {
+	const scan = Promise.withResolvers<void>();
+	vi.mocked(scanOnce).mockReturnValue(scan.promise);
+	const step = makeStep();
+	let completed = false;
+	const running = runSyncCalendars(step).then((result) => {
+		completed = true;
+		return result;
+	});
+	await Promise.resolve();
+	expect(completed).toBe(false);
+	const metricsBeforeCompletion = await calendarSyncTotal.get();
+	expect(metricsBeforeCompletion.values).toEqual([]);
+	expect(scanOnce).toHaveBeenCalledWith(ctx);
+	scan.resolve();
+	const scanResult = await running;
+	expect(scanResult).toBe('requested');
+	expect(step.configs).toEqual([{ name: 'scan-calendars' }]);
+	const metricsAfterCompletion = await calendarSyncTotal.get();
+	expect(metricsAfterCompletion.values).toEqual([
+		expect.objectContaining({ labels: { status: 'success' }, value: 1 })
+	]);
+});
+
+test('sync callback errors escape with failure metrics', async () => {
+	const error = new Error('local scan read failed');
+	vi.mocked(scanOnce).mockRejectedValue(error);
+	await expect(runSyncCalendars(makeStep())).rejects.toBe(error);
+	const failureMetrics = await calendarSyncTotal.get();
+	expect(failureMetrics.values).toEqual([
+		expect.objectContaining({ labels: { status: 'failure' }, value: 1 })
+	]);
+});
+
+test('an opaque engine escape is not counted as a failed scan', async () => {
+	const error = new Error('opaque engine escape');
+	const step = { run: vi.fn().mockRejectedValue(error) };
+	await expect(runSyncCalendars(step)).rejects.toBe(error);
+	expect(scanOnce).not.toHaveBeenCalled();
+	const metrics = await calendarSyncTotal.get();
+	expect(metrics.values).toEqual([]);
+});
+
+test('refresh awaits the cycle with injectable options and native step defaults', async () => {
+	const cycle = Promise.withResolvers<void>();
+	vi.mocked(refreshCycle).mockReturnValue(cycle.promise);
+	const opts = { now: Temporal.Instant.from(at), via: 'refresh' as const };
+	const step = makeStep();
+	let completed = false;
+	const running = runRefreshCalendars(step, opts).then((result) => {
+		completed = true;
+		return result;
+	});
+	await Promise.resolve();
+	expect(completed).toBe(false);
+	expect(refreshCycle).toHaveBeenCalledWith(ctx, opts);
+	cycle.resolve();
+	const refreshResult = await running;
+	expect(refreshResult).toBe('refreshed');
+	expect(step.configs).toEqual([{ name: 'refresh-calendars' }]);
+});
+
+test('refresh local errors and engine escapes propagate unchanged', async () => {
+	const error = new Error('mirror write failed');
+	vi.mocked(refreshCycle).mockRejectedValue(error);
+	await expect(runRefreshCalendars(makeStep())).rejects.toBe(error);
+	const engineError = new Error('opaque engine escape');
+	await expect(runRefreshCalendars({ run: vi.fn().mockRejectedValue(engineError) })).rejects.toBe(
+		engineError
+	);
+});
+
+test('expiry uses the injected clock, is idempotent, and logs the changed count', async () => {
+	const cases = [
+		{ id: 'stale', status: 'pending', start: '2026-04-01T11:00:00Z', expectedStatus: 'expired' },
+		{ id: 'boundary', status: 'pending', start: at, expectedStatus: 'expired' },
+		{ id: 'future', status: 'pending', start: '2026-04-01T13:00:00Z', expectedStatus: 'pending' },
+		{
+			id: 'confirmed',
+			status: 'confirmed',
+			start: '2026-04-01T11:00:00Z',
+			expectedStatus: 'confirmed'
+		},
+		{ id: 'purged', status: 'purged', start: '2026-04-01T11:00:00Z', expectedStatus: 'purged' }
+	] as const;
+	for (const { id, status, start } of cases) {
+		await ctx.db
+			.insertInto('appointments')
+			.values({
+				...sampleAppointment,
+				id,
+				event_type_id: id,
+				cancel_token: `token-${id}`,
+				status,
+				start_time: start,
+				action_log: '[]'
+			})
+			.execute();
+	}
+	const step = makeStep();
+	const firstExpiryResult = await runExpirePendingAppointments(step, clock);
+	expect(firstExpiryResult).toBe('expired');
+	const repeatedExpiryResult = await runExpirePendingAppointments(step, clock);
+	expect(repeatedExpiryResult).toBe('expired');
+	for (const { id, expectedStatus } of cases) {
+		const row = await ctx.db
+			.selectFrom('appointments')
+			.selectAll()
+			.where('id', '=', id)
+			.executeTakeFirstOrThrow();
+		expect(row.status).toBe(expectedStatus);
+		if (expectedStatus === 'expired') {
+			expect(parseActionLog(row.action_log)).toEqual([{ action: 'expire', actor: 'system', at }]);
+		} else {
+			expect(parseActionLog(row.action_log)).toEqual([]);
+		}
+	}
+	expect(ctx.logger.info).toHaveBeenCalledExactlyOnceWith(
+		{ count: 2 },
+		'expired stale pending requests'
+	);
+	expect(step.configs).toEqual([
+		{ name: 'expire-pending-appointments' },
+		{ name: 'expire-pending-appointments' }
+	]);
+});
+
+test('expiry local SQL errors escape rather than completing the pass', async () => {
+	await ctx.db.schema.dropTable('appointments').execute();
+	await expect(runExpirePendingAppointments(makeStep(), clock)).rejects.toThrow();
+	expect(ctx.logger.info).not.toHaveBeenCalled();
+});

apps/worker/src/workflows/refresh-calendars.ts

@@ -0,0 +1,22 @@
+import { refreshCalendars, type RefreshCalendarsResult } from '@when/jobs';
+import { refreshCycle, type RefreshOptions } from '../calendar/refresh.js';
+import { getWorkerContext } from '../services/context.js';
+import { implementObservedWorkflow } from '../services/metrics.js';
+
+interface RefreshStep {
+	run<T>(config: { name: string }, fn: () => Promise<T>): Promise<T>;
+}
+
+export async function runRefreshCalendars(
+	step: RefreshStep,
+	opts: RefreshOptions = {}
+): Promise<RefreshCalendarsResult> {
+	await step.run({ name: 'refresh-calendars' }, async () => {
+		await refreshCycle(getWorkerContext(), opts);
+	});
+	return 'refreshed';
+}
+
+export function registerRefreshCalendarsWorkflow(): void {
+	implementObservedWorkflow(refreshCalendars, ({ step }) => runRefreshCalendars(step));
+}

apps/worker/src/workflows/sync-calendars.ts

@@ -1,10 +1,26 @@
 import { syncCalendars, type SyncCalendarsResult } from '@when/jobs';
-import type { CalendarSyncScanner } from '../calendar/sync-scanner.js';
-import { implementObservedWorkflow } from '../services/metrics.js';
+import { scanOnce } from '../calendar/sync.js';
+import { getWorkerContext } from '../services/context.js';
+import { calendarSyncTotal, implementObservedWorkflow } from '../services/metrics.js';
 
-export function registerSyncCalendarsWorkflow(scanner: CalendarSyncScanner): void {
-	implementObservedWorkflow(syncCalendars, (): SyncCalendarsResult => {
-		scanner.requestScan();
-		return 'requested';
+interface SyncStep {
+	run<T>(config: { name: string }, fn: () => Promise<T>): Promise<T>;
+}
+
+export async function runSyncCalendars(step: SyncStep): Promise<SyncCalendarsResult> {
+	await step.run({ name: 'scan-calendars' }, async () => {
+		const ctx = getWorkerContext();
+		try {
+			await scanOnce(ctx);
+			calendarSyncTotal.inc({ status: 'success' });
+		} catch (error) {
+			calendarSyncTotal.inc({ status: 'failure' });
+			throw error;
+		}
 	});
+	return 'requested';
+}
+
+export function registerSyncCalendarsWorkflow(): void {
+	implementObservedWorkflow(syncCalendars, ({ step }) => runSyncCalendars(step));
 }

docs/architecture.md

@@ -51,16 +51,37 @@ appointment request ─► web: write appointment row ─► enqueue job ─►
 
 Jobs run on openworkflow over `node:sqlite`. `packages/jobs` is the single source of
 truth for **what** jobs exist and their input/output shapes
-(`packages/jobs/src/specs.ts`): the web app triggers a run from a spec
-(\`runWorkflow(sendAppointmentEmail, …)\`), and the worker provides the implementation. Both
-sides share types and resolve a workflow by name, so web never imports worker code.
+(`packages/jobs/src/specs.ts`): the web app enqueues `reconcile-appointment` from its
+spec, and the worker provides the implementation. Both sides share types and resolve a
+workflow by name, so web never imports worker code. Web and CLI initialize producer-only
+clients: they neither start a worker nor persist schedules.
 
-- **Durable steps.** Each side-effect (one SMTP send) is a memoized `step.run(...)` with
-  its own retry policy, so a replay never re-sends something already sent.
-- **Idempotency keys** dedupe enqueues within openworkflow's dedup window. Appointment emails
-  key on `appointmentId:kind:ics_sequence` — `ics_sequence` bumps on every reschedule, so
-  a repeat same-kind email isn't swallowed as a duplicate. Calendar sync uses a random
-  key per call (each appointment change should trigger a scan).
+- **Durable steps.** Saved successes are reused on replay. Reconciliation checkpoints
+  prepared role-tagged email envelopes, then starts independent guest/host delivery
+  children with native retries. Native retry waits park and release the execution slot.
+  A crash after a remote effect but before its checkpoint can still duplicate effects
+  or logs: execution is at-least-once, not exactly-once.
+- **Idempotency keys** dedupe enqueues for 24 hours, including matching terminal history;
+  they do not automatically rerun old jobs. Appointment emails key on
+  `appointmentId:kind:ics_sequence`; email-free reconciliation uses
+  `appointmentId:sync:ics_sequence`. Purge keys use sorted chain IDs. Calendar sync
+  uses a random key per call, and the producer awaits enqueueing only.
+- **Native maintenance.** Worker `start()` persists declared UTC schedules, preserving
+  pause state: refresh every five minutes, pending expiry hourly, repair every ten
+  minutes. Startup then independently enqueues all three `{}` passes without options.
+  Cron is wall-clock aligned and catches up once per overdue schedule, not once per
+  missed tick. Pausing cron does not suppress startup passes or already queued work.
+- **One execution slot.** Maintenance handlers await their actual pass; sync's
+  `requested` result means its scan finished. There are no application maintenance
+  timers or coalescers. Repeated requests each get a pass and can build a backlog;
+  slow active maintenance blocks appointment jobs and probes. Expected provider
+  failures are recorded for later passes; unexpected local errors escape to bounded
+  native retries. A completed pass does not mean every provider succeeded.
+
+Email action logs describe attempts; engine history determines terminal outcomes.
+Failed purges retain hidden local personal data and require explicit operator review.
+See [interpreting outcomes](deployment.md#interpreting-outcomes) for detailed semantics
+and the [rollout runbook](deployment.md#grouped-workflow-rollout) for backup requirements.
 
 ## Calendar I/O (off the request path)
 
@@ -92,8 +113,10 @@ delivery:
 
 `packages/db` wraps `node:sqlite` behind a small Kysely dialect (no native addons) and
 owns the schema types, migrations, and runner. Migrations run on boot and are idempotent.
-Both web and worker open the same `when.sqlite`; the worker additionally owns the
-openworkflow queue database.
+Both web and worker open the same `when.sqlite`; web/CLI producers and the worker
+also open the shared openworkflow queue database. Worker shutdown stops config watching and
+HTTP admission, awaits worker stop and HTTP closure, then closes queue and app DBs.
+Worker stop waits for active work to finish or park, not for pending jobs or future retries.
 
 ## Configuration system
 
@@ -145,14 +168,14 @@ moment, not a zone.
   the `auth` section of `docs/config.md` for when it fits.
 - Secrets are injected via `${ENV_VAR}` interpolation in `when.yaml`, never committed.
 - Nothing in SQLite is encrypted at the column level. The database holds guest names,
-  emails and booking history in plain text, so the data directory deserves the same
-  handling as `.env` — and a backup that leaves the host should be encrypted as a whole,
-  which protects that data rather than one field of it.
+  emails and booking history in plain text. The queue also holds prepared email
+  envelopes, personal data, and guest capability links. Treat both databases and their
+  backups like `.env`; encrypt backups that leave the host.
 - Never log raw request bodies, session tokens, `cancel_token` values, or secrets.
 
 #### The guest capability model
 
-Guests have no account, so their link *is* their credential — a `cancel_token` in the query
+Guests have no account, so their link _is_ their credential — a `cancel_token` in the query
 string. A phone-only booking may not even have an email address to send a second factor to, so
 the goal is not to make a leak impossible. It is to keep the blast radius small, bounded in
 time, and reversible by the host.
@@ -166,13 +189,13 @@ time, and reversible by the host.
   token for it. A stale link therefore shows that the meeting moved and stops there; the guest
   was sent the new link when it moved.
 - **Everything expires.** `APPOINTMENT_VIEW_GRACE_DAYS` (7) past the appointment's end, and the
-  same window gates the `.ics` download. A superseded row counts those 7 days from the *move*,
+  same window gates the `.ics` download. A superseded row counts those 7 days from the _move_,
   not from its old slot, which would otherwise still be months away.
 - **The host can revoke.** Rotating from the guest-link dialog mints a fresh token, so every
   link issued so far stops resolving. Offered only on the chain tip, enforced in SQL. No email
   is sent — re-delivery is the host's call, since the address may itself be the leak path.
 
-What a leaked *live* link still permits: cancelling, or moving the appointment — which on a
+What a leaked _live_ link still permits: cancelling, or moving the appointment — which on a
 `require_approval: true` meeting drops it back to `pending` and re-arms host approval. Both
 are written to the action log, both are visible to the host, and both are recoverable. That is
 accepted as proportionate. View-only tokens and emailed confirmation before destructive actions
@@ -185,7 +208,7 @@ These are understood and deliberately not fixed. Recorded so a future reader doe
 them for oversights.
 
 - **`cancel_token` and bearer tokens compare with `!==`, not in constant time.** They are
-  full-entropy random values, which makes remote timing extraction impractical. The *password*
+  full-entropy random values, which makes remote timing extraction impractical. The _password_
   comparison is constant-time, because a human-chosen secret is a different problem.
 - **`/public/[...file]` follows symlinks.** `resolve()` blocks `../`, but a symlink planted
   inside the public directory serves its target. That directory is operator-mounted, and an

docs/deployment.md

@@ -146,15 +146,132 @@ State is two SQLite files under the data directory the compose file mounts
 (`./data` → `/app/data`):
 
 - `when.sqlite` — appointments, mirrored busy times, and service status (shared by web + worker).
-- `openworkflow.sqlite` — the background job queue (owned by the worker).
+- `openworkflow.sqlite` — the background job queue (shared by web/CLI producers and worker).
 
 Branding overrides live in a separate `public/` directory mounted alongside
 (`./public` → `/app/public`, overridable with `WHEN_PUBLIC_DIR`); the web app serves
 them at `/public/...`. The bundled defaults under `/assets/*` are baked into the image
 and need no mount.
 
-Back up the data directory (and `public/`, if you use it) to preserve appointments and
-settings across restarts. Database paths are configurable under `database` in `when.yaml`.
+Back up both databases consistently using SQLite-aware backup handling, not live file
+copies that omit committed WAL data. Include `when.yaml`, its environment secrets, and
+`public/` overrides separately. Database paths are configurable under `database`.
+Both databases are sensitive: the queue includes prepared email envelopes, personal data
+and guest capability links. Encrypt off-host backups and restrict local access.
+
+## Grouped workflow rollout
+
+Deploy web, worker and CLI as one stopped, grouped release. Do not deploy intermediate
+PR images automatically; release/version/tag operations require separate authorization.
+Application tests and the read-only drain/backup procedure are the gates, not an
+assumption of old/new queue compatibility.
+
+1. **Verify the release.** Run affected jobs/worker/web/CLI checks, lint, coverage and
+   builds, workspace checks and browser regressions. Confirm both CI paths execute the
+   co-located integration tests. Keep the currently deployed artifacts available.
+2. **Stop producers**, including web, CLI commands and any other enqueuers. Let the
+   **old worker** drain every unfinished old run. Worker shutdown alone is not draining:
+   future pending jobs, parked retries/parents, running jobs and legacy `sleeping` work
+   all count. Inspect the old queue read-only; do not initialize the new jobs client,
+   since connection runs migrations. If recurring dispatch exists, pause it with the
+   matching deployed API so the queue can drain.
+3. **Halt if it cannot drain.** Require an explicit operator retry/cancel/reset decision
+   after reviewing local and remote state. Never automatically wipe/reset a queue, even
+   an apparently empty one, edit checkpoints, or resume changed old application graphs.
+   Inspect again after any authorized action; cancelling a parent need not cancel its
+   already-started children.
+4. **Stop the old worker**, recheck that no unfinished runs remain, then take a consistent
+   paired app/queue backup while all writers stay stopped. Include committed WAL data
+   with SQLite-aware handling and verify restoration on copies (examples below).
+5. **Deploy the grouped artifacts once**, against the same config and database paths,
+   with no old worker or competing maintenance timers. Native `worker.start()` persists
+   schedules and preserves pause; startup then enqueues all three `{}` passes without
+   options, independently of pause. Inspect startup passes, schedules, service status
+   and queue progress. Cron catches up once per overdue schedule and can coexist with
+   startup work. Repeated scans are not coalesced; slow active maintenance blocks other
+   jobs at concurrency one, while native retry waits park. Do not replay old graphs or
+   automatically rerun terminal history; dedup may return an existing terminal run.
+6. **Rollback only with evidence.** Use a proven compatible binary, or obtain explicit
+   approval for restoring the paired backups and losing all later writes. Old binaries
+   are not assumed compatible with migrated queues. Stop every writer before a paired
+   restore; restore into clean destinations without stale WAL/SHM sidecars. Remote
+   calendar/email effects are not undone by restoring SQLite.
+
+### Read-only drain inspection
+
+Set `QUEUE` to the actual old queue path. With the SQLite CLI, inspect the schema first:
+
+```sh
+sqlite3 -readonly "$QUEUE" '.schema workflow_runs'
+```
+
+The installed `0.11.0` schema uses `workflow_runs`, with `namespace_id`, `id`,
+`workflow_name`, `status`, and `available_at`. Verify these columns against the deployed
+schema; a mismatch requires investigation, not opening it with a newer client. Installed
+public status types/terminal logic recognize `completed`, legacy `succeeded`, `failed`,
+and **`canceled`** (one l) as terminal. `pending`, `running`, and legacy `sleeping` are
+unfinished. Unknown statuses halt the rollout, even if they look terminal.
+
+```sh
+sqlite3 -readonly -header -column "$QUEUE" <<'SQL'
+PRAGMA query_only = ON;
+SELECT status, COUNT(*) AS runs FROM workflow_runs GROUP BY status;
+SELECT namespace_id, id, workflow_name, status, available_at
+FROM workflow_runs
+WHERE status NOT IN ('completed', 'succeeded', 'failed', 'canceled')
+ORDER BY namespace_id, created_at, id;
+SQL
+```
+
+Proceed only when the second query returns no rows and every observed status is known.
+There is deliberately **no due-time filter**: future pending work is unfinished. Inspect
+all namespaces and child runs, not just top-level appointment runs. Do not dump inputs,
+envelopes, errors or capability links into shared logs.
+
+### Paired backup and restore check
+
+With producers and worker stopped, set `APP`, `QUEUE`, and `BACKUP` to the real paths
+and a fresh restricted backup directory. For ordinary paths without single quotes:
+
+```sh
+mkdir -m 700 "$BACKUP"
+sqlite3 -readonly "$APP" ".backup '$BACKUP/when.sqlite'"
+sqlite3 -readonly "$QUEUE" ".backup '$BACKUP/openworkflow.sqlite'"
+mkdir -m 700 "$BACKUP/restore-check"
+cp "$BACKUP/when.sqlite" "$BACKUP/restore-check/when.sqlite"
+cp "$BACKUP/openworkflow.sqlite" "$BACKUP/restore-check/openworkflow.sqlite"
+sqlite3 -readonly "$BACKUP/restore-check/when.sqlite" 'PRAGMA integrity_check;'
+sqlite3 -readonly "$BACKUP/restore-check/openworkflow.sqlite" 'PRAGMA integrity_check;'
+```
+
+Both integrity checks must report `ok`. Compare restored appointment counts and queue
+status counts to the stopped sources, and rerun drain inspection on the restored queue.
+SQLite `.backup` includes committed WAL state; do not copy live main files alone or
+remove source WAL files to make backups look clean. Because every writer is stopped,
+the sequential backups form a consistent pair. Keep that pair together; test copies
+must never reach real providers or become the running deployment.
+
+### Interpreting outcomes
+
+An email action `failed` is an **observed provider attempt**, possibly followed by retries
+and recovery; local errors are not provider-failure observations. `done` requires a
+nonempty intended set where every delivery reported `sent` and the row is still eligible
+at the final write. Mixed/all skipped sets and missing/purged rows get no summary.
+Engine history is authoritative for terminal failure; no aggregate terminal email-failure
+summary is synthesized. An already-started sibling can continue independently. SMTP
+service status is the **last recorded physical attempt outcome**, not an aggregate
+recipient/job result. Handler metrics include replays/suspensions; email metrics count
+physical attempt outcomes. Crashes before checkpoints can duplicate effects or log
+entries: this is at-least-once execution, not an exactly-once ledger.
+
+A failed purge retains the current and later hidden `purged` rows, including local
+personal data. Earlier successful deletions remain deleted. Web selection skips
+already-purged rows, so clicking purge again is not promised recovery. Review retained
+rows, engine history and remote state before choosing explicit operator cleanup; there
+is no automatic terminal rerun or remote-orphan fallback. A completed maintenance pass
+likewise does not guarantee every provider succeeded. See the
+[worker README](../apps/worker/README.md#maintenance-and-shutdown) for cadence and the
+[jobs README](../packages/jobs/README.md#schedule-management) for public schedule management.
 
 ## Config reloads
 

docs/development.md

@@ -105,6 +105,8 @@ suite in both pipelines.
 - **Type safety at boundaries:** validate every external input (YAML, forms, API) against
   the config JSON Schema (ajv) — the repo does not use Zod. Database access goes through
   Kysely.
+- **Async readability:** assign awaited results to named variables before returning
+  them or passing them as arguments.
 - **Time:** never call `new Date()` / `Date.now()` in domain logic; inject a `Clock`.
 - **Comments:** comment non-obvious rationale only. Don't restate what a name already
   says.

packages/jobs/README.md

@@ -12,7 +12,10 @@ recurring schedules.
 
 - `reconcileAppointment` — coordinate appointment calendar/video updates and email.
 - `purgeAppointment` — clean up purged appointment chains.
-- `syncCalendars` — request the worker's calendar repair scan.
+- `syncCalendars` — `{}` → `requested`, after the actual calendar repair scan finishes.
+- `refreshCalendars` — `{}` → `refreshed`, after refresh and conflict checks.
+- `expirePendingAppointments` — `{}` → `expired`, after idempotent expiry SQL.
+  Completed maintenance passes can include recorded provider failures.
 - `testEmail`, `testProvider`, `listProviderCalendars`, `testCalendar`,
   `generateVideoChat` — on-demand worker operations.
 - `initOpenWorkflow({ dbPath, logger? })` — open and migrate the queue database and
@@ -28,10 +31,63 @@ recurring schedules.
 Specs and their input/output types live in `src/specs.ts`. The `./specs` entrypoint
 exports contracts without the process-wide client helpers.
 
-Queue migrations run on connection, including producer connections. Existing step
-results survive the engine upgrade, but that does not make a changed application
-workflow graph safe to replay: drain old appointment work before deploying changed
-handlers. Back up both SQLite databases consistently before an engine upgrade.
+Queue migrations run on connection, including producer connections. Do not start a
+new producer against the old deployment during cutover. Saved results do not make
+changed application graphs safe to replay. Follow the
+[grouped rollout runbook](../../docs/deployment.md#grouped-workflow-rollout): drain all
+unfinished old work before deployment, then stop and back up both SQLite databases
+consistently. Never automatically rerun terminal history. Reusing an enqueue key
+can return an existing terminal run within the engine's 24-hour dedup window.
+
+Prepared envelopes contain personal data and guest capability links; queue files and
+backups are sensitive. Remote effects and logs can duplicate after a crash before
+checkpointing, so neither durable steps nor enqueue keys provide exactly-once effects.
+
+## Schedule management
+
+Only native worker `start()` persists the declarations from registered implementations.
+Startup awaits it, then enqueues refresh, expiry and sync independently with `{}` and
+no run options. Their UTC schedules are `*/5 * * * *`, `0 * * * *`, and `*/10 * * * *`,
+with IDs `when:refresh-calendars`, `when:expire-pending-appointments`, and
+`when:sync-calendars`. Upserts preserve pause; startup runs do not depend on cron's pause
+state. Cron is wall-clock aligned and catches up once per overdue schedule, not per
+missed tick. No application timers or coalescers supplement it.
+
+These are the installed `0.11.0` public APIs, not `when-cli` commands. Given an initialized
+client connected to the intended queue, list every page:
+
+```ts
+const client = getOpenWorkflow();
+let after: string | undefined;
+do {
+	const page = await client.schedules.list({ limit: 100, after });
+	console.table(
+		page.data.map(({ id, cron, timezone, nextRunAt }) => ({
+			id,
+			cron,
+			timezone,
+			nextRunAt
+		}))
+	);
+	after = page.pagination.next ?? undefined;
+} while (after);
+```
+
+Manage one schedule explicitly (the following are separate operator actions):
+
+```ts
+await client.schedules.pause('when:sync-calendars');
+await client.schedules.resume('when:sync-calendars');
+await client.schedules.delete('when:sync-calendars');
+```
+
+`nextRunAt === null` means paused; resume computes the next future occurrence. Pause
+and delete do not cancel already queued runs or suppress independent startup runs.
+Removing a declaration alone leaves its persisted schedule active. To retire a target,
+stop the worker, delete its schedule through the public API, and remove the declaration
+before restarting; otherwise native startup recreates it. Schedule-management connections
+also migrate queues, so use the matching deployed version, not a new client for old-queue
+read-only inspection.
 
 ## Scripts
 

packages/jobs/src/index.ts

@@ -1,6 +1,8 @@
 export {
 	reconcileAppointment,
 	syncCalendars,
+	refreshCalendars,
+	expirePendingAppointments,
 	purgeAppointment,
 	testEmail,
 	testProvider,
@@ -14,6 +16,10 @@ export type {
 	ReconcileAppointmentResult,
 	SyncCalendarsInput,
 	SyncCalendarsResult,
+	RefreshCalendarsInput,
+	RefreshCalendarsResult,
+	ExpirePendingAppointmentsInput,
+	ExpirePendingAppointmentsResult,
 	PurgeAppointmentRow,
 	PurgeAppointmentInput,
 	PurgeAppointmentResult,

packages/jobs/src/specs.test.ts

@@ -1,7 +1,13 @@
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
 import { expect, test } from 'vitest';
+import { initOpenWorkflow, closeOpenWorkflow } from './openworkflow.js';
 import {
 	reconcileAppointment,
 	syncCalendars,
+	refreshCalendars,
+	expirePendingAppointments,
 	testEmail,
 	testProvider,
 	listProviderCalendars,
@@ -15,8 +21,51 @@ test('reconcileAppointment carries the shared workflow name', () => {
 	expect(reconcileAppointment.name).toBe('reconcile-appointment');
 });
 
-test('syncCalendars carries the shared workflow name', () => {
-	expect(syncCalendars.name).toBe('sync-calendars');
+test.each([
+	[refreshCalendars, 'refresh-calendars', 'when:refresh-calendars', '*/5 * * * *'],
+	[
+		expirePendingAppointments,
+		'expire-pending-appointments',
+		'when:expire-pending-appointments',
+		'0 * * * *'
+	],
+	[syncCalendars, 'sync-calendars', 'when:sync-calendars', '*/10 * * * *']
+])('%s declares its stable payload-free UTC schedule', (spec, name, id, cron) => {
+	expect(spec.name).toBe(name);
+	expect(spec.version).toBeUndefined();
+	expect(spec.schedules).toEqual([{ id, cron, timezone: 'UTC', input: {} }]);
+});
+
+test('producer initialization and maintenance enqueues do not persist schedules', async () => {
+	const directory = await mkdtemp(join(tmpdir(), 'when-producer-schedules-'));
+	try {
+		const client = initOpenWorkflow({ dbPath: join(directory, 'queue.sqlite') });
+		const schedulesBeforeEnqueue = await client.schedules.list();
+		expect(schedulesBeforeEnqueue.data).toEqual([]);
+		for (const spec of [refreshCalendars, expirePendingAppointments, syncCalendars]) {
+			const handle = await client.runWorkflow(spec, {});
+			expect(handle.workflowRun).toMatchObject({
+				workflowName: spec.name,
+				status: 'pending',
+				input: {},
+				workflowScheduleId: null
+			});
+		}
+		const schedulesAfterEnqueue = await client.schedules.list();
+		expect(schedulesAfterEnqueue.data).toEqual([]);
+	} finally {
+		await closeOpenWorkflow();
+		await rm(directory, { recursive: true, force: true });
+	}
+});
+
+test('sync retains its workflow-level backstop', () => {
+	expect(syncCalendars.retryPolicy).toEqual({
+		maximumAttempts: 1,
+		initialInterval: '1s',
+		backoffCoefficient: 2,
+		maximumInterval: '1s'
+	});
 });
 
 test('testEmail carries the shared workflow name', () => {

packages/jobs/src/specs.ts

@@ -152,13 +152,36 @@ export const testCalendar: WorkflowSpec<TestCalendarInput, TestCalendarResult> =
 	}
 });
 
+export type RefreshCalendarsInput = Record<string, never>;
+export type RefreshCalendarsResult = 'refreshed';
+
+export const refreshCalendars: WorkflowSpec<RefreshCalendarsInput, RefreshCalendarsResult> =
+	defineWorkflowSpec<RefreshCalendarsInput, RefreshCalendarsResult>({
+		name: 'refresh-calendars',
+		schedules: [{ id: 'when:refresh-calendars', cron: '*/5 * * * *', timezone: 'UTC', input: {} }]
+	});
+
+export type ExpirePendingAppointmentsInput = Record<string, never>;
+export type ExpirePendingAppointmentsResult = 'expired';
+
+export const expirePendingAppointments: WorkflowSpec<
+	ExpirePendingAppointmentsInput,
+	ExpirePendingAppointmentsResult
+> = defineWorkflowSpec<ExpirePendingAppointmentsInput, ExpirePendingAppointmentsResult>({
+	name: 'expire-pending-appointments',
+	schedules: [
+		{ id: 'when:expire-pending-appointments', cron: '0 * * * *', timezone: 'UTC', input: {} }
+	]
+});
+
 export type SyncCalendarsInput = Record<string, never>;
 export type SyncCalendarsResult = 'requested';
 
-// Payload-free wake-up the worker handles by running its calendar sync.
+// A requested result means the pass finished, not that every provider succeeded.
 export const syncCalendars: WorkflowSpec<SyncCalendarsInput, SyncCalendarsResult> =
 	defineWorkflowSpec<SyncCalendarsInput, SyncCalendarsResult>({
 		name: 'sync-calendars',
+		schedules: [{ id: 'when:sync-calendars', cron: '*/10 * * * *', timezone: 'UTC', input: {} }],
 		retryPolicy: {
 			maximumAttempts: 1,
 			initialInterval: '1s',