From 2cdf43a25801660472e683fcf9cf0c07db3feafb Mon Sep 17 00:00:00 2001 From: ggbdpq Date: Tue, 20 Oct 2026 10:01:48 +0800 Subject: [PATCH] feat(wp1): egress serializers map thinking, tool calls, stop reason for chat/responses/anthropic Co-authored-by: Cursor --- .../src/protocols/anthropic/messages.ts | 124 +++++--- .../gateway/src/protocols/anthropic/stream.ts | 28 +- apps/gateway/src/protocols/openai/chat.ts | 100 +++--- .../gateway/src/protocols/openai/responses.ts | 201 +++++++++--- apps/gateway/src/protocols/openai/stream.ts | 39 ++- .../test/protocols/egress/egress.test.ts | 288 ++++++++++++++++++ 6 files changed, 658 insertions(+), 122 deletions(-) create mode 100644 apps/gateway/test/protocols/egress/egress.test.ts diff --git a/apps/gateway/src/protocols/anthropic/messages.ts b/apps/gateway/src/protocols/anthropic/messages.ts index d1c4cec..7a9b57b 100644 --- a/apps/gateway/src/protocols/anthropic/messages.ts +++ b/apps/gateway/src/protocols/anthropic/messages.ts @@ -4,42 +4,47 @@ import type { ProtocolHandlerArgs } from "../../server/http.ts"; import { beginSse, writeAnthropicSse } from "../../server/sse.ts"; import { anthropicToCanonical } from "./request.ts"; import { + STOP_REASON, + TEXT_BLOCK, + THINKING_BLOCK, + contentBlockInputJsonDelta, contentBlockStart, contentBlockStop, contentBlockTextDelta, + contentBlockThinkingDelta, messageDelta, messageStart, messageStop, streamError, } from "./stream.ts"; +import type { AggregateResult } from "../../core/aggregate.ts"; import type { UsageEvent } from "../../core/events.ts"; -// Provider usage 的最小形状（AggregateUsage / UsageEvent 的公共部分）。 -export interface AnthropicUsageSummary { - inputTokens?: number; - outputTokens?: number; -} - -export function buildAnthropicMessage({ - id, - model, - text, - usage, -}: { - id: string; - model: string; - text: string; - usage?: AnthropicUsageSummary; -}) { +export function buildAnthropicMessage(id: string, model: string, result: AggregateResult) { + const content: unknown[] = []; + if (result.thinking) content.push({ ...THINKING_BLOCK, thinking: result.thinking }); + if (result.text || !result.toolCalls.length) content.push({ type: "text", text: result.text }); + for (const call of result.toolCalls) { + // 参数不是合法 JSON 属于上游协议错误，交给 server 的错误映射，不静默吞掉。 + content.push({ + type: "tool_use", + id: call.id, + name: call.name, + input: call.arguments ? JSON.parse(call.arguments) : {}, + }); + } return { id, type: "message", role: "assistant", model, - content: [{ type: "text", text }], - stop_reason: "end_turn", + content, + stop_reason: STOP_REASON[result.stopReason], stop_sequence: null, - usage: { input_tokens: usage?.inputTokens ?? 0, output_tokens: usage?.outputTokens ?? 0 }, + usage: { + input_tokens: result.usage.inputTokens ?? 0, + output_tokens: result.usage.outputTokens ?? 0, + }, }; } @@ -59,27 +64,78 @@ export async function handleAnthropicMessages({ if (body.stream) { beginSse(res); writeAnthropicSse(res, "message_start", messageStart(id, model)); - writeAnthropicSse(res, "content_block_start", contentBlockStart(0)); let usage: UsageEvent | undefined; + let stopReason: string = STOP_REASON.stop; + // 同一时刻只有一个打开的 content block，换类型时先 stop 上一个；index 按出现顺序编号。 + let index = -1; + let current: string | null = null; + const closeBlock = () => { + if (current === null) return; + writeAnthropicSse(res, "content_block_stop", contentBlockStop(index)); + current = null; + }; + const openBlock = (key: string, block: object) => { + if (current === key) return; + closeBlock(); + current = key; + writeAnthropicSse(res, "content_block_start", contentBlockStart(++index, block)); + }; + for await (const event of run.stream(request)) { - if (event.type === "text.delta") { - writeAnthropicSse(res, "content_block_delta", contentBlockTextDelta(0, event.text)); - } else if (event.type === "usage") { - usage = event; - } else if (event.type === "error") { - // 官方 Anthropic 流式语义：错误以 error 事件收尾，不再发 message_stop。 - writeAnthropicSse(res, "error", streamError(event.error)); - res.end(); - return; + switch (event.type) { + case "text.delta": + openBlock("text", TEXT_BLOCK); + writeAnthropicSse(res, "content_block_delta", contentBlockTextDelta(index, event.text)); + break; + case "thinking.delta": + openBlock("thinking", THINKING_BLOCK); + writeAnthropicSse( + res, + "content_block_delta", + contentBlockThinkingDelta(index, event.text), + ); + break; + case "tool.start": + openBlock(`tool:${event.index}`, { + type: "tool_use", + id: event.id, + name: event.name, + input: {}, + }); + break; + case "tool.arguments.delta": + if (current === `tool:${event.index}`) { + writeAnthropicSse( + res, + "content_block_delta", + contentBlockInputJsonDelta(index, event.json), + ); + } + break; + case "tool.end": + if (current === `tool:${event.index}`) closeBlock(); + break; + case "message.end": + stopReason = STOP_REASON[event.stopReason]; + break; + case "usage": + usage = event; + break; + case "error": + // 官方 Anthropic 流式语义：错误以 error 事件收尾，不再发 message_stop。 + writeAnthropicSse(res, "error", streamError(event.error)); + res.end(); + return; } } - writeAnthropicSse(res, "content_block_stop", contentBlockStop(0)); - writeAnthropicSse(res, "message_delta", messageDelta("end_turn", usage?.outputTokens ?? 0)); + // 没有任何内容块时保持 v0.1 的空 text 块。 + if (index === -1) openBlock("text", TEXT_BLOCK); + closeBlock(); + writeAnthropicSse(res, "message_delta", messageDelta(stopReason, usage?.outputTokens ?? 0)); writeAnthropicSse(res, "message_stop", messageStop()); res.end(); return; } - const result = await run.aggregate(request); - writeJson(res, 200, buildAnthropicMessage({ id, model, text: result.text, usage: result.usage })); + writeJson(res, 200, buildAnthropicMessage(id, model, await run.aggregate(request))); } diff --git a/apps/gateway/src/protocols/anthropic/stream.ts b/apps/gateway/src/protocols/anthropic/stream.ts index 23d05a1..9f868ff 100644 --- a/apps/gateway/src/protocols/anthropic/stream.ts +++ b/apps/gateway/src/protocols/anthropic/stream.ts @@ -17,14 +17,38 @@ export function messageStart(id: string, model: string) { }; } -export function contentBlockStart(index: number) { - return { type: "content_block_start", index, content_block: { type: "text", text: "" } }; +export const TEXT_BLOCK = { type: "text", text: "" }; +// 上游不提供 thinking signature，按 Anthropic 形状给空串。 +export const THINKING_BLOCK = { type: "thinking", thinking: "", signature: "" }; + +// Canonical stopReason → Anthropic stop_reason。 +export const STOP_REASON = { + stop: "end_turn", + tool_use: "tool_use", + length: "max_tokens", + cancelled: "end_turn", +} as const; + +export function contentBlockStart(index: number, contentBlock: object = TEXT_BLOCK) { + return { type: "content_block_start", index, content_block: contentBlock }; } export function contentBlockTextDelta(index: number, text: string) { return { type: "content_block_delta", index, delta: { type: "text_delta", text } }; } +export function contentBlockThinkingDelta(index: number, thinking: string) { + return { type: "content_block_delta", index, delta: { type: "thinking_delta", thinking } }; +} + +export function contentBlockInputJsonDelta(index: number, partialJson: string) { + return { + type: "content_block_delta", + index, + delta: { type: "input_json_delta", partial_json: partialJson }, + }; +} + export function contentBlockStop(index: number) { return { type: "content_block_stop", index }; } diff --git a/apps/gateway/src/protocols/openai/chat.ts b/apps/gateway/src/protocols/openai/chat.ts index ba269c4..461f3ad 100644 --- a/apps/gateway/src/protocols/openai/chat.ts +++ b/apps/gateway/src/protocols/openai/chat.ts @@ -3,8 +3,9 @@ import { wireId, writeJson } from "../../server/http.ts"; import type { ProtocolHandlerArgs } from "../../server/http.ts"; import { beginSse, writeSseData, writeSseDone } from "../../server/sse.ts"; import { assertHasText, chatToCanonical } from "./request.ts"; -import { chatChunk } from "./stream.ts"; +import { FINISH_REASON, chatChunk } from "./stream.ts"; import type { ChatCompletionChunk, UsageSummary } from "./stream.ts"; +import type { AggregateResult } from "../../core/aggregate.ts"; import type { UsageEvent } from "../../core/events.ts"; // Provider usage（inputTokens/outputTokens）→ OpenAI usage；无数据保持 0。 @@ -14,32 +15,26 @@ function toOpenAiUsage(usage?: UsageSummary) { return { prompt_tokens: input, completion_tokens: output, total_tokens: input + output }; } -export function buildChatCompletion({ - id, - model, - text, - finishReason = "stop", - usage, -}: { - id: string; - model: string; - text: string; - finishReason?: string; - usage?: UsageSummary; -}) { +export function buildChatCompletion(id: string, model: string, result: AggregateResult) { + const message: Record = { + role: "assistant", + content: result.text || (result.toolCalls.length ? null : ""), + }; + if (result.thinking) message.reasoning_content = result.thinking; + if (result.toolCalls.length) { + message.tool_calls = result.toolCalls.map((call) => ({ + id: call.id, + type: "function", + function: { name: call.name, arguments: call.arguments }, + })); + } return { id, object: "chat.completion", created: Math.floor(Date.now() / 1000), model, - choices: [ - { - index: 0, - message: { role: "assistant", content: text || "" }, - finish_reason: finishReason, - }, - ], - usage: toOpenAiUsage(usage), + choices: [{ index: 0, message, finish_reason: FINISH_REASON[result.stopReason] }], + usage: toOpenAiUsage(result.usage), }; } @@ -57,22 +52,54 @@ export async function handleChatCompletions({ beginSse(res); writeSseData(res, chatChunk(id, model, { role: "assistant" })); let usage: UsageEvent | undefined; + let finishReason: string = FINISH_REASON.stop; + // Canonical tool index → 下游 tool_calls[].index，按 tool.start 出现顺序编号。 + const toolIndex = new Map(); for await (const event of run.stream(request)) { - if (event.type === "text.delta") { - writeSseData(res, chatChunk(id, model, { content: event.text })); - } else if (event.type === "usage") { - usage = event; - } else if (event.type === "error") { - // v0.1 在此会触发 headers-sent 崩溃并挂住连接；现按 OpenAI 惯例发 error 数据后干净收尾。 - writeSseData(res, { - error: { message: event.error.message, type: "server_error", code: event.error.code }, - }); - writeSseDone(res); - res.end(); - return; + switch (event.type) { + case "text.delta": + writeSseData(res, chatChunk(id, model, { content: event.text })); + break; + case "thinking.delta": + writeSseData(res, chatChunk(id, model, { reasoning_content: event.text })); + break; + case "tool.start": { + const index = toolIndex.size; + toolIndex.set(event.index, index); + const call = { name: event.name, arguments: "" }; + writeSseData( + res, + chatChunk(id, model, { + tool_calls: [{ index, id: event.id, type: "function", function: call }], + }), + ); + break; + } + case "tool.arguments.delta": { + const index = toolIndex.get(event.index) ?? event.index; + writeSseData( + res, + chatChunk(id, model, { tool_calls: [{ index, function: { arguments: event.json } }] }), + ); + break; + } + case "message.end": + finishReason = FINISH_REASON[event.stopReason]; + break; + case "usage": + usage = event; + break; + case "error": + // v0.1 在此会触发 headers-sent 崩溃并挂住连接；现按 OpenAI 惯例发 error 数据后干净收尾。 + writeSseData(res, { + error: { message: event.error.message, type: "server_error", code: event.error.code }, + }); + writeSseDone(res); + res.end(); + return; } } - writeSseData(res, chatChunk(id, model, {}, "stop")); + writeSseData(res, chatChunk(id, model, {}, finishReason)); if (usage) { // OpenAI include_usage 约定：空 choices 的收尾 usage chunk。 const final: ChatCompletionChunk = chatChunk(id, model, {}, "stop"); @@ -85,6 +112,5 @@ export async function handleChatCompletions({ return; } - const result = await run.aggregate(request); - writeJson(res, 200, buildChatCompletion({ id, model, text: result.text, usage: result.usage })); + writeJson(res, 200, buildChatCompletion(id, model, await run.aggregate(request))); } diff --git a/apps/gateway/src/protocols/openai/responses.ts b/apps/gateway/src/protocols/openai/responses.ts index 8707b37..646e173 100644 --- a/apps/gateway/src/protocols/openai/responses.ts +++ b/apps/gateway/src/protocols/openai/responses.ts @@ -1,4 +1,5 @@ // OpenAI Responses 端点适配器：Canonical Event → wire。 +import type { ServerResponse } from "node:http"; import { wireId, writeJson } from "../../server/http.ts"; import type { ProtocolHandlerArgs } from "../../server/http.ts"; import { beginSse, writeSseData, writeSseDone } from "../../server/sse.ts"; @@ -9,46 +10,128 @@ import { responseCompleted, responseCreated, responseFailed, + responseStatus, } from "./stream.ts"; -import type { UsageSummary } from "./stream.ts"; +import type { AggregateResult } from "../../core/aggregate.ts"; import type { UsageEvent } from "../../core/events.ts"; -export function buildResponsesCompletion({ - id, - model, - text, - status = "completed", - usage, -}: { - id: string; - model: string; - text: string; - status?: string; - usage?: UsageSummary; -}) { +const functionCallItem = (call: { id: string; name: string }) => ({ + id: `fc_${call.id}`, + type: "function_call", + call_id: call.id, + name: call.name, +}); + +export function buildResponsesCompletion(id: string, model: string, result: AggregateResult) { + const { text, thinking, toolCalls, usage } = result; const outputId = `msg_${id.replace(/[^a-zA-Z0-9_]/g, "")}`; + const output: unknown[] = []; + if (thinking) { + output.push({ + id: `rs_${id}`, + type: "reasoning", + summary: [{ type: "summary_text", text: thinking }], + }); + } + if (text || !toolCalls.length) { + output.push({ + id: outputId, + type: "message", + status: "completed", + role: "assistant", + content: [{ type: "output_text", text: text || "", annotations: [] }], + }); + } + for (const call of toolCalls) { + output.push({ ...functionCallItem(call), arguments: call.arguments, status: "completed" }); + } const input = usage?.inputTokens ?? 0; - const output = usage?.outputTokens ?? 0; + const outputTokens = usage?.outputTokens ?? 0; return { id, object: "response", created_at: Math.floor(Date.now() / 1000), - status, + ...responseStatus(result.stopReason), model, - output: [ - { - id: outputId, - type: "message", - status: "completed", - role: "assistant", - content: [{ type: "output_text", text: text || "", annotations: [] }], - }, - ], + output, output_text: text || "", usage: { input_tokens: input, - output_tokens: output, - total_tokens: input + output, + output_tokens: outputTokens, + total_tokens: input + outputTokens, + }, + }; +} + +// 流式 output item：打开时写 added，delta 写增量，close 写 done。 +interface OpenItem { + key: string; + delta(chunk: string): void; + close(): void; +} + +function messageItem(res: ServerResponse, id: string, index: number) { + return { + delta: (text: string) => writeSseData(res, outputTextDelta(id, text, index)), + close: () => writeSseData(res, outputTextDone(id, index)), + }; +} + +function reasoningItem(res: ServerResponse, id: string, index: number) { + const item = { id: `rs_${id}_${index}`, type: "reasoning" }; + const ref = { response_id: id, item_id: item.id, output_index: index, summary_index: 0 }; + let text = ""; + writeSseData(res, { + type: "response.output_item.added", + response_id: id, + output_index: index, + item: { ...item, summary: [] }, + }); + return { + delta(chunk: string) { + text += chunk; + writeSseData(res, { type: "response.reasoning_summary_text.delta", ...ref, delta: chunk }); + }, + close() { + writeSseData(res, { type: "response.reasoning_summary_text.done", ...ref, text }); + writeSseData(res, { + type: "response.output_item.done", + response_id: id, + output_index: index, + item: { ...item, summary: [{ type: "summary_text", text }] }, + }); + }, + }; +} + +function functionCallStreamItem( + res: ServerResponse, + id: string, + index: number, + call: { id: string; name: string }, +) { + const item = functionCallItem(call); + const ref = { response_id: id, item_id: item.id, output_index: index }; + let args = ""; + writeSseData(res, { + type: "response.output_item.added", + response_id: id, + output_index: index, + item: { ...item, arguments: "", status: "in_progress" }, + }); + return { + delta(chunk: string) { + args += chunk; + writeSseData(res, { type: "response.function_call_arguments.delta", ...ref, delta: chunk }); + }, + close() { + writeSseData(res, { type: "response.function_call_arguments.done", ...ref, arguments: args }); + writeSseData(res, { + type: "response.output_item.done", + response_id: id, + output_index: index, + item: { ...item, arguments: args, status: "completed" }, + }); }, }; } @@ -67,28 +150,60 @@ export async function handleResponses({ beginSse(res); writeSseData(res, responseCreated(id, model)); let usage: UsageEvent | undefined; + let stopReason: string | undefined; + // 同一时刻只有一个打开的 item，换类型时先收尾上一个；output_index 按出现顺序编号。 + // ponytail: 假设上游工具调用按顺序到达（不交错）；交错的并行工具调用需按 index 分别维护。 + // 初值用断言：current 只在闭包 switchTo 里赋值，直接写 `= null` 会被控制流收窄成 never。 + let current = null as OpenItem | null; + let nextIndex = 0; + const switchTo = (key: string, open: (index: number) => Omit) => { + if (current?.key !== key) { + current?.close(); + current = { key, ...open(nextIndex++) }; + } + return current; + }; + for await (const event of run.stream(request)) { - if (event.type === "text.delta") { - writeSseData(res, outputTextDelta(id, event.text)); - } else if (event.type === "usage") { - usage = event; - } else if (event.type === "error") { - writeSseData(res, responseFailed(id, model, event.error)); - res.end(); - return; + switch (event.type) { + case "text.delta": + switchTo("message", (i) => messageItem(res, id, i)).delta(event.text); + break; + case "thinking.delta": + switchTo("reasoning", (i) => reasoningItem(res, id, i)).delta(event.text); + break; + case "tool.start": + switchTo(`tool:${event.index}`, (i) => functionCallStreamItem(res, id, i, event)); + break; + case "tool.arguments.delta": + if (current?.key === `tool:${event.index}`) current.delta(event.json); + break; + case "tool.end": + if (current?.key === `tool:${event.index}`) { + current.close(); + current = null; + } + break; + case "message.end": + stopReason = event.stopReason; + break; + case "usage": + usage = event; + break; + case "error": + writeSseData(res, responseFailed(id, model, event.error)); + res.end(); + return; } } - writeSseData(res, outputTextDone(id)); - writeSseData(res, responseCompleted(id, model, usage)); + // 没有任何 output item 时保持 v0.1 的空 output_text.done 收尾。 + if (nextIndex === 0) writeSseData(res, outputTextDone(id)); + current?.close(); + writeSseData(res, responseCompleted(id, model, usage, stopReason)); writeSseDone(res); res.end(); return; } - const result = await run.aggregate(request); - writeJson( - res, - 200, - buildResponsesCompletion({ id, model, text: result.text, usage: result.usage }), - ); + writeJson(res, 200, buildResponsesCompletion(id, model, await run.aggregate(request))); } diff --git a/apps/gateway/src/protocols/openai/stream.ts b/apps/gateway/src/protocols/openai/stream.ts index bd9e57d..966819e 100644 --- a/apps/gateway/src/protocols/openai/stream.ts +++ b/apps/gateway/src/protocols/openai/stream.ts @@ -11,8 +11,23 @@ export interface UsageSummary { export interface ChatChunkDelta { role?: string; content?: string; + reasoning_content?: string; + tool_calls?: Array<{ + index: number; + id?: string; + type?: "function"; + function: { name?: string; arguments: string }; + }>; } +// Canonical stopReason → OpenAI Chat finish_reason。 +export const FINISH_REASON = { + stop: "stop", + tool_use: "tool_calls", + length: "length", + cancelled: "stop", +} as const; + export interface ChatCompletionChunk { id: string; object: string; @@ -44,35 +59,47 @@ export function responseCreated(id: string, model: string) { }; } -export function outputTextDelta(id: string, delta: string) { +export function outputTextDelta(id: string, delta: string, outputIndex = 0) { return { type: "response.output_text.delta", response_id: id, item_id: `msg_${id}`, - output_index: 0, + output_index: outputIndex, content_index: 0, delta, }; } -export function outputTextDone(id: string) { +export function outputTextDone(id: string, outputIndex = 0) { return { type: "response.output_text.done", response_id: id, item_id: `msg_${id}`, - output_index: 0, + output_index: outputIndex, content_index: 0, }; } -export function responseCompleted(id: string, model: string, usage?: UsageSummary) { +// stopReason length → incomplete，其余 completed。 +export function responseStatus(stopReason?: string) { + return stopReason === "length" + ? { status: "incomplete", incomplete_details: { reason: "max_output_tokens" } } + : { status: "completed" }; +} + +export function responseCompleted( + id: string, + model: string, + usage?: UsageSummary, + stopReason?: string, +) { const response: { id: string; object: string; status: string; model: string; usage?: { input_tokens: number; output_tokens: number; total_tokens: number }; - } = { id, object: "response", status: "completed", model }; + } = { id, object: "response", ...responseStatus(stopReason), model }; if (usage) { const input = usage.inputTokens ?? 0; const output = usage.outputTokens ?? 0; diff --git a/apps/gateway/test/protocols/egress/egress.test.ts b/apps/gateway/test/protocols/egress/egress.test.ts new file mode 100644 index 0000000..d97a95b --- /dev/null +++ b/apps/gateway/test/protocols/egress/egress.test.ts @@ -0,0 +1,288 @@ +// WP1 出口映射：thinking-tools.sse 经真实 GatewayProvider normalize → 三个出口（流式 + 非流式）。 +// 期望值按 WP1 映射表手写；纯文本路径由 characterization golden 兜住逐字节不变。 +import test from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; +import { once } from "node:events"; +import type { AddressInfo } from "node:net"; +import { GatewayProvider } from "../../../src/providers/gateway/index.ts"; +import { aggregateStream } from "../../../src/core/aggregate.ts"; +import { readJsonBody } from "../../../src/server/http.ts"; +import type { ProtocolHandlerArgs } from "../../../src/server/http.ts"; +import { handleChatCompletions } from "../../../src/protocols/openai/chat.ts"; +import { handleResponses } from "../../../src/protocols/openai/responses.ts"; +import { handleAnthropicMessages } from "../../../src/protocols/anthropic/messages.ts"; +import { createFakeGateway } from "../../helpers/fakeGateway.ts"; + +type Handler = (args: ProtocolHandlerArgs) => Promise; +const MODEL = "Qwen3.7-Max"; + +async function call(handler: Handler, body: Record) { + const fake = createFakeGateway({ sseFixture: "thinking-tools.sse" }); + fake.listen(0, "127.0.0.1"); + await once(fake, "listening"); + const { port: fakePort } = fake.address() as AddressInfo; + const provider = new GatewayProvider({ + pat: "qoder_pat_fake", + fetchImpl: (url, init) => { + const target = new URL(url as string | URL); + return fetch(`http://127.0.0.1:${fakePort}${target.pathname}${target.search}`, init); + }, + requestTimeoutMs: 5000, + }); + const ctx = () => ({ requestId: "r1", signal: new AbortController().signal }); + const server = http.createServer(async (req, res) => { + await handler({ + req, + res, + body: await readJsonBody(req), + model: MODEL, + run: { + stream: (request) => provider.stream(request, ctx()), + aggregate: (request) => aggregateStream(provider.stream(request, ctx())), + }, + }); + }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + try { + const res = await fetch(`http://127.0.0.1:${(server.address() as AddressInfo).port}`, { + method: "POST", + body: JSON.stringify(body), + }); + return await res.text(); + } finally { + server.close(); + fake.close(); + } +} + +function sse(text: string) { + return text + .split("\n\n") + .filter((block) => block.trim()) + .map((block) => { + const data = block + .split("\n") + .filter((line) => line.startsWith("data: ")) + .map((line) => line.slice(6)) + .join("\n"); + return data === "[DONE]" ? "[DONE]" : JSON.parse(data); + }); +} + +const chatBody = { messages: [{ role: "user", content: "weather?" }] }; +const responsesBody = { input: "weather?" }; +const anthropicBody = { max_tokens: 100, messages: [{ role: "user", content: "weather?" }] }; +const toolCall = { id: "call_1", name: "get_weather", arguments: '{"city":"Hanoi"}' }; + +test("egress chat stream: reasoning_content、分片 tool_calls、finish_reason tool_calls", async () => { + const chunks = sse(await call(handleChatCompletions, { ...chatBody, stream: true })); + assert.equal(chunks.at(-1), "[DONE]"); + assert.deepEqual( + chunks + .slice(0, -1) + .map((c) => ({ delta: c.choices[0].delta, finish: c.choices[0].finish_reason })), + [ + { delta: { role: "assistant" }, finish: null }, + { delta: { reasoning_content: "let me think" }, finish: null }, + { delta: { content: "answer" }, finish: null }, + { + delta: { + tool_calls: [ + { + index: 0, + id: "call_1", + type: "function", + function: { name: "get_weather", arguments: "" }, + }, + ], + }, + finish: null, + }, + { delta: { tool_calls: [{ index: 0, function: { arguments: '{"ci' } }] }, finish: null }, + { + delta: { tool_calls: [{ index: 0, function: { arguments: 'ty":"Hanoi"}' } }] }, + finish: null, + }, + { delta: {}, finish: "tool_calls" }, + ], + ); +}); + +test("egress chat non-stream: message 带 reasoning_content 与 tool_calls", async () => { + const body = JSON.parse(await call(handleChatCompletions, chatBody)); + assert.deepEqual(body.choices[0], { + index: 0, + message: { + role: "assistant", + content: "answer", + reasoning_content: "let me think", + tool_calls: [ + { + id: toolCall.id, + type: "function", + function: { name: toolCall.name, arguments: toolCall.arguments }, + }, + ], + }, + finish_reason: "tool_calls", + }); +}); + +test("egress responses stream: reasoning item、message、function_call item 依次开合", async () => { + const events = sse(await call(handleResponses, { ...responsesBody, stream: true })); + const rs = events[1].item.id; + const msg = events.find((e) => e.type === "response.output_text.delta").item_id; + const fc = { id: "fc_call_1", type: "function_call", call_id: "call_1", name: "get_weather" }; + assert.deepEqual( + events.slice(1, -2).map(({ response_id: _, ...e }) => e), + [ + { + type: "response.output_item.added", + output_index: 0, + item: { id: rs, type: "reasoning", summary: [] }, + }, + { + type: "response.reasoning_summary_text.delta", + item_id: rs, + output_index: 0, + summary_index: 0, + delta: "let me think", + }, + { + type: "response.reasoning_summary_text.done", + item_id: rs, + output_index: 0, + summary_index: 0, + text: "let me think", + }, + { + type: "response.output_item.done", + output_index: 0, + item: { + id: rs, + type: "reasoning", + summary: [{ type: "summary_text", text: "let me think" }], + }, + }, + { + type: "response.output_text.delta", + item_id: msg, + output_index: 1, + content_index: 0, + delta: "answer", + }, + { type: "response.output_text.done", item_id: msg, output_index: 1, content_index: 0 }, + { + type: "response.output_item.added", + output_index: 2, + item: { ...fc, arguments: "", status: "in_progress" }, + }, + { + type: "response.function_call_arguments.delta", + item_id: fc.id, + output_index: 2, + delta: '{"ci', + }, + { + type: "response.function_call_arguments.delta", + item_id: fc.id, + output_index: 2, + delta: 'ty":"Hanoi"}', + }, + { + type: "response.function_call_arguments.done", + item_id: fc.id, + output_index: 2, + arguments: toolCall.arguments, + }, + { + type: "response.output_item.done", + output_index: 2, + item: { ...fc, arguments: toolCall.arguments, status: "completed" }, + }, + ], + ); + assert.equal(events.at(-2).type, "response.completed"); + assert.equal(events.at(-2).response.status, "completed"); + assert.equal(events.at(-1), "[DONE]"); +}); + +test("egress responses non-stream: output 含 reasoning、message、function_call", async () => { + const body = JSON.parse(await call(handleResponses, responsesBody)); + assert.equal(body.status, "completed"); + assert.equal(body.output_text, "answer"); + assert.deepEqual( + body.output.map(({ id: _, ...item }: { id: string }) => item), + [ + { type: "reasoning", summary: [{ type: "summary_text", text: "let me think" }] }, + { + type: "message", + status: "completed", + role: "assistant", + content: [{ type: "output_text", text: "answer", annotations: [] }], + }, + { + type: "function_call", + call_id: "call_1", + name: "get_weather", + arguments: toolCall.arguments, + status: "completed", + }, + ], + ); +}); + +test("egress anthropic stream: thinking / text / tool_use 三个块与 stop_reason tool_use", async () => { + const events = sse(await call(handleAnthropicMessages, { ...anthropicBody, stream: true })); + assert.equal(events[0].type, "message_start"); + assert.deepEqual(events.slice(1), [ + { + type: "content_block_start", + index: 0, + content_block: { type: "thinking", thinking: "", signature: "" }, + }, + { + type: "content_block_delta", + index: 0, + delta: { type: "thinking_delta", thinking: "let me think" }, + }, + { type: "content_block_stop", index: 0 }, + { type: "content_block_start", index: 1, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 1, delta: { type: "text_delta", text: "answer" } }, + { type: "content_block_stop", index: 1 }, + { + type: "content_block_start", + index: 2, + content_block: { type: "tool_use", id: "call_1", name: "get_weather", input: {} }, + }, + { + type: "content_block_delta", + index: 2, + delta: { type: "input_json_delta", partial_json: '{"ci' }, + }, + { + type: "content_block_delta", + index: 2, + delta: { type: "input_json_delta", partial_json: 'ty":"Hanoi"}' }, + }, + { type: "content_block_stop", index: 2 }, + { + type: "message_delta", + delta: { stop_reason: "tool_use", stop_sequence: null }, + usage: { output_tokens: 0 }, + }, + { type: "message_stop" }, + ]); +}); + +test("egress anthropic non-stream: thinking、text、tool_use 块，input 为解析后的对象", async () => { + const body = JSON.parse(await call(handleAnthropicMessages, anthropicBody)); + assert.equal(body.stop_reason, "tool_use"); + assert.deepEqual(body.content, [ + { type: "thinking", thinking: "let me think", signature: "" }, + { type: "text", text: "answer" }, + { type: "tool_use", id: "call_1", name: "get_weather", input: { city: "Hanoi" } }, + ]); +});