From bbd2195c64a0bcdfd3f41833a670165e31325628 Mon Sep 17 00:00:00 2001 From: Michal Date: Mon, 10 Aug 2026 22:27:21 +0100 Subject: [PATCH] test(mcplocal): prove chat SSE streams live through the proxy, not buffered MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `mcpctl chat reviewer` showed nothing until the turn finished, then dumped the whole answer at once. mcpd streams token deltas and the CLI renders them incrementally — the sole buffering point was mcplocal's catch-all /api/v1/* proxy reading the whole SSE body via res.text() before replying. The previous commit (cherry-picked from feat/agentic-teams) pipes the body through instead; this one adds the cover that was missing: - proxy-long-running.test.ts: a progressive-delivery test in which the stand-in mcpd withholds its final frame until the client has observed the first delta through the proxy. A buffering proxy cannot satisfy that ordering — verified: the test fails in 3s (no hang) against a res.text() proxy and passes against the piped one. The existing SSE test used inject(), which collects the whole body and so passes either way. - agent-chat.smoke.test.ts: a live smoke that posts through mcplocal on localhost:3200 (the path `mcpctl chat` actually takes — every other chat smoke uses --direct and bypasses the proxy entirely) and asserts delta frames arrive spread across the generation window, not in one burst at stream end. Uses its own agent: the shared smoke agent pins replies to a single token, too short to tell live streaming from a buffer dump. Also settles the strict-boolean-expressions lint on the auth-header check the streaming split touched. Local: workspace 2546 passed, proxy suite 10/10, smoke file loads + self-skips. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016dNpnBqyyz9GxfznVcX2sP --- src/mcplocal/src/http/routes/proxy.ts | 4 +- src/mcplocal/tests/proxy-long-running.test.ts | 59 ++++++++++++ .../tests/smoke/agent-chat.smoke.test.ts | 95 +++++++++++++++++++ 3 files changed, 157 insertions(+), 1 deletion(-) diff --git a/src/mcplocal/src/http/routes/proxy.ts b/src/mcplocal/src/http/routes/proxy.ts index d6ff369..9b2bebd 100644 --- a/src/mcplocal/src/http/routes/proxy.ts +++ b/src/mcplocal/src/http/routes/proxy.ts @@ -71,7 +71,9 @@ export function registerProxyRoutes(app: FastifyInstance, client: McpdClient): v // Forward the user's auth token to mcpd so RBAC applies per-user. // If no user token is present, mcpd will use its auth hook to reject. const authHeader = request.headers['authorization'] as string | undefined; - const userToken = authHeader?.startsWith('Bearer ') ? authHeader.slice(7) : undefined; + const userToken = authHeader !== undefined && authHeader.startsWith('Bearer ') + ? authHeader.slice(7) + : undefined; if (isLongRunning(path)) { return proxyStreaming(reply, client, request.method, path, querystring, body, userToken); diff --git a/src/mcplocal/tests/proxy-long-running.test.ts b/src/mcplocal/tests/proxy-long-running.test.ts index 0c5fd04..a793084 100644 --- a/src/mcplocal/tests/proxy-long-running.test.ts +++ b/src/mcplocal/tests/proxy-long-running.test.ts @@ -1,3 +1,5 @@ +import http from 'node:http'; + import Fastify, { type FastifyInstance } from 'fastify'; import { describe, it, expect, afterEach } from 'vitest'; @@ -144,6 +146,63 @@ describe('proxy — long-running route budget', () => { expect(res.body).toContain('[DONE]'); }); + it('delivers each SSE frame while the upstream is still generating', async () => { + // The buffering regression is invisible to the pass-through test above: + // `inject()` collects the whole body, so a proxy that buffers via + // res.text() still passes it. This test proves *progressive* delivery by + // making the upstream withhold its final frame until the client has + // observed the first one. A buffering proxy can never satisfy that + // ordering — the 3s guard resolves the gate so the run fails cleanly + // instead of deadlocking. + let openGate: (seen: boolean) => void = () => {}; + const clientSawFirstFrame = new Promise((r) => { openGate = r; }); + const guard = setTimeout(() => openGate(false), 3_000); + + const base = await startUpstream((a) => { + a.post('/api/v1/agents/:name/chat', async (_req, reply) => { + reply.raw.writeHead(200, { 'Content-Type': 'text/event-stream' }); + reply.raw.write('data: {"type":"text","delta":"live"}\n\n'); + await clientSawFirstFrame; + reply.raw.write('data: {"type":"final"}\n\n'); + reply.raw.write('data: [DONE]\n\n'); + reply.raw.end(); + return reply; + }); + }); + const proxy = await startProxy(base, 50); + await proxy.listen({ port: 0, host: '127.0.0.1' }); + const addr = proxy.server.address(); + if (addr === null || typeof addr === 'string') throw new Error('no address'); + + const body = await new Promise((resolve, reject) => { + const req = http.request({ + hostname: '127.0.0.1', + port: addr.port, + path: '/api/v1/agents/reviewer/chat', + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + }, (res) => { + let acc = ''; + res.setEncoding('utf-8'); + res.on('data', (chunk: string) => { + acc += chunk; + if (acc.includes('"delta":"live"')) openGate(true); + }); + res.on('end', () => resolve(acc)); + res.on('error', reject); + }); + req.on('error', reject); + req.end(JSON.stringify({ message: 'hi', stream: true })); + }); + clearTimeout(guard); + + // The ordering proof: the first frame reached the client while the + // upstream was still holding the stream open. + await expect(clientSawFirstFrame).resolves.toBe(true); + expect(body).toContain('"type":"final"'); + expect(body).toContain('[DONE]'); + }); + it('relays a non-200 status from a streaming route', async () => { const base = await startUpstream((a) => { a.post('/api/v1/agents/:name/chat', async (_req, reply) => { diff --git a/src/mcplocal/tests/smoke/agent-chat.smoke.test.ts b/src/mcplocal/tests/smoke/agent-chat.smoke.test.ts index e445511..3261c43 100644 --- a/src/mcplocal/tests/smoke/agent-chat.smoke.test.ts +++ b/src/mcplocal/tests/smoke/agent-chat.smoke.test.ts @@ -18,8 +18,12 @@ import { describe, it, expect, beforeAll, afterAll } from 'vitest'; import http from 'node:http'; import https from 'node:https'; import { spawnSync, execSync } from 'node:child_process'; +import { existsSync, readFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { homedir } from 'node:os'; const MCPD_URL = process.env.MCPD_URL ?? 'https://mcpctl.ad.itaz.eu'; +const MCPLOCAL_URL = process.env.MCPLOCAL_URL ?? 'http://localhost:3200'; const LLM_URL = process.env.MCPCTL_SMOKE_LLM_URL; const LLM_MODEL = process.env.MCPCTL_SMOKE_LLM_MODEL ?? 'qwen3-thinking'; const LLM_KEY = process.env.MCPCTL_SMOKE_LLM_KEY; @@ -27,6 +31,10 @@ const SUFFIX = Date.now().toString(36); const SECRET_NAME = `smoke-chat-sec-${SUFFIX}`; const LLM_NAME = `smoke-chat-llm-${SUFFIX}`; const AGENT_NAME = `smoke-chat-agent-${SUFFIX}`; +// Dedicated agent for the streaming-timing test: the shared agent's system +// prompt pins the reply to a single token, which is too short to distinguish +// live streaming from an end-of-turn buffer dump. +const STREAM_AGENT_NAME = `smoke-stream-agent-${SUFFIX}`; interface CliResult { code: number; stdout: string; stderr: string } @@ -99,6 +107,7 @@ describe('agent chat smoke (live LLM)', () => { afterAll(() => { if (!liveLlmConfigured || !mcpdUp) return; run(`delete agent ${AGENT_NAME}`); + run(`delete agent ${STREAM_AGENT_NAME}`); run(`delete llm ${LLM_NAME}`); run(`delete secret ${SECRET_NAME}`); }); @@ -139,6 +148,92 @@ describe('agent chat smoke (live LLM)', () => { expect(result.stderr).toMatch(/thread:\s+c[a-z0-9]+/); }); + it('streams progressively THROUGH mcplocal — frames arrive during generation, not in one burst', async () => { + if (!liveLlmConfigured || !mcpdUp) return; + // The regression this pins: mcplocal's /api/v1/* proxy buffered SSE via + // res.text(), so the CLI showed nothing until the turn finished and then + // dumped the whole answer at once. The --direct tests above bypass + // mcplocal entirely and cannot catch that. This one posts to the local + // proxy (the path `mcpctl chat` actually takes) and asserts frames are + // spread across the generation window: with buffering, everything lands + // within a few ms of stream end. + if (!(await healthz(MCPLOCAL_URL))) { + // eslint-disable-next-line no-console + console.warn(`\n ○ mcplocal streaming smoke: skipped — ${MCPLOCAL_URL}/healthz unreachable.\n`); + return; + } + let token = ''; + try { + const credsPath = join(homedir(), '.mcpctl', 'credentials'); + if (existsSync(credsPath)) { + const creds = JSON.parse(readFileSync(credsPath, 'utf-8')) as { token?: string }; + if (creds.token !== undefined) token = creds.token; + } + } catch { /* unauthenticated — the request will 401 and fail loudly */ } + + run(`delete agent ${STREAM_AGENT_NAME}`); + const agent = run([ + `create agent ${STREAM_AGENT_NAME}`, + `--llm ${LLM_NAME}`, + `--description "mcplocal streaming smoke"`, + `--system-prompt "You are a smoke test. Follow the user's instructions exactly."`, + '--default-temperature 0', + '--default-max-tokens 512', + ].join(' ')); + expect(agent.code, agent.stderr).toBe(0); + + const url = new URL(`${MCPLOCAL_URL.replace(/\/$/, '')}/api/v1/agents/${STREAM_AGENT_NAME}/chat`); + const deltaTimes: number[] = []; + let endTime = 0; + let status = 0; + let raw = ''; + + await new Promise((resolve, reject) => { + const req = http.request({ + hostname: url.hostname, + port: url.port || 80, + path: url.pathname, + method: 'POST', + timeout: 120_000, + headers: { + 'Content-Type': 'application/json', + ...(token !== '' ? { Authorization: `Bearer ${token}` } : {}), + }, + }, (res) => { + status = res.statusCode ?? 0; + res.setEncoding('utf-8'); + let buf = ''; + res.on('data', (chunk: string) => { + raw += chunk; + buf += chunk; + let nl: number; + while ((nl = buf.indexOf('\n\n')) !== -1) { + const frame = buf.slice(0, nl); + buf = buf.slice(nl + 2); + if (/"type":"(text|thinking)"/.test(frame)) deltaTimes.push(Date.now()); + } + }); + res.on('end', () => { endTime = Date.now(); resolve(); }); + res.on('error', reject); + }); + req.on('error', reject); + req.on('timeout', () => { req.destroy(); reject(new Error('stream timed out')); }); + req.end(JSON.stringify({ + message: 'Count from 1 to 40, one number per line. No other text.', + stream: true, + max_tokens: 400, + })); + }); + + expect(status, raw.slice(0, 500)).toBe(200); + expect(deltaTimes.length).toBeGreaterThanOrEqual(2); + // The buffering signature: every frame lands in the same final burst as + // stream end. Live streaming puts the first delta well before the end — + // a 40-line generation spans seconds; 300ms is a conservative floor. + const firstDelta = deltaTimes[0]!; + expect(endTime - firstDelta).toBeGreaterThanOrEqual(300); + }, 150_000); + it('streaming `mcpctl chat` emits text deltas', () => { if (!liveLlmConfigured || !mcpdUp) return; // Default mode is streaming. Pipe stdout/stderr separately.