kick!

src/executor.ts

@@ -559,13 +559,20 @@ async function executeBatchParallel(
 		requestBatchRender();
 	}, 100);
 
-	const results: Array<{ task: Task; result: Promise<any> }> = [];
+	// Semaphore-based concurrency control:
+	// Start up to maxParallel tasks immediately. When ANY task completes,
+	// start the next pending task. This ensures slots fill as soon as they
+	// open, instead of blocking on the oldest task (FIFO pattern).
+	const pending = [...tasks];
+	const running = new Set<Promise<void>>();
 
-	for (const task of tasks) {
+	/** Start the next pending task if a slot is available. */
-		const assignedModel = roundRobin?.assign(task.id);
+	const kick = (): void => {
-		results.push({
+		while (running.size < maxParallel && pending.length > 0) {
-			task,
+			const task = pending.shift()!;
-			result: executeTask(
+			const assignedModel = roundRobin?.assign(task.id);
+
+			const p = executeTask(
 				task,
 				project,
 				config,
@@ -577,35 +584,43 @@ async function executeBatchParallel(
 				assignedModel,
 				roundRobin,
 				requestBatchRender,
-			).catch((error) => {
+			)
-				// Safety net: one task failure should never crash the batch.
+				.catch((error) => {
-				// executeTask already marks failed and notifies, but catch as
+					// Safety net: one task failure should never crash the batch.
-				// a last resort so the error doesn't propagate and crash pi.
+					// executeTask already marks failed and notifies, but catch as
-				roundRobin?.release(task.id);
+					// a last resort so the error doesn't propagate and crash pi.
-				requestBatchRender();
+					roundRobin?.release(task.id);
-				const errorMsg = error instanceof Error ? error.message : String(error);
+					requestBatchRender();
-				progress.markFailed(task.id, errorMsg);
+					const errorMsg =
-				// Auto-update the PRD source file checkbox
+						error instanceof Error ? error.message : String(error);
-				try {
+					progress.markFailed(task.id, errorMsg);
-					updateTaskInFile(project.sourcePath, task.id, "failed");
+					// Auto-update the PRD source file checkbox
-				} catch {
+					try {
-					// Best-effort
+						updateTaskInFile(project.sourcePath, task.id, "failed");
-				}
+					} catch {
-				sendChatMessage?.(`✗ ${task.id} · ${task.title} — ${errorMsg}`);
+						// Best-effort
-				ctx.ui.notify(`Task ${task.id} failed: ${errorMsg}`, "error");
+					}
-			}),
+					sendChatMessage?.(`✗ ${task.id} · ${task.title} — ${errorMsg}`);
-		});
+					ctx.ui.notify(`Task ${task.id} failed: ${errorMsg}`, "error");
+				})
+				.finally(() => {
+					// Remove from running set and start next pending task
+					running.delete(p);
+					requestBatchRender();
+					kick();
+				});
 
-		// Limit concurrency
+			running.add(p);
-		if (results.length >= maxParallel) {
-			const first = results.shift();
-			if (first) await first.result;
 		}
-	}
+	};
 
-	// Wait for remaining tasks
+	// Kick off initial batch of tasks (up to maxParallel)
-	for (const { result } of results) {
+	kick();
-		await result;
+
+	// Wait for all tasks to complete (kick() adds new promises to `running`
+	// when completed tasks free up slots, so we iterate until the set is empty).
+	while (running.size > 0) {
+		await Promise.race(running);
 	}
 
 	clearInterval(spinnerTimer);