Compare commits
2 Commits
fix/chat-s
...
fix/chat-r
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ac5dee906e | ||
| d4c33baf03 |
@@ -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) => {
|
||||
|
||||
116
src/cli/tests/commands/chat-thread-brick.test.ts
Normal file
116
src/cli/tests/commands/chat-thread-brick.test.ts
Normal 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');
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user