diff --git a/src/browser/App.tsx b/src/browser/App.tsx index e8c5b67331..a0e13d545c 100644 --- a/src/browser/App.tsx +++ b/src/browser/App.tsx @@ -77,6 +77,7 @@ import { useTelemetry } from "./hooks/useTelemetry"; import { getRuntimeTypeForTelemetry } from "@/common/telemetry"; import { useStartWorkspaceCreation } from "./hooks/useStartWorkspaceCreation"; import { useAPI } from "@/browser/contexts/API"; +import { requestActiveTurnThinkingLevel } from "@/browser/utils/activeTurnThinking"; import { clearPendingWorkspaceAiSettings, markPendingWorkspaceAiSettings, @@ -534,6 +535,10 @@ function AppInner() { clearPendingWorkspaceAiSettings(workspaceId, normalizedAgentId); // Best-effort only. }); + + // Mid-turn change: also apply to the active turn's next model step so + // the palette/keybind path behaves like the slider (ThinkingProvider). + requestActiveTurnThinkingLevel(api, workspaceId, normalized); } // Dispatch toast notification event for UI feedback diff --git a/src/browser/contexts/ThinkingContext.test.tsx b/src/browser/contexts/ThinkingContext.test.tsx index aacc694331..7ca00695aa 100644 --- a/src/browser/contexts/ThinkingContext.test.tsx +++ b/src/browser/contexts/ThinkingContext.test.tsx @@ -590,6 +590,81 @@ describe("ThinkingContext", () => { } }); + test("requests a mid-turn override for the active workspace turn on slider changes", async () => { + const workspaceId = "ws-set-thinking-mid-turn"; + const setActiveTurnThinkingLevel = mock< + (args: { + workspaceId: string; + thinkingLevel: ThinkingLevel; + }) => Promise<{ success: true; data: { accepted: boolean } }> + >(() => Promise.resolve({ success: true as const, data: { accepted: true } })); + currentClientMock = { + workspace: { + updateAgentAISettings: mock(() => + Promise.resolve({ success: true as const, data: undefined }) + ), + setActiveTurnThinkingLevel, + }, + }; + + setWorkspaceMetadata(createWorkspaceMetadata({ id: workspaceId })); + + const view = renderWithWorkspaceMetadata({ + workspaceId, + modelOverride: null, + children: ( + + + + ), + }); + + const button = await view.findByTestId("set-thinking-medium", undefined, METADATA_WAIT_OPTIONS); + act(() => { + button.click(); + }); + + await waitFor(() => { + expect(setActiveTurnThinkingLevel).toHaveBeenCalledWith({ + workspaceId, + thinkingLevel: "medium", + }); + }, METADATA_WAIT_OPTIONS); + }); + + test("does not request a mid-turn override in project scope (no workspaceId)", async () => { + const projectPath = "/Users/dev/mid-turn-scope"; + const setActiveTurnThinkingLevel = mock(() => + Promise.resolve({ success: true as const, data: { accepted: false } }) + ); + currentClientMock = { + workspace: { setActiveTurnThinkingLevel }, + }; + + const view = renderWithAPI( + + + + ); + + const button = await view.findByTestId("set-thinking-medium", undefined, METADATA_WAIT_OPTIONS); + act(() => { + button.click(); + }); + + // Project/global scopes have no active turn to override; the route must + // not fire (persisted settings alone drive the next turn). + await waitFor(() => { + expect( + readPersistedState( + getThinkingLevelKey(getProjectScopeId(projectPath)), + null + ) + ).toBe("medium"); + }, METADATA_WAIT_OPTIONS); + expect(setActiveTurnThinkingLevel).not.toHaveBeenCalled(); + }); + test("cycles thinking level via keybind in project-scoped (creation) flow", async () => { const projectPath = "/Users/dev/my-project"; diff --git a/src/browser/contexts/ThinkingContext.tsx b/src/browser/contexts/ThinkingContext.tsx index 9dfea68d0e..bc59b88cf8 100644 --- a/src/browser/contexts/ThinkingContext.tsx +++ b/src/browser/contexts/ThinkingContext.tsx @@ -27,6 +27,7 @@ import { enforceThinkingPolicy, getAvailableThinkingLevels } from "@/common/util import { useMinThinkingLevels } from "@/browser/hooks/useMinThinkingLevels"; import { useProvidersConfig } from "@/browser/hooks/useProvidersConfig"; import { useAPI } from "@/browser/contexts/API"; +import { requestActiveTurnThinkingLevel } from "@/browser/utils/activeTurnThinking"; import { clearPendingWorkspaceAiSettings, getWorkspaceAiSettingsFromMetadata, @@ -226,12 +227,20 @@ export const ThinkingProvider: React.FC = (props) => { thinkingLevel: level, reasoningMode: getCurrentReasoningMode(), }); + // Mid-turn change: also request the new level for the active turn's next + // model step. Non-workspace (project/global) mounts have no workspaceId + // and skip this inside the helper. + if (props.workspaceId) { + requestActiveTurnThinkingLevel(api, props.workspaceId, level); + } }, [ + api, defaultModel, getCurrentReasoningMode, metadataSettings.model, persistAgentAiSettings, + props.workspaceId, scopeId, setThinkingLevelInternal, ] diff --git a/src/browser/utils/activeTurnThinking.ts b/src/browser/utils/activeTurnThinking.ts new file mode 100644 index 0000000000..e82b94775c --- /dev/null +++ b/src/browser/utils/activeTurnThinking.ts @@ -0,0 +1,25 @@ +import type { APIClient } from "@/browser/contexts/API"; +import type { ThinkingLevel } from "@/common/types/thinking"; + +/** + * Best-effort request to apply a thinking-level change to the active turn's + * NEXT model step (mid-turn override). Shared by every UI path that changes a + * workspace's thinking level (slider, keybinds, command palette) so they all + * behave identically during a stream. + * + * No streaming gate on purpose: the backend no-ops cheaply (accepted: false) + * when the workspace is idle or unknown, and the persisted workspace setting + * (updated separately by callers) still covers future turns. + */ +export function requestActiveTurnThinkingLevel( + api: APIClient | null | undefined, + workspaceId: string, + thinkingLevel: ThinkingLevel +): void { + if (!api || !workspaceId) { + return; + } + api.workspace.setActiveTurnThinkingLevel({ workspaceId, thinkingLevel }).catch(() => { + // Best-effort: transient IPC failure loses only the mid-turn nudge. + }); +} diff --git a/src/common/orpc/schemas/api.ts b/src/common/orpc/schemas/api.ts index 7e759783fa..3c55ef5eb3 100644 --- a/src/common/orpc/schemas/api.ts +++ b/src/common/orpc/schemas/api.ts @@ -1202,6 +1202,17 @@ export const workspace = { }), output: ResultSchema(z.void(), z.string()), }, + // Mid-turn thinking change: request that the active turn's NEXT model step + // uses this level. `accepted: false` (success) = no turn active — persisted + // settings already cover the next turn. `accepted: true` = the level applies + // to the current turn's next step if one occurs (expires silently otherwise). + setActiveTurnThinkingLevel: { + input: z.object({ + workspaceId: z.string(), + thinkingLevel: ThinkingLevelSchema, + }), + output: ResultSchema(z.object({ accepted: z.boolean() }), z.string()), + }, preflightArchive: { input: z.object({ workspaceId: z.string() }), output: ResultSchema(ArchivePreflightResultSchema, z.string()), diff --git a/src/common/types/thinking.ts b/src/common/types/thinking.ts index a8a4b13b8e..f8139247c4 100644 --- a/src/common/types/thinking.ts +++ b/src/common/types/thinking.ts @@ -168,14 +168,8 @@ export const ANTHROPIC_THINKING_BUDGETS: Record = { /** * Anthropic effort type - matches SDK's AnthropicProviderOptions["effort"]. - * - * Note: Opus 4.7 and Sonnet 5 introduced a native "xhigh" effort level in the API, - * but the SDK's Zod validator still rejects "xhigh". Mux handles this by sending - * "max" through the SDK and rewriting `output_config.effort` to "xhigh" in a fetch - * wrapper for native-xhigh Anthropic models when the user selected the xhigh ThinkingLevel. - * See `wrapFetchWithAnthropicCacheControl` and `buildRequestHeaders`. */ -export type AnthropicEffortLevel = "low" | "medium" | "high" | "max"; +export type AnthropicEffortLevel = "low" | "medium" | "high" | "xhigh" | "max"; /** * Anthropic effort parameter mapping (Opus 4.5+) @@ -183,22 +177,38 @@ export type AnthropicEffortLevel = "low" | "medium" | "high" | "max"; * The effort parameter controls how much computational work the model applies. * - Opus 4.5 supports: low, medium, high (policy clamps xhigh → high) * - Opus 4.6 supports: low, medium, high, max (xhigh maps to "max" effort) - * - Opus 4.7+ and Sonnet 5+ support: low, medium, high, xhigh, max (xhigh requires wire override) + * - Opus 4.7+ and Sonnet 5+ support: low, medium, high, xhigh, max (native xhigh) + * + * The table keeps `xhigh: "max"` as the non-native fallback; `getAnthropicEffort` + * upgrades it to the native "xhigh" wire value on models that support it. * - * Because the @ai-sdk/anthropic Zod schema doesn't accept "xhigh" yet, we send - * "max" through the SDK for native-xhigh Anthropic models and rewrite - * `output_config.effort` to "xhigh" in the Anthropic fetch wrapper. + * SDK coupling note: explicit `providerOptions.anthropic.effort` values flow + * verbatim into `output_config.effort` as of @ai-sdk/anthropic 4.0.11 (its Zod + * schema accepts "xhigh"; its `supportsXhighEffort` model gating applies only to + * the SDK's top-level `reasoning` CallOption mapping, which Mux does not use). + * Re-verify this pass-through on major SDK upgrades. */ const ANTHROPIC_EFFORT: Record = { off: "low", low: "low", medium: "medium", high: "high", - xhigh: "max", // SDK placeholder; fetch wrapper rewrites to "xhigh" on native-xhigh models + xhigh: "max", // Non-native fallback; native-xhigh models get "xhigh" via getAnthropicEffort max: "max", }; -export function getAnthropicEffort(level: ThinkingLevel): AnthropicEffortLevel { +/** + * Model-aware Anthropic effort resolution: `xhigh` maps to the native "xhigh" + * wire value only on models that support it (Opus 4.7+, Sonnet 5+, Mythos); + * everywhere else it falls back to "max" per the table above. + */ +export function getAnthropicEffort( + level: ThinkingLevel, + capabilityModel: string +): AnthropicEffortLevel { + if (level === "xhigh" && anthropicSupportsNativeXhigh(capabilityModel)) { + return "xhigh"; + } return ANTHROPIC_EFFORT[level]; } diff --git a/src/common/utils/ai/providerOptions.test.ts b/src/common/utils/ai/providerOptions.test.ts index 93426c502d..64cb6ffdf4 100644 --- a/src/common/utils/ai/providerOptions.test.ts +++ b/src/common/utils/ai/providerOptions.test.ts @@ -16,7 +16,6 @@ import { preserveAnthropic1MContextForFollowUp, resolveProviderOptionsNamespaceKey, ANTHROPIC_1M_CONTEXT_HEADER, - MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER, MUX_WORKSPACE_ID_HEADER, } from "./providerOptions"; @@ -111,7 +110,9 @@ describe("buildProviderOptions - Anthropic", () => { }); }); - for (const model of ["claude-opus-4-6", "claude-sonnet-4-6", "claude-sonnet-5"] as const) { + // Non-native-xhigh adaptive models: xhigh falls back to "max" effort and + // adaptive thinking carries no `display` field. + for (const model of ["claude-opus-4-6", "claude-sonnet-4-6"] as const) { describe(`${model} (adaptive thinking + effort)`, () => { for (const { thinking, expectedThinking, effort } of [ { thinking: "medium", expectedThinking: { type: "adaptive" }, effort: "medium" }, @@ -134,12 +135,49 @@ describe("buildProviderOptions - Anthropic", () => { }); } + // Native-xhigh models (Opus 4.7+ / Sonnet 5+): xhigh is a distinct native + // effort and adaptive thinking requires `display: "summarized"` to return + // thinking content. + for (const model of ["claude-opus-4-7", "claude-sonnet-5"] as const) { + describe(`${model} (native xhigh effort + summarized display)`, () => { + for (const { thinking, expectedThinking, effort } of [ + { + thinking: "medium", + expectedThinking: { type: "adaptive", display: "summarized" }, + effort: "medium", + }, + { + thinking: "xhigh", + expectedThinking: { type: "adaptive", display: "summarized" }, + effort: "xhigh", + }, + { + thinking: "max", + expectedThinking: { type: "adaptive", display: "summarized" }, + effort: "max", + }, + { thinking: "off", expectedThinking: { type: "disabled" }, effort: "low" }, + ] as const) { + test(`maps ${thinking} to ${effort} effort`, () => { + const anthropic = anthropicProviderOptions( + buildProviderOptions(`anthropic:${model}`, thinking) + ); + + expect(anthropic.thinking).toEqual(expectedThinking); + expect(anthropic.effort).toBe(effort); + }); + } + }); + } + describe("claude-fable-5 (Mythos-class: API rejects disabled thinking)", () => { test("maps medium to adaptive thinking like other adaptive models", () => { const anthropic = anthropicProviderOptions( buildProviderOptions("anthropic:claude-fable-5", "medium") ); - expect(anthropic.thinking).toEqual({ type: "adaptive" }); + // Mythos-class models are native-xhigh tier, so adaptive thinking carries + // the summarized display flag. + expect(anthropic.thinking).toEqual({ type: "adaptive", display: "summarized" }); expect(anthropic.effort).toBe("medium"); }); @@ -1590,67 +1628,12 @@ describe("buildRequestHeaders", () => { }); } - describe("Opus 4.7+ xhigh effort override", () => { - for (const { name, model, routeProvider, thinkingLevel, expected } of [ - { - name: "emits override header when thinkingLevel=xhigh for Opus 4.7", - model: "anthropic:claude-opus-4-7", - routeProvider: undefined, - thinkingLevel: "xhigh", - expected: { [MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER]: "xhigh" }, - }, - { - name: "emits override header for gateway-routed Opus 4.7 with xhigh (passthrough)", - model: "mux-gateway:anthropic/claude-opus-4-7", - routeProvider: "mux-gateway", - thinkingLevel: "xhigh", - expected: { [MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER]: "xhigh" }, - }, - { - name: "emits override header for Opus 4.8", - model: "anthropic:claude-opus-4-8", - routeProvider: undefined, - thinkingLevel: "xhigh", - expected: { [MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER]: "xhigh" }, - }, - { - // Sonnet 5 added native xhigh for the Sonnet tier, so it needs the wire rewrite too. - name: "emits override header for Sonnet 5", - model: "anthropic:claude-sonnet-5", - routeProvider: undefined, - thinkingLevel: "xhigh", - expected: { [MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER]: "xhigh" }, - }, - { - name: "does not emit override header for Opus 4.7 with thinkingLevel=max", - model: "anthropic:claude-opus-4-7", - routeProvider: undefined, - thinkingLevel: "max", - expected: undefined, - }, - { - name: "does not emit override header for Opus 4.6 with xhigh", - // Opus 4.6 maps xhigh -> "max" effort; SDK accepts "max" so no wire rewrite needed. - model: "anthropic:claude-opus-4-6", - routeProvider: undefined, - thinkingLevel: "xhigh", - expected: undefined, - }, - { - name: "does not emit override header for non-passthrough gateway (openrouter)", - // Non-passthrough gateways must not receive this Mux-internal header. - model: "anthropic:claude-opus-4-7", - routeProvider: "openrouter", - thinkingLevel: "xhigh", - expected: undefined, - }, - ] as const) { - test(name, () => { - expect( - buildRequestHeaders(model, undefined, undefined, undefined, routeProvider, thinkingLevel) - ).toEqual(expected); - }); - } + // Native xhigh effort no longer needs a Mux-internal override header: the + // SDK accepts effort "xhigh" directly (see buildProviderOptions tests above), + // so buildRequestHeaders is thinking-level-independent. + test("does not emit any Mux-internal effort header for native-xhigh models", () => { + expect(buildRequestHeaders("anthropic:claude-opus-4-7")).toBeUndefined(); + expect(buildRequestHeaders("anthropic:claude-sonnet-5")).toBeUndefined(); }); describe("openaiProModeAvailable", () => { diff --git a/src/common/utils/ai/providerOptions.ts b/src/common/utils/ai/providerOptions.ts index e439e42c9f..77c7882037 100644 --- a/src/common/utils/ai/providerOptions.ts +++ b/src/common/utils/ai/providerOptions.ts @@ -42,15 +42,6 @@ import { export { resolveProviderOptionsNamespaceKey } from "./models"; export { openaiProModeAvailable } from "./proMode"; -/** - * Request header used to override Anthropic's `output_config.effort` at the - * wire level. The @ai-sdk/anthropic Zod schema rejects "xhigh", so for - * Opus 4.7+ with xhigh ThinkingLevel we send "max" through the SDK and ask the - * fetch wrapper to rewrite it to "xhigh" via this header (which is then stripped - * before the request reaches Anthropic). - */ -export const MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER = "x-mux-anthropic-effort"; - /** * OpenRouter reasoning options * @see https://openrouter.ai/docs/use-cases/reasoning-tokens @@ -281,11 +272,9 @@ export function buildProviderOptions( const usesAdaptiveThinking = isOpus46 || supportsNativeXhigh || isSonnet46; if (isOpus45 || usesAdaptiveThinking) { - // Map to SDK-accepted effort. For Opus 4.7+ / Sonnet 5+ with xhigh ThinkingLevel, the - // SDK gets "max" as a placeholder and the Anthropic fetch wrapper rewrites - // `output_config.effort` to "xhigh" via the X-Mux-Anthropic-Effort header - // (added in buildRequestHeaders). - const effortLevel = getAnthropicEffort(effectiveThinking); + // Model-aware effort: native-xhigh models (Opus 4.7+ / Sonnet 5+ / Mythos) + // get the native "xhigh" wire value; other adaptive models keep xhigh → "max". + const effortLevel = getAnthropicEffort(effectiveThinking, capabilityModel); const budgetTokens = ANTHROPIC_THINKING_BUDGETS[effectiveThinking]; // Opus 4.6+ / Sonnet 4.6 / Sonnet 5: adaptive thinking when on, disabled when off // Opus 4.5: enabled thinking with budgetTokens ceiling (only when not "off") @@ -293,12 +282,19 @@ export function buildProviderOptions( // excludes "off" for them and AIService clamps the effective level via // resolveEffectiveThinkingLevel, so "off" should not reach here — but if a stray // path does, omit `thinking` (API defaults to adaptive) rather than hard-erroring. + // + // Native-xhigh models require `thinking.display: "summarized"` to return + // thinking content on adaptive requests; non-native adaptive models + // (Opus 4.6 / Sonnet 4.6) must not receive `display`. + // See https://platform.claude.com/docs/en/build-with-claude/adaptive-thinking#summarized-thinking const thinking: AnthropicProviderOptions["thinking"] = usesAdaptiveThinking ? effectiveThinking === "off" ? anthropicRejectsDisabledThinking(capabilityModel) ? undefined : { type: "disabled" } - : { type: "adaptive" } + : supportsNativeXhigh + ? { type: "adaptive", display: "summarized" } + : { type: "adaptive" } : budgetTokens > 0 ? { type: "enabled", budgetTokens } : undefined; @@ -309,10 +305,6 @@ export function buildProviderOptions( thinkingLevel: effectiveThinking, }); - // Note: Opus 4.7+ requires `thinking.display: "summarized"` to receive - // thinking content, but the SDK's Zod schema strips unknown keys so we - // can't add it here. The Anthropic fetch wrapper injects `display` on the - // wire for Opus 4.7+ adaptive thinking requests. const anthropicOptions: AnthropicProviderOptions = { disableParallelToolUse: false, sendReasoning: true, @@ -614,8 +606,7 @@ export function buildRequestHeaders( muxProviderOptions?: MuxProviderOptions, workspaceId?: string, providersConfig?: ProvidersConfigMap | null, - routeProvider?: ProviderName, - thinkingLevel?: ThinkingLevel + routeProvider?: ProviderName ): Record | undefined { const headers: Record = {}; @@ -638,22 +629,5 @@ export function buildRequestHeaders( headers["anthropic-beta"] = ANTHROPIC_1M_CONTEXT_HEADER; } - // Native-xhigh Anthropic models use an "xhigh" effort level that the - // @ai-sdk/anthropic Zod schema doesn't accept yet. Emit a Mux-internal header - // so the Anthropic fetch wrapper can rewrite `output_config.effort` to "xhigh" on the wire. - // - // Only emit when the route will pass through our Anthropic fetch wrapper - // (direct Anthropic or passthrough gateways like mux-gateway). Non-passthrough - // gateways (OpenRouter, Bedrock, github-copilot) must not see this header — - // they would forward it externally verbatim and strict proxies may reject it. - if ( - origin === "anthropic" && - routePassesHeaders && - thinkingLevel === "xhigh" && - anthropicSupportsNativeXhigh(modelString) - ) { - headers[MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER] = "xhigh"; - } - return Object.keys(headers).length > 0 ? headers : undefined; } diff --git a/src/common/utils/thinking/policy.test.ts b/src/common/utils/thinking/policy.test.ts index b7256e7411..08abefeb3d 100644 --- a/src/common/utils/thinking/policy.test.ts +++ b/src/common/utils/thinking/policy.test.ts @@ -9,6 +9,7 @@ import { resolveMinimumThinkingLevel, resolveEffectiveThinkingLevel, getAvailableThinkingLevels, + isXaiGrokFastVariantSwap, } from "./policy"; describe("getThinkingPolicyForModel", () => { @@ -922,3 +923,16 @@ describe("enforceThinkingPolicy with a minimum floor", () => { expect(enforceThinkingPolicy("anthropic:claude-sonnet-4-5", "off")).toBe("off"); }); }); + +describe("isXaiGrokFastVariantSwap", () => { + test("flags only off<->on transitions on xai:grok-4-1-fast", () => { + // off <-> non-off swaps the underlying reasoning/non-reasoning variant. + expect(isXaiGrokFastVariantSwap("xai:grok-4-1-fast", "off", "high")).toBe(true); + expect(isXaiGrokFastVariantSwap("xai:grok-4-1-fast", "high", "off")).toBe(true); + // non-off -> non-off stays on the reasoning variant (no swap). + expect(isXaiGrokFastVariantSwap("xai:grok-4-1-fast", "low", "high")).toBe(false); + // Other models never swap instances on thinking-level changes. + expect(isXaiGrokFastVariantSwap("xai:grok-4-1-fast-reasoning", "off", "high")).toBe(false); + expect(isXaiGrokFastVariantSwap("anthropic:claude-sonnet-4-5", "off", "high")).toBe(false); + }); +}); diff --git a/src/common/utils/thinking/policy.ts b/src/common/utils/thinking/policy.ts index 874a679d7a..efe58b78ce 100644 --- a/src/common/utils/thinking/policy.ts +++ b/src/common/utils/thinking/policy.ts @@ -347,6 +347,25 @@ export function enforceThinkingPolicy( return closest; } +/** + * Whether a thinking-level transition would swap the underlying model instance + * for xAI's grok-4-1-fast (providerModelFactory maps off → non-reasoning and + * non-off → reasoning variants at model creation). Such transitions cannot be + * applied to an in-flight stream via provider options, so mid-turn overrides + * skip them (the persisted setting still applies to the next turn). + */ +export function isXaiGrokFastVariantSwap( + modelString: string, + currentLevel: ThinkingLevel, + nextLevel: ThinkingLevel +): boolean { + const [provider, modelId] = modelString.trim().toLowerCase().split(":", 2); + if (provider !== "xai" || modelId !== "grok-4-1-fast") { + return false; + } + return (currentLevel === "off") !== (nextLevel === "off"); +} + /** * Resolve a parsed thinking input to a concrete ThinkingLevel for a given model. * diff --git a/src/node/orpc/router.ts b/src/node/orpc/router.ts index 266e099a6a..79e791b21a 100644 --- a/src/node/orpc/router.ts +++ b/src/node/orpc/router.ts @@ -4095,6 +4095,15 @@ export const router = (authToken?: string) => { input.aiSettings ); }), + setActiveTurnThinkingLevel: t + .input(schemas.workspace.setActiveTurnThinkingLevel.input) + .output(schemas.workspace.setActiveTurnThinkingLevel.output) + .handler(({ context, input }) => { + return context.workspaceService.setActiveTurnThinkingLevel( + input.workspaceId, + input.thinkingLevel + ); + }), updateTitle: t .input(schemas.workspace.updateTitle.input) .output(schemas.workspace.updateTitle.output) diff --git a/src/node/services/agentSession.thinkingOverride.test.ts b/src/node/services/agentSession.thinkingOverride.test.ts new file mode 100644 index 0000000000..e4ccca47b7 --- /dev/null +++ b/src/node/services/agentSession.thinkingOverride.test.ts @@ -0,0 +1,163 @@ +import { describe, expect, it, mock } from "bun:test"; + +import type { MuxMessage } from "@/common/types/message"; +import { Ok, Err } from "@/common/types/result"; +import type { AIService, StreamMessageOptions } from "@/node/services/aiService"; +import { createAgentSessionHarness } from "./agentSession.testHarness"; +import type { ActiveTurnThinkingOverride } from "./thinkingOverride"; + +const MODEL = "anthropic:claude-sonnet-4-5"; + +describe("AgentSession.setActiveTurnThinkingLevel", () => { + it("reports accepted:false while idle (persisted settings cover the next turn)", async () => { + const { session, cleanup } = await createAgentSessionHarness({ + workspaceId: "thinking-override-idle", + }); + try { + expect(session.setActiveTurnThinkingLevel("high")).toEqual({ accepted: false }); + } finally { + session.dispose(); + await cleanup(); + } + }); + + it("accepts writes during PREPARING and streaming, threads the holder to the stream, and expires it on turn end", async () => { + const capturedHolders: Array = []; + const capturedPendingAtStreamStart: Array = []; + let sessionRef: { + setActiveTurnThinkingLevel: (level: "low" | "high" | "medium") => { accepted: boolean }; + } | null = null; + + const streamMessage = mock((opts: StreamMessageOptions) => { + capturedHolders.push(opts.activeTurnThinkingOverride); + capturedPendingAtStreamStart.push(opts.activeTurnThinkingOverride?.pending); + // Simulate a slider change while the stream is active: both writes must + // land in the SAME holder object the stream received (last write wins). + const first = sessionRef?.setActiveTurnThinkingLevel("high"); + const second = sessionRef?.setActiveTurnThinkingLevel("low"); + expect(first).toEqual({ accepted: true }); + expect(second).toEqual({ accepted: true }); + expect(opts.activeTurnThinkingOverride?.pending).toBe("low"); + return Promise.resolve(Ok(undefined)); + }); + + const { session, cleanup } = await createAgentSessionHarness({ + workspaceId: "thinking-override-active", + aiServiceOverrides: { + streamMessage: streamMessage as unknown as AIService["streamMessage"], + }, + }); + sessionRef = session; + + try { + const result = await session.sendMessage("hello", { model: MODEL, agentId: "exec" }); + expect(result.success).toBe(true); + expect(streamMessage.mock.calls).toHaveLength(1); + expect(capturedHolders[0]).toBeDefined(); + + // The turn ended (mocked stream resolved without events -> IDLE): the + // holder expired, so late writes fall back to persisted settings. + await session.waitForIdle(); + expect(session.setActiveTurnThinkingLevel("high")).toEqual({ accepted: false }); + + // A follow-up turn gets a FRESH holder with no inherited pending level. + const second = await session.sendMessage("again", { model: MODEL, agentId: "exec" }); + expect(second.success).toBe(true); + expect(capturedHolders[1]).toBeDefined(); + expect(capturedHolders[1]).not.toBe(capturedHolders[0]); + expect(capturedPendingAtStreamStart[1]).toBeUndefined(); + } finally { + session.dispose(); + await cleanup(); + } + }); + + it("delivers a change made during the PREPARING window to the turn's stream options", async () => { + let pendingSeenByStream: string | undefined; + const streamMessage = mock((opts: StreamMessageOptions) => { + pendingSeenByStream = opts.activeTurnThinkingOverride?.pending; + return Promise.resolve(Ok(undefined)); + }); + + const { session, cleanup } = await createAgentSessionHarness({ + workspaceId: "thinking-override-preparing", + aiServiceOverrides: { + streamMessage: streamMessage as unknown as AIService["streamMessage"], + }, + }); + + try { + // onAccepted runs after the holder is created but before the stream + // starts — exactly the PREPARING window a fast slider change can hit. + const result = await session.sendMessage( + "hello", + { model: MODEL, agentId: "exec" }, + { + onAccepted: () => { + expect(session.setActiveTurnThinkingLevel("high")).toEqual({ accepted: true }); + }, + } + ); + expect(result.success).toBe(true); + expect(streamMessage.mock.calls).toHaveLength(1); + // The pre-stream write reaches the turn's first provider request via the + // holder (prepareStep runs before step 1). + expect(pendingSeenByStream).toBe("high"); + } finally { + session.dispose(); + await cleanup(); + } + }); + + it("clears the holder when a pre-stream failure aborts the accepted turn", async () => { + const streamMessage = mock((_opts: StreamMessageOptions) => + Promise.resolve(Err({ type: "unknown" as const, raw: "startup failed" })) + ); + + const { session, cleanup } = await createAgentSessionHarness({ + workspaceId: "thinking-override-prestream-failure", + aiServiceOverrides: { + streamMessage: streamMessage as unknown as AIService["streamMessage"], + }, + }); + + try { + const result = await session.sendMessage("hello", { model: MODEL, agentId: "exec" }); + expect(result.success).toBe(false); + await session.waitForIdle(); + expect(session.setActiveTurnThinkingLevel("high")).toEqual({ accepted: false }); + } finally { + session.dispose(); + await cleanup(); + } + }); + + it("clears the holder when an onAccepted failure aborts the turn before streaming", async () => { + const streamMessage = mock((_history: MuxMessage[]) => Promise.resolve(Ok(undefined))); + const { session, cleanup } = await createAgentSessionHarness({ + workspaceId: "thinking-override-onaccepted-failure", + aiServiceOverrides: { + streamMessage: streamMessage as unknown as AIService["streamMessage"], + }, + }); + + try { + const result = await session.sendMessage( + "hello", + { model: MODEL, agentId: "exec" }, + { + onAccepted: () => { + throw new Error("acceptance hook failed"); + }, + } + ); + expect(result.success).toBe(false); + // The stream never started; the holder must not leak into idle state. + expect(streamMessage.mock.calls).toHaveLength(0); + expect(session.setActiveTurnThinkingLevel("high")).toEqual({ accepted: false }); + } finally { + session.dispose(); + await cleanup(); + } + }); +}); diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index d31d0de360..601a7665a5 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -60,6 +60,7 @@ import { type ThinkingLevel, } from "@/common/types/thinking"; import { enforceThinkingPolicy, resolveMinimumThinkingLevel } from "@/common/utils/thinking/policy"; +import type { ActiveTurnThinkingOverride } from "@/node/services/thinkingOverride"; import { createMuxMessage, dedupeAgentSkillRefs, @@ -359,6 +360,14 @@ export class AgentSession { private disposed = false; private turnPhase: TurnPhase = TurnPhase.IDLE; private activePreparedTurnAbortController: AbortController | null = null; + /** + * Per-turn holder for mid-turn thinking-level overrides. Created when a turn + * is durably accepted (before any await that could let the renderer's slider + * route race in), threaded by reference into StreamManager, and cleared when + * the turn ends (setTurnPhase → IDLE). Null while idle: the slider route then + * reports accepted:false and persisted settings cover the next turn. + */ + private activeTurnThinkingOverride: ActiveTurnThinkingOverride | null = null; // When true, stream-end skips auto-flushing queued messages so an edit can truncate first. private deferQueuedFlushUntilAfterEdit = false; // Provider-executed tools (for example native web_search/web_fetch) complete inside one @@ -2806,6 +2815,13 @@ export class AgentSession { return Ok(undefined); } + // Turn durably accepted + options finalized: open the mid-turn thinking + // override window BEFORE the user-message emit / onAccepted / any further + // await, so a slider change during PREPARING (runtime warmup, model + // creation) lands in the holder the stream's prepareStep will read. + const turnThinkingOverride: ActiveTurnThinkingOverride = {}; + this.activeTurnThinkingOverride = turnThinkingOverride; + // Emit snapshots only for immediately-sent turns. On on-send compaction paths, // snapshots are deferred with the follow-up message to avoid duplicate ephemeral // snapshot rows that were never persisted. @@ -2848,6 +2864,11 @@ export class AgentSession { try { await internal?.onAccepted?.(); } catch (error) { + // Pre-stream failure: identity-guarded so a replacement turn's holder + // (created while this one unwound) is never cleared by mistake. + if (this.activeTurnThinkingOverride === turnThinkingOverride) { + this.activeTurnThinkingOverride = null; + } return Err(createUnknownSendMessageError(getErrorMessage(error))); } @@ -2887,7 +2908,8 @@ export class AgentSession { undefined, agentInitiated, preparedTurnAbortController.signal, - goalKind + goalKind, + turnThinkingOverride ); } finally { // Success should advance via stream events; if startup never emitted any, don't leave the @@ -2963,6 +2985,10 @@ export class AgentSession { // accept its options, even if startup fails before the stream fully begins. this.setAutoRetryResumeState(optionsForStream, internal?.agentInitiated, internal?.goalKind); this.setTurnPhase(TurnPhase.PREPARING); + // Open the mid-turn thinking override window for the resumed turn (after + // setTurnPhase(PREPARING), which clears the holder on the IDLE transition). + const turnThinkingOverride: ActiveTurnThinkingOverride = {}; + this.activeTurnThinkingOverride = turnThinkingOverride; try { // Must await here so the finally block runs after streaming completes, // not immediately when the Promise is returned. @@ -2973,7 +2999,8 @@ export class AgentSession { undefined, internal?.agentInitiated, undefined, - internal?.goalKind + internal?.goalKind, + turnThinkingOverride ); if (!result.success) { return result; @@ -3507,7 +3534,11 @@ export class AgentSession { disablePostCompactionAttachments?: boolean, agentInitiated?: boolean, abortSignal?: AbortSignal, - goalKind?: GoalSyntheticMessageKind + goalKind?: GoalSyntheticMessageKind, + // Session-owned per-turn holder for mid-turn thinking changes. Passed + // explicitly (not read from the field) so a preempted turn can never pick + // up its replacement's holder. Absent for internal retry paths. + activeTurnThinkingOverride?: ActiveTurnThinkingOverride ): Promise> { const isStartupAbortRequested = (): boolean => abortSignal?.aborted === true; @@ -3692,6 +3723,10 @@ export class AgentSession { disableWorkspaceAgents: options?.disableWorkspaceAgents, hasQueuedMessages: this.hasQueuedMessages.bind(this), openaiTruncationModeOverride, + // Mid-turn thinking overrides clamp against the same floor as the + // send-time level above (single source of truth for the floor). + minThinkingLevel, + activeTurnThinkingOverride, }); if (!streamResult.success) { @@ -5014,6 +5049,10 @@ export class AgentSession { this.emitStreamLifecycleIfChanged(); if (next === TurnPhase.IDLE) { + // Turn ended: expire any mid-turn thinking override. Safe unconditionally + // because a replacement turn (e.g. an edit) only creates its holder after + // the preempted turn has already been transitioned to IDLE. + this.activeTurnThinkingOverride = null; const waiters = this.idleWaiters; this.idleWaiters = []; for (const resolve of waiters) { @@ -5026,6 +5065,23 @@ export class AgentSession { return this.turnPhase !== TurnPhase.IDLE; } + /** + * Mid-turn thinking change: request that the active turn's next model step + * uses `level`. Returns accepted:false when no turn is active — the caller + * already persisted the setting, which covers the next turn. Last write wins + * across consecutive calls; the pending value expires silently if the turn + * ends before another model step occurs. + */ + setActiveTurnThinkingLevel(level: ThinkingLevel): { accepted: boolean } { + this.assertNotDisposed("setActiveTurnThinkingLevel"); + const holder = this.activeTurnThinkingOverride; + if (!holder) { + return { accepted: false }; + } + holder.pending = level; + return { accepted: true }; + } + isPreparingTurn(): boolean { return this.turnPhase === TurnPhase.PREPARING; } diff --git a/src/node/services/aiService.test.ts b/src/node/services/aiService.test.ts index 69ba723c46..c7b1b3c8bf 100644 --- a/src/node/services/aiService.test.ts +++ b/src/node/services/aiService.test.ts @@ -57,6 +57,10 @@ import type { import { log } from "./log"; import type { SessionUsageService } from "./sessionUsageService"; import type { ModelFallbackOptions, StreamManager } from "./streamManager"; +import type { + ActiveTurnThinkingOverride, + RebuildProviderOptionsForThinkingLevel, +} from "./thinkingOverride"; import { ExperimentsService } from "./experimentsService"; import type { DevToolsService } from "./devToolsService"; import { TelemetryService } from "@/node/services/telemetryService"; @@ -2955,6 +2959,161 @@ describe("AIService.streamMessage compaction boundary slicing", () => { }) ); }); + + describe("mid-turn thinking override rebuild closure", () => { + const START_STREAM_THINKING_OVERRIDE_STATE_INDEX = 26; + const START_STREAM_THINKING_REBUILD_INDEX = 27; + + function getThinkingOverrideStartStreamArgs(harness: StreamMessageHarness): { + holder: unknown; + rebuild: RebuildProviderOptionsForThinkingLevel; + } { + expect(harness.startStreamCalls).toHaveLength(1); + const call = harness.startStreamCalls[0]; + if (!call) { + throw new Error("Expected streamManager.startStream call arguments"); + } + const holder = call[START_STREAM_THINKING_OVERRIDE_STATE_INDEX]; + const rebuild = call[START_STREAM_THINKING_REBUILD_INDEX]; + expect(typeof rebuild).toBe("function"); + return { holder, rebuild: rebuild as RebuildProviderOptionsForThinkingLevel }; + } + + it("threads the session holder by reference and rebuilds options through the same pipeline", async () => { + using muxHome = new DisposableTempDir("ai-service-thinking-override"); + const projectPath = path.join(muxHome.path, "project"); + await fs.mkdir(projectPath, { recursive: true }); + + const workspaceId = "workspace-thinking-override"; + const metadata = createLocalWorkspaceMetadata(workspaceId, projectPath); + const harness = createHarness(muxHome.path, metadata, { + useRequestedModelString: true, + canonicalProviderName: "anthropic", + }); + + const sessionHolder: ActiveTurnThinkingOverride = {}; + const result = await harness.service.streamMessage({ + messages: [createMuxMessage("latest-user", "user", "hello")], + workspaceId, + // Budget-token Anthropic model (no adaptive effort): level changes show + // up as thinking.budgetTokens differences. + modelString: "anthropic:claude-sonnet-4-5", + thinkingLevel: "low", + minThinkingLevel: "off", + activeTurnThinkingOverride: sessionHolder, + }); + expect(result.success).toBe(true); + + const { holder, rebuild } = getThinkingOverrideStartStreamArgs(harness); + // Same object: AgentSession's setter writes must be visible to prepareStep. + expect(holder).toBe(sessionHolder); + + // No-op: requested level equals the current effective level. + expect(rebuild("low")).toBeNull(); + + // Real transition: rebuilt provider options reflect the new level. + const rebuilt = rebuild("high"); + expect(rebuilt?.effectiveLevel).toBe("high"); + const anthropic = rebuilt?.providerOptions.anthropic as + | { thinking?: { type: string; budgetTokens?: number } } + | undefined; + expect(anthropic?.thinking).toEqual({ type: "enabled", budgetTokens: 20000 }); + + // The closure diffs against the LIVE level, not the send-time one: + // repeating the applied level is now a no-op. + expect(rebuild("high")).toBeNull(); + }); + + it("clamps mid-turn requests against the session-provided floor", async () => { + using muxHome = new DisposableTempDir("ai-service-thinking-floor"); + const projectPath = path.join(muxHome.path, "project"); + await fs.mkdir(projectPath, { recursive: true }); + + const workspaceId = "workspace-thinking-floor"; + const metadata = createLocalWorkspaceMetadata(workspaceId, projectPath); + const harness = createHarness(muxHome.path, metadata, { + useRequestedModelString: true, + canonicalProviderName: "anthropic", + }); + + const result = await harness.service.streamMessage({ + messages: [createMuxMessage("latest-user", "user", "hello")], + workspaceId, + modelString: KNOWN_MODELS.SONNET.id, + thinkingLevel: "medium", + minThinkingLevel: "medium", + activeTurnThinkingOverride: {}, + }); + expect(result.success).toBe(true); + + const { rebuild } = getThinkingOverrideStartStreamArgs(harness); + // Below-floor requests clamp up to the floor, which equals the current + // level here — so they must be treated as no-ops, not as downgrades. + expect(rebuild("off")).toBeNull(); + expect(rebuild("low")).toBeNull(); + // Above-floor requests still apply. + expect(rebuild("high")?.effectiveLevel).toBe("high"); + }); + + it("applies Anthropic native-xhigh transitions as plain provider-option rebuilds", async () => { + using muxHome = new DisposableTempDir("ai-service-thinking-xhigh"); + const projectPath = path.join(muxHome.path, "project"); + await fs.mkdir(projectPath, { recursive: true }); + + const workspaceId = "workspace-thinking-xhigh"; + const metadata = createLocalWorkspaceMetadata(workspaceId, projectPath); + const harness = createHarness(muxHome.path, metadata, { + useRequestedModelString: true, + canonicalProviderName: "anthropic", + }); + + const result = await harness.service.streamMessage({ + messages: [createMuxMessage("latest-user", "user", "hello")], + workspaceId, + modelString: "anthropic:claude-opus-4-7", + thinkingLevel: "high", + activeTurnThinkingOverride: {}, + }); + expect(result.success).toBe(true); + + const { rebuild } = getThinkingOverrideStartStreamArgs(harness); + const rebuilt = rebuild("xhigh"); + expect(rebuilt?.effectiveLevel).toBe("xhigh"); + const anthropic = rebuilt?.providerOptions.anthropic as + | { effort?: string; thinking?: unknown } + | undefined; + // Post-wire-hack: the native effort flows directly via provider options. + expect(anthropic?.effort).toBe("xhigh"); + expect(anthropic?.thinking).toEqual({ type: "adaptive", display: "summarized" }); + }); + + it("skips the grok-4-1-fast off<->on transition (model-instance swap)", async () => { + using muxHome = new DisposableTempDir("ai-service-thinking-grok"); + const projectPath = path.join(muxHome.path, "project"); + await fs.mkdir(projectPath, { recursive: true }); + + const workspaceId = "workspace-thinking-grok"; + const metadata = createLocalWorkspaceMetadata(workspaceId, projectPath); + const harness = createHarness(muxHome.path, metadata, { + useRequestedModelString: true, + canonicalProviderName: "xai" as ProviderName, + }); + + const result = await harness.service.streamMessage({ + messages: [createMuxMessage("latest-user", "user", "hello")], + workspaceId, + modelString: "xai:grok-4-1-fast", + thinkingLevel: "off", + activeTurnThinkingOverride: {}, + }); + expect(result.success).toBe(true); + + const { rebuild } = getThinkingOverrideStartStreamArgs(harness); + // off -> high selects a different model instance at creation time; the + // in-flight stream cannot express it via provider options. + expect(rebuild("high")).toBeNull(); + }); + }); }); describe("AIService.streamMessage multi-project trust gating", () => { diff --git a/src/node/services/aiService.ts b/src/node/services/aiService.ts index 2d915016db..377e7d6297 100644 --- a/src/node/services/aiService.ts +++ b/src/node/services/aiService.ts @@ -116,9 +116,14 @@ import { } from "@/common/types/thinking"; import { enforceThinkingPolicy, + isXaiGrokFastVariantSwap, resolveEffectiveThinkingLevel, resolveMinimumThinkingLevel, } from "@/common/utils/thinking/policy"; +import type { + ActiveTurnThinkingOverride, + RebuildProviderOptionsForThinkingLevel, +} from "@/node/services/thinkingOverride"; import type { ErrorEvent, @@ -269,6 +274,19 @@ export interface StreamMessageOptions { hasQueuedMessages?: (dispatchMode?: "tool-end" | "turn-end") => boolean; muxMetadata?: MuxMessageMetadata; openaiTruncationModeOverride?: "auto" | "disabled"; + /** + * Model floor already resolved by AgentSession (config.json + * minThinkingLevelByModel → resolveMinimumThinkingLevel). Passed down so + * mid-turn overrides clamp against the same floor as the send-time level; + * internal callers may omit it (re-resolved from defaults). + */ + minThinkingLevel?: ThinkingLevel; + /** + * Session-owned per-turn holder for mid-turn thinking-level overrides. + * When absent (compaction, sub-agent paths), the feature is inert for the + * stream. See src/node/services/thinkingOverride.ts. + */ + activeTurnThinkingOverride?: ActiveTurnThinkingOverride; } /** @@ -1027,6 +1045,8 @@ export class AIService extends EventEmitter { hasQueuedMessages, openaiTruncationModeOverride, muxMetadata, + minThinkingLevel: providedMinThinkingLevel, + activeTurnThinkingOverride, } = opts; // Support interrupts during startup (before StreamManager emits stream-start). // We register an AbortController up-front and let stopStream() abort it. @@ -2416,8 +2436,7 @@ export class AIService extends EventEmitter { effectiveMuxProviderOptions, workspaceId, this.providerService.getConfig(), - routeProvider, - effectiveThinkingLevel + routeProvider ); // --- Model parameter overrides from providers.jsonc --- @@ -2433,21 +2452,29 @@ export class AIService extends EventEmitter { // Recursive merge within the provider namespace preserves non-conflicting nested // subfields (e.g., user reasoning.max_tokens alongside Mux reasoning.enabled). // Mux-built values win on leaf conflicts for safety of thinking/reasoning/cache. + // Shared by the initial build and mid-turn thinking-level rebuilds so both + // produce identically-shaped options. const providerOptionsNamespaceKey = resolveProviderOptionsNamespaceKey( canonicalProviderName, routeProvider ); - const muxProviderNamespace = (providerOptions as Record)?.[ - providerOptionsNamespaceKey - ]; - const mergedProviderOptions = resolvedOverrides.providerExtras - ? { - ...providerOptions, - [providerOptionsNamespaceKey]: isPlainObject(muxProviderNamespace) - ? mergeProviderExtrasUnderMux(resolvedOverrides.providerExtras, muxProviderNamespace) - : resolvedOverrides.providerExtras, - } - : providerOptions; + const mergeModelParameterExtras = ( + builtOptions: Record + ): Record => { + if (!resolvedOverrides.providerExtras) { + return builtOptions; + } + const muxProviderNamespace = builtOptions[providerOptionsNamespaceKey]; + return { + ...builtOptions, + [providerOptionsNamespaceKey]: isPlainObject(muxProviderNamespace) + ? mergeProviderExtrasUnderMux(resolvedOverrides.providerExtras, muxProviderNamespace) + : resolvedOverrides.providerExtras, + }; + }; + const mergedProviderOptions = mergeModelParameterExtras( + providerOptions as Record + ); recordStartupPhaseTiming("buildRequestConfigMs", buildRequestConfigStartedAt); @@ -2458,6 +2485,63 @@ export class AIService extends EventEmitter { ); } + // --- Mid-turn thinking-level override support --- + // Floor resolved by AgentSession when present; internal callers fall back + // to the model default so clamping never loosens below policy. + const minThinkingLevel = + providedMinThinkingLevel ?? + resolveMinimumThinkingLevel(modelString, undefined, this.providerService.getConfig()); + // Rebuilds provider options for a new level using the exact same pipeline + // as the initial build (policy clamp → resolveEffectiveThinkingLevel → + // buildProviderOptions → providers.jsonc extras merge). Consumed by + // StreamManager's prepareStep; `null` ⇒ skip (no-op or model-swap level). + const currentEffectiveLevelRef = { current: effectiveThinkingLevel }; + const rebuildProviderOptionsForThinkingLevel: RebuildProviderOptionsForThinkingLevel = ( + level + ) => { + const clamped = enforceThinkingPolicy( + modelString, + level, + minThinkingLevel, + this.providerService.getConfig() + ); + const effective = resolveEffectiveThinkingLevel( + modelString, + clamped, + this.providerService.getConfig() + ); + if (effective === currentEffectiveLevelRef.current) { + return null; + } + // off ↔ non-off on grok-4-1-fast selects a different model instance — + // not expressible via provider options on the in-flight stream. + if ( + isXaiGrokFastVariantSwap( + canonicalModelString, + currentEffectiveLevelRef.current, + effective + ) + ) { + return null; + } + const rebuilt = buildProviderOptions( + modelString, + effective, + providerRequestMessages, + (id) => this.streamManager.isResponseIdLost(id), + effectiveMuxProviderOptions, + workspaceId, + truncationMode, + this.providerService.getConfig(), + routeProvider, + promptCacheScope, + reasoningMode + ); + const merged = mergeModelParameterExtras(rebuilt as Record); + currentEffectiveLevelRef.current = effective; + return { effectiveLevel: effective, providerOptions: merged }; + }; + // Debug dump: Log the complete LLM request when MUX_DEBUG_LLM_REQUEST is set if (process.env.MUX_DEBUG_LLM_REQUEST === "1") { log.info( @@ -2577,16 +2661,19 @@ export class AIService extends EventEmitter { // clamped level may violate the next model's policy/floor (the // providerOptions builders require a policy-valid level, e.g. an // "off" source level on a fixed-effort model like gpt-5-pro). + // A mid-turn thinking override folded in by StreamManager wins + // over the send-time level. + const nextMinThinkingLevel = resolveMinimumThinkingLevel( + nextModelString, + this.config.loadConfigOrDefault().minThinkingLevelByModel?.[ + normalizeToCanonical(nextModelString) + ], + this.providerService.getConfig() + ); const nextThinkingLevel = enforceThinkingPolicy( nextModelString, - effectiveThinkingLevel, - resolveMinimumThinkingLevel( - nextModelString, - this.config.loadConfigOrDefault().minThinkingLevelByModel?.[ - normalizeToCanonical(nextModelString) - ], - this.providerService.getConfig() - ), + prepareOptions?.thinkingLevelOverride ?? effectiveThinkingLevel, + nextMinThinkingLevel, this.providerService.getConfig() ); @@ -2729,8 +2816,7 @@ export class AIService extends EventEmitter { effectiveMuxProviderOptions, workspaceId, this.providerService.getConfig(), - next.routeProvider, - nextThinkingLevel + next.routeProvider ); if (pendingRunMetadataId != null) { // Keep DevTools run correlation on fallback requests too. @@ -2750,20 +2836,76 @@ export class AIService extends EventEmitter { next.canonicalProviderName, next.routeProvider ); - const nextMuxNamespace = (nextProviderOptions as Record)?.[ - nextNamespaceKey - ]; - const nextMergedProviderOptions = nextOverrides.providerExtras - ? { - ...nextProviderOptions, - [nextNamespaceKey]: isPlainObject(nextMuxNamespace) - ? mergeProviderExtrasUnderMux( - nextOverrides.providerExtras, - nextMuxNamespace - ) - : nextOverrides.providerExtras, + // Mirrors mergeModelParameterExtras for the fallback model; + // shared by this baseline build and mid-turn rebuilds below. + const mergeNextModelParameterExtras = ( + builtOptions: Record + ): Record => { + if (!nextOverrides.providerExtras) { + return builtOptions; + } + const nextMuxNamespace = builtOptions[nextNamespaceKey]; + return { + ...builtOptions, + [nextNamespaceKey]: isPlainObject(nextMuxNamespace) + ? mergeProviderExtrasUnderMux( + nextOverrides.providerExtras, + nextMuxNamespace + ) + : nextOverrides.providerExtras, + }; + }; + const nextMergedProviderOptions = mergeNextModelParameterExtras( + nextProviderOptions as Record + ); + + // Rebuild closure bound to the FALLBACK model so mid-turn + // thinking changes keep working after the hop. + const nextCurrentEffectiveLevelRef = { current: nextThinkingLevel }; + const rebuildNextProviderOptionsForThinkingLevel: RebuildProviderOptionsForThinkingLevel = + (level) => { + const clamped = enforceThinkingPolicy( + next.canonicalModelString, + level, + nextMinThinkingLevel, + this.providerService.getConfig() + ); + const effective = resolveEffectiveThinkingLevel( + next.canonicalModelString, + clamped, + this.providerService.getConfig() + ); + if (effective === nextCurrentEffectiveLevelRef.current) { + return null; + } + if ( + isXaiGrokFastVariantSwap( + next.canonicalModelString, + nextCurrentEffectiveLevelRef.current, + effective + ) + ) { + return null; } - : nextProviderOptions; + const rebuilt = buildProviderOptions( + next.canonicalModelString, + effective, + nextProviderRequestMessages, + (id) => this.streamManager.isResponseIdLost(id), + effectiveMuxProviderOptions, + workspaceId, + truncationMode, + this.providerService.getConfig(), + next.routeProvider, + promptCacheScope, + reasoningMode + ); + const merged = mergeNextModelParameterExtras( + rebuilt as Record + ); + nextCurrentEffectiveLevelRef.current = effective; + return { effectiveLevel: effective, providerOptions: merged }; + }; return Ok({ model: next.model, @@ -2776,6 +2918,8 @@ export class AIService extends EventEmitter { callSettingsOverrides: nextOverrides.standard, anthropicCacheTtl: effectiveMuxProviderOptions.anthropic?.cacheTtl ?? undefined, thinkingLevel: nextThinkingLevel, + rebuildProviderOptionsForThinkingLevel: + rebuildNextProviderOptionsForThinkingLevel, initialMetadataPatch: { routedThroughGateway: next.routedThroughGateway, ...(next.routeProvider != null ? { routeProvider: next.routeProvider } : {}), @@ -2843,7 +2987,9 @@ export class AIService extends EventEmitter { : undefined, runtimeTempDir, modelFallback, - toolSearchRuntime?.state + toolSearchRuntime?.state, + activeTurnThinkingOverride, + rebuildProviderOptionsForThinkingLevel ); recordStartupPhaseTiming("startStreamMs", startStreamStartedAt); diff --git a/src/node/services/providerModelFactory.test.ts b/src/node/services/providerModelFactory.test.ts index 09e61dcacb..54b3595a9f 100644 --- a/src/node/services/providerModelFactory.test.ts +++ b/src/node/services/providerModelFactory.test.ts @@ -21,7 +21,6 @@ import { resolveOpenAIWebSocketResponsesUrl, wrapFetchWithAnthropicCacheControl, } from "./providerModelFactory"; -import { MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER } from "@/common/utils/ai/providerOptions"; import { hasLanguageModelCleanup } from "./languageModelCleanup"; import type { DevToolsService } from "./devToolsService"; import { CodexOauthService } from "./codexOauthService"; @@ -1634,82 +1633,44 @@ function parseSentBody(call: CapturedFetchCall): Record { return JSON.parse(call.init.body as string) as Record; } -describe("wrapFetchWithAnthropicCacheControl — Opus 4.7+ / Sonnet 5+ wire transforms", () => { - for (const model of ["claude-opus-4-7", "claude-opus-4-8", "claude-sonnet-5"] as const) { - it(`injects thinking.display=summarized for ${model} adaptive thinking`, async () => { - const { calls, fakeFetch } = createCapturingFetch(); - const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch); - const body = JSON.stringify({ - model, - thinking: { type: "adaptive" }, - output_config: { effort: "medium" }, - }); - await wrapped("https://api.anthropic.com/v1/messages", { method: "POST", body }); - expect(calls.length).toBe(1); - const sent = parseSentBody(calls[0]); - expect(sent.thinking).toEqual({ type: "adaptive", display: "summarized" }); - }); - } - - it("preserves a user-supplied display value on Opus 4.7", async () => { +// Effort "xhigh" and thinking.display flow through the SDK directly as of +// @ai-sdk/anthropic 4.0.11 (see buildProviderOptions), so the wrapper must NOT +// rewrite reasoning fields — it only normalizes cache_control. +describe("wrapFetchWithAnthropicCacheControl — reasoning fields pass through unchanged", () => { + it("passes native xhigh effort and summarized display through on the direct body", async () => { const { calls, fakeFetch } = createCapturingFetch(); - const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch); + const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch, null, { + injectCacheControl: false, + }); const body = JSON.stringify({ model: "claude-opus-4-7", - thinking: { type: "adaptive", display: "omitted" }, + thinking: { type: "adaptive", display: "summarized" }, + output_config: { effort: "xhigh" }, }); await wrapped("https://api.anthropic.com/v1/messages", { method: "POST", body }); - const sent = parseSentBody(calls[0]) as { thinking: { display: string } }; - expect(sent.thinking.display).toBe("omitted"); + expect(calls.length).toBe(1); + const sent = parseSentBody(calls[0]); + expect(sent.thinking).toEqual({ type: "adaptive", display: "summarized" }); + expect(sent.output_config).toEqual({ effort: "xhigh" }); }); - it("rewrites output_config.effort to xhigh when override header is present", async () => { + it("does not inject display or rewrite effort for adaptive requests without them", async () => { const { calls, fakeFetch } = createCapturingFetch(); - const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch); - const body = JSON.stringify({ - model: "claude-opus-4-7", - thinking: { type: "adaptive" }, - output_config: { effort: "max" }, - }); - await wrapped("https://api.anthropic.com/v1/messages", { - method: "POST", - body, - headers: { [MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER]: "xhigh" }, + const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch, null, { + injectCacheControl: false, }); - const sent = parseSentBody(calls[0]) as { output_config: { effort: string } }; - expect(sent.output_config.effort).toBe("xhigh"); - // Override header is stripped before forwarding - const outHeaders = new Headers(calls[0].init.headers); - expect(outHeaders.get(MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER)).toBeNull(); - }); - - it("does not inject display for Opus 4.6 adaptive thinking", async () => { - const { calls, fakeFetch } = createCapturingFetch(); - const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch); const body = JSON.stringify({ model: "claude-opus-4-6", thinking: { type: "adaptive" }, + output_config: { effort: "max" }, }); await wrapped("https://api.anthropic.com/v1/messages", { method: "POST", body }); const sent = parseSentBody(calls[0]); expect(sent.thinking).toEqual({ type: "adaptive" }); + expect(sent.output_config).toEqual({ effort: "max" }); }); - it("does not inject display when thinking is disabled", async () => { - const { calls, fakeFetch } = createCapturingFetch(); - const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch); - const body = JSON.stringify({ - model: "claude-opus-4-7", - thinking: { type: "disabled" }, - }); - await wrapped("https://api.anthropic.com/v1/messages", { method: "POST", body }); - const sent = parseSentBody(calls[0]); - expect(sent.thinking).toEqual({ type: "disabled" }); - }); -}); - -describe("wrapFetchWithAnthropicCacheControl — gateway (AI SDK) body shape", () => { - it("injects display=summarized into providerOptions.anthropic.thinking for Opus 4.7 via gateway", async () => { + it("passes gateway (AI SDK) body providerOptions through unchanged", async () => { const { calls, fakeFetch } = createCapturingFetch(); const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch, null, { injectCacheControl: false, @@ -1717,7 +1678,7 @@ describe("wrapFetchWithAnthropicCacheControl — gateway (AI SDK) body shape", ( const body = JSON.stringify({ prompt: [{ role: "user", content: [{ type: "text", text: "hi" }] }], providerOptions: { - anthropic: { thinking: { type: "adaptive" }, effort: "medium" }, + anthropic: { thinking: { type: "adaptive", display: "summarized" }, effort: "xhigh" }, }, }); await wrapped("https://gateway.example.com/v1/language-model", { @@ -1726,36 +1687,12 @@ describe("wrapFetchWithAnthropicCacheControl — gateway (AI SDK) body shape", ( headers: { "ai-model-id": "anthropic/claude-opus-4-7" }, }); const sent = parseSentBody(calls[0]) as { - providerOptions: { anthropic: { thinking: unknown } }; + providerOptions: { anthropic: { thinking: unknown; effort: string } }; }; expect(sent.providerOptions.anthropic.thinking).toEqual({ type: "adaptive", display: "summarized", }); - }); - - it("rewrites providerOptions.anthropic.effort to xhigh via gateway", async () => { - const { calls, fakeFetch } = createCapturingFetch(); - const wrapped = wrapFetchWithAnthropicCacheControl(fakeFetch, null, { - injectCacheControl: false, - }); - const body = JSON.stringify({ - prompt: [{ role: "user", content: [{ type: "text", text: "hi" }] }], - providerOptions: { - anthropic: { thinking: { type: "adaptive" }, effort: "max" }, - }, - }); - await wrapped("https://gateway.example.com/v1/language-model", { - method: "POST", - body, - headers: { - "ai-model-id": "anthropic/claude-opus-4-7", - [MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER]: "xhigh", - }, - }); - const sent = parseSentBody(calls[0]) as { - providerOptions: { anthropic: { effort: string } }; - }; expect(sent.providerOptions.anthropic.effort).toBe("xhigh"); }); }); diff --git a/src/node/services/providerModelFactory.ts b/src/node/services/providerModelFactory.ts index 959e2e5620..d7404556f9 100644 --- a/src/node/services/providerModelFactory.ts +++ b/src/node/services/providerModelFactory.ts @@ -3,7 +3,7 @@ import { createOpenAICompatible } from "@ai-sdk/openai-compatible"; import type { XaiProviderOptions } from "@ai-sdk/xai"; import { fromNodeProviderChain } from "@aws-sdk/credential-providers"; import { wrapLanguageModel, type LanguageModel } from "ai"; -import { anthropicSupportsNativeXhigh, type ThinkingLevel } from "@/common/types/thinking"; +import type { ThinkingLevel } from "@/common/types/thinking"; import { Ok, Err } from "@/common/types/result"; import type { Result } from "@/common/types/result"; import type { SendMessageError } from "@/common/types/errors"; @@ -49,10 +49,7 @@ import { } from "@/node/services/languageModelCleanup"; import { createOpenAIWebSocketTransportFetch } from "@/node/services/openAIWebSocketTransportFetch"; import { log } from "@/node/services/log"; -import { - MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER, - resolveProviderOptionsNamespaceKey, -} from "@/common/utils/ai/providerOptions"; +import { resolveProviderOptionsNamespaceKey } from "@/common/utils/ai/providerOptions"; import { resolveRoute, type RouteContext } from "@/common/routing"; import { getExplicitGatewayPrefix as getExplicitGatewayProvider, @@ -337,67 +334,9 @@ export function wrapFetchWithAnthropicCacheControl( return baseFetch(input, init); } - // Detect and strip Mux-internal headers before forwarding. - const incomingHeaders = new Headers(init.headers); - const effortOverride = incomingHeaders.get(MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER); - let headersModified = false; - if (effortOverride != null) { - incomingHeaders.delete(MUX_ANTHROPIC_EFFORT_OVERRIDE_HEADER); - headersModified = true; - } - try { const json = JSON.parse(init.body) as Record; - // Native-xhigh adaptive-thinking models (Opus 4.7+ and Sonnet 5+) require - // `thinking.display: "summarized"` to return thinking content in the response. - // Inject it on the wire for any adaptive thinking request targeting those models. - // See https://platform.claude.com/docs/en/build-with-claude/adaptive-thinking#summarized-thinking - // - // Two body shapes to handle: - // - Direct Anthropic API: `{ model, thinking: { type: "adaptive" }, output_config }` - // - AI SDK gateway (mux-gateway): `{ prompt, providerOptions: { anthropic: { thinking } } }` - // with the model exposed via the ai-model-id header. - const directModel = typeof json.model === "string" ? json.model : ""; - const headerModelId = incomingHeaders.get("ai-model-id") ?? ""; - // Reuse the shared native-xhigh detector so the wire-level regex stays in - // one place (src/common/types/thinking.ts) — it also normalizes provider - // prefixes (e.g., `anthropic/claude-opus-4-7`, `anthropic/claude-sonnet-5`). - const targetsNativeXhighModel = [directModel, headerModelId].some( - anthropicSupportsNativeXhigh - ); - - const directThinking = isRecord(json.thinking) ? json.thinking : undefined; - const providerOpts = isRecord(json.providerOptions) ? json.providerOptions : undefined; - const anthropicOpts = - providerOpts && isRecord(providerOpts.anthropic) ? providerOpts.anthropic : undefined; - const gatewayThinking = - anthropicOpts && isRecord(anthropicOpts.thinking) ? anthropicOpts.thinking : undefined; - - if (targetsNativeXhighModel) { - if (directThinking?.type === "adaptive") { - directThinking.display ??= "summarized"; - log.debug("Anthropic wrapper: injected thinking.display=summarized (direct body)"); - } - if (gatewayThinking?.type === "adaptive") { - gatewayThinking.display ??= "summarized"; - log.debug("Anthropic wrapper: injected thinking.display=summarized (gateway body)"); - } - } - - // Native-xhigh Anthropic models use an "xhigh" effort level. The @ai-sdk/anthropic - // Zod schema still rejects "xhigh", so providerOptions sends "max" through - // the SDK and we rewrite effort here based on the Mux-internal override - // header — for both direct and gateway body shapes. - if (effortOverride) { - if (isRecord(json.output_config)) { - json.output_config.effort = effortOverride; - } - if (anthropicOpts) { - anthropicOpts.effort = effortOverride; - } - } - // Inject cache_control on the last tool if tools array exists. // If the SDK already populated cache_control, preserve it but override ttl // when a higher-level cacheTtl is configured. @@ -444,15 +383,11 @@ export function wrapFetchWithAnthropicCacheControl( // Update body with modified JSON const newBody = JSON.stringify(json); - const outHeaders = incomingHeaders; + const outHeaders = new Headers(init.headers); outHeaders.delete("content-length"); // Body size changed return baseFetch(input, { ...init, headers: outHeaders, body: newBody }); } catch { - // If JSON parsing fails we can't touch the body, but we still need to - // strip Mux-internal headers if any were present so Anthropic doesn't see them. - if (headersModified) { - return baseFetch(input, { ...init, headers: incomingHeaders }); - } + // If JSON parsing fails we can't touch the body; forward unchanged. return baseFetch(input, init); } }; @@ -1202,8 +1137,7 @@ export class ProviderModelFactory { // Use getProviderFetch to preserve any user-configured custom fetch (e.g., proxies) const baseFetch = getProviderFetch(providerConfig); const disableBeta = muxProviderOptions?.anthropic?.disableBetaFeatures === true; - // Always wrap to apply Opus 4.7+ wire-level transforms (display + effort - // override); skip cache_control injection when beta features are off. + // Wrap for cache_control normalization; skip injection when beta features are off. const fetchWithCacheControl = wrapFetchWithAnthropicCacheControl( baseFetch, effectiveAnthropicCacheTtl, @@ -1716,8 +1650,8 @@ export class ProviderModelFactory { const baseFetch = getProviderFetch(providerConfig); const isAnthropicModel = modelId.startsWith("anthropic/"); const disableBeta = muxProviderOptions?.anthropic?.disableBetaFeatures === true; - // For Anthropic models via gateway, always wrap to apply Opus 4.7+ wire - // transforms; skip cache_control injection when beta features are off. + // For Anthropic models via gateway, wrap for cache_control normalization; + // skip injection when beta features are off. const fetchWithCacheControl = isAnthropicModel ? wrapFetchWithAnthropicCacheControl(baseFetch, effectiveAnthropicCacheTtl, { injectCacheControl: !disableBeta, diff --git a/src/node/services/streamManager.test.ts b/src/node/services/streamManager.test.ts index 8a7af7d8d4..cb0af7106c 100644 --- a/src/node/services/streamManager.test.ts +++ b/src/node/services/streamManager.test.ts @@ -15,6 +15,10 @@ import { Ok, Err } from "@/common/types/result"; import type { ToolPolicy } from "@/common/utils/tools/toolPolicy"; import type { ToolSearchStreamState } from "@/common/utils/tools/toolCatalog"; import { StreamManager, type ModelFallbackPrepareOptions } from "./streamManager"; +import type { + ActiveTurnThinkingOverride, + RebuildProviderOptionsForThinkingLevel, +} from "./thinkingOverride"; import { stripEncryptedContent } from "@/node/utils/messages/stripEncryptedContent"; import * as aiSdk from "ai"; import { @@ -5215,3 +5219,329 @@ describe("StreamManager - tool search activeTools scoping", () => { expect(request.toolSearchState).toBe(toolSearchState); }); }); + +describe("StreamManager - mid-turn thinking override", () => { + interface OverrideRequestForTests { + model: unknown; + messages: ModelMessage[]; + system?: string; + providerOptions?: Record; + thinkingOverrideState?: ActiveTurnThinkingOverride; + rebuildProviderOptionsForThinkingLevel?: RebuildProviderOptionsForThinkingLevel; + } + + type BuildStreamRequestConfig = (...args: unknown[]) => OverrideRequestForTests; + type CreateStreamResult = ( + request: OverrideRequestForTests, + abortController: AbortController + ) => unknown; + type CapturedPrepareStep = (options: { messages: ModelMessage[] }) => Promise< + | { + messages?: ModelMessage[]; + activeTools?: string[]; + providerOptions?: Record; + } + | undefined + >; + + const model = createAnthropic({ apiKey: "test" })("claude-sonnet-4-5"); + const messages: ModelMessage[] = [{ role: "user", content: "hello" }]; + + function getRequestHelpers(streamManager: StreamManager): { + buildRequestConfig: BuildStreamRequestConfig; + createStreamResult: CreateStreamResult; + } { + const buildRequestConfig = Reflect.get(streamManager, "buildStreamRequestConfig") as + | ((...args: unknown[]) => OverrideRequestForTests) + | undefined; + const createStreamResultMethod = Reflect.get(streamManager, "createStreamResult") as + | CreateStreamResult + | undefined; + expect(typeof buildRequestConfig).toBe("function"); + expect(typeof createStreamResultMethod).toBe("function"); + if (!buildRequestConfig || !createStreamResultMethod) { + throw new Error("Expected StreamManager private helpers to exist"); + } + return { + buildRequestConfig: (...args) => buildRequestConfig.apply(streamManager, args), + createStreamResult: (request, abortController) => + createStreamResultMethod.call(streamManager, request, abortController), + }; + } + + function setupStreamTextSpy() { + return spyOn(aiSdk, "streamText").mockReturnValue({ + fullStream: (async function* asyncGenerator() { + yield* [] as unknown[]; + await Promise.resolve(); + })(), + usage: Promise.resolve(undefined), + providerMetadata: Promise.resolve(undefined), + totalUsage: Promise.resolve(undefined), + steps: Promise.resolve([]), + } as unknown as ReturnType); + } + + function capturePrepareStep( + streamTextSpy: ReturnType + ): CapturedPrepareStep { + const prepareStep = streamTextSpy.mock.calls[0]?.[0]?.prepareStep as + | CapturedPrepareStep + | undefined; + expect(typeof prepareStep).toBe("function"); + if (!prepareStep) { + throw new Error("Expected prepareStep to be captured"); + } + return prepareStep; + } + + afterEach(() => { + mock.restore(); + }); + + test("applies a pending override in place: same object identity, old keys deleted, sink invoked", async () => { + const streamManager = new StreamManager(historyService); + const { createStreamResult } = getRequestHelpers(streamManager); + const streamTextSpy = setupStreamTextSpy(); + + const appliedLevels: string[] = []; + const state: ActiveTurnThinkingOverride = { + pending: "high", + onApplied: (level) => appliedLevels.push(level), + }; + const originalProviderOptions: Record = { + anthropic: { effort: "low", thinking: { type: "enabled", budgetTokens: 4000 } }, + staleNamespace: { key: "must-be-deleted" }, + }; + const rebuilt = { anthropic: { effort: "high", thinking: { type: "enabled" } } }; + const rebuild = mock((level: string) => + level === "high" ? { effectiveLevel: "high" as const, providerOptions: rebuilt } : null + ); + + const request: OverrideRequestForTests = { + model, + messages, + system: "system", + providerOptions: originalProviderOptions, + thinkingOverrideState: state, + rebuildProviderOptionsForThinkingLevel: + rebuild as unknown as RebuildProviderOptionsForThinkingLevel, + }; + createStreamResult(request, new AbortController()); + const prepareStep = capturePrepareStep(streamTextSpy); + + const step = await prepareStep({ messages }); + // Rebuilt options are returned for the step (defense in depth) … + expect(step?.providerOptions).toEqual(rebuilt); + // … and the live request object is replaced IN PLACE (same identity), + // deleting keys the SDK's deep-merge could never remove. + expect(request.providerOptions).toBe(originalProviderOptions); + expect(request.providerOptions).toEqual(rebuilt); + expect("staleNamespace" in originalProviderOptions).toBe(false); + // Consume-once bookkeeping + metadata sink. + expect(state.pending).toBeUndefined(); + expect(state.applied).toBe("high"); + expect(appliedLevels).toEqual(["high"]); + // Without a new pending value the next step is a no-op again. + expect(await prepareStep({ messages })).toBeUndefined(); + expect(rebuild).toHaveBeenCalledTimes(1); + }); + + test("clears pending without touching options when the rebuild reports not-applicable", async () => { + const streamManager = new StreamManager(historyService); + const { createStreamResult } = getRequestHelpers(streamManager); + const streamTextSpy = setupStreamTextSpy(); + + const state: ActiveTurnThinkingOverride = { pending: "off" }; + const originalProviderOptions: Record = { xai: { some: "config" } }; + const rebuild = mock(() => null); + + const request: OverrideRequestForTests = { + model, + messages, + system: "system", + providerOptions: originalProviderOptions, + thinkingOverrideState: state, + rebuildProviderOptionsForThinkingLevel: + rebuild as unknown as RebuildProviderOptionsForThinkingLevel, + }; + createStreamResult(request, new AbortController()); + const prepareStep = capturePrepareStep(streamTextSpy); + + expect(await prepareStep({ messages })).toBeUndefined(); + expect(request.providerOptions).toEqual({ xai: { some: "config" } }); + // Consume-once: a skipped application must not retry on every later step. + expect(state.pending).toBeUndefined(); + expect(state.applied).toBeUndefined(); + expect(await prepareStep({ messages })).toBeUndefined(); + expect(rebuild).toHaveBeenCalledTimes(1); + }); + + test("stays byte-identical to the feature-off behavior when no override state exists", async () => { + const streamManager = new StreamManager(historyService); + const { createStreamResult } = getRequestHelpers(streamManager); + const streamTextSpy = setupStreamTextSpy(); + + createStreamResult({ model, messages, system: "system" }, new AbortController()); + const prepareStep = capturePrepareStep(streamTextSpy); + expect(await prepareStep({ messages })).toBeUndefined(); + }); + + test("buildStreamRequestConfig normalizes providerOptions to a stable mutable object only when a rebuild closure exists", () => { + const streamManager = new StreamManager(historyService); + const { buildRequestConfig, createStreamResult } = getRequestHelpers(streamManager); + const streamTextSpy = setupStreamTextSpy(); + + const state: ActiveTurnThinkingOverride = {}; + const rebuild: RebuildProviderOptionsForThinkingLevel = () => null; + + // Without the closure, an absent providerOptions stays absent (no behavior change). + const plainRequest = buildRequestConfig( + model, + KNOWN_MODELS.SONNET.id, + messages, + "system", + undefined, // routeProvider + undefined, // tools + undefined, // providerOptions + undefined, // maxOutputTokens + undefined, // callSettingsOverrides + undefined, // toolPolicy + undefined, // hasQueuedMessages + undefined, // headers + undefined, // anthropicCacheTtlOverride + undefined, // onChunk + undefined, // onStepMessages + undefined // toolSearchState + ); + expect(plainRequest.providerOptions).toBeUndefined(); + + // With the closure, undefined normalizes to a mutable object whose identity + // is exactly what streamText() captures — otherwise in-place mutation at + // prepareStep time would be unobservable to the SDK's per-step merge. + const request = buildRequestConfig( + model, + KNOWN_MODELS.SONNET.id, + messages, + "system", + undefined, + undefined, + undefined, // providerOptions intentionally absent + undefined, + undefined, + undefined, + undefined, + undefined, + undefined, + undefined, + undefined, + undefined, + undefined, // onToolExecutionStart + state, + rebuild + ); + expect(request.providerOptions).toEqual({}); + expect(request.thinkingOverrideState).toBe(state); + expect(request.rebuildProviderOptionsForThinkingLevel).toBe(rebuild); + + createStreamResult(request, new AbortController()); + const streamTextArgs = streamTextSpy.mock.calls[0]?.[0]; + expect(streamTextArgs?.providerOptions).toBe( + request.providerOptions as NonNullable["providerOptions"] + ); + }); + + test("model fallback folds the pending override into prepare() and rebinds holder + closure on the swapped request", async () => { + const streamManager = new StreamManager(historyService); + Reflect.set(streamManager, "tokenTracker", { + setModel: () => Promise.resolve(undefined), + countTokens: () => Promise.resolve(0), + }); + + const workspaceId = "thinking-fallback-workspace"; + const messageId = "thinking-fallback-message"; + const historySequence = 1; + const fallbackModel = KNOWN_MODELS.GPT.id; + + await appendPartialAssistantForTests(workspaceId, messageId, historySequence); + const processStreamWithCleanup = getProcessStreamWithCleanupForTests(streamManager); + + const capturedNextRequests: OverrideRequestForTests[] = []; + const createStreamResult = mock((request: OverrideRequestForTests) => { + capturedNextRequests.push(request); + return createStreamResultForTests( + (async function* () { + await Promise.resolve(); + yield { type: "text-delta", text: "fallback answer" }; + yield { type: "finish", finishReason: "stop" }; + })(), + { inputTokens: 5, outputTokens: 3, totalTokens: 8 } + ); + }); + expect(Reflect.set(streamManager, "createStreamResult", createStreamResult)).toBe(true); + + const fallbackLanguageModel = createTestLanguageModel("fallback-model"); + const fallbackRebuild: RebuildProviderOptionsForThinkingLevel = () => null; + const prepare = mock((nextModelString: string, options?: ModelFallbackPrepareOptions) => { + expect(options?.thinkingLevelOverride).toBe("high"); + return Promise.resolve( + Ok({ + model: fallbackLanguageModel, + modelString: nextModelString, + messages: [], + system: "fallback system", + tools: undefined, + thinkingLevel: "high", + rebuildProviderOptionsForThinkingLevel: fallbackRebuild, + }) + ); + }); + + // Pending override that never got a next step on the refusing stream: the + // fallback hop must not silently revert it. + const holder: ActiveTurnThinkingOverride = { pending: "high" }; + const startTime = Date.now() - 250; + const streamInfo = createStreamInfoForTests({ + streamResult: createStreamResultForTests( + (async function* () { + await Promise.resolve(); + yield { type: "finish", finishReason: "content-filter", rawFinishReason: "refusal" }; + })(), + { inputTokens: 10, outputTokens: 0, totalTokens: 10 } + ), + messageId, + startTime, + lastPartTimestamp: startTime, + model: KNOWN_MODELS.SONNET.id, + metadataModel: KNOWN_MODELS.SONNET.id, + historySequence, + runtime: LOCAL_TEST_RUNTIME, + request: { + model: createTestLanguageModel("refused-model"), + messages: [], + providerOptions: undefined, + thinkingOverrideState: holder, + }, + modelFallback: { + options: { chain: [fallbackModel], prepare }, + requestedModel: KNOWN_MODELS.SONNET.id, + refusedModels: [], + original: { maxOutputTokens: undefined }, + }, + }); + + await processStreamWithCleanup.call(streamManager, workspaceId, streamInfo, historySequence); + + expect(prepare).toHaveBeenCalledTimes(1); + // Pending was folded into the fallback baseline (consumed, kept as applied + // for potential second hops). + expect(holder.pending).toBeUndefined(); + expect(holder.applied).toBe("high"); + // The swapped request carries the SAME holder (the session setter keeps + // working) and the fallback-bound rebuild closure. + expect(createStreamResult).toHaveBeenCalledTimes(1); + const nextRequest = capturedNextRequests[0]; + expect(nextRequest?.thinkingOverrideState).toBe(holder); + expect(nextRequest?.rebuildProviderOptionsForThinkingLevel).toBe(fallbackRebuild); + }); +}); diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index 9fdb103e64..e3b41efb24 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -14,6 +14,7 @@ import { RetryError, } from "ai"; import type { LanguageModelV2Usage } from "@ai-sdk/provider"; +import type { ProviderOptions } from "@ai-sdk/provider-utils"; import type { Result } from "@/common/types/result"; import assert from "@/common/utils/assert"; import { Ok, Err } from "@/common/types/result"; @@ -32,6 +33,10 @@ import type { import type { SendMessageError, StreamErrorType } from "@/common/types/errors"; import type { MuxMetadata, MuxMessage, PersistedToolModelUsage } from "@/common/types/message"; import type { ThinkingLevel } from "@/common/types/thinking"; +import type { + ActiveTurnThinkingOverride, + RebuildProviderOptionsForThinkingLevel, +} from "@/node/services/thinkingOverride"; import type { NestedToolCall } from "@/common/orpc/schemas/message"; import type { ProvidersConfigMap } from "@/common/orpc/types"; import { @@ -190,6 +195,19 @@ interface StreamRequestConfig { * `activeTools`. Absent when the feature is inactive. */ toolSearchState?: ToolSearchStreamState; + /** + * Mid-turn thinking override holder — the SESSION'S object, by reference + * (never created inside StreamManager: a locally-created object would be + * invisible to AgentSession.setActiveTurnThinkingLevel). prepareStep + * consumes `pending` before every model step. + */ + thinkingOverrideState?: ActiveTurnThinkingOverride; + /** + * Closure built by AIService that re-runs the effective-level pipeline + * (policy clamp + resolveEffectiveThinkingLevel) and rebuilds provider + * options for the stream's model. `null` ⇒ not applicable / no-op. + */ + rebuildProviderOptionsForThinkingLevel?: RebuildProviderOptionsForThinkingLevel; } /** @@ -220,6 +238,12 @@ export interface PreparedModelFallback { thinkingLevel?: string; /** Route attribution corrections (routedThroughGateway, routeProvider, costsIncluded). */ initialMetadataPatch?: Partial; + /** + * Rebuild closure bound to the FALLBACK model so mid-turn thinking changes + * keep working after a fallback hop (the source model's closure would build + * options for the wrong model). + */ + rebuildProviderOptionsForThinkingLevel?: RebuildProviderOptionsForThinkingLevel; } export interface ModelFallbackPrepareOptions { @@ -229,6 +253,12 @@ export interface ModelFallbackPrepareOptions { * receives it so provider-message preparation cannot mutate live UI parts. */ continuation?: { assistantMessage: MuxMessage }; + /** + * Mid-turn thinking level to fold into the fallback's baseline: pending (not + * yet applied) or already-applied override from the refused stream. The + * fallback prepare re-clamps it for the next model. + */ + thinkingLevelOverride?: ThinkingLevel; } export interface ModelFallbackOptions { @@ -1573,9 +1603,16 @@ export class StreamManager extends EventEmitter { onChunk?: StreamTextOnChunk, onStepMessages?: (messages: ModelMessage[]) => void, toolSearchState?: ToolSearchStreamState, - onToolExecutionStart?: (toolCallId: string) => void + onToolExecutionStart?: (toolCallId: string) => void, + thinkingOverrideState?: ActiveTurnThinkingOverride, + rebuildProviderOptionsForThinkingLevel?: RebuildProviderOptionsForThinkingLevel ): StreamRequestConfig { - const finalProviderOptions = providerOptions; + // Mid-turn thinking overrides mutate providerOptions IN PLACE (the SDK's + // per-step deep-merge reads the object passed at streamText() time, so + // identity must stay stable). An initially-undefined value would make that + // mutation unobservable — normalize to a guaranteed mutable object. + const finalProviderOptions = + rebuildProviderOptionsForThinkingLevel != null ? (providerOptions ?? {}) : providerOptions; // Apply cache control for Anthropic models let finalMessages = messages; @@ -1648,6 +1685,8 @@ export class StreamManager extends EventEmitter { onStepMessages, toolPolicy, toolSearchState, + thinkingOverrideState, + rebuildProviderOptionsForThinkingLevel, }; } @@ -1703,6 +1742,57 @@ export class StreamManager extends EventEmitter { ]; } + /** + * Consume a pending mid-turn thinking-level override for the next step. + * + * Consume-once: `pending` is always cleared (a failed/no-op application must + * not retry on every subsequent step). On success the CONTENT of + * `request.providerOptions` is replaced in place — object identity is + * preserved because the SDK's per-step deep-merge reads the reference passed + * at streamText() time, and deep-merge alone cannot delete keys (e.g. the + * Anthropic `thinking` object when moving to "off"). + * + * Returns the rebuilt options (for the prepareStep return value) or + * undefined when there is nothing to apply. + */ + private applyPendingThinkingOverride( + request: StreamRequestConfig + ): Record | undefined { + const state = request.thinkingOverrideState; + const pending = state?.pending; + if (state == null || pending == null) { + return undefined; + } + state.pending = undefined; + const rebuild = request.rebuildProviderOptionsForThinkingLevel; + if (rebuild == null) { + return undefined; + } + const rebuilt = rebuild(pending); + if (rebuilt == null) { + log.debug("Mid-turn thinking override skipped (not applicable / no-op)", { + requestedLevel: pending, + appliedLevel: state.applied, + }); + return undefined; + } + const target = request.providerOptions; + // buildStreamRequestConfig normalizes providerOptions to a stable object + // whenever a rebuild closure is present, so target must exist here. + assert(target != null, "providerOptions must be normalized when a rebuild closure is present"); + for (const key of Object.keys(target)) { + delete target[key]; + } + Object.assign(target, rebuilt.providerOptions); + state.applied = rebuilt.effectiveLevel; + state.onApplied?.(rebuilt.effectiveLevel); + log.debug("Mid-turn thinking override applied", { + requestedLevel: pending, + effectiveLevel: rebuilt.effectiveLevel, + }); + return rebuilt.providerOptions; + } + private createStreamResult( request: StreamRequestConfig, abortController: AbortController, @@ -1740,10 +1830,25 @@ export class StreamManager extends EventEmitter { // undefined when the feature is inactive, keeping the return value // byte-identical to the pre-feature behavior. const activeTools = computeActiveToolNames(request.toolSearchState); - if (rewritten === stepMessages && activeTools === undefined) return undefined; + // Mid-turn thinking-level change: consume a pending override before + // this step's provider request is built. + const thinkingOverride = this.applyPendingThinkingOverride(request); + if ( + rewritten === stepMessages && + activeTools === undefined && + thinkingOverride === undefined + ) { + return undefined; + } return { ...(rewritten === stepMessages ? {} : { messages: rewritten }), ...(activeTools !== undefined ? { activeTools } : {}), + // Defense in depth: the in-place request mutation is authoritative + // (per-step deep-merge cannot delete keys); returning the rebuilt + // options also covers any future SDK options snapshotting. + ...(thinkingOverride !== undefined + ? { providerOptions: thinkingOverride as ProviderOptions } + : {}), }; }, onChunk: request.onChunk, @@ -1786,7 +1891,9 @@ export class StreamManager extends EventEmitter { onChunk?: StreamTextOnChunk, onStepMessages?: (messages: ModelMessage[]) => void, modelFallback?: ModelFallbackOptions, - toolSearchState?: ToolSearchStreamState + toolSearchState?: ToolSearchStreamState, + thinkingOverrideState?: ActiveTurnThinkingOverride, + rebuildProviderOptionsForThinkingLevel?: RebuildProviderOptionsForThinkingLevel ): WorkspaceStreamInfo { // abortController is created and linked to the caller-provided abortSignal in startStream(). @@ -1809,7 +1916,9 @@ export class StreamManager extends EventEmitter { onChunk, onStepMessages, toolSearchState, - (toolCallId) => this.handleToolExecutionStart(workspaceId, messageId, toolCallId) + (toolCallId) => this.handleToolExecutionStart(workspaceId, messageId, toolCallId), + thinkingOverrideState, + rebuildProviderOptionsForThinkingLevel ); // Start streaming - this can throw immediately if API key is missing @@ -1872,6 +1981,19 @@ export class StreamManager extends EventEmitter { cumulativeProviderMetadata: undefined, }; + // Mid-turn thinking override: route applied levels into this stream's + // metadata (partials, stream-end, final assistant message). Wired before + // any step can run; the catch-up sync covers a holder that already applied + // a level (e.g. re-attachment on retry paths). + if (request.thinkingOverrideState) { + request.thinkingOverrideState.onApplied = (level) => { + streamInfo.thinkingLevel = level; + }; + if (request.thinkingOverrideState.applied) { + streamInfo.thinkingLevel = request.thinkingOverrideState.applied; + } + } + // Atomically register the stream this.workspaceStreams.set(workspaceId, streamInfo); @@ -2436,14 +2558,32 @@ export class StreamManager extends EventEmitter { // it would be categorized as a retryable api/unknown error and re-enter the // unbounded auto-retry loop this feature exists to prevent. Aborts rethrow so // a user interrupt during prepare stays an abort instead of a refusal. + // Fold a mid-turn thinking override (pending or already applied) into the + // fallback's baseline so the hop doesn't silently revert the user's + // mid-turn change. Pending is cleared here — prepare() re-clamps it for + // the fallback model and bakes it into the rebuilt provider options. + const overrideHolder = streamInfo.request.thinkingOverrideState; + const thinkingLevelOverride = overrideHolder?.pending ?? overrideHolder?.applied; + if (overrideHolder?.pending != null) { + overrideHolder.pending = undefined; + } + if (overrideHolder != null && thinkingLevelOverride != null) { + // Keep the override visible to later hops: prepare() bakes it into this + // hop's baseline, but a second hop re-folds from `applied`. + overrideHolder.applied = thinkingLevelOverride; + } + const prepareCallOptions: ModelFallbackPrepareOptions | undefined = + continuation?.success === true || thinkingLevelOverride != null + ? { + ...(continuation?.success === true + ? { continuation: { assistantMessage: continuation.data } } + : {}), + ...(thinkingLevelOverride != null ? { thinkingLevelOverride } : {}), + } + : undefined; let prepared: Result; try { - prepared = await fallbackState.options.prepare( - nextModelString, - continuation?.success === true - ? { continuation: { assistantMessage: continuation.data } } - : undefined - ); + prepared = await fallbackState.options.prepare(nextModelString, prepareCallOptions); } catch (error) { if (streamInfo.abortController.signal.aborted) { throw error; @@ -2485,7 +2625,12 @@ export class StreamManager extends EventEmitter { // Same state object: aiService's fallback prepare() rebuilt it in place // against the fallback toolset, so prepareStep keeps reading live state. streamInfo.request.toolSearchState, - (toolCallId) => this.handleToolExecutionStart(workspaceId, streamInfo.messageId, toolCallId) + (toolCallId) => this.handleToolExecutionStart(workspaceId, streamInfo.messageId, toolCallId), + // Same holder object (the session's setter keeps working across the + // hop) with a closure bound to the FALLBACK model. Attached before + // createStreamResult below in case the SDK eagerly prepares step 1. + streamInfo.request.thinkingOverrideState, + prepared.data.rebuildProviderOptionsForThinkingLevel ); // createStreamResult may eagerly prepare the first fallback step and update // latestMessages. Clear stale source-step messages before starting it so a @@ -3935,7 +4080,9 @@ export class StreamManager extends EventEmitter { onStepMessages?: (messages: ModelMessage[]) => void, providedRuntimeTempDir?: string, modelFallback?: ModelFallbackOptions, - toolSearchState?: ToolSearchStreamState + toolSearchState?: ToolSearchStreamState, + thinkingOverrideState?: ActiveTurnThinkingOverride, + rebuildProviderOptionsForThinkingLevel?: RebuildProviderOptionsForThinkingLevel ): Promise> { const typedWorkspaceId = workspaceId as WorkspaceId; @@ -4019,7 +4166,9 @@ export class StreamManager extends EventEmitter { onChunk, onStepMessages, modelFallback, - toolSearchState + toolSearchState, + thinkingOverrideState, + rebuildProviderOptionsForThinkingLevel ); // Guard against a narrow race: diff --git a/src/node/services/thinkingOverride.ts b/src/node/services/thinkingOverride.ts new file mode 100644 index 0000000000..46348811e5 --- /dev/null +++ b/src/node/services/thinkingOverride.ts @@ -0,0 +1,40 @@ +/** + * Mid-turn thinking-level override state. + * + * One holder object is created per turn by AgentSession (at durable turn + * acceptance) and threaded BY REFERENCE down through AIService into + * StreamManager's request config — the same shared-mutable-state pattern as + * ToolSearchStreamState. The renderer's thinking slider writes `pending` via + * AgentSession.setActiveTurnThinkingLevel; StreamManager consumes it at the + * next prepareStep (which runs before every model step, including step 1, so + * changes during the PREPARING window apply to the turn's first request). + * + * Node-runtime-local: carries a callback, so it does not belong in + * src/common/types (never crosses IPC). + */ +import type { ThinkingLevel } from "@/common/types/thinking"; + +export interface ActiveTurnThinkingOverride { + /** Raw level requested mid-turn; consumed at the next prepareStep (incl. step 1). */ + pending?: ThinkingLevel; + /** Effective level after the most recent successful application. */ + applied?: ThinkingLevel; + /** Sink wired by StreamManager to the owning streamInfo (metadata). */ + onApplied?: (level: ThinkingLevel) => void; +} + +/** + * Result of rebuilding provider options for a mid-turn thinking-level change. + * `null` from the rebuild closure means "not applicable / no-op" (e.g. the + * clamped level equals the current one, or the transition would require a + * different model instance) — the pending override is then dropped for this + * turn and the persisted setting still covers the next turn. + */ +export interface RebuiltThinkingProviderOptions { + effectiveLevel: ThinkingLevel; + providerOptions: Record; +} + +export type RebuildProviderOptionsForThinkingLevel = ( + level: ThinkingLevel +) => RebuiltThinkingProviderOptions | null; diff --git a/src/node/services/workspaceService.test.ts b/src/node/services/workspaceService.test.ts index 2126b580c9..56a3719d37 100644 --- a/src/node/services/workspaceService.test.ts +++ b/src/node/services/workspaceService.test.ts @@ -218,6 +218,16 @@ function createFrontendWorkspaceMetadata( }; } +describe("WorkspaceService.setActiveTurnThinkingLevel", () => { + test("returns accepted:false when the workspace has no session", () => { + const workspaceService = createWorkspaceServiceForTest({ config: {} }); + // No session was ever created for this workspace: nothing is running, so + // the mid-turn override is a no-op and persisted settings cover the next turn. + const result = workspaceService.setActiveTurnThinkingLevel("unknown-workspace", "high"); + expect(result).toEqual(Ok({ accepted: false })); + }); +}); + describe("WorkspaceService bash monitor wakes", () => { test("sends a synthetic wake and marks the record delivered when monitor output matches", async () => { const { config, cleanup } = await createTestHistoryService(); diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index a7647ea41b..5557d90015 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -7125,6 +7125,27 @@ export class WorkspaceService extends EventEmitter { } } + /** + * Mid-turn thinking change: forward the requested level to the workspace's + * active turn (if any). Deliberately `sessions.get`, not getOrCreateSession — + * creating a session to tell it "change the turn you don't have" is pointless; + * persisted settings already cover the next turn. + */ + setActiveTurnThinkingLevel( + workspaceId: string, + level: ThinkingLevel + ): Result<{ accepted: boolean }, string> { + try { + const session = this.sessions.get(workspaceId.trim()); + if (!session) { + return Ok({ accepted: false }); + } + return Ok(session.setActiveTurnThinkingLevel(level)); + } catch (error) { + return Err(`Failed to set active-turn thinking level: ${getErrorMessage(error)}`); + } + } + async fork( sourceWorkspaceId: string, newName?: string,