diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 10b673459e56..7788751b0a60 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -2912,15 +2912,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( : null, ), Effect.orElseSucceed(() => null), - Effect.flatMap((response) => - response === null - ? client.raw - .request("thread/resume", { - threadId: input.nativeThreadId, - excludeTurns: true, - }) - .pipe(Effect.flatMap(decodeCodexChildModel)) - : Effect.succeed(response), + Effect.filterOrElse( + (response) => !(response === null), + (_response) => + client.raw + .request("thread/resume", { + threadId: input.nativeThreadId, + excludeTurns: true, + }) + .pipe(Effect.flatMap(decodeCodexChildModel)), ), Effect.timeout("5 seconds"), Effect.flatMap((response) => diff --git a/apps/server/src/orchestration-v2/CommandOutputHub.test.ts b/apps/server/src/orchestration-v2/CommandOutputHub.test.ts index 06475371f4ce..3740f4f9a240 100644 --- a/apps/server/src/orchestration-v2/CommandOutputHub.test.ts +++ b/apps/server/src/orchestration-v2/CommandOutputHub.test.ts @@ -163,8 +163,9 @@ const subscribeFrames = (threadId: ThreadId) => const takeFrame = (frames: Queue.Queue) => Queue.take(frames).pipe( - Effect.flatMap((frame) => - frame === "ended" ? Effect.die("command output stream ended early") : Effect.succeed(frame), + Effect.filterOrElse( + (frame) => !(frame === "ended"), + (_frame) => Effect.die("command output stream ended early"), ), ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 2a8956913da2..abfdd9127a7b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11981,21 +11981,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } const plan = yield* dispatchOnce(command).pipe( - Effect.flatMap((planned) => - planned.events.length > 0 || - command.type === "thread.background-work.settle" || - command.type === "thread.stop" || - command.type === "subagent.stop" || - // A second Stop sharing, or one after the share already ended, is not an error. - command.type === "thread.guest-share.close" - ? Effect.succeed(planned) - : Effect.fail( - new OrchestratorDispatchError({ - commandId: command.commandId, - commandType: command.type, - cause: "Command produced no domain events.", - }), - ), + Effect.filterOrFail( + (planned) => + planned.events.length > 0 || + command.type === "thread.background-work.settle" || + command.type === "thread.stop" || + command.type === "subagent.stop" || + // Closing a share again, or after it ended, is not an error. + command.type === "thread.guest-share.close", + (_planned) => + new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "Command produced no domain events.", + }), ), Effect.catch((cause) => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 480c7c6c738e..e9b45b5418b2 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -830,6 +830,7 @@ export const layer: Layer.Layer< ), ), ), + // @effect-diagnostics-next-line flatMapConditionalToFilterOrFail:off - The predicate narrows the captured event for the subagent update. Effect.flatMap((storedEvents) => storedEvents.length === 0 || input.event.type !== "subagent.updated" ? Effect.succeed(storedEvents) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 39baad936ba9..5cda4c23e533 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -707,15 +707,14 @@ const make = Effect.gen(function* () { cause, }), ), - Effect.flatMap((projection) => - projection.thread.projectId === input.projectId && projection.thread.deletedAt === null - ? Effect.succeed(projection) - : Effect.fail( - new ThreadManagementThreadNotFoundError({ - projectId: input.projectId, - threadId: input.threadId, - }), - ), + Effect.filterOrFail( + (projection) => + projection.thread.projectId === input.projectId && projection.thread.deletedAt === null, + (_projection) => + new ThreadManagementThreadNotFoundError({ + projectId: input.projectId, + threadId: input.threadId, + }), ), ); @@ -734,15 +733,14 @@ const make = Effect.gen(function* () { cause, }), ), - Effect.flatMap((projection) => - projection.thread.projectId === input.projectId && projection.thread.deletedAt === null - ? Effect.succeed(projection) - : Effect.fail( - new ThreadManagementThreadNotFoundError({ - projectId: input.projectId, - threadId: input.threadId, - }), - ), + Effect.filterOrFail( + (projection) => + projection.thread.projectId === input.projectId && projection.thread.deletedAt === null, + (_projection) => + new ThreadManagementThreadNotFoundError({ + projectId: input.projectId, + threadId: input.threadId, + }), ), ); diff --git a/apps/server/src/vcs/GitVcsDriverCore.ts b/apps/server/src/vcs/GitVcsDriverCore.ts index c689d27be226..9b79c2110e28 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.ts @@ -3961,8 +3961,9 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* }, ); return run.pipe( - Effect.flatMap((result) => - result.exitCode !== 0 && isRefLockFailure(result.stderr) ? run : Effect.succeed(result), + Effect.filterOrElse( + (result) => !(result.exitCode !== 0 && isRefLockFailure(result.stderr)), + (_result) => run, ), ); }; diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 0f917e81f1b9..183bb208ffec 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -1275,20 +1275,20 @@ export function createServerEnvironmentAtoms( refreshTrigger: ({ environmentId }) => usageScanSettingsAtom(environmentId), execute: (input, emit) => request(WS_METHODS.serverGetUsageSummary, input).pipe( - Effect.flatMap((summary) => - summary.sources.some((source) => source.refreshing === true) - ? emit(summary).pipe( - Effect.andThen( - request(WS_METHODS.serverGetUsageSummary, { ...input, awaitRefresh: true }), - ), - Effect.catch((error) => - Effect.logWarning("Could not refresh slow usage sources.").pipe( - Effect.annotateLogs({ ...safeErrorLogAttributes(error) }), - Effect.as(summary), - ), + Effect.filterOrElse( + (summary) => !summary.sources.some((source) => source.refreshing === true), + (summary) => + emit(summary).pipe( + Effect.andThen( + request(WS_METHODS.serverGetUsageSummary, { ...input, awaitRefresh: true }), + ), + Effect.catch((error) => + Effect.logWarning("Could not refresh slow usage sources.").pipe( + Effect.annotateLogs({ ...safeErrorLogAttributes(error) }), + Effect.as(summary), ), - ) - : Effect.succeed(summary), + ), + ), ), ), }), diff --git a/packages/provider-cursor/src/server/driver.ts b/packages/provider-cursor/src/server/driver.ts index 6937c7aa88a7..b61301390d61 100644 --- a/packages/provider-cursor/src/server/driver.ts +++ b/packages/provider-cursor/src/server/driver.ts @@ -225,21 +225,24 @@ export const CursorDriver: ProviderDriver< }, auth.usesApiKey ? "api-key" : "browser", ).pipe( - Effect.flatMap((snapshot) => - effectiveConfig.enabled && - snapshot.installed && - snapshot.auth.status === "authenticated" - ? host.settings.get.pipe( - Effect.flatMap((settings) => - readCursorUsageLimits( - effectiveConfig, - { ...processEnv, CURSOR_API_KEY: apiKey }, - settings.cursorKeychainUsageEnabled, - ), + Effect.filterOrElse( + (snapshot) => + !( + effectiveConfig.enabled && + snapshot.installed && + snapshot.auth.status === "authenticated" + ), + (snapshot) => + host.settings.get.pipe( + Effect.flatMap((settings) => + readCursorUsageLimits( + effectiveConfig, + { ...processEnv, CURSOR_API_KEY: apiKey }, + settings.cursorKeychainUsageEnabled, ), - Effect.map((usageLimits) => ({ ...snapshot, usageLimits })), - ) - : Effect.succeed(snapshot), + ), + Effect.map((usageLimits) => ({ ...snapshot, usageLimits })), + ), ), ), ), diff --git a/packages/provider-grok/src/server/driver.ts b/packages/provider-grok/src/server/driver.ts index 23ecf5815853..84cc590c5455 100644 --- a/packages/provider-grok/src/server/driver.ts +++ b/packages/provider-grok/src/server/driver.ts @@ -168,16 +168,21 @@ export const GrokDriver: ProviderDriver const textGeneration = yield* makeGrokTextGeneration(effectiveConfig, processEnv); const checkProvider = checkGrokProviderStatus(effectiveConfig, processEnv, cwd).pipe( - Effect.flatMap((snapshot) => - effectiveConfig.enabled && snapshot.installed && snapshot.auth.status === "authenticated" - ? readGrokAccount(processEnv).pipe( - Effect.map(({ email, usageLimits }) => ({ - ...snapshot, - auth: email ? { ...snapshot.auth, email } : snapshot.auth, - usageLimits, - })), - ) - : Effect.succeed(snapshot), + Effect.filterOrElse( + (snapshot) => + !( + effectiveConfig.enabled && + snapshot.installed && + snapshot.auth.status === "authenticated" + ), + (snapshot) => + readGrokAccount(processEnv).pipe( + Effect.map(({ email, usageLimits }) => ({ + ...snapshot, + auth: email ? { ...snapshot.auth, email } : snapshot.auth, + usageLimits, + })), + ), ), Effect.map(stampIdentity), Effect.provideService(HttpClient.HttpClient, httpClient), diff --git a/packages/source-control-github/src/server/GitHubPullRequestApi.ts b/packages/source-control-github/src/server/GitHubPullRequestApi.ts index 1099ee1e094c..49aaaa833186 100644 --- a/packages/source-control-github/src/server/GitHubPullRequestApi.ts +++ b/packages/source-control-github/src/server/GitHubPullRequestApi.ts @@ -1578,18 +1578,16 @@ export const make = Effect.gen(function* () { ), ), listActionRequiredRuns(input).pipe( - Effect.flatMap((runs) => - runs.length > workflowApprovalLimit - ? Effect.fail( - new GitHubWorkflowApprovalRefusedError({ - cwd: input.cwd, - number: input.number, - reason: "run-list-truncated", - observedCount: runs.length, - limit: workflowApprovalLimit, - }), - ) - : Effect.succeed(runs), + Effect.filterOrFail( + (runs) => !(runs.length > workflowApprovalLimit), + (runs) => + new GitHubWorkflowApprovalRefusedError({ + cwd: input.cwd, + number: input.number, + reason: "run-list-truncated", + observedCount: runs.length, + limit: workflowApprovalLimit, + }), ), ), ], @@ -1836,40 +1834,41 @@ export const make = Effect.gen(function* () { (batch) => batch.items.length > 0 || input.cursor !== undefined || hasQuery, () => read(false), ), - Effect.flatMap((batch) => { - if (!stacks || batch.items.length === 0) return Effect.succeed(batch); - return Effect.forEach( - batch.items, - (item) => { - const lookup = pullRequestLookup({ - repository: input.repository, - number: item.number, - }); - if (lookup === null) return Effect.succeed(item); - return GraphQLResolver.request(lookup, stackMembershipResolver).pipe( - Effect.map((result) => { - const pullRequest = result.repository?.pullRequest; - const stack = pullRequest == null ? undefined : toStackMembership(pullRequest); - return stack === undefined ? item : { ...item, stack }; - }), - Effect.catch((error) => - Effect.logWarning("Pull request stack membership enrichment failed", { - operation: "listPullRequestStackMemberships", - host: input.host, - error: error._tag, - }).pipe(Effect.as(item)), - ), - ); - }, - { concurrency: "unbounded" }, - ).pipe( - Effect.provideService(GitHubGraphQlClient.GitHubGraphQlTarget, { - host: input.host, - operation: "listPullRequestStackMemberships", - }), - Effect.map((items) => ({ ...batch, items })), - ); - }), + Effect.filterOrElse( + (batch) => !stacks || batch.items.length === 0, + (batch) => + Effect.forEach( + batch.items, + (item) => { + const lookup = pullRequestLookup({ + repository: input.repository, + number: item.number, + }); + if (lookup === null) return Effect.succeed(item); + return GraphQLResolver.request(lookup, stackMembershipResolver).pipe( + Effect.map((result) => { + const pullRequest = result.repository?.pullRequest; + const stack = pullRequest == null ? undefined : toStackMembership(pullRequest); + return stack === undefined ? item : { ...item, stack }; + }), + Effect.catch((error) => + Effect.logWarning("Pull request stack membership enrichment failed", { + operation: "listPullRequestStackMemberships", + host: input.host, + error: error._tag, + }).pipe(Effect.as(item)), + ), + ); + }, + { concurrency: "unbounded" }, + ).pipe( + Effect.provideService(GitHubGraphQlClient.GitHubGraphQlTarget, { + host: input.host, + operation: "listPullRequestStackMemberships", + }), + Effect.map((items) => ({ ...batch, items })), + ), + ), ); }, diff --git a/tsconfig.base.json b/tsconfig.base.json index f4ad537cf4ec..06233a044f66 100644 --- a/tsconfig.base.json +++ b/tsconfig.base.json @@ -24,6 +24,7 @@ "namespaceImportPackages": ["@effect/platform-node", "effect"], "effectFn": ["span", "inferred-span", "suggested-span"], "diagnosticSeverity": { + "flatMapConditionalToFilterOrFail": "error", "catchConditionalRefailToCatchIf": "error", "tryCatchInEffectGen": "error", "schemaStructWithTag": "error",