diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 9c4d86ce16bf6..f0946f09186e8 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -1677,6 +1677,13 @@ function ThreadDetailBody( key={props.selectedThread.latestRun?.runId} thread={props.selectedThread} environmentId={props.environmentId} + resetCredits={ + props.serverConfig?.providers.find( + (provider) => + provider.instanceId === + props.selectedThread.runtime?.providerInstanceId, + )?.usageLimits?.resetCredits + } /> ({ })); vi.mock("../../state/threads", () => ({ threadEnvironment: { updateMetadata: {} } })); vi.mock("../../state/use-atom-command", () => ({ useAtomCommand: () => vi.fn() })); +vi.mock("../usage/useResetCredit", () => ({ useResetCredit: vi.fn() })); import { UsageLimitRecoveryCard } from "./UsageLimitRecoveryCard"; diff --git a/apps/mobile/src/features/threads/UsageLimitRecoveryCard.tsx b/apps/mobile/src/features/threads/UsageLimitRecoveryCard.tsx index 201fc78234f5a..cd76c9d77468e 100644 --- a/apps/mobile/src/features/threads/UsageLimitRecoveryCard.tsx +++ b/apps/mobile/src/features/threads/UsageLimitRecoveryCard.tsx @@ -2,7 +2,11 @@ import { squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime" import { AsyncResult } from "effect/reactivity"; import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell"; import { limitRecoveryOwnsSnooze } from "@t3tools/client-runtime/state/thread-execution"; -import type { EnvironmentId } from "@t3tools/contracts"; +import type { + EnvironmentId, + ProviderInstanceId, + ServerProviderResetCredits, +} from "@t3tools/contracts"; import { CHATGPT_USAGE_LIMIT_CODE, CHATGPT_USAGE_URL, @@ -17,13 +21,17 @@ import { AppText as Text } from "../../components/ui/AppText"; import { useDeadlinePassed } from "../../lib/clock"; import { threadEnvironment } from "../../state/threads"; import { useAtomCommand } from "../../state/use-atom-command"; +import { useResetCredit } from "../usage/useResetCredit"; export function UsageLimitRecoveryCard({ thread, environmentId, + resetCredits, }: { thread: EnvironmentThreadShell; environmentId: EnvironmentId; + /** The limited account's banked reset credits, when it reports any. */ + resetCredits?: ServerProviderResetCredits | undefined; }) { const updateMetadata = useAtomCommand(threadEnvironment.updateMetadata); const [pending, setPending] = useState(false); @@ -48,6 +56,7 @@ export function UsageLimitRecoveryCard({ : resetAt; const [copied, setCopied] = useState(false); const [actionPending, setActionPending] = useState(false); + const instanceId = thread.runtime?.providerInstanceId ?? null; const code = thread.runtime?.lastErrorCode; // OpenAI's partner guidelines specify the ChatGPT notice and its action. const workspaceLimit = code === CHATGPT_WORKSPACE_USAGE_LIMIT_CODE; @@ -59,6 +68,7 @@ export function UsageLimitRecoveryCard({ !runId ) return null; + const canReset = instanceId !== null && (resetCredits?.availableCount ?? 0) > 0; const resetText = displayAt ? `${canSchedule && scheduled ? "Resumes" : "Resets"} ${new Date(displayAt).toLocaleString( undefined, @@ -155,9 +165,12 @@ export function UsageLimitRecoveryCard({ ? `Usage limit reached. ${resetText}` : "Usage limit reached. The provider did not report a reset time. Retry manually when your limit is available."} - {chatGptLimit || canSchedule ? ( + {chatGptLimit || canSchedule || canReset ? ( // The margins cancel the 44pt touch height so the links sit 8pt under the text. + {canReset && instanceId !== null ? ( + + ) : null} {chatGptLimit ? ( + + {status ? {status} : null} + + ); +} + function RecoveryAction(props: { readonly label: string; readonly role?: "button" | "link"; diff --git a/apps/mobile/src/features/usage/UsageLimitsSection.tsx b/apps/mobile/src/features/usage/UsageLimitsSection.tsx index 214e64fab628b..1833de5976a58 100644 --- a/apps/mobile/src/features/usage/UsageLimitsSection.tsx +++ b/apps/mobile/src/features/usage/UsageLimitsSection.tsx @@ -1,10 +1,7 @@ -import * as Dialog from "../../components/ui/Dialog"; import { RegistryContext, useAtomMount, useAtomValue } from "@effect/atom-react"; import { normalizeProviderAccentColor } from "@t3tools/client-runtime/state/provider-instance-display"; import { - AuthProvidersManageScope, type EnvironmentId, - type ProviderConsumeResetCreditOutcome, type ProviderConsumeResetCreditInput, type ServerProvider, type ServerProviderResetCredits, @@ -25,9 +22,6 @@ import { Linking, Pressable, View } from "react-native"; import { AppText as Text } from "../../components/ui/AppText"; import { ProviderIcon } from "../../components/ProviderIcon"; import { environmentPresentations } from "../../state/presentation"; -import { serverEnvironment } from "../../state/server"; -import { readEnvironmentScope, useEnvironmentScope } from "../../state/session"; -import { useAtomCommand } from "../../state/use-atom-command"; import { useNowMinuteMs } from "../../lib/clock"; import { environmentIdsKey, @@ -36,6 +30,7 @@ import { refreshEnvironmentLimits, } from "./usageLimitsRefresh"; import { usageProviderKind, useProviderColors } from "./usageProviders"; +import { useResetCredit } from "./useResetCredit"; const PACE_LABEL = { ahead: "ahead of pace", on: "on pace", under: "under pace" } as const; @@ -220,13 +215,6 @@ export function AccountLimits(props: { ); } -const OUTCOME_TEXT: Record = { - reset: "Reset applied. Your windows have cleared.", - nothingToReset: "Nothing to reset right now.", - noCredit: "No reset credit left.", - alreadyRedeemed: "That credit was already redeemed.", -}; - export function ResetCredits(props: { readonly environmentId: EnvironmentId; readonly input: ProviderConsumeResetCreditInput; @@ -235,12 +223,7 @@ export function ResetCredits(props: { readonly dense?: boolean; }) { const { environmentId, input, credits, now, dense = false } = props; - const canManageProviders = useEnvironmentScope(environmentId, AuthProvidersManageScope); - const consume = useAtomCommand(serverEnvironment.consumeResetCredit, { - reportFailure: false, - }); - const [busy, setBusy] = useState(false); - const [status, setStatus] = useState(null); + const { canManageProviders, busy, status, confirm } = useResetCredit(environmentId, input); if (dense && credits.availableCount === 0 && status === null) return null; const expiresIn = credits.nextExpiresAt @@ -253,35 +236,6 @@ export function ResetCredits(props: { expiresIn ? ` · next expires in ${expiresIn}` : "" }`; - const redeem = async () => { - if (!readEnvironmentScope(environmentId, AuthProvidersManageScope)) return; - setBusy(true); - setStatus(null); - const result = await consume({ environmentId, input }); - setBusy(false); - if (result._tag === "Success") { - setStatus(result.value.warning ?? OUTCOME_TEXT[result.value.outcome]); - return; - } - setStatus( - "error" in result.cause && result.cause.error instanceof Error - ? result.cause.error.message - : "Could not use the reset credit.", - ); - }; - - const confirm = () => { - if (!readEnvironmentScope(environmentId, AuthProvidersManageScope)) return; - Dialog.alert( - "Use a reset credit?", - "This redeems one credit on your account and clears the current rate-limit windows. It cannot be undone.", - [ - { text: "Cancel", style: "cancel" }, - { text: "Use credit", onPress: () => void redeem() }, - ], - ); - }; - return ( {summary} diff --git a/apps/mobile/src/features/usage/useResetCredit.test.tsx b/apps/mobile/src/features/usage/useResetCredit.test.tsx new file mode 100644 index 0000000000000..2c2432de4c42a --- /dev/null +++ b/apps/mobile/src/features/usage/useResetCredit.test.tsx @@ -0,0 +1,112 @@ +import { createRef, useImperativeHandle } from "react"; +import { EnvironmentId, ProviderInstanceId } from "@t3tools/contracts"; +import { act } from "react"; +import { createRoot } from "react-dom/client"; +import { expect, it, vi } from "vite-plus/test"; + +const mocks = vi.hoisted(() => ({ consume: vi.fn(), canManage: vi.fn() })); +vi.mock("../../components/ui/Dialog", () => ({ alert: vi.fn() })); +vi.mock("../../state/server", () => ({ serverEnvironment: { consumeResetCredit: {} } })); +vi.mock("../../state/session", () => ({ + readEnvironmentScope: mocks.canManage, + useEnvironmentScope: () => true, +})); +vi.mock("../../state/use-atom-command", () => ({ useAtomCommand: () => mocks.consume })); + +import * as Dialog from "../../components/ui/Dialog"; +import { useResetCredit } from "./useResetCredit"; + +const environmentId = EnvironmentId.make("reset-environment"); +const input = { instanceId: ProviderInstanceId.make("codex") }; +const success = { _tag: "Success", value: { outcome: "reset" } } as const; +type Result = typeof success | { readonly _tag: "Failure"; readonly cause: { error: Error } }; + +function setup() { + mocks.consume.mockReset().mockResolvedValue(success); + mocks.canManage.mockReset().mockReturnValue(true); + vi.mocked(Dialog.alert).mockClear(); + const root = createRoot(document.createElement("div")); + const ref = createRef>(); + function Probe() { + const value = useResetCredit(environmentId, input); + useImperativeHandle(ref, () => value); + return null; + } + act(() => root.render()); + return { + get value() { + if (ref.current === null) throw new Error("Reset-credit hook was not mounted"); + return ref.current; + }, + [Symbol.dispose]: () => act(() => root.unmount()), + }; +} + +function approval() { + const action = vi + .mocked(Dialog.alert) + .mock.lastCall?.[2]?.find((button) => button.text === "Use credit")?.onPress; + if (action === undefined) throw new Error("Missing reset-credit confirmation"); + return action; +} + +it.each(["success", "failure", "rejection"] as const)( + "coalesces overlapping approvals and permits a new attempt after %s", + async (outcome) => { + using fixture = setup(); + let resolve!: (value: Result) => void; + let reject!: (reason: Error) => void; + const promise = new Promise((complete, fail) => { + resolve = complete; + reject = fail; + }); + const pending = { promise, resolve, reject }; + mocks.consume.mockReturnValueOnce(pending.promise); + fixture.value.confirm(); + const first = approval(); + fixture.value.confirm(); + const second = approval(); + act(() => { + first(); + second(); + }); + expect(mocks.consume).toHaveBeenCalledTimes(1); + expect(mocks.consume).toHaveBeenCalledWith({ environmentId, input }); + expect(fixture.value.busy).toBe(true); + const prompts = vi.mocked(Dialog.alert).mock.calls.length; + fixture.value.confirm(); + expect(Dialog.alert).toHaveBeenCalledTimes(prompts); + + await act(async () => { + if (outcome === "rejection") pending.reject(new Error("Disconnected")); + else if (outcome === "failure") { + pending.resolve({ _tag: "Failure", cause: { error: new Error("Denied") } }); + } else pending.resolve(success); + await pending.promise.catch(() => undefined); + }); + expect(fixture.value.busy).toBe(false); + expect(fixture.value.status).toBe( + outcome === "success" + ? "Reset applied. Your windows have cleared." + : outcome === "failure" + ? "Denied" + : "Could not use the reset credit.", + ); + await act(async () => { + fixture.value.confirm(); + approval()(); + }); + expect(mocks.consume).toHaveBeenCalledTimes(2); + expect(fixture.value.busy).toBe(false); + }, +); + +it("rechecks management permission after the confirmation opens", () => { + using fixture = setup(); + fixture.value.confirm(); + const confirm = approval(); + mocks.canManage.mockReturnValue(false); + act(() => confirm()); + expect(mocks.consume).not.toHaveBeenCalled(); + expect(fixture.value.busy).toBe(false); +}); diff --git a/apps/mobile/src/features/usage/useResetCredit.ts b/apps/mobile/src/features/usage/useResetCredit.ts new file mode 100644 index 0000000000000..64c7bb8b518b9 --- /dev/null +++ b/apps/mobile/src/features/usage/useResetCredit.ts @@ -0,0 +1,66 @@ +import { + AuthProvidersManageScope, + type EnvironmentId, + type ProviderConsumeResetCreditInput, + type ProviderConsumeResetCreditOutcome, +} from "@t3tools/contracts"; +import { useRef, useState } from "react"; + +import * as Dialog from "../../components/ui/Dialog"; +import { serverEnvironment } from "../../state/server"; +import { readEnvironmentScope, useEnvironmentScope } from "../../state/session"; +import { useAtomCommand } from "../../state/use-atom-command"; + +const OUTCOME_TEXT: Record = { + reset: "Reset applied. Your windows have cleared.", + nothingToReset: "Nothing to reset right now.", + noCredit: "No reset credit left.", + alreadyRedeemed: "That credit was already redeemed.", +}; + +/** Spending a credit asks for confirmation first because it cannot be undone. */ +export function useResetCredit( + environmentId: EnvironmentId, + input: ProviderConsumeResetCreditInput, +) { + const canManageProviders = useEnvironmentScope(environmentId, AuthProvidersManageScope); + const consume = useAtomCommand(serverEnvironment.consumeResetCredit, { + reportFailure: false, + }); + const inFlight = useRef(false); + const [busy, setBusy] = useState(false); + const [status, setStatus] = useState(null); + + const redeem = async () => { + if (inFlight.current || !readEnvironmentScope(environmentId, AuthProvidersManageScope)) return; + inFlight.current = true; + setBusy(true); + setStatus(null); + const result = await consume({ environmentId, input }).catch(() => null); + inFlight.current = false; + setBusy(false); + if (result?._tag === "Success") { + setStatus(result.value.warning ?? OUTCOME_TEXT[result.value.outcome]); + return; + } + setStatus( + result !== null && "error" in result.cause && result.cause.error instanceof Error + ? result.cause.error.message + : "Could not use the reset credit.", + ); + }; + + const confirm = () => { + if (inFlight.current || !readEnvironmentScope(environmentId, AuthProvidersManageScope)) return; + Dialog.alert( + "Use a reset credit?", + "This redeems one credit on your account and clears the current rate-limit windows. It cannot be undone.", + [ + { text: "Cancel", style: "cancel" }, + { text: "Use credit", onPress: () => void redeem() }, + ], + ); + }; + + return { canManageProviders, busy, status, confirm }; +} diff --git a/apps/mobile/vite.config.ts b/apps/mobile/vite.config.ts index 6291a1ff27708..e92cbbb93b7fa 100644 --- a/apps/mobile/vite.config.ts +++ b/apps/mobile/vite.config.ts @@ -53,6 +53,7 @@ const reactTests = [ "src/features/connection/LinkedEnvironmentsSection.test.tsx", "src/features/devices/device-stream.browser.test.ts", "src/features/keyboard/keyboardControllerState.test.tsx", + "src/features/usage/useResetCredit.test.tsx", ]; const sourceRoot = `${normalizePath(NodeURL.fileURLToPath(new URL("./src", import.meta.url)))}/`; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 0a6c689e5ccd2..1aa4ee632c694 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1346,6 +1346,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { beforeEmitInbound === undefined ? {} : { beforeEmitInbound }, ); const clientFactory: CodexAdapterV2.CodexAppServerClientFactoryShape = { + threadReleases: yield* CodexAdapterV2.makeCodexThreadReleases, open: (openInput) => Layer.build(CodexReplay.layerReplayWithDriver(driver)).pipe( Effect.mapError( @@ -3958,6 +3959,118 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("does not retry an external writer because of an old completed release", () => + Effect.gen(function* () { + const releases = yield* CodexAdapterV2.makeCodexThreadReleases; + yield* releases.begin("old-thread"); + yield* releases.begin("pending-thread"); + yield* releases.finish("old-thread"); + yield* TestClock.adjust("91 seconds"); + assert.equal(yield* releases.awaitRelease("old-thread"), false); + yield* releases.begin("recent-thread"); + yield* releases.finish("recent-thread"); + assert.equal(yield* releases.awaitRelease("recent-thread"), true); + assert.equal(yield* releases.awaitRelease("recent-thread"), false); + const cancelledWait = yield* releases + .awaitRelease("pending-thread") + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Fiber.interrupt(cancelledWait); + yield* releases.finish("pending-thread"); + assert.equal(yield* releases.awaitRelease("pending-thread"), true); + }), + ); + + it.effect("resumes a thread once this server's unsubscribed copy closes", () => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "thread-switching-accounts"; + const resume = (id: number): CodexReplay.CodexAppServerReplayEntry => ({ + type: "expect_outbound", + label: `thread/resume ${id}`, + frame: { + id, + method: "thread/resume", + params: { + threadId: nativeThreadId, + modelProvider: "openai", + excludeTurns: true, + cwd: CODEX_TEST_RUNTIME_POLICY.cwd, + model: CODEX_TEST_MODEL_SELECTION.model, + config: CodexAdapterV2.CODEX_THREAD_CONFIG, + }, + }, + }); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "codex-resume-after-release", + entries: [ + ...codexReplayPreamble({ + nativeThreadId, + nativeTurnId: "unused-turn", + prompt: "unused-prompt", + }).slice(0, 5), + { + type: "expect_outbound", + label: "unsubscribe", + frame: { + id: 3, + method: "thread/unsubscribe", + params: { threadId: nativeThreadId }, + }, + }, + { + type: "emit_inbound", + label: "unsubscribed", + frame: { id: 3, result: { status: "unsubscribed" } }, + }, + resume(4), + { + type: "emit_inbound", + label: "active writer", + frame: { + id: 4, + error: { + code: -32600, + message: `thread ${nativeThreadId} already has an active writer`, + }, + }, + }, + { + type: "emit_inbound", + label: "closed", + frame: { method: "thread/closed", params: { threadId: nativeThreadId } }, + }, + resume(5), + { + type: "emit_inbound", + label: "resumed", + frame: { id: 5, result: { thread: { id: nativeThreadId, updatedAt: 1782622450 } } }, + }, + ], + }), + ); + + yield* harness.runtime.unloadThread!({ providerThread: harness.providerThread }); + const resumed = yield* harness.runtime.resumeThread({ + providerThread: harness.providerThread, + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + }); + + assert.equal(resumed.nativeThreadRef?.nativeId, nativeThreadId); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + it.effect("reports a thread another Codex client is writing to as in use", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index dda4e561d6bcc..2671f3a75a192 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -82,10 +82,12 @@ import * as CodexClient from "effect-codex-app-server/client"; import * as CodexErrors from "effect-codex-app-server/errors"; import * as CodexSchema from "effect-codex-app-server/schema"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import * as Digest from "effect/crypto/Digest"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; @@ -1392,8 +1394,76 @@ export interface CodexAppServerClientFactoryShape { ProviderAdapter.ProviderAdapterOpenSessionError, Scope.Scope >; + /** Shared by every Codex instance, since accounts share one sessions directory. */ + readonly threadReleases?: CodexThreadReleases; } +// Codex keeps an unsubscribed thread's writer lock until its idle sweep closes it, +// about a minute later, so another account resuming it right away is refused. +const CODEX_THREAD_RELEASE_TIMEOUT = Duration.seconds(90); + +export interface CodexThreadReleases { + readonly begin: (nativeThreadId: string) => Effect.Effect; + readonly finish: (nativeThreadId: string) => Effect.Effect; + /** Waits for a release this server started; false when there is none or it timed out. */ + readonly awaitRelease: (nativeThreadId: string) => Effect.Effect; +} + +export const makeCodexThreadReleases = Clock.clockWith((clock) => + Effect.sync((): CodexThreadReleases => { + const releases = new Map>(); + // Keep close-before-resume evidence briefly, including idle unloads never resumed here. + const completedAt = new Map(); + const pruneCompleted = () => { + const cutoff = + clock.currentTimeMillisUnsafe() - Duration.toMillis(CODEX_THREAD_RELEASE_TIMEOUT); + for (const [nativeThreadId, at] of completedAt) { + if (at > cutoff) break; + completedAt.delete(nativeThreadId); + releases.delete(nativeThreadId); + } + }; + return { + begin: (nativeThreadId) => + Effect.sync(() => { + pruneCompleted(); + const current = releases.get(nativeThreadId); + if (current === undefined || Deferred.isDoneUnsafe(current)) { + completedAt.delete(nativeThreadId); + releases.set(nativeThreadId, Deferred.makeUnsafe()); + } + }), + finish: (nativeThreadId) => + Effect.suspend(() => { + const release = releases.get(nativeThreadId); + if (release === undefined) return Effect.void; + if (!completedAt.has(nativeThreadId)) { + completedAt.set(nativeThreadId, clock.currentTimeMillisUnsafe()); + } + return Deferred.succeed(release, undefined); + }), + awaitRelease: (nativeThreadId) => + Effect.suspend(() => { + pruneCompleted(); + const release = releases.get(nativeThreadId); + return release === undefined + ? Effect.succeed(false) + : Deferred.await(release).pipe( + Effect.timeoutOption(CODEX_THREAD_RELEASE_TIMEOUT), + Effect.map(Option.isSome), + Effect.tap(() => + Effect.sync(() => { + if (releases.get(nativeThreadId) !== release) return; + releases.delete(nativeThreadId); + completedAt.delete(nativeThreadId); + }), + ), + ); + }), + }; + }), +); + export class CodexAppServerClientFactory extends Context.Service< CodexAppServerClientFactory, CodexAppServerClientFactoryShape @@ -1631,6 +1701,7 @@ export const layerAppServerClientFactory: Layer.Layer< ); return CodexAppServerClientFactory.of({ + threadReleases: yield* makeCodexThreadReleases, open: Effect.fn("CodexAdapterV2.open")(function* (input) { const scope = yield* Scope.Scope; const environment = { @@ -1817,6 +1888,17 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( settings: resolvedRuntime?.config ?? adapterOptions.settings, environment, }); + const { threadReleases } = clientFactory; + const releasingThreadIds = new Set(); + const finishThreadRelease = (nativeThreadId: string) => + Effect.suspend(() => { + releasingThreadIds.delete(nativeThreadId); + return threadReleases?.finish(nativeThreadId) ?? Effect.void; + }); + // The app-server drops every writer lock when it exits. + yield* Effect.addFinalizer(() => + Effect.forEach([...releasingThreadIds], finishThreadRelease, { discard: true }), + ); const additionalContextByThread = yield* Ref.make( new Map< string, @@ -4918,6 +5000,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( // their command items, while the shared provider session stays alive. yield* client.handleServerNotification("thread/closed", (payload) => Effect.gen(function* () { + yield* finishThreadRelease(payload.threadId); const completedAt = yield* DateTime.now; for (const context of (yield* Ref.get(settledTurns)).values()) { if (context.providerThread.nativeThreadRef?.nativeId !== payload.threadId) continue; @@ -6869,6 +6952,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( .pipe(Effect.andThen(resume)) : Effect.fail(cause), }), + // Another account's app-server here may still hold the lock it is releasing. + Effect.catchIf(isCodexActiveWriterError, (cause) => + (clientFactory.threadReleases?.awaitRelease(threadId) ?? Effect.succeed(false)).pipe( + Effect.flatMap((released) => (released ? resume : Effect.fail(cause))), + ), + ), ); }); @@ -7431,7 +7520,13 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( function* (unloadInput) { const nativeThreadId = yield* getNativeThreadId(unloadInput.providerThread); sessionApprovals.delete(nativeThreadId); - yield* client.request("thread/unsubscribe", { threadId: nativeThreadId }); + releasingThreadIds.add(nativeThreadId); + yield* threadReleases?.begin(nativeThreadId) ?? Effect.void; + const { status } = yield* client + .request("thread/unsubscribe", { threadId: nativeThreadId }) + .pipe(Effect.onError(() => finishThreadRelease(nativeThreadId))); + // Only the last subscriber's unsubscribe schedules a close. + if (status !== "unsubscribed") yield* finishThreadRelease(nativeThreadId); }, (effect, unloadInput) => effect.pipe( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 4ae31a714ebcc..2c084866ce94a 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -4090,6 +4090,80 @@ it.effect( }), ); +it.effect( + "ProviderSessionManagerV2 unloads a native thread already rebound to another account's session", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("account-switch"); + const providerSessionId = idAllocator.derive.providerSession({ + providerInstanceId: modelSelection.instanceId, + }); + const providerThread = makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId, now }), + { + id: yield* idAllocator.allocate.event({ threadId }), + type: "provider-thread.updated", + threadId, + occurredAt: now, + payload: providerThread, + }, + ], + }); + const runtime = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + yield* runtime.resumeThread({ providerThread, threadId, modelSelection, runtimePolicy }); + const otherAccount = ProviderInstanceId.make("codex_2"); + yield* eventSink.write({ + events: [ + { + id: yield* idAllocator.allocate.event({ threadId }), + type: "provider-thread.updated", + threadId, + occurredAt: now, + payload: { + ...providerThread, + providerInstanceId: otherAccount, + providerSessionId: idAllocator.derive.providerSession({ + providerInstanceId: otherAccount, + }), + }, + }, + { + // The account-switch commit detaches the old session before its detach + // effect runs, so the projection no longer returns that session. + id: yield* idAllocator.allocate.event({ threadId }), + type: "provider-session.detached", + threadId, + occurredAt: now, + payload: { providerSessionId, detachedAt: now, reason: "provider_switch" }, + }, + ], + }); + + yield* manager.detach({ providerSessionId, threadId }); + + assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, ["native-thread"]); + }).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })), Effect.scoped); + }), +); + it.effect( "ProviderSessionManagerV2 re-attaching a thread waits for its in-flight unload, then reloads it", () => diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 7081207f8e85a..3fc414caa3785 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -2435,7 +2435,19 @@ export const layerWithOptions = ( .filter((thread) => thread.providerSessionId === input.providerSessionId) .map((thread) => [thread.id, thread] as const), ); - detachedProviderThreads = [...providerThreads.values()]; + // The detached event already removed this session's binding, so the + // projection no longer returns it; the live runtime still knows its driver. + const driver = currentEntry.runtime.driver; + // A switch to another account rebinds this thread's native threads to the new + // session in the same commit, but this process still holds their writer locks. + const reboundProviderThreads = projection.value.providerThreads.filter( + (thread) => + !providerThreads.has(thread.id) && + thread.driver === driver && + thread.nativeThreadRef !== null && + (thread.appThreadId ?? input.threadId) === input.threadId, + ); + detachedProviderThreads = [...providerThreads.values(), ...reboundProviderThreads]; const activeTurns = projection.value.providerTurns.filter( (turn) => turn.status === "running" && providerThreads.has(turn.providerThreadId), ); diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 016571d10dbb0..ba5553843d4df 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -2563,6 +2563,20 @@ export default function ChatView(props: ChatViewProps) { }); return item ? [item] : []; }); + const limitRecoveryInstanceId = serverRuntime?.providerInstanceId ?? null; + const limitRecoveryCredits = + limitRecoveryRun === null || limitRecoveryInstanceId === null + ? undefined + : providerStatuses.find((provider) => provider.instanceId === limitRecoveryInstanceId) + ?.usageLimits?.resetCredits; + const limitRecoveryResetCredit = + limitRecoveryCredits === undefined || limitRecoveryInstanceId === null + ? null + : { + environmentId, + input: { instanceId: limitRecoveryInstanceId }, + credits: limitRecoveryCredits, + }; const limitRecoveryBanner = limitRecoveryRun !== null && activeThreadShell !== null ? usageLimitRecoveryBannerItem({ @@ -2575,6 +2589,7 @@ export default function ChatView(props: ChatViewProps) { recovery: activeThreadShell.limitRecovery ?? null, snoozedUntil: activeThreadShell.snoozedUntil, timestampFormat, + resetCredit: limitRecoveryResetCredit, onChange: async (limitRecovery) => { const result = await updateThreadMetadata({ environmentId, diff --git a/apps/web/src/components/chat/UsageLimitRecoveryBanner.test.tsx b/apps/web/src/components/chat/UsageLimitRecoveryBanner.test.tsx index 84b71be9b62d9..621ec3d3e6456 100644 --- a/apps/web/src/components/chat/UsageLimitRecoveryBanner.test.tsx +++ b/apps/web/src/components/chat/UsageLimitRecoveryBanner.test.tsx @@ -17,6 +17,7 @@ function bannerDescription(resetAt: Date, timestampFormat: TimestampFormat) { snoozedUntil: null, recovery: null, timestampFormat, + resetCredit: null, onChange: async () => {}, }).description; } diff --git a/apps/web/src/components/chat/UsageLimitRecoveryBanner.tsx b/apps/web/src/components/chat/UsageLimitRecoveryBanner.tsx index 744db5ea3a1e9..fe3ab49e03fe4 100644 --- a/apps/web/src/components/chat/UsageLimitRecoveryBanner.tsx +++ b/apps/web/src/components/chat/UsageLimitRecoveryBanner.tsx @@ -1,7 +1,11 @@ import { + AuthProvidersManageScope, + type EnvironmentId, type OrchestrationV2LimitRecovery, type OrchestrationV2LimitRecoveryUpdate, + type ProviderConsumeResetCreditInput, type RunId, + type ServerProviderResetCredits, } from "@t3tools/contracts"; import { type TimestampFormat } from "@t3tools/contracts/settings"; import { deadlineStore } from "@t3tools/client-runtime/clock"; @@ -15,6 +19,8 @@ import { useState } from "react"; import { useDeadlinePassed } from "../../hooks/useDeadline"; import { OpenAI } from "../Icons"; import { ChatGptUsageButton } from "../settings/ChatGptUsageButton"; +import { readEnvironmentScope } from "../../state/session"; +import { ResetCreditDialog, useResetCredit } from "../usage/UsageLimits"; import { Button, InlineButton } from "../ui/button"; import { formatUpcomingTimestamp } from "../../timestampFormat"; import { toastManager } from "../ui/toast"; @@ -31,6 +37,12 @@ type RecoveryProps = { snoozedUntil: string | null; recovery: OrchestrationV2LimitRecovery | null; timestampFormat: TimestampFormat; + /** Present when the limited account has banked reset credits. */ + resetCredit: { + environmentId: EnvironmentId; + input: ProviderConsumeResetCreditInput; + credits: ServerProviderResetCredits; + } | null; onChange: (recovery: OrchestrationV2LimitRecoveryUpdate) => Promise; }; @@ -53,9 +65,23 @@ export function usageLimitRecoveryBannerItem(props: RecoveryProps): ComposerBann canSchedule && recovery?.runId === runId && recovery.resetAt === resetAt && recovery.autoResume; const workspaceLimit = errorCode === CHATGPT_WORKSPACE_USAGE_LIMIT_CODE; const chatGptLimit = workspaceLimit || errorCode === CHATGPT_USAGE_LIMIT_CODE; - const recoveryActions = canSchedule ? ( - - ) : null; + const resetCredit = + props.resetCredit !== null && props.resetCredit.credits.availableCount > 0 + ? props.resetCredit + : null; + const recoveryActions = + canSchedule || resetCredit !== null ? ( + + {resetCredit !== null ? ( + + ) : null} + {canSchedule ? : null} + + ) : null; const displayAt = resumes && resetAt && snoozedUntil && Date.parse(snoozedUntil) > Date.parse(resetAt) ? snoozedUntil @@ -99,6 +125,47 @@ export function usageLimitRecoveryBannerItem(props: RecoveryProps): ComposerBann }; } +function ResetCreditAction({ + environmentId, + input, +}: { + environmentId: EnvironmentId; + input: ProviderConsumeResetCreditInput; +}) { + const { canManageProviders, confirming, setConfirming, busy, redeem } = useResetCredit( + environmentId, + input, + ); + async function confirm() { + const result = await redeem(); + if (result === undefined) return; + toastManager.add({ + type: result.reset ? "success" : "error", + title: result.reset ? "Usage reset" : "Could not reset usage", + description: result.reset ? "Send a message to continue." : result.message, + }); + } + return ( + <> + { + if (readEnvironmentScope(environmentId, AuthProvidersManageScope)) setConfirming(true); + }} + > + {busy ? "Resetting..." : "Reset usage"} + + void confirm()} + disabled={!canManageProviders} + /> + + ); +} + function RecoveryActions({ runId, resetAt, recovery, snoozedUntil, onChange }: RecoveryProps) { const [pending, setPending] = useState(false); @@ -139,7 +206,7 @@ function RecoveryActions({ runId, resetAt, recovery, snoozedUntil, onChange }: R } // Inline errors would widen the actions column and crush the title. return ( - + <> void toggle("resume")}> {pending ? "Saving..." : scheduled ? "Cancel auto-resume" : "Resume at reset"} @@ -150,6 +217,6 @@ function RecoveryActions({ runId, resetAt, recovery, snoozedUntil, onChange }: R > {pending ? "Saving..." : snoozed ? "Wake now" : "Snooze until reset"} - + ); } diff --git a/apps/web/src/components/usage/UsageLimits.resetCredit.browser.test.tsx b/apps/web/src/components/usage/UsageLimits.resetCredit.browser.test.tsx new file mode 100644 index 0000000000000..7384ff62fa2f1 --- /dev/null +++ b/apps/web/src/components/usage/UsageLimits.resetCredit.browser.test.tsx @@ -0,0 +1,96 @@ +import { createRef, useImperativeHandle } from "react"; +import { EnvironmentId, ProviderInstanceId } from "@t3tools/contracts"; +import { expect, it, vi } from "vite-plus/test"; + +import { act, render } from "../../../test/render.tsx"; + +const mocks = vi.hoisted(() => ({ consume: vi.fn(), canManage: vi.fn() })); +vi.mock("../../state/session", () => ({ + readEnvironmentScope: mocks.canManage, + useEnvironmentScope: () => true, +})); +vi.mock("../../state/server", () => ({ serverEnvironment: { consumeResetCredit: {} } })); +vi.mock("../../state/use-atom-command", () => ({ useAtomCommand: () => mocks.consume })); +vi.mock("../../hooks/useSettings", () => ({ usePrimarySettings: vi.fn() })); +vi.mock("../../state/presentation", () => ({ environmentPresentations: {} })); +vi.mock("./UsageLimitsPooled", () => ({ UsageLimitsPooled: () => null })); + +import { useResetCredit } from "./UsageLimits"; + +const environmentId = EnvironmentId.make("reset-environment"); +const input = { instanceId: ProviderInstanceId.make("codex") }; +const success = { _tag: "Success", value: { outcome: "reset" } } as const; +type Result = typeof success | { readonly _tag: "Failure"; readonly cause: { error: Error } }; + +function setup() { + mocks.consume.mockReset().mockResolvedValue(success); + mocks.canManage.mockReset().mockReturnValue(true); + const ref = createRef>(); + function Probe() { + const value = useResetCredit(environmentId, input); + useImperativeHandle(ref, () => value); + return null; + } + return { + Probe, + get value() { + if (ref.current === null) throw new Error("Reset-credit hook was not mounted"); + return ref.current; + }, + }; +} + +it.each(["success", "failure", "rejection"] as const)( + "coalesces overlapping redemptions and permits a new attempt after %s", + async (outcome) => { + const fixture = setup(); + await using _screen = await render(); + let resolve!: (value: Result) => void; + let reject!: (reason: Error) => void; + const promise = new Promise((complete, fail) => { + resolve = complete; + reject = fail; + }); + const pending = { promise, resolve, reject }; + mocks.consume.mockReturnValueOnce(pending.promise); + const attempts: Array> = []; + await act(() => { + attempts.push(fixture.value.redeem(), fixture.value.redeem()); + }); + expect(mocks.consume).toHaveBeenCalledTimes(1); + expect(mocks.consume).toHaveBeenCalledWith({ environmentId, input }); + expect(await attempts[1]).toBeUndefined(); + expect(fixture.value.busy).toBe(true); + + const message = + outcome === "success" + ? "Reset applied. Your windows have cleared." + : outcome === "failure" + ? "Denied" + : "Could not use the reset credit."; + await act(async () => { + if (outcome === "rejection") pending.reject(new Error("Disconnected")); + else if (outcome === "failure") { + pending.resolve({ _tag: "Failure", cause: { error: new Error("Denied") } }); + } else pending.resolve(success); + expect(await attempts[0]).toEqual({ reset: outcome === "success", message }); + }); + expect(fixture.value.busy).toBe(false); + expect(fixture.value.status).toBe(message); + + mocks.canManage.mockReturnValue(false); + await act(async () => { + expect(await fixture.value.redeem()).toBeUndefined(); + }); + expect(mocks.consume).toHaveBeenCalledTimes(1); + mocks.canManage.mockReturnValue(true); + await act(async () => { + expect(await fixture.value.redeem()).toEqual({ + reset: true, + message: "Reset applied. Your windows have cleared.", + }); + }); + expect(mocks.consume).toHaveBeenCalledTimes(2); + expect(fixture.value.busy).toBe(false); + }, +); diff --git a/apps/web/src/components/usage/UsageLimits.tsx b/apps/web/src/components/usage/UsageLimits.tsx index 43ee023af2400..f11fbc9cbc958 100644 --- a/apps/web/src/components/usage/UsageLimits.tsx +++ b/apps/web/src/components/usage/UsageLimits.tsx @@ -19,7 +19,7 @@ import { remainingPercent, } from "@t3tools/shared/usageLimits"; import { GaugeIcon, TrendingDownIcon, TrendingUpIcon } from "lucide-react"; -import { Fragment, useLayoutEffect, useState } from "react"; +import { Fragment, useLayoutEffect, useRef, useState } from "react"; import { usePrimarySettings } from "../../hooks/useSettings"; import { observeResize } from "../../lib/observeResize"; @@ -243,26 +243,28 @@ export function useResetCredit( ) { const canManageProviders = useEnvironmentScope(environmentId, AuthProvidersManageScope); const consume = useAtomCommand(serverEnvironment.consumeResetCredit, { reportFailure: false }); + const inFlight = useRef(false); const [confirming, setConfirming] = useState(false); const [busy, setBusy] = useState(false); const [status, setStatus] = useState(null); const redeem = async () => { setConfirming(false); - if (!readEnvironmentScope(environmentId, AuthProvidersManageScope)) return; + if (inFlight.current || !readEnvironmentScope(environmentId, AuthProvidersManageScope)) return; + inFlight.current = true; setBusy(true); setStatus(null); - const result = await consume({ environmentId, input }); + const result = await consume({ environmentId, input }).catch(() => null); + inFlight.current = false; setBusy(false); - if (result._tag === "Success") { - setStatus(result.value.warning ?? OUTCOME_TEXT[result.value.outcome]); - return; - } - setStatus( - "error" in result.cause && result.cause.error instanceof Error - ? result.cause.error.message - : "Could not use the reset credit.", - ); + const message = + result?._tag === "Success" + ? (result.value.warning ?? OUTCOME_TEXT[result.value.outcome]) + : result !== null && "error" in result.cause && result.cause.error instanceof Error + ? result.cause.error.message + : "Could not use the reset credit."; + setStatus(message); + return { reset: result?._tag === "Success" && result.value.outcome === "reset", message }; }; return { canManageProviders, confirming, setConfirming, busy, status, redeem };