diff --git a/apps/server/src/orchestration-v2/CommandPolicy.test.ts b/apps/server/src/orchestration-v2/CommandPolicy.test.ts index 85babb8a869bf..4a7af95214a5a 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.test.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.test.ts @@ -60,6 +60,7 @@ function dispatchProjection( sessionCapabilities === undefined ? [] : [{ runAttemptId: activeAttemptId, status: "running" }], + runtimeRequests: [], } as unknown as OrchestrationV2ThreadProjection; } @@ -244,6 +245,58 @@ it.each(["/compact", " /LOGOUT "])("queues a steer sent while %s runs", (text) = ); }); +it.each(["user_input", "command"] as const)( + "queues every steer while the active turn waits on a pending %s request", + (kind) => { + const projection = dispatchProjection(baseCapabilities); + const waiting = { + ...projection, + runtimeRequests: [{ kind, status: "pending", responseCapability: { type: "live" } }], + } as unknown as OrchestrationV2ThreadProjection; + for (const deliveryIntent of ["auto", "steer"] as const) { + assert.deepEqual( + CommandPolicy.resolveMessageDispatchIntent( + waiting, + { type: "start_immediately" }, + deliveryIntent, + ), + { type: "queue_after_active" }, + ); + } + assert.deepEqual( + CommandPolicy.resolveMessageDispatchIntent(waiting, { + type: "steer_active", + targetRunId: activeRunId, + }), + { type: "queue_after_active" }, + ); + assert.deepEqual( + CommandPolicy.resolveMessageDispatchIntent(waiting, { type: "start_immediately" }, "restart"), + { type: "restart_active", targetRunId: activeRunId }, + ); + // An answered request, or one answered by a later message, has no callback to abort. + for (const request of [ + { kind, status: "resolved", responseCapability: { type: "live" } }, + { kind, status: "pending", responseCapability: { type: "message" } }, + ]) { + const nonBlocking = { + ...projection, + runtimeRequests: [request], + } as unknown as OrchestrationV2ThreadProjection; + for (const deliveryIntent of ["auto", "steer"] as const) { + assert.deepEqual( + CommandPolicy.resolveMessageDispatchIntent( + nonBlocking, + { type: "start_immediately" }, + deliveryIntent, + ), + { type: "steer_active", targetRunId: activeRunId }, + ); + } + } + }, +); + const layer = it.layer(CommandPolicy.layer); layer("CommandPolicyV2", (it) => { diff --git a/apps/server/src/orchestration-v2/CommandPolicy.ts b/apps/server/src/orchestration-v2/CommandPolicy.ts index cd56460369bd6..faab7a1080092 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.ts @@ -90,6 +90,19 @@ export function goalCommandStopsActiveRun( } } +/** + * A steer aborts the provider's pending question or approval callback before the person + * answers it, so while one is pending, any steer waits behind the turn instead. A request + * answered by a message has no callback and does not hold steers. + */ +export function waitsOnPendingRequest( + projection: Pick, +): boolean { + return projection.runtimeRequests.some( + (request) => request.status === "pending" && request.responseCapability.type !== "message", + ); +} + export function resolveMessageDispatchIntent( projection: OrchestrationV2ThreadProjection, requestedMode: MessageDispatchMode, @@ -97,6 +110,7 @@ export function resolveMessageDispatchIntent( ): MessageDispatchMode { const mode = resolveDeliveryIntent(projection, requestedMode, deliveryIntent); if (mode.type !== "steer_active") return mode; + if (waitsOnPendingRequest(projection)) return { type: "queue_after_active" }; // Compaction and sign-out cannot accept steers; queue follow-ups. const target = projection.runs.find((run) => run.id === mode.targetRunId); const targetMessage = projection.messages.find((message) => message.id === target?.userMessageId); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c7cd7a9ff288e..189ae8ff1d91d 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -114,6 +114,7 @@ import { isActiveRun, isNativeMaintenanceCommand, resolveMessageDispatchIntent, + waitsOnPendingRequest, } from "./CommandPolicy.ts"; import { CommandReceiptStoreV2, type CommandReceiptV2 } from "./CommandReceiptStore.ts"; import { @@ -6737,6 +6738,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (message) => message.id === target.userMessageId, ); if ( + (resolvedMode.type === "restart_active" || !waitsOnPendingRequest(projection)) && isLiveRun(target) && goalCommandStopsActiveRun(goalCommand, goal?.status ?? null) && (targetMessage === undefined || !isNativeMaintenanceCommand(targetMessage)) && @@ -10869,6 +10871,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: { ...queuedRun, queuePosition: 1 }, }); const stops = + !waitsOnPendingRequest(projection) && target !== undefined && isLiveRun(target) && goalCommandStopsActiveRun(goalCommand, goal?.status ?? null) && @@ -10885,6 +10888,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio now, }); } + if (waitsOnPendingRequest(projection)) { + // It goes first in the queue instead and runs after the answer. + yield* emitEvent({ + type: "run.updated", + threadId: command.threadId, + runId: queuedRun.id, + nodeId: queuedRootNode.id, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...queuedRun, queuePosition: 1 }, + }); + return yield* stopRunForQueuedCommand({ + command, + events, + effects, + projection, + queuedRunId: queuedRun.id, + targetRunId: undefined, + reason: "Queued behind a pending request", + now, + }); + } if ( (target?.status === "preparing" || target?.status === "starting" || diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index ad0c179d783fe..2be65143baf0a 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -9,8 +9,10 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, + ProviderSessionId, ProviderThreadId, ProviderTurnId, + RuntimeRequestId, ThreadId, TurnItemId, type ModelSelection, @@ -2270,6 +2272,126 @@ it.effect("keeps a queued message queued when promoted into a run that Stop reac ), ); +it.effect.each(["ordinary", "goal-steer", "goal-promotion", "goal-restart"] as const)( + "queues a steer and puts a promoted message first while the turn waits on the person: %s", + (route) => + Effect.scoped( + Effect.gen(function* () { + const { started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + `steering-pending-request-${route}`, + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const sink = yield* EventSink.EventSinkV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + const messageText = (id: string) => + ((route === "goal-steer" || route === "goal-restart") && id === "steered") || + (route === "goal-promotion" && id === "promoted") + ? "/goal ship it" + : id; + const send = (id: string, delivery: "queue" | "steer") => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(id), + threadId, + messageId: MessageId.make(`message:${id}`), + text: messageText(id), + attachments: [], + modelSelection: runSelection, + ...(delivery === "queue" + ? { dispatchMode: { type: "queue_after_active" } } + : { + dispatchMode: { type: "start_immediately" }, + deliveryIntent: + route === "goal-restart" && id === "steered" ? "restart" : "steer", + }), + createdBy: "user", + creationSource: "web", + }); + yield* send("queued", "queue"); + yield* send("promoted", "queue"); + const promotedRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === MessageId.make("message:promoted"), + )!; + const now = yield* DateTime.now; + yield* sink.write({ + events: [ + { + id: EventId.make("request:question"), + type: "runtime-request.updated", + threadId, + runId: first.runId, + nodeId: first.rootNodeId, + occurredAt: now, + payload: { + id: RuntimeRequestId.make("request:question"), + nodeId: first.rootNodeId, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { + type: "live", + providerSessionId: ProviderSessionId.make("session:waiting"), + }, + createdAt: now, + resolvedAt: null, + }, + }, + ], + }); + + yield* orchestrator.dispatch({ + type: "queued-message.promote-to-steer", + commandId: CommandId.make("promote"), + threadId, + queuedRunId: promotedRun.id, + targetRunId: first.runId, + }); + yield* send("steered", "steer"); + yield* worker.drain(); + + const projection = yield* orchestrator.getThreadProjection(threadId); + if (route === "goal-restart") { + assert.isTrue( + projection.turnItems.some((item) => item.type === "run_interrupt_request"), + ); + assert.deepEqual( + started.map((turn) => turn.message.text), + ["first", "/goal ship it"], + ); + assert.deepEqual( + projection.runtimeRequests.map((request) => request.status), + ["cancelled"], + ); + return; + } + assert.isFalse( + projection.turnItems.some((item) => item.type === "run_interrupt_request"), + ); + assert.deepEqual(steered, []); + assert.equal(started.length, 1); + assert.deepEqual( + projection.runs + .filter((run) => run.status === "queued") + .sort((left, right) => (left.queuePosition ?? 0) - (right.queuePosition ?? 0)) + .map( + (run) => + projection.messages.find((message) => message.id === run.userMessageId)?.text, + ), + ["promoted", "queued", "steered"].map(messageText), + ); + assert.deepEqual( + projection.runtimeRequests.map((request) => request.status), + ["pending"], + ); + }).pipe(Effect.provide(layer)); + }), + ), +); + // regression: #32891 a restart that Stop already reached is lost it.effect.each([false, true])( "refuses a steer after Stop (provider turn ready: %s)", diff --git a/docs/user/composer.md b/docs/user/composer.md index df880e49c6ca5..aed972b6739e2 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -69,6 +69,9 @@ are saved on the server and can be edited or removed before they send. When Enter sends, `Cmd+Enter` on macOS or `Ctrl+Enter` on Windows and Linux uses the opposite action: it steers when your default is Queue and queues when your default is Steer. +While the agent waits on your answer to a question or an approval, a steer queues +instead, and a queued message you steer moves to the front of the queue, so the +question stays until you answer it. If your **Send shortcut** requires `Cmd+Enter` or `Ctrl+Enter`, that shortcut uses your default follow-up behavior; add **Shift** to use the opposite action. This applies only to multiline messages when you choose the multiline send option.