diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 1499ca975de7..b2a1122cb0b5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3301,8 +3301,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); const emitEvent = emit(events, command); const activeRunIds = new Set( - projection.runs.filter((run) => run.status === "queued").map((run) => run.id), + projection.runs + .filter((run) => run.status === "queued" || run.status === "preparing") + .map((run) => run.id), ); + const preparingRuns = projection.runs.filter((run) => run.status === "preparing"); + const preparationItems = + preparingRuns.length === 0 + ? [] + : (yield* loadProjectionForCommand(command, ["turnItems"], { + turnItemTypes: ["command_execution"], + turnItemRunIds: preparingRuns.map((run) => run.id), + })).turnItems; for (const run of projection.runs.filter((candidate) => activeRunIds.has(candidate.id))) { yield* emitEvent({ type: "run.updated", @@ -3313,6 +3323,31 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: { ...run, status: "cancelled", queuePosition: null, completedAt: now }, }); } + for (const run of preparingRuns) { + const preparationItem = preparationItems.find( + (item) => + item.runId === run.id && + item.type === "command_execution" && + item.input === WORKSPACE_PREPARATION_INPUT && + item.status === "running", + ); + if (preparationItem === undefined) continue; + yield* emitEvent({ + type: "turn-item.updated", + threadId: command.threadId, + runId: run.id, + nodeId: preparationItem.nodeId ?? undefined, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { + ...preparationItem, + status: "cancelled", + title: "Workspace preparation cancelled", + completedAt: now, + updatedAt: now, + }, + }); + } for (const attempt of projection.attempts.filter( (candidate) => activeRunIds.has(candidate.runId) && diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 086c52296e9d..d4d1fc668b9e 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -3580,6 +3580,113 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("cancels preparing work on archive and keeps it cancelled through replay", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const threadId = ThreadId.make("runtime-layer-archive-preparing-thread"); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("archive-preparing-create"), + threadId, + projectId: ProjectId.make("archive-preparing-project"), + title: "Archive preparing work", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("archive-preparing-message"), + threadId, + messageId: MessageId.make("archive-preparing-message"), + text: "Prepare my workspace.", + attachments: [], + modelSelection, + dispatchMode: { type: "defer_start" }, + }); + const before = yield* orchestrator.getThreadProjection(threadId); + assert.lengthOf(before.runs, 1); + assert.isEmpty(before.subagents); + const run = before.runs[0]!; + assert.equal(run.status, "preparing"); + const preparation = before.turnItems.find((item) => item.type === "command_execution")!; + assert.equal(preparation.status, "running"); + assert.isEmpty((yield* orchestrator.getThreadShell(threadId))!.pendingBackgroundTasks); + + const archive = { + type: "thread.archive" as const, + commandId: CommandId.make("archive-preparing-archive"), + threadId, + }; + const archivedResult = yield* orchestrator.dispatch(archive); + assert.equal((yield* orchestrator.dispatch(archive)).sequence, archivedResult.sequence); + + const assertCancelled = Effect.gen(function* () { + const projection = yield* orchestrator.getThreadProjection(threadId); + const cancelledRun = projection.runs.find((candidate) => candidate.id === run.id)!; + assert.equal(cancelledRun.status, "cancelled"); + assert.isNull(cancelledRun.queuePosition); + assert.isNotNull(cancelledRun.completedAt); + assert.equal(projection.attempts[0]?.status, "cancelled"); + assert.isNotNull(projection.attempts[0]?.completedAt); + assert.equal(projection.nodes[0]?.status, "cancelled"); + assert.isNotNull(projection.nodes[0]?.completedAt); + const item = projection.turnItems.find((candidate) => candidate.id === preparation.id)!; + assert.equal(item.status, "cancelled"); + assert.equal(item.title, "Workspace preparation cancelled"); + assert.deepEqual(item.completedAt, cancelledRun.completedAt); + assert.deepEqual(item.updatedAt, cancelledRun.completedAt); + assert.isEmpty((yield* orchestrator.getThreadShell(threadId))!.pendingBackgroundTasks); + assert.equal(yield* orchestrator.resumeQueuedRuns, 0); + }); + assert.isNotNull((yield* orchestrator.getThreadProjection(threadId)).thread.archivedAt); + yield* assertCancelled; + assert.isTrue((yield* maintenance.rebuild).valid); + yield* assertCancelled; + + // Preparation may finish after archive. Even reopening must not let its + // late callbacks revive the cancelled run or its background activity. + yield* orchestrator.dispatch({ + type: "thread.unarchive", + commandId: CommandId.make("archive-preparing-unarchive"), + threadId, + }); + for (const callback of [ + { type: "prepared-run.progress", phase: "setup" }, + { type: "prepared-run.release" }, + { + type: "prepared-run.fail", + failure: { + class: "validation_error", + message: "Late preparation failure", + code: null, + retryable: false, + }, + }, + ] as const) { + const error = yield* orchestrator + .dispatch({ + ...callback, + commandId: CommandId.make(`archive-preparing-late:${callback.type}`), + threadId, + runId: run.id, + }) + .pipe(Effect.flip); + assert.equal(error._tag, "OrchestratorDispatchError"); + } + yield* assertCancelled; + assert.isTrue((yield* maintenance.rebuild).valid); + yield* assertCancelled; + }), + ); + it.effect("cancels queued work when a thread is archived", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2;