diff --git a/apps/server/src/orchestration-v2/CommandPolicy.ts b/apps/server/src/orchestration-v2/CommandPolicy.ts index cd56460369bd..c8ddbde3c29f 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.ts @@ -1,5 +1,6 @@ import { type ChatAttachment, + type ModelSelection, CommandId, type OrchestrationV2Command, OrchestrationV2ProviderCapabilities, @@ -16,6 +17,16 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; +// Account choice is preflighted outside the dispatch lock; a raced user action wins. +export const AccountFallback = Context.Reference< + | { + readonly commandId: CommandId; + readonly threadSelection: ModelSelection; + readonly requestedSelection: ModelSelection | undefined; + } + | undefined +>("t3/orchestration-v2/AccountFallback", { defaultValue: () => undefined }); + export const SteeringExecutionPolicyV2 = Schema.Literals(["active_steering", "interrupt_restart"]); export type SteeringExecutionPolicyV2 = typeof SteeringExecutionPolicyV2.Type; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3b15161670c0..8ceb2b662621 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1,3 +1,4 @@ +import { AccountFallback } from "./CommandPolicy.ts"; import type { OrchestrationV2ThreadLinkMetadata, GuestShareId, @@ -6794,6 +6795,25 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); return; } + const fallback = yield* AccountFallback; + const automaticFallback = fallback?.commandId === command.commandId ? fallback : undefined; + if (automaticFallback && command.manualContinuationOfRunId === undefined) { + const mode = resolveMessageDispatchIntent( + projection, + command.dispatchMode, + command.deliveryIntent, + ); + if ( + mode.type === "steer_active" || + mode.type === "restart_active" || + !modelSelectionsEqual(projection.thread.modelSelection, automaticFallback.threadSelection) + ) { + command = { + ...command, + modelSelection: automaticFallback.requestedSelection ?? projection.thread.modelSelection, + }; + } + } const activeProviderInstanceId = projection.providerThreads.find( (candidate) => candidate.id === projection.thread.activeProviderThreadId, )?.providerInstanceId; @@ -6817,7 +6837,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio latestExecutedRun(projection.runs)?.id !== source.id || projection.thread.archivedAt !== null || projection.thread.deletedAt !== null || - projection.runtimeRequests.some((request) => request.status === "pending") + projection.runtimeRequests.some((request) => request.status === "pending") || + (automaticFallback !== undefined && + (projection.thread.settledOverride === "settled" || + !modelSelectionsEqual(source.modelSelection, automaticFallback.threadSelection) || + !modelSelectionsEqual( + projection.thread.modelSelection, + automaticFallback.threadSelection, + ) || + (projection.thread.snoozedUntil != null && + DateTime.toEpochMillis(projection.thread.snoozedUntil) > + DateTime.toEpochMillis(yield* DateTime.now)) || + (yield* stopReachedRun(command, command.threadId, source.id)))) ) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 156223fee207..2289ef428016 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -8307,6 +8307,8 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { if (lastErrorClass === "usage_limit") { assert.deepEqual(candidate, { id: sqlShell.id, + projectId: sqlShell.projectId, + modelSelection: sqlShell.modelSelection, status: sqlShell.status, lastErrorClass: sqlShell.lastErrorClass, usageLimitResetAt: sqlShell.usageLimitResetAt, @@ -8475,6 +8477,33 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { (row) => row.id === threadId, ), ); + for (const [since, found] of [ + [DateTime.subtract(now, { minutes: 1 }), true], + [DateTime.add(now, { minutes: 1 }), false], + ] as const) { + assert.strictEqual( + (yield* store.getLimitRecoveryCandidates({ + ...recoveryOptions, + switchAccountsSince: since, + })).some((row) => row.id === threadId), + found, + ); + } + yield* sql`UPDATE orchestration_v2_projection_turn_items + SET payload_json = json_remove(payload_json, '$.failure.resetAt') WHERE turn_item_id = ${limitItem.id}`; + assert.isTrue( + (yield* store.getLimitRecoveryCandidates({ + ...recoveryOptions, + switchAccountsSince: DateTime.subtract(now, { minutes: 1 }), + })).some((row) => row.id === threadId), + ); + assert.isFalse( + (yield* store.getLimitRecoveryCandidates({ ...recoveryOptions, autoResume: true })).some( + (row) => row.id === threadId, + ), + ); + yield* sql`UPDATE orchestration_v2_projection_turn_items + SET payload_json = json_set(payload_json, '$.failure.resetAt', ${limitItem.failure.resetAt}) WHERE turn_item_id = ${limitItem.id}`; const reset = DateTime.makeUnsafe(limitItem.failure.resetAt); const recovery = { runId: original.id, @@ -8739,6 +8768,17 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { deletedParentChild, activeParentChild, ]; + assert.deepEqual( + (yield* store.getLimitRecoveryCandidates({ + now, + autoResume: false, + snooze: false, + switchAccountsSince: DateTime.subtract(now, { minutes: 1 }), + })) + .filter((row) => children.includes(row.id)) + .map((row) => row.id), + [activeParentChild], + ); for (const options of [ { now, autoResume: true, snooze: false }, { now, autoResume: false, snooze: true }, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index a003ade1ae8f..083ab4b41d12 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -184,6 +184,8 @@ export type ProjectionRecoveryKind = export type ProjectionLimitRecoveryCandidate = Pick< OrchestrationV2ThreadShell, | "id" + | "projectId" + | "modelSelection" | "status" | "lastErrorClass" | "latestRunId" @@ -622,6 +624,8 @@ export interface ProjectionStoreV2Shape { readonly now: DateTime.Utc; readonly autoResume: boolean; readonly snooze: boolean; + /** Also returns limits hit since then whose window has not reset, to move to another account. */ + readonly switchAccountsSince?: DateTime.Utc; }) => Effect.Effect, ProjectionStoreV2Error>; /** Epoch ms of the next armed reset or snooze end after `now`, when a waiting thread becomes a candidate. */ readonly getLimitRecoveryWakeAt: ( @@ -4662,11 +4666,28 @@ export const layer: Layer.Layer = AND json_extract(t.payload_json, '$.archivedAt') IS NULL AND json_extract(t.payload_json, '$.settledOverride') IS NOT 'settled' AND json_extract(item.payload_json, '$.failure.class') = 'usage_limit' - AND json_extract(item.payload_json, '$.failure.resetAt') IS NOT NULL - AND julianday(json_extract(item.payload_json, '$.failure.resetAt')) > julianday(COALESCE(r.completed_at, json_extract(t.payload_json, '$.updatedAt'))) AND ( ( - json_extract(t.payload_json, '$.limitRecovery.runId') IS r.run_id + ${booleanInt(options.switchAccountsSince !== undefined)} + AND julianday(COALESCE(r.completed_at, json_extract(t.payload_json, '$.updatedAt'))) >= julianday(${DateTime.formatIso(options.switchAccountsSince ?? options.now)}) + AND ( + json_extract(item.payload_json, '$.failure.resetAt') IS NULL + OR julianday(json_extract(item.payload_json, '$.failure.resetAt')) > julianday(${DateTime.formatIso(options.now)}) + ) + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_threads parent + WHERE json_extract(t.payload_json, '$.lineage.relationshipToParent') = 'subagent' + AND parent.thread_id = json_extract(t.payload_json, '$.lineage.parentThreadId') + AND ( + json_extract(parent.payload_json, '$.settledOverride') = 'settled' + OR json_extract(parent.payload_json, '$.archivedAt') IS NOT NULL + OR parent.deleted_at IS NOT NULL + ) + ) + ) + OR ( + julianday(json_extract(item.payload_json, '$.failure.resetAt')) > julianday(COALESCE(r.completed_at, json_extract(t.payload_json, '$.updatedAt'))) + AND json_extract(t.payload_json, '$.limitRecovery.runId') IS r.run_id AND json_extract(t.payload_json, '$.limitRecovery.resetAt') IS json_extract(item.payload_json, '$.failure.resetAt') AND json_extract(t.payload_json, '$.limitRecovery.autoResume') = 1 AND julianday(json_extract(item.payload_json, '$.failure.resetAt')) <= julianday(${DateTime.formatIso(options.now)}) @@ -4676,6 +4697,8 @@ export const layer: Layer.Layer = ) ) OR ( + julianday(json_extract(item.payload_json, '$.failure.resetAt')) > julianday(COALESCE(r.completed_at, json_extract(t.payload_json, '$.updatedAt'))) + AND ( json_extract(t.payload_json, '$.limitRecovery.runId') IS NOT r.run_id OR json_extract(t.payload_json, '$.limitRecovery.resetAt') IS NOT json_extract(item.payload_json, '$.failure.resetAt') @@ -4714,6 +4737,8 @@ export const layer: Layer.Layer = if (summary.lastErrorClass !== "usage_limit") continue; candidates.push({ id: thread.id, + projectId: thread.projectId, + modelSelection: thread.modelSelection, status: "failed", lastErrorClass: summary.lastErrorClass, usageLimitResetAt: summary.usageLimitResetAt, diff --git a/apps/server/src/orchestration-v2/ThreadMessageIntake.test.ts b/apps/server/src/orchestration-v2/ThreadMessageIntake.test.ts index a92cdd22aa51..7c72e4e0eb7d 100644 --- a/apps/server/src/orchestration-v2/ThreadMessageIntake.test.ts +++ b/apps/server/src/orchestration-v2/ThreadMessageIntake.test.ts @@ -1,4 +1,5 @@ import * as DocumentText from "../provider/DocumentText.ts"; +import { AccountFallback } from "./CommandPolicy.ts"; // @effect-diagnostics nodeBuiltinImport:off import * as NodeFS from "node:fs"; import * as NodePath from "node:path"; @@ -10,6 +11,9 @@ import { EventId, GuestMessageId, MessageId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, RunId, RuntimeRequestId, ThreadId, @@ -17,6 +21,7 @@ import { type OrchestrationV2Command, type OrchestrationV2ServerCommand, type OrchestrationV2StoredEvent, + type ServerProvider, type UserInputAttachments, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; @@ -28,6 +33,7 @@ import { resolveAttachmentPath, } from "../attachmentStore.ts"; import * as ServerConfig from "../config.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; import * as ServerSettings from "../serverSettings.ts"; import { OrchestratorCommandPreviouslyRejectedError, @@ -41,6 +47,9 @@ const layerIntakeTest = ServerConfig.layerTest(process.cwd(), { }).pipe( Layer.provideMerge(DocumentText.layer.pipe(Layer.provideMerge(NodeServices.layer))), Layer.provideMerge(ServerSettings.layerTest()), + Layer.provideMerge( + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + ), ); const layerFailingDispatchHandler = @@ -428,6 +437,112 @@ it.effect("sends a guest message with the guest's files, never ones the client n }).pipe(Effect.provide(layerIntakeTest)), ); +it.effect.each([false, true])( + "routes an exhausted account while preserving manual resume: %s", + (manual) => + Effect.gen(function* () { + const claude = (id: string, usedPercent: number): ServerProvider => ({ + instanceId: ProviderInstanceId.make(id), + driver: ProviderDriverKind.make("claudeAgent"), + enabled: true, + installed: true, + version: null, + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-09-13T00:00:00.000Z", + models: [ + { slug: "claude-opus-5-5", name: "Opus 5.5", isCustom: false, capabilities: null }, + ], + slashCommands: [], + skills: [], + usageLimits: { + checkedAt: "2026-09-13T00:00:00.000Z", + windows: [ + { + id: "five_hour", + kind: "session", + label: "Session", + usedPercent, + resetsAt: "2099-01-01T00:00:00.000Z", + }, + ], + }, + }); + const captured: OrchestrationV2ServerCommand[] = []; + const selection = { + instanceId: ProviderInstanceId.make("claude_work"), + model: "claude-opus-5-5", + }; + yield* dispatchCommand({ + type: "message.dispatch", + commandId: CommandId.make("send-limited"), + threadId: ThreadId.make("thread-limited"), + messageId: MessageId.make("message-limited"), + ...(manual ? { manualContinuationOfRunId: RunId.make("limited-run") } : {}), + text: "go", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }).pipe( + Effect.provide( + Layer.mergeAll( + Layer.mock(ThreadManagementService.ThreadManagementService)({ + dispatch: (command) => + AccountFallback.pipe( + Effect.flatMap((fallback) => { + expect(fallback).toEqual( + manual + ? undefined + : { + commandId: command.commandId, + threadSelection: selection, + requestedSelection: undefined, + }, + ); + return layerFailingDispatchHandler(captured)(command); + }), + ), + getThreadRecords: () => + Effect.succeed({ + thread: { + projectId: ProjectId.make("project-limited"), + modelSelection: selection, + }, + } as never), + }), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getProviders: Effect.succeed([ + claude("claude_work", 100), + claude("claude_personal", 30), + ]), + }), + ServerSettings.layerTest({ + fallbackToAvailableAccounts: true, + providerInstances: Object.fromEntries( + ["claude_work", "claude_personal"].map((id) => [ + ProviderInstanceId.make(id), + { driver: ProviderDriverKind.make("claudeAgent"), enabled: true }, + ]), + ), + }), + ), + ), + Effect.result, + ); + if (manual) { + expect(captured[0]).not.toHaveProperty("modelSelection"); + return; + } + expect(captured[0]).toMatchObject({ + modelSelection: { + instanceId: "claude_personal", + model: "claude-opus-5-5", + }, + }); + }).pipe(Effect.provide(layerIntakeTest)), +); + it.effect("refuses a queued edit that keeps a file another edit already deleted", () => Effect.gen(function* () { const config = yield* ServerConfig.ServerConfig; diff --git a/apps/server/src/orchestration-v2/ThreadMessageIntake.ts b/apps/server/src/orchestration-v2/ThreadMessageIntake.ts index 6e9fd503baec..2f1c87d64e41 100644 --- a/apps/server/src/orchestration-v2/ThreadMessageIntake.ts +++ b/apps/server/src/orchestration-v2/ThreadMessageIntake.ts @@ -1,14 +1,24 @@ import { remapComposerContextAttachments } from "@t3tools/shared/composerContextReferences"; import { appendUserInputAttachmentPaths } from "../provider/userInputAttachments.ts"; -import type { ChatAttachment, OrchestrationV2Command } from "@t3tools/contracts"; +import type { + ChatAttachment, + ModelSelection, + OrchestrationV2Command, + ProjectId, +} from "@t3tools/contracts"; +import { availableAccountSelection } from "@t3tools/shared/usageLimits"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Schema from "effect/Schema"; import { resolveAttachmentPath } from "../attachmentStore.ts"; import * as ServerConfig from "../config.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as ServerSettings from "../serverSettings.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as AttachmentClaims from "./AttachmentClaims.ts"; +import { AccountFallback } from "./CommandPolicy.ts"; import * as ThreadLaunch from "./ThreadLaunchService.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; @@ -85,11 +95,65 @@ const withGuestAttachments = Effect.fn("ThreadMessageIntake.withGuestAttachments return { ...command, attachments: message?.attachments ?? [] }; }); +/** Null unless the user lets sends move off an account that is out of usage. */ +const readAccountChoice = Effect.fn("ThreadMessageIntake.readAccountChoice")(function* () { + const settings = yield* ServerSettings.ServerSettingsService.pipe( + Effect.flatMap((service) => service.getSettings), + Effect.orElseSucceed(() => null), + ); + if (!settings?.fallbackToAvailableAccounts) return null; + const providers = yield* ProviderRegistry.ProviderRegistry.pipe( + Effect.flatMap((registry) => registry.getProviders), + ); + const now = DateTime.toEpochMillis(yield* DateTime.now); + return (selection: ModelSelection, projectId: ProjectId) => + availableAccountSelection({ selection, providers, now, settings, projectId }) ?? selection; +}); + +const withAvailableAccount = Effect.fn("ThreadMessageIntake.withAvailableAccount")(function* ( + command: Extract, +) { + // Steering stays on the active run's account; server continuations validate their own. + if ( + command.dispatchMode.type === "steer_active" || + command.dispatchMode.type === "restart_active" || + command.manualContinuationOfRunId !== undefined || + command.usageLimitContinuationOfRunId !== undefined || + command.restartContinuationOfRunId !== undefined + ) { + return { command, fallback: undefined }; + } + const choose = yield* readAccountChoice(); + if (choose === null) return { command, fallback: undefined }; + const threads = yield* ThreadManagement.ThreadManagementService; + const records = yield* threads + .getThreadRecords(command.threadId, []) + .pipe(Effect.orElseSucceed(() => null)); + if (records === null) return { command, fallback: undefined }; + const { thread } = records; + const selection = command.modelSelection ?? thread.modelSelection; + const next = choose(selection, thread.projectId); + return { + command: next === selection ? command : { ...command, modelSelection: next }, + fallback: + next === selection + ? undefined + : { + commandId: command.commandId, + threadSelection: thread.modelSelection, + requestedSelection: command.modelSelection, + }, + }; +}); + export const dispatchCommand = Effect.fn("ThreadMessageIntake.dispatchCommand")(function* ( incoming: OrchestrationV2Command, ) { - const command = - incoming.type === "message.dispatch" ? yield* withGuestAttachments(incoming) : incoming; + const prepared = + incoming.type === "message.dispatch" + ? yield* withAvailableAccount(yield* withGuestAttachments(incoming)) + : { command: incoming, fallback: undefined }; + const { command } = prepared; const threads = yield* ThreadManagement.ThreadManagementService; if (command.type === "runtime-request.respond" && command.attachmentsByQuestionId) { const config = yield* ServerConfig.ServerConfig; @@ -197,6 +261,7 @@ export const dispatchCommand = Effect.fn("ThreadMessageIntake.dispatchCommand")( : {}), }) .pipe( + Effect.provideService(AccountFallback, prepared.fallback), Effect.tap((result) => releaseUnusedClaims( claimed.claimedPaths, @@ -229,9 +294,14 @@ export const sendToThread = Effect.fn("ThreadMessageIntake.sendToThread")(functi }); export const launchThread = Effect.fn("ThreadMessageIntake.launchThread")(function* ( - input: ThreadLaunch.ThreadLaunchInput, + requested: ThreadLaunch.ThreadLaunchInput, ) { const launches = yield* ThreadLaunch.ThreadLaunchService; + const choose = yield* readAccountChoice(); + const input = + choose === null + ? requested + : { ...requested, modelSelection: choose(requested.modelSelection, requested.projectId) }; yield* AttachmentClaims.validateAttachmentLimits(input.initialMessage?.attachments ?? []); if (!input.initialMessage?.attachments.some(AttachmentClaims.attachmentIsPendingUpload)) { return yield* launches.launch(input); diff --git a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts index 430018adcb06..aef259b8b781 100644 --- a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts +++ b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts @@ -1,5 +1,14 @@ import { assert, describe, it } from "@effect/vitest"; -import { RunId, ThreadId, type OrchestrationV2DomainEvent } from "@t3tools/contracts"; +import { + DEFAULT_SERVER_SETTINGS, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + RunId, + ThreadId, + type OrchestrationV2DomainEvent, + type ServerProvider, +} from "@t3tools/contracts"; import * as Clock from "effect/Clock"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -8,13 +17,14 @@ import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; import * as UsageLimitRecoveryWorker from "./UsageLimitRecoveryWorker.ts"; -import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts"; +import { limitRecoveryCommand, limitSwitchCommand } from "./UsageLimitRecoveryWorker.ts"; it.effect("idle sweeps skip SQL until the next wake time or a relevant event", () => Effect.gen(function* () { @@ -39,6 +49,7 @@ it.effect("idle sweeps skip SQL until the next wake time or a relevant event", ( ), }), Layer.mock(ThreadManagement.ThreadManagementService)({}), + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), Layer.mock(Orchestrator.OrchestratorV2)({ streamDomainEvents: Stream.fromQueue(events), }), @@ -77,6 +88,11 @@ const failedThread = (failedAgoMs: number, resetAgoMs: number) => { const failedAt = DateTime.makeUnsafe(nowMs - failedAgoMs); const candidate: ProjectionStore.ProjectionLimitRecoveryCandidate = { id: ThreadId.make("thread:limited"), + projectId: ProjectId.make("project:limited"), + modelSelection: { + instanceId: ProviderInstanceId.make("claude_work"), + model: "claude-opus-5-5", + }, status: "failed", lastErrorClass: "usage_limit", usageLimitResetAt: DateTime.formatIso(DateTime.makeUnsafe(nowMs - resetAgoMs)), @@ -116,3 +132,67 @@ describe("limitRecoveryCommand arming", () => { assert.isNull(limitRecoveryCommand(thread, true, nowMs)); }); }); + +describe("limitSwitchCommand", () => { + const claude = (id: string, usedPercent: number): ServerProvider => ({ + instanceId: ProviderInstanceId.make(id), + driver: ProviderDriverKind.make("claudeAgent"), + enabled: true, + installed: true, + version: null, + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-10-19T11:00:00.000Z", + models: [{ slug: "claude-opus-5-5", name: "Opus 5.5", isCustom: false, capabilities: null }], + slashCommands: [], + skills: [], + usageLimits: { + checkedAt: "2026-10-19T11:00:00.000Z", + windows: [{ id: "five_hour", kind: "session", label: "Session", usedPercent }], + }, + }); + const settings = { + ...DEFAULT_SERVER_SETTINGS, + fallbackToAvailableAccounts: true, + providerInstances: { + [ProviderInstanceId.make("claude_work")]: { + driver: ProviderDriverKind.make("claudeAgent"), + enabled: true, + }, + [ProviderInstanceId.make("claude_personal")]: { + driver: ProviderDriverKind.make("claudeAgent"), + enabled: true, + }, + }, + }; + // The work account's bars are stale: the turn that just failed is the proof it is spent. + const providers = [claude("claude_work", 40), claude("claude_personal", 20)]; + const input = { providers, settings, nowMs, limitedUntil: new Map() }; + const justLimited = failedThread(60_000, -HOUR_MS); + + it("continues the same model on another account with usage left", () => { + assert.deepInclude(limitSwitchCommand(justLimited, input), { + type: "message.dispatch", + manualContinuationOfRunId: RunId.make("run:limited"), + modelSelection: { + instanceId: ProviderInstanceId.make("claude_personal"), + model: "claude-opus-5-5", + }, + }); + }); + + it.each([ + ["the setting is off", justLimited, { ...input, settings: DEFAULT_SERVER_SETTINGS }], + ["the limit was hit long ago", failedThread(HOUR_MS, -HOUR_MS), input], + [ + "the other account hit its own limit here and has not reset", + justLimited, + { + ...input, + limitedUntil: new Map([[ProviderInstanceId.make("claude_personal"), nowMs + HOUR_MS]]), + }, + ], + ])("waits when %s", (_name, thread, options) => { + assert.isNull(limitSwitchCommand(thread, options)); + }); +}); diff --git a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts index 3992d46e6bf9..e64c5ba182fa 100644 --- a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts +++ b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts @@ -1,4 +1,12 @@ -import { CommandId, MessageId, type OrchestrationV2Command } from "@t3tools/contracts"; +import { + CommandId, + MessageId, + type OrchestrationV2Command, + type ProviderInstanceId, + type ServerProvider, + type ServerSettings as ServerSettingsValue, +} from "@t3tools/contracts"; +import { availableAccountSelection } from "@t3tools/shared/usageLimits"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -6,13 +14,70 @@ import * as Stream from "effect/Stream"; import type { OrchestrationV2DomainEvent } from "@t3tools/contracts"; import * as DueHint from "../scheduling/DueHint.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; import { forkParked } from "../serverActivation.ts"; import * as ServerSettings from "../serverSettings.ts"; +import { AccountFallback } from "./CommandPolicy.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; export const MAX_OVERDUE_ARM_MS = 24 * 60 * 60 * 1000; +// Only limits hit this recently move accounts, so turning the setting on never wakes old threads. +const ACCOUNT_SWITCH_WINDOW_MS = 10 * 60 * 1000; + +/** Continues a thread that just hit a usage limit on another account with usage left. */ +export function limitSwitchCommand( + thread: ProjectionStore.ProjectionLimitRecoveryCandidate, + input: { + readonly providers: ReadonlyArray; + readonly settings: ServerSettingsValue; + readonly nowMs: number; + /** Accounts that hit a limit here until the given reset, whatever their stale bars say. */ + readonly limitedUntil: ReadonlyMap; + }, +): OrchestrationV2Command | null { + if ( + thread.status !== "failed" || + thread.lastErrorClass !== "usage_limit" || + !thread.latestRunId || + thread.archivedAt !== null || + thread.settledOverride === "settled" || + thread.pendingRuntimeRequest !== null || + input.nowMs - DateTime.toEpochMillis(thread.latestRunCompletedAt ?? thread.updatedAt) > + ACCOUNT_SWITCH_WINDOW_MS || + (thread.snoozedUntil != null && DateTime.toEpochMillis(thread.snoozedUntil) > input.nowMs) + ) + return null; + const selection = thread.modelSelection; + const next = availableAccountSelection({ + selection, + providers: input.providers.filter( + (provider) => + provider.instanceId === selection.instanceId || + (input.limitedUntil.get(provider.instanceId) ?? 0) <= input.nowMs, + ), + now: input.nowMs, + settings: input.settings, + projectId: thread.projectId, + exhausted: true, + }); + if (next === null) return null; + const identity = `${thread.id}:${thread.latestRunId}`; + return { + type: "message.dispatch", + commandId: CommandId.make(`limit-switch:${identity}`), + messageId: MessageId.make(`limit-switch:${identity}`), + threadId: thread.id, + manualContinuationOfRunId: thread.latestRunId, + modelSelection: next, + text: "Continue where you left off.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "server", + }; +} export function limitRecoveryCommand( thread: ProjectionStore.ProjectionLimitRecoveryCandidate, @@ -92,6 +157,10 @@ const makeSweep = Effect.gen(function* () { const threads = yield* ThreadManagement.ThreadManagementService; const orchestrator = yield* Orchestrator.OrchestratorV2; const settings = yield* ServerSettings.ServerSettingsService; + const registry = yield* ProviderRegistry.ProviderRegistry; + const limitedUntil = new Map(); + // A rejected switch is not retried every tick; the thread falls back to waiting for the reset. + const attemptedSwitches = new Set(); const due = DueHint.make(); yield* forkParked( Stream.runForEach(orchestrator.streamDomainEvents, (event) => @@ -109,9 +178,43 @@ const makeSweep = Effect.gen(function* () { now, autoResume: preferences.autoResumeLimitedThreads, snooze: preferences.snoozeLimitedThreads, + ...(preferences.fallbackToAvailableAccounts + ? { switchAccountsSince: DateTime.makeUnsafe(nowMs - ACCOUNT_SWITCH_WINDOW_MS) } + : {}), }); + const providers = preferences.fallbackToAvailableAccounts ? yield* registry.getProviders : []; let acted = false; for (const thread of candidates) { + const switchCommand = preferences.fallbackToAvailableAccounts + ? limitSwitchCommand(thread, { providers, settings: preferences, nowMs, limitedUntil }) + : null; + if (switchCommand !== null && !attemptedSwitches.has(switchCommand.commandId)) { + attemptedSwitches.add(switchCommand.commandId); + acted = true; + const resetMs = Date.parse(thread.usageLimitResetAt ?? ""); + yield* threads.dispatch(switchCommand).pipe( + Effect.provideService(AccountFallback, { + commandId: switchCommand.commandId, + threadSelection: thread.modelSelection, + requestedSelection: undefined, + }), + Effect.tap(() => + Effect.sync(() => + limitedUntil.set( + thread.modelSelection.instanceId, + Number.isFinite(resetMs) ? resetMs : nowMs + ACCOUNT_SWITCH_WINDOW_MS, + ), + ), + ), + Effect.catchCause((cause) => + Effect.logWarning("orchestration-v2.limit-recovery.account-switch-failed", { + threadId: thread.id, + cause, + }), + ), + ); + continue; + } const command = limitRecoveryCommand( thread, preferences.autoResumeLimitedThreads, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index d488c76f313b..bc4b9e39d045 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -8,6 +8,7 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { vi } from "vite-plus/test"; +import { AccountFallback } from "./CommandPolicy.ts"; import { type ApplicationStoredEvent, CheckpointId, @@ -1771,18 +1772,30 @@ it.layer(layerTest)("OrchestrationV2LayerLive", (it) => { const steerCommandId = CommandId.make("runtime-delivery-intent-auto"); const steerMessageId = MessageId.make("runtime-delivery-intent-auto"); - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: "user", - creationSource: "web", - commandId: steerCommandId, - threadId, - messageId: steerMessageId, - text: "Include this in the active work.", - attachments: [], - dispatchMode: { type: "start_immediately" }, - deliveryIntent: "auto", - }); + yield* orchestrator + .dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: steerCommandId, + threadId, + messageId: steerMessageId, + text: "Include this in the active work.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + deliveryIntent: "auto", + modelSelection: { + ...modelSelection, + instanceId: ProviderInstanceId.make("exhausted-fallback"), + }, + }) + .pipe( + Effect.provideService(AccountFallback, { + commandId: steerCommandId, + threadSelection: modelSelection, + requestedSelection: undefined, + }), + ); const steered = yield* orchestrator.getThreadProjection(threadId); assert.lengthOf(steered.runs, 1); assert.equal( @@ -11822,6 +11835,32 @@ it.layer(layerTest)("usage-limit recovery", (it) => { yield* orchestrator.dispatch(freshResume!); assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 2); } + if (["stop", "settled", "archive", "replacement"].includes(scenario)) { + const switchId = CommandId.make(`account-switch:${scenario}`); + const result = yield* orchestrator + .dispatch({ + type: "message.dispatch", + commandId: switchId, + threadId, + messageId: MessageId.make(switchId), + manualContinuationOfRunId: run.id, + text: "Continue where you left off.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "server", + }) + .pipe( + Effect.provideService(AccountFallback, { + commandId: switchId, + threadSelection: modelSelection, + requestedSelection: undefined, + }), + Effect.exit, + ); + assert.equal(result._tag, "Failure"); + assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 1); + } const before = yield* orchestrator.getThreadProjection(threadId); if (resume !== null) { yield* orchestrator.dispatch(resume); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 09c6d86c050f..d2bd5dffb38d 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1434,6 +1434,7 @@ export const layerWsRpc = ( | FileSystem.FileSystem | ServerConfig.ServerConfig | ServerSettings.ServerSettingsService + | ProviderRegistry.ProviderRegistry | DocumentText.DocumentText >(); const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 5db96b73fc6b..df76535c4c64 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -2628,6 +2628,9 @@ export default function ChatView(props: ChatViewProps) { const fallbackWindow = fallback ? `${fallback.provider.instanceId}:${fallback.resetsAt ?? ""}` : null; + const fallbackProvider = providerStatuses.find( + (provider) => provider.instanceId === fallbackSelection?.instanceId, + ); const fallbackItems: ComposerBannerStackItem[] = fallback && fallbackSelection && @@ -2636,10 +2639,9 @@ export default function ChatView(props: ChatViewProps) { ? [ modelFallbackBannerItem({ fallbackModelName: - providerStatuses - .find((provider) => provider.instanceId === fallbackSelection.instanceId) - ?.models.find((model) => model.slug === fallbackSelection.model)?.name ?? - fallbackSelection.model, + fallbackProvider?.models.find((model) => model.slug === fallbackSelection.model) + ?.name ?? fallbackSelection.model, + fallbackProviderName: `${fallbackProvider?.displayName?.trim() || fallbackProvider?.driver || "Account"} [${fallbackSelection.instanceId}]`, defaultProviderName: fallback.provider.displayName ?? fallback.provider.driver, defaultModelName: fallback.provider.models.find((model) => model.slug === defaultSelection.model) diff --git a/apps/web/src/components/chat/ComposerUsageLimits.tsx b/apps/web/src/components/chat/ComposerUsageLimits.tsx index b46614a8b418..6aab26765600 100644 --- a/apps/web/src/components/chat/ComposerUsageLimits.tsx +++ b/apps/web/src/components/chat/ComposerUsageLimits.tsx @@ -71,6 +71,7 @@ export function usageLimitsBannerItem( export function modelFallbackBannerItem({ fallbackModelName, + fallbackProviderName, defaultProviderName, defaultModelName, resetsAt, @@ -79,6 +80,7 @@ export function modelFallbackBannerItem({ onDismiss, }: { readonly fallbackModelName: string; + readonly fallbackProviderName: string; readonly defaultProviderName: string; readonly defaultModelName: string; readonly resetsAt: string | null; @@ -90,7 +92,7 @@ export function modelFallbackBannerItem({ id: "fallback-model", variant: "info", icon: , - title: `Using ${fallbackModelName}`, + title: `Using ${fallbackModelName} on ${fallbackProviderName}`, description: `${defaultProviderName} is out of usage${resetsAt ? ` · resets ${formatUpcomingTimestamp(resetsAt, timestampFormat)}` : ""}`, actions: (