diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index ffe99dd293f39..22782ebdaa02c 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -2699,11 +2699,15 @@ describe("ClaudeAdapterV2 background wake turns", () => { uuid: "00000000-0000-4000-8000-000000000101", session_id: WAKE_NATIVE_SESSION, }); - const makeAssistantTextFrame = (input: { readonly uuid: string; readonly text: string }) => + const makeAssistantTextFrame = (input: { + readonly uuid: string; + readonly text: string; + readonly model?: string; + }) => claudeSdkFrame({ type: "assistant", message: { - model: "claude-sonnet-4-6", + model: input.model ?? "claude-sonnet-4-6", id: `msg_${input.uuid}`, type: "message", role: "assistant", @@ -3120,6 +3124,519 @@ describe("ClaudeAdapterV2 background wake turns", () => { }); const makeWakeHarness = makeWakeHarnessWithOptions(); + describe("Claude model downgrade", () => { + const selected = { + ...CLAUDE_TEST_MODEL_SELECTION, + model: "claude-opus-5-5", + options: [{ id: "effort", value: "high" }], + } satisfies ModelSelection; + const modelTestLayer = Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + ); + const makeModelTurn = Effect.fn("ClaudeAdapterV2.test.makeModelTurn")(function* ( + thread: Pick, + id: string, + overrides: Partial[0]> = {}, + ) { + return makeClaudeTestTurnInput({ + threadId: thread.threadId, + providerThread: thread.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make(id), + text: "Continue.", + attachments: [], + modelSelection: selected, + ...overrides, + }); + }); + const nextModelTurn = ( + previous: ProviderAdapter.ProviderAdapterV2TurnInput, + id: string, + overrides: Partial = {}, + ): ProviderAdapter.ProviderAdapterV2TurnInput => ({ + ...previous, + providerTurnOrdinal: previous.providerTurnOrdinal + 1, + attemptId: RunAttemptId.make(id), + runId: RunId.make(id), + rootNodeId: NodeId.make(id), + ...overrides, + message: { + ...previous.message, + ...overrides.message, + messageId: MessageId.make(id), + }, + }); + const latestProviderThreadUpdate = ( + events: ReadonlyArray, + ) => { + const event = events.findLast((event) => event.type === "provider_thread.updated"); + if (event === undefined) throw new Error("Expected provider-thread update."); + return event; + }; + const fallbackFrame = () => + claudeSdkFrame({ + type: "system", + subtype: "model_refusal_fallback", + uuid: "00000000-0000-4000-8000-000000000701", + session_id: WAKE_NATIVE_SESSION, + content: "Switched to Opus 5.", + original_model: selected.model, + fallback_model: "claude-opus-5", + request_id: null, + trigger: "refusal", + direction: "retry", + scope: "session", + } satisfies SDKMessage); + + it.effect.each([null, selected.model, "claude-sonnet-4-6"])( + "follows the downgrade and honors the next manual selection %s", + (manualModel) => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const first = yield* makeModelTurn(harness, "fallback-first", { + text: "First message.", + }); + yield* harness.runtime.startTurn(first); + yield* harness.offerAndWait(fallbackFrame()); + yield* Queue.take(harness.systemNoticeReceipts); + const reported = latestProviderThreadUpdate(harness.events); + const fallbackSelection = { ...first.modelSelection, model: "claude-opus-5" }; + assert.deepEqual( + reported.providerThread.nativeMetadata?.modelSelection, + fallbackSelection, + ); + assert.deepEqual(reported.modelSelectionChange, { + expected: first.modelSelection, + next: fallbackSelection, + }); + assert.isEmpty(harness.liveSelectionChanges); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "fallback-answer", + text: "Done.", + model: "claude-opus-5", + }), + ); + yield* harness.offerAndWait( + makeResultFrame({ uuid: "fallback-result", result: "Done." }), + ); + yield* Queue.take(harness.terminalReceipts); + const completed = latestProviderThreadUpdate(harness.events); + assert.deepEqual( + completed.providerThread.nativeMetadata?.modelSelection, + fallbackSelection, + ); + + const selection = + manualModel !== null ? { ...selected, model: manualModel } : fallbackSelection; + yield* harness.runtime.startTurn( + nextModelTurn(first, "fallback-next", { + appThread: { ...first.appThread, modelSelection: selection }, + providerThread: completed.providerThread, + modelSelection: selection, + message: { ...first.message, text: "Next." }, + }), + ); + assert.deepEqual( + harness.liveSelectionChanges.filter( + (change) => typeof change === "object" && change !== null && "model" in change, + ), + manualModel === null ? [] : [{ model: selection.model }], + ); + assert.lengthOf(harness.offeredMessages, 2); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "next-answer", + text: "Done.", + model: selection.model, + }), + ); + yield* harness.offerAndWait(makeResultFrame({ uuid: "next-result", result: "Done." })); + yield* Queue.take(harness.terminalReceipts); + const completedNext = latestProviderThreadUpdate(harness.events); + assert.deepEqual(completedNext.providerThread.nativeMetadata?.modelSelection, selection); + }).pipe(Effect.provide(modelTestLayer)), + ); + + it.effect("follows a repeated downgrade immediately after manually restoring the model", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const first = yield* makeModelTurn(harness, "repeated-fallback-first", { + text: "First message.", + }); + yield* harness.runtime.startTurn(first); + yield* harness.offerAndWait(fallbackFrame()); + yield* Queue.take(harness.systemNoticeReceipts); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "repeated-fallback-answer", + text: "Done.", + model: "claude-opus-5", + }), + ); + yield* harness.offerAndWait( + makeResultFrame({ uuid: "repeated-fallback-result", result: "Done." }), + ); + yield* Queue.take(harness.terminalReceipts); + const completed = latestProviderThreadUpdate(harness.events); + yield* harness.runtime.startTurn( + nextModelTurn(first, "repeated-fallback-next", { + appThread: { ...first.appThread, modelSelection: selected }, + providerThread: completed.providerThread, + }), + ); + assert.deepEqual(harness.liveSelectionChanges[0], { model: selected.model }); + const frame = fallbackFrame(); + if (frame.type !== "system" || frame.subtype !== "model_refusal_fallback") + throw new Error("Expected fallback."); + yield* harness.offerAndWait({ + ...frame, + uuid: "00000000-0000-4000-8000-000000000703", + }); + yield* Queue.take(harness.systemNoticeReceipts); + const selectionChanges = () => + harness.events.flatMap((event) => + event.type === "provider_thread.updated" && event.modelSelectionChange !== undefined + ? [event.modelSelectionChange] + : [], + ); + const expected = { expected: selected, next: { ...selected, model: frame.fallback_model } }; + assert.deepEqual(selectionChanges(), [expected, expected]); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "repeated-fallback-next-answer", + text: "Done again.", + model: frame.fallback_model, + }), + ); + assert.deepEqual(selectionChanges(), [expected, expected]); + yield* harness.offerAndWait( + makeResultFrame({ uuid: "repeated-fallback-next-result", result: "Done again." }), + ); + yield* Queue.take(harness.terminalReceipts); + }).pipe(Effect.provide(modelTestLayer)), + ); + + it.effect.each(["claude-opus-5", selected.model])( + "resumes a downgraded session with the selected model %s", + (model) => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + yield* harness.runtime.startTurn( + yield* makeModelTurn(harness, "fallback-resume", { + providerThread: { + ...harness.providerThread, + nativeMetadata: { + modelSelection: { ...selected, model: "claude-opus-5" }, + }, + }, + modelSelection: { ...selected, model }, + nativeThreadHasTurns: true, + text: "Continue on the selected model.", + }), + ); + if (model === "claude-opus-5") { + assert.isEmpty(harness.liveSelectionChanges); + } else { + assert.deepEqual(harness.liveSelectionChanges[0], { model }); + } + assert.lengthOf(harness.offeredMessages, 1); + }).pipe(Effect.provide(modelTestLayer)), + ); + + it.effect("leaves a local fallback out of the main thread's model state", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + yield* harness.runtime.startTurn(yield* makeModelTurn(harness, "fallback-local")); + const frame = fallbackFrame(); + if (frame.type !== "system" || frame.subtype !== "model_refusal_fallback") + throw new Error("Expected fallback."); + yield* harness.offerAndWait({ ...frame, scope: "local" as const }); + yield* Queue.take(harness.systemNoticeReceipts); + assert.isEmpty( + harness.events.filter( + (event) => + event.type === "provider_thread.updated" && event.modelSelectionChange !== undefined, + ), + ); + assert.isUndefined( + latestProviderThreadUpdate(harness.events).providerThread.nativeMetadata?.modelSelection, + ); + assert.isEmpty(harness.liveSelectionChanges); + }).pipe(Effect.provide(modelTestLayer)), + ); + + it.effect("does not switch the model while draining a native background continuation", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const first = yield* makeModelTurn(harness, "fallback-background", { + text: "Run a background task.", + }); + yield* harness.runtime.startTurn(first); + yield* harness.offerAndWait(fallbackFrame()); + yield* Queue.take(harness.systemNoticeReceipts); + yield* harness.offerAndWait(wakeTaskStarted); + yield* harness.offerAndWait(turnOneResult); + yield* Queue.take(harness.terminalReceipts); + yield* harness.offerAndWait(wakeNotification); + yield* harness.offerAndWait(wakeTurnInit); + yield* Queue.take(harness.continuationReceipts); + const latest = latestProviderThreadUpdate(harness.events); + yield* harness.runtime.startTurn( + nextModelTurn(first, "fallback-background-continuation", { + providerThread: latest.providerThread, + message: { ...first.message, createdBy: "agent", creationSource: "provider" }, + }), + ); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "fallback-wake-answer", + text: WAKE_ASSISTANT_TEXT, + model: "claude-opus-5", + }), + ); + yield* harness.offerAndWait(wakeResult); + yield* Queue.take(harness.terminalReceipts); + assert.isEmpty(harness.liveSelectionChanges); + assert.lengthOf(harness.offeredMessages, 1); + }).pipe(Effect.provide(modelTestLayer)), + ); + + it.effect.each([false, true])( + "replays a model downgrade before continuation attachment with fallback before wake init %s", + (beforeWakeInit) => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const first = yield* makeModelTurn(harness, "buffered-fallback-first", { + text: "Run a background task.", + }); + yield* harness.runtime.startTurn(first); + yield* harness.offerAndWait(wakeTaskStarted); + yield* harness.offerAndWait(turnOneResult); + yield* Queue.take(harness.terminalReceipts); + yield* harness.offerAndWait(wakeNotification); + assert.isEmpty(harness.continuationRequests); + if (beforeWakeInit) { + yield* harness.offerAndWait(fallbackFrame()); + assert.isEmpty(harness.continuationRequests); + } + yield* harness.offerAndWait(wakeTurnInit); + yield* Queue.take(harness.continuationReceipts); + if (!beforeWakeInit) yield* harness.offerAndWait(fallbackFrame()); + assert.lengthOf(harness.continuationRequests, 1); + const latest = latestProviderThreadUpdate(harness.events); + yield* harness.runtime.startTurn( + nextModelTurn(first, "buffered-fallback-continuation", { + providerThread: latest.providerThread, + message: { ...first.message, createdBy: "agent", creationSource: "provider" }, + }), + ); + yield* harness.offerAndWait(wakeResult); + yield* Queue.take(harness.terminalReceipts); + const next = { ...selected, model: "claude-opus-5" }; + const changes = harness.events.flatMap((event) => + event.type === "provider_thread.updated" && event.modelSelectionChange !== undefined + ? [event.modelSelectionChange] + : [], + ); + assert.deepEqual(changes, [{ expected: selected, next }]); + const completed = latestProviderThreadUpdate(harness.events); + assert.deepEqual(completed.providerThread.nativeMetadata?.modelSelection, next); + assert.isEmpty(harness.liveSelectionChanges); + assert.lengthOf(harness.offeredMessages, 1); + yield* harness.runtime.startTurn( + nextModelTurn(first, "buffered-fallback-reselect", { + providerThread: completed.providerThread, + providerTurnOrdinal: 3, + }), + ); + assert.deepEqual(harness.liveSelectionChanges[0], { model: selected.model }); + }).pipe(Effect.provide(modelTestLayer)), + ); + + it.effect.each([false, true])( + "preserves a newer manual selection during a background fallback after prior fallback %s", + (priorFallback) => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const first = yield* makeModelTurn(harness, "fallback-manual-background", { + text: "Run a background task.", + }); + yield* harness.runtime.startTurn(first); + if (priorFallback) { + yield* harness.offerAndWait(fallbackFrame()); + yield* Queue.take(harness.systemNoticeReceipts); + } + yield* harness.offerAndWait(wakeTaskStarted); + yield* harness.offerAndWait(turnOneResult); + yield* Queue.take(harness.terminalReceipts); + yield* harness.offerAndWait(wakeNotification); + yield* harness.offerAndWait(wakeTurnInit); + yield* Queue.take(harness.continuationReceipts); + const latest = latestProviderThreadUpdate(harness.events); + const manual = { ...selected, model: "claude-haiku-4-5-20251001" }; + yield* harness.runtime.startTurn( + nextModelTurn(first, "fallback-manual-continuation", { + appThread: { ...first.appThread, modelSelection: manual }, + providerThread: latest.providerThread, + message: { ...first.message, createdBy: "agent", creationSource: "provider" }, + }), + ); + const frame = fallbackFrame(); + if (frame.type !== "system" || frame.subtype !== "model_refusal_fallback") + throw new Error("Expected fallback."); + const model = priorFallback ? "claude-sonnet-4-6" : frame.fallback_model; + yield* harness.offerAndWait({ ...frame, fallback_model: model }); + yield* Queue.take(harness.systemNoticeReceipts); + const reported = latestProviderThreadUpdate(harness.events); + assert.deepEqual(reported.modelSelectionChange, { + expected: priorFallback ? { ...selected, model: frame.fallback_model } : selected, + next: { ...selected, model }, + }); + assert.isEmpty(harness.liveSelectionChanges); + assert.lengthOf(harness.offeredMessages, 1); + yield* harness.offerAndWait(wakeResult); + yield* Queue.take(harness.terminalReceipts); + }).pipe(Effect.provide(modelTestLayer)), + ); + + it.effect.each([false, true])( + "preserves a manual gateway alias after a downgrade with resumed session %s", + (resumed) => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const first = yield* makeModelTurn(harness, "fallback-gateway-first"); + let providerThread: OrchestrationV2ProviderThread = { + ...harness.providerThread, + nativeMetadata: { + modelSelection: { ...selected, model: "claude-opus-5" }, + }, + }; + if (!resumed) { + yield* harness.runtime.startTurn(first); + yield* harness.offerAndWait(fallbackFrame()); + yield* Queue.take(harness.systemNoticeReceipts); + yield* harness.offerAndWait( + makeResultFrame({ uuid: "fallback-gateway-first-result", result: "Done." }), + ); + yield* Queue.take(harness.terminalReceipts); + const latest = latestProviderThreadUpdate(harness.events); + providerThread = latest.providerThread; + } + const gateway = { ...selected, model: "gateway-opus" }; + const next = nextModelTurn(first, "fallback-gateway-next", { + appThread: { ...first.appThread, modelSelection: gateway }, + providerThread, + modelSelection: gateway, + nativeThreadHasTurns: true, + }); + yield* harness.runtime.startTurn(next); + assert.deepEqual(harness.liveSelectionChanges[0], { model: gateway.model }); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "fallback-gateway-answer", + text: "Done.", + model: selected.model, + }), + ); + yield* harness.offerAndWait( + makeResultFrame({ uuid: "fallback-gateway-result", result: "Done." }), + ); + yield* Queue.take(harness.terminalReceipts); + const selectionChanges = harness.events.flatMap((event) => + event.type === "provider_thread.updated" && event.modelSelectionChange !== undefined + ? [event.modelSelectionChange] + : [], + ); + assert.deepEqual( + selectionChanges, + resumed ? [] : [{ expected: selected, next: { ...selected, model: "claude-opus-5" } }], + ); + const completed = latestProviderThreadUpdate(harness.events); + assert.deepEqual(completed.providerThread.nativeMetadata?.modelSelection, gateway); + const changesBeforeNext = harness.liveSelectionChanges.length; + yield* harness.runtime.startTurn( + nextModelTurn(next, "fallback-gateway-third", { + providerThread: completed.providerThread, + }), + ); + assert.lengthOf(harness.liveSelectionChanges, changesBeforeNext); + }).pipe(Effect.provide(modelTestLayer)), + ); + + it.effect("does not mistake a gateway's canonical model name for a downgrade", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + yield* harness.runtime.startTurn( + yield* makeModelTurn(harness, "fallback-gateway", { + modelSelection: { ...selected, model: "gateway-opus" }, + }), + ); + yield* harness.offerAndWait( + makeAssistantTextFrame({ uuid: "gateway-answer", text: "Done.", model: selected.model }), + ); + yield* harness.offerAndWait(makeResultFrame({ uuid: "gateway-result", result: "Done." })); + yield* Queue.take(harness.terminalReceipts); + const latest = latestProviderThreadUpdate(harness.events); + assert.isUndefined(latest.providerThread.nativeMetadata?.modelSelection); + assert.isEmpty(harness.liveSelectionChanges); + }).pipe(Effect.provide(modelTestLayer)), + ); + + it.effect.each([ + { selection: selected, apiModelId: selected.model }, + { + selection: { + ...selected, + model: "claude-opus-4-6", + options: [...selected.options, { id: "contextWindow", value: "1m" }], + }, + apiModelId: "claude-opus-4-6[1m]", + }, + ])( + "follows a provider revert to $apiModelId after a downgrade within the same turn", + ({ selection, apiModelId }) => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const first = yield* makeModelTurn(harness, "fallback-revert", { + modelSelection: selection, + }); + yield* harness.runtime.startTurn(first); + const frame = fallbackFrame(); + if (frame.type !== "system" || frame.subtype !== "model_refusal_fallback") + throw new Error("Expected fallback."); + yield* harness.offerAndWait({ ...frame, original_model: apiModelId }); + yield* Queue.take(harness.systemNoticeReceipts); + yield* harness.offerAndWait({ + ...frame, + uuid: "00000000-0000-4000-8000-000000000702", + original_model: apiModelId, + direction: "revert" as const, + }); + yield* Queue.take(harness.systemNoticeReceipts); + const latest = latestProviderThreadUpdate(harness.events); + assert.deepEqual(latest.modelSelectionChange, { + expected: { ...selection, model: "claude-opus-5" }, + next: selection, + }); + assert.deepEqual(latest.providerThread.nativeMetadata?.modelSelection, selection); + yield* harness.offerAndWait( + makeAssistantTextFrame({ uuid: "revert-answer", text: "Done.", model: apiModelId }), + ); + yield* harness.offerAndWait(makeResultFrame({ uuid: "revert-result", result: "Done." })); + yield* Queue.take(harness.terminalReceipts); + yield* harness.runtime.startTurn( + nextModelTurn(first, "revert-next", { + providerThread: latestProviderThreadUpdate(harness.events).providerThread, + }), + ); + assert.isEmpty(harness.liveSelectionChanges); + }).pipe(Effect.provide(modelTestLayer)), + ); + }); + it.effect.each(["release-after", "cancel-after", "release-before", "resume-pending"] as const)( "keeps attachment reservations across compaction: %s boundary", (outcome) => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index a280bd8440a9a..9385a7d7b2d5b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -62,7 +62,7 @@ import { parseCliArgs } from "@t3tools/shared/cliArgs"; import { fromLenientJson } from "@t3tools/shared/schemaJson"; import { AgentScope } from "@t3tools/shared/AgentScope"; import * as HostProcess from "@t3tools/shared/HostProcess"; -import { applyClaudePromptEffortPrefix } from "@t3tools/shared/model"; +import { applyClaudePromptEffortPrefix, modelSelectionsEqual } from "@t3tools/shared/model"; import { CLAUDE_RESUME_COMPACTION_COMPACT_ANSWER, CLAUDE_RESUME_COMPACTION_NEVER_ANSWER, @@ -3973,6 +3973,7 @@ function formatClaudeUsageLimitWait(waitMs: number): string { interface ActiveClaudeTurnContext { readonly queryPolicy: ClaudeRuntimeQueryPolicy; readonly input: ProviderAdapter.ProviderAdapterV2TurnInput; + followedModelSelection: ModelSelection; readonly nativeThreadId: string; readonly nativeTurnId: string; nativeMessageCursor: string | null; @@ -4070,6 +4071,7 @@ interface ClaudeLiveQueryContext { readonly query: ClaudeAgentSdkQuerySession; readonly queryPolicyKey: string; selection: CompiledClaudeModelSelection; + currentModel: string; readonly closed: Deferred.Deferred; lastTurn: ActiveClaudeTurnContext | null; // "early" echoes on the first frame; "acknowledged" confirms uuid acceptance, but resume turns can still precede the prompt. @@ -5253,6 +5255,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const now = yield* DateTime.now; const providerThread: OrchestrationV2ProviderThread = { ...input.providerThread, + nativeMetadata: + (yield* Ref.get(lastProviderThreadByNativeThread)).get(input.nativeThreadId) + ?.nativeMetadata ?? input.providerThread.nativeMetadata, providerSessionId: session.id, ...(input.status === undefined ? {} : { status: input.status }), pendingBackgroundTasks: claudePendingBackgroundTasksFromRoster(roster, awaitedTaskIds), @@ -5269,6 +5274,39 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); }); + // Session model events may repeat; publish only changes. + const reportClaudeModel = Effect.fnUntraced(function* ( + context: ActiveClaudeTurnContext, + model: string, + ) { + const current = + (yield* Ref.get(lastProviderThreadByNativeThread)).get(context.nativeThreadId) ?? + context.input.providerThread; + const selection = { ...context.input.modelSelection, model }; + const change = !modelSelectionsEqual(context.followedModelSelection, selection) + ? { expected: context.followedModelSelection, next: selection } + : undefined; + if ( + change === undefined && + current.nativeMetadata?.modelSelection !== undefined && + modelSelectionsEqual(current.nativeMetadata.modelSelection, selection) + ) + return; + const providerThread = { + ...current, + nativeMetadata: { ...current.nativeMetadata, modelSelection: selection }, + updatedAt: yield* DateTime.now, + }; + yield* rememberProviderThread(providerThread); + yield* emitProviderEvent({ + type: "provider_thread.updated", + driver: CLAUDE_PROVIDER, + providerThread, + ...(change === undefined ? {} : { modelSelectionChange: change }), + }); + if (change !== undefined) context.followedModelSelection = selection; + }); + const trackClaudeGoal = Effect.fnUntraced(function* (input: { readonly nativeThreadId: string; readonly context: ActiveClaudeTurnContext; @@ -7600,6 +7638,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } const providerThread: OrchestrationV2ProviderThread = { ...input.context.input.providerThread, + nativeMetadata: + (nativeThreadId === null + ? undefined + : (yield* Ref.get(lastProviderThreadByNativeThread)).get(nativeThreadId) + ?.nativeMetadata) ?? input.context.input.providerThread.nativeMetadata, providerSessionId: session.id, ...(clearConversationHead ? { nativeConversationHeadRef: null } : {}), ...(nativeThreadId !== null && goalsByNativeThread.has(nativeThreadId) @@ -7886,6 +7929,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( isKnownSubagentTaskStarted || isNewSubagentTaskStarted || isWakeTurnStart || + (message.type === "system" && message.subtype === "model_refusal_fallback") || message.type === "assistant" || message.type === "user" || message.type === "result" || @@ -8649,6 +8693,16 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( (message.subtype === "model_refusal_fallback" || (message.subtype === "informational" && message.level !== "info")) ) { + if (message.subtype === "model_refusal_fallback" && message.scope !== "local") { + const model = + message.direction === "revert" ? message.original_model : message.fallback_model; + liveQuery.currentModel = model; + const reportedModel = + model === compileClaudeModelSelection(context.input.modelSelection).apiModelId + ? context.input.modelSelection.model + : model; + yield* reportClaudeModel(context, reportedModel); + } const now = yield* DateTime.now; const nativeItemId = message.uuid; context.assistant.noticeTexts.add(message.content); @@ -10115,7 +10169,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( existing.nativeThreadId === nativeThreadId && existing.queryPolicyKey === queryPolicyKey && !existing.stopping && - existing.selection.queryIdentity !== compiledSelection.queryIdentity && + (existing.selection.queryIdentity !== compiledSelection.queryIdentity || + existing.currentModel !== compiledSelection.apiModelId) && canApplyClaudeSelectionLive(existing.selection, compiledSelection) && (existing.selection.apiModelId === compiledSelection.apiModelId || existing.openedPermissionMode === queryPolicy.permissionMode) @@ -10124,7 +10179,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( // Preserve launcher-selected models unless the model changed or a failed update left it unknown. yield* ( live.selection.queryIdentity === "" || - live.selection.apiModelId !== compiledSelection.apiModelId + live.currentModel !== compiledSelection.apiModelId ? live.query.setModel(compiledSelection.apiModelId) : Effect.void ).pipe( @@ -10138,13 +10193,15 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ), ); live.selection = compiledSelection; + live.currentModel = compiledSelection.apiModelId; } if ( existing !== null && existing.nativeThreadId === nativeThreadId && existing.queryPolicyKey === queryPolicyKey && !existing.stopping && - existing.selection.queryIdentity === compiledSelection.queryIdentity + existing.selection.queryIdentity === compiledSelection.queryIdentity && + existing.currentModel === compiledSelection.apiModelId ) { // A denied ExitPlanMode leaves Claude in plan mode. Restore only from plan: // Auto/default may reflect a policy restriction that forbids restoring bypass. @@ -10290,6 +10347,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( query: querySession, queryPolicyKey, selection: compiledSelection, + currentModel: + shouldResume && turnInput.providerThread.nativeMetadata?.modelSelection !== undefined + ? compileClaudeModelSelection( + turnInput.providerThread.nativeMetadata.modelSelection, + ).apiModelId + : compiledSelection.apiModelId, closed, lastTurn: null, promptEchoMode: "unknown", @@ -10363,6 +10426,16 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( Effect.ensuring(Deferred.succeed(closed, undefined)), Effect.forkIn(sessionScope), ); + if ( + context.currentModel !== compiledSelection.apiModelId && + !isClaudeProviderContinuationTurn(turnInput) + ) { + yield* context.query.setModel(compiledSelection.apiModelId); + yield* context.query.applyFlagSettings( + claudeLiveSelectionFlagSettings(compiledSelection), + ); + context.currentModel = compiledSelection.apiModelId; + } return context; }); @@ -10441,6 +10514,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const querySession = yield* openQuery(turnInput, nativeThreadId); const context: ActiveClaudeTurnContext = { input: turnInput, + followedModelSelection: isClaudeProviderContinuationTurn(turnInput) + ? (querySession.lastTurn?.followedModelSelection ?? + turnInput.providerThread.nativeMetadata?.modelSelection ?? + turnInput.modelSelection) + : turnInput.modelSelection, queryPolicy: querySession.queryPolicy, nativeThreadId, nativeTurnId, @@ -10544,6 +10622,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); if (userMessage !== null) { yield* attachTurn; + if (turnInput.providerThread.nativeMetadata?.modelSelection !== undefined) { + yield* reportClaudeModel(context, turnInput.modelSelection.model); + } yield* startUserTurnForWakeReports(nativeThreadId); yield* Effect.uninterruptibleMask((restore) => restore( diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index 7f1006f1c5ec5..92ff3b091d972 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -434,6 +434,192 @@ it.effect("records accepted billed turn usage once without billing the context w }).pipe(Effect.provide(layerTest.pipe(Layer.provide(layerAnalytics)))); }); +it.effect.each([ + { label: "current selection", changedSelection: false, changedThread: false, follows: true }, + { label: "new manual selection", changedSelection: true, changedThread: false, follows: false }, + { + label: "different provider thread", + changedSelection: false, + changedThread: true, + follows: false, + }, +])("follows a model downgrade only for the $label", (testCase) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const sink = yield* EventSink.EventSinkV2; + const store = yield* ProjectionStore.ProjectionStoreV2; + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; + const ids = yield* IdAllocator.IdAllocatorV2; + const created = yield* threadCreatedEvent(now); + if (created.type !== "thread.created") throw new Error("Expected a thread."); + const original = created.payload; + const manual = { ...modelSelection, model: "another-model" }; + const next = { ...modelSelection, model: "fallback-model" }; + yield* sink.write({ + events: [ + { + ...created, + payload: { + ...original, + modelSelection: testCase.changedSelection ? manual : modelSelection, + activeProviderThreadId: testCase.changedThread ? null : original.activeProviderThreadId, + }, + }, + ], + }); + const providerSessionId = yield* ids.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId: original.id, + }); + const providerThread: OrchestrationV2ProviderThread = { + id: original.activeProviderThreadId!, + driver: CODEX_DRIVER, + providerInstanceId: modelSelection.instanceId, + providerSessionId, + appThreadId: original.id, + ownerNodeId: null, + nativeThreadRef: { driver: CODEX_DRIVER, nativeId: "native-thread", strength: "strong" }, + nativeConversationHeadRef: null, + status: "active", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + pendingBackgroundTasks: [], + contextUsage: null, + nativeMetadata: { modelSelection: next }, + goal: null, + createdAt: now, + updatedAt: now, + }; + const events = yield* ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId: original.id, + event: { + type: "provider_thread.updated", + driver: CODEX_DRIVER, + providerThread, + modelSelectionChange: { expected: modelSelection, next }, + }, + }); + const stored = yield* store.getThread(original.id); + assert.deepEqual( + stored.modelSelection, + testCase.follows ? next : testCase.changedSelection ? manual : modelSelection, + ); + assert.deepEqual( + events.map((stored) => stored.event.type), + testCase.follows + ? ["provider-thread.updated", "thread.model-selection-updated"] + : ["provider-thread.updated"], + ); + const projection = yield* store.getThreadProjection(original.id); + assert.deepEqual(projection.providerThreads[0]?.nativeMetadata?.modelSelection, next); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("preserves a concurrent thread edit after a downgrade is normalized", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const sink = yield* EventSink.EventSinkV2; + const store = yield* ProjectionStore.ProjectionStoreV2; + const ids = yield* IdAllocator.IdAllocatorV2; + const commands = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const created = yield* threadCreatedEvent(now); + if (created.type !== "thread.created") throw new Error("Expected a thread."); + const original = created.payload; + yield* sink.write({ events: [created] }); + const next = { ...modelSelection, model: "fallback-model" }; + const manual = { ...modelSelection, model: "manually-selected-model" }; + const providerSessionId = yield* ids.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId: original.id, + }); + const normalized = yield* Deferred.make< + void, + ProviderEventIngestor.ProviderEventIngestorV2Error + >(); + const allowCommit = yield* Deferred.make(); + const gatedSink = EventSink.EventSinkV2.of({ + ...sink, + write: (input) => + Effect.gen(function* () { + assert.include(yield* commands.activeKeys, original.id); + yield* Deferred.succeed(normalized, undefined); + yield* Deferred.await(allowCommit); + return yield* sink.write(input); + }), + }); + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2.pipe( + Effect.provide(Layer.fresh(ProviderEventIngestor.layer)), + Effect.provideService(EventSink.EventSinkV2, gatedSink), + ); + const input: ProviderEventIngestor.ProviderEventIngestInput = { + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId: yield* ids.allocate.thread({ fixtureName: "downgrade-route" }), + event: { + type: "provider_thread.updated", + driver: CODEX_DRIVER, + providerThread: { + id: original.activeProviderThreadId!, + driver: CODEX_DRIVER, + providerInstanceId: modelSelection.instanceId, + providerSessionId, + appThreadId: original.id, + ownerNodeId: null, + nativeThreadRef: { driver: CODEX_DRIVER, nativeId: "native-thread", strength: "strong" }, + nativeConversationHeadRef: null, + status: "active", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + nativeMetadata: { modelSelection: next }, + createdAt: now, + updatedAt: now, + }, + modelSelectionChange: { expected: modelSelection, next }, + }, + }; + const downgrade = yield* ingestor.ingestNormalized(input).pipe( + Effect.tapCause((cause) => Deferred.failCause(normalized, cause)), + Effect.forkScoped, + ); + yield* Deferred.await(normalized); + const editRequested = yield* Deferred.make(); + const edit = yield* Deferred.succeed(editRequested, undefined).pipe( + Effect.andThen( + commands.withLock( + original.id, + Effect.gen(function* () { + const current = yield* store.getThread(original.id); + yield* sink.write({ + events: [ + { + ...created, + id: yield* ids.allocate.event({ threadId: original.id }), + type: "thread.metadata-updated", + payload: { ...current, title: "Manual title", modelSelection: manual }, + }, + ], + }); + }), + ), + ), + Effect.forkScoped, + ); + yield* Deferred.await(editRequested); + yield* Deferred.succeed(allowCommit, undefined); + assert.lengthOf(yield* Fiber.join(downgrade), 2); + yield* Fiber.join(edit); + const updated = yield* store.getThread(original.id); + assert.equal(updated.title, "Manual title"); + assert.deepEqual(updated.modelSelection, manual); + }).pipe(Effect.provide(layerTest)), +); + layer("ProviderEventIngestorV2", (it) => { it.effect("normalizes provider events through the real event log and projection store", () => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 5a2aa39faece4..c9dd8d10b00bd 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -591,10 +591,34 @@ export const layer: Layer.Layer< threadId, payload: input.event.providerThread, }); + const events = [updated]; + const change = input.event.modelSelectionChange; + if (change !== undefined) { + const thread = yield* projections.getThread(threadId); + if ( + thread.activeProviderThreadId === input.event.providerThread.id && + thread.providerInstanceId === input.event.providerThread.providerInstanceId && + change.next.instanceId === thread.providerInstanceId && + modelSelectionsEqual(thread.modelSelection, change.expected) && + !modelSelectionsEqual(thread.modelSelection, change.next) + ) { + events.push( + yield* makeDomainEvent(input, { + type: "thread.model-selection-updated", + threadId, + payload: { + ...thread, + modelSelection: change.next, + updatedAt: updated.occurredAt, + }, + }), + ); + } + } return input.event.retainedNativeTurnIds === undefined - ? [updated] + ? events : [ - updated, + ...events, ...(yield* rollBackRewoundRuns(input, { threadId, providerThreadId: input.event.providerThread.id, @@ -902,10 +926,16 @@ export const layer: Layer.Layer< ); }), ), - (effect) => - input.event.type === "app_thread.created" - ? threadCommands.withLock(input.event.appThread.id, effect) - : effect, + (effect) => { + const threadId = + input.event.type === "app_thread.created" + ? input.event.appThread.id + : input.event.type === "provider_thread.updated" && + input.event.modelSelectionChange !== undefined + ? (input.event.providerThread.appThreadId ?? input.threadId) + : null; + return threadId === null ? effect : threadCommands.withLock(threadId, effect); + }, ), ), }); diff --git a/docs/user/providers-claude.md b/docs/user/providers-claude.md index c94fc3420f875..e9df0c3dad8fa 100644 --- a/docs/user/providers-claude.md +++ b/docs/user/providers-claude.md @@ -41,6 +41,12 @@ Claude Code's verbose mode can stay enabled when you use Claude for text generat thread titles, branch names, commit messages, and pull request descriptions. On a remote connection, T3 Code uses the Claude configuration on the connected server. +## Model downgrades + +Claude may switch to a fallback model when it flags a request. T3 Code +updates the thread's model selection to match, and later messages keep using that +model. You can choose another model in the same thread. + ## Compact long conversations Set **Auto-compact after** in the Claude provider settings to an integer between diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index cadf4173a60b1..980ac203c7e63 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -92,6 +92,9 @@ export const ProviderAdapterV2Event = Schema.Union([ driver: ProviderDriverKind, providerThread: OrchestrationV2ProviderThread, retainedNativeTurnIds: Schema.optional(Schema.Array(Schema.String)), + modelSelectionChange: Schema.optional( + Schema.Struct({ expected: ModelSelection, next: ModelSelection }), + ), }), Schema.Struct({ type: Schema.Literal("provider_turn.updated"),