diff --git a/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx index 793e3ae75412c..3504f8c3f9413 100644 --- a/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx @@ -61,6 +61,7 @@ const PAGE_PROJECT_KEYS: Record - {!projectSelected ? ( - - writeToAll({ autoResumeLimitedThreads: value })} - /> - writeToAll({ snoozeLimitedThreads: value })} - /> - - ) : null} + + writeToAll({ autoResumeLimitedThreads: value })} + /> + writeToAll({ snoozeLimitedThreads: value })} + /> + { expect(uniformMobileSetting([], key)).toBeNull(); }, ); + + it("overrides resume per project while snooze stays environment-wide", () => { + const settings: ServerSettings = { + ...DEFAULT_SERVER_SETTINGS, + autoResumeLimitedThreads: false, + }; + const targets = resolveMobileSettingsTargets( + [environment(firstId, settings)], + [{ environmentId: firstId, id: firstProject }], + ); + + expect( + planMobileScopedSettingsPatch(targets, true, { autoResumeLimitedThreads: true }), + ).toEqual([ + { + environmentId: firstId, + patch: { projectSettingsOverrides: { [firstProject]: { autoResumeLimitedThreads: true } } }, + }, + ]); + expect(planMobileScopedSettingsPatch(targets, true, { snoozeLimitedThreads: true })).toEqual( + [], + ); + }); }); function environment(environmentId: EnvironmentId, settings: ServerSettings): SettingsTarget { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 8ff160f0cc2ac..6e8aeb9ba1d83 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -5795,6 +5795,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { if (lastErrorClass === "usage_limit") { assert.deepEqual(candidate, { id: sqlShell.id, + projectId: sqlShell.projectId, status: sqlShell.status, lastErrorClass: sqlShell.lastErrorClass, usageLimitResetAt: sqlShell.usageLimitResetAt, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index dea68ee7723d1..e1d43ced2564b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -175,6 +175,7 @@ export type ProjectionRecoveryKind = export type ProjectionLimitRecoveryCandidate = Pick< OrchestrationV2ThreadShell, | "id" + | "projectId" | "status" | "lastErrorClass" | "latestRunId" @@ -548,7 +549,10 @@ export interface ProjectionStoreV2Shape { ) => Effect.Effect; readonly getLimitRecoveryCandidates: (options: { readonly now: DateTime.Utc; + /** Arms unarmed candidates by default; the project lists flip it, and the worker still decides per candidate. */ readonly autoResume: boolean; + readonly autoResumeEnabledProjectIds?: ReadonlyArray; + readonly autoResumeDisabledProjectIds?: ReadonlyArray; readonly snooze: boolean; }) => Effect.Effect, ProjectionStoreV2Error>; /** Epoch ms of the next armed reset or snooze end after `now`, when a waiting thread becomes a candidate. */ @@ -4053,7 +4057,11 @@ export const layer: Layer.Layer = OR json_extract(t.payload_json, '$.limitRecovery.resetAt') IS NOT json_extract(item.payload_json, '$.failure.resetAt') ) AND ( - ${booleanInt(options.autoResume)} + CASE + WHEN t.project_id IN (SELECT value FROM json_each(${encodeIdList(options.autoResumeDisabledProjectIds ?? [])})) THEN 0 + WHEN t.project_id IN (SELECT value FROM json_each(${encodeIdList(options.autoResumeEnabledProjectIds ?? [])})) THEN 1 + ELSE ${booleanInt(options.autoResume)} + END OR (${booleanInt(options.snooze)} AND julianday(json_extract(item.payload_json, '$.failure.resetAt')) > julianday(${DateTime.formatIso(options.now)})) ) -- A subagent of a settled, archived or deleted parent is abandoned work. @@ -4086,6 +4094,7 @@ export const layer: Layer.Layer = if (summary.lastErrorClass !== "usage_limit") continue; candidates.push({ id: thread.id, + projectId: thread.projectId, status: "failed", lastErrorClass: summary.lastErrorClass, usageLimitResetAt: summary.usageLimitResetAt, diff --git a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts index 430018adcb06a..ceed3bbda898f 100644 --- a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts +++ b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts @@ -1,5 +1,5 @@ import { assert, describe, it } from "@effect/vitest"; -import { RunId, ThreadId, type OrchestrationV2DomainEvent } from "@t3tools/contracts"; +import { ProjectId, RunId, ThreadId, type OrchestrationV2DomainEvent } from "@t3tools/contracts"; import * as Clock from "effect/Clock"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -46,6 +46,7 @@ it.effect("idle sweeps skip SQL until the next wake time or a relevant event", ( getSettings: Effect.succeed({ autoResumeLimitedThreads: true, snoozeLimitedThreads: false, + projectSettingsOverrides: {}, } as never), streamChanges: Stream.never, }), @@ -77,6 +78,7 @@ 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"), status: "failed", lastErrorClass: "usage_limit", usageLimitResetAt: DateTime.formatIso(DateTime.makeUnsafe(nowMs - resetAgoMs)), diff --git a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts index 985815bd73eb4..8542711951228 100644 --- a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts +++ b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts @@ -1,9 +1,16 @@ -import { CommandId, MessageId, type OrchestrationV2Command } from "@t3tools/contracts"; +import { + CommandId, + MessageId, + type OrchestrationV2Command, + type OrchestrationV2DomainEvent, + ProjectId, + type ServerSettings as ServerSettingsValue, +} from "@t3tools/contracts"; +import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; 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 { forkParked } from "../serverActivation.ts"; @@ -14,6 +21,33 @@ import * as ThreadManagement from "./ThreadManagementService.ts"; const MAX_OVERDUE_ARM_MS = 24 * 60 * 60 * 1000; +/** The one place a limited thread's recovery policy is resolved. */ +export function resolveLimitRecoveryPolicy( + settings: ServerSettingsValue, + thread: Pick, +): { readonly autoResume: boolean; readonly snooze: boolean } { + return { + autoResume: resolveProjectSettings(settings, thread.projectId).settings + .autoResumeLimitedThreads, + snooze: settings.snoozeLimitedThreads, + }; +} + +/** Narrows the sweep's query to candidates whose project could auto-arm. */ +function limitAutoResumeScope(settings: ServerSettingsValue) { + const enabled: Array = []; + const disabled: Array = []; + for (const [projectId, overrides] of Object.entries(settings.projectSettingsOverrides)) { + if (overrides.autoResumeLimitedThreads === true) enabled.push(ProjectId.make(projectId)); + if (overrides.autoResumeLimitedThreads === false) disabled.push(ProjectId.make(projectId)); + } + return { + autoResume: settings.autoResumeLimitedThreads, + autoResumeEnabledProjectIds: enabled, + autoResumeDisabledProjectIds: disabled, + }; +} + export function limitRecoveryCommand( thread: ProjectionStore.ProjectionLimitRecoveryCandidate, autoResume: boolean, @@ -87,7 +121,7 @@ const affectsLimitRecovery = (event: OrchestrationV2DomainEvent) => event.type.startsWith("provider-session.") || (event.type === "turn-item.updated" && event.payload.type === "error"); -const makeSweep = Effect.gen(function* () { +export const makeSweep = Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; const threads = yield* ThreadManagement.ThreadManagementService; const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -107,17 +141,13 @@ const makeSweep = Effect.gen(function* () { const preferences = yield* settings.getSettings; const candidates = yield* projections.getLimitRecoveryCandidates({ now, - autoResume: preferences.autoResumeLimitedThreads, + ...limitAutoResumeScope(preferences), snooze: preferences.snoozeLimitedThreads, }); let acted = false; for (const thread of candidates) { - const command = limitRecoveryCommand( - thread, - preferences.autoResumeLimitedThreads, - nowMs, - preferences.snoozeLimitedThreads, - ); + const policy = resolveLimitRecoveryPolicy(preferences, thread); + const command = limitRecoveryCommand(thread, policy.autoResume, nowMs, policy.snooze); if (command === null) continue; acted = true; yield* Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 170ceee73a331..98ef898be82d5 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -1,6 +1,10 @@ import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider"; import { SourceControlProviderKind } from "@t3tools/contracts"; -import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts"; +import { + limitRecoveryCommand, + makeSweep as makeLimitRecoverySweep, + resolveLimitRecoveryPolicy, +} from "./UsageLimitRecoveryWorker.ts"; import { createOrchestrationV2TurnItemVisibility } from "@t3tools/shared/orchestrationV2Timeline"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts"; @@ -14,6 +18,7 @@ import { CheckpointRef, CommandId, ContextTransferId, + DEFAULT_SERVER_SETTINGS, EventId, MessageId, NodeId, @@ -7977,6 +7982,50 @@ it.layer(layerSharedApplicationDataPlaneTest)("shared application data plane", ( }); it.layer(layerTest)("usage-limit recovery", (it) => { + it.effect.each([ + { environment: false, project: true, arms: true }, + { environment: true, project: false, arms: false }, + { environment: true, project: undefined, arms: true }, + { environment: false, project: undefined, arms: false }, + ])( + "auto-arms with environment $environment and project override $project: $arms", + ({ environment, project, arms }) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const projectId = ProjectId.make("limit-policy:project"); + const settings = { + ...DEFAULT_SERVER_SETTINGS, + autoResumeLimitedThreads: environment, + projectSettingsOverrides: + project === undefined ? {} : { [projectId]: { autoResumeLimitedThreads: project } }, + }; + const candidate: ProjectionStore.ProjectionLimitRecoveryCandidate = { + id: ThreadId.make("limit-policy:thread"), + projectId, + status: "failed", + lastErrorClass: "usage_limit", + latestRunId: RunId.make("limit-policy:run"), + usageLimitResetAt: DateTime.formatIso(DateTime.add(now, { hours: 1 })), + archivedAt: null, + settledOverride: null, + pendingRuntimeRequest: null, + latestRunCompletedAt: now, + updatedAt: now, + limitRecovery: null, + snoozedUntil: null, + }; + const policy = resolveLimitRecoveryPolicy(settings, candidate); + const arm = limitRecoveryCommand( + candidate, + policy.autoResume, + DateTime.toEpochMillis(now), + policy.snooze, + ); + if (!arms) return assert.isNull(arm); + assert(arm?.type === "thread.metadata.update"); + assert.isTrue(arm.limitRecovery?.autoResume); + }), + ); it.effect.each(["interrupted", "usage_limit"] as const)( "manually resumes an %s run ahead of its queued message only once", (reason) => @@ -8785,6 +8834,230 @@ it.layer(layerTest)("usage-limit recovery", (it) => { ); }); +const seedLimitedThread = Effect.fn("test.seedLimitedThread")(function* ( + key: string, + options: { readonly resetInPast?: boolean } = {}, +) { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const events = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make(`limit-policy:${key}`); + const projectId = ProjectId.make(`limit-policy:project:${key}`); + const now = yield* DateTime.now; + yield* seedProject({ + projectId, + title: "Limit policy project", + workspaceRoot: process.cwd(), + defaultModelSelection: modelSelection, + createdAt: DateTime.formatIso(now), + }); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`limit-policy:create:${key}`), + threadId, + projectId, + title: "Limited thread", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`limit-policy:message:${key}`), + threadId, + messageId: MessageId.make(`limit-policy:message:${key}`), + text: "Work on this.", + attachments: [], + dispatchMode: { type: "defer_start" }, + createdBy: "user", + creationSource: "web", + }); + const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + const completedAt = options.resetInPast ? DateTime.subtract(now, { minutes: 10 }) : now; + const resetAt = DateTime.formatIso( + options.resetInPast ? DateTime.subtract(now, { minutes: 5 }) : DateTime.add(now, { hours: 1 }), + ); + yield* events.write({ + commandId: CommandId.make(`limit-policy:failure:${key}`), + events: [ + { + id: EventId.make(`limit-policy:run:${key}`), + type: "run.updated", + threadId, + occurredAt: completedAt, + payload: { ...run, status: "failed", completedAt }, + }, + { + id: EventId.make(`limit-policy:error:${key}`), + type: "turn-item.updated", + threadId, + occurredAt: completedAt, + payload: { + id: TurnItemId.make(`limit-policy:error:${key}`), + type: "error", + threadId, + runId: run.id, + nodeId: run.rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "failed", + title: "Usage limit reached", + startedAt: completedAt, + completedAt, + updatedAt: completedAt, + failure: { + class: "usage_limit", + message: "Plan limit reached.", + code: "usageLimitExceeded", + retryable: null, + resetAt, + }, + }, + }, + ], + }); + return { threadId, projectId, runId: run.id, resetAt }; +}); + +const sweepLimitRecovery = Effect.fn("test.sweepLimitRecovery")(function* ( + patch: Parameters[0], +) { + yield* (yield* ServerSettings.ServerSettingsService).updateSettings(patch); + yield* (yield* makeLimitRecoverySweep)(); +}); + +const limitRecoveryOf = Effect.fn("test.limitRecoveryOf")(function* (threadId: ThreadId) { + const orchestrator = yield* Orchestrator.OrchestratorV2; + return (yield* orchestrator.getThreadProjection(threadId)).thread.limitRecovery ?? null; +}); + +// Same as layerTest, but the sweep under test needs to change the settings the runtime reads. +const layerLimitPolicyTest = Layer.mergeAll( + RuntimeLayer.layer, + RuntimeLayer.layerEventSink, + ProjectStore.layer, + ProjectionStore.layer, + EffectOutbox.layer, + ThreadCommandExecutor.layer, +).pipe( + Layer.provide(McpSessionRegistryTestkit.layer), + Layer.provide(McpProviderSessions.layer), + Layer.provide(SqlitePersistence.layerMemory), + Layer.provide(layerCheckpointStoreTest), + Layer.provide(layerServerConfig), + Layer.provideMerge(ServerSettings.layerTest()), + Layer.provide(layerTestProviderInstanceRegistry), + Layer.provide(layerGitWorkflowTest), + Layer.provide(layerProjectServiceTest), + Layer.provide(layerPlatformTest), +); + +it.layer(layerLimitPolicyTest)("usage-limit recovery project policy", (it) => { + it.effect("keeps a manually armed resume when the project turns auto-resume off", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const thread = yield* seedLimitedThread("manual-arm", { resetInPast: true }); + yield* orchestrator.dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make("limit-policy:manual-arm"), + threadId: thread.threadId, + limitRecovery: { runId: thread.runId, resetAt: thread.resetAt, autoResume: true }, + }); + const before = (yield* orchestrator.getThreadProjection(thread.threadId)).runs.length; + yield* sweepLimitRecovery({ + autoResumeLimitedThreads: false, + projectSettingsOverrides: { + [thread.projectId]: { autoResumeLimitedThreads: false }, + }, + }); + assert.lengthOf((yield* orchestrator.getThreadProjection(thread.threadId)).runs, before + 1); + }), + ); + + it.effect("only snoozes when the project turns auto-resume off and snooze is on", () => + Effect.gen(function* () { + const thread = yield* seedLimitedThread("snooze-only"); + yield* sweepLimitRecovery({ + autoResumeLimitedThreads: true, + snoozeLimitedThreads: true, + projectSettingsOverrides: { + [thread.projectId]: { autoResumeLimitedThreads: false }, + }, + }); + assert.deepInclude(yield* limitRecoveryOf(thread.threadId), { + runId: thread.runId, + resetAt: thread.resetAt, + autoResume: false, + snooze: true, + }); + }), + ); + + it.effect("arms only the project that turns auto-resume on", () => + Effect.gen(function* () { + const enabled = yield* seedLimitedThread("enabled-project"); + const other = yield* seedLimitedThread("other-project"); + const projections = yield* ProjectionStore.ProjectionStoreV2; + const candidates = yield* projections.getLimitRecoveryCandidates({ + now: yield* DateTime.now, + autoResume: false, + autoResumeEnabledProjectIds: [enabled.projectId], + snooze: false, + }); + assert.deepEqual( + candidates + .map((row) => row.id) + .filter((id) => id === enabled.threadId || id === other.threadId), + [enabled.threadId], + ); + yield* sweepLimitRecovery({ + autoResumeLimitedThreads: false, + snoozeLimitedThreads: false, + projectSettingsOverrides: { + [enabled.projectId]: { autoResumeLimitedThreads: true }, + }, + }); + assert.deepInclude(yield* limitRecoveryOf(enabled.threadId), { autoResume: true }); + assert.isNull(yield* limitRecoveryOf(other.threadId)); + }), + ); + + it.effect("skips the project that turns environment auto-resume off", () => + Effect.gen(function* () { + const disabled = yield* seedLimitedThread("disabled-project"); + const inherited = yield* seedLimitedThread("inherited-project"); + const projections = yield* ProjectionStore.ProjectionStoreV2; + const candidates = yield* projections.getLimitRecoveryCandidates({ + now: yield* DateTime.now, + autoResume: true, + autoResumeDisabledProjectIds: [disabled.projectId], + snooze: false, + }); + assert.deepEqual( + candidates + .map((row) => row.id) + .filter((id) => id === disabled.threadId || id === inherited.threadId), + [inherited.threadId], + ); + yield* sweepLimitRecovery({ + autoResumeLimitedThreads: true, + snoozeLimitedThreads: false, + projectSettingsOverrides: { + [disabled.projectId]: { autoResumeLimitedThreads: false }, + }, + }); + assert.isNull(yield* limitRecoveryOf(disabled.threadId)); + assert.deepInclude(yield* limitRecoveryOf(inherited.threadId), { autoResume: true }); + }), + ); +}); + const queueResumeDisabledInstanceProjectId = ProjectId.make( "project:queue-resume-disabled-instance", ); diff --git a/apps/web/src/components/settings/settingsSearch.ts b/apps/web/src/components/settings/settingsSearch.ts index 60c89ffc6b732..730e140b87e6a 100644 --- a/apps/web/src/components/settings/settingsSearch.ts +++ b/apps/web/src/components/settings/settingsSearch.ts @@ -446,6 +446,7 @@ export const SETTINGS_SEARCH_ITEMS = [ id: "auto-resume-limited-threads", title: "Resume after usage limits", to: "/settings/threads", + scope: "project-defaults", searchTerms: ["usage quota rate limit reset recover continue"], }, { diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 94c9cc03c8de9..ee0c5aafaffc9 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -256,6 +256,7 @@ the continuation finishes. If the queue was held by a restart, resume it then. When the provider reports a reset time, choose **Resume at reset** to schedule a continuation. You can cancel it from the thread. Enable **Resume after usage limits** in **Settings → Threads** to schedule limit stops by default. +Select a project there to turn it on or off for that project only. The environment must be running when the reset arrives; it resumes overdue continuations after a restart. Stopping the thread, sending a new message, archiving, or settling it prevents a pending continuation from starting. diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index f6409a87074fd..37c40452fd17a 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -1104,6 +1104,7 @@ export const PROJECT_SCOPED_SERVER_SETTING_KEYS = [ "sidebarAutoSettleAfterDays", "sidebarAutoSettleKeepPullRequests", "continueThreadsAfterServerUpdate", + "autoResumeLimitedThreads", "responseStreamingMode", "speechProjectCustomWords", ] as const; @@ -1142,6 +1143,7 @@ export const ProjectSettingsOverrides = Schema.Struct({ sidebarAutoSettleAfterDays: Schema.optionalKey(Schema.NullOr(SidebarAutoSettleAfterDays)), sidebarAutoSettleKeepPullRequests: Schema.optionalKey(Schema.Boolean), continueThreadsAfterServerUpdate: Schema.optionalKey(Schema.Boolean), + autoResumeLimitedThreads: Schema.optionalKey(Schema.Boolean), responseStreamingMode: Schema.optionalKey(ResponseStreamingMode), speechProjectCustomWords: Schema.optionalKey(SpeechCustomWords), providerInstances: Schema.optionalKey(