From 5ea80c8d80fded3d08c8c7332adf210a7aab4ffe Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 4 Jul 2026 09:08:37 -0400 Subject: [PATCH] fix: improve agent runtime correctness from upstream Pi (#99949) * fix: improve agent runtime correctness * test: track agent tool temp directories * test: update helper routing expectation --- npm-shrinkwrap.json | 19 +- package.json | 3 +- packages/agent-core/src/agent-loop.test.ts | 54 +++++ packages/agent-core/src/agent-loop.ts | 8 + .../src/harness/compaction/compaction.test.ts | 79 ++++++- .../src/harness/compaction/compaction.ts | 31 ++- pnpm-lock.yaml | 29 +-- scripts/lib/dependency-ownership.json | 6 + src/agents/embedded-agent-helpers/errors.ts | 2 + .../provider-error-patterns.test.ts | 8 + src/agents/sessions/agent-session.ts | 7 +- src/agents/sessions/tools/bash.test.ts | 22 +- src/agents/sessions/tools/bash.ts | 5 + src/agents/sessions/tools/edit-diff.ts | 129 ++++++++-- src/agents/sessions/tools/edit.test.ts | 88 ++++++- src/agents/sessions/tools/edit.ts | 24 +- src/agents/sessions/tools/find.fd.test.ts | 64 ++++- src/agents/sessions/tools/find.ts | 31 ++- src/agents/sessions/tools/read.test.ts | 41 +++- src/agents/sessions/tools/read.ts | 49 ++-- src/agents/utils/child-process.test.ts | 70 ++++++ src/agents/utils/child-process.ts | 35 ++- src/agents/utils/image-resize.ts | 69 ++++++ src/agents/utils/mime.ts | 50 ++++ .../openai-chatgpt-responses.test.ts | 211 +++++++++++++++++ src/llm/providers/openai-chatgpt-responses.ts | 220 ++++++++++++------ src/llm/utils/overflow.test.ts | 34 +++ src/llm/utils/overflow.ts | 9 + src/llm/utils/retry.test.ts | 57 +++++ src/llm/utils/retry.ts | 61 +++++ .../image-ops.rastermill-adapter.test.ts | 26 +++ src/media/image-ops.ts | 27 +++ src/media/photon.runtime.ts | 12 + src/scripts/test-projects.test.ts | 2 + 34 files changed, 1389 insertions(+), 193 deletions(-) create mode 100644 src/agents/utils/child-process.test.ts create mode 100644 src/llm/utils/overflow.test.ts create mode 100644 src/llm/utils/retry.test.ts create mode 100644 src/llm/utils/retry.ts create mode 100644 src/media/photon.runtime.ts diff --git a/npm-shrinkwrap.json b/npm-shrinkwrap.json index 6378bad9fa55..2bf88798aaf9 100644 --- a/npm-shrinkwrap.json +++ b/npm-shrinkwrap.json @@ -14,7 +14,7 @@ "@anthropic-ai/sdk": "0.100.1", "@clack/core": "1.3.1", "@clack/prompts": "1.4.0", - "@earendil-works/pi-tui": "0.78.0", + "@earendil-works/pi-tui": "0.80.3", "@google/genai": "2.7.0", "@grammyjs/runner": "2.0.3", "@grammyjs/transformer-throttler": "1.2.1", @@ -25,6 +25,7 @@ "@mozilla/readability": "0.6.0", "@openclaw/fs-safe": "0.3.0", "@openclaw/proxyline": "0.3.3", + "@silvia-odwyer/photon-node": "0.3.4", "chalk": "5.6.2", "chokidar": "5.0.0", "clawpdf": "0.3.0", @@ -153,13 +154,13 @@ } }, "node_modules/@earendil-works/pi-tui": { - "version": "0.78.0", - "resolved": "https://registry.npmjs.org/@earendil-works/pi-tui/-/pi-tui-0.78.0.tgz", - "integrity": "sha512-3a705FnsVVUhAyceShNB3kS2rpxcxLcx+hqB0u6MMMpHwQGbW+m++MqA6r7eOzq/8FLx5e3vDh38h/SVTk2qzw==", + "version": "0.80.3", + "resolved": "https://registry.npmjs.org/@earendil-works/pi-tui/-/pi-tui-0.80.3.tgz", + "integrity": "sha512-2BJI6qwRQfnM0Q7seL1+SbacU/jRRjBnN7Hu3n9BjAn7/s5FaBNnvdD1qBQYRsFTHfjqMaDsjYqanPyqwXj99w==", "license": "MIT", "dependencies": { "get-east-asian-width": "1.6.0", - "marked": "15.0.12" + "marked": "18.0.5" }, "engines": { "node": ">=22.19.0" @@ -2110,15 +2111,15 @@ } }, "node_modules/marked": { - "version": "15.0.12", - "resolved": "https://registry.npmjs.org/marked/-/marked-15.0.12.tgz", - "integrity": "sha512-8dD6FusOQSrpv9Z1rdNMdlSgQOIP880DHqnohobOmYLElGEqAL/JvxvuxZO16r4HtjTlfPRDC1hbvxC9dPN2nA==", + "version": "18.0.5", + "resolved": "https://registry.npmjs.org/marked/-/marked-18.0.5.tgz", + "integrity": "sha512-S6GcvALHg6K4ohtu4E7x0a1AqhAjp6cV8KhLSyN9qVapnzJkusVBxZRcIU9AeYsbe6P1hKDusSbEOzGyyuce6w==", "license": "MIT", "bin": { "marked": "bin/marked.js" }, "engines": { - "node": ">= 18" + "node": ">= 20" } }, "node_modules/math-intrinsics": { diff --git a/package.json b/package.json index c2033685a2e8..04b271a15050 100644 --- a/package.json +++ b/package.json @@ -1972,7 +1972,7 @@ "@anthropic-ai/sdk": "0.100.1", "@clack/core": "1.3.1", "@clack/prompts": "1.4.0", - "@earendil-works/pi-tui": "0.78.0", + "@earendil-works/pi-tui": "0.80.3", "@google/genai": "2.7.0", "@grammyjs/runner": "2.0.3", "@grammyjs/transformer-throttler": "1.2.1", @@ -1983,6 +1983,7 @@ "@mozilla/readability": "0.6.0", "@openclaw/fs-safe": "0.3.0", "@openclaw/proxyline": "0.3.3", + "@silvia-odwyer/photon-node": "0.3.4", "chalk": "5.6.2", "chokidar": "5.0.0", "clawpdf": "0.3.0", diff --git a/packages/agent-core/src/agent-loop.test.ts b/packages/agent-core/src/agent-loop.test.ts index 495df409e640..f12fe2de13ad 100644 --- a/packages/agent-core/src/agent-loop.test.ts +++ b/packages/agent-core/src/agent-loop.test.ts @@ -789,6 +789,60 @@ describe("agentLoop tool termination", () => { ).toBe(true); }); + it("ignores progress updates after a tool execution settles", async () => { + let delayedUpdate: ((result: AgentToolResult) => void) | undefined; + const tool: AgentTool = { + name: "delayed_tool", + label: "delayed_tool", + description: "captures progress callbacks", + parameters: Type.Object({}, { additionalProperties: false }), + execute: async (_toolCallId, _args, _signal, onUpdate) => { + delayedUpdate = onUpdate; + onUpdate?.({ + content: [{ type: "text", text: "running" }], + details: { status: "running" }, + }); + return { + content: [{ type: "text", text: "done" }], + details: { status: "done" }, + terminate: true, + }; + }, + }; + const streamFn: StreamFn = () => { + const stream = createAssistantMessageEventStream(); + queueMicrotask(() => { + const message = makeAssistantMessage([ + { type: "toolCall", id: "call-delayed", name: tool.name, arguments: {} }, + ]); + stream.push({ type: "done", reason: "toolUse", message }); + stream.end(); + }); + return stream; + }; + + const events = await collectEvents( + agentLoop( + [{ role: "user", content: "run", timestamp: 1 }], + { systemPrompt: "", messages: [], tools: [tool] }, + { ...config, toolExecution: "sequential" }, + undefined, + streamFn, + ), + ); + const countAfterRun = events.length; + delayedUpdate?.({ + content: [{ type: "text", text: "late" }], + details: { status: "late" }, + }); + await new Promise((resolve) => { + setTimeout(resolve, 0); + }); + + expect(events).toHaveLength(countAfterRun); + expect(events.filter((event) => event.type === "tool_execution_update")).toHaveLength(1); + }); + it("continues after a side-effect tool result when afterToolCall records it without terminate", async () => { const executed: string[] = []; let turn = 0; diff --git a/packages/agent-core/src/agent-loop.ts b/packages/agent-core/src/agent-loop.ts index c77bb3abf13f..a8609462c11d 100644 --- a/packages/agent-core/src/agent-loop.ts +++ b/packages/agent-core/src/agent-loop.ts @@ -960,6 +960,7 @@ async function executePreparedToolCall( emit: AgentEventSink, ): Promise { const updateEvents: Promise[] = []; + let acceptingUpdates = true; try { const result = await prepared.tool.execute( @@ -967,6 +968,9 @@ async function executePreparedToolCall( prepared.args as never, signal, (partialResult) => { + if (!acceptingUpdates) { + return; + } updateEvents.push( Promise.resolve( emit({ @@ -983,14 +987,18 @@ async function executePreparedToolCall( ); }, ); + acceptingUpdates = false; await Promise.all(updateEvents); return { result, isError: false }; } catch (error) { + acceptingUpdates = false; await Promise.all(updateEvents); return { result: createErrorToolResult(error instanceof Error ? error.message : String(error)), isError: true, }; + } finally { + acceptingUpdates = false; } } diff --git a/packages/agent-core/src/harness/compaction/compaction.test.ts b/packages/agent-core/src/harness/compaction/compaction.test.ts index 7245d3db05ba..e8b948697da2 100644 --- a/packages/agent-core/src/harness/compaction/compaction.test.ts +++ b/packages/agent-core/src/harness/compaction/compaction.test.ts @@ -1,7 +1,8 @@ import { describe, expect, it, vi } from "vitest"; import { createAssistantMessageEventStream } from "../../llm.js"; import type { AssistantMessage, Model, StreamFn } from "../../llm.js"; -import { generateSummary } from "./compaction.js"; +import { compact, generateSummary } from "./compaction.js"; +import { createFileOps } from "./utils.js"; describe("generateSummary thinking options", () => { it("maps explicit Fable off to low effort for compaction", async () => { @@ -35,8 +36,10 @@ describe("generateSummary thinking options", () => { stopReason: "stop", timestamp: 1, }; - const streamFn = vi.fn((_model, _context, options) => { + const streamFn = vi.fn((_model, context, options) => { expect(options?.reasoning).toBe("low"); + expect(context.systemPrompt).toContain("user and an AI assistant"); + expect(context.systemPrompt).not.toContain("AI coding assistant"); const stream = createAssistantMessageEventStream(); stream.push({ type: "done", reason: "stop", message: summaryMessage }); stream.end(); @@ -60,3 +63,75 @@ describe("generateSummary thinking options", () => { expect(streamFn).toHaveBeenCalledOnce(); }); }); + +describe("split-turn compaction", () => { + it("serializes history and turn-prefix summaries", async () => { + const model: Model = { + id: "summary-model", + name: "Summary Model", + api: "test-api", + provider: "test-provider", + baseUrl: "https://example.test", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 100_000, + maxTokens: 8_000, + }; + let active = 0; + let maxActive = 0; + let callCount = 0; + const streamFn = vi.fn(() => { + active++; + maxActive = Math.max(maxActive, active); + callCount++; + const stream = createAssistantMessageEventStream(); + setTimeout(() => { + active--; + const message: AssistantMessage = { + role: "assistant", + content: [{ type: "text", text: `summary-${callCount}` }], + api: model.api, + provider: model.provider, + model: model.id, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: 1, + }; + stream.push({ type: "done", reason: "stop", message }); + stream.end(); + }, 5); + return stream; + }); + + const result = await compact( + { + firstKeptEntryId: "kept-entry", + messagesToSummarize: [{ role: "user", content: "history", timestamp: 1 }], + turnPrefixMessages: [{ role: "user", content: "prefix", timestamp: 2 }], + isSplitTurn: true, + tokensBefore: 100, + fileOps: createFileOps(), + settings: { enabled: true, reserveTokens: 1_000, keepRecentTokens: 100 }, + }, + model, + undefined, + undefined, + undefined, + undefined, + undefined, + streamFn, + ); + + expect(result.ok).toBe(true); + expect(streamFn).toHaveBeenCalledTimes(2); + expect(maxActive).toBe(1); + }); +}); diff --git a/packages/agent-core/src/harness/compaction/compaction.ts b/packages/agent-core/src/harness/compaction/compaction.ts index b264795b5b41..c778727d8f64 100644 --- a/packages/agent-core/src/harness/compaction/compaction.ts +++ b/packages/agent-core/src/harness/compaction/compaction.ts @@ -441,7 +441,7 @@ export function findCutPoint( }; } -export const SUMMARIZATION_SYSTEM_PROMPT = `You are a context summarization assistant. Your task is to read a conversation between a user and an AI coding assistant, then produce a structured summary following the exact format specified. +export const SUMMARIZATION_SYSTEM_PROMPT = `You are a context summarization assistant. Your task is to read a conversation between a user and an AI assistant, then produce a structured summary following the exact format specified. Do NOT continue the conversation. Do NOT respond to any questions in the conversation. ONLY output the structured summary.`; @@ -800,9 +800,9 @@ export async function compact( let summary: string; if (isSplitTurn && turnPrefixMessages.length > 0) { - const [historyResult, turnPrefixResult] = await Promise.all([ + const historyResult = messagesToSummarize.length > 0 - ? generateSummary( + ? await generateSummary( messagesToSummarize, model, settings.reserveTokens, @@ -815,22 +815,21 @@ export async function compact( streamFn, runtime, ) - : Promise.resolve(ok("No prior history.")), - generateTurnPrefixSummary( - turnPrefixMessages, - model, - settings.reserveTokens, - apiKey, - headers, - signal, - thinkingLevel, - streamFn, - runtime, - ), - ]); + : ok("No prior history."); if (!historyResult.ok) { return err(historyResult.error); } + const turnPrefixResult = await generateTurnPrefixSummary( + turnPrefixMessages, + model, + settings.reserveTokens, + apiKey, + headers, + signal, + thinkingLevel, + streamFn, + runtime, + ); if (!turnPrefixResult.ok) { return err(turnPrefixResult.error); } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index ea2ad91007f9..e5f07e3d4949 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -56,8 +56,8 @@ importers: specifier: 1.4.0 version: 1.4.0 '@earendil-works/pi-tui': - specifier: 0.78.0 - version: 0.78.0 + specifier: 0.80.3 + version: 0.80.3 '@google/genai': specifier: 2.7.0 version: 2.7.0(@modelcontextprotocol/sdk@1.29.0(zod@4.4.3)) @@ -88,6 +88,9 @@ importers: '@openclaw/proxyline': specifier: 0.3.3 version: 0.3.3(undici@8.5.0) + '@silvia-odwyer/photon-node': + specifier: 0.3.4 + version: 0.3.4 chalk: specifier: 5.6.2 version: 5.6.2 @@ -2489,8 +2492,8 @@ packages: resolution: {integrity: sha512-3yJ255e4ag3wfZu/DSxeOZK1UtnqNxnspmLaQetGT0pDkThNZoHs+Zg6dgZZ19JEVomXygvfHn9lNpICZuYtEA==} engines: {node: '>=22.12.0'} - '@earendil-works/pi-tui@0.78.0': - resolution: {integrity: sha512-3a705FnsVVUhAyceShNB3kS2rpxcxLcx+hqB0u6MMMpHwQGbW+m++MqA6r7eOzq/8FLx5e3vDh38h/SVTk2qzw==} + '@earendil-works/pi-tui@0.80.3': + resolution: {integrity: sha512-2BJI6qwRQfnM0Q7seL1+SbacU/jRRjBnN7Hu3n9BjAn7/s5FaBNnvdD1qBQYRsFTHfjqMaDsjYqanPyqwXj99w==} engines: {node: '>=22.19.0'} '@emnapi/core@1.10.0': @@ -6129,16 +6132,16 @@ packages: markdown-table@2.0.0: resolution: {integrity: sha512-Ezda85ToJUBhM6WGaG6veasyym+Tbs3cMAw/ZhOPqXiYsr0jgocBV3j3nx+4lk47plLlIqjwuTm/ywVI+zjJ/A==} - marked@15.0.12: - resolution: {integrity: sha512-8dD6FusOQSrpv9Z1rdNMdlSgQOIP880DHqnohobOmYLElGEqAL/JvxvuxZO16r4HtjTlfPRDC1hbvxC9dPN2nA==} - engines: {node: '>= 18'} - hasBin: true - marked@18.0.4: resolution: {integrity: sha512-c/BTaKzg0G6ezQx97DAkYU7k0HM6ys0FqYeKBL6hlBByZwy+ycA1+f0vDdjMHKKeEjdgkx0GOv9Il6D+85cOqA==} engines: {node: '>= 20'} hasBin: true + marked@18.0.5: + resolution: {integrity: sha512-S6GcvALHg6K4ohtu4E7x0a1AqhAjp6cV8KhLSyN9qVapnzJkusVBxZRcIU9AeYsbe6P1hKDusSbEOzGyyuce6w==} + engines: {node: '>= 20'} + hasBin: true + math-intrinsics@1.1.0: resolution: {integrity: sha512-/IXtbwEk5HTPyEwyKX6hGkYXxM9nbj64B+ilVJnC/R6B0pH5G4V3b0pVbL7DBj4tkhBAppbQUlf6F6Xl9LHu1g==} engines: {node: '>= 0.4'} @@ -8597,10 +8600,10 @@ snapshots: - opusscript - utf-8-validate - '@earendil-works/pi-tui@0.78.0': + '@earendil-works/pi-tui@0.80.3': dependencies: get-east-asian-width: 1.6.0 - marked: 15.0.12 + marked: 18.0.5 '@emnapi/core@1.10.0': dependencies: @@ -12271,10 +12274,10 @@ snapshots: dependencies: repeat-string: 1.6.1 - marked@15.0.12: {} - marked@18.0.4: {} + marked@18.0.5: {} + math-intrinsics@1.1.0: {} matrix-events-sdk@0.0.1: {} diff --git a/scripts/lib/dependency-ownership.json b/scripts/lib/dependency-ownership.json index 95dbdf5b6ab8..92d680c6a23a 100644 --- a/scripts/lib/dependency-ownership.json +++ b/scripts/lib/dependency-ownership.json @@ -72,6 +72,12 @@ "class": "core-runtime", "risk": ["filesystem", "path-safety"] }, + "@silvia-odwyer/photon-node": { + "owner": "core:media", + "class": "core-runtime", + "activation": ["agent.tool.read.image.bmp"], + "risk": ["wasm", "parser", "untrusted-files"] + }, "chalk": { "owner": "core:cli", "class": "core-runtime", diff --git a/src/agents/embedded-agent-helpers/errors.ts b/src/agents/embedded-agent-helpers/errors.ts index a3c85fca14e7..9bddf718f6e0 100644 --- a/src/agents/embedded-agent-helpers/errors.ts +++ b/src/agents/embedded-agent-helpers/errors.ts @@ -7,6 +7,7 @@ import { } from "@openclaw/normalization-core/string-coerce"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import type { AssistantMessage } from "../../llm/types.js"; +import { isConfiguredContextSizeOverflowError } from "../../llm/utils/overflow.js"; import { createSubsystemLogger } from "../../logging/subsystem.js"; import { extractLeadingHttpStatus, @@ -131,6 +132,7 @@ export function isContextOverflowError(errorMessage?: string): boolean { hasContextWindow && (lower.includes("ran out of room") || lower.includes("ran out of space")); return ( lower.includes("request_too_large") || + isConfiguredContextSizeOverflowError(errorMessage) || (lower.includes("invalid_argument") && lower.includes("maximum number of tokens")) || lower.includes("request exceeds the maximum size") || lower.includes("context length exceeded") || diff --git a/src/agents/embedded-agent-helpers/provider-error-patterns.test.ts b/src/agents/embedded-agent-helpers/provider-error-patterns.test.ts index 70ad6aeceef4..57bb55cfd812 100644 --- a/src/agents/embedded-agent-helpers/provider-error-patterns.test.ts +++ b/src/agents/embedded-agent-helpers/provider-error-patterns.test.ts @@ -134,6 +134,14 @@ describe("isContextOverflowError with provider patterns", () => { ).toBe(true); }); + it("detects DS4 configured context size overflow", () => { + expect( + isContextOverflowError( + "400 Prompt has 256468 tokens, but the configured context size is 256000 tokens", + ), + ).toBe(true); + }); + it("still detects standard context overflow patterns", () => { expect(isContextOverflowError("context length exceeded")).toBe(true); expect(isContextOverflowError("prompt is too long: 150000 tokens > 128000 maximum")).toBe(true); diff --git a/src/agents/sessions/agent-session.ts b/src/agents/sessions/agent-session.ts index 91489a208c2e..2e03408f0a1b 100644 --- a/src/agents/sessions/agent-session.ts +++ b/src/agents/sessions/agent-session.ts @@ -32,6 +32,7 @@ import type { TextContent, } from "../../llm/types.js"; import { isContextOverflow } from "../../llm/utils/overflow.js"; +import { isRetryableAssistantError } from "../../llm/utils/retry.js"; import type { Agent, AgentEvent, @@ -2609,11 +2610,7 @@ export class AgentSession { return false; } - const err = message.errorMessage; - // Match: overloaded_error, provider returned error, rate limit, 429, 500, 502, 503, 504, service unavailable, network/connection errors (including connection lost), WebSocket transport closes/errors, fetch failed, premature stream endings, HTTP/2 closed before response, terminated, retry delay exceeded - return /overloaded|provider.?returned.?error|rate.?limit|too many requests|429|500|502|503|504|service.?unavailable|server.?error|internal.?error|network.?error|connection.?error|connection.?refused|connection.?lost|websocket.?closed|websocket.?error|other side closed|fetch failed|upstream.?connect|reset before headers|socket hang up|ended without|stream ended before message_stop|http2 request did not get a response|timed? out|timeout|terminated|retry delay/i.test( - err, - ); + return isRetryableAssistantError(message); } /** diff --git a/src/agents/sessions/tools/bash.test.ts b/src/agents/sessions/tools/bash.test.ts index ee550a3faab1..c9cf58fbad73 100644 --- a/src/agents/sessions/tools/bash.test.ts +++ b/src/agents/sessions/tools/bash.test.ts @@ -2,7 +2,7 @@ // timer-safe millisecond values. import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion"; import { describe, expect, it } from "vitest"; -import { resolveBashTimeoutMs } from "./bash.js"; +import { createBashTool, resolveBashTimeoutMs, type BashOperations } from "./bash.js"; describe("bash tool timeout helpers", () => { it("converts positive timeout seconds to timer-safe milliseconds", () => { @@ -23,3 +23,23 @@ describe("bash tool timeout helpers", () => { expect(resolveBashTimeoutMs(-1)).toBeUndefined(); }); }); + +describe("bash tool output lifecycle", () => { + it("ignores output callbacks after execution settles", async () => { + const operations: BashOperations = { + exec: async (_command, _cwd, { onData }) => { + onData(Buffer.from("before\n")); + setTimeout(() => onData(Buffer.from("late\n")), 0); + return { exitCode: 0 }; + }, + }; + const tool = createBashTool(process.cwd(), { operations }); + + const result = await tool.execute("call-late-output", { command: "ignored" }); + await new Promise((resolve) => { + setTimeout(resolve, 20); + }); + + expect(result.content[0]).toEqual({ type: "text", text: "before\n" }); + }); +}); diff --git a/src/agents/sessions/tools/bash.ts b/src/agents/sessions/tools/bash.ts index cb53c87bca3b..8f4952d88ceb 100644 --- a/src/agents/sessions/tools/bash.ts +++ b/src/agents/sessions/tools/bash.ts @@ -304,6 +304,7 @@ export function createBashToolDefinition( const resolvedCommand = commandPrefix ? `${commandPrefix}\n${command}` : command; const spawnContext = resolveSpawnContext(resolvedCommand, cwd, spawnHook); const output = new OutputAccumulator({ tempFilePrefix: "openclaw-bash" }); + let acceptingOutput = true; let updateTimer: NodeJS.Timeout | undefined; let updateDirty = false; let lastUpdateAt = 0; @@ -353,11 +354,15 @@ export function createBashToolDefinition( } const handleData = (data: Buffer) => { + if (!acceptingOutput) { + return; + } output.append(data); scheduleOutputUpdate(); }; const finishOutput = async () => { + acceptingOutput = false; output.finish(); clearUpdateTimer(); emitOutputUpdate(); diff --git a/src/agents/sessions/tools/edit-diff.ts b/src/agents/sessions/tools/edit-diff.ts index e6c1678eeff3..46ebc167bfe8 100644 --- a/src/agents/sessions/tools/edit-diff.ts +++ b/src/agents/sessions/tools/edit-diff.ts @@ -93,6 +93,109 @@ interface MatchedEdit { newText: string; } +type TextReplacement = Pick; + +interface LineSpan { + start: number; + end: number; +} + +function splitLinesWithEndings(content: string): string[] { + return content.match(/[^\n]*\n|[^\n]+/g) ?? []; +} + +function getLineSpans(content: string): LineSpan[] { + let offset = 0; + return splitLinesWithEndings(content).map((line) => { + const span = { start: offset, end: offset + line.length }; + offset = span.end; + return span; + }); +} + +function getReplacementLineRange(lines: LineSpan[], replacement: TextReplacement) { + const replacementStart = replacement.matchIndex; + const replacementEnd = replacement.matchIndex + replacement.matchLength; + const startLine = lines.findIndex( + (line) => replacementStart >= line.start && replacementStart < line.end, + ); + if (startLine === -1) { + throw new Error("Replacement range is outside the base content."); + } + + let endLine = startLine; + while (endLine < lines.length && lines[endLine].end < replacementEnd) { + endLine++; + } + if (endLine >= lines.length) { + throw new Error("Replacement range is outside the base content."); + } + return { startLine, endLine: endLine + 1 }; +} + +function applyReplacements(content: string, replacements: TextReplacement[], offset = 0): string { + let result = content; + for (let i = replacements.length - 1; i >= 0; i--) { + const replacement = replacements[i]; + const matchIndex = replacement.matchIndex - offset; + result = + result.slice(0, matchIndex) + + replacement.newText + + result.slice(matchIndex + replacement.matchLength); + } + return result; +} + +/** + * Rewrite only lines touched by fuzzy replacements. Untouched lines retain + * their original bytes even though matching used normalized content. + */ +export function applyReplacementsPreservingUnchangedLines( + originalContent: string, + baseContent: string, + replacements: TextReplacement[], +): string { + const originalLines = splitLinesWithEndings(originalContent); + const baseLines = getLineSpans(baseContent); + if (originalLines.length !== baseLines.length) { + throw new Error( + "Cannot preserve unchanged lines because the base content has a different line count.", + ); + } + + const groups: Array<{ + startLine: number; + endLine: number; + replacements: TextReplacement[]; + }> = []; + const sortedReplacements = replacements.toSorted((a, b) => a.matchIndex - b.matchIndex); + for (const replacement of sortedReplacements) { + const range = getReplacementLineRange(baseLines, replacement); + const current = groups.at(-1); + if (current && range.startLine < current.endLine) { + current.endLine = Math.max(current.endLine, range.endLine); + current.replacements.push(replacement); + } else { + groups.push({ ...range, replacements: [replacement] }); + } + } + + let originalLineIndex = 0; + let result = ""; + for (const group of groups) { + result += originalLines.slice(originalLineIndex, group.startLine).join(""); + const groupStartOffset = baseLines[group.startLine].start; + const groupEndOffset = baseLines[group.endLine - 1].end; + result += applyReplacements( + baseContent.slice(groupStartOffset, groupEndOffset), + group.replacements, + groupStartOffset, + ); + originalLineIndex = group.endLine; + } + return result + originalLines.slice(originalLineIndex).join(""); +} + /** * Find oldText in content, trying exact match first, then fuzzy match. * When fuzzy matching is used, the returned contentForReplacement is the @@ -202,8 +305,7 @@ function getNoChangeError(path: string, totalEdits: number): EditNoChangeError { * * All edits are matched against the same original content. Replacements are * then applied in reverse order so offsets remain stable. If any edit needs - * fuzzy matching, the operation runs in fuzzy-normalized content space to - * preserve current single-edit behavior. + * fuzzy matching, only touched lines are rewritten from normalized content. */ export function applyEditsToNormalizedContent( normalizedContent: string, @@ -224,19 +326,20 @@ export function applyEditsToNormalizedContent( const initialMatches = normalizedEdits.map((edit) => fuzzyFindText(normalizedContent, edit.oldText), ); - const baseContent = initialMatches.some((match) => match.usedFuzzyMatch) + const usedFuzzyMatch = initialMatches.some((match) => match.usedFuzzyMatch); + const replacementBaseContent = usedFuzzyMatch ? normalizeForFuzzyMatch(normalizedContent) : normalizedContent; const matchedEdits: MatchedEdit[] = []; for (let i = 0; i < normalizedEdits.length; i++) { const edit = normalizedEdits[i]; - const matchResult = fuzzyFindText(baseContent, edit.oldText); + const matchResult = fuzzyFindText(replacementBaseContent, edit.oldText); if (!matchResult.found) { throw getNotFoundError(path, i, normalizedEdits.length); } - const occurrences = countOccurrences(baseContent, edit.oldText); + const occurrences = countOccurrences(replacementBaseContent, edit.oldText); if (occurrences > 1) { throw getDuplicateError(path, i, normalizedEdits.length, occurrences); } @@ -260,14 +363,14 @@ export function applyEditsToNormalizedContent( } } - let newContent = baseContent; - for (let i = matchedEdits.length - 1; i >= 0; i--) { - const edit = matchedEdits[i]; - newContent = - newContent.slice(0, edit.matchIndex) + - edit.newText + - newContent.slice(edit.matchIndex + edit.matchLength); - } + const baseContent = normalizedContent; + const newContent = usedFuzzyMatch + ? applyReplacementsPreservingUnchangedLines( + normalizedContent, + replacementBaseContent, + matchedEdits, + ) + : applyReplacements(replacementBaseContent, matchedEdits); if (baseContent === newContent) { throw getNoChangeError(path, normalizedEdits.length); diff --git a/src/agents/sessions/tools/edit.test.ts b/src/agents/sessions/tools/edit.test.ts index ebe12b923827..82468196fe7c 100644 --- a/src/agents/sessions/tools/edit.test.ts +++ b/src/agents/sessions/tools/edit.test.ts @@ -3,9 +3,16 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; +import { applyPatch } from "diff"; +import { Value } from "typebox/value"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { Theme } from "../../modes/interactive/theme/theme.js"; -import { createEditTool, createEditToolDefinition, type EditOperations } from "./edit.js"; +import { + createEditTool, + createEditToolDefinition, + type EditOperations, + type EditToolDetails, +} from "./edit.js"; const testTheme = { bg: (_name: string, text: string) => text, @@ -157,6 +164,85 @@ describe("edit tool", () => { }); }); + it("preserves untouched lines during fuzzy multi-edits", async () => { + const original = [ + "keep before ", + "first target ", + "first after", + "keep middle ", + "second target ", + "second after", + "keep after ", + "", + ].join("\n"); + const filePath = await createTempFile(original); + const tool = createEditTool(tmpDir); + + const result = await tool.execute( + "call-fuzzy", + { + path: filePath, + edits: [ + { oldText: "first target\nfirst after", newText: "FIRST\nFIRST2" }, + { oldText: "second target\nsecond after", newText: "SECOND\nSECOND2" }, + ], + }, + undefined, + ); + + const expected = [ + "keep before ", + "FIRST", + "FIRST2", + "keep middle ", + "SECOND", + "SECOND2", + "keep after ", + "", + ].join("\n"); + await expect(fs.readFile(filePath, "utf-8")).resolves.toBe(expected); + const details = result.details as EditToolDetails; + expect(applyPatch(original, details.patch)).toBe(expected); + }); + + it("preserves the correct duplicate line after a fuzzy replacement", async () => { + const original = "replace me \nafter \n"; + const filePath = await createTempFile(original); + const tool = createEditTool(tmpDir); + + const result = await tool.execute( + "call-duplicate", + { + path: filePath, + edits: [{ oldText: "replace me\n", newText: "after\n" }], + }, + undefined, + ); + + const expected = "after\nafter \n"; + await expect(fs.readFile(filePath, "utf-8")).resolves.toBe(expected); + const details = result.details as EditToolDetails; + expect(applyPatch(original, details.patch)).toBe(expected); + }); + + it("strips model-added metadata while retaining the strict edit schema", async () => { + const filePath = await createTempFile("before\n"); + const tool = createEditTool(tmpDir); + const prepared = tool.prepareArguments?.({ + path: filePath, + reason: "model explanation", + edits: [{ oldText: "before", newText: "after", reason: "why" }], + }); + + expect(prepared).toEqual({ + path: filePath, + edits: [{ oldText: "before", newText: "after" }], + }); + expect(Value.Check(tool.parameters, prepared)).toBe(true); + await tool.execute("call-metadata", prepared as never, undefined); + await expect(fs.readFile(filePath, "utf-8")).resolves.toBe("after\n"); + }); + it("renders previews through custom edit operations", async () => { // Preview rendering must use injected operations so remote/sandbox files are // shown without accidentally reading from the host filesystem. diff --git a/src/agents/sessions/tools/edit.ts b/src/agents/sessions/tools/edit.ts index 5c3cb573fa60..976f5e02622d 100644 --- a/src/agents/sessions/tools/edit.ts +++ b/src/agents/sessions/tools/edit.ts @@ -109,7 +109,7 @@ function prepareEditArguments(input: unknown): EditToolInput { return input as EditToolInput; } - const args = input as Record; + const args = { ...(input as Record) }; // Some models (Opus 4.6, GLM-5.1) send edits as a JSON string instead of an array if (typeof args.edits === "string") { @@ -122,14 +122,24 @@ function prepareEditArguments(input: unknown): EditToolInput { } const legacy = args as LegacyEditToolInput; - if (typeof legacy.oldText !== "string" || typeof legacy.newText !== "string") { - return args as unknown as EditToolInput; + if (typeof legacy.oldText === "string" && typeof legacy.newText === "string") { + const edits = Array.isArray(legacy.edits) ? [...legacy.edits] : []; + edits.push({ oldText: legacy.oldText, newText: legacy.newText }); + args.edits = edits; } - const edits = Array.isArray(legacy.edits) ? [...legacy.edits] : []; - edits.push({ oldText: legacy.oldText, newText: legacy.newText }); - const { oldText: _oldText, newText: _newText, ...rest } = legacy; - return { ...rest, edits } as EditToolInput; + const edits = Array.isArray(args.edits) + ? args.edits.map((edit) => { + if (!edit || typeof edit !== "object" || Array.isArray(edit)) { + return edit; + } + const candidate = edit as Record; + return { oldText: candidate.oldText, newText: candidate.newText }; + }) + : args.edits; + + // Keep the strict provider schema while tolerating model-added metadata. + return { path: args.path, edits } as EditToolInput; } function validateEditInput(input: EditToolInput): { diff --git a/src/agents/sessions/tools/find.fd.test.ts b/src/agents/sessions/tools/find.fd.test.ts index ac053cbc7549..a269a04e633d 100644 --- a/src/agents/sessions/tools/find.fd.test.ts +++ b/src/agents/sessions/tools/find.fd.test.ts @@ -1,7 +1,10 @@ import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; import { EventEmitter } from "node:events"; +import fs from "node:fs/promises"; +import path from "node:path"; import { PassThrough } from "node:stream"; -import { expect, it, vi } from "vitest"; +import { afterEach, expect, it, vi } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../../../test/helpers/temp-dir.js"; import { ensureTool } from "../../utils/tools-manager.js"; import { createFindToolDefinition } from "./find.js"; @@ -13,25 +16,66 @@ vi.mock("../../utils/tools-manager.js", () => ({ ensureTool: vi.fn(), })); -it("rejects partial fd output when fd exits with an error", async () => { - const stdout = new PassThrough(); - const stderr = new PassThrough(); - const child = Object.assign(new EventEmitter(), { +const tempDirs = useAutoCleanupTempDirTracker(afterEach); +type MockChild = ChildProcessWithoutNullStreams & { stdout: PassThrough; stderr: PassThrough }; + +afterEach(() => { + vi.clearAllMocks(); +}); + +function createChild(): MockChild { + return Object.assign(new EventEmitter(), { stdin: new PassThrough(), - stdout, - stderr, + stdout: new PassThrough(), + stderr: new PassThrough(), killed: false, kill: vi.fn(() => true), - }) as unknown as ChildProcessWithoutNullStreams; + }) as unknown as MockChild; +} + +it("rejects partial fd output when fd exits with an error", async () => { + const child = createChild(); vi.mocked(spawn).mockReturnValue(child); vi.mocked(ensureTool).mockResolvedValue("fd"); const tool = createFindToolDefinition("/workspace"); const result = tool.execute("call-1", { pattern: "*.ts" }, undefined, undefined, {} as never); await vi.waitFor(() => expect(spawn).toHaveBeenCalledOnce()); - stdout.end("/workspace/partial.ts\n"); - stderr.end("fd failed while reading subtree\n"); + child.stdout.end("/workspace/partial.ts\n"); + child.stderr.end("fd failed while reading subtree\n"); child.emit("close", 2, null); await expect(result).rejects.toThrow("fd failed while reading subtree"); }); + +it.each([ + { name: "inside a repository", gitBoundary: true, expected: false }, + { name: "outside a repository", gitBoundary: false, expected: true }, +])("sets --no-require-git only $name", async ({ gitBoundary, expected }) => { + const tempDir = tempDirs.make("openclaw-find-fd-"); + const searchPath = path.join(tempDir, "nested"); + await fs.mkdir(searchPath, { recursive: true }); + if (gitBoundary) { + await fs.writeFile(path.join(tempDir, ".git"), "gitdir: /tmp/example\n"); + } + + const child = createChild(); + vi.mocked(spawn).mockReturnValue(child); + vi.mocked(ensureTool).mockResolvedValue("fd"); + const tool = createFindToolDefinition(tempDir); + const result = tool.execute( + "call-git-boundary", + { pattern: "AGENTS.md", path: searchPath }, + undefined, + undefined, + {} as never, + ); + await vi.waitFor(() => expect(spawn).toHaveBeenCalledOnce()); + child.stdout.end(); + child.stderr.end(); + child.emit("close", 0, null); + await result; + + const args = vi.mocked(spawn).mock.calls[0]?.[1] as string[]; + expect(args.includes("--no-require-git")).toBe(expected); +}); diff --git a/src/agents/sessions/tools/find.ts b/src/agents/sessions/tools/find.ts index 6ffafe6937e0..440e7d2c63f0 100644 --- a/src/agents/sessions/tools/find.ts +++ b/src/agents/sessions/tools/find.ts @@ -26,6 +26,19 @@ import type { FindToolDetails } from "./tool-contracts.js"; import { wrapToolDefinition } from "./tool-definition-wrapper.js"; import { DEFAULT_MAX_BYTES, formatSize, truncateHead } from "./truncate.js"; +function isInsideGitRepository(searchPath: string): boolean { + for (let current = searchPath; ; ) { + if (existsSync(path.join(current, ".git"))) { + return true; + } + const parent = path.dirname(current); + if (parent === current) { + return false; + } + current = parent; + } +} + const findSchema = Type.Object({ pattern: Type.String({ description: "Glob pattern to match files, e.g. '*.ts', '**/*.json', or 'src/**/*.spec.ts'", @@ -244,17 +257,13 @@ export function createFindToolDefinition( return; } - // Build fd arguments. --no-require-git makes fd apply hierarchical .gitignore - // semantics whether or not the search path is inside a git repository, without - // leaking sibling-directory rules the way --ignore-file (a global source) would. - const args: string[] = [ - "--glob", - "--color=never", - "--hidden", - "--no-require-git", - "--max-results", - String(effectiveLimit), - ]; + const args: string[] = ["--glob", "--color=never", "--hidden"]; + // Outside a repo, fd needs this flag to honor standalone ignore files. + // Inside a repo, default git-aware traversal preserves nested repo boundaries. + if (!isInsideGitRepository(searchPath)) { + args.push("--no-require-git"); + } + args.push("--max-results", String(effectiveLimit)); // fd --glob matches against the basename unless --full-path is set; in --full-path // mode it matches against the absolute candidate path, so a path-containing diff --git a/src/agents/sessions/tools/read.test.ts b/src/agents/sessions/tools/read.test.ts index 97b1f0f7a8e5..ff7714c18814 100644 --- a/src/agents/sessions/tools/read.test.ts +++ b/src/agents/sessions/tools/read.test.ts @@ -4,12 +4,14 @@ import { Buffer } from "node:buffer"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../../../test/helpers/temp-dir.js"; import { withEnvAsync } from "../../../test-utils/env.js"; import { createReadToolDefinition } from "./read.js"; import { DEFAULT_MAX_BYTES } from "./truncate.js"; const decodeWindowsTextFileBufferMock = vi.hoisted(() => vi.fn(() => "")); +const tempDirs = useAutoCleanupTempDirTracker(afterEach); vi.mock("../../../infra/windows-encoding.js", () => ({ decodeWindowsTextFileBuffer: decodeWindowsTextFileBufferMock, @@ -18,6 +20,21 @@ vi.mock("../../../infra/windows-encoding.js", () => ({ const ONE_PIXEL_PNG_BASE64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+/p9sAAAAASUVORK5CYII="; +function createTinyBmp(): Buffer { + const buffer = Buffer.alloc(58); + buffer.write("BM", 0, "ascii"); + buffer.writeUInt32LE(buffer.length, 2); + buffer.writeUInt32LE(54, 10); + buffer.writeUInt32LE(40, 14); + buffer.writeInt32LE(1, 18); + buffer.writeInt32LE(1, 22); + buffer.writeUInt16LE(1, 26); + buffer.writeUInt16LE(24, 28); + buffer.writeUInt32LE(4, 34); + buffer[56] = 0xff; + return buffer; +} + function textContent( result: Awaited["execute"]>>, ): string { @@ -64,6 +81,28 @@ describe("read tool", () => { } }); + it("converts BMP files to PNG attachments", async () => { + const tempDir = tempDirs.make("openclaw-read-bmp-"); + const filePath = path.join(tempDir, "pixel.bmp"); + await fs.writeFile(filePath, createTinyBmp()); + const tool = createReadToolDefinition(tempDir, { autoResizeImages: false }); + const result = await tool.execute( + "call-bmp", + { path: filePath }, + undefined, + undefined, + {} as never, + ); + + expect(textContent(result)).toContain("Read image file [image/png]"); + expect(textContent(result)).toContain("converted from image/bmp to image/png"); + const image = result.content.find((part) => part.type === "image"); + expect(image).toMatchObject({ type: "image", mimeType: "image/png" }); + expect(Buffer.from(image?.type === "image" ? image.data : "", "base64").subarray(0, 8)).toEqual( + Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]), + ); + }); + it("shell-quotes the long-first-line fallback path", async () => { // The fallback command is shown to the model; quote the path so suggested // follow-up commands cannot execute path text as shell syntax. diff --git a/src/agents/sessions/tools/read.ts b/src/agents/sessions/tools/read.ts index fabf5b1d89ba..43a3865c3f63 100644 --- a/src/agents/sessions/tools/read.ts +++ b/src/agents/sessions/tools/read.ts @@ -25,7 +25,7 @@ import { type Theme, } from "../../modes/interactive/theme/theme.js"; import type { AgentTool } from "../../runtime/index.js"; -import { formatDimensionNote, resizeImage } from "../../utils/image-resize.js"; +import { processImage } from "../../utils/image-resize.js"; import { detectSupportedImageMimeTypeFromFile } from "../../utils/mime.js"; import { formatPathRelativeToCwdOrAbsolute } from "../../utils/paths.js"; import type { ToolDefinition, ToolRenderResultOptions } from "../extensions/types.js"; @@ -259,7 +259,7 @@ export function createReadToolDefinition( return { name: "read", label: "read", - description: `Read the contents of a file. Supports text files and images (jpg, png, gif, webp). Images are sent as attachments. For text files, output is truncated to ${DEFAULT_MAX_LINES} lines or ${DEFAULT_MAX_BYTES / 1024}KB (whichever is hit first). Use offset/limit for large files. When you need the full file, continue with offset until complete.`, + description: `Read the contents of a file. Supports text files and images (jpg, png, gif, webp, bmp). Images are sent as attachments. For text files, output is truncated to ${DEFAULT_MAX_LINES} lines or ${DEFAULT_MAX_BYTES / 1024}KB (whichever is hit first). Use offset/limit for large files. When you need the full file, continue with offset until complete.`, promptSnippet: "Read file contents", promptGuidelines: ["Use read to examine files instead of cat or sed."], parameters: readSchema, @@ -305,38 +305,25 @@ export function createReadToolDefinition( // Read image as binary. const buffer = await ops.readFile(absolutePath); const base64 = buffer.toString("base64"); - if (autoResizeImages) { - // Resize image if needed before sending it back to the model. - const resized = await resizeImage({ type: "image", data: base64, mimeType }); - if (!resized) { - let textNote = `Read image file [${mimeType}]\n[Image omitted: could not be resized below the inline image size limit.]`; - if (nonVisionImageNote) { - textNote += `\n${nonVisionImageNote}`; - } - content = [{ type: "text", text: textNote }]; - } else { - const dimensionNote = formatDimensionNote(resized); - let textNote = `Read image file [${resized.mimeType}]`; - if (dimensionNote) { - textNote += `\n${dimensionNote}`; - } - if (nonVisionImageNote) { - textNote += `\n${nonVisionImageNote}`; - } - content = [ - { type: "text", text: textNote }, - { type: "image", data: resized.data, mimeType: resized.mimeType }, - ]; - } - } else { - let textNote = `Read image file [${mimeType}]`; + const processed = await processImage( + { type: "image", data: base64, mimeType }, + { autoResizeImages }, + ); + if (!processed.ok) { + let textNote = `Read image file [${mimeType}]\n${processed.message}`; if (nonVisionImageNote) { textNote += `\n${nonVisionImageNote}`; } - content = [ - { type: "text", text: textNote }, - { type: "image", data: base64, mimeType }, - ]; + content = [{ type: "text", text: textNote }]; + } else { + let textNote = `Read image file [${processed.image.mimeType}]`; + if (processed.hints.length > 0) { + textNote += `\n${processed.hints.join("\n")}`; + } + if (nonVisionImageNote) { + textNote += `\n${nonVisionImageNote}`; + } + content = [{ type: "text", text: textNote }, processed.image]; } } else { // Read text content. diff --git a/src/agents/utils/child-process.test.ts b/src/agents/utils/child-process.test.ts new file mode 100644 index 000000000000..191ac22c3e4f --- /dev/null +++ b/src/agents/utils/child-process.test.ts @@ -0,0 +1,70 @@ +import { spawn, type ChildProcessByStdio } from "node:child_process"; +import type { Readable } from "node:stream"; +import { afterEach, describe, expect, it } from "vitest"; +import { waitForChildProcess } from "./child-process.js"; + +describe.skipIf(process.platform === "win32")("waitForChildProcess", () => { + let child: ChildProcessByStdio | undefined; + + afterEach(() => { + if (child?.pid) { + try { + process.kill(-child.pid, "SIGKILL"); + } catch {} + } + child = undefined; + }); + + it("drains active descendant output after the parent exits", async () => { + const command = + 'printf "HEAD\\n"; ( for i in 1 2 3 4 5 6; do sleep 0.05; printf "TICK$i\\n"; done ) &'; + child = spawn("/bin/sh", ["-c", command], { + stdio: ["ignore", "pipe", "pipe"], + detached: true, + }); + let output = ""; + child.stdout.on("data", (chunk: Buffer) => { + output += chunk.toString(); + }); + + await expect(waitForChildProcess(child)).resolves.toBe(0); + expect(output).toContain("HEAD"); + expect(output).toContain("TICK6"); + }); + + it("releases a quiet inherited pipe after the idle grace", async () => { + child = spawn("/bin/sh", ["-c", 'printf "DONE\\n"; ( sleep 30 ) &'], { + stdio: ["ignore", "pipe", "pipe"], + detached: true, + }); + let output = ""; + child.stdout.on("data", (chunk: Buffer) => { + output += chunk.toString(); + }); + + const startedAt = Date.now(); + await expect(waitForChildProcess(child)).resolves.toBe(0); + expect(output).toContain("DONE"); + expect(Date.now() - startedAt).toBeLessThan(2_000); + }); + + it("bounds draining from a continuously writing descendant", async () => { + child = spawn( + "/bin/sh", + ["-c", 'printf "HEAD\\n"; ( while :; do printf "TICK\\n"; sleep 0.03; done ) &'], + { + stdio: ["ignore", "pipe", "pipe"], + detached: true, + }, + ); + let output = ""; + child.stdout.on("data", (chunk: Buffer) => { + output += chunk.toString(); + }); + + const startedAt = Date.now(); + await expect(waitForChildProcess(child)).resolves.toBe(0); + expect(output).toContain("TICK"); + expect(Date.now() - startedAt).toBeLessThan(2_000); + }); +}); diff --git a/src/agents/utils/child-process.ts b/src/agents/utils/child-process.ts index 74fb492969f2..d12c4cdfcfae 100644 --- a/src/agents/utils/child-process.ts +++ b/src/agents/utils/child-process.ts @@ -6,14 +6,14 @@ import type { ChildProcess } from "node:child_process"; const EXIT_STDIO_GRACE_MS = 100; +const EXIT_STDIO_MAX_DRAIN_MS = 1_000; /** * Wait for a child process to terminate without hanging on inherited stdio handles. * - * On Windows, daemonized descendants can inherit the child's stdout/stderr pipe - * handles. In that case the child emits `exit`, but `close` can hang forever even - * though the original process is already gone. We wait briefly for stdio to end, - * then forcibly stop tracking the inherited handles. + * A detached descendant may keep stdout/stderr open after the child exits. Wait + * until those pipes are idle, re-arming the grace timer for every late chunk, so + * active output drains without hanging forever on an inherited handle. */ export function waitForChildProcess(child: ChildProcess): Promise { return new Promise((resolve, reject) => { @@ -21,6 +21,7 @@ export function waitForChildProcess(child: ChildProcess): Promise let exited = false; let exitCode: number | null = null; let postExitTimer: NodeJS.Timeout | undefined; + let postExitDeadlineTimer: NodeJS.Timeout | undefined; let stdoutEnded = child.stdout === null; let stderrEnded = child.stderr === null; @@ -29,11 +30,17 @@ export function waitForChildProcess(child: ChildProcess): Promise clearTimeout(postExitTimer); postExitTimer = undefined; } + if (postExitDeadlineTimer) { + clearTimeout(postExitDeadlineTimer); + postExitDeadlineTimer = undefined; + } child.removeListener("error", onError); child.removeListener("exit", onExit); child.removeListener("close", onClose); child.stdout?.removeListener("end", onStdoutEnd); child.stderr?.removeListener("end", onStderrEnd); + child.stdout?.removeListener("data", onData); + child.stderr?.removeListener("data", onData); }; const finalize = (code: number | null) => { @@ -56,6 +63,19 @@ export function waitForChildProcess(child: ChildProcess): Promise } }; + const armIdleTimer = () => { + if (postExitTimer) { + clearTimeout(postExitTimer); + } + postExitTimer = setTimeout(() => finalize(exitCode), EXIT_STDIO_GRACE_MS); + }; + + const onData = () => { + if (exited && !settled) { + armIdleTimer(); + } + }; + const onStdoutEnd = () => { stdoutEnded = true; maybeFinalizeAfterExit(); @@ -80,7 +100,10 @@ export function waitForChildProcess(child: ChildProcess): Promise exitCode = code; maybeFinalizeAfterExit(); if (!settled) { - postExitTimer = setTimeout(() => finalize(code), EXIT_STDIO_GRACE_MS); + // Drain finite descendant tails, but never let a chatty inherited pipe + // keep an already-exited command alive indefinitely. + postExitDeadlineTimer = setTimeout(() => finalize(exitCode), EXIT_STDIO_MAX_DRAIN_MS); + armIdleTimer(); } }; @@ -90,6 +113,8 @@ export function waitForChildProcess(child: ChildProcess): Promise child.stdout?.once("end", onStdoutEnd); child.stderr?.once("end", onStderrEnd); + child.stdout?.on("data", onData); + child.stderr?.on("data", onData); child.once("error", onError); child.once("exit", onExit); child.once("close", onClose); diff --git a/src/agents/utils/image-resize.ts b/src/agents/utils/image-resize.ts index 6b7bf3762d08..805f313d9515 100644 --- a/src/agents/utils/image-resize.ts +++ b/src/agents/utils/image-resize.ts @@ -5,6 +5,7 @@ */ import type { ImageContent } from "../../llm/types.js"; import { + convertImageToPng, createImageProcessor, isImageProcessorUnavailableError, type ImageProbe, @@ -27,6 +28,74 @@ interface ResizedImage { wasResized: boolean; } +export type ProcessImageResult = + | { ok: true; image: ImageContent; hints: string[] } + | { ok: false; message: string }; + +const INLINE_IMAGE_MIME_TYPES = new Set(["image/jpeg", "image/png", "image/gif", "image/webp"]); + +function baseMimeType(mimeType: string | undefined): string { + const normalized = mimeType?.split(";")[0]?.trim().toLowerCase(); + return normalized === "image/jpg" ? "image/jpeg" : (normalized ?? ""); +} + +async function normalizeImageForProvider( + image: ImageContent, +): Promise<{ image: ImageContent; convertedFrom?: string } | null> { + const mimeType = baseMimeType(image.mimeType); + if (INLINE_IMAGE_MIME_TYPES.has(mimeType)) { + return { image: { ...image, mimeType } }; + } + try { + const output = await convertImageToPng(Buffer.from(image.data, "base64")); + return { + image: { type: "image", data: output.toString("base64"), mimeType: "image/png" }, + convertedFrom: mimeType || image.mimeType, + }; + } catch { + return null; + } +} + +/** Normalize image formats for model input, then enforce inline size limits when enabled. */ +export async function processImage( + image: ImageContent, + options: { autoResizeImages: boolean }, +): Promise { + const normalized = await normalizeImageForProvider(image); + if (!normalized) { + return { + ok: false, + message: "[Image omitted: could not be converted to a supported inline image format.]", + }; + } + + const hints: string[] = []; + if (normalized.convertedFrom) { + hints.push(`[Image converted from ${normalized.convertedFrom} to image/png.]`); + } + if (!options.autoResizeImages) { + return { ok: true, image: normalized.image, hints }; + } + + const resized = await resizeImage(normalized.image); + if (!resized) { + return { + ok: false, + message: "[Image omitted: could not be resized below the inline image size limit.]", + }; + } + const dimensionNote = formatDimensionNote(resized); + if (dimensionNote) { + hints.push(dimensionNote); + } + return { + ok: true, + image: { type: "image", data: resized.data, mimeType: resized.mimeType }, + hints, + }; +} + // 4.5MB of base64 payload. Provides headroom below Anthropic's 5MB limit. const DEFAULT_MAX_BYTES = 4.5 * 1024 * 1024; diff --git a/src/agents/utils/mime.ts b/src/agents/utils/mime.ts index 05a28de04b1a..775f5ee077b9 100644 --- a/src/agents/utils/mime.ts +++ b/src/agents/utils/mime.ts @@ -23,6 +23,9 @@ function detectSupportedImageMimeType(buffer: Uint8Array): string | null { if (startsWithAscii(buffer, 0, "RIFF") && startsWithAscii(buffer, 8, "WEBP")) { return "image/webp"; } + if (startsWithAscii(buffer, 0, "BM") && isBmp(buffer)) { + return "image/bmp"; + } return null; } @@ -70,6 +73,44 @@ function isAnimatedPng(buffer: Uint8Array): boolean { return false; } +function isBmp(buffer: Uint8Array): boolean { + if (buffer.length < 26) { + return false; + } + const declaredFileSize = readUint32LE(buffer, 2); + const pixelDataOffset = readUint32LE(buffer, 10); + const dibHeaderSize = readUint32LE(buffer, 14); + if (declaredFileSize !== 0 && declaredFileSize < 26) { + return false; + } + if (pixelDataOffset < 14 + dibHeaderSize) { + return false; + } + if (declaredFileSize !== 0 && pixelDataOffset >= declaredFileSize) { + return false; + } + + let colorPlanes: number; + let bitsPerPixel: number; + if (dibHeaderSize === 12) { + colorPlanes = readUint16LE(buffer, 22); + bitsPerPixel = readUint16LE(buffer, 24); + } else if (dibHeaderSize >= 40 && dibHeaderSize <= 124) { + if (buffer.length < 30) { + return false; + } + colorPlanes = readUint16LE(buffer, 26); + bitsPerPixel = readUint16LE(buffer, 28); + } else { + return false; + } + return colorPlanes === 1 && [1, 4, 8, 16, 24, 32].includes(bitsPerPixel); +} + +function readUint16LE(buffer: Uint8Array, offset: number): number { + return (buffer[offset] ?? 0) + ((buffer[offset + 1] ?? 0) << 8); +} + function readUint32BE(buffer: Uint8Array, offset: number): number { return ( (buffer[offset] ?? 0) * 0x1000000 + @@ -79,6 +120,15 @@ function readUint32BE(buffer: Uint8Array, offset: number): number { ); } +function readUint32LE(buffer: Uint8Array, offset: number): number { + return ( + (buffer[offset] ?? 0) + + ((buffer[offset + 1] ?? 0) << 8) + + ((buffer[offset + 2] ?? 0) << 16) + + (buffer[offset + 3] ?? 0) * 0x1000000 + ); +} + function startsWith(buffer: Uint8Array, bytes: number[]): boolean { if (buffer.length < bytes.length) { return false; diff --git a/src/llm/providers/openai-chatgpt-responses.test.ts b/src/llm/providers/openai-chatgpt-responses.test.ts index 85257c83f293..d5c327897e6c 100644 --- a/src/llm/providers/openai-chatgpt-responses.test.ts +++ b/src/llm/providers/openai-chatgpt-responses.test.ts @@ -1,9 +1,12 @@ +import { arch, platform, release } from "node:os"; +import { zstdDecompressSync } from "node:zlib"; // ChatGPT Responses provider tests cover stream handling and timeout behavior. import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion"; import { afterEach, describe, expect, it, vi } from "vitest"; import { SYSTEM_PROMPT_CACHE_BOUNDARY } from "../../agents/system-prompt-cache-boundary.js"; import type { Context, Model } from "../types.js"; import { + closeOpenAICodexWebSocketSessions, extractOpenAICodexAccountId, parseSSEForTest, resetOpenAICodexWebSocketDebugStats, @@ -59,6 +62,22 @@ function stubHangingFetch(timeoutMs: number): void { ); } +function completedSseResponse(responseId = "resp_test"): Response { + const event = { + type: "response.completed", + response: { + id: responseId, + status: "completed", + output: [], + usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8 }, + }, + }; + return new Response(`data: ${JSON.stringify(event)}\n\n`, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); +} + describe("extractOpenAICodexAccountId", () => { it("decodes URL-safe base64 JWT payloads", () => { const accessToken = createJwt({ @@ -80,8 +99,10 @@ describe("extractOpenAICodexAccountId", () => { describe("streamOpenAICodexResponses transport", () => { afterEach(() => { + closeOpenAICodexWebSocketSessions(); vi.restoreAllMocks(); vi.unstubAllGlobals(); + vi.useRealTimers(); resetOpenAICodexWebSocketDebugStats(); }); @@ -102,6 +123,196 @@ describe("streamOpenAICodexResponses transport", () => { messages: [{ role: "user", content: "hi", timestamp: 1 }], } satisfies Context; + it("builds the first Node request with an OS-specific user agent", async () => { + vi.resetModules(); + const freshProvider = await import("./openai-chatgpt-responses.js"); + let userAgent: string | null = null; + vi.stubGlobal( + "fetch", + vi.fn(async (_input, init) => { + userAgent = new Headers(init?.headers).get("user-agent"); + return completedSseResponse(); + }), + ); + + await freshProvider + .streamOpenAICodexResponses(model, context, { + apiKey: createJwt({ + "https://api.openai.com/auth": { chatgpt_account_id: "acct-1" }, + }), + transport: "sse", + }) + .result(); + + expect(userAgent).toBe(`openclaw (${platform()} ${release()}; ${arch()})`); + }); + + it("zstd-compresses SSE bodies without overriding an existing encoding", async () => { + const captured: Array<{ body: BodyInit | null | undefined; encoding: string | null }> = []; + vi.stubGlobal( + "fetch", + vi.fn(async (_input, init) => { + captured.push({ + body: init?.body, + encoding: new Headers(init?.headers).get("content-encoding"), + }); + return completedSseResponse(`resp_${captured.length}`); + }), + ); + const apiKey = createJwt({ + "https://api.openai.com/auth": { chatgpt_account_id: "acct-1" }, + }); + + await streamOpenAICodexResponses(model, context, { apiKey, transport: "sse" }).result(); + await streamOpenAICodexResponses(model, context, { + apiKey, + transport: "sse", + headers: { "content-encoding": "identity" }, + }).result(); + + expect(captured[0]?.encoding).toBe("zstd"); + expect(captured[0]?.body).toBeInstanceOf(Uint8Array); + const decoded = JSON.parse( + Buffer.from(zstdDecompressSync(captured[0]?.body as Uint8Array)).toString("utf8"), + ) as { model?: string }; + expect(decoded.model).toBe(model.id); + expect(captured[1]).toMatchObject({ encoding: "identity", body: expect.any(String) }); + }); + + it("keeps JSON request bodies for custom ChatGPT relays", async () => { + let capturedBody: BodyInit | null | undefined; + let capturedEncoding: string | null = null; + vi.stubGlobal( + "fetch", + vi.fn(async (_input, init) => { + capturedBody = init?.body; + capturedEncoding = new Headers(init?.headers).get("content-encoding"); + return completedSseResponse(); + }), + ); + + await streamOpenAICodexResponses( + { ...model, provider: "custom-relay", baseUrl: "https://relay.test/backend-api" }, + context, + { + apiKey: createJwt({ + "https://api.openai.com/auth": { chatgpt_account_id: "acct-1" }, + }), + transport: "sse", + }, + ).result(); + + expect(capturedEncoding).toBeNull(); + expect(capturedBody).toEqual(expect.any(String)); + expect(JSON.parse(capturedBody as string)).toMatchObject({ model: model.id }); + }); + + it("reconnects once when the websocket connection limit is reached", async () => { + let connections = 0; + class ConnectionLimitWebSocket extends EventTarget { + private readonly limitReached = connections++ === 0; + + constructor() { + super(); + queueMicrotask(() => this.dispatchEvent(new Event("open"))); + } + + send(): void { + const event = this.limitReached + ? { type: "error", error: { code: "websocket_connection_limit_reached" } } + : { + type: "response.completed", + response: { + id: "resp_ws", + status: "completed", + output: [], + usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8 }, + }, + }; + queueMicrotask(() => { + this.dispatchEvent(Object.assign(new Event("message"), { data: JSON.stringify(event) })); + }); + } + + close(): void {} + } + const fetchMock = vi.fn(); + vi.stubGlobal("WebSocket", ConnectionLimitWebSocket); + vi.stubGlobal("fetch", fetchMock); + + const result = await streamOpenAICodexResponses(model, context, { + apiKey: createJwt({ + "https://api.openai.com/auth": { chatgpt_account_id: "acct-1" }, + }), + transport: "websocket", + }).result(); + + expect(result.stopReason).toBe("stop"); + expect(connections).toBe(2); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it("rotates cached websockets before the backend connection age limit", async () => { + vi.useFakeTimers(); + const startedAt = new Date("2026-07-03T00:00:00Z"); + vi.setSystemTime(startedAt); + let connections = 0; + const sentConnectionIds: number[] = []; + + class AgedWebSocket extends EventTarget { + readonly connectionId = ++connections; + readyState = 1; + + constructor() { + super(); + queueMicrotask(() => this.dispatchEvent(new Event("open"))); + } + + send(): void { + sentConnectionIds.push(this.connectionId); + queueMicrotask(() => { + this.dispatchEvent( + Object.assign(new Event("message"), { + data: JSON.stringify({ + type: "response.completed", + response: { + id: `resp_${this.connectionId}`, + status: "completed", + output: [], + usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8 }, + }, + }), + }), + ); + }); + } + + close(): void { + this.readyState = 3; + } + } + vi.stubGlobal("WebSocket", AgedWebSocket); + const apiKey = createJwt({ + "https://api.openai.com/auth": { chatgpt_account_id: "acct-1" }, + }); + const sessionId = "aged-session"; + + await streamOpenAICodexResponses(model, context, { + apiKey, + sessionId, + transport: "websocket-cached", + }).result(); + vi.setSystemTime(new Date(startedAt.getTime() + 56 * 60 * 1000)); + await streamOpenAICodexResponses(model, context, { + apiKey, + sessionId, + transport: "websocket-cached", + }).result(); + + expect(sentConnectionIds).toEqual([1, 2]); + expect(connections).toBe(2); + }); + it("preserves max for GPT-5.6 simple Codex Responses requests", async () => { let capturedPayload: Record | undefined; const stream = streamSimpleOpenAICodexResponses( diff --git a/src/llm/providers/openai-chatgpt-responses.ts b/src/llm/providers/openai-chatgpt-responses.ts index d685d3ec90ef..98cc4317f547 100644 --- a/src/llm/providers/openai-chatgpt-responses.ts +++ b/src/llm/providers/openai-chatgpt-responses.ts @@ -1,5 +1,6 @@ // OpenAI ChatGPT Responses provider handles ChatGPT-authenticated response streams. import type * as NodeOs from "node:os"; +import type * as NodeZlib from "node:zlib"; import type { Tool as OpenAITool, ResponseCreateParamsStreaming, @@ -7,20 +8,24 @@ import type { ResponseStreamEvent, } from "openai/resources/responses/responses.js"; -// NEVER convert to top-level runtime imports - breaks browser/Vite builds -let os: typeof NodeOs | null = null; - type DynamicImport = (specifier: string) => Promise; const dynamicImport: DynamicImport = (specifier) => import(specifier); -const NODE_OS_SPECIFIER = "node:os"; -if (typeof process !== "undefined" && (process.versions?.node || process.versions?.bun)) { - void dynamicImport(NODE_OS_SPECIFIER).then((m) => { - os = m as typeof NodeOs; - }); +type ProcessWithOsBuiltinModule = typeof process & { + getBuiltinModule?: (id: "node:os") => typeof NodeOs; +}; + +function loadNodeOs(): typeof NodeOs | null { + if (typeof process === "undefined" || !(process.versions?.node || process.versions?.bun)) { + return null; + } + return (process as ProcessWithOsBuiltinModule).getBuiltinModule?.("node:os") ?? null; } +// NEVER convert to top-level runtime imports - breaks browser/Vite builds +const os = loadNodeOs(); + import { resolveTimerTimeoutMs, clampTimerTimeoutMs, @@ -68,10 +73,12 @@ import { buildBaseOptions } from "./simple-options.js"; const DEFAULT_CODEX_BASE_URL = "https://chatgpt.com/backend-api"; const MAX_RETRIES = 3; const BASE_DELAY_MS = 1000; +const REQUEST_COMPRESSION_ZSTD_LEVEL = 3; const RETRY_AFTER_HTTP_DATE_RE = /^(?:(?:Mon|Tue|Wed|Thu|Fri|Sat|Sun), \d{2} (?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec) \d{4} \d{2}:\d{2}:\d{2} GMT|(?:Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday), \d{2}-(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)-\d{2} \d{2}:\d{2}:\d{2} GMT|(?:Mon|Tue|Wed|Thu|Fri|Sat|Sun) (?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec) [ \d]\d \d{2}:\d{2}:\d{2} \d{4})$/; const CODEX_TOOL_CALL_PROVIDERS = new Set(["openai", "opencode"]); const WEBSOCKET_MESSAGE_TOO_BIG_CLOSE_CODE = 1009; +const WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE = "websocket_connection_limit_reached"; const OPENAI_CHATGPT_RESPONSES_ERROR_BODY_MAX_BYTES = 16 * 1024; const OPENAI_CHATGPT_RESPONSES_SUCCESS_BODY_MAX_BYTES = 16 * 1024 * 1024; @@ -195,6 +202,30 @@ function formatRequestTimeoutError(timeoutMs: number, cause: unknown): Error { }); } +type ProcessWithZlibBuiltinModule = typeof process & { + getBuiltinModule?: (id: "node:zlib") => typeof NodeZlib; +}; + +function compressRequestBodyZstd(bodyJson: string): Uint8Array | null { + if (typeof process === "undefined" || !(process.versions?.node || process.versions?.bun)) { + return null; + } + const zlib = (process as ProcessWithZlibBuiltinModule).getBuiltinModule?.("node:zlib"); + if (!zlib || typeof zlib.zstdCompressSync !== "function") { + return null; + } + try { + const compressed = zlib.zstdCompressSync(bodyJson, { + params: { + [zlib.constants.ZSTD_c_compressionLevel]: REQUEST_COMPRESSION_ZSTD_LEVEL, + }, + }); + return Uint8Array.from(compressed); + } catch { + return null; + } +} + // ============================================================================ // Main Stream Function // ============================================================================ @@ -275,58 +306,76 @@ export const streamOpenAICodexResponses: StreamFunction< if (transport !== "sse" && !websocketDisabledForSession) { let websocketStarted = false; - try { - await processWebSocketStream( - resolveCodexWebSocketUrl(model.baseUrl), - body, - websocketHeaders, - output, - stream, - model, - () => { - websocketStarted = true; - }, - requestOptions, - firstEventAbort.abort, - ); + let retriedWebSocketConnectionLimit = false; + while (true) { + websocketStarted = false; + try { + await processWebSocketStream( + resolveCodexWebSocketUrl(model.baseUrl), + body, + websocketHeaders, + output, + stream, + model, + () => { + websocketStarted = true; + }, + requestOptions, + firstEventAbort.abort, + ); - if (activeSignal?.aborted) { - throw new Error("Request was aborted"); + if (activeSignal?.aborted) { + throw new Error("Request was aborted"); + } + stream.push({ + type: "done", + reason: output.stopReason as "stop" | "length" | "toolUse", + message: output, + }); + stream.end(); + return; + } catch (error) { + const aborted = activeSignal?.aborted; + const connectionLimitBeforeStart = + !websocketStarted && isWebSocketConnectionLimitReachedError(error); + if (!aborted && connectionLimitBeforeStart && !retriedWebSocketConnectionLimit) { + retriedWebSocketConnectionLimit = true; + continue; + } + if (aborted || (isCodexNonTransportError(error) && !connectionLimitBeforeStart)) { + throw error; + } + appendAssistantMessageDiagnostic( + output, + createAssistantMessageDiagnostic("provider_transport_failure", error, { + configuredTransport: transport, + fallbackTransport: transport === "auto" && !websocketStarted ? "sse" : undefined, + eventsEmitted: websocketStarted, + phase: websocketStarted + ? "after_message_stream_start" + : "before_message_stream_start", + requestBytes: new TextEncoder().encode(bodyJson).byteLength, + }), + ); + recordWebSocketFailure(options?.sessionId, error, { + activateSseFallback: transport === "auto", + }); + if (websocketStarted || transport !== "auto") { + throw error; + } + recordWebSocketSseFallback(options?.sessionId); + break; } - stream.push({ - type: "done", - reason: output.stopReason as "stop" | "length" | "toolUse", - message: output, - }); - stream.end(); - return; - } catch (error) { - const aborted = activeSignal?.aborted; - if (aborted || isCodexNonTransportError(error)) { - throw error; - } - appendAssistantMessageDiagnostic( - output, - createAssistantMessageDiagnostic("provider_transport_failure", error, { - configuredTransport: transport, - fallbackTransport: transport === "auto" && !websocketStarted ? "sse" : undefined, - eventsEmitted: websocketStarted, - phase: websocketStarted - ? "after_message_stream_start" - : "before_message_stream_start", - requestBytes: new TextEncoder().encode(bodyJson).byteLength, - }), - ); - recordWebSocketFailure(options?.sessionId, error, { - activateSseFallback: transport === "auto", - }); - if (websocketStarted || transport !== "auto") { - throw error; - } - recordWebSocketSseFallback(options?.sessionId); } } + const canCompressSseBody = model.provider === "openai" && !sseHeaders.has("content-encoding"); + const compressedBody = canCompressSseBody ? compressRequestBodyZstd(bodyJson) : null; + if (compressedBody) { + sseHeaders.set("content-encoding", "zstd"); + } + const sseBody: BodyInit = compressedBody ?? bodyJson; + // Fetch with retry logic for rate limits and transient errors let response: Response | undefined; let lastError: Error | undefined; @@ -340,7 +389,7 @@ export const streamOpenAICodexResponses: StreamFunction< response = await fetch(resolveCodexUrl(model.baseUrl), { method: "POST", headers: sseHeaders, - body: bodyJson, + body: sseBody, signal: activeSignal, }); await options?.onResponse?.( @@ -670,6 +719,34 @@ function isCodexNonTransportError(error: unknown): boolean { return error instanceof CodexApiError || error instanceof CodexProtocolError; } +function isWebSocketConnectionLimitReachedError(error: unknown): boolean { + return error instanceof CodexApiError && error.code === WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE; +} + +function extractCodexEventError(event: Record): { + code?: string; + message?: string; +} { + const nested = + event.error && typeof event.error === "object" + ? (event.error as Record) + : undefined; + return { + code: + typeof event.code === "string" + ? event.code + : typeof nested?.code === "string" + ? nested.code + : undefined, + message: + typeof event.message === "string" + ? event.message + : typeof nested?.message === "string" + ? nested.message + : undefined, + }; +} + async function* mapCodexEvents( events: AsyncIterable>, ): AsyncGenerator { @@ -680,10 +757,9 @@ async function* mapCodexEvents( } if (type === "error") { - const code = (event as { code?: string }).code || ""; - const message = (event as { message?: string }).message || ""; + const { code, message } = extractCodexEventError(event); throw new CodexApiError(`Codex error: ${message || code || JSON.stringify(event)}`, { - code: code || undefined, + code, payload: event, }); } @@ -803,6 +879,7 @@ export const parseSSEForTest = parseSSE; const OPENAI_BETA_RESPONSES_WEBSOCKETS = "responses_websockets=2026-02-06"; const SESSION_WEBSOCKET_CACHE_TTL_MS = 5 * 60 * 1000; +const SESSION_WEBSOCKET_MAX_AGE_MS = 55 * 60 * 1000; type WebSocketEventType = "open" | "message" | "error" | "close"; type WebSocketListener = (event: unknown) => void; @@ -823,10 +900,16 @@ interface CachedWebSocketContinuationState { interface CachedWebSocketConnection { socket: WebSocketLike; busy: boolean; + createdAt: number; idleTimer?: ReturnType; continuation?: CachedWebSocketContinuationState; } +type WebSocketConstructor = new ( + url: string, + protocols?: string | string[] | { headers?: Record }, +) => WebSocketLike; + export interface OpenAICodexWebSocketDebugStats { requests: number; connectionsCreated: number; @@ -847,6 +930,7 @@ export interface OpenAICodexWebSocketDebugStats { const websocketSessionCache = new Map(); const websocketDebugStats = new Map(); const websocketSseFallbackSessions = new Set(); +let cachedWebsocket: WebSocketConstructor | null = null; function getOrCreateWebSocketDebugStats(sessionId: string): OpenAICodexWebSocketDebugStats { let stats = websocketDebugStats.get(sessionId); @@ -869,6 +953,7 @@ function getOrCreateWebSocketDebugStats(sessionId: string): OpenAICodexWebSocket } export function resetOpenAICodexWebSocketDebugStats(sessionId?: string): void { + cachedWebsocket = null; if (sessionId) { websocketDebugStats.delete(sessionId); websocketSseFallbackSessions.delete(sessionId); @@ -932,12 +1017,6 @@ function recordWebSocketFailure( stats.websocketFallbackActive = isWebSocketSseFallbackActive(sessionId); } -type WebSocketConstructor = new ( - url: string, - protocols?: string | string[] | { headers?: Record }, -) => WebSocketLike; - -let cachedWebsocket: WebSocketConstructor | null = null; async function getWebSocketConstructor(): Promise { if (cachedWebsocket) { return cachedWebsocket; @@ -1006,6 +1085,10 @@ function isWebSocketReusable(socket: WebSocketLike): boolean { return readyState === undefined || readyState === 1; } +function isWebSocketSessionExpired(entry: CachedWebSocketConnection): boolean { + return Date.now() - entry.createdAt >= SESSION_WEBSOCKET_MAX_AGE_MS; +} + function closeWebSocketSilently(socket: WebSocketLike, code = 1000, reason = "done"): void { try { socket.close(code, reason); @@ -1136,7 +1219,10 @@ async function acquireWebSocket( clearTimeout(cached.idleTimer); cached.idleTimer = undefined; } - if (!cached.busy && isWebSocketReusable(cached.socket)) { + if (!cached.busy && isWebSocketSessionExpired(cached)) { + closeWebSocketSilently(cached.socket, 1000, "connection_age_limit"); + websocketSessionCache.delete(sessionId); + } else if (!cached.busy && isWebSocketReusable(cached.socket)) { cached.busy = true; return { socket: cached.socket, @@ -1170,7 +1256,7 @@ async function acquireWebSocket( } const socket = await connectWebSocket(url, headers, signal); - const entry: CachedWebSocketConnection = { socket, busy: true }; + const entry: CachedWebSocketConnection = { socket, busy: true, createdAt: Date.now() }; websocketSessionCache.set(sessionId, entry); return { socket, diff --git a/src/llm/utils/overflow.test.ts b/src/llm/utils/overflow.test.ts new file mode 100644 index 000000000000..976d0ad3089b --- /dev/null +++ b/src/llm/utils/overflow.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from "vitest"; +import type { AssistantMessage } from "../types.js"; +import { isConfiguredContextSizeOverflowError, isContextOverflow } from "./overflow.js"; + +function errorMessage(message: string): AssistantMessage { + return { + role: "assistant", + content: [], + api: "test-api", + provider: "test-provider", + model: "test-model", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "error", + errorMessage: message, + timestamp: 1, + }; +} + +describe("configured context size overflow", () => { + it.each([ + "400 Prompt has 256468 tokens, but the configured context size is 256000 tokens", + "Prompt has 5,958,968 tokens, but the configured context size is 256,000 tokens", + ])("detects %s", (text) => { + expect(isConfiguredContextSizeOverflowError(text)).toBe(true); + expect(isContextOverflow(errorMessage(text), 256_000)).toBe(true); + }); +}); diff --git a/src/llm/utils/overflow.ts b/src/llm/utils/overflow.ts index 4e3fb931d67e..6d5a683d6b8b 100644 --- a/src/llm/utils/overflow.ts +++ b/src/llm/utils/overflow.ts @@ -1,6 +1,14 @@ // Overflow helpers classify provider overflow errors and retryable responses. import type { AssistantMessage } from "../types.js"; +const CONFIGURED_CONTEXT_SIZE_OVERFLOW_RE = + /prompt has [\d,]+ tokens?, but the configured context size is [\d,]+ tokens?/i; + +/** Detects DS4-style raw token-count context overflow errors. */ +export function isConfiguredContextSizeOverflowError(errorMessage: string): boolean { + return CONFIGURED_CONTEXT_SIZE_OVERFLOW_RE.test(errorMessage); +} + /** * Regex patterns to detect context overflow errors from different providers. * @@ -48,6 +56,7 @@ const OVERFLOW_PATTERNS = [ /context window exceeds limit/i, // MiniMax /exceeded model token limit/i, // Kimi For Coding /too large for model with \d+ maximum context length/i, // Mistral + CONFIGURED_CONTEXT_SIZE_OVERFLOW_RE, // DS4 server /model_context_window_exceeded/i, // z.ai non-standard finish_reason surfaced as error text /prompt too long; exceeded (?:max )?context length/i, // Ollama explicit overflow error /context[_ ]length[_ ]exceeded/i, // Generic fallback diff --git a/src/llm/utils/retry.test.ts b/src/llm/utils/retry.test.ts new file mode 100644 index 000000000000..a47cc2710271 --- /dev/null +++ b/src/llm/utils/retry.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, it } from "vitest"; +import type { AssistantMessage } from "../types.js"; +import { isRetryableAssistantError } from "./retry.js"; + +function errorMessage(message: string): AssistantMessage { + return { + role: "assistant", + content: [], + api: "test-api", + provider: "test-provider", + model: "test-model", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "error", + errorMessage: message, + timestamp: 1, + }; +} + +describe("isRetryableAssistantError", () => { + it.each([ + "An error occurred while processing your request. You can retry your request.", + "The system encountered an unexpected error. Try your request again.", + "Temporary provider failure; please retry your request.", + ])("accepts explicit retry guidance: %s", (text) => { + expect(isRetryableAssistantError(errorMessage(text))).toBe(true); + }); + + it("keeps concrete quota failures non-retryable", () => { + expect(isRetryableAssistantError(errorMessage("429 insufficient_quota"))).toBe(false); + expect(isRetryableAssistantError(errorMessage("Monthly usage limit reached"))).toBe(false); + }); + + it("retries transient billing-service failures", () => { + expect( + isRetryableAssistantError( + errorMessage("503 billing service unavailable; please retry your request"), + ), + ).toBe(true); + }); + + it("retries short-window quota exhaustion", () => { + expect( + isRetryableAssistantError( + errorMessage( + "429 RESOURCE_EXHAUSTED: Quota exceeded for quota metric requests per minute; please retry your request", + ), + ), + ).toBe(true); + }); +}); diff --git a/src/llm/utils/retry.ts b/src/llm/utils/retry.ts new file mode 100644 index 000000000000..d45d97580c4e --- /dev/null +++ b/src/llm/utils/retry.ts @@ -0,0 +1,61 @@ +import type { AssistantMessage } from "../types.js"; + +function buildProviderErrorPattern(patterns: readonly string[]): RegExp { + return new RegExp(patterns.join("|"), "i"); +} + +const NON_RETRYABLE_PROVIDER_LIMIT_ERROR_PATTERN = buildProviderErrorPattern([ + "GoUsageLimitError", + "FreeUsageLimitError", + "Monthly usage limit reached", + "available balance", + "insufficient_quota", + "out of budget", +]); + +const RETRYABLE_PROVIDER_ERROR_PATTERN = buildProviderErrorPattern([ + "overloaded", + "rate.?limit", + "too many requests", + "429", + "500", + "502", + "503", + "504", + "service.?unavailable", + "server.?error", + "internal.?error", + "provider.?returned.?error", + "network.?error", + "connection.?error", + "connection.?refused", + "connection.?lost", + "other side closed", + "fetch failed", + "upstream.?connect", + "reset before headers", + "socket hang up", + "timed? out", + "timeout", + "terminated", + "websocket.?closed", + "websocket.?error", + "ended without", + "stream ended before message_stop", + "http2 request did not get a response", + "retry delay", + "you can retry your request", + "try your request again", + "please retry your request", +]); + +/** Classify transient provider/transport failures for outer retry policy. */ +export function isRetryableAssistantError(message: AssistantMessage): boolean { + if (message.stopReason !== "error" || !message.errorMessage) { + return false; + } + if (NON_RETRYABLE_PROVIDER_LIMIT_ERROR_PATTERN.test(message.errorMessage)) { + return false; + } + return RETRYABLE_PROVIDER_ERROR_PATTERN.test(message.errorMessage); +} diff --git a/src/media/image-ops.rastermill-adapter.test.ts b/src/media/image-ops.rastermill-adapter.test.ts index 7dd74e155353..1e376f7d693c 100644 --- a/src/media/image-ops.rastermill-adapter.test.ts +++ b/src/media/image-ops.rastermill-adapter.test.ts @@ -4,10 +4,36 @@ import { afterEach, describe, expect, it, vi } from "vitest"; describe("image ops Rastermill adapter", () => { afterEach(() => { vi.doUnmock("rastermill"); + vi.doUnmock("@silvia-odwyer/photon-node"); vi.doUnmock("../infra/resolve-system-bin.js"); vi.resetModules(); }); + it("does not load Photon for Rastermill-backed operations", async () => { + const encode = vi.fn(async () => ({ data: Buffer.from("jpeg") })); + const photonModuleFactory = vi.fn(() => { + throw new Error("Photon loaded eagerly"); + }); + + vi.doMock("@silvia-odwyer/photon-node", photonModuleFactory); + vi.doMock("rastermill", () => ({ + RastermillUnavailableError: class RastermillUnavailableError extends Error { + causes = []; + }, + createRastermill: vi.fn(() => ({ encode })), + isRastermillUnavailableError: () => false, + readImageMetadataFromHeader: vi.fn(() => ({ width: 1, height: 1 })), + readImageProbeFromHeader: vi.fn(() => ({ width: 1, height: 1, format: "jpeg" })), + })); + + const { resizeToJpeg } = await import("./image-ops.js"); + + await expect( + resizeToJpeg({ buffer: Buffer.from("input"), maxSide: 1, quality: 80 }), + ).resolves.toEqual(Buffer.from("jpeg")); + expect(photonModuleFactory).not.toHaveBeenCalled(); + }); + it("configures Rastermill with OpenClaw limits, temp root, and command resolution", async () => { const encode = vi.fn(async () => ({ data: Buffer.from("jpeg") })); const createRastermill = vi.fn((_options: unknown) => ({ encode })); diff --git a/src/media/image-ops.ts b/src/media/image-ops.ts index af67abe7e356..785e141f77cf 100644 --- a/src/media/image-ops.ts +++ b/src/media/image-ops.ts @@ -11,6 +11,7 @@ import { } from "rastermill"; import { resolveSystemBin } from "../infra/resolve-system-bin.js"; import { resolvePreferredOpenClawTmpDir } from "../infra/tmp-openclaw-dir.js"; +import { createLazyRuntimeModule } from "../shared/lazy-runtime.js"; export type { ImageMetadata, ImageProbe }; @@ -51,6 +52,8 @@ export const IMAGE_REDUCE_QUALITY_STEPS = [85, 75, 65, 55, 45, 35] as const; /** Shared input/output pixel cap for Rastermill-backed image operations. */ export const MAX_IMAGE_INPUT_PIXELS = 25_000_000; +const loadPhotonRuntime = createLazyRuntimeModule(() => import("./photon.runtime.js")); + /** Creates a Rastermill processor with OpenClaw temp-dir, pixel-limit, and command trust policy. */ export function createImageProcessor() { return createRastermill({ @@ -151,6 +154,30 @@ export async function convertHeicToJpeg(buffer: Buffer): Promise { } } +/** Converts image bytes to PNG, including BMP fallback unsupported by Rastermill's Photon gate. */ +export async function convertImageToPng(buffer: Buffer): Promise { + try { + return (await createImageProcessor().encode(buffer, { format: "png" })).data; + } catch (error) { + const probe = readRastermillImageProbeFromHeader(buffer); + const withinPixelLimit = + probe && + probe.format === "bmp" && + probe.width > 0 && + probe.height > 0 && + probe.width <= MAX_IMAGE_INPUT_PIXELS / probe.height; + if (!withinPixelLimit) { + throw error; + } + + try { + return (await loadPhotonRuntime()).convertBmpToPngWithPhoton(buffer); + } catch { + throw error; + } + } +} + /** Detects alpha support using a full transparency probe, falling back to trusted header metadata. */ export async function hasAlphaChannel(buffer: Buffer): Promise { try { diff --git a/src/media/photon.runtime.ts b/src/media/photon.runtime.ts new file mode 100644 index 000000000000..13d3831c81c3 --- /dev/null +++ b/src/media/photon.runtime.ts @@ -0,0 +1,12 @@ +import photon from "@silvia-odwyer/photon-node"; + +/** Decode validated BMP bytes only after Rastermill rejects the format. */ +export function convertBmpToPngWithPhoton(buffer: Buffer): Buffer { + let image: InstanceType | undefined; + try { + image = photon.PhotonImage.new_from_byteslice(buffer); + return Buffer.from(image.get_bytes()); + } finally { + image?.free(); + } +} diff --git a/src/scripts/test-projects.test.ts b/src/scripts/test-projects.test.ts index 01ce0ff077f0..7ac95ea57dec 100644 --- a/src/scripts/test-projects.test.ts +++ b/src/scripts/test-projects.test.ts @@ -975,6 +975,8 @@ describe("test-projects args", () => { "src/agents/cli-runner.reliability.test.ts", "src/agents/models-config.file-mode.test.ts", "src/agents/sandbox/ssh.test.ts", + "src/agents/sessions/tools/find.fd.test.ts", + "src/agents/sessions/tools/read.test.ts", ], watchMode: false, },