diff --git a/apps/server/src/orchestration-v2/CommandPolicy.test.ts b/apps/server/src/orchestration-v2/CommandPolicy.test.ts index 1bf294641a01c..17c61dc921766 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; } @@ -197,6 +198,52 @@ 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" }], + } 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 }, + ); + const answered = { + ...projection, + runtimeRequests: [{ kind, status: "resolved" }], + } as unknown as OrchestrationV2ThreadProjection; + for (const deliveryIntent of ["auto", "steer"] as const) { + assert.deepEqual( + CommandPolicy.resolveMessageDispatchIntent( + answered, + { 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 302bb12f797cb..83db884d89f56 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.ts @@ -89,6 +89,16 @@ 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. + */ +export function waitsOnPendingRequest( + projection: Pick, +): boolean { + return projection.runtimeRequests.some((request) => request.status === "pending"); +} + export function resolveMessageDispatchIntent( projection: OrchestrationV2ThreadProjection, requestedMode: MessageDispatchMode, @@ -96,6 +106,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 91acb881637c1..2b94856e2776e 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -101,6 +101,7 @@ import { goalCommandStopsActiveRun, isNativeMaintenanceCommand, resolveMessageDispatchIntent, + waitsOnPendingRequest, } from "./CommandPolicy.ts"; import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts"; import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts"; @@ -5240,6 +5241,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)) && @@ -8994,6 +8996,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) && @@ -9010,6 +9013,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, + }); + } yield* emitEvent({ type: "run.updated", threadId: command.threadId, diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index d8ac4769e140f..f3909e6d1969d 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -9,6 +9,7 @@ import { ProviderInstanceId, ProviderThreadId, ProviderTurnId, + RuntimeRequestId, ThreadId, TurnItemId, type ModelSelection, @@ -1107,6 +1108,123 @@ 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: "message" }, + 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)); + }), + ), +); + it.effect("queues a plain send behind the stopped queue and releases its hold", () => Effect.scoped( Effect.gen(function* () { diff --git a/docs/user/composer.md b/docs/user/composer.md index 714572dfd5005..30d7edaa2d5ad 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -59,6 +59,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.