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
5 changes: 3 additions & 2 deletions apps/server/src/assets/AttachmentUpload.ts
Original file line number Diff line number Diff line change
Expand Up @@ -230,8 +230,9 @@ export const storeAttachmentUpload = Effect.fn("AttachmentUpload.store")(functio
}),
}).pipe(
Effect.as(true),
Effect.catch((error) =>
error.reason._tag === "AlreadyExists" ? Effect.succeed(false) : Effect.fail(error),
Effect.catchIf(
(error) => error.reason._tag === "AlreadyExists",
() => Effect.succeed(false),
),
);
}),
Expand Down
28 changes: 12 additions & 16 deletions apps/server/src/orchestration-v2/CheckpointRestoreSafety.ts
Original file line number Diff line number Diff line change
Expand Up @@ -34,15 +34,12 @@ export const isCheckpointRestoreIsolated = Effect.fn("orchestrationV2.isCheckpoi
if (!shared && worktreePath !== null) {
// Storage cleanup may have removed the worktree; rollback recreates it.
const resolve = (target: string) =>
fileSystem
.realPath(target)
.pipe(
Effect.catch((error) =>
error.reason._tag === "NotFound"
? Effect.succeed(path.resolve(target))
: Effect.fail(error),
),
);
fileSystem.realPath(target).pipe(
Effect.catchIf(
(error) => error.reason._tag === "NotFound",
(_error) => Effect.succeed(path.resolve(target)),
),
);
const cwd = yield* resolve(scope.cwd);
const worktreeCwd = yield* resolve(worktreePath);
shared = cwd !== worktreeCwd;
Expand Down Expand Up @@ -74,13 +71,12 @@ export const isCheckpointRestoreIsolated = Effect.fn("orchestrationV2.isCheckpoi
for (const candidate of paths) {
if (checkedPaths.has(candidate)) continue;
checkedPaths.add(candidate);
const otherCwd = yield* fileSystem
.realPath(candidate)
.pipe(
Effect.catch((error) =>
error.reason._tag === "NotFound" ? Effect.succeed(null) : Effect.fail(error),
),
);
const otherCwd = yield* fileSystem.realPath(candidate).pipe(
Effect.catchIf(
(error) => error.reason._tag === "NotFound",
(_error) => Effect.succeed(null),
),
);
if (otherCwd !== null && (contains(cwd, otherCwd) || contains(otherCwd, cwd))) {
shared = true;
break;
Expand Down
16 changes: 8 additions & 8 deletions apps/server/src/orchestration-v2/EffectWorker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -177,14 +177,14 @@ export const layerExecutor: Layer.Layer<
cause,
}),
),
Effect.catch((cause) =>
request.subagent === undefined &&
isNonRetryableProviderTurnControlFailure(
request.type,
Cause.pretty(Cause.fail(cause)),
)
? Effect.void
: Effect.fail(cause),
Effect.catchIf(
(cause) =>
request.subagent === undefined &&
isNonRetryableProviderTurnControlFailure(
request.type,
Cause.pretty(Cause.fail(cause)),
),
() => Effect.void,
),
Effect.andThen((outcome) =>
request.subagent !== undefined && outcome === undefined
Expand Down
16 changes: 8 additions & 8 deletions apps/server/src/orchestration-v2/ProviderTurnStartService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -505,14 +505,14 @@ export const layer: Layer.Layer<
path: worktreePath,
}),
),
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.failCause(cause)
: Effect.logWarning("provider turn start failed to recreate worktree", {
threadId: projection.thread.id,
worktreePath,
cause: Cause.pretty(cause),
}),
Effect.catchCauseIf(
(cause) => !Cause.hasInterruptsOnly(cause),
(cause) =>
Effect.logWarning("provider turn start failed to recreate worktree", {
threadId: projection.thread.id,
worktreePath,
cause: Cause.pretty(cause),
}),
),
);
}
Expand Down
14 changes: 7 additions & 7 deletions apps/server/src/orchestration-v2/RunExecutionService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -926,13 +926,13 @@ export const layer: Layer.Layer<
ordinalWithinScope: Math.max(0, input.run.ordinal - 1),
})
.pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.failCause(cause)
: Effect.logWarning(
"orchestration V2 checkpoint baseline capture failed; starting provider without a baseline",
{ runId: input.run.id },
),
Effect.catchCauseIf(
(cause) => !Cause.hasInterruptsOnly(cause),
(_cause) =>
Effect.logWarning(
"orchestration V2 checkpoint baseline capture failed; starting provider without a baseline",
{ runId: input.run.id },
),
),
);
if (
Expand Down
14 changes: 7 additions & 7 deletions apps/server/src/orchestration-v2/ThreadPullRequestService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -318,13 +318,13 @@ export const make = Effect.gen(function* () {
});
pendingBackfill.delete(thread.id);
}).pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.failCause(cause)
: Effect.logWarning("thread pull request update failed", {
threadId: thread.id,
cause: Cause.pretty(cause),
}).pipe(Effect.tap(() => Effect.sync(() => failBackfill([thread])))),
Effect.catchCauseIf(
(cause) => !Cause.hasInterruptsOnly(cause),
(cause) =>
Effect.logWarning("thread pull request update failed", {
threadId: thread.id,
cause: Cause.pretty(cause),
}).pipe(Effect.tap(() => Effect.sync(() => failBackfill([thread])))),
),
),
{ discard: true },
Expand Down
54 changes: 27 additions & 27 deletions apps/server/src/orchestration-v2/ThreadSettlementService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -350,13 +350,13 @@ export const make = Effect.gen(function* () {
},
(effect, thread) =>
effect.pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.failCause(cause)
: Effect.logWarning("automatic thread settlement skipped", {
threadId: thread.id,
cause: Cause.pretty(cause),
}).pipe(Effect.as(null)),
Effect.catchCauseIf(
(cause) => !Cause.hasInterruptsOnly(cause),
(cause) =>
Effect.logWarning("automatic thread settlement skipped", {
threadId: thread.id,
cause: Cause.pretty(cause),
}).pipe(Effect.as(null)),
),
),
);
Expand Down Expand Up @@ -507,13 +507,13 @@ export const make = Effect.gen(function* () {
discard: true,
});
}).pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.failCause(cause)
: Effect.logWarning("automatic thread settlement skipped", {
threadIds: group.map((thread) => thread.id),
cause: Cause.pretty(cause),
}),
Effect.catchCauseIf(
(cause) => !Cause.hasInterruptsOnly(cause),
(cause) =>
Effect.logWarning("automatic thread settlement skipped", {
threadIds: group.map((thread) => thread.id),
cause: Cause.pretty(cause),
}),
),
),
{ concurrency: 8, discard: true },
Expand All @@ -525,12 +525,12 @@ export const make = Effect.gen(function* () {
threadId?: ThreadId,
) =>
sweep(mergedPullRequest, threadId).pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.failCause(cause)
: Effect.logWarning("automatic thread settlement sweep failed", {
cause: Cause.pretty(cause),
}),
Effect.catchCauseIf(
(cause) => !Cause.hasInterruptsOnly(cause),
(cause) =>
Effect.logWarning("automatic thread settlement sweep failed", {
cause: Cause.pretty(cause),
}),
),
);
// A thread that just stopped must not wait behind a full sweep or another thread's host
Expand Down Expand Up @@ -590,13 +590,13 @@ export const make = Effect.gen(function* () {
},
(effect, threadId) =>
effect.pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.failCause(cause)
: Effect.logWarning("cleaning up a settled thread failed", {
threadId,
cause: Cause.pretty(cause),
}),
Effect.catchCauseIf(
(cause) => !Cause.hasInterruptsOnly(cause),
(cause) =>
Effect.logWarning("cleaning up a settled thread failed", {
threadId,
cause: Cause.pretty(cause),
}),
),
),
);
Expand Down
13 changes: 6 additions & 7 deletions apps/server/src/persistence/initializeV2Database.ts
Original file line number Diff line number Diff line change
Expand Up @@ -39,13 +39,12 @@ export const initializeV2Database = Effect.fn("initializeV2Database")(function*
}
});
// Publish only a complete snapshot, without replacing an existing V2 database.
yield* fs
.link(snapshotPath, destinationPath)
.pipe(
Effect.catch((error) =>
error.reason._tag === "AlreadyExists" ? Effect.void : Effect.fail(error),
),
);
yield* fs.link(snapshotPath, destinationPath).pipe(
Effect.catchIf(
(error) => error.reason._tag === "AlreadyExists",
(_error) => Effect.void,
),
);
}).pipe(
Effect.scoped,
Effect.mapError((cause) => new V2DatabaseImportError({ sourcePath, destinationPath, cause })),
Expand Down
10 changes: 6 additions & 4 deletions apps/server/src/pullRequest/PullRequestService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2954,8 +2954,9 @@ export const make = Effect.gen(function* () {
]);
// A replacement reader can join a lookup still finishing its previous reader's cancellation.
return Cache.get(listCache, key).pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause) ? Cache.get(listCache, key) : Effect.failCause(cause),
Effect.catchCauseIf(
(cause) => Cause.hasInterruptsOnly(cause),
(_cause) => Cache.get(listCache, key),
),
);
};
Expand Down Expand Up @@ -3385,8 +3386,9 @@ export const make = Effect.gen(function* () {
if (missing.size === 0) return { stats: held };
const key = statsBatchKey(missing.values());
const { result, at } = yield* Cache.get(listStatsCache, key).pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause) ? Cache.get(listStatsCache, key) : Effect.failCause(cause),
Effect.catchCauseIf(
(cause) => Cause.hasInterruptsOnly(cause),
(_cause) => Cache.get(listStatsCache, key),
),
);
for (const [key, ref] of missing) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -382,6 +382,7 @@ export const make = Effect.gen(function* () {
options,
).pipe(
Effect.as(prepared.remoteUrl),
// @effect-diagnostics-next-line catchConditionalRefailToCatchIf:off - The condition also narrows httpsUrl for the fallback closure.
Effect.catch((sshError) =>
httpsUrl === null || !isSshRejection(sshError)
? Effect.fail(sshError)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1294,8 +1294,9 @@ export const makeAcpRegistryCatalog = Effect.fn("AcpRegistryCatalog.make")(funct
for (let attempt = 0; attempt < INSTALL_LOCK_RETRY_COUNT; attempt += 1) {
const acquired = yield* fileSystem.writeFileString(lockPath, "", { flag: "wx" }).pipe(
Effect.as(true),
Effect.catch((error) =>
isAlreadyExists(error) ? Effect.succeed(false) : Effect.fail(error),
Effect.catchIf(
(error) => isAlreadyExists(error),
() => Effect.succeed(false),
),
);
if (acquired) return;
Expand Down
21 changes: 11 additions & 10 deletions packages/provider-acp/src/server/AcpSessionRuntime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2221,10 +2221,10 @@ export const make = Effect.fn("AcpSessionRuntime.make")(function* (
);
}
const { sessionId, sessionSetupResult } = yield* setupSession.pipe(
Effect.catch((error) =>
!isAcpAuthenticationRequired(error) || options.authenticateOnAuthRequired === false
? Effect.fail(error)
: authenticateAfterRequired(error).pipe(Effect.andThen(setupSession)),
Effect.catchIf(
(error) =>
!(!isAcpAuthenticationRequired(error) || options.authenticateOnAuthRequired === false),
(error) => authenticateAfterRequired(error).pipe(Effect.andThen(setupSession)),
),
);

Expand Down Expand Up @@ -2578,12 +2578,13 @@ export const make = Effect.fn("AcpSessionRuntime.make")(function* (
),
(activePrompt) =>
Fiber.join(activePrompt.fiber).pipe(
Effect.catchCause((cause) =>
options.cancelBehavior !== "wait-for-prompt" && Cause.hasInterruptsOnly(cause)
? Effect.succeed({
stopReason: "cancelled",
} satisfies EffectAcpSchema.PromptResponse)
: Effect.failCause(cause),
Effect.catchCauseIf(
(cause) =>
options.cancelBehavior !== "wait-for-prompt" && Cause.hasInterruptsOnly(cause),
(_cause) =>
Effect.succeed({
stopReason: "cancelled",
} satisfies EffectAcpSchema.PromptResponse),
),
Effect.tap(() =>
closeActiveAssistantSegment({ queue: eventQueue, assistantSegmentRef }),
Expand Down
7 changes: 3 additions & 4 deletions packages/provider-grok/src/server/xaiAcpExtension.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1524,10 +1524,9 @@ export const makeXAiPromptCompletionRuntime = Effect.fn("makeXAiPromptCompletion
.pipe(Effect.forkChild);
return yield* Effect.raceFirst(
Fiber.join(promptRpcFiber).pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.succeed(cancelledResponse)
: Effect.failCause(cause),
Effect.catchCauseIf(
(cause) => Cause.hasInterruptsOnly(cause),
(_cause) => Effect.succeed(cancelledResponse),
),
),
Deferred.await(fallback.deferred),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -907,6 +907,7 @@ export const make = Effect.gen(function* () {
: readLineStats(input, pull.version).pipe(Effect.orElseSucceed(() => null)),
// Reviewers need `pr:review:read`, which a contents-only token lacks.
readPages("getChangeRequest", input, `${path}/reviewers`, Json.GitCafeActorEntries).pipe(
// @effect-diagnostics-next-line catchConditionalRefailToCatchIf:off - catchIf widens this paginated result to any in tsc-rs.
Effect.catch((error) =>
isForbidden(error) ? Effect.succeed({ items: [], next: null }) : Effect.fail(error),
),
Expand Down Expand Up @@ -989,6 +990,7 @@ export const make = Effect.gen(function* () {
: // A closed pull request whose branch is gone keeps its head, but git.cafe answers
// 404 for commits it no longer has; the conversation still stands without them.
readPages(operation, input, `${path}/commits`, Json.GitCafeCommits).pipe(
// @effect-diagnostics-next-line catchConditionalRefailToCatchIf:off - catchIf widens this paginated result to any in tsc-rs.
Effect.catch((error) =>
error.reason === "not-found"
? Effect.succeed({ items: [], next: null })
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": {
"catchConditionalRefailToCatchIf": "error",
"tryCatchInEffectGen": "error",
"schemaStructWithTag": "error",
"preferTypedSchemaDecoder": "error",
Expand Down
Loading