fix(worker): stop() waits for signal acks and concurrent steps

AGENTS.md

@@ -147,6 +147,7 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe
 - `invoke()` inserts an independent execution and memoizes its id as a step in one statement. Only with `cancelWithParent` does it set `parent_execution_id`, together with `cancel_with_parent`; settlement fails a completed child that has a parent but neither a waiter nor that flag
 - Memoized by step key, so a replay returns the stored result: `step()`, `sleep()`, `invoke()`, `invokeAndWait()`, `waitForResult()`, `waitForEvent()`, `subscribe()`, `waitForAny()` and `signalWithStart()`. `emit()` (deduplicated only with a `dedupe_key`), `sendSignal()` (deduplicated by signal id), `schedule()` (replaces the pending occurrence), `unschedule()` and `cancel()` are not memoized and run again on every replay
 - `checkpoint()` hangs up when the abort signal fired or the time window closed; it persists nothing
+- Each attempt tracks its step callbacks and their saves (`TaskContextOptions.steps`). Before it reports a result, the worker waits for those still running, bounded by `timeoutMs`, so a sibling's hangup or error (or `stop()`) does not end an attempt while a concurrent step runs. A timeout, cancellation or signal failure ends it at once
 - `step()` checkpoints after reading the step, so `fn` never starts once the attempt aborted, and hangs up as suspended when the claim fence rejects its save
 - Hangup pattern: abort execution and return never-resolving promise
 - Resume happens automatically when database wakes execution
@@ -271,6 +272,7 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe
 - `_private_execution_signals` stores one row per `(execution_id, id)`; `seq` orders delivery and `delivered_at` is set when a `ctx.onSignal` handler returned (at least once per id)
 - Senders lock the execution row and increment `_private_executions.signal_count`. Settlement reads `signal_count` from its locked row, which is newer than its snapshot: a completion whose count exceeds the signals it can see, or that has undelivered signals for names the attempt listened for (`signal_names` in the result), is re-queued instead of completed
 - Claiming an execution reads its undelivered signals under the row lock senders take, so signals stored later have a higher `seq` and later claims read them with `seq > after_seq`. Signal reads and acknowledgements are fenced by the claim token. `SignalInbox` delivers in `seq` order and skips a signal whose name has no handler yet; it stays undelivered, so settlement re-queues if a handler for it registers later. On completion the worker drains the inbox (`close({ drain: true })`): signals already read are delivered before settling, and only ones stored after the last read re-queue
+- A delivery failure (a throwing `ctx.onSignal` handler or a failed acknowledgement) aborts the attempt as `signal-failed`, except after the worker signal aborted: stopping ends acknowledgement retries, so the handler keeps its slot and the attempt is released once it returns
 - Re-queues are unbounded by design: each needs a signal stored after the previous attempt started, which the next attempt receives at claim time, so an execution re-runs only while signals keep arriving
 - `key` on `_private_executions` with a partial unique index makes at most one non-terminal, non-cancelled execution per `(task_key, key, queue)`; `signal_with_start()` takes a transaction advisory lock on the key, then signals the existing execution or inserts one
 

docs/content/crafting-tasks/retries-and-steps.md

@@ -85,6 +85,8 @@ Steps are therefore at-least-once around their callback. For side effects outsid
 
 The handler always runs from the start: on every retry and every resume after `ctx.sleep()` or a wait, code outside steps runs again.
 
+Steps can run concurrently, for example with `Promise.all()`. When one of them throws or the handler hangs up in another (`ctx.sleep()`, a wait, or `orchestrator.stop()`), the attempt ends only after the callbacks still running return and save their results, or once `timeoutMs` elapses. A timeout or cancellation ends the attempt at once.
+
 ## Working with Loops
 
 Loops without steps re-process all items on retry:

docs/content/task-execution/signals.md

@@ -95,4 +95,4 @@ A handler does not have to wait for its signals: one that registers listeners an
 
 ## Errors
 
-A signal handler that throws fails the attempt like an uncaught error in the handler, with the error `Signal handler for "<name>" failed: <message>`. The signal is not marked delivered, so the retry receives it again.
+A signal handler that throws fails the attempt like an uncaught error in the handler, with the error `Signal handler for "<name>" failed: <message>`. The signal is not marked delivered, so the retry receives it again. During `orchestrator.stop()`, a signal handler that throws, or a signal that cannot be marked delivered, releases the attempt once the handler returns, without using up a retry.

