test(mcplocal): prove chat SSE streams live through the proxy, not buffered
Some checks failed
CI/CD / typecheck (pull_request) Successful in 1m17s
CI/CD / test (pull_request) Successful in 1m27s
CI/CD / lint (pull_request) Successful in 2m55s
CI/CD / smoke (pull_request) Failing after 1m59s
CI/CD / build (pull_request) Successful in 4m33s
CI/CD / publish (pull_request) Has been skipped
Some checks failed
CI/CD / typecheck (pull_request) Successful in 1m17s
CI/CD / test (pull_request) Successful in 1m27s
CI/CD / lint (pull_request) Successful in 2m55s
CI/CD / smoke (pull_request) Failing after 1m59s
CI/CD / build (pull_request) Successful in 4m33s
CI/CD / publish (pull_request) Has been skipped
`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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016dNpnBqyyz9GxfznVcX2sP
This commit is contained in:
@@ -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);
|
||||
|
||||
@@ -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<boolean>((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<string>((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) => {
|
||||
|
||||
@@ -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<void>((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.
|
||||
|
||||
Reference in New Issue
Block a user