diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 26482feb129fd..7874eb6ffb34e 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -9720,7 +9720,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if (task.result !== null) { // A Stop or cancel here that lost the race to the result still owes its stop. if (command.stoppedThere === undefined) { - return yield* emitRemoteStopOwed(events, command, task, task.remoteChild!); + yield* emitRemoteStopOwed(events, command, task, task.remoteChild!); + return; } const owesStop = task.status === "cancelled" || @@ -9816,6 +9817,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } as OrchestrationV2TurnItem, }); } + return plan.parentRun?.id; }); /** @@ -14860,7 +14862,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* dispatchRemoteDelegatedTaskRequest(command, events); break; case "delegated_task.remote.complete": - yield* dispatchRemoteDelegatedTaskComplete(command, events); + updateDelegatedCompletionOnlyRunId = yield* dispatchRemoteDelegatedTaskComplete( + command, + events, + ); break; case "delegated_task.remote.stop": yield* dispatchRemoteDelegatedTaskStop(command, events); diff --git a/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts b/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts index f075b43d9f882..e824c69af7aea 100644 --- a/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts +++ b/apps/server/src/orchestration-v2/RemoteDelegatedTask.test.ts @@ -1,5 +1,6 @@ -import { assert, it } from "@effect/vitest"; +import { assert, it, vi } from "@effect/vitest"; import { + CheckpointId, CommandId, EnvironmentId, EventId, @@ -419,4 +420,79 @@ it.layer(layerTest)("a delegated task in a linked environment", (it) => { assert.equal(offered[0]?.threadId, parent.threadId); }), ); + it.effect("keeps a parent run that settled between its result's read and commit", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const parent = yield* seedLiveParent("settles-mid-result"); + yield* orchestrator.dispatch({ + type: "delegated_task.remote.request", + commandId: CommandId.make("command:remote-delegate:settles-mid-result"), + parentThreadId: parent.threadId, + parentRunId: parent.runId, + parentNodeId: parent.rootNodeId, + task: "Run the suite there.", + driver: ProviderDriverKind.make("remote"), + modelSelection, + remoteChild, + completionWake: "always", + }); + const projection = yield* orchestrator.getThreadProjection(parent.threadId); + const task = projection.subagents[0]!; + const run = projection.runs.find((candidate) => candidate.id === parent.runId)!; + const checkpointId = CheckpointId.make("checkpoint:settles-mid-result"); + const commit = sink.commitCommand; + let settled = false; + yield* Effect.acquireUseRelease( + Effect.sync(() => + vi.spyOn(sink, "commitCommand").mockImplementation((input) => + Effect.gen(function* () { + if (!settled && input.commandType === "delegated_task.remote.complete") { + settled = true; + // The parent's checkpoint lands after the result read its run. + yield* commit({ + commandId: CommandId.make("command:checkpoint:settles-mid-result"), + threadId: parent.threadId, + commandType: "checkpoint.capture", + acceptedAt: now, + effects: [], + events: [ + { + id: EventId.make("event:settles-mid-result:settled"), + type: "run.updated", + threadId: parent.threadId, + runId: parent.runId, + occurredAt: now, + payload: { ...run, status: "completed", completedAt: now, checkpointId }, + }, + ], + }); + } + return yield* commit(input); + }), + ), + ), + () => + orchestrator.dispatch({ + type: "delegated_task.remote.complete", + commandId: CommandId.make("command:remote-complete:settles-mid-result"), + parentThreadId: parent.threadId, + taskId: task.id, + status: "completed", + result: "All 412 tests pass on the box.", + stoppedThere: "ran-out", + }), + (spy) => Effect.sync(() => spy.mockRestore()), + ); + assert.isTrue(settled); + const after = (yield* orchestrator.getThreadProjection(parent.threadId)).runs.find( + (candidate) => candidate.id === parent.runId, + )!; + assert.deepEqual( + { status: after.status, completedAt: after.completedAt, checkpointId: after.checkpointId }, + { status: "completed", completedAt: now, checkpointId }, + ); + }), + ); });