From 3261b9ce83fcc61d58bf292762f12a4a98207701 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 1 Sep 2026 17:36:53 +0000 Subject: [PATCH] refactor(dev): MCP session lifecycles as Effect scoped resources (wave 3.5 stage 3) --- .changeset/effect-mcp-session-lifecycles.md | 5 + .../dev/mcp-session/mcp-session-service.ts | 232 ++++++++++------- .../src/dev/mcp-session/mcp-session.ts | 241 +++++++++++------- packages/agent-bundle/src/effect/lift.ts | 15 ++ 4 files changed, 309 insertions(+), 184 deletions(-) create mode 100644 .changeset/effect-mcp-session-lifecycles.md create mode 100644 packages/agent-bundle/src/effect/lift.ts diff --git a/.changeset/effect-mcp-session-lifecycles.md b/.changeset/effect-mcp-session-lifecycles.md new file mode 100644 index 000000000..c3f90304a --- /dev/null +++ b/.changeset/effect-mcp-session-lifecycles.md @@ -0,0 +1,5 @@ +--- +"agent-bundle": patch +--- + +Rewrite MCP session lifecycles on Effect: the open acquisition chain is scoped `acquireRelease` resources, session teardown is one structured Effect, and the #134 fail-closed stale-epoch contract rides the typed error channel. No public API or wire-contract change. diff --git a/packages/agent-bundle/src/dev/mcp-session/mcp-session-service.ts b/packages/agent-bundle/src/dev/mcp-session/mcp-session-service.ts index 6ada681a2..6a50cd41f 100644 --- a/packages/agent-bundle/src/dev/mcp-session/mcp-session-service.ts +++ b/packages/agent-bundle/src/dev/mcp-session/mcp-session-service.ts @@ -4,6 +4,7 @@ import { type Transport, } from '@modelcontextprotocol/client'; import { StdioClientTransport } from '@modelcontextprotocol/client/stdio'; +import { Cause, Effect, Exit } from 'effect'; import { randomUUID } from 'node:crypto'; import { mkdtemp, readFile, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; @@ -14,6 +15,8 @@ import { validateArtifact } from '../../build/validate-artifact.ts'; import { DiagnosticError } from '../../core/diagnostics.ts'; import { joinArtifact } from '../../core/paths.ts'; import { isRecord, parseJsonWithoutDuplicateKeys } from '../../core/strict-json.ts'; +import { runPromise } from '../../effect/boundary.ts'; +import { liftPromise, liftTry } from '../../effect/lift.ts'; import { readTargetMcpServer, type ModernMcpServer, @@ -258,12 +261,12 @@ export class McpSessionService { let cleanupFailed = false; let cleanupFailure: unknown; try { - return await this.#open({ ...options, signal, timeoutMs }, (error) => { + return await runPromise(this.#openEffect({ ...options, signal, timeoutMs }, (error) => { if (!cleanupFailed) { cleanupFailed = true; cleanupFailure = error; } - }); + })); } finally { this.#openingSessions.delete(opening); opening.finish(cleanupFailed @@ -272,82 +275,102 @@ export class McpSessionService { } } - async #open( + /** + * The open acquisition chain as scoped resources: the epoch lease and the + * plugin-data directory are `acquireRelease`d into the open scope, and + * their releases run — newest first, failures collected, never thrown from + * a finalizer — only while the session has not been constructed. Once the + * `McpSession` exists it owns every resource, the scope finalizers disarm, + * and a later open failure is cleaned up by `session.close()` instead. + */ + #openEffect( options: OpenMcpSessionOptions, reportCleanupFailure: (error: unknown) => void, - ): Promise { - const target = options.target; - const runtime = this.#runtime(target); - if (options.serverName.trim().length === 0) throw new Error('MCP server name must be nonempty.'); - const epochReference = await this.#epochStore.acquireEpochReference(options.epochId); - let pluginData: string | undefined; - let session: McpSession | undefined; - try { - const epochRoot = epochReference.root; - const diagnostics = await validateArtifact({ - allowEpochStagingMarker: true, - artifactRoot: epochRoot, - registry: this.#registry, - }); - const errors = diagnostics.filter((diagnostic) => diagnostic.severity === 'error'); - if (errors.length > 0) throw new DiagnosticError(errors); - const targetRoot = joinArtifact(epochRoot, target); - const server = await this.#server(targetRoot, target, runtime, options.serverName); - pluginData = await mkdtemp(resolve(tmpdir(), 'agent-bundle-mcp-')); - const sessionId = randomUUID(); - session = new McpSession({ - assertEpochAvailable: async () => { - const probe = await this.#epochStore.acquireEpochReference(options.epochId); - await probe.close(); - }, - binding: { epochId: options.epochId, serverName: options.serverName, target }, - createClient: this.#createClient, - createStdioTransport: this.#createStdioTransport, - createStreamableHttpTransport: this.#createStreamableHttpTransport, - epochReference, - id: sessionId, - onClose: () => this.#invalidateSession(sessionId, new Error('MCP session closed.')), - onClosing: () => this.#invalidateSession(sessionId, new Error('MCP session is closing.')), - pluginData, - resolved: { runtime, server, target, targetRoot }, - timeoutMs: options.timeoutMs, - ...(this.#traceSink === undefined ? {} : { traceSink: this.#traceSink }), - workspaceRoot: resolve(options.workspaceRoot ?? this.#projectRoot), - }); - await session.initialize({ signal: options.signal }); - if (this.#closed) throw new Error('MCP session service is closed.'); - this.#sessions.set(sessionId, { - appLeaseCount: 0, - closeWatchers: new Set(), - closed: false, - session, - }); - return session; - } catch (error) { - if (session !== undefined) { - try { - await session.close(); - } catch (cleanupError) { - reportCleanupFailure(cleanupError); - throw cleanupError; - } - } else { - const cleanupFailures: unknown[] = []; - try { - if (pluginData !== undefined) await rm(pluginData, { force: true, recursive: true }); - } catch (cleanupError) { - cleanupFailures.push(cleanupError); - } - try { - await epochReference.close(); - } catch (cleanupError) { - cleanupFailures.push(cleanupError); + ): Effect.Effect { + return Effect.suspend(() => { + const cleanupFailures: unknown[] = []; + let constructed: McpSession | undefined; + const releaseUnlessTransferred = (release: () => Promise): Effect.Effect => + Effect.promise(async () => { + if (constructed !== undefined) return; + try { + await release(); + } catch (error) { + cleanupFailures.push(error); + } + }); + const program = Effect.gen({ self: this }, function* (this: McpSessionService) { + const target = options.target; + const runtime = yield* liftTry(() => this.#runtime(target)); + if (options.serverName.trim().length === 0) { + return yield* Effect.fail(new Error('MCP server name must be nonempty.')); } - for (const failure of cleanupFailures) reportCleanupFailure(failure); - if (cleanupFailures.length > 0) throw cleanupFailures[cleanupFailures.length - 1]; - } - throw error; - } + const epochReference = yield* Effect.acquireRelease( + liftPromise(() => this.#epochStore.acquireEpochReference(options.epochId)), + (reference) => releaseUnlessTransferred(() => reference.close()), + ); + const epochRoot = epochReference.root; + const diagnostics = yield* liftPromise(() => validateArtifact({ + allowEpochStagingMarker: true, + artifactRoot: epochRoot, + registry: this.#registry, + })); + const errors = diagnostics.filter((diagnostic) => diagnostic.severity === 'error'); + if (errors.length > 0) return yield* Effect.fail(new DiagnosticError(errors)); + const targetRoot = yield* liftTry(() => joinArtifact(epochRoot, target)); + const server = yield* liftPromise(() => this.#server(targetRoot, target, runtime, options.serverName)); + const pluginData = yield* Effect.acquireRelease( + liftPromise(() => mkdtemp(resolve(tmpdir(), 'agent-bundle-mcp-'))), + (directory) => releaseUnlessTransferred(() => rm(directory, { force: true, recursive: true })), + ); + const sessionId = randomUUID(); + const session = yield* liftTry(() => new McpSession({ + assertEpochAvailable: async () => { + const probe = await this.#epochStore.acquireEpochReference(options.epochId); + await probe.close(); + }, + binding: { epochId: options.epochId, serverName: options.serverName, target }, + createClient: this.#createClient, + createStdioTransport: this.#createStdioTransport, + createStreamableHttpTransport: this.#createStreamableHttpTransport, + epochReference, + id: sessionId, + onClose: () => this.#invalidateSession(sessionId, new Error('MCP session closed.')), + onClosing: () => this.#invalidateSession(sessionId, new Error('MCP session is closing.')), + pluginData, + resolved: { runtime, server, target, targetRoot }, + timeoutMs: options.timeoutMs, + ...(this.#traceSink === undefined ? {} : { traceSink: this.#traceSink }), + workspaceRoot: resolve(options.workspaceRoot ?? this.#projectRoot), + })); + constructed = session; + yield* liftPromise(() => session.initialize({ signal: options.signal })); + if (this.#closed) return yield* Effect.fail(new Error('MCP session service is closed.')); + this.#sessions.set(sessionId, { + appLeaseCount: 0, + closeWatchers: new Set(), + closed: false, + session, + }); + return session; + }); + return Effect.scoped(program).pipe( + Effect.catch((error) => Effect.suspend(() => { + const session = constructed; + if (session !== undefined) { + return liftPromise(() => session.close()).pipe( + Effect.catch((cleanupError) => Effect.suspend(() => { + reportCleanupFailure(cleanupError); + return Effect.fail(cleanupError); + })), + Effect.andThen(Effect.fail(error)), + ); + } + for (const failure of cleanupFailures) reportCleanupFailure(failure); + return Effect.fail(cleanupFailures.length > 0 ? cleanupFailures[cleanupFailures.length - 1] : error); + })), + ); + }); } get(id: McpSessionId): McpSession | undefined { @@ -375,28 +398,49 @@ export class McpSessionService { const entry = this.#invalidateSession(id, new Error('MCP session service is closed.')); return entry === undefined ? [] : [[id, entry.session] as const]; }); - this.#closePromise = this.#close(sessions); + const openings = [...this.#openingSessions]; + for (const opening of openings) opening.abort.abort(new Error('MCP session service is closed.')); + this.#closePromise = runPromise(this.#closeEffect(openings, sessions)); return this.#closePromise; } - async #close(sessions: readonly (readonly [string, McpSession])[]): Promise { - const openings = [...this.#openingSessions]; - for (const opening of openings) opening.abort.abort(new Error('MCP session service is closed.')); - const openingResults = await Promise.allSettled(openings.map((opening) => opening.done)); - const sessionResults = await Promise.allSettled(sessions.map(([, session]) => session.close())); - const failures = Object.freeze([ - ...openingResults.flatMap((result): readonly McpSessionServiceCloseFailure[] => - result.status === 'rejected' - ? [Object.freeze({ error: result.reason, resource: 'opening' as const })] - : []), - ...sessionResults.flatMap((result, index): readonly McpSessionServiceCloseFailure[] => { - const sessionId = sessions[index]?.[0]; - return result.status === 'rejected' && sessionId !== undefined - ? [Object.freeze({ error: result.reason, resource: 'session' as const, sessionId })] - : []; - }), - ]); - if (failures.length > 0) throw new McpSessionServiceCloseError(failures); + /** + * Waits for every tracked lifecycle — openings first, then active + * sessions, each phase settling concurrently via per-element `Exit` — and + * fails with one `McpSessionServiceCloseError` naming every resource that + * could not be released. + */ + #closeEffect( + openings: readonly OpeningSession[], + sessions: readonly (readonly [string, McpSession])[], + ): Effect.Effect { + return Effect.gen(function* () { + const openingResults = yield* Effect.forEach( + openings, + (opening) => Effect.exit(liftPromise(() => opening.done)), + { concurrency: 'unbounded' }, + ); + const sessionResults = yield* Effect.forEach( + sessions, + ([, session]) => Effect.exit(liftPromise(() => session.close())), + { concurrency: 'unbounded' }, + ); + const failures = Object.freeze([ + ...openingResults.flatMap((result): readonly McpSessionServiceCloseFailure[] => + Exit.isFailure(result) + ? [Object.freeze({ error: Cause.squash(result.cause), resource: 'opening' as const })] + : []), + ...sessionResults.flatMap((result, index): readonly McpSessionServiceCloseFailure[] => { + const sessionId = sessions[index]?.[0]; + return Exit.isFailure(result) && sessionId !== undefined + ? [Object.freeze({ error: Cause.squash(result.cause), resource: 'session' as const, sessionId })] + : []; + }), + ]); + if (failures.length > 0) { + return yield* Effect.fail(new McpSessionServiceCloseError(failures)); + } + }); } #invalidateSession(id: string, reason: unknown): ActiveSession | undefined { diff --git a/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts b/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts index 5d44d59f7..130f341e1 100644 --- a/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts +++ b/packages/agent-bundle/src/dev/mcp-session/mcp-session.ts @@ -7,12 +7,14 @@ import type { Tool, Transport, } from '@modelcontextprotocol/client'; +import { Effect, Semaphore } from 'effect'; import { randomUUID } from 'node:crypto'; import { rm } from 'node:fs/promises'; import type { Stream } from 'node:stream'; -import { serialQueue } from '../../core/async.ts'; import { isRecord, snapshotStrictJsonValue } from '../../core/strict-json.ts'; +import { runPromise, runSync } from '../../effect/boundary.ts'; +import { liftPromise, liftTry } from '../../effect/lift.ts'; import type { EpochReference } from '../epoch-store.ts'; import type { McpSessionBinding, @@ -151,7 +153,8 @@ export class McpSession { #closed = false; #connection: McpSessionConnectionState | undefined; #droppedThroughSequence = 0; - readonly #lifecycle = serialQueue(); + /** Serializes initialize / restart / close, replacing the pre-Effect serial queue. */ + readonly #lifecycle: Semaphore.Semaphore = runSync(Semaphore.make(1)); #sequence = 0; #stderrOutput = ''; #stderrOverflow = false; @@ -253,12 +256,14 @@ export class McpSession { } async initialize(options?: McpSessionRequestOptions): Promise { - return this.#operation('initialize', () => this.#lifecycle.run(async () => { - this.#assertOpen(); - if (this.#connection === undefined) await this.#connect(options); - this.#assertOpen(); - return this.connection; - })); + return this.#operation('initialize', () => runPromise(this.#lifecycle.withPermit( + Effect.gen({ self: this }, function* (this: McpSession) { + yield* this.#assertOpenEffect(); + if (this.#connection === undefined) yield* this.#connectEffect(options); + yield* this.#assertOpenEffect(); + return this.connection; + }), + ))); } async listTools(options?: McpSessionRequestOptions): Promise { @@ -305,37 +310,58 @@ export class McpSession { if (options.signal?.aborted) { throw options.signal.reason ?? new Error('MCP session tool call was aborted.'); } - return this.#operation('callTool', async () => { - await this.#assertEpochCurrent(); + return this.#operation('callTool', () => runPromise(this.#callToolEffect(options))); + } + + #callToolEffect(options: McpSessionToolCallOptions): Effect.Effect { + return this.#assertEpochCurrentEffect().pipe(Effect.andThen(Effect.suspend(() => { const requestId = options.requestId ?? randomUUID(); - if (requestId.trim().length === 0) throw new Error('MCP session requestId must be nonempty.'); - if (this.#requests.has(requestId)) throw new Error(`MCP session request ${JSON.stringify(requestId)} is already active.`); + if (requestId.trim().length === 0) { + return Effect.fail(new Error('MCP session requestId must be nonempty.')); + } + if (this.#requests.has(requestId)) { + return Effect.fail(new Error(`MCP session request ${JSON.stringify(requestId)} is already active.`)); + } const controller = new AbortController(); const onAbort = () => controller.abort(options.signal?.reason); options.signal?.addEventListener('abort', onAbort, { once: true }); this.#requests.set(requestId, controller); - try { - const result = await this.#clientFor().callTool({ arguments: options.arguments, name: options.name }, { + return Effect.gen({ self: this }, function* (this: McpSession) { + const client = yield* liftTry(() => this.#clientFor()); + const result = yield* liftPromise(() => client.callTool({ arguments: options.arguments, name: options.name }, { signal: controller.signal, timeout: requestOptions(options, this.#timeoutMs).timeout, + })); + yield* liftTry(() => { + this.#throwIfStderrExceeded(); }); - this.#throwIfStderrExceeded(); return result; - } catch (error) { - // A call that failed while the epoch vanished mid-flight reports the - // stale epoch, not the incidental abort or timeout it produced. - if (this.#staleEpochFailure === undefined && !this.#closed && this.#assertEpochAvailable !== undefined) { - try { - await this.#assertEpochAvailable(); - } catch (cause) { - this.#failStaleEpoch(cause); - } - } - throw this.#staleEpochFailure ?? error; - } finally { - options.signal?.removeEventListener('abort', onAbort); - this.#requests.delete(requestId); + }).pipe( + Effect.catch((error) => this.#substituteStaleEpochFailure(error)), + Effect.ensuring(Effect.sync(() => { + options.signal?.removeEventListener('abort', onAbort); + this.#requests.delete(requestId); + })), + ); + }))); + } + + /** + * A call that failed while the epoch vanished mid-flight reports the + * stale epoch, not the incidental abort or timeout it produced. + */ + #substituteStaleEpochFailure(error: unknown): Effect.Effect { + return Effect.suspend(() => { + const probe = this.#assertEpochAvailable; + if (this.#staleEpochFailure !== undefined || this.#closed || probe === undefined) { + return Effect.fail(this.#staleEpochFailure ?? error); } + return liftPromise(() => probe()).pipe( + Effect.catch((cause) => Effect.sync(() => { + this.#failStaleEpoch(cause); + })), + Effect.andThen(Effect.suspend(() => Effect.fail(this.#staleEpochFailure ?? error))), + ); }); } @@ -352,16 +378,18 @@ export class McpSession { } async restart(options?: McpSessionRequestOptions): Promise { - return this.#operation('restart', () => this.#lifecycle.run(async () => { - this.#assertOpen(); - this.#cancelAll('MCP session restarted.'); - await this.#closeClient(); - this.#assertOpen(); - this.#connection = undefined; - await this.#connect(options); - this.#assertOpen(); - return this.connection; - })); + return this.#operation('restart', () => runPromise(this.#lifecycle.withPermit( + Effect.gen({ self: this }, function* (this: McpSession) { + yield* this.#assertOpenEffect(); + this.#cancelAll('MCP session restarted.'); + yield* liftPromise(() => this.#closeClient()); + yield* this.#assertOpenEffect(); + this.#connection = undefined; + yield* this.#connectEffect(options); + yield* this.#assertOpenEffect(); + return this.connection; + }), + ))); } close(): Promise { @@ -374,46 +402,69 @@ export class McpSession { } catch { // Lifecycle observers cannot prevent client, temporary-data, or epoch cleanup. } - void this.#operation('close', () => this.#lifecycle.run(() => this.#close())).then(closing.resolve, closing.reject); + void this.#operation('close', () => runPromise(this.#lifecycle.withPermit(this.#closeEffect()))) + .then(closing.resolve, closing.reject); return closing.promise; } - async #close(): Promise { - this.#cancelAll('MCP session closed.'); - try { - await this.#closeClient(); - } finally { - try { - await rm(this.#pluginData, { force: true, recursive: true }); - } finally { - try { - await this.#epochReference.close(); - } finally { - this.#onClose(); + /** + * Session teardown as one Effect. Every release step always runs, in + * order — drain the client, remove plugin data, release the epoch lease, + * notify the owner — and the last failing step's error is re-raised once + * every resource has been visited (the pre-Effect nested `finally` chain + * had the same last-failure-wins contract). + */ + #closeEffect(): Effect.Effect { + return Effect.suspend(() => { + const failures: unknown[] = []; + const step = (release: () => Promise): Effect.Effect => + liftPromise(release).pipe( + Effect.catch((error) => Effect.sync(() => { + failures.push(error); + })), + Effect.asVoid, + ); + return Effect.gen({ self: this }, function* (this: McpSession) { + this.#cancelAll('MCP session closed.'); + yield* step(() => this.#closeClient()); + yield* step(() => rm(this.#pluginData, { force: true, recursive: true })); + yield* step(() => this.#epochReference.close()); + this.#onClose(); + if (failures.length > 0) { + return yield* Effect.fail(failures[failures.length - 1]); } - } - } + }); + }); } #assertOpen(): void { if (this.#closed) throw new Error('MCP session is closed.'); } + #assertOpenEffect(): Effect.Effect { + return Effect.suspend(() => this.#closed + ? Effect.fail(new Error('MCP session is closed.')) + : Effect.void); + } + /** * Fails a tool call closed when the pinned epoch no longer exists — the * project changed underneath the session (often another process's build * retention, which cannot observe this process's epoch leases). Discovery * cancels every in-flight request with the same typed failure and closes - * the session, mirroring the stderr-overflow contract. + * the session, mirroring the stderr-overflow contract. The #134 contract + * rides the typed error channel as `McpSessionStaleEpochError`. */ - async #assertEpochCurrent(): Promise { - if (this.#staleEpochFailure !== undefined) throw this.#staleEpochFailure; - if (this.#assertEpochAvailable === undefined) return; - try { - await this.#assertEpochAvailable(); - } catch (cause) { - throw this.#failStaleEpoch(cause); - } + #assertEpochCurrentEffect(): Effect.Effect { + return Effect.suspend(() => { + if (this.#staleEpochFailure !== undefined) return Effect.fail(this.#staleEpochFailure); + const probe = this.#assertEpochAvailable; + if (probe === undefined) return Effect.void; + return liftPromise(() => probe()).pipe( + Effect.catch((cause) => Effect.fail(this.#failStaleEpoch(cause))), + Effect.asVoid, + ); + }); } #failStaleEpoch(cause: unknown): McpSessionStaleEpochError { @@ -445,33 +496,43 @@ export class McpSession { return this.#client!; } - async #connect(options?: McpSessionRequestOptions): Promise { - const client = this.#createClient(); - let capture: StderrCapture | undefined; - try { - const transport = this.#transport((nextCapture) => { - capture = nextCapture; - }); - const recording = new RecordingTransport(transport, (direction, message) => this.#recordFrame(direction, message)); - await client.connect(recording, requestOptions(options, this.#timeoutMs)); - this.#throwIfStderrExceeded(capture); - this.#assertOpen(); - this.#client = client; - this.#capture = capture; - this.#connection = Object.freeze({ - capabilities: client.getServerCapabilities(), - protocolEra: client.getProtocolEra?.(), - protocolVersion: client.getNegotiatedProtocolVersion?.(), - server: client.getServerVersion(), - }); - } catch (error) { - try { - await client.close(); - } finally { - capture?.stop(); - } - throw error; - } + #connectEffect(options?: McpSessionRequestOptions): Effect.Effect { + return Effect.suspend(() => { + const client = this.#createClient(); + const connectState: { capture?: StderrCapture } = {}; + return Effect.gen({ self: this }, function* (this: McpSession) { + const recording = yield* liftTry(() => { + const transport = this.#transport((nextCapture) => { + connectState.capture = nextCapture; + }); + return new RecordingTransport(transport, (direction, message) => this.#recordFrame(direction, message)); + }); + yield* liftPromise(() => client.connect(recording, requestOptions(options, this.#timeoutMs))); + yield* liftTry(() => { + this.#throwIfStderrExceeded(connectState.capture); + this.#assertOpen(); + this.#client = client; + this.#capture = connectState.capture; + this.#connection = Object.freeze({ + capabilities: client.getServerCapabilities(), + protocolEra: client.getProtocolEra?.(), + protocolVersion: client.getNegotiatedProtocolVersion?.(), + server: client.getServerVersion(), + }); + }); + }).pipe( + // A failed connect drains the replacement client and stops its + // stderr capture before the failure re-raises; a cleanup failure + // replaces the original error, exactly as the pre-Effect catch did. + Effect.catch((error) => liftPromise(async () => { + try { + await client.close(); + } finally { + connectState.capture?.stop(); + } + }).pipe(Effect.andThen(Effect.fail(error)))), + ); + }); } async #closeClient(): Promise { diff --git a/packages/agent-bundle/src/effect/lift.ts b/packages/agent-bundle/src/effect/lift.ts new file mode 100644 index 000000000..5fec1275d --- /dev/null +++ b/packages/agent-bundle/src/effect/lift.ts @@ -0,0 +1,15 @@ +import { Effect } from 'effect'; + +/** + * Lifts for the dev seam's existing Promise/sync helpers. Both keep the + * thrown/rejected value untouched in the error channel — the dev seam's + * typed contracts are plain `Error` subclasses that must cross + * `src/effect/boundary.ts` identity-preserved, and several call sites + * re-raise non-Error values (for example an `AbortSignal.reason`) verbatim. + */ + +export const liftPromise = (evaluate: () => PromiseLike): Effect.Effect => + Effect.tryPromise({ catch: (error) => error, try: evaluate }); + +export const liftTry = (evaluate: () => A): Effect.Effect => + Effect.try({ catch: (error) => error, try: evaluate });