Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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) =>
Expand Down
5 changes: 3 additions & 2 deletions apps/server/src/orchestration-v2/CommandOutputHub.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -163,8 +163,9 @@ const subscribeFrames = (threadId: ThreadId) =>

const takeFrame = (frames: Queue.Queue<OrchestrationV2CommandOutputFrame | "ended">) =>
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"),
),
);

Expand Down
29 changes: 14 additions & 15 deletions apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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* () {
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/orchestration-v2/ProviderEventIngestor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
34 changes: 16 additions & 18 deletions apps/server/src/orchestration-v2/ThreadManagementService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
}),
),
);

Expand All @@ -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,
}),
),
);

Expand Down
5 changes: 3 additions & 2 deletions apps/server/src/vcs/GitVcsDriverCore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
),
);
};
Expand Down
26 changes: 13 additions & 13 deletions packages/client-runtime/src/state/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1275,20 +1275,20 @@ export function createServerEnvironmentAtoms<R, E>(
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),
),
),
),
),
}),
Expand Down
31 changes: 17 additions & 14 deletions packages/provider-cursor/src/server/driver.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 })),
),
),
),
),
Expand Down
25 changes: 15 additions & 10 deletions packages/provider-grok/src/server/driver.ts
Original file line number Diff line number Diff line change
Expand Up @@ -168,16 +168,21 @@ export const GrokDriver: ProviderDriver<GrokSettings, GrokDriverEnv, Path.Path>
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),
Expand Down
91 changes: 45 additions & 46 deletions packages/source-control-github/src/server/GitHubPullRequestApi.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
}),
),
),
],
Expand Down Expand Up @@ -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 })),
),
),
);
},

Expand Down
1 change: 1 addition & 0 deletions tsconfig.base.json
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
"namespaceImportPackages": ["@effect/platform-node", "effect"],
"effectFn": ["span", "inferred-span", "suggested-span"],
"diagnosticSeverity": {
"flatMapConditionalToFilterOrFail": "error",
"catchConditionalRefailToCatchIf": "error",
"tryCatchInEffectGen": "error",
"schemaStructWithTag": "error",
Expand Down
Loading