diff --git a/extensions/anthropic/session-catalog.test.ts b/extensions/anthropic/session-catalog.test.ts index 4db4af0fed3f..3884678d0246 100644 --- a/extensions/anthropic/session-catalog.test.ts +++ b/extensions/anthropic/session-catalog.test.ts @@ -1614,6 +1614,39 @@ describe("Claude session catalog", () => { expect(readFileSpy.mock.calls.filter(([filePath]) => isCatalogFile(filePath))).toEqual([]); }); + it("bounds append-only snapshot staleness to 15 seconds", async () => { + const home = await createHome(); + const projectDir = path.join(home, ".claude", "projects", "-workspace"); + const sessionId = "append-staleness"; + const transcriptPath = path.join(projectDir, `${sessionId}.jsonl`); + await writeProject({ + home, + entries: [], + transcripts: { [sessionId]: [sdkCliMessage(sessionId, "Initial")] }, + }); + const baseNow = Date.now(); + const fixedDirectoryTime = new Date(baseNow - 10_000); + await fs.utimes(projectDir, fixedDirectoryTime, fixedDirectoryTime); + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(baseNow); + const initial = await listLocalClaudeSessionPage({}, home); + const initialUpdatedAt = initial.sessions[0]?.updatedAt; + + await fs.appendFile(transcriptPath, `${JSON.stringify({ type: "progress" })}\n`); + const appendedAt = new Date(baseNow + 2_000); + await fs.utimes(transcriptPath, appendedAt, appendedAt); + // Content writes do not portably change the parent directory mtime. Pin it so this test owns + // only the documented TTL path on every filesystem used by CI. + await fs.utimes(projectDir, fixedDirectoryTime, fixedDirectoryTime); + + nowSpy.mockReturnValue(baseNow + 14_999); + const staleWithinBound = await listLocalClaudeSessionPage({}, home); + expect(staleWithinBound.sessions[0]?.updatedAt).toBe(initialUpdatedAt); + + nowSpy.mockReturnValue(baseNow + 15_000); + const refreshedAtBound = await listLocalClaudeSessionPage({}, home); + expect(refreshedAtBound.sessions[0]?.updatedAt).toBe(appendedAt.getTime()); + }); + it("invalidates the assembled scan on a project directory mtime change", async () => { const home = await createHome(); const projectDir = path.join(home, ".claude", "projects", "-workspace"); diff --git a/extensions/anthropic/session-catalog.ts b/extensions/anthropic/session-catalog.ts index a9bf7d885b67..ed0ac6622341 100644 --- a/extensions/anthropic/session-catalog.ts +++ b/extensions/anthropic/session-catalog.ts @@ -65,7 +65,7 @@ const MAX_CATALOG_DISCOVERY_FILES = 10_000; const MAX_CATALOG_DISCOVERY_CACHE_ENTRIES = 20_000; const MAX_CATALOG_JSON_CACHE_ENTRIES = 4_000; const MAX_CLAUDE_SESSION_SCAN_CACHE_ENTRIES = 8; -const CLAUDE_SESSION_SCAN_TTL_MS = 5_000; +const CLAUDE_SESSION_SCAN_TTL_MS = 15_000; const CLAUDE_METADATA_PREFIX_BYTES = 1024 * 1024; const CLAUDE_METADATA_READ_CHUNK_BYTES = 16 * 1024; const MAX_CATALOG_METADATA_SCAN_BYTES = 64 * 1024 * 1024; @@ -791,7 +791,8 @@ async function listClaudeSessions( const cached = claudeSessionScanCache.get(root); // This tree stamp is the catalog cache's cheap freshness owner: exact child membership makes // same-tick adds, deletes, and renames visible even when a filesystem reuses the directory mtime. - // File appends and Desktop-only changes may stay stale for at most the five-second TTL. + // File appends and Desktop-only changes may keep this snapshot stale for at most 15 seconds, + // below the UI's 30-second cadence. Clients observe them on the first poll after cache expiry. if ( options.forceRefresh !== true && cached && diff --git a/extensions/codex/src/session-catalog-terminal.ts b/extensions/codex/src/session-catalog-terminal.ts index 57a10dd281f6..fe4b8ae21c23 100644 --- a/extensions/codex/src/session-catalog-terminal.ts +++ b/extensions/codex/src/session-catalog-terminal.ts @@ -58,12 +58,31 @@ export async function requireCatalogEligibleThread( control: CodexSessionCatalogControl, threadId: string, ): Promise { + // Mutating actions use a fresh pinned control and authoritative thread/read. For passive callers, + // keep the three-second positive hit; only a miss needs to bypass the list memo. + const cached = await findCatalogEligibleThread(control, threadId, false); + if (cached) { + return cached; + } + const refreshed = await findCatalogEligibleThread(control, threadId, true); + if (refreshed) { + return refreshed; + } + throw new CatalogParamsError("Codex session is not a non-archived interactive Codex session"); +} + +async function findCatalogEligibleThread( + control: CodexSessionCatalogControl, + threadId: string, + forceRefresh: boolean, +): Promise { let cursor: string | undefined; const seenCursors = new Set(); for (let pageIndex = 0; pageIndex < MAX_ACTION_CATALOG_PAGES; pageIndex += 1) { const page = await control.listPage({ limit: CODEX_SESSION_CATALOG_MAX_PAGE_LIMIT, ...(cursor ? { cursor } : {}), + ...(forceRefresh ? { forceRefresh: true } : {}), }); const candidate = page.sessions.find((session) => session.threadId === threadId); if (candidate) { @@ -74,7 +93,7 @@ export async function requireCatalogEligibleThread( } const nextCursor = page.nextCursor?.trim(); if (!nextCursor) { - throw new CatalogParamsError("Codex session is not a non-archived interactive Codex session"); + return undefined; } if (seenCursors.has(nextCursor)) { throw new CatalogParamsError("Codex session eligibility could not be verified"); diff --git a/extensions/codex/src/session-catalog-types.ts b/extensions/codex/src/session-catalog-types.ts index a0420de5ee05..314a165b6b61 100644 --- a/extensions/codex/src/session-catalog-types.ts +++ b/extensions/codex/src/session-catalog-types.ts @@ -41,6 +41,8 @@ export type CodexSessionCatalogPageParams = { limit?: number; searchTerm?: string; cwd?: string; + /** Bypasses the brief list memo after a specific thread lookup misses. */ + forceRefresh?: boolean; }; export type CodexSessionCatalogControl = { diff --git a/extensions/codex/src/session-catalog.test.ts b/extensions/codex/src/session-catalog.test.ts index 2227753c3dbf..5c384803d089 100644 --- a/extensions/codex/src/session-catalog.test.ts +++ b/extensions/codex/src/session-catalog.test.ts @@ -19,7 +19,10 @@ import { } from "./app-server/session-binding.test-helpers.js"; import { listPairedNode } from "./session-catalog-node-continue.js"; import { catalogError, parseCatalogPage } from "./session-catalog-parsing.js"; -import { CODEX_TERMINAL_RESUME_COMMAND } from "./session-catalog-terminal.js"; +import { + CODEX_TERMINAL_RESUME_COMMAND, + requireCatalogEligibleThread, +} from "./session-catalog-terminal.js"; import { CODEX_LOCAL_SESSION_HOST_ID, codexSessionCatalogRuntime, @@ -534,6 +537,67 @@ describe("Codex supervision catalog", () => { }); }); + it("memoizes cloned request options until runtime config identity changes", async () => { + let runtimeConfig = { agents: { defaults: { workspace: "/workspace/a" } } } as OpenClawConfig; + commandRpcMocks.codexControlRequest.mockResolvedValue({ thread: idleThread() }); + const control = createCodexSessionCatalogControl({ + getPluginConfig: () => ({ supervision: { enabled: true } }), + getRuntimeConfig: () => runtimeConfig, + }); + const cloneSpy = vi.spyOn(globalThis, "structuredClone"); + + await control.readThread("thread-1"); + await control.readThread("thread-1"); + expect(cloneSpy).toHaveBeenCalledTimes(2); + + runtimeConfig = { agents: { defaults: { workspace: "/workspace/b" } } } as OpenClawConfig; + await control.readThread("thread-1"); + expect(cloneSpy).toHaveBeenCalledTimes(4); + }); + + it("briefly memoizes thread lists and invalidates on TTL or config identity", async () => { + let now = 1_000; + let runtimeConfig = {} as OpenClawConfig; + commandRpcMocks.codexControlRequest.mockResolvedValue({ + data: [idleThread({ source: "cli" })], + }); + const control = createCodexSessionCatalogControl({ + getPluginConfig: () => ({ supervision: { enabled: true } }), + getRuntimeConfig: () => runtimeConfig, + now: () => now, + }); + + await control.listPage({ limit: 25 }); + await control.listPage({ limit: 25 }); + expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledOnce(); + + now += 3_001; + await control.listPage({ limit: 25 }); + expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledTimes(2); + + runtimeConfig = { agents: {} } as OpenClawConfig; + await control.listPage({ limit: 25 }); + expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledTimes(3); + }); + + it("force-refreshes after a cached specific-thread miss", async () => { + let includeThread = false; + commandRpcMocks.codexControlRequest.mockImplementation(async () => ({ + data: includeThread ? [idleThread({ id: "thread-new", source: "cli" })] : [], + })); + const control = createCodexSessionCatalogControl({ + getPluginConfig: () => ({ supervision: { enabled: true } }), + getRuntimeConfig: () => config, + }); + await control.listPage({ limit: 100 }); + includeThread = true; + + await expect(requireCatalogEligibleThread(control, "thread-new")).resolves.toMatchObject({ + threadId: "thread-new", + }); + expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledTimes(2); + }); + it("scans bounded native pages for complete title-only search results", async () => { const pluginConfig = { supervision: { enabled: true } }; commandRpcMocks.codexControlRequest.mockImplementation( @@ -773,14 +837,16 @@ describe("Codex supervision catalog", () => { it("keeps catalog reads and writes available when supervision is disabled live", async () => { let pluginConfig: unknown = { supervision: { enabled: true } }; + let runtimeConfig = {} as OpenClawConfig; commandRpcMocks.codexControlRequest.mockResolvedValue({ data: [] }); const control = createCodexSessionCatalogControl({ getPluginConfig: () => pluginConfig, - getRuntimeConfig: () => config, + getRuntimeConfig: () => runtimeConfig, }); await expect(control.listPage({})).resolves.toEqual({ sessions: [] }); pluginConfig = { supervision: { enabled: false } }; + runtimeConfig = { plugins: {} } as OpenClawConfig; await expect(control.listPage({})).resolves.toEqual({ sessions: [] }); expect(commandRpcMocks.codexControlRequest).toHaveBeenCalledTimes(2); @@ -3392,6 +3458,8 @@ describe("Codex supervision actions", () => { const binDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-codex-terminal-")); tempDirs.push(binDir); process.env.PATH = binDir; + let now = Date.now(); + vi.spyOn(Date, "now").mockImplementation(() => now); const executable = path.join(binDir, process.platform === "win32" ? "codex.cmd" : "codex"); const control = createEligibleControl({ listPage: vi.fn(async () => ({ @@ -3468,6 +3536,7 @@ describe("Codex supervision actions", () => { if (process.platform !== "win32") { await fs.chmod(executable, 0o755); } + now += 60_001; await expect(getProvider()?.list({})).resolves.toEqual( expect.arrayContaining([ expect.objectContaining({ diff --git a/extensions/codex/src/session-catalog.ts b/extensions/codex/src/session-catalog.ts index e34532e39084..d53c5913fdb8 100644 --- a/extensions/codex/src/session-catalog.ts +++ b/extensions/codex/src/session-catalog.ts @@ -16,7 +16,10 @@ import { import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; import { CODEX_CONTROL_METHODS } from "./app-server/capabilities.js"; import { resolveCodexAppServerClientInstanceId } from "./app-server/client.js"; -import { resolveCodexSupervisionAppServerRuntimeOptions } from "./app-server/config.js"; +import { + resolveCodexSupervisionAppServerRuntimeOptions, + type CodexAppServerStartOptions, +} from "./app-server/config.js"; import { buildCodexAppServerConnectionFingerprint } from "./app-server/plugin-app-cache-key.js"; import { assertCodexThreadForkParams } from "./app-server/protocol-validators.js"; import type { @@ -103,6 +106,8 @@ import { toGenericTranscriptItem } from "./session-catalog-transcript-item.js"; import type { CodexSessionCatalogControl, CodexSessionCatalogHost, + CodexSessionCatalogPage, + CodexSessionCatalogPageParams, CodexSessionCatalogParams, CodexSessionCatalogResult, CodexSessionCatalogSession, @@ -119,6 +124,28 @@ const boundCatalogSessionId = (value: unknown) => boundedCatalogString(value, MAX_SESSION_ID_LENGTH); const CODEX_SUPERVISION_SESSION_KEY_PREFIX = "harness:codex:supervision:"; +const CODEX_SESSION_CATALOG_LIST_TTL_MS = 3_000; +const CODEX_SESSION_CATALOG_LIST_CACHE_MAX_ENTRIES = 32; + +type CodexCatalogRequestOptions = { + config: OpenClawConfig | undefined; + startOptions: CodexAppServerStartOptions; +}; + +type CodexCatalogPageCacheEntry = { + expiresAt: number; + page: Promise; +}; + +function codexCatalogPageCacheKey(params: CodexSessionCatalogPageParams): string { + // Mirror listPage's search/cwd normalization; these trimmed values are what reach app-server. + return JSON.stringify([ + params.cursor ?? null, + params.limit ?? null, + params.searchTerm?.trim().toLocaleLowerCase() || null, + params.cwd?.trim() || null, + ]); +} export { CODEX_LOCAL_SESSION_HOST_ID, @@ -239,13 +266,35 @@ export function createCodexSessionCatalogControl(params: { }): CodexSessionCatalogControl { const now = params.now ?? Date.now; const getPluginConfig = () => params.getPluginConfig(); + const requestOptionsByConfig = new WeakMap(); + const catalogPagesByConfig = new WeakMap< + OpenClawConfig, + Map + >(); + const resolveRequestOptions = ( + startOptions: CodexAppServerStartOptions, + ): CodexCatalogRequestOptions => { + const runtimeConfig = params.getRuntimeConfig(); + if (!runtimeConfig) { + return { config: undefined, startOptions: structuredClone(startOptions) }; + } + const cached = requestOptionsByConfig.get(runtimeConfig); + if (cached) { + // Plugin start options derive from this same immutable config snapshot. Config reload changes + // object identity; re-cloning on every poll only adds CPU and allocation to the catalog path. + return cached; + } + const resolved = { + config: structuredClone(runtimeConfig), + startOptions: structuredClone(startOptions), + }; + requestOptionsByConfig.set(runtimeConfig, resolved); + return resolved; + }; const createRequestSnapshot = (): CodexSessionCatalogRequestSnapshot => { const pluginConfig = getPluginConfig(); const runtime = resolveCodexSupervisionAppServerRuntimeOptions({ pluginConfig }); - const requestOptions = { - config: structuredClone(params.getRuntimeConfig()), - startOptions: structuredClone(runtime.start), - }; + const requestOptions = resolveRequestOptions(runtime.start); return { requestTimeoutMs: runtime.requestTimeoutMs, listThreads: async (listParams, timeoutMs) => @@ -290,8 +339,7 @@ export function createCodexSessionCatalogControl(params: { const withPinnedConnection: CodexSessionCatalogControl["withPinnedConnection"] = async (run) => { const pluginConfig = getPluginConfig(); const runtime = resolveCodexSupervisionAppServerRuntimeOptions({ pluginConfig }); - const runtimeConfig = structuredClone(params.getRuntimeConfig()); - const startOptions = structuredClone(runtime.start); + const { config: runtimeConfig, startOptions } = resolveRequestOptions(runtime.start); const client = await getLeasedSharedCodexAppServerClient({ config: runtimeConfig, startOptions, @@ -361,11 +409,55 @@ export function createCodexSessionCatalogControl(params: { } }; - return createCodexSessionCatalogControlFromRequests({ + const control = createCodexSessionCatalogControlFromRequests({ createRequestSnapshot, now, withPinnedConnection, }); + return { + ...control, + async listPage(pageParams) { + const runtimeConfig = params.getRuntimeConfig(); + if (!runtimeConfig) { + return await control.listPage(pageParams); + } + let cache = catalogPagesByConfig.get(runtimeConfig); + if (!cache) { + cache = new Map(); + catalogPagesByConfig.set(runtimeConfig, cache); + } + const key = codexCatalogPageCacheKey(pageParams); + const cached = cache.get(key); + if (pageParams.forceRefresh !== true && cached && cached.expiresAt > now()) { + // The app-server may scan rollout metadata for thread/list. Share a page for three seconds; + // config identity and forceRefresh invalidate it so specific actions cannot miss new rows. + cache.delete(key); + cache.set(key, cached); + return await cached.page; + } + if (cached) { + cache.delete(key); + } + const page = control.listPage(pageParams); + const entry = { expiresAt: now() + CODEX_SESSION_CATALOG_LIST_TTL_MS, page }; + cache.set(key, entry); + while (cache.size > CODEX_SESSION_CATALOG_LIST_CACHE_MAX_ENTRIES) { + const oldest = cache.keys().next(); + if (oldest.done) { + break; + } + cache.delete(oldest.value); + } + try { + return await page; + } catch (error) { + if (cache.get(key) === entry) { + cache.delete(key); + } + throw error; + } + }, + }; } async function listGatewayHost(params: { diff --git a/src/config/runtime-snapshot.ts b/src/config/runtime-snapshot.ts index 9917a1089833..c99c7d22a4a2 100644 --- a/src/config/runtime-snapshot.ts +++ b/src/config/runtime-snapshot.ts @@ -1,5 +1,6 @@ // Produces redacted runtime config snapshots for diagnostics and UI surfaces. import { sha256Base64Url } from "../infra/crypto-digest.js"; +import { clearExecutablePathCache } from "../infra/executable-path.js"; import { resetPublishedConfigRuntimeEnv, type PreparedConfigRuntimeEnv, @@ -165,6 +166,7 @@ export function setRuntimeConfigSnapshot( config: OpenClawConfig, sourceConfig?: OpenClawConfig, ): void { + clearExecutablePathCache(); runtimeConfigSnapshot = config; runtimeConfigSourceSnapshot = sourceConfig ?? null; runtimeConfigSnapshotMetadata = createRuntimeConfigSnapshotMetadata(config, sourceConfig); @@ -195,6 +197,7 @@ export function setRuntimeConfigSourceSnapshotIfCurrent(params: { } export function resetConfigRuntimeState(): void { + clearExecutablePathCache(); runtimeConfigSnapshot = null; runtimeConfigSourceSnapshot = null; runtimeConfigSnapshotMetadata = null; diff --git a/src/gateway/server-methods/session-catalog.test.ts b/src/gateway/server-methods/session-catalog.test.ts index 5d5419155ff2..742746234d46 100644 --- a/src/gateway/server-methods/session-catalog.test.ts +++ b/src/gateway/server-methods/session-catalog.test.ts @@ -72,15 +72,29 @@ async function call( config: Record = {}, client?: { connect?: { scopes?: string[] }; connId?: string }, contextOverrides: Record = {}, +) { + const pending = startCall(method, params, config, client, contextOverrides); + await pending.completion; + return pending.respond; +} + +function startCall( + method: keyof typeof sessionCatalogHandlers, + params: unknown, + config: Record = {}, + client?: { connect?: { scopes?: string[] }; connId?: string }, + contextOverrides: Record = {}, ) { const respond = vi.fn(); - await sessionCatalogHandlers[method]?.({ - params, - respond, - client, - context: { getRuntimeConfig: () => config, ...contextOverrides }, - } as never); - return respond; + const completion = Promise.resolve( + sessionCatalogHandlers[method]?.({ + params, + respond, + client, + context: { getRuntimeConfig: () => config, ...contextOverrides }, + } as never), + ); + return { completion, respond }; } describe("session catalog Gateway methods", () => { @@ -162,6 +176,62 @@ describe("session catalog Gateway methods", () => { }); }); + it("single-flights identical concurrent lists and gives followers only the final result", async () => { + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const host = { + hostId: "gateway:local", + label: "Local", + kind: "gateway" as const, + connected: true, + sessions: [], + }; + const list = vi.fn(async ({ onHost }: { onHost?: (value: typeof host) => void }) => { + onHost?.(host); + await gate; + return [host]; + }); + hoisted.activeRegistry.sessionCatalogs = [{ provider: provider("codex", { list }) }]; + const config = { agents: { list: [{ id: "main" }, { id: "research" }] } }; + const leaderBroadcast = vi.fn(); + const followerBroadcast = vi.fn(); + const leader = startCall( + "sessions.catalog.list", + { progressId: "leader-progress" }, + config, + { connId: "leader" }, + { broadcastToConnIds: leaderBroadcast }, + ); + const follower = startCall( + "sessions.catalog.list", + { progressId: "follower-progress" }, + config, + { connId: "follower" }, + { broadcastToConnIds: followerBroadcast }, + ); + const otherAgent = startCall("sessions.catalog.list", { agentId: "research" }, config); + const otherParams = startCall("sessions.catalog.list", { search: "other" }, config); + + await vi.waitFor(() => expect(list).toHaveBeenCalledTimes(3)); + release(); + await Promise.all([ + leader.completion, + follower.completion, + otherAgent.completion, + otherParams.completion, + ]); + + expect(leaderBroadcast).toHaveBeenCalledOnce(); + expect(followerBroadcast).not.toHaveBeenCalled(); + for (const pending of [leader, follower, otherAgent, otherParams]) { + expect(pending.respond).toHaveBeenCalledWith(true, { + catalogs: [expect.objectContaining({ id: "codex", hosts: [host] })], + }); + } + }); + it("projects authoritative creator ownership onto streamed and final catalog rows", async () => { const broadcastToConnIds = vi.fn(); const host = { @@ -559,21 +629,23 @@ describe("session catalog Gateway methods", () => { }); }); - it("refreshes a provider's core new-session target when listing", async () => { + it("memoizes a provider's create target until runtime config identity changes", async () => { let createSession: { model: string; agentRuntime: string } | undefined = { model: "anthropic/claude-opus-4-8", agentRuntime: "claude-cli", }; + const resolveCreateSession = vi.fn(() => createSession); hoisted.activeRegistry.sessionCatalogs = [ { pluginId: "anthropic", provider: provider("claude", { - resolveCreateSession: () => createSession, + resolveCreateSession, }), }, ]; + const config = {}; - const respond = await call("sessions.catalog.list", {}); + const respond = await call("sessions.catalog.list", {}, config); expect(respond).toHaveBeenCalledWith(true, { catalogs: [ @@ -589,7 +661,19 @@ describe("session catalog Gateway methods", () => { }); createSession = undefined; - const refreshed = await call("sessions.catalog.list", {}); + const cached = await call("sessions.catalog.list", {}, config); + expect(cached).toHaveBeenCalledWith(true, { + catalogs: [ + expect.objectContaining({ + capabilities: expect.objectContaining({ + createSession: { model: "anthropic/claude-opus-4-8" }, + }), + }), + ], + }); + expect(resolveCreateSession).toHaveBeenCalledOnce(); + + const refreshed = await call("sessions.catalog.list", {}, {}); expect(refreshed).toHaveBeenCalledWith(true, { catalogs: [ expect.objectContaining({ @@ -601,6 +685,43 @@ describe("session catalog Gateway methods", () => { }), ], }); + expect(resolveCreateSession).toHaveBeenCalledTimes(2); + }); + + it("retries an exception-derived create target failure without a config reload", async () => { + const resolveCreateSession = vi + .fn() + .mockImplementationOnce(() => { + throw new Error("provider warming"); + }) + .mockReturnValue({ + model: "anthropic/claude-opus-4-8", + agentRuntime: "claude-cli", + }); + hoisted.activeRegistry.sessionCatalogs = [ + { provider: provider("claude", { resolveCreateSession }) }, + ]; + const config = {}; + + const unavailable = await call("sessions.catalog.list", {}, config); + expect(unavailable).toHaveBeenCalledWith(true, { + catalogs: [ + expect.objectContaining({ + capabilities: { continueSession: false, archive: false }, + }), + ], + }); + const recovered = await call("sessions.catalog.list", {}, config); + expect(recovered).toHaveBeenCalledWith(true, { + catalogs: [ + expect.objectContaining({ + capabilities: expect.objectContaining({ + createSession: { model: "anthropic/claude-opus-4-8" }, + }), + }), + ], + }); + expect(resolveCreateSession).toHaveBeenCalledTimes(2); }); it("keeps creation available when catalog history listing fails", async () => { @@ -683,7 +804,7 @@ describe("session catalog Gateway methods", () => { }, ]; - expect(resolveSessionCatalogCreateTarget("claude", "research")).toEqual({ + expect(resolveSessionCatalogCreateTarget("claude", "research", {})).toEqual({ ok: true, target: { model: "anthropic/claude-opus-4-8", @@ -691,7 +812,7 @@ describe("session catalog Gateway methods", () => { pluginOwnerId: "anthropic", }, }); - expect(resolveSessionCatalogCreateTarget("missing", "research")).toEqual({ + expect(resolveSessionCatalogCreateTarget("missing", "research", {})).toEqual({ ok: false, message: "unknown session catalog: missing", unknownCatalog: true, diff --git a/src/gateway/server-methods/session-catalog.ts b/src/gateway/server-methods/session-catalog.ts index 11129244d08c..bc14734d7b08 100644 --- a/src/gateway/server-methods/session-catalog.ts +++ b/src/gateway/server-methods/session-catalog.ts @@ -13,6 +13,7 @@ import { validateSessionsCatalogListParams, validateSessionsCatalogReadParams, } from "../../../packages/gateway-protocol/src/index.js"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { getActivePluginSessionExtensionRegistry } from "../../plugins/runtime.js"; import { gatewaySubagentState } from "../../plugins/runtime/gateway-bindings.js"; import type { @@ -87,26 +88,68 @@ type ProviderCreateTargetResolution = | { ok: true; target: SessionCatalogCreateTarget } | { ok: false; message: string }; +const providerCreateTargetsByConfig = new WeakMap< + OpenClawConfig, + WeakMap> +>(); + +const catalogListsByConfig = new WeakMap< + OpenClawConfig, + Map> +>(); + +function providerCreateTargetCache( + config: OpenClawConfig, + provider: SessionCatalogProvider, +): Map { + let byProvider = providerCreateTargetsByConfig.get(config); + if (!byProvider) { + byProvider = new WeakMap(); + providerCreateTargetsByConfig.set(config, byProvider); + } + let byAgent = byProvider.get(provider); + if (!byAgent) { + byAgent = new Map(); + byProvider.set(provider, byAgent); + } + return byAgent; +} + function resolveProviderCreateTarget( provider: SessionCatalogProvider, - agentId?: string, + agentId: string, + config: OpenClawConfig, ): ProviderCreateTargetResolution { + const cache = providerCreateTargetCache(config, provider); + const cached = cache.get(agentId); + if (cached) { + // The provider contract makes create targets config-derived. A reload changes config identity; + // retaining the old target would advertise a model no longer allowed. + return cached; + } + let resolution: ProviderCreateTargetResolution; try { const target = provider.resolveCreateSession?.({ agentId }); const model = target?.model.trim(); const agentRuntime = target?.agentRuntime.trim(); - return model && agentRuntime - ? { ok: true, target: { model, agentRuntime } } - : { ok: false, message: `session catalog ${provider.id} cannot create sessions` }; + resolution = + model && agentRuntime + ? { ok: true, target: { model, agentRuntime } } + : { ok: false, message: `session catalog ${provider.id} cannot create sessions` }; } catch (error) { + // Resolver exceptions are not config state. Retry them on the next request so a transient + // provider initialization failure cannot suppress session creation until config reload. return { ok: false, message: catalogError(error).message }; } + cache.set(agentId, resolution); + return resolution; } /** Resolves a catalog-owned create target at the start of sessions.create. */ export function resolveSessionCatalogCreateTarget( catalogId: string, agentId: string, + config: OpenClawConfig, ): SessionCatalogCreateTargetResolution { const registration = registrations().find((entry) => entry.provider.id === catalogId); if (!registration) { @@ -116,12 +159,43 @@ export function resolveSessionCatalogCreateTarget( unknownCatalog: true, }; } - const resolved = resolveProviderCreateTarget(registration.provider, agentId); + const resolved = resolveProviderCreateTarget(registration.provider, agentId, config); return resolved.ok ? { ok: true, target: { ...resolved.target, pluginOwnerId: registration.pluginId } } : resolved; } +function sessionCatalogListKey(params: { + agentId: string; + request: SessionsCatalogListParams; + search?: string; +}): string { + const cursors = params.request.cursors + ? Object.entries(params.request.cursors).toSorted(([left], [right]) => + left.localeCompare(right), + ) + : null; + return JSON.stringify([ + params.agentId, + params.request.catalogId ?? null, + params.search ?? null, + params.request.limitPerHost ?? null, + params.request.hostIds ?? null, + cursors, + ]); +} + +function catalogListInflightMap( + config: OpenClawConfig, +): Map> { + let inFlight = catalogListsByConfig.get(config); + if (!inFlight) { + inFlight = new Map(); + catalogListsByConfig.set(config, inFlight); + } + return inFlight; +} + function providerOrRespond( catalogId: string, respond: RespondFn, @@ -216,58 +290,82 @@ export const sessionCatalogHandlers: GatewayRequestHandlers = { const search = normalizeSessionCatalogSearch(request.search); const progressId = request.progressId; const progressConnId = progressId && client?.connId ? client.connId : undefined; - const requestEntries = createSessionCatalogRequestEntrySnapshot({ - cfg: config, - fallbackAgentId: resolvedAgent.agentId, + const listKey = sessionCatalogListKey({ + agentId: resolvedAgent.agentId, + request, + search, }); - const listNodes = createSessionCatalogRequestNodeSnapshot(); - const catalogList = await Promise.all( - selected.map(async (provider): Promise => { - const createTarget = resolveProviderCreateTarget(provider, resolvedAgent.agentId); - const createSession = createTarget.ok ? { model: createTarget.target.model } : undefined; - const onHost = progressConnId - ? (host: SessionCatalog["hosts"][number]) => { - // Progressive frames are an optimization. The final RPC response remains - // authoritative when a slow client drops an intermediate host update. - context.broadcastToConnIds( - "sessions.catalog.host", - { - progressId, - agentId: resolvedAgent.agentId, - catalog: catalogResult( - provider, - [requestEntries.projectHostCreatedActors(host)], - undefined, - createSession, - ), - }, - new Set([progressConnId]), - { dropIfSlow: true }, - ); - } - : undefined; - try { - const hosts = await provider.list({ - search, - limitPerHost: request.limitPerHost, - hostIds: request.hostIds, - ...(request.cursors !== undefined ? { cursors: request.cursors } : {}), - sessionEntries: requestEntries.sessionEntries, - listNodes, - ...(onHost ? { onHost } : {}), - }); - return catalogResult( - provider, - hosts.map(requestEntries.projectHostCreatedActors), - undefined, - createSession, - ); - } catch (error) { - return catalogResult(provider, [], catalogError(error), createSession); - } - }), - ); - respond(true, { catalogs: catalogList }); + const inFlight = catalogListInflightMap(config); + const pending = inFlight.get(listKey); + if (pending) { + // progressId is connection-owned and excluded from the work key. Followers skip progressive + // frames and receive only the authoritative final result emitted for every caller below. + respond(true, await pending); + return; + } + const operation = (async () => { + const requestEntries = createSessionCatalogRequestEntrySnapshot({ + cfg: config, + fallbackAgentId: resolvedAgent.agentId, + }); + const listNodes = createSessionCatalogRequestNodeSnapshot(); + const catalogList = await Promise.all( + selected.map(async (provider): Promise => { + const createTarget = resolveProviderCreateTarget(provider, resolvedAgent.agentId, config); + const createSession = createTarget.ok ? { model: createTarget.target.model } : undefined; + const onHost = progressConnId + ? (host: SessionCatalog["hosts"][number]) => { + // Progressive frames are an optimization. The final RPC response remains + // authoritative when a slow client drops an intermediate host update. + context.broadcastToConnIds( + "sessions.catalog.host", + { + progressId, + agentId: resolvedAgent.agentId, + catalog: catalogResult( + provider, + [requestEntries.projectHostCreatedActors(host)], + undefined, + createSession, + ), + }, + new Set([progressConnId]), + { dropIfSlow: true }, + ); + } + : undefined; + try { + const hosts = await provider.list({ + search, + limitPerHost: request.limitPerHost, + hostIds: request.hostIds, + ...(request.cursors !== undefined ? { cursors: request.cursors } : {}), + sessionEntries: requestEntries.sessionEntries, + listNodes, + ...(onHost ? { onHost } : {}), + }); + return catalogResult( + provider, + hosts.map(requestEntries.projectHostCreatedActors), + undefined, + createSession, + ); + } catch (error) { + return catalogResult(provider, [], catalogError(error), createSession); + } + }), + ); + return { catalogs: catalogList }; + })(); + // Sharing ends when this exact promise settles; later polls always execute against fresh state. + inFlight.set(listKey, operation); + try { + respond(true, await operation); + } finally { + if (inFlight.get(listKey) === operation) { + inFlight.delete(listKey); + } + } }, "sessions.catalog.read": async ({ params, respond }) => { diff --git a/src/gateway/server-methods/sessions-create.ts b/src/gateway/server-methods/sessions-create.ts index 9c65b2b60151..7d86e6019ca2 100644 --- a/src/gateway/server-methods/sessions-create.ts +++ b/src/gateway/server-methods/sessions-create.ts @@ -87,7 +87,7 @@ export const sessionCreateHandlers: GatewayRequestHandlers = { } const catalogTarget = catalogId && catalogAgentId - ? resolveSessionCatalogCreateTarget(catalogId, catalogAgentId) + ? resolveSessionCatalogCreateTarget(catalogId, catalogAgentId, cfg) : undefined; if (catalogTarget && !catalogTarget.ok) { respond( diff --git a/src/infra/executable-path.test.ts b/src/infra/executable-path.test.ts index 59d866ddf513..72af5886709b 100644 --- a/src/infra/executable-path.test.ts +++ b/src/infra/executable-path.test.ts @@ -1,10 +1,12 @@ // Covers executable path detection and PATH lookup helpers. +import nodeFs from "node:fs"; import fs from "node:fs/promises"; import path from "node:path"; -import { afterEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { withTempDir } from "../test-helpers/temp-dir.js"; import { withMockedPlatform } from "../test-utils/vitest-spies.js"; import { + clearExecutablePathCache, isRegularFile, resolveExecutable, resolveExecutableFromPathEnv, @@ -12,6 +14,10 @@ import { resolveExecutablePathCandidate, } from "./executable-path.js"; +beforeEach(() => { + clearExecutablePathCache(); +}); + function restoreEnvValue(name: string, value: string | undefined): void { if (value === undefined) { delete process.env[name]; @@ -62,6 +68,54 @@ describe("executable path helpers", () => { }); }); + it("memoizes PATH hits and misses until explicit invalidation", async () => { + await withTempDir({ prefix: "openclaw-exec-path-" }, async (base) => { + const binDir = path.join(base, "bin"); + await fs.mkdir(binDir); + const executable = path.join(binDir, "runner"); + await fs.writeFile(executable, "#!/bin/sh\nexit 0\n", { mode: 0o755 }); + const statSpy = vi.spyOn(nodeFs, "statSync"); + + expect(resolveExecutableFromPathEnv("runner", binDir)).toBe(executable); + const hitProbeCount = statSpy.mock.calls.length; + expect(resolveExecutableFromPathEnv("runner", binDir)).toBe(executable); + expect(statSpy).toHaveBeenCalledTimes(hitProbeCount); + + expect(resolveExecutableFromPathEnv("missing", binDir)).toBeUndefined(); + const missProbeCount = statSpy.mock.calls.length; + expect(resolveExecutableFromPathEnv("missing", binDir)).toBeUndefined(); + expect(statSpy).toHaveBeenCalledTimes(missProbeCount); + + clearExecutablePathCache(); + expect(resolveExecutableFromPathEnv("runner", binDir)).toBe(executable); + expect(resolveExecutableFromPathEnv("missing", binDir)).toBeUndefined(); + expect(statSpy.mock.calls.length).toBeGreaterThan(missProbeCount); + }); + }); + + it("does not reuse relative PATH probes after cwd changes", async () => { + await withTempDir({ prefix: "openclaw-exec-path-" }, async (base) => { + const firstCwd = path.join(base, "first"); + const secondCwd = path.join(base, "second"); + const relativeBin = "bin"; + await fs.mkdir(path.join(firstCwd, relativeBin), { recursive: true }); + await fs.mkdir(secondCwd); + const executable = path.join(firstCwd, relativeBin, "runner"); + await fs.writeFile(executable, "#!/bin/sh\nexit 0\n", { mode: 0o755 }); + const originalCwd = process.cwd(); + try { + process.chdir(firstCwd); + expect(resolveExecutableFromPathEnv("runner", relativeBin)).toBe( + path.join(relativeBin, "runner"), + ); + process.chdir(secondCwd); + expect(resolveExecutableFromPathEnv("runner", relativeBin)).toBeUndefined(); + } finally { + process.chdir(originalCwd); + } + }); + }); + it("resolves absolute, home-relative, and Path-cased env executables", async () => { await withTempDir({ prefix: "openclaw-exec-path-" }, async (base) => { const homeDir = path.join(base, "home"); diff --git a/src/infra/executable-path.ts b/src/infra/executable-path.ts index dc67062e311d..244786008670 100644 --- a/src/infra/executable-path.ts +++ b/src/infra/executable-path.ts @@ -112,6 +112,54 @@ function isExecutableFile(filePath: string, options?: { env?: NodeJS.ProcessEnv } const WINDOWS_NATIVE_EXECUTABLE_EXTENSIONS = new Set([".com", ".exe", ".bat", ".cmd"]); +const EXECUTABLE_PATH_CACHE_TTL_MS = 60_000; +const EXECUTABLE_PATH_CACHE_MAX_ENTRIES = 128; + +type ExecutablePathCacheEntry = { + expiresAt: number; + resolved: string | null; +}; + +const executablePathCache = new Map(); + +function cacheExecutablePath(key: string, resolved: string | undefined): void { + executablePathCache.set(key, { + expiresAt: Date.now() + EXECUTABLE_PATH_CACHE_TTL_MS, + resolved: resolved ?? null, + }); + while (executablePathCache.size > EXECUTABLE_PATH_CACHE_MAX_ENTRIES) { + const oldest = executablePathCache.keys().next(); + if (oldest.done) { + break; + } + executablePathCache.delete(oldest.value); + } +} + +function executablePathCacheKey( + executable: string, + pathEnv: string, + env: NodeJS.ProcessEnv | undefined, + includeExtensionless: boolean | undefined, +): string { + const pathExt = + resolveEnvironmentValue(env, "PATHEXT") ?? + resolveEnvironmentValue(process.env, "PATHEXT") ?? + ""; + let cwd = ""; + try { + // Relative PATH entries resolve against cwd, so changing directories must invalidate them. + cwd = process.cwd(); + } catch { + // A deleted cwd already makes relative probes fail; keep the cache key stable for that state. + } + return `${process.platform}\0${executable}\0${pathEnv}\0${pathExt}\0${includeExtensionless !== false}\0${cwd}`; +} + +/** Clears process-local PATH probe results after the runtime environment changes. */ +export function clearExecutablePathCache(): void { + executablePathCache.clear(); +} export function resolveExecutableFromPathEnv( executable: string, @@ -119,6 +167,18 @@ export function resolveExecutableFromPathEnv( env?: NodeJS.ProcessEnv, options?: { includeExtensionless?: boolean }, ): string | undefined { + const cacheKey = executablePathCacheKey(executable, pathEnv, env, options?.includeExtensionless); + const cached = executablePathCache.get(cacheKey); + if (cached && cached.expiresAt > Date.now()) { + // PATH probes synchronously stat every candidate. Reuse hits and misses until config reload or + // this short TTL expires; otherwise catalog polling repeatedly blocks the Gateway event loop. + executablePathCache.delete(cacheKey); + executablePathCache.set(cacheKey, cached); + return cached.resolved ?? undefined; + } + if (cached) { + executablePathCache.delete(cacheKey); + } const delimiter = process.platform === "win32" ? ";" : path.delimiter; const entries = pathEnv.split(delimiter).filter(Boolean); const extensions = resolveWindowsExecutableExtensions( @@ -137,10 +197,12 @@ export function resolveExecutableFromPathEnv( if ( hasNativeWindowsExtension ? isRegularFile(candidate) : isExecutableFile(candidate, { env }) ) { + cacheExecutablePath(cacheKey, candidate); return candidate; } } } + cacheExecutablePath(cacheKey, undefined); return undefined; } diff --git a/src/plugins/session-catalog.ts b/src/plugins/session-catalog.ts index 912de7c44dcf..daf656a07253 100644 --- a/src/plugins/session-catalog.ts +++ b/src/plugins/session-catalog.ts @@ -137,7 +137,7 @@ type SessionCatalogCreateParams = { export type SessionCatalogProvider = { id: string; label: string; - /** Resolves the current core new-session target for the requested agent. */ + /** Config-derived target; the Gateway memoizes it for one runtime-config object identity. */ resolveCreateSession?: ( params: SessionCatalogCreateParams, ) => SessionCatalogCreateTarget | undefined;