src/cli/src/commands/chat.ts

@@ -70,7 +70,7 @@ export function createChatCommand(deps: ChatCommandDeps): Command {
 }
 
 /** What the chat is bound to: a named Agent or a Project. */
-interface ChatSubject {
+export interface ChatSubject {
   kind: 'agent' | 'project';
   name: string;
   /** URL segment, e.g. `agents/reviewer` or `projects/sre` (name url-encoded). */
@@ -97,14 +97,17 @@ function resolveSubject(agent: string | undefined, opts: ChatOpts): ChatSubject
  * `personality` overlay (the project schema rejects unknown fields) and adds
  * `allowSecrets` when requested.
  */
-function chatBody(subject: ChatSubject, message: string, threadId: string | undefined, overrides: Overrides, stream?: boolean): Record<string, unknown> {
+export function chatBody(subject: ChatSubject, message: string, threadId: string | undefined, overrides: Overrides, stream?: boolean): Record<string, unknown> {
   const o: Record<string, unknown> = { ...overrides };
   if (subject.kind === 'project') {
     delete o.personality;
     if (subject.allowSecrets) o.allowSecrets = true;
   }
   const body: Record<string, unknown> = { message, ...o };
-  if (threadId !== undefined) body.threadId = threadId;
+  // Guard the empty string, not just undefined: a turn that dies before its
+  // `final` frame yields no thread id, and sending `threadId: ""` trips mcpd's
+  // min(1) validation — bricking every later message in the REPL with a 400.
+  if (threadId !== undefined && threadId !== '') body.threadId = threadId;
   if (stream === true) body.stream = true;
   return body;
 }
@@ -205,7 +208,11 @@ async function runOneShot(
   const bar = installStatusBar();
   try {
     const finalThread = await streamOnce(deps, subject, message, threadId, overrides, bar);
+    if (finalThread !== undefined) {
       process.stderr.write(`\n(thread: ${finalThread})\n`);
+    } else {
+      process.stderr.write('\n');
+    }
   } finally {
     bar?.teardown();
   }
@@ -262,7 +269,9 @@ async function runRepl(
           const answered = formatAnswered(res.llm, res.model, res.failedOver);
           if (answered !== '') process.stderr.write(`${styleStats(`(${answered})`)}\n`);
         } else {
-          threadId = await streamOnce(deps, subject, line, threadId, overrides, bar);
+          // A failed turn resolves undefined — keep the previous thread (or
+          // none) instead of overwriting it, so the next message still works.
+          threadId = await streamOnce(deps, subject, line, threadId, overrides, bar) ?? threadId;
           process.stdout.write('\n');
         }
       } catch (err) {
@@ -502,15 +511,21 @@ async function chatRequestNonStream(
   });
 }
 
-/** Stream a single chat call. Returns the resolved threadId. */
+/**
-async function streamOnce(
+ * Stream a single chat call. Returns the resolved threadId, or undefined when
+ * the turn never produced a `final` frame (upstream error, early disconnect).
+ * Returning undefined — instead of the old '' — lets callers keep their
+ * previous thread state rather than poisoning the next request with an empty
+ * id that mcpd's validation rejects.
+ */
+export async function streamOnce(
   deps: ChatCommandDeps,
   subject: ChatSubject,
   message: string,
   threadId: string | undefined,
   overrides: Overrides,
   bar: StatusBar | null = null,
-): Promise<string> {
+): Promise<string | undefined> {
   const url = new URL(`${deps.baseUrl}/api/v1/${subject.path}/chat`);
   const body = JSON.stringify(chatBody(subject, message, threadId, overrides, true));
 
@@ -531,7 +546,7 @@ async function streamOnce(
     }
   }
 
-  return new Promise<string>((resolve, reject) => {
+  return new Promise<string | undefined>((resolve, reject) => {
     const driver = url.protocol === 'https:' ? https : http;
     const req = driver.request({
       hostname: url.hostname,
@@ -552,7 +567,7 @@ async function streamOnce(
         return;
       }
       let buf = '';
-      let resolvedThread = threadId ?? '';
+      let resolvedThread: string | undefined = threadId;
       let answered = '';
       res.setEncoding('utf-8');
       res.on('data', (chunk: string) => {

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

@@ -0,0 +1,116 @@
+/**
+ * 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');
+  });
+});