From 30c79e4db10051dc3980362d8f12e66827f64979 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 2 Aug 2026 14:42:01 -0700 Subject: [PATCH] fix(inference): prevent lost cancellation, stuck compaction, and runaway tools (#118146) * fix(inference): preserve cancellations, compaction, and tool-loop safety * fix(inference): satisfy Codex and mock-provider type checks * fix(codex): secure tool-free recovery and restricted turns * fix(codex): satisfy host-prompt provenance lint --- .../app-server/attempt-client-cleanup.test.ts | 24 ++ .../src/app-server/attempt-client-cleanup.ts | 23 ++ .../codex/src/app-server/bounded-turn.test.ts | 5 + .../codex/src/app-server/bounded-turn.ts | 20 +- .../codex/src/app-server/compact.test.ts | 213 ++++++++++++++++-- extensions/codex/src/app-server/compact.ts | 201 +++++++++-------- .../src/app-server/dynamic-tool-build.test.ts | 37 +++ .../src/app-server/dynamic-tool-build.ts | 4 +- .../src/app-server/dynamic-tools.test.ts | 106 ++++++++- .../codex/src/app-server/dynamic-tools.ts | 68 +++--- .../app-server/event-projector-assistant.ts | 13 +- .../src/app-server/event-projector-values.ts | 26 +-- .../event-projector.commentary.test.ts | 105 +++++++++ .../src/app-server/run-attempt-resources.ts | 1 + .../run-attempt-thread-cleanup.test.ts | 128 ++++++++++- .../app-server/run-attempt-turn-request.ts | 22 +- .../src/app-server/run-attempt-turn-start.ts | 2 +- .../app-server/settled-turn-context.test.ts | 87 ++++++- .../src/app-server/settled-turn-context.ts | 82 ++++++- .../src/app-server/side-question.test.ts | 100 ++++++++ .../codex/src/app-server/side-question.ts | 91 +++++--- .../src/app-server/thread-lifecycle.test.ts | 3 + .../codex/src/app-server/thread-requests.ts | 4 +- .../codex/src/conversation-binding.test.ts | 84 +++++++ extensions/codex/src/conversation-binding.ts | 38 ++-- .../codex/src/web-search-provider.test.ts | 13 +- extensions/ollama/src/cjk-char-estimate.ts | 47 ---- extensions/ollama/src/stream.ts | 3 +- .../src/providers/mock-openai/server.test.ts | 80 +++++++ .../src/providers/mock-openai/server.ts | 36 ++- .../openai-responses-stream-internal.ts | 18 +- ...enai-responses-stream-terminal-internal.ts | 40 ++-- ...responses-stream-terminal-recovery.test.ts | 147 ++++++++++++ .../goal-context-survives-compaction.yaml | 42 ++++ .../agent-tools.before-tool-call.e2e.test.ts | 31 ++- .../agent-tools.before-tool-call.policy.ts | 2 +- src/gateway/config-reload-plan.ts | 4 +- src/gateway/config-reload.test.ts | 35 ++- .../gateway.compaction-hot-reload.e2e.test.ts | 39 +++- .../server-methods/sessions-compact.ts | 36 ++- .../server.sessions.compaction.test.ts | 14 +- src/plugin-sdk/text-utility-runtime.ts | 9 + 42 files changed, 1702 insertions(+), 381 deletions(-) delete mode 100644 extensions/ollama/src/cjk-char-estimate.ts diff --git a/extensions/codex/src/app-server/attempt-client-cleanup.test.ts b/extensions/codex/src/app-server/attempt-client-cleanup.test.ts index 04c7e4644bf3..95c9492e396f 100644 --- a/extensions/codex/src/app-server/attempt-client-cleanup.test.ts +++ b/extensions/codex/src/app-server/attempt-client-cleanup.test.ts @@ -1,13 +1,37 @@ // Codex tests cover attempt client cleanup plugin behavior. import { describe, expect, it, vi } from "vitest"; import { + closeCodexStartupClientBestEffort, interruptCodexTurnAndWaitBestEffort, + retireUnsafeCodexTurnClientBestEffort, retireCodexAppServerClientAfterTimedOutTurn, unsubscribeCodexThreadBestEffort, } from "./attempt-client-cleanup.js"; import { createClientHarness } from "./test-support.js"; describe("Codex app-server attempt client cleanup", () => { + it("keeps strict startup retirement failures visible to lifecycle owners", async () => { + const closeAndWait = vi.fn(async () => { + throw new Error("strict client retirement failed"); + }); + + await expect(closeCodexStartupClientBestEffort({ closeAndWait } as never)).rejects.toThrow( + "strict client retirement failed", + ); + }); + + it("preserves the primary failure when unsafe turn retirement rejects", async () => { + const close = vi.fn(); + const closeAndWait = vi.fn(async () => { + throw new Error("unsafe client retirement failed"); + }); + + await expect( + retireUnsafeCodexTurnClientBestEffort({ close, closeAndWait } as never, "startup interrupt"), + ).resolves.toBeUndefined(); + expect(close).toHaveBeenCalledOnce(); + }); + it("waits for the matching terminal after an interrupt is acknowledged", async () => { const harness = createClientHarness(); const completion = interruptCodexTurnAndWaitBestEffort(harness.client, { diff --git a/extensions/codex/src/app-server/attempt-client-cleanup.ts b/extensions/codex/src/app-server/attempt-client-cleanup.ts index d5eac41b2569..d48ede326d0a 100644 --- a/extensions/codex/src/app-server/attempt-client-cleanup.ts +++ b/extensions/codex/src/app-server/attempt-client-cleanup.ts @@ -76,6 +76,29 @@ export async function closeCodexStartupClientBestEffort( } } +/** Retires an unsafe turn client without replacing an already-authoritative failure. */ +export async function retireUnsafeCodexTurnClientBestEffort( + client: CodexAppServerClient, + operation: string, +): Promise { + try { + await closeCodexStartupClientBestEffort(client); + } catch (error) { + embeddedAgentLog.debug("codex app-server unsafe turn client retirement failed", { + operation, + error, + }); + try { + client.close(); + } catch (closeError) { + embeddedAgentLog.debug("codex app-server unsafe turn client close failed", { + operation, + error: closeError, + }); + } + } +} + /** Sends a bounded turn interrupt and waits for Codex to confirm terminal abort handling. */ export async function interruptCodexTurnAndWaitBestEffort( client: CodexAppServerClient, diff --git a/extensions/codex/src/app-server/bounded-turn.test.ts b/extensions/codex/src/app-server/bounded-turn.test.ts index 590f5786a863..ecc94adb2f26 100644 --- a/extensions/codex/src/app-server/bounded-turn.test.ts +++ b/extensions/codex/src/app-server/bounded-turn.test.ts @@ -423,13 +423,18 @@ describe("runBoundedCodexAppServerTurn settled finalization isolation", () => { dynamicTools: [], ephemeral: true, config: { + "agents.enabled": false, "features.hooks": false, "features.multi_agent": false, + "features.multi_agent_v2": false, "skills.include_instructions": false, include_environment_context: false, mcp_servers: { inherited: { enabled: false } }, }, }); + const turnParams = fake.request.mock.calls.find(([method]) => method === "turn/start")?.[1]; + expect(turnParams).not.toHaveProperty("cwd"); + expect(turnParams).not.toHaveProperty("environments"); expect(fake.request).toHaveBeenCalledWith( "thread/inject_items", { threadId: "thread-finalizer", items: historyItems }, diff --git a/extensions/codex/src/app-server/bounded-turn.ts b/extensions/codex/src/app-server/bounded-turn.ts index 60f2152e0d28..f9d546ab3ec3 100644 --- a/extensions/codex/src/app-server/bounded-turn.ts +++ b/extensions/codex/src/app-server/bounded-turn.ts @@ -48,10 +48,11 @@ import { readCodexInheritedMcpServerNames, } from "./thread-requests.js"; -const CODEX_PRIVATE_STDIO_ARGS = ["app-server", "--listen", "stdio://"]; const CODEX_APP_SERVER_ARGS_ENV_KEY = "OPENCLAW_CODEX_APP_SERVER_ARGS"; const CODEX_BOUNDED_THREAD_CONFIG: JsonObject = { + "agents.enabled": false, "features.multi_agent": false, + "features.multi_agent_v2": false, "features.apps": false, "features.plugins": false, "features.image_generation": false, @@ -282,12 +283,12 @@ async function runBoundedCodexAppServerTurnInWorkspace( ); try { const turn = assertCodexTurnStartResponse( + // Inherit the empty thread environment; a cwd override recreates native tools. await client.request( "turn/start", { threadId: thread.thread.id, input: params.input, - cwd: workspace.cwd, approvalPolicy: "on-request", model, effort: "low", @@ -375,6 +376,19 @@ function buildPrivateCodexAppServerStartOptions( start: ReturnType["start"], codexHome: string, ): ReturnType["start"] { + // Provider identity and model catalogs must survive isolation; hooks, MCP, + // sandbox policy, and other process overrides must not cross that boundary. + const providerArgs = start.args.flatMap((arg, index) => { + const override = + arg === "-c" || arg === "--config" + ? start.args[index + 1] + : arg.startsWith("--config=") + ? arg.slice("--config=".length) + : undefined; + return override && /^\s*(?:openai_base_url|model_catalog_json)\s*=/u.test(override) + ? ["-c", override] + : []; + }); const privateEnv = Object.fromEntries( Object.entries(start.env ?? {}).filter( ([name]) => name.trim().toUpperCase() !== CODEX_APP_SERVER_ARGS_ENV_KEY, @@ -386,7 +400,7 @@ function buildPrivateCodexAppServerStartOptions( }); return { ...start, - args: [...CODEX_PRIVATE_STDIO_ARGS], + args: ["app-server", ...providerArgs, "--listen", "stdio://"], env: { ...privateEnv, CODEX_HOME: codexHome, diff --git a/extensions/codex/src/app-server/compact.test.ts b/extensions/codex/src/app-server/compact.test.ts index dbf45f5c97f7..fd5699ba9e73 100644 --- a/extensions/codex/src/app-server/compact.test.ts +++ b/extensions/codex/src/app-server/compact.test.ts @@ -7,6 +7,11 @@ import { type HarnessContextEngine as ContextEngine, } from "openclaw/plugin-sdk/agent-harness-runtime"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + consumeCodexAppServerLiveThread, + ensureCodexAppServerClientRuntime, + retainCodexAppServerLiveThread, +} from "./client-runtime.js"; import { CodexAppServerRpcError, type CodexAppServerClient } from "./client.js"; import { maybeCompactCodexAppServerSession as maybeCompactCodexAppServerSessionImpl } from "./compact.js"; import { resolveCodexSupervisionAppServerRuntimeOptions } from "./config.js"; @@ -23,6 +28,7 @@ import { writeCodexAppServerBinding, } from "./session-binding.test-helpers.js"; import type { CodexAppServerClientFactory } from "./shared-client.js"; +import { CODEX_APP_SERVER_VERSION } from "./version.js"; let tempDir: string; let codexAppServerClientFactoryForTest: CodexAppServerClientFactory | undefined; @@ -202,6 +208,88 @@ describe("maybeCompactCodexAppServerSession", () => { expect(details.completed).toBe(true); }); + it("resubscribes an evicted session before compacting without displacing its sibling", async () => { + const fake = createFakeCodexClient(); + setCodexAppServerClientFactoryForTest(async () => fake.client); + const sessionFile = await writeTestBinding(); + + await fake.client.request("thread/resume", { threadId: "thread-2", excludeTurns: true }); + await retainCodexAppServerLiveThread( + fake.client, + "thread-2", + async (threadId) => { + await fake.client.request("thread/unsubscribe", { threadId }); + }, + "config-thread-2", + ); + fake.request.mockClear(); + + await expect(startCompaction(sessionFile)).resolves.toMatchObject({ + ok: true, + compacted: true, + }); + + expect(fake.request.mock.calls.map(([method]) => method)).toEqual([ + "thread/resume", + "thread/compact/start", + "thread/unsubscribe", + ]); + expect(fake.request).toHaveBeenCalledWith( + "thread/resume", + { threadId: "thread-1", excludeTurns: true }, + expect.objectContaining({ timeoutMs: expect.any(Number) }), + ); + await expect( + consumeCodexAppServerLiveThread(fake.client, "thread-2", "config-thread-2"), + ).resolves.toBe(true); + }); + + it("keeps an owned thread subscribed when a sibling finishes during compaction", async () => { + const fake = createFakeCodexClient({ autoCompleteCompaction: false }); + setCodexAppServerClientFactoryForTest(async () => fake.client); + const sessionFile = await writeTestBinding(); + const pending = startCompaction(sessionFile); + await vi.waitFor(() => { + expect(fake.request).toHaveBeenCalledWith("thread/compact/start", { threadId: "thread-1" }); + }); + + await fake.client.request("thread/resume", { threadId: "thread-2", excludeTurns: true }); + await retainCodexAppServerLiveThread(fake.client, "thread-2", undefined, "config-thread-2"); + fake.completeCompaction(); + + await expect(pending).resolves.toMatchObject({ ok: true, compacted: true }); + expect(fake.request).toHaveBeenCalledWith( + "thread/unsubscribe", + { threadId: "thread-1" }, + { timeoutMs: 5_000 }, + ); + await expect( + consumeCodexAppServerLiveThread(fake.client, "thread-2", "config-thread-2"), + ).resolves.toBe(true); + }); + + it("preserves an incognito thread's separately owned live subscription", async () => { + const fake = createFakeCodexClient({ + retainedThreadId: null, + subscribedThreadIds: ["thread-1"], + }); + setCodexAppServerClientFactoryForTest(async () => fake.client); + const sessionKey = "agent:main:dashboard:incognito-compact"; + const sessionFile = await writeTestBinding({}, sessionKey); + + await expect( + maybeCompactCodexAppServerSession({ + sessionId: "session-1", + sessionKey, + sessionFile, + workspaceDir: tempDir, + trigger: "manual", + }), + ).resolves.toMatchObject({ ok: true, compacted: true }); + + expect(fake.request.mock.calls.map(([method]) => method)).toEqual(["thread/compact/start"]); + }); + it("uses the exact prepared Platform key for native compaction", async () => { const fake = createFakeCodexClient(); const factory = vi.fn(async () => fake.client); @@ -289,7 +377,7 @@ describe("maybeCompactCodexAppServerSession", () => { }); it("uses the native supervision runtime and auth for supervised bindings", async () => { - const fake = createFakeCodexClient(); + const fake = createFakeCodexClient({ retainedThreadId: null }); const factory = vi.fn(async () => fake.client); const sessionFile = await writeSupervisedTestBinding({ authProfileId: "openai:binding-profile", @@ -319,6 +407,11 @@ describe("maybeCompactCodexAppServerSession", () => { startOptions: expect.objectContaining({ homeScope: "user" }), }), ); + expect(fake.request.mock.calls.map(([method]) => method)).toEqual([ + "thread/resume", + "thread/compact/start", + "thread/unsubscribe", + ]); }); it("fails closed when a supervised binding is no longer enabled", async () => { @@ -382,7 +475,7 @@ describe("maybeCompactCodexAppServerSession", () => { }); it("starts native app-server compaction for post-context-engine budget requests", async () => { - const fake = createFakeCodexClient(); + const fake = createFakeCodexClient({ retainedThreadId: null }); setCodexAppServerClientFactoryForTest(async () => fake.client); const sessionFile = await writeTestBinding({ contextEngine: { @@ -417,6 +510,11 @@ describe("maybeCompactCodexAppServerSession", () => { { threadId: "thread-1" }, { timeoutMs: 60_000 }, ); + expect(fake.request.mock.calls.map(([method]) => method)).toEqual([ + "thread/resume", + "thread/compact/start", + "thread/unsubscribe", + ]); expect(result.ok).toBe(true); expect(result.compacted).toBe(true); expect(result.reason).toBeUndefined(); @@ -588,7 +686,10 @@ describe("maybeCompactCodexAppServerSession", () => { let externalWriteStarted = false; let externalWriteFinished = false; const fake = createFakeCodexClient(); - fake.request.mockImplementation(async () => { + fake.request.mockImplementation(async (method) => { + if (method === "thread/unsubscribe") { + return {}; + } const response = await expectExternalMutationBlockedDuringNativeRequest({ releaseExternalMutation: releaseExternalWrite, isExternalMutationStarted: () => externalWriteStarted, @@ -672,7 +773,10 @@ describe("maybeCompactCodexAppServerSession", () => { let externalClearStarted = false; let externalClearFinished = false; const fake = createFakeCodexClient(); - fake.request.mockImplementation(async () => { + fake.request.mockImplementation(async (method) => { + if (method === "thread/unsubscribe") { + return {}; + } const response = await expectExternalMutationBlockedDuringNativeRequest({ releaseExternalMutation: releaseExternalClear, isExternalMutationStarted: () => externalClearStarted, @@ -1293,7 +1397,9 @@ describe("maybeCompactCodexAppServerSession", () => { fake.completeCompaction(); await expect(first).resolves.toMatchObject({ ok: true, compacted: true }); await vi.waitFor(() => { - expect(fake.request).toHaveBeenCalledTimes(2); + expect( + fake.request.mock.calls.filter(([method]) => method === "thread/compact/start"), + ).toHaveLength(2); }); fake.completeCompaction(); @@ -1393,7 +1499,9 @@ describe("maybeCompactCodexAppServerSession", () => { fake.completeCompaction(); await expect(first).resolves.toMatchObject({ ok: true, compacted: true }); await vi.waitFor(() => { - expect(fake.request).toHaveBeenCalledTimes(2); + expect( + fake.request.mock.calls.filter(([method]) => method === "thread/compact/start"), + ).toHaveLength(2); }); fake.completeCompaction(); @@ -1487,7 +1595,7 @@ describe("maybeCompactCodexAppServerSession", () => { }); it("keeps the lifecycle fence when an unconfirmed stdio process does not stop", async () => { - const fake = createFakeCodexClient(); + const fake = createFakeCodexClient({ retainedThreadId: "thread-stuck-stdio" }); fake.request.mockRejectedValueOnce(new Error("thread/compact/start timed out")); fake.closeAndWait.mockResolvedValueOnce(false); setCodexAppServerClientFactoryForTest(async () => fake.client); @@ -1831,7 +1939,7 @@ describe("maybeCompactCodexAppServerSession", () => { }); it("forwards compaction to native Codex even when a context engine owns compaction", async () => { - const fake = createFakeCodexClient(); + const fake = createFakeCodexClient({ retainedThreadId: null }); setCodexAppServerClientFactoryForTest(async () => fake.client); const sessionFile = await writeTestBinding(); const compact = vi.fn(async () => ({ @@ -1872,6 +1980,11 @@ describe("maybeCompactCodexAppServerSession", () => { ); expect(fake.request).toHaveBeenCalledWith("thread/compact/start", { threadId: "thread-1" }); + expect(fake.request.mock.calls.map(([method]) => method)).toEqual([ + "thread/resume", + "thread/compact/start", + "thread/unsubscribe", + ]); expect(result.ok).toBe(true); expect(result.compacted).toBe(true); expect(compactDetails(result)).toMatchObject({ @@ -1928,6 +2041,8 @@ function createFakeCodexClient( autoCompleteCompaction?: boolean; interruptError?: Error; rejectInterrupt?: boolean; + retainedThreadId?: string | null; + subscribedThreadIds?: readonly string[]; } = {}, ): { client: CodexAppServerClient; @@ -1939,7 +2054,16 @@ function createFakeCodexClient( } { const handlers = new Set<(notification: CodexServerNotification) => void>(); const closeHandlers = new Set<() => void>(); + const retainedThreadId = + options.retainedThreadId === undefined ? "thread-1" : options.retainedThreadId; + const subscribedThreadIds = new Set( + options.subscribedThreadIds ?? (retainedThreadId ? [retainedThreadId] : []), + ); const emit = (notification: CodexServerNotification): void => { + const threadId = (notification.params as { threadId?: string } | undefined)?.threadId; + if (threadId && !subscribedThreadIds.has(threadId)) { + return; + } for (const handler of handlers) { handler(notification); } @@ -1978,6 +2102,46 @@ function createFakeCodexClient( }; const request = vi.fn( async (method: string, params?: unknown) => { + const threadId = (params as { threadId?: string } | undefined)?.threadId; + if (method === "thread/resume" && threadId) { + subscribedThreadIds.add(threadId); + return { + thread: { + id: threadId, + sessionId: "session-1", + forkedFromId: null, + preview: "", + ephemeral: false, + modelProvider: "openai", + createdAt: 1, + updatedAt: 1, + status: { type: "idle" }, + path: null, + cwd: tempDir, + cliVersion: CODEX_APP_SERVER_VERSION, + source: "unknown", + agentNickname: null, + agentRole: null, + gitInfo: null, + name: null, + turns: [], + }, + model: "gpt-5.5-codex", + modelProvider: "openai", + serviceTier: null, + cwd: tempDir, + instructionSources: [], + approvalPolicy: "never", + approvalsReviewer: "user", + sandbox: { type: "dangerFullAccess" }, + permissionProfile: null, + reasoningEffort: null, + }; + } + if (method === "thread/unsubscribe" && threadId) { + subscribedThreadIds.delete(threadId); + return {}; + } if (method === "turn/interrupt" && options.interruptError) { throw options.interruptError; } @@ -1985,7 +2149,6 @@ function createFakeCodexClient( throw new Error("interrupt unavailable"); } if (method === "thread/compact/start" && options.autoCompleteCompaction !== false) { - const threadId = (params as { threadId?: unknown }).threadId; if (typeof threadId !== "string") { throw new Error("thread/compact/start requires threadId"); } @@ -2039,17 +2202,29 @@ function createFakeCodexClient( return () => handlers.delete(handler); }, ); + const client = { + request, + close, + closeAndWait, + addNotificationHandler, + addRequestHandler: vi.fn(() => () => undefined), + addCloseHandler: vi.fn((handler: () => void) => { + closeHandlers.add(handler); + return () => closeHandlers.delete(handler); + }), + } as unknown as CodexAppServerClient; + ensureCodexAppServerClientRuntime(client, { agentDir: tempDir }); + addNotificationHandler.mockClear(); + if (retainedThreadId) { + void retainCodexAppServerLiveThread( + client, + retainedThreadId, + undefined, + `config-${retainedThreadId}`, + ); + } return { - client: { - request, - close, - closeAndWait, - addNotificationHandler, - addCloseHandler: vi.fn((handler: () => void) => { - closeHandlers.add(handler); - return () => closeHandlers.delete(handler); - }), - } as unknown as CodexAppServerClient, + client, request, close, closeAndWait, diff --git a/extensions/codex/src/app-server/compact.ts b/extensions/codex/src/app-server/compact.ts index 87f3f064643a..3ecb41eadb9a 100644 --- a/extensions/codex/src/app-server/compact.ts +++ b/extensions/codex/src/app-server/compact.ts @@ -8,9 +8,19 @@ import { type EmbeddedAgentCompactResult, } from "openclaw/plugin-sdk/agent-harness-runtime"; import { resolveAgentDir, resolveDefaultAgentId } from "openclaw/plugin-sdk/agent-runtime"; -import { isCodexAlreadyTerminalInterruptError } from "./attempt-client-cleanup.js"; +import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue"; +import { asOptionalRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { isIncognitoSessionKey } from "../incognito-session.js"; +import { + CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, + closeCodexStartupClientBestEffort, + CodexAppServerUnsafeSubscriptionError, + isCodexAlreadyTerminalInterruptError, + unsubscribeCodexThreadBestEffort, +} from "./attempt-client-cleanup.js"; import { readCodexNotificationItem } from "./attempt-notifications.js"; import { resolveCodexBindingAppServerConnection } from "./binding-connection.js"; +import { consumeCodexAppServerLiveThread } from "./client-runtime.js"; import { CodexAppServerRpcError, type CodexAppServerClient } from "./client.js"; import { readCodexNotificationThreadId, @@ -30,9 +40,10 @@ import { releaseLeasedSharedCodexAppServerClient, type CodexAppServerClientFactory, } from "./shared-client.js"; +import { resumeCodexAppServerThread } from "./thread-resume.js"; const warnedIgnoredCompactionOverrides = new Set(); -const codexNativeCompactionQueues = new Map>(); +const codexNativeCompactionQueue = new KeyedAsyncQueue(); const CODEX_NATIVE_COMPACTION_INTERRUPT_GRACE_MS = 30_000; type CodexAppServerCompactOptions = { bindingStore: CodexAppServerBindingStore; @@ -287,51 +298,30 @@ async function runExclusiveCodexNativeCompaction( signal: AbortSignal | undefined, run: () => Promise, ): Promise { - const previous = codexNativeCompactionQueues.get(threadId) ?? Promise.resolve(); - let releaseCurrent!: () => void; - const current = new Promise((resolve) => { - releaseCurrent = resolve; - }); - const queued = previous.then( - () => current, - () => current, - ); - codexNativeCompactionQueues.set(threadId, queued); - try { - await waitForCodexNativeCompactionQueue(previous, signal); + signal?.throwIfAborted(); + let started = false; + const queued = codexNativeCompactionQueue.enqueue(threadId, async () => { + started = true; signal?.throwIfAborted(); - return await run(); - } finally { - releaseCurrent(); - // A canceled waiter must remain in the chain until its predecessor settles; - // otherwise a later request can skip the still-active compaction. - void queued.then(() => { - if (codexNativeCompactionQueues.get(threadId) === queued) { - codexNativeCompactionQueues.delete(threadId); - } - }); - } -} - -async function waitForCodexNativeCompactionQueue( - previous: Promise, - signal: AbortSignal | undefined, -): Promise { + return run(); + }); if (!signal) { - await previous.catch(() => undefined); - return; + return queued; } - signal.throwIfAborted(); let removeAbortListener = () => {}; const aborted = new Promise((_, reject) => { const onAbort = () => { - reject(signal.reason instanceof Error ? signal.reason : new Error("compaction aborted")); + if (!started) { + reject(signal.reason instanceof Error ? signal.reason : new Error("compaction aborted")); + } }; removeAbortListener = () => signal.removeEventListener("abort", onAbort); signal.addEventListener("abort", onAbort, { once: true }); }); try { - await Promise.race([previous.catch(() => undefined), aborted]); + // The canceled promise settles immediately, but its queued task remains + // behind its predecessor so later compactions cannot overtake active work. + return await Promise.race([queued, aborted]); } finally { removeAbortListener(); } @@ -413,8 +403,7 @@ function readCompactionOverrideEntries(params: CompactEmbeddedAgentSessionParams inheritedRecord?: Record; inheritedPath?: string; }> = []; - const defaultCompaction = readRecord(readRecord(params.config?.agents)?.defaults)?.compaction; - const defaultRecord = readRecord(defaultCompaction); + const defaultRecord = asOptionalRecord(params.config?.agents?.defaults?.compaction); if (defaultRecord) { entries.push({ path: "agents.defaults", record: defaultRecord }); } @@ -427,8 +416,7 @@ function readCompactionOverrideEntries(params: CompactEmbeddedAgentSessionParams const id = typeof agent?.id === "string" ? agent.id.trim().toLowerCase() : ""; return id === agentId; }); - const agentCompaction = readRecord(activeAgent)?.compaction; - const agentRecord = readRecord(agentCompaction); + const agentRecord = asOptionalRecord(activeAgent?.compaction); if (agentRecord) { entries.push({ path: `agents.list.${agentId}`, @@ -448,12 +436,6 @@ function readAgentIdFromSessionKey(sessionKey: string | undefined): string | und return parts[1]?.trim() || undefined; } -function readRecord(value: unknown): Record | undefined { - return value && typeof value === "object" && !Array.isArray(value) - ? (value as Record) - : undefined; -} - async function compactCodexNativeThread( params: CompactEmbeddedAgentSessionParams, options: CodexAppServerCompactOptions, @@ -464,22 +446,16 @@ async function compactCodexNativeThread( sessionKey: params.sessionKey, trigger: params.trigger, }); - return { - ok: true, + return codexNativeCompactionResult(params, { compacted: false, reason: "codex app-server owns automatic compaction", - result: { - summary: "", - firstKeptEntryId: "", - tokensBefore: params.currentTokenCount ?? 0, - details: { - backend: "codex-app-server", - skipped: true, - reason: "non_manual_trigger", - trigger: params.trigger ?? "unknown", - }, + details: { + backend: "codex-app-server", + skipped: true, + reason: "non_manual_trigger", + trigger: params.trigger ?? "unknown", }, - }; + }); } const nativeExecutionBlock = resolveCodexNativeExecutionBlock({ config: params.config, @@ -567,6 +543,21 @@ async function compactCodexNativeThread( agentDir: params.agentDir, config: params.config, }); + let releaseThreadSubscription: (() => Promise) | undefined; + const releaseCompactionThread = async (threadId: string) => { + if ( + await unsubscribeCodexThreadBestEffort(client, { + threadId, + timeoutMs: CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, + }) + ) { + return; + } + await closeCodexStartupClientBestEffort(client); + throw new CodexAppServerUnsafeSubscriptionError( + `Codex compaction thread subscription could not be released: ${threadId}`, + ); + }; const completionWatch = watchCodexNativeCompactionCompletion({ client, threadId: binding.threadId, @@ -575,6 +566,7 @@ async function compactCodexNativeThread( interruptGraceMs: options.nativeInterruptGraceMs ?? CODEX_NATIVE_COMPACTION_INTERRUPT_GRACE_MS, retireUnconfirmed: async () => { + releaseThreadSubscription = undefined; const transportStopped = await client.closeAndWait({ exitTimeoutMs: 5_000, forceKillDelayMs: 250, @@ -610,6 +602,22 @@ async function compactCodexNativeThread( throw new Error("failed to detach unconfirmed codex app-server thread binding"); }, }); + const acquireThreadSubscription = async (timeoutMs?: number) => { + if (!isIncognitoSessionKey(params.sessionKey)) { + // Remove any idle ownership first: sibling cleanup must not evict + // this subscription while compaction still awaits terminal events. + if (!(await consumeCodexAppServerLiveThread(client, binding.threadId))) { + await resumeCodexAppServerThread({ + client, + abandonClient: async () => closeCodexStartupClientBestEffort(client), + request: { threadId: binding.threadId, excludeTurns: true }, + timeoutMs: timeoutMs ?? appServer.requestTimeoutMs, + ...(params.abortSignal ? { signal: params.abortSignal } : {}), + }); + } + releaseThreadSubscription = async () => releaseCompactionThread(binding.threadId); + } + }; const beginNativeCompactionRequest = async (timeoutMs?: number) => { completionWatch.beginRequest(); const requestParams = { threadId: binding.threadId }; @@ -668,6 +676,11 @@ async function compactCodexNativeThread( }; } binding = currentBinding; + const guardedRequestTimeoutMs = Math.min( + appServer.requestTimeoutMs, + CODEX_APP_SERVER_BINDING_GUARDED_REQUEST_TIMEOUT_MS, + ); + await acquireThreadSubscription(guardedRequestTimeoutMs); await clearContextEngineProjectionBeforeNativeCompaction({ sessionId: params.sessionId, bindingStore: options.bindingStore, @@ -675,12 +688,7 @@ async function compactCodexNativeThread( binding, }); try { - await beginNativeCompactionRequest( - Math.min( - appServer.requestTimeoutMs, - CODEX_APP_SERVER_BINDING_GUARDED_REQUEST_TIMEOUT_MS, - ), - ); + await beginNativeCompactionRequest(guardedRequestTimeoutMs); return { started: true as const, accepted: true as const }; } catch (error) { await options.bindingStore.mutate(bindingIdentity, { @@ -702,6 +710,7 @@ async function compactCodexNativeThread( } } else { params.abortSignal?.throwIfAborted(); + await acquireThreadSubscription(); try { await beginNativeCompactionRequest(); } catch (error) { @@ -742,8 +751,12 @@ async function compactCodexNativeThread( }; } finally { completionWatch.cancel(); - if (shouldReleaseDefaultLease) { - releaseLeasedSharedCodexAppServerClient(client); + try { + await releaseThreadSubscription?.(); + } finally { + if (shouldReleaseDefaultLease) { + releaseLeasedSharedCodexAppServerClient(client); + } } } const resultDetails: JsonObject = { @@ -759,16 +772,7 @@ async function compactCodexNativeThread( } : {}), }; - return { - ok: true, - compacted: true, - result: { - summary: "", - firstKeptEntryId: "", - tokensBefore: params.currentTokenCount ?? 0, - details: resultDetails, - }, - }; + return codexNativeCompactionResult(params, { compacted: true, details: resultDetails }); }, ); } catch (error) { @@ -791,6 +795,23 @@ async function compactCodexNativeThread( } } +function codexNativeCompactionResult( + params: CompactEmbeddedAgentSessionParams, + outcome: { compacted: boolean; reason?: string; details: JsonObject }, +): EmbeddedAgentCompactResult { + return { + ok: true, + compacted: outcome.compacted, + ...(outcome.reason ? { reason: outcome.reason } : {}), + result: { + summary: "", + firstKeptEntryId: "", + tokensBefore: params.currentTokenCount ?? 0, + details: outcome.details, + }, + }; +} + function skippedCodexNativeCompactionResult( params: CompactEmbeddedAgentSessionParams, skipped: { @@ -800,25 +821,19 @@ function skippedCodexNativeCompactionResult( currentThreadId?: string; }, ): EmbeddedAgentCompactResult { - return { - ok: true, + return codexNativeCompactionResult(params, { compacted: false, reason: skipped.reason, - result: { - summary: "", - firstKeptEntryId: "", - tokensBefore: params.currentTokenCount ?? 0, - details: { - backend: "codex-app-server", - skipped: true, - reason: skipped.code, - request: "after_context_engine", - trigger: params.trigger ?? "unknown", - ...(skipped.expectedThreadId ? { expectedThreadId: skipped.expectedThreadId } : {}), - ...(skipped.currentThreadId ? { currentThreadId: skipped.currentThreadId } : {}), - }, + details: { + backend: "codex-app-server", + skipped: true, + reason: skipped.code, + request: "after_context_engine", + trigger: params.trigger ?? "unknown", + ...(skipped.expectedThreadId ? { expectedThreadId: skipped.expectedThreadId } : {}), + ...(skipped.currentThreadId ? { currentThreadId: skipped.currentThreadId } : {}), }, - }; + }); } function failedCodexThreadBindingCompactionResult( diff --git a/extensions/codex/src/app-server/dynamic-tool-build.test.ts b/extensions/codex/src/app-server/dynamic-tool-build.test.ts index 2d98b08ca39f..a3cfc1044a5c 100644 --- a/extensions/codex/src/app-server/dynamic-tool-build.test.ts +++ b/extensions/codex/src/app-server/dynamic-tool-build.test.ts @@ -17,6 +17,7 @@ import { resolveCodexAppServerExecutionCwd, resolveCodexExternalSandboxPolicyForOpenClawSandbox, resolveCodexMessageToolProvider, + resolveCodexSandboxEnvironmentSelection, shouldEnableCodexAppServerNativeToolSurface, } from "./dynamic-tool-build.js"; import { @@ -188,6 +189,42 @@ describe("Codex app-server dynamic tool build", () => { ).toBe("discord"); }); + const sandboxEnvironment = { environmentId: "sandbox-1", cwd: "/workspace" }; + + it.each([ + { + name: "restricted without a sandbox", + environment: undefined, + nativeToolSurfaceEnabled: false, + expected: [], + }, + { + name: "restricted with a sandbox", + environment: sandboxEnvironment, + nativeToolSurfaceEnabled: false, + expected: [], + }, + { + name: "native without a sandbox", + environment: undefined, + nativeToolSurfaceEnabled: true, + expected: undefined, + }, + { + name: "native with a sandbox", + environment: sandboxEnvironment, + nativeToolSurfaceEnabled: true, + expected: [sandboxEnvironment], + }, + ])("preserves the explicit Codex environment selection when $name", (testCase) => { + expect( + resolveCodexSandboxEnvironmentSelection( + testCase.environment, + testCase.nativeToolSurfaceEnabled, + ), + ).toEqual(testCase.expected); + }); + it("maps sandbox exec-server cwd through the remote workspace mapping", () => { expect( resolveCodexAppServerExecutionCwd({ diff --git a/extensions/codex/src/app-server/dynamic-tool-build.ts b/extensions/codex/src/app-server/dynamic-tool-build.ts index 6bab8d9860d6..8b6779efe56f 100644 --- a/extensions/codex/src/app-server/dynamic-tool-build.ts +++ b/extensions/codex/src/app-server/dynamic-tool-build.ts @@ -614,7 +614,9 @@ export function resolveCodexSandboxEnvironmentSelection( environment: CodexSandboxExecEnvironment | undefined, nativeToolSurfaceEnabled: boolean, ): CodexTurnEnvironmentParams[] | undefined { - return environment && nativeToolSurfaceEnabled ? [environment] : undefined; + // Omitting this selection while a turn sets cwd restores Codex's local + // environment; an explicit empty selection keeps native tools disabled. + return nativeToolSurfaceEnabled ? (environment ? [environment] : undefined) : []; } /** Chooses the cwd visible to Codex native execution after sandbox exec-server setup. */ export function resolveCodexAppServerExecutionCwd(params: { diff --git a/extensions/codex/src/app-server/dynamic-tools.test.ts b/extensions/codex/src/app-server/dynamic-tools.test.ts index 2bf5daf479c2..c4cac0e3d4c9 100644 --- a/extensions/codex/src/app-server/dynamic-tools.test.ts +++ b/extensions/codex/src/app-server/dynamic-tools.test.ts @@ -27,6 +27,7 @@ import { setActivePluginRegistry, } from "openclaw/plugin-sdk/plugin-test-runtime"; import { createOpenClawTestState } from "openclaw/plugin-sdk/test-state"; +import { estimateToolResultTextChars } from "openclaw/plugin-sdk/text-utility-runtime"; import { afterEach, describe, expect, it, vi } from "vitest"; import { createCodexDynamicToolBridge } from "./dynamic-tools.js"; import { @@ -928,6 +929,73 @@ describe("createCodexDynamicToolBridge", () => { expect(text).toContain("rerun with narrower args"); }); + it.each([ + { label: "ASCII", character: "a", truncated: false }, + { label: "dense CJK", character: "你", truncated: true }, + ])( + "budgets $label tool results by their shared token weight", + async ({ character, truncated }) => { + const original = character.repeat(16_000); + const bridge = createBridgeWithToolResult("large_lookup", textToolResult(original), { + contextWindowTokens: 128_000, + }); + + const result = await bridge.handleToolCall({ + threadId: "thread-1", + turnId: "turn-1", + callId: "call-weighted", + namespace: null, + tool: "large_lookup", + arguments: {}, + }); + const firstItem = result.contentItems[0]; + if (firstItem?.type !== "inputText" || typeof firstItem.text !== "string") { + throw new Error("expected inputText tool result"); + } + expect(estimateToolResultTextChars(firstItem.text)).toBeLessThanOrEqual(32_000); + if (truncated) { + expect(firstItem.text).toContain("original 16000 chars, weighted budget 32000"); + expect(firstItem.text.length).toBeLessThan(9_000); + } else { + expect(firstItem.text).toBe(original); + } + }, + ); + + it.each([ + { label: "missing", contextWindowTokens: undefined, maxChars: 16_000 }, + { label: "zero", contextWindowTokens: 0, maxChars: 16_000 }, + { label: "negative", contextWindowTokens: -1, maxChars: 16_000 }, + { label: "non-finite", contextWindowTokens: Number.POSITIVE_INFINITY, maxChars: 16_000 }, + { label: "tiny", contextWindowTokens: 1, maxChars: 1 }, + { label: "extra-large", contextWindowTokens: 200_000, maxChars: 64_000 }, + ])( + "preserves the canonical cap for $label contexts", + async ({ contextWindowTokens, maxChars }) => { + const bridge = createBridgeWithToolResult( + "large_lookup", + textToolResult("x".repeat(70_000)), + { + contextWindowTokens, + }, + ); + + const result = await bridge.handleToolCall({ + threadId: "thread-1", + turnId: "turn-1", + callId: "call-context-cap", + namespace: null, + tool: "large_lookup", + arguments: {}, + }); + const firstItem = result.contentItems[0]; + if (firstItem?.type !== "inputText" || typeof firstItem.text !== "string") { + throw new Error("expected inputText tool result"); + } + expect(firstItem.text.length).toBe(maxChars); + }, + ); + it("applies the context-share ceiling for small effective windows", async () => { const bridge = createCodexDynamicToolBridge({ tools: [ @@ -959,7 +1027,7 @@ describe("createCodexDynamicToolBridge", () => { it("keeps a whole code point when dynamic tool text crosses the automatic boundary", async () => { const maxChars = 16_000; const totalChars = 20_000; - const noticeText = `...(OpenClaw truncated dynamic tool result: original ${totalChars} chars, showing ${maxChars}; rerun with narrower args.)`; + const noticeText = `...(OpenClaw truncated dynamic tool result: original ${totalChars} chars, weighted budget ${maxChars}; rerun with narrower args.)`; const textBudget = maxChars - noticeText.length - 1; const prefix = "a".repeat(textBudget - 1); const longText = `${prefix}😀${"z".repeat(totalChars - prefix.length - 2)}`; @@ -1021,6 +1089,42 @@ describe("createCodexDynamicToolBridge", () => { expect(text).not.toContain("b".repeat(10_000)); }); + it("shares weighted budget across mixed text blocks while preserving images", async () => { + const bridge = createBridgeWithToolResult( + "mixed_lookup", + { + content: [ + { type: "text", text: "a".repeat(4_000) }, + { type: "image", mimeType: "image/png", data: COMPUTER_FRAME_IMAGE }, + { type: "text", text: "你".repeat(9_000) }, + ], + details: {}, + }, + { contextWindowTokens: 128_000 }, + ); + + const result = await bridge.handleToolCall({ + threadId: "thread-1", + turnId: "turn-1", + callId: "call-mixed-weighted", + namespace: null, + tool: "mixed_lookup", + arguments: {}, + }); + const text = result.contentItems + .map((item) => (item.type === "inputText" && typeof item.text === "string" ? item.text : "")) + .join(""); + + expect(result.contentItems.map((item) => item.type)).toEqual([ + "inputText", + "inputImage", + "inputText", + ]); + expect(result.contentItems[0]).toEqual({ type: "inputText", text: "a".repeat(4_000) }); + expect(estimateToolResultTextChars(text)).toBeLessThanOrEqual(32_000); + expect(text).toContain("original 13000 chars, weighted budget 32000"); + }); + it.each([ { toolName: "tts", mediaUrl: "/tmp/reply.opus", audioAsVoice: true }, { toolName: "image_generate", mediaUrl: "/tmp/generated.png" }, diff --git a/extensions/codex/src/app-server/dynamic-tools.ts b/extensions/codex/src/app-server/dynamic-tools.ts index 95e10707bcc7..8944c60a96ec 100644 --- a/extensions/codex/src/app-server/dynamic-tools.ts +++ b/extensions/codex/src/app-server/dynamic-tools.ts @@ -46,7 +46,12 @@ import { expectDefined } from "openclaw/plugin-sdk/expect-runtime"; import type { ImageContent, TextContent } from "openclaw/plugin-sdk/llm"; import { normalizeOpenAIToolSchemas } from "openclaw/plugin-sdk/provider-tools"; import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; -import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; +import { + DEFAULT_MAX_LIVE_TOOL_RESULT_CHARS, + estimateToolResultTextChars, + resolveLiveToolResultMaxChars, + sliceToolResultTextToBudget, +} from "openclaw/plugin-sdk/text-utility-runtime"; import type { CodexDynamicToolsLoading } from "./config.js"; import { createFailedDynamicToolResponse, @@ -398,29 +403,6 @@ const EXPLICIT_MESSAGE_PROVIDER_KEYS = ["channel", "provider"]; const EXPLICIT_MESSAGE_TARGET_KEYS = ["target", "to", "channelId"]; const EXPLICIT_MESSAGE_THREAD_KEYS = ["threadId", "thread_id", "messageThreadId", "topicId"]; const EXPLICIT_MESSAGE_REPLY_KEYS = ["replyTo", "replyToId", "replyToIdFull"]; -const DEFAULT_CODEX_DYNAMIC_TOOL_RESULT_MAX_CHARS = 16_000; -const LARGE_CONTEXT_CODEX_DYNAMIC_TOOL_RESULT_MAX_CHARS = 32_000; -const XL_CONTEXT_CODEX_DYNAMIC_TOOL_RESULT_MAX_CHARS = 64_000; - -function resolveCodexDynamicToolResultMaxChars(contextWindowTokens?: number): number { - if ( - typeof contextWindowTokens !== "number" || - !Number.isFinite(contextWindowTokens) || - contextWindowTokens <= 0 - ) { - return DEFAULT_CODEX_DYNAMIC_TOOL_RESULT_MAX_CHARS; - } - const tokens = Math.floor(contextWindowTokens); - const autoCap = - tokens >= 200_000 - ? XL_CONTEXT_CODEX_DYNAMIC_TOOL_RESULT_MAX_CHARS - : tokens >= 100_000 - ? LARGE_CONTEXT_CODEX_DYNAMIC_TOOL_RESULT_MAX_CHARS - : DEFAULT_CODEX_DYNAMIC_TOOL_RESULT_MAX_CHARS; - // Match the core live-result context-share ceiling without importing core - // internals across the bundled-plugin boundary. - return Math.min(autoCap, Math.max(1, Math.floor(tokens * 0.3) * 4)); -} function computerFrameImageIdentity( content: AgentToolResult["content"] | undefined, @@ -466,9 +448,13 @@ export function createCodexDynamicToolBridge(params: { directToolNames?: Iterable; }): CodexDynamicToolBridge { const toolResultHookContext = toToolResultHookContext(params.hookContext); - const toolResultMaxChars = resolveCodexDynamicToolResultMaxChars( - params.hookContext?.contextWindowTokens, - ); + const contextWindowTokens = params.hookContext?.contextWindowTokens; + const toolResultMaxChars = + typeof contextWindowTokens === "number" && + Number.isFinite(contextWindowTokens) && + contextWindowTokens > 0 + ? Math.max(1, resolveLiveToolResultMaxChars({ contextWindowTokens })) + : DEFAULT_MAX_LIVE_TOOL_RESULT_CHARS; const availableProjection = projectCodexDynamicTools(params.tools); const registeredProjection = params.registeredTools ? projectCodexDynamicTools(params.registeredTools) @@ -1433,23 +1419,28 @@ function withDynamicToolAsyncStarted( function normalizeToolResultMaxChars(maxChars: number): number { return typeof maxChars === "number" && Number.isFinite(maxChars) && maxChars > 0 ? Math.floor(maxChars) - : DEFAULT_CODEX_DYNAMIC_TOOL_RESULT_MAX_CHARS; + : DEFAULT_MAX_LIVE_TOOL_RESULT_CHARS; } function convertToolContents( content: Array, - toolResultMaxChars = DEFAULT_CODEX_DYNAMIC_TOOL_RESULT_MAX_CHARS, + toolResultMaxChars = DEFAULT_MAX_LIVE_TOOL_RESULT_CHARS, ): CodexDynamicToolCallOutputContentItem[] { const maxChars = normalizeToolResultMaxChars(toolResultMaxChars); const totalTextChars = content.reduce( (total, item) => total + (item.type === "text" ? item.text.length : 0), 0, ); - if (totalTextChars <= maxChars) { + const totalTextBudget = content.reduce( + (total, item) => total + (item.type === "text" ? estimateToolResultTextChars(item.text) : 0), + 0, + ); + if (totalTextBudget <= maxChars) { return content.flatMap(convertToolContent); } - const noticeText = `...(OpenClaw truncated dynamic tool result: original ${totalTextChars} chars, showing ${maxChars}; rerun with narrower args.)`; + const noticeText = `...(OpenClaw truncated dynamic tool result: original ${totalTextChars} chars, weighted budget ${maxChars}; rerun with narrower args.)`; const notice = `\n${noticeText}`; - const textBudget = Math.max(0, maxChars - notice.length); + const noticeChars = estimateToolResultTextChars(notice); + const textBudget = Math.max(0, maxChars - noticeChars); let remainingTextBudget = textBudget; let appendedNotice = false; const output: CodexDynamicToolCallOutputContentItem[] = []; @@ -1461,15 +1452,14 @@ function convertToolContents( if (appendedNotice) { continue; } - if (notice.length >= maxChars) { - output.push({ type: "inputText", text: truncateUtf16Safe(noticeText, maxChars) }); + if (noticeChars >= maxChars) { + output.push({ type: "inputText", text: sliceToolResultTextToBudget(noticeText, maxChars) }); appendedNotice = true; continue; } - const sliceLength = Math.min(item.text.length, remainingTextBudget); - remainingTextBudget -= sliceLength; - const shouldAppendNotice = remainingTextBudget <= 0; - const text = truncateUtf16Safe(item.text, sliceLength); + const text = sliceToolResultTextToBudget(item.text, remainingTextBudget); + remainingTextBudget -= estimateToolResultTextChars(text); + const shouldAppendNotice = remainingTextBudget <= 0 || text.length < item.text.length; if (shouldAppendNotice) { // The notice budget is reserved before slicing text, so the combined // result is already bounded without another boundary-sensitive cut. @@ -1480,7 +1470,7 @@ function convertToolContents( } } if (!appendedNotice) { - output.push({ type: "inputText", text: truncateUtf16Safe(noticeText, maxChars) }); + output.push({ type: "inputText", text: sliceToolResultTextToBudget(noticeText, maxChars) }); } return output; } diff --git a/extensions/codex/src/app-server/event-projector-assistant.ts b/extensions/codex/src/app-server/event-projector-assistant.ts index 64d39ee81371..19ae5b4a05f1 100644 --- a/extensions/codex/src/app-server/event-projector-assistant.ts +++ b/extensions/codex/src/app-server/event-projector-assistant.ts @@ -253,7 +253,13 @@ export class CodexAssistantProjection { this.assistantTextByItem.set(typedItemId, text); return; } - if (!text) { + if ( + text === undefined || + (!text && + (phase === "commentary" || + activeItemIds.size > 0 || + readString(item, "type") !== "message")) + ) { return; } const itemId = rawItemId ?? `raw-assistant-${this.assistantItemOrder.length + 1}`; @@ -263,6 +269,7 @@ export class CodexAssistantProjection { pendingTerminalAssistantEchoItemId === undefined && activeItemIds.size === 0; if ( + text && phase !== "commentary" && candidateWasSupersededBeforeRaw && itemId !== this.streamedPartialAssistantItemId && @@ -275,6 +282,10 @@ export class CodexAssistantProjection { } this.rememberAssistantItem(itemId); this.assistantTextByItem.set(itemId, text); + // Empty raw finals prove an actual stop; retain that fact without publishing fake output. + if (!text) { + return; + } this.rawPromotedAssistantItemIds.add(itemId); if (phase === "commentary") { this.emitCommentaryProgress({ itemId, text }); diff --git a/extensions/codex/src/app-server/event-projector-values.ts b/extensions/codex/src/app-server/event-projector-values.ts index d182788bf0fd..1e620831f15e 100644 --- a/extensions/codex/src/app-server/event-projector-values.ts +++ b/extensions/codex/src/app-server/event-projector-values.ts @@ -82,20 +82,18 @@ export function splitPlanText(text: string): string[] { export function extractRawAssistantText(item: JsonObject): string | undefined { const content = Array.isArray(item.content) ? item.content : []; - const text = content - .flatMap((entry) => { - if (!isJsonObject(entry)) { - return []; - } - const type = readString(entry, "type"); - if (type !== "output_text" && type !== "text") { - return []; - } - const value = readString(entry, "text"); - return value ? [value] : []; - }) - .join(""); - return text.trim() || undefined; + const parts = content.flatMap((entry) => { + if (!isJsonObject(entry)) { + return []; + } + const type = readString(entry, "type"); + if (type !== "output_text" && type !== "text") { + return []; + } + const value = readString(entry, "text"); + return value === undefined ? [] : [value]; + }); + return parts.length > 0 ? parts.join("").trim() : undefined; } export function readItemString(item: CodexThreadItem, key: string): string | undefined { diff --git a/extensions/codex/src/app-server/event-projector.commentary.test.ts b/extensions/codex/src/app-server/event-projector.commentary.test.ts index 123b9aa189ca..f9acff07f43c 100644 --- a/extensions/codex/src/app-server/event-projector.commentary.test.ts +++ b/extensions/codex/src/app-server/event-projector.commentary.test.ts @@ -98,6 +98,111 @@ describe("CodexAppServerEventProjector commentary projection", () => { expect(result.replayMetadata).toEqual({ hadPotentialSideEffects: false, replaySafe: true }); }); + it.each([ + { itemId: "msg_mock_1", text: "" }, + { itemId: "msg_mock_1", text: " \n " }, + { itemId: undefined, text: "" }, + { itemId: undefined, text: " \n " }, + ])( + "preserves an explicit raw empty stop ($itemId) after a settled write", + async ({ itemId, text }) => { + const onAgentEvent = vi.fn(); + const projector = await createProjector({ ...(await createParams()), onAgentEvent }); + const priorAssistant = { type: "agentMessage", id: "msg-before-write", text: "" }; + await projector.handleNotification(forCurrentTurn("item/started", { item: priorAssistant })); + await projector.handleNotification( + forCurrentTurn("item/completed", { item: priorAssistant }), + ); + const item = { + type: "dynamicToolCall", + id: "call-write", + namespace: null, + tool: "write", + arguments: { path: "note.txt", content: "written once" }, + status: "inProgress", + contentItems: null, + success: null, + durationMs: null, + }; + await projector.handleNotification(forCurrentTurn("item/started", { item })); + projector.recordDynamicToolCall({ + callId: item.id, + tool: item.tool, + arguments: item.arguments, + }); + projector.recordDynamicToolResult({ + callId: item.id, + tool: item.tool, + success: true, + sideEffectEvidence: true, + contentItems: [{ type: "inputText", text: "written once" }], + }); + await projector.handleNotification( + forCurrentTurn("item/completed", { + item: { + ...item, + status: "completed", + contentItems: [{ type: "inputText", text: "written once" }], + success: true, + durationMs: 1, + }, + }), + ); + await projector.handleNotification( + forCurrentTurn("rawResponseItem/completed", { + item: { + type: "message", + ...(itemId ? { id: itemId } : {}), + role: "assistant", + content: [{ type: "output_text", text }], + }, + }), + ); + await projector.handleNotification(turnCompleted([])); + + const result = projector.buildResult(buildEmptyToolTelemetry()); + expect(result.assistantTexts).toEqual([]); + expect(result.lastAssistant).toBeUndefined(); + expect(result.currentAttemptAssistant).toMatchObject({ + stopReason: "stop", + content: [{ type: "text", text: "" }], + }); + expect(result.replayMetadata).toEqual({ hadPotentialSideEffects: true, replaySafe: false }); + expect(result.itemLifecycle).toEqual({ startedCount: 2, completedCount: 2, activeCount: 0 }); + expect(result.messagesSnapshot.filter((message) => message.role === "assistant")).toEqual([ + expect.objectContaining({ content: [expect.objectContaining({ type: "toolCall" })] }), + ]); + expect(onAgentEvent.mock.calls.some(([event]) => event.stream === "assistant")).toBe(false); + }, + ); + + it.each([ + { label: "missing content", content: [] }, + { label: "missing text", content: [{ type: "output_text" }] }, + { label: "non-text content", content: [{ type: "reasoning", text: "" }] }, + { label: "commentary", content: [{ type: "output_text", text: "" }], phase: "commentary" }, + { label: "active tool", content: [{ type: "output_text", text: "" }], active: true }, + ])("does not fabricate a terminal assistant for $label", async ({ content, phase, active }) => { + const projector = await createProjector(); + if (active) { + await projector.handleNotification( + forCurrentTurn("item/started", { + item: { type: "commandExecution", id: "pending-tool", status: "inProgress" }, + }), + ); + } + await projector.handleNotification( + forCurrentTurn("rawResponseItem/completed", { + item: { type: "message", role: "assistant", ...(phase ? { phase } : {}), content }, + }), + ); + await projector.handleNotification(turnCompleted([])); + + expect( + projector.buildResult(buildEmptyToolTelemetry()).currentAttemptAssistant, + ).toBeUndefined(); + }); + it("streams commentary agent messages as keyed progress events", async () => { const onAgentEvent = vi.fn(); const onPartialReply = vi.fn(); diff --git a/extensions/codex/src/app-server/run-attempt-resources.ts b/extensions/codex/src/app-server/run-attempt-resources.ts index 638380f459e0..310852118b20 100644 --- a/extensions/codex/src/app-server/run-attempt-resources.ts +++ b/extensions/codex/src/app-server/run-attempt-resources.ts @@ -75,6 +75,7 @@ export function prepareCodexAttemptResources(prompt: CodexAttemptPrompt) { | CodexNativePreToolUseFailure["disposition"] | undefined, releaseSharedClientLease: undefined as (() => void) | undefined, + startupClientUnsafe: false, sharedCodexClientRetiredForOneShotCleanup: false, sandboxExecEnvironmentAcquired: false, codexEnvironmentSelection: undefined as CodexTurnEnvironmentParams[] | undefined, diff --git a/extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts b/extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts index 3e15c9e5e5f0..7be99beab6b7 100644 --- a/extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts +++ b/extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts @@ -328,7 +328,16 @@ describe("Codex app-server main thread cleanup", () => { expect(requests[0]?.params).toEqual(expect.objectContaining({ ephemeral: true })); }); - it("unsubscribes an incognito Codex thread when turn start fails", async () => { + it.each([ + { reason: "fails", error: new Error("turn start exploded") }, + { + reason: "is cancelled before its request is written", + error: Object.assign(new Error("turn/start aborted"), { + code: "CODEX_APP_SERVER_LOCAL_REQUEST_CANCELLED", + mayHaveWritten: false, + }), + }, + ])("unsubscribes an incognito Codex thread when turn start $reason", async ({ error }) => { const sessionFile = path.join(tempDir, "session.jsonl"); const workspaceDir = path.join(tempDir, "workspace"); const sessionKey = "agent:main:dashboard:incognito-failed-turn"; @@ -339,7 +348,7 @@ describe("Codex app-server main thread cleanup", () => { return threadStartResult(); } if (method === "turn/start") { - throw new Error("turn start exploded"); + throw error; } return {}; }); @@ -359,7 +368,7 @@ describe("Codex app-server main thread cleanup", () => { bindingStore: testCodexAppServerBindingStore, clientFactory, }), - ).rejects.toThrow("turn start exploded"); + ).rejects.toThrow(error.message); expect(requests.map((entry) => entry.method)).toEqual([ "thread/start", "turn/start", @@ -373,6 +382,115 @@ describe("Codex app-server main thread cleanup", () => { await expect(readCodexAppServerBinding(sessionFile)).resolves.toBeUndefined(); }); + it.each([ + { label: "confirms", interruptFails: false }, + { label: "cannot confirm", interruptFails: true }, + ])( + "$label an indeterminate native turn before releasing its thread", + async ({ interruptFails }) => { + const sessionFile = path.join(tempDir, "cancelled-start-session.jsonl"); + const workspaceDir = path.join(tempDir, "cancelled-start-workspace"); + const harness = createClientHarness(); + const abort = new AbortController(); + vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client); + + const params = createParams(sessionFile, workspaceDir); + params.abortSignal = abort.signal; + const run = runCodexAppServerAttempt(params, { + bindingStore: testCodexAppServerBindingStore, + }); + const failure = run.then( + () => undefined, + (error: unknown) => error, + ); + const initialize = await waitForHarnessRequest(harness, "initialize"); + harness.send({ + id: initialize.id, + result: { userAgent: `openclaw/${CODEX_APP_SERVER_VERSION} (macOS; test)` }, + }); + const threadStart = await waitForHarnessRequest(harness, "thread/start"); + harness.send({ id: threadStart.id, result: threadStartResult() }); + const turnStart = await waitForHarnessRequest(harness, "turn/start"); + + abort.abort("cancelled"); + const interrupt = await waitForHarnessRequest(harness, "turn/interrupt"); + expect(JSON.parse(harness.writes.at(-1) ?? "{}")).toMatchObject({ + method: "turn/interrupt", + params: { threadId: "thread-1", turnId: "" }, + }); + harness.send({ id: turnStart.id, result: turnStartResult() }); + harness.send( + interruptFails + ? { id: interrupt.id, error: { code: -32_000, message: "startup interrupt failed" } } + : { id: interrupt.id, result: {} }, + ); + if (!interruptFails) { + const unsubscribe = await waitForHarnessRequest(harness, "thread/unsubscribe"); + harness.send({ id: unsubscribe.id, result: {} }); + } + await expect(failure).resolves.toMatchObject({ message: "turn/start aborted" }); + expect(harness.writes.map((entry) => JSON.parse(entry).method)).toEqual([ + "initialize", + "initialized", + "thread/start", + "turn/start", + "turn/interrupt", + ...(!interruptFails ? ["thread/unsubscribe"] : []), + ]); + expect(harness.stdinDestroyed).toBe(interruptFails); + }, + ); + + it("preserves startup cancellation when unsafe client retirement rejects", async () => { + const sessionFile = path.join(tempDir, "retirement-failure-session.jsonl"); + const workspaceDir = path.join(tempDir, "retirement-failure-workspace"); + const startupError = Object.assign(new Error("turn/start aborted"), { + code: "CODEX_APP_SERVER_LOCAL_REQUEST_CANCELLED", + mayHaveWritten: true, + }); + const close = vi.fn(); + const closeAndWait = vi.fn(async () => { + throw new Error("client retirement failed"); + }); + const request = vi.fn(async (method: string) => { + if (method === "thread/start") { + return threadStartResult(); + } + if (method === "turn/start") { + throw startupError; + } + if (method === "turn/interrupt") { + throw new Error("startup interrupt failed"); + } + throw new Error(`unexpected cleanup request: ${method}`); + }); + const clientFactory: CodexAppServerClientFactory = multiplexedClientFactory(async () => { + return { + ...mockClientRuntimeMethods(), + request, + close, + closeAndWait, + addNotificationHandler: () => () => undefined, + addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, + } as never; + }); + + await expect( + runCodexAppServerAttempt(createParams(sessionFile, workspaceDir), { + bindingStore: testCodexAppServerBindingStore, + clientFactory, + }), + ).rejects.toBe(startupError); + expect(request.mock.calls.map(([method]) => method)).toEqual([ + "thread/start", + "turn/start", + "turn/interrupt", + ]); + expect(closeAndWait).toHaveBeenCalledOnce(); + expect(close).toHaveBeenCalledOnce(); + }); + it("keeps an interrupted shared turn subscribed until its exact terminal arrives", async () => { const sessionFile = path.join(tempDir, "cancelled-session.jsonl"); const workspaceDir = path.join(tempDir, "cancelled-workspace"); @@ -404,6 +522,10 @@ describe("Codex app-server main thread cleanup", () => { abort.abort("cancelled"); const interrupt = await waitForHarnessRequest(harness, "turn/interrupt"); + expect(JSON.parse(harness.writes.at(-1) ?? "{}")).toMatchObject({ + method: "turn/interrupt", + params: { threadId: "thread-1", turnId: "turn-1" }, + }); harness.send({ id: interrupt.id, result: {} }); await new Promise((resolve) => { setImmediate(resolve); diff --git a/extensions/codex/src/app-server/run-attempt-turn-request.ts b/extensions/codex/src/app-server/run-attempt-turn-request.ts index 4d559d1ea9a1..fb163c8e4584 100644 --- a/extensions/codex/src/app-server/run-attempt-turn-request.ts +++ b/extensions/codex/src/app-server/run-attempt-turn-request.ts @@ -1,8 +1,13 @@ import { embeddedAgentLog, formatErrorMessage } from "openclaw/plugin-sdk/agent-harness-runtime"; +import { + interruptCodexTurnAndWaitBestEffort, + retireUnsafeCodexTurnClientBestEffort, +} from "./attempt-client-cleanup.js"; import { createCodexModelCallDiagnosticEmitter, utf8JsonByteLength, } from "./attempt-diagnostics.js"; +import { isCodexAppServerIndeterminateRequestCancellationError } from "./client.js"; import { assertCodexTurnStartResponse } from "./protocol-validators.js"; import type { CodexTurnStartResponse } from "./protocol.js"; import { readCodexRateLimitsRevision } from "./rate-limit-cache.js"; @@ -136,9 +141,20 @@ export async function prepareCodexAttemptTurnRequest( throwIfTurnStartAcceptedAfterAbort(); return startedTurn; } catch (error) { - if (acceptedTurnId) { - await turnRuntime.interruptTurn(acceptedTurnId); - releaseCurrentRoute(); + if (acceptedTurnId || isCodexAppServerIndeterminateRequestCancellationError(error)) { + // Codex serializes start/interrupt per thread; an empty id interrupts + // the accepted native turn even when local cancellation hid its response. + try { + resourceState.startupClientUnsafe = !(await interruptCodexTurnAndWaitBestEffort( + resourceState.client, + { threadId: resourceState.thread.threadId, turnId: acceptedTurnId ?? "" }, + )); + if (resourceState.startupClientUnsafe) { + await retireUnsafeCodexTurnClientBestEffort(resourceState.client, "startup interrupt"); + } + } finally { + releaseCurrentRoute(); + } } else { await activeTurnRoute.cancelTurn(); } diff --git a/extensions/codex/src/app-server/run-attempt-turn-start.ts b/extensions/codex/src/app-server/run-attempt-turn-start.ts index 200ba0bac9be..e6b343ff9230 100644 --- a/extensions/codex/src/app-server/run-attempt-turn-start.ts +++ b/extensions/codex/src/app-server/run-attempt-turn-start.ts @@ -242,7 +242,7 @@ export async function startCodexAttemptTurn( threadId: resourceState.thread.threadId, }) : true; - if (!state.timedOut && bindingReleased) { + if (!state.timedOut && bindingReleased && !resourceState.startupClientUnsafe) { const released = await unsubscribeCodexThreadBestEffort(resourceState.client, { threadId: resourceState.thread.threadId, timeoutMs: CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, diff --git a/extensions/codex/src/app-server/settled-turn-context.test.ts b/extensions/codex/src/app-server/settled-turn-context.test.ts index cdef35a40b02..eef2dd652924 100644 --- a/extensions/codex/src/app-server/settled-turn-context.test.ts +++ b/extensions/codex/src/app-server/settled-turn-context.test.ts @@ -1,7 +1,13 @@ import type { AgentMessage } from "openclaw/plugin-sdk/agent-harness-runtime"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { captureCodexSettledTurnFinalizationContext } from "./settled-turn-context.js"; -import { attachCodexMirrorIdentity } from "./upstream-prompt-provenance.js"; +import { attachCodexMirrorAttestation } from "./transcript-mirror-attestation.js"; +import { + attachCodexMirrorIdentity, + attachUpstreamUserText, + readMirrorIdentity, + readUpstreamUserText, +} from "./upstream-prompt-provenance.js"; const mocks = vi.hoisted(() => ({ readHistory: vi.fn(), @@ -37,6 +43,30 @@ function settledTurn() { ]; } +function settledHostPromptTurn() { + const settledMessages = settledTurn(); + settledMessages[0] = attachUpstreamUserText( + message( + { role: "user", content: "Send it.", idempotencyKey: "durable-user-turn" }, + "turn-2:prompt", + ), + "Decorated upstream prompt: Send it.", + ); + const persistedPrompt = { + role: "user", + content: "Send it.", + timestamp: 1, + idempotencyKey: "durable-user-turn", + __openclaw: { senderIsOwner: true, transport: { messageId: "transport-message" } }, + } as AgentMessage; + return { + settledMessages, + persistedPrompt, + historyMessages: [persistedPrompt, ...settledMessages.slice(1)], + mirroredMessages: settledMessages.slice(1), + }; +} + async function captureContext(params: { historyMessages: AgentMessage[]; mirroredMessages: AgentMessage[]; @@ -79,6 +109,61 @@ describe("captureCodexSettledTurnFinalizationContext", () => { expect(context?.messages).not.toBe(historyMessages); }); + it("adopts an exact host-persisted prompt without rewriting its canonical metadata", async () => { + const { persistedPrompt, ...turn } = settledHostPromptTurn(); + + const context = await captureContext(turn); + + expect(context).toEqual({ source: "openclaw-transcript", messages: turn.historyMessages }); + expect(Object.isFrozen(context?.messages)).toBe(true); + expect(context?.messages[0]).toEqual(persistedPrompt); + expect(readMirrorIdentity(context!.messages[0]!)).toBeUndefined(); + expect(readUpstreamUserText(context!.messages[0]!)).toBeUndefined(); + expect(context?.messages[0]).toMatchObject({ + __openclaw: { senderIsOwner: true, transport: { messageId: "transport-message" } }, + }); + }); + + it.each([ + { + name: "missing persisted key", + change: (prompt: AgentMessage) => ({ ...prompt, idempotencyKey: undefined }), + }, + { + name: "different persisted key", + change: (prompt: AgentMessage) => ({ ...prompt, idempotencyKey: "different-user-turn" }), + }, + { + name: "changed prompt content", + change: (prompt: AgentMessage) => ({ ...prompt, content: "Send something else." }), + }, + { + name: "conflicting mirror identity", + change: (prompt: AgentMessage) => attachCodexMirrorIdentity(prompt, "foreign-turn:prompt"), + }, + { + name: "stale Codex mirror attestation", + change: (prompt: AgentMessage) => attachCodexMirrorAttestation(prompt, "stale-fingerprint"), + }, + { + name: "conflicting upstream prompt", + change: (prompt: AgentMessage) => + attachUpstreamUserText(prompt, "Untrusted upstream prompt."), + }, + ])("rejects host-persisted prompt adoption with $name", async ({ change }) => { + const turn = settledHostPromptTurn(); + turn.historyMessages[0] = change(turn.persistedPrompt) as AgentMessage; + + await expect(captureContext(turn)).resolves.toBeUndefined(); + }); + + it("rejects duplicate host-persisted prompt idempotency keys", async () => { + const turn = settledHostPromptTurn(); + turn.historyMessages.unshift({ ...turn.persistedPrompt }); + + await expect(captureContext(turn)).resolves.toBeUndefined(); + }); + it.each([ { name: "missing current prompt", diff --git a/extensions/codex/src/app-server/settled-turn-context.ts b/extensions/codex/src/app-server/settled-turn-context.ts index 7fee60f29c57..2a7bbaa8a440 100644 --- a/extensions/codex/src/app-server/settled-turn-context.ts +++ b/extensions/codex/src/app-server/settled-turn-context.ts @@ -8,8 +8,16 @@ import { readCodexMirroredSessionHistoryMessages, type CodexMirroredSessionHistoryTarget, } from "./session-history.js"; -import { serializeCodexMirrorSourceEvidence } from "./transcript-mirror-attestation.js"; -import { readMirrorIdentity } from "./upstream-prompt-provenance.js"; +import { + readCodexMirrorSourceFingerprint, + serializeCodexMirrorSourceEvidence, +} from "./transcript-mirror-attestation.js"; +import { + attachCodexMirrorIdentity, + attachUpstreamUserText, + readMirrorIdentity, + readUpstreamUserText, +} from "./upstream-prompt-provenance.js"; type SettledTurnFinalizationContext = EmbeddedRunAttemptResult["settledTurnFinalizationContext"]; @@ -30,6 +38,66 @@ function collectUniqueMessageIdentities( return identities; } +function adoptPersistedHostPrompt(params: { + historyMessages: readonly AgentMessage[]; + mirroredMessages: readonly AgentMessage[]; + settledMessages: readonly AgentMessage[]; + turnId: string; +}): { historyMessages: readonly AgentMessage[]; mirroredMessages: readonly AgentMessage[] } { + const promptIdentity = `${params.turnId}:prompt`; + if (params.mirroredMessages.some((message) => readMirrorIdentity(message) === promptIdentity)) { + return params; + } + const sourcePrompt = params.settledMessages[0]; + const sourceKey = (sourcePrompt as { idempotencyKey?: unknown } | undefined)?.idempotencyKey; + if ( + sourcePrompt?.role !== "user" || + readMirrorIdentity(sourcePrompt) !== promptIdentity || + typeof sourceKey !== "string" || + sourceKey.trim().length === 0 + ) { + return params; + } + const matches = params.historyMessages.flatMap((message, index) => + message.role === "user" && + (message as { idempotencyKey?: unknown }).idempotencyKey === sourceKey + ? [{ index, message }] + : [], + ); + const persistedPrompt = matches.length === 1 ? matches[0] : undefined; + const persistedMetadata = persistedPrompt?.message as + | { __openclaw?: { mirrorOrigin?: unknown } } + | undefined; + if ( + !persistedPrompt || + readMirrorIdentity(persistedPrompt.message) !== undefined || + readCodexMirrorSourceFingerprint(persistedPrompt.message) !== undefined || + persistedMetadata?.["__openclaw"]?.mirrorOrigin === "codex-app-server" + ) { + return params; + } + const sourceUpstreamText = readUpstreamUserText(sourcePrompt); + const persistedUpstreamText = readUpstreamUserText(persistedPrompt.message); + if (persistedUpstreamText !== undefined && persistedUpstreamText !== sourceUpstreamText) { + return params; + } + let logicalPrompt = attachCodexMirrorIdentity(persistedPrompt.message, promptIdentity); + if (sourceUpstreamText !== undefined) { + logicalPrompt = attachUpstreamUserText(logicalPrompt, sourceUpstreamText); + } + if ( + serializeCodexMirrorSourceEvidence(logicalPrompt) !== + serializeCodexMirrorSourceEvidence(sourcePrompt) + ) { + return params; + } + // Host-owned prompts are authoritative by durable turn key; adopt their identity + // only in verification views so canonical channel metadata remains untouched. + const historyMessages = [...params.historyMessages]; + historyMessages[persistedPrompt.index] = logicalPrompt; + return { historyMessages, mirroredMessages: [logicalPrompt, ...params.mirroredMessages] }; +} + /** Freezes one complete active transcript branch through the settled tool-result boundary. */ function buildCodexSettledTurnFinalizationContext(params: { historyMessages: readonly AgentMessage[]; @@ -37,6 +105,7 @@ function buildCodexSettledTurnFinalizationContext(params: { settledMessages: readonly AgentMessage[]; turnId: string; }): SettledTurnFinalizationContext | undefined { + const { historyMessages, mirroredMessages } = adoptPersistedHostPrompt(params); const boundaryMessage = params.settledMessages.findLast( (message) => message.role === "toolResult", ); @@ -62,8 +131,8 @@ function buildCodexSettledTurnFinalizationContext(params: { return undefined; } - const historyIdentities = collectUniqueMessageIdentities(params.historyMessages); - const mirroredIdentities = collectUniqueMessageIdentities(params.mirroredMessages); + const historyIdentities = collectUniqueMessageIdentities(historyMessages); + const mirroredIdentities = collectUniqueMessageIdentities(mirroredMessages); if (!historyIdentities || !mirroredIdentities) { return undefined; } @@ -71,7 +140,7 @@ function buildCodexSettledTurnFinalizationContext(params: { if (mirroredBoundaryIndex === undefined) { return undefined; } - const mirroredThroughBoundary = params.mirroredMessages.slice(0, mirroredBoundaryIndex + 1); + const mirroredThroughBoundary = mirroredMessages.slice(0, mirroredBoundaryIndex + 1); if ( mirroredThroughBoundary.length !== requiredIdentities.length || mirroredThroughBoundary.some( @@ -88,8 +157,7 @@ function buildCodexSettledTurnFinalizationContext(params: { for (const mirroredMessage of mirroredThroughBoundary) { const identity = readMirrorIdentity(mirroredMessage); const historyIndex = identity ? historyIdentities.get(identity) : undefined; - const historyMessage = - historyIndex === undefined ? undefined : params.historyMessages[historyIndex]; + const historyMessage = historyIndex === undefined ? undefined : historyMessages[historyIndex]; if ( historyIndex === undefined || historyIndex <= previousHistoryIndex || diff --git a/extensions/codex/src/app-server/side-question.test.ts b/extensions/codex/src/app-server/side-question.test.ts index 54e5b59ebb20..cca69d6d06cd 100644 --- a/extensions/codex/src/app-server/side-question.test.ts +++ b/extensions/codex/src/app-server/side-question.test.ts @@ -21,6 +21,7 @@ import { createCodexTestBindingStore, type CodexAppServerBindingStore, } from "./session-binding.test-helpers.js"; +import { createClientHarness } from "./test-support.js"; const readCodexAppServerBindingMock = vi.fn(); const isCodexAppServerNativeAuthProfileMock = vi.fn(); @@ -69,6 +70,7 @@ vi.mock("./shared-client.js", () => ({ releaseCodexAppServerClientLease: vi.fn((lease: { client?: unknown }) => { lease.client = undefined; }), + retireSharedCodexAppServerClientIfCurrent: vi.fn(), withLeasedCodexAppServerClientStartSelectionRetry: (params: SelectionRetryParams) => withLeasedCodexAppServerClientStartSelectionRetryMock(params), })); @@ -3077,6 +3079,104 @@ describe("runCodexAppServerSideQuestion", () => { ); }); + it.each([ + { label: "after its request is written", written: true, interruptFails: false }, + { label: "before its request is written", written: false, interruptFails: false }, + { + label: "when Codex proves its native turn already ended", + written: true, + interruptFails: false, + alreadyTerminal: true, + }, + { + label: "when its native thread cannot unsubscribe", + written: true, + interruptFails: false, + unsubscribeFails: true, + }, + { label: "when its startup interrupt fails", written: true, interruptFails: true }, + { + label: "when its startup interrupt and client retirement fail", + written: true, + interruptFails: true, + retirementFails: true, + }, + ])( + "scopes side-turn abort cleanup $label", + async ({ written, interruptFails, retirementFails, alreadyTerminal, unsubscribeFails }) => { + const controller = new AbortController(); + const harness = createClientHarness(); + if (retirementFails) { + vi.spyOn(harness.client, "closeAndWait").mockRejectedValueOnce( + new Error("side client retirement failed"), + ); + } + getSharedCodexAppServerClientMock.mockResolvedValue(harness.client); + const waitForRequest = async (method: string) => + await vi.waitFor( + () => { + const request = harness.writes + .map((write) => JSON.parse(write) as { id: number; method: string; params: unknown }) + .find((message) => message.method === method); + if (!request) { + throw new Error(`Codex side harness did not write ${method}`); + } + return request; + }, + { interval: 1, timeout: 5_000 }, + ); + const run = runCodexAppServerSideQuestion( + sideParams({ opts: { abortSignal: controller.signal } }), + ); + const failure = run.then( + () => undefined, + (error: unknown) => error, + ); + const fork = await waitForRequest("thread/fork"); + harness.send({ id: fork.id, result: threadResult("side-thread") }); + const inject = await waitForRequest("thread/inject_items"); + harness.send({ id: inject.id, result: {} }); + + if (written) { + const turnStart = await waitForRequest("turn/start"); + controller.abort("side-start-cancelled"); + const interrupt = await waitForRequest("turn/interrupt"); + expect(interrupt.params).toEqual({ threadId: "side-thread", turnId: "" }); + harness.send({ id: turnStart.id, result: turnStartResult("turn-1") }); + harness.send( + alreadyTerminal + ? { + id: interrupt.id, + error: { code: -32_600, message: "no active turn to interrupt" }, + } + : interruptFails + ? { id: interrupt.id, error: { code: -32_000, message: "side interrupt failed" } } + : { id: interrupt.id, result: {} }, + ); + } else { + controller.abort("side-start-cancelled"); + } + + if (!interruptFails) { + const unsubscribe = await waitForRequest("thread/unsubscribe"); + harness.send( + unsubscribeFails + ? { id: unsubscribe.id, error: { code: -32_000, message: "side unsubscribe failed" } } + : { id: unsubscribe.id, result: {} }, + ); + } + await expect(failure).resolves.toMatchObject({ message: "turn/start aborted" }); + expect(harness.writes.map((write) => JSON.parse(write).method)).toEqual([ + "thread/fork", + "thread/inject_items", + ...(written ? ["turn/start", "turn/interrupt"] : []), + ...(!interruptFails ? ["thread/unsubscribe"] : []), + ]); + expect(harness.stdinDestroyed).toBe(interruptFails || unsubscribeFails === true); + harness.client.close(); + }, + ); + it("interrupts and unsubscribes the ephemeral thread on abort", async () => { const controller = new AbortController(); const client = createFakeClient(); diff --git a/extensions/codex/src/app-server/side-question.ts b/extensions/codex/src/app-server/side-question.ts index c713c0314816..5e9fd98c1e3e 100644 --- a/extensions/codex/src/app-server/side-question.ts +++ b/extensions/codex/src/app-server/side-question.ts @@ -20,13 +20,22 @@ import { import { loadExecApprovals } from "openclaw/plugin-sdk/exec-approvals-runtime"; import { resolveCodexAppServerForModelProvider } from "./app-server-policy.js"; import { handleCodexAppServerApprovalRequest } from "./approval-bridge.js"; +import { + isCodexAlreadyTerminalInterruptError, + retireUnsafeCodexTurnClientBestEffort, + unsubscribeCodexThreadBestEffort, +} from "./attempt-client-cleanup.js"; import { resolveCodexAppServerPreparedAuthHandoff } from "./auth-bridge.js"; import { requireCodexSupervisionModelSelection, resolveCodexBindingAppServerConnection, } from "./binding-connection.js"; import { ensureCodexAppServerClientRuntime } from "./client-runtime.js"; -import { isCodexAppServerApprovalRequest, type CodexAppServerClient } from "./client.js"; +import { + isCodexAppServerApprovalRequest, + isCodexAppServerIndeterminateRequestCancellationError, + type CodexAppServerClient, +} from "./client.js"; import { canUseCodexModelBackedApprovalsReviewerForModel, readCodexPluginConfig, @@ -659,31 +668,39 @@ export async function runCodexAppServerSideQuestion( readCodexSupportedReasoningEfforts(params.runtimeModel?.compat), ); const turnResponse = assertCodexTurnStartResponse( - await client.request( - "turn/start", - { - threadId: childThreadId, - input: [{ type: "text", text: params.question.trim(), text_elements: [] }], - cwd, - model: modelSelection.model, - ...(usesSupervisionConnection ? {} : { personality: CODEX_NATIVE_PERSONALITY_NONE }), - ...(serviceTier ? { serviceTier } : {}), - ...(usesSupervisionConnection - ? {} - : { - effort, - collaborationMode: { - mode: "default" as const, - settings: { - model: modelSelection.model, - reasoning_effort: effort, - developer_instructions: null, + await client + .request( + "turn/start", + { + threadId: childThreadId, + input: [{ type: "text", text: params.question.trim(), text_elements: [] }], + cwd, + model: modelSelection.model, + ...(usesSupervisionConnection ? {} : { personality: CODEX_NATIVE_PERSONALITY_NONE }), + ...(serviceTier ? { serviceTier } : {}), + ...(usesSupervisionConnection + ? {} + : { + effort, + collaborationMode: { + mode: "default" as const, + settings: { + model: modelSelection.model, + reasoning_effort: effort, + developer_instructions: null, + }, }, - }, - }), - }, - { timeoutMs: appServer.requestTimeoutMs, signal: params.opts?.abortSignal }, - ), + }), + }, + { timeoutMs: appServer.requestTimeoutMs, signal: params.opts?.abortSignal }, + ) + .catch((error: unknown) => { + if (isCodexAppServerIndeterminateRequestCancellationError(error)) { + // Codex serializes an empty-id startup interrupt after this written turn/start. + turnId = ""; + } + throw error; + }), ); turnId = turnResponse.turn.id; collector.setTurn(childThreadId, turnId); @@ -1119,7 +1136,7 @@ async function cleanupCodexSideThread( if (!params.threadId) { return; } - if (params.interrupt && params.turnId) { + if (params.interrupt && params.turnId !== undefined) { try { await client.request( "turn/interrupt", @@ -1127,17 +1144,21 @@ async function cleanupCodexSideThread( { timeoutMs: params.timeoutMs }, ); } catch (error) { - embeddedAgentLog.debug("codex /btw side thread interrupt cleanup failed", { error }); + if (!isCodexAlreadyTerminalInterruptError(error)) { + embeddedAgentLog.debug("codex /btw side thread interrupt cleanup failed", { error }); + await retireUnsafeCodexTurnClientBestEffort(client, "side turn interrupt"); + // An unconfirmed native turn must never lose its only visible subscription. + return; + } } } - try { - await client.request( - "thread/unsubscribe", - { threadId: params.threadId }, - { timeoutMs: params.timeoutMs }, - ); - } catch (error) { - embeddedAgentLog.debug("codex /btw side thread unsubscribe cleanup failed", { error }); + if ( + !(await unsubscribeCodexThreadBestEffort(client, { + threadId: params.threadId, + timeoutMs: params.timeoutMs, + })) + ) { + await retireUnsafeCodexTurnClientBestEffort(client, "side thread unsubscribe"); } } diff --git a/extensions/codex/src/app-server/thread-lifecycle.test.ts b/extensions/codex/src/app-server/thread-lifecycle.test.ts index 64e3d42bc389..11f89880e239 100644 --- a/extensions/codex/src/app-server/thread-lifecycle.test.ts +++ b/extensions/codex/src/app-server/thread-lifecycle.test.ts @@ -83,6 +83,7 @@ describe("Codex ring-zero thread config", () => { expect(start.environments).toEqual([]); expect(start.baseInstructions).toBe(""); for (const config of [start.config, resume.config]) { + expect(config?.["agents.enabled"]).toBe(false); expect(config?.["tools.experimental_request_user_input.enabled"]).toBe(false); expect(config?.["features.multi_agent"]).toBe(false); expect(config?.["features.multi_agent_v2"]).toBe(false); @@ -134,6 +135,7 @@ describe("Codex delegation capability", () => { }); for (const request of [start, resume]) { + expect(request.config?.["agents.enabled"]).toBe(false); expect(request.config?.["features.multi_agent"]).toBe(false); expect(request.config?.["features.multi_agent_v2"]).toBe(false); expect(request.config?.["features.goals"]).toBe(false); @@ -190,6 +192,7 @@ describe("Codex delegation capability", () => { for (const request of [start, resume]) { for (const disabledFeature of [ + "agents.enabled", "features.apps", "features.current_time_reminder", "features.deferred_executor", diff --git a/extensions/codex/src/app-server/thread-requests.ts b/extensions/codex/src/app-server/thread-requests.ts index bb54ca10dc9b..962fe5437a58 100644 --- a/extensions/codex/src/app-server/thread-requests.ts +++ b/extensions/codex/src/app-server/thread-requests.ts @@ -58,11 +58,13 @@ const CODEX_TOOL_SEARCH_UNSUPPORTED_THREAD_CONFIG: JsonObject = { }; const CODEX_DELEGATION_DISABLED_THREAD_CONFIG: JsonObject = { + "agents.enabled": false, "features.multi_agent": false, "features.multi_agent_v2": false, }; const CODEX_RING_ZERO_THREAD_CONFIG: JsonObject = { + ...CODEX_DELEGATION_DISABLED_THREAD_CONFIG, "features.apps": false, "features.current_time_reminder": false, "features.deferred_executor": false, @@ -71,8 +73,6 @@ const CODEX_RING_ZERO_THREAD_CONFIG: JsonObject = { "features.hooks": false, "features.image_generation": false, "features.memories": false, - "features.multi_agent": false, - "features.multi_agent_v2": false, "features.plugins": false, "features.standalone_web_search": false, "features.token_budget": false, diff --git a/extensions/codex/src/conversation-binding.test.ts b/extensions/codex/src/conversation-binding.test.ts index f1225f6d898e..cd447da9dce5 100644 --- a/extensions/codex/src/conversation-binding.test.ts +++ b/extensions/codex/src/conversation-binding.test.ts @@ -107,6 +107,7 @@ import { type CodexAppServerThreadBinding, writeCodexAppServerBinding, } from "./app-server/session-binding.test-helpers.js"; +import { createClientHarness } from "./app-server/test-support.js"; import { getCodexAppServerTurnRouter } from "./app-server/turn-router.js"; import { legacyCodexConversationBindingId } from "./conversation-binding-data.js"; import { codexConversationBindingRuntime } from "./conversation-binding.js"; @@ -2350,6 +2351,89 @@ describe("codex conversation binding", () => { }); }); + it.each([ + { label: "ordinary", sessionKey: undefined, interruptFails: false }, + { + label: "incognito", + sessionKey: "agent:main:dashboard:incognito-start-timeout", + interruptFails: false, + }, + { label: "ordinary with a failed interrupt", sessionKey: undefined, interruptFails: true }, + { + label: "incognito with a failed interrupt", + sessionKey: "agent:main:dashboard:incognito-start-interrupt-failure", + interruptFails: true, + }, + ])( + "interrupts an indeterminate $label native turn before cleanup", + async ({ sessionKey, interruptFails }) => { + const sessionFile = path.join(tempDir, "session.jsonl"); + const harness = createClientHarness(); + await writeTestConversationBinding(sessionFile, { + threadId: "thread-1", + clientId: harness.client.getInstanceId(), + cwd: tempDir, + }); + sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(harness.client); + if (interruptFails) { + sharedClientMocks.retireSharedCodexAppServerClientIfCurrent.mockReturnValueOnce({ + activeLeases: 2, + closed: false, + }); + } + const waitForRequest = async (method: string) => + await vi.waitFor( + () => { + const request = harness.writes + .map((write) => JSON.parse(write) as { id: number; method: string; params: unknown }) + .find((message) => message.method === method); + if (!request) { + throw new Error(`Codex conversation harness did not write ${method}`); + } + return request; + }, + { interval: 1, timeout: 5_000 }, + ); + const { event, ctx } = boundConversationClaim(sessionFile, sessionKey); + const result = handleCodexConversationInboundClaim(event, ctx, { + pluginConfig: { appServer: { requestTimeoutMs: 100 } }, + }); + const turnStart = await waitForRequest("turn/start"); + const interrupt = await waitForRequest("turn/interrupt"); + expect(interrupt.params).toEqual({ threadId: "thread-1", turnId: "" }); + harness.send({ id: turnStart.id, result: { turn: { id: "turn-1" } } }); + harness.send( + interruptFails + ? { id: interrupt.id, error: { code: -32_000, message: "startup interrupt failed" } } + : { id: interrupt.id, result: {} }, + ); + if (sessionKey && !interruptFails) { + const unsubscribe = await waitForRequest("thread/unsubscribe"); + harness.send({ id: unsubscribe.id, result: {} }); + } + + await expect(result).resolves.toEqual({ + handled: true, + reply: { text: "Codex app-server turn failed: turn/start timed out" }, + }); + expect(harness.writes.map((write) => JSON.parse(write).method)).toEqual([ + "turn/start", + "turn/interrupt", + ...(sessionKey && !interruptFails ? ["thread/unsubscribe"] : []), + ]); + expect(sharedClientMocks.retireSharedCodexAppServerClientIfCurrent).toHaveBeenCalledTimes( + interruptFails ? 1 : 0, + ); + expect( + readCodexConversationActiveTurn(testConversationIdentity(sessionFile)), + ).toBeUndefined(); + if (sessionKey) { + await expect(readTestConversationBinding(sessionFile)).resolves.toBeUndefined(); + } + harness.client.close(); + }, + ); + it.each([ { label: "ordinary", sessionKey: undefined }, { label: "incognito", sessionKey: "agent:main:dashboard:incognito-turn-timeout" }, diff --git a/extensions/codex/src/conversation-binding.ts b/extensions/codex/src/conversation-binding.ts index 1fd874e0db27..09e9b18db52e 100644 --- a/extensions/codex/src/conversation-binding.ts +++ b/extensions/codex/src/conversation-binding.ts @@ -1,6 +1,5 @@ // Codex plugin module implements conversation binding behavior. import { - embeddedAgentLog, formatErrorMessage, resolveSandboxContext, } from "openclaw/plugin-sdk/agent-harness-runtime"; @@ -19,11 +18,15 @@ import { CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, closeCodexStartupClientBestEffort, interruptCodexTurnAndWaitBestEffort, + retireUnsafeCodexTurnClientBestEffort, unsubscribeCodexThreadBestEffort, } from "./app-server/attempt-client-cleanup.js"; import { resolveCodexAppServerAuthProfileIdForAgent } from "./app-server/auth-bridge.js"; import { CODEX_CONTROL_METHODS } from "./app-server/capabilities.js"; -import type { CodexAppServerClient } from "./app-server/client.js"; +import { + isCodexAppServerIndeterminateRequestCancellationError, + type CodexAppServerClient, +} from "./app-server/client.js"; import { canUseCodexModelBackedApprovalsReviewerForModel, codexSandboxPolicyForTurn, @@ -883,19 +886,21 @@ async function runBoundTurn(params: { }, }; } catch (error) { - const timedOut = error instanceof CodexConversationTurnTimeoutError; - if (timedOut && activeTurnId) { - // Keep the exact turn, notification handlers, and lease alive until - // Codex confirms the terminal notification; otherwise inference survives. + if ( + (error instanceof CodexConversationTurnTimeoutError && activeTurnId) || + (turnRoute && isCodexAppServerIndeterminateRequestCancellationError(error)) + ) { + // Per-thread serialization makes an empty startup interrupt follow an + // accepted turn whose id was lost to local request cancellation. const completed = await interruptCodexTurnAndWaitBestEffort(client, { threadId, - turnId: activeTurnId, + turnId: activeTurnId ?? "", }); if (!completed) { // Retirement detaches the physical client while sibling leases finish; // never send another cleanup request or retire that detached client twice. retiredUnsafeClient = client; - await retireUnsafeCodexConversationClientBestEffort(client, "turn interrupt"); + await retireUnsafeCodexTurnClientBestEffort(client, "turn interrupt"); } } if (isIncognitoSessionKey(params.sessionKey)) { @@ -909,7 +914,7 @@ async function runBoundTurn(params: { timeoutMs: CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, }); if (!unsubscribed) { - await retireUnsafeCodexConversationClientBestEffort(client, "thread unsubscribe"); + await retireUnsafeCodexTurnClientBestEffort(client, "thread unsubscribe"); } } } @@ -921,21 +926,6 @@ async function runBoundTurn(params: { } } -async function retireUnsafeCodexConversationClientBestEffort( - client: CodexAppServerClient, - operation: "turn interrupt" | "thread unsubscribe", -): Promise { - try { - await closeCodexStartupClientBestEffort(client); - } catch (error) { - // Cleanup must not replace the original turn or timeout failure. - embeddedAgentLog.debug("codex conversation client retirement failed during cleanup", { - operation, - error, - }); - } -} - function assertNativeConversationApprovalPolicySupported(params: { execPolicy?: OpenClawExecPolicyForCodexAppServer; approvalPolicy: ReturnType["approvalPolicy"]; diff --git a/extensions/codex/src/web-search-provider.test.ts b/extensions/codex/src/web-search-provider.test.ts index d477beeb466c..4dcb1649f785 100644 --- a/extensions/codex/src/web-search-provider.test.ts +++ b/extensions/codex/src/web-search-provider.test.ts @@ -242,6 +242,9 @@ describe("codex web search provider", () => { "--listen", "stdio://", "-c", + "openai_base_url=http://127.0.0.1:44080/v1", + "--config=model_catalog_json=/tmp/qa catalog/models.json", + "-c", "mcp_servers.external.command='unsafe'", ], clearEnv: ["CODEX_HOME", "KEEP_CLEARED"], @@ -311,7 +314,15 @@ describe("codex web search provider", () => { const threadStartCwd = (requests[1]?.params as { cwd?: string } | undefined)?.cwd; const isolatedCodexHome = isolatedStartOptions?.env?.CODEX_HOME; expect(threadStartCwd).not.toBe("/tmp/openclaw-agent"); - expect(isolatedStartOptions?.args).toEqual(["app-server", "--listen", "stdio://"]); + expect(isolatedStartOptions?.args).toEqual([ + "app-server", + "-c", + "openai_base_url=http://127.0.0.1:44080/v1", + "-c", + "model_catalog_json=/tmp/qa catalog/models.json", + "--listen", + "stdio://", + ]); expect(isolatedStartOptions?.clearEnv).toEqual([ "KEEP_CLEARED", "OPENCLAW_CODEX_APP_SERVER_ARGS", diff --git a/extensions/ollama/src/cjk-char-estimate.ts b/extensions/ollama/src/cjk-char-estimate.ts deleted file mode 100644 index db04e2267ad2..000000000000 --- a/extensions/ollama/src/cjk-char-estimate.ts +++ /dev/null @@ -1,47 +0,0 @@ -/** - * CJK-aware character weighting for Ollama usage fallback estimates. - * - * This stays plugin-private because exposing it through the Plugin SDK would - * create a stable public contract for one provider-specific fallback. Keep - * the weighting aligned with normalization-core's CJK budget heuristic. - */ - -const CHARS_PER_TOKEN_ESTIMATE = 4; - -const NON_ASCII_RE = /[\u0080-\u{10FFFF}]/u; -const COMMON_CJK_RE = /[\u00B7\u3000-\u319F\u4E00-\u9FA5\uAC00-\uD7AF\uFF01-\uFF60]/gu; -const RARE_BMP_CJK_RE = - /[\u1100-\u11FF\u2E80-\u2FFF\u31A0-\u4DFF\u9FA6-\u9FFF\uA000-\uA4FF\uA700-\uA707\uA960-\uA97F\uD7B0-\uD7FF\uF900-\uFAFF]/gu; -const TWO_TOKEN_CJK_RE = - /[\u{02C7}\u{02C9}-\u{02CB}\u{02D9}\u{02EA}-\u{02EB}\uFE10-\uFE4F\uFF61-\uFFDC\uFFE0-\uFFE6]|\u{0305}|\u{0323}/gu; -const THREE_TOKEN_SUPPLEMENTARY_CJK_RE = /[\u{1D360}-\u{1D371}]/gu; -const SUPPLEMENTARY_CJK_RE = - /[\u{16FE0}-\u{16FFF}\u{1AFF0}-\u{1AFFF}\u{1B000}-\u{1B16F}\u{1F200}-\u{1F2FF}\u{20000}-\u{2FA1F}\u{30000}-\u{3347F}]/gu; -const SPECIAL_CJK_RE = - /[\u{02C7}\u{02C9}-\u{02CB}\u{02D9}\u{02EA}-\u{02EB}\u1100-\u11FF\u2E80-\u2FFF\u31A0-\u4DFF\u9FA6-\u9FFF\uA000-\uA4FF\uA700-\uA707\uA960-\uA97F\uD7B0-\uD7FF\uF900-\uFAFF\uFE10-\uFE4F\uFF61-\uFFDC\uFFE0-\uFFE6\u{16FE0}-\u{16FFF}\u{1AFF0}-\u{1AFFF}\u{1B000}-\u{1B16F}\u{1D360}-\u{1D371}\u{1F200}-\u{1F2FF}\u{20000}-\u{2FA1F}\u{30000}-\u{3347F}]|\u{0305}|\u{0323}/u; - -function countMatches(text: string, pattern: RegExp): number { - return (text.match(pattern) ?? []).length; -} - -export function estimateStringChars(text: string): number { - if (!NON_ASCII_RE.test(text)) { - return text.length; - } - const commonCjkCount = countMatches(text, COMMON_CJK_RE); - const commonEstimate = text.length + commonCjkCount * (CHARS_PER_TOKEN_ESTIMATE - 1); - if (!SPECIAL_CJK_RE.test(text)) { - return commonEstimate; - } - const rareBmpCjkCount = countMatches(text, RARE_BMP_CJK_RE); - const twoTokenCjkCount = countMatches(text, TWO_TOKEN_CJK_RE); - const threeTokenSupplementaryCjkCount = countMatches(text, THREE_TOKEN_SUPPLEMENTARY_CJK_RE); - const supplementaryCjkCount = countMatches(text, SUPPLEMENTARY_CJK_RE); - return ( - commonEstimate + - rareBmpCjkCount * (CHARS_PER_TOKEN_ESTIMATE * 3 - 1) + - twoTokenCjkCount * (CHARS_PER_TOKEN_ESTIMATE * 2 - 1) + - threeTokenSupplementaryCjkCount * (CHARS_PER_TOKEN_ESTIMATE * 3 - 2) + - supplementaryCjkCount * (CHARS_PER_TOKEN_ESTIMATE * 4 - 2) - ); -} diff --git a/extensions/ollama/src/stream.ts b/extensions/ollama/src/stream.ts index 508bc7edd9a3..22482d1a9554 100644 --- a/extensions/ollama/src/stream.ts +++ b/extensions/ollama/src/stream.ts @@ -32,8 +32,7 @@ import { import { createSubsystemLogger } from "openclaw/plugin-sdk/runtime-env"; import { fetchWithSsrFGuard, isLoopbackHost } from "openclaw/plugin-sdk/ssrf-runtime"; import { isRecord, readStringValue } from "openclaw/plugin-sdk/string-coerce-runtime"; -import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; -import { estimateStringChars } from "./cjk-char-estimate.js"; +import { estimateStringChars, truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import { OLLAMA_CLOUD_BASE_URL, OLLAMA_DEFAULT_BASE_URL } from "./defaults.js"; import { shouldWrapOllamaCompatMoonshotThinking } from "./model-behavior.js"; import { normalizeOllamaWireModelId } from "./model-id.js"; diff --git a/extensions/qa-lab/src/providers/mock-openai/server.test.ts b/extensions/qa-lab/src/providers/mock-openai/server.test.ts index 109dca06425a..68b4e86cb1cf 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.test.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.test.ts @@ -3624,6 +3624,86 @@ describe("qa mock openai server", () => { expect(await firstB.text()).toContain('\\"label\\":\\"qa-fanout-alpha\\"'); }); + it("isolates interleaved session state while preserving cross-provider ownership", async () => { + const server = await startMockServer(); + const handoffPrompt = + "Delegate one bounded QA task to a subagent. Wait for the subagent to finish."; + const fanoutPrompt = + "Subagent fanout synthesis check: delegate two bounded subagents sequentially, then report both results together."; + const sessions = ["qa-session-alpha", "qa-session-beta"] as const; + const runtimePrompt = (sessionId: string) => + `Runtime: agent=main | sessionId=${sessionId} | channel=qa`; + const postSession = (sessionId: string, input: unknown[], cacheBoundary = 0) => + expectNonStreamingResponsesJson(server, { + model: "gpt-5.6-luna", + prompt_cache_key: `${sessionId}:${cacheBoundary}`, + tools: [SESSIONS_SPAWN_TOOL], + input: [makeDeveloperInput(runtimePrompt(sessionId)), ...input], + }); + + const handoffs = await Promise.all( + sessions.map((sessionId) => postSession(sessionId, [makeUserInput(handoffPrompt)])), + ); + for (const handoff of handoffs) { + expect(outputToolArgsFromItem(outputToolCall(handoff, "sessions_spawn"))).toMatchObject({ + label: "qa-sidecar", + }); + } + + const crossProviderContinuation = await postJson(server, "/v1/messages", { + model: "claude-opus-4-8", + max_tokens: 256, + system: [{ type: "text", text: runtimePrompt(sessions[0]) }], + tools: [{ name: "sessions_spawn", input_schema: { type: "object", properties: {} } }], + messages: [makeAnthropicUserText(handoffPrompt)], + }); + expect(crossProviderContinuation.status).toBe(200); + const continued = requireRecord( + await crossProviderContinuation.json(), + "cross-provider handoff continuation", + ); + expect(continued.stop_reason).toBe("end_turn"); + + const anthropicHandoff = await postJson(server, "/v1/messages", { + model: "claude-opus-4-8", + max_tokens: 256, + system: [{ type: "text", text: runtimePrompt("qa-session-anthropic") }], + tools: [{ name: "sessions_spawn", input_schema: { type: "object", properties: {} } }], + messages: [makeAnthropicUserText(handoffPrompt)], + }); + expect(anthropicHandoff.status).toBe(200); + expect( + requireRecord(await anthropicHandoff.json(), "independent Anthropic handoff"), + ).toMatchObject({ stop_reason: "tool_use" }); + + const firstFanoutCalls = await Promise.all( + sessions.map((sessionId) => postSession(sessionId, [makeUserInput(fanoutPrompt)], 1)), + ); + for (const fanout of firstFanoutCalls) { + expect(outputToolArgsFromItem(outputToolCall(fanout, "sessions_spawn"))).toMatchObject({ + label: "qa-fanout-alpha", + }); + } + + const secondFanoutCalls = await Promise.all( + sessions.map((sessionId) => + postSession( + sessionId, + [ + makeUserInput(fanoutPrompt), + makeToolOutput('{"status":"accepted","childSessionKey":"alpha","note":"ALPHA-OK"}'), + ], + 2, + ), + ), + ); + for (const fanout of secondFanoutCalls) { + expect(outputToolArgsFromItem(outputToolCall(fanout, "sessions_spawn"))).toMatchObject({ + label: "qa-fanout-beta", + }); + } + }); + it("answers heartbeat prompts without spawning extra subagents", async () => { const server = await startMockServer(); diff --git a/extensions/qa-lab/src/providers/mock-openai/server.ts b/extensions/qa-lab/src/providers/mock-openai/server.ts index da2b227800ed..d8b851745dd0 100644 --- a/extensions/qa-lab/src/providers/mock-openai/server.ts +++ b/extensions/qa-lab/src/providers/mock-openai/server.ts @@ -9,6 +9,7 @@ import { listMockOpenAiServerModelIds, } from "../shared/mock-model-config.js"; import { buildMessagesPayload } from "./mock-anthropic-messages.js"; +import { convertAnthropicMessagesToResponsesInput } from "./mock-anthropic-wire.js"; import { buildAssistantText, isCanonicalCompactionRetryWriteResult, @@ -1807,12 +1808,32 @@ export async function startQaMockOpenAiServer(params?: { }) { const host = params?.host ?? "127.0.0.1"; const finalOnlyMarkerPauseMs = params?.finalOnlyMarkerPauseMs ?? 1_500; - const scenarioState: MockScenarioState = { - anthropicThinkingErrorScenarioKeys: new Set(), - subagentFanoutCompletedWorkers: new Set<"alpha" | "beta">(), - subagentFanoutPhase: 0, - subagentHandoffSpawned: false, - toolLoopReadAttempts: 0, + const scenarioStates = new Map(); + const scenarioStateFor = (body: Record): MockScenarioState => { + const input = Array.isArray(body.input) + ? body.input + : convertAnthropicMessagesToResponsesInput({ + system: body.system as AnthropicMessagesRequest["system"], + messages: [], + }); + const systemPrompt = extractAllRequestTexts( + input.filter((item) => item.role === "developer" || item.role === "system"), + body, + ); + const sessionId = + /\bRuntime:\s*[^\n]*\bsessionId=([^\s|]+)/u.exec(systemPrompt)?.[1] ?? + (body.client_metadata as { session_id?: unknown } | undefined)?.session_id; + const key = typeof sessionId === "string" ? sessionId : ""; + // Runtime session identity survives provider switches and cache-boundary changes. + const state = scenarioStates.get(key) ?? { + anthropicThinkingErrorScenarioKeys: new Set(), + subagentFanoutCompletedWorkers: new Set<"alpha" | "beta">(), + subagentFanoutPhase: 0, + subagentHandoffSpawned: false, + toolLoopReadAttempts: 0, + }; + scenarioStates.set(key, state); + return state; }; let lastRequest: MockOpenAiRequestSnapshot | null = null; const requests: MockOpenAiRequestSnapshot[] = []; @@ -1845,7 +1866,7 @@ export async function startQaMockOpenAiServer(params?: { inflightRequests.set(inflightRequestId, { prompt, allInputText }); let events: StreamEvent[]; try { - events = await buildResponsesPayload(request.body, scenarioState); + events = await buildResponsesPayload(request.body, scenarioStateFor(request.body)); } finally { inflightRequests.delete(inflightRequestId); } @@ -2067,6 +2088,7 @@ export async function startQaMockOpenAiServer(params?: { }); return; } + const scenarioState = scenarioStateFor(body as Record); const { events, input, diff --git a/packages/ai/src/transports/openai-responses-stream-internal.ts b/packages/ai/src/transports/openai-responses-stream-internal.ts index 14754843465a..e18118dfeb11 100644 --- a/packages/ai/src/transports/openai-responses-stream-internal.ts +++ b/packages/ai/src/transports/openai-responses-stream-internal.ts @@ -39,7 +39,7 @@ import { } from "./openai-responses-stream-slots-internal.js"; import { createResponsesTerminalController, - resolveCompletedToolCallName, + resolveCompletedResponsesToolCall, resolveResponsesToolCallId, type ResponsesEventSink, type ResponsesThinkingBlock, @@ -618,7 +618,6 @@ export async function processResponsesStream( if (!streamingToolCall && streamingToolCalls.hasActive()) { continue; } - const completedName = resolveCompletedToolCallName(streamingToolCall, item.name); const streamedArguments = streamingToolCall?.block.partialJson ?? ""; const completedArguments = typeof item.arguments === "string" ? item.arguments : undefined; @@ -633,8 +632,11 @@ export async function processResponsesStream( completedArguments !== undefined && (completedArguments.length > 0 || !streamedArguments) ? completedArguments - : streamedArguments || "{}"; - const args = parseStreamingJson(finalArguments); + : streamedArguments; + const validated = resolveCompletedResponsesToolCall(item, { + name: streamingToolCall?.block.name, + arguments: finalArguments, + }); let toolCall: ToolCall; let contentIndex: number; @@ -644,10 +646,10 @@ export async function processResponsesStream( // the canonical id on completion. Upgrade the same public block so // replay and its function_call_output retain both identities. block.id = resolveResponsesToolCallId(item, block.id); - block.name = completedName; + block.name = validated.name; // Finalize in-place and strip the scratch buffer so replay only // carries parsed arguments. - block.arguments = args; + block.arguments = validated.arguments; delete (block as { partialJson?: string }).partialJson; toolCall = block; contentIndex = streamingToolCall.contentIndex; @@ -655,8 +657,8 @@ export async function processResponsesStream( toolCall = { type: "toolCall", id: resolveResponsesToolCallId(item), - name: completedName, - arguments: args, + name: validated.name, + arguments: validated.arguments, }; // Some compatible streams only send the completed item. Preserve // the normal balanced lifecycle and persist the call for replay. diff --git a/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts b/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts index 0295cc3eee9d..18308a1e3ade 100644 --- a/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts +++ b/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts @@ -80,12 +80,15 @@ export function resolveResponsesToolCallId( return resolvedItemId ? `${generated}|${resolvedItemId}` : generated; } -export function resolveCompletedToolCallName( - toolCall: { block: { name: string } } | undefined, - value: unknown, -): string { - const streamedName = toolCall?.block.name.trim() || undefined; - const completedName = typeof value === "string" ? value.trim() || undefined : undefined; +export function resolveCompletedResponsesToolCall( + item: Extract, + streamed?: { name?: string; arguments?: string }, +): Pick { + if (item.status && item.status !== "completed") { + throw new Error("Responses stream completed with an incomplete terminal tool call"); + } + const streamedName = streamed?.name?.trim() || undefined; + const completedName = typeof item.name === "string" ? item.name.trim() || undefined : undefined; if (streamedName && completedName && streamedName !== completedName) { throw new Error( `Responses stream changed tool-call function name from ${streamedName} to ${completedName}`, @@ -95,7 +98,13 @@ export function resolveCompletedToolCallName( if (!name) { throw new Error("Responses stream completed tool call without a function name"); } - return name; + const argumentsValue = parseJsonObjectPreservingUnsafeIntegers( + streamed?.arguments ?? item.arguments, + ); + if (!argumentsValue) { + throw new Error("Responses stream completed tool call with invalid JSON arguments"); + } + return { name, arguments: argumentsValue }; } export function createResponsesTerminalController(params: { @@ -195,21 +204,8 @@ export function createResponsesTerminalController(params: { stream.push({ type: "text_start", contentIndex: index, partial: output as never }); stream.push({ type: "text_end", contentIndex: index, content: text, partial: output as never }); }; - const validateCompletedToolCall = ( - item: Extract, - ) => { - if (item.status && item.status !== "completed") { - throw new Error("Responses stream completed with an incomplete terminal tool call"); - } - const name = resolveCompletedToolCallName(undefined, item.name); - const argumentsValue = parseJsonObjectPreservingUnsafeIntegers(item.arguments); - if (!argumentsValue) { - throw new Error("Responses stream completed tool call with invalid JSON arguments"); - } - return { name, arguments: argumentsValue }; - }; const appendToolCall = (item: Extract) => { - const validated = validateCompletedToolCall(item); + const validated = resolveCompletedResponsesToolCall(item); const toolCall: ToolCall = { type: "toolCall", id: resolveResponsesToolCallId(item), @@ -249,7 +245,7 @@ export function createResponsesTerminalController(params: { throw new Error("Responses stream omitted an output item before completed output"); } if (item.type === "function_call") { - validateCompletedToolCall(item); + resolveCompletedResponsesToolCall(item); } } for (const item of items) { diff --git a/packages/ai/src/transports/openai-responses-stream-terminal-recovery.test.ts b/packages/ai/src/transports/openai-responses-stream-terminal-recovery.test.ts index fc7e15d44abb..bf8fec77ab26 100644 --- a/packages/ai/src/transports/openai-responses-stream-terminal-recovery.test.ts +++ b/packages/ai/src/transports/openai-responses-stream-terminal-recovery.test.ts @@ -85,6 +85,127 @@ const rejectedTerminalToolBatchFixture = (params: { }, }); +const rejectedStreamedToolFixture = (params: { + name: string; + status: "completed" | "incomplete"; + arguments?: string; + error: string; + started?: boolean; +}): ParityFixture => { + const item = { + id: "fc_rejected_streamed", + call_id: "call_rejected_streamed", + type: "function_call", + name: "lookup", + ...(params.arguments === undefined ? {} : { arguments: params.arguments }), + status: params.status, + }; + const started = params.started + ? [ + { + type: "response.output_item.added", + output_index: 0, + item: { ...item, arguments: "", status: "in_progress" }, + }, + { + type: "response.function_call_arguments.delta", + output_index: 0, + item_id: item.id, + delta: "{", + }, + ] + : []; + return { + name: params.name, + events: [ + ...started, + { type: "response.output_item.done", output_index: 0, item }, + completed("resp_rejected_streamed_tool", [item]), + ], + canonical: { + events: params.started + ? [ + { type: "toolcall_start", contentIndex: 0 }, + { type: "toolcall_delta", contentIndex: 0, delta: "{" }, + ] + : [], + content: params.started + ? [ + { + type: "toolCall", + id: "call_rejected_streamed|fc_rejected_streamed", + name: "lookup", + arguments: {}, + partialJson: false, + }, + ] + : [], + responseId: null, + stopReason: "stop", + error: params.error, + }, + }; +}; + +const unsafeIntegerToolFixture = (source: "streamed" | "terminal"): ParityFixture => { + const item = { + id: "fc_unsafe_integer", + call_id: "call_unsafe_integer", + type: "function_call", + name: "send_message", + arguments: + '{"to":1481220477346119781,"safe":42,"maxSafe":9007199254740991,"nested":{"ids":[9007199254740993,-9007199254740992]}}', + status: "completed", + }; + const streamed = + source === "streamed" + ? [ + { + type: "response.output_item.added", + output_index: 0, + item: { ...item, arguments: "", status: "in_progress" }, + }, + { + type: "response.function_call_arguments.delta", + output_index: 0, + item_id: item.id, + delta: '{"to":', + }, + { type: "response.output_item.done", output_index: 0, item }, + ] + : []; + return { + name: `${source} completed tool arguments preserve unsafe integers`, + events: [...streamed, completed("resp_unsafe_integer_tool", [item])], + canonical: { + events: [ + { type: "toolcall_start", contentIndex: 0 }, + ...(source === "streamed" + ? [{ type: "toolcall_delta", contentIndex: 0, delta: '{"to":' }] + : []), + { type: "toolcall_end", contentIndex: 0 }, + ], + content: [ + { + type: "toolCall", + id: "call_unsafe_integer|fc_unsafe_integer", + name: "send_message", + arguments: { + to: "1481220477346119781", + safe: 42, + maxSafe: 9007199254740991, + nested: { ids: ["9007199254740993", "-9007199254740992"] }, + }, + partialJson: false, + }, + ], + responseId: "resp_unsafe_integer_tool", + stopReason: "toolUse", + error: null, + }, + }; +}; + const rejectedOutOfOrderTerminalFixture = ( state: "completed" | "started" | "reasoning", ): ParityFixture => { @@ -342,6 +463,32 @@ const fixtures: ParityFixture[] = [ arguments: '{"q":', error: "Responses stream completed tool call with invalid JSON arguments", }), + rejectedStreamedToolFixture({ + name: "streamed malformed tool arguments never complete their started call", + status: "completed", + arguments: '{"q":', + error: "Responses stream completed tool call with invalid JSON arguments", + started: true, + }), + rejectedStreamedToolFixture({ + name: "streamed non-object tool arguments never start a call", + status: "completed", + arguments: "[]", + error: "Responses stream completed tool call with invalid JSON arguments", + }), + rejectedStreamedToolFixture({ + name: "streamed missing tool arguments never fabricate an empty call", + status: "completed", + error: "Responses stream completed tool call with invalid JSON arguments", + }), + rejectedStreamedToolFixture({ + name: "streamed incomplete tool calls never start a call", + status: "incomplete", + arguments: '{"q":"x"}', + error: "Responses stream completed with an incomplete terminal tool call", + }), + unsafeIntegerToolFixture("streamed"), + unsafeIntegerToolFixture("terminal"), rejectedTerminalToolBatchFixture({ name: "terminal tool batch rejects a later incomplete call before executing earlier calls", status: "incomplete", diff --git a/qa/scenarios/goals/goal-context-survives-compaction.yaml b/qa/scenarios/goals/goal-context-survives-compaction.yaml index 4eff9f7bba52..c53721f7447d 100644 --- a/qa/scenarios/goals/goal-context-survives-compaction.yaml +++ b/qa/scenarios/goals/goal-context-survives-compaction.yaml @@ -13,6 +13,7 @@ scenario: objective: Verify compaction does not remove active session-goal context from the next model turn. successCriteria: - An authorized channel turn starts a session goal through `/goal start`. + - A separate conversation completes a turn before the original session is compacted. - The same session completes a manual `/compact` operation. - The first post-compaction inbound turn reaches the mock model with the exact active-goal context line. - The assertion uses the captured post-compaction provider request. @@ -39,6 +40,7 @@ scenario: goalObjective: Preserve the deployment audit objective expectedGoalContext: "Active goal: Preserve the deployment audit objective — advance; keep active until fully achieved; block only after the same blocker on 3 consecutive turns; after update_goal, provide the requested visible final." contextSeedMarker: GOAL-COMPACTION-CONTEXT-SEED + otherConversationMarker: GOAL-COMPACTION-SECOND-SESSION-READY postCompactionMessage: GOAL-CONTEXT-POST-COMPACTION unrelated lunch question flow: @@ -104,6 +106,46 @@ flow: ref: seedOutboundIndex timeoutMs: expr: liveTurnTimeoutMs(env, 60000) + - set: otherConversationId + value: + expr: "`${conversationId}-other`" + - set: otherDelivery + value: + expr: "transport.buildAgentDelivery({ target: `group:${otherConversationId}` })" + - set: otherSessionKey + value: + expr: "buildAgentSessionKey({ agentId: 'qa', channel: otherDelivery.channel, accountId: transport.accountId, peer: { kind: 'group', id: otherDelivery.replyTo }, dmScope: env.cfg.session?.dmScope, identityLinks: env.cfg.session?.identityLinks })" + - set: otherOutboundIndex + value: + expr: "state.getSnapshot().messages.filter((message) => message.direction === 'outbound').length" + - sendInbound: + conversation: + id: + ref: otherConversationId + kind: group + senderId: + ref: config.senderId + senderName: QA Goal Operator + text: + expr: "`@openclaw reply exactly: ${config.otherConversationMarker}`" + - waitForOutbound: + conversation: + id: + ref: otherConversationId + kind: group + sinceIndex: + ref: otherOutboundIndex + textIncludes: + ref: config.otherConversationMarker + timeoutMs: + expr: liveTurnTimeoutMs(env, 60000) + - call: waitForCondition + args: + - lambda: + async: true + expr: "env.gateway.call('sessions.list', {}, { timeoutMs: 30000 }).then((result) => result.sessions?.find((session) => session.key === otherSessionKey && session.hasActiveRun === false))" + - expr: liveTurnTimeoutMs(env, 60000) + - 100 - set: compactOutboundIndex value: expr: "state.getSnapshot().messages.filter((message) => message.direction === 'outbound').length" diff --git a/src/agents/agent-tools.before-tool-call.e2e.test.ts b/src/agents/agent-tools.before-tool-call.e2e.test.ts index 0cb0baac5559..ec18985e2a12 100644 --- a/src/agents/agent-tools.before-tool-call.e2e.test.ts +++ b/src/agents/agent-tools.before-tool-call.e2e.test.ts @@ -279,13 +279,16 @@ describe("before_tool_call loop detection behavior", () => { } } - function createGenericReadRepeatFixture() { + function createGenericReadRepeatFixture( + loopDetectionContext?: Parameters[2], + ) { const execute = vi.fn().mockResolvedValue({ content: [{ type: "text", text: "same output" }], details: { ok: true }, }); return { - tool: createWrappedTool("read", execute), + tool: createWrappedTool("read", execute, loopDetectionContext), + execute, params: { path: "/tmp/file" }, }; } @@ -961,12 +964,29 @@ describe("before_tool_call loop detection behavior", () => { it("escalates repeated critical vetoes to the global circuit breaker", async () => { await withToolLoopEvents(async (emitted) => { - const { tool, params } = createGenericReadRepeatFixture(); + const runId = "codex-native-global-breaker"; + const { tool, params, execute } = createGenericReadRepeatFixture({ + ...enabledLoopDetectionContext, + runId, + }); - for (let i = 0; i <= 30; i += 1) { - await tool.execute(`read-global-${i}`, params, undefined, undefined); + for (let i = 0; i <= GLOBAL_CIRCUIT_BREAKER_THRESHOLD; i += 1) { + const toolCallId = `read-global-${i}`; + const nativeOutcome = await runBeforeToolCallHook({ + toolName: "read", + params, + toolCallId, + ctx: { + agentId: enabledLoopDetectionContext.agentId, + sessionKey: enabledLoopDetectionContext.sessionKey, + runId, + }, + }); + expect(nativeOutcome.blocked).toBe(false); + await tool.execute(toolCallId, params, undefined, undefined); } + expect(execute).toHaveBeenCalledTimes(CRITICAL_THRESHOLD); expect(emitted.at(-1)).toMatchObject({ type: "tool.loop", level: "critical", @@ -1188,6 +1208,7 @@ describe("before_tool_call loop detection behavior", () => { agentId: "main", sessionKey: "session-key", runId: "run-1", + loopDetection: { enabled: true }, }, ); diff --git a/src/agents/agent-tools.before-tool-call.policy.ts b/src/agents/agent-tools.before-tool-call.policy.ts index f9bc9fd89dc5..238c4db1578a 100644 --- a/src/agents/agent-tools.before-tool-call.policy.ts +++ b/src/agents/agent-tools.before-tool-call.policy.ts @@ -184,7 +184,7 @@ export async function runBeforeToolCallHook(args: { } } - if (args.ctx.loopDetection?.enabled !== false) { + if (args.ctx.loopDetection?.enabled === true) { recordToolCall( sessionState, toolName, diff --git a/src/gateway/config-reload-plan.ts b/src/gateway/config-reload-plan.ts index 04a2eed0c464..9467e542475c 100644 --- a/src/gateway/config-reload-plan.ts +++ b/src/gateway/config-reload-plan.ts @@ -95,7 +95,7 @@ const BASE_RELOAD_RULES: ReloadRule[] = [ kind: "hot", actions: ["restart-heartbeat"], }, - { prefix: "agents.defaults.compaction", kind: "hot" }, + { prefix: "agents.defaults", kind: "hot" }, { prefix: "agents.defaults.models", kind: "hot", @@ -137,7 +137,7 @@ const BASE_RELOAD_RULES_TAIL: ReloadRule[] = [ { prefix: "wizard", kind: "none" }, { prefix: "logging", kind: "none" }, { prefix: "agents", kind: "none" }, - { prefix: "tools", kind: "none" }, + { prefix: "tools", kind: "hot" }, { prefix: "bindings", kind: "none" }, { prefix: "audio", kind: "none" }, { prefix: "agent", kind: "none" }, diff --git a/src/gateway/config-reload.test.ts b/src/gateway/config-reload.test.ts index 092cd47da315..7c15c5e83896 100644 --- a/src/gateway/config-reload.test.ts +++ b/src/gateway/config-reload.test.ts @@ -334,6 +334,10 @@ describe("buildGatewayReloadPlan", () => { path: "agents.defaults.models", expected: { restartHeartbeat: true }, }, + { + path: "agents.defaults.heartbeat.every", + expected: { restartHeartbeat: true }, + }, { path: "agents.defaults.modelPolicy.allow", expected: { restartHeartbeat: true }, @@ -359,11 +363,22 @@ describe("buildGatewayReloadPlan", () => { }); it.each([ + "agents.defaults", "agents.defaults.compaction", "agents.defaults.compaction.model", "agents.defaults.compaction.maxActiveTranscriptBytes", "agents.defaults.compaction.memoryFlush.model", - ])("refreshes compaction config without restarting subsystems: %s", (path) => { + "agents.defaults.contextTokens", + "agents.defaults.contextPruning.mode", + "agents.defaults.contextLimits.postCompactionMaxChars", + "agents.defaults.timeoutSeconds", + "agents.defaults.userTimezone", + "tools", + "tools.deny", + "tools.allow", + "tools.profile", + "tools.byProvider.openai.deny", + ])("refreshes prepared model runtime policy without restarting subsystems: %s", (path) => { const plan = buildGatewayReloadPlan([path]); expect(plan).toMatchObject({ @@ -1741,7 +1756,7 @@ describe("startGatewayConfigReloader", () => { } satisfies OpenClawConfig; const terminalPolicy = createTerminalLaunchPolicy(initialConfig); const events: string[] = []; - const onNoopConfigCommit = async ( + const onHotReload = async ( plan: GatewayReloadPlan, nextConfig: OpenClawConfig, ownership: GatewayConfigReloadTransactionOwnership, @@ -1763,7 +1778,7 @@ describe("startGatewayConfigReloader", () => { const harness = createReloaderHarness(vi.fn(), { initialConfig, initialCompareConfig: initialConfig, - onNoopConfigCommit, + onHotReload, onConfigApplied: () => { events.push("applied"); terminalPolicy.commitConfig(); @@ -1811,7 +1826,7 @@ describe("startGatewayConfigReloader", () => { agents: { defaults: { sandbox: { mode: "all" as const } } }, } satisfies OpenClawConfig; const terminalPolicy = createTerminalLaunchPolicy(initialConfig); - const onNoopConfigCommit = async ( + const onHotReload = async ( plan: GatewayReloadPlan, nextConfig: OpenClawConfig, ownership: GatewayConfigReloadTransactionOwnership, @@ -1825,7 +1840,7 @@ describe("startGatewayConfigReloader", () => { { initialConfig, initialCompareConfig: initialConfig, - onNoopConfigCommit, + onHotReload, onConfigApplied: () => terminalPolicy.commitConfig(), }, ); @@ -1867,7 +1882,7 @@ describe("startGatewayConfigReloader", () => { } satisfies OpenClawConfig; const terminalPolicy = createTerminalLaunchPolicy(initialConfig); const events: string[] = []; - const onNoopConfigCommit = async ( + const onHotReload = async ( plan: GatewayReloadPlan, nextConfig: OpenClawConfig, ownership: GatewayConfigReloadTransactionOwnership, @@ -1888,7 +1903,7 @@ describe("startGatewayConfigReloader", () => { const harness = createReloaderHarness(vi.fn(), { initialConfig, initialCompareConfig: initialConfig, - onNoopConfigCommit, + onHotReload, onConfigApplied: () => { events.push("applied"); terminalPolicy.commitConfig(); @@ -2460,7 +2475,7 @@ describe("startGatewayConfigReloader", () => { expect(stopResolved).toBe(true); }); - it("notifies lifecycle owners for no-op sandbox policy changes", async () => { + it("hot-reloads sandbox policy for prepared model lifecycle owners", async () => { const initialConfig: OpenClawConfig = { gateway: { reload: {} }, agents: { defaults: { sandbox: { mode: "off" } } }, @@ -2475,12 +2490,12 @@ describe("startGatewayConfigReloader", () => { await flushWatcherChange(harness); expect(harness.onConfigChange).toHaveBeenCalledTimes(1); - expect(harness.onConfigChange.mock.calls[0]?.[0].noopPaths).toContain( + expect(harness.onConfigChange.mock.calls[0]?.[0].hotReasons).toContain( "agents.defaults.sandbox.mode", ); expect(harness.onConfigChange.mock.calls[0]?.[1]).toBe(nextConfig); expect(harness.onConfigApplied).toHaveBeenCalledTimes(1); - expect(harness.onHotReload).not.toHaveBeenCalled(); + expect(harness.onHotReload).toHaveBeenCalledTimes(1); expect(harness.onRestart).not.toHaveBeenCalled(); await harness.reloader.stop(); }); diff --git a/src/gateway/gateway.compaction-hot-reload.e2e.test.ts b/src/gateway/gateway.compaction-hot-reload.e2e.test.ts index 8c68d12b38d9..4abcaa543589 100644 --- a/src/gateway/gateway.compaction-hot-reload.e2e.test.ts +++ b/src/gateway/gateway.compaction-hot-reload.e2e.test.ts @@ -61,7 +61,7 @@ describe("gateway compaction hot reload", () => { afterEach(resetGatewayState); it( - "applies a hot-reloaded compaction model to automatic chat preflight without restarting", + "applies hot-reloaded tool policy, context budgets, and compaction models without restarting", { timeout: 90_000 }, async () => { const envSnapshot = captureEnv([...ISOLATED_GATEWAY_ENV_KEYS]); @@ -96,6 +96,7 @@ describe("gateway compaction hot reload", () => { deleteTestEnvValue("OPENCLAW_TEST_MINIMAL_GATEWAY"); const providerRequests: string[] = []; + const providerRequestToolNames: string[][] = []; const summaryByModel = new Map(); const providerServer = createServer((request, response) => { void (async () => { @@ -107,9 +108,15 @@ describe("gateway compaction hot reload", () => { response.writeHead(404).end(); return; } - const body = JSON.parse(Buffer.concat(chunks).toString("utf8")) as { model?: string }; + const body = JSON.parse(Buffer.concat(chunks).toString("utf8")) as { + model?: string; + tools?: Array<{ name?: string; function?: { name?: string } }>; + }; const model = body.model ?? ""; providerRequests.push(model); + providerRequestToolNames.push( + body.tools?.map((tool) => tool.name ?? tool.function?.name ?? "") ?? [], + ); const text = summaryByModel.get(model) ?? "Primary assistant answer."; const message = { type: "message", @@ -269,14 +276,39 @@ describe("gateway compaction hot reload", () => { await sendChatAndWait("Warm the existing prepared model runtime."); expect(providerRequests).toContain(primaryModel.modelId); + expect(providerRequestToolNames.at(-1)).toContain("exec"); expect(loadSessionEntry(scope)?.compactionCount ?? 0).toBe(0); + const reloadedTools = { deny: ["exec"] }; + await writeConfigFile({ ...initialConfig, tools: reloadedTools }); + await expect + .poll(() => getRuntimeConfig().tools?.deny, { timeout: 5_000, interval: 50 }) + .toEqual(["exec"]); + await sendChatAndWait("Apply the hot-reloaded tool deny policy to the existing runtime."); + expect(providerRequestToolNames.at(-1)).not.toContain("exec"); + + const contextTokens = 48_000; + const reloadedAgentDefaults = { ...initialConfig.agents.defaults, contextTokens }; + await writeConfigFile({ + ...initialConfig, + agents: { ...initialConfig.agents, defaults: reloadedAgentDefaults }, + tools: reloadedTools, + }); + await expect + .poll(() => getRuntimeConfig().agents?.defaults?.contextTokens, { + timeout: 5_000, + interval: 50, + }) + .toBe(contextTokens); + await sendChatAndWait("Apply the hot-reloaded context budget to the existing runtime."); + expect(loadSessionEntry(scope)?.contextTokens).toBe(contextTokens); + await writeConfigFile({ ...initialConfig, agents: { ...initialConfig.agents, defaults: { - ...initialConfig.agents.defaults, + ...reloadedAgentDefaults, compaction: { ...initialConfig.agents.defaults.compaction, model: newCompactionModel.modelRef, @@ -284,6 +316,7 @@ describe("gateway compaction hot reload", () => { }, }, }, + tools: reloadedTools, }); await expect .poll(() => getRuntimeConfig().agents?.defaults?.compaction, { diff --git a/src/gateway/server-methods/sessions-compact.ts b/src/gateway/server-methods/sessions-compact.ts index 7b84aaf56012..587d726532cf 100644 --- a/src/gateway/server-methods/sessions-compact.ts +++ b/src/gateway/server-methods/sessions-compact.ts @@ -6,7 +6,7 @@ import { validateSessionsCompactParams, } from "../../../packages/gateway-protocol/src/index.js"; import { resolveDefaultAgentId } from "../../agents/agent-scope.js"; -import { clearSessionQueues } from "../../auto-reply/reply/queue/cleanup.js"; +import { resolveEmbeddedSessionLane } from "../../agents/embedded-agent-runner/lanes.js"; import { hasPendingFollowupQueueWork } from "../../auto-reply/reply/queue/state.js"; import { resolveSessionWorkStartError, @@ -19,6 +19,7 @@ import { trimSessionTranscriptForManualCompact, } from "../../config/sessions/session-accessor.js"; import { formatErrorMessage } from "../../infra/errors.js"; +import { getCommandLaneSnapshot } from "../../process/command-queue.js"; import { isCompetingSessionWorkAdmissionActive, runExclusiveSessionLifecycleMutation, @@ -160,17 +161,12 @@ export const sessionCompactHandlers: GatewayRequestHandlers = { } const lifecycleRevision = entry.lifecycleRevision; - const lifecycleIdentities = [ - key, - target.canonicalKey, - compactTarget.primaryKey, - sessionId, - lifecycleRevision, - ]; + const queueIdentities = [key, target.canonicalKey, compactTarget.primaryKey, sessionId]; + const lifecycleIdentities = [...queueIdentities, lifecycleRevision]; let sessionStillCurrent = true; let compactionNoopReason: string | undefined; let blockedByActiveRun = false; - let blockedByQueuedFollowup = false; + let blockedByQueuedWork = false; try { await runExclusiveSessionLifecycleMutation({ scope: storePath, @@ -221,18 +217,14 @@ export const sessionCompactHandlers: GatewayRequestHandlers = { agentId: requestedAgentId, defaultAgentId: resolveDefaultAgentId(cfg), }); - blockedByQueuedFollowup = hasPendingFollowupQueueWork([ - key, - target.canonicalKey, - compactTarget.primaryKey, - sessionId, - ]); - if (blockedByActiveRun || blockedByQueuedFollowup) { - return; - } - // Queued follow-ups are accepted user intent, not stale transcript work. - // Refuse compaction until the queue drains so cleanup cannot erase them. - clearSessionQueues([key, target.canonicalKey, compactTarget.primaryKey, sessionId]); + // Accepted work can live only in its command lane; waiting behind it + // while holding the lifecycle fence would deadlock or drop that turn. + blockedByQueuedWork = + hasPendingFollowupQueueWork(queueIdentities) || + queueIdentities.some( + (identity) => + getCommandLaneSnapshot(resolveEmbeddedSessionLane(identity)).queuedCount > 0, + ); }, run: async () => { if (!sessionStillCurrent) { @@ -260,7 +252,7 @@ export const sessionCompactHandlers: GatewayRequestHandlers = { ); return; } - if (blockedByQueuedFollowup) { + if (blockedByQueuedWork) { respond( false, undefined, diff --git a/src/gateway/server.sessions.compaction.test.ts b/src/gateway/server.sessions.compaction.test.ts index 71571e1218fe..5e826cdde829 100644 --- a/src/gateway/server.sessions.compaction.test.ts +++ b/src/gateway/server.sessions.compaction.test.ts @@ -1441,7 +1441,10 @@ test("sessions.compact preserves accepted queued follow-up work", async () => { } }); -test("sessions.compact preserves an in-flight collected follow-up waiting in the command lane", async () => { +test.each([ + { name: "follow-up-backed", withFollowup: true }, + { name: "lane-only", withFollowup: false }, +])("sessions.compact preserves accepted $name command-lane work", async ({ withFollowup }) => { const { storePath } = await createSessionStoreDir(); const sessionId = "sess-compact-command-queue"; const sessionKey = "agent:main:main"; @@ -1462,8 +1465,9 @@ test("sessions.compact preserves an in-flight collected follow-up waiting in the enqueuedAt: Date.now(), run: {}, } as unknown as FollowupRun; - const queue = getFollowupQueue(sessionKey, { mode: "collect" }); - queue.inFlight.add(queuedRun); + if (withFollowup) { + getFollowupQueue(sessionKey, { mode: "collect" }).inFlight.add(queuedRun); + } setCommandLaneConcurrency(lane, 0); let commandRan = false; const queuedCommand = enqueueCommandInLane(lane, async () => { @@ -1484,7 +1488,9 @@ test("sessions.compact preserves an in-flight collected follow-up waiting in the code: "INVALID_REQUEST", message: "Session main has queued work; retry after it finishes.", }); - expect(getExistingFollowupQueue(sessionKey)?.inFlight).toContain(queuedRun); + expect(Boolean(getExistingFollowupQueue(sessionKey)?.inFlight.has(queuedRun))).toBe( + withFollowup, + ); expect(getCommandLaneSnapshot(lane).queuedCount).toBe(1); expect(commandRan).toBe(false); expect(embeddedRunMock.compactEmbeddedAgentSession).not.toHaveBeenCalled(); diff --git a/src/plugin-sdk/text-utility-runtime.ts b/src/plugin-sdk/text-utility-runtime.ts index 777847aef0c7..f519c5edf615 100644 --- a/src/plugin-sdk/text-utility-runtime.ts +++ b/src/plugin-sdk/text-utility-runtime.ts @@ -2,7 +2,16 @@ import type { BaseProbeResult } from "../channels/plugins/types.public.js"; import { withTimeout } from "../utils/with-timeout.js"; +export { + estimateToolResultTextChars, + sliceToolResultTextToBudget, +} from "../agents/embedded-agent-runner/tool-result-text-budget.js"; +export { + DEFAULT_MAX_LIVE_TOOL_RESULT_CHARS, + resolveLiveToolResultMaxChars, +} from "../agents/tool-result-limits.js"; export { escapeHtml } from "../shared/html-escape.js"; +export { estimateStringChars } from "../utils/cjk-chars.js"; type ChannelProbeResult = BaseProbeResult & { elapsedMs?: number };