Compare commits

..

2 Commits

Author SHA1 Message Date
Michal
ac5dee906e fix(cli): don't brick the chat REPL when the first turn fails upstream
Some checks failed
CI/CD / typecheck (pull_request) Successful in 1m20s
CI/CD / lint (pull_request) Successful in 2m48s
CI/CD / test (pull_request) Successful in 1m24s
CI/CD / build (pull_request) Successful in 2m40s
CI/CD / smoke (pull_request) Failing after 3m46s
CI/CD / publish (pull_request) Has been skipped
Observed live: turn 1 died with an anthropic 429 before the stream's `final`
frame, streamOnce resolved '' as the thread id, the REPL stored it, and every
later message sent `threadId: ""` — rejected by mcpd's z.string().min(1) with
HTTP 400. Permanently stuck: no turn could succeed again, so no `final` frame
could ever repair the id.

Two independent layers:
- streamOnce now resolves `string | undefined` — undefined when no `final`
  frame arrived — and the REPL keeps its previous thread state on undefined
  instead of overwriting it. One-shot mode skips the `(thread: ...)` footer
  when there is none to report.
- chatBody refuses to serialize an empty threadId at all, so even a leaked ''
  can never reach the wire.

Regression cover in chat-thread-brick.test.ts (7 tests), including the full
REPL chain: failed turn 1 → turn 2 body carries no threadId key. The
assertions are the direct inverse of the old behavior, so they fail pre-fix
by construction.

CLI suite 726 passed, lint clean, tsc clean.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016dNpnBqyyz9GxfznVcX2sP
2026-08-10 22:54:53 +01:00
d4c33baf03 Merge pull request 'fix(mcplocal): stream chat SSE through the proxy instead of buffering it' (#109) from fix/chat-sse-streaming into main
Some checks failed
CI/CD / lint (push) Successful in 1m12s
CI/CD / test (push) Successful in 1m27s
CI/CD / typecheck (push) Successful in 3m4s
CI/CD / smoke (push) Failing after 2m0s
CI/CD / build (push) Successful in 2m21s
CI/CD / publish (push) Has been skipped
2026-08-10 21:28:00 +00:00
2 changed files with 141 additions and 10 deletions

View File

@@ -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);
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<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) => {

View File

@@ -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');
});
});