diff --git a/src/agents/cli-runner.spawn.test.ts b/src/agents/cli-runner.spawn.test.ts index 00c37632e840..62b27038b051 100644 --- a/src/agents/cli-runner.spawn.test.ts +++ b/src/agents/cli-runner.spawn.test.ts @@ -43,9 +43,9 @@ import { attachCliMessagingDeliveryEvidence, getCliMessagingDeliveryEvidence, } from "./cli-runner/delivery-evidence.js"; +import { logCliInvocation } from "./cli-runner/execute-logging.js"; import { executePreparedCliRun } from "./cli-runner/execute.js"; import { - buildCliEnvAuthLog, buildCliExecLogLine, createManagedRun, setCliRunnerExecuteTestDeps, @@ -2617,33 +2617,40 @@ describe("runCliAgent spawn path", () => { expect(input.env?.OTEL_SDK_DISABLED).toBeUndefined(); }); - it("formats CLI auth env diagnostics as key names without secret values", () => { + it("logs CLI auth env diagnostics as key names without secret values", () => { vi.stubEnv("ANTHROPIC_API_KEY", "sk-ant-host"); vi.stubEnv("ANTHROPIC_API_TOKEN", "token-host"); vi.stubEnv("GEMINI_CLI_SYSTEM_SETTINGS_PATH", "/tmp/host-gemini-settings.json"); vi.stubEnv("OPENAI_API_KEY", "sk-openai-host"); + const log = vi.fn(); - const log = buildCliEnvAuthLog({ - ANTHROPIC_API_TOKEN: "token-child", - CLAUDE_CODE_PROVIDER_MANAGED_BY_HOST: "1", - GEMINI_CLI_HOME: "/tmp/child-gemini-home", - OPENAI_API_KEY: "sk-openai-child", + logCliInvocation({ + args: [], + command: "claude", + env: { + ANTHROPIC_API_TOKEN: "token-child", + CLAUDE_CODE_PROVIDER_MANAGED_BY_HOST: "1", + GEMINI_CLI_HOME: "/tmp/child-gemini-home", + OPENAI_API_KEY: "sk-openai-child", + }, + log, }); - expect(log).toMatch(/host=.*ANTHROPIC_API_KEY/); - expect(log).toMatch(/host=.*ANTHROPIC_API_TOKEN/); - expect(log).toMatch(/host=.*OPENAI_API_KEY/); - expect(log).toMatch(/child=.*ANTHROPIC_API_TOKEN/); - expect(log).toMatch(/child=.*CLAUDE_CODE_PROVIDER_MANAGED_BY_HOST/); - expect(log).toMatch(/child=.*OPENAI_API_KEY/); - expect(log).toMatch(/cleared=.*ANTHROPIC_API_KEY/); - expect(log).toMatch(/runtimeHost=.*GEMINI_CLI_SYSTEM_SETTINGS_PATH/); - expect(log).toMatch(/runtimeChild=.*GEMINI_CLI_HOME/); - expect(log).toMatch(/runtimeCleared=.*GEMINI_CLI_SYSTEM_SETTINGS_PATH/); - expect(log).not.toContain("sk-ant-host"); - expect(log).not.toContain("token-child"); - expect(log).not.toContain("/tmp/child-gemini-home"); - expect(log).not.toContain("sk-openai-child"); + const authLog = log.mock.calls.map(([message]) => String(message)).join("\n"); + expect(authLog).toMatch(/host=.*ANTHROPIC_API_KEY/); + expect(authLog).toMatch(/host=.*ANTHROPIC_API_TOKEN/); + expect(authLog).toMatch(/host=.*OPENAI_API_KEY/); + expect(authLog).toMatch(/child=.*ANTHROPIC_API_TOKEN/); + expect(authLog).toMatch(/child=.*CLAUDE_CODE_PROVIDER_MANAGED_BY_HOST/); + expect(authLog).toMatch(/child=.*OPENAI_API_KEY/); + expect(authLog).toMatch(/cleared=.*ANTHROPIC_API_KEY/); + expect(authLog).toMatch(/runtimeHost=.*GEMINI_CLI_SYSTEM_SETTINGS_PATH/); + expect(authLog).toMatch(/runtimeChild=.*GEMINI_CLI_HOME/); + expect(authLog).toMatch(/runtimeCleared=.*GEMINI_CLI_SYSTEM_SETTINGS_PATH/); + expect(authLog).not.toContain("sk-ant-host"); + expect(authLog).not.toContain("token-child"); + expect(authLog).not.toContain("/tmp/child-gemini-home"); + expect(authLog).not.toContain("sk-openai-child"); }); it("prepends bootstrap warnings to the CLI prompt body", async () => { diff --git a/src/agents/cli-runner/execute-logging.ts b/src/agents/cli-runner/execute-logging.ts index 98ce3cbea2f9..1c44921c5cd1 100644 --- a/src/agents/cli-runner/execute-logging.ts +++ b/src/agents/cli-runner/execute-logging.ts @@ -172,7 +172,7 @@ export function buildCliExecLogLine(params: { } /** Summarizes auth-related env keys preserved or cleared for a CLI child process. */ -export function buildCliEnvAuthLog(childEnv: Record): string { +function buildCliEnvAuthLog(childEnv: Record): string { const hostKeys = listPresentCliEnvKeys(process.env, CLI_ENV_AUTH_LOG_KEYS); const childKeys = listPresentCliEnvKeys(childEnv, CLI_ENV_AUTH_LOG_KEYS); const childKeySet = new Set(childKeys); diff --git a/src/agents/cli-runner/execute.test-support.ts b/src/agents/cli-runner/execute.test-support.ts index 66e21572009e..cd14730349a3 100644 --- a/src/agents/cli-runner/execute.test-support.ts +++ b/src/agents/cli-runner/execute.test-support.ts @@ -4,8 +4,8 @@ import { vi } from "vitest"; import type { requestHeartbeat } from "../../infra/heartbeat-wake.js"; import type { enqueueSystemEvent } from "../../infra/system-events.js"; import type { getProcessSupervisor } from "../../process/supervisor/index.js"; -import "./execute.js"; -import type { CliReusableSession } from "./types.js"; +import { executeDeps } from "./execute-deps.js"; +export { buildCliExecLogLine } from "./execute-logging.js"; type ProcessSupervisor = ReturnType; type SupervisorSpawnFn = ProcessSupervisor["spawn"]; @@ -13,40 +13,8 @@ type EnqueueSystemEventFn = typeof enqueueSystemEvent; type RequestHeartbeatFn = typeof requestHeartbeat; type UnknownMock = Mock<(...args: unknown[]) => unknown>; -type BuildCliExecLogLineParams = { - provider: string; - model: string; - promptChars: number; - trigger?: string; - useResume: boolean; - cliSessionId?: string; - resolvedSessionId?: string; - reusableSession: CliReusableSession; - hasHistoryPrompt: boolean; -}; - -type CliRunnerExecuteTestApi = { - buildCliEnvAuthLog(childEnv: Record): string; - buildCliExecLogLine(params: BuildCliExecLogLineParams): string; - setCliRunnerExecuteTestDeps(overrides: Record): void; -}; - -function getTestApi(): CliRunnerExecuteTestApi { - return (globalThis as Record)[ - Symbol.for("openclaw.cliRunnerExecuteTestApi") - ] as CliRunnerExecuteTestApi; -} - -export function buildCliEnvAuthLog(childEnv: Record): string { - return getTestApi().buildCliEnvAuthLog(childEnv); -} - -export function buildCliExecLogLine(params: BuildCliExecLogLineParams): string { - return getTestApi().buildCliExecLogLine(params); -} - -export function setCliRunnerExecuteTestDeps(overrides: Record): void { - getTestApi().setCliRunnerExecuteTestDeps(overrides); +export function setCliRunnerExecuteTestDeps(overrides: Partial): void { + Object.assign(executeDeps, overrides); } export const supervisorSpawnMock: UnknownMock = vi.fn(); diff --git a/src/agents/cli-runner/execute.ts b/src/agents/cli-runner/execute.ts index 2c5a2c008a6c..e2a683e3798e 100644 --- a/src/agents/cli-runner/execute.ts +++ b/src/agents/cli-runner/execute.ts @@ -28,7 +28,6 @@ import { prepareClaudeCliSkillsPlugin } from "./claude-skills-plugin.js"; import { executeDeps } from "./execute-deps.js"; import { createCliEventHandlers } from "./execute-events.js"; import { - buildCliEnvAuthLog, buildCliExecLogLine, CLAUDE_SELECTED_AUTH_ENV_KEYS, CLI_BACKEND_PRESERVE_ENV, @@ -114,16 +113,6 @@ function assertExactToolAvailabilityRuntimeVersion(params: { }); } -if (process.env.VITEST || process.env.NODE_ENV === "test") { - (globalThis as Record)[Symbol.for("openclaw.cliRunnerExecuteTestApi")] = { - buildCliEnvAuthLog, - buildCliExecLogLine, - setCliRunnerExecuteTestDeps: (overrides: Record) => { - Object.assign(executeDeps, overrides as Partial); - }, - }; -} - type ExecutePreparedCliRunOptions = { onPhase?: (phase: "send" | "resolve" | "cleanup") => void; }; diff --git a/src/agents/code-mode-bridge.ts b/src/agents/code-mode-bridge.ts index dc3ce00a9c18..17d879726fe9 100644 --- a/src/agents/code-mode-bridge.ts +++ b/src/agents/code-mode-bridge.ts @@ -30,20 +30,6 @@ import { ToolInputError } from "./tools/common.js"; import { resolveEligibleNodeFromList } from "./tools/nodes-utils.js"; import { resolveInternalSessionKey, resolveMainSessionAlias } from "./tools/sessions-helpers.js"; -type CodeModeSwarmDeps = { - emitSessionLifecycleEvent: typeof emitSessionLifecycleEvent; - getSwarmRunByLaunchReplayKey: typeof getSwarmRunByLaunchReplayKey; - initSubagentRegistry: typeof initSubagentRegistry; - waitForCollectorCompletion: typeof waitForCollectorCompletion; -}; - -const defaultCodeModeSwarmDeps: CodeModeSwarmDeps = { - emitSessionLifecycleEvent, - getSwarmRunByLaunchReplayKey, - initSubagentRegistry, - waitForCollectorCompletion, -}; - const CODE_MODE_NODES_TOOL_ID = "openclaw:core:nodes"; type CodeModeNode = { @@ -162,8 +148,6 @@ async function runNodesBridge(params: { throw new ToolInputError("unsupported nodes bridge action."); } -let codeModeSwarmDeps = defaultCodeModeSwarmDeps; - export function codeModeReplayIdForToolCall( ctx: ToolSearchToolContext, toolCallId: string, @@ -285,7 +269,7 @@ async function runAgentSpawnBridge(params: { // The registry persists this exact tuple and payload hash before launch. const idempotencyKey = `${params.codeModeRunId}:${params.request.id}`; const requesterSessionKey = resolveCodeModeRequesterSessionKey(params.ctx); - let existing = codeModeSwarmDeps.getSwarmRunByLaunchReplayKey( + let existing = getSwarmRunByLaunchReplayKey( idempotencyKey, requesterSessionKey, params.ctx.agentId, @@ -299,13 +283,10 @@ async function runAgentSpawnBridge(params: { throw new ToolInputError("agents.run persisted launch reservation cannot be recovered."); } // Cold-start restore idempotently re-enqueues this durable launch before agentWait parks. - codeModeSwarmDeps.initSubagentRegistry(); + initSubagentRegistry(); existing = - codeModeSwarmDeps.getSwarmRunByLaunchReplayKey( - idempotencyKey, - requesterSessionKey, - params.ctx.agentId, - ) ?? existing; + getSwarmRunByLaunchReplayKey(idempotencyKey, requesterSessionKey, params.ctx.agentId) ?? + existing; if (existing.swarmLaunchPending === true && !existing.queuedLaunch) { throw new ToolInputError("agents.run persisted launch reservation cannot be recovered."); } @@ -350,7 +331,7 @@ async function runAgentWaitBridge(params: { throw new ToolInputError("agents.run wait requires session identity."); } const requesterSessionKey = resolveCodeModeRequesterSessionKey(params.ctx); - return await codeModeSwarmDeps.waitForCollectorCompletion({ + return await waitForCollectorCompletion({ runId: runId.trim(), currentSessionKeys: new Set([rawSessionKey, requesterSessionKey]), currentAgentId: params.ctx.agentId, @@ -374,7 +355,7 @@ function runSwarmNoteBridge(params: { if (!sessionKey) { throw new ToolInputError("swarmNote requires session identity."); } - codeModeSwarmDeps.emitSessionLifecycleEvent({ + emitSessionLifecycleEvent({ sessionKey, reason: "swarm-note", swarmGroupId: resolveCodeModeSwarmGroupId(params.ctx), @@ -552,9 +533,3 @@ export async function runBridgeRequest(params: { return { id: params.request.id, ok: false, error: formatErrorMessage(error) }; } } - -export function setCodeModeSwarmDepsForTest(overrides?: Partial): void { - codeModeSwarmDeps = overrides - ? { ...defaultCodeModeSwarmDeps, ...overrides } - : defaultCodeModeSwarmDeps; -} diff --git a/src/agents/code-mode-headless.test.ts b/src/agents/code-mode-headless.test.ts index 1aa1b239960d..bb7d696cceef 100644 --- a/src/agents/code-mode-headless.test.ts +++ b/src/agents/code-mode-headless.test.ts @@ -1,4 +1,12 @@ -import { afterEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const loadCodeModeTypeScriptRuntime = vi.hoisted(() => + vi.fn<() => Promise>(), +); + +vi.mock("./code-mode-typescript-runtime.js", () => ({ + loadCodeModeTypeScriptRuntime, +})); import { createDeferred } from "../../test/helpers/promise.js"; import { prepareSource } from "./code-mode-runtime.js"; import { runCodeModeScriptHeadless, type CodeModeHeadlessResult } from "./code-mode.js"; @@ -57,9 +65,13 @@ function expectFailed(result: CodeModeHeadlessResult) { } describe("headless Code Mode", () => { + beforeEach(async () => { + loadCodeModeTypeScriptRuntime.mockResolvedValue(await import("typescript")); + }); + afterEach(() => { vi.useRealTimers(); - testing.setTypescriptRuntimeForTest(null); + loadCodeModeTypeScriptRuntime.mockReset(); expect(testing.activeRuns.size).toBe(0); testing.activeRuns.clear(); testing.resumingRunIds.clear(); @@ -827,7 +839,7 @@ describe("headless Code Mode", () => { }); it("times out an unfinished headless TypeScript runtime load", async () => { - testing.setTypescriptRuntimeForTest(new Promise(() => {})); + loadCodeModeTypeScriptRuntime.mockReturnValue(new Promise(() => {})); const result = expectFailed( await runCodeModeScriptHeadless({ @@ -847,7 +859,7 @@ describe("headless Code Mode", () => { }); it("aborts an unfinished headless TypeScript runtime load", async () => { - testing.setTypescriptRuntimeForTest(new Promise(() => {})); + loadCodeModeTypeScriptRuntime.mockReturnValue(new Promise(() => {})); const controller = new AbortController(); const resultPromise = runCodeModeScriptHeadless({ ctx: createHeadlessHarness(), @@ -868,31 +880,20 @@ describe("headless Code Mode", () => { it("keeps worker-leg wall-clock expiry classified as timeout", async () => { const ctx = createHeadlessHarness(); - const config = testing.resolveCodeModeHeadlessConfig(ctx); - const headlessScope = testing.createHeadlessAbortScope(undefined, 100); - try { - const result = await testing.runCodeModeWorker( - { - kind: "exec", - source: "while (true) {}", - config, - catalog: [], - apiFiles: [], - namespaces: [], - }, - 5_000, - undefined, - headlessScope.signal, - ); + expectCompleted(await runCodeModeScriptHeadless({ ctx, code: "return true;" })); - expect(result).toMatchObject({ - status: "failed", - code: "timeout", - error: "code mode timeout exceeded", - }); - } finally { - headlessScope.cleanup(); - } + expect( + expectFailed( + await runCodeModeScriptHeadless({ + ctx, + code: "while (true) {}", + wallClockMs: 100, + }), + ), + ).toMatchObject({ + code: "timeout", + error: "code mode timeout exceeded", + }); }); it("classifies syntax errors", async () => { diff --git a/src/agents/code-mode-headless.ts b/src/agents/code-mode-headless.ts index 900e9ea63133..4b3dcef61553 100644 --- a/src/agents/code-mode-headless.ts +++ b/src/agents/code-mode-headless.ts @@ -44,7 +44,7 @@ import { import { ToolSearchRuntime, type ToolSearchToolContext } from "./tool-search.js"; import { ToolInputError } from "./tools/common.js"; -export function createHeadlessAbortScope( +function createHeadlessAbortScope( signal: AbortSignal | undefined, wallClockMs: number, ): { signal: AbortSignal; cleanup: () => void } { diff --git a/src/agents/code-mode-runtime.ts b/src/agents/code-mode-runtime.ts index 8821e54a9cba..eb04a55cd11d 100644 --- a/src/agents/code-mode-runtime.ts +++ b/src/agents/code-mode-runtime.ts @@ -3,7 +3,6 @@ import { uniqueValues } from "@openclaw/normalization-core/string-normalization" import { parse, tokenizer } from "acorn"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { formatErrorMessage } from "../infra/errors.js"; -import { createLazyPromiseLoader } from "../shared/lazy-runtime.js"; import { clampNumber } from "../utils.js"; import { resolveAgentConfig } from "./agent-scope-config.js"; import { boundCodeModeResult } from "./code-mode-json.js"; @@ -16,6 +15,7 @@ import { CODE_MODE_SHELL_SOURCE_ERROR, isShellLikeCodeModeSource, } from "./code-mode-shell-source.js"; +import { loadCodeModeTypeScriptRuntime } from "./code-mode-typescript-runtime.js"; import type { CodeModeFailurePhase, CodeModeWorkerThreadResult } from "./code-mode-worker-types.js"; import type { ToolSearchConfig, ToolSearchToolContext } from "./tool-search.js"; import { asToolParamsRecord, ToolInputError } from "./tools/common.js"; @@ -96,14 +96,6 @@ export type CodeModeWorkerResult = output: unknown[]; }; -const typescriptRuntimeLoader = createLazyPromiseLoader(() => import("typescript"), { - cacheRejections: true, -}); -let typescriptRuntimeForTest: - | typeof import("typescript") - | Promise - | null = null; - function normalizeCodeModeRawConfig(value: unknown): Record | undefined { const codeMode = value; if (codeMode === true) { @@ -526,13 +518,6 @@ function rejectsModuleAccess( return /\bimport\b\s*(?:\.|\(|["'`{*]|\w)|\brequire\b\s*\(/u.test(source); } -async function loadTypeScriptRuntime(): Promise { - if (typescriptRuntimeForTest) { - return await typescriptRuntimeForTest; - } - return await typescriptRuntimeLoader.load(); -} - export async function prepareSource(input: { code: string; language?: CodeModeLanguage; @@ -551,7 +536,7 @@ export async function prepareSource(input: { } return input.code; } - const ts = await loadTypeScriptRuntime(); + const ts = await loadCodeModeTypeScriptRuntime(); if (rejectsModuleAccess(input.code, ts)) { throw new ToolInputError("code mode module access is disabled."); } @@ -599,13 +584,3 @@ export function enforceSnapshotPayloadLimits(params: { throw new CodeModeLimitError("code mode snapshot limit exceeded"); } } - -export const codeModeRuntimeTesting = { - getTypescriptRuntimePromise: (): Promise | null => - typescriptRuntimeLoader.peek() ?? null, - setTypescriptRuntimeForTest: ( - runtime: typeof import("typescript") | Promise | null, - ) => { - typescriptRuntimeForTest = runtime; - }, -}; diff --git a/src/agents/code-mode-swarm.test.ts b/src/agents/code-mode-swarm.test.ts index 4cf704582c25..d6e45eafc838 100644 --- a/src/agents/code-mode-swarm.test.ts +++ b/src/agents/code-mode-swarm.test.ts @@ -1,13 +1,43 @@ import { createHash } from "node:crypto"; import { stableStringify } from "@openclaw/normalization-core"; -import { afterEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { createCodeModeNamespaceRuntime } from "./code-mode-namespaces.js"; -import { resolveCodeModeConfig } from "./code-mode.js"; -import { testing } from "./code-mode.test-support.js"; +import { applyCodeModeCatalog, resolveCodeModeConfig } from "./code-mode.js"; +import { + createCodeModeHarness, + fakeTool, + runUntilCompleted, + testing, +} from "./code-mode.test-support.js"; +import type { SubagentRunRecord } from "./subagents/registry/subagent-registry.types.js"; import { SWARM_CODE_MODE_IDEMPOTENCY_KEY, SWARM_CODE_MODE_REQUEST_FINGERPRINT, } from "./subagents/swarm/swarm-code-mode.js"; +import { jsonResult, type AnyAgentTool } from "./tools/common.js"; + +const swarmMocks = vi.hoisted(() => ({ + emitSessionLifecycleEvent: vi.fn(), + getSwarmRunByLaunchReplayKey: vi.fn(), + initSubagentRegistry: vi.fn(), + waitForCollectorCompletion: vi.fn(), +})); + +vi.mock("../sessions/session-lifecycle-events.js", async (importOriginal) => ({ + ...(await importOriginal()), + emitSessionLifecycleEvent: swarmMocks.emitSessionLifecycleEvent, +})); + +vi.mock("./subagents/registry/subagent-registry.js", async (importOriginal) => ({ + ...(await importOriginal()), + getSwarmRunByLaunchReplayKey: swarmMocks.getSwarmRunByLaunchReplayKey, + initSubagentRegistry: swarmMocks.initSubagentRegistry, +})); + +vi.mock("./tools/agents-wait-tool.js", async (importOriginal) => ({ + ...(await importOriginal()), + waitForCollectorCompletion: swarmMocks.waitForCollectorCompletion, +})); const config = resolveCodeModeConfig({ tools: { codeMode: true } } as never); @@ -66,9 +96,78 @@ function swarmContext() { }; } +function collectorRecord(overrides: Partial = {}): SubagentRunRecord { + return { + runId: "collector-1", + childSessionKey: "agent:main:subagent:1", + requesterSessionKey: "agent:main:main", + requesterDisplayKey: "agent:main:main", + task: "Research", + cleanup: "delete", + createdAt: 1, + execution: { status: "running" }, + ...overrides, + }; +} + +function collectorFingerprint(task = "Research"): string { + return `sha256:${createHash("sha256") + .update( + stableStringify({ + task, + collect: true, + groupId: "swarm:agent:main:main:run-swarm", + }), + ) + .digest("hex")}`; +} + +function createSwarmHarness(execute?: AnyAgentTool["execute"]) { + const harness = createCodeModeHarness(); + const toolsConfig = (harness.config as { tools: Record }).tools; + toolsConfig.swarm = { enabled: true }; + Object.assign(harness.ctx, { + sessionId: "session-swarm", + runId: "run-swarm", + }); + const spawnTool = fakeTool("sessions_spawn", "Spawn a collector"); + spawnTool.execute = vi.fn( + execute ?? (async () => jsonResult({ status: "accepted", runId: "collector-1" })), + ) as AnyAgentTool["execute"]; + applyCodeModeCatalog({ + tools: [...harness.tools, spawnTool], + config: harness.config, + sessionId: harness.ctx.sessionId, + sessionKey: harness.ctx.sessionKey, + runId: harness.ctx.runId, + catalogRef: harness.catalogRef, + }); + return { ...harness, spawnTool }; +} + +async function runSwarmCode(harness: ReturnType, code: string) { + const execTool = harness.tools[0]; + const waitTool = harness.tools[1]; + if (!execTool || !waitTool) { + throw new Error("expected Code Mode exec and wait tools"); + } + return await runUntilCompleted({ execTool, waitTool, code }); +} + +beforeEach(() => { + swarmMocks.emitSessionLifecycleEvent.mockReset(); + swarmMocks.getSwarmRunByLaunchReplayKey.mockReset().mockReturnValue(undefined); + swarmMocks.initSubagentRegistry.mockReset(); + swarmMocks.waitForCollectorCompletion.mockReset().mockResolvedValue({ + runId: "collector-1", + status: "done", + result: "restored", + sessionKey: "agent:main:subagent:1", + }); +}); + afterEach(() => { testing.activeRuns.clear(); - testing.setSwarmDepsForTest(); }); describe("Code Mode swarm guest", () => { @@ -237,25 +336,10 @@ describe("Code Mode swarm host bridge", () => { }); it("dispatches notes with the canonical swarm group", async () => { - const emitSessionLifecycleEvent = vi.fn(); - testing.setSwarmDepsForTest({ emitSessionLifecycleEvent }); + const result = await runSwarmCode(createSwarmHarness(), 'phase("Plan"); return "ok";'); - const result = await testing.runBridgeRequest({ - runtime: {}, - namespaceRuntime: {}, - parentToolCallId: "parent", - codeModeRunId: "cm-note", - maxOutputBytes: 64 * 1024, - ctx: swarmContext(), - request: { - id: "bridge:1", - method: "swarmNote", - args: [{ kind: "phase", text: "Plan" }], - }, - }); - - expect(result).toMatchObject({ ok: true, value: { ok: true } }); - expect(emitSessionLifecycleEvent).toHaveBeenCalledWith({ + expect(result).toMatchObject({ status: "completed", value: "ok" }); + expect(swarmMocks.emitSessionLifecycleEvent).toHaveBeenCalledWith({ sessionKey: "agent:main:main", reason: "swarm-note", swarmGroupId: "swarm:agent:main:main:run-swarm", @@ -265,307 +349,81 @@ describe("Code Mode swarm host bridge", () => { }); it("re-settles a persisted collector after restart without double-spawn", async () => { - let persisted: Record | undefined; - let replayId = ""; - const callExactId = vi.fn(async (_id: string, input: Record) => { - const idempotencyKey = input[SWARM_CODE_MODE_IDEMPOTENCY_KEY]; - const requestFingerprint = input[SWARM_CODE_MODE_REQUEST_FINGERPRINT]; - expect(idempotencyKey).toBe(`${replayId}:bridge:1`); + let persisted: SubagentRunRecord | undefined; + const harness = createSwarmHarness(async (_toolCallId, input) => { + const spawnInput = input as Record; + const replayKey = spawnInput[SWARM_CODE_MODE_IDEMPOTENCY_KEY]; + const requestFingerprint = spawnInput[SWARM_CODE_MODE_REQUEST_FINGERPRINT]; + expect(replayKey).toEqual( + expect.stringMatching(/^cm_replay_[0-9a-f]{24}:bridge:agentSpawn:1$/u), + ); expect(requestFingerprint).toMatch(/^sha256:[0-9a-f]{64}$/u); - persisted = { - runId: "collector-1", + persisted = collectorRecord({ swarmRunId: "collector-1", - childSessionKey: "agent:main:subagent:1", collect: true, - swarmLaunchReplayKey: idempotencyKey, - swarmLaunchRequestFingerprint: requestFingerprint, - }; - return { - result: { details: { status: "accepted", runId: "collector-1" } }, - }; - }); - const runtime = { - namespaceEntries: () => [ - { id: "openclaw:core:sessions_spawn", source: "openclaw", name: "sessions_spawn" }, - ], - callExactId, - }; - const getSwarmRunByLaunchReplayKey = vi.fn(() => persisted); - const waitForCollectorCompletion = vi.fn(async () => ({ - runId: "collector-1", - status: "done", - result: "restored", - sessionKey: "agent:main:subagent:1", - })); - testing.setSwarmDepsForTest({ - getSwarmRunByLaunchReplayKey, - waitForCollectorCompletion, - }); - const spawnRequest = { - id: "bridge:1", - method: "agentSpawn", - args: ["Research", { label: "facts" }], - }; - const globalAliasContext = { - ...swarmContext(), - sessionKey: "main", - config: { tools: { codeMode: true, swarm: { enabled: true } }, session: { scope: "global" } }, - runtimeConfig: { - tools: { codeMode: true, swarm: { enabled: true } }, - session: { scope: "global" }, - }, - } as const; - const code = 'return await agents.run("Research", { label: "facts" });'; - replayId = testing.codeModeReplayIdForToolCall( - globalAliasContext, - "call_0", - code, - "response-turn-1", - ); - const restoredReplayId = testing.codeModeReplayIdForToolCall( - globalAliasContext, - "call_0", - code, - structuredClone("response-turn-1"), - ); - expect(restoredReplayId).toBe(replayId); - const bridgeBase = { - runtime, - namespaceRuntime: {}, - parentToolCallId: "parent", - codeModeRunId: restoredReplayId, - maxOutputBytes: 64 * 1024, - ctx: globalAliasContext, - }; - - const first = await testing.runBridgeRequest({ ...bridgeBase, request: spawnRequest }); - const replayed = await testing.runBridgeRequest({ ...bridgeBase, request: spawnRequest }); - const waited = await testing.runBridgeRequest({ - ...bridgeBase, - request: { id: "bridge:2", method: "agentWait", args: ["collector-1"] }, - }); - - expect(first).toMatchObject({ ok: true, value: { runId: "collector-1" } }); - expect(replayed).toMatchObject({ ok: true, value: { runId: "collector-1" } }); - expect(waited).toMatchObject({ ok: true, value: { status: "done", result: "restored" } }); - expect(callExactId).toHaveBeenCalledTimes(1); - expect(getSwarmRunByLaunchReplayKey).toHaveBeenNthCalledWith( - 1, - `${replayId}:bridge:1`, - "global", - undefined, - ); - expect(getSwarmRunByLaunchReplayKey).toHaveBeenNthCalledWith( - 2, - `${replayId}:bridge:1`, - "global", - undefined, - ); - expect(waitForCollectorCompletion).toHaveBeenCalledWith({ - config: globalAliasContext.config, - currentAgentId: undefined, - runId: "collector-1", - currentSessionKeys: new Set(["main", "global"]), - signal: undefined, - }); - }); - - it("spawns two collectors when later turns reuse the tool-call id and source", async () => { - const persistedByReplayKey = new Map>(); - const callExactId = vi.fn(async (_id: string, input: Record) => { - const replayKey = String(input[SWARM_CODE_MODE_IDEMPOTENCY_KEY]); - const runId = `collector-${persistedByReplayKey.size + 1}`; - persistedByReplayKey.set(replayKey, { - runId, - childSessionKey: `agent:main:subagent:${persistedByReplayKey.size + 1}`, - swarmLaunchReplayKey: replayKey, - swarmLaunchRequestFingerprint: input[SWARM_CODE_MODE_REQUEST_FINGERPRINT], + swarmLaunchReplayKey: String(replayKey), + swarmLaunchRequestFingerprint: String(requestFingerprint), }); - return { result: { details: { status: "accepted", runId } } }; + return jsonResult({ status: "accepted", runId: "collector-1" }); }); - testing.setSwarmDepsForTest({ - getSwarmRunByLaunchReplayKey: (key) => persistedByReplayKey.get(key), - }); - const runtime = { - namespaceEntries: () => [ - { id: "openclaw:core:sessions_spawn", source: "openclaw", name: "sessions_spawn" }, - ], - callExactId, - }; - const ctx = swarmContext(); + swarmMocks.getSwarmRunByLaunchReplayKey.mockImplementation(() => persisted); const code = 'return await agents.run("Research");'; - const firstReplayId = testing.codeModeReplayIdForToolCall( - ctx, - "call_0", - code, - "response-turn-1", - ); - const secondReplayId = testing.codeModeReplayIdForToolCall( - ctx, - "call_0", - code, - "response-turn-2", - ); - const bridgeBase = { - runtime, - namespaceRuntime: {}, - parentToolCallId: "parent", - maxOutputBytes: 64 * 1024, - ctx, - request: { id: "bridge:1", method: "agentSpawn", args: ["Research", {}] }, - }; - const first = await testing.runBridgeRequest({ - ...bridgeBase, - codeModeRunId: firstReplayId, - }); - const second = await testing.runBridgeRequest({ - ...bridgeBase, - codeModeRunId: secondReplayId, - }); + const first = await runSwarmCode(harness, code); + const replayed = await runSwarmCode(harness, code); - expect(secondReplayId).not.toBe(firstReplayId); - expect(first).toMatchObject({ ok: true, value: { runId: "collector-1" } }); - expect(second).toMatchObject({ ok: true, value: { runId: "collector-2" } }); - expect(callExactId).toHaveBeenCalledTimes(2); - expect([...persistedByReplayKey.keys()]).toEqual([ - `${firstReplayId}:bridge:1`, - `${secondReplayId}:bridge:1`, - ]); + expect(first).toMatchObject({ status: "completed", value: "restored" }); + expect(replayed).toMatchObject({ status: "completed", value: "restored" }); + expect(harness.spawnTool.execute).toHaveBeenCalledTimes(1); + expect(swarmMocks.getSwarmRunByLaunchReplayKey).toHaveBeenCalledTimes(2); + expect(swarmMocks.waitForCollectorCompletion).toHaveBeenCalledTimes(2); }); - it("rejects replay when the collector request payload changes", async () => { - const callExactId = vi.fn(async (_id: string, input: Record) => ({ - result: { details: { status: "accepted", runId: "collector-1" } }, - fingerprint: input[SWARM_CODE_MODE_REQUEST_FINGERPRINT], - })); - const runtime = { - namespaceEntries: () => [ - { id: "openclaw:core:sessions_spawn", source: "openclaw", name: "sessions_spawn" }, - ], - callExactId, - }; - const persisted: { value?: Record } = {}; - testing.setSwarmDepsForTest({ - getSwarmRunByLaunchReplayKey: () => persisted.value, - }); - const bridgeBase = { - runtime, - namespaceRuntime: {}, - parentToolCallId: "parent", - codeModeRunId: "cm-restart", - maxOutputBytes: 64 * 1024, - ctx: swarmContext(), - }; - const first = await testing.runBridgeRequest({ - ...bridgeBase, - request: { id: "bridge:1", method: "agentSpawn", args: ["Research one", {}] }, - }); - expect(first.ok).toBe(true); - const spawnInput = callExactId.mock.calls[0]?.[1] as Record; - persisted.value = { - runId: "collector-1", - childSessionKey: "agent:main:subagent:1", - swarmLaunchRequestFingerprint: spawnInput[SWARM_CODE_MODE_REQUEST_FINGERPRINT], - }; + it("rejects a persisted collector whose request fingerprint does not match", async () => { + swarmMocks.getSwarmRunByLaunchReplayKey.mockReturnValue( + collectorRecord({ swarmLaunchRequestFingerprint: collectorFingerprint("Different task") }), + ); + const harness = createSwarmHarness(); - const replay = await testing.runBridgeRequest({ - ...bridgeBase, - request: { id: "bridge:1", method: "agentSpawn", args: ["Research two", {}] }, - }); + const result = await runSwarmCode(harness, 'return await agents.run("Research");'); - expect(replay).toMatchObject({ ok: false }); - expect(replay.ok ? "" : replay.error).toContain("does not match the persisted collector"); - expect(callExactId).toHaveBeenCalledTimes(1); + expect(result).toMatchObject({ status: "failed", code: "internal_error" }); + expect(String(result.error)).toContain("does not match the persisted collector"); + expect(harness.spawnTool.execute).not.toHaveBeenCalled(); }); it("rejects a pending reservation without durable launch state", async () => { - const callExactId = vi.fn(async (_id: string, _input: Record) => ({ - result: { details: { status: "accepted", runId: "collector-1" } }, - })); - const runtime = { - namespaceEntries: () => [ - { id: "openclaw:core:sessions_spawn", source: "openclaw", name: "sessions_spawn" }, - ], - callExactId, - }; - const persisted: { value?: Record } = {}; - const initSubagentRegistry = vi.fn(); - testing.setSwarmDepsForTest({ - getSwarmRunByLaunchReplayKey: () => persisted.value, - initSubagentRegistry, - }); - const bridgeBase = { - runtime, - namespaceRuntime: {}, - parentToolCallId: "parent", - codeModeRunId: "cm-restart", - maxOutputBytes: 64 * 1024, - ctx: swarmContext(), - }; - await testing.runBridgeRequest({ - ...bridgeBase, - request: { id: "bridge:1", method: "agentSpawn", args: ["Research", {}] }, - }); - const spawnInput = callExactId.mock.calls[0]?.[1] as Record; - persisted.value = { - runId: "collector-1", - childSessionKey: "agent:main:subagent:1", - swarmLaunchPending: true, - swarmLaunchRequestFingerprint: spawnInput[SWARM_CODE_MODE_REQUEST_FINGERPRINT], - }; + swarmMocks.getSwarmRunByLaunchReplayKey.mockReturnValue( + collectorRecord({ + swarmLaunchPending: true, + swarmLaunchRequestFingerprint: collectorFingerprint(), + }), + ); + const harness = createSwarmHarness(); - const replay = await testing.runBridgeRequest({ - ...bridgeBase, - request: { id: "bridge:1", method: "agentSpawn", args: ["Research", {}] }, - }); + const result = await runSwarmCode(harness, 'return await agents.run("Research");'); - expect(replay).toMatchObject({ ok: false }); - expect(replay.ok ? "" : replay.error).toContain("launch reservation cannot be recovered"); - expect(initSubagentRegistry).not.toHaveBeenCalled(); - expect(callExactId).toHaveBeenCalledTimes(1); + expect(result).toMatchObject({ status: "failed", code: "internal_error" }); + expect(String(result.error)).toContain("launch reservation cannot be recovered"); + expect(swarmMocks.initSubagentRegistry).not.toHaveBeenCalled(); + expect(harness.spawnTool.execute).not.toHaveBeenCalled(); }); it("re-enqueues a durable pending reservation before returning its handle", async () => { - const initSubagentRegistry = vi.fn(); - testing.setSwarmDepsForTest({ - initSubagentRegistry, - getSwarmRunByLaunchReplayKey: () => ({ - runId: "collector-1", - childSessionKey: "agent:main:subagent:1", + swarmMocks.getSwarmRunByLaunchReplayKey.mockReturnValue( + collectorRecord({ swarmLaunchPending: true, - swarmLaunchRequestFingerprint: `sha256:${createHash("sha256") - .update( - stableStringify({ - task: "Research", - collect: true, - groupId: "swarm:agent:main:main:run-swarm", - }), - ) - .digest("hex")}`, + swarmLaunchRequestFingerprint: collectorFingerprint(), queuedLaunch: { request: {}, timeoutMs: 1, schedulerGroupKey: "group", maxConcurrent: 1 }, }), - }); - const runtime = { - namespaceEntries: () => [ - { id: "openclaw:core:sessions_spawn", source: "openclaw", name: "sessions_spawn" }, - ], - callExactId: vi.fn(), - }; + ); + const harness = createSwarmHarness(); - const replay = await testing.runBridgeRequest({ - runtime, - namespaceRuntime: {}, - parentToolCallId: "parent", - codeModeRunId: "cm-restart", - maxOutputBytes: 64 * 1024, - ctx: swarmContext(), - request: { id: "bridge:1", method: "agentSpawn", args: ["Research", {}] }, - }); + const result = await runSwarmCode(harness, 'return await agents.run("Research");'); - expect(replay).toMatchObject({ ok: true, value: { runId: "collector-1" } }); - expect(initSubagentRegistry).toHaveBeenCalledOnce(); - expect(runtime.callExactId).not.toHaveBeenCalled(); + expect(result).toMatchObject({ status: "completed", value: "restored" }); + expect(swarmMocks.initSubagentRegistry).toHaveBeenCalledOnce(); + expect(harness.spawnTool.execute).not.toHaveBeenCalled(); }); it("renews expired snapshots while agentWait remains pending", () => { diff --git a/src/agents/code-mode-typescript-runtime.ts b/src/agents/code-mode-typescript-runtime.ts new file mode 100644 index 000000000000..f13badc6aaa8 --- /dev/null +++ b/src/agents/code-mode-typescript-runtime.ts @@ -0,0 +1,9 @@ +import { createLazyPromiseLoader } from "../shared/lazy-runtime.js"; + +const typescriptRuntimeLoader = createLazyPromiseLoader(() => import("typescript"), { + cacheRejections: true, +}); + +export function loadCodeModeTypeScriptRuntime(): Promise { + return typescriptRuntimeLoader.load(); +} diff --git a/src/agents/code-mode-worker.ts b/src/agents/code-mode-worker.ts index 353bf9fe6425..01eebb1a8d16 100644 --- a/src/agents/code-mode-worker.ts +++ b/src/agents/code-mode-worker.ts @@ -1,10 +1,9 @@ import { readFile } from "node:fs/promises"; import { createRequire } from "node:module"; -import path from "node:path"; -import { fileURLToPath, pathToFileURL } from "node:url"; import { Worker } from "node:worker_threads"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { formatErrorMessage } from "../infra/errors.js"; +import { resolveRuntimeWorkerUrl } from "../infra/runtime-worker-url.js"; import type { CodeModeFailureCode, CodeModeWorkerResult } from "./code-mode-runtime.js"; let quickJsWasmModulePromise: Promise | undefined; @@ -23,20 +22,12 @@ function getQuickJsWasmModule(): Promise { return quickJsWasmModulePromise; } -export function resolveCodeModeWorkerUrl(currentModuleUrl: string): URL { - const currentPath = fileURLToPath(currentModuleUrl); - const distMarker = `${path.sep}dist${path.sep}`; - const distIndex = currentPath.lastIndexOf(distMarker); - if (distIndex >= 0) { - const distRoot = currentPath.slice(0, distIndex + distMarker.length - 1); - return pathToFileURL(path.join(distRoot, "agents", "code-mode.worker.js")); - } - const extension = path.extname(currentPath) || ".js"; - return new URL(`./code-mode.worker${extension}`, currentModuleUrl); -} - function codeModeWorkerUrl(): URL { - return resolveCodeModeWorkerUrl(import.meta.url); + return resolveRuntimeWorkerUrl({ + currentModuleUrl: import.meta.url, + sourceWorkerName: "code-mode.worker", + distWorkerPath: "agents/code-mode.worker.js", + }); } function failedCodeModeWorkerResult( diff --git a/src/agents/code-mode.guest.test.ts b/src/agents/code-mode.guest.test.ts index e3dae9c257d9..3ed135c981f7 100644 --- a/src/agents/code-mode.guest.test.ts +++ b/src/agents/code-mode.guest.test.ts @@ -474,28 +474,6 @@ describe("Code Mode guest execution", () => { ); }); - it("does not load TypeScript for plain JavaScript code mode runs", async () => { - const { config, catalogRef, tools: codeModeTools } = createCodeModeHarness(); - applyCodeModeCatalog({ - tools: [...codeModeTools, pluginTool("fake_noop", "Noop")], - config, - sessionId: "session-code-mode", - sessionKey: "agent:main:main", - runId: "run-code-mode", - catalogRef, - }); - - const details = await runUntilCompleted({ - execTool: expectDefined(codeModeTools[0], "codeModeTools[0] test invariant"), - waitTool: expectDefined(codeModeTools[1], "codeModeTools[1] test invariant"), - code: "return 42;", - }); - - expect(details.status).toBe("completed"); - expect(details.value).toBe(42); - expect(testing.getTypescriptRuntimePromise()).toBeNull(); - }); - it("allows identifiers and strings that contain import without module access", async () => { const { config, catalogRef, tools: codeModeTools } = createCodeModeHarness(); applyCodeModeCatalog({ diff --git a/src/agents/code-mode.test-support.ts b/src/agents/code-mode.test-support.ts index 36a48a265ece..69177a5d1208 100644 --- a/src/agents/code-mode.test-support.ts +++ b/src/agents/code-mode.test-support.ts @@ -1,136 +1,27 @@ import { expect, vi } from "vitest"; import { setPluginToolMeta } from "../plugins/tools.js"; +import { codeModeReplayIdForToolCall } from "./code-mode-bridge.js"; +import { resolveCodeModeHeadlessConfig } from "./code-mode-runtime.js"; import type { CodeModeSkill } from "./code-mode-skills.js"; +import { activeRuns, removeExpiredRuns, resumingRunIds } from "./code-mode-state.js"; +import { normalizeCodeModeWorkerResult, runCodeModeWorker } from "./code-mode-worker.js"; import { createCodeModeTools } from "./code-mode.js"; -import { - createToolSearchCatalogRef, - type ToolSearchCatalogRef, - type ToolSearchToolContext, -} from "./tool-search.js"; +import { createToolSearchCatalogRef, type ToolSearchCatalogRef } from "./tool-search.js"; import { jsonResult, type AnyAgentTool } from "./tools/common.js"; -type CodeModeConfig = { - enabled: boolean; - runtime: "quickjs-wasi"; - mode: "only"; - languages: ("javascript" | "typescript")[]; - timeoutMs: number; - memoryLimitBytes: number; - maxOutputBytes: number; - maxSnapshotBytes: number; - maxPendingToolCalls: number; - snapshotTtlSeconds: number; - searchDefaultLimit: number; - maxSearchLimit: number; +export const testing = { + activeRuns, + resumingRunIds, + codeModeReplayIdForToolCall, + removeExpiredRuns, + normalizeCodeModeWorkerResult, + runCodeModeWorker, + resolveCodeModeHeadlessConfig, }; -type CodeModeFailureCode = - | "aborted" - | "invalid_input" - | "runtime_unavailable" - | "timeout" - | "output_limit_exceeded" - | "snapshot_limit_exceeded" - | "internal_error"; - -type CodeModeWorkerResult = - | { status: "completed"; value: unknown; output: unknown[] } - | { - status: "waiting"; - snapshotBytes: Uint8Array; - pendingRequests: Array<{ id: string; method: string; args: unknown[] }>; - output: unknown[]; - } - | { - status: "failed"; - error: string; - code: CodeModeFailureCode; - failurePhase: "input" | "guest" | "bridge" | "host"; - bridgeDispatchStarted: boolean; - output: unknown[]; - }; - -type CodeModeTestApi = { - activeRuns: Map< - string, - { - runId: string; - config: CodeModeConfig; - expiresAt: number; - replayId?: string; - agentWaitRetainUntil?: number; - pending: Array<{ - id: string; - method: string; - args: unknown[]; - promise: Promise; - settled?: unknown; - cancel?: () => void; - }>; - } - >; - resumingRunIds: Set; - codeModeReplayIdForToolCall( - ctx: ToolSearchToolContext, - toolCallId: string, - code: string, - assistantTurnId?: string, - ): string; - removeExpiredRuns(now?: number): void; - runBridgeRequest( - params: Record, - ): Promise<{ id: string; ok: true; value: unknown } | { id: string; ok: false; error: string }>; - createHeadlessAbortScope( - signal: AbortSignal | undefined, - wallClockMs: number, - ): { signal: AbortSignal; cleanup: () => void }; - normalizeCodeModeWorkerResult(result: CodeModeWorkerResult): CodeModeWorkerResult; - runCodeModeWorker( - workerData: unknown, - timeoutMs: number, - workerUrl?: URL, - signal?: AbortSignal, - ): Promise; - resolveCodeModeHeadlessConfig( - ctx: ToolSearchToolContext, - overrides?: Partial< - Pick< - CodeModeConfig, - | "timeoutMs" - | "memoryLimitBytes" - | "maxOutputBytes" - | "maxSnapshotBytes" - | "maxPendingToolCalls" - > - >, - ): CodeModeConfig; - resolveCodeModeWorkerUrl(currentModuleUrl: string): URL; - getTypescriptRuntimePromise(): Promise | null; - setTypescriptRuntimeForTest( - runtime: typeof import("typescript") | Promise | null, - ): void; - setSwarmDepsForTest(overrides?: { - emitSessionLifecycleEvent?: (event: Record) => void; - getSwarmRunByLaunchReplayKey?: (key: string, requesterSessionKey?: string) => unknown; - initSubagentRegistry?: () => void; - waitForCollectorCompletion?: (params: Record) => Promise; - }): void; -}; - -function getTestApi(): CodeModeTestApi { - const api = (globalThis as Record)[Symbol.for("openclaw.codeModeTestApi")]; - if (!api) { - throw new Error("code mode test API is unavailable"); - } - return api as CodeModeTestApi; -} - -export const testing = getTestApi(); - export function resetCodeModeTestState(): void { testing.activeRuns.clear(); testing.resumingRunIds.clear(); - testing.setTypescriptRuntimeForTest(null); } export function fakeTool(name: string, description: string): AnyAgentTool { diff --git a/src/agents/code-mode.test.ts b/src/agents/code-mode.test.ts index 17e20fce54c3..0a4f2782962a 100644 --- a/src/agents/code-mode.test.ts +++ b/src/agents/code-mode.test.ts @@ -15,7 +15,6 @@ import { pluginTool, mcpTool, createCodeModeHarness, - testing, } from "./code-mode.test-support.js"; import { createToolSearchCatalogRef, @@ -35,15 +34,6 @@ describe("Code Mode catalog and model-visible surface", () => { resetCodeModeTestState(); }); - it("resolves the packaged worker URL from stable and hashed dist modules", () => { - expect(testing.resolveCodeModeWorkerUrl("file:///repo/dist/agents/code-mode.js").pathname).toBe( - "/repo/dist/agents/code-mode.worker.js", - ); - expect(testing.resolveCodeModeWorkerUrl("file:///repo/dist/selection-abc123.js").pathname).toBe( - "/repo/dist/agents/code-mode.worker.js", - ); - }); - it("hides all normal tools behind exec and wait", () => { const { config, catalogRef, tools: codeModeTools } = createCodeModeHarness(); const shellExec = fakeTool("exec", "Run shell command"); diff --git a/src/agents/code-mode.ts b/src/agents/code-mode.ts index f2d8582fb69a..dab6a459d24e 100644 --- a/src/agents/code-mode.ts +++ b/src/agents/code-mode.ts @@ -6,11 +6,6 @@ import { Type } from "typebox"; import { getAgentToolExecutionContext } from "../../packages/agent-core/src/tool-execution-context.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import type { HookContext } from "./agent-tools.before-tool-call.js"; -import { - codeModeReplayIdForToolCall, - runBridgeRequest, - setCodeModeSwarmDepsForTest, -} from "./code-mode-bridge.js"; import { CODE_MODE_EXEC_TOOL_NAME, CODE_MODE_WAIT_TOOL_NAME, @@ -18,22 +13,16 @@ import { markCodeModeControlTool, } from "./code-mode-control-tools.js"; import { runCodeModeExec, runWait } from "./code-mode-execution.js"; -import { createHeadlessAbortScope, runCodeModeScriptHeadless } from "./code-mode-headless.js"; +import { runCodeModeScriptHeadless } from "./code-mode-headless.js"; import { describeCodeModeNamespacesForPrompt } from "./code-mode-namespaces.js"; import { - codeModeRuntimeTesting, isCodeModeEngagedForModel, readCode, readRunId, resolveCodeModeConfig, - resolveCodeModeHeadlessConfig, } from "./code-mode-runtime.js"; -import { activeRuns, removeExpiredRuns, resumingRunIds } from "./code-mode-state.js"; import { normalizeCodeModeTimeoutResult, - normalizeCodeModeWorkerResult, - resolveCodeModeWorkerUrl, - runCodeModeWorker, CodeModeHeadlessAbortError, CodeModeHeadlessTimeoutError, } from "./code-mode-worker.js"; @@ -351,24 +340,3 @@ export function addClientToolsToCodeModeCatalog(params: { enabled: resolveCodeModeConfig(params.config, params.agentId).enabled !== false, }); } - -/** Test-only hooks and state accessors for Code Mode worker orchestration. */ -const testing = { - activeRuns, - resumingRunIds, - codeModeReplayIdForToolCall, - removeExpiredRuns, - runBridgeRequest, - createHeadlessAbortScope, - normalizeCodeModeWorkerResult, - runCodeModeWorker, - resolveCodeModeHeadlessConfig, - resolveCodeModeWorkerUrl, - getTypescriptRuntimePromise: codeModeRuntimeTesting.getTypescriptRuntimePromise, - setTypescriptRuntimeForTest: codeModeRuntimeTesting.setTypescriptRuntimeForTest, - setSwarmDepsForTest: setCodeModeSwarmDepsForTest, -}; - -if (process.env.VITEST || process.env.NODE_ENV === "test") { - (globalThis as Record)[Symbol.for("openclaw.codeModeTestApi")] = testing; -} diff --git a/src/agents/code-mode.typescript.test.ts b/src/agents/code-mode.typescript.test.ts index 99ffc8ad039b..6bee5e85eceb 100644 --- a/src/agents/code-mode.typescript.test.ts +++ b/src/agents/code-mode.typescript.test.ts @@ -2,6 +2,14 @@ import { expectDefined } from "@openclaw/normalization-core"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const loadCodeModeTypeScriptRuntime = vi.hoisted(() => + vi.fn<() => Promise>(), +); + +vi.mock("./code-mode-typescript-runtime.js", () => ({ + loadCodeModeTypeScriptRuntime, +})); import { applyCodeModeCatalog } from "./code-mode.js"; import { resetCodeModeTestState, @@ -13,28 +21,18 @@ import { } from "./code-mode.test-support.js"; describe("Code Mode TypeScript execution", () => { - beforeEach(() => { + beforeEach(async () => { vi.useRealTimers(); + loadCodeModeTypeScriptRuntime.mockResolvedValue(await import("typescript")); }); afterEach(() => { vi.useRealTimers(); + loadCodeModeTypeScriptRuntime.mockReset(); resetCodeModeTestState(); }); it("supports TypeScript source transform", async () => { - testing.setTypescriptRuntimeForTest({ - ...(await import("typescript")), - transpileModule: vi.fn((code: string) => ({ - outputText: code.replace(": number", ""), - diagnostics: [], - })), - ScriptTarget: { ES2022: 9 }, - ModuleKind: { ESNext: 99 }, - ImportsNotUsedAsValues: { Remove: 0 }, - DiagnosticCategory: { Error: 1 }, - flattenDiagnosticMessageText: (message: unknown) => String(message), - } as never); const { config, catalogRef, tools: codeModeTools } = createCodeModeHarness(); applyCodeModeCatalog({ tools: [...codeModeTools, pluginTool("fake_noop", "Noop")], @@ -131,7 +129,7 @@ describe("Code Mode TypeScript execution", () => { runId: "run-code-mode", catalogRef, }); - testing.setTypescriptRuntimeForTest(new Promise(() => {})); + loadCodeModeTypeScriptRuntime.mockReturnValue(new Promise(() => {})); const result = resultDetails( await expectDefined(codeModeTools[0], "Code Mode exec test invariant").execute( @@ -161,7 +159,7 @@ describe("Code Mode TypeScript execution", () => { runId: "run-code-mode", catalogRef, }); - testing.setTypescriptRuntimeForTest(new Promise(() => {})); + loadCodeModeTypeScriptRuntime.mockReturnValue(new Promise(() => {})); const controller = new AbortController(); const resultPromise = expectDefined(codeModeTools[0], "Code Mode exec test invariant").execute( "code-call-typescript-load-abort", diff --git a/src/agents/compaction-planning-worker-runtime.ts b/src/agents/compaction-planning-worker-runtime.ts new file mode 100644 index 000000000000..1c75286a5290 --- /dev/null +++ b/src/agents/compaction-planning-worker-runtime.ts @@ -0,0 +1,117 @@ +import { Worker } from "node:worker_threads"; +import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion"; +import { toErrorObject } from "../infra/errors.js"; +import { resolveRuntimeWorkerUrl } from "../infra/runtime-worker-url.js"; +import type { + CompactionPlanningWorkerInput, + CompactionPlanningWorkerResult, + CompactionPlanningWorkerValue, +} from "./compaction-planning.worker.js"; + +const COMPACTION_PLANNING_WORKER_TIMEOUT_MS = 60_000; + +export class CompactionPlanningWorkerError extends Error { + constructor( + message: string, + readonly code: "unavailable" | "timeout" | "failed", + ) { + super(message); + this.name = "CompactionPlanningWorkerError"; + } +} + +function compactionPlanningWorkerUrl(): URL { + return resolveRuntimeWorkerUrl({ + currentModuleUrl: import.meta.url, + sourceWorkerName: "compaction-planning.worker", + distWorkerPath: "agents/compaction-planning.worker.js", + }); +} + +export function runCompactionPlanningWorker(params: { + input: CompactionPlanningWorkerInput; + signal?: AbortSignal; + timeoutMs?: number; + workerUrl?: URL; +}): Promise { + const abortError = () => + toErrorObject( + params.signal?.reason ?? new Error("compaction planning aborted"), + "Non-Error rejection", + ); + if (params.signal?.aborted) { + return Promise.reject(abortError()); + } + + const workerUrl = params.workerUrl ?? compactionPlanningWorkerUrl(); + const sourceWorkerExecArgv = workerUrl.pathname.endsWith(".ts") ? ["--import", "tsx"] : undefined; + let worker: Worker; + try { + worker = new Worker(workerUrl, { + workerData: params.input, + execArgv: sourceWorkerExecArgv, + }); + } catch (error) { + return Promise.reject( + new CompactionPlanningWorkerError( + error instanceof Error ? error.message : String(error), + "unavailable", + ), + ); + } + + worker.unref?.(); + + return new Promise((resolve, reject) => { + let settled = false; + const timeout = setTimeout( + () => + fail(new CompactionPlanningWorkerError("compaction planning worker timed out", "timeout")), + resolveTimerTimeoutMs(params.timeoutMs, COMPACTION_PLANNING_WORKER_TIMEOUT_MS), + ); + const abort = () => fail(abortError()); + + const settle = (finish: () => void, terminate: boolean) => { + if (settled) { + return; + } + settled = true; + clearTimeout(timeout); + params.signal?.removeEventListener("abort", abort); + worker.removeAllListeners(); + if (terminate) { + void worker.terminate(); + } + finish(); + }; + const fail = (error: Error, terminate = true) => settle(() => reject(error), terminate); + + params.signal?.addEventListener("abort", abort, { once: true }); + + worker.once("message", (message: CompactionPlanningWorkerResult) => { + settle(() => { + if (message.status === "ok") { + resolve(message.value); + return; + } + reject(new CompactionPlanningWorkerError(message.error, "failed")); + }, false); + }); + worker.once("error", (error) => { + const message = error instanceof Error ? error.message : String(error); + fail(new CompactionPlanningWorkerError(message, "unavailable")); + }); + worker.once("exit", (code) => { + if (code === 0) { + return; + } + fail( + new CompactionPlanningWorkerError( + `compaction planning worker exited with code ${code}`, + "unavailable", + ), + false, + ); + }); + }); +} diff --git a/src/agents/compaction-planning-worker.test-support.ts b/src/agents/compaction-planning-worker.test-support.ts deleted file mode 100644 index 6859d993e6a8..000000000000 --- a/src/agents/compaction-planning-worker.test-support.ts +++ /dev/null @@ -1,19 +0,0 @@ -import type { - CompactionPlanningWorkerInput, - CompactionPlanningWorkerValue, -} from "./compaction-planning.worker.js"; -import "./compaction-planning-worker.js"; - -type CompactionPlanningWorkerTestApi = { - resolveCompactionPlanningWorkerUrl(currentModuleUrl?: string): URL; - runCompactionPlanningWorker(params: { - input: CompactionPlanningWorkerInput; - signal?: AbortSignal; - timeoutMs?: number; - workerUrl?: URL; - }): Promise; -}; - -export const compactionPlanningWorkerTesting = (globalThis as Record)[ - Symbol.for("openclaw.compactionPlanningWorkerTestApi") -] as CompactionPlanningWorkerTestApi; diff --git a/src/agents/compaction-planning-worker.test.ts b/src/agents/compaction-planning-worker.test.ts index e22b963ccf31..4ff95bd3857d 100644 --- a/src/agents/compaction-planning-worker.test.ts +++ b/src/agents/compaction-planning-worker.test.ts @@ -2,11 +2,11 @@ import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion"; import { serializeConversation } from "openclaw/plugin-sdk/agent-core"; import { beforeAll, describe, expect, it, vi } from "vitest"; +import { runCompactionPlanningWorker } from "./compaction-planning-worker-runtime.js"; import { buildOversizedFallbackPlanWithWorker, buildSummaryChunksWithWorker, } from "./compaction-planning-worker.js"; -import { compactionPlanningWorkerTesting } from "./compaction-planning-worker.test-support.js"; import { estimateMessagesTokens } from "./compaction-planning.js"; import { runCompactionPlanningWorkerInput } from "./compaction-planning.worker.js"; import type { AgentMessage } from "./runtime/index.js"; @@ -27,12 +27,10 @@ function createSyntheticWorkerUrl(source: string): URL { } describe("compaction planning worker", () => { - let packagedSummaryChunks: Awaited< - ReturnType - >; + let packagedSummaryChunks: Awaited>; beforeAll(async () => { - packagedSummaryChunks = await compactionPlanningWorkerTesting.runCompactionPlanningWorker({ + packagedSummaryChunks = await runCompactionPlanningWorker({ input: { kind: "summaryChunks", messages: [makeMessage(1), makeMessage(2), makeMessage(3)], @@ -42,21 +40,6 @@ describe("compaction planning worker", () => { }); }); - it("resolves the packaged worker URL from stable and hashed dist modules", () => { - // Hashed bundle names still resolve to the stable worker sibling emitted by - // the build, so runtime imports do not depend on the main chunk hash. - expect( - compactionPlanningWorkerTesting.resolveCompactionPlanningWorkerUrl( - "file:///repo/dist/agents/compaction-planning-worker.js", - ).pathname, - ).toBe("/repo/dist/agents/compaction-planning.worker.js"); - expect( - compactionPlanningWorkerTesting.resolveCompactionPlanningWorkerUrl( - "file:///repo/dist/selection-abc123.js", - ).pathname, - ).toBe("/repo/dist/agents/compaction-planning.worker.js"); - }); - it("rejects invalid and retired worker input", () => { for (const input of [ { kind: "summaryChunks" }, @@ -235,7 +218,7 @@ describe("compaction planning worker", () => { `); const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); try { - await compactionPlanningWorkerTesting.runCompactionPlanningWorker({ + await runCompactionPlanningWorker({ input: { kind: "summaryChunks", messages: [makeMessage(1), makeMessage(2), makeMessage(3)], @@ -254,7 +237,7 @@ describe("compaction planning worker", () => { it("classifies missing worker runtime as unavailable", async () => { await expect( - compactionPlanningWorkerTesting.runCompactionPlanningWorker({ + runCompactionPlanningWorker({ input: { kind: "summaryChunks", messages: [makeMessage(1)], @@ -284,20 +267,18 @@ describe("compaction planning worker", () => { const timer = new Promise<"timer">((resolve) => { setTimeout(() => resolve("timer"), 0); }); - const planning = compactionPlanningWorkerTesting - .runCompactionPlanningWorker({ - input: { - kind: "stageSplit", - messages: Array.from({ length: 180 }, (_, index) => - makeMessage(index + 1, "x".repeat(12_000)), - ), - maxChunkTokens: 8000, - parts: 4, - }, - timeoutMs: 30_000, - workerUrl, - }) - .then(() => "planning" as const); + const planning = runCompactionPlanningWorker({ + input: { + kind: "stageSplit", + messages: Array.from({ length: 180 }, (_, index) => + makeMessage(index + 1, "x".repeat(12_000)), + ), + maxChunkTokens: 8000, + parts: 4, + }, + timeoutMs: 30_000, + workerUrl, + }).then(() => "planning" as const); await expect(Promise.race([timer, planning])).resolves.toBe("timer"); await expect(planning).resolves.toBe("planning"); diff --git a/src/agents/compaction-planning-worker.ts b/src/agents/compaction-planning-worker.ts index ed139bcffc55..8aea43bd9ba6 100644 --- a/src/agents/compaction-planning-worker.ts +++ b/src/agents/compaction-planning-worker.ts @@ -1,12 +1,11 @@ +import { + CompactionPlanningWorkerError, + runCompactionPlanningWorker, +} from "./compaction-planning-worker-runtime.js"; /** * Runs CPU-heavy compaction planning in a worker thread when histories are * large enough to risk starving the main event loop. */ -import path from "node:path"; -import { fileURLToPath, pathToFileURL } from "node:url"; -import { Worker } from "node:worker_threads"; -import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion"; -import { toErrorObject } from "../infra/errors.js"; import { buildOversizedFallbackPlan, buildStageSplitPlan, @@ -19,127 +18,14 @@ import { } from "./compaction-planning.js"; import type { CompactionPlanningWorkerInput, - CompactionPlanningWorkerResult, CompactionPlanningWorkerValue, } from "./compaction-planning.worker.js"; import type { AgentMessage } from "./runtime/index.js"; -const COMPACTION_PLANNING_WORKER_TIMEOUT_MS = 60_000; // Worker startup is more expensive than local planning for tiny histories. // Keep small compactions synchronous; move only starvation-sized plans off-thread. const COMPACTION_PLANNING_WORKER_MIN_MESSAGES = 64; -class CompactionPlanningWorkerError extends Error { - constructor( - message: string, - readonly code: "unavailable" | "timeout" | "failed", - ) { - super(message); - this.name = "CompactionPlanningWorkerError"; - } -} - -function resolveCompactionPlanningWorkerUrl(currentModuleUrl = import.meta.url): URL { - const currentPath = fileURLToPath(currentModuleUrl); - const normalized = currentPath.replaceAll(path.sep, "/"); - const distMarker = "/dist/"; - const distIndex = normalized.lastIndexOf(distMarker); - if (distIndex >= 0) { - const distRoot = currentPath.slice(0, distIndex + distMarker.length); - return pathToFileURL(path.join(distRoot, "agents", "compaction-planning.worker.js")); - } - const extension = path.extname(currentPath) || ".js"; - return new URL(`./compaction-planning.worker${extension}`, currentModuleUrl); -} - -function runCompactionPlanningWorker(params: { - input: CompactionPlanningWorkerInput; - signal?: AbortSignal; - timeoutMs?: number; - workerUrl?: URL; -}): Promise { - const abortError = () => - toErrorObject( - params.signal?.reason ?? new Error("compaction planning aborted"), - "Non-Error rejection", - ); - if (params.signal?.aborted) { - return Promise.reject(abortError()); - } - - const workerUrl = params.workerUrl ?? resolveCompactionPlanningWorkerUrl(); - const sourceWorkerExecArgv = workerUrl.pathname.endsWith(".ts") ? ["--import", "tsx"] : undefined; - let worker: Worker; - try { - worker = new Worker(workerUrl, { - workerData: params.input, - execArgv: sourceWorkerExecArgv, - }); - } catch (error) { - return Promise.reject( - new CompactionPlanningWorkerError( - error instanceof Error ? error.message : String(error), - "unavailable", - ), - ); - } - - worker.unref?.(); - - return new Promise((resolve, reject) => { - let settled = false; - const timeout = setTimeout( - () => - fail(new CompactionPlanningWorkerError("compaction planning worker timed out", "timeout")), - resolveTimerTimeoutMs(params.timeoutMs, COMPACTION_PLANNING_WORKER_TIMEOUT_MS), - ); - const abort = () => fail(abortError()); - - const settle = (finish: () => void, terminate: boolean) => { - if (settled) { - return; - } - settled = true; - clearTimeout(timeout); - params.signal?.removeEventListener("abort", abort); - worker.removeAllListeners(); - if (terminate) { - void worker.terminate(); - } - finish(); - }; - const fail = (error: Error, terminate = true) => settle(() => reject(error), terminate); - - params.signal?.addEventListener("abort", abort, { once: true }); - - worker.once("message", (message: CompactionPlanningWorkerResult) => { - settle(() => { - if (message.status === "ok") { - resolve(message.value); - return; - } - reject(new CompactionPlanningWorkerError(message.error, "failed")); - }, false); - }); - worker.once("error", (error) => { - const message = error instanceof Error ? error.message : String(error); - fail(new CompactionPlanningWorkerError(message, "unavailable")); - }); - worker.once("exit", (code) => { - if (code === 0) { - return; - } - fail( - new CompactionPlanningWorkerError( - `compaction planning worker exited with code ${code}`, - "unavailable", - ), - false, - ); - }); - }); -} - function restoreIndexedMessages(source: AgentMessage[], indexes: number[]): AgentMessage[] { return indexes.map((index) => { const message = source.at(index); @@ -264,14 +150,3 @@ export async function computeAdaptiveChunkRatioWithWorker(params: { restore: (value) => value.ratio, }); } - -const compactionPlanningWorkerTesting = { - resolveCompactionPlanningWorkerUrl, - runCompactionPlanningWorker, -}; - -if (process.env.VITEST || process.env.NODE_ENV === "test") { - (globalThis as Record)[ - Symbol.for("openclaw.compactionPlanningWorkerTestApi") - ] = compactionPlanningWorkerTesting; -} diff --git a/src/infra/runtime-worker-url.test.ts b/src/infra/runtime-worker-url.test.ts new file mode 100644 index 000000000000..9a514dd4f378 --- /dev/null +++ b/src/infra/runtime-worker-url.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, it } from "vitest"; +import { resolveRuntimeWorkerUrl } from "./runtime-worker-url.js"; + +describe("resolveRuntimeWorkerUrl", () => { + it("resolves source siblings and stable packaged worker paths", () => { + expect( + resolveRuntimeWorkerUrl({ + currentModuleUrl: "file:///repo/src/agents/code-mode-worker.ts", + sourceWorkerName: "code-mode.worker", + distWorkerPath: "agents/code-mode.worker.js", + }).pathname, + ).toBe("/repo/src/agents/code-mode.worker.ts"); + + for (const currentModuleUrl of [ + "file:///repo/dist/agents/code-mode.js", + "file:///repo/dist/selection-abc123.js", + ]) { + expect( + resolveRuntimeWorkerUrl({ + currentModuleUrl, + sourceWorkerName: "code-mode.worker", + distWorkerPath: "agents/code-mode.worker.js", + }).pathname, + ).toBe("/repo/dist/agents/code-mode.worker.js"); + } + }); +}); diff --git a/src/infra/runtime-worker-url.ts b/src/infra/runtime-worker-url.ts new file mode 100644 index 000000000000..981d8f9dce33 --- /dev/null +++ b/src/infra/runtime-worker-url.ts @@ -0,0 +1,20 @@ +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; + +/** Resolve a source worker sibling or its stable packaged path under dist. */ +export function resolveRuntimeWorkerUrl(params: { + currentModuleUrl: string; + sourceWorkerName: string; + distWorkerPath: string; +}): URL { + const currentPath = fileURLToPath(params.currentModuleUrl); + const normalized = currentPath.replaceAll(path.sep, "/"); + const distMarker = "/dist/"; + const distIndex = normalized.lastIndexOf(distMarker); + if (distIndex >= 0) { + const distRoot = currentPath.slice(0, distIndex + distMarker.length); + return pathToFileURL(path.join(distRoot, params.distWorkerPath)); + } + const extension = path.extname(currentPath) || ".js"; + return new URL(`./${params.sourceWorkerName}${extension}`, params.currentModuleUrl); +}