Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
37 changes: 36 additions & 1 deletion apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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) &&
Expand Down
107 changes: 107 additions & 0 deletions apps/server/src/orchestration-v2/runtimeLayer.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading