Skip to content

🤖 refactor: split maybeStartQueuedTasksFromReservations, part 3: drain preamble #7747

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ThomasK33 merged 3 commits into from
Oct 30, 2026
Merged

🤖 refactor: split maybeStartQueuedTasksFromReservations, part 3: drain preamble #7747

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4053fc6
🤖 refactor: split maybeStartQueuedTasksFromReservations, part 3: drai…
ThomasK33 Oct 30, 2026
d74050c
Merge branch 'refactor/task-queue-start-split-2' into refactor/task-q…
ThomasK33 Oct 30, 2026
b8608c8
Merge branch 'refactor/task-queue-start-split-2' into refactor/task-q…
ThomasK33 Oct 30, 2026
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
172 changes: 103 additions & 69 deletions src/node/services/taskService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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) {

Check warning on line 16818 in src/node/services/taskService.ts

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: >= -> > 

No unit test failed when maybeStartQueuedTasksFromReservations had >= -> >. Add a test that tells them apart, or say why they are equivalent.
break;
}
const reservation = await this.reserveQueuedTaskSlot(task, drain, plans);
Expand All @@ -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;

Check warning on line 16835 in src/node/services/taskService.ts

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: 0 -> 1 

No unit test failed when listQueuedAgentTasks had 0 -> 1. Add a test that tells them apart, or say why they are equivalent.
const bTime = b.createdAt ? Date.parse(b.createdAt) : 0;

Check warning on line 16836 in src/node/services/taskService.ts

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: 0 -> 1 

No unit test failed when listQueuedAgentTasks had 0 -> 1. Add a test that tells them apart, or say why they are equivalent.
return aTime - bTime;

Check warning on line 16837 in src/node/services/taskService.ts

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: - -> + 

No unit test failed when listQueuedAgentTasks had - -> +. Add a test that tells them apart, or say why they are equivalent.
});
}

/**
* 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<QueuedDrainOpening> {
let config = this.config.loadConfigOrDefault();
const taskSettings: TaskSettings = config.taskSettings ?? DEFAULT_TASK_SETTINGS;

Check warning on line 16847 in src/node/services/taskService.ts

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: ?? -> || 

No unit test failed when openQueuedDrain had ?? -> ||. Add a test that tells them apart, or say why they are equivalent.
let taskIndex = this.buildAgentTaskIndex(config);
let queuedTasks = this.listQueuedAgentTasks(config);

let interruptedInactiveWorkflowQueuedTask = false;

Check warning on line 16851 in src/node/services/taskService.ts

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: false -> true 

No unit test failed when openQueuedDrain had false -> true. Add a test that tells them apart, or say why they are equivalent.
for (const task of queuedTasks) {
const taskId = task.id;
assert(taskId != null && taskId.length > 0, "queued task id is required");

Check warning on line 16854 in src/node/services/taskService.ts

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: && -> || 

No unit test failed when openQueuedDrain had && -> ||. Add a test that tells them apart, or say why they are equivalent.
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<QueuedDrainOpening> {
// 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.
Expand Down
Loading