Skip to content

🤖 refactor: split startReservedAgentTaskLeased, part 1: admission, send and status #7756

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 1 commit into from
Oct 30, 2026
Merged

🤖 refactor: split startReservedAgentTaskLeased, part 1: admission, send and status #7756

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
133 changes: 82 additions & 51 deletions src/node/services/taskService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1475,6 +1475,18 @@
reusedExistingCheckout: boolean;
}

/**
* A reserved launch's checkout once materialized: what the materialize step returned, with the
* launch's init log and its shared-checkout facts. Built once after the fork, never re-read.
*/
interface ReservedLaunchCheckout extends Readonly<MaterializedTaskLaunch> {
readonly initLogger: InitLogger;
/** The row was queued as isolation: "none" (read when the launch started). */
readonly taskWasShared: boolean;
/** The task reuses its parent's checkout: it was shared and that checkout still existed. */
readonly sharesParentCheckout: boolean;
}

export type TaskMessageQueueDispatchMode = "tool-end" | "turn-end";

export interface SendAgentTaskMessageResult {
Expand Down Expand Up @@ -7320,14 +7332,19 @@
}
}

/**
* The launch under its lease, in steps. Each step calls the next one right after its own last
* await instead of returning a value to await, so no write, admission or send waits an extra
* microtask. startReservedAgentTask awaits the whole chain before it releases the lease.
*/
private async startReservedAgentTaskLeased(plan: TaskLaunchPlan): Promise<void> {
assert(plan.parentWorkspaceId.length > 0, "startReservedAgentTask requires parentWorkspaceId");

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

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: > -> >= 

No unit test failed when startReservedAgentTaskLeased had > -> >=. Add a test that tells them apart, or say why they are equivalent.

Check warning on line 7341 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 startReservedAgentTaskLeased had 0 -> 1. Add a test that tells them apart, or say why they are equivalent.
if (plan.start.kind === "sendMessage") {
assert(plan.start.prompt.length > 0, "startReservedAgentTask requires prompt");

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

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: > -> >= 

No unit test failed when startReservedAgentTaskLeased had > -> >=. Add a test that tells them apart, or say why they are equivalent.
}

const entryAtStart = this.readLaunchRow(plan.taskId);
if (entryAtStart?.workspace.taskStatus !== "starting") {

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

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: ?. -> . 

No unit test failed when startReservedAgentTaskLeased had ?. -> .. Add a test that tells them apart, or say why they are equivalent.
return;
}
// Stop-cascade barrier: no launch while its own or its parent's stop latch is held. The
Expand All @@ -7341,11 +7358,11 @@
await this.editActiveWorkspaceEntry(
plan.taskId,
(workspace) => {
if (workspace.taskStatus !== "starting") return;

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

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: !== -> === 

No unit test failed when startReservedAgentTaskLeased had !== -> ===. Add a test that tells them apart, or say why they are equivalent.
if (rowSupersedes(workspace, plan.attemptId)) return;
workspace.taskStatus = "queued";
},
{ allowMissing: true }

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

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: true -> false 

No unit test failed when startReservedAgentTaskLeased had true -> false. Add a test that tells them apart, or say why they are equivalent.
);
} catch (error) {
await this.markTaskLaunchFailed(plan.taskId, getErrorMessage(error), plan);
Expand All @@ -7354,7 +7371,7 @@
await this.emitWorkspaceMetadata(plan.taskId);
// The release-time drain may have seen "starting" while this rollback was awaiting disk.
// Complete the handoff after publishing "queued", or leave the next release owning it.
if (!this.taskStopLatch.deferLaunchWhileStopInProgress(plan.taskId, plan.parentWorkspaceId)) {

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

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: delete ! 

No unit test failed when startReservedAgentTaskLeased had delete !. Add a test that tells them apart, or say why they are equivalent.
this.scheduleMaybeStartQueuedTasks();
}
return;
Expand All @@ -7379,13 +7396,13 @@
// first, marked unproven — an older build's stale-starting revert may have relaunched an
// admitted execution without any marker.
let launchAttemptId = entryAtStart.workspace.taskAttemptId;
let launchReceiptEligible = entryAtStart.workspace.taskAttemptUnproven !== true;

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

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: true -> false 

No unit test failed when startReservedAgentTaskLeased had true -> false. Add a test that tells them apart, or say why they are equivalent.

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

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: !== -> === 

No unit test failed when startReservedAgentTaskLeased had !== -> ===. Add a test that tells them apart, or say why they are equivalent.
if (!isTaskAttemptId(launchAttemptId)) {
launchReceiptEligible = false;
const stamped = newTaskAttemptId();
try {
await this.editActiveWorkspaceEntry(plan.taskId, (workspace) => {
if (workspace.taskStatus !== "starting" || isTaskAttemptId(workspace.taskAttemptId)) {

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

View workflow job for this annotation

GitHub Actions / Quality / Mutation

Mutation survived: || -> && 

No unit test failed when startReservedAgentTaskLeased had || -> &&. Add a test that tells them apart, or say why they are equivalent.
return;
}
workspace.taskAttemptId = stamped;
Expand Down Expand Up @@ -7452,6 +7469,12 @@

// Track reuse explicitly: owner-derived paths can change during launch, so equality is unsafe.
const sharesParentCheckout = taskWasShared && materialized.reusedExistingCheckout;
const checkout: ReservedLaunchCheckout = {
...materialized,
initLogger,
taskWasShared,
sharesParentCheckout,
};
// The row reads between the fork and the sanitize step. A throw fails the launch, which
// leaves the task resumable: first quarantine a checkout the sanitize step has not run in
// (#4674: sends and MCP refuse it), then end the init so init waiters don't hang.
Expand All @@ -7464,26 +7487,15 @@
throw error;
}
};
const cancelMaterializedLaunch = () =>
this.cancelReservedLaunch(plan, initLogger, {
runtime: materialized.runtimeForTaskWorkspace,
preservePhysicalWorkspace: sharesParentCheckout,
});
if (plan.abortSignal?.aborted) {
await cancelMaterializedLaunch();
await this.cancelMaterializedLaunch(plan, checkout);
return;
}

const entryAfterMaterialize = readLaunchRowBeforeSanitize();
if (!entryAfterMaterialize) {
initLogger.logComplete(-1);
await this.cleanupMaterializedTaskWorkspace(
materialized.runtimeForTaskWorkspace,
plan.parentMeta.projectPath,
plan.workspaceName,
plan.taskId,
{ preservePhysicalWorkspace: sharesParentCheckout }
);
await this.cleanupLaunchCheckout(plan, checkout);
return;
}
if (entryAfterMaterialize.workspace.taskStatus !== "starting") {
Expand Down Expand Up @@ -7531,20 +7543,14 @@
);
await this.emitWorkspaceMetadata(plan.taskId);
if (plan.abortSignal?.aborted) {
await cancelMaterializedLaunch();
await this.cancelMaterializedLaunch(plan, checkout);
return;
}

const entryBeforeSend = readLaunchRowBeforeSanitize();
if (!entryBeforeSend) {
initLogger.logComplete(-1);
await this.cleanupMaterializedTaskWorkspace(
runtimeForTaskWorkspace,
plan.parentMeta.projectPath,
plan.workspaceName,
plan.taskId,
{ preservePhysicalWorkspace: sharesParentCheckout }
);
await this.cleanupLaunchCheckout(plan, checkout);
return;
}
if (
Expand Down Expand Up @@ -7616,7 +7622,7 @@
// already running at its mark. So recheck all four right before starting the init (the
// abort signal alone misses the Stop and the removal).
if (plan.abortSignal?.aborted) {
await cancelMaterializedLaunch();
await this.cancelMaterializedLaunch(plan, checkout);
return;
}
let entryBeforeInit: ReturnType<typeof findWorkspaceEntry>;
Expand All @@ -7634,13 +7640,7 @@
}
if (!entryBeforeInit) {
initLogger.logComplete(-1);
await this.cleanupMaterializedTaskWorkspace(
runtimeForTaskWorkspace,
plan.parentMeta.projectPath,
plan.workspaceName,
plan.taskId,
{ preservePhysicalWorkspace: sharesParentCheckout }
);
await this.cleanupLaunchCheckout(plan, checkout);
return;
}
if (
Expand All @@ -7653,13 +7653,7 @@
if (entryBeforeInit.workspace.pendingRemoval != null) {
// As the admission below would refuse it: the removal owns the row and its checkout.
initLogger.logComplete(-1);
await this.cleanupMaterializedTaskWorkspace(
runtimeForTaskWorkspace,
plan.parentMeta.projectPath,
plan.workspaceName,
plan.taskId,
{ preservePhysicalWorkspace: sharesParentCheckout }
);
await this.cleanupLaunchCheckout(plan, checkout);
throw new Error(pendingRemovalAdmissionMessage(entryBeforeInit.workspace.pendingRemoval));
}
this.registerTaskBackgroundInit(
Expand All @@ -7669,6 +7663,17 @@
);
}

await this.admitReservedLaunchPrompt(plan, checkout);
}

/**
* The launch's linearization point and launch fence: a cancel seen here never launched, and
* past the admission only a Stop ends the execution. Then the send, in the same tick.
*/
private async admitReservedLaunchPrompt(
plan: TaskLaunchPlan,
checkout: ReservedLaunchCheckout
): Promise<void> {
const startOptions = {
model: plan.taskModelString,
agentId: plan.agentId,
Expand All @@ -7680,7 +7685,7 @@
// Linearization point: cancellation observed here guarantees "never launched"; past the
// admission below only a Stop (Layer 2) ends the execution.
if (plan.abortSignal?.aborted) {
await cancelMaterializedLaunch();
await this.cancelMaterializedLaunch(plan, checkout);
return;
}
// Launch fence (same-attempt send under the reservation's id): bind the obligation before
Expand All @@ -7692,16 +7697,23 @@
});
if (admission.kind !== "admitted") {
const message = launchAdmissionRefusalMessage(admission);
await this.cleanupMaterializedTaskWorkspace(
runtimeForTaskWorkspace,
plan.parentMeta.projectPath,
plan.workspaceName,
plan.taskId,
{ preservePhysicalWorkspace: sharesParentCheckout }
);
await this.cleanupLaunchCheckout(plan, checkout);
throw new Error(message);
}
plan.sendAdmitted = true;
await this.sendReservedLaunchPrompt(plan, checkout, admission, startOptions);
}

/**
* Sends the admitted prompt (or resumes the stream) under the admission token. A failed send
* reclaims the checkout and fails the launch. A sent launch becomes `running`.
*/
private async sendReservedLaunchPrompt(
plan: TaskLaunchPlan,
checkout: ReservedLaunchCheckout,
admission: Extract<TaskTurnAdmission, { kind: "admitted" }>,
startOptions: SendMessageOptions
): Promise<void> {
// The token doubles as the send's staleness probe: a guarded send is never a user resume, so
// WorkspaceService skips markInterruptedTaskRunning for it (that rescue serializes on the
// global mutex createMany may still hold while this launch starts). The resume-kind start has
Expand Down Expand Up @@ -7739,13 +7751,7 @@
});
if (!sendResult.success) {
const message = launchSendErrorMessage(sendResult.error);
await this.cleanupMaterializedTaskWorkspace(
runtimeForTaskWorkspace,
plan.parentMeta.projectPath,
plan.workspaceName,
plan.taskId,
{ preservePhysicalWorkspace: sharesParentCheckout }
);
await this.cleanupLaunchCheckout(plan, checkout);
throw new Error(message);
}

Expand All @@ -7764,6 +7770,31 @@
this.scheduleMaybeStartQueuedTasks();
}

/** Cancels a launch after its fork, with the checkout's runtime and its shared flag. */
private cancelMaterializedLaunch(
plan: TaskLaunchPlan,
checkout: ReservedLaunchCheckout
): Promise<void> {
return this.cancelReservedLaunch(plan, checkout.initLogger, {
runtime: checkout.runtimeForTaskWorkspace,
preservePhysicalWorkspace: checkout.sharesParentCheckout,
});
}

/** Reclaims the launch's checkout (never the parent's shared one) when the launch ends early. */
private cleanupLaunchCheckout(
plan: TaskLaunchPlan,
checkout: ReservedLaunchCheckout
): Promise<void> {
return this.cleanupMaterializedTaskWorkspace(
checkout.runtimeForTaskWorkspace,
plan.parentMeta.projectPath,
plan.workspaceName,
plan.taskId,
{ preservePhysicalWorkspace: checkout.sharesParentCheckout }
);
}

createFromRpc(args: RpcTaskCreateArgs): Promise<Result<TaskCreateResult, string>> {
return this.create({
...args,
Expand Down
Loading