From 412dcec3d7d30e0411a87e3d0f5d4f72ed5d7f14 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Mon, 31 Aug 2026 19:41:32 +0000 Subject: [PATCH 1/2] =?UTF-8?q?=F0=9F=A4=96=20refactor:=20convert=20provid?= =?UTF-8?q?erService=20mutation=20internals=20to=20Effect?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2b of the progressive Effect migration. Effect.gen pipelines behind thin Effect.runPromise Promise facades; provider oRPC mutations ride handlerGen. Public API and observable behavior unchanged; all existing tests pass unmodified. --- src/node/orpc/router.ts | 31 +- src/node/services/providerService.ts | 768 +++++++++++++++++---------- 2 files changed, 512 insertions(+), 287 deletions(-) diff --git a/src/node/orpc/router.ts b/src/node/orpc/router.ts index 45bb3225f2..415a50df24 100644 --- a/src/node/orpc/router.ts +++ b/src/node/orpc/router.ts @@ -485,27 +485,44 @@ export const router = (authToken?: string) => { .input(schemas.providers.getConfig.input) .output(schemas.providers.getConfig.output) .handler(({ context }) => context.providerService.getConfig()), + // Provider mutations run Effect generators via handlerGen (client aborts + // interrupt the fiber); the wire contracts are unchanged. Sync reads + // (list/getConfig) and subscriptions stay plain handlers. addCustomProvider: t .input(schemas.providers.addCustomProvider.input) .output(schemas.providers.addCustomProvider.output) - .handler(({ context, input }) => context.providerService.addCustomProvider(input)), + .handler( + handlerGen(function* ({ context }, input) { + return yield* context.providerService.addCustomProviderEffect(input); + }) + ), removeCustomProvider: t .input(schemas.providers.removeCustomProvider.input) .output(schemas.providers.removeCustomProvider.output) - .handler(({ context, input }) => - context.providerService.removeCustomProvider(input.provider) + .handler( + handlerGen(function* ({ context }, input) { + return yield* context.providerService.removeCustomProviderEffect(input.provider); + }) ), setProviderConfig: t .input(schemas.providers.setProviderConfig.input) .output(schemas.providers.setProviderConfig.output) - .handler(({ context, input }) => - context.providerService.setConfig(input.provider, input.keyPath, input.value) + .handler( + handlerGen(function* ({ context }, input) { + return yield* context.providerService.setConfigEffect( + input.provider, + input.keyPath, + input.value + ); + }) ), setModels: t .input(schemas.providers.setModels.input) .output(schemas.providers.setModels.output) - .handler(({ context, input }) => - context.providerService.setModels(input.provider, input.models) + .handler( + handlerGen(function* ({ context }, input) { + return yield* context.providerService.setModelsEffect(input.provider, input.models); + }) ), onConfigChanged: t .input(schemas.providers.onConfigChanged.input) diff --git a/src/node/services/providerService.ts b/src/node/services/providerService.ts index 586a68f660..c87bedae36 100644 --- a/src/node/services/providerService.ts +++ b/src/node/services/providerService.ts @@ -1,4 +1,24 @@ +/** + * Provider configuration service (providers.jsonc + provider-relevant main + * config state). + * + * Mutation internals are Effect-native: each fallible pipeline is an + * `Effect.gen` program and the public Promise methods are thin + * `Effect.runPromise` facades, so pre-Effect callers (OAuth services, tests) + * keep working unchanged while oRPC routes ride `handlerGen` directly (client + * aborts interrupt the fiber). The wire `Result` unions stay in the success + * channel — callers branch on `success`, not on thrown errors — and the only + * typed failure is `ProviderPersistenceError`, which carries the + * `getErrorMessage` string each facade folds into its own wire error shape + * (`persistence_failed` codes / `Failed to ...` strings), exactly like the + * old per-method try/catch blocks did. + * + * Synchronous read paths (`list`, `getConfig`, `validateRouteOverrides`) stay + * plain methods: they compose no async work, so an Effect conversion would + * add fiber overhead without composition benefit. + */ import { EventEmitter } from "events"; +import { Effect, Schema } from "effect"; import type { Config, ProjectsConfig } from "@/node/config"; import { FileLeaseManager, ProvidersConfigStore } from "@/node/config"; import { @@ -128,6 +148,21 @@ function getProviderConfigRecord(config: unknown): Record; } +/** + * Typed failure for providers.jsonc / main-config write pipelines. `message` + * carries `getErrorMessage(cause)` so each facade folds it into its wire + * error shape without reformatting. A single tag is deliberate: no caller + * branches on WHICH write failed, only on the folded wire `Result`. + */ +class ProviderPersistenceError extends Schema.TaggedError()( + "ProviderPersistenceError", + { message: Schema.String } +) {} + +function toPersistenceError(error: unknown): ProviderPersistenceError { + return new ProviderPersistenceError({ message: getErrorMessage(error) }); +} + export class ProviderService { private readonly policyService: PolicyService | null; private readonly emitter = new EventEmitter(); @@ -220,6 +255,31 @@ export class ProviderService { this.notifyConfigChanged(); } + /** + * `notifyFromMutation` with a synchronously throwing subscriber routed into + * the error channel — the pre-Effect mutation methods invoked it inside + * their try/catch, so a subscriber throw folded into the wire error. + */ + private notifyFromMutationEffect(): Effect.Effect { + return Effect.try({ try: () => this.notifyFromMutation(), catch: toPersistenceError }); + } + + /** + * Run a providers.jsonc read-modify-write callback under the cross-process + * lock (see FileLeaseManager.withProvidersFileLock). Lock acquisition + * failures and callback throws land in the error channel. + */ + private providersFileLockEffect( + callback: () => Promise | T + ): Effect.Effect { + return Effect.tryPromise({ + // async thunk: mirrors the old `await this.fileLeaseManager...`, which + // coerces non-Promise returns (e.g. a test double's synchronous lock). + try: async () => this.fileLeaseManager.withProvidersFileLock(callback), + catch: toPersistenceError, + }); + } + private listBuiltInProviders(): ProviderName[] { const providers = [...SUPPORTED_PROVIDERS]; @@ -634,41 +694,49 @@ export class ProviderService { .filter((modelId) => !allowedModels.includes(modelId)); } - public async addCustomProvider( + public addCustomProvider( input: AddCustomProviderInput ): Promise> { + return Effect.runPromise(this.addCustomProviderEffect(input)); + } + + public addCustomProviderEffect( + input: AddCustomProviderInput + ): Effect.Effect> { + // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` + const self = this; const provider = input.provider.trim(); - if (isBuiltInProvider(provider)) { - return { - success: false, - error: { - code: "built_in_provider", - message: `Provider ${provider} is built in and cannot be added as custom.`, - }, - }; - } + return Effect.gen(function* () { + if (isBuiltInProvider(provider)) { + return { + success: false as const, + error: { + code: "built_in_provider" as const, + message: `Provider ${provider} is built in and cannot be added as custom.`, + }, + }; + } - const validation = validateCustomProviderId(provider); - if (!validation.ok) { - return { - success: false, - error: addErrorReason( - { code: "invalid_provider_id", message: "Invalid custom provider id." }, - validation.reason - ), - }; - } + const validation = validateCustomProviderId(provider); + if (!validation.ok) { + return { + success: false as const, + error: addErrorReason( + { code: "invalid_provider_id" as const, message: "Invalid custom provider id." }, + validation.reason + ), + }; + } - const baseUrl = input.baseUrl.trim(); + const baseUrl = input.baseUrl.trim(); - try { // Read-modify-write under the cross-process lock so concurrent writers // (other windows, CLI processes) cannot clobber each other's saves. // The callback returns an error result to bail, or null once saved. - const lockError = await this.fileLeaseManager.withProvidersFileLock( + const lockError = yield* self.providersFileLockEffect( (): CustomProviderMutationResult | null => { const providersConfig = getProviderConfigRecord( - this.providersConfigStore.loadProvidersConfig() ?? {} + self.providersConfigStore.loadProvidersConfig() ?? {} ); if (Object.hasOwn(providersConfig, provider)) { return { @@ -691,19 +759,19 @@ export class ProviderService { }; } - if (this.policyService?.isEnforced() && !this.policyService.isProviderAllowed(provider)) { + if (self.policyService?.isEnforced() && !self.policyService.isProviderAllowed(provider)) { return { success: false, - error: this.getPolicyDeniedError(`Provider ${provider} is not allowed by policy.`), + error: self.getPolicyDeniedError(`Provider ${provider} is not allowed by policy.`), }; } - const providerPolicy = this.getProviderPolicy(provider); + const providerPolicy = self.getProviderPolicy(provider); const persistedBaseUrl = providerPolicy.forcedBaseUrl ?? baseUrl; if (providerPolicy.forcedBaseUrl && baseUrl !== providerPolicy.forcedBaseUrl) { return { success: false, - error: this.getPolicyDeniedError( + error: self.getPolicyDeniedError( `Provider ${provider} base URL is locked by policy.`, `Expected ${providerPolicy.forcedBaseUrl}.` ), @@ -711,11 +779,11 @@ export class ProviderService { } const normalizedModels = normalizeProviderModelEntries(input.models); - const disallowedModels = this.getDisallowedModelsByPolicy(provider, normalizedModels); + const disallowedModels = self.getDisallowedModelsByPolicy(provider, normalizedModels); if (disallowedModels.length > 0) { return { success: false, - error: this.getPolicyDeniedError( + error: self.getPolicyDeniedError( `One or more models are not allowed by policy: ${disallowedModels.join(", ")}` ), }; @@ -736,7 +804,7 @@ export class ProviderService { }; providersConfig[provider] = providerConfig; - this.providersConfigStore.saveProvidersConfig(providersConfig); + self.providersConfigStore.saveProvidersConfig(providersConfig); return null; } ); @@ -744,89 +812,111 @@ export class ProviderService { return lockError; } - const providerInfo = this.getConfig()[provider]; - if (!providerInfo) { - return { + // Reload + notify stay inside the guarded region: the pre-Effect + // try/catch covered them, so a throwing reload or subscriber folds + // into the persistence_failed wire error below. + return yield* Effect.try({ + try: (): CustomProviderMutationResult => { + const providerInfo = self.getConfig()[provider]; + if (!providerInfo) { + return { + success: false, + error: { + code: "persistence_failed", + message: `Provider ${provider} was saved but could not be reloaded.`, + }, + }; + } + + self.notifyFromMutation(); + return { success: true, data: providerInfo }; + }, + catch: toPersistenceError, + }); + }).pipe( + Effect.catchTag("ProviderPersistenceError", (error) => + Effect.succeed>({ success: false, - error: { - code: "persistence_failed", - message: `Provider ${provider} was saved but could not be reloaded.`, - }, - }; - } + error: addErrorReason( + { code: "persistence_failed", message: `Failed to add provider ${provider}.` }, + error.message + ), + }) + ) + ); + } - this.notifyFromMutation(); - return { success: true, data: providerInfo }; - } catch (error) { - return { - success: false, - error: addErrorReason( - { code: "persistence_failed", message: `Failed to add provider ${provider}.` }, - getErrorMessage(error) - ), - }; - } + public removeCustomProvider(providerInput: string): Promise> { + return Effect.runPromise(this.removeCustomProviderEffect(providerInput)); } - public async removeCustomProvider( + public removeCustomProviderEffect( providerInput: string - ): Promise> { - const provider = providerInput.trim(); - const providersConfig = getProviderConfigRecord( - this.providersConfigStore.loadProvidersConfig() ?? {} - ); - const providerConfig = providersConfig[provider]; - // Manual providers.jsonc edits can shadow a built-in id. Removing that entry - // restores the built-in default, so only reject bona fide built-in configs. - const isShadowedCustomProvider = - isBuiltInProvider(provider) && isCustomProviderConfig(providerConfig); - - if (isBuiltInProvider(provider) && !isShadowedCustomProvider) { - return { - success: false, - error: { - code: "built_in_provider", - message: `Provider ${provider} is built in and cannot be removed as custom.`, - }, - }; - } + ): 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 provider = providerInput.trim(); + const providersConfig = getProviderConfigRecord( + self.providersConfigStore.loadProvidersConfig() ?? {} + ); + const providerConfig = providersConfig[provider]; + // Manual providers.jsonc edits can shadow a built-in id. Removing that entry + // restores the built-in default, so only reject bona fide built-in configs. + const isShadowedCustomProvider = + isBuiltInProvider(provider) && isCustomProviderConfig(providerConfig); - if (!isShadowedCustomProvider) { - const validation = validateCustomProviderId(provider); - if (!validation.ok) { + if (isBuiltInProvider(provider) && !isShadowedCustomProvider) { return { - success: false, - error: addErrorReason( - { code: "invalid_provider_id", message: "Invalid custom provider id." }, - validation.reason - ), + success: false as const, + error: { + code: "built_in_provider" as const, + message: `Provider ${provider} is built in and cannot be removed as custom.`, + }, }; } - } - if (!Object.hasOwn(providersConfig, provider)) { - return { - success: false, - error: { code: "unknown_provider", message: `Provider ${provider} does not exist.` }, - }; - } + if (!isShadowedCustomProvider) { + const validation = validateCustomProviderId(provider); + if (!validation.ok) { + return { + success: false as const, + error: addErrorReason( + { code: "invalid_provider_id" as const, message: "Invalid custom provider id." }, + validation.reason + ), + }; + } + } - if (!isCustomProviderConfig(providersConfig[provider])) { - return { - success: false, - error: { - code: "not_custom_provider", - message: `Provider ${provider} is not a custom provider.`, - }, - }; - } + if (!Object.hasOwn(providersConfig, provider)) { + return { + success: false as const, + error: { + code: "unknown_provider" as const, + message: `Provider ${provider} does not exist.`, + }, + }; + } + + if (!isCustomProviderConfig(providersConfig[provider])) { + return { + success: false as const, + error: { + code: "not_custom_provider" as const, + message: `Provider ${provider} is not a custom provider.`, + }, + }; + } - try { // Re-validate and delete under the cross-process lock (see setConfigValue). - const lockError = await this.fileLeaseManager.withProvidersFileLock( - (): CustomProviderMutationResult | null => { + // A lock/save failure folds into persistence_failed here — narrower than + // the whole-pipeline fold because the repair step below owns a different + // wire error code. + const lockOutcome = yield* self + .providersFileLockEffect((): CustomProviderMutationResult | null => { const latestProvidersConfig = getProviderConfigRecord( - this.providersConfigStore.loadProvidersConfig() ?? {} + self.providersConfigStore.loadProvidersConfig() ?? {} ); if (!isCustomProviderConfig(latestProvidersConfig[provider])) { return { @@ -839,45 +929,59 @@ export class ProviderService { } delete latestProvidersConfig[provider]; - this.providersConfigStore.saveProvidersConfig(latestProvidersConfig); + self.providersConfigStore.saveProvidersConfig(latestProvidersConfig); return null; - } - ); - if (lockError) { - return lockError; + }) + .pipe( + Effect.catchTag("ProviderPersistenceError", (error) => + Effect.succeed | null>({ + success: false, + error: addErrorReason( + { code: "persistence_failed", message: `Failed to remove provider ${provider}.` }, + error.message + ), + }) + ) + ); + if (lockOutcome) { + return lockOutcome; } - } catch (error) { - return { - success: false, - error: addErrorReason( - { code: "persistence_failed", message: `Failed to remove provider ${provider}.` }, - getErrorMessage(error) - ), - }; - } - try { - await this.config.editConfig((config) => - this.repairRemovedCustomProviderReferences(config, provider) + const repairOutcome = yield* Effect.tryPromise({ + // async thunk: a synchronously throwing editConfig mock still lands in + // the error channel, mirroring the old `await` + try/catch. + try: async () => + self.config.editConfig((config) => + self.repairRemovedCustomProviderReferences(config, provider) + ), + catch: toPersistenceError, + }).pipe( + Effect.map((): CustomProviderMutationResult | null => null), + Effect.catchTag("ProviderPersistenceError", (error) => + Effect.sync((): CustomProviderMutationResult | null => { + // The provider is already deleted from providers.jsonc. Notify subscribers so they + // re-sync even when durable model reference cleanup needs another attempt. + self.notifyFromMutation(); + return { + success: false, + error: addErrorReason( + { + code: "config_repair_failed", + message: `Provider ${provider} was removed, but saved model references could not be repaired.`, + }, + error.message + ), + }; + }) + ) ); - } catch (error) { - // The provider is already deleted from providers.jsonc. Notify subscribers so they - // re-sync even when durable model reference cleanup needs another attempt. - this.notifyFromMutation(); - return { - success: false, - error: addErrorReason( - { - code: "config_repair_failed", - message: `Provider ${provider} was removed, but saved model references could not be repaired.`, - }, - getErrorMessage(error) - ), - }; - } + if (repairOutcome) { + return repairOutcome; + } - this.notifyFromMutation(); - return { success: true, data: undefined }; + self.notifyFromMutation(); + return { success: true as const, data: undefined }; + }); } private repairRemovedCustomProviderReferences( @@ -993,48 +1097,61 @@ export class ProviderService { /** * Set custom models for a provider */ - public async setModels( + public setModels(provider: string, models: ProviderModelEntry[]): Promise> { + return Effect.runPromise(this.setModelsEffect(provider, models)); + } + + public setModelsEffect( provider: string, models: ProviderModelEntry[] - ): Promise> { - try { - const normalizedModels = normalizeProviderModelEntries(models); + ): 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 normalizedModels = yield* Effect.try({ + try: () => normalizeProviderModelEntries(models), + catch: toPersistenceError, + }); // Read-modify-write under the cross-process lock (see setConfigValue). // The callback returns a policy denial to bail, or null once saved. - const policyDenial = await this.fileLeaseManager.withProvidersFileLock((): string | null => { - const denial = this.validateModelsEditPolicy(provider, normalizedModels); + const policyDenial = yield* self.providersFileLockEffect((): string | null => { + const denial = self.validateModelsEditPolicy(provider, normalizedModels); if (denial != null) { return denial; } - const providersConfig = this.providersConfigStore.loadProvidersConfig() ?? {}; + const providersConfig = self.providersConfigStore.loadProvidersConfig() ?? {}; if (!providersConfig[provider]) { providersConfig[provider] = {}; } if (provider === "coder") { - this.applyCoderModelEdit( + self.applyCoderModelEdit( providersConfig.coder as Record, normalizedModels ); } else { providersConfig[provider].models = normalizedModels; } - this.providersConfigStore.saveProvidersConfig(providersConfig); + self.providersConfigStore.saveProvidersConfig(providersConfig); return null; }); if (policyDenial != null) { - return { success: false, error: policyDenial }; + return { success: false as const, error: policyDenial }; } - this.notifyFromMutation(); + yield* self.notifyFromMutationEffect(); - return { success: true, data: undefined }; - } catch (error) { - const message = getErrorMessage(error); - return { success: false, error: `Failed to set models: ${message}` }; - } + return { success: true as const, data: undefined }; + }).pipe( + Effect.catchTag("ProviderPersistenceError", (error) => + Effect.succeed>({ + success: false, + error: `Failed to set models: ${error.message}`, + }) + ) + ); } /** @@ -1107,58 +1224,83 @@ export class ProviderService { * Gateways that are explicitly disabled or fully deconfigured are removed; * configured-but-not-auto-eligible gateways keep any manual route. */ - private async syncGatewayLifecycle(provider: string): Promise { - if (!(provider in PROVIDER_DEFINITIONS)) return; - const providerName = provider as ProviderName; - const def = PROVIDER_DEFINITIONS[providerName]; - if (def.kind !== "gateway") return; + private syncGatewayLifecycleEffect( + provider: 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* () { + if (!(provider in PROVIDER_DEFINITIONS)) return; + const providerName = provider as ProviderName; + const def = PROVIDER_DEFINITIONS[providerName]; + if (def.kind !== "gateway") return; + + // Everything up to the branch decision is synchronous config/policy + // reading; one Effect.try keeps a thrown read in the error channel + // exactly like the old `await this.syncGatewayLifecycle(...)` inside + // the callers' try/catch. It returns the main-config edit to apply, + // or null when routePriority already matches. + const edit = yield* Effect.try({ + try: (): ((c: ProjectsConfig) => ProjectsConfig) | null => { + const providersConfig = self.providersConfigStore.loadProvidersConfig() ?? {}; + const rawProviderConfig = providersConfig[providerName] ?? {}; + // Coder credentials are issuer-bound and its deploymentUrl field stays + // editable under an enforced forcedBaseUrl: lifecycle checks must resolve + // against the forced URL (mirroring getConfig and routing), or editing + // the unlocked field would evict coder from routePriority while Settings + // and runtime model creation stay connected to the forced deployment. + const forcedBaseUrl = self.policyService?.isEnforced() + ? self.policyService.getForcedBaseUrl(providerName) + : undefined; + const providerConfig = + providerName === "coder" && forcedBaseUrl !== undefined + ? { ...rawProviderConfig, deploymentUrl: forcedBaseUrl } + : rawProviderConfig; + const isAutoRouteEligible = isProviderAutoRouteEligible(providerName, providerConfig); + const config = self.config.loadConfigOrDefault(); + const priority = config.routePriority ?? ["direct"]; + + if (isAutoRouteEligible && !priority.includes(providerName)) { + // Insert before "direct" to stay reachable while preserving the + // relative order of any user-configured routes already present. + const directIndex = priority.indexOf("direct"); + const insertIndex = directIndex === -1 ? priority.length : directIndex; + const nextPriority = [...priority]; + nextPriority.splice(insertIndex, 0, providerName); + return (c) => ({ + ...c, + routePriority: nextPriority, + // Clear legacy disable — routePriority presence is now the authoritative + // routing signal, so a stale muxGatewayEnabled: false must not veto it. + ...(providerName === "mux-gateway" ? { muxGatewayEnabled: undefined } : {}), + }); + } else if (!isAutoRouteEligible && priority.includes(providerName)) { + // Only remove a gateway from routePriority when it is truly deconfigured + // or explicitly disabled. Configured-but-not-auto-eligible providers + // (e.g., Bedrock with IAM role auth that has no observable credentials) + // should keep any manually added route. + const credentials = resolveProviderCredentials(providerName, providerConfig); + const shouldRemove = + !credentials.isConfigured || isProviderDisabledInConfig(providerConfig); + if (shouldRemove) { + return (c) => ({ + ...c, + routePriority: priority.filter((p) => p !== providerName), + }); + } + } + return null; + }, + catch: toPersistenceError, + }); - const providersConfig = this.providersConfigStore.loadProvidersConfig() ?? {}; - const rawProviderConfig = providersConfig[providerName] ?? {}; - // Coder credentials are issuer-bound and its deploymentUrl field stays - // editable under an enforced forcedBaseUrl: lifecycle checks must resolve - // against the forced URL (mirroring getConfig and routing), or editing - // the unlocked field would evict coder from routePriority while Settings - // and runtime model creation stay connected to the forced deployment. - const forcedBaseUrl = this.policyService?.isEnforced() - ? this.policyService.getForcedBaseUrl(providerName) - : undefined; - const providerConfig = - providerName === "coder" && forcedBaseUrl !== undefined - ? { ...rawProviderConfig, deploymentUrl: forcedBaseUrl } - : rawProviderConfig; - const isAutoRouteEligible = isProviderAutoRouteEligible(providerName, providerConfig); - const config = this.config.loadConfigOrDefault(); - const priority = config.routePriority ?? ["direct"]; - - if (isAutoRouteEligible && !priority.includes(providerName)) { - // Insert before "direct" to stay reachable while preserving the - // relative order of any user-configured routes already present. - const directIndex = priority.indexOf("direct"); - const insertIndex = directIndex === -1 ? priority.length : directIndex; - const nextPriority = [...priority]; - nextPriority.splice(insertIndex, 0, providerName); - await this.config.editConfig((c) => ({ - ...c, - routePriority: nextPriority, - // Clear legacy disable — routePriority presence is now the authoritative - // routing signal, so a stale muxGatewayEnabled: false must not veto it. - ...(providerName === "mux-gateway" ? { muxGatewayEnabled: undefined } : {}), - })); - } else if (!isAutoRouteEligible && priority.includes(providerName)) { - // Only remove a gateway from routePriority when it is truly deconfigured - // or explicitly disabled. Configured-but-not-auto-eligible providers - // (e.g., Bedrock with IAM role auth that has no observable credentials) - // should keep any manually added route. - const credentials = resolveProviderCredentials(providerName, providerConfig); - const shouldRemove = !credentials.isConfigured || isProviderDisabledInConfig(providerConfig); - if (shouldRemove) { - await this.config.editConfig((c) => ({ - ...c, - routePriority: priority.filter((p) => p !== providerName), - })); + if (edit !== null) { + yield* Effect.tryPromise({ + try: async () => self.config.editConfig(edit), + catch: toPersistenceError, + }); } - } + }); } /** @@ -1191,29 +1333,39 @@ export class ProviderService { * Intended for persisted auth blobs (e.g. Codex OAuth tokens) that should never * cross the frontend boundary. */ - public async setConfigValue( + public setConfigValue( provider: string, keyPath: string[], value: unknown ): Promise> { - const deniedSegment = keyPath.find((segment) => DENIED_KEY_PATH_SEGMENTS.has(segment)); - if (deniedSegment) { - // Match the agentic config mutation path so legacy ORPC callers cannot write into prototypes. - return { success: false, error: `Denied key path segment: "${deniedSegment}"` }; - } + return Effect.runPromise(this.setConfigValueEffect(provider, keyPath, value)); + } + + private setConfigValueEffect( + provider: string, + keyPath: string[], + value: unknown + ): 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 deniedSegment = keyPath.find((segment) => DENIED_KEY_PATH_SEGMENTS.has(segment)); + if (deniedSegment) { + // Match the agentic config mutation path so legacy ORPC callers cannot write into prototypes. + return { success: false as const, error: `Denied key path segment: "${deniedSegment}"` }; + } - try { // Read-modify-write under the cross-process lock: every providers.jsonc // writer must cooperate or a whole-file save from one process could // resurrect credentials another process just rotated/cleared. // The callback returns a policy denial to bail, or null once saved. - const policyDenial = await this.fileLeaseManager.withProvidersFileLock((): string | null => { - const denial = this.validateProviderEditPolicy(provider, keyPath); + const policyDenial = yield* self.providersFileLockEffect((): string | null => { + const denial = self.validateProviderEditPolicy(provider, keyPath); if (denial != null) { return denial; } - const providersConfig = this.providersConfigStore.loadProvidersConfig() ?? {}; + const providersConfig = self.providersConfigStore.loadProvidersConfig() ?? {}; // Ensure provider exists if (!providersConfig[provider]) { @@ -1249,20 +1401,24 @@ export class ProviderService { } // Save updated config - this.providersConfigStore.saveProvidersConfig(providersConfig); + self.providersConfigStore.saveProvidersConfig(providersConfig); return null; }); if (policyDenial != null) { - return { success: false, error: policyDenial }; + return { success: false as const, error: policyDenial }; } - this.notifyFromMutation(); - await this.syncGatewayLifecycle(provider); + yield* self.notifyFromMutationEffect(); + yield* self.syncGatewayLifecycleEffect(provider); - return { success: true, data: undefined }; - } catch (error) { - const message = getErrorMessage(error); - return { success: false, error: `Failed to set provider config: ${message}` }; - } + return { success: true as const, data: undefined }; + }).pipe( + Effect.catchTag("ProviderPersistenceError", (error) => + Effect.succeed>({ + success: false, + error: `Failed to set provider config: ${error.message}`, + }) + ) + ); } /** @@ -1280,24 +1436,37 @@ export class ProviderService { * credential-management primitive (clearing dead tokens, persisting * rotations), not a user-driven config edit. */ - public async updateConfigValue( + public updateConfigValue( provider: string, keyPath: string[], update: (current: unknown) => { value: unknown } | null ): Promise> { - const deniedSegment = keyPath.find((segment) => DENIED_KEY_PATH_SEGMENTS.has(segment)); - if (deniedSegment) { - return { success: false, error: `Denied key path segment: "${deniedSegment}"` }; - } - if (keyPath.length === 0) { - return { success: false, error: "updateConfigValue requires a non-empty key path" }; - } + return Effect.runPromise(this.updateConfigValueEffect(provider, keyPath, update)); + } - try { - const applied = await this.fileLeaseManager.withProvidersFileLock(() => { + private updateConfigValueEffect( + provider: string, + keyPath: string[], + update: (current: unknown) => { value: unknown } | null + ): 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 deniedSegment = keyPath.find((segment) => DENIED_KEY_PATH_SEGMENTS.has(segment)); + if (deniedSegment) { + return { success: false as const, error: `Denied key path segment: "${deniedSegment}"` }; + } + if (keyPath.length === 0) { + return { + success: false as const, + error: "updateConfigValue requires a non-empty key path", + }; + } + + const applied = yield* self.providersFileLockEffect(() => { // Load, decide, and write under the lock — no awaits in between, so // the predicate result cannot be invalidated by any cooperating writer. - const providersConfig = this.providersConfigStore.loadProvidersConfig() ?? {}; + const providersConfig = self.providersConfigStore.loadProvidersConfig() ?? {}; let current: unknown = providersConfig[provider]; for (const key of keyPath) { current = @@ -1329,16 +1498,20 @@ export class ProviderService { target[lastKey] = decision.value; } - this.providersConfigStore.saveProvidersConfig(providersConfig); + self.providersConfigStore.saveProvidersConfig(providersConfig); return true; }); - await this.afterAppliedMutation(provider, applied); - return { success: true, data: { applied } }; - } catch (error) { - const message = getErrorMessage(error); - return { success: false, error: `Failed to update provider config: ${message}` }; - } + yield* self.afterAppliedMutationEffect(provider, applied); + return { success: true as const, data: { applied } }; + }).pipe( + Effect.catchTag("ProviderPersistenceError", (error) => + Effect.succeed>({ + success: false, + error: `Failed to update provider config: ${error.message}`, + }) + ) + ); } /** @@ -1350,16 +1523,22 @@ export class ProviderService { * any non-success result, which would strand a credential that IS stored * (and reported as connected) in a revoked state. */ - private async afterAppliedMutation(provider: string, applied: boolean): Promise { - if (!applied) { - return; - } - try { - this.notifyFromMutation(); - await this.syncGatewayLifecycle(provider); - } catch (error) { - log.error(`Post-write route sync failed for provider ${provider}:`, error); - } + private afterAppliedMutationEffect(provider: string, applied: boolean): 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 (!applied) { + return; + } + yield* self.notifyFromMutationEffect(); + yield* self.syncGatewayLifecycleEffect(provider); + }).pipe( + Effect.catch((error) => + Effect.sync(() => { + log.error(`Post-write route sync failed for provider ${provider}:`, error); + }) + ) + ); } /** @@ -1374,15 +1553,26 @@ export class ProviderService { * Internal credential-management primitive: skips policy gating like * updateConfigValue. */ - public async updateProviderSection( + public updateProviderSection( provider: string, update: ( section: Record | undefined ) => { value: Record } | null ): Promise> { - try { - const applied = await this.fileLeaseManager.withProvidersFileLock(() => { - const providersConfig = this.providersConfigStore.loadProvidersConfig() ?? {}; + return Effect.runPromise(this.updateProviderSectionEffect(provider, update)); + } + + private updateProviderSectionEffect( + provider: string, + update: ( + section: Record | undefined + ) => { value: Record } | null + ): 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 applied = yield* self.providersFileLockEffect(() => { + const providersConfig = self.providersConfigStore.loadProvidersConfig() ?? {}; const section = providersConfig[provider] as Record | undefined; const decision = update(section); @@ -1398,35 +1588,49 @@ export class ProviderService { } providersConfig[provider] = decision.value as BaseProviderConfig; - this.providersConfigStore.saveProvidersConfig(providersConfig); + self.providersConfigStore.saveProvidersConfig(providersConfig); return true; }); // Best-effort: a landed write must not be reported as failed (see - // afterAppliedMutation). - await this.afterAppliedMutation(provider, applied); - return { success: true, data: { applied } }; - } catch (error) { - const message = getErrorMessage(error); - return { success: false, error: `Failed to update provider config: ${message}` }; - } + // afterAppliedMutationEffect). + yield* self.afterAppliedMutationEffect(provider, applied); + return { success: true as const, data: { applied } }; + }).pipe( + Effect.catchTag("ProviderPersistenceError", (error) => + Effect.succeed>({ + success: false, + error: `Failed to update provider config: ${error.message}`, + }) + ) + ); } - public async setConfig( + public setConfig( provider: string, keyPath: string[], value: string | boolean ): Promise> { - const deniedSegment = keyPath.find((segment) => DENIED_KEY_PATH_SEGMENTS.has(segment)); - if (deniedSegment) { - // Match the agentic config mutation path so legacy ORPC callers cannot write into prototypes. - return { success: false, error: `Denied key path segment: "${deniedSegment}"` }; - } + return Effect.runPromise(this.setConfigEffect(provider, keyPath, value)); + } + + public setConfigEffect( + provider: string, + keyPath: string[], + value: string | boolean + ): 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 deniedSegment = keyPath.find((segment) => DENIED_KEY_PATH_SEGMENTS.has(segment)); + if (deniedSegment) { + // Match the agentic config mutation path so legacy ORPC callers cannot write into prototypes. + return { success: false as const, error: `Denied key path segment: "${deniedSegment}"` }; + } - try { const isProviderTypeEdit = keyPath.length === 1 && keyPath[0] === "providerType"; if (isProviderTypeEdit && !isCustomProviderType(value)) { - return { success: false, error: `Invalid custom provider type: ${String(value)}` }; + return { success: false as const, error: `Invalid custom provider type: ${String(value)}` }; } // Value-only guard shared by every provider: no SDK adapter survives a @@ -1437,19 +1641,19 @@ export class ProviderService { if (isBaseUrlEdit && typeof value === "string" && value !== "") { const queryFragmentError = baseUrlQueryFragmentError(value); if (queryFragmentError != null) { - return { success: false, error: queryFragmentError }; + return { success: false as const, error: queryFragmentError }; } } // Read-modify-write under the cross-process lock (see setConfigValue). // The callback returns a policy denial to bail, or null once saved. - const policyDenial = await this.fileLeaseManager.withProvidersFileLock((): string | null => { - const denial = this.validateProviderEditPolicy(provider, keyPath); + const policyDenial = yield* self.providersFileLockEffect((): string | null => { + const denial = self.validateProviderEditPolicy(provider, keyPath); if (denial != null) { return denial; } - const providersConfig = this.providersConfigStore.loadProvidersConfig() ?? {}; + const providersConfig = self.providersConfigStore.loadProvidersConfig() ?? {}; // The add-time id collision rule applies only when this write would // CONVERT a non-custom entry into a custom provider. An entry that is @@ -1533,20 +1737,24 @@ export class ProviderService { } // Save updated config - this.providersConfigStore.saveProvidersConfig(providersConfig); + self.providersConfigStore.saveProvidersConfig(providersConfig); return null; }); if (policyDenial != null) { - return { success: false, error: policyDenial }; + return { success: false as const, error: policyDenial }; } - this.notifyFromMutation(); - await this.syncGatewayLifecycle(provider); + yield* self.notifyFromMutationEffect(); + yield* self.syncGatewayLifecycleEffect(provider); - return { success: true, data: undefined }; - } catch (error) { - const message = getErrorMessage(error); - return { success: false, error: `Failed to set provider config: ${message}` }; - } + return { success: true as const, data: undefined }; + }).pipe( + Effect.catchTag("ProviderPersistenceError", (error) => + Effect.succeed>({ + success: false, + error: `Failed to set provider config: ${error.message}`, + }) + ) + ); } updateRoutePreferences( From 8917825235d001f6e4f08ab5465f8a48fe02399c Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Mon, 31 Aug 2026 19:54:29 +0000 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=A4=96=20fix:=20make=20provider=20mut?= =?UTF-8?q?ation=20pipelines=20uninterruptible?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex P2: a client abort interrupting the handler fiber mid-lock could persist the providers.jsonc write while skipping notify/lifecycle/repair steps. asAtomicMutation (Effect.uninterruptible) restores the pre-Effect run-to-completion semantics; red/green test added. --- src/node/orpc/router.ts | 8 +++-- src/node/services/providerService.test.ts | 36 ++++++++++++++++++++ src/node/services/providerService.ts | 40 ++++++++++++++++++----- 3 files changed, 72 insertions(+), 12 deletions(-) diff --git a/src/node/orpc/router.ts b/src/node/orpc/router.ts index 415a50df24..4424442aeb 100644 --- a/src/node/orpc/router.ts +++ b/src/node/orpc/router.ts @@ -485,9 +485,11 @@ export const router = (authToken?: string) => { .input(schemas.providers.getConfig.input) .output(schemas.providers.getConfig.output) .handler(({ context }) => context.providerService.getConfig()), - // Provider mutations run Effect generators via handlerGen (client aborts - // interrupt the fiber); the wire contracts are unchanged. Sync reads - // (list/getConfig) and subscriptions stay plain handlers. + // Provider mutations run Effect generators via handlerGen; the wire + // contracts are unchanged. The service pipelines are uninterruptible + // (see asAtomicMutation in providerService.ts), so a client abort + // cannot strand a persisted write without its post-write steps. Sync + // reads (list/getConfig) and subscriptions stay plain handlers. addCustomProvider: t .input(schemas.providers.addCustomProvider.input) .output(schemas.providers.addCustomProvider.output) diff --git a/src/node/services/providerService.test.ts b/src/node/services/providerService.test.ts index ce50c4d085..60a89ace63 100644 --- a/src/node/services/providerService.test.ts +++ b/src/node/services/providerService.test.ts @@ -1,5 +1,6 @@ import { FileLeaseManager, ProvidersConfigStore } from "@/node/config"; import { describe, expect, it, spyOn } from "bun:test"; +import { Effect, Fiber } from "effect"; import * as fs from "fs"; import * as fsPromises from "fs/promises"; import { writeFile } from "node:fs/promises"; @@ -2347,3 +2348,38 @@ describe("ProviderService gateway lifecycle", () => { }); }); }); + +describe("ProviderService mutation interruption", () => { + it("runs the write and post-write steps to completion when interrupted mid-mutation", async () => { + await withTempConfigAsync(async (config, service) => { + await saveRoutePriority(config, ["direct"]); + let notified = 0; + const unsubscribe = service.onConfigChanged(() => { + notified += 1; + }); + try { + // runFork executes synchronously up to the first async yield (the + // providers-file lock); interrupting there mirrors an oRPC client + // abort landing while the mutation is in flight (handlerGen + // interrupts the handler fiber on abort). + const fiber = Effect.runFork( + service.setConfigEffect("mux-gateway", ["couponCode"], "gateway-token") + ); + await Effect.runPromise(Fiber.interrupt(fiber)); + + // The mutation pipeline is uninterruptible: the persisted write, the + // change notification, and the gateway routePriority sync must all + // have completed — a write that lands without its post-write steps + // would leave observers and routing state inconsistent. + const stored = new ProvidersConfigStore(config.rootDir).loadProvidersConfig()?.[ + "mux-gateway" + ] as Record; + expect(stored.couponCode).toBe("gateway-token"); + expect(notified).toBe(1); + expect(config.loadConfigOrDefault().routePriority).toEqual(["mux-gateway", "direct"]); + } finally { + unsubscribe(); + } + }); + }); +}); diff --git a/src/node/services/providerService.ts b/src/node/services/providerService.ts index c87bedae36..6b0c27425d 100644 --- a/src/node/services/providerService.ts +++ b/src/node/services/providerService.ts @@ -5,8 +5,11 @@ * Mutation internals are Effect-native: each fallible pipeline is an * `Effect.gen` program and the public Promise methods are thin * `Effect.runPromise` facades, so pre-Effect callers (OAuth services, tests) - * keep working unchanged while oRPC routes ride `handlerGen` directly (client - * aborts interrupt the fiber). The wire `Result` unions stay in the success + * keep working unchanged while oRPC routes ride `handlerGen` directly. Every + * mutation pipeline is uninterruptible (see `asAtomicMutation`), so a client + * abort defers until the write and its post-write consistency steps finish — + * matching the pre-Effect Promise chains, which aborts never cancelled. + * The wire `Result` unions stay in the success * channel — callers branch on `success`, not on thrown errors — and the only * typed failure is `ProviderPersistenceError`, which carries the * `getErrorMessage` string each facade folds into its own wire error shape @@ -163,6 +166,19 @@ function toPersistenceError(error: unknown): ProviderPersistenceError { return new ProviderPersistenceError({ message: getErrorMessage(error) }); } +/** + * Provider mutations must run their write + post-write consistency steps + * (change notification, gateway routePriority sync, durable-reference + * repair) to completion once started: a client abort interrupting the + * handler fiber mid-lock would otherwise persist the callback's write while + * skipping the follow-ups, leaving observers and routing state inconsistent. + * The pre-Effect implementation had exactly these semantics — an aborted + * oRPC request never cancelled the running Promise chain. + */ +function asAtomicMutation(effect: Effect.Effect): Effect.Effect { + return Effect.uninterruptible(effect); +} + export class ProviderService { private readonly policyService: PolicyService | null; private readonly emitter = new EventEmitter(); @@ -842,7 +858,8 @@ export class ProviderService { error.message ), }) - ) + ), + asAtomicMutation ); } @@ -981,7 +998,7 @@ export class ProviderService { self.notifyFromMutation(); return { success: true as const, data: undefined }; - }); + }).pipe(asAtomicMutation); } private repairRemovedCustomProviderReferences( @@ -1150,7 +1167,8 @@ export class ProviderService { success: false, error: `Failed to set models: ${error.message}`, }) - ) + ), + asAtomicMutation ); } @@ -1417,7 +1435,8 @@ export class ProviderService { success: false, error: `Failed to set provider config: ${error.message}`, }) - ) + ), + asAtomicMutation ); } @@ -1510,7 +1529,8 @@ export class ProviderService { success: false, error: `Failed to update provider config: ${error.message}`, }) - ) + ), + asAtomicMutation ); } @@ -1602,7 +1622,8 @@ export class ProviderService { success: false, error: `Failed to update provider config: ${error.message}`, }) - ) + ), + asAtomicMutation ); } @@ -1753,7 +1774,8 @@ export class ProviderService { success: false, error: `Failed to set provider config: ${error.message}`, }) - ) + ), + asAtomicMutation ); }