fix: keep quiet MCP SSE streams alive

.changeset/quiet-mcp-sse-keepalive.md

@@ -0,0 +1,5 @@
+---
+"executor": patch
+---
+
+Keep quiet MCP GET streams alive with SSE comment heartbeats so Bun clients do not time out.

apps/local/src/mcp-sse-heartbeat.test.ts

@@ -0,0 +1,145 @@
+// ---------------------------------------------------------------------------
+// Quiet standalone GET `/mcp` under Bun — regression for #1983
+// ---------------------------------------------------------------------------
+//
+// After initialize, the SDK GET stream is a 200 `text/event-stream` with no
+// body bytes until a server-initiated message. Bun fetch does not settle a
+// silent stream even when `Bun.serve` has `idleTimeout: 0`. The handler must
+// emit a legal SSE comment so the GET resolves promptly, then drop the
+// upstream stream on cancel so a reconnect is not 409.
+// ---------------------------------------------------------------------------
+
+import { describe, expect, it } from "@effect/vitest";
+import { Effect } from "effect";
+
+import type { ExecutionEngine } from "@executor-js/execution";
+
+import { createMcpRequestHandler } from "./mcp";
+
+const MCP_POST_HEADERS = {
+  "content-type": "application/json",
+  accept: "application/json, text/event-stream",
+} as const;
+
+const stubEngine: ExecutionEngine<never> = {
+  execute: () => Effect.succeed({ result: "unused" }),
+  executeWithPause: () => Effect.succeed({ status: "completed", result: { result: "unused" } }),
+  resume: () => Effect.succeed(null),
+  getPausedExecution: () => Effect.succeed(null),
+  pausedExecutionCount: () => Effect.succeed(0),
+  hasPausedExecutions: () => Effect.succeed(false),
+  getDescription: Effect.succeed("test executor"),
+  shutdown: Effect.void,
+};
+
+const initializeBody = {
+  jsonrpc: "2.0",
+  id: 1,
+  method: "initialize",
+  params: {
+    protocolVersion: "2025-06-18",
+    capabilities: {},
+    clientInfo: { name: "bun-sse-test", version: "1.0.0" },
+  },
+};
+
+const openLiveSession = async (
+  origin: string,
+  path = "/mcp",
+): Promise<{ readonly sessionId: string }> => {
+  const init = await fetch(`${origin}${path}`, {
+    method: "POST",
+    headers: MCP_POST_HEADERS,
+    body: JSON.stringify(initializeBody),
+  });
+  expect(init.status).toBe(200);
+  const sessionId = init.headers.get("mcp-session-id");
+  expect(sessionId).toBeTruthy();
+  await init.body?.cancel();
+
+  const initialized = await fetch(`${origin}${path}`, {
+    method: "POST",
+    headers: { ...MCP_POST_HEADERS, "mcp-session-id": sessionId! },
+    body: JSON.stringify({ jsonrpc: "2.0", method: "notifications/initialized" }),
+  });
+  expect(initialized.status).toBe(202);
+  await initialized.body?.cancel();
+  return { sessionId: sessionId! };
+};
+
+describe("local MCP handler, quiet GET stream", () => {
+  it("resolves a live-session GET under Bun with a keepalive comment and reconnects after cancel", async () => {
+    const handler = createMcpRequestHandler({ engine: stubEngine });
+    const server = Bun.serve({
+      port: 0,
+      idleTimeout: 0,
+      fetch: (request) => handler.handleRequest(request),
+    });
+
+    // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: always stop the server
+    try {
+      const origin = `http://127.0.0.1:${server.port}`;
+      const { sessionId } = await openLiveSession(origin);
+
+      const getAbort = new AbortController();
+      const getTimeout = setTimeout(() => getAbort.abort(), 1_000);
+      const get = await fetch(`${origin}/mcp`, {
+        method: "GET",
+        headers: { accept: "text/event-stream", "mcp-session-id": sessionId },
+        signal: getAbort.signal,
+      });
+      expect(get.status).toBe(200);
+      expect(get.headers.get("content-type")).toContain("text/event-stream");
+      expect(get.headers.get("mcp-session-id")).toBe(sessionId);
+
+      const reader = get.body!.getReader();
+      const first = await reader.read();
+      expect(new TextDecoder().decode(first.value)).toBe(": keepalive\n\n");
+      await reader.cancel();
+      getAbort.abort();
+      clearTimeout(getTimeout);
+
+      const reconnect = await fetch(`${origin}/mcp`, {
+        method: "GET",
+        headers: { accept: "text/event-stream", "mcp-session-id": sessionId },
+        signal: AbortSignal.timeout(1_000),
+      });
+      expect(reconnect.status).toBe(200);
+      expect(reconnect.headers.get("content-type")).toContain("text/event-stream");
+      await reconnect.body?.cancel();
+    } finally {
+      server.stop(true);
+      await handler.close();
+    }
+  });
+
+  it("keeps the same GET contract on a toolkit MCP path", async () => {
+    const handler = createMcpRequestHandler({ engine: stubEngine });
+    const server = Bun.serve({
+      port: 0,
+      idleTimeout: 0,
+      fetch: (request) => handler.handleRequest(request),
+    });
+
+    // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: always stop the server
+    try {
+      const origin = `http://127.0.0.1:${server.port}`;
+      const path = "/mcp/toolkits/deploy";
+      const { sessionId } = await openLiveSession(origin, path);
+
+      const get = await fetch(`${origin}${path}`, {
+        method: "GET",
+        headers: { accept: "text/event-stream", "mcp-session-id": sessionId },
+        signal: AbortSignal.timeout(1_000),
+      });
+      expect(get.status).toBe(200);
+      expect(get.headers.get("content-type")).toContain("text/event-stream");
+      const reader = get.body!.getReader();
+      expect(new TextDecoder().decode((await reader.read()).value)).toBe(": keepalive\n\n");
+      await reader.cancel();
+    } finally {
+      server.stop(true);
+      await handler.close();
+    }
+  });
+});

