diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 378e399e2532b..7c63bdb6024a6 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -11148,6 +11148,29 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return false; }); + const applySelectionLive = Effect.fnUntraced(function* ( + live: ClaudeLiveQueryContext, + selection: CompiledClaudeModelSelection, + ) { + // Preserve launcher-selected models unless the model changed or a failed update left it unknown. + yield* ( + live.selection.queryIdentity === "" || + live.selection.apiModelId !== selection.apiModelId + ? live.query.setModel(selection.apiModelId) + : Effect.void + ).pipe( + Effect.andThen( + live.query.applyFlagSettings(claudeLiveSelectionFlagSettings(selection)), + ), + Effect.tapError(() => + Effect.sync(() => { + live.selection = { ...live.selection, queryIdentity: "" }; + }), + ), + ); + live.selection = selection; + }); + const openQuery = Effect.fnUntraced(function* ( turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, nativeThreadId: string, @@ -11188,24 +11211,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( (existing.selection.apiModelId === compiledSelection.apiModelId || existing.openedPermissionMode === queryPolicy.permissionMode) ) { - const live = existing; - // Preserve launcher-selected models unless the model changed or a failed update left it unknown. - yield* ( - live.selection.queryIdentity === "" || - live.selection.apiModelId !== compiledSelection.apiModelId - ? live.query.setModel(compiledSelection.apiModelId) - : Effect.void - ).pipe( - Effect.andThen( - live.query.applyFlagSettings(claudeLiveSelectionFlagSettings(compiledSelection)), - ), - Effect.tapError(() => - Effect.sync(() => { - live.selection = { ...live.selection, queryIdentity: "" }; - }), - ), - ); - live.selection = compiledSelection; + yield* applySelectionLive(existing, compiledSelection); } if ( existing !== null && @@ -12403,6 +12409,30 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ); }), }, + applyModelSelection: Effect.fn("ClaudeAdapterV2.applyModelSelection")(function* ({ + providerThread, + modelSelection, + }) { + const live = yield* Ref.get(queryContext); + const selection = compileClaudeModelSelection(modelSelection, modelCatalog); + // A model change still waits for the next turn, which checks it against the turn's permission mode. + if ( + live === null || + live.stopping || + live.nativeThreadId !== providerThread.nativeThreadRef?.nativeId || + live.selection.queryIdentity === selection.queryIdentity || + live.selection.apiModelId !== selection.apiModelId || + !canApplyClaudeSelectionLive(live.selection, selection) + ) { + return; + } + // A failed update clears the live identity, so the next turn applies the selection again. + yield* applySelectionLive(live, selection).pipe( + Effect.catch((cause) => + Effect.logWarning("orchestration-v2.claude-live-selection-failed", { cause }), + ), + ); + }), stopTask: Effect.fn("ClaudeAdapterV2.stopTask")(function* ({ providerThread, taskId }) { const liveQuery = yield* Ref.get(queryContext); const task = (yield* Ref.get(sessionSubagentsByTaskId)).get(taskId); diff --git a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts index 46994a891f705..9700757eb8bbe 100644 --- a/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts +++ b/apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts @@ -494,3 +494,119 @@ it.effect.each(deliveries)( ), ), ); + +it.effect("applies an effort change to the running Claude query without waiting for a turn", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("claude-live-effort"); + const sdkMessages = yield* Queue.unbounded(); + const flagSettings: unknown[] = []; + const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({ + instanceId: modelSelection.instanceId, + settings, + environment: {}, + attachmentsDir: cwd, + fileSystem: yield* FileSystem.FileSystem, + path: yield* Path.Path, + random: yield* SecureRandom.SecureRandom, + idAllocator: yield* IdAllocator.IdAllocatorV2, + queryRunner: { + allocateSessionId: Effect.succeed(sessionId), + open: () => + Effect.succeed({ + messages: Stream.fromQueue(sdkMessages), + offer: () => Effect.void, + setModel: () => Effect.void, + applyFlagSettings: (flags) => + Effect.sync(() => { + flagSettings.push(flags); + }), + 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.die("an effort change must not interrupt the turn"), + close: Effect.void, + }), + forkSession: () => Effect.die("unused"), + subagentLaunchToolUseId: () => Effect.succeed(null), + assertComplete: Effect.void, + }, + }); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make("thread:live-effort"); + const high = { ...modelSelection, options: [{ id: "effort", value: "high" }] }; + const running = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId: ProjectId.make("project:live-effort"), + title: "Live effort", + modelSelection: high, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("first"), + threadId, + messageId: MessageId.make("first"), + text: "Keep working.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(running); + assert.deepEqual(flagSettings, []); + + yield* orchestrator.dispatch({ + type: "thread.model-selection.set", + commandId: CommandId.make("medium"), + threadId, + modelSelection: { ...modelSelection, options: [{ id: "effort", value: "medium" }] }, + }); + yield* worker.drain(); + + assert.deepEqual(flagSettings, [ + { effortLevel: "medium", fastMode: null, ultracode: null }, + ]); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { name: "claude-live-effort" }, + ProviderAdapterRegistry.layerSingle(adapter), + { runEffectWorker: false }, + ), + ), + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), +); diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 39128063e356e..2e94254c8a805 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -3,6 +3,7 @@ import { CheckpointScopeId, CommandId, MessageId, + ModelSelection, NodeId, ProviderSessionId, RunAttemptId, @@ -40,6 +41,12 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ detail: Schema.optional(Schema.String), revokeMcpCredential: Schema.optional(Schema.Boolean), }), + Schema.Struct({ + type: Schema.Literal("provider-session.apply-model-selection"), + providerSessionId: ProviderSessionId, + providerThreadId: ProviderThreadId, + modelSelection: ModelSelection, + }), Schema.Struct({ type: Schema.Literal("provider-turn.start"), runId: RunId, @@ -136,6 +143,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [ ] as const satisfies ReadonlyArray; export const PROCESS_BOUND_EFFECT_TYPES = [ + "provider-session.apply-model-selection", "provider-turn.start", "provider-turn.interrupt", "provider-turn.steer", diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index f0fba6c0f61da..cc43601543eec 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -93,6 +93,7 @@ function layerExecutorFor(input: { ProviderTurnControlService.ProviderTurnControlServiceV2.of({ interrupt: input.interrupt ?? (() => Effect.void), steer: () => Effect.void, + applyModelSelection: () => Effect.void, interruptAndAwaitTerminal: (request) => record( request.replacementProviderSessionId === undefined diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 2c180231c6ace..0fdfcc3a34f91 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -138,6 +138,24 @@ export const layerExecutor: Layer.Layer< }), ), ); + case "provider-session.apply-model-selection": + return providerTurnControl + .applyModelSelection({ + threadId: effect.threadId, + providerSessionId: effect.request.providerSessionId, + providerThreadId: effect.request.providerThreadId, + modelSelection: effect.request.modelSelection, + }) + .pipe( + Effect.mapError( + (cause) => + new OrchestrationEffectExecutionError({ + effectId: effect.id, + effectType: effect.request.type, + cause, + }), + ), + ); case "provider-turn.start": return providerTurnStart .start({ threadId: effect.threadId, runId: effect.request.runId, willRetry }) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 7af71e48d67ea..4393d3692a641 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -4624,6 +4624,46 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Archiving ends the guest share, so its waiting files go too. yield* releaseGuestAttachments(command, thread.guestShare, updatedThread.guestShare, effects); + if ( + command.type === "thread.model-selection.set" && + providerSwitchPlan !== null && + !providerSwitchPlan.instanceChanged && + providerSwitchPlan.releaseProviderSessionIds.length === 0 + ) { + const { providerThread, providerTurn } = yield* projectionStore + .getRunningTurnContext(command.threadId) + .pipe(mapDispatchError(command)); + const providerSessionId = providerThread?.providerSessionId; + if (providerThread !== undefined && providerTurn !== undefined && providerSessionId) { + const session = yield* providerSessions.get(providerSessionId).pipe( + Effect.mapError( + (cause) => + new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause, + }), + ), + ); + if (Option.isSome(session) && session.value.applyModelSelection !== undefined) { + yield* Ref.update(effects, (existing) => [ + ...existing, + { + id: `effect:${command.commandId}:provider-session.apply-model-selection`, + commandId: command.commandId, + threadId: command.threadId, + request: { + type: "provider-session.apply-model-selection", + providerSessionId, + providerThreadId: providerThread.id, + modelSelection: command.modelSelection, + }, + } satisfies PendingOrchestrationEffectV2, + ]); + } + } + } + if (command.type === "thread.metadata.update" && command.regenerateTitle === true) { yield* Ref.update(effects, (existing) => [ ...existing, diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index e628fc5bcc4bd..0197a10fae2e2 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -771,6 +771,7 @@ it.effect( assert.equal(summary.stoppedSessions, 1); assert.equal(summary.retiredEffects, 2); assert.deepEqual(committedInput?.cancelUnsettledEffects?.effectTypes, [ + "provider-session.apply-model-selection", "provider-turn.start", "provider-turn.interrupt", "provider-turn.steer", diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts index bd225bb41773b..2537018b57e31 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts @@ -1,6 +1,7 @@ import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences"; import { MessageId, + type ModelSelection, type NodeId, ProviderSessionId, ProviderThreadId, @@ -68,6 +69,18 @@ export interface ProviderTurnControlServiceV2Shape { readonly providerTurnId: ProviderTurnId; readonly messageId: MessageId; }) => Effect.Effect; + /** Best effort: a turn that ended or moved sessions picks the selection up when it next starts. */ + readonly applyModelSelection: (input: { + readonly threadId: ThreadId; + readonly providerSessionId: ProviderSessionId; + readonly providerThreadId: ProviderThreadId; + readonly modelSelection: ModelSelection; + }) => Effect.Effect< + void, + | ProjectionStore.ProjectionStoreV2Error + | ProviderSessionManager.ProviderSessionManagerV2Error + | ProviderAdapter.ProviderAdapterV2Error + >; readonly interruptAndAwaitTerminal: (input: { readonly threadId: ThreadId; readonly providerSessionId: ProviderSessionId; @@ -185,6 +198,26 @@ export const layer: Layer.Layer< }); return ProviderTurnControlServiceV2.of({ + applyModelSelection: Effect.fn("ProviderTurnControlService.applyModelSelection")( + function* (input) { + const { providerThread, providerTurn } = yield* projections.getRunningTurnContext( + input.threadId, + ); + if ( + providerThread?.id !== input.providerThreadId || + providerThread.providerSessionId !== input.providerSessionId || + providerTurn === undefined + ) { + return; + } + const session = yield* sessions.get(input.providerSessionId); + if (Option.isNone(session) || session.value.applyModelSelection === undefined) return; + yield* session.value.applyModelSelection({ + providerThread, + modelSelection: input.modelSelection, + }); + }, + ), interrupt: Effect.fn("ProviderTurnControlService.interrupt")( function* (input) { const loaded = yield* load({ ...input, operation: "interrupt" }); diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 98f71a7473a84..6f09312a3cbea 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -676,6 +676,11 @@ export interface ProviderAdapterV2SessionRuntime { readonly steerTurn: ( input: ProviderAdapterV2SteerInput, ) => Effect.Effect; + /** Applies a thread's new selection to its running process where the provider can; otherwise it waits for the next turn. */ + readonly applyModelSelection?: (input: { + readonly providerThread: OrchestrationV2ProviderThread; + readonly modelSelection: ModelSelection; + }) => Effect.Effect; readonly stopTask?: (input: { readonly providerThread: OrchestrationV2ProviderThread; readonly taskId: string;