import { appendToolCallArgumentDelta } from "../utils/toolCallArguments.ts"; import { sanitizeErrorMessage } from "../utils/error.ts"; /** * Extract a provider error message from a buffered SSE stream that carries an * error-only chunk (`data: {"error":...}`) and no content chunks. * * Some executors always return `text/event-stream` even on failure (e.g. the * Devin/Windsurf CLI executors emit `data: {"error":{"message":"Devin CLI not * found..."}}`). Those chunks have no `choices`/Claude/Responses content, so the * content parsers (parseSSEToOpenAIResponse etc.) correctly return `null`. Without * this helper the caller would replace the real upstream error with a generic * "Invalid SSE response" 502, swallowing the actionable message (#3324). * * Provider-agnostic: matches any `data:` chunk that has an `error` field but no * `choices` array. The returned message is always run through sanitizeErrorMessage * so stack traces / absolute source paths never leak (Hard Rule #12). Returns * `null` when no error-only chunk is present (so valid-content streams are left * to the normal parsers). */ export function extractSSEErrorMessage(rawSSE: unknown): string | null { const lines = String(rawSSE || "").split("\n"); for (const line of lines) { const trimmed = line.trim(); if (!trimmed.startsWith("data:")) continue; const payload = trimmed.slice(5).trim(); if (!payload || payload === "[DONE]") continue; let parsed: unknown; try { parsed = JSON.parse(payload); } catch { continue; // Ignore malformed lines and keep scanning. } if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) continue; const record = parsed as Record; // A chunk with content (choices) is not an error-only chunk — defer to the // normal content parsers so the valid-SSE path is never short-circuited. if (Array.isArray(record.choices)) continue; const err = record.error; if (err == null) continue; let message = ""; if (typeof err === "string") { message = err; } else if (typeof err === "object" && !Array.isArray(err)) { const errRecord = err as Record; if (typeof errRecord.message === "string") { message = errRecord.message; } else { message = JSON.stringify(err); } } else { message = String(err); } const sanitized = sanitizeErrorMessage(message); if (sanitized) return sanitized; } return null; } /** * Convert OpenAI-style SSE chunks into a single non-streaming JSON response. * Used as a fallback when upstream returns text/event-stream for stream=false. */ function readSSEEvents(rawSSE) { const lines = String(rawSSE || "").split("\n"); const events = []; let currentEvent = ""; let currentData = []; const flush = () => { if (currentData.length === 0) { currentEvent = ""; return; } const payload = currentData.join("\n").trim(); currentData = []; if (!payload || payload === "[DONE]") { currentEvent = ""; return; } try { const data = JSON.parse(payload); if ( currentEvent && data && typeof data === "object" && !Array.isArray(data) && typeof data.type !== "string" ) { data.type = currentEvent; } events.push({ event: currentEvent || undefined, data, }); } catch { // Ignore malformed SSE events and continue best-effort parsing. } currentEvent = ""; }; for (const rawLine of lines) { const line = rawLine.replace(/\r$/, ""); if (line.trim() === "") { flush(); continue; } if (line.startsWith("event:")) { // Some relays omit the blank separator between Claude events. Flush the // previous event before accepting the next event name. if (currentData.length > 0) flush(); currentEvent = line.slice(6).trim(); continue; } if (line.startsWith("data:")) { const dataLine = line.slice(5).trimStart(); if (dataLine.trim() === "[DONE]") { flush(); currentEvent = ""; continue; } currentData.push(dataLine); } } flush(); return events; } function toRecord(value) { return value && typeof value === "object" && !Array.isArray(value) ? value : {}; } function toString(value, fallback = "") { return typeof value === "string" ? value : fallback; } function toNumber(value, fallback = 0) { if (typeof value === "number" && Number.isFinite(value)) return value; if (typeof value === "string" && value.trim().length > 0) { const parsed = Number(value); return Number.isFinite(parsed) ? parsed : fallback; } return fallback; } export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) { const lines = String(rawSSE || "").split("\n"); const chunks = []; let sawChoices = false; for (const line of lines) { const trimmed = line.trim(); if (!trimmed.startsWith("data:")) continue; const payload = trimmed.slice(5).trim(); if (!payload || payload === "[DONE]") continue; try { const parsed = JSON.parse(payload); if (Array.isArray(parsed?.choices)) { sawChoices = true; } chunks.push(parsed); } catch { // Ignore malformed SSE lines and continue best-effort parsing. } } if (chunks.length === 0 || !sawChoices) return null; const first = chunks[0]; const contentParts = []; const reasoningParts = []; type AccumulatedToolCall = { id: string | null; index: number; type: string; function: { name: string; arguments: string }; }; const accumulatedToolCalls = new Map(); let unknownToolCallSeq = 0; let finishReason = "stop"; let usage = null; const getToolCallKey = (toolCall: Record) => { if (Number.isInteger(toolCall?.index)) return `idx:${toolCall.index}`; if (toolCall?.id != null) return `id:${String(toolCall.id)}`; unknownToolCallSeq += 1; return `seq:${unknownToolCallSeq}`; }; for (const chunk of chunks) { const choice = chunk?.choices?.[0]; const delta = choice?.delta || {}; if (typeof delta.content === "string" && delta.content.length > 0) { contentParts.push(delta.content); } if (typeof delta.reasoning_content === "string" && delta.reasoning_content.length > 0) { reasoningParts.push(delta.reasoning_content); } // Normalize `reasoning` alias (NVIDIA kimi-k2.5 etc.) if ( typeof delta.reasoning === "string" && delta.reasoning.length > 0 && !delta.reasoning_content ) { reasoningParts.push(delta.reasoning); } // T18: Accumulate tool calls correctly across streamed chunks if (delta.tool_calls) { for (const tc of delta.tool_calls) { const key = getToolCallKey(tc); const existing = accumulatedToolCalls.get(key); const deltaArgs = typeof tc?.function?.arguments === "string" ? tc.function.arguments : ""; if (!existing) { accumulatedToolCalls.set(key, { id: tc?.id != null ? String(tc.id) : null, index: Number.isInteger(tc?.index) ? tc.index : accumulatedToolCalls.size, type: tc?.type || "function", function: { name: tc?.function?.name || "unknown", arguments: deltaArgs, }, }); } else { existing.id = existing.id || (tc?.id != null ? String(tc.id) : null); if (!Number.isInteger(existing.index) && Number.isInteger(tc?.index)) { existing.index = tc.index; } if (tc?.function?.name && !existing.function?.name) { existing.function = existing.function || {}; existing.function.name = tc.function.name; } existing.function = existing.function || {}; existing.function.arguments = appendToolCallArgumentDelta( existing.function.arguments, deltaArgs ); accumulatedToolCalls.set(key, existing); } } } if (choice?.finish_reason) { finishReason = choice.finish_reason; } if (chunk?.usage && typeof chunk.usage === "object") { usage = chunk.usage; } } const joinedContent = contentParts.length > 0 ? contentParts.join("").trim() : ""; const joinedReasoning = reasoningParts.length > 0 ? reasoningParts.join("").trim() : null; const message: Record = { role: "assistant", content: joinedContent, }; if (joinedReasoning) { message.reasoning_content = joinedReasoning; } const finalToolCalls = [...accumulatedToolCalls.values()].filter(Boolean).sort((a, b) => { const ai = Number.isInteger(a?.index) ? a.index : 0; const bi = Number.isInteger(b?.index) ? b.index : 0; return ai - bi; }); if (finalToolCalls.length > 0) { finishReason = "tool_calls"; // T18 normalization message.tool_calls = finalToolCalls; } const result: Record = { id: first.id != null ? String(first.id) : `chatcmpl-${Date.now()}`, object: "chat.completion", created: first.created || Math.floor(Date.now() / 1000), model: first.model || fallbackModel || "unknown", choices: [ { index: 0, message, finish_reason: finishReason, }, ], }; if (usage) { result.usage = usage; } return result; } /** * Convert Claude-style SSE events into a single non-streaming message object. * Used when Claude-compatible upstreams stream even for stream=false. */ export function parseSSEToClaudeResponse(rawSSE, fallbackModel) { const payloads = readSSEEvents(rawSSE) .map((event) => toRecord(event.data)) .filter((payload) => Object.keys(payload).length > 0); if (payloads.length === 0) return null; const blocks = new Map(); const usage = {}; let messageId = ""; let model = fallbackModel || "claude"; let role = "assistant"; let stopReason = "end_turn"; let stopSequence = null; let sawClaudeEvent = false; const mergeUsage = (incoming) => { const usageRecord = toRecord(incoming); for (const [key, value] of Object.entries(usageRecord)) { if (typeof value === "number" && Number.isFinite(value)) { usage[key] = value; } else if (value && typeof value === "object" && !Array.isArray(value)) { usage[key] = { ...toRecord(usage[key]), ...toRecord(value) }; } else if (typeof value === "string" && value.trim().length > 0) { usage[key] = value; } } }; const tryParseJson = (raw) => { try { return JSON.parse(raw); } catch { return raw; } }; for (const payload of payloads) { const eventType = toString(payload.type); if (eventType === "message_start") { sawClaudeEvent = true; const message = toRecord(payload.message); messageId = toString(message.id, messageId || `msg_${Date.now()}`); model = toString(message.model, model); role = toString(message.role, role); mergeUsage(message.usage); continue; } if (eventType === "content_block_start") { sawClaudeEvent = true; const index = toNumber(payload.index, blocks.size); const contentBlock = toRecord(payload.content_block); const blockType = toString(contentBlock.type); if (blockType === "thinking") { blocks.set(index, { type: "thinking", index, thinking: toString(contentBlock.thinking), signature: toString(contentBlock.signature) || undefined, }); } else if (blockType === "tool_use") { blocks.set(index, { type: "tool_use", index, id: toString(contentBlock.id, `toolu_${Date.now()}_${index}`), name: toString(contentBlock.name), input: contentBlock.input ?? {}, inputJson: "", }); } else { blocks.set(index, { type: "text", index, text: toString(contentBlock.text), }); } continue; } if (eventType === "content_block_delta") { sawClaudeEvent = true; const index = toNumber(payload.index, 0); const delta = toRecord(payload.delta); const deltaType = toString(delta.type); const existing = blocks.get(index); if (deltaType === "input_json_delta") { const toolUse = existing && existing.type === "tool_use" ? existing : { type: "tool_use", index, id: `toolu_${Date.now()}_${index}`, name: "", input: {}, inputJson: "", }; toolUse.inputJson += toString(delta.partial_json); blocks.set(index, toolUse); continue; } const isThinkingDelta = deltaType === "thinking_delta" || typeof delta.thinking === "string"; const isSignatureDelta = deltaType === "signature_delta" || typeof delta.signature === "string"; if (isThinkingDelta || isSignatureDelta) { const thinking = existing && existing.type === "thinking" ? existing : { type: "thinking", index, thinking: "", signature: undefined }; if (isThinkingDelta) thinking.thinking += toString(delta.thinking); const signature = toString(delta.signature); if (signature) thinking.signature = `${thinking.signature || ""}${signature}`; blocks.set(index, thinking); continue; } const textBlock = existing && existing.type === "text" ? existing : { type: "text", index, text: "", }; textBlock.text += toString(delta.text); blocks.set(index, textBlock); continue; } if (eventType === "message_delta") { sawClaudeEvent = true; const delta = toRecord(payload.delta); stopReason = toString(delta.stop_reason, stopReason); stopSequence = typeof delta.stop_sequence === "string" ? String(delta.stop_sequence) : stopSequence; mergeUsage(payload.usage); continue; } mergeUsage(payload.usage); } if (!sawClaudeEvent) return null; const content = []; for (const block of [...blocks.values()].sort((a, b) => a.index - b.index)) { if (block.type === "text") { if (block.text) content.push({ type: "text", text: block.text }); continue; } if (block.type === "thinking") { const hasSignature = typeof block.signature === "string" && block.signature.length > 0; if (block.thinking || hasSignature) { content.push({ type: "thinking", thinking: block.thinking || "", ...(hasSignature ? { signature: block.signature } : {}), }); } continue; } const input = block.inputJson.trim().length > 0 ? tryParseJson(block.inputJson) : block.input; content.push({ type: "tool_use", id: block.id, name: block.name, input }); } return { id: messageId || `msg_${Date.now()}`, type: "message", role, model, content, stop_reason: stopReason, ...(stopSequence ? { stop_sequence: stopSequence } : {}), ...(Object.keys(usage).length > 0 ? { usage } : {}), }; } /** * Convert Responses API SSE events into a single non-streaming response object. * Expects events such as response.created / response.in_progress / response.completed. */ const RESPONSES_TERMINAL_EVENT_TYPES = new Set([ "response.completed", "response.done", "response.cancelled", "response.canceled", "response.failed", "response.incomplete", ]); function toOutputIndex(value) { if (typeof value === "number" && Number.isInteger(value)) return value; if (typeof value === "string" && value.trim().length > 0) { const parsed = Number(value); if (Number.isInteger(parsed)) return parsed; } return null; } function toIdString(value) { return value === null || value === undefined ? "" : String(value); } function cloneResponseItem(item) { const record = toRecord(item); return { ...record, id: record.id != null ? String(record.id) : record.id, call_id: record.call_id != null ? String(record.call_id) : record.call_id, ...(Array.isArray(record.content) ? { content: record.content.map((contentPart) => { const part = toRecord(contentPart); return { ...part }; }), } : {}), ...(Array.isArray(record.summary) ? { summary: record.summary.map((summaryPart) => { const part = toRecord(summaryPart); return { ...part }; }), } : {}), }; } function ensureResponsesMessageItem(outputItems, outputIndex) { const existing = outputItems.get(outputIndex); if (existing?.type === "message") return existing; const next = { ...(existing && typeof existing === "object" ? existing : {}), id: existing?.id != null ? String(existing.id) : `msg_${Date.now()}_${outputIndex}`, type: "message", role: "assistant", content: Array.isArray(existing?.content) ? existing.content.map((contentPart) => ({ ...toRecord(contentPart) })) : [{ type: "output_text", annotations: [], text: "" }], }; if (next.content.length === 0) { next.content.push({ type: "output_text", annotations: [], text: "" }); } outputItems.set(outputIndex, next); return next; } function ensureResponsesReasoningItem(outputItems, outputIndex, itemId) { const existing = outputItems.get(outputIndex); if (existing?.type === "reasoning") return existing; const next = { ...(existing && typeof existing === "object" ? existing : {}), id: itemId || (existing?.id != null ? String(existing.id) : null) || `rs_${Date.now()}_${outputIndex}`, type: "reasoning", summary: Array.isArray(existing?.summary) ? existing.summary.map((summaryPart) => ({ ...toRecord(summaryPart) })) : [{ type: "summary_text", text: "" }], }; if (next.summary.length === 0) { next.summary.push({ type: "summary_text", text: "" }); } outputItems.set(outputIndex, next); return next; } function ensureResponsesFunctionCallItem(outputItems, outputIndex, itemId, callId, name) { const existing = outputItems.get(outputIndex); const normalizedItemId = toIdString(itemId); const normalizedCallId = toIdString(callId); const existingId = existing?.id != null ? String(existing.id) : ""; const existingCallId = existing?.call_id != null ? String(existing.call_id) : ""; if (existing?.type === "function_call") { if (existing.call_id != null) existing.call_id = String(existing.call_id); if (existing.id != null) existing.id = String(existing.id); if (normalizedCallId && !existing.call_id) existing.call_id = normalizedCallId; if (name && !existing.name) existing.name = name; if (normalizedItemId && !existing.id) existing.id = normalizedItemId; return existing; } const next = { ...(existing && typeof existing === "object" ? existing : {}), id: normalizedItemId || existingId || `fc_${normalizedCallId || `${Date.now()}_${outputIndex}`}`, type: "function_call", call_id: normalizedCallId || existingCallId || "", name: name || existing?.name || "", arguments: typeof existing?.arguments === "string" ? existing.arguments : "", }; outputItems.set(outputIndex, next); return next; } function mergeResponseItems(existing, incoming) { const next = cloneResponseItem(incoming); if (!existing || typeof existing !== "object") return next; return { ...existing, ...next, ...(Array.isArray(next.content) ? { content: next.content, } : {}), ...(Array.isArray(next.summary) ? { summary: next.summary, } : {}), }; } export function parseSSEToResponsesOutput(rawSSE, fallbackModel) { const lines = String(rawSSE || "").split("\n"); const events = []; for (const line of lines) { const trimmed = line.trim(); if (!trimmed.startsWith("data:")) continue; const payload = trimmed.slice(5).trim(); if (!payload || payload === "[DONE]") continue; try { const parsed = JSON.parse(payload); const record = toRecord(parsed); if (Object.keys(record).length > 0) { events.push(record); } } catch { // Ignore malformed lines and continue best-effort parsing. } } if (events.length === 0) return null; let terminalResponse = null; let terminalEventType = ""; let latestResponse = null; const outputItems = new Map(); for (const evt of events) { const eventType = toString(evt?.type); const outputIndex = toOutputIndex(evt?.output_index); const item = toRecord(evt?.item); if (outputIndex !== null && eventType === "response.output_item.added") { outputItems.set(outputIndex, cloneResponseItem(item)); } if (outputIndex !== null && eventType === "response.output_item.done") { const existing = outputItems.get(outputIndex); outputItems.set(outputIndex, mergeResponseItems(existing, item)); } if (outputIndex !== null && eventType === "response.output_text.delta") { const messageItem = ensureResponsesMessageItem(outputItems, outputIndex); const content = Array.isArray(messageItem.content) ? messageItem.content : []; const firstPart = content.length > 0 ? { ...toRecord(content[0]) } : { type: "output_text", annotations: [] }; firstPart.type = firstPart.type || "output_text"; firstPart.annotations = Array.isArray(firstPart.annotations) ? firstPart.annotations : []; firstPart.text = `${toString(firstPart.text)}${toString(evt.delta)}`; content[0] = firstPart; messageItem.content = content; } if (outputIndex !== null && eventType === "response.output_text.done") { const messageItem = ensureResponsesMessageItem(outputItems, outputIndex); const content = Array.isArray(messageItem.content) ? messageItem.content : []; const firstPart = content.length > 0 ? { ...toRecord(content[0]) } : { type: "output_text", annotations: [] }; firstPart.type = firstPart.type || "output_text"; firstPart.annotations = Array.isArray(firstPart.annotations) ? firstPart.annotations : []; firstPart.text = toString(evt.text, toString(firstPart.text)); content[0] = firstPart; messageItem.content = content; } if (outputIndex !== null && eventType === "response.reasoning_summary_text.delta") { const reasoningItem = ensureResponsesReasoningItem( outputItems, outputIndex, toIdString(evt.item_id) ); const summary = Array.isArray(reasoningItem.summary) ? reasoningItem.summary : []; const firstPart = summary.length > 0 ? { ...toRecord(summary[0]) } : { type: "summary_text", text: "" }; firstPart.type = firstPart.type || "summary_text"; firstPart.text = `${toString(firstPart.text)}${toString(evt.delta)}`; summary[0] = firstPart; reasoningItem.summary = summary; } if (outputIndex !== null && eventType === "response.reasoning_summary_text.done") { const reasoningItem = ensureResponsesReasoningItem( outputItems, outputIndex, toIdString(evt.item_id) ); const summary = Array.isArray(reasoningItem.summary) ? reasoningItem.summary : []; const firstPart = summary.length > 0 ? { ...toRecord(summary[0]) } : { type: "summary_text", text: "" }; firstPart.type = firstPart.type || "summary_text"; firstPart.text = toString(evt.text, toString(firstPart.text)); summary[0] = firstPart; reasoningItem.summary = summary; } if (outputIndex !== null && eventType === "response.function_call_arguments.delta") { const functionCallItem = ensureResponsesFunctionCallItem( outputItems, outputIndex, toIdString(evt.item_id), "", "" ); functionCallItem.arguments = `${toString(functionCallItem.arguments)}${toString(evt.delta)}`; } if (outputIndex !== null && eventType === "response.function_call_arguments.done") { const functionCallItem = ensureResponsesFunctionCallItem( outputItems, outputIndex, toIdString(evt.item_id), "", "" ); functionCallItem.arguments = toString(evt.arguments, toString(functionCallItem.arguments)); } if (RESPONSES_TERMINAL_EVENT_TYPES.has(eventType) && evt.response) { terminalResponse = evt.response; terminalEventType = eventType; } if (evt?.response && typeof evt.response === "object") { latestResponse = evt.response; } else if (evt?.object === "response") { latestResponse = evt; } } const picked = terminalResponse || latestResponse; if (!picked || typeof picked !== "object") return null; const reconstructedOutput = [...outputItems.entries()] .sort((a, b) => a[0] - b[0]) .map(([, item]) => item) .filter((item) => item && typeof item === "object"); const pickedOutput = Array.isArray(picked.output) ? picked.output : []; // #3948 — A Responses-API terminal snapshot (`response.completed`) can carry a // non-empty `output` that LACKS the assistant message item (e.g. only a // `reasoning` item) even though the streamed `output_text` deltas reconstructed // a full message. Preferring such a textless terminal output drops the // assistant text → empty content on `stream:false` (n8n combo). When the // terminal output has no message item but the reconstructed delta output does, // use the reconstructed output (a superset carrying the message). The terminal // snapshot still wins whenever it already contains the message item. const outputHasMessage = (items: unknown[]) => items.some((item) => toRecord(item).type === "message"); const chosenOutput = pickedOutput.length > 0 && !outputHasMessage(pickedOutput) && outputHasMessage(reconstructedOutput) ? reconstructedOutput : pickedOutput.length > 0 ? pickedOutput : reconstructedOutput; const statusFallback = terminalEventType === "response.cancelled" ? "cancelled" : terminalEventType === "response.canceled" ? "canceled" : terminalEventType === "response.failed" ? "failed" : terminalEventType === "response.incomplete" ? "incomplete" : terminalResponse ? "completed" : "in_progress"; return { id: picked.id != null ? String(picked.id) : `resp_${Date.now()}`, object: picked.object || "response", model: picked.model || fallbackModel || "unknown", output: chosenOutput.map((item) => { const record = toRecord(item); return { ...record, id: record.id != null ? String(record.id) : record.id, call_id: record.call_id != null ? String(record.call_id) : record.call_id, }; }), usage: picked.usage || null, status: picked.status || statusFallback, created_at: picked.created_at || Math.floor(Date.now() / 1000), metadata: picked.metadata || {}, }; }