apps/local/src/mcp.ts

@@ -10,6 +10,7 @@ import {
   preInitializeMethodNotFound,
   type McpResource,
 } from "@executor-js/host-mcp";
+import { withMcpSseHeartbeat } from "@executor-js/host-mcp/sse-heartbeat";
 import {
   createExecutorMcpServer,
   type ExecutorMcpServerConfig,
@@ -196,7 +197,7 @@ export const createMcpRequestHandler = (
         if (!sessionResource || mcpResourceKey(sessionResource) !== mcpResourceKey(resource)) {
           return jsonError(403, -32003, "Session belongs to a different MCP resource");
         }
-        return transport.handleRequest(request);
+        return withMcpSseHeartbeat(request, await transport.handleRequest(request));
       }
 
       // Pre-initialize dispatch: only `initialize` opens a session here, so a
@@ -263,7 +264,7 @@ export const createMcpRequestHandler = (
           }),
         );
         await created.connect(transport);
-        const response = await transport.handleRequest(request);
+        const response = withMcpSseHeartbeat(request, await transport.handleRequest(request));
 
         if (!transport.sessionId) {
           await ignoreClose(() => transport.close());

e2e/local/mcp-standalone-get.test.ts

@@ -0,0 +1,95 @@
+// Black-box regression for #1983: a quiet Streamable HTTP GET `/mcp` on the
+// local daemon must resolve under Bun with a legal SSE comment, then release
+// the standalone stream on cancel so a reconnect is not 409.
+import { expect } from "@effect/vitest";
+import { Effect } from "effect";
+
+import { scenario } from "../src/scenario";
+import { Cli, RunDir } from "../src/services";
+import { withLocalServer } from "./local-server";
+
+const MCP_POST_HEADERS = {
+  "content-type": "application/json",
+  accept: "application/json, text/event-stream",
+} as const;
+
+scenario(
+  "Local · a quiet MCP GET stream stays alive with an SSE keepalive comment",
+  { timeout: 300_000 },
+  Effect.gen(function* () {
+    const cli = yield* Cli;
+    const runDir = yield* RunDir;
+
+    yield* withLocalServer(cli, runDir, (server) =>
+      Effect.gen(function* () {
+        const auth = { authorization: `Bearer ${server.token}` };
+
+        const init = yield* Effect.promise(() =>
+          fetch(`${server.origin}/mcp`, {
+            method: "POST",
+            headers: { ...MCP_POST_HEADERS, ...auth },
+            body: JSON.stringify({
+              jsonrpc: "2.0",
+              id: 1,
+              method: "initialize",
+              params: {
+                protocolVersion: "2025-06-18",
+                capabilities: {},
+                clientInfo: { name: "e2e-local-sse-keepalive", version: "1.0.0" },
+              },
+            }),
+          }),
+        );
+        expect(init.status).toBe(200);
+        const sessionId = init.headers.get("mcp-session-id");
+        expect(sessionId).toBeTruthy();
+        yield* Effect.promise(() => init.body?.cancel() ?? Promise.resolve());
+
+        const initialized = yield* Effect.promise(() =>
+          fetch(`${server.origin}/mcp`, {
+            method: "POST",
+            headers: { ...MCP_POST_HEADERS, ...auth, "mcp-session-id": sessionId! },
+            body: JSON.stringify({ jsonrpc: "2.0", method: "notifications/initialized" }),
+          }),
+        );
+        expect(initialized.status).toBe(202);
+        yield* Effect.promise(() => initialized.body?.cancel() ?? Promise.resolve());
+
+        const get = yield* Effect.promise(() =>
+          fetch(`${server.origin}/mcp`, {
+            method: "GET",
+            headers: {
+              ...auth,
+              accept: "text/event-stream",
+              "mcp-session-id": sessionId!,
+            },
+            signal: AbortSignal.timeout(5_000),
+          }),
+        );
+        expect(get.status).toBe(200);
+        expect(get.headers.get("content-type")).toContain("text/event-stream");
+        expect(get.headers.get("mcp-session-id")).toBe(sessionId);
+
+        const reader = get.body!.getReader();
+        const first = yield* Effect.promise(() => reader.read());
+        expect(new TextDecoder().decode(first.value)).toBe(": keepalive\n\n");
+        yield* Effect.promise(() => reader.cancel());
+
+        const reconnect = yield* Effect.promise(() =>
+          fetch(`${server.origin}/mcp`, {
+            method: "GET",
+            headers: {
+              ...auth,
+              accept: "text/event-stream",
+              "mcp-session-id": sessionId!,
+            },
+            signal: AbortSignal.timeout(5_000),
+          }),
+        );
+        expect(reconnect.status).toBe(200);
+        expect(reconnect.headers.get("content-type")).toContain("text/event-stream");
+        yield* Effect.promise(() => reconnect.body?.cancel() ?? Promise.resolve());
+      }),
+    );
+  }),
+);

packages/hosts/mcp/package.json

@@ -28,6 +28,10 @@
       "types": "./src/in-memory-session-store.ts",
       "default": "./src/in-memory-session-store.ts"
     },
