refactor(dev): type MCP session lifecycle failures and scope the tool-call abort controller (#512)

* refactor(dev): type MCP session lifecycle failures and scope the tool-call abort controller

Expected MCP session failures (closed session or service, uninitialized
client, invalid or duplicate requestId, empty server name) now ride the
Effect error channel as McpSessionError (a CodedError) instead of bare
Error values; messages are unchanged so the routes' message match and
every Promise caller see the same text.

The tool-call request slot is a scoped resource: Effect.acquireRelease
admits the requestId (fails closed on a duplicate) and owns the request's
AbortController; release aborts it and frees the slot, so an interrupted
fiber cannot leak a live SDK request. The host signal composes in with
AbortSignal.any instead of a hand-rolled listener.

* chore(changeset): record the coded MCP session errors (#512)

Author: ScriptedAlchemy

.changeset/512-mcp-session-typed-errors.md

@@ -0,0 +1,5 @@
+---
+"agent-bundle": patch
+---
+
+Reject closed, uninitialized, or misused `agent-bundle dev` MCP sessions with a coded `McpSessionError` (`code` one of `session-closed`, `not-initialized`, `invalid-request-id`, `duplicate-request-id`, `invalid-server-name`, `service-closed`) instead of a bare `Error`; every message is unchanged, so existing message matches keep working. A tool call's request slot is now released — and its in-flight SDK request aborted — whenever the call is interrupted, not only when it settles. (#512)

packages/agent-bundle/src/dev/mcp-session/mcp-session-service.ts

@@ -44,6 +44,7 @@ import {
 } from './mcp-session-apps.ts';
 
 import {
+  McpSessionError,
   McpSessionServiceCloseError,
   McpSessionStaleEpochError,
   type McpClient,
@@ -55,7 +56,8 @@ import {
   type StdioTransport,
 } from './mcp-session-types.ts';
 
-export { McpSessionServiceCloseError, McpSessionStaleEpochError };
+export { McpSessionError, McpSessionServiceCloseError, McpSessionStaleEpochError };
+export type { McpSessionErrorCode } from './mcp-session-types.ts';
 export { mcpAppClientCapabilities };
 export { McpSession } from './mcp-session.ts';
 export type {
@@ -251,7 +253,7 @@ export class McpSessionService {
   }
 
   async open(options: OpenMcpSessionOptions): Promise<McpSession> {
-    if (this.#closed) throw new Error('MCP session service is closed.');
+    if (this.#closed) throw McpSessionError.serviceClosed();
     const timeoutMs = requestOptions(options).timeout;
     const opening = openingSession();
     this.#openingSessions.add(opening);
@@ -303,7 +305,7 @@ export class 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.'));
+          return yield* Effect.fail(McpSessionError.invalidServerName());
         }
         const epochReference = yield* Effect.acquireRelease(
           liftPromise(() => this.#epochStore.acquireEpochReference(options.epochId)),
@@ -345,7 +347,7 @@ export class McpSessionService {
         }));
         constructed = session;
         yield* liftPromise(() => session.initialize({ signal: options.signal }));
-        if (this.#closed) return yield* Effect.fail(new Error('MCP session service is closed.'));
+        if (this.#closed) return yield* Effect.fail(McpSessionError.serviceClosed());
         this.#sessions.set(sessionId, {
           appLeaseCount: 0,
           closeWatchers: new Set(),

packages/agent-bundle/src/dev/mcp-session/mcp-session-types.ts

@@ -19,6 +19,7 @@ import type {
   McpSessionReplayOverflow,
 } from './mcp-session-protocol.ts';
 import type { McpSessionTraceSink } from './mcp-session-trace.ts';
+import { CodedError } from '../../core/errors.ts';
 import { deepFreeze } from '../../core/freeze.ts';
 
 
@@ -146,6 +147,55 @@ export interface McpSessionServiceCloseFailure {
   readonly sessionId?: McpSessionId;
 }
 
+export type McpSessionErrorCode =
+  | 'duplicate-request-id'
+  | 'invalid-request-id'
+  | 'invalid-server-name'
+  | 'not-initialized'
+  | 'service-closed'
+  | 'session-closed';
+
+/**
+ * Expected session-lifecycle failures on the Effect error channel: the
+ * session or its service is closed, a protocol call ran before `initialize`,
+ * or a request was admitted with an invalid or already-active `requestId`.
+ * These ride the fail channel as a `CodedError` (never `Effect.die`) and
+ * rethrow unchanged at `src/effect/boundary.ts`, so Promise callers keep the
+ * exact messages they saw before the class existed.
+ */
+export class McpSessionError extends CodedError<McpSessionErrorCode> {
+  constructor(code: McpSessionErrorCode, message: string) {
+    super('McpSessionError', code, message);
+  }
+
+  static closed(): McpSessionError {
+    return new McpSessionError('session-closed', 'MCP session is closed.');
+  }
+
+  static duplicateRequestId(requestId: string): McpSessionError {
+    return new McpSessionError(
+      'duplicate-request-id',
+      `MCP session request ${JSON.stringify(requestId)} is already active.`,
+    );
+  }
+
+  static invalidRequestId(): McpSessionError {
+    return new McpSessionError('invalid-request-id', 'MCP session requestId must be nonempty.');
+  }
+
+  static invalidServerName(): McpSessionError {
+    return new McpSessionError('invalid-server-name', 'MCP server name must be nonempty.');
+  }
+
+  static notInitialized(): McpSessionError {
+    return new McpSessionError('not-initialized', 'MCP session must initialize before protocol operations.');
+  }
+
+  static serviceClosed(): McpSessionError {
+    return new McpSessionError('service-closed', 'MCP session service is closed.');
+  }
+}
+
 /**
  * The session's pinned artifact epoch is no longer available: the project
  * changed underneath the session (typically another process's build

packages/agent-bundle/src/dev/mcp-session/mcp-session.ts

@@ -7,7 +7,7 @@ import type {
   Tool,
   Transport,
 } from '@modelcontextprotocol/client';
-import { Effect, Semaphore } from 'effect';
+import { Effect, type Scope, Semaphore } from 'effect';
 import { randomUUID } from 'node:crypto';
 import { rm } from 'node:fs/promises';
 import type { Stream } from 'node:stream';
@@ -36,6 +36,7 @@ import {
 import { McpSessionTraceLog, type McpSessionTraceSink } from './mcp-session-trace.ts';
 import { RecordingTransport } from './mcp-recording-transport.ts';
 import {
+  McpSessionError,
   McpSessionStaleEpochError,
   type McpClient,
   type McpRequestOptions as RequestOptions,
@@ -145,6 +146,11 @@ export class McpSession {
   readonly #workspaceRoot: string;
   readonly #frames: McpSessionFrame[] = [];
   readonly #events: McpSessionEvent[] = [];
+  /**
+   * In-flight tool calls by `requestId`. Each controller is owned by its
+   * request's scope (see {@link #admitRequest}); `cancel()` and `#cancelAll`
+   * abort it with their reason and the SDK decides how the call rejects.
+   */
   readonly #requests = new Map<string, AbortController>();
   #capture: StderrCapture | undefined;
   #client: McpClient | undefined;
@@ -313,46 +319,70 @@ export class McpSession {
     return this.#operation('callTool', () => runPromise(this.#callToolEffect(options)));
   }
 
+  /**
+   * One tool call as a scoped Effect. The request slot is the scoped
+   * resource: `acquireRelease` admits the `requestId` (failing closed on a
+   * duplicate) and owns its `AbortController`; release aborts the controller
+   * and frees the slot, so an interrupted fiber can no longer leak a live SDK
+   * request. `cancel()` and `#cancelAll` abort the same controller with their
+   * reason, and the host signal is composed in with `AbortSignal.any`, so the
+   * SDK still decides how an aborted request rejects — exactly as before.
+   *
+   * `scopedAbortSignal` (`Effect.abortSignal`) is the same
+   * `acquireRelease(new AbortController, abort)` shape, but it exposes only
+   * the signal and aborts without a reason; this contract needs both.
+   */
   #callToolEffect(options: McpSessionToolCallOptions): Effect.Effect<CallToolResult, unknown> {
     return this.#assertEpochCurrentEffect().pipe(Effect.andThen(Effect.suspend(() => {
       if (options.signal?.aborted) {
         return Effect.fail(options.signal.reason ?? new Error('MCP session tool call was aborted.'));
       }
       const requestId = options.requestId ?? randomUUID();
       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.`));
+        return Effect.fail(McpSessionError.invalidRequestId());
       }
-      const controller = new AbortController();
-      const onAbort = () => controller.abort(options.signal?.reason);
-      options.signal?.addEventListener('abort', onAbort, { once: true });
-      this.#requests.set(requestId, controller);
-      return Effect.gen({ self: this }, function* (this: McpSession) {
+      const call = Effect.gen({ self: this }, function* (this: McpSession) {
+        const controller = yield* this.#admitRequest(requestId);
         const client = yield* liftTry(() => this.#clientFor());
         const result = yield* liftPromise(() => client.callTool({
           ...(options._meta === undefined ? {} : { _meta: options._meta }),
           arguments: options.arguments,
           name: options.name,
         }, {
-          signal: controller.signal,
+          signal: options.signal === undefined ? controller.signal : AbortSignal.any([controller.signal, options.signal]),
           timeout: requestOptions(options, this.#timeoutMs).timeout,
         }));
         yield* liftTry(() => {
           this.#throwIfStderrExceeded();
         });
         return result;
-      }).pipe(
+      });
+      return Effect.scoped(call).pipe(
         Effect.catch((error) => this.#substituteStaleEpochFailure(error)),
-        Effect.ensuring(Effect.sync(() => {
-          options.signal?.removeEventListener('abort', onAbort);
-          this.#requests.delete(requestId);
-        })),
       );
     })));
   }
 
+  /**
+   * Admits one in-flight request as a scoped resource. Acquire fails closed
+   * when the `requestId` is already active; release aborts the request's
+   * controller (a no-op once the SDK call settled) and frees the slot.
+   */
+  #admitRequest(requestId: string): Effect.Effect<AbortController, McpSessionError, Scope.Scope> {
+    return Effect.acquireRelease(
+      Effect.suspend(() => {
+        if (this.#requests.has(requestId)) return Effect.fail(McpSessionError.duplicateRequestId(requestId));
+        const controller = new AbortController();
+        this.#requests.set(requestId, controller);
+        return Effect.succeed(controller);
+      }),
+      (controller) => Effect.sync(() => {
+        this.#requests.delete(requestId);
+        controller.abort();
+      }),
+    );
+  }
+
   /**
    * A call that failed while the epoch vanished mid-flight reports the
    * stale epoch, not the incidental abort or timeout it produced.
@@ -444,12 +474,12 @@ export class McpSession {
   }
 
   #assertOpen(): void {
-    if (this.#closed) throw new Error('MCP session is closed.');
+    if (this.#closed) throw McpSessionError.closed();
   }
 
-  #assertOpenEffect(): Effect.Effect<void, Error> {
+  #assertOpenEffect(): Effect.Effect<void, McpSessionError> {
     return Effect.suspend(() => this.#closed
-      ? Effect.fail(new Error('MCP session is closed.'))
+      ? Effect.fail(McpSessionError.closed())
       : Effect.void);
   }
 
@@ -495,9 +525,7 @@ export class McpSession {
 
   #clientFor(): McpClient {
     this.#assertOpen();
-    if (this.#connection === undefined) {
-      throw new Error('MCP session must initialize before protocol operations.');
-    }
+    if (this.#connection === undefined) throw McpSessionError.notInitialized();
     this.#throwIfStderrExceeded();
     return this.#client!;
   }

packages/agent-bundle/tests/mcp-session-service.test.ts

@@ -13,7 +13,12 @@ import { normalizeProject } from '../src/config/normalize.ts';
 
 import { EpochStore } from '../src/dev/epoch-store.ts';
 import { McpAppBindingService, type McpAppSessionAuthority } from '../src/dev/mcp-apps/mcp-app-binding-service.ts';
-import { mcpAppClientCapabilities, McpSession, McpSessionService } from '../src/dev/mcp-session/mcp-session-service.ts';
+import {
+  mcpAppClientCapabilities,
+  McpSession,
+  McpSessionError,
+  McpSessionService,
+} from '../src/dev/mcp-session/mcp-session-service.ts';
 import type { ArtifactEpoch } from '../src/dev/types.ts';
 import { pathTokens, type NormalizationTargetRegistry } from '../src/core/types.ts';
 import { agentBundleNodeModules } from './helpers/workspace-paths.ts';
@@ -1078,6 +1083,90 @@ it('fails and closes the session as soon as stderr exceeds its output bound', as
   }
 }, 30_000);
 
+it('fails admission, lifecycle, and service misuse closed with coded McpSessionError values', async () => {
+  const root = await mkdtemp(join(tmpdir(), 'agent-bundle-persistent-mcp-typed-errors-'));
+  try {
+    const epochStore = await publishFixtureEpoch(root, 'epoch-1');
+    const releases: Array<() => void> = [];
+    const signals: AbortSignal[] = [];
+    const service = new McpSessionService({
+      createClient: () => ({
+        callTool: async (_params: unknown, options?: { readonly signal?: AbortSignal }) => {
+          if (options?.signal !== undefined) signals.push(options.signal);
+          await new Promise<void>((resolvePromise) => {
+            releases.push(resolvePromise);
+          });
+          return { content: [] };
+        },
+        close: async () => undefined,
+        connect: async () => undefined,
+        ...mcpCatalogStub(),
+      }),
+      createStdioTransport: () => stdioTransportStub() as never,
+      epochStore,
+      projectRoot: root,
+    });
+
+    const expectSessionError = async (
+      rejection: Promise<unknown>,
+      code: string,
+      message: string,
+    ): Promise<void> => {
+      const error = await rejection.then(
+        () => { throw new Error(`Expected ${code} to reject.`); },
+        (failure: unknown) => failure,
+      );
+      expect(error).toBeInstanceOf(McpSessionError);
+      expect(error).toEqual(expect.objectContaining({ code, message, name: 'McpSessionError' }));
+    };
+
+    await expectSessionError(
+      service.open({ epochId: 'epoch-1', serverName: '  ', target: 'portable' }),
+      'invalid-server-name',
+      'MCP server name must be nonempty.',
+    );
+
+    const session = await service.open({ epochId: 'epoch-1', serverName: 'fixture', target: 'portable' });
+    await expectSessionError(
+      session.callTool({ arguments: {}, name: 'fixture', requestId: ' ' }),
+      'invalid-request-id',
+      'MCP session requestId must be nonempty.',
+    );
+
+    const first = session.callTool({ arguments: {}, name: 'fixture', requestId: 'shared' });
+    await new Promise((resolvePromise) => setTimeout(resolvePromise, 10));
+    expect(signals).toHaveLength(1);
+    await expectSessionError(
+      session.callTool({ arguments: {}, name: 'fixture', requestId: 'shared' }),
+      'duplicate-request-id',
+      'MCP session request "shared" is already active.',
+    );
+    expect(signals[0]?.aborted).toBe(false);
+    releases.shift()?.();
+    await expect(first).resolves.toEqual({ content: [] });
+    // Releasing the request slot aborts its controller and frees the id.
+    expect(signals[0]?.aborted).toBe(true);
+    const reused = session.callTool({ arguments: {}, name: 'fixture', requestId: 'shared' });
+    await new Promise((resolvePromise) => setTimeout(resolvePromise, 10));
+    expect(signals).toHaveLength(2);
+    releases.shift()?.();
+    await expect(reused).resolves.toEqual({ content: [] });
+
+    await session.close();
+    await expectSessionError(session.restart(), 'session-closed', 'MCP session is closed.');
+    await expectSessionError(session.listTools(), 'session-closed', 'MCP session is closed.');
+
+    await service.close();
+    await expectSessionError(
+      service.open({ epochId: 'epoch-1', serverName: 'fixture', target: 'portable' }),
+      'service-closed',
+      'MCP session service is closed.',
+    );
+  } finally {
+    await rm(root, { force: true, recursive: true });
+  }
+}, 30_000);
+
 it('opens a generated streamable HTTP server through its modern transport', async () => {
   const root = await mkdtemp(join(tmpdir(), 'agent-bundle-persistent-mcp-remote-'));
   try {