refactor(planning): give planning workers one lifetime owner

Planning, counterfactual analysis and checkpoint explanation now share one internal worker-job owner for creation, handler cleanup, termination and supersession, and one request/response message contract between the coordinator and the worker. Job-specific progress, checkpoint recovery and completion rules stay explicit. Refs #277 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-17 11:11:37 +02:00
1 parent 7fc2b33587
commit b69b3c3690
@@ -17,6 +17,7 @@ import type {
 	StaleCounterfactualAnalysis,
 } from './leave-planner.core';
 import { planningDateInParis } from './planning-date';
+import type { WorkerRequest, WorkerResponse } from './leave-planner.worker-messages';

 export type {
 	IfThenPlanningRule,
@@ -278,20 +279,6 @@ function invalidDeadline(request: TransportSnapshot): InvalidInputOutcome | unde
 	return undefined;
 }

-type PlannerWorkerResponse =
-	| { type: 'heartbeat' }
-	| { type: 'progress'; stage: string }
-	| { type: 'checkpoint'; plan: PlanOutcome; state?: PlanningStateSnapshot }
-	| { type: 'result'; outcome: PlanningOutcome }
-	| { type: 'artifacts'; artifacts: PlanningArtifacts }
-	| {
-			type: 'explanation';
-			explanation?: PlanningExplanation;
-			explanationError?: PlanningReason;
-			finalState?: PlanningStateSnapshot;
-	  }
-	| { type: 'error'; code?: string; message: string };
-
 interface PlanningRun {
 	outcome: Promise<PlanningOutcome>;
 	explanation: Promise<PlanningExplanation | undefined>;
@@ -320,6 +307,47 @@ function interruptedOutcome(
 const latestWorkerCancels = new Set<() => void>();
 let latestCounterfactualCancel: (() => void) | undefined;

+interface WorkerJob {
+	post(message: WorkerRequest, onFailure: (error: unknown) => void): void;
+	release(): void;
+}
+
+// Single owner of a worker's lifetime: creation, handlers, cancellation registration and an idempotent release.
+function startWorkerJob(
+	supersede: () => void,
+	onMessage: (message: WorkerResponse) => void,
+	onError: () => void,
+): WorkerJob {
+	const worker = new Worker(new URL('./leave-planner.worker', import.meta.url), {
+		type: 'module',
+	});
+	let released = false;
+	worker.onmessage = (event: MessageEvent<WorkerResponse>) => {
+		if (!released && event.data) onMessage(event.data);
+	};
+	worker.onerror = () => {
+		if (!released) onError();
+	};
+	latestWorkerCancels.add(supersede);
+	return {
+		post(message, onFailure) {
+			try {
+				worker.postMessage(message);
+			} catch (error) {
+				onFailure(error);
+			}
+		},
+		release() {
+			if (released) return;
+			released = true;
+			worker.onmessage = null;
+			worker.onerror = null;
+			worker.terminate();
+			latestWorkerCancels.delete(supersede);
+		},
+	};
+}
+
 function cancelLatestWorkers(): void {
 	for (const cancel of [...latestWorkerCancels]) cancel();
 	latestWorkerCancels.clear();
@@ -349,9 +377,11 @@ function planInWorker(snapshot: TransportSnapshot): PlanningRun {
 	});
 	let cancel: (() => void) | undefined;
 	try {
-		const worker = new Worker(new URL('./leave-planner.worker', import.meta.url), {
-			type: 'module',
-		});
+		const job = startWorkerJob(
+			() => supersede(),
+			(message) => handleMessage(message),
+			() => fail('worker-error', 'Planning worker failed before returning explanation data.'),
+		);
 		let outcomeSettled = false;
 		let cancellationRequested = false;
 		let cancelledRace: PlanningIncompleteOutcome | undefined;
@@ -364,9 +394,7 @@ function planInWorker(snapshot: TransportSnapshot): PlanningRun {
 		let receivedFinalState: PlanningStateSnapshot | undefined;
 		let livenessWatchdog: ReturnType<typeof setTimeout>;
 		const terminate = () => {
-			worker.onmessage = null;
-			worker.onerror = null;
-			worker.terminate();
+			job.release();
 			clearTimeout(livenessWatchdog);
 		};
 		const finishOutcome = (result: PlanningOutcome) => {
@@ -434,7 +462,6 @@ function planInWorker(snapshot: TransportSnapshot): PlanningRun {
 			resolveArtifacts(complete);
 			resolveTimings(complete.timings ?? {});
 			terminate();
-			latestWorkerCancels.delete(supersede);
 		};
 		const supersede = () => {
 			if (artifactsSettled) return;
@@ -456,7 +483,6 @@ function planInWorker(snapshot: TransportSnapshot): PlanningRun {
 				finalState: receivedFinalState ?? checkpointState,
 			});
 		};
-		latestWorkerCancels.add(supersede);
 		const fail = (code: string, message: string) => {
 			const reason = { code, message };
 			if (!outcomeSettled) finishOutcome(interruptedOutcome(checkpoint, code, message));
@@ -508,15 +534,11 @@ function planInWorker(snapshot: TransportSnapshot): PlanningRun {
 		cancel = () => {
 			if (outcomeSettled || cancellationRequested) return;
 			cancellationRequested = true;
-			try {
-				worker.postMessage({ type: 'cancel' });
-			} catch {
-				fail('worker-error', 'The cancellation request could not reach the worker.');
-			}
+			job.post({ type: 'cancel' }, () =>
+				fail('worker-error', 'The cancellation request could not reach the worker.'),
+			);
 		};
-		worker.onmessage = (event: MessageEvent<PlannerWorkerResponse>) => {
-			const message = event.data;
-			if (!message) return;
+		const handleMessage = (message: WorkerResponse) => {
 			if (message.type === 'heartbeat') {
 				scheduleLivenessWatchdog();
 				return;
@@ -552,13 +574,9 @@ function planInWorker(snapshot: TransportSnapshot): PlanningRun {
 			if (message.type === 'error') fail(message.code ?? 'worker-error', message.message);
 			else fail('worker-error', 'Planning worker returned an unexpected response.');
 		};
-		worker.onerror = () =>
-			fail('worker-error', 'Planning worker failed before returning explanation data.');
-		try {
-			worker.postMessage({ type: 'plan', request: { ...snapshot, progress: undefined } });
-		} catch {
-			fail('worker-error', 'Planning input could not be sent to the isolated worker.');
-		}
+		job.post({ type: 'plan', request: { ...snapshot, progress: undefined } }, () =>
+			fail('worker-error', 'Planning input could not be sent to the isolated worker.'),
+		);
 	} catch {
 		const result = interruptedOutcome(
 			undefined,
@@ -677,62 +695,43 @@ function counterfactualInWorker(
 ): Promise<CounterfactualAnalysis> {
 	return new Promise((resolve, reject) => {
 		latestCounterfactualCancel?.();
-		let worker: Worker;
+		let job: WorkerJob;
+		const finish = (callback: () => void) => {
+			if (settled) return;
+			settled = true;
+			job.release();
+			if (latestCounterfactualCancel === cancel) latestCounterfactualCancel = undefined;
+			callback();
+		};
+		let settled = false;
+		const cancel = () =>
+			finish(() => reject(new Error('Counterfactual analysis was superseded.')));
 		try {
-			worker = new Worker(new URL('./leave-planner.worker', import.meta.url), {
-				type: 'module',
-			});
+			job = startWorkerJob(
+				cancel,
+				(message) => {
+					if (message.type === 'counterfactual-result' && message.analysis)
+						finish(() => resolve(message.analysis));
+					else if (message.type === 'error')
+						finish(() =>
+							reject(
+								new Error(
+									message.message ??
+										`Counterfactual analysis failed (${message.code ?? 'worker-error'}).`,
+								),
+							),
+						);
+				},
+				() => finish(() => reject(new Error('Counterfactual worker failed.'))),
+			);
 		} catch (error) {
 			reject(error);
 			return;
 		}
-		let settled = false;
-		const finish = (callback: () => void) => {
-			if (settled) return;
-			settled = true;
-			worker.onmessage = null;
-			worker.onerror = null;
-			worker.terminate();
-			if (cancel) latestWorkerCancels.delete(cancel);
-			if (latestCounterfactualCancel === cancel) latestCounterfactualCancel = undefined;
-			callback();
-		};
-		const cancel = () =>
-			finish(() => reject(new Error('Counterfactual analysis was superseded.')));
 		latestCounterfactualCancel = cancel;
-		latestWorkerCancels.add(cancel);
-		worker.onmessage = (
-			event: MessageEvent<{
-				type: string;
-				analysis?: CounterfactualAnalysis;
-				code?: string;
-				message?: string;
-			}>,
-		) => {
-			if (event.data?.type === 'counterfactual-result' && event.data.analysis)
-				finish(() => resolve(event.data.analysis!));
-			else if (event.data?.type === 'error')
-				finish(() =>
-					reject(
-						new Error(
-							event.data.message ??
-								`Counterfactual analysis failed (${event.data.code ?? 'worker-error'}).`,
-						),
-					),
-				);
-		};
-		worker.onerror = () => finish(() => reject(new Error('Counterfactual worker failed.')));
-		try {
-			worker.postMessage({
-				type: 'counterfactual',
-				request: snapshot,
-				plan,
-				placement,
-				state,
-			});
-		} catch (error) {
-			finish(() => reject(error));
-		}
+		job.post({ type: 'counterfactual', request: snapshot, plan, placement, state }, (error) =>
+			finish(() => reject(error)),
+		);
 	});
 }

@@ -743,53 +742,39 @@ function explainCheckpointInWorker(
 	interruption: PlanningReason,
 ): Promise<Pick<PlanningArtifacts, 'explanation' | 'explanationError' | 'finalState'>> {
 	return new Promise((resolve, reject) => {
-		let worker: Worker;
-		try {
-			worker = new Worker(new URL('./leave-planner.worker', import.meta.url), {
-				type: 'module',
-			});
-		} catch (error) {
-			reject(error);
-			return;
-		}
+		let job: WorkerJob;
 		let settled = false;
 		const finish = (callback: () => void) => {
 			if (settled) return;
 			settled = true;
-			worker.onmessage = null;
-			worker.onerror = null;
-			worker.terminate();
-			if (cancel) latestWorkerCancels.delete(cancel);
+			job.release();
 			callback();
 		};
-		const cancel = () =>
-			finish(() => reject(new Error('Checkpoint explanation was superseded.')));
-		latestWorkerCancels.add(cancel);
-		worker.onmessage = (event: MessageEvent<PlannerWorkerResponse>) => {
-			const response = event.data;
-			if (response?.type === 'explanation')
-				finish(() =>
-					resolve({
-						explanation: response.explanation,
-						explanationError: response.explanationError,
-						finalState: response.finalState,
-					}),
-				);
-			else if (response?.type === 'error') finish(() => reject(new Error(response.message)));
-		};
-		worker.onerror = () =>
-			finish(() => reject(new Error('Checkpoint explanation worker failed.')));
 		try {
-			worker.postMessage({
-				type: 'explain-checkpoint',
-				request: snapshot,
-				plan,
-				state,
-				interruption,
-			});
+			job = startWorkerJob(
+				() => finish(() => reject(new Error('Checkpoint explanation was superseded.'))),
+				(response) => {
+					if (response.type === 'explanation')
+						finish(() =>
+							resolve({
+								explanation: response.explanation,
+								explanationError: response.explanationError,
+								finalState: response.finalState,
+							}),
+						);
+					else if (response.type === 'error')
+						finish(() => reject(new Error(response.message)));
+				},
+				() => finish(() => reject(new Error('Checkpoint explanation worker failed.'))),
+			);
 		} catch (error) {
-			finish(() => reject(error));
+			reject(error);
+			return;
 		}
+		job.post(
+			{ type: 'explain-checkpoint', request: snapshot, plan, state, interruption },
+			(error) => finish(() => reject(error)),
+		);
 	});
 }