diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 6c3114329da21..37c4a75f663b6 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6063,10 +6063,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio taskIds: delivery.taskIds, }; } - // Route durable mailbox deliveries and delegated task questions under the - // thread lock, using the live session's capabilities. A question queued - // behind a long parent turn is never seen. Never interrupt/restart a turn - // for a notification. + // Steer mailbox deliveries, delegated task questions, and command notices + // into an active goal under the thread lock: queued behind a long turn or a + // goal run, they are never seen. Never interrupt/restart a turn for one. + const noticeKind = command.notification?.source.kind; + const commandNotice = noticeKind === "command" && command.creationSource === "server"; if ( !freshContext && (delegatedCompletion !== undefined @@ -6074,7 +6075,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (id) => projection.subagents.find((task) => task.id === id)?.completionWake === "always", ) - : command.notification?.source.kind === "delegated_task") + : noticeKind === "delegated_task" || commandNotice) ) { const active = projection.runs.find((run) => run.status === "running"); const providerThread = projection.providerThreads.find( @@ -6090,6 +6091,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio active !== undefined && activeTurn !== undefined && providerThread?.providerSessionId != null && + (!commandNotice || providerThread.goal?.status === "active") && (activeMessage === undefined || !isNativeMaintenanceCommand(activeMessage)) ) { const session = yield* providerSessions diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 2b08c0b8345b6..2c311f2569753 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -651,63 +651,97 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* return { cwd, events, started, steered, interrupted, layer, startFirstTurn }; }); -it.effect("steers a delegated task question notice into the running parent turn", () => - Effect.scoped( - Effect.gen(function* () { - const { started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( - "delegated-question-steer", - ); - yield* Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - const threadId = yield* startFirstTurn; - const { thread } = yield* orchestrator.getThreadProjection(threadId); - yield* (yield* EventSink.EventSinkV2).write({ - events: [ - { - id: EventId.make("event:composer-selection"), - type: "thread.model-selection-updated", - threadId, - providerInstanceId: instanceId, - occurredAt: yield* DateTime.now, - payload: { ...thread, modelSelection: composerSelection }, - }, - ], - }); - const messageId = MessageId.make("message:delegated-task-question:child-question"); - const notification = { - source: { - kind: "delegated_task" as const, - taskIds: [NodeId.make("task:question")], - childThreadId: ThreadId.make("thread:question-child"), - }, - outcome: "updated" as const, - summary: "Delegated task has a question", - }; - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make("question-wake"), - threadId, - messageId, - text: "Delegated task asked a question", - notification, - attachments: [], - dispatchMode: { type: "queue_after_active" }, - createdBy: "agent", - creationSource: "server", - }); - yield* worker.drain(); - const after = yield* orchestrator.getThreadProjection(threadId); - assert.deepEqual(steered, ["Delegated task asked a question"]); - assert.lengthOf(started, 1); - assert.isFalse(after.runs.some((run) => run.status === "queued")); - const message = after.messages.find((candidate) => candidate.id === messageId); - assert.equal(message?.runId, started[0]?.runId); - assert.deepEqual(message?.notification, notification); - assert.deepEqual(after.thread.modelSelection, composerSelection); - }).pipe(Effect.provide(layer)); - }), - ), +it.effect.each([ + { notice: "delegated task question", goal: null, steers: true }, + { notice: "background command", goal: "active", steers: true }, + { notice: "background command", goal: null, steers: false }, +] as const)( + "a $notice notice steers into the running turn: $steers (goal $goal)", + ({ notice, goal, steers }) => + Effect.scoped( + Effect.gen(function* () { + const { started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + `notice-steer-${notice.replaceAll(" ", "-")}-${goal ?? "none"}`, + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const { thread } = yield* orchestrator.getThreadProjection(threadId); + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("event:composer-selection"), + type: "thread.model-selection-updated", + threadId, + providerInstanceId: instanceId, + occurredAt: yield* DateTime.now, + payload: { ...thread, modelSelection: composerSelection }, + }, + ], + }); + if (goal !== null) { + const providerThread = (yield* orchestrator.getThreadProjection(threadId)) + .providerThreads[0]!; + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("event:goal-active"), + type: "provider-thread.updated", + threadId, + occurredAt: yield* DateTime.now, + payload: { ...providerThread, goal: { objective: "Ship it", status: goal } }, + }, + ], + }); + } + const messageId = MessageId.make("message:notice"); + const notification = + notice === "delegated task question" + ? { + source: { + kind: "delegated_task" as const, + taskIds: [NodeId.make("task:question")], + childThreadId: ThreadId.make("thread:question-child"), + }, + outcome: "updated" as const, + summary: "Delegated task has a question", + } + : { + source: { kind: "command" as const }, + outcome: "completed" as const, + summary: "Background command finished", + }; + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("notice-wake"), + threadId, + messageId, + text: `${notification.summary}\nexit 0: built 12 files`, + notification, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "server", + }); + yield* worker.drain(); + const after = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + steered, + steers ? [`${notification.summary}\nexit 0: built 12 files`] : [], + ); + assert.lengthOf(started, 1); + assert.equal( + after.runs.some((run) => run.status === "queued"), + !steers, + ); + const message = after.messages.find((candidate) => candidate.id === messageId); + assert.deepEqual(message?.notification, notification); + if (steers) assert.equal(message?.runId, started[0]?.runId); + assert.deepEqual(after.thread.modelSelection, composerSelection); + }).pipe(Effect.provide(layer)); + }), + ), ); it.effect("sends a selected held message once and preserves the remaining queue order", () =>