diff --git a/apps/server/src/provider/CodexAccountSwitch.test.ts b/apps/server/src/provider/CodexAccountSwitch.test.ts new file mode 100644 index 0000000000000..e72872eeaa25e --- /dev/null +++ b/apps/server/src/provider/CodexAccountSwitch.test.ts @@ -0,0 +1,88 @@ +import { assert, it } from "@effect/vitest"; +import { + ProviderDriverKind, + ProviderInstanceId, + type ServerProvider, + type ServerProviderAuth, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; + +import * as ProviderSessionManager from "../orchestration-v2/ProviderSessionManager.ts"; +import * as CodexAccountSwitch from "./CodexAccountSwitch.ts"; +import * as ProviderRegistry from "./ProviderRegistry.ts"; + +const codex = ProviderInstanceId.make("codex"); +const codexWork = ProviderInstanceId.make("codex-work"); +const claude = ProviderInstanceId.make("claude"); + +const provider = ( + instanceId: ProviderInstanceId, + auth: ServerProviderAuth, + driver = ProviderDriverKind.make("codex"), +): ServerProvider => ({ + instanceId, + driver, + enabled: true, + installed: true, + version: "1.0.0", + status: "ready", + auth, + checkedAt: "2026-11-12T00:00:00.000Z", + models: [], + slashCommands: [], + skills: [], +}); + +const signedIn = (email: string, workspaceId?: string): ServerProviderAuth => ({ + status: "authenticated", + email, + ...(workspaceId === undefined ? {} : { workspaceId }), +}); + +it.effect("closes a Codex instance's sessions only when its signed-in account changes", () => + Effect.gen(function* () { + const changes = yield* Queue.unbounded>(); + const closed = yield* Queue.unbounded(); + const claudeDriver = ProviderDriverKind.make("claude"); + yield* Layer.build( + CodexAccountSwitch.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getProviders: Effect.succeed([ + provider(codex, signedIn("alice@example.com", "ws-alice")), + provider(codexWork, signedIn("team@example.com", "ws-personal")), + ]), + streamChanges: Stream.fromQueue(changes), + }), + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ + closeInstance: (instanceId) => Queue.offer(closed, instanceId).pipe(Effect.asVoid), + }), + ), + ), + ), + ); + + yield* Queue.offerAll(changes, [ + // A token refresh rewrites auth.json without changing the account. + [provider(codex, signedIn("alice@example.com", "ws-alice"))], + [provider(codex, signedIn("alice@example.com"))], + [provider(codex, { status: "unknown" })], + [provider(claude, signedIn("one@example.com"), claudeDriver)], + [provider(claude, signedIn("two@example.com"), claudeDriver)], + [ + provider(codex, signedIn("bob@example.com")), + provider(codexWork, signedIn("team@example.com", "ws-team")), + ], + ]); + + assert.deepStrictEqual( + [yield* Queue.take(closed), yield* Queue.take(closed)], + [codex, codexWork], + ); + assert.strictEqual(yield* Queue.size(closed), 0); + }).pipe(Effect.scoped), +); diff --git a/apps/server/src/provider/CodexAccountSwitch.ts b/apps/server/src/provider/CodexAccountSwitch.ts new file mode 100644 index 0000000000000..5bb4721448e6c --- /dev/null +++ b/apps/server/src/provider/CodexAccountSwitch.ts @@ -0,0 +1,75 @@ +import { + ProviderDriverKind, + type ProviderInstanceId, + type ServerProvider, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import * as ProviderSessionManager from "../orchestration-v2/ProviderSessionManager.ts"; +import * as ProviderRegistry from "./ProviderRegistry.ts"; + +const CODEX_DRIVER = ProviderDriverKind.make("codex"); + +interface CodexAccount { + readonly email: string; + readonly workspaceId: string | undefined; +} + +const signedInAccount = (provider: ServerProvider): CodexAccount | undefined => + provider.driver === CODEX_DRIVER && + provider.auth.status === "authenticated" && + provider.auth.email !== undefined + ? { email: provider.auth.email, workspaceId: provider.auth.workspaceId } + : undefined; + +// The workspace id comes from an optional rate-limit read, so compare it only when both have one. +const isSameAccount = (previous: CodexAccount, next: CodexAccount) => + previous.email === next.email && + (previous.workspaceId === undefined || + next.workspaceId === undefined || + previous.workspaceId === next.workspaceId); + +const recordAccounts = ( + accounts: Map, + providers: ReadonlyArray, +): ReadonlyArray => { + const switched: Array = []; + for (const provider of providers) { + const next = signedInAccount(provider); + if (next === undefined) continue; + const previous = accounts.get(provider.instanceId); + if (previous !== undefined && !isSameAccount(previous, next)) { + switched.push(provider.instanceId); + } + accounts.set(provider.instanceId, { + email: next.email, + workspaceId: + next.workspaceId ?? (previous?.email === next.email ? previous.workspaceId : undefined), + }); + } + return switched; +}; + +// A Codex app-server reads auth.json once, so after `codex login` switches accounts outside T3 +// its live sessions keep the old account until they are closed. +export const layer = Layer.effectDiscard( + Effect.gen(function* () { + const providers = yield* ProviderRegistry.ProviderRegistry; + const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; + const accounts = new Map(); + recordAccounts(accounts, yield* providers.getProviders); + yield* providers.streamChanges.pipe( + Stream.runForEach((snapshot) => + Effect.forEach( + recordAccounts(accounts, snapshot), + (instanceId) => + sessions.closeInstance(instanceId).pipe(Effect.ignoreCause({ log: true })), + { discard: true }, + ), + ), + Effect.forkScoped, + ); + }), +); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 293d6670bbf0b..b4b541e73115c 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -114,6 +114,7 @@ import * as ProviderInstallationGate from "./provider/ProviderInstallationGate.t import * as ProviderInstanceRegistry from "./provider/ProviderInstanceRegistry.ts"; import * as ProviderAdapterRegistry from "./orchestration-v2/ProviderAdapterRegistry.ts"; import * as ProviderRegistry from "./provider/ProviderRegistry.ts"; +import * as CodexAccountSwitch from "./provider/CodexAccountSwitch.ts"; import * as ProviderUsageLimitsIngestion from "./provider/ProviderUsageLimitsIngestion.ts"; import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; import * as ProjectFaviconResolver from "./project/ProjectFaviconResolver.ts"; @@ -682,6 +683,7 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( ).pipe(Layer.provide(DelegatedTaskQuestionReactor.layer), Layer.provide(ProjectionStoreV2.layer)), ProviderUsageLimitsIngestion.layer, layerProviderInstallationRefresh, + CodexAccountSwitch.layer, ProviderInstallation.layer, ReplayMarkers.layer, Layer.effectDiscard(ProjectSettingsOverrideSweep.sweep),