diff --git a/docs-site/src/content/docs/reference/adapters.md b/docs-site/src/content/docs/reference/adapters.md index 6a9251504cb..3b0e9d3cc04 100644 --- a/docs-site/src/content/docs/reference/adapters.md +++ b/docs-site/src/content/docs/reference/adapters.md @@ -89,6 +89,11 @@ transport; it does not infer subscription attribution from the inbound protocol. Adjacent text blocks share inline-tag parsing; typed thinking resets it. Unsupported or malformed array blocks return an upstream error. Other hosts keep their existing behavior. This does not add Mistral signature replay. +- Native `/v1/chat/completions` passthrough to `https://api.mistral.ai/v1` (key auth) + converts the same arrays into Chat strings: thinking arrives as `reasoning_content` + and text as `content`, streamed in source order. A non-streaming reply joins each + into `message.reasoning_content` and `message.content`. Unsupported blocks return + an upstream error. - Suppresses bare `` text when it duplicates a structured call, and collapses two immediately adjacent identical blocks when exactly one structured call agrees with their function and input. A doubled `input` is reduced to one copy, joined either directly or by one newline, diff --git a/scripts/test-layout/layout-chat.json b/scripts/test-layout/layout-chat.json index 602fafabfc1..791372006bf 100644 --- a/scripts/test-layout/layout-chat.json +++ b/scripts/test-layout/layout-chat.json @@ -8,5 +8,6 @@ "client-error-body.test.ts": "lib", "chat-error-body-normalization.test.ts": "responses", "chat-error-collector-verdicts.test.ts": "responses", - "server-chat-error-body-evidence.test.ts": "server" + "server-chat-error-body-evidence.test.ts": "server", + "chat-native-mistral-content.test.ts": "responses" } diff --git a/src/server/chat-native-sse.ts b/src/server/chat-native-sse.ts index 74b5017f8af..75e4e217b6f 100644 --- a/src/server/chat-native-sse.ts +++ b/src/server/chat-native-sse.ts @@ -1,3 +1,4 @@ +import { chatContentEvents } from "../adapters/openai-chat/response-events"; import { ownedChatErrorPayload, canonicalSseErrorBlock } from "../lib/client-error-body"; import { legacyFailureVerdict, type FailureVerdictSink } from "../chat/failure-verdict"; import { chatCompletionsErrorBody, type ChatCompletionsCollectorSource } from "../chat/outbound"; @@ -114,6 +115,82 @@ function normalizedChunk(value: Rec, requestedModel: string): Rec { }; } +type ChatStringField = "content" | "reasoning_content"; + +/** Mistral typed blocks as Chat string runs in source order; null when a block is unsupported. */ +function mistralContentRuns(content: unknown[], baseUrl: string): { field: ChatStringField; text: string }[] | null { + const runs: { field: ChatStringField; text: string }[] = []; + for (const event of chatContentEvents(content, baseUrl)) { + if (event.type === "error") return null; + const field = event.type === "text_delta" ? "content" : "reasoning_content"; + const last = runs.at(-1); + if (last?.field === field) last.text += event.text; + else runs.push({ field, text: event.text }); + } + return runs; +} + +/** + * Mistral reasoning models stream `delta.content` as typed `thinking`/`text` blocks, while native + * Chat clients parse strings. Each run becomes its own chunk in source order, thinking as + * `reasoning_content` and text as `content`. `role` rides the first chunk; the last keeps the + * remaining delta fields, `finish_reason` and `usage`. Every emitted choice carries its index. + * Chunks without typed content pass through unchanged. + */ +export function mistralTypedContentChunks(value: Rec, baseUrl: string): Rec[] | null { + const choices = Array.isArray(value.choices) ? value.choices : []; + if (!choices.some(choice => isRec(choice) && isRec(choice.delta) && Array.isArray(choice.delta.content))) return [value]; + const { choices: _choices, usage, ...chunk } = value; + const split: unknown[] = []; + for (const [position, choice] of choices.entries()) { + if (!isRec(choice) || !isRec(choice.delta) || !Array.isArray(choice.delta.content)) { + split.push(isRec(choice) && typeof choice.index !== "number" ? { ...choice, index: position } : choice); + continue; + } + const runs = mistralContentRuns(choice.delta.content, baseUrl); + if (!runs) return null; + const { content: _content, role, ...delta } = choice.delta; + const index = typeof choice.index === "number" ? choice.index : position; + const first = role === undefined ? {} : { role }; + const last = runs.pop() ?? { field: "content", text: "" }; + runs.forEach((run, ordinal) => split.push({ + index, delta: { ...(ordinal === 0 ? first : {}), [run.field]: run.text }, finish_reason: null, + })); + const prior = typeof delta[last.field] === "string" ? delta[last.field] as string : ""; + split.push({ ...choice, index, delta: { ...(runs.length === 0 ? first : {}), ...delta, [last.field]: prior + last.text } }); + } + return split.map((choice, position) => ({ + ...chunk, + choices: [choice], + ...(position === split.length - 1 && usage !== undefined ? { usage } : {}), + })); +} + +/** Buffered form of {@link mistralTypedContentChunks}: text joins `content`, thinking joins `reasoning_content`. */ +export function mistralTypedContentCompletion(value: Rec, baseUrl: string): Rec | null { + if (!Array.isArray(value.choices)) return value; + let changed = false; + const choices: unknown[] = []; + for (const choice of value.choices) { + if (!isRec(choice) || !isRec(choice.message) || !Array.isArray(choice.message.content)) { + choices.push(choice); + continue; + } + const runs = mistralContentRuns(choice.message.content, baseUrl); + if (!runs) return null; + const joined = (field: ChatStringField) => runs.filter(run => run.field === field).map(run => run.text).join(""); + const reasoning = joined("reasoning_content"); + const prior = typeof choice.message.reasoning_content === "string" ? choice.message.reasoning_content : ""; + choices.push({ + ...choice, + // `null` for a thinking- or tool-only reply, matching the streamed collector's message. + message: { ...choice.message, content: joined("content") || null, ...(reasoning ? { reasoning_content: prior + reasoning } : {}) }, + }); + changed = true; + } + return changed ? { ...value, choices } : value; +} + export function jsonCompletionSse(value: Rec, requestedModel: string, budget?: TranslatorBudget): string { const id = typeof value.id === "string" ? value.id : `chatcmpl-${crypto.randomUUID().replace(/-/g, "").slice(0, 24)}`; const created = typeof value.created === "number" ? value.created : Math.floor(Date.now() / 1000); @@ -188,6 +265,8 @@ interface NativeChatSseOptions { * `stallTimeoutSec` then resolves to disabled so a slow local model is not cut mid-turn. */ localUpstream?: boolean; + /** Canonical Mistral Chat base URL; set only for that destination, it converts typed content blocks. */ + mistralBaseUrl?: string; clientDeclaredEmptyToolCatalog?: boolean; declaredLegacyFunctionNames?: ReadonlySet; onFirstOutput?: () => void; @@ -397,6 +476,13 @@ export function nativeChatSse( fail(controller, UNDECLARED_CHAT_TOOL_MESSAGE, UNDECLARED_TOOL_CALL_ERROR_CODE); return; } + const normalized = normalizedChunk(parsed, options.requestedModel); + // Replay capture above keeps reading the upstream chunk; only the client copy is converted. + const clientChunks = options.mistralBaseUrl ? mistralTypedContentChunks(normalized, options.mistralBaseUrl) : [normalized]; + if (!clientChunks) { + fail(controller, "unsupported Mistral content block", "upstream_sse_invalid"); + return; + } const choices = Array.isArray(parsed.choices) ? parsed.choices : []; const finished = choices.some(choice => isRec(choice) && typeof choice.finish_reason === "string" && choice.finish_reason.length > 0); if (hasPendingToolName() && choices.some((choice, position) => isRec(choice) @@ -406,7 +492,7 @@ export function nativeChatSse( return; } if (finished) sawFinish = true; - const hasText = choices.some(choice => { + const hasText = clientChunks.flatMap(chunk => Array.isArray(chunk.choices) ? chunk.choices : []).some(choice => { if (!isRec(choice) || !isRec(choice.delta)) return false; const delta = choice.delta; return (typeof delta.content === "string" && delta.content.length > 0) @@ -445,10 +531,12 @@ export function nativeChatSse( options.collectorSource?.retain(JSON.stringify(payload), parsed); relay(controller, (eventName === "error" ? "event: error\n" : "") + canonicalSseErrorBlock(payload)); } else { - const normalized = normalizedChunk(parsed, options.requestedModel); - const choiceError = choices.some(choice => isRec(choice) && choice.error !== undefined && choice.error !== null); - relay(controller, choiceError ? canonicalSseErrorBlock(normalizeChatChoiceErrors(normalized)) - : replaceSseDataPayload(block, JSON.stringify(normalized)) + delimiter); + for (const chunk of clientChunks) { + const chunkChoices = Array.isArray(chunk.choices) ? chunk.choices : []; + const choiceError = chunkChoices.some(choice => isRec(choice) && choice.error !== undefined && choice.error !== null); + relay(controller, choiceError ? canonicalSseErrorBlock(normalizeChatChoiceErrors(chunk)) + : replaceSseDataPayload(block, JSON.stringify(chunk)) + delimiter); + } } }; diff --git a/src/server/chat-native.ts b/src/server/chat-native.ts index 123659d41d5..e1d72dbe49b 100644 --- a/src/server/chat-native.ts +++ b/src/server/chat-native.ts @@ -82,7 +82,10 @@ import { recordAttemptCredentialSource, type RequestLogContext, } from "./request-log"; -import { chatResponseHasUndeclaredToolCalls, UNDECLARED_CHAT_TOOL_MESSAGE, jsonCompletionSse, nativeChatSse, structuredError, usageFromChat } from "./chat-native-sse"; +import { + chatResponseHasUndeclaredToolCalls, UNDECLARED_CHAT_TOOL_MESSAGE, jsonCompletionSse, mistralTypedContentCompletion, nativeChatSse, + structuredError, usageFromChat, +} from "./chat-native-sse"; import { beginInferenceAttempt, type InferenceAttempt } from "./inference/attempt"; import { createFinalRequestLog } from "./inference/final-log"; import { registerTurn, unregisterTurn } from "./lifecycle"; @@ -92,7 +95,7 @@ import type { ComboProtocolSource } from "./responses/core-combo-native"; import type { ProtocolEnvelope } from "../protocols/envelope"; import { protocolTraceSuppressed } from "../protocols/trace"; import { nativeChatReplayScope, bindNativeChatReplayScope, nativeChatReasoningCapture } from "./chat-native-reasoning"; -import { resolveMistralTransport } from "../providers/mistral-transport"; +import { isCanonicalMistralChat, resolveMistralTransport } from "../providers/mistral-transport"; export { isNativeChatRouteEligible, nativeChatDeclineReason } from "./chat-native-eligibility"; @@ -805,6 +808,7 @@ export async function runNativeChatAttempt( const contentType = response.headers.get("content-type")?.toLowerCase() ?? ""; const reasoningCapture = nativeChatReasoningCapture(activeProvider, route.modelId, replayScope, translatorBudget); + const mistralBaseUrl = isCanonicalMistralChat(activeProvider) ? activeProvider.baseUrl : undefined; if (contentType.includes("text/event-stream") && response.body) { if (requestedStream) transferTurnToStream(); let terminalStatus: number | undefined; @@ -845,6 +849,7 @@ export async function runNativeChatAttempt( signal: upstream.signal, stallTimeoutSec: config.stallTimeoutSec, localUpstream, + mistralBaseUrl, clientDeclaredEmptyToolCatalog, declaredLegacyFunctionNames, onFirstOutput, @@ -956,11 +961,14 @@ export async function runNativeChatAttempt( if (clientDeclaredEmptyToolCatalog && chatResponseHasUndeclaredToolCalls(completion, declaredLegacyFunctionNames)) { return failUpstream(502, UNDECLARED_CHAT_TOOL_MESSAGE, "upstream_error", UNDECLARED_TOOL_CALL_ERROR_CODE); } + // Replay capture above keeps reading the upstream body; only the client copy is converted. + const clientCompletion = mistralBaseUrl ? mistralTypedContentCompletion(completion, mistralBaseUrl) : completion; + if (!clientCompletion) return failUpstream(502, "unsupported Mistral content block", "upstream_error"); onFirstOutput?.(); try { const serialized = requestedStream - ? jsonCompletionSse(completion, requestedModel, translatorBudget) - : JSON.stringify(completion); + ? jsonCompletionSse(clientCompletion, requestedModel, translatorBudget) + : JSON.stringify(clientCompletion); if (!requestedStream) translatorBudget.chargeRetained(Buffer.byteLength(serialized) * 2, { kind: "live_transient" }); finishLog(200); return new Response(serialized, { diff --git a/structure/providers/chat-compat.md b/structure/providers/chat-compat.md index 9a979b3e4b4..e5d37e9bc52 100644 --- a/structure/providers/chat-compat.md +++ b/structure/providers/chat-compat.md @@ -48,8 +48,9 @@ Shared parsing and streaming follow the [request-copy](../transports/byte-accoun ## Reasoning and tool-result compatibility For `api.mistral.ai`, `src/adapters/openai-chat/response-events.ts` preserves typed text/thinking arrays in both response paths and rejects unsupported blocks. Inline-tag parsing crosses adjacent text blocks but resets at typed reasoning. It adds no signature replay. +Native Chat passthrough reuses that converter for canonical key-auth Mistral (`isCanonicalMistralChat`), because its clients parse Chat strings. `src/server/chat-native-sse.ts` splits a typed stream delta into one chunk per run in source order, thinking as `reasoning_content` and text as `content`; the last chunk keeps the remaining delta fields, `finish_reason` and `usage`. A buffered `message.content` joins text into `content` and thinking into `reasoning_content`. An unsupported block fails the turn with a 502. Native reasoning replay capture still reads the upstream chunk, so request history is unchanged. On that same host, `src/adapters/openai-chat/messages.ts` reconstructs historical thinking as native `content` chunks before assistant text, for every hosted model family. An explicit `preserveReasoningContentModels` list, including `[]`, still controls enrollment; the default preserves supplied thinking. The shared prompt projection counts that thinking. Replay can recover actual text from the existing scoped tool-call cache, but never invents placeholders or forwards generic IR signatures. Original chunk metadata and exact chunk boundaries are not retained by this text-only reconstruction. Native Chat passthrough and other destinations retain their existing behavior. -Tests: `tests/adapters/openai/openai-chat-mistral-content.test.ts`; byte budgets retain their existing limits. +Tests: `tests/adapters/openai/openai-chat-mistral-content.test.ts`, `tests/responses/chat-native-mistral-content.test.ts`; byte budgets retain their existing limits. ### Inline think-tag recovery diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 801b3c2155b..66f66b124f6 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -453,6 +453,7 @@ "chat-media-translation.test.ts": "responses", "chat-native-developer-role.test.ts": "responses", "chat-native-reasoning-replay.test.ts": "responses", "chat-native-reasoning-scope.test.ts": "responses", + "chat-native-mistral-content.test.ts": "responses", "chat-native-image-normalization.test.ts": "responses", "chat-native-spend.test.ts": "responses", "chat-refusal-scope.test.ts": "responses", "chat-refusal.test.ts": "responses", "chat-responses-control-integration.test.ts": "responses", "chat-responses-control-scope.test.ts": "responses", diff --git a/tests/responses/chat-native-mistral-content.test.ts b/tests/responses/chat-native-mistral-content.test.ts new file mode 100644 index 00000000000..544dfccf06d --- /dev/null +++ b/tests/responses/chat-native-mistral-content.test.ts @@ -0,0 +1,154 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { handleChatCompletions } from "../../src/server/chat-completions"; +import type { OcxConfig } from "../../src/types"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; + +// Native /v1/chat/completions passthrough to canonical Mistral. Reasoning models answer with typed +// `thinking`/`text` content blocks; the relay must hand Chat clients plain strings. +const MISTRAL = "https://api.mistral.ai/v1"; +const ANSWER = '{"outcome":"deny"}'; +const USAGE = { prompt_tokens: 10, completion_tokens: 5 }; +const thinking = (text: string) => ({ type: "thinking", thinking: [{ type: "text", text }], closed: true }); +const text = (value: string) => ({ type: "text", text: value }); +const delta = (value: Record, extra: Record = {}) => + ({ id: "synthetic", choices: [{ index: 0, delta: value, finish_reason: null }], ...extra }); +const terminal = { id: "synthetic", choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: USAGE }; + +type Chunk = { choices: { delta: Record; finish_reason: string | null }[]; usage?: unknown }; + +const originalFetch = globalThis.fetch; +let releaseSpendHome: (() => void) | undefined; + +afterEach(() => { + globalThis.fetch = originalFetch; + releaseSpendHome?.(); + releaseSpendHome = undefined; +}); + +function sse(chunks: unknown[]): Response { + return new Response(chunks.map(chunk => `data: ${JSON.stringify(chunk)}\n\n`).join("") + "data: [DONE]\n\n", { + headers: { "content-type": "text/event-stream" }, + }); +} + +function completion(content: unknown) { + return { id: "synthetic", choices: [{ index: 0, message: { role: "assistant", content }, finish_reason: "stop" }], usage: USAGE }; +} + +async function chat(upstream: Response, stream: boolean, baseUrl = MISTRAL): Promise { + globalThis.fetch = (async () => upstream) as unknown as typeof fetch; + releaseSpendHome ??= acquireOwnedSpendHome(); + return handleChatCompletions(new Request("http://localhost/v1/chat/completions", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "fixture/zai-glm-5-3", messages: [{ role: "user", content: "Synthetic fixture" }], stream }), + }), { port: 0, defaultProvider: "fixture", providers: { fixture: { + adapter: "openai-chat", baseUrl, authMode: "key", apiKey: "synthetic-native-chat-key", + } } } as OcxConfig, { model: "", provider: "" }); +} + +async function frames(response: Response): Promise<{ chunks: Chunk[]; raw: string }> { + const raw = await response.text(); + const chunks = raw.split("\n\n") + .map(block => block.split("\n").find(line => line.startsWith("data: "))?.slice(6)) + .filter((data): data is string => data !== undefined && data !== "[DONE]") + .map(data => JSON.parse(data) as Chunk); + return { chunks, raw }; +} + +/** Client-visible string runs in arrival order, adjacent runs of one field merged. */ +function runs(chunks: Chunk[]): [string, string][] { + const out: [string, string][] = []; + for (const chunk of chunks) { + for (const field of ["reasoning_content", "content"] as const) { + const value = chunk.choices?.[0]?.delta?.[field]; + if (typeof value !== "string" || value.length === 0) continue; + if (out.at(-1)?.[0] === field) out.at(-1)![1] += value; + else out.push([field, value]); + } + } + return out; +} + +describe("native Chat passthrough of Mistral typed content", () => { + test("streamed thinking and text arrive as Chat strings in source order", async () => { + const response = await chat(sse([ + delta({ role: "assistant", content: "" }), + delta({ content: [thinking("Synthetic reasoning.")] }), + delta({ content: [thinking(" More."), text('{"outcome')] }), + delta({ content: '":"deny"}' }), + terminal, + ]), true); + expect(response.status).toBe(200); + const { chunks } = await frames(response); + for (const chunk of chunks) { + const content = chunk.choices?.[0]?.delta?.content; + expect(content === undefined || typeof content === "string").toBe(true); + } + expect(runs(chunks)).toEqual([["reasoning_content", "Synthetic reasoning. More."], ["content", ANSWER]]); + expect(chunks.at(-1)).toMatchObject({ choices: [{ finish_reason: "stop" }], usage: USAGE }); + }); + + test("interleaved blocks in one chunk keep their order; role leads, finish_reason and usage trail", async () => { + const response = await chat(sse([{ + id: "synthetic", usage: USAGE, + choices: [{ index: 0, delta: { role: "assistant", content: [text("A"), thinking("first"), text("B")] }, finish_reason: "stop" }], + }]), true); + const { chunks } = await frames(response); + expect(runs(chunks)).toEqual([["content", "A"], ["reasoning_content", "first"], ["content", "B"]]); + expect(chunks.map(chunk => chunk.choices[0]!.delta.role)).toEqual(["assistant", undefined, undefined]); + expect(chunks.map(chunk => chunk.choices[0]!.finish_reason)).toEqual([null, null, "stop"]); + expect(chunks.map(chunk => chunk.usage === undefined)).toEqual([true, true, false]); + }); + + test("a typed-only stream cut before its terminal frame still counts as output and closes cleanly", async () => { + const response = await chat(new Response(`data: ${JSON.stringify(delta({ content: [thinking("t"), text("partial")] }))}\n\n`, { + headers: { "content-type": "text/event-stream" }, + }), true); + const { chunks, raw } = await frames(response); + expect(runs(chunks)).toEqual([["reasoning_content", "t"], ["content", "partial"]]); + expect(raw).toContain("data: [DONE]"); + expect(raw).not.toContain("upstream_sse_truncated"); + }); + + test("a buffered reply joins thinking into reasoning_content and text into content", async () => { + const response = await chat(Response.json(completion([thinking("Synthetic reasoning."), text(ANSWER)])), false); + expect(response.status).toBe(200); + const body = await response.json() as { choices: { message: Record }[] }; + expect(body.choices[0]!.message).toMatchObject({ content: ANSWER, reasoning_content: "Synthetic reasoning." }); + }); + + test("a buffered thinking-only reply has null content, like the streamed collector", async () => { + const response = await chat(Response.json(completion([thinking("Synthetic reasoning.")])), false); + const body = await response.json() as { choices: { message: Record }[] }; + expect(body.choices[0]!.message).toMatchObject({ content: null, reasoning_content: "Synthetic reasoning." }); + }); + + test("a buffered upstream reply to a streaming client is synthesized from the converted strings", async () => { + const response = await chat(Response.json(completion([thinking("Synthetic reasoning."), text(ANSWER)])), true); + const { chunks } = await frames(response); + expect(runs(chunks)).toEqual([["reasoning_content", "Synthetic reasoning."], ["content", ANSWER]]); + }); + + test("a streamed upstream reply to a non-streaming client collects the converted strings", async () => { + const response = await chat(sse([delta({ content: [thinking("Synthetic reasoning."), text(ANSWER)] }), terminal]), false); + expect(response.status).toBe(200); + const body = await response.json() as { choices: { message: Record }[] }; + expect(body.choices[0]!.message).toMatchObject({ content: ANSWER, reasoning_content: "Synthetic reasoning." }); + }); + + test("an unsupported block fails closed in both delivery shapes", async () => { + const streamed = await frames(await chat(sse([delta({ content: [{ type: "image_url", image_url: "x" }] }), terminal]), true)); + expect(streamed.raw).toContain("upstream_sse_invalid"); + expect(runs(streamed.chunks)).toEqual([]); + const buffered = await chat(Response.json(completion([{ type: "audio" }])), false); + expect(buffered.status).toBe(502); + expect(await buffered.json()).toMatchObject({ error: { type: "upstream_error" } }); + }); + + test("other destinations keep relaying their content unchanged", async () => { + const content = [thinking("kept"), text("as-is")]; + const response = await chat(sse([delta({ content }), terminal]), true, "https://chat.example.test/v1"); + const { chunks } = await frames(response); + expect(chunks[0]!.choices[0]!.delta.content).toEqual(content); + }); +});