diff --git a/src/cli/src/commands/chat.ts b/src/cli/src/commands/chat.ts index 628642b..864cf46 100644 --- a/src/cli/src/commands/chat.ts +++ b/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 { +export function chatBody(subject: ChatSubject, message: string, threadId: string | undefined, overrides: Overrides, stream?: boolean): Record { const o: Record = { ...overrides }; if (subject.kind === 'project') { delete o.personality; if (subject.allowSecrets) o.allowSecrets = true; } const body: Record = { 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); - process.stderr.write(`\n(thread: ${finalThread})\n`); + 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 { +): Promise { 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((resolve, reject) => { + return new Promise((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) => { diff --git a/src/cli/tests/commands/chat-thread-brick.test.ts b/src/cli/tests/commands/chat-thread-brick.test.ts new file mode 100644 index 0000000..0e09392 --- /dev/null +++ b/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((r) => server!.close(() => r())); + server = null; + } +}); + +/** Serve one SSE response body for any POST, return the base URL. */ +async function serveSse(frames: string[]): Promise { + 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((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'); + }); +});