diff --git a/src/node/orpc/router.ts b/src/node/orpc/router.ts index f53dad4e1d..545e11da63 100644 --- a/src/node/orpc/router.ts +++ b/src/node/orpc/router.ts @@ -575,11 +575,14 @@ export const router = (authToken?: string) => { ), }, + // OAuth procedures (gateway/copilot/governor/codex) run Effect generators + // via handlerGen; the wire contracts are unchanged. Flow-starting + // mutations are uninterruptible in the services (see the respective + // startDesktopFlowEffect/startDeviceFlowEffect), so a client abort cannot + // leak a loopback server or strand a flow record; waits are interruptible + // (abandoning a wait leaves the flow's own lifecycle intact) and cancels + // run guaranteed-finalizer cleanup. muxGatewayOauth: { - // startDesktopFlow rides handlerGen; its service pipeline is - // uninterruptible (see startDesktopFlowEffect) so a client abort cannot - // leak the loopback server. waitFor/cancel stay plain handlers until the - // batch OAuth-service conversion migrates the remaining procedures. startDesktopFlow: t .input(schemas.muxGatewayOauth.startDesktopFlow.input) .output(schemas.muxGatewayOauth.startDesktopFlow.output) @@ -591,103 +594,141 @@ export const router = (authToken?: string) => { waitForDesktopFlow: t .input(schemas.muxGatewayOauth.waitForDesktopFlow.input) .output(schemas.muxGatewayOauth.waitForDesktopFlow.output) - .handler(({ context, input }) => - context.muxGatewayOauthService.waitForDesktopFlow(input.flowId, { - timeoutMs: input.timeoutMs, + .handler( + handlerGen(function* ({ context }, input) { + return yield* context.muxGatewayOauthService.waitForDesktopFlowEffect(input.flowId, { + timeoutMs: input.timeoutMs, + }); }) ), cancelDesktopFlow: t .input(schemas.muxGatewayOauth.cancelDesktopFlow.input) .output(schemas.muxGatewayOauth.cancelDesktopFlow.output) - .handler(async ({ context, input }) => { - await context.muxGatewayOauthService.cancelDesktopFlow(input.flowId); - }), + .handler( + handlerGen(function* ({ context }, input) { + yield* context.muxGatewayOauthService.cancelDesktopFlowEffect(input.flowId); + }) + ), }, copilotOauth: { startDeviceFlow: t .input(schemas.copilotOauth.startDeviceFlow.input) .output(schemas.copilotOauth.startDeviceFlow.output) - .handler(({ context }) => context.copilotOauthService.startDeviceFlow()), + .handler( + handlerGen(function* ({ context }) { + return yield* context.copilotOauthService.startDeviceFlowEffect(); + }) + ), waitForDeviceFlow: t .input(schemas.copilotOauth.waitForDeviceFlow.input) .output(schemas.copilotOauth.waitForDeviceFlow.output) - .handler(({ context, input }) => - context.copilotOauthService.waitForDeviceFlow(input.flowId, { - timeoutMs: input.timeoutMs, + .handler( + handlerGen(function* ({ context }, input) { + return yield* context.copilotOauthService.waitForDeviceFlowEffect(input.flowId, { + timeoutMs: input.timeoutMs, + }); }) ), cancelDeviceFlow: t .input(schemas.copilotOauth.cancelDeviceFlow.input) .output(schemas.copilotOauth.cancelDeviceFlow.output) - .handler(({ context, input }) => { - context.copilotOauthService.cancelDeviceFlow(input.flowId); - }), + .handler( + handlerGen(function* ({ context }, input) { + yield* context.copilotOauthService.cancelDeviceFlowEffect(input.flowId); + }) + ), }, muxGovernorOauth: { startDesktopFlow: t .input(schemas.muxGovernorOauth.startDesktopFlow.input) .output(schemas.muxGovernorOauth.startDesktopFlow.output) - .handler(({ context, input }) => - context.muxGovernorOauthService.startDesktopFlow({ - governorOrigin: input.governorOrigin, + .handler( + handlerGen(function* ({ context }, input) { + return yield* context.muxGovernorOauthService.startDesktopFlowEffect({ + governorOrigin: input.governorOrigin, + }); }) ), waitForDesktopFlow: t .input(schemas.muxGovernorOauth.waitForDesktopFlow.input) .output(schemas.muxGovernorOauth.waitForDesktopFlow.output) - .handler(({ context, input }) => - context.muxGovernorOauthService.waitForDesktopFlow(input.flowId, { - timeoutMs: input.timeoutMs, + .handler( + handlerGen(function* ({ context }, input) { + return yield* context.muxGovernorOauthService.waitForDesktopFlowEffect(input.flowId, { + timeoutMs: input.timeoutMs, + }); }) ), cancelDesktopFlow: t .input(schemas.muxGovernorOauth.cancelDesktopFlow.input) .output(schemas.muxGovernorOauth.cancelDesktopFlow.output) - .handler(async ({ context, input }) => { - await context.muxGovernorOauthService.cancelDesktopFlow(input.flowId); - }), + .handler( + handlerGen(function* ({ context }, input) { + yield* context.muxGovernorOauthService.cancelDesktopFlowEffect(input.flowId); + }) + ), }, codexOauth: { startDesktopFlow: t .input(schemas.codexOauth.startDesktopFlow.input) .output(schemas.codexOauth.startDesktopFlow.output) - .handler(({ context }) => context.codexOauthService.startDesktopFlow()), + .handler( + handlerGen(function* ({ context }) { + return yield* context.codexOauthService.startDesktopFlowEffect(); + }) + ), waitForDesktopFlow: t .input(schemas.codexOauth.waitForDesktopFlow.input) .output(schemas.codexOauth.waitForDesktopFlow.output) - .handler(({ context, input }) => - context.codexOauthService.waitForDesktopFlow(input.flowId, { - timeoutMs: input.timeoutMs, + .handler( + handlerGen(function* ({ context }, input) { + return yield* context.codexOauthService.waitForDesktopFlowEffect(input.flowId, { + timeoutMs: input.timeoutMs, + }); }) ), cancelDesktopFlow: t .input(schemas.codexOauth.cancelDesktopFlow.input) .output(schemas.codexOauth.cancelDesktopFlow.output) - .handler(async ({ context, input }) => { - await context.codexOauthService.cancelDesktopFlow(input.flowId); - }), + .handler( + handlerGen(function* ({ context }, input) { + yield* context.codexOauthService.cancelDesktopFlowEffect(input.flowId); + }) + ), startDeviceFlow: t .input(schemas.codexOauth.startDeviceFlow.input) .output(schemas.codexOauth.startDeviceFlow.output) - .handler(({ context }) => context.codexOauthService.startDeviceFlow()), + .handler( + handlerGen(function* ({ context }) { + return yield* context.codexOauthService.startDeviceFlowEffect(); + }) + ), waitForDeviceFlow: t .input(schemas.codexOauth.waitForDeviceFlow.input) .output(schemas.codexOauth.waitForDeviceFlow.output) - .handler(({ context, input }) => - context.codexOauthService.waitForDeviceFlow(input.flowId, { - timeoutMs: input.timeoutMs, + .handler( + handlerGen(function* ({ context }, input) { + return yield* context.codexOauthService.waitForDeviceFlowEffect(input.flowId, { + timeoutMs: input.timeoutMs, + }); }) ), cancelDeviceFlow: t .input(schemas.codexOauth.cancelDeviceFlow.input) .output(schemas.codexOauth.cancelDeviceFlow.output) - .handler(async ({ context, input }) => { - await context.codexOauthService.cancelDeviceFlow(input.flowId); - }), + .handler( + handlerGen(function* ({ context }, input) { + yield* context.codexOauthService.cancelDeviceFlowEffect(input.flowId); + }) + ), disconnect: t .input(schemas.codexOauth.disconnect.input) .output(schemas.codexOauth.disconnect.output) - .handler(({ context }) => context.codexOauthService.disconnect()), + .handler( + handlerGen(function* ({ context }) { + return yield* context.codexOauthService.disconnectEffect(); + }) + ), }, coderOauth: { startDesktopFlow: t diff --git a/src/node/services/codexOauthService.ts b/src/node/services/codexOauthService.ts index 353f197d99..4b03ca842c 100644 --- a/src/node/services/codexOauthService.ts +++ b/src/node/services/codexOauthService.ts @@ -1,4 +1,17 @@ +/** + * Codex (ChatGPT) OAuth service. + * + * Internals are Effect-native (see muxGatewayOauthService.ts for the shape): + * fallible pipelines are `Effect.gen` programs whose error channel carries a + * single reason-carrying tagged error, and the public Promise methods are + * thin `Effect.runPromise` facades folding back into the wire + * `Result<_, string>` shape, so pre-Effect callers keep working unchanged. + * Device-flow cancellation stays on the AbortController seam (the polling + * loop is a forked fiber, but its lifecycle is controlled through + * `finishDeviceFlow`'s abort, not fiber interruption). + */ import * as crypto from "crypto"; +import { Duration, Effect, Schema } from "effect"; import type { Result } from "@/common/types/result"; import { Err, Ok } from "@/common/types/result"; import { @@ -24,7 +37,7 @@ import { parseCodexOauthAuth, type CodexOauthAuth, } from "@/node/utils/codexOauthAuth"; -import { createDeferred } from "@/node/utils/oauthUtils"; +import { createDeferred, toWireResult } from "@/node/utils/oauthUtils"; import { startLoopbackServer } from "@/node/utils/oauthLoopbackServer"; import { OAuthFlowManager } from "@/node/utils/oauthFlowManager"; import { getErrorMessage } from "@/common/utils/errors"; @@ -96,6 +109,15 @@ function isInvalidGrantError(errorText: string): boolean { return lower.includes("invalid_grant") || lower.includes("revoked"); } +/** + * Typed failure for Codex OAuth errors. `reason` carries the exact + * user-facing string the wire `Result` contract expects, so facades map it + * 1:1 onto `Err(reason)` without reformatting. + */ +export class CodexOauthError extends Schema.TaggedError()("CodexOauthError", { + reason: Schema.String, +}) {} + export class CodexOauthService { private readonly desktopFlows = new OAuthFlowManager(); private readonly deviceFlows = new Map(); @@ -113,103 +135,207 @@ export class CodexOauthService { ) {} async disconnect(): Promise> { - // Clear stored ChatGPT OAuth tokens so Codex-only models are hidden again. - this.cachedAuth = null; - return await this.providerService.setConfigValue("openai", ["codexOauth"], undefined); + return Effect.runPromise(this.disconnectEffect()); + } + + /** + * Wire-shaped Effect surface for handlerGen router handlers. Uninterruptible + * (mirrors asAtomicMutation in providerService.ts): a client abort must not + * skip the persisted-credential clear after the in-memory cache was already + * invalidated. + */ + disconnectEffect(): Effect.Effect> { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.uninterruptible( + Effect.gen(function* () { + // Clear stored ChatGPT OAuth tokens so Codex-only models are hidden again. + self.cachedAuth = null; + // setConfigValue resolves with a wire Result; a rejection stays a + // defect, matching the previously un-caught await. + return yield* Effect.promise(() => + self.providerService.setConfigValue("openai", ["codexOauth"], undefined) + ); + }) + ); } async startDesktopFlow(): Promise> { - const flowId = randomBase64Url(); - - const codeVerifier = randomBase64Url(); - const codeChallenge = sha256Base64Url(codeVerifier); - const redirectUri = CODEX_OAUTH_BROWSER_REDIRECT_URI; - - let loopback: Awaited>; - try { - loopback = await startLoopbackServer({ - port: 1455, - host: "localhost", - callbackPath: "/auth/callback", - validateLoopback: true, - expectedState: flowId, - deferSuccessResponse: true, + return Effect.runPromise(this.startDesktopFlowEffect()); + } + + /** + * Wire-shaped Effect surface for handlerGen router handlers. Uninterruptible + * (mirrors startDesktopFlowEffect in muxGatewayOauthService.ts): a client + * abort between the loopback-server acquisition and `desktopFlows.register` + * would leak the server with nothing left to close it. Flow startup is quick + * and local, so running it to completion on abort is cheap; an abandoned + * flow still self-cleans via the registered timeout. + */ + startDesktopFlowEffect(): Effect.Effect< + Result<{ flowId: string; authorizeUrl: string }, string> + > { + return Effect.uninterruptible(toWireResult(this.launchDesktopFlowEffect())); + } + + private launchDesktopFlowEffect(): Effect.Effect< + { flowId: string; authorizeUrl: string }, + CodexOauthError + > { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + const flowId = randomBase64Url(); + + const codeVerifier = randomBase64Url(); + const codeChallenge = sha256Base64Url(codeVerifier); + const redirectUri = CODEX_OAUTH_BROWSER_REDIRECT_URI; + + const loopback = yield* Effect.tryPromise({ + try: () => + startLoopbackServer({ + port: 1455, + host: "localhost", + callbackPath: "/auth/callback", + validateLoopback: true, + expectedState: flowId, + deferSuccessResponse: true, + }), + catch: (error) => + new CodexOauthError({ + reason: `Failed to start OAuth callback listener: ${getErrorMessage(error)}`, + }), }); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Failed to start OAuth callback listener: ${message}`); - } - const resultDeferred = createDeferred>(); + const resultDeferred = createDeferred>(); + + self.desktopFlows.register(flowId, { + server: loopback.server, + resultDeferred, + // Keep server-side timeout tied to flow lifetime so abandoned flows + // (e.g. callers that never invoke waitForDesktopFlow) still self-clean. + timeoutHandle: setTimeout(() => { + Effect.runFork( + self.desktopFlows.finishEffect(flowId, Err("Timed out waiting for OAuth callback")) + ); + }, DEFAULT_DESKTOP_TIMEOUT_MS), + }); - this.desktopFlows.register(flowId, { - server: loopback.server, - resultDeferred, - // Keep server-side timeout tied to flow lifetime so abandoned flows - // (e.g. callers that never invoke waitForDesktopFlow) still self-clean. - timeoutHandle: setTimeout(() => { - void this.desktopFlows.finish(flowId, Err("Timed out waiting for OAuth callback")); - }, DEFAULT_DESKTOP_TIMEOUT_MS), - }); + const authorizeUrl = buildCodexAuthorizeUrl({ + redirectUri, + state: flowId, + codeChallenge, + }); - const authorizeUrl = buildCodexAuthorizeUrl({ - redirectUri, - state: flowId, - codeChallenge, + // Background fiber: wait for the loopback callback, exchange code for + // tokens, then finish the flow. Races against resultDeferred (which + // resolves on cancel/timeout) so the fiber exits cleanly if the flow is + // cancelled. + Effect.runFork( + self.desktopCallbackPipeline({ + flowId, + redirectUri, + codeVerifier, + loopback, + resultDeferred, + }) + ); + + log.debug(`[Codex OAuth] Desktop flow started (flowId=${flowId})`); + + return { flowId, authorizeUrl }; }); + } - // Background task: wait for the loopback callback, exchange code for tokens, - // then finish the flow. Races against resultDeferred (which resolves on - // cancel/timeout) so the task exits cleanly if the flow is cancelled. - void (async () => { - const callbackResult = await Promise.race([ - loopback.result, - resultDeferred.promise.then(() => null), - ]); + /** + * Desktop-flow completion pipeline, forked from `startDesktopFlowEffect`. + * Races the loopback callback against resultDeferred so that if the flow is + * cancelled/timed out externally, this fiber exits cleanly instead of + * dangling on loopback.result. + */ + private desktopCallbackPipeline(args: { + flowId: string; + redirectUri: string; + codeVerifier: string; + loopback: Awaited>; + resultDeferred: ReturnType>>; + }): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + const callbackResult = yield* Effect.promise(() => + Promise.race([args.loopback.result, args.resultDeferred.promise.then((): null => null)]) + ); // null means the flow was finished externally (cancel/timeout). if (!callbackResult) return; if (!callbackResult.success) { - await this.desktopFlows.finish(flowId, Err(callbackResult.error)); + yield* self.desktopFlows.finishEffect(args.flowId, Err(callbackResult.error)); return; } - const exchangeResult = await this.handleDesktopCallbackAndExchange({ - flowId, - redirectUri, - codeVerifier, - code: callbackResult.data.code, - error: null, - errorDescription: undefined, - }); + const exchangeResult: Result = yield* toWireResult( + self.handleDesktopCallbackAndExchange({ + flowId: args.flowId, + redirectUri: args.redirectUri, + codeVerifier: args.codeVerifier, + code: callbackResult.data.code, + error: null, + errorDescription: undefined, + }) + ); if (exchangeResult.success) { - loopback.sendSuccessResponse(); + args.loopback.sendSuccessResponse(); } else { - loopback.sendFailureResponse(exchangeResult.error); + args.loopback.sendFailureResponse(exchangeResult.error); } - await this.desktopFlows.finish(flowId, exchangeResult); - })(); - - log.debug(`[Codex OAuth] Desktop flow started (flowId=${flowId})`); - - return Ok({ flowId, authorizeUrl }); + yield* self.desktopFlows.finishEffect(args.flowId, exchangeResult); + }); } async waitForDesktopFlow( flowId: string, opts?: { timeoutMs?: number } ): Promise> { - return this.desktopFlows.waitFor(flowId, opts?.timeoutMs ?? DEFAULT_DESKTOP_TIMEOUT_MS); + return Effect.runPromise(this.waitForDesktopFlowEffect(flowId, opts)); + } + + /** + * Wire-shaped Effect surface for handlerGen router handlers. Left + * interruptible: abandoning the wait does not affect the flow itself (the + * shared deferred and registered timeout keep the flow's lifecycle intact). + */ + waitForDesktopFlowEffect( + flowId: string, + opts?: { timeoutMs?: number } + ): Effect.Effect> { + return this.desktopFlows.waitForEffect(flowId, opts?.timeoutMs ?? DEFAULT_DESKTOP_TIMEOUT_MS); } async cancelDesktopFlow(flowId: string): Promise { - if (this.desktopFlows.has(flowId)) { - log.debug(`[Codex OAuth] Desktop flow cancelled (flowId=${flowId})`); - } - await this.desktopFlows.cancel(flowId); + return Effect.runPromise(this.cancelDesktopFlowEffect(flowId)); + } + + /** + * Wire-shaped Effect surface for handlerGen router handlers. + * Uninterruptible: once the cancel begins, the teardown must complete — a + * client abort mid-cancel must not leave the flow registered (its callback + * could still persist credentials after the user asked to cancel). + */ + cancelDesktopFlowEffect(flowId: string): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.uninterruptible( + Effect.gen(function* () { + if (self.desktopFlows.has(flowId)) { + log.debug(`[Codex OAuth] Desktop flow cancelled (flowId=${flowId})`); + } + yield* self.desktopFlows.cancelEffect(flowId); + }) + ); } async startDeviceFlow(): Promise< @@ -223,133 +349,208 @@ export class CodexOauthService { string > > { - const flowId = randomBase64Url(); - - const deviceAuthResult = await this.requestDeviceUserCode(); - if (!deviceAuthResult.success) { - return Err(deviceAuthResult.error); - } - - const { deviceAuthId, userCode, intervalSeconds, expiresAtMs } = deviceAuthResult.data; - const verifyUrl = CODEX_OAUTH_DEVICE_VERIFY_URL; - - const { promise: resultPromise, resolve: resolveResult } = - createDeferred>(); - - const abortController = new AbortController(); - - const timeoutMs = Math.min(DEFAULT_DEVICE_TIMEOUT_MS, Math.max(0, expiresAtMs - Date.now())); - const timeout = setTimeout(() => { - void this.finishDeviceFlow(flowId, Err("Device code expired")); - }, timeoutMs); - - this.deviceFlows.set(flowId, { - flowId, - deviceAuthId, - userCode, - verifyUrl, - intervalSeconds, - expiresAtMs, - abortController, - pollingStarted: false, - timeout, - cleanupTimeout: null, - resultPromise, - resolveResult, - settled: false, - }); - - log.debug(`[Codex OAuth] Device flow started (flowId=${flowId})`); + return Effect.runPromise(this.startDeviceFlowEffect()); + } - return Ok({ flowId, userCode, verifyUrl, intervalSeconds }); + /** + * Wire-shaped Effect surface for handlerGen router handlers. Uninterruptible: + * preserves the pre-handlerGen run-to-completion semantics — a client abort + * must not allocate a device code upstream without registering the local + * flow record (and its expiry timeout) that lets callers re-attach or the + * flow self-clean. + */ + startDeviceFlowEffect(): Effect.Effect< + Result<{ flowId: string; userCode: string; verifyUrl: string; intervalSeconds: number }, string> + > { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.uninterruptible( + toWireResult( + Effect.gen(function* () { + const flowId = randomBase64Url(); + + const { deviceAuthId, userCode, intervalSeconds, expiresAtMs } = + yield* self.requestDeviceUserCode(); + const verifyUrl = CODEX_OAUTH_DEVICE_VERIFY_URL; + + const { promise: resultPromise, resolve: resolveResult } = + createDeferred>(); + + const abortController = new AbortController(); + + const timeoutMs = Math.min( + DEFAULT_DEVICE_TIMEOUT_MS, + Math.max(0, expiresAtMs - Date.now()) + ); + const timeout = setTimeout(() => { + Effect.runFork(self.finishDeviceFlowEffect(flowId, Err("Device code expired"))); + }, timeoutMs); + + self.deviceFlows.set(flowId, { + flowId, + deviceAuthId, + userCode, + verifyUrl, + intervalSeconds, + expiresAtMs, + abortController, + pollingStarted: false, + timeout, + cleanupTimeout: null, + resultPromise, + resolveResult, + settled: false, + }); + + log.debug(`[Codex OAuth] Device flow started (flowId=${flowId})`); + + return { flowId, userCode, verifyUrl, intervalSeconds }; + }) + ) + ); } async waitForDeviceFlow( flowId: string, opts?: { timeoutMs?: number } ): Promise> { - const flow = this.deviceFlows.get(flowId); - if (!flow) { - return Err("OAuth flow not found"); - } + return Effect.runPromise(this.waitForDeviceFlowEffect(flowId, opts)); + } - if (!flow.pollingStarted) { - flow.pollingStarted = true; - this.pollDeviceFlow(flowId).catch((error) => { - // The polling loop is responsible for resolving the flow; if we reach - // here something unexpected happened. - const message = getErrorMessage(error); - log.warn(`[Codex OAuth] Device polling crashed (flowId=${flowId}): ${message}`); - void this.finishDeviceFlow(flowId, Err(`Device polling crashed: ${message}`)); + /** + * Wire-shaped Effect surface for handlerGen router handlers. Left + * interruptible: the polling fiber is forked inside a single sync step (so + * an interrupt cannot mark polling started without launching it), and + * abandoning the wait leaves the shared deferred and flow timeouts intact. + */ + waitForDeviceFlowEffect( + flowId: string, + opts?: { timeoutMs?: number } + ): Effect.Effect> { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + const flow = self.deviceFlows.get(flowId); + if (!flow) { + return Err("OAuth flow not found"); + } + + yield* Effect.sync(() => { + if (flow.pollingStarted) return; + flow.pollingStarted = true; + Effect.runFork( + self.pollDeviceFlowEffect(flowId).pipe( + Effect.catchDefect((defect) => + Effect.gen(function* () { + // The polling loop is responsible for resolving the flow; if we + // reach here something unexpected happened. + const message = getErrorMessage(defect); + log.warn(`[Codex OAuth] Device polling crashed (flowId=${flowId}): ${message}`); + yield* self.finishDeviceFlowEffect( + flowId, + Err(`Device polling crashed: ${message}`) + ); + }) + ) + ) + ); }); - } - const timeoutMs = opts?.timeoutMs ?? DEFAULT_DEVICE_TIMEOUT_MS; + const timeoutMs = opts?.timeoutMs ?? DEFAULT_DEVICE_TIMEOUT_MS; + + // Effect.timeout bounds this wait call only: on timeout it interrupts + // the promise-wait fiber (the shared deferred is unaffected for other + // waiters), and its timer is cleared when the deferred wins. + const result: Result = yield* Effect.promise( + async () => flow.resultPromise + ).pipe( + Effect.timeout(Duration.millis(timeoutMs)), + Effect.catch(() => + Effect.succeed>(Err("Timed out waiting for device authorization")) + ) + ); + + if (!result.success) { + // Ensure polling is cancelled on timeout/errors. + yield* self.finishDeviceFlowEffect(flowId, result); + } - let timeoutHandle: ReturnType | null = null; - const timeoutPromise = new Promise>((resolve) => { - timeoutHandle = setTimeout(() => { - resolve(Err("Timed out waiting for device authorization")); - }, timeoutMs); + return result; }); - - const result = await Promise.race([flow.resultPromise, timeoutPromise]); - - if (timeoutHandle !== null) { - clearTimeout(timeoutHandle); - } - - if (!result.success) { - // Ensure polling is cancelled on timeout/errors. - void this.finishDeviceFlow(flowId, result); - } - - return result; } async cancelDeviceFlow(flowId: string): Promise { - const flow = this.deviceFlows.get(flowId); - if (!flow) return; + return Effect.runPromise(this.cancelDeviceFlowEffect(flowId)); + } - log.debug(`[Codex OAuth] Device flow cancelled (flowId=${flowId})`); - await this.finishDeviceFlow(flowId, Err("OAuth flow cancelled")); + /** + * Wire-shaped Effect surface for handlerGen router handlers. + * Uninterruptible: once the cancel begins, the finish bookkeeping must + * complete — a client abort mid-cancel must not leave the flow polling (it + * could still persist credentials after the user asked to cancel). + */ + cancelDeviceFlowEffect(flowId: string): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.uninterruptible( + Effect.gen(function* () { + const flow = self.deviceFlows.get(flowId); + if (!flow) return; + + log.debug(`[Codex OAuth] Device flow cancelled (flowId=${flowId})`); + yield* self.finishDeviceFlowEffect(flowId, Err("OAuth flow cancelled")); + }) + ); } async getValidAuth(): Promise> { - const stored = this.readStoredAuth(); - if (!stored) { - return Err("Codex OAuth is not configured"); - } - - if (!isCodexOauthAuthExpired(stored)) { - return Ok(stored); - } - - await using _lock = await this.refreshMutex.acquire(); - - // Re-read after acquiring lock in case another caller refreshed first. - const latest = this.readStoredAuth(); - if (!latest) { - return Err("Codex OAuth is not configured"); - } + return Effect.runPromise(this.getValidAuthEffect()); + } - if (!isCodexOauthAuthExpired(latest)) { - return Ok(latest); - } + getValidAuthEffect(): Effect.Effect> { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + const stored = self.readStoredAuth(); + if (!stored) { + return Err("Codex OAuth is not configured"); + } - const refreshed = await this.refreshTokens(latest); - if (!refreshed.success) { - return Err(refreshed.error); - } + if (!isCodexOauthAuthExpired(stored)) { + return Ok(stored); + } - return Ok(refreshed.data); + // acquireUseRelease guarantees the mutex is released on every exit path + // (refresh success/failure, defects, interruption) — the Effect + // equivalent of the pre-Effect `await using` lock. + return yield* Effect.acquireUseRelease( + Effect.promise(() => self.refreshMutex.acquire()), + () => + Effect.gen(function* () { + // Re-read after acquiring lock in case another caller refreshed first. + const latest = self.readStoredAuth(); + if (!latest) { + return Err("Codex OAuth is not configured"); + } + + if (!isCodexOauthAuthExpired(latest)) { + return Ok(latest); + } + + return yield* toWireResult(self.refreshTokens(latest)); + }), + (lock) => Effect.promise(() => lock[Symbol.asyncDispose]()) + ); + }); } async dispose(): Promise { await this.desktopFlows.shutdownAll(); const deviceIds = [...this.deviceFlows.keys()]; - await Promise.all(deviceIds.map((id) => this.finishDeviceFlow(id, Err("App shutting down")))); + for (const id of deviceIds) { + Effect.runSync(this.finishDeviceFlowEffect(id, Err("App shutting down"))); + } for (const flow of this.deviceFlows.values()) { clearTimeout(flow.timeout); @@ -372,80 +573,107 @@ export class CodexOauthService { return auth; } - private async persistAuth(auth: CodexOauthAuth): Promise> { - const result = await this.providerService.setConfigValue("openai", ["codexOauth"], auth); - // Invalidate cache so the next readStoredAuth() picks up the persisted value from disk. - // We clear rather than set because setConfigValue may have side-effects (e.g. file-write - // failures) and we want the next read to be authoritative. - this.cachedAuth = null; - return result; + private persistAuth(auth: CodexOauthAuth): Effect.Effect> { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + // setConfigValue resolves with a wire Result; a rejection stays a + // defect, matching the previously un-caught await. + const result = yield* Effect.promise(() => + self.providerService.setConfigValue("openai", ["codexOauth"], auth) + ); + // Invalidate cache so the next readStoredAuth() picks up the persisted value from disk. + // We clear rather than set because setConfigValue may have side-effects (e.g. file-write + // failures) and we want the next read to be authoritative. + self.cachedAuth = null; + return result; + }); } - private async handleDesktopCallbackAndExchange(input: { + private handleDesktopCallbackAndExchange(input: { flowId: string; redirectUri: string; codeVerifier: string; code: string | null; error: string | null; errorDescription?: string; - }): Promise> { - if (input.error) { - const message = input.errorDescription - ? `${input.error}: ${input.errorDescription}` - : input.error; - return Err(`Codex OAuth error: ${message}`); - } - - if (!input.code) { - return Err("Missing OAuth code"); - } + }): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + if (input.error) { + const message = input.errorDescription + ? `${input.error}: ${input.errorDescription}` + : input.error; + return yield* Effect.fail(new CodexOauthError({ reason: `Codex OAuth error: ${message}` })); + } - const tokenResult = await this.exchangeCodeForTokens({ - code: input.code, - redirectUri: input.redirectUri, - codeVerifier: input.codeVerifier, - }); - if (!tokenResult.success) { - return Err(tokenResult.error); - } + if (!input.code) { + return yield* Effect.fail(new CodexOauthError({ reason: "Missing OAuth code" })); + } - const persistResult = await this.persistAuth(tokenResult.data); - if (!persistResult.success) { - return Err(persistResult.error); - } + const auth = yield* self.exchangeCodeForTokens({ + code: input.code, + redirectUri: input.redirectUri, + codeVerifier: input.codeVerifier, + }); - log.debug(`[Codex OAuth] Desktop exchange completed (flowId=${input.flowId})`); + const persistResult = yield* self.persistAuth(auth); + if (!persistResult.success) { + return yield* Effect.fail(new CodexOauthError({ reason: persistResult.error })); + } - this.windowService?.focusMainWindow(); + log.debug(`[Codex OAuth] Desktop exchange completed (flowId=${input.flowId})`); - return Ok(undefined); + self.windowService?.focusMainWindow(); + }); } - private async exchangeCodeForTokens(input: { + private exchangeCodeForTokens(input: { code: string; redirectUri: string; codeVerifier: string; - }): Promise> { - try { - const response = await fetch(CODEX_OAUTH_TOKEN_URL, { - method: "POST", - headers: { "Content-Type": "application/x-www-form-urlencoded" }, - body: buildCodexTokenExchangeBody({ - code: input.code, - redirectUri: input.redirectUri, - codeVerifier: input.codeVerifier, - }), + }): Effect.Effect { + return Effect.gen(function* () { + const response = yield* Effect.tryPromise({ + // async thunk: mirrors the old `await fetch(...)`, which coerces + // non-Promise returns (e.g. a test's synchronous fetch mock). + try: async () => + fetch(CODEX_OAUTH_TOKEN_URL, { + method: "POST", + headers: { "Content-Type": "application/x-www-form-urlencoded" }, + body: buildCodexTokenExchangeBody({ + code: input.code, + redirectUri: input.redirectUri, + codeVerifier: input.codeVerifier, + }), + }), + catch: (error) => + new CodexOauthError({ + reason: `Codex OAuth exchange failed: ${getErrorMessage(error)}`, + }), }); if (!response.ok) { - const errorText = await response.text().catch(() => ""); + // Preserve the HTTP status fallback when the response body is unreadable. + const errorText = yield* Effect.promise(() => response.text().catch(() => "")); const prefix = `Codex OAuth exchange failed (${response.status})`; - return Err(errorText ? `${prefix}: ${errorText}` : prefix); + return yield* Effect.fail( + new CodexOauthError({ reason: errorText ? `${prefix}: ${errorText}` : prefix }) + ); } - const json = (await response.json()) as unknown; + const json = yield* Effect.tryPromise({ + try: async (): Promise => response.json(), + catch: (error) => + new CodexOauthError({ + reason: `Codex OAuth exchange failed: ${getErrorMessage(error)}`, + }), + }); if (!isPlainObject(json)) { - return Err("Codex OAuth exchange returned an invalid JSON payload"); + return yield* Effect.fail( + new CodexOauthError({ reason: "Codex OAuth exchange returned an invalid JSON payload" }) + ); } const accessToken = typeof json.access_token === "string" ? json.access_token : null; @@ -454,48 +682,63 @@ export class CodexOauthService { const idToken = typeof json.id_token === "string" ? json.id_token : undefined; if (!accessToken) { - return Err("Codex OAuth exchange response missing access_token"); + return yield* Effect.fail( + new CodexOauthError({ reason: "Codex OAuth exchange response missing access_token" }) + ); } if (!refreshToken) { - return Err("Codex OAuth exchange response missing refresh_token"); + return yield* Effect.fail( + new CodexOauthError({ reason: "Codex OAuth exchange response missing refresh_token" }) + ); } if (expiresIn === null) { - return Err("Codex OAuth exchange response missing expires_in"); + return yield* Effect.fail( + new CodexOauthError({ reason: "Codex OAuth exchange response missing expires_in" }) + ); } const accountId = extractAccountIdFromTokens({ accessToken, idToken }) ?? undefined; - return Ok({ + const auth: CodexOauthAuth = { type: "oauth", access: accessToken, refresh: refreshToken, expires: Date.now() + Math.max(0, Math.floor(expiresIn * 1000)), accountId, - }); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Codex OAuth exchange failed: ${message}`); - } + }; + return auth; + }); } - private async refreshTokens(current: CodexOauthAuth): Promise> { - try { - const response = await fetch(CODEX_OAUTH_TOKEN_URL, { - method: "POST", - headers: { "Content-Type": "application/x-www-form-urlencoded" }, - body: buildCodexRefreshBody({ refreshToken: current.refresh }), + private refreshTokens(current: CodexOauthAuth): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + const response = yield* Effect.tryPromise({ + // async thunk: mirrors the old `await fetch(...)`, which coerces + // non-Promise returns (e.g. a test's synchronous fetch mock). + try: async () => + fetch(CODEX_OAUTH_TOKEN_URL, { + method: "POST", + headers: { "Content-Type": "application/x-www-form-urlencoded" }, + body: buildCodexRefreshBody({ refreshToken: current.refresh }), + }), + catch: (error) => + new CodexOauthError({ + reason: `Codex OAuth refresh failed: ${getErrorMessage(error)}`, + }), }); if (!response.ok) { - const errorText = await response.text().catch(() => ""); + const errorText = yield* Effect.promise(() => response.text().catch(() => "")); // When the refresh token is invalid/revoked, clear persisted auth so subsequent // requests fall back to the existing "not connected" behavior. if (isInvalidGrantError(errorText)) { log.debug("[Codex OAuth] Refresh token rejected; clearing stored auth"); - const disconnectResult = await this.disconnect(); + const disconnectResult = yield* self.disconnectEffect(); if (!disconnectResult.success) { log.warn( `[Codex OAuth] Failed to clear stored auth after refresh failure: ${disconnectResult.error}` @@ -504,12 +747,22 @@ export class CodexOauthService { } const prefix = `Codex OAuth refresh failed (${response.status})`; - return Err(errorText ? `${prefix}: ${errorText}` : prefix); + return yield* Effect.fail( + new CodexOauthError({ reason: errorText ? `${prefix}: ${errorText}` : prefix }) + ); } - const json = (await response.json()) as unknown; + const json = yield* Effect.tryPromise({ + try: async (): Promise => response.json(), + catch: (error) => + new CodexOauthError({ + reason: `Codex OAuth refresh failed: ${getErrorMessage(error)}`, + }), + }); if (!isPlainObject(json)) { - return Err("Codex OAuth refresh returned an invalid JSON payload"); + return yield* Effect.fail( + new CodexOauthError({ reason: "Codex OAuth refresh returned an invalid JSON payload" }) + ); } const accessToken = typeof json.access_token === "string" ? json.access_token : null; @@ -518,11 +771,15 @@ export class CodexOauthService { const idToken = typeof json.id_token === "string" ? json.id_token : undefined; if (!accessToken) { - return Err("Codex OAuth refresh response missing access_token"); + return yield* Effect.fail( + new CodexOauthError({ reason: "Codex OAuth refresh response missing access_token" }) + ); } if (expiresIn === null) { - return Err("Codex OAuth refresh response missing expires_in"); + return yield* Effect.fail( + new CodexOauthError({ reason: "Codex OAuth refresh response missing expires_in" }) + ); } const accountId = extractAccountIdFromTokens({ accessToken, idToken }) ?? current.accountId; @@ -535,45 +792,70 @@ export class CodexOauthService { accountId, }; - const persistResult = await this.persistAuth(next); + const persistResult = yield* self.persistAuth(next); if (!persistResult.success) { - return Err(persistResult.error); + return yield* Effect.fail(new CodexOauthError({ reason: persistResult.error })); } - return Ok(next); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Codex OAuth refresh failed: ${message}`); - } + return next; + }).pipe( + // Mirror the pre-Effect whole-body try/catch: an unexpected throw — + // e.g. a rejected persistAuth/disconnect config write, which + // Effect.promise surfaces as a defect — must fold into the wire error + // so getValidAuth() keeps returning Err(...) instead of rejecting. + Effect.catchDefect((defect) => + Effect.fail( + new CodexOauthError({ reason: `Codex OAuth refresh failed: ${getErrorMessage(defect)}` }) + ) + ) + ); } - private async requestDeviceUserCode(): Promise< - Result< - { - deviceAuthId: string; - userCode: string; - intervalSeconds: number; - expiresAtMs: number; - }, - string - > + private requestDeviceUserCode(): Effect.Effect< + { + deviceAuthId: string; + userCode: string; + intervalSeconds: number; + expiresAtMs: number; + }, + CodexOauthError > { - try { - const response = await fetch(CODEX_OAUTH_DEVICE_USERCODE_URL, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ client_id: CODEX_OAUTH_CLIENT_ID }), + return Effect.gen(function* () { + const response = yield* Effect.tryPromise({ + // async thunk: mirrors the old `await fetch(...)` coercion (see above). + try: async () => + fetch(CODEX_OAUTH_DEVICE_USERCODE_URL, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ client_id: CODEX_OAUTH_CLIENT_ID }), + }), + catch: (error) => + new CodexOauthError({ + reason: `Codex OAuth device auth request failed: ${getErrorMessage(error)}`, + }), }); if (!response.ok) { - const errorText = await response.text().catch(() => ""); + const errorText = yield* Effect.promise(() => response.text().catch(() => "")); const prefix = `Codex OAuth device auth request failed (${response.status})`; - return Err(errorText ? `${prefix}: ${errorText}` : prefix); + return yield* Effect.fail( + new CodexOauthError({ reason: errorText ? `${prefix}: ${errorText}` : prefix }) + ); } - const json = (await response.json()) as unknown; + const json = yield* Effect.tryPromise({ + try: async (): Promise => response.json(), + catch: (error) => + new CodexOauthError({ + reason: `Codex OAuth device auth request failed: ${getErrorMessage(error)}`, + }), + }); if (!isPlainObject(json)) { - return Err("Codex OAuth device auth response returned an invalid JSON payload"); + return yield* Effect.fail( + new CodexOauthError({ + reason: "Codex OAuth device auth response returned an invalid JSON payload", + }) + ); } const deviceAuthId = typeof json.device_auth_id === "string" ? json.device_auth_id : null; @@ -582,7 +864,11 @@ export class CodexOauthService { const expiresIn = parseOptionalNumber(json.expires_in); if (!deviceAuthId || !userCode) { - return Err("Codex OAuth device auth response missing required fields"); + return yield* Effect.fail( + new CodexOauthError({ + reason: "Codex OAuth device auth response missing required fields", + }) + ); } const intervalSeconds = interval !== null ? Math.max(1, Math.floor(interval)) : 5; @@ -591,86 +877,119 @@ export class CodexOauthService { ? Date.now() + Math.max(0, Math.floor(expiresIn * 1000)) : Date.now() + DEFAULT_DEVICE_TIMEOUT_MS; - return Ok({ deviceAuthId, userCode, intervalSeconds, expiresAtMs }); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Codex OAuth device auth request failed: ${message}`); - } + return { deviceAuthId, userCode, intervalSeconds, expiresAtMs }; + }); } - private async pollDeviceFlow(flowId: string): Promise { - const flow = this.deviceFlows.get(flowId); - if (!flow || flow.settled) { - return; - } - - const intervalSeconds = flow.intervalSeconds; - - while (Date.now() < flow.expiresAtMs) { - if (flow.abortController.signal.aborted) { - await this.finishDeviceFlow(flowId, Err("OAuth flow cancelled")); + /** + * Device-token polling loop, forked from `waitForDeviceFlowEffect`. + * Cancellation flows through the flow's AbortController (aborted by + * `finishDeviceFlow`), not fiber interruption, so the loop always exits via + * its own checks. + */ + private pollDeviceFlowEffect(flowId: string): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + const flow = self.deviceFlows.get(flowId); + if (!flow || flow.settled) { return; } - const attempt = await this.pollDeviceTokenOnce(flow); - if (attempt.kind === "success") { - const persistResult = await this.persistAuth(attempt.auth); - if (!persistResult.success) { - await this.finishDeviceFlow(flowId, Err(persistResult.error)); + const intervalSeconds = flow.intervalSeconds; + + while (Date.now() < flow.expiresAtMs) { + if (flow.abortController.signal.aborted) { + yield* self.finishDeviceFlowEffect(flowId, Err("OAuth flow cancelled")); return; } - log.debug(`[Codex OAuth] Device authorization completed (flowId=${flowId})`); - this.windowService?.focusMainWindow(); - await this.finishDeviceFlow(flowId, Ok(undefined)); - return; - } + const attempt = yield* self.pollDeviceTokenOnce(flow); + if (attempt.kind === "success") { + const persistResult = yield* self.persistAuth(attempt.auth); + if (!persistResult.success) { + yield* self.finishDeviceFlowEffect(flowId, Err(persistResult.error)); + return; + } - if (attempt.kind === "fatal") { - await this.finishDeviceFlow(flowId, Err(attempt.message)); - return; - } + log.debug(`[Codex OAuth] Device authorization completed (flowId=${flowId})`); + self.windowService?.focusMainWindow(); + yield* self.finishDeviceFlowEffect(flowId, Ok(undefined)); + return; + } - try { - // OpenCode guide: intervalSeconds * 1000 + 3000 - await sleepWithAbort(intervalSeconds * 1000 + 3000, flow.abortController.signal); - } catch { - // Abort is handled via cancelDeviceFlow()/finishDeviceFlow(). - return; + if (attempt.kind === "fatal") { + yield* self.finishDeviceFlowEffect(flowId, Err(attempt.message)); + return; + } + + // OpenCode guide: intervalSeconds * 1000 + 3000. sleepWithAbort keeps + // cancellation on the AbortController seam; an abort rejection exits + // the loop like the pre-Effect try/catch did. + const slept = yield* Effect.promise(() => + sleepWithAbort(intervalSeconds * 1000 + 3000, flow.abortController.signal).then( + () => true, + () => false + ) + ); + if (!slept) { + // Abort is handled via cancelDeviceFlow()/finishDeviceFlow(). + return; + } } - } - await this.finishDeviceFlow(flowId, Err("Device code expired")); + yield* self.finishDeviceFlowEffect(flowId, Err("Device code expired")); + }); } - private async pollDeviceTokenOnce( + private pollDeviceTokenOnce( flow: DeviceFlow - ): Promise< + ): Effect.Effect< | { kind: "success"; auth: CodexOauthAuth } | { kind: "pending" } | { kind: "fatal"; message: string } > { - try { - const response = await fetch(CODEX_OAUTH_DEVICE_TOKEN_POLL_URL, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ device_auth_id: flow.deviceAuthId, user_code: flow.userCode }), - signal: flow.abortController.signal, + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + const response = yield* Effect.tryPromise({ + try: async () => + fetch(CODEX_OAUTH_DEVICE_TOKEN_POLL_URL, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ device_auth_id: flow.deviceAuthId, user_code: flow.userCode }), + signal: flow.abortController.signal, + }), + catch: (error) => + new CodexOauthError({ + // Abort is treated as cancellation. + reason: flow.abortController.signal.aborted + ? "OAuth flow cancelled" + : `Device authorization failed: ${getErrorMessage(error)}`, + }), }); if (response.status === 403 || response.status === 404) { - return { kind: "pending" }; + return { kind: "pending" as const }; } if (response.status !== 200) { - const errorText = await response.text().catch(() => ""); + const errorText = yield* Effect.promise(() => response.text().catch(() => "")); const prefix = `Codex OAuth device token poll failed (${response.status})`; - return { kind: "fatal", message: errorText ? `${prefix}: ${errorText}` : prefix }; + return { + kind: "fatal" as const, + message: errorText ? `${prefix}: ${errorText}` : prefix, + }; } - const json = (await response.json().catch(() => null)) as unknown; + const json = yield* Effect.promise( + async (): Promise => response.json().catch(() => null) + ); if (!isPlainObject(json)) { - return { kind: "fatal", message: "Codex OAuth device token poll returned invalid JSON" }; + return { + kind: "fatal" as const, + message: "Codex OAuth device token poll returned invalid JSON", + }; } const authorizationCode = @@ -679,54 +998,52 @@ export class CodexOauthService { if (!authorizationCode || !codeVerifier) { return { - kind: "fatal", + kind: "fatal" as const, message: "Codex OAuth device token poll response missing required fields", }; } - const tokenResult = await this.exchangeCodeForTokens({ + const auth = yield* self.exchangeCodeForTokens({ code: authorizationCode, redirectUri: "https://auth.openai.com/deviceauth/callback", codeVerifier, }); - if (!tokenResult.success) { - return { kind: "fatal", message: tokenResult.error }; - } - - return { kind: "success", auth: tokenResult.data }; - } catch (error) { - // Abort is treated as cancellation. - if (flow.abortController.signal.aborted) { - return { kind: "fatal", message: "OAuth flow cancelled" }; - } - - const message = getErrorMessage(error); - return { kind: "fatal", message: `Device authorization failed: ${message}` }; - } + return { kind: "success" as const, auth }; + }).pipe( + // Fold exchange/poll failures into the fatal branch (message is the + // exact wire error string, matching the pre-Effect returns). + Effect.catchTag("CodexOauthError", (error) => + Effect.succeed({ kind: "fatal" as const, message: error.reason }) + ) + ); } - private finishDeviceFlow(flowId: string, result: Result): Promise { - const flow = this.deviceFlows.get(flowId); - if (!flow || flow.settled) { - return Promise.resolve(); - } + /** Idempotent device-flow finish: all-sync bookkeeping + deferred resolve. */ + private finishDeviceFlowEffect( + flowId: string, + result: Result + ): Effect.Effect { + return Effect.sync(() => { + const flow = this.deviceFlows.get(flowId); + if (!flow || flow.settled) { + return; + } - flow.settled = true; - clearTimeout(flow.timeout); - flow.abortController.abort(); + flow.settled = true; + clearTimeout(flow.timeout); + flow.abortController.abort(); - try { - flow.resolveResult(result); - } finally { - if (flow.cleanupTimeout !== null) { - clearTimeout(flow.cleanupTimeout); + try { + flow.resolveResult(result); + } finally { + if (flow.cleanupTimeout !== null) { + clearTimeout(flow.cleanupTimeout); + } + flow.cleanupTimeout = setTimeout(() => { + this.deviceFlows.delete(flowId); + }, COMPLETED_FLOW_TTL_MS); } - flow.cleanupTimeout = setTimeout(() => { - this.deviceFlows.delete(flowId); - }, COMPLETED_FLOW_TTL_MS); - } - - return Promise.resolve(); + }); } } diff --git a/src/node/services/copilotOauthService.ts b/src/node/services/copilotOauthService.ts index e876b13486..6bb2e6a80b 100644 --- a/src/node/services/copilotOauthService.ts +++ b/src/node/services/copilotOauthService.ts @@ -1,4 +1,17 @@ +/** + * GitHub Copilot OAuth (device-flow) service. + * + * Internals are Effect-native (see muxGatewayOauthService.ts for the shape): + * fallible pipelines are `Effect.gen` programs whose error channel carries a + * single reason-carrying tagged error, and the public Promise methods are + * thin `Effect.runPromise` facades folding back into the wire + * `Result<_, string>` shape, so pre-Effect callers keep working unchanged. + * Device-flow cancellation stays on the `flow.cancelled` seam (the polling + * loop is a forked fiber, but its lifecycle is controlled through + * `finishFlow`, not fiber interruption). + */ import * as crypto from "crypto"; +import { Duration, Effect, Schema } from "effect"; import type { Result } from "@/common/types/result"; import { Err, Ok } from "@/common/types/result"; import type { ProviderService } from "@/node/services/providerService"; @@ -6,7 +19,7 @@ import type { WindowService } from "@/node/services/windowService"; import { log } from "@/node/services/log"; import { getErrorMessage } from "@/common/utils/errors"; import { COPILOT_MODEL_PREFIXES } from "@/common/utils/copilot/modelRouting"; -import { createDeferred } from "@/node/utils/oauthUtils"; +import { createDeferred, toWireResult } from "@/node/utils/oauthUtils"; const GITHUB_COPILOT_CLIENT_ID = "Ov23li8tweQw6odWQebz"; const SCOPE = "read:user"; @@ -30,6 +43,16 @@ interface DeviceFlow { resolveResult: (result: Result) => void; } +/** + * Typed failure for Copilot OAuth errors. `reason` carries the exact + * user-facing string the wire `Result` contract expects, so facades map it + * 1:1 onto `Err(reason)` without reformatting. + */ +export class CopilotOauthError extends Schema.TaggedError()( + "CopilotOauthError", + { reason: Schema.String } +) {} + export class CopilotOauthService { private readonly flows = new Map(); @@ -41,103 +64,181 @@ export class CopilotOauthService { async startDeviceFlow(): Promise< Result<{ flowId: string; verificationUri: string; userCode: string }, string> > { - const flowId = crypto.randomUUID(); - - try { - const res = await fetch(GITHUB_DEVICE_CODE_URL, { - method: "POST", - headers: { - Accept: "application/json", - "Content-Type": "application/x-www-form-urlencoded", - }, - body: new URLSearchParams({ - client_id: GITHUB_COPILOT_CLIENT_ID, - scope: SCOPE, - }), - }); - - if (!res.ok) { - const text = await res.text().catch(() => ""); - return Err(`GitHub device code request failed (${res.status}): ${text}`); - } - - const data = (await res.json()) as { - verification_uri?: string; - user_code?: string; - device_code?: string; - interval?: number; - }; - - if (!data.verification_uri || !data.user_code || !data.device_code) { - return Err("Invalid response from GitHub device code endpoint"); - } + return Effect.runPromise(this.startDeviceFlowEffect()); + } - const { promise: resultPromise, resolve: resolveResult } = - createDeferred>(); - - const timeout = setTimeout(() => { - void this.finishFlow(flowId, Err("Timed out waiting for GitHub authorization")); - }, DEFAULT_TIMEOUT_MS); - - this.flows.set(flowId, { - flowId, - deviceCode: data.device_code, - interval: data.interval ?? 5, - cancelled: false, - pollingStarted: false, - timeout, - cleanupTimeout: null, - resultPromise, - resolveResult, - }); + /** + * Wire-shaped Effect surface for handlerGen router handlers. Uninterruptible: + * preserves the pre-handlerGen run-to-completion semantics — a client abort + * must not allocate a GitHub device code without registering the local flow + * record (and its expiry timeout) that lets callers re-attach or the flow + * self-clean. + */ + startDeviceFlowEffect(): Effect.Effect< + Result<{ flowId: string; verificationUri: string; userCode: string }, string> + > { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.uninterruptible( + toWireResult( + Effect.gen(function* () { + const flowId = crypto.randomUUID(); + + const res = yield* Effect.tryPromise({ + // async thunk: mirrors the old `await fetch(...)`, which coerces + // non-Promise returns (e.g. a test's synchronous fetch mock). + try: async () => + fetch(GITHUB_DEVICE_CODE_URL, { + method: "POST", + headers: { + Accept: "application/json", + "Content-Type": "application/x-www-form-urlencoded", + }, + body: new URLSearchParams({ + client_id: GITHUB_COPILOT_CLIENT_ID, + scope: SCOPE, + }), + }), + catch: (error) => + new CopilotOauthError({ + reason: `Failed to start device flow: ${getErrorMessage(error)}`, + }), + }); + + if (!res.ok) { + const text = yield* Effect.promise(() => res.text().catch(() => "")); + return yield* Effect.fail( + new CopilotOauthError({ + reason: `GitHub device code request failed (${res.status}): ${text}`, + }) + ); + } - log.debug(`Copilot OAuth device flow started (flowId=${flowId})`); + const data = yield* Effect.tryPromise({ + try: async () => { + const json = (await res.json()) as unknown; + // Validate inside the caught region: a null/non-object JSON + // body folds into the wire error instead of a defect. + if (json === null || typeof json !== "object") { + throw new TypeError("GitHub device code response was not a JSON object"); + } + return json as { + verification_uri?: string; + user_code?: string; + device_code?: string; + interval?: number; + }; + }, + catch: (error) => + new CopilotOauthError({ + reason: `Failed to start device flow: ${getErrorMessage(error)}`, + }), + }); + + if (!data.verification_uri || !data.user_code || !data.device_code) { + return yield* Effect.fail( + new CopilotOauthError({ reason: "Invalid response from GitHub device code endpoint" }) + ); + } - return Ok({ - flowId, - verificationUri: data.verification_uri, - userCode: data.user_code, - }); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Failed to start device flow: ${message}`); - } + const { promise: resultPromise, resolve: resolveResult } = + createDeferred>(); + + const timeout = setTimeout(() => { + self.finishFlow(flowId, Err("Timed out waiting for GitHub authorization")); + }, DEFAULT_TIMEOUT_MS); + + self.flows.set(flowId, { + flowId, + deviceCode: data.device_code, + interval: data.interval ?? 5, + cancelled: false, + pollingStarted: false, + timeout, + cleanupTimeout: null, + resultPromise, + resolveResult, + }); + + log.debug(`Copilot OAuth device flow started (flowId=${flowId})`); + + return { + flowId, + verificationUri: data.verification_uri, + userCode: data.user_code, + }; + }) + ) + ); } async waitForDeviceFlow( flowId: string, opts?: { timeoutMs?: number } ): Promise> { - const flow = this.flows.get(flowId); - if (!flow) { - return Err("Device flow not found"); - } + return Effect.runPromise(this.waitForDeviceFlowEffect(flowId, opts)); + } - // Start polling in background (guard against re-entrant calls, e.g. React StrictMode re-mount) - if (!flow.pollingStarted) { - flow.pollingStarted = true; - void this.pollForToken(flow); - } + /** + * Wire-shaped Effect surface for handlerGen router handlers. Left + * interruptible: the polling fiber is forked inside a single sync step (so + * an interrupt cannot mark polling started without launching it), and + * abandoning the wait leaves the shared deferred and flow timeouts intact. + */ + waitForDeviceFlowEffect( + flowId: string, + opts?: { timeoutMs?: number } + ): Effect.Effect> { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + const flow = self.flows.get(flowId); + if (!flow) { + return Err("Device flow not found"); + } - const timeoutMs = opts?.timeoutMs ?? DEFAULT_TIMEOUT_MS; - let timeoutHandle: ReturnType | null = null; - const timeoutPromise = new Promise>((resolve) => { - timeoutHandle = setTimeout(() => { - resolve(Err("Timed out waiting for GitHub authorization")); - }, timeoutMs); - }); + // Start polling in background (guard against re-entrant calls, e.g. React StrictMode re-mount) + yield* Effect.sync(() => { + if (flow.pollingStarted) return; + flow.pollingStarted = true; + Effect.runFork(self.pollForTokenEffect(flow)); + }); - const result = await Promise.race([flow.resultPromise, timeoutPromise]); + const timeoutMs = opts?.timeoutMs ?? DEFAULT_TIMEOUT_MS; + + // Effect.timeout bounds this wait call only: on timeout it interrupts + // the promise-wait fiber (the shared deferred is unaffected for other + // waiters), and its timer is cleared when the deferred wins. + const result: Result = yield* Effect.promise( + async () => flow.resultPromise + ).pipe( + Effect.timeout(Duration.millis(timeoutMs)), + Effect.catch(() => + Effect.succeed>(Err("Timed out waiting for GitHub authorization")) + ) + ); - if (timeoutHandle !== null) { - clearTimeout(timeoutHandle); - } + if (!result.success) { + yield* Effect.sync(() => self.finishFlow(flowId, result)); + } - if (!result.success) { - void this.finishFlow(flowId, result); - } + return result; + }); + } - return result; + /** + * Wire-shaped Effect surface for handlerGen router handlers. + * Uninterruptible: once the cancel begins, the finish bookkeeping must + * complete — a client abort mid-cancel must not leave the flow polling (it + * could still persist credentials after the user asked to cancel). The + * bookkeeping itself is a single sync step. + */ + cancelDeviceFlowEffect(flowId: string): Effect.Effect { + return Effect.uninterruptible( + Effect.sync(() => { + this.cancelDeviceFlow(flowId); + }) + ); } cancelDeviceFlow(flowId: string): void { @@ -165,99 +266,178 @@ export class CopilotOauthService { this.flows.clear(); } - private async pollForToken(flow: DeviceFlow): Promise { - while (!flow.cancelled) { - try { - const res = await fetch(GITHUB_ACCESS_TOKEN_URL, { - method: "POST", - headers: { - Accept: "application/json", - "Content-Type": "application/x-www-form-urlencoded", - }, - body: new URLSearchParams({ - client_id: GITHUB_COPILOT_CLIENT_ID, - device_code: flow.deviceCode, - grant_type: "urn:ietf:params:oauth:grant-type:device_code", - }), - }); + /** + * Device-token polling loop, forked from `waitForDeviceFlowEffect`. + * Cancellation flows through `flow.cancelled` (set by `finishFlow`), not + * fiber interruption, so the loop always exits via its own checks. + */ + private pollForTokenEffect(flow: DeviceFlow): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + while (!flow.cancelled) { + const outcome = yield* self.pollOnceEffect(flow).pipe( + // Transient errors (network failures, setConfig rejections) are + // logged and retried, matching the pre-Effect try/catch loop. + Effect.catch((error) => + Effect.sync((): "continue" | "stop" => { + if (flow.cancelled) return "stop"; + log.warn(`Copilot OAuth polling error (will retry): ${getErrorMessage(error)}`); + return "continue"; + }) + ) + ); + if (outcome === "stop") return; + + // Sleep before next iteration (placed at end so the first poll happens immediately) + yield* Effect.sleep(Duration.millis(flow.interval * 1000 + POLLING_SAFETY_MARGIN_MS)); + } + }); + } - const data = (await res.json()) as { - access_token?: string; - error?: string; - interval?: number; - }; - - // Re-check cancellation after the fetch round-trip to avoid - // persisting credentials for a flow that was cancelled mid-request. - if (flow.cancelled) return; - - if (data.access_token) { - // Store token as apiKey for the github-copilot provider - const persistResult = await this.providerService.setConfig( - "github-copilot", - ["apiKey"], - data.access_token - ); - - if (!persistResult.success) { - void this.finishFlow(flow.flowId, Err(persistResult.error)); - return; + /** + * One poll of GitHub's access-token endpoint. Returns whether the loop + * should keep polling; any failure in the error channel takes the + * transient-retry path in `pollForTokenEffect`. + */ + private pollOnceEffect(flow: DeviceFlow): Effect.Effect<"continue" | "stop", unknown> { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + const data = yield* Effect.tryPromise({ + try: async () => { + const res = await fetch(GITHUB_ACCESS_TOKEN_URL, { + method: "POST", + headers: { + Accept: "application/json", + "Content-Type": "application/x-www-form-urlencoded", + }, + body: new URLSearchParams({ + client_id: GITHUB_COPILOT_CLIENT_ID, + device_code: flow.deviceCode, + grant_type: "urn:ietf:params:oauth:grant-type:device_code", + }), + }); + + const json = (await res.json()) as unknown; + // Validate inside the caught region: a null/non-object JSON body + // takes the transient-retry path (like the pre-Effect try/catch) + // instead of killing the polling fiber with a defect. + if (json === null || typeof json !== "object") { + throw new TypeError("GitHub token response was not a JSON object"); } + return json as { + access_token?: string; + error?: string; + interval?: number; + }; + }, + catch: (error) => error, + }); - // Fetch available models from Copilot API (best-effort, non-blocking on failure) - try { - const modelsRes = await fetch(`${COPILOT_API_BASE_URL}/models`, { - headers: { - Authorization: `Bearer ${data.access_token}`, - "Openai-Intent": "conversation-edits", - Accept: "application/json", - }, - }); - - if (modelsRes.ok) { - const modelsData = (await modelsRes.json()) as { - data?: Array<{ id: string }>; - }; - if (modelsData.data && modelsData.data.length > 0) { - const modelIds = modelsData.data - .map((m) => m.id) - .filter((id) => COPILOT_MODEL_PREFIXES.some((prefix) => id.startsWith(prefix))); - if (modelIds.length > 0) { - await this.providerService.setModels("github-copilot", modelIds); - } - } - } - } catch (e) { - log.debug("Failed to fetch Copilot models after login", e); - } + // Re-check cancellation after the fetch round-trip to avoid + // persisting credentials for a flow that was cancelled mid-request. + if (flow.cancelled) return "stop" as const; + + const accessToken = data.access_token; + if (accessToken) { + // Store token as apiKey for the github-copilot provider + const persistResult = yield* Effect.tryPromise({ + try: async () => + self.providerService.setConfig("github-copilot", ["apiKey"], accessToken), + catch: (error) => error, + }); - log.debug(`Copilot OAuth completed successfully (flowId=${flow.flowId})`); - this.windowService?.focusMainWindow(); - void this.finishFlow(flow.flowId, Ok(undefined)); - return; + if (!persistResult.success) { + self.finishFlow(flow.flowId, Err(persistResult.error)); + return "stop" as const; } - if (data.error === "authorization_pending") { - // Expected during normal flow — will retry after sleep below - } else if (data.error === "slow_down") { - flow.interval = data.interval ?? flow.interval + 5; - } else if (data.error) { - // Any other error - void this.finishFlow(flow.flowId, Err(`GitHub OAuth error: ${data.error}`)); - return; - } - } catch (error) { - if (flow.cancelled) return; - const message = getErrorMessage(error); - log.warn(`Copilot OAuth polling error (will retry): ${message}`); - // Transient errors — fall through to sleep, then retry + yield* self.fetchModelsAfterLoginEffect(accessToken); + + log.debug(`Copilot OAuth completed successfully (flowId=${flow.flowId})`); + self.windowService?.focusMainWindow(); + self.finishFlow(flow.flowId, Ok(undefined)); + return "stop" as const; } - // Sleep before next iteration (placed at end so the first poll happens immediately) - await new Promise((resolve) => - setTimeout(resolve, flow.interval * 1000 + POLLING_SAFETY_MARGIN_MS) - ); - } + if (data.error === "authorization_pending") { + // Expected during normal flow — will retry after sleep below + } else if (data.error === "slow_down") { + flow.interval = data.interval ?? flow.interval + 5; + } else if (data.error) { + // Any other error + self.finishFlow(flow.flowId, Err(`GitHub OAuth error: ${data.error}`)); + return "stop" as const; + } + + return "continue" as const; + }); + } + + /** + * Fetch available models from Copilot API (best-effort, non-blocking on + * failure): every failure — including a setModels rejection — is logged at + * debug level and swallowed, matching the pre-Effect inner try/catch. + */ + private fetchModelsAfterLoginEffect(accessToken: string): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + const modelsRes = yield* Effect.tryPromise({ + try: async () => + fetch(`${COPILOT_API_BASE_URL}/models`, { + headers: { + Authorization: `Bearer ${accessToken}`, + "Openai-Intent": "conversation-edits", + Accept: "application/json", + }, + }), + catch: (error) => error, + }); + + if (!modelsRes.ok) return; + + const modelsData = yield* Effect.tryPromise({ + try: async () => { + const json = (await modelsRes.json()) as unknown; + // Validate inside the caught region: a null/non-object JSON body + // must stay best-effort instead of becoming a defect. + if (json === null || typeof json !== "object") { + throw new TypeError("Copilot models response was not a JSON object"); + } + return json as { data?: Array<{ id: string }> }; + }, + catch: (error) => error, + }); + if (!Array.isArray(modelsData.data) || modelsData.data.length === 0) return; + + const modelIds = modelsData.data + .map((m) => m.id) + .filter((id) => COPILOT_MODEL_PREFIXES.some((prefix) => id.startsWith(prefix))); + if (modelIds.length === 0) return; + + yield* Effect.tryPromise({ + try: async () => self.providerService.setModels("github-copilot", modelIds), + catch: (error) => error, + }); + }).pipe( + Effect.catch((error) => + Effect.sync(() => { + log.debug("Failed to fetch Copilot models after login", error); + }) + ), + // Defensive: any unexpected throw outside the caught thunks (e.g. a + // surprising payload item shape in the map/filter above) must also stay + // best-effort — the pre-Effect code ran this whole block inside one + // try/catch, and a defect here would kill the polling fiber after the + // token was persisted but before finishFlow reports success. + Effect.catchDefect((defect) => + Effect.sync(() => { + log.debug("Failed to fetch Copilot models after login", defect); + }) + ) + ); } private finishFlow(flowId: string, result: Result): void { diff --git a/src/node/services/muxGatewayOauthService.ts b/src/node/services/muxGatewayOauthService.ts index d78168f44c..8f08bdea81 100644 --- a/src/node/services/muxGatewayOauthService.ts +++ b/src/node/services/muxGatewayOauthService.ts @@ -18,7 +18,7 @@ import * as crypto from "crypto"; import { Effect, Schema } from "effect"; import type { Result } from "@/common/types/result"; -import { Err, Ok } from "@/common/types/result"; +import { Err } from "@/common/types/result"; import { buildAuthorizeUrl, buildExchangeBody, @@ -31,7 +31,7 @@ import type { ProviderService } from "@/node/services/providerService"; import { resolveProviderCredentials } from "@/node/utils/providerRequirements"; import type { WindowService } from "@/node/services/windowService"; import { log } from "@/node/services/log"; -import { createDeferred, renderOAuthCallbackHtml } from "@/node/utils/oauthUtils"; +import { createDeferred, renderOAuthCallbackHtml, toWireResult } from "@/node/utils/oauthUtils"; import { startLoopbackServer } from "@/node/utils/oauthLoopbackServer"; import { OAuthFlowManager } from "@/node/utils/oauthFlowManager"; import { getErrorMessage } from "@/common/utils/errors"; @@ -64,18 +64,6 @@ export class MuxGatewayOAuthError extends Schema.TaggedError( - effect: Effect.Effect -): Effect.Effect> { - return effect.pipe( - Effect.map((value): Result => Ok(value)), - Effect.catchTag("MuxGatewayOAuthError", (error) => - Effect.succeed>(Err(error.reason)) - ) - ); -} - export class MuxGatewayOauthService { private readonly desktopFlows = new OAuthFlowManager(); private readonly serverFlows = new Map(); @@ -189,6 +177,13 @@ export class MuxGatewayOauthService { }), }); + // Guard before field access: a null/non-object JSON body must fold into + // the invalid-payload error instead of a fiber-killing defect. + if (json === null || typeof json !== "object") { + return yield* Effect.fail( + new MuxGatewayOAuthError({ reason: "Xum Gateway returned an invalid balance payload" }) + ); + } const payload = json as { remaining_microdollars?: unknown; ai_gateway_concurrent_requests_per_user?: unknown; @@ -363,13 +358,41 @@ export class MuxGatewayOauthService { flowId: string, opts?: { timeoutMs?: number } ): Promise> { - return this.desktopFlows.waitFor(flowId, opts?.timeoutMs ?? DEFAULT_DESKTOP_TIMEOUT_MS); + return Effect.runPromise(this.waitForDesktopFlowEffect(flowId, opts)); + } + + /** + * Wire-shaped Effect surface for handlerGen router handlers. Left + * interruptible: abandoning the wait does not affect the flow itself (the + * shared deferred and registered timeout keep the flow's lifecycle intact). + */ + waitForDesktopFlowEffect( + flowId: string, + opts?: { timeoutMs?: number } + ): Effect.Effect> { + return this.desktopFlows.waitForEffect(flowId, opts?.timeoutMs ?? DEFAULT_DESKTOP_TIMEOUT_MS); } async cancelDesktopFlow(flowId: string): Promise { - if (!this.desktopFlows.has(flowId)) return; - log.debug(`Xum Gateway OAuth desktop flow cancelled (flowId=${flowId})`); - await this.desktopFlows.cancel(flowId); + return Effect.runPromise(this.cancelDesktopFlowEffect(flowId)); + } + + /** + * Wire-shaped Effect surface for handlerGen router handlers. + * Uninterruptible: once the cancel begins, the teardown must complete — a + * client abort mid-cancel must not leave the flow registered (its callback + * could still persist credentials after the user asked to cancel). + */ + cancelDesktopFlowEffect(flowId: string): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.uninterruptible( + Effect.gen(function* () { + if (!self.desktopFlows.has(flowId)) return; + log.debug(`Xum Gateway OAuth desktop flow cancelled (flowId=${flowId})`); + yield* self.desktopFlows.cancelEffect(flowId); + }) + ); } startServerFlow(input: { redirectUri: string }): { authorizeUrl: string; state: string } { @@ -505,7 +528,15 @@ export class MuxGatewayOauthService { } const json = yield* Effect.tryPromise({ - try: async () => (await response.json()) as { access_token?: unknown }, + try: async () => { + const parsed = (await response.json()) as unknown; + // Validate inside the caught region: a null/non-object JSON body + // folds into the wire error instead of a defect. + if (parsed === null || typeof parsed !== "object") { + throw new TypeError("Response was not a JSON object"); + } + return parsed as { access_token?: unknown }; + }, catch: (error) => new MuxGatewayOAuthError({ reason: `Xum Gateway exchange failed: ${getErrorMessage(error)}`, diff --git a/src/node/services/muxGovernorOauthService.ts b/src/node/services/muxGovernorOauthService.ts index 5859ea36f5..9f86161702 100644 --- a/src/node/services/muxGovernorOauthService.ts +++ b/src/node/services/muxGovernorOauthService.ts @@ -4,9 +4,16 @@ * Similar pattern to XumGatewayOauthService but: * - Takes a user-provided governor origin (not hardcoded) * - Persists credentials to config.json (muxGovernorUrl + muxGovernorToken) + * + * Internals are Effect-native (see muxGatewayOauthService.ts for the shape): + * fallible pipelines are `Effect.gen` programs whose error channel carries a + * single reason-carrying tagged error, and the public Promise methods are + * thin `Effect.runPromise` facades folding back into the wire + * `Result<_, string>` shape, so pre-Effect callers keep working unchanged. */ import * as crypto from "crypto"; +import { Effect, Schema } from "effect"; import type { Result } from "@/common/types/result"; import { Err, Ok } from "@/common/types/result"; import { @@ -19,7 +26,7 @@ import type { Config } from "@/node/config"; import type { PolicyService } from "@/node/services/policyService"; import type { WindowService } from "@/node/services/windowService"; import { log } from "@/node/services/log"; -import { createDeferred, renderOAuthCallbackHtml } from "@/node/utils/oauthUtils"; +import { createDeferred, renderOAuthCallbackHtml, toWireResult } from "@/node/utils/oauthUtils"; import { startLoopbackServer } from "@/node/utils/oauthLoopbackServer"; import { OAuthFlowManager } from "@/node/utils/oauthFlowManager"; import { getErrorMessage } from "@/common/utils/errors"; @@ -33,6 +40,16 @@ interface ServerFlow { expiresAtMs: number; } +/** + * Typed failure for governor OAuth errors. `reason` carries the exact + * user-facing string the wire `Result` contract expects, so facades map it + * 1:1 onto `Err(reason)` without reformatting. + */ +export class MuxGovernorOAuthError extends Schema.TaggedError()( + "MuxGovernorOAuthError", + { reason: Schema.String } +) {} + export class MuxGovernorOauthService { private readonly desktopFlows = new OAuthFlowManager(); private readonly serverFlows = new Map(); @@ -46,74 +63,129 @@ export class MuxGovernorOauthService { async startDesktopFlow(input: { governorOrigin: string; }): Promise> { - // Normalize and validate the governor origin - let governorOrigin: string; - try { - governorOrigin = normalizeGovernorUrl(input.governorOrigin); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Invalid Governor URL: ${message}`); - } + return Effect.runPromise(this.startDesktopFlowEffect(input)); + } - const flowId = crypto.randomUUID(); + /** + * Wire-shaped Effect surface for handlerGen router handlers. Uninterruptible + * (mirrors startDesktopFlowEffect in muxGatewayOauthService.ts): a client + * abort between the loopback-server acquisition and `desktopFlows.register` + * would leak the server with nothing left to close it. Flow startup is quick + * and local, so running it to completion on abort is cheap; an abandoned + * flow still self-cleans via the registered timeout. + */ + startDesktopFlowEffect(input: { + governorOrigin: string; + }): Effect.Effect> { + return Effect.uninterruptible(toWireResult(this.launchDesktopFlowEffect(input))); + } - let loopback: Awaited>; - try { - loopback = await startLoopbackServer({ - expectedState: flowId, - deferSuccessResponse: true, - renderHtml: (r) => - renderOAuthCallbackHtml({ - title: r.success ? "Enrollment complete" : "Enrollment failed", - message: r.success - ? "You can return to Xum. You may now close this tab." - : (r.error ?? "Unknown error"), - success: r.success, + private launchDesktopFlowEffect(input: { + governorOrigin: string; + }): Effect.Effect< + { flowId: string; authorizeUrl: string; redirectUri: string }, + MuxGovernorOAuthError + > { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + // Normalize and validate the governor origin + const governorOrigin = yield* Effect.try({ + try: () => normalizeGovernorUrl(input.governorOrigin), + catch: (error) => + new MuxGovernorOAuthError({ reason: `Invalid Governor URL: ${getErrorMessage(error)}` }), + }); + + const flowId = crypto.randomUUID(); + + const loopback = yield* Effect.tryPromise({ + try: () => + startLoopbackServer({ + expectedState: flowId, + deferSuccessResponse: true, + renderHtml: (r) => + renderOAuthCallbackHtml({ + title: r.success ? "Enrollment complete" : "Enrollment failed", + message: r.success + ? "You can return to Xum. You may now close this tab." + : (r.error ?? "Unknown error"), + success: r.success, + }), + }), + catch: (error) => + new MuxGovernorOAuthError({ + reason: `Failed to start OAuth callback listener: ${getErrorMessage(error)}`, }), }); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Failed to start OAuth callback listener: ${message}`); - } - const authorizeUrl = buildGovernorAuthorizeUrl({ - governorOrigin, - redirectUri: loopback.redirectUri, - state: flowId, - }); + const authorizeUrl = buildGovernorAuthorizeUrl({ + governorOrigin, + redirectUri: loopback.redirectUri, + state: flowId, + }); - const resultDeferred = createDeferred>(); + const resultDeferred = createDeferred>(); + + self.desktopFlows.register(flowId, { + server: loopback.server, + resultDeferred, + // Keep server-side timeout tied to flow lifetime so abandoned flows + // (e.g. callers that never invoke waitForDesktopFlow) still self-clean. + timeoutHandle: setTimeout(() => { + Effect.runFork( + self.desktopFlows.finishEffect(flowId, Err("Timed out waiting for OAuth callback")) + ); + }, DEFAULT_DESKTOP_TIMEOUT_MS), + }); - this.desktopFlows.register(flowId, { - server: loopback.server, - resultDeferred, - // Keep server-side timeout tied to flow lifetime so abandoned flows - // (e.g. callers that never invoke waitForDesktopFlow) still self-clean. - timeoutHandle: setTimeout(() => { - void this.desktopFlows.finish(flowId, Err("Timed out waiting for OAuth callback")); - }, DEFAULT_DESKTOP_TIMEOUT_MS), + // Background fiber: await loopback callback, do token exchange, finish + // flow. Races against resultDeferred so that if the flow is cancelled/ + // timed out externally, this fiber exits cleanly instead of dangling on + // loopback.result. + Effect.runFork( + self.desktopCallbackPipeline(flowId, governorOrigin, loopback, resultDeferred) + ); + + log.debug( + `Xum Governor OAuth desktop flow started (flowId=${flowId}, origin=${governorOrigin})` + ); + + return { flowId, authorizeUrl, redirectUri: loopback.redirectUri }; }); + } - // Background task: await loopback callback, do token exchange, finish flow. - // Race against resultDeferred so that if the flow is cancelled/timed out - // externally, this task exits cleanly instead of dangling on loopback.result. - void (async () => { - const callbackOrDone = await Promise.race([ - loopback.result, - resultDeferred.promise.then((): null => null), - ]); + /** + * Desktop-flow completion pipeline, forked from `startDesktopFlowEffect`. + * Races the loopback callback against resultDeferred so that if the flow is + * cancelled/timed out externally, this fiber exits cleanly instead of + * dangling on loopback.result. + */ + private desktopCallbackPipeline( + flowId: string, + governorOrigin: string, + loopback: Awaited>, + resultDeferred: ReturnType>> + ): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + const callbackOrDone = yield* Effect.promise(() => + Promise.race([loopback.result, resultDeferred.promise.then((): null => null)]) + ); // Flow was already finished externally (timeout or cancel). if (callbackOrDone === null) return; let result: Result; if (callbackOrDone.success) { - result = await this.handleCallbackAndExchange({ - state: flowId, - governorOrigin, - code: callbackOrDone.data.code, - error: null, - }); + result = yield* toWireResult( + self.handleCallbackAndExchange({ + state: flowId, + governorOrigin, + code: callbackOrDone.data.code, + error: null, + }) + ); } else { result = Err(`Xum Governor OAuth error: ${callbackOrDone.error}`); } @@ -125,27 +197,49 @@ export class MuxGovernorOauthService { loopback.sendFailureResponse(result.error); } - await this.desktopFlows.finish(flowId, result); - })(); - - log.debug( - `Xum Governor OAuth desktop flow started (flowId=${flowId}, origin=${governorOrigin})` - ); - - return Ok({ flowId, authorizeUrl, redirectUri: loopback.redirectUri }); + yield* self.desktopFlows.finishEffect(flowId, result); + }); } async waitForDesktopFlow( flowId: string, opts?: { timeoutMs?: number } ): Promise> { - return this.desktopFlows.waitFor(flowId, opts?.timeoutMs ?? DEFAULT_DESKTOP_TIMEOUT_MS); + return Effect.runPromise(this.waitForDesktopFlowEffect(flowId, opts)); + } + + /** + * Wire-shaped Effect surface for handlerGen router handlers. Left + * interruptible: abandoning the wait does not affect the flow itself (the + * shared deferred and registered timeout keep the flow's lifecycle intact). + */ + waitForDesktopFlowEffect( + flowId: string, + opts?: { timeoutMs?: number } + ): Effect.Effect> { + return this.desktopFlows.waitForEffect(flowId, opts?.timeoutMs ?? DEFAULT_DESKTOP_TIMEOUT_MS); } async cancelDesktopFlow(flowId: string): Promise { - if (!this.desktopFlows.has(flowId)) return; - log.debug(`Xum Governor OAuth desktop flow cancelled (flowId=${flowId})`); - await this.desktopFlows.cancel(flowId); + return Effect.runPromise(this.cancelDesktopFlowEffect(flowId)); + } + + /** + * Wire-shaped Effect surface for handlerGen router handlers. + * Uninterruptible: once the cancel begins, the teardown must complete — a + * client abort mid-cancel must not leave the flow registered (its callback + * could still persist credentials after the user asked to cancel). + */ + cancelDesktopFlowEffect(flowId: string): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.uninterruptible( + Effect.gen(function* () { + if (!self.desktopFlows.has(flowId)) return; + log.debug(`Xum Governor OAuth desktop flow cancelled (flowId=${flowId})`); + yield* self.desktopFlows.cancelEffect(flowId); + }) + ); } startServerFlow(input: { @@ -194,32 +288,40 @@ export class MuxGovernorOauthService { error: string | null; errorDescription?: string; }): Promise> { - const state = input.state; - if (!state) { - return Err("Missing OAuth state"); - } - - const flow = this.serverFlows.get(state); - if (!flow) { - return Err("Unknown OAuth state"); - } - - if (Date.now() > flow.expiresAtMs) { - this.serverFlows.delete(state); - return Err("OAuth flow expired"); - } - - // Regardless of outcome, this flow should not be reused. - const governorOrigin = flow.governorOrigin; - this.serverFlows.delete(state); - - return this.handleCallbackAndExchange({ - state, - governorOrigin, - code: input.code, - error: input.error, - errorDescription: input.errorDescription, - }); + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.runPromise( + toWireResult( + Effect.gen(function* () { + const state = input.state; + if (!state) { + return yield* Effect.fail(new MuxGovernorOAuthError({ reason: "Missing OAuth state" })); + } + + const flow = self.serverFlows.get(state); + if (!flow) { + return yield* Effect.fail(new MuxGovernorOAuthError({ reason: "Unknown OAuth state" })); + } + + if (Date.now() > flow.expiresAtMs) { + self.serverFlows.delete(state); + return yield* Effect.fail(new MuxGovernorOAuthError({ reason: "OAuth flow expired" })); + } + + // Regardless of outcome, this flow should not be reused. + const governorOrigin = flow.governorOrigin; + self.serverFlows.delete(state); + + yield* self.handleCallbackAndExchange({ + state, + governorOrigin, + code: input.code, + error: input.error, + errorDescription: input.errorDescription, + }); + }) + ) + ); } async dispose(): Promise { @@ -227,85 +329,118 @@ export class MuxGovernorOauthService { this.serverFlows.clear(); } - private async handleCallbackAndExchange(input: { + private handleCallbackAndExchange(input: { state: string; governorOrigin: string; code: string | null; error: string | null; errorDescription?: string; - }): Promise> { - if (input.error) { - const message = input.errorDescription - ? `${input.error}: ${input.errorDescription}` - : input.error; - return Err(`Xum Governor OAuth error: ${message}`); - } - - if (!input.code) { - return Err("Missing OAuth code"); - } + }): Effect.Effect { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; + return Effect.gen(function* () { + if (input.error) { + const message = input.errorDescription + ? `${input.error}: ${input.errorDescription}` + : input.error; + return yield* Effect.fail( + new MuxGovernorOAuthError({ reason: `Xum Governor OAuth error: ${message}` }) + ); + } - const tokenResult = await this.exchangeCodeForToken(input.code, input.governorOrigin); - if (!tokenResult.success) { - return Err(tokenResult.error); - } + if (!input.code) { + return yield* Effect.fail(new MuxGovernorOAuthError({ reason: "Missing OAuth code" })); + } - // Persist to config.json - try { - await this.config.editConfig((config) => ({ - ...config, - muxGovernorUrl: input.governorOrigin, - muxGovernorToken: tokenResult.data, - })); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Failed to save Governor credentials: ${message}`); - } + const token = yield* self.exchangeCodeForToken(input.code, input.governorOrigin); + + // Persist to config.json + yield* Effect.tryPromise({ + try: async () => + self.config.editConfig((config) => ({ + ...config, + muxGovernorUrl: input.governorOrigin, + muxGovernorToken: token, + })), + catch: (error) => + new MuxGovernorOAuthError({ + reason: `Failed to save Governor credentials: ${getErrorMessage(error)}`, + }), + }); - log.debug(`Xum Governor OAuth exchange completed (state=${input.state})`); + log.debug(`Xum Governor OAuth exchange completed (state=${input.state})`); - this.windowService?.focusMainWindow(); + self.windowService?.focusMainWindow(); - const refreshResult = await this.policyService?.refreshNow(); - if (refreshResult && !refreshResult.success) { - log.warn("Policy refresh after Governor enrollment failed", { - error: refreshResult.error, - }); - } - return Ok(undefined); + // refreshNow resolves with a wire Result; a rejection stays a defect, + // matching the previously un-caught await. + const refreshResult = yield* Effect.promise(async () => self.policyService?.refreshNow()); + if (refreshResult && !refreshResult.success) { + log.warn("Policy refresh after Governor enrollment failed", { + error: refreshResult.error, + }); + } + }); } - private async exchangeCodeForToken( + private exchangeCodeForToken( code: string, governorOrigin: string - ): Promise> { - const exchangeUrl = buildGovernorExchangeUrl(governorOrigin); - - try { - const response = await fetch(exchangeUrl, { - method: "POST", - headers: { - "Content-Type": "application/x-www-form-urlencoded", - }, - body: buildGovernorExchangeBody({ code }), + ): Effect.Effect { + return Effect.gen(function* () { + const exchangeUrl = buildGovernorExchangeUrl(governorOrigin); + + const response = yield* Effect.tryPromise({ + // async thunk: mirrors the old `await fetch(...)`, which coerces + // non-Promise returns (e.g. a test's synchronous fetch mock). + try: async () => + fetch(exchangeUrl, { + method: "POST", + headers: { + "Content-Type": "application/x-www-form-urlencoded", + }, + body: buildGovernorExchangeBody({ code }), + }), + catch: (error) => + new MuxGovernorOAuthError({ + reason: `Xum Governor exchange failed: ${getErrorMessage(error)}`, + }), }); if (!response.ok) { - const errorText = await response.text().catch(() => ""); + // Preserve the HTTP status fallback when the response body is unreadable. + const errorText = yield* Effect.promise(() => response.text().catch(() => "")); const prefix = `Xum Governor exchange failed (${response.status})`; - return Err(errorText ? `${prefix}: ${errorText}` : prefix); + return yield* Effect.fail( + new MuxGovernorOAuthError({ reason: errorText ? `${prefix}: ${errorText}` : prefix }) + ); } - const json = (await response.json()) as { access_token?: unknown }; + const json = yield* Effect.tryPromise({ + try: async () => { + const parsed = (await response.json()) as unknown; + // Validate inside the caught region: a null/non-object JSON body + // folds into the wire error instead of a defect. + if (parsed === null || typeof parsed !== "object") { + throw new TypeError("Response was not a JSON object"); + } + return parsed as { access_token?: unknown }; + }, + catch: (error) => + new MuxGovernorOAuthError({ + reason: `Xum Governor exchange failed: ${getErrorMessage(error)}`, + }), + }); const token = typeof json.access_token === "string" ? json.access_token : null; if (!token) { - return Err("Xum Governor exchange response missing access_token"); + return yield* Effect.fail( + new MuxGovernorOAuthError({ + reason: "Xum Governor exchange response missing access_token", + }) + ); } - return Ok(token); - } catch (error) { - const message = getErrorMessage(error); - return Err(`Xum Governor exchange failed: ${message}`); - } + return token; + }); } } diff --git a/src/node/utils/oauthFlowManager.ts b/src/node/utils/oauthFlowManager.ts index ea7b674208..6cb25a8596 100644 --- a/src/node/utils/oauthFlowManager.ts +++ b/src/node/utils/oauthFlowManager.ts @@ -279,10 +279,20 @@ export class OAuthFlowManager { // up by the server close — same fire-and-forget shape as the // pre-Effect `void this.finish(...)`, but as a supervised fiber that // survives this effect's completion. - const release = self.beginFinish(flowId, result); - if (release !== null) { - yield* Effect.forkDetach(release); - } + // + // Uninterruptible: waitForEffect now runs on interruptible handler + // fibers (handlerGen), and an interrupt landing between beginFinish + // (sync unregister) and the release fork would strand a flow that is + // no longer in the map but whose scope release (timeout clear, + // deferred settle, server close) never runs. + yield* Effect.uninterruptible( + Effect.gen(function* () { + const release = self.beginFinish(flowId, result); + if (release !== null) { + yield* Effect.forkDetach(release); + } + }) + ); } return result; @@ -318,16 +328,23 @@ export class OAuthFlowManager { * * Idempotent — no-op if the flow was already removed. Mirrors the * `finishDesktopFlow` pattern. Never fails: release defects are logged. + * + * Uninterruptible: finish is the atomic teardown primitive — once it + * begins, the sync unregister and the scope release must complete together, + * even when run on an interruptible handler fiber (a client abort mid-finish + * must not strand an unregistered flow with live resources). */ finishEffect(flowId: string, result: Result): Effect.Effect { // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` const self = this; - return Effect.gen(function* () { - const release = self.beginFinish(flowId, result); - if (release !== null) { - yield* release; - } - }); + return Effect.uninterruptible( + Effect.gen(function* () { + const release = self.beginFinish(flowId, result); + if (release !== null) { + yield* release; + } + }) + ); } async finish(flowId: string, result: Result): Promise { diff --git a/src/node/utils/oauthUtils.ts b/src/node/utils/oauthUtils.ts index 7d22219f4e..ad28ca83bf 100644 --- a/src/node/utils/oauthUtils.ts +++ b/src/node/utils/oauthUtils.ts @@ -1,4 +1,7 @@ import type http from "node:http"; +import { Effect } from "effect"; +import type { Result } from "@/common/types/result"; +import { Err, Ok } from "@/common/types/result"; /** * Shared OAuth utility functions extracted from the individual OAuth service files. @@ -7,6 +10,24 @@ import type http from "node:http"; * muxGatewayOauthService, muxGovernorOauthService, and mcpOauthService. */ +/** + * Fold an Effect pipeline's typed failure channel back into the wire + * `Result<_, string>` shape shared by the OAuth services' Promise facades and + * handlerGen router procedures. The error type is constrained to carry the + * exact user-facing string in `reason`, so the fold maps it 1:1 onto + * `Err(reason)` without reformatting; failures without `reason` (e.g. a tag a + * caller must branch on first, like MuxGatewaySessionExpiredError) are + * rejected at compile time. + */ +export function toWireResult( + effect: Effect.Effect +): Effect.Effect> { + return effect.pipe( + Effect.map((value): Result => Ok(value)), + Effect.catch((error) => Effect.succeed>(Err(error.reason))) + ); +} + /** A deferred promise with an externally-accessible `resolve` handle. */ export interface Deferred { promise: Promise;