mcpctl/src/cli/tests/commands/chat-thread-brick.test.ts

/**
 * Regression: a failed first turn must not brick the REPL.
 *
 * Observed live: the first message died upstream (anthropic 429) before the
 * stream's `final` frame, streamOnce resolved '' as the thread id, the REPL
 * stored it, and every later message sent `threadId: ""` — which mcpd's
 * `z.string().min(1)` rejects with HTTP 400. The session was permanently
 * stuck: no turn could succeed again, so no `final` frame could ever repair
 * the thread id.
 *
 * The fix has two independent layers, pinned separately below:
 *   1. streamOnce resolves `undefined` (not '') when no `final` frame arrived,
 *      and the REPL keeps its previous thread state on undefined;
 *   2. chatBody never serializes an empty threadId, even if one leaks in.
 */
import http from 'node:http';
import type { AddressInfo } from 'node:net';

import { describe, it, expect, afterEach } from 'vitest';

import { chatBody, streamOnce } from '../../src/commands/chat.js';
import type { ChatCommandDeps, ChatSubject } from '../../src/commands/chat.js';
import type { ApiClient } from '../../src/api-client.js';

const subject: ChatSubject = {
  kind: 'agent',
  name: 'reviewer',
  path: 'agents/reviewer',
  allowSecrets: false,
};

// streamOnce only touches baseUrl + token; the ApiClient is for the
// non-streaming path and never dereferenced here.
function depsFor(baseUrl: string): ChatCommandDeps {
  return { client: null as unknown as ApiClient, baseUrl, log: () => {} };
}

let server: http.Server | null = null;

afterEach(async () => {
  if (server !== null) {
    await new Promise<void>((r) => server!.close(() => r()));
    server = null;
  }
});

/** Serve one SSE response body for any POST, return the base URL. */
async function serveSse(frames: string[]): Promise<string> {
  server = http.createServer((_req, res) => {
    res.writeHead(200, { 'Content-Type': 'text/event-stream' });
    for (const f of frames) res.write(`data: ${f}\n\n`);
    res.end();
  });
  await new Promise<void>((r) => server!.listen(0, '127.0.0.1', r));
  const { port } = server.address() as AddressInfo;
  return `http://127.0.0.1:${String(port)}`;
}

describe('chatBody — threadId serialization', () => {
  it('omits threadId when undefined', () => {
    expect(chatBody(subject, 'hi', undefined, {})).not.toHaveProperty('threadId');
  });

  it('omits threadId when empty — the exact payload that 400s against mcpd', () => {
    expect(chatBody(subject, 'hi', '', {})).not.toHaveProperty('threadId');
  });

  it('includes a real threadId', () => {
    expect(chatBody(subject, 'hi', 'cthread123', {})).toHaveProperty('threadId', 'cthread123');
  });
});

describe('streamOnce — thread id after a failed turn', () => {
  it('resolves undefined when the stream errors before any final frame', async () => {
    const base = await serveSse([
      '{"type":"error","message":"anthropic stream: HTTP 429"}',
      '[DONE]',
    ]);
    const resolved = await streamOnce(depsFor(base), subject, 'hi', undefined, {});
    expect(resolved).toBeUndefined();
  });

  it('keeps the caller-supplied thread when the turn fails mid-conversation', async () => {
    const base = await serveSse([
      '{"type":"error","message":"upstream died"}',
      '[DONE]',
    ]);
    const resolved = await streamOnce(depsFor(base), subject, 'hi', 'cexisting1', {});
    expect(resolved).toBe('cexisting1');
  });

  it('resolves the threadId announced by the final frame', async () => {
    const base = await serveSse([
      '{"type":"text","delta":"pong"}',
      '{"type":"final","threadId":"cfresh42"}',
      '[DONE]',
    ]);
    const resolved = await streamOnce(depsFor(base), subject, 'hi', undefined, {});
    expect(resolved).toBe('cfresh42');
  });

  it('REPL chain: failed turn 1 leaves turn 2 sendable (the brick)', async () => {
    const base = await serveSse([
      '{"type":"error","message":"anthropic stream: HTTP 429"}',
      '[DONE]',
    ]);
    // Mirrors runRepl's assignment: threadId = streamOnce(...) ?? threadId
    let threadId: string | undefined = undefined;
    threadId = (await streamOnce(depsFor(base), subject, 'hi', threadId, {})) ?? threadId;

    // Turn 2's body must be valid for mcpd: no threadId key at all.
    const body = chatBody(subject, 'hi again', threadId, {}, true);
    expect(body).not.toHaveProperty('threadId');
    expect(body).toHaveProperty('message', 'hi again');
  });
});