hosted: run sse-retry on val.town

sse-retry was excluded as single-process only. Two things kept it from
working across isolates:

- The reconnecting GET stays open for as long as the client listens, and
  the fetch bridge buffered every response until end(), so the client
  never received the tool result. The bridge now hands a server-sent
  event stream back at its first write and feeds later writes into it
  (cancelling it emits "close" on the response, as a socket would);
  listenFetch streams bodies too. Because such a stream can outlive the
  handler, dispatch writes the cell through on each event it sends, and
  withServerTiming holds the stream's end until the store flush.
- The GET could reach an isolate that never saw the tool call, and the
  verdict compared performance.now() readings a fresh judging instance
  does not have. The scenario now sets answersFromLog and reads the
  pending tool call, the next event id and the GETs so far from its log,
  echoes the session id the client sends, and, when judging a merged log,
  reads the reconnect delay from the checks' timestamps. In one process
  the log says what memory did, so the CLI runs as before.

Only elicitation-sep1034-client-defaults stays excluded on val.town. A
val.town-shaped test drives the SDK client through one process, two
sharing a store and two that forget every cell after each request, and
fails a client that reconnects early whichever process it reaches.

Author: pcarleton

examples/hosted/fetch-bridge.test.ts

@@ -0,0 +1,54 @@
+import { describe, it, expect } from 'vitest';
+import { toFetchHandler } from './fetch-bridge';
+
+const text = (b: Uint8Array | undefined) => new TextDecoder().decode(b);
+
+describe('fetch bridge', () => {
+  it('hands back a server-sent event stream at its first write, and tells the listener when the client leaves', async () => {
+    let closed = false;
+    let later!: () => void;
+    const handle = toFetchHandler((_req, res) => {
+      res.on('close', () => {
+        closed = true;
+      });
+      res.writeHead(200, { 'content-type': 'text/event-stream' });
+      res.write('data: one\n\n');
+      later = () => res.write('data: two\n\n');
+    });
+    // Resolves although the listener never ends the response.
+    const r = await handle(new Request('http://test/'));
+    expect(r.status).toBe(200);
+    expect(r.headers.get('content-type')).toBe('text/event-stream');
+    const reader = r.body!.getReader();
+    expect(text((await reader.read()).value)).toBe('data: one\n\n');
+    later();
+    expect(text((await reader.read()).value)).toBe('data: two\n\n');
+    await reader.cancel();
+    expect(closed).toBe(true);
+  });
+
+  it('closes the stream when the listener ends it', async () => {
+    const handle = toFetchHandler((_req, res) => {
+      res.setHeader('content-type', 'text/event-stream');
+      res.write('data: one\n\n');
+      setTimeout(() => res.end('data: last\n\n'), 10);
+    });
+    const r = await handle(new Request('http://test/'));
+    expect(await r.text()).toBe('data: one\n\ndata: last\n\n');
+  });
+
+  it('buffers any other response until end()', async () => {
+    let ended = false;
+    const handle = toFetchHandler((_req, res) => {
+      res.writeHead(200, { 'content-type': 'application/json' });
+      res.write('{"a":');
+      setTimeout(() => {
+        ended = true;
+        res.end('1}');
+      }, 10);
+    });
+    const r = await handle(new Request('http://test/'));
+    expect(ended).toBe(true);
+    expect(await r.json()).toEqual({ a: 1 });
+  });
+});

examples/hosted/fetch-bridge.ts

@@ -81,10 +81,6 @@ export function toFetchHandler(
       captureHeaders(h);
       return nodeRes;
     }) as ServerResponse['writeHead'];
-    nodeRes.write = ((c: string | Buffer, enc?: BufferEncoding) => {
-      if (c) chunks.push(typeof c === 'string' ? Buffer.from(c, enc) : c);
-      return true;
-    }) as ServerResponse['write'];
     nodeRes.flushHeaders = () => {};
     Object.defineProperty(nodeRes, 'statusCode', {
       get: () => status,
@@ -93,9 +89,53 @@ export function toFetchHandler(
       }
     });
 
