import { createServer } from "node:http"; import { createOpenAICompletionsTransportStreamFn } from "@openclaw/ai/transports"; import type { StreamFn } from "openclaw/plugin-sdk/agent-core"; import { attachModelProviderRequestTransport } from "openclaw/plugin-sdk/agent-harness-runtime"; import { createAssistantMessageEventStream, type Context, type Model, type ProviderContext, } from "openclaw/plugin-sdk/llm"; import { registerSingleProviderPlugin } from "openclaw/plugin-sdk/plugin-test-runtime"; import { describe, expect, it, vi } from "vitest"; import plugin from "./index.js"; import { wrapMoonshotStream } from "./native-video.js"; import { MOONSHOT_BASE_URL } from "./provider-catalog.js"; const MP4_A = "data:video/mp4;base64,YWFhYWFhYWFhYWFhYWFhYQ=="; const MP4_B = "data:video/mp4;base64,YmJiYmJiYmJiYmJiYmJiYg=="; const WEBM = "data:video/webm;base64,d2VibQ=="; function model(overrides: Partial = {}): Model { return { id: "kimi-k3", name: "Kimi K3", provider: "moonshot", api: "openai-completions", baseUrl: MOONSHOT_BASE_URL, reasoning: true, input: ["text", "image", "video"] as never, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, contextWindow: 1_048_576, maxTokens: 1_048_576, ...overrides, } as Model; } function genericPayload(videoUrls = [MP4_A]) { return { model: "kimi-k3", messages: [ { role: "user", content: [ { type: "text", text: "before" }, { type: "image_url", image_url: { url: "data:image/png;base64,aW1hZ2U=" } }, ...videoUrls.map((url) => ({ type: "video_url", video_url: { url } })), { type: "text", text: "after" }, ], }, ], }; } function capturePayloadStream(payload: unknown, capture: (value: unknown) => void): StreamFn { return async (payloadModel, _context, options) => { const replacement = await options?.onPayload?.(payload, payloadModel); capture(replacement === undefined ? payload : replacement); return createAssistantMessageEventStream(); }; } function createNativeWrapper(streamFn: StreamFn, requestBytesExclusive = 100_000_000): StreamFn { return wrapMoonshotStream( { provider: "moonshot", modelId: "kimi-k3", streamFn } as never, false, requestBytesExclusive, ); } describe("Moonshot native video wrapper", () => { it("preserves serialized current-user MP4 parts and order", async () => { const payload = genericPayload(); let dispatched: unknown; const caller = vi.fn((value: unknown) => { expect(JSON.stringify(value)).not.toContain("__openclaw"); expect(JSON.stringify(value)).not.toContain("/private/"); expect((value as typeof payload).messages[0]?.content.map((part) => part.type)).toEqual([ "text", "image_url", "video_url", "text", ]); }); const wrapped = createNativeWrapper( capturePayloadStream(payload, (value) => (dispatched = value)), ); await wrapped(model(), { messages: [] } as Context, { onPayload: caller }); expect(caller).toHaveBeenCalledOnce(); expect((dispatched as typeof payload).messages[0]?.content[2]).toEqual({ type: "video_url", video_url: { url: MP4_A }, }); }); it("allows valid hook clones and injections while omitting non-MP4 video", async () => { const payload = genericPayload([MP4_A, WEBM]); let dispatched: unknown; const wrapped = createNativeWrapper( capturePayloadStream(payload, (value) => (dispatched = value)), ); await wrapped(model(), { messages: [] } as Context, { onPayload(value) { const content = (value as typeof payload).messages[0]!.content; const valid = content[2]! as Record; content.push(structuredClone(valid) as never); content.push({ type: "video_url", video_url: { url: MP4_B } } as never); content.push({ type: "image_url", image_url: { url: WEBM } } as never); }, }); const body = JSON.stringify(dispatched); expect(body.split(MP4_A)).toHaveLength(3); expect(body).toContain("YmJiYmJi"); expect(body).not.toContain("d2VibQ"); expect(body.match(/video omitted/gu)).toHaveLength(2); }); it("validates a caller replacement after Moonshot thinking post-processing", async () => { const payload = genericPayload(); let dispatched: unknown; const wrapped = createNativeWrapper( capturePayloadStream(payload, (value) => (dispatched = value)), ); await wrapped(model(), { messages: [] } as Context, { onPayload(value) { return structuredClone(value); }, }); expect(dispatched).toMatchObject({ reasoning_effort: "max" }); expect((dispatched as typeof payload).messages[0]!.content[2]).toEqual({ type: "video_url", video_url: { url: MP4_A }, }); }); it("evicts later admitted videos in place to satisfy the exclusive final size", async () => { const payload = genericPayload([MP4_A, MP4_B]); const projectedWithSecondOmitted = genericPayload([MP4_A]); projectedWithSecondOmitted.messages[0]!.content.splice(3, 0, { type: "text", text: "(video omitted: Moonshot request size limit)", } as never); Object.assign(projectedWithSecondOmitted, { reasoning_effort: "max" }); const ceiling = Buffer.byteLength(JSON.stringify(projectedWithSecondOmitted), "utf8") + 1; let dispatched: unknown; const wrapped = createNativeWrapper( capturePayloadStream(payload, (value) => (dispatched = value)), ceiling, ); await wrapped(model(), { messages: [] } as Context, {}); const content = (dispatched as typeof payload).messages[0]!.content; expect(content[2]).toEqual({ type: "video_url", video_url: { url: MP4_A } }); expect(content[3]).toMatchObject({ type: "text", text: expect.stringContaining("size limit") }); expect(Buffer.byteLength(JSON.stringify(dispatched), "utf8")).toBeLessThan(ceiling); }); it("measures caller replacements and rejects an oversized non-video body", async () => { const wrapped = createNativeWrapper( capturePayloadStream(genericPayload(), () => undefined), 200, ); await expect( wrapped(model(), { messages: [] } as Context, { onPayload: () => ({ model: "kimi-k3", messages: [], padding: "x".repeat(300) }), }), ).rejects.toThrow("Moonshot request body must be smaller than 200 bytes"); }); it.each(["wrapStreamFn", "wrapSimpleCompletionStreamFn"] as const)( "keeps thinking outside native video for %s", async (hookName) => { const provider = await registerSingleProviderPlugin(plugin); const payload = genericPayload(); let dispatched: unknown; const wrapped = provider[hookName]?.({ provider: "moonshot", modelId: "kimi-k3", thinkingLevel: "off", streamFn: capturePayloadStream(payload, (value) => (dispatched = value)), } as never); if (!wrapped) { throw new Error(`Moonshot did not register ${hookName}`); } await wrapped(model(), { messages: [] } as Context, { onPayload(value) { const record = value as Record; expect(record.reasoning_effort).toBe("max"); expect(JSON.stringify(record)).toContain('"type":"video_url"'); record.reasoning_effort = "low"; }, }); expect(dispatched).toMatchObject({ reasoning_effort: "max" }); }, ); }); describe("Moonshot registered transport boundary", () => { it("sends ordered video_url content through the real Chat Completions transport", async () => { let requestBody: Record | undefined; const server = createServer((request, response) => { let body = ""; request.setEncoding("utf8"); request.on("data", (chunk) => (body += chunk)); request.on("end", () => { requestBody = JSON.parse(body) as Record; response.writeHead(200, { "content-type": "text/event-stream" }); response.end( `data: {"id":"chatcmpl-test","object":"chat.completion.chunk","model":"kimi-k3","choices":[{"index":0,"delta":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n`, ); }); }); await new Promise((resolve) => { server.listen(0, "127.0.0.1", () => resolve()); }); try { const address = server.address(); if (!address || typeof address === "string") { throw new Error("missing loopback address"); } const provider = await registerSingleProviderPlugin(plugin); const officialModel = model(); const normalized = provider.normalizeResolvedModel?.({ provider: "moonshot", modelId: "kimi-k3", model: officialModel, } as never) as Model | undefined; expect(normalized?.input).toContain("video"); const transport = createOpenAICompletionsTransportStreamFn(); const loopbackTransport: StreamFn = (runtimeModel, context, options) => transport( attachModelProviderRequestTransport( { ...runtimeModel, provider: "custom", baseUrl: `http://127.0.0.1:${address.port}/v1`, }, { allowPrivateNetwork: true }, ), context, options, ); const wrapped = provider.wrapStreamFn?.({ provider: "moonshot", modelId: "kimi-k3", thinkingLevel: "off", streamFn: loopbackTransport, } as never); if (!wrapped || !normalized) { throw new Error("Moonshot registered transport unavailable"); } const context: ProviderContext = { messages: [ { role: "user", content: [ { type: "text", text: "before" }, { type: "image", mimeType: "image/png", data: "aW1hZ2U=" }, { type: "video", mimeType: "video/mp4", data: "dmlkZW8=" }, { type: "text", text: "after" }, ], timestamp: 1, }, ], }; let callerPayload: unknown; const caller = vi.fn((payload: unknown) => (callerPayload = payload)); const stream = await wrapped(normalized, context as never, { apiKey: "test-key", maxRetries: 0, onPayload: caller, }); let streamError: unknown; for await (const event of stream) { if (event.type === "error") { streamError = event.error; } } expect(caller, JSON.stringify(streamError)).toHaveBeenCalledOnce(); expect(JSON.stringify(callerPayload)).not.toContain("/private/"); expect(JSON.stringify(callerPayload)).toContain("data:video/mp4;base64,dmlkZW8="); expect(requestBody, JSON.stringify(streamError)).toBeDefined(); const messages = requestBody?.messages as Array<{ content?: Array> }>; expect(messages[0]?.content).toEqual([ { type: "text", text: "before" }, { type: "image_url", image_url: { url: "data:image/png;base64,aW1hZ2U=" } }, { type: "video_url", video_url: { url: "data:video/mp4;base64,dmlkZW8=" } }, { type: "text", text: "after" }, ]); } finally { await new Promise((resolve, reject) => { server.close((error) => { if (error) { reject(error); } else { resolve(); } }); }); } }); });