mcpctl/src/mcplocal/tests/proxy-long-running.test.ts

import http from 'node:http';

import Fastify, { type FastifyInstance } from 'fastify';
import { describe, it, expect, afterEach } from 'vitest';

import {
  McpdClient,
  UpstreamTimeoutError,
  ConnectionError,
  LONG_RUNNING_TIMEOUT_MS,
  DEFAULT_TIMEOUT_MS,
} from '../src/http/mcpd-client.js';
import { registerProxyRoutes } from '../src/http/routes/proxy.js';

/**
 * Regression cover for the 30s proxy timeout that made `mcpctl chat` fail with
 * a misleading "Cannot reach mcpd daemon" 503 while mcpd was answering
 * /healthz in 32ms.
 *
 * Three separate defects are pinned here:
 *   1. chat routes inherited the 30s CRUD budget, so any turn longer than 30s
 *      failed — and an agent turn is a tool-use loop that routinely exceeds it;
 *   2. a timeout was reported as a connection failure, sending diagnosis after
 *      a network fault that did not exist;
 *   3. SSE was buffered through res.text(), so streaming never reached the
 *      client even when the turn finished in time.
 */
let app: FastifyInstance | null = null;
let upstream: FastifyInstance | null = null;

afterEach(async () => {
  if (app) { await app.close(); app = null; }
  if (upstream) { await upstream.close(); upstream = null; }
});

/** A stand-in mcpd. Returns its base URL. */
async function startUpstream(register: (a: FastifyInstance) => void): Promise<string> {
  upstream = Fastify();
  register(upstream);
  await upstream.listen({ port: 0, host: '127.0.0.1' });
  const addr = upstream.server.address();
  if (addr === null || typeof addr === 'string') throw new Error('no address');
  return `http://127.0.0.1:${String(addr.port)}`;
}

async function startProxy(baseUrl: string, timeoutMs?: number): Promise<FastifyInstance> {
  app = Fastify();
  registerProxyRoutes(app, new McpdClient(baseUrl, 'test-token', {}, timeoutMs));
  await app.ready();
  return app;
}

describe('proxy — long-running route budget', () => {
  it('gives chat routes the long budget, not the 30s CRUD default', () => {
    // The constants themselves are the contract: a 30s cap on an agent turn is
    // a guaranteed failure, not a safety net.
    expect(DEFAULT_TIMEOUT_MS).toBe(30_000);
    expect(LONG_RUNNING_TIMEOUT_MS).toBeGreaterThanOrEqual(600_000);
  });

  it('does not abort an agent chat that outlives the CRUD budget', async () => {
    const base = await startUpstream((a) => {
      a.post('/api/v1/agents/:name/chat', async () => {
        // Longer than the (deliberately tiny) CRUD budget below. Before the
        // fix this inherited that budget and 503'd.
        await new Promise((r) => setTimeout(r, 250));
        return { answer: 'pong' };
      });
    });
    // CRUD budget of 50ms — a chat route must NOT inherit it.
    const proxy = await startProxy(base, 50);

    const res = await proxy.inject({
      method: 'POST',
      url: '/api/v1/agents/reviewer/chat',
      payload: { message: 'hi' },
    });

    expect(res.statusCode).toBe(200);
    expect(res.json()).toEqual({ answer: 'pong' });
  });

  it('still applies the short budget to ordinary CRUD routes', async () => {
    const base = await startUpstream((a) => {
      a.get('/api/v1/servers', async () => {
        await new Promise((r) => setTimeout(r, 300));
        return [];
      });
    });
    const proxy = await startProxy(base, 50);

    const res = await proxy.inject({ method: 'GET', url: '/api/v1/servers' });
    // Times out — and is now reported honestly as a timeout, not a connection fault.
    expect(res.statusCode).toBe(504);
    expect(res.json().error).toBe('upstream_timeout');
  });

  it('reports a timeout as 504, never as "cannot reach mcpd"', async () => {
    const base = await startUpstream((a) => {
      a.get('/api/v1/servers', async () => {
        await new Promise((r) => setTimeout(r, 300));
        return [];
      });
    });
    const proxy = await startProxy(base, 50);

    const res = await proxy.inject({ method: 'GET', url: '/api/v1/servers' });
    const body = res.json();
    expect(body.message).toMatch(/did not respond within/);
    expect(body.message).not.toMatch(/Cannot reach mcpd/);
    expect(body.message).toMatch(/reachable/);
  });

  it('streams SSE through instead of buffering it', async () => {
    const base = await startUpstream((a) => {
      a.post('/api/v1/agents/:name/chat', async (_req, reply) => {
        reply.raw.writeHead(200, {
          'Content-Type': 'text/event-stream',
          'Cache-Control': 'no-cache',
          'X-Accel-Buffering': 'no',
        });
        reply.raw.write('data: {"type":"text","delta":"po"}\n\n');
        reply.raw.write('data: {"type":"text","delta":"ng"}\n\n');
        reply.raw.write('data: [DONE]\n\n');
        reply.raw.end();
        return reply;
      });
    });
    const proxy = await startProxy(base, 50);

    const res = await proxy.inject({
      method: 'POST',
      url: '/api/v1/agents/reviewer/chat',
      payload: { message: 'hi', stream: true },
    });

    expect(res.statusCode).toBe(200);
    // Content-type must survive — a client that gets application/json will not
    // parse the event stream.
    expect(res.headers['content-type']).toMatch(/text\/event-stream/);
    // x-accel-buffering=no must survive too, or intermediaries re-buffer the
    // stream and reintroduce the stall.
    expect(res.headers['x-accel-buffering']).toBe('no');
    expect(res.body).toContain('"delta":"po"');
    expect(res.body).toContain('"delta":"ng"');
    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) => {
        return reply.code(404).send({ error: 'Agent not found' });
      });
    });
    const proxy = await startProxy(base, 50);

    const res = await proxy.inject({
      method: 'POST',
      url: '/api/v1/agents/ghost/chat',
      payload: { message: 'hi' },
    });
    expect(res.statusCode).toBe(404);
    expect(res.body).toContain('Agent not found');
  });

  it('still reports a genuinely unreachable daemon as 503', async () => {
    // Port 1 is reserved and refuses instantly.
    const proxy = await startProxy('http://127.0.0.1:1', 500);
    const res = await proxy.inject({ method: 'GET', url: '/api/v1/servers' });

    expect(res.statusCode).toBe(503);
    expect(res.json().error).toBe('service_unavailable');
  });

  it('propagates 401 from a streaming route so login guidance still fires', async () => {
    const base = await startUpstream((a) => {
      a.post('/api/v1/agents/:name/chat', async (_req, reply) => reply.code(401).send({}));
    });
    const proxy = await startProxy(base, 50);

    const res = await proxy.inject({
      method: 'POST',
      url: '/api/v1/agents/reviewer/chat',
      payload: { message: 'hi' },
    });
    expect(res.statusCode).toBe(401);
    expect(res.json().message).toMatch(/mcpctl login/);
  });
});

describe('error taxonomy', () => {
  it('keeps timeout and unreachable as distinct types', () => {
    const timeout = new UpstreamTimeoutError('http://mcpd', 30_000);
    expect(timeout).not.toBeInstanceOf(ConnectionError);
    expect(timeout.timeoutMs).toBe(30_000);
    expect(timeout.message).toMatch(/did not respond within 30000ms/);
  });
});