From 97d1304f369cadbf0b7ae0ac3dad6398d29540a7 Mon Sep 17 00:00:00 2001 From: Arya Buddha <40647186+AryaBuddha@users.noreply.github.com> Date: Fri, 18 Sep 2026 11:00:36 -0400 Subject: [PATCH] feat: continue a stopped thread when its usage limit resets When Codex or Claude stops a turn on a usage limit, the thread error only carried the rendered sentence, so the user had to come back and resend by hand. The failed turn now carries the provider-reported reset instant (turn.completed usageLimitResetsAt -> session.lastErrorLimitResetsAt), the error banner offers "Continue when limit resets", and a persisted per-thread schedule (thread.auto-continue.set/.clear, autoContinueAt) is fired by a new sweep reactor that starts a Continue turn once the reset passes. The schedule survives restarts, is cancellable from the same banner on web and mobile, and any manual turn, settle, or archive clears it. Built by Claude Fable 5 via Cursor. Co-authored-by: Cursor --- .../features/threads/AutoContinueBanner.tsx | 95 ++++++ .../features/threads/ThreadDetailScreen.tsx | 5 + .../OrchestrationEngineHarness.integration.ts | 7 + .../src/environment/ServerEnvironment.ts | 1 + .../orchestration/AutoContinueReactor.test.ts | 165 ++++++++++ .../src/orchestration/AutoContinueReactor.ts | 106 ++++++ .../Layers/OrchestrationReactor.test.ts | 11 + .../Layers/OrchestrationReactor.ts | 3 + .../Layers/ProjectionPipeline.ts | 32 ++ .../Layers/ProjectionSnapshotQuery.test.ts | 4 + .../Layers/ProjectionSnapshotQuery.ts | 17 + .../Layers/ProviderRuntimeIngestion.test.ts | 61 ++++ .../Layers/ProviderRuntimeIngestion.ts | 14 + apps/server/src/orchestration/Schemas.ts | 4 + .../decider.autoContinue.test.ts | 310 ++++++++++++++++++ apps/server/src/orchestration/decider.ts | 183 ++++++++++- .../src/orchestration/projector.test.ts | 1 + apps/server/src/orchestration/projector.ts | 35 ++ .../Layers/ProjectionThreadSessions.ts | 4 + .../persistence/Layers/ProjectionThreads.ts | 5 + apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/054_AutoContinue.ts | 26 ++ .../Services/ProjectionThreadSessions.ts | 4 + .../persistence/Services/ProjectionThreads.ts | 3 + .../src/provider/Layers/ClaudeAdapter.ts | 38 ++- .../src/provider/Layers/CodexAdapter.ts | 4 + .../src/provider/Layers/claudeUsageLimits.ts | 3 +- .../provider/Layers/codexUsageLimits.test.ts | 45 +++ .../src/provider/Layers/codexUsageLimits.ts | 45 ++- apps/server/src/server.ts | 2 + apps/web/src/components/ChatView.tsx | 55 +++- .../src/components/chat/ThreadErrorBanner.tsx | 68 +++- apps/web/src/hooks/useThreadActions.ts | 64 ++++ apps/web/src/state/entities.ts | 9 + docs/user/usage.md | 7 + .../client-runtime/src/operations/commands.ts | 20 ++ .../src/state/threadCommands.ts | 32 ++ .../client-runtime/src/state/threadReducer.ts | 20 ++ packages/contracts/src/environment.ts | 4 + packages/contracts/src/orchestration.ts | 76 +++++ packages/contracts/src/providerRuntime.ts | 4 + 41 files changed, 1562 insertions(+), 32 deletions(-) create mode 100644 apps/mobile/src/features/threads/AutoContinueBanner.tsx create mode 100644 apps/server/src/orchestration/AutoContinueReactor.test.ts create mode 100644 apps/server/src/orchestration/AutoContinueReactor.ts create mode 100644 apps/server/src/orchestration/decider.autoContinue.test.ts create mode 100644 apps/server/src/persistence/Migrations/054_AutoContinue.ts diff --git a/apps/mobile/src/features/threads/AutoContinueBanner.tsx b/apps/mobile/src/features/threads/AutoContinueBanner.tsx new file mode 100644 index 000000000000..9003c4979e39 --- /dev/null +++ b/apps/mobile/src/features/threads/AutoContinueBanner.tsx @@ -0,0 +1,95 @@ +import type { EnvironmentId, OrchestrationThreadShell } from "@t3tools/contracts"; +import { formatDuration } from "@t3tools/shared/usageLimits"; +import { useCallback } from "react"; +import { Pressable, View } from "react-native"; + +import { AppText as Text } from "../../components/AppText"; +import { appAtomRegistry } from "../../state/atom-registry"; +import { environmentServerConfigsAtom } from "../../state/server"; +import { threadEnvironment } from "../../state/threads"; +import { useAtomCommand } from "../../state/use-atom-command"; + +function environmentSupportsAutoContinue(environmentId: EnvironmentId) { + return ( + appAtomRegistry.get(environmentServerConfigsAtom).get(environmentId)?.environment.capabilities + .threadAutoContinue === true + ); +} + +/** + * "Continue when the limit resets" for a thread stopped by a provider usage + * limit: offers scheduling while the session error carries a structured reset + * instant, and shows the pending continuation with a cancel once scheduled. + * Hidden on servers that predate thread.auto-continue (version skew). + */ +export function AutoContinueBanner(props: { + readonly environmentId: EnvironmentId; + readonly thread: OrchestrationThreadShell; +}) { + const setAutoContinue = useAtomCommand(threadEnvironment.setAutoContinue, { + reportFailure: false, + }); + const clearAutoContinue = useAtomCommand(threadEnvironment.clearAutoContinue, { + reportFailure: false, + }); + const { environmentId, thread } = props; + const resetsAt = thread.session?.lastErrorLimitResetsAt ?? null; + const scheduledFor = thread.autoContinueAt ?? null; + + const schedule = useCallback(() => { + if (resetsAt === null) return; + void setAutoContinue({ + environmentId, + input: { threadId: thread.id, autoContinueAt: resetsAt }, + }); + }, [environmentId, resetsAt, setAutoContinue, thread.id]); + const cancel = useCallback(() => { + void clearAutoContinue({ + environmentId, + input: { threadId: thread.id, reason: "user" }, + }); + }, [clearAutoContinue, environmentId, thread.id]); + + if (!environmentSupportsAutoContinue(environmentId)) return null; + const now = Date.now(); + if (scheduledFor !== null) { + const waitMs = Date.parse(scheduledFor) - now; + return ( + + + + {waitMs > 0 + ? `Continuing when the limit resets, in ${formatDuration(waitMs)}` + : "Continuing when the limit resets"} + + + Cancel + + + + ); + } + const isLimitError = + thread.session?.status === "error" && resetsAt !== null && Date.parse(resetsAt) > now; + if (!isLimitError || resetsAt === null) return null; + return ( + + + + Usage limit reached. Resets in {formatDuration(Date.parse(resetsAt) - now)}. + + + Continue on reset + + + + ); +} diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index e7b4650e2117..4a3ff9b27f52 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -80,6 +80,7 @@ import type { PendingUserInputDraftAnswer, ThreadFeedEntry, } from "../../lib/threadActivity"; +import { AutoContinueBanner } from "./AutoContinueBanner"; import { PendingApprovalCard } from "./PendingApprovalCard"; import { ComposerFeedback } from "./ComposerFeedback"; import { ComposerUsageLimits } from "./ComposerUsageLimits"; @@ -1016,6 +1017,10 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread ) : null} + {/* Hidden (not unmounted) while a user-input request owns the composer slot, so composer drafts and editor state survive. A rejected creation has no thread to send to; the failure card diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index ace2b0fc43bf..83b92b8d0ff0 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -68,6 +68,7 @@ import { } from "../src/orchestration/Services/OrchestrationEngine.ts"; import { ThreadDeletionReactor } from "../src/orchestration/Services/ThreadDeletionReactor.ts"; import * as ThreadSettlementReactor from "../src/orchestration/ThreadSettlementReactor.ts"; +import * as AutoContinueReactor from "../src/orchestration/AutoContinueReactor.ts"; import * as PullRequestSyncReactor from "../src/orchestration/PullRequestSyncReactor.ts"; import * as ThreadPullRequestReactor from "../src/orchestration/ThreadPullRequestReactor.ts"; import { OrchestrationReactor } from "../src/orchestration/Services/OrchestrationReactor.ts"; @@ -412,6 +413,12 @@ export const makeOrchestrationIntegrationHarness = ( drain: Effect.void, }), ), + Layer.provideMerge( + Layer.succeed(AutoContinueReactor.AutoContinueReactor, { + start: () => Effect.void, + drain: Effect.void, + }), + ), Layer.provideMerge( Layer.succeed(PullRequestSyncReactor.PullRequestSyncReactor, { start: () => Effect.void, diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index a36c0a03b6f2..c2f3e90a5053 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -230,6 +230,7 @@ export const make = Effect.gen(function* () { threadRestartContinuation: true, projectSettingsOverrides: true, threadSnooze: true, + threadAutoContinue: true, environmentThemes: true, usageLimitSources: true, usagePriceOverrides: true, diff --git a/apps/server/src/orchestration/AutoContinueReactor.test.ts b/apps/server/src/orchestration/AutoContinueReactor.test.ts new file mode 100644 index 000000000000..63db703b46db --- /dev/null +++ b/apps/server/src/orchestration/AutoContinueReactor.test.ts @@ -0,0 +1,165 @@ +import { + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationCommand, + type OrchestrationShellSnapshot, + type OrchestrationThreadShell, +} from "@t3tools/contracts"; +import { assert, describe, it } from "@effect/vitest"; +import * as Crypto from "effect/Crypto"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import { TestClock } from "effect/testing"; + +import { ServerActivation } from "../serverActivation.ts"; +import * as AutoContinueReactor from "./AutoContinueReactor.ts"; +import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts"; + +const NOW = "2026-08-28T12:00:00.000Z"; +const DUE_AT = "2026-08-28T11:00:00.000Z"; +const LATER_AT = "2026-08-28T14:00:00.000Z"; +const PROJECT_ID = ProjectId.make("auto-continue-project"); + +type FireCommand = Extract; + +const testCrypto = Crypto.make({ + randomBytes: (size) => new Uint8Array(size).fill(1), + digest: (_algorithm, data) => Effect.succeed(data), +}); + +function makeThread( + id: string, + overrides: Partial = {}, +): OrchestrationThreadShell { + return { + id: ThreadId.make(id), + projectId: PROJECT_ID, + title: id, + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5", + }, + runtimeMode: "full-access", + interactionMode: "default", + pullRequests: [], + branch: null, + worktreePath: null, + latestTurn: null, + createdAt: "2026-08-01T00:00:00.000Z", + updatedAt: NOW, + archivedAt: null, + settledOverride: null, + settledAt: null, + session: null, + latestUserMessageAt: null, + hasPendingApprovals: false, + hasPendingUserInput: false, + hasActionableProposedPlan: false, + ...overrides, + }; +} + +function makeSnapshot( + threads: ReadonlyArray, +): OrchestrationShellSnapshot { + return { + snapshotSequence: 1, + projects: [], + threads, + updatedAt: NOW, + }; +} + +const makeHarness = Effect.fn("makeAutoContinueHarness")(function* ( + snapshot: OrchestrationShellSnapshot, +) { + const activation = yield* Deferred.make(); + const snapshotReads = yield* Queue.unbounded(); + const commands = yield* Ref.make>([]); + + const dependencies = Layer.mergeAll( + Layer.mock(ProjectionSnapshotQuery)({ + getShellSnapshot: () => Queue.offer(snapshotReads, undefined).pipe(Effect.as(snapshot)), + }), + Layer.mock(OrchestrationEngineService)({ + dispatch: (command) => { + if (command.type !== "thread.auto-continue.fire") { + return Effect.die(new Error(`Unexpected command: ${command.type}`)); + } + return Ref.update(commands, (recorded) => [...recorded, command]).pipe( + Effect.as({ sequence: 1 }), + ); + }, + }), + Layer.succeed(ServerActivation, Deferred.await(activation)), + Layer.succeed(Crypto.Crypto, testCrypto), + ); + + return { + activation, + snapshotReads, + commands, + layer: AutoContinueReactor.layer.pipe(Layer.provide(dependencies)), + }; +}); + +describe("AutoContinueReactor", () => { + it.effect("fires due continuations as thread.auto-continue.fire commands", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const harness = yield* makeHarness( + makeSnapshot([makeThread("due-thread", { autoContinueAt: DUE_AT })]), + ); + yield* Effect.provide( + Effect.gen(function* () { + const reactor = yield* AutoContinueReactor.AutoContinueReactor; + yield* reactor.start(); + yield* Deferred.succeed(harness.activation, undefined); + yield* Queue.take(harness.snapshotReads); + yield* reactor.drain; + }), + harness.layer, + ); + const dispatched = yield* Ref.get(harness.commands); + assert.strictEqual(dispatched.length, 1); + assert.strictEqual(dispatched[0]?.threadId, "due-thread"); + assert.strictEqual(dispatched[0]?.autoContinueAt, DUE_AT); + assert.isTrue(dispatched[0]?.commandId.startsWith("server:auto-continue:due-thread:")); + assert.isString(dispatched[0]?.messageId); + }), + ), + ); + + it.effect("leaves future schedules, archived threads, and unscheduled threads alone", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const harness = yield* makeHarness( + makeSnapshot([ + makeThread("later-thread", { autoContinueAt: LATER_AT }), + makeThread("archived-thread", { autoContinueAt: DUE_AT, archivedAt: NOW }), + makeThread("plain-thread"), + ]), + ); + yield* Effect.provide( + Effect.gen(function* () { + const reactor = yield* AutoContinueReactor.AutoContinueReactor; + yield* reactor.start(); + yield* Deferred.succeed(harness.activation, undefined); + yield* Queue.take(harness.snapshotReads); + yield* reactor.drain; + }), + harness.layer, + ); + const dispatched = yield* Ref.get(harness.commands); + assert.strictEqual(dispatched.length, 0); + }), + ), + ); +}); diff --git a/apps/server/src/orchestration/AutoContinueReactor.ts b/apps/server/src/orchestration/AutoContinueReactor.ts new file mode 100644 index 000000000000..59810a713ed2 --- /dev/null +++ b/apps/server/src/orchestration/AutoContinueReactor.ts @@ -0,0 +1,106 @@ +import { CommandId, MessageId } from "@t3tools/contracts"; +import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schedule from "effect/Schedule"; +import type * as Scope from "effect/Scope"; + +import { forkParked } from "../serverActivation.ts"; +import * as OrchestrationEngine from "./Services/OrchestrationEngine.ts"; +import * as ProjectionSnapshotQuery from "./Services/ProjectionSnapshotQuery.ts"; + +/** + * Fires scheduled "continue when the limit resets" turns. The schedule is a + * persisted thread field (`autoContinueAt`, written by thread.auto-continue.set), + * so it survives restarts: each sweep re-reads the shell snapshot and triggers + * every due thread. The decider validates the schedule again at fire time + * (compare-and-set on the timestamp), so a cancel or reschedule that raced a + * sweep wins over the stale trigger, and duplicate sweeps are no-ops. + */ +export class AutoContinueReactor extends Context.Service< + AutoContinueReactor, + { + readonly start: () => Effect.Effect; + readonly drain: Effect.Effect; + } +>()("t3/orchestration/AutoContinueReactor") {} + +/** @public Service construction is part of the canonical Effect module API. */ +export const make = Effect.gen(function* () { + const engine = yield* OrchestrationEngine.OrchestrationEngineService; + const snapshots = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; + const crypto = yield* Crypto.Crypto; + + const sweep = Effect.fn("AutoContinueReactor.sweep")(function* () { + const snapshot = yield* snapshots.getShellSnapshot(); + const nowMs = Date.parse(DateTime.formatIso(yield* DateTime.now)); + // Archived threads keep no schedule (archive clears it in the decider), + // but guard anyway: firing into an archived thread would be rejected and + // logged every minute until the stale row is repaired. + const due = snapshot.threads.filter( + (thread) => + thread.autoContinueAt != null && + Date.parse(thread.autoContinueAt) <= nowMs && + thread.archivedAt === null, + ); + yield* Effect.forEach( + due, + (thread) => + Effect.gen(function* () { + const autoContinueAt = thread.autoContinueAt; + if (autoContinueAt == null) return; + const commandUuid = yield* crypto.randomUUIDv4; + const messageUuid = yield* crypto.randomUUIDv4; + const createdAt = DateTime.formatIso(yield* DateTime.now); + yield* engine.dispatch({ + type: "thread.auto-continue.fire", + commandId: CommandId.make(`server:auto-continue:${thread.id}:${commandUuid}`), + threadId: thread.id, + autoContinueAt, + messageId: MessageId.make(messageUuid), + createdAt, + }); + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("auto-continue fire skipped", { + threadId: thread.id, + cause: Cause.pretty(cause), + }), + ), + ), + { concurrency: 4, discard: true }, + ); + }); + + const runSweep = sweep().pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("auto-continue sweep failed", { cause: Cause.pretty(cause) }), + ), + ); + const worker = yield* makeDrainableWorker((_: void) => runSweep); + + const start: AutoContinueReactor["Service"]["start"] = Effect.fn("AutoContinueReactor.start")( + function* () { + // The first sweep on start fires continuations whose reset passed while + // the server was down — exactly what the user asked for by scheduling. + yield* forkParked( + Effect.gen(function* () { + yield* worker.enqueue(undefined); + yield* worker.drain; + }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid), + ); + }, + ); + + return { start, drain: worker.drain } satisfies AutoContinueReactor["Service"]; +}); + +export const layer = Layer.effect(AutoContinueReactor, make); diff --git a/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts b/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts index d25c442ac8b2..ce5ac257fe9b 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts @@ -10,6 +10,7 @@ import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts"; import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; import { ThreadDeletionReactor } from "../Services/ThreadDeletionReactor.ts"; import * as ThreadSettlementReactor from "../ThreadSettlementReactor.ts"; +import * as AutoContinueReactor from "../AutoContinueReactor.ts"; import * as PullRequestSyncReactor from "../PullRequestSyncReactor.ts"; import * as ThreadPullRequestReactor from "../ThreadPullRequestReactor.ts"; import { OrchestrationReactor } from "../Services/OrchestrationReactor.ts"; @@ -95,6 +96,15 @@ describe("OrchestrationReactor", () => { drain: Effect.void, }), ), + Layer.provideMerge( + Layer.succeed(AutoContinueReactor.AutoContinueReactor, { + start: () => { + started.push("auto-continue-reactor"); + return Effect.void; + }, + drain: Effect.void, + }), + ), Layer.provideMerge( Layer.succeed(PullRequestSyncReactor.PullRequestSyncReactor, { start: () => { @@ -128,6 +138,7 @@ describe("OrchestrationReactor", () => { "thread-deletion-reactor", "thread-pull-request-reactor", "thread-settlement-reactor", + "auto-continue-reactor", "pull-request-sync-reactor", "agent-awareness-relay", "storage-cleanup", diff --git a/apps/server/src/orchestration/Layers/OrchestrationReactor.ts b/apps/server/src/orchestration/Layers/OrchestrationReactor.ts index da5cc337e420..97dc498401d8 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationReactor.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationReactor.ts @@ -10,6 +10,7 @@ import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts"; import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; import { ThreadDeletionReactor } from "../Services/ThreadDeletionReactor.ts"; import * as ThreadSettlementReactor from "../ThreadSettlementReactor.ts"; +import * as AutoContinueReactor from "../AutoContinueReactor.ts"; import * as PullRequestSyncReactor from "../PullRequestSyncReactor.ts"; import * as ThreadPullRequestReactor from "../ThreadPullRequestReactor.ts"; import * as AgentAwarenessRelay from "../../relay/AgentAwarenessRelay.ts"; @@ -21,6 +22,7 @@ export const makeOrchestrationReactor = Effect.gen(function* () { const checkpointReactor = yield* CheckpointReactor; const threadDeletionReactor = yield* ThreadDeletionReactor; const threadSettlementReactor = yield* ThreadSettlementReactor.ThreadSettlementReactor; + const autoContinueReactor = yield* AutoContinueReactor.AutoContinueReactor; const pullRequestSyncReactor = yield* PullRequestSyncReactor.PullRequestSyncReactor; const threadPullRequestReactor = yield* ThreadPullRequestReactor.ThreadPullRequestReactor; const agentAwarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay; @@ -33,6 +35,7 @@ export const makeOrchestrationReactor = Effect.gen(function* () { yield* threadDeletionReactor.start(); yield* threadPullRequestReactor.start(); yield* threadSettlementReactor.start(); + yield* autoContinueReactor.start(); yield* pullRequestSyncReactor.start(); yield* agentAwarenessRelay.start(); yield* storageCleanup.start(); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index b5e6cb0cdd54..5f99bb766b18 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -630,6 +630,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti unsettledAt: null, snoozedUntil: null, snoozedAt: null, + autoContinueAt: null, pinnedAt: null, pinOrderKey: null, activeOrderKey: null, @@ -748,6 +749,36 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } + case "thread.auto-continue-scheduled": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) { + return; + } + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + autoContinueAt: event.payload.autoContinueAt, + updatedAt: event.payload.updatedAt, + }); + return; + } + + case "thread.auto-continue-cleared": { + const existingRow = yield* projectionThreadRepository.getById({ + threadId: event.payload.threadId, + }); + if (Option.isNone(existingRow)) { + return; + } + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + autoContinueAt: null, + updatedAt: event.payload.updatedAt, + }); + return; + } + case "thread.pinned": { const existingRow = yield* projectionThreadRepository.getById({ threadId: event.payload.threadId, @@ -1355,6 +1386,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti runtimeMode: event.payload.session.runtimeMode, activeTurnId: event.payload.session.activeTurnId, lastError: event.payload.session.lastError, + lastErrorLimitResetsAt: event.payload.session.lastErrorLimitResetsAt ?? null, updatedAt: event.payload.session.updatedAt, }); }); diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 843eb8343d84..a1a916017cd9 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -482,6 +482,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { unsettledAt: null, snoozedUntil: null, snoozedAt: null, + autoContinueAt: null, pinnedAt: "2026-02-24T00:00:01.000Z", pinOrderKey: "gm", activeOrderKey: "hq", @@ -539,6 +540,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { runtimeMode: "approval-required", activeTurnId: asTurnId("turn-1"), lastError: null, + lastErrorLimitResetsAt: null, updatedAt: "2026-02-24T00:00:07.000Z", }, }, @@ -608,6 +610,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { unsettledAt: null, snoozedUntil: null, snoozedAt: null, + autoContinueAt: null, pinnedAt: "2026-02-24T00:00:01.000Z", pinOrderKey: "gm", activeOrderKey: "hq", @@ -620,6 +623,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { runtimeMode: "approval-required", activeTurnId: asTurnId("turn-1"), lastError: null, + lastErrorLimitResetsAt: null, updatedAt: "2026-02-24T00:00:07.000Z", }, latestUserMessageAt: "2026-02-24T00:00:04.000Z", diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 03bf17329286..964ad0f75766 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -382,6 +382,7 @@ function mapSessionRow( runtimeMode: row.runtimeMode, activeTurnId: row.activeTurnId, lastError: row.lastError, + lastErrorLimitResetsAt: row.lastErrorLimitResetsAt ?? null, updatedAt: row.updatedAt, }; } @@ -579,6 +580,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { unsettled_at AS "unsettledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + auto_continue_at AS "autoContinueAt", pinned_at AS "pinnedAt", pin_order_key AS "pinOrderKey", active_order_key AS "activeOrderKey", @@ -620,6 +622,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { unsettled_at AS "unsettledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + auto_continue_at AS "autoContinueAt", pinned_at AS "pinnedAt", pin_order_key AS "pinOrderKey", active_order_key AS "activeOrderKey", @@ -693,6 +696,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { unsettled_at AS "unsettledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + auto_continue_at AS "autoContinueAt", pinned_at AS "pinnedAt", pin_order_key AS "pinOrderKey", active_order_key AS "activeOrderKey", @@ -857,6 +861,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { runtime_mode AS "runtimeMode", active_turn_id AS "activeTurnId", last_error AS "lastError", + last_error_limit_resets_at AS "lastErrorLimitResetsAt", updated_at AS "updatedAt" FROM projection_thread_sessions ORDER BY thread_id ASC @@ -878,6 +883,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { sessions.runtime_mode AS "runtimeMode", sessions.active_turn_id AS "activeTurnId", sessions.last_error AS "lastError", + sessions.last_error_limit_resets_at AS "lastErrorLimitResetsAt", sessions.updated_at AS "updatedAt" FROM projection_thread_sessions sessions INNER JOIN projection_threads threads @@ -903,6 +909,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { sessions.runtime_mode AS "runtimeMode", sessions.active_turn_id AS "activeTurnId", sessions.last_error AS "lastError", + sessions.last_error_limit_resets_at AS "lastErrorLimitResetsAt", sessions.updated_at AS "updatedAt" FROM projection_thread_sessions sessions INNER JOIN projection_threads threads @@ -1258,6 +1265,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { unsettled_at AS "unsettledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + auto_continue_at AS "autoContinueAt", pinned_at AS "pinnedAt", pin_order_key AS "pinOrderKey", active_order_key AS "activeOrderKey", @@ -1293,6 +1301,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { sessions.runtime_mode AS "runtimeMode", sessions.active_turn_id AS "activeTurnId", sessions.last_error AS "lastError", + sessions.last_error_limit_resets_at AS "lastErrorLimitResetsAt", sessions.updated_at AS "updatedAt" FROM projection_threads AS threads LEFT JOIN projection_thread_sessions AS sessions @@ -1611,6 +1620,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { runtime_mode AS "runtimeMode", active_turn_id AS "activeTurnId", last_error AS "lastError", + last_error_limit_resets_at AS "lastErrorLimitResetsAt", updated_at AS "updatedAt" FROM projection_thread_sessions WHERE thread_id = ${threadId} @@ -2284,6 +2294,7 @@ pending_approval_requests AS ( runtimeMode: row.runtimeMode, activeTurnId: row.activeTurnId, lastError: row.lastError, + lastErrorLimitResetsAt: row.lastErrorLimitResetsAt ?? null, updatedAt: row.updatedAt, }); } @@ -2333,6 +2344,7 @@ pending_approval_requests AS ( unsettledAt: row.unsettledAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, + autoContinueAt: row.autoContinueAt ?? null, pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, activeOrderKey: row.activeOrderKey ?? null, @@ -2578,6 +2590,7 @@ pending_approval_requests AS ( unsettledAt: row.unsettledAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, + autoContinueAt: row.autoContinueAt ?? null, pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, activeOrderKey: row.activeOrderKey ?? null, @@ -2734,6 +2747,7 @@ pending_approval_requests AS ( unsettledAt: row.unsettledAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, + autoContinueAt: row.autoContinueAt ?? null, pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, activeOrderKey: row.activeOrderKey ?? null, @@ -2897,6 +2911,7 @@ pending_approval_requests AS ( unsettledAt: row.unsettledAt, snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, + autoContinueAt: row.autoContinueAt ?? null, pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, activeOrderKey: row.activeOrderKey ?? null, @@ -3253,6 +3268,7 @@ pending_approval_requests AS ( unsettledAt: threadRow.value.unsettledAt, snoozedUntil: threadRow.value.snoozedUntil, snoozedAt: threadRow.value.snoozedAt, + autoContinueAt: threadRow.value.autoContinueAt ?? null, pinnedAt: threadRow.value.pinnedAt, pinOrderKey: threadRow.value.pinOrderKey ?? null, activeOrderKey: threadRow.value.activeOrderKey ?? null, @@ -3554,6 +3570,7 @@ pending_approval_requests AS ( unsettledAt: threadRow.value.unsettledAt, snoozedUntil: threadRow.value.snoozedUntil, snoozedAt: threadRow.value.snoozedAt, + autoContinueAt: threadRow.value.autoContinueAt ?? null, pinnedAt: threadRow.value.pinnedAt, pinOrderKey: threadRow.value.pinOrderKey ?? null, activeOrderKey: threadRow.value.activeOrderKey ?? null, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index d61739f72c21..295c70316b95 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -482,6 +482,67 @@ describe("ProviderRuntimeIngestion", () => { expect(thread.session?.lastError).toBe("turn failed"); }); + it("carries a usage-limit reset instant onto the session error and clears it on recovery", async () => { + const harness = await createHarness(); + const resetsAt = "2026-01-01T05:00:00.000Z"; + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-limit-turn-started"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:00.000Z", + turnId: asTurnId("turn-limit"), + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-limit-turn-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:01.000Z", + turnId: asTurnId("turn-limit"), + payload: { + state: "failed", + errorMessage: "Codex usage limit reached. The session limit resets in 5h.", + usageLimitResetsAt: resetsAt, + }, + }); + + const failed = await waitForThread( + harness.readModel, + (entry) => + entry.session?.status === "error" && entry.session?.lastErrorLimitResetsAt === resetsAt, + ); + expect(failed.session?.lastErrorLimitResetsAt).toBe(resetsAt); + + // A later successful turn clears the offer along with the error. + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-limit-turn-retry"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T06:00:00.000Z", + turnId: asTurnId("turn-retry"), + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-limit-turn-retry-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T06:00:01.000Z", + turnId: asTurnId("turn-retry"), + payload: { + state: "completed", + }, + }); + + const recovered = await waitForThread( + harness.readModel, + (entry) => entry.session?.status === "ready" && entry.session?.lastError === null, + ); + expect(recovered.session?.lastErrorLimitResetsAt ?? null).toBeNull(); + }); + it.each([ { delivery: "buffered", responseStreamingMode: "paragraph" as const }, { delivery: "streamed", responseStreamingMode: "token" as const }, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 84cb34a4e783..4928ce013be9 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -1886,6 +1886,16 @@ const make = Effect.gen(function* () { : status === "ready" || status === "interrupted" ? null : (thread.session?.lastError ?? null); + // Travels with lastError: set from a usage-limit failure that named + // its reset instant, cleared whenever lastError clears, and kept + // through unrelated status writes so the offer survives reconnects. + const lastErrorLimitResetsAt = + event.type === "turn.completed" && + normalizeRuntimeTurnState(event.payload.state) === "failed" + ? (event.payload.usageLimitResetsAt ?? null) + : status === "ready" || status === "interrupted" + ? null + : (thread.session?.lastErrorLimitResetsAt ?? null); if (shouldApplyThreadLifecycle) { if (event.type === "turn.started" && acceptedTurnStartedSourcePlan !== null) { @@ -1922,6 +1932,7 @@ const make = Effect.gen(function* () { runtimeMode: thread.session?.runtimeMode ?? "full-access", activeTurnId: nextActiveTurnId, lastError, + lastErrorLimitResetsAt, updatedAt: now, }, createdAt: now, @@ -2422,6 +2433,9 @@ const make = Effect.gen(function* () { runtimeMode: thread.session?.runtimeMode ?? "full-access", activeTurnId: eventTurnId ?? null, lastError: runtimeErrorMessage, + // Adapters that know the reset instant emit it on the failed + // turn.completed, which follows this event and wins the session. + lastErrorLimitResetsAt: null, updatedAt: now, }, createdAt: now, diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 29468dc3f84e..1db37e12bc3c 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -13,6 +13,8 @@ import { ThreadUnsettledPayload as ContractsThreadUnsettledPayloadSchema, ThreadSnoozedPayload as ContractsThreadSnoozedPayloadSchema, ThreadUnsnoozedPayload as ContractsThreadUnsnoozedPayloadSchema, + ThreadAutoContinueScheduledPayload as ContractsThreadAutoContinueScheduledPayloadSchema, + ThreadAutoContinueClearedPayload as ContractsThreadAutoContinueClearedPayloadSchema, ThreadPinnedPayload as ContractsThreadPinnedPayloadSchema, ThreadUnpinnedPayload as ContractsThreadUnpinnedPayloadSchema, ThreadPinReorderedPayload as ContractsThreadPinReorderedPayloadSchema, @@ -48,6 +50,8 @@ export const ThreadUnarchivedPayload = ContractsThreadUnarchivedPayloadSchema; export const ThreadUnsettledPayload = ContractsThreadUnsettledPayloadSchema; export const ThreadSnoozedPayload = ContractsThreadSnoozedPayloadSchema; export const ThreadUnsnoozedPayload = ContractsThreadUnsnoozedPayloadSchema; +export const ThreadAutoContinueScheduledPayload = ContractsThreadAutoContinueScheduledPayloadSchema; +export const ThreadAutoContinueClearedPayload = ContractsThreadAutoContinueClearedPayloadSchema; export const ThreadPinnedPayload = ContractsThreadPinnedPayloadSchema; export const ThreadUnpinnedPayload = ContractsThreadUnpinnedPayloadSchema; export const ThreadPinReorderedPayload = ContractsThreadPinReorderedPayloadSchema; diff --git a/apps/server/src/orchestration/decider.autoContinue.test.ts b/apps/server/src/orchestration/decider.autoContinue.test.ts new file mode 100644 index 000000000000..02fb05c35927 --- /dev/null +++ b/apps/server/src/orchestration/decider.autoContinue.test.ts @@ -0,0 +1,310 @@ +import { + CommandId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationReadModel, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { AUTO_CONTINUE_MESSAGE_TEXT, decideOrchestrationCommand } from "./decider.ts"; + +const NOW = "2026-01-01T00:00:00.000Z"; +// The decider's clock is the Effect test clock, pinned to the epoch, so +// "future" resume times are relative to 1970-01-01T00:00:00.000Z. +const FUTURE_RESET = "1970-01-01T05:00:00.000Z"; +const DUE_RESET = "1969-12-31T21:00:00.000Z"; + +function makeReadModel(input: { + readonly autoContinueAt?: string | null; + readonly archivedAt?: string | null; + readonly settledOverride?: "settled" | "active" | null; + readonly settledAt?: string | null; +}): OrchestrationReadModel { + return { + snapshotSequence: 0, + projects: [], + threads: [ + { + id: ThreadId.make("thread-1"), + projectId: ProjectId.make("project-1"), + title: "Thread", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + pullRequests: [], + latestTurn: null, + createdAt: NOW, + updatedAt: NOW, + archivedAt: input.archivedAt ?? null, + settledOverride: input.settledOverride ?? null, + settledAt: input.settledAt ?? null, + snoozedUntil: null, + snoozedAt: null, + autoContinueAt: input.autoContinueAt ?? null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, + }, + ], + updatedAt: NOW, + }; +} + +it.layer(NodeServices.layer)("auto-continue decider", (it) => { + it.effect("schedules a continuation for a future reset time", () => + Effect.gen(function* () { + const event = yield* decideOrchestrationCommand({ + command: { + type: "thread.auto-continue.set", + commandId: CommandId.make("cmd-ac-set"), + threadId: ThreadId.make("thread-1"), + autoContinueAt: FUTURE_RESET, + }, + readModel: makeReadModel({}), + }); + const events = Array.isArray(event) ? event : [event]; + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe("thread.auto-continue-scheduled"); + if (events[0]?.type === "thread.auto-continue-scheduled") { + expect(events[0].payload.autoContinueAt).toBe(FUTURE_RESET); + } + }), + ); + + it.effect("rejects a resume time that is not in the future", () => + Effect.gen(function* () { + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.auto-continue.set", + commandId: CommandId.make("cmd-ac-set-past"), + threadId: ThreadId.make("thread-1"), + autoContinueAt: DUE_RESET, + }, + readModel: makeReadModel({}), + }).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + + it.effect("rejects an unparseable resume time", () => + Effect.gen(function* () { + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.auto-continue.set", + commandId: CommandId.make("cmd-ac-set-garbage"), + threadId: ThreadId.make("thread-1"), + autoContinueAt: "not-a-date", + }, + readModel: makeReadModel({}), + }).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + + it.effect("re-emits idempotently for a duplicate schedule to the same instant", () => + Effect.gen(function* () { + const reEmit = yield* decideOrchestrationCommand({ + command: { + type: "thread.auto-continue.set", + commandId: CommandId.make("cmd-ac-set-again"), + threadId: ThreadId.make("thread-1"), + autoContinueAt: FUTURE_RESET, + }, + readModel: makeReadModel({ autoContinueAt: FUTURE_RESET }), + }); + const events = Array.isArray(reEmit) ? reEmit : [reEmit]; + expect(events).toHaveLength(1); + if (events[0]?.type === "thread.auto-continue-scheduled") { + // No state change — keep the existing updatedAt. + expect(events[0].payload.updatedAt).toBe(NOW); + } + }), + ); + + it.effect("clears with reason user and re-emits idempotently when unscheduled", () => + Effect.gen(function* () { + const event = yield* decideOrchestrationCommand({ + command: { + type: "thread.auto-continue.clear", + commandId: CommandId.make("cmd-ac-clear"), + threadId: ThreadId.make("thread-1"), + reason: "user", + }, + readModel: makeReadModel({ autoContinueAt: FUTURE_RESET }), + }); + const events = Array.isArray(event) ? event : [event]; + expect(events[0]?.type).toBe("thread.auto-continue-cleared"); + if (events[0]?.type === "thread.auto-continue-cleared") { + expect(events[0].payload.reason).toBe("user"); + expect(events[0].payload.updatedAt).not.toBe(NOW); + } + + const alreadyClear = yield* decideOrchestrationCommand({ + command: { + type: "thread.auto-continue.clear", + commandId: CommandId.make("cmd-ac-clear-again"), + threadId: ThreadId.make("thread-1"), + reason: "user", + }, + readModel: makeReadModel({}), + }); + const clearEvents = Array.isArray(alreadyClear) ? alreadyClear : [alreadyClear]; + expect(clearEvents[0]?.type).toBe("thread.auto-continue-cleared"); + if (clearEvents[0]?.type === "thread.auto-continue-cleared") { + // No state change — keep the existing updatedAt. + expect(clearEvents[0].payload.updatedAt).toBe(NOW); + } + }), + ); + + it.effect("any turn start spends the schedule (activity clear)", () => + Effect.gen(function* () { + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.turn.start", + commandId: CommandId.make("cmd-turn-start"), + threadId: ThreadId.make("thread-1"), + message: { + messageId: MessageId.make("message-1"), + role: "user", + text: "Keep going", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + createdAt: NOW, + }, + readModel: makeReadModel({ autoContinueAt: FUTURE_RESET }), + }); + const events = Array.isArray(result) ? result : [result]; + const cleared = events.find((entry) => entry.type === "thread.auto-continue-cleared"); + expect(cleared).toBeDefined(); + if (cleared?.type === "thread.auto-continue-cleared") { + expect(cleared.payload.reason).toBe("activity"); + } + }), + ); + + it.effect("settling clears a pending continuation", () => + Effect.gen(function* () { + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.settle", + commandId: CommandId.make("cmd-settle"), + threadId: ThreadId.make("thread-1"), + }, + readModel: makeReadModel({ autoContinueAt: FUTURE_RESET }), + }); + const events = Array.isArray(result) ? result : [result]; + expect(events.map((entry) => entry.type)).toEqual([ + "thread.settled", + "thread.auto-continue-cleared", + ]); + }), + ); + + it.effect("archiving clears a pending continuation", () => + Effect.gen(function* () { + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.archive", + commandId: CommandId.make("cmd-archive"), + threadId: ThreadId.make("thread-1"), + }, + readModel: makeReadModel({ autoContinueAt: FUTURE_RESET }), + }); + const events = Array.isArray(result) ? result : [result]; + expect(events.map((entry) => entry.type)).toEqual([ + "thread.archived", + "thread.auto-continue-cleared", + ]); + }), + ); + + it.effect("fires a due continuation as a Continue turn that spends the schedule", () => + Effect.gen(function* () { + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.auto-continue.fire", + commandId: CommandId.make("server:auto-continue:thread-1:uuid"), + threadId: ThreadId.make("thread-1"), + autoContinueAt: DUE_RESET, + messageId: MessageId.make("message-continue"), + createdAt: NOW, + }, + readModel: makeReadModel({ autoContinueAt: DUE_RESET }), + }); + const events = Array.isArray(result) ? result : [result]; + expect(events.map((entry) => entry.type)).toEqual([ + "thread.auto-continue-cleared", + "thread.message-sent", + "thread.turn-start-requested", + ]); + const messageSent = events.find((entry) => entry.type === "thread.message-sent"); + if (messageSent?.type === "thread.message-sent") { + expect(messageSent.payload.text).toBe(AUTO_CONTINUE_MESSAGE_TEXT); + expect(messageSent.payload.role).toBe("user"); + expect(messageSent.payload.messageId).toBe("message-continue"); + } + }), + ); + + it.effect("rejects firing when the schedule changed since the sweep read it", () => + Effect.gen(function* () { + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.auto-continue.fire", + commandId: CommandId.make("server:auto-continue:thread-1:stale"), + threadId: ThreadId.make("thread-1"), + autoContinueAt: DUE_RESET, + messageId: MessageId.make("message-continue"), + createdAt: NOW, + }, + // Cancelled (or rescheduled) after the sweep read the shell. + readModel: makeReadModel({}), + }).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + + it.effect("rejects firing a schedule that is not due yet", () => + Effect.gen(function* () { + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.auto-continue.fire", + commandId: CommandId.make("server:auto-continue:thread-1:early"), + threadId: ThreadId.make("thread-1"), + autoContinueAt: FUTURE_RESET, + messageId: MessageId.make("message-continue"), + createdAt: NOW, + }, + readModel: makeReadModel({ autoContinueAt: FUTURE_RESET }), + }).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + + it.effect("rejects scheduling on an archived thread", () => + Effect.gen(function* () { + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.auto-continue.set", + commandId: CommandId.make("cmd-ac-set-archived"), + threadId: ThreadId.make("thread-1"), + autoContinueAt: FUTURE_RESET, + }, + readModel: makeReadModel({ archivedAt: NOW }), + }).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); +}); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 0119c0e8599a..a0f8d860e3a1 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -51,6 +51,13 @@ const monogramSegmenter = new Intl.Segmenter(undefined, { granularity: "grapheme const isScriptRunCommand = Schema.is(SCRIPT_RUN_COMMAND_PATTERN); +/** + * The user message a fired auto-continue sends. Deliberately a plain + * "Continue": the provider session already holds the stopped turn's context, + * and the visible message tells the user exactly what the server did. + */ +export const AUTO_CONTINUE_MESSAGE_TEXT = "Continue"; + const nowIso = Effect.map(DateTime.now, DateTime.formatIso); const decodeUserInputRequestedPayload = Schema.decodeUnknownOption(UserInputRequestedPayload); const threadPullRequestLinksEqual = Schema.toEquivalence(Schema.NullOr(ThreadLinkedPullRequest)); @@ -433,26 +440,48 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" } case "thread.archive": { - yield* requireThreadNotArchived({ + const thread = yield* requireThreadNotArchived({ readModel, command, threadId: command.threadId, }); const occurredAt = yield* nowIso; - return { + const archivedEvent = { ...(yield* withEventBase({ aggregateKind: "thread", aggregateId: command.threadId, occurredAt, commandId: command.commandId, })), - type: "thread.archived", + type: "thread.archived" as const, payload: { threadId: command.threadId, archivedAt: occurredAt, updatedAt: occurredAt, }, }; + // Archiving parks the thread: a pending limit-reset continuation must + // not fire into it — or surprise the user weeks later on unarchive. + if (thread.autoContinueAt == null) { + return archivedEvent; + } + return [ + archivedEvent, + { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + })), + type: "thread.auto-continue-cleared" as const, + payload: { + threadId: command.threadId, + reason: "user" as const, + updatedAt: occurredAt, + }, + }, + ]; } case "thread.unarchive": { @@ -598,6 +627,24 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }, }); } + // A settled thread is done: a pending limit-reset continuation would + // restart work the user just called finished. + if (thread.autoContinueAt != null) { + companionEvents.push({ + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + })), + type: "thread.auto-continue-cleared", + payload: { + threadId: command.threadId, + reason: "user", + updatedAt: occurredAt, + }, + }); + } return companionEvents.length > 0 ? [settledEvent, ...companionEvents] : settledEvent; } @@ -725,6 +772,118 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }; } + case "thread.auto-continue.set": { + const thread = yield* requireThreadNotArchived({ + readModel, + command, + threadId: command.threadId, + }); + const occurredAt = yield* nowIso; + // Mirrors thread.snooze: a resume time that already passed would fire + // on the next sweep as a surprise turn, and the negated comparison also + // rejects unparseable timestamps (NaN fails every comparison). + if (!(Date.parse(command.autoContinueAt) > Date.parse(occurredAt))) { + return yield* Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} auto-continue time ${command.autoContinueAt} is not in the future`, + }), + ); + } + // Re-scheduling to the SAME instant is a duplicate (double-click, + // raced clients): re-emit with the original updatedAt so the + // projection is a no-op. A different instant is a real change. + const alreadyScheduled = thread.autoContinueAt === command.autoContinueAt; + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + })), + type: "thread.auto-continue-scheduled", + payload: { + threadId: command.threadId, + autoContinueAt: command.autoContinueAt, + updatedAt: alreadyScheduled ? thread.updatedAt : occurredAt, + }, + }; + } + + case "thread.auto-continue.clear": { + const thread = yield* requireThreadNotArchived({ + readModel, + command, + threadId: command.threadId, + }); + // Idempotent by re-emission (see thread.unsnooze): clearing an + // unscheduled thread lands on the same null state without churning + // updatedAt. + const alreadyClear = thread.autoContinueAt == null; + const occurredAt = yield* nowIso; + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + })), + type: "thread.auto-continue-cleared", + payload: { + threadId: command.threadId, + reason: command.reason, + updatedAt: alreadyClear ? thread.updatedAt : occurredAt, + }, + }; + } + + case "thread.auto-continue.fire": { + const thread = yield* requireThreadNotArchived({ + readModel, + command, + threadId: command.threadId, + }); + // Compare-and-set: the sweep read a schedule; if a cancel or a + // reschedule landed since, that decision wins over the stale trigger. + if (thread.autoContinueAt == null || thread.autoContinueAt !== command.autoContinueAt) { + return yield* Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} auto-continue schedule changed before firing`, + }), + ); + } + const occurredAt = yield* nowIso; + if (Date.parse(command.autoContinueAt) > Date.parse(occurredAt)) { + return yield* Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} auto-continue at ${command.autoContinueAt} is not due yet`, + }), + ); + } + // The continuation is an ordinary turn start; its lifecycle resets + // spend the schedule the same way a manual send would, so no separate + // cleared event is needed here. + return yield* decideOrchestrationCommand({ + command: { + type: "thread.turn.start", + commandId: command.commandId, + threadId: command.threadId, + message: { + messageId: command.messageId, + role: "user", + text: AUTO_CONTINUE_MESSAGE_TEXT, + attachments: [], + }, + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + createdAt: command.createdAt, + }, + readModel, + }); + } + case "thread.pin": { const thread = yield* requireThreadNotArchived({ readModel, @@ -1494,6 +1653,24 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }, }); } + // Any turn start spends a pending limit-reset continuation: the work + // is moving again, so firing another turn later would double-send. + if (targetThread.autoContinueAt != null) { + lifecycleResetEvents.push({ + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.auto-continue-cleared", + payload: { + threadId: command.threadId, + reason: "activity", + updatedAt: command.createdAt, + }, + }); + } return [ ...lifecycleResetEvents, ...(userMessageEvent ? [userMessageEvent] : []), diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index c4e1996f1ddd..7e55a6070a9c 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -98,6 +98,7 @@ describe("orchestration projector", () => { unsettledAt: null, snoozedUntil: null, snoozedAt: null, + autoContinueAt: null, deletedAt: null, messages: [], proposedPlans: [], diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 53013770b15b..abfcbf9a8d3c 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -50,6 +50,8 @@ import { ThreadUnarchivedPayload, ThreadUnsettledPayload, ThreadUnsnoozedPayload, + ThreadAutoContinueScheduledPayload, + ThreadAutoContinueClearedPayload, ThreadRevertedPayload, ThreadSessionSetPayload, ThreadTurnDiffCompletedPayload, @@ -442,6 +444,7 @@ export function projectEvent( activeOrderKey: null, snoozedUntil: null, snoozedAt: null, + autoContinueAt: null, deletedAt: null, messages: [], activities: [], @@ -554,6 +557,38 @@ export function projectEvent( })), ); + case "thread.auto-continue-scheduled": + return decodeForEvent( + ThreadAutoContinueScheduledPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + autoContinueAt: payload.autoContinueAt, + updatedAt: payload.updatedAt, + }), + })), + ); + + case "thread.auto-continue-cleared": + return decodeForEvent( + ThreadAutoContinueClearedPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + autoContinueAt: null, + updatedAt: payload.updatedAt, + }), + })), + ); + case "thread.pinned": return decodeForEvent(ThreadPinnedPayload, event.payload, event.type, "payload").pipe( Effect.map((payload) => ({ diff --git a/apps/server/src/persistence/Layers/ProjectionThreadSessions.ts b/apps/server/src/persistence/Layers/ProjectionThreadSessions.ts index dcb750983a00..d1f8569cff9c 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreadSessions.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreadSessions.ts @@ -28,6 +28,7 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () { runtime_mode, active_turn_id, last_error, + last_error_limit_resets_at, updated_at ) VALUES ( @@ -38,6 +39,7 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () { ${row.runtimeMode}, ${row.activeTurnId}, ${row.lastError}, + ${row.lastErrorLimitResetsAt ?? null}, ${row.updatedAt} ) ON CONFLICT (thread_id) @@ -48,6 +50,7 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () { runtime_mode = excluded.runtime_mode, active_turn_id = excluded.active_turn_id, last_error = excluded.last_error, + last_error_limit_resets_at = excluded.last_error_limit_resets_at, updated_at = excluded.updated_at `, }); @@ -65,6 +68,7 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () { runtime_mode AS "runtimeMode", active_turn_id AS "activeTurnId", last_error AS "lastError", + last_error_limit_resets_at AS "lastErrorLimitResetsAt", updated_at AS "updatedAt" FROM projection_thread_sessions WHERE thread_id = ${threadId} diff --git a/apps/server/src/persistence/Layers/ProjectionThreads.ts b/apps/server/src/persistence/Layers/ProjectionThreads.ts index 4feaf3a185b9..533bc61f7004 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -54,6 +54,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { unsettled_at, snoozed_until, snoozed_at, + auto_continue_at, pinned_at, pin_order_key, active_order_key, @@ -86,6 +87,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.unsettledAt}, ${row.snoozedUntil}, ${row.snoozedAt}, + ${row.autoContinueAt ?? null}, ${row.pinnedAt}, ${row.pinOrderKey ?? null}, ${row.activeOrderKey ?? null}, @@ -118,6 +120,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { unsettled_at = excluded.unsettled_at, snoozed_until = excluded.snoozed_until, snoozed_at = excluded.snoozed_at, + auto_continue_at = excluded.auto_continue_at, pinned_at = excluded.pinned_at, pin_order_key = excluded.pin_order_key, active_order_key = excluded.active_order_key, @@ -157,6 +160,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { unsettled_at AS "unsettledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + auto_continue_at AS "autoContinueAt", pinned_at AS "pinnedAt", pin_order_key AS "pinOrderKey", active_order_key AS "activeOrderKey", @@ -198,6 +202,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { unsettled_at AS "unsettledAt", snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", + auto_continue_at AS "autoContinueAt", pinned_at AS "pinnedAt", pin_order_key AS "pinOrderKey", active_order_key AS "activeOrderKey", diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 18aae09febf8..a41e8254d694 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -65,6 +65,7 @@ import Migration0050 from "./Migrations/050_ProjectionThreadPullRequests.ts"; import Migration0051 from "./Migrations/051_ProjectionThreadMessageContext.ts"; import Migration0052 from "./Migrations/052_ProjectionThreadTitleState.ts"; import Migration0053 from "./Migrations/053_PullRequestFilesViewed.ts"; +import Migration0054 from "./Migrations/054_AutoContinue.ts"; /** * Migration loader with all migrations defined inline. @@ -130,6 +131,7 @@ const migrationEntries = [ [51, "ProjectionThreadMessageContext", Migration0051], [52, "ProjectionThreadTitleState", Migration0052], [53, "PullRequestFilesViewed", Migration0053], + [54, "AutoContinue", Migration0054], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/054_AutoContinue.ts b/apps/server/src/persistence/Migrations/054_AutoContinue.ts new file mode 100644 index 000000000000..fd65c7662745 --- /dev/null +++ b/apps/server/src/persistence/Migrations/054_AutoContinue.ts @@ -0,0 +1,26 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const threadColumns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_threads) + `; + if (!threadColumns.some((column) => column.name === "auto_continue_at")) { + yield* sql` + ALTER TABLE projection_threads + ADD COLUMN auto_continue_at TEXT + `; + } + + const sessionColumns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_thread_sessions) + `; + if (!sessionColumns.some((column) => column.name === "last_error_limit_resets_at")) { + yield* sql` + ALTER TABLE projection_thread_sessions + ADD COLUMN last_error_limit_resets_at TEXT + `; + } +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreadSessions.ts b/apps/server/src/persistence/Services/ProjectionThreadSessions.ts index 7cecac33eb6a..b6ef850dc870 100644 --- a/apps/server/src/persistence/Services/ProjectionThreadSessions.ts +++ b/apps/server/src/persistence/Services/ProjectionThreadSessions.ts @@ -29,6 +29,10 @@ export const ProjectionThreadSession = Schema.Struct({ runtimeMode: RuntimeMode, activeTurnId: Schema.NullOr(TurnId), lastError: Schema.NullOr(Schema.String), + // The limit-reset instant beside a usage-limit lastError; see + // OrchestrationSession.lastErrorLimitResetsAt. Optional so rows written + // before the column existed still decode. + lastErrorLimitResetsAt: Schema.optional(Schema.NullOr(IsoDateTime)), updatedAt: IsoDateTime, }); export type ProjectionThreadSession = typeof ProjectionThreadSession.Type; diff --git a/apps/server/src/persistence/Services/ProjectionThreads.ts b/apps/server/src/persistence/Services/ProjectionThreads.ts index 2c8186a321c0..157141a08852 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -47,6 +47,9 @@ export const ProjectionThread = Schema.Struct({ unsettledAt: Schema.NullOr(IsoDateTime), snoozedUntil: Schema.NullOr(IsoDateTime), snoozedAt: Schema.NullOr(IsoDateTime), + // Pending limit-reset continuation; see OrchestrationThread.autoContinueAt. + // Optional so rows written before the column existed still decode. + autoContinueAt: Schema.optional(Schema.NullOr(IsoDateTime)), pinnedAt: Schema.NullOr(IsoDateTime), pinOrderKey: Schema.optional(Schema.NullOr(Schema.String)), activeOrderKey: Schema.optional(Schema.NullOr(Schema.String)), diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 13645a33a05b..234fbac00231 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -27,7 +27,11 @@ import { } from "@anthropic-ai/claude-agent-sdk"; import { parseCliArgs } from "@t3tools/shared/cliArgs"; import { isWorkspaceImagePreviewPath } from "@t3tools/shared/filePreview"; -import { type ClaudeScopedLimitNames, claudeRateLimitEventToUpdate } from "./claudeUsageLimits.ts"; +import { + type ClaudeScopedLimitNames, + claudeRateLimitEventToUpdate, + isoFromEpochSeconds, +} from "./claudeUsageLimits.ts"; import { ApprovalRequestId, classifyTaskAgentKind, @@ -277,6 +281,9 @@ interface ClaudeTurnState { nextSyntheticAssistantBlockIndex: number; authenticationFailureMessage: string | undefined; rejectedRateLimitTypes: Set; + // Reset instants (ISO) of the windows in rejectedRateLimitTypes, so a + // limit-stopped turn can name when a continuation becomes possible. + readonly rejectedRateLimitResetsAt: Map; latestAssistantRateLimited: boolean; emittedThinkingText: boolean; readonly thinkingSnapshotIds: Set; @@ -2662,6 +2669,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( status: ProviderRuntimeTurnStatus, errorMessage?: string, result?: SDKResultMessage, + usageLimitResetsAt?: string, ) { const resultContextWindow = maxClaudeContextWindowFromModelUsage(result?.modelUsage); if (resultContextWindow !== undefined) { @@ -2835,6 +2843,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ? { totalCostUsd: result.total_cost_usd } : {}), ...(errorMessage ? { errorMessage } : {}), + ...(status === "failed" && usageLimitResetsAt ? { usageLimitResetsAt } : {}), tokenUsage: normalizeClaudeTurnTokenUsage(result, turnState.hasSubagents, status), }, providerRefs: nativeProviderRefs(context), @@ -3388,6 +3397,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( nextSyntheticAssistantBlockIndex: -1, authenticationFailureMessage: undefined, rejectedRateLimitTypes: new Set(), + rejectedRateLimitResetsAt: new Map(), latestAssistantRateLimited: false, emittedThinkingText: false, thinkingSnapshotIds: new Set(), @@ -3500,7 +3510,20 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( yield* emitRuntimeError(context, errorMessage ?? "Claude turn failed."); } - yield* completeTurn(context, status, errorMessage, message); + // The latest reset among the windows that blocked this turn: continuing + // earlier would stop on the still-parked window again. Only a limit stop + // names one — an auth failure with a stale window entry must not. + const usageLimitResetsAt = + status === "failed" && + turn !== undefined && + turn.authenticationFailureMessage === undefined && + (turn.rejectedRateLimitTypes.size > 0 || turn.latestAssistantRateLimited) + ? [...turn.rejectedRateLimitResetsAt.values()].sort( + (left, right) => Date.parse(right) - Date.parse(left), + )[0] + : undefined; + + yield* completeTurn(context, status, errorMessage, message, usageLimitResetsAt); }); /** @@ -4109,13 +4132,19 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( // Current blocking evidence is independent of whether its warning has // already been shown. A recovery can omit or advance the reset time; // its window type remains stable without clearing another window. - if (blocked) context.turnState.rejectedRateLimitTypes.add(limitType); - else if ( + if (blocked) { + context.turnState.rejectedRateLimitTypes.add(limitType); + const resetsAtIso = isoFromEpochSeconds(rateLimitInfo.resetsAt); + if (resetsAtIso !== undefined) { + context.turnState.rejectedRateLimitResetsAt.set(limitType, resetsAtIso); + } + } else if ( rateLimitInfo.status === "allowed" || rateLimitInfo.status === "allowed_warning" || overageAllowed ) { context.turnState.rejectedRateLimitTypes.delete(limitType); + context.turnState.rejectedRateLimitResetsAt.delete(limitType); } } if (blocked && context.turnState !== undefined) { @@ -5215,6 +5244,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( nextSyntheticAssistantBlockIndex: -1, authenticationFailureMessage: undefined, rejectedRateLimitTypes: new Set(), + rejectedRateLimitResetsAt: new Map(), latestAssistantRateLimited: false, emittedThinkingText: false, thinkingSnapshotIds: new Set(), diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index ba61d14f5cf5..3d22bb6b136d 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -75,6 +75,7 @@ import { type CodexRateLimitSnapshot, codexRateLimitsToUpdate, codexUsageLimitMessage, + codexUsageLimitResetsAt, mergeCodexRateLimits, } from "./codexUsageLimits.ts"; const isCodexAppServerProcessExitedError = Schema.is(CodexErrors.CodexAppServerProcessExitedError); @@ -2373,6 +2374,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( let usageLimitError: ProviderRuntimeEvent | undefined; let usageLimitMessage: string | undefined; + let usageLimitResetsAt: string | undefined; if (event.method === "turn/completed") { const completedPayload = readPayload( EffectCodexSchema.V2TurnCompletedNotification, @@ -2384,6 +2386,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( : undefined; if (turnError?.codexErrorInfo === "usageLimitExceeded") { usageLimitMessage = codexUsageLimitMessage(rateLimits, event.createdAt); + usageLimitResetsAt = codexUsageLimitResetsAt(rateLimits, event.createdAt); usageLimitError = { ...runtimeEventBase(event, event.threadId), type: "runtime.error", @@ -2403,6 +2406,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( payload: { ...runtimeEvent.payload, ...(usageLimitMessage ? { errorMessage: usageLimitMessage } : {}), + ...(usageLimitResetsAt ? { usageLimitResetsAt } : {}), tokenUsage: completeCodexTurnTokenUsage( turnTokenUsage, String(runtimeEvent.turnId), diff --git a/apps/server/src/provider/Layers/claudeUsageLimits.ts b/apps/server/src/provider/Layers/claudeUsageLimits.ts index b67645102e5a..2fc172f28920 100644 --- a/apps/server/src/provider/Layers/claudeUsageLimits.ts +++ b/apps/server/src/provider/Layers/claudeUsageLimits.ts @@ -100,7 +100,8 @@ function readModelScoped(rateLimits: object): ReadonlyArray { ); } -function isoFromEpochSeconds(value: number | undefined): string | undefined { +/** Claude reports reset instants as epoch seconds; windows store ISO strings. */ +export function isoFromEpochSeconds(value: number | undefined): string | undefined { if (value === undefined || !Number.isFinite(value) || value <= 0) return undefined; const dt = DateTime.make(value * 1000); return Option.isSome(dt) ? DateTime.formatIso(dt.value) : undefined; diff --git a/apps/server/src/provider/Layers/codexUsageLimits.test.ts b/apps/server/src/provider/Layers/codexUsageLimits.test.ts index 37006789986c..4316ed9750dc 100644 --- a/apps/server/src/provider/Layers/codexUsageLimits.test.ts +++ b/apps/server/src/provider/Layers/codexUsageLimits.test.ts @@ -7,6 +7,7 @@ import { codexRateLimitsToUpdate, codexResetCreditsToContract, codexUsageLimitMessage, + codexUsageLimitResetsAt, mergeCodexRateLimits, } from "./codexUsageLimits.ts"; @@ -265,6 +266,50 @@ describe("codexUsageLimitMessage", () => { }); }); +describe("codexUsageLimitResetsAt", () => { + const at = "2026-01-01T00:00:00.000Z"; + const atSeconds = Date.parse(at) / 1000; + + it("returns the exhausted window's reset instant the message names", () => { + expect( + codexUsageLimitResetsAt( + { + limitId: "codex", + primary: { usedPercent: 40, resetsAt: atSeconds + 3_600, windowDurationMins: 300 }, + secondary: { usedPercent: 100, resetsAt: atSeconds + 86_400, windowDurationMins: 10_080 }, + }, + at, + ), + ).toBe("2026-01-02T00:00:00.000Z"); + }); + + it("returns the latest reset when several windows are exhausted", () => { + expect( + codexUsageLimitResetsAt( + { + limitId: "codex", + primary: { usedPercent: 100, resetsAt: atSeconds + 3_600, windowDurationMins: 300 }, + secondary: { usedPercent: 100, resetsAt: atSeconds + 86_400, windowDurationMins: 10_080 }, + }, + at, + ), + ).toBe("2026-01-02T00:00:00.000Z"); + }); + + it("returns undefined when no exhausted window reports a future reset", () => { + expect( + codexUsageLimitResetsAt( + { + limitId: "codex", + primary: { usedPercent: 100, resetsAt: atSeconds - 60, windowDurationMins: 300 }, + }, + at, + ), + ).toBeUndefined(); + expect(codexUsageLimitResetsAt(undefined, at)).toBeUndefined(); + }); +}); + describe("mergeCodexRateLimits", () => { it("keeps windows an update does not carry", () => { const merged = mergeCodexRateLimits( diff --git a/apps/server/src/provider/Layers/codexUsageLimits.ts b/apps/server/src/provider/Layers/codexUsageLimits.ts index 74ce04a7895f..5619857f62b3 100644 --- a/apps/server/src/provider/Layers/codexUsageLimits.ts +++ b/apps/server/src/provider/Layers/codexUsageLimits.ts @@ -209,6 +209,25 @@ function codexUsageLimitNextStep(rateLimitReachedType: string | null | undefined } } +/** The exhausted window that has yet to reset, latest reset first. */ +function latestExhaustedWindow( + snapshot: CodexRateLimitSnapshot | undefined, + atIso: string, +): { readonly kind: string; readonly resetsAt: string; readonly resetMs: number } | undefined { + const atMs = Date.parse(atIso); + const windows = snapshot && Number.isFinite(atMs) ? codexRateLimitsToWindows(snapshot) : []; + let latest: { kind: string; resetsAt: string; resetMs: number } | undefined; + for (const window of windows) { + if (window.usedPercent < 100 || !window.resetsAt) continue; + const resetMs = Date.parse(window.resetsAt); + if (!Number.isFinite(resetMs) || resetMs <= atMs || (latest && resetMs <= latest.resetMs)) { + continue; + } + latest = { kind: window.kind, resetsAt: window.resetsAt, resetMs }; + } + return latest; +} + /** * The message a usage-limit stop shows instead of the provider sentence, which * on a Business workspace blames credits for a window that simply ran out. The @@ -220,15 +239,21 @@ export function codexUsageLimitMessage( atIso: string, ): string { const atMs = Date.parse(atIso); - const windows = snapshot && Number.isFinite(atMs) ? codexRateLimitsToWindows(snapshot) : []; - let reset = ""; - let latestResetMs = Number.NEGATIVE_INFINITY; - for (const window of windows) { - if (window.usedPercent < 100 || !window.resetsAt) continue; - const resetMs = Date.parse(window.resetsAt); - if (!Number.isFinite(resetMs) || resetMs <= atMs || resetMs <= latestResetMs) continue; - latestResetMs = resetMs; - reset = ` The ${window.kind} limit resets in ${formatCodexUsageLimitWait(resetMs - atMs)}.`; - } + const window = latestExhaustedWindow(snapshot, atIso); + const reset = window + ? ` The ${window.kind} limit resets in ${formatCodexUsageLimitWait(window.resetMs - atMs)}.` + : ""; return `Codex usage limit reached.${reset}${codexUsageLimitNextStep(snapshot?.rateLimitReachedType)}`; } + +/** + * The structured reset instant behind codexUsageLimitMessage's sentence, for + * the failed turn's `usageLimitResetsAt`. Undefined when no exhausted window + * reports a future reset — then there is nothing to schedule against. + */ +export function codexUsageLimitResetsAt( + snapshot: CodexRateLimitSnapshot | undefined, + atIso: string, +): string | undefined { + return latestExhaustedWindow(snapshot, atIso)?.resetsAt; +} diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 3fd0bb7274a0..7be43fafaa54 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -85,6 +85,7 @@ import { ProviderCommandReactorLive } from "./orchestration/Layers/ProviderComma import { CheckpointReactorLive } from "./orchestration/Layers/CheckpointReactor.ts"; import { ThreadDeletionReactorLive } from "./orchestration/Layers/ThreadDeletionReactor.ts"; import * as ThreadSettlementReactor from "./orchestration/ThreadSettlementReactor.ts"; +import * as AutoContinueReactor from "./orchestration/AutoContinueReactor.ts"; import * as StorageCleanup from "./storageCleanup.ts"; import * as PullRequestSyncReactor from "./orchestration/PullRequestSyncReactor.ts"; import * as ThreadPullRequestReactor from "./orchestration/ThreadPullRequestReactor.ts"; @@ -251,6 +252,7 @@ const ReactorLayerLive = Layer.empty.pipe( Layer.provideMerge(StorageCleanup.layer), Layer.provideMerge(ThreadDeletionReactorLive), Layer.provideMerge(ThreadSettlementReactor.layer), + Layer.provideMerge(AutoContinueReactor.layer), Layer.provideMerge(PullRequestSyncReactor.layer), Layer.provideMerge(ThreadPullRequestReactor.layer), Layer.provideMerge(AgentAwarenessRelay.layer.pipe(Layer.provide(ServerSecretStore.layer))), diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 79d393fc8d1f..5a8b1f4ba7ef 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -1474,7 +1474,13 @@ export default function ChatView(props: ChatViewProps) { const threadSyncPhase = routeKind === "server" ? (props.threadSyncPhase ?? null) : null; const threadDetailLoading = threadSyncPhase === "loading"; const handleNewThread = useNewThreadHandler(); - const { settleThread, pinThread, confirmAndUnpinThread } = useThreadActions(); + const { + settleThread, + pinThread, + confirmAndUnpinThread, + setThreadAutoContinue, + clearThreadAutoContinue, + } = useThreadActions(); const routeThreadRef = useMemo( () => scopeThreadRef(environmentId, threadId), [environmentId, threadId], @@ -6036,6 +6042,52 @@ export default function ChatView(props: ChatViewProps) { if (activeThreadRef === null || activeThreadWokeAt === null) return; markThreadVisited(scopedThreadKey(activeThreadRef), activeThreadWokeAt); }, [activeThreadRef, activeThreadWokeAt, markThreadVisited]); + // "Continue when the limit resets": offered beside a usage-limit failure + // whose reset instant the provider reported, and shown as a cancellable + // pending state once scheduled. nowMinute drives the future check without + // a repaint loop; the decider re-validates at fire time regardless. + const supportsAutoContinue = serverConfig?.environment.capabilities.threadAutoContinue === true; + const autoContinueResetsAt = activeServerThread?.session?.lastErrorLimitResetsAt ?? null; + const autoContinueScheduledFor = activeServerThread?.autoContinueAt ?? null; + const scheduleAutoContinue = useCallback(() => { + if (activeThreadRef === null || autoContinueResetsAt === null) return; + void setThreadAutoContinue(activeThreadRef, autoContinueResetsAt); + }, [activeThreadRef, autoContinueResetsAt, setThreadAutoContinue]); + const cancelAutoContinue = useCallback(() => { + if (activeThreadRef === null) return; + void clearThreadAutoContinue(activeThreadRef); + }, [activeThreadRef, clearThreadAutoContinue]); + const threadErrorAutoContinue = useMemo(() => { + if (!supportsAutoContinue || activeThreadRef === null) return null; + if (autoContinueScheduledFor !== null) { + return { + resetsAt: autoContinueResetsAt ?? autoContinueScheduledFor, + scheduledFor: autoContinueScheduledFor, + onSchedule: scheduleAutoContinue, + onCancel: cancelAutoContinue, + }; + } + if ( + autoContinueResetsAt === null || + !(Date.parse(autoContinueResetsAt) > Date.parse(nowMinute)) + ) { + return null; + } + return { + resetsAt: autoContinueResetsAt, + scheduledFor: null, + onSchedule: scheduleAutoContinue, + onCancel: cancelAutoContinue, + }; + }, [ + activeThreadRef, + autoContinueResetsAt, + autoContinueScheduledFor, + cancelAutoContinue, + nowMinute, + scheduleAutoContinue, + supportsAutoContinue, + ]); // Mirror of the sidebar's Woke pill for the open thread. const activeThreadLastVisitedAt = useUiStateStore((store) => activeThreadKey === null ? undefined : store.threadLastVisitedAtById[activeThreadKey], @@ -9851,6 +9903,7 @@ export default function ChatView(props: ChatViewProps) { /> { setThreadError(activeThread.id, null); dismissThreadErrorBannerForSession(threadErrorBannerKey); diff --git a/apps/web/src/components/chat/ThreadErrorBanner.tsx b/apps/web/src/components/chat/ThreadErrorBanner.tsx index 29a6a6db1846..84a6caa909d8 100644 --- a/apps/web/src/components/chat/ThreadErrorBanner.tsx +++ b/apps/web/src/components/chat/ThreadErrorBanner.tsx @@ -1,8 +1,10 @@ import { memo } from "react"; import { Alert, AlertAction, AlertDescription } from "../ui/alert"; import { Button } from "../ui/button"; -import { CircleAlertIcon, XIcon } from "lucide-react"; +import { CircleAlertIcon, ClockIcon, XIcon } from "lucide-react"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; +import { usePrimarySettings } from "../../hooks/useSettings"; +import { formatUpcomingTimestamp } from "../../timestampFormat"; export function getThreadErrorBannerKey(threadKey: string, error: string | null): string | null { return error === null ? null : `${threadKey}\u0000${error}`; @@ -33,38 +35,78 @@ export function isThreadErrorBannerDismissedForSession(bannerKey: string | null) return bannerKey !== null && sessionDismissedThreadErrorBannerKeys.has(bannerKey); } +/** + * Limit-reset continuation the banner can offer beside a usage-limit error. + * `resetsAt` is the provider-reported reset instant; `scheduledFor` is the + * thread's pending schedule, when one exists. + */ +export interface ThreadErrorBannerAutoContinue { + readonly resetsAt: string; + readonly scheduledFor: string | null; + readonly onSchedule: () => void; + readonly onCancel: () => void; +} + export const ThreadErrorBanner = memo(function ThreadErrorBanner({ error, onDismiss, + autoContinue, }: { error: string | null; onDismiss?: () => void; + autoContinue?: ThreadErrorBannerAutoContinue | null; }) { - if (!error) return null; + const timestampFormat = usePrimarySettings((settings) => settings.timestampFormat); + const scheduledFor = autoContinue?.scheduledFor ?? null; + // A pending continuation stays visible after the error text is dismissed: + // a scheduled turn the user could neither see nor cancel is a one-way door. + if (!error && scheduledFor === null) return null; return (
- + {error ? : } - - }>{error} - - {error} - - + {error ? ( + + }>{error} + + {error} + + + ) : null} + {autoContinue ? ( +
+ {scheduledFor !== null ? ( + <> + + Continuing when the limit resets ·{" "} + {formatUpcomingTimestamp(scheduledFor, timestampFormat)} + + + + ) : ( + + )} +
+ ) : null} - {onDismiss && ( + {error && onDismiss ? ( - )} + ) : null}
); diff --git a/apps/web/src/hooks/useThreadActions.ts b/apps/web/src/hooks/useThreadActions.ts index 55ff9992ce57..5fa1deb24111 100644 --- a/apps/web/src/hooks/useThreadActions.ts +++ b/apps/web/src/hooks/useThreadActions.ts @@ -26,6 +26,7 @@ import { refreshArchivedThreadsForEnvironment } from "../lib/archivedThreadsStat import { releaseComposerDraftUploads } from "../lib/composerDraftUploads"; import { readLocalApi } from "../localApi"; import { + readEnvironmentSupportsAutoContinue, readEnvironmentSupportsPinning, readEnvironmentSupportsPinReorder, readEnvironmentSupportsActiveReorder, @@ -92,6 +93,18 @@ export class ThreadSnoozeBlockedError extends Schema.TaggedError()( + "ThreadAutoContinueUnsupportedError", + { + environmentId: EnvironmentId, + threadId: ThreadId, + }, +) { + override get message(): string { + return "This environment's server does not support continuing when a limit resets yet. Update the server to use it."; + } +} + /** Key that sorts before every arranged pinned thread, so a fresh pin lands at the top of the run. Undefined (keyless, sorts with the legacy block) when key math can't produce one — pinning must never fail on placement. */ @@ -210,6 +223,12 @@ export function useThreadActions() { const unsnoozeThreadMutation = useAtomCommand(threadEnvironment.unsnooze, { reportFailure: false, }); + const setAutoContinueMutation = useAtomCommand(threadEnvironment.setAutoContinue, { + reportFailure: false, + }); + const clearAutoContinueMutation = useAtomCommand(threadEnvironment.clearAutoContinue, { + reportFailure: false, + }); const stopThreadSession = useAtomCommand(threadEnvironment.stopSession); const removeWorktree = useAtomCommand(vcsEnvironment.removeWorktree, { reportFailure: false, @@ -731,6 +750,47 @@ export function useThreadActions() { [unsnoozeThreadMutation], ); + const setThreadAutoContinue = useCallback( + async (target: ScopedThreadRef, autoContinueAt: string) => { + // Version skew: never send the command to a server that predates it. + if (!readEnvironmentSupportsAutoContinue(target.environmentId)) { + return AsyncResult.failure( + Cause.fail( + new ThreadAutoContinueUnsupportedError({ + environmentId: target.environmentId, + threadId: target.threadId, + }), + ), + ); + } + return setAutoContinueMutation({ + environmentId: target.environmentId, + input: { threadId: target.threadId, autoContinueAt }, + }); + }, + [setAutoContinueMutation], + ); + + const clearThreadAutoContinue = useCallback( + async (target: ScopedThreadRef) => { + if (!readEnvironmentSupportsAutoContinue(target.environmentId)) { + return AsyncResult.failure( + Cause.fail( + new ThreadAutoContinueUnsupportedError({ + environmentId: target.environmentId, + threadId: target.threadId, + }), + ), + ); + } + return clearAutoContinueMutation({ + environmentId: target.environmentId, + input: { threadId: target.threadId, reason: "user" }, + }); + }, + [clearAutoContinueMutation], + ); + const confirmAndDeleteThread = useCallback( async (target: ScopedThreadRef) => { const localApi = readLocalApi(); @@ -770,6 +830,8 @@ export function useThreadActions() { unsettleThread, snoozeThread, unsnoozeThread, + setThreadAutoContinue, + clearThreadAutoContinue, pinThread, unpinThread, confirmAndUnpinThread, @@ -778,12 +840,14 @@ export function useThreadActions() { }), [ archiveThread, + clearThreadAutoContinue, confirmAndDeleteThread, confirmAndUnpinThread, deleteThread, pinThread, reorderPinnedThread, reorderActiveThread, + setThreadAutoContinue, settleThread, snoozeThread, unarchiveThread, diff --git a/apps/web/src/state/entities.ts b/apps/web/src/state/entities.ts index d9610e20717f..d9276259af0f 100644 --- a/apps/web/src/state/entities.ts +++ b/apps/web/src/state/entities.ts @@ -202,6 +202,15 @@ export function readEnvironmentSupportsSnooze(environmentId: EnvironmentId): boo ); } +/** Whether the environment's server understands thread.auto-continue.set/.clear + and fires the continuation itself. Same version-skew contract as snooze. */ +export function readEnvironmentSupportsAutoContinue(environmentId: EnvironmentId): boolean { + return ( + appAtomRegistry.get(environmentServerConfigsAtom).get(environmentId)?.environment.capabilities + .threadAutoContinue === true + ); +} + /** Whether the environment's server understands thread.pin/unpin. Same version-skew contract as settlement. */ export function readEnvironmentSupportsPinning(environmentId: EnvironmentId): boolean { diff --git a/docs/user/usage.md b/docs/user/usage.md index 2e6b4acd9e4c..a122ec89ab85 100644 --- a/docs/user/usage.md +++ b/docs/user/usage.md @@ -69,6 +69,13 @@ current model's limits without leaving the conversation. The result opens above closes when you dismiss it or send your next message. It uses the same snapshot as **Usage → Limits**, so it does not run the agent or refresh anything. The command is offered only for providers that appear under **Usage → Limits**. +When a turn stops on a usage limit that reports its reset time, the failure banner offers +**Continue when limit resets**. The environment then sends "Continue" to the agent by itself once +the limit resets — the schedule lives on the server, so it fires even with every client closed, +and survives server restarts. Cancel it from the same banner, or just keep working: sending any +message, settling, or archiving the thread clears the scheduled continuation. Codex and Claude +report reset times when a limit stops a turn; other providers do not offer this. + OpenCode Go reports its session, weekly, and monthly allowance when OpenCode runs locally in the environment. T3 cannot report limits for external OpenCode servers because their credentials belong to the remote server. Cursor reports diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 8f313c1632ff..30afd34d6c6d 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -39,6 +39,8 @@ export type SettleThreadInput = CommandInput<"thread.settle">; export type UnsettleThreadInput = CommandInput<"thread.unsettle">; export type SnoozeThreadInput = CommandInput<"thread.snooze">; export type UnsnoozeThreadInput = CommandInput<"thread.unsnooze">; +export type SetThreadAutoContinueInput = CommandInput<"thread.auto-continue.set">; +export type ClearThreadAutoContinueInput = CommandInput<"thread.auto-continue.clear">; export type PinThreadInput = CommandInput<"thread.pin">; export type UnpinThreadInput = CommandInput<"thread.unpin">; export type ReorderPinnedThreadInput = CommandInput<"thread.pin.reorder">; @@ -206,6 +208,24 @@ export const unsnoozeThread: (input: UnsnoozeThreadInput) => CommandEffect = Eff }); }); +export const setThreadAutoContinue: (input: SetThreadAutoContinueInput) => CommandEffect = + Effect.fn("EnvironmentCommands.setThreadAutoContinue")(function* (input) { + return yield* dispatch({ + ...input, + type: "thread.auto-continue.set", + commandId: yield* commandId(input), + }); + }); + +export const clearThreadAutoContinue: (input: ClearThreadAutoContinueInput) => CommandEffect = + Effect.fn("EnvironmentCommands.clearThreadAutoContinue")(function* (input) { + return yield* dispatch({ + ...input, + type: "thread.auto-continue.clear", + commandId: yield* commandId(input), + }); + }); + export const pinThread: (input: PinThreadInput) => CommandEffect = Effect.fn( "EnvironmentCommands.pinThread", )(function* (input) { diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 93e22cfd0c70..94a397aa3c21 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -30,6 +30,8 @@ import { type ReorderPinnedThreadInput, type ReorderActiveThreadInput, type SettleThreadInput, + type SetThreadAutoContinueInput, + type ClearThreadAutoContinueInput, type SnoozeThreadInput, type StartThreadTurnInput, type StopThreadSessionInput, @@ -40,6 +42,7 @@ import { type UnsnoozeThreadInput, type UpdateThreadMetadataInput, archiveThread, + clearThreadAutoContinue, createThread, deleteThread, interruptThreadTurn, @@ -48,6 +51,7 @@ import { respondToThreadUserInput, dismissThreadUserInput, revertThreadCheckpoint, + setThreadAutoContinue, setThreadInteractionMode, setThreadRuntimeMode, pinThread, @@ -68,6 +72,7 @@ import type { EnvironmentRegistry } from "../connection/registry.ts"; export type { ArchiveThreadInput, + ClearThreadAutoContinueInput, CreateThreadInput, DeleteThreadInput, InterruptThreadTurnInput, @@ -82,6 +87,7 @@ export type { ReorderPinnedThreadInput, ReorderActiveThreadInput, SettleThreadInput, + SetThreadAutoContinueInput, SnoozeThreadInput, StartThreadTurnInput, StopThreadSessionInput, @@ -152,6 +158,18 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + setAutoContinue: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:set-auto-continue", + execute: (input: SetThreadAutoContinueInput) => setThreadAutoContinue(input), + scheduler, + concurrency, + }), + clearAutoContinue: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:clear-auto-continue", + execute: (input: ClearThreadAutoContinueInput) => clearThreadAutoContinue(input), + scheduler, + concurrency, + }), pin: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:pin", execute: (input: PinThreadInput) => pinThread(input), @@ -302,6 +320,20 @@ export function createThreadEnvironmentAtoms( snoozedUntil: null, snoozedAt: null, })), + setAutoContinue: optimistic.wrap(commands.setAutoContinue, (thread, input, now) => + // Mirrors the decider's future-time invariant so a rejected command + // never leaves an optimistic schedule behind. + !(Date.parse(input.autoContinueAt) > Date.parse(now)) + ? thread + : { + ...thread, + autoContinueAt: input.autoContinueAt, + }, + ), + clearAutoContinue: optimistic.wrap(commands.clearAutoContinue, (thread) => ({ + ...thread, + autoContinueAt: null, + })), pin: optimistic.wrap(commands.pin, (thread, input, now) => ({ ...thread, pinnedAt: thread.pinnedAt ?? now, diff --git a/packages/client-runtime/src/state/threadReducer.ts b/packages/client-runtime/src/state/threadReducer.ts index 101bb34fba91..b497804fabe0 100644 --- a/packages/client-runtime/src/state/threadReducer.ts +++ b/packages/client-runtime/src/state/threadReducer.ts @@ -215,6 +215,26 @@ export function applyThreadDetailEvent( }, }; + case "thread.auto-continue-scheduled": + return { + kind: "updated", + thread: { + ...thread, + autoContinueAt: event.payload.autoContinueAt, + updatedAt: event.payload.updatedAt, + }, + }; + + case "thread.auto-continue-cleared": + return { + kind: "updated", + thread: { + ...thread, + autoContinueAt: null, + updatedAt: event.payload.updatedAt, + }, + }; + case "thread.pinned": return { kind: "updated", diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 825dca87b631..f0001dcea370 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -116,6 +116,10 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ /** Server understands thread.snooze / thread.unsnooze commands. Same version-skew contract as threadSettlement. */ threadSnooze: Schema.optionalKey(Schema.Boolean), + /** Server understands thread.auto-continue.set / .clear and starts the + continuation turn itself once the limit resets. Same version-skew + contract as threadSettlement. */ + threadAutoContinue: Schema.optionalKey(Schema.Boolean), /** Server streams themes an environment publishes. Absent on servers from before environment themes shipped, which never emit the events -- so a client reconnecting to one must drop published themes rather than keep diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index e4a6e1166e85..52f427d54681 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -603,6 +603,10 @@ export const OrchestrationSession = Schema.Struct({ runtimeMode: RuntimeMode.pipe(Schema.withDecodingDefault(Effect.succeed(DEFAULT_RUNTIME_MODE))), activeTurnId: Schema.NullOr(TurnId), lastError: Schema.NullOr(TrimmedNonEmptyString), + // Set beside lastError when the failure was a provider usage-limit stop + // that reported when its window resets. Clients use it to offer "continue + // when the limit resets". Optional so pre-limit-reset payloads decode. + lastErrorLimitResetsAt: Schema.optional(Schema.NullOr(IsoDateTime)), updatedAt: IsoDateTime, }); export type OrchestrationSession = typeof OrchestrationSession.Type; @@ -805,6 +809,10 @@ export const OrchestrationThread = Schema.Struct({ // Optional so payloads from pre-snooze servers still decode. snoozedUntil: Schema.optional(Schema.NullOr(IsoDateTime)), snoozedAt: Schema.optional(Schema.NullOr(IsoDateTime)), + // A scheduled hands-off continuation: once this instant passes, the server + // starts a turn that resumes the work a usage limit stopped. Any manually + // started turn clears it. Optional so pre-auto-continue payloads decode. + autoContinueAt: Schema.optional(Schema.NullOr(IsoDateTime)), // Active pinned threads render in the pinned block. Settled and snoozed // threads remain in their respective shelves even when pinned. // Optional so payloads from pre-pinning servers still decode. @@ -884,6 +892,8 @@ export const OrchestrationThreadShell = Schema.Struct({ unsettledAt: Schema.optional(Schema.NullOr(IsoDateTime)), snoozedUntil: Schema.optional(Schema.NullOr(IsoDateTime)), snoozedAt: Schema.optional(Schema.NullOr(IsoDateTime)), + // See OrchestrationThread.autoContinueAt: pending limit-reset continuation. + autoContinueAt: Schema.optional(Schema.NullOr(IsoDateTime)), pinnedAt: Schema.optional(Schema.NullOr(IsoDateTime)), pinOrderKey: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), activeOrderKey: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), @@ -1170,6 +1180,40 @@ const ThreadUnsnoozeCommand = Schema.Struct({ reason: Schema.Literal("user"), }); +/** Schedules a hands-off continuation turn for when a usage limit resets. */ +const ThreadAutoContinueSetCommand = Schema.Struct({ + type: Schema.Literal("thread.auto-continue.set"), + commandId: CommandId, + threadId: ThreadId, + // When to start the continuation — the provider limit's reset instant. + autoContinueAt: IsoDateTime, +}); + +const ThreadAutoContinueClearCommand = Schema.Struct({ + type: Schema.Literal("thread.auto-continue.clear"), + commandId: CommandId, + threadId: ThreadId, + // Commands only carry "user": activity clears are decided server-side (any + // turn start spends the schedule with reason "activity"). + reason: Schema.Literal("user"), +}); + +/** + * Server-only trigger the auto-continue reactor dispatches when a scheduled + * continuation comes due. Carries the schedule it read as a compare-and-set + * guard, so a reschedule or cancel that raced the sweep wins. + */ +const ThreadAutoContinueFireCommand = Schema.Struct({ + type: Schema.Literal("thread.auto-continue.fire"), + commandId: CommandId, + threadId: ThreadId, + autoContinueAt: IsoDateTime, + // Identity of the continuation's user message, minted by the reactor so + // the decider stays deterministic per command. + messageId: MessageId, + createdAt: IsoDateTime, +}); + const ThreadPinCommand = Schema.Struct({ type: Schema.Literal("thread.pin"), commandId: CommandId, @@ -1402,6 +1446,8 @@ const DispatchableClientOrchestrationCommand = Schema.Union([ ThreadUnsettleCommand, ThreadSnoozeCommand, ThreadUnsnoozeCommand, + ThreadAutoContinueSetCommand, + ThreadAutoContinueClearCommand, ThreadPinCommand, ThreadUnpinCommand, ThreadPinReorderCommand, @@ -1435,6 +1481,8 @@ export const ClientOrchestrationCommand = Schema.Union([ ThreadUnsettleCommand, ThreadSnoozeCommand, ThreadUnsnoozeCommand, + ThreadAutoContinueSetCommand, + ThreadAutoContinueClearCommand, ThreadPinCommand, ThreadUnpinCommand, ThreadPinReorderCommand, @@ -1624,6 +1672,7 @@ const ThreadPullRequestLinkSyncCommand = Schema.Struct({ const InternalOrchestrationCommand = Schema.Union([ ThreadAutoSettleCommand, + ThreadAutoContinueFireCommand, ThreadPullRequestSyncCommand, ThreadPullRequestLinkSyncCommand, ThreadSessionSetCommand, @@ -1663,6 +1712,8 @@ export const OrchestrationEventType = Schema.Literals([ "thread.unsettled", "thread.snoozed", "thread.unsnoozed", + "thread.auto-continue-scheduled", + "thread.auto-continue-cleared", "thread.pinned", "thread.unpinned", "thread.pin-reordered", @@ -1784,6 +1835,21 @@ export const ThreadUnsnoozedPayload = Schema.Struct({ updatedAt: IsoDateTime, }); +export const ThreadAutoContinueScheduledPayload = Schema.Struct({ + threadId: ThreadId, + autoContinueAt: IsoDateTime, + updatedAt: IsoDateTime, +}); + +export const ThreadAutoContinueClearedPayload = Schema.Struct({ + threadId: ThreadId, + // user: the cancel button. activity: a turn start spent the schedule (a + // manual send or the scheduled continuation itself), or a settle/archive + // parked the thread — mirrors thread.unsnoozed's activity resets. + reason: Schema.Literals(["user", "activity"]), + updatedAt: IsoDateTime, +}); + export const ThreadPinnedPayload = Schema.Struct({ threadId: ThreadId, pinnedAt: IsoDateTime, @@ -2052,6 +2118,16 @@ export const OrchestrationEvent = Schema.Union([ type: Schema.Literal("thread.unsnoozed"), payload: ThreadUnsnoozedPayload, }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.auto-continue-scheduled"), + payload: ThreadAutoContinueScheduledPayload, + }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.auto-continue-cleared"), + payload: ThreadAutoContinueClearedPayload, + }), Schema.Struct({ ...EventBaseFields, type: Schema.Literal("thread.pinned"), diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index af1baac74f9d..937315e76215 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -404,6 +404,10 @@ const TurnCompletedPayload = Schema.Struct({ modelUsage: Schema.optional(UnknownRecordSchema), totalCostUsd: Schema.optional(Schema.Number), errorMessage: Schema.optional(TrimmedNonEmptyStringSchema), + // Present on failed turns stopped by a usage limit whose reset instant the + // provider reported. Lets clients schedule a continuation for that moment; + // errorMessage still carries the human sentence. + usageLimitResetsAt: Schema.optional(IsoDateTime), tokenUsage: Schema.optional(TurnTokenUsage), }); export type TurnCompletedPayload = typeof TurnCompletedPayload.Type;