diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 0d1366dbd858e..fb584170a280c 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -12102,7 +12102,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); it.effect( - "uses task_started as an incremental roster fallback, keeps its tool call, and clears internal work on empty snapshot", + "uses task_started as an incremental roster fallback and keeps its owed wake after an empty snapshot", () => Effect.gen(function* () { const harness = yield* makeWakeHarness; @@ -12250,7 +12250,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ).map((task) => task.taskId), WAKE_TASK_ID, ); - assert.isFalse(yield* harness.hasPendingBackgroundWork); + assert.isTrue(yield* harness.hasPendingBackgroundWork); }).pipe( Effect.provide( Layer.mergeAll( @@ -15308,9 +15308,18 @@ describe("ClaudeAdapterV2 background wake turns", () => { assert.isTrue(yield* harness.hasPendingBackgroundWork); assert.lengthOf(harness.continuationRequests, 0); + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "system", + subtype: "background_tasks_changed", + tasks: [], + uuid: "roster-before-notification", + }), + ); + assert.isTrue(yield* harness.hasPendingBackgroundWork); yield* harness.offerAndWait(wakeNotification); assert.lengthOf(harness.continuationRequests, 0); - assert.isFalse(yield* harness.hasPendingBackgroundWork); + assert.isTrue(yield* harness.hasPendingBackgroundWork); yield* Queue.offer(harness.sdkMessages, wakeAssistant); yield* harness.awaitContinuation(1); @@ -16199,8 +16208,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* harness.offerAndWait(emptyRoster); yield* harness.offerAndWait(wakeNotification); if (emptyLevel === "after") yield* harness.offerAndWait(emptyRoster); - // Query replacement reads the internal roster, which the exit already cleared. - assert.isFalse(yield* harness.hasPendingBackgroundWork); + assert.isTrue(yield* harness.hasPendingBackgroundWork); yield* harness.offerAndWait(wakeTurnInit); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -20662,7 +20670,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); yield* Queue.offer(firstProcess, turnOneResult); yield* changes.until(() => events.some((event) => event.type === "turn.terminal")); - assert.isFalse(yield* hasPendingBackgroundWork); + assert.isTrue(yield* hasPendingBackgroundWork); const alternateModel = CLAUDE_TEST_RESTART_MODEL_SELECTION; const activeProviderThread = { @@ -20861,7 +20869,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* changes.until( () => events.filter((event) => event.type === "turn.terminal").length === 2, ); - assert.isFalse(yield* hasPendingBackgroundWork); + assert.isTrue(yield* hasPendingBackgroundWork); assert.lengthOf(continuationRequests, 0); yield* Queue.offer(secondProcess, wakeResult); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 0ac551140704f..cd07511b83096 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -13749,6 +13749,14 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return true; } } + // An empty roster can precede task_notification; the wake it owes is still pending. + for (const taskIds of (yield* Ref.get( + wakeEligibleBackgroundTasksByNativeThread, + )).values()) { + if (taskIds.size > 0) { + return true; + } + } const buffers = yield* Ref.get(wakeBuffers); for (const entry of buffers.values()) { if ( @@ -13756,7 +13764,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( (message) => message.type === "user" || message.type === "assistant" || - message.type === "result", + message.type === "result" || + // The roster empties on the notification, before the wake turn's frames arrive. + (message.type === "system" && message.subtype === "task_notification"), ) ) { return true; diff --git a/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts new file mode 100644 index 0000000000000..1eb614adf4296 --- /dev/null +++ b/apps/server/src/orchestration-v2/ClaudeAccountSwitchWake.integration.test.ts @@ -0,0 +1,509 @@ +import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + ClaudeSettings, + CommandId, + MessageId, + ProviderInstanceId, + ProjectId, + ThreadId, + type ModelSelection, + type OrchestrationV2DomainEvent, +} from "@t3tools/contracts"; +import * as SecureRandom from "effect/crypto/SecureRandom"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; +import * as Exit from "effect/Exit"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import { ProviderAdapterOpenSessionError } from "@t3tools/provider-core/server/ProviderAdapter"; +import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; +import type * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; +import * as ClaudeAdapterV2 from "./Adapters/ClaudeAdapterV2.ts"; +import { ClaudeProviderCapabilitiesV2 } from "./Adapters/ClaudeAdapterV2.ts"; +import * as EventSink from "./EventSink.ts"; +import * as EffectWorker from "./EffectWorker.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; +import * as ProviderSessionManager from "./ProviderSessionManager.ts"; + +const nativeSessionId = "account-switch-wake-session"; +const settings = Schema.decodeSync(ClaudeSettings)({}); +const firstAccount: ModelSelection = { + instanceId: ProviderInstanceId.make("claudeAgent"), + model: "claude-sonnet-4-6", +}; +const secondAccount: ModelSelection = { + instanceId: ProviderInstanceId.make("claude-second-account"), + model: "claude-sonnet-4-6", +}; + +function frame(value: Record): SDKMessage { + return { session_id: nativeSessionId, ...value } as unknown as SDKMessage; +} + +const resultFrame = (uuid: string, text: string, wake = false) => + frame({ + type: "result", + subtype: "success", + uuid, + is_error: false, + num_turns: 1, + result: text, + stop_reason: "end_turn", + terminal_reason: "completed", + permission_denials: [], + duration_ms: 1, + duration_api_ms: 1, + total_cost_usd: 0, + usage: { input_tokens: 1, output_tokens: 1 }, + modelUsage: {}, + ...(wake ? { origin: { kind: "task-notification" } } : {}), + }); + +const assistantFrame = (uuid: string, text: string) => + frame({ + type: "assistant", + uuid, + parent_tool_use_id: null, + message: { + id: `msg_${uuid}`, + role: "assistant", + type: "message", + model: firstAccount.model, + content: [{ type: "text", text }], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 1, output_tokens: 1 }, + }, + }); + +const backgroundBashStarted = frame({ + type: "system", + subtype: "task_started", + task_id: "bash-build", + tool_use_id: "toolu_bash_build", + description: "Build in the background", + is_backgrounded: true, + task_type: "local_bash", + uuid: "bash-started", +}); +// Claude may publish the emptied roster before the notification that owes a wake turn. +const emptyRoster = frame({ + type: "system", + subtype: "background_tasks_changed", + tasks: [], + uuid: "roster-empty", +}); +const bashNotification = frame({ + type: "system", + subtype: "task_notification", + task_id: "bash-build", + tool_use_id: "toolu_bash_build", + status: "completed", + output_file: "/tmp/claude-replay/tasks/bash-build.output", + summary: 'Background command "Build in the background" completed (exit code 0)', + uuid: "bash-notification", +}); +const WAKE_REPLY = "BUILD_REPORT_DELIVERED"; + +const makeAccount = Effect.fn("makeAccount")(function* (input: { + readonly instanceId: ProviderInstanceId; + readonly cwd: string; + readonly continuations: Queue.Queue; + readonly closed: Array; +}) { + const processes: Array> = []; + const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({ + instanceId: input.instanceId, + settings, + environment: {}, + attachmentsDir: input.cwd, + fileSystem: yield* FileSystem.FileSystem, + path: yield* Path.Path, + random: yield* SecureRandom.SecureRandom, + idAllocator: yield* IdAllocator.IdAllocatorV2, + continuationRequests: { + offer: (request) => Queue.offer(input.continuations, request).pipe(Effect.asVoid), + }, + queryRunner: { + allocateSessionId: Effect.succeed(nativeSessionId), + open: () => + Effect.gen(function* () { + const messages = yield* Queue.unbounded(); + processes.push(messages); + return { + messages: Stream.fromQueue(messages), + offer: () => Effect.void, + setModel: () => Effect.void, + renameSession: () => Effect.void, + applyFlagSettings: () => Effect.void, + setPermissionMode: () => Effect.void, + backgroundTasks: () => Effect.succeed(false), + mcpServerStatus: Effect.succeed([]), + readMcpResource: () => Effect.die("unexpected MCP resource read"), + callMcpTool: () => Effect.die("unexpected MCP tool call"), + listPermissionRules: Effect.die("unexpected permission rules read"), + stopTask: () => Effect.void, + interrupt: Effect.void, + close: Effect.sync(() => void input.closed.push(input.instanceId)).pipe( + Effect.andThen(Queue.shutdown(messages)), + ), + }; + }), + forkSession: () => Effect.die("unused"), + subagentLaunchToolUseId: () => Effect.succeed(null), + subagentModel: () => Effect.succeed(null), + assertComplete: Effect.void, + }, + }); + return { adapter, processes }; +}); + +it.live.each([ + "sent after", + "queued behind", + "stopped after direct handoff", + "stopped after failed target start", +] as const)( + "preserves Claude background-work ownership and target selection for a switch %s", + (variant) => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("claude-account-switch-wake"); + const continuations = + yield* Queue.unbounded(); + const closed: Array = []; + const first = yield* makeAccount({ + instanceId: firstAccount.instanceId, + cwd, + continuations, + closed, + }); + const second = yield* makeAccount({ + instanceId: secondAccount.instanceId, + cwd, + continuations, + closed, + }); + const failedStart = variant === "stopped after failed target start"; + const stopping = variant.startsWith("stopped after"); + const adapters = [ + first.adapter, + failedStart + ? { + ...second.adapter, + openSession: (input: Parameters[0]) => + Effect.fail( + new ProviderAdapterOpenSessionError({ + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + providerSessionId: input.providerSessionId, + requiresSignIn: true, + cause: "Controlled target sign-in refusal before first provider turn", + }), + ), + } + : second.adapter, + ]; + // Two accounts of one Claude install resume each other's native history. + const registry = Layer.succeed( + ProviderAdapterRegistry.ProviderAdapterRegistryV2, + ProviderAdapterRegistry.ProviderAdapterRegistryV2.of({ + get: (instanceId) => { + const adapter = adapters.find((candidate) => candidate.instanceId === instanceId); + return adapter === undefined + ? Effect.fail( + new ProviderAdapterRegistry.ProviderAdapterRegistryLookupError({ instanceId }), + ) + : Effect.succeed(adapter); + }, + list: () => Effect.succeed(adapters.map((adapter) => adapter.instanceId)), + getMetadata: (instanceId) => + Effect.succeed({ + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + continuationKey: stopping + ? `claude:${instanceId}` + : "claude:shared-native-account-history", + enabled: true, + capabilities: ClaudeProviderCapabilitiesV2, + }), + }), + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make("thread:claude-account-switch-wake"); + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const runStatus = (messageId: string, status: string) => + watch( + (event) => + event.type === "run.updated" && + event.payload.userMessageId === messageId && + event.payload.status === status, + ); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId: ProjectId.make("project:claude-account-switch-wake"), + title: "Account switch during a background build", + modelSelection: firstAccount, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const queued = variant === "queued behind"; + const send = (ordinal: number, modelSelection: ModelSelection) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`send-${ordinal}`), + threadId, + messageId: MessageId.make(`send-${ordinal}`), + text: `Prompt ${ordinal}`, + attachments: [], + modelSelection, + dispatchMode: { type: queued ? "queue_after_active" : "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + + const firstRunning = yield* watch( + (event) => event.type === "provider-turn.updated" && event.payload.status === "running", + ); + yield* send(1, firstAccount); + yield* worker.drain(); + yield* Fiber.join(firstRunning); + if (queued) yield* send(2, secondAccount); + const firstSettling = yield* runStatus("send-1", "waiting"); + const firstProcess = first.processes[0]!; + yield* Queue.offer(firstProcess, backgroundBashStarted); + yield* Queue.offer(firstProcess, resultFrame("turn-1", "Started the build.")); + yield* Fiber.join(firstSettling); + yield* worker.drain(); + + if (!queued) yield* send(2, secondAccount); + yield* worker.drain(); + if (stopping) { + if (!failedStart) { + const secondSettling = yield* runStatus("send-2", "waiting"); + yield* Queue.offer(second.processes[0]!, resultFrame("turn-2", "Handoff complete.")); + yield* Fiber.join(secondSettling); + yield* worker.drain(); + } + const beforeStop = yield* orchestrator.getThreadProjection(threadId); + const source = beforeStop.providerThreads.find( + (thread) => thread.providerInstanceId === firstAccount.instanceId, + )!; + assert.deepEqual( + beforeStop.runs.map((run) => run.status), + ["completed", failedStart ? "failed" : "completed"], + ); + assert.equal(source.pendingBackgroundTasks?.[0]?.taskId, "bash-build"); + assert.isFalse( + beforeStop.turnItems.some( + (item) => item.type === "command_execution" && item.status === "running", + ), + "the background command is reported only by the roster", + ); + assert.deepEqual(closed, []); + const sourceSession = Option.getOrThrow( + yield* (yield* ProviderSessionManager.ProviderSessionManagerV2).get( + source.providerSessionId!, + ), + ); + assert.isTrue(yield* sourceSession.hasPendingBackgroundWorkForThread!(source)); + const target = beforeStop.runs[1]!; + if (failedStart) { + assert.isFalse( + beforeStop.providerTurns.some( + (turn) => turn.providerThreadId === target.providerThreadId, + ), + ); + const directInterrupt = yield* orchestrator + .dispatch({ + type: "run.interrupt", + commandId: CommandId.make("direct-target-stop"), + threadId, + runId: target.id, + }) + .pipe(Effect.exit); + assert.isTrue( + Exit.isFailure(directInterrupt), + "a failed target alone is not interruptible", + ); + } + yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("stop"), + threadId, + }); + yield* worker.drain(); + const stopped = yield* orchestrator.getThreadProjection(threadId); + assert.include( + closed, + firstAccount.instanceId, + "Stop must close the retained source process", + ); + assert.deepEqual( + stopped.providerThreads.find((thread) => thread.id === source.id) + ?.pendingBackgroundTasks, + [], + ); + assert.isFalse(yield* sourceSession.hasPendingBackgroundWorkForThread!(source)); + if (failedStart) { + assert.deepEqual(stopped.runs[1], target, "Stop preserves the target failure"); + assert.deepEqual(stopped.attempts, beforeStop.attempts); + assert.deepEqual(stopped.nodes, beforeStop.nodes); + assert.deepEqual(stopped.providerTurns, beforeStop.providerTurns); + assert.deepEqual( + stopped.turnItems.filter((item) => item.type === "error"), + beforeStop.turnItems.filter((item) => item.type === "error"), + ); + assert.isTrue( + stopped.turnItems.some( + (item) => item.runId === target.id && item.type === "run_interrupt_request", + ), + "Stop still guards late tool calls on the failed target", + ); + } + assert.equal( + stopped.thread.activeProviderThreadId, + beforeStop.thread.activeProviderThreadId, + ); + assert.deepEqual(stopped.thread.modelSelection, secondAccount); + assert.equal( + stopped.providerThreads.find( + (thread) => thread.id === stopped.thread.activeProviderThreadId, + )?.providerInstanceId, + secondAccount.instanceId, + ); + return; + } + const waiting = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + waiting.runs.map((run) => [run.providerInstanceId, run.status]), + [ + [firstAccount.instanceId, "completed"], + [secondAccount.instanceId, "queued"], + ], + ); + + // The settled roster gives queue resumption its chance before the notification. + const rosterSettled = yield* orchestrator + .streamStoredEventsFrom({ + threadId, + afterSequence: yield* (yield* EventSink.EventSinkV2).latestSequence(), + eventType: "provider-thread.updated", + }) + .pipe(Stream.runHead, Effect.forkScoped); + yield* Queue.offer(firstProcess, emptyRoster); + yield* Fiber.join(rosterSettled); + yield* worker.drain(); + yield* orchestrator.resumeQueuedRuns; + yield* worker.drain(); + const afterRoster = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual(closed, [], "the notification is still owed by the first account"); + assert.equal(afterRoster.runs[1]?.status, "queued"); + assert.equal( + afterRoster.providerSessions.find( + (session) => session.providerInstanceId === firstAccount.instanceId, + )?.status, + "ready", + ); + + // The notification and wake reply arrive on the still-open process. + yield* Queue.offer(firstProcess, bashNotification); + yield* Queue.offer(firstProcess, assistantFrame("wake-assistant", WAKE_REPLY)); + yield* Queue.offer(firstProcess, resultFrame("wake-result", WAKE_REPLY, true)); + const request = yield* Queue.take(continuations); + const wakeSettling = yield* runStatus("wake", "waiting"); + const switched = yield* runStatus("send-2", "running"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("wake"), + threadId, + messageId: MessageId.make("wake"), + text: request.detail ?? "Background task completed.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "provider", + continuationProviderThreadId: request.providerThreadId, + notification: request.notification ?? { + source: { kind: "background_task" }, + outcome: "updated", + summary: "Background activity updated", + }, + }); + yield* worker.drain(); + yield* Fiber.join(wakeSettling); + yield* worker.drain(); + yield* Fiber.join(switched); + yield* worker.drain(); + + const after = yield* orchestrator.getThreadProjection(threadId); + assert.isTrue( + after.messages.some((message) => message.text === WAKE_REPLY), + "the wake reply must reach the thread", + ); + const wakeRun = after.runs.find((run) => run.userMessageId === "wake"); + assert.equal(wakeRun?.providerInstanceId, firstAccount.instanceId); + assert.equal(wakeRun?.status, "completed"); + assert.equal( + after.runs.find((run) => run.userMessageId === "send-2")?.providerInstanceId, + secondAccount.instanceId, + ); + // Only now may the switch release the first account's process. + assert.deepEqual(closed, [firstAccount.instanceId]); + assert.lengthOf(second.processes, 1); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { + name: "claude-account-switch-wake", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + registry, + { runEffectWorker: false }, + ), + ), + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), +); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 53f726bf4628d..18f31dc000fee 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -82,6 +82,7 @@ import { import * as KeyedLock from "@t3tools/shared/KeyedLock"; import { modelSelectionsEqual } from "@t3tools/shared/model"; import { modelSwitchSince } from "./modelSwitch.ts"; +import { resolveBufferedProviderContinuation } from "./bufferedProviderContinuation.ts"; import { isPlaceholderThreadTitle } from "./placeholderThreadTitle.ts"; import { combinePullRequestWatchWakes, type PullRequestWatchWake } from "./pullRequestWatch.ts"; import { recordThreadOpen } from "@t3tools/shared/paletteSearch"; @@ -989,7 +990,20 @@ function worktreeSetupPending( function nextQueuedRun( projection: Pick, ): OrchestrationV2Run | undefined { - return queuedRunsInDeliveryOrder(projection).find((run) => run.queueHeld !== true); + const queued = queuedRunsInDeliveryOrder(projection).filter((run) => run.queueHeld !== true); + const next = queued[0]; + // Retained sessions stay busy until their buffered output is consumed. + const bufferedMessageIds = new Set( + projection.messages.filter(isBufferedProviderNotification).map((message) => message.id), + ); + if (next === undefined || bufferedMessageIds.has(next.userMessageId)) return next; + return ( + queued.find( + (run) => + run.providerInstanceId !== next.providerInstanceId && + bufferedMessageIds.has(run.userMessageId), + ) ?? next + ); } // Delegated completions, monitor notifications and unpinned scheduled deliveries never choose a @@ -1433,7 +1447,7 @@ function conversationProviderThread( ); } -/** Wakes and fresh starts act on the active session; a message continues the conversation. */ +/** Provider wakes and fresh starts act on the active session; other sends continue the conversation. */ function dispatchProviderThread( projection: Pick< OrchestrationV2ThreadProjection, @@ -1442,9 +1456,7 @@ function dispatchProviderThread( message: Pick, freshContext: boolean, ): OrchestrationV2ProviderThread | undefined { - return freshContext || - message.notification !== undefined || - message.delegatedCompletion !== undefined + return freshContext || message.notification?.source.kind === "background_task" ? projection.providerThreads.find( (candidate) => candidate.id === projection.thread.activeProviderThreadId, ) @@ -1737,6 +1749,25 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return false; }); + const busyProviderThreadIds = Effect.fn("Orchestrator.busyProviderThreadIds")(function* ( + providerThreads: OrchestrationV2ThreadProjection["providerThreads"], + ) { + const busy = new Set(); + for (const owner of providerThreads) { + if (owner.providerSessionId === null) continue; + const runtime = yield* providerSessions + .get(owner.providerSessionId) + .pipe(Effect.orElseSucceed(() => Option.none())); + if ( + Option.isSome(runtime) && + (yield* sessionHasBackgroundWork(runtime.value.providerSession, [owner])) + ) { + busy.add(owner.id); + } + } + return busy; + }); + // Runs in every command transaction; avoid reading thousands of finished leaf nodes. const readCommandProjection = ( threadId: ThreadId, @@ -2489,11 +2520,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } // Failed worktree creation would run queued work in the project checkout; wait for the user. const failedRun = latestExecutedRun(projection.runs); - const endedRunId = options?.latestRunEnded === true ? failedRun?.id : options?.endedRunId; + const latestRunEnded = options?.latestRunEnded === true && queuedRun.queueHeld !== false; + const endedRunId = latestRunEnded ? failedRun?.id : options?.endedRunId; const failedRunId = - options?.latestRunEnded === true && failedRun?.status === "failed" - ? failedRun.id - : options?.failedRunId; + latestRunEnded && failedRun?.status === "failed" ? failedRun.id : options?.failedRunId; const failureClass = failedRun && failedRunId !== undefined && failedRun.id === failedRunId ? latestRootProviderFailure(failedRun, projection.turnItems)?.class @@ -2611,13 +2641,24 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), ), ); + const bufferedContinuation = isBufferedProviderNotification(queuedMessage); const followsCurrent = followsCurrentConversation(queuedMessage); const freshContext = queuedRun.freshContext === true; - const activeProviderThread = dispatchProviderThread(projection, queuedMessage, freshContext); + const selectedProviderThread = dispatchProviderThread( + projection, + queuedMessage, + freshContext, + ); + const activeProviderThread = + selectedProviderThread !== undefined && + providerThreadPredatesBoundary(selectedProviderThread, contextBoundaryOrdinal(projection)) + ? rootProviderThreadsForProvider(projection, projection.thread.providerInstanceId)[0] + : selectedProviderThread; const contextBoundary = freshContext ? ordinal : contextBoundaryOrdinal(projection); // A run queued before a fresh run started is bound to the conversation that run left behind. const followsLatestContext = !freshContext && + !bufferedContinuation && (followsCurrent || providerThreadPredatesBoundary(storedProviderThread, contextBoundary)); const targetProviderThread = followsLatestContext ? activeProviderThread?.providerInstanceId === queuedRun.providerInstanceId && @@ -2654,13 +2695,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio createdAt: now, updatedAt: now, }; - const selectionChanged = !modelSelectionsEqual( - projection.thread.modelSelection, - queuedRun.modelSelection, - ); + const selectionChanged = + !bufferedContinuation && + !modelSelectionsEqual(projection.thread.modelSelection, queuedRun.modelSelection); // The picker may already have moved ahead of the native conversation. const switchFromActiveProvider = - followsCurrent && + !bufferedContinuation && activeProviderThread !== undefined && activeProviderThread.providerInstanceId !== queuedRun.providerInstanceId; const switchPlan = @@ -2715,22 +2755,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio !freshContext && switchPlan?.transition.type === "restart_and_resume" ) { - yield* Effect.logInfo( - "Holding the queue: the replaced provider session still runs background work", - { threadId, runId: queuedRun.id, providerSessionIds: Array.from(busySessionIds) }, - ); - yield* writeSystemEvents( - projection.runs - .filter((run) => run.status === "queued") - .map((run) => ({ - type: "run.updated" as const, - threadId, - runId: run.id, - providerInstanceId: run.providerInstanceId, - occurredAt: now, - payload: { ...run, queueHeld: true }, - })), - ); return; } const sessionsToDetach = releasableSessions.filter( @@ -7064,7 +7088,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ]); projection = applyToProjection(projection, event); } - let modelSelection = command.modelSelection ?? projection.thread.modelSelection; + const continuation = resolveBufferedProviderContinuation(projection, command); + const continuationProviderThread = continuation?.providerThread; + if (command.continuationProviderThreadId !== undefined && continuation === undefined) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "Buffered continuations require their owning provider thread.", + }); + } + let modelSelection = + continuation?.modelSelection ?? command.modelSelection ?? projection.thread.modelSelection; let dispatchMode = resolveMessageDispatchIntent( projection, command.dispatchMode, @@ -7415,11 +7449,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return; } - const activeProviderThread = dispatchProviderThread( + const selectedProviderThread = dispatchProviderThread( projection, { ...command, delegatedCompletion }, freshContext, ); + // A retained session's late roster can point back across a fresh-context boundary. + const activeProviderThread = + continuationProviderThread ?? + (selectedProviderThread !== undefined && + providerThreadPredatesBoundary(selectedProviderThread, contextBoundaryOrdinal(projection)) + ? rootProviderThreadsForProvider(projection, projection.thread.providerInstanceId)[0] + : selectedProviderThread); const activeRun = projection.runs.find(isActiveRun); // Messages after failed or stopped worktree setup must wait for a worktree, // rather than start in the project checkout. @@ -7432,7 +7473,79 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio projection.thread.worktreePath === null ? latestRun : undefined; - const queueBehindRun = activeRun ?? worktreeMissingRun; + const isProviderSwitch = + continuationProviderThread === undefined && + activeRun === undefined && + activeProviderThread !== undefined && + activeProviderThread.providerInstanceId !== modelSelection.instanceId; + // Account overlays share native history. Selection commands may already + // have updated the app thread, so classify against the native thread's owner. + const providerSwitchPlan = isProviderSwitch + ? yield* providerSwitchService + .plan({ + projection: { + ...projection, + thread: { + ...projection.thread, + modelSelection: { + ...projection.thread.modelSelection, + instanceId: activeProviderThread.providerInstanceId, + }, + }, + }, + targetModelSelection: modelSelection, + }) + .pipe(mapDispatchError(command)) + : null; + const canResumeAcrossInstances = + !freshContext && + isProviderSwitch && + activeProviderThread.nativeThreadRef !== null && + providerSwitchPlan?.transition.type === "restart_and_resume"; + const releasableSessions = uniqueSessions([ + ...projection.providerSessions.filter( + (session) => + providerSwitchPlan?.releaseProviderSessionIds.includes(session.id) && + session.status !== "stopped" && + session.status !== "error", + ), + ...(freshContext && activeRun === undefined + ? sessionsReleasedByFreshContext(projection, activeProviderThread) + : []), + ]); + const busySessionIds = new Set(); + for (const session of releasableSessions) { + if (yield* sessionHasBackgroundWork(session, projection.providerThreads)) { + busySessionIds.add(session.id); + } + } + const sessionsToDetach = releasableSessions.filter( + (session) => !busySessionIds.has(session.id), + ); + const waitsForBackgroundWork = canResumeAcrossInstances && busySessionIds.size > 0; + + const queuedSession = + activeRun === undefined && nextQueuedRun(projection) !== undefined + ? projection.providerSessions.find( + (session) => session.id === activeProviderThread?.providerSessionId, + ) + : undefined; + const bufferedProviderWake = + command.createdBy === "agent" && + command.creationSource === "provider" && + command.notification !== undefined; + // Buffered output is part of the busy session; only its owning turn can drain it. + const queueWaitingOnBackgroundWork = + !bufferedProviderWake && + queuedSession !== undefined && + (busySessionIds.has(queuedSession.id) || + (yield* sessionHasBackgroundWork(queuedSession, projection.providerThreads))); + const queueBehindRun = + activeRun ?? + worktreeMissingRun ?? + (waitsForBackgroundWork || queueWaitingOnBackgroundWork + ? (latestRun ?? undefined) + : undefined); const pendingMergeBackTransfers = pendingMergeBackTransfersForThread(projection); const startsOrQueuesRun = dispatchMode.type === "defer_start" || @@ -7853,8 +7966,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); const now = yield* DateTime.now; if ( - !modelSelectionsEqual(projection.thread.modelSelection, modelSelection) || - projection.thread.providerInstanceId !== modelSelection.instanceId + continuationProviderThread === undefined && + (!modelSelectionsEqual(projection.thread.modelSelection, modelSelection) || + projection.thread.providerInstanceId !== modelSelection.instanceId) ) { yield* emit( events, @@ -7894,57 +8008,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio projection.thread.historyOrigin === "v1_import" && contextBoundary === 0 ? yield* readHandoffItems(command.threadId, [null]) : []; - const isProviderSwitch = - activeProviderThread !== undefined && - activeProviderThread.providerInstanceId !== modelSelection.instanceId; - // Account overlays share native history. Selection commands may already - // have updated the app thread, so classify against the native thread's owner. - const providerSwitchPlan = isProviderSwitch - ? yield* providerSwitchService - .plan({ - projection: { - ...projection, - thread: { - ...projection.thread, - modelSelection: { - ...projection.thread.modelSelection, - instanceId: activeProviderThread.providerInstanceId, - }, - }, - }, - targetModelSelection: modelSelection, - }) - .pipe(mapDispatchError(command)) - : null; - const canResumeAcrossInstances = - !freshContext && - isProviderSwitch && - activeProviderThread.nativeThreadRef !== null && - providerSwitchPlan?.transition.type === "restart_and_resume"; - const releasableSessions = uniqueSessions([ - ...projection.providerSessions.filter( - (session) => - providerSwitchPlan?.releaseProviderSessionIds.includes(session.id) && - session.status !== "stopped" && - session.status !== "error", - ), - ...(freshContext ? sessionsReleasedByFreshContext(projection, activeProviderThread) : []), - ]); - const sessionsToDetach: Array = []; - for (const detaching of releasableSessions) { - if (yield* sessionHasBackgroundWork(detaching, projection.providerThreads)) { - // A schedule has nobody to wait or press Stop; like a queued fresh start, leave it running. - if (freshContext && command.scheduledTaskId !== undefined) continue; - return yield* new OrchestratorDispatchError({ - commandId: command.commandId, - commandType: command.type, - cause: freshContext - ? "The current session is still running background agents or commands, and a fresh session would end them. Wait for them to finish, or press Stop, then send the message again." - : "The previous provider is still running background agents or commands, and switching would end them. Wait for them to finish, or press Stop, then send the message again.", - }); - } - sessionsToDetach.push(detaching); - } if ( pendingForkTransfer === undefined && @@ -13106,15 +13169,21 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio run?.providerThreadId === null ? undefined : projection.providerThreads.find((candidate) => candidate.id === run?.providerThreadId); + const busyOwners = + command.holdQueue === true + ? yield* busyProviderThreadIds(projection.providerThreads) + : new Set(); const hasBackgroundWork = - run?.id === projection.runs.findLast((candidate) => candidate.status !== "queued")?.id && - derivePendingBackgroundWork({ - latestRun: run, - providerThreads: projection.providerThreads, - turnItems: projection.turnItems, - activeProviderThreadId: projection.thread.activeProviderThreadId, - runs: projection.runs, - }).length > 0; + busyOwners.size > 0 || + (run?.id === projection.runs.findLast((candidate) => candidate.status !== "queued")?.id && + derivePendingBackgroundWork({ + latestRun: run, + providerThreads: projection.providerThreads, + turnItems: projection.turnItems, + activeProviderThreadId: + command.holdQueue === true ? null : projection.thread.activeProviderThreadId, + runs: projection.runs, + }).length > 0); // A failed start has no provider turn. Background work still belongs // to the provider thread, so Stop reaches its latest accepted turn. const providerTurn = @@ -13216,6 +13285,23 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio latestTurnByProviderThread.set(turn.providerThreadId, { turn, runOrdinal }); } } + for (const owner of command.holdQueue === true ? projection.providerThreads : []) { + if ( + owner.id === providerThread.id || + ((owner.pendingBackgroundTasks?.length ?? 0) === 0 && !busyOwners.has(owner.id)) + ) { + continue; + } + const turn = projection.providerTurns.findLast( + (candidate) => candidate.providerThreadId === owner.id, + ); + if (turn === undefined) continue; + const runOrdinal = owner.lastRunOrdinal ?? -1; + const latest = latestTurnByProviderThread.get(owner.id); + if (latest === undefined || runOrdinal > latest.runOrdinal) { + latestTurnByProviderThread.set(owner.id, { turn, runOrdinal }); + } + } for (const { turn } of latestTurnByProviderThread.values()) { const owner = projection.providerThreads.find( (candidate) => candidate.id === turn.providerThreadId, @@ -13395,6 +13481,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } if (providerTurn === undefined) { + if (command.holdQueue === true && otherProviderInterrupts.length > 0) { + yield* emitEvent({ + type: "turn-item.updated", + threadId: command.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: interruptRequestItem, + }); + yield* stopRemainingWork(); + yield* Ref.update(effects, (existing) => [...existing, ...otherProviderInterrupts]); + return undefined; + } return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, @@ -13712,6 +13812,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? Effect.void : Ref.update(effects, (existing) => [...existing, delegatedTasksStopEffect(command)]); const latestRun = projection.runs.findLast((run) => run.status !== "queued"); + const busyOwners = yield* busyProviderThreadIds(projection.providerThreads); const target = projection.runs.findLast( (run) => @@ -13721,9 +13822,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio latestRun, providerThreads: projection.providerThreads, turnItems: projection.turnItems, - activeProviderThreadId: projection.thread.activeProviderThreadId, runs: projection.runs, - }).length > 0 + }).length > 0 || busyOwners.size > 0 ? latestRun : undefined); if (target !== undefined) { @@ -15610,6 +15710,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( command.type === "queue.resume" || command.type === "queued-run.start" || + (command.type === "message.dispatch" && + plan.events.some( + (event) => event.type === "run.created" && event.payload.status === "queued", + )) || (command.type === "thread.runtime-mode.set" && runtimeModeCeiling !== undefined) ) { yield* mapDispatchError(command)( @@ -15878,22 +15982,58 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio afterSequence: terminalEventsAfterSequence, eventType: "provider-thread.updated", }), - eventSink.stream({ afterSequence: terminalEventsAfterSequence, eventType: "subagent.updated" }), + Stream.merge( + eventSink.stream({ + afterSequence: terminalEventsAfterSequence, + eventType: "subagent.updated", + }), + // Codex MCP and dynamic tools that outlive their turn settle with only a node update. + eventSink + .stream({ afterSequence: terminalEventsAfterSequence, eventType: "node.updated" }) + .pipe( + Stream.filter( + ({ event }) => + event.type === "node.updated" && + event.payload.kind === "tool_call" && + !isOrchestrationV2WorkActive(event.payload.status), + ), + ), + ), ).pipe( - Stream.filter(settlesChildWork), - Stream.runForEach((stored) => + Stream.map((stored) => ({ stored, settlesChild: settlesChildWork(stored) })), + Stream.filter( + ({ stored, settlesChild }) => + settlesChild || + (stored.event.type === "provider-thread.updated" && + !backgroundWorkHoldsCompletion(stored.event.payload.pendingBackgroundTasks ?? [])) || + (stored.event.type === "node.updated" && + stored.event.payload.kind === "tool_call" && + !isOrchestrationV2WorkActive(stored.event.payload.status)), + ), + Stream.runForEach(({ stored, settlesChild }) => Effect.gen(function* () { const childThreadId = stored.event.threadId; + // Commands already promote their queues; only unsolicited settlement needs this retry. + if ( + stored.commandId === null && + (yield* projectionStore.canStartQueuedRun(childThreadId)) + ) { + yield* threadDispatch.withLock(childThreadId, startNextQueuedRun(childThreadId)); + } + if (!settlesChild) return; const parentThreadId = yield* appOwnedSubagentParentThreadId(childThreadId); if (parentThreadId === undefined) return; yield* threadDispatch.withLock(parentThreadId, finalizeAppOwnedSubagent(childThreadId)); }).pipe( Effect.catchCause((cause) => - Effect.logWarning("Failed to finalize a delegated child after its work settled", { - threadId: stored.event.threadId, - sequence: stored.sequence, - cause, - }), + Effect.logWarning( + "Failed to resume a queue or finalize a delegated child after its work settled", + { + threadId: stored.event.threadId, + sequence: stored.sequence, + cause, + }, + ), ), ), ), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 9a7e4fcb64490..23fda5deb3467 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -902,6 +902,23 @@ export function emptyProjection( }; } +// Retained sessions report after a handoff or fresh start without reclaiming the route. +function providerThreadUpdateSelectsRoute( + thread: OrchestrationV2ThreadProjection["thread"], + providerThread: OrchestrationV2ProviderThread, + activeProviderThread: OrchestrationV2ProviderThread | undefined, +): boolean { + return ( + providerThread.appThreadId === thread.id && + !isQueuedProviderThreadPlaceholder(providerThread) && + (thread.activeProviderThreadId === null || + providerThread.id === thread.activeProviderThreadId || + (providerThread.providerInstanceId === thread.providerInstanceId && + (activeProviderThread?.providerInstanceId !== providerThread.providerInstanceId || + (providerThread.firstRunOrdinal ?? -1) >= (activeProviderThread.firstRunOrdinal ?? -1)))) + ); +} + export function applyToProjection( projection: OrchestrationV2ThreadProjection, event: OrchestrationV2DomainEvent, @@ -1025,14 +1042,15 @@ export function applyToProjection( case "provider-thread.updated": return { ...base, - thread: - event.payload.appThreadId === base.thread.id && - !isQueuedProviderThreadPlaceholder(event.payload) - ? { - ...base.thread, - activeProviderThreadId: event.payload.id, - } - : base.thread, + thread: providerThreadUpdateSelectsRoute( + base.thread, + event.payload, + base.providerThreads.find( + (candidate) => candidate.id === base.thread.activeProviderThreadId, + ), + ) + ? { ...base.thread, activeProviderThreadId: event.payload.id } + : base.thread, providerThreads: upsertById(base.providerThreads, event.payload), }; case "provider-turn.updated": @@ -2723,8 +2741,27 @@ export const layer: Layer.Layer = LIMIT 1 `; const threadRow = threadRows[0]; - if (threadRow !== undefined) { - const thread = yield* decodeThreadPayload(threadRow.payload_json); + const thread = + threadRow === undefined + ? undefined + : yield* decodeThreadPayload(threadRow.payload_json); + const activeRows = + thread?.activeProviderThreadId == null || + thread.activeProviderThreadId === event.payload.id || + thread.providerInstanceId !== event.payload.providerInstanceId + ? [] + : yield* sql` + SELECT payload_json FROM orchestration_v2_projection_provider_threads + WHERE provider_thread_id = ${thread.activeProviderThreadId} LIMIT 1 + `; + const activeProviderThread = + activeRows[0] === undefined + ? undefined + : yield* decodeProviderThreadPayload(activeRows[0].payload_json); + if ( + thread !== undefined && + providerThreadUpdateSelectsRoute(thread, event.payload, activeProviderThread) + ) { const updatedThread = { ...thread, activeProviderThreadId: event.payload.id, diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.ts index 01f6e0acb989d..84a963fc005b7 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.ts @@ -258,6 +258,9 @@ export const layer = Layer.effectDiscard( createdBy: "agent", // Only buffered wakes use creationSource "provider"; Claude would drain an empty buffer instead of prompting message_text. creationSource: request.delivery === "message_text" ? "server" : "provider", + ...(request.delivery === "message_text" + ? {} + : { continuationProviderThreadId: request.providerThreadId }), }) .pipe(Effect.tap((dispatched) => releaseIfNeverStarted(request, messageId, dispatched))); // Catch after dispatchIfCurrent so adapters see raw failures first; clearIfCurrent is idempotent and covers others. diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 611d9a3244f24..5a2798872b9a4 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -70,6 +70,7 @@ import { DispatchModeLimit, exceededDispatchModeLimit } from "./DispatchModeLimi import { projectTurnItemForDetail } from "./WireProjection.ts"; import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts"; import * as ServerSettings from "../serverSettings.ts"; +import { resolveBufferedProviderContinuation } from "./bufferedProviderContinuation.ts"; import { ThreadWorktreeRestore } from "../workspace/restoreMissingWorktree.ts"; export type ThreadManagementSendMode = "auto" | "queue" | "steer" | "restart"; @@ -768,15 +769,24 @@ const make = Effect.gen(function* () { ), ); case "message.dispatch": - return orchestrator.getThreadRecords(command.threadId, []).pipe( - Effect.flatMap((projection) => - assertProviderInstanceEnabled({ - projectId: projection.thread.projectId, - commandId: command.commandId, - instanceId: (command.modelSelection ?? projection.thread.modelSelection).instanceId, - }), - ), - ); + return orchestrator + .getThreadRecords( + command.threadId, + command.continuationProviderThreadId === undefined ? [] : ["providerThreads", "runs"], + ) + .pipe( + Effect.flatMap((projection) => + assertProviderInstanceEnabled({ + projectId: projection.thread.projectId, + commandId: command.commandId, + instanceId: ( + resolveBufferedProviderContinuation(projection, command)?.modelSelection ?? + command.modelSelection ?? + projection.thread.modelSelection + ).instanceId, + }), + ), + ); case "delegated_task.request": return orchestrator.getThreadRecords(command.parentThreadId, []).pipe( Effect.flatMap((parent) => diff --git a/apps/server/src/orchestration-v2/bufferedProviderContinuation.ts b/apps/server/src/orchestration-v2/bufferedProviderContinuation.ts new file mode 100644 index 0000000000000..6bc3bc2b5dd90 --- /dev/null +++ b/apps/server/src/orchestration-v2/bufferedProviderContinuation.ts @@ -0,0 +1,28 @@ +import type { + OrchestrationV2ServerCommand, + OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; + +export function resolveBufferedProviderContinuation( + projection: Pick, + command: Extract, +) { + if ( + command.continuationProviderThreadId === undefined || + command.createdBy !== "agent" || + command.creationSource !== "provider" || + command.notification === undefined || + command.freshContext === true + ) { + return undefined; + } + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === command.continuationProviderThreadId, + ); + const modelSelection = projection.runs.findLast( + (run) => run.providerThreadId === providerThread?.id, + )?.modelSelection; + return providerThread === undefined || modelSelection === undefined + ? undefined + : { providerThread, modelSelection }; +} diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index d9c431925a6bb..50477619b591d 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -354,7 +354,7 @@ const layerTest = Layer.mergeAll( Layer.provide(SqlitePersistence.layerMemory), Layer.provide(layerCheckpointStoreTest), Layer.provide(layerServerConfig), - Layer.provide(ServerSettings.layerTest()), + Layer.provideMerge(ServerSettings.layerTest()), Layer.provide(layerTestProviderInstanceRegistry), Layer.provide(layerGitWorkflowTest), Layer.provide(layerProjectServiceTest), @@ -8756,67 +8756,160 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); - it.effect("keeps a buffered provider wake on its own provider behind a queued pick", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const threadId = ThreadId.make("runtime-layer-buffered-wake-behind-queued-pick"); - yield* orchestrator.dispatch({ - type: "thread.create", - commandId: CommandId.make(`${threadId}:create`), - threadId, - projectId: ProjectId.make(`${threadId}:project`), - createdBy: "user", - creationSource: "web", - title: "Buffered wake behind a pick", - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: process.cwd(), - }); - for (const [key, selection] of [ - ["active", modelSelection], - ["picked", { ...modelSelection, instanceId: alternateInstanceId }], - ] as const) { + it.effect.each(["neither", "owner", "selected"] as const)( + "keeps a buffered provider wake on its own provider behind a queued pick, disabled: %s", + (disabled) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const threads = yield* ThreadManagementService.ThreadManagementService; + const settings = yield* ServerSettings.ServerSettingsService; + const threadId = ThreadId.make( + `runtime-layer-buffered-wake-behind-queued-pick-${disabled}`, + ); + const projectId = ProjectId.make(`${threadId}:project`); yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make(`${threadId}:${key}`), + type: "thread.create", + commandId: CommandId.make(`${threadId}:create`), threadId, - messageId: MessageId.make(`${threadId}:${key}`), - text: key, - attachments: [], - modelSelection: selection, - dispatchMode: { type: key === "active" ? "start_immediately" : "queue_after_active" }, + projectId, createdBy: "user", creationSource: "web", + title: "Buffered wake behind a pick", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), }); - } - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make(`${threadId}:wake`), - threadId, - messageId: MessageId.make(`${threadId}:wake`), - text: "Background activity updated", - attachments: [], - dispatchMode: { type: "queue_after_active" }, - createdBy: "agent", - creationSource: "provider", - notification: { - source: { kind: "background_task" }, - outcome: "updated", - summary: "Background activity updated", - }, - }); - const projection = yield* orchestrator.getThreadProjection(threadId); - const active = projection.runs.find( - (run) => run.userMessageId === MessageId.make(`${threadId}:active`), - ); - const wake = projection.runs.find( - (run) => run.userMessageId === MessageId.make(`${threadId}:wake`), - ); - assert.deepEqual(wake?.modelSelection, modelSelection); - assert.equal(wake?.providerThreadId, active?.providerThreadId); - }), + for (const [key, selection] of [ + ["active", modelSelection], + ["picked", { ...modelSelection, instanceId: alternateInstanceId }], + ] as const) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${threadId}:${key}`), + threadId, + messageId: MessageId.make(`${threadId}:${key}`), + text: key, + attachments: [], + modelSelection: selection, + dispatchMode: { type: key === "active" ? "start_immediately" : "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + } + const before = yield* orchestrator.getThreadProjection(threadId); + const source = before.runs.find((run) => run.userMessageId === `${threadId}:active`)!; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + if (disabled !== "neither") { + yield* eventSink.write({ + events: [ + { + id: EventId.make(`${threadId}:source-completed`), + type: "run.updated", + threadId, + runId: source.id, + occurredAt: now, + payload: { ...source, status: "completed", completedAt: now }, + }, + ], + }); + yield* orchestrator.resumeQueuedRuns; + const switched = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + switched.runs.find((run) => run.userMessageId === `${threadId}:picked`)?.status, + "starting", + ); + assert.equal(switched.thread.modelSelection.instanceId, alternateInstanceId); + } + yield* settings.updateSettings({ + projectSettingsOverrides: { + [projectId]: { + providerInstances: { + [modelSelection.instanceId]: { enabled: disabled !== "owner" }, + [alternateInstanceId]: { enabled: disabled !== "selected" }, + }, + }, + }, + }); + const dispatched = yield* Effect.exit( + threads.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${threadId}:wake`), + threadId, + messageId: MessageId.make(`${threadId}:wake`), + text: "Background activity updated", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "provider", + ...(disabled === "neither" + ? {} + : { continuationProviderThreadId: source.providerThreadId! }), + notification: { + source: { kind: "background_task" }, + outcome: "updated", + summary: "Background activity updated", + }, + }), + ); + const projection = yield* orchestrator.getThreadProjection(threadId); + if (disabled === "owner") { + assert.equal(dispatched._tag, "Failure"); + if (dispatched._tag === "Failure") { + const reason = dispatched.cause.reasons.find((reason) => reason._tag === "Fail"); + assert.instanceOf( + reason?.error, + ThreadManagementService.ThreadManagementProviderInstanceDisabledError, + ); + assert.deepInclude(reason?.error, { projectId, instanceId: modelSelection.instanceId }); + } + assert.equal(projection.runs.length, before.runs.length); + return; + } + assert.equal(dispatched._tag, "Success"); + const active = projection.runs.find( + (run) => run.userMessageId === MessageId.make(`${threadId}:active`), + ); + const wake = projection.runs.find( + (run) => run.userMessageId === MessageId.make(`${threadId}:wake`), + ); + assert.deepEqual(wake?.modelSelection, modelSelection); + assert.equal(wake?.providerThreadId, active?.providerThreadId); + if (disabled === "neither") { + assert.deepEqual(projection.thread.modelSelection, modelSelection); + return; + } + assert.deepEqual(projection.thread.modelSelection, { + ...modelSelection, + instanceId: alternateInstanceId, + }); + const picked = projection.runs.find((run) => run.userMessageId === `${threadId}:picked`)!; + yield* eventSink.write({ + events: [ + { + id: EventId.make(`${threadId}:picked-completed`), + type: "run.updated", + threadId, + runId: picked.id, + occurredAt: now, + payload: { ...picked, status: "completed", completedAt: now }, + }, + ], + }); + yield* orchestrator.resumeQueuedRuns; + const resumed = yield* orchestrator.getThreadProjection(threadId); + const delivered = resumed.runs.find((run) => run.id === wake?.id)!; + assert.equal(delivered.status, "starting"); + assert.equal(delivered.providerThreadId, source.providerThreadId); + assert.equal( + resumed.attempts.find((attempt) => attempt.id === delivered.activeAttemptId) + ?.providerInstanceId, + modelSelection.instanceId, + ); + assert.equal(resumed.thread.modelSelection.instanceId, alternateInstanceId); + }), ); it.effect.each(["", "pause", "resume", "clear", "edit"])( @@ -10068,6 +10161,85 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("starts a background-work wake beside a queue held after a failed run", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const events = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make("queue-hold-background-wake"); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:create`), + threadId, + projectId: ProjectId.make(`${threadId}:project`), + title: "Background wake beside a held queue", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), + }); + const send = (index: number, text: string) => + orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:message:${index}`), + threadId, + messageId: MessageId.make(`${threadId}:message:${index}`), + text, + attachments: [], + modelSelection, + dispatchMode: { type: index === 0 ? "start_immediately" : "queue_after_active" }, + }); + yield* send(0, "Active"); + yield* send(1, "Queued"); + const before = yield* orchestrator.getThreadProjection(threadId); + const [failed, held] = [before.runs[0]!, before.runs[1]!]; + const now = yield* DateTime.now; + yield* events.write({ + events: [ + { + id: EventId.make(`${threadId}:failed`), + type: "run.updated", + threadId, + runId: failed.id, + occurredAt: now, + payload: { ...failed, status: "failed", startedAt: now, completedAt: now }, + }, + { + id: EventId.make(`${threadId}:held`), + type: "run.updated", + threadId, + runId: held.id, + occurredAt: now, + payload: { ...held, queueHeld: true }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "provider", + commandId: CommandId.make(`${threadId}:wake`), + threadId, + messageId: MessageId.make(`${threadId}:wake`), + text: "Background task completed.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + notification: { + source: { kind: "background_task" }, + outcome: "updated", + summary: "Background task completed.", + }, + }); + const after = yield* orchestrator.getThreadProjection(threadId); + assert.equal(after.runs[2]?.status, "starting"); + assert.isTrue(after.runs.find((run) => run.id === held.id)?.queueHeld); + }), + ); + it.effect("edits and removes queued runs", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index f07d942cbac2f..1cd86cb939a41 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -6,6 +6,7 @@ import { type ChatAttachment, EventId, MessageId, + NodeId, type ModelSelection, type OrchestrationV2Command, type OrchestrationV2DomainEvent, @@ -66,6 +67,7 @@ import { } from "./fixtures/shared.ts"; import * as ProviderReplayHarness from "./ProviderReplayHarness.ts"; import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; +import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); @@ -130,13 +132,18 @@ function makeTestAdapter(input: { }; readonly backgroundTasksEndWithProcess?: true; readonly backgroundWork?: Ref.Ref; + readonly bufferedWake?: Ref.Ref; + readonly restartForSelectionChange?: true; readonly closedSessions?: Ref.Ref>; }): ProviderAdapter.ProviderAdapterV2["Service"] { return { instanceId: input.instanceId, driver: input.driver, getCapabilities: () => Effect.succeed(input.capabilities), - planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + planSelectionTransition: () => + Effect.succeed({ + type: input.restartForSelectionChange ? "restart_session" : "apply_on_next_turn", + }), openSession: (sessionInput) => Effect.gen(function* () { const events = yield* PubSub.unbounded(); @@ -188,7 +195,12 @@ function makeTestAdapter(input: { : { // Like Claude: native subagents are visible only to the session-wide // probe, never to the per-thread roster probe. - hasPendingBackgroundWork: Ref.get(input.backgroundWork), + hasPendingBackgroundWork: Effect.all([ + Ref.get(input.backgroundWork), + input.bufferedWake === undefined + ? Effect.succeed(false) + : Ref.get(input.bufferedWake), + ]).pipe(Effect.map(([background, buffered]) => background || buffered)), hasPendingBackgroundWorkForThread: () => Effect.succeed(false), }), ensureThread: (threadInput) => @@ -258,6 +270,11 @@ function makeTestAdapter(input: { compactThread: (turnInput) => runtime.startTurn(turnInput), startTurn: (turnInput) => Effect.gen(function* () { + if ( + input.bufferedWake !== undefined && + turnInput.message.creationSource === "provider" + ) + yield* Ref.set(input.bufferedWake, false); if ( input.failStartOnce !== undefined && (yield* Ref.getAndSet(input.failStartOnce, false)) @@ -444,7 +461,10 @@ function makeTestAdapter(input: { } }), steerTurn: () => Effect.void, - interruptTurn: () => Effect.void, + interruptTurn: () => + input.pendingBackgroundWork === undefined + ? Effect.void + : Ref.set(input.pendingBackgroundWork, false), respondToRuntimeRequest: () => Effect.void, readThreadSnapshot: () => unimplemented(input.driver, "readThreadSnapshot unused in provider switch test"), @@ -2795,31 +2815,58 @@ describe("orchestration v2 provider switching", () => { it.live.each([ [ "detaches the previous instance's session when a send hands off to another instance", - { background: false, queued: false, sharedHistory: false }, + { background: false, queued: false, sharedHistory: false, buffered: false }, + ], + [ + "keeps a busy previous session running when a direct send hands off", + { background: true, queued: false, sharedHistory: false, buffered: false }, + ], + [ + "resumes a direct account switch when a background tool finishes", + { background: true, queued: false, sharedHistory: true, buffered: false }, ], [ - "refuses the hand-off while the previous instance still runs background work", - { background: true, queued: false, sharedHistory: false }, + "resumes an explicit account switch after a failed run's background work settles", + { background: true, queued: true, sharedHistory: true, buffered: false, failed: true }, + ], + [ + "keeps explicit resume when a buffered wake follows failure", + { background: true, queued: true, sharedHistory: true, buffered: true, failed: true }, ], [ "keeps a busy previous session running when a queued hand-off starts", - { background: true, queued: true, sharedHistory: false }, + { background: true, queued: true, sharedHistory: false, buffered: false }, + ], + [ + "waits for background work before resuming a queued account switch", + { background: true, queued: true, sharedHistory: true, buffered: false }, ], [ - "holds a queued account switch while the replaced session still runs background work", - { background: true, queued: true, sharedHistory: true }, + "drains buffered provider output before resuming a waiting account switch", + { background: true, queued: false, sharedHistory: true, buffered: true }, ], - ] as const)("%s", ([, { background, queued, sharedHistory }]) => + [ + "drains buffered output queued before the predecessor settles", + { background: true, queued: true, sharedHistory: true, buffered: true }, + ], + ] as const)("%s", ([, options]) => Effect.gen(function* () { + const { background, queued, sharedHistory, buffered } = options; + const failed = "failed" in options && options.failed; + const omitThird = failed && buffered; const cwd = yield* checkpointWorkspace("handoff-releases-session"); const capturedTurns = yield* Ref.make>([]); const backgroundWork = yield* Ref.make(background); + const bufferedWake = yield* Ref.make(false); const closedSessions = yield* Ref.make>([]); const firstStarted = yield* Deferred.make(); const releaseFirst = yield* Deferred.make(); const secondSelection: ModelSelection = { ...CLAUDE_MODEL_SELECTION, - instanceId: ProviderInstanceId.make("claude-second-account"), + instanceId: failed + ? CLAUDE_MODEL_SELECTION.instanceId + : ProviderInstanceId.make("claude-second-account"), + ...(failed ? { model: "claude-opus-4-6" } : {}), }; const adapters = [ makeTestAdapter({ @@ -2830,6 +2877,14 @@ describe("orchestration v2 provider switching", () => { responseByRunOrdinal: { 1: "First account complete" }, capturedTurns, backgroundWork, + bufferedWake, + ...(failed + ? { + restartForSelectionChange: true, + failure: makeProviderFailure({ class: "provider_error" }), + failedRunOrdinals: new Set([1]), + } + : {}), closedSessions, ...(queued ? { holdFirstTurn: firstStarted, releaseFirstTurn: releaseFirst } : {}), }), @@ -2872,19 +2927,36 @@ describe("orchestration v2 provider switching", () => { yield* Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - const send = (ordinal: number, modelSelection: ModelSelection, attempt = "") => + const send = (ordinal: number, modelSelection: ModelSelection) => orchestrator.dispatch({ type: "message.dispatch", createdBy: "user", creationSource: "web", - commandId: CommandId.make(`command:handoff-releases-session:${ordinal}${attempt}`), + commandId: CommandId.make(`command:handoff-releases-session:${ordinal}`), threadId: handoffThreadId, - messageId: MessageId.make(`message:handoff-releases-session:${ordinal}${attempt}`), + messageId: MessageId.make(`message:handoff-releases-session:${ordinal}`), text: `Prompt ${ordinal}`, attachments: [], modelSelection, dispatchMode: { type: queued ? "queue_after_active" : "start_immediately" }, }); + const dispatchBufferedWake = () => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:handoff-buffered-wake"), + threadId: handoffThreadId, + messageId: MessageId.make("message:handoff-buffered-wake"), + text: "Buffered output", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "provider", + notification: { + source: { kind: "background_task" }, + outcome: "updated", + summary: "Background work completed", + }, + }); yield* orchestrator.dispatch({ type: "thread.create", createdBy: "user", @@ -2903,91 +2975,205 @@ describe("orchestration v2 provider switching", () => { if (queued) { yield* Deferred.await(firstStarted); - yield* send(2, secondSelection); - const queuedRunId = (yield* orchestrator.getThreadProjection(handoffThreadId)).runs[1] - ?.id; - assert.isDefined(queuedRunId); + } else { + yield* worker.drain(); + yield* waitForIdle(handoffThreadId); + } + const afterFirst = yield* orchestrator.getThreadProjection(handoffThreadId); + const firstSession = afterFirst.providerSessions.find( + (session) => session.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, + )!; + yield* send(2, secondSelection); + if (queued && buffered && !failed) { + yield* send(3, secondSelection); + yield* Ref.set(backgroundWork, false); + yield* Ref.set(bufferedWake, true); + yield* dispatchBufferedWake(); + assert.lengthOf(yield* Ref.get(capturedTurns), 1); yield* Deferred.succeed(releaseFirst, undefined); yield* orchestrator.streamStoredEvents.pipe( Stream.filter( ({ event }) => event.type === "run.updated" && - event.runId === queuedRunId && - (event.payload.queueHeld === true || event.payload.status === "completed"), + event.payload.userMessageId === "message:handoff-releases-session:3" && + event.payload.status === "completed", ), Stream.runHead, ); yield* worker.drain(); - const afterQueue = yield* orchestrator.getThreadProjection(handoffThreadId); - const firstSession = afterQueue.providerSessions.find( - (session) => session.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, + const turns = yield* Ref.get(capturedTurns); + assert.deepEqual( + turns.map((turn) => turn.text), + ["Prompt 1", "Buffered output", "Prompt 2", "Prompt 3"], ); - // Detaching would end the background agents, so the busy session must stay up. + assert.equal(turns[1]?.nativeThreadId, turns[0]?.nativeThreadId); + assert.equal(turns[2]?.nativeThreadId, turns[0]?.nativeThreadId); + assert.deepEqual(yield* Ref.get(closedSessions), [firstSession.id]); + return; + } + if (queued) { + yield* Deferred.succeed(releaseFirst, undefined); + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.ordinal === 1 && + event.payload.status === (failed ? "failed" : "completed"), + ), + Stream.runHead, + ); + } + yield* worker.drain(); + if (failed) { + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.status === "queued" && + event.payload.queueHeld === true, + ), + Stream.runHead, + ); + yield* orchestrator.dispatch({ + type: "queue.resume", + commandId: CommandId.make("explicit-resume"), + threadId: handoffThreadId, + }); + } + const waiting = yield* orchestrator.getThreadProjection(handoffThreadId); + if (background) { assert.deepEqual(yield* Ref.get(closedSessions), []); assert.equal(firstSession?.status, "ready"); - if (!sharedHistory) { - assert.deepEqual( - afterQueue.runs.map((run) => [run.providerInstanceId, run.status]), - [ - [CLAUDE_MODEL_SELECTION.instanceId, "completed"], - [secondSelection.instanceId, "completed"], - ], - ); - return; - } - // Resuming the native thread beside the busy session would fork its history. + } + if (sharedHistory) { + // Native history cannot resume beside its busy owner, but needs no manual queue hold. assert.deepEqual( - afterQueue.runs.map((run) => [run.status, run.queueHeld === true]), + waiting.runs.map((run) => [run.status, run.queueHeld === true]), [ - ["completed", false], - ["queued", true], + [failed ? "failed" : "completed", false], + ["queued", false], ], ); assert.lengthOf(yield* Ref.get(capturedTurns), 1); + if (!omitThird) yield* send(3, secondSelection); + const behind = yield* orchestrator.getThreadProjection(handoffThreadId); + assert.deepEqual( + behind.runs.map((run) => run.status), + [failed ? "failed" : "completed", "queued", ...(omitThird ? [] : ["queued"])], + ); + yield* Ref.set(bufferedWake, buffered); yield* Ref.set(backgroundWork, false); - yield* orchestrator.dispatch({ - type: "queue.resume", - commandId: CommandId.make("command:handoff-releases-session:resume"), + if (buffered) { + yield* dispatchBufferedWake(); + yield* worker.drain(); + assert.isFalse( + (yield* orchestrator.getThreadProjection(handoffThreadId)).runs.find( + (run) => run.userMessageId === "message:handoff-releases-session:2", + )?.queueHeld === true, + ); + assert.isFalse( + yield* Ref.get(bufferedWake), + "The owning session must drain buffered output before the account switch", + ); + } + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const settledEvent = { + id: EventId.make("event:handoff-releases-session:background-ended"), threadId: handoffThreadId, + providerInstanceId: CLAUDE_MODEL_SELECTION.instanceId, + occurredAt: now, + }; + yield* eventSink.write({ + events: [ + queued + ? { + ...settledEvent, + type: "provider-thread.updated", + payload: { ...afterFirst.providerThreads[0]!, pendingBackgroundTasks: [] }, + } + : !buffered + ? { + // Like a Codex MCP tool finishing after its turn: no roster or subagent update. + ...settledEvent, + type: "node.updated", + payload: { + id: NodeId.make("node:handoff-background-tool"), + threadId: handoffThreadId, + runId: afterFirst.runs[0]!.id, + parentNodeId: afterFirst.runs[0]!.rootNodeId!, + rootNodeId: afterFirst.runs[0]!.rootNodeId!, + kind: "tool_call", + status: "completed", + countsForRun: true, + providerThreadId: afterFirst.providerThreads[0]!.id, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: now, + }, + } + : { + ...settledEvent, + type: "subagent.updated", + payload: { + id: NodeId.make("subagent:handoff-background"), + threadId: handoffThreadId, + runId: afterFirst.runs[0]!.id, + parentNodeId: afterFirst.runs[0]!.rootNodeId!, + origin: "provider_native", + createdBy: "agent", + driver: CLAUDE_DRIVER, + providerInstanceId: CLAUDE_MODEL_SELECTION.instanceId, + providerThreadId: afterFirst.providerThreads[0]!.id, + childThreadId: null, + nativeTaskRef: null, + prompt: "Background work", + title: null, + model: null, + status: "completed", + result: "Done", + startedAt: now, + completedAt: now, + updatedAt: now, + }, + }, + ], }); + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.userMessageId === + `message:handoff-releases-session:${omitThird ? 2 : 3}` && + event.payload.status === "completed", + ), + Stream.runHead, + ); yield* worker.drain(); - const resumed = yield* waitForIdle(handoffThreadId); + const resumed = yield* orchestrator.getThreadProjection(handoffThreadId); assert.deepEqual( - resumed.runs.map((run) => [run.providerInstanceId, run.status]), + resumed.runs + .filter((run) => run.userMessageId !== "message:handoff-buffered-wake") + .map((run) => [run.providerInstanceId, run.status]), [ - [CLAUDE_MODEL_SELECTION.instanceId, "completed"], + [CLAUDE_MODEL_SELECTION.instanceId, failed ? "failed" : "completed"], [secondSelection.instanceId, "completed"], + ...(omitThird ? [] : [[secondSelection.instanceId, "completed"]]), ], ); + const turns = yield* Ref.get(capturedTurns); + const sentTurns = turns.filter((turn) => turn.text !== "Buffered output"); + assert.equal(sentTurns[1]?.nativeThreadId, sentTurns[0]?.nativeThreadId); + assert.equal(sentTurns[1]?.text, "Prompt 2"); + if (!omitThird) assert.equal(sentTurns[2]?.text, "Prompt 3"); + if (buffered) assert.equal(turns[1]?.nativeThreadId, turns[0]?.nativeThreadId); assert.deepEqual(yield* Ref.get(closedSessions), [firstSession!.id]); return; } - yield* worker.drain(); - const afterFirst = yield* waitForIdle(handoffThreadId); - const firstSession = afterFirst.providerSessions.find( - (session) => session.providerInstanceId === CLAUDE_MODEL_SELECTION.instanceId, - ); - assert.equal(firstSession?.status, "ready"); - - if (background) { - // Native subagents live only in the adapter's roster, so the projection's - // pendingBackgroundTasks stays empty while the probe still reports work. - assert.deepEqual(afterFirst.providerThreads[0]?.pendingBackgroundTasks ?? [], []); - const refused = yield* send(2, secondSelection, ":refused").pipe(Effect.flip); - assert.include(String(refused.cause), "still running background agents or commands"); - const unchanged = yield* orchestrator.getThreadProjection(handoffThreadId); - assert.lengthOf(unchanged.runs, 1); - assert.deepEqual(yield* Ref.get(closedSessions), []); - assert.equal(unchanged.thread.providerInstanceId, CLAUDE_MODEL_SELECTION.instanceId); - assert.deepEqual( - unchanged.providerSessions.map((session) => [session.id, session.status]), - [[firstSession!.id, "ready"]], - ); - yield* Ref.set(backgroundWork, false); - } - - yield* send(2, secondSelection); yield* worker.drain(); const afterSecond = yield* waitForIdle(handoffThreadId); yield* worker.drain(); @@ -2996,17 +3182,18 @@ describe("orchestration v2 provider switching", () => { assert.deepEqual( projection.runs.map((run) => [run.providerInstanceId, run.status]), [ - [CLAUDE_MODEL_SELECTION.instanceId, "completed"], + [CLAUDE_MODEL_SELECTION.instanceId, failed ? "failed" : "completed"], [secondSelection.instanceId, "completed"], ], ); assert.lengthOf(afterSecond.contextHandoffs, 1); assert.deepEqual( projection.providerSessions.map((session) => session.providerInstanceId), - [secondSelection.instanceId], + background + ? [CLAUDE_MODEL_SELECTION.instanceId, secondSelection.instanceId] + : [secondSelection.instanceId], ); - // The detached event alone hides the session; the runtime must also close. - assert.deepEqual(yield* Ref.get(closedSessions), [firstSession!.id]); + assert.deepEqual(yield* Ref.get(closedSessions), background ? [] : [firstSession!.id]); }).pipe( Effect.provide( ProviderReplayHarness.layerWithRegistry( @@ -3029,7 +3216,6 @@ describe("orchestration v2 provider switching", () => { }), ); - // regression: #16000 thread setting changes must not end a Claude session's background work it.live.each([ ["runtime mode", "runtime-mode"], ["worktree", "worktree"], @@ -3229,7 +3415,13 @@ describe("orchestration v2 provider switching", () => { codexSession!.id, ); - yield* Ref.set(codexBackgroundWork, false); + yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("command:retained-session-background:stop"), + threadId, + }); + yield* worker.drain(); + assert.isFalse(yield* Ref.get(codexBackgroundWork)); yield* selectGrok("idle"); yield* worker.drain(); const switched = yield* orchestrator.getThreadProjection(threadId); @@ -3288,17 +3480,21 @@ describe("orchestration v2 provider switching", () => { }), ); - it.live.each(["immediately", "behind a running turn"] as const)( + it.live.each([ + "immediately", + "immediately with background work", + "behind a running turn", + ] as const)( "starts a fresh provider session %s and never hands earlier runs to a later provider", (variant) => Effect.gen(function* () { - const queued = variant !== "immediately"; - const key = queued ? "queued" : "immediate"; + const queued = variant === "behind a running turn"; + const key = variant.replaceAll(" ", "-"); const cwd = yield* checkpointWorkspace(`fresh-context-${key}`); const capturedTurns = yield* Ref.make>([]); const closedSessions = yield* Ref.make>([]); - // A queued start has nobody to refuse, so a busy session must stay up. - const backgroundWork = yield* Ref.make(queued); + const busy = variant !== "immediately"; + const backgroundWork = yield* Ref.make(busy); const started = yield* Deferred.make(); const release = yield* Deferred.make(); const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ @@ -3405,8 +3601,7 @@ describe("orchestration v2 provider switching", () => { } yield* worker.drain(); yield* waitForIdle(freshThreadId); - // Claude hosts one native thread per session, so an idle old one is closed. - assert.deepEqual(yield* Ref.get(closedSessions), queued ? [] : [firstSession.id]); + assert.deepEqual(yield* Ref.get(closedSessions), busy ? [] : [firstSession.id]); yield* Ref.set(backgroundWork, false); yield* send("delta", CODEX_MODEL_SELECTION); yield* worker.drain(); @@ -3556,6 +3751,187 @@ describe("orchestration v2 provider switching", () => { }), ); + it.live.each(["handoff", "fresh", "queued fresh", "queued fresh before handoff"] as const)( + "delivers a retained session's buffered continuation on its owner after %s", + (variant) => + Effect.gen(function* () { + const laterHandoff = variant === "queued fresh before handoff"; + const queuedFresh = variant === "queued fresh" || laterHandoff; + const cwd = yield* checkpointWorkspace(`continuation-owner-${variant}`); + const capturedTurns = yield* Ref.make>([]); + const closedSessions = yield* Ref.make>([]); + const started = yield* Deferred.make(); + const release = yield* Deferred.make(); + const source = CLAUDE_MODEL_SELECTION; + const target = + variant === "handoff" ? CODEX_MODEL_SELECTION : { ...source, model: "claude-opus-4-6" }; + const registry = ProviderAdapterRegistry.layerFromAdapters([ + makeTestAdapter({ + instanceId: source.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: source, + responseByRunOrdinal: {}, + capturedTurns, + closedSessions, + backgroundWork: yield* Ref.make(true), + nativeThreadGeneration: yield* Ref.make(0), + ...(queuedFresh + ? { holdFirstTurn: started, releaseFirstTurn: release, holdRunOrdinal: 2 } + : {}), + }), + makeTestAdapter({ + instanceId: CODEX_MODEL_SELECTION.instanceId, + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + }), + ]); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + const threadId = ThreadId.make("thread:continuation-owner"); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId: ProjectId.make("project:continuation-owner"), + title: "Continuation owner", + modelSelection: source, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const send = (name: string, modelSelection: ModelSelection, freshContext = false) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(name), + messageId: MessageId.make(name), + threadId, + text: name, + attachments: [], + modelSelection, + freshContext, + dispatchMode: { type: "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + yield* send("source", source); + yield* worker.drain(); + const before = yield* waitForIdle(threadId); + const owner = before.providerThreads.find( + (row) => row.id === before.thread.activeProviderThreadId, + )!; + yield* send("target", target, variant !== "handoff"); + if (queuedFresh) yield* Deferred.await(started); + else { + yield* worker.drain(); + yield* waitForIdle(threadId); + } + if (laterHandoff) yield* send("later", CODEX_MODEL_SELECTION); + const beforeWake = yield* orchestrator.getThreadProjection(threadId); + const selectedRoute = beforeWake.thread.activeProviderThreadId; + const wakeCreated = yield* orchestrator + .streamStoredEventsFrom({ + threadId, + afterSequence: yield* (yield* EventSink.EventSinkV2).latestSequence(), + eventType: "run.created", + }) + .pipe( + Stream.filter( + ({ event }) => + event.type === "run.created" && event.payload.userMessageId === "wake", + ), + Stream.runHead, + Effect.forkScoped, + ); + yield* requests.offer({ + threadId, + providerThreadId: owner.id, + driver: CLAUDE_DRIVER, + messageId: MessageId.make("wake"), + detail: "Owned buffered output", + }); + yield* Fiber.join(wakeCreated); + if (queuedFresh) { + const queued = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + queued.runs.find((run) => run.userMessageId === "wake")?.providerThreadId, + owner.id, + ); + yield* Deferred.succeed(release, undefined); + } + yield* worker.drain(); + const after = yield* waitForIdle(threadId); + const turns = yield* Ref.get(capturedTurns); + const wake = turns.find((turn) => turn.text === "Owned buffered output")!; + assert.isDefined(wake); + assert.equal(wake.providerThreadId, owner.id); + assert.equal(wake.nativeThreadId, owner.nativeThreadRef?.nativeId); + assert.deepEqual(wake.modelSelection, source); + assert.deepEqual( + after.thread.modelSelection, + laterHandoff ? CODEX_MODEL_SELECTION : target, + ); + if (laterHandoff) { + assert.isBelow( + turns.indexOf(wake), + turns.findIndex((turn) => turn.text.endsWith("later")), + ); + assert.equal( + after.providerThreads.find((row) => row.id === after.thread.activeProviderThreadId) + ?.providerInstanceId, + CODEX_MODEL_SELECTION.instanceId, + ); + } else assert.equal(after.thread.activeProviderThreadId, selectedRoute); + assert.deepEqual(yield* Ref.get(closedSessions), []); + yield* orchestrator.dispatch({ + type: "provider-session.detach", + commandId: CommandId.make("retire-source"), + threadId, + providerSessionId: owner.providerSessionId!, + }); + yield* worker.drain(); + yield* (yield* EffectOutbox.EffectOutboxV2).awaitSettled( + `effect:retire-source:provider-session.detach:${owner.providerSessionId!}`, + ); + assert.include(yield* Ref.get(closedSessions), owner.providerSessionId!); + yield* Effect.logInfo("continuation-owner-proof", { + variant, + owner: owner.id, + delivered: wake.providerThreadId, + selectedRoute, + selectedInstance: after.thread.providerInstanceId, + retired: owner.providerSessionId, + }); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { + name: `continuation-owner-${variant}`, + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + registry, + { runContinuationWorker: true }, + ).pipe(Layer.provideMerge(ProviderContinuationRequests.layer)), + ), + ); + }), + ); it.live("keeps the user's conversation after a scheduled fresh run in the same thread", () => Effect.gen(function* () { const cwd = yield* checkpointWorkspace("scheduled-fresh-user-chat"); @@ -3657,6 +4033,24 @@ describe("orchestration v2 provider switching", () => { yield* worker.drain(); yield* waitForIdle(threadId); } + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + commandId: CommandId.make("scheduled-fresh-monitor"), + messageId: MessageId.make("scheduled-fresh-monitor"), + threadId, + text: "monitor-wake", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + notification: { + source: { kind: "monitor" }, + outcome: "updated", + summary: "monitor-wake", + }, + }); + yield* worker.drain(); + yield* waitForIdle(threadId); const turns = yield* Ref.get(capturedTurns); const turnFor = (name: string) => turns.find((turn) => turn.text.endsWith(name)); assert.notEqual( @@ -3666,6 +4060,10 @@ describe("orchestration v2 provider switching", () => { // The scheduled run's fresh session must not replace the user's conversation. assert.equal(turnFor("user-second")!.nativeThreadId, turnFor("user-first")!.nativeThreadId); assert.equal(turnFor("user-third")!.nativeThreadId, turnFor("user-first")!.nativeThreadId); + assert.equal( + turnFor("monitor-wake")!.nativeThreadId, + turnFor("user-first")!.nativeThreadId, + ); }).pipe( Effect.provide( ProviderReplayHarness.layerWithRegistry( diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 44e9aa8290e9a..4a8e8c1c782d8 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3347,6 +3347,8 @@ export const OrchestrationV2Command = Schema.Union([ Schema.Struct({ type: Schema.Literal("message.dispatch"), notification: Schema.optional(OrchestrationV2Notification), + /** Buffered provider output must drain on the session that produced it. */ + continuationProviderThreadId: Schema.optional(ProviderThreadId), /** Sends this pending guest message; the server uses the guest's text, not `text`. */ guestMessageId: Schema.optional(GuestMessageId), ...OrchestrationV2CreationFields,