diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index f21a31410bed..7a689bf40b28 100644 --- a/config/assertion-safety-baseline.txt +++ b/config/assertion-safety-baseline.txt @@ -2884,7 +2884,6 @@ src/gateway/conversation-turn.ts 1 src/gateway/desktop/host-guidance.ts 1 src/gateway/desktop/managed-linux.ts 1 src/gateway/desktop/node-source-context.ts 1 -src/gateway/embeddings-http.ts 1 src/gateway/exec-approval-manager.ts 7 src/gateway/gateway-cli-backend.live-helpers.ts 4 src/gateway/gateway-cli-backend.live-probe-helpers.ts 7 @@ -2911,7 +2910,7 @@ src/gateway/node-pairing-ssh-verify.ts 1 src/gateway/node-registry-private.ts 3 src/gateway/node-registry.ts 27 src/gateway/openai-agent-run-usage.ts 1 -src/gateway/openai-http.ts 30 +src/gateway/openai-http.ts 29 src/gateway/openresponses-http.ts 4 src/gateway/operator-approval-snapshot.ts 4 src/gateway/operator-approval-store.ts 8 diff --git a/src/gateway/embeddings-http.test.ts b/src/gateway/embeddings-http.test.ts index c34780d4ce2d..54c72f4afc56 100644 --- a/src/gateway/embeddings-http.test.ts +++ b/src/gateway/embeddings-http.test.ts @@ -440,6 +440,22 @@ describe("OpenAI-compatible embeddings HTTP API (e2e)", () => { await expectInvalidEmbeddingRequest(res); }); + it.each([ + { name: "unsupported encoding", option: { encoding_format: "hex" } }, + { name: "numeric encoding", option: { encoding_format: 1 } }, + { name: "zero dimensions", option: { dimensions: 0 } }, + { name: "negative dimensions", option: { dimensions: -1 } }, + { name: "fractional dimensions", option: { dimensions: 1.5 } }, + { name: "string dimensions", option: { dimensions: "768" } }, + { name: "unsafe dimensions", option: { dimensions: Number.MAX_SAFE_INTEGER + 1 } }, + ])("rejects $name before creating an embedding provider", async ({ option }) => { + const providersCreatedBefore = createEmbeddingProviderMock.mock.calls.length; + const res = await postEmbeddings({ model: "openclaw/default", input: "hello", ...option }); + + await expectInvalidEmbeddingRequest(res); + expect(createEmbeddingProviderMock).toHaveBeenCalledTimes(providersCreatedBefore); + }); + it.each([ { name: "an empty string", input: "" }, { name: "an empty batch", input: [] }, diff --git a/src/gateway/embeddings-http.ts b/src/gateway/embeddings-http.ts index 1609b4300f06..5cbbbebf66b6 100644 --- a/src/gateway/embeddings-http.ts +++ b/src/gateway/embeddings-http.ts @@ -6,6 +6,7 @@ import { normalizeLowercaseStringOrEmpty, normalizeOptionalString, } from "@openclaw/normalization-core/string-coerce"; +import { z } from "zod"; import { resolveAgentDir } from "../agents/agent-scope.js"; import { resolveMemorySearchConfig } from "../agents/memory-search.js"; import { createConfiguredProviderLocalServiceAcquirer } from "../agents/provider-local-service.js"; @@ -40,13 +41,13 @@ type OpenAiEmbeddingsHttpOptions = { rateLimiter?: AuthRateLimiter; }; -type EmbeddingsRequest = { - model?: unknown; - input?: unknown; - encoding_format?: unknown; - dimensions?: unknown; - user?: unknown; -}; +const EmbeddingsRequestSchema = z.object({ + model: z.string().optional(), + input: z.union([z.string(), z.array(z.string())]).optional(), + encoding_format: z.enum(["float", "base64"]).optional(), + dimensions: z.number().int().positive().optional(), + user: z.string().optional(), +}); const DEFAULT_EMBEDDINGS_BODY_BYTES = 5 * 1024 * 1024; const MAX_EMBEDDING_INPUTS = 128; @@ -174,10 +175,6 @@ export async function drainRetainedOpenAiEmbeddingProviders(): Promise { } } -function coerceRequest(value: unknown): EmbeddingsRequest { - return value && typeof value === "object" ? (value as EmbeddingsRequest) : {}; -} - function resolveInputTexts(input: unknown): string[] | null { if (typeof input === "string") { return [input]; @@ -331,7 +328,18 @@ export async function handleOpenAiEmbeddingsHttpRequest( return true; } - const payload = coerceRequest(handled.body); + const parsed = EmbeddingsRequestSchema.safeParse(handled.body); + if (!parsed.success) { + const issue = parsed.error.issues[0]; + sendJson(res, 400, { + error: { + message: issue ? `${issue.path.join(".")}: ${issue.message}` : "Invalid request body", + type: "invalid_request_error", + }, + }); + return true; + } + const payload = parsed.data; const requestModel = normalizeOptionalString(payload.model) ?? ""; if (!requestModel) { sendJson(res, 400, { @@ -425,10 +433,7 @@ export async function handleOpenAiEmbeddingsHttpRequest( memorySearch: memorySearch ? { ...memorySearch, - outputDimensionality: - typeof payload.dimensions === "number" && payload.dimensions > 0 - ? Math.floor(payload.dimensions) - : memorySearch.outputDimensionality, + outputDimensionality: payload.dimensions ?? memorySearch.outputDimensionality, } : undefined, }), diff --git a/src/gateway/openai-http.test.ts b/src/gateway/openai-http.test.ts index ad5c80d6c31d..84c8d63eb7ef 100644 --- a/src/gateway/openai-http.test.ts +++ b/src/gateway/openai-http.test.ts @@ -2755,6 +2755,66 @@ describe("OpenAI-compatible HTTP API (e2e)", () => { }, ); + it.each([ + { name: "a string", stream: "false" }, + { name: "a number", stream: 1 }, + { name: "an array", stream: [] }, + { name: "an object", stream: {} }, + ])("rejects $name stream mode before dispatching an agent", async ({ stream }) => { + agentCommandMock.mockClear(); + const response = await postChatCompletions(enabledPort, { + model: "openclaw", + stream, + messages: [{ role: "user", content: "hi" }], + }); + + expect(response.status).toBe(400); + expect(response.headers.get("content-type")).toContain("application/json"); + await expect(response.json()).resolves.toMatchObject({ + error: { type: "invalid_request_error", message: expect.stringContaining("stream") }, + }); + expect(agentCommandMock).not.toHaveBeenCalled(); + }); + + it("keeps one created timestamp across every streamed completion chunk", async () => { + let now = 1_700_000_000_000; + const nowSpy = vi.spyOn(Date, "now").mockImplementation(() => (now += 1_000)); + try { + agentCommandMock.mockClear(); + agentCommandMock.mockImplementationOnce((async (opts: unknown) => { + const runId = (opts as { runId?: string }).runId ?? ""; + emitAgentEvent({ runId, stream: "assistant", data: { delta: "commentary" } }); + return { + payloads: [{ text: "commentary" }], + meta: { + stopReason: "tool_calls", + pendingToolCalls: [ + { id: "call_1", name: "lookup", arguments: JSON.stringify({ q: "x".repeat(300) }) }, + ], + agentMeta: { usage: { input: 4, output: 1, total: 5 } }, + }, + }; + }) as never); + + const response = await postChatCompletions(enabledPort, { + model: "openclaw", + stream: true, + stream_options: { include_usage: true }, + messages: [{ role: "user", content: "hi" }], + }); + expect(response.status).toBe(200); + const chunks = parseSseDataLines(await response.text()) + .filter((data) => data !== "[DONE]") + .map((data) => JSON.parse(data) as { created?: number }); + + expect(chunks.length).toBeGreaterThanOrEqual(6); + expect(chunks.every((chunk) => typeof chunk.created === "number")).toBe(true); + expect(new Set(chunks.map((chunk) => chunk.created))).toEqual(new Set([chunks[0]?.created])); + } finally { + nowSpy.mockRestore(); + } + }); + it("streams SSE chunks when stream=true", async () => { const port = enabledPort; try { diff --git a/src/gateway/openai-http.ts b/src/gateway/openai-http.ts index 3692def56a3a..0e86d179d644 100644 --- a/src/gateway/openai-http.ts +++ b/src/gateway/openai-http.ts @@ -9,6 +9,7 @@ import { normalizeOptionalString, } from "@openclaw/normalization-core/string-coerce"; import { avoidTrailingHighSurrogateBreak } from "@openclaw/normalization-core/utf16-slice"; +import { z } from "zod"; import type { AdmittedRunContext } from "../agents/admitted-run-context.js"; import { isClientToolNameConflictError } from "../agents/agent-tool-definition-adapter.js"; import type { AgentStreamParams, ClientToolDefinition } from "../agents/command/shared-types.js"; @@ -102,25 +103,26 @@ type OpenAiChatMessage = { stopReason?: unknown; }; -type OpenAiChatCompletionRequest = { - model?: unknown; - stream?: unknown; - // Naming/style reference: src/agents/openai-transport-stream.ts:1262-1273 - stream_options?: unknown; - tools?: unknown; - tool_choice?: unknown; - messages?: unknown; - user?: unknown; - max_tokens?: unknown; - max_completion_tokens?: unknown; - temperature?: unknown; - top_p?: unknown; - response_format?: unknown; - frequency_penalty?: unknown; - presence_penalty?: unknown; - seed?: unknown; - stop?: unknown; -}; +const OpenAiChatCompletionRequestSchema = z.object({ + model: z.string().optional(), + stream: z.boolean().nullish(), + stream_options: z.object({ include_usage: z.boolean().optional() }).passthrough().nullish(), + tools: z.array(z.unknown()).optional(), + tool_choice: z.unknown().optional(), + messages: z.array(z.unknown()).optional(), + user: z.string().optional(), + max_tokens: z.number().int().positive().nullish(), + max_completion_tokens: z.number().int().positive().nullish(), + temperature: z.number().nullish(), + top_p: z.number().nullish(), + response_format: z.unknown().optional(), + frequency_penalty: z.number().nullish(), + presence_penalty: z.number().nullish(), + seed: z.number().nullish(), + stop: z.union([z.string(), z.array(z.string())]).nullish(), +}); + +type OpenAiChatCompletionRequest = z.infer; const DEFAULT_OPENAI_CHAT_COMPLETIONS_BODY_BYTES = 20 * 1024 * 1024; const DEFAULT_OPENAI_MAX_IMAGE_PARTS = 8; @@ -281,11 +283,13 @@ function applyChatToolChoice(params: { tools: ClientToolDefinition[]; toolChoice throw new Error(`tool_choice ${choiceType} is not supported`); } -function writeAssistantRoleChunk(res: ServerResponse, params: { runId: string; model: string }) { +type ChatCompletionStreamIdentity = { runId: string; model: string; created: number }; + +function writeAssistantRoleChunk(res: ServerResponse, params: ChatCompletionStreamIdentity) { writeSse(res, { id: params.runId, object: "chat.completion.chunk", - created: Math.floor(Date.now() / 1000), + created: params.created, model: params.model, choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }], }); @@ -293,12 +297,12 @@ function writeAssistantRoleChunk(res: ServerResponse, params: { runId: string; m function writeAssistantContentChunk( res: ServerResponse, - params: { runId: string; model: string; content: string }, + params: ChatCompletionStreamIdentity & { content: string }, ) { writeSse(res, { id: params.runId, object: "chat.completion.chunk", - created: Math.floor(Date.now() / 1000), + created: params.created, model: params.model, choices: [ { @@ -312,12 +316,12 @@ function writeAssistantContentChunk( function writeAssistantFinishChunk( res: ServerResponse, - params: { runId: string; model: string; finishReason: "stop" | "tool_calls" }, + params: ChatCompletionStreamIdentity & { finishReason: "stop" | "tool_calls" }, ) { writeSse(res, { id: params.runId, object: "chat.completion.chunk", - created: Math.floor(Date.now() / 1000), + created: params.created, model: params.model, choices: [ { @@ -349,9 +353,7 @@ function splitArgumentsForStreaming(argumentsValue: string): string[] { function writeAssistantToolCallsIncrementalChunks( res: ServerResponse, - params: { - runId: string; - model: string; + params: ChatCompletionStreamIdentity & { toolCalls: Array<{ id: string; name: string; arguments: string }>; }, ) { @@ -359,7 +361,7 @@ function writeAssistantToolCallsIncrementalChunks( writeSse(res, { id: params.runId, object: "chat.completion.chunk", - created: Math.floor(Date.now() / 1000), + created: params.created, model: params.model, choices: [ { @@ -383,7 +385,7 @@ function writeAssistantToolCallsIncrementalChunks( writeSse(res, { id: params.runId, object: "chat.completion.chunk", - created: Math.floor(Date.now() / 1000), + created: params.created, model: params.model, choices: [ { @@ -406,16 +408,14 @@ function writeAssistantToolCallsIncrementalChunks( function writeUsageChunk( res: ServerResponse, - params: { - runId: string; - model: string; + params: ChatCompletionStreamIdentity & { usage: OpenAiChatCompletionsUsage; }, ) { writeSse(res, { id: params.runId, object: "chat.completion.chunk", - created: Math.floor(Date.now() / 1000), + created: params.created, model: params.model, choices: [], usage: params.usage, @@ -739,13 +739,6 @@ function buildAgentPrompt( }; } -function coerceRequest(val: unknown): OpenAiChatCompletionRequest { - if (!val || typeof val !== "object") { - return {}; - } - return val as OpenAiChatCompletionRequest; -} - function resolveAgentResponseText(result: unknown): string { const payloads = (result as { payloads?: Array<{ text?: string }> } | null)?.payloads; if (!Array.isArray(payloads) || payloads.length === 0) { @@ -899,8 +892,19 @@ export async function handleOpenAiHttpRequest( return true; } const senderIsOwner = resolveOpenAiCompatibleHttpSenderIsOwner(req, handled.requestAuth); - const payload = coerceRequest(handled.body); - const stream = Boolean(payload.stream); + const parsed = OpenAiChatCompletionRequestSchema.safeParse(handled.body); + if (!parsed.success) { + const issue = parsed.error.issues[0]; + sendJson(res, 400, { + error: { + message: issue ? `${issue.path.join(".")}: ${issue.message}` : "Invalid request body", + type: "invalid_request_error", + }, + }); + return true; + } + const payload = parsed.data; + const stream = payload.stream === true; const streamIncludeUsage = stream && resolveIncludeUsageForStreaming(payload); const model = typeof payload.model === "string" ? payload.model : "openclaw"; const user = typeof payload.user === "string" ? payload.user : undefined; @@ -1091,6 +1095,8 @@ export async function handleOpenAiHttpRequest( } const runId = `chatcmpl_${randomUUID()}`; + const created = Math.floor(Date.now() / 1000); + const streamIdentity = { runId, model, created }; const deps = createDefaultDeps(); const abortController = new AbortController(); const mergedExtraSystemPrompt = [prompt.extraSystemPrompt, toolChoicePrompt] @@ -1164,7 +1170,7 @@ export async function handleOpenAiHttpRequest( sendJson(res, 200, { id: runId, object: "chat.completion", - created: Math.floor(Date.now() / 1000), + created, model, choices: [ { @@ -1190,7 +1196,7 @@ export async function handleOpenAiHttpRequest( sendJson(res, 200, { id: runId, object: "chat.completion", - created: Math.floor(Date.now() / 1000), + created, model, choices: [ { @@ -1270,11 +1276,11 @@ export async function handleOpenAiHttpRequest( stopWatchingDisconnect(); unsubscribe(); if (!wroteStopChunk) { - writeAssistantFinishChunk(res, { runId, model, finishReason: finalizeFinishReason }); + writeAssistantFinishChunk(res, { ...streamIdentity, finishReason: finalizeFinishReason }); wroteStopChunk = true; } if (streamIncludeUsage && finalUsage) { - writeUsageChunk(res, { runId, model, usage: finalUsage }); + writeUsageChunk(res, { ...streamIdentity, usage: finalUsage }); } writeDone(res); res.end(); @@ -1347,13 +1353,12 @@ export async function handleOpenAiHttpRequest( if (!wroteRole) { wroteRole = true; - writeAssistantRoleChunk(res, { runId, model }); + writeAssistantRoleChunk(res, streamIdentity); } sawAssistantDelta = true; writeAssistantContentChunk(res, { - runId, - model, + ...streamIdentity, content, }); return; @@ -1408,7 +1413,7 @@ export async function handleOpenAiHttpRequest( }); wroteRole = true; - writeAssistantRoleChunk(res, { runId, model }); + writeAssistantRoleChunk(res, streamIdentity); void (async () => { try { @@ -1459,7 +1464,7 @@ export async function handleOpenAiHttpRequest( if (stopReason === "tool_calls" && pendingToolCalls && pendingToolCalls.length > 0) { if (!wroteRole) { wroteRole = true; - writeAssistantRoleChunk(res, { runId, model }); + writeAssistantRoleChunk(res, streamIdentity); } if (!sawAssistantDelta) { const commentary = @@ -1469,15 +1474,13 @@ export async function handleOpenAiHttpRequest( if (commentary) { sawAssistantDelta = true; writeAssistantContentChunk(res, { - runId, - model, + ...streamIdentity, content: commentary, }); } } writeAssistantToolCallsIncrementalChunks(res, { - runId, - model, + ...streamIdentity, toolCalls: pendingToolCalls, }); requestFinalize("tool_calls"); @@ -1487,7 +1490,7 @@ export async function handleOpenAiHttpRequest( if (!sawAssistantDelta) { if (!wroteRole) { wroteRole = true; - writeAssistantRoleChunk(res, { runId, model }); + writeAssistantRoleChunk(res, streamIdentity); } const content = @@ -1498,8 +1501,7 @@ export async function handleOpenAiHttpRequest( sawAssistantDelta = true; writeAssistantContentChunk(res, { - runId, - model, + ...streamIdentity, content, }); } diff --git a/src/gateway/openresponses-http.test.ts b/src/gateway/openresponses-http.test.ts index 4af733669128..385a0c6b361e 100644 --- a/src/gateway/openresponses-http.test.ts +++ b/src/gateway/openresponses-http.test.ts @@ -1378,6 +1378,44 @@ describe("OpenResponses HTTP API (e2e)", () => { expect(prompt).toContain('<< { + let now = 1_700_000_000_000; + const nowSpy = vi.spyOn(Date, "now").mockImplementation(() => (now += 1_000)); + try { + agentCommandMock.mockClear(); + agentCommandMock.mockImplementationOnce((async (opts: unknown) => + buildAssistantDeltaResult({ + opts, + emit: emitAgentEvent, + deltas: ["hello"], + text: "hello", + })) as never); + + const response = await postResponses(enabledPort, { + stream: true, + model: "openclaw", + input: "hi", + }); + expect(response.status).toBe(200); + const createdAt = parseSseEvents(await response.text()) + .filter((event) => + ["response.created", "response.in_progress", "response.completed"].includes( + event.event ?? "", + ), + ) + .map( + (event) => + (parseSseData(event) as { response?: { created_at?: number } }).response?.created_at, + ); + + expect(createdAt).toHaveLength(3); + expect(createdAt.every((value) => typeof value === "number")).toBe(true); + expect(new Set(createdAt)).toEqual(new Set([createdAt[0]])); + } finally { + nowSpy.mockRestore(); + } + }); + it("streams OpenResponses SSE events", async () => { const port = enabledPort; try { diff --git a/src/gateway/openresponses-http.ts b/src/gateway/openresponses-http.ts index 3b84dbbcce9e..fad5d8de1425 100644 --- a/src/gateway/openresponses-http.ts +++ b/src/gateway/openresponses-http.ts @@ -378,6 +378,7 @@ function resolveStopReasonAndPendingToolCalls(meta: unknown): { function createResponseResource(params: { id: string; + createdAt: number; model: string; status: ResponseResource["status"]; output: OutputItem[]; @@ -387,7 +388,7 @@ function createResponseResource(params: { return { id: params.id, object: "response", - created_at: Math.floor(Date.now() / 1000), + created_at: params.createdAt, status: params.status, model: params.model, output: params.output, @@ -717,6 +718,7 @@ export async function handleOpenResponsesHttpRequest( } const responseId = `resp_${randomUUID()}`; + const responseIdentity = { id: responseId, createdAt: Math.floor(Date.now() / 1000) }; const rememberResponseSession = () => storeResponseSession(responseId, sessionKey, responseSessionScope); const outputItemId = `msg_${randomUUID()}`; @@ -775,7 +777,7 @@ export async function handleOpenResponsesHttpRequest( !isToolChoiceConstraintSatisfied({ constraint: toolChoiceConstraint, pendingToolCalls }) ) { const failed = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "failed", output: [], @@ -819,7 +821,7 @@ export async function handleOpenResponsesHttpRequest( } const response = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "completed", output, @@ -831,7 +833,7 @@ export async function handleOpenResponsesHttpRequest( } const response = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "completed", output: [ @@ -854,7 +856,7 @@ export async function handleOpenResponsesHttpRequest( logWarn(`openresponses: non-stream response failed: ${String(err)}`); if (isClientToolNameConflictError(err)) { const response = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "failed", output: [], @@ -864,7 +866,7 @@ export async function handleOpenResponsesHttpRequest( return true; } const response = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "failed", output: [], @@ -873,7 +875,7 @@ export async function handleOpenResponsesHttpRequest( const mapped = resolveOpenAiCompatError(err); if (mapped) { const mappedResponse = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "failed", output: [], @@ -974,7 +976,7 @@ export async function handleOpenResponsesHttpRequest( }); const finalResponse = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: finalizeRequested.status, output: [completedItem], @@ -1034,7 +1036,7 @@ export async function handleOpenResponsesHttpRequest( rememberResponseSession(); finalizeFailedResponse( createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "failed", output: [], @@ -1049,7 +1051,7 @@ export async function handleOpenResponsesHttpRequest( // Send initial events const initialResponse = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "in_progress", output: [], @@ -1218,7 +1220,7 @@ export async function handleOpenResponsesHttpRequest( rememberResponseSession(); finalizeFailedResponse( createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "failed", output: [], @@ -1252,7 +1254,7 @@ export async function handleOpenResponsesHttpRequest( !isToolChoiceConstraintSatisfied({ constraint: toolChoiceConstraint, pendingToolCalls }) ) { const failed = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "failed", output: [], @@ -1358,7 +1360,7 @@ export async function handleOpenResponsesHttpRequest( } const completedResponse = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "completed", output: [completedItem, ...functionCallItems], @@ -1405,7 +1407,7 @@ export async function handleOpenResponsesHttpRequest( finalUsage = finalUsage ?? createEmptyUsage(); if (isClientToolNameConflictError(err)) { const errorResponse = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "failed", output: [], @@ -1417,7 +1419,7 @@ export async function handleOpenResponsesHttpRequest( return; } const errorResponse = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "failed", output: [], @@ -1428,7 +1430,7 @@ export async function handleOpenResponsesHttpRequest( const mapped = resolveOpenAiCompatError(err); if (mapped) { const mappedResponse = createResponseResource({ - id: responseId, + ...responseIdentity, model, status: "failed", output: [],