packages/pgconductor-js/src/lib/map-concurrent.ts

@@ -89,5 +89,7 @@ export async function* mapConcurrent<T, R>(
 		}
 	} finally {
 		itemWait?.cancel();
+		// Also on an error or an early exit, the consumer continues once every mapper settled
+		await Promise.allSettled(Array.from(active.values(), (task) => task.promise));
 	}
 }

packages/pgconductor-js/src/task-context.ts

@@ -133,6 +133,8 @@ export type TaskContextOptions = {
 	window?: [string, string];
 	telemetry: Telemetry;
 	signals: SignalInbox;
+	/** Step callbacks and their saves in flight */
+	steps: Set<Promise<unknown>>;
 };
 
 /** A signal sent to an execution with `conductor.signal()` or `signalWithStart()`. */
@@ -243,18 +245,23 @@ export class TaskContext<
 				"pgconductor.queue": this.opts.execution.queue,
 			},
 			run: async () => {
-				const result = await fn();
-				const saved = await this.opts.db.saveStep(
-					{
-						executionId: this.opts.execution.id,
-						queue: this.opts.execution.queue,
-						orchestratorId: this.opts.execution.locked_by,
-						claimToken: this.opts.execution.claim_token,
-						key: name,
-						result: { result: result as JsonValue },
-					},
-					{ signal: this.signal },
-				);
+				const running = (async () => {
+					const result = await fn();
+					const saved = await this.opts.db.saveStep(
+						{
+							executionId: this.opts.execution.id,
+							queue: this.opts.execution.queue,
+							orchestratorId: this.opts.execution.locked_by,
+							claimToken: this.opts.execution.claim_token,
+							key: name,
+							result: { result: result as JsonValue },
+						},
+						{ signal: this.signal },
+					);
+					return { result, saved };
+				})();
+				this.opts.steps.add(running);
+				const { result, saved } = await running.finally(() => this.opts.steps.delete(running));
 				return saved ? result : this.abortAndHangup({ reason: "suspended" });
 			},
 		});

packages/pgconductor-js/src/worker.ts

@@ -543,9 +543,14 @@ export class Worker<
 	): Promise<ExecutionResult | ExecutionResult[]> {
 		const taskAbortController = new TypedAbortController<TaskAbortReasons>();
 		const signal = AbortSignal.any([taskAbortController.signal, this.signal]);
+		const timedOut = new Deferred<void>();
 		const timeout = task.timeoutMs
-			? setTimeout(() => taskAbortController.abort({ reason: "timed-out" }), task.timeoutMs)
+			? setTimeout(() => {
+					timedOut.resolve();
+					taskAbortController.abort({ reason: "timed-out" });
+				}, task.timeoutMs)
 			: undefined;
+		const steps = new Set<Promise<unknown>>();
 
 		// Wrapped, so no handler output can be mistaken for an abort
 		const aborted = new Promise<{ abort: TaskAbortReasons }>((resolve) => {
@@ -573,6 +578,9 @@ export class Worker<
 					{ signal: this.signal },
 				),
 			onError: (error) => {
+				// Stopping ends the retries of an acknowledgement. The handler keeps its slot, and the
+				// attempt is released with the signal undelivered.
+				if (this.signal.aborted) return;
 				taskAbortController.abort({ reason: "signal-failed", error: error.message });
 			},
 		});
@@ -607,6 +615,7 @@ export class Worker<
 				window: task.window,
 				telemetry: this.telemetry,
 				signals,
+				steps,
 			},
 			extraContext,
 		);
@@ -705,6 +714,7 @@ export class Worker<
 			}
 
 			if (outcome.signalFailure) {
+				if (this.signal.aborted) return { ...identity, status: "released" } as const;
 				return {
 					...identity,
 					status: "failed",
@@ -735,6 +745,12 @@ export class Worker<
 				error: coerceError(err).message,
 			} as const;
 		} finally {
+			// Steps still running when a sibling hung up or threw end with the attempt, unless the
+			// worker aborted it. They save their results before the attempt settles.
+			const reason = taskAbortController.signal.reason?.reason;
+			if (reason !== "timed-out" && reason !== "cancelled" && reason !== "signal-failed") {
+				await Promise.race([Promise.allSettled(steps), timedOut.promise]);
+			}
 			clearTimeout(timeout);
 			this._running.delete(exec.id);
 			// A hung-up or failed attempt leaves undelivered signals for the next one