Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -263,4 +263,47 @@ describe('UnifiedSwarmCoordinator event-driven task wait (Dream Cycle 2026-09-20
expect(result).toBeDefined();
expect(result?.status).toBe('cancelled');
});

it('settles in-flight waits on shutdown() (tasks cleared, no per-task event) instead of hanging until taskTimeoutMs', async () => {
vi.useFakeTimers();

const submit = (name: string) => coordinator.submitTask({
type: 'coding',
name,
description: 'shutdown-while-waiting test',
priority: 'normal',
dependencies: [],
input: {},
timeoutMs: 10000,
retries: 0,
maxRetries: 3,
metadata: {},
});
const queuedId = await submit('queued wait');
const completionId = await submit('completion wait');

const internals = coordinator as unknown as {
waitForQueuedTask(taskId: string, domain: 'core', startTime: number): Promise<{ success: boolean; error?: Error }>;
waitForTaskCompletion(taskId: string, timeoutMs: number): Promise<unknown>;
};

let queuedResult: { success: boolean; error?: Error } | undefined;
void internals.waitForQueuedTask(queuedId, 'core', performance.now()).then((r) => { queuedResult = r; });

let completionError: Error | undefined;
void internals.waitForTaskCompletion(completionId, 10000).catch((e: Error) => { completionError = e; });

await coordinator.shutdown();
await Promise.resolve();
await Promise.resolve();

// No timer advanced: both must have settled off the 'swarm.stopped' event.
expect(queuedResult?.success).toBe(false);
expect(queuedResult?.error?.message).toBe(`Task ${queuedId} not found`);
expect(completionError?.message).toBe(`Task ${completionId} not found`);
// And no listener or pending timeout is left behind.
expect(coordinator.listenerCount('task.completed')).toBe(0);
expect(coordinator.listenerCount('swarm.stopped')).toBe(0);
expect(vi.getTimerCount()).toBe(0);
});
});
45 changes: 37 additions & 8 deletions v3/@claude-flow/swarm/src/unified-coordinator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1341,26 +1341,38 @@ export class UnifiedSwarmCoordinator extends EventEmitter implements IUnifiedSwa
}

let timeoutHandle: ReturnType<typeof setTimeout>;
const settle = (result: ParallelExecutionResult): void => {
clearTimeout(timeoutHandle);
const detach = (): void => {
this.off('task.completed', onTaskEvent);
this.off('task.failed', onTaskEvent);
this.off('swarm.stopped', onSwarmStopped);
};
const settle = (result: ParallelExecutionResult): void => {
clearTimeout(timeoutHandle);
detach();
resolve(result);
};
const onTaskEvent = (event: SwarmEvent): void => {
if (event.data.taskId !== taskId) return;
const result = resultForCurrentStatus();
if (result) settle(result);
};
// shutdown() clears state.tasks without a per-task terminal event; the
// old 100ms poll noticed that ("not found") almost immediately. Re-check
// on 'swarm.stopped' so an in-flight wait settles instead of hanging
// (and holding the event loop open) until taskTimeoutMs.
const onSwarmStopped = (): void => {
const result = resultForCurrentStatus();
if (result) settle(result);
};

this.on('task.completed', onTaskEvent);
this.on('task.failed', onTaskEvent);
this.on('swarm.stopped', onSwarmStopped);

// Timeout after configured duration — a client-side deadline, not a
// coordinator-observed state, so it stays a timer rather than an event.
timeoutHandle = setTimeout(() => {
this.off('task.completed', onTaskEvent);
this.off('task.failed', onTaskEvent);
detach();
const task = this.state.tasks.get(taskId);
if (task && task.status !== 'completed') {
task.status = 'timeout';
Expand Down Expand Up @@ -1392,10 +1404,14 @@ export class UnifiedSwarmCoordinator extends EventEmitter implements IUnifiedSwa
}

let timeoutHandle: ReturnType<typeof setTimeout>;
const cleanup = (): void => {
clearTimeout(timeoutHandle);
const detach = (): void => {
this.off('task.completed', onTaskEvent);
this.off('task.failed', onTaskEvent);
this.off('swarm.stopped', onSwarmStopped);
};
const cleanup = (): void => {
clearTimeout(timeoutHandle);
detach();
};
const onTaskEvent = (event: SwarmEvent): void => {
if (event.data.taskId !== taskId) return;
Expand All @@ -1405,13 +1421,26 @@ export class UnifiedSwarmCoordinator extends EventEmitter implements IUnifiedSwa
resolve(task);
}
};
// See waitForQueuedTask(): shutdown() clears state.tasks with no
// per-task event, so re-check on 'swarm.stopped' rather than hanging
// until timeoutMs (matches the old poll's "not found" rejection).
const onSwarmStopped = (): void => {
const task = this.state.tasks.get(taskId);
if (!task) {
cleanup();
reject(new Error(`Task ${taskId} not found`));
} else if (isTerminal(task)) {
cleanup();
resolve(task);
}
};

this.on('task.completed', onTaskEvent);
this.on('task.failed', onTaskEvent);
this.on('swarm.stopped', onSwarmStopped);

timeoutHandle = setTimeout(() => {
this.off('task.completed', onTaskEvent);
this.off('task.failed', onTaskEvent);
detach();
const task = this.state.tasks.get(taskId);
if (task) {
task.status = 'timeout';
Expand Down
Loading