From 4053fc64e2bc616711626b6d60b0eb5aa30de910 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 30 Oct 2026 03:01:46 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=A4=96=20refactor:=20split=20maybeStartQu?= =?UTF-8?q?euedTasksFromReservations,=20part=203:=20drain=20preamble?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part 3 of 3. It moves the head under the mutex into steps: the reads, the inactive-workflow prepass, the stale-turn sweep bookkeeping and the capacity gates. maybeStartQueuedTasksFromReservations ends at CC 8 (from 76 before the stack). It keeps the lease disposer, the mutex block, the per-task loop and the launch of the plans after the mutex. - openQueuedDrain reads the config, the task settings, the task index and the queued tasks, runs the prepass, and re-reads all three when the prepass interrupted a task. Then it calls countQueuedDrainSlots right after the prepass's last await. - countQueuedDrainSlots reads staleTurnSweepGeneration before the turn count, completes the sweep only when the scan saw every owner, and returns "idle" (nothing queued, or no free slot) or "open" with the drain record the loop reads. The turn count still runs after the prepass. - listQueuedAgentTasks is the former listQueuedTasks closure, now a method because both reads in openQueuedDrain call it. No behaviour change: same awaits on collaborators, same calls in the same order (compared with a script), same log messages and fields. --- _Generated with `xum` • Model: `anthropic:claude-opus-5-5` • Thinking: `xhigh`_ --- src/node/services/taskService.ts | 172 ++++++++++++++++++------------- 1 file changed, 103 insertions(+), 69 deletions(-) diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 232c88af361..2d5ae85fab8 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -1030,6 +1030,17 @@ interface QueuedDrainSnapshot { readonly taskIndex: AgentTaskIndex; } +/** A drain that has queued tasks and a free slot: what its per-task loop reads. */ +interface QueuedDrain extends QueuedDrainSnapshot { + readonly queuedTasks: readonly AgentTaskWorkspaceEntry[]; + readonly availableSlots: number; +} + +/** A drain's reads under the mutex: idle (nothing queued, or no free slot), or open. */ +type QueuedDrainOpening = + | { readonly kind: "idle" } + | { readonly kind: "open"; readonly drain: QueuedDrain }; + /** A queued task's parent, as the drain's config names it, and the runtimes its launch uses. */ interface QueuedTaskParent { readonly parentWorkspaceId: string; @@ -16799,76 +16810,12 @@ export class TaskService implements AgentTaskIntegration { { await using _lock = await this.mutex.acquire(); - let config = this.config.loadConfigOrDefault(); - const taskSettings: TaskSettings = config.taskSettings ?? DEFAULT_TASK_SETTINGS; - const listQueuedTasks = (sourceConfig: ProjectsConfig): AgentTaskWorkspaceEntry[] => - this.listAgentTaskWorkspaces(sourceConfig) - .filter((task) => task.taskStatus === "queued" && typeof task.id === "string") - .sort((a, b) => { - const aTime = a.createdAt ? Date.parse(a.createdAt) : 0; - const bTime = b.createdAt ? Date.parse(b.createdAt) : 0; - return aTime - bTime; - }); - let taskIndex = this.buildAgentTaskIndex(config); - let queuedTasks = listQueuedTasks(config); - - let interruptedInactiveWorkflowQueuedTask = false; - for (const task of queuedTasks) { - const taskId = task.id; - assert(taskId != null && taskId.length > 0, "queued task id is required"); - if ( - await this.interruptTaskRecoveryForInactiveWorkflowOwner( - taskId, - config, - "queued-inactive-workflow-owner-prepass", - taskIndex, - { scheduleQueueDrain: false } - ) - ) { - interruptedInactiveWorkflowQueuedTask = true; - } - } - if (interruptedInactiveWorkflowQueuedTask) { - config = this.config.loadConfigOrDefault(); - taskIndex = this.buildAgentTaskIndex(config); - queuedTasks = listQueuedTasks(config); - } - - // Captured before the count reads any turn file: only requests armed before then are swept. - const staleTurnSweepGeneration = this.staleTurnSweepRequestedGeneration; - if ( - queuedTasks.length === 0 && - staleTurnSweepGeneration === this.staleTurnSweepDoneGeneration - ) { - taskQueueDebug("TaskService.maybeStartQueuedTasks no queued tasks", {}); - return; - } - const activeAgentTasks = this.countActiveAgentTasks(config); - const { count: activeWorkspaceTurns, complete: sweptEveryOwner } = - await this.getWorkspaceTurnManager().countActiveWorkspaceTurnsInGlobalScan(); - // A count that throws, or a scan that skipped an unreadable owner directory, leaves the - // request pending so a later drain retries the sweep (empty-queue drains included). - if (sweptEveryOwner) { - this.staleTurnSweepDoneGeneration = Math.max( - this.staleTurnSweepDoneGeneration, - staleTurnSweepGeneration - ); - } - if (queuedTasks.length === 0) return; - const availableSlots = Math.max( - 0, - taskSettings.maxParallelAgentTasks - (activeAgentTasks + activeWorkspaceTurns) - ); - taskQueueDebug("TaskService.maybeStartQueuedTasks reservation summary", { - maxParallelAgentTasks: taskSettings.maxParallelAgentTasks, - availableSlots, - }); - if (availableSlots === 0) return; - - const drain: QueuedDrainSnapshot = { config, taskIndex }; + const opening = await this.openQueuedDrain(); + if (opening.kind === "idle") return; + const { drain } = opening; let reservedSlots = 0; - for (const task of queuedTasks) { - if (reservedSlots >= availableSlots) { + for (const task of drain.queuedTasks) { + if (reservedSlots >= drain.availableSlots) { break; } const reservation = await this.reserveQueuedTaskSlot(task, drain, plans); @@ -16880,6 +16827,93 @@ export class TaskService implements AgentTaskIntegration { await this.launchReservedPlans(plans); } + /** A config's queued tasks, oldest first. */ + private listQueuedAgentTasks(sourceConfig: ProjectsConfig): AgentTaskWorkspaceEntry[] { + return this.listAgentTaskWorkspaces(sourceConfig) + .filter((task) => task.taskStatus === "queued" && typeof task.id === "string") + .sort((a, b) => { + const aTime = a.createdAt ? Date.parse(a.createdAt) : 0; + const bTime = b.createdAt ? Date.parse(b.createdAt) : 0; + return aTime - bTime; + }); + } + + /** + * The drain's first reads under the mutex: the queued tasks, after a prepass that interrupts + * the ones whose workflow owner is no longer active (and re-reads the config if it did). + */ + private async openQueuedDrain(): Promise { + let config = this.config.loadConfigOrDefault(); + const taskSettings: TaskSettings = config.taskSettings ?? DEFAULT_TASK_SETTINGS; + let taskIndex = this.buildAgentTaskIndex(config); + let queuedTasks = this.listQueuedAgentTasks(config); + + let interruptedInactiveWorkflowQueuedTask = false; + for (const task of queuedTasks) { + const taskId = task.id; + assert(taskId != null && taskId.length > 0, "queued task id is required"); + if ( + await this.interruptTaskRecoveryForInactiveWorkflowOwner( + taskId, + config, + "queued-inactive-workflow-owner-prepass", + taskIndex, + { scheduleQueueDrain: false } + ) + ) { + interruptedInactiveWorkflowQueuedTask = true; + } + } + if (interruptedInactiveWorkflowQueuedTask) { + config = this.config.loadConfigOrDefault(); + taskIndex = this.buildAgentTaskIndex(config); + queuedTasks = this.listQueuedAgentTasks(config); + } + return await this.countQueuedDrainSlots(taskSettings, { config, taskIndex }, queuedTasks); + } + + /** + * The drain's free slots, from the active tasks and the turn count (which also completes a + * pending stale-turn sweep). Idle when no task is queued or no slot is free. + */ + private async countQueuedDrainSlots( + taskSettings: TaskSettings, + snapshot: QueuedDrainSnapshot, + queuedTasks: readonly AgentTaskWorkspaceEntry[] + ): Promise { + // Captured before the count reads any turn file: only requests armed before then are swept. + const staleTurnSweepGeneration = this.staleTurnSweepRequestedGeneration; + if ( + queuedTasks.length === 0 && + staleTurnSweepGeneration === this.staleTurnSweepDoneGeneration + ) { + taskQueueDebug("TaskService.maybeStartQueuedTasks no queued tasks", {}); + return { kind: "idle" }; + } + const activeAgentTasks = this.countActiveAgentTasks(snapshot.config); + const { count: activeWorkspaceTurns, complete: sweptEveryOwner } = + await this.getWorkspaceTurnManager().countActiveWorkspaceTurnsInGlobalScan(); + // A count that throws, or a scan that skipped an unreadable owner directory, leaves the + // request pending so a later drain retries the sweep (empty-queue drains included). + if (sweptEveryOwner) { + this.staleTurnSweepDoneGeneration = Math.max( + this.staleTurnSweepDoneGeneration, + staleTurnSweepGeneration + ); + } + if (queuedTasks.length === 0) return { kind: "idle" }; + const availableSlots = Math.max( + 0, + taskSettings.maxParallelAgentTasks - (activeAgentTasks + activeWorkspaceTurns) + ); + taskQueueDebug("TaskService.maybeStartQueuedTasks reservation summary", { + maxParallelAgentTasks: taskSettings.maxParallelAgentTasks, + availableSlots, + }); + if (availableSlots === 0) return { kind: "idle" }; + return { kind: "open", drain: { ...snapshot, queuedTasks, availableSlots } }; + } + /** * One queued task of a drain, in queue order: the workflow-owner and stop checks, a task that * is already streaming and the launch refusals, then its launch from the parent metadata read.