diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3607e7272273d..a95e698a1fc93 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6064,7 +6064,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command.dispatchMode, command.deliveryIntent, ); - if (dispatchMode.type === "steer_active") { + if (dispatchMode.type === "steer_active" || dispatchMode.type === "restart_active") { const targetRunId = dispatchMode.targetRunId; const target = projection.runs.find((run) => run.id === targetRunId); const turn = latestProviderTurnForAttempt( @@ -6076,9 +6076,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( target !== undefined && (target.status === "completed" || + (dispatchMode.type === "restart_active" && !isActiveRun(target)) || ((target.status === "running" || target.status === "waiting") && turn !== undefined && - turn.status === "completed")) + (turn.status === "completed" || + (dispatchMode.type === "restart_active" && + turn.status !== "pending" && + turn.status !== "running")))) ) { dispatchMode = { type: "start_immediately" }; } @@ -6086,7 +6090,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Stop leaves the run reading as running until its provider stops. A plain // send in that window queues behind the stopped queue instead of steering // text the agent never receives, and like any user send it releases the - // hold; an explicit steer or restart still fails below. + // hold; an explicit steer still fails below. if ( command.deliveryIntent === "auto" && (dispatchMode.type === "steer_active" || dispatchMode.type === "restart_active") && @@ -6094,6 +6098,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) { dispatchMode = { type: "queue_after_active" }; } + // An explicit restart that Stop reached has only its start left, so it + // queues behind the stopping run like any other start. + if ( + dispatchMode.type === "restart_active" && + (yield* stopReachedRun(command, command.threadId, dispatchMode.targetRunId)) + ) { + dispatchMode = { type: "start_immediately" }; + } const freshContext = command.freshContext === true; if ( freshContext && diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 733ace700ba9e..37b2ab259f840 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1773,6 +1773,225 @@ it.effect("keeps a queued message queued when promoted into a run that Stop reac ), ); +// regression: #32891 a restart that Stop already reached is lost +it.effect("refuses a steer but starts a restart sent into a run that Stop reached", () => + Effect.scoped( + Effect.gen(function* () { + const { started, steered, layer, startFirstTurn } = + yield* nextTurnSelectionHarness("restart-after-stop"); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + // The worker has not delivered the interrupt yet, so the run still reads as running. + yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make("stop"), + threadId, + runId: first.runId, + }); + const steerError = yield* orchestrator + .dispatch({ + type: "message.dispatch", + commandId: CommandId.make("steer"), + threadId, + messageId: MessageId.make("message:steer"), + text: "steer", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "steer_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }) + .pipe(Effect.flip); + assert.equal(steerError._tag, "OrchestratorDispatchError"); + assert.include(String(steerError.cause), "is stopping and cannot be steered"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("restart"), + threadId, + messageId: MessageId.make("message:restart"), + text: "restart", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "restart_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + + assert.deepEqual(steered, []); + assert.deepEqual( + started.map((turn) => turn.message.text), + ["first", "restart"], + ); + }).pipe(Effect.provide(layer)); + }), + ), +); + +// regression: #32891 a restart that arrives after the turn ended is lost +it.effect.each([ + { status: "completed", settled: true }, + { status: "completed", settled: false }, + { status: "interrupted", settled: true }, + { status: "interrupted", settled: false }, + { status: "cancelled", settled: false }, +] as const)( + "starts a restart that missed the turn ($status, run settled: $settled) as a new turn", + ({ status, settled: runSettled }) => + Effect.scoped( + Effect.gen(function* () { + const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + `restart-after-${status}-${runSettled ? "settled" : "unsettled"}`, + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + const turnEnded = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.payload.id === turn.id && + event.payload.status === status, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status, completedAt: yield* DateTime.now }, + }); + yield* Fiber.join(turnEnded); + const settleRun = Effect.gen(function* () { + const settled = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === first.runId && + event.payload.status !== "running", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + runOrdinal: first.runOrdinal, + status, + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); + }); + if (runSettled) yield* settleRun; + + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("restart"), + threadId, + messageId: MessageId.make("message:restart"), + text: "restart", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "restart_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + if (!runSettled) { + const restartLeftQueue = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.userMessageId === "message:restart" && + event.payload.status !== "queued", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* settleRun; + yield* Fiber.join(restartLeftQueue); + yield* worker.drain(); + } + + assert.deepEqual(steered, []); + assert.deepEqual( + started.map((turn) => turn.message.text), + ["first", "restart"], + ); + }).pipe(Effect.provide(layer)); + }), + ), +); + +// regression: #32891 a restart into a failed provider turn is lost +it.effect("queues a restart sent into a provider turn that failed before its run settled", () => + Effect.scoped( + Effect.gen(function* () { + const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + "restart-after-failed-unsettled", + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + const turnFailed = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.payload.id === turn.id && + event.payload.status === "failed", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status: "failed", completedAt: yield* DateTime.now }, + }); + yield* Fiber.join(turnFailed); + + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("restart"), + threadId, + messageId: MessageId.make("message:restart"), + text: "restart", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "restart_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + + assert.deepEqual(steered, []); + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + projection.runs.find((run) => run.userMessageId === "message:restart")?.status, + "queued", + ); + }).pipe(Effect.provide(layer)); + }), + ), +); + it.effect("queues a plain send behind the stopped queue and releases its hold", () => Effect.scoped( Effect.gen(function* () {