+    // A server-sent event stream can stay open for as long as the client
+    // listens (a GET stream, a response held open), so it is handed back at
+    // its first write and fed as the listener writes; any other response is
+    // buffered until end().
+    let stream: ReadableStreamDefaultController<Uint8Array> | undefined;
+    let cancelled = false;
+    const isEventStream = () =>
+      (headers.get('content-type') ?? '').includes('text/event-stream');
+
     return new Promise<Response>((resolve) => {
-      nodeRes.end = ((c?: string | Buffer, enc?: BufferEncoding) => {
-        if (c) chunks.push(typeof c === 'string' ? Buffer.from(c, enc) : c);
+      const push = (c: string | Uint8Array, enc?: BufferEncoding) => {
+        const buf = typeof c === 'string' ? Buffer.from(c, enc) : c;
+        if (!stream) {
+          chunks.push(Buffer.from(buf));
+          return;
+        }
+        if (!cancelled) stream.enqueue(buf);
+      };
+      nodeRes.write = ((c: string | Uint8Array, enc?: BufferEncoding) => {
+        if (!c) return true;
+        push(c, enc);
+        if (!stream && isEventStream()) {
+          resolve(
+            new Response(
+              new ReadableStream<Uint8Array>({
+                start(controller) {
+                  stream = controller;
+                  for (const b of chunks.splice(0)) controller.enqueue(b);
+                },
+                // The client went away: tell the listener, as a socket would.
+                cancel() {
+                  cancelled = true;
+                  nodeRes.emit('close');
+                }
+              }),
+              { status, headers }
+            )
+          );
+        }
+        return true;
+      }) as ServerResponse['write'];
+      nodeRes.end = ((c?: string | Uint8Array, enc?: BufferEncoding) => {
+        if (c) push(c, enc);
+        if (stream) {
+          if (!cancelled) stream.close();
+          return nodeRes;
+        }
         resolve(
           new Response(chunks.length ? Buffer.concat(chunks) : null, {
             status,

examples/hosted/local-relay.ts

@@ -40,7 +40,17 @@ export function listenFetch(
       })
     );
     res.writeHead(out.status, Object.fromEntries(out.headers));
-    res.end(Buffer.from(await out.arrayBuffer()));
+    // Passed on as it arrives: a server-sent event stream may stay open.
+    if (out.body) {
+      const reader = out.body.getReader();
+      res.on('close', () => void reader.cancel().catch(() => {}));
+      for (;;) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        res.write(value);
+      }
+    }
+    res.end();
   });
   return new Promise((resolve) => server.listen(port, () => resolve(server)));
 }

examples/hosted/valtown-sse-retry.test.ts

@@ -0,0 +1,230 @@
+/**
+ * sse-retry on a val.town-shaped deployment: each hosted app behind the
+ * fetch bridge and withServerTiming, as valtown.ts serves it, with requests
+ * dealt round-robin between apps that share only a store. The tool call and
+ * the GET that resumes it can then reach different processes.
+ */
+
+import { describe, it, expect, afterEach } from 'vitest';
+import type { Server } from 'http';
+import { Client } from '@modelcontextprotocol/sdk/client/index.js';
+import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
+import { CallToolResultSchema } from '@modelcontextprotocol/sdk/types.js';
+import { createHostedApp } from '../../src/hosted/server';
+import { MemoryRunStore } from '../../src/hosted/store';
+import { toFetchHandler } from './fetch-bridge';
+import { listenFetch } from './local-relay';
+import { withServerTiming } from './valtown';
+
+const REV = '2025-11-25';
+
+interface Deployment {
+  origin: string;
+  apps: ReturnType<typeof createHostedApp>[];
+  server: Server;
+}
+
+const open: Deployment[] = [];
+
+async function deploy(
+  processes: number,
+  { store = false, evict = false } = {}
+): Promise<Deployment> {
+  const shared = store ? new MemoryRunStore() : undefined;
+  const apps = Array.from({ length: processes }, () =>
+    createHostedApp({ store: shared })
+  );
+  const handlers = apps.map(({ app, sessions }) => {
+    const handle = withServerTiming(
+      toFetchHandler(app),
+      () => sessions.flush(),
+      0
+    );
+    return async (req: Request) => {
+      const res = await handle(req);
+      // An isolate that drops every cell between requests.
+      if (evict) {
+        for (const run of sessions.list())
+          await sessions.destroy(run.id, false);
+      }
+      return res;
+    };
+  });
+  let turn = 0;
+  const server = await listenFetch(0, (req) =>
+    handlers[turn++ % handlers.length](req)
+  );
+  const dep = {
+    origin: `http://localhost:${(server.address() as { port: number }).port}`,
+    apps,
+    server
+  };
+  open.push(dep);
+  return dep;
+}
+
+afterEach(async () => {
+  for (const dep of open.splice(0)) {
+    for (const { sessions } of dep.apps) await sessions.close();
+    dep.server.closeAllConnections?.();
+    await new Promise<void>((r) => dep.server.close(() => r()));
+  }
+});
+
+async function results(dep: Deployment, cell: string) {
+  for (const { sessions } of dep.apps) await sessions.flush();
+  return fetch(`${dep.origin}/results/${cell}`).then((r) =>
+    r.json()
+  ) as Promise<{
+    verdict: string;
+    checks: Array<{
+      id: string;
+      status: string;
+      details?: Record<string, unknown>;
+    }>;
+  }>;
+}
+
+const statusOf = (checks: Array<{ id: string; status: string }>, id: string) =>
+  checks.find((c) => c.id === id)?.status;
+
+describe.each([
+  { label: 'one process', processes: 1, store: false, evict: false },
+  {
+    label: 'two processes sharing a store',
+    processes: 2,
+    store: true,
+    evict: false
+  },
+  {
+    label: 'two processes that forget every cell after each request',
+    processes: 2,
+    store: true,
+    evict: true
+  }
+])('sse-retry through the val.town entry ($label)', (opts) => {
+  it('passes the SDK client, whichever process the reconnect reaches', async () => {
+    const dep = await deploy(opts.processes, opts);
+    const cell = `sse/${REV}/sse-retry`;
+    // What examples/clients/typescript/sse-retry-test.ts does.
+    const client = new Client(
+      { name: 'sse-retry-test-client', version: '1.0.0' },
+      { capabilities: {} }
+    );
+    const transport = new StreamableHTTPClientTransport(
+      new URL(`${dep.origin}/s/${cell}/mcp`),
+      {
+        reconnectionOptions: {
+          initialReconnectionDelay: 1000,
+          maxReconnectionDelay: 10000,
+          reconnectionDelayGrowFactor: 1.5,
+          maxRetries: 3
+        }
+      }
+    );
+    await client.connect(transport);
+    const result = await client.request(
+      {
+        method: 'tools/call',
+        params: { name: 'test_reconnection', arguments: {} }
+      },
+      CallToolResultSchema
+    );
+    expect(result.content).toEqual([
+      { type: 'text', text: 'Reconnection test completed successfully' }
+    ]);
+    await transport.close();
+
+    const { verdict, checks } = await results(dep, cell);
+    expect(checks.filter((c) => c.status === 'FAILURE')).toEqual([]);
+    expect({
+      verdict,
+      reconnect: statusOf(checks, 'client-sse-graceful-reconnect'),
+      timing: statusOf(checks, 'client-sse-retry-timing'),
+      lastEventId: statusOf(checks, 'client-sse-last-event-id')
+    }).toEqual({
+      verdict: 'pass',
+      reconnect: 'SUCCESS',
+      timing: 'SUCCESS',
+      lastEventId: 'SUCCESS'
+    });
+  }, 20_000);
+});
+
+describe('sse-retry across processes, by hand', () => {
+  it('answers the tool call on the other process and still fails a reconnect that comes too early', async () => {
+    // Requests alternate between the two processes: initialize and the tool
+    // call reach the first, the notification and the GET the second.
+    const dep = await deploy(2, { store: true });
+    const cell = `early/${REV}/sse-retry`;
+    const url = `${dep.origin}/s/${cell}/mcp`;
+    const headers = {
+      'content-type': 'application/json',
+      accept: 'application/json, text/event-stream',
+      'mcp-protocol-version': REV
+    };
+    const init = await fetch(url, {
+      method: 'POST',
+      headers,
+      body: JSON.stringify({
+        jsonrpc: '2.0',
+        id: 1,
+        method: 'initialize',
+        params: {
+          protocolVersion: REV,
+          clientInfo: { name: 'early', version: '0' },
+          capabilities: {}
+        }
+      })
+    });
+    await init.text();
+    const session = { 'mcp-session-id': init.headers.get('mcp-session-id')! };
+    await fetch(url, {
+      method: 'POST',
+      headers: { ...headers, ...session },
+      body: JSON.stringify({
+        jsonrpc: '2.0',
+        method: 'notifications/initialized'
+      })
+    }).then((r) => r.text());
+    // The tool call's stream: a priming event, then closed by the server.
+    const call = await fetch(url, {
+      method: 'POST',
+      headers: { ...headers, ...session },
+      body: JSON.stringify({
+        jsonrpc: '2.0',
+        id: 7,
+        method: 'tools/call',
+        params: { name: 'test_reconnection', arguments: {} }
+      })
+    });
+    expect(await call.text()).toContain('id: event-1');
+    // Reconnect at once, ignoring the 500 ms retry.
+    const get = await fetch(url, {
+      headers: {
+        accept: 'text/event-stream',
+        'mcp-protocol-version': REV,
+        'last-event-id': 'event-1',
+        ...session
+      }
+    });
+    expect(get.headers.get('mcp-session-id')).toBe(session['mcp-session-id']);
+    const reader = get.body!.getReader();
+    let seen = '';
+    while (!seen.includes('Reconnection test completed')) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      seen += new TextDecoder().decode(value);
+    }
+    await reader.cancel();
+    // Answered on the process that never saw the tool call, with its id.
+    expect(seen).toContain('"id":7');
+    expect(seen).toContain('id: event-3');
+
+    const { checks } = await results(dep, cell);
+    const timing = checks.find((c) => c.id === 'client-sse-retry-timing');
+    expect(timing?.status).toBe('FAILURE');
+    expect(timing?.details?.actualDelayMs as number).toBeLessThan(450);
+    expect(statusOf(checks, 'client-sse-last-event-id')).toBe('SUCCESS');
+  });
+});