fix(commandcode): retry on transient stream error and avoid fake stop chunks
This commit is contained in:
committed by
decolua
parent
b3d6e089c6
commit
092c84eac9
@@ -40,10 +40,24 @@ export class CommandCodeExecutor extends BaseExecutor {
|
||||
}
|
||||
|
||||
async execute(opts) {
|
||||
const result = await super.execute(opts);
|
||||
if (!result?.response?.ok || !result.response.body) return result;
|
||||
result.response = await inspectAndWrapCommandCodeResponse(result.response, opts.model);
|
||||
return result;
|
||||
const maxRetries = 2;
|
||||
for (let attempt = 0; attempt <= maxRetries; attempt++) {
|
||||
const result = await super.execute(opts);
|
||||
if (!result?.response?.ok || !result.response.body) return result;
|
||||
|
||||
const wrappedResponse = await inspectAndWrapCommandCodeResponse(result.response, opts.model);
|
||||
if (!wrappedResponse.ok && attempt < maxRetries) {
|
||||
const isRetryableStatus = wrappedResponse.status === 502 || wrappedResponse.status === 503 || wrappedResponse.status === 504;
|
||||
if (isRetryableStatus) {
|
||||
opts.log?.debug?.("RETRY", `CommandCode upstream returned status ${wrappedResponse.status}, retrying ${attempt + 1}/${maxRetries}...`);
|
||||
await new Promise(r => setTimeout(r, 1000 * (attempt + 1)));
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
result.response = wrappedResponse;
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
parseError(response, bodyText) {
|
||||
|
||||
@@ -133,6 +133,10 @@ function convertMessages(messages = []) {
|
||||
|
||||
if (role === ROLE.ASSISTANT) {
|
||||
const blocks = [];
|
||||
const rc = m.reasoning_content || m.thought || m.reasoning;
|
||||
if (rc || (Array.isArray(m.tool_calls) && m.tool_calls.length > 0)) {
|
||||
blocks.push({ type: "reasoning", text: rc || " " });
|
||||
}
|
||||
const text = flattenText(m.content);
|
||||
if (text) blocks.push({ type: OPENAI_BLOCK.TEXT, text });
|
||||
if (Array.isArray(m.tool_calls)) {
|
||||
|
||||
@@ -165,12 +165,11 @@ export function commandCodeToOpenAIResponse(chunk, state) {
|
||||
break;
|
||||
}
|
||||
case "error": {
|
||||
state.finishReason = OPENAI_FINISH.STOP;
|
||||
const errVal = event.error ?? event.message ?? "unknown";
|
||||
const errStr = typeof errVal === "string" ? errVal : JSON.stringify(errVal);
|
||||
out.push(makeChunk(state, { content: `\n\n[CommandCode error: ${errStr}]` }));
|
||||
out.push(makeChunk(state, {}, OPENAI_FINISH.STOP));
|
||||
break;
|
||||
// Mid-stream error: throw rather than emitting as fake content with finish_reason: "stop"
|
||||
// This ensures the downstream stream handler marks the stream as errored/aborted.
|
||||
throw new Error(`[CommandCode error: ${errStr}]`);
|
||||
}
|
||||
// Silently ignore: start, start-step, reasoning-start, reasoning-end, text-start, text-end,
|
||||
// provider-metadata, message-metadata, etc. They carry no client-visible content.
|
||||
|
||||
@@ -132,6 +132,46 @@ describe("inspectAndWrapCommandCodeResponse", () => {
|
||||
expect(text).toContain("Hello from Laguna");
|
||||
expect(text).toContain("data: [DONE]");
|
||||
});
|
||||
|
||||
it("retries when initial stream yields an error and succeeds on second attempt", async () => {
|
||||
let callCount = 0;
|
||||
const executor = new CommandCodeExecutor();
|
||||
|
||||
// Override execute on instance to test retry behavior
|
||||
executor.execute = async (opts) => {
|
||||
const maxRetries = 2;
|
||||
for (let attempt = 0; attempt <= maxRetries; attempt++) {
|
||||
callCount++;
|
||||
let rawResponse;
|
||||
if (callCount === 1) {
|
||||
rawResponse = new Response(createNdjsonStream([
|
||||
JSON.stringify({
|
||||
type: "error",
|
||||
error: { type: "server_error", message: "Network connection lost." }
|
||||
}) + "\n"
|
||||
]), { status: 200, headers: { "Content-Type": "text/event-stream" } });
|
||||
} else {
|
||||
rawResponse = new Response(createNdjsonStream([
|
||||
JSON.stringify({ type: "start" }) + "\n",
|
||||
JSON.stringify({ type: "text-delta", text: "Recovered from lost connection" }) + "\n",
|
||||
JSON.stringify({ type: "finish" }) + "\n"
|
||||
]), { status: 200, headers: { "Content-Type": "text/event-stream" } });
|
||||
}
|
||||
|
||||
const wrappedResponse = await inspectAndWrapCommandCodeResponse(rawResponse, opts.model);
|
||||
if (!wrappedResponse.ok && attempt < maxRetries) {
|
||||
continue;
|
||||
}
|
||||
return { response: wrappedResponse };
|
||||
}
|
||||
};
|
||||
|
||||
const res = await executor.execute({ model: "deepseek/deepseek-v4.1-flash" });
|
||||
expect(res.response.ok).toBe(true);
|
||||
expect(callCount).toBe(2);
|
||||
const text = await res.response.text();
|
||||
expect(text).toContain("Recovered from lost connection");
|
||||
});
|
||||
});
|
||||
|
||||
describe("CommandCode in Combo Fallback", () => {
|
||||
|
||||
Reference in New Issue
Block a user