import { describe, expect, it } from "vitest"; import type { UIMessage, UIMessageChunk } from "ai"; import { createResponseRequest, responsesToUiStream } from "./responses-transport"; const bytes = (events: Record[], partial = false) => { const text = events.map((event) => `event: ${String(event.type)}\r\ndata: ${JSON.stringify(event)}\r\n\r\n`).join(""); const encoded = new TextEncoder().encode(text); return new ReadableStream({ start(controller) { if (partial) for (let i = 0; i < encoded.length; i += 7) controller.enqueue(encoded.slice(i, i + 7)); else controller.enqueue(encoded); controller.close(); } }); }; const collect = async (stream: ReadableStream) => { const parts: UIMessageChunk[] = []; const reader = stream.getReader(); while (true) { const next = await reader.read(); if (next.done) break; parts.push(next.value); } return parts; }; const events = (): Record[] => [ { type: "response.created", sequence_number: 0, response: { id: "resp_1" } }, { type: "response.output_item.added", sequence_number: 1, item: { type: "message", id: "msg_1", role: "assistant", content: [] } }, { type: "response.content_part.added", sequence_number: 2, item_id: "msg_1", content_index: 0, part: { type: "output_text", text: "" } }, { type: "response.output_text.delta", sequence_number: 3, item_id: "msg_1", content_index: 0, delta: "压力待核验" }, { type: "tjwater.permission_request", sequence_number: 4, response_id: "resp_1", data: { request_id: "permission_1" } }, { type: "response.output_text.done", sequence_number: 5, item_id: "msg_1", content_index: 0, text: "压力待核验" }, { type: "response.completed", sequence_number: 6, response: { id: "resp_1", status: "completed" } }, ]; describe("Responses transport", () => { it("sends only the latest user input and keeps business options explicit", () => { const messages = [{ id: "old", role: "assistant", parts: [{ type: "text", text: "旧回答" }] }, { id: "new", role: "user", parts: [{ type: "text", text: "继续核查" }] }] as UIMessage[]; expect(createResponseRequest(messages, { conversation: "owned", approvalMode: "always" })).toEqual({ input: [{ role: "user", content: [{ type: "input_text", text: "继续核查" }] }], stream: true, conversation: "owned", tjwater: { client_message_id: "new", approval_mode: "always", capabilities: [] }, }); }); it("decodes split UTF-8 and CRLF while retaining approval events", async () => { expect(await collect(responsesToUiStream(bytes(events(), true)))).toEqual([ { type: "start", messageId: "msg_1" }, { type: "text-start", id: "msg_1:0" }, { type: "text-delta", id: "msg_1:0", delta: "压力待核验" }, { type: "data-permission_request", data: { request_id: "permission_1" } }, { type: "text-end", id: "msg_1:0" }, { type: "finish", finishReason: "stop" }, ]); }); it("reports premature disconnects, mismatched owners and invalid event order", async () => { await expect(collect(responsesToUiStream(bytes(events().slice(0, 5))))).rejects.toThrow("断开"); const wrong = events(); wrong[4] = { ...wrong[4], response_id: "other" }; await expect(collect(responsesToUiStream(bytes(wrong)))).rejects.toThrow("归属"); await expect(collect(responsesToUiStream(bytes([events()[0]!, events()[3]!])))).rejects.toThrow(); }); it("turns failed and cancelled Responses into visible errors", async () => { const failed = events(); failed[6] = { type: "response.failed", sequence_number: 6, response: { id: "resp_1", error: { message: "数据源不可用" } } }; expect((await collect(responsesToUiStream(bytes(failed)))).at(-2)).toEqual({ type: "error", errorText: "数据源不可用" }); }); });