+    "./sse-heartbeat": {
+      "types": "./src/sse-heartbeat.ts",
+      "default": "./src/sse-heartbeat.ts"
+    },
     "./browser-approval": {
       "types": "./src/browser-approval.ts",
       "default": "./src/browser-approval.ts"

packages/hosts/mcp/src/envelope.test.ts

@@ -225,6 +225,34 @@ it("dispatches toolkit MCP routes with the parsed toolkit resource", async () =>
   });
 });
 
+it("forwards a live GET SSE body without injecting keepalive comments", async () => {
+  const StoreLive = Layer.succeed(McpSessionStore)({
+    dispatch: (): Effect.Effect<McpDispatchResult> =>
+      Effect.succeed(
+        new Response("data: already-kept-alive\n\n", {
+          status: 200,
+          headers: { "content-type": "text/event-stream", "mcp-session-id": "s1" },
+        }),
+      ),
+    dispose: () => Effect.void,
+  });
+
+  const handler = buildHandler(StoreLive, McpErrorReporterNoop);
+  const response = await handler(
+    new Request("https://host.test/mcp", {
+      method: "GET",
+      headers: {
+        authorization: "Bearer x",
+        accept: "text/event-stream",
+        "mcp-session-id": "s1",
+      },
+    }),
+  );
+
+  expect(response.status).toBe(200);
+  expect(await response.text()).toBe("data: already-kept-alive\n\n");
+});
+
 // ---------------------------------------------------------------------------
 // The pre-initialize dispatch guard. Session-less, only `initialize` is servable,
 // and the transport's answer for everything else is a connection-killing HTTP