From 05fb55eb7b555ec60214c86d2b07935814db0e90 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 13 Jul 2026 21:45:53 -0700 Subject: [PATCH] feat: discover OpenCode and Pi sessions on paired nodes (#106941) * feat(sessions): add opencode and pi catalogs * fix(sessions): satisfy catalog lint * fix(sessions): validate pi storage paths * fix(sessions): index growing pi transcripts * refactor(sessions): split pi path resolution * docs(sessions): refresh catalog map * refactor(sessions): privatize pi catalog internals * fix(sessions): validate node catalog payloads * fix(sessions): harden external catalog discovery * fix(sessions): normalize catalog transcript data * refactor(gateway): keep node scope gate compact * fix(opencode): reject option-like session ids * fix(sessions): align external id validation --- docs/docs_map.md | 3 + docs/nodes/index.md | 15 + docs/plugins/reference/acpx.md | 14 +- docs/plugins/reference/opencode.md | 18 +- extensions/acpx/index.ts | 2 + extensions/acpx/openclaw.plugin.json | 14 + extensions/acpx/src/config-schema.ts | 6 + .../acpx/src/pi-session-catalog-plugin.ts | 355 +++++++++ .../acpx/src/pi-session-catalog.test.ts | 723 ++++++++++++++++++ extensions/acpx/src/pi-session-catalog.ts | 373 +++++++++ extensions/acpx/src/pi-session-paths.ts | 97 +++ extensions/acpx/src/pi-session-store.ts | 489 ++++++++++++ extensions/opencode/index.ts | 2 + extensions/opencode/openclaw.plugin.json | 21 +- extensions/opencode/session-catalog-plugin.ts | 388 ++++++++++ extensions/opencode/session-catalog.test.ts | 406 ++++++++++ extensions/opencode/session-catalog.ts | 457 +++++++++++ src/gateway/server-plugins.test.ts | 28 + src/gateway/server-plugins.ts | 13 +- .../e2e/external-session-catalogs.e2e.test.ts | 136 ++++ 20 files changed, 3546 insertions(+), 14 deletions(-) create mode 100644 extensions/acpx/src/pi-session-catalog-plugin.ts create mode 100644 extensions/acpx/src/pi-session-catalog.test.ts create mode 100644 extensions/acpx/src/pi-session-catalog.ts create mode 100644 extensions/acpx/src/pi-session-paths.ts create mode 100644 extensions/acpx/src/pi-session-store.ts create mode 100644 extensions/opencode/session-catalog-plugin.ts create mode 100644 extensions/opencode/session-catalog.test.ts create mode 100644 extensions/opencode/session-catalog.ts create mode 100644 ui/src/e2e/external-session-catalogs.e2e.test.ts diff --git a/docs/docs_map.md b/docs/docs_map.md index a8d2f3add2c1..00ea58485070 100644 --- a/docs/docs_map.md +++ b/docs/docs_map.md @@ -4884,6 +4884,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H3: Local model inference - H3: Codex sessions and transcripts - H3: Claude sessions and transcripts + - H3: OpenCode and Pi sessions - H2: Invoking commands - H2: Command policy - H2: Config (openclaw.json) @@ -6008,6 +6009,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H1: ACPx plugin - H2: Distribution - H2: Surface + - H2: Pi native sessions - H2: Related docs ## plugins/reference/admin-http-rpc.md @@ -6797,6 +6799,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H1: OpenCode plugin - H2: Distribution - H2: Surface + - H2: Native sessions - H2: Related docs ## plugins/reference/openrouter.md diff --git a/docs/nodes/index.md b/docs/nodes/index.md index ad8859897cc3..99394c2ad30e 100644 --- a/docs/nodes/index.md +++ b/docs/nodes/index.md @@ -369,6 +369,21 @@ node does not advertise this command yet, so its rows remain view-only. See [Anthropic: Claude sessions across computers](/providers/anthropic#claude-sessions-across-computers) for the Control UI behavior and storage sources. +### OpenCode and Pi sessions + +The bundled OpenCode and ACPX plugins also discover read-only native session +catalogs on the Gateway and paired nodes. A node advertises +`opencode.sessions.list.v1` / `opencode.sessions.read.v1` when the `opencode` +CLI is installed, and `acpx.pi.sessions.list.v1` / `acpx.pi.sessions.read.v1` +when Pi's session directory exists. Approve the node pairing upgrade when new +commands first appear. + +OpenCode reads through its official CLI JSON/export surface. Pi reads its +documented JSONL session store, including project and global `settings.json` +session directories plus `PI_CODING_AGENT_DIR` and +`PI_CODING_AGENT_SESSION_DIR` overrides. Both catalogs are enabled by default; +turn them off in the Web UI under **Config > Plugins**. + ## Invoking commands Low-level (raw RPC): diff --git a/docs/plugins/reference/acpx.md b/docs/plugins/reference/acpx.md index 11b4c93e52b5..f19a977818fe 100644 --- a/docs/plugins/reference/acpx.md +++ b/docs/plugins/reference/acpx.md @@ -16,7 +16,19 @@ OpenClaw ACP runtime backend with plugin-owned session and transport management. ## Surface -skills +skills; session catalog: pi + +## Pi native sessions + +The bundled runtime auto-detects Pi's session store on the Gateway and paired +nodes. Stored sessions appear in the **Pi** sessions-sidebar group, with +read-only transcript browsing from Pi's documented JSONL session format. The +catalog honors project and global `settings.json` session directories plus +`PI_CODING_AGENT_DIR` and `PI_CODING_AGENT_SESSION_DIR`. Relative paths resolve +from the directory containing their `settings.json` file. + +Turn **Pi Session Catalog** off under **Config > Plugins > ACPX Runtime** to +disable discovery. It is enabled by default. ## Related docs diff --git a/docs/plugins/reference/opencode.md b/docs/plugins/reference/opencode.md index 592968174557..72ab23e7e18e 100644 --- a/docs/plugins/reference/opencode.md +++ b/docs/plugins/reference/opencode.md @@ -1,5 +1,5 @@ --- -summary: "Adds OpenCode model provider support to OpenClaw." +summary: "Adds OpenCode model provider and native session catalog support to OpenClaw." read_when: - You are installing, configuring, or auditing the opencode plugin title: "OpenCode plugin" @@ -7,7 +7,7 @@ title: "OpenCode plugin" # OpenCode plugin -Adds OpenCode model provider support to OpenClaw. +Adds OpenCode model provider and native session catalog support to OpenClaw. ## Distribution @@ -16,7 +16,19 @@ Adds OpenCode model provider support to OpenClaw. ## Surface -providers: opencode; contracts: mediaUnderstandingProviders +providers: opencode; contracts: mediaUnderstandingProviders; session catalog: opencode + +## Native sessions + +OpenClaw auto-detects the `opencode` CLI on the Gateway and paired nodes. Stored +sessions then appear in the **OpenCode** sessions-sidebar group, with read-only +transcript browsing through the official `opencode --pure db ... --format json` +and `opencode --pure export` commands. The restricted environment and `--pure` +mode prevent catalog browsing from loading project plugins or inheriting unrelated +Gateway credentials. + +Turn **OpenCode Session Catalog** off under **Config > Plugins > OpenCode** to +disable discovery. It is enabled by default. ## Related docs diff --git a/extensions/acpx/index.ts b/extensions/acpx/index.ts index 2124f56f272a..4d7f80592bb9 100644 --- a/extensions/acpx/index.ts +++ b/extensions/acpx/index.ts @@ -12,6 +12,7 @@ import type { PluginHookReplyDispatchResult, } from "./runtime-api.js"; import { DEFAULT_ACPX_TIMEOUT_SECONDS } from "./src/config-schema.js"; +import { registerPiSessionCatalog } from "./src/pi-session-catalog-plugin.js"; function resolveReplyDispatchTimeoutMs(pluginConfig?: Record): number { const timeoutSeconds = pluginConfig?.timeoutSeconds; @@ -49,6 +50,7 @@ const plugin = { description: "Embedded ACP runtime backend with plugin-owned session and transport management.", register(api: OpenClawPluginApi) { const replyDispatchTimeoutMs = resolveReplyDispatchTimeoutMs(api.pluginConfig); + registerPiSessionCatalog(api); api.registerService( createAcpxRuntimeService({ pluginConfig: api.pluginConfig, diff --git a/extensions/acpx/openclaw.plugin.json b/extensions/acpx/openclaw.plugin.json index 43c189fbe61f..49111dc30874 100644 --- a/extensions/acpx/openclaw.plugin.json +++ b/extensions/acpx/openclaw.plugin.json @@ -49,6 +49,16 @@ "type": "number", "minimum": 0 }, + "piSessionCatalog": { + "type": "object", + "additionalProperties": false, + "properties": { + "enabled": { + "type": "boolean", + "default": true + } + } + }, "mcpServers": { "type": "object", "additionalProperties": { @@ -134,6 +144,10 @@ "help": "Reserved compatibility field for the older embedded ACPX queue-owner path. Accepted for compatibility and logged as ignored.", "advanced": true }, + "piSessionCatalog.enabled": { + "label": "Pi Session Catalog", + "help": "Auto-detect Pi sessions on the Gateway and paired nodes, then show them in the sessions sidebar." + }, "probeAgent": { "label": "Health Probe Agent", "help": "Agent id used for the embedded ACP runtime health probe. Defaults to the first `acp.allowedAgents` entry when that allowlist is set, otherwise to the runtime built-in probe agent (codex). Set this explicitly (for example `opencode` or `claude`) when the default probe agent is not installed or not authenticated, so the whole embedded ACP backend does not get marked unavailable.", diff --git a/extensions/acpx/src/config-schema.ts b/extensions/acpx/src/config-schema.ts index 5c0aa398caee..865db2d7b240 100644 --- a/extensions/acpx/src/config-schema.ts +++ b/extensions/acpx/src/config-schema.ts @@ -42,6 +42,7 @@ export type AcpxPluginConfig = { strictWindowsCmdWrapper?: boolean; timeoutSeconds?: number; queueOwnerTtlSeconds?: number; + piSessionCatalog?: { enabled?: boolean }; mcpServers?: Record; agents?: Record; }; @@ -117,6 +118,11 @@ export const AcpxPluginConfigSchema = z.strictObject({ .number({ error: "queueOwnerTtlSeconds must be a number >= 0" }) .min(0, { error: "queueOwnerTtlSeconds must be a number >= 0" }) .optional(), + piSessionCatalog: z + .strictObject({ + enabled: z.boolean({ error: "piSessionCatalog.enabled must be a boolean" }).default(true), + }) + .optional(), mcpServers: z.record(z.string(), McpServerConfigSchema).optional(), agents: z .record( diff --git a/extensions/acpx/src/pi-session-catalog-plugin.ts b/extensions/acpx/src/pi-session-catalog-plugin.ts new file mode 100644 index 000000000000..b54e673b24ce --- /dev/null +++ b/extensions/acpx/src/pi-session-catalog-plugin.ts @@ -0,0 +1,355 @@ +import process from "node:process"; +import type { + OpenClawPluginApi, + OpenClawPluginNodeHostCommand, + OpenClawPluginNodeInvokePolicy, +} from "openclaw/plugin-sdk/plugin-entry"; +import type { PluginRuntime } from "openclaw/plugin-sdk/plugin-runtime"; +import type { + SessionCatalogHost, + SessionCatalogProvider, + SessionCatalogSession, + SessionCatalogTranscriptItem, + SessionsCatalogReadResult, +} from "openclaw/plugin-sdk/session-catalog"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { + listLocalPiSessionPage, + optionalPiString, + readLocalPiTranscriptPage, + type PiSessionPage, +} from "./pi-session-catalog.js"; +import { piSessionStoreAvailable } from "./pi-session-paths.js"; + +const PI_SESSIONS_LIST_COMMAND = "acpx.pi.sessions.list.v1"; +const PI_SESSION_READ_COMMAND = "acpx.pi.sessions.read.v1"; + +const CAPABILITY = "pi-sessions"; +const LOCAL_HOST_ID = "gateway"; +const MAX_PAGE_LIMIT = 100; +const MAX_HOSTS = 100; +const MAX_CURSOR_LENGTH = 128; +const MAX_SEARCH_LENGTH = 500; +const NODE_TIMEOUT_MS = 20_000; +const SESSION_ID_PATTERN = /^(?!-)[A-Za-z0-9._:-]{1,256}$/u; +const TRANSCRIPT_ITEM_TYPES = new Set([ + "userMessage", + "agentMessage", + "reasoning", + "toolCall", + "toolResult", + "other", +]); + +function isOptionalString(value: unknown): boolean { + return value === undefined || typeof value === "string"; +} + +function isOptionalNumber(value: unknown): boolean { + return value === undefined || typeof value === "number"; +} + +function isNodeSession(value: unknown): value is SessionCatalogSession { + return ( + isRecord(value) && + typeof value.threadId === "string" && + SESSION_ID_PATTERN.test(value.threadId) && + typeof value.status === "string" && + value.status.length > 0 && + typeof value.archived === "boolean" && + typeof value.canContinue === "boolean" && + typeof value.canArchive === "boolean" && + isOptionalString(value.name) && + isOptionalString(value.cwd) && + isOptionalString(value.source) && + isOptionalString(value.modelProvider) && + isOptionalString(value.cliVersion) && + isOptionalString(value.gitBranch) && + isOptionalString(value.openClawSessionKey) && + isOptionalNumber(value.createdAt) && + isOptionalNumber(value.updatedAt) && + isOptionalNumber(value.recencyAt) + ); +} + +function isNodeTranscriptItem(value: unknown): value is SessionCatalogTranscriptItem { + return ( + isRecord(value) && + typeof value.type === "string" && + TRANSCRIPT_ITEM_TYPES.has(value.type) && + isOptionalString(value.id) && + isOptionalString(value.text) && + isOptionalString(value.timestamp) && + isOptionalString(value.model) && + (value.truncated === undefined || typeof value.truncated === "boolean") + ); +} + +function parseNodeParams(paramsJSON?: string | null): unknown { + if (!paramsJSON) { + return undefined; + } + try { + return JSON.parse(paramsJSON) as unknown; + } catch (error) { + throw new Error("Pi session parameters must be valid JSON", { cause: error }); + } +} + +function fullConfigCatalogEnabled(config: unknown): boolean { + if (!isRecord(config) || !isRecord(config.plugins) || !isRecord(config.plugins.entries)) { + return true; + } + const entry = config.plugins.entries.acpx; + if (!isRecord(entry) || !isRecord(entry.config) || !isRecord(entry.config.piSessionCatalog)) { + return true; + } + return entry.config.piSessionCatalog.enabled !== false; +} + +function isPiSessionCatalogEnabled(pluginConfig: unknown): boolean { + return ( + !isRecord(pluginConfig) || + !isRecord(pluginConfig.piSessionCatalog) || + pluginConfig.piSessionCatalog.enabled !== false + ); +} + +function createPiSessionNodeHostCommands(): OpenClawPluginNodeHostCommand[] { + const available = ({ config, env }: { config: unknown; env: NodeJS.ProcessEnv }) => + fullConfigCatalogEnabled(config) && piSessionStoreAvailable(env); + return [ + { + command: PI_SESSIONS_LIST_COMMAND, + cap: CAPABILITY, + dangerous: false, + isAvailable: available, + handle: async (paramsJSON) => + JSON.stringify(await listLocalPiSessionPage(parseNodeParams(paramsJSON))), + }, + { + command: PI_SESSION_READ_COMMAND, + cap: CAPABILITY, + dangerous: false, + isAvailable: available, + handle: async (paramsJSON) => + JSON.stringify(await readLocalPiTranscriptPage(parseNodeParams(paramsJSON))), + }, + ]; +} + +function createPiSessionNodeInvokePolicies(): OpenClawPluginNodeInvokePolicy[] { + return [ + { + commands: [PI_SESSIONS_LIST_COMMAND, PI_SESSION_READ_COMMAND], + defaultPlatforms: ["macos", "linux", "windows"], + handle: (context) => context.invokeNode(), + }, + ]; +} + +function nodeLabel(node: { displayName?: string; remoteIp?: string; nodeId: string }): string { + return node.displayName?.trim() || node.remoteIp?.trim() || node.nodeId; +} + +function unwrapNodePayload(value: unknown): unknown { + return isRecord(value) && typeof value.payloadJSON === "string" + ? (JSON.parse(value.payloadJSON) as unknown) + : value; +} + +type CatalogNode = Awaited>["nodes"][number]; + +async function listPiNodeHost( + runtime: PluginRuntime, + query: Parameters[0], + node: CatalogNode, +): Promise { + const hostId = `node:${node.nodeId}`; + const common = { + hostId, + label: nodeLabel(node), + kind: "node" as const, + connected: node.connected === true, + nodeId: node.nodeId, + }; + if (node.connected !== true) { + return { + ...common, + sessions: [], + error: { code: "NODE_OFFLINE", message: "Paired node is offline" }, + }; + } + try { + const raw = await runtime.nodes.invoke({ + nodeId: node.nodeId, + command: PI_SESSIONS_LIST_COMMAND, + params: { + ...(query.limitPerHost ? { limit: query.limitPerHost } : {}), + ...(query.search?.trim() + ? { searchTerm: query.search.trim().slice(0, MAX_SEARCH_LENGTH) } + : {}), + ...(query.cursors?.[hostId] ? { cursor: query.cursors[hostId] } : {}), + }, + timeoutMs: NODE_TIMEOUT_MS, + scopes: ["operator.write"], + }); + return { ...common, ...parseNodeSessionPage(unwrapNodePayload(raw)) }; + } catch { + return { + ...common, + sessions: [], + error: { code: "NODE_INVOKE_FAILED", message: "Paired node Pi sessions are unavailable" }, + }; + } +} + +function parseNodeSessionPage(value: unknown): PiSessionPage { + if ( + !isRecord(value) || + !Array.isArray(value.sessions) || + value.sessions.length > MAX_PAGE_LIMIT + ) { + throw new Error("Pi node returned an invalid session page"); + } + if (!value.sessions.every(isNodeSession)) { + throw new Error("Pi node returned an invalid session page"); + } + const sessions = value.sessions; + const nextCursor = optionalPiString(value.nextCursor, MAX_CURSOR_LENGTH); + if (value.nextCursor !== undefined && !nextCursor) { + throw new Error("Pi node returned an invalid cursor"); + } + return { sessions, ...(nextCursor ? { nextCursor } : {}) }; +} + +function parseNodeTranscriptPage(value: unknown, threadId: string): SessionsCatalogReadResult { + if ( + !isRecord(value) || + value.threadId !== threadId || + !Array.isArray(value.items) || + value.items.length > MAX_PAGE_LIMIT || + !value.items.every(isNodeTranscriptItem) + ) { + throw new Error("Pi node returned an invalid transcript page"); + } + const nextCursor = optionalPiString(value.nextCursor, MAX_CURSOR_LENGTH); + if (value.nextCursor !== undefined && !nextCursor) { + throw new Error("Pi node returned an invalid cursor"); + } + return { + hostId: LOCAL_HOST_ID, + threadId, + items: value.items, + ...(nextCursor ? { nextCursor } : {}), + }; +} + +async function listPiHosts( + runtime: PluginRuntime, + query: Parameters[0], +): Promise { + const requested = query.hostIds ? new Set(query.hostIds) : undefined; + const searchTerm = query.search?.trim().slice(0, MAX_SEARCH_LENGTH) || undefined; + const hosts: SessionCatalogHost[] = []; + if ((!requested || requested.has(LOCAL_HOST_ID)) && piSessionStoreAvailable(process.env)) { + try { + hosts.push({ + hostId: LOCAL_HOST_ID, + label: "Local Pi", + kind: "gateway", + connected: true, + ...(await listLocalPiSessionPage({ + limit: query.limitPerHost, + ...(searchTerm ? { searchTerm } : {}), + cursor: query.cursors?.[LOCAL_HOST_ID], + })), + }); + } catch { + hosts.push({ + hostId: LOCAL_HOST_ID, + label: "Local Pi", + kind: "gateway", + connected: true, + sessions: [], + error: { code: "LOCAL_READ_FAILED", message: "Local Pi sessions are unavailable" }, + }); + } + } + let nodes: Awaited>["nodes"]; + try { + nodes = (await runtime.nodes.list()).nodes; + } catch { + return hosts; + } + const eligible = nodes + .filter( + (node) => + node.commands?.includes(PI_SESSIONS_LIST_COMMAND) && + (!requested || requested.has(`node:${node.nodeId}`)), + ) + .toSorted((left, right) => nodeLabel(left).localeCompare(nodeLabel(right))) + .slice(0, MAX_HOSTS - hosts.length); + const nodeHosts = await Promise.all(eligible.map((node) => listPiNodeHost(runtime, query, node))); + return [...hosts, ...nodeHosts]; +} + +async function readPiTranscript( + runtime: PluginRuntime, + request: Parameters[0], +): Promise { + if (request.hostId === LOCAL_HOST_ID) { + return await readLocalPiTranscriptPage({ + threadId: request.threadId, + ...(request.limit ? { limit: request.limit } : {}), + ...(request.cursor ? { cursor: request.cursor } : {}), + }); + } + if (!request.hostId.startsWith("node:")) { + throw new Error("hostId is invalid"); + } + const nodeId = request.hostId.slice("node:".length); + const node = (await runtime.nodes.list()).nodes.find( + (candidate) => + candidate.nodeId === nodeId && + candidate.connected === true && + candidate.commands?.includes(PI_SESSION_READ_COMMAND), + ); + if (!node) { + throw new Error("paired-node Pi session host is unavailable"); + } + const raw = await runtime.nodes.invoke({ + nodeId, + command: PI_SESSION_READ_COMMAND, + params: { + threadId: request.threadId, + ...(request.limit ? { limit: request.limit } : {}), + ...(request.cursor ? { cursor: request.cursor } : {}), + }, + timeoutMs: NODE_TIMEOUT_MS, + scopes: ["operator.write"], + }); + return { + ...parseNodeTranscriptPage(unwrapNodePayload(raw), request.threadId), + hostId: request.hostId, + label: nodeLabel(node), + }; +} + +export function registerPiSessionCatalog(api: OpenClawPluginApi): void { + if (!isPiSessionCatalogEnabled(api.pluginConfig)) { + return; + } + api.registerSessionCatalog({ + id: "pi", + label: "Pi", + list: async (query) => await listPiHosts(api.runtime, query), + read: async (request) => await readPiTranscript(api.runtime, request), + }); + for (const command of createPiSessionNodeHostCommands()) { + api.registerNodeHostCommand(command); + } + for (const policy of createPiSessionNodeInvokePolicies()) { + api.registerNodeInvokePolicy(policy); + } +} diff --git a/extensions/acpx/src/pi-session-catalog.test.ts b/extensions/acpx/src/pi-session-catalog.test.ts new file mode 100644 index 000000000000..b2eafc0dace4 --- /dev/null +++ b/extensions/acpx/src/pi-session-catalog.test.ts @@ -0,0 +1,723 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { registerPiSessionCatalog } from "./pi-session-catalog-plugin.js"; +import { listLocalPiSessionPage, readLocalPiTranscriptPage } from "./pi-session-catalog.js"; +import { piSessionStore } from "./pi-session-paths.js"; + +const PI_SESSIONS_LIST_COMMAND = "acpx.pi.sessions.list.v1"; +const PI_SESSION_READ_COMMAND = "acpx.pi.sessions.read.v1"; + +const temporaryDirectories: string[] = []; +const originalSessionDir = process.env.PI_CODING_AGENT_SESSION_DIR; +const originalAgentDir = process.env.PI_CODING_AGENT_DIR; +const originalHome = process.env.HOME; +const originalUserProfile = process.env.USERPROFILE; + +async function createPiStore(assistantText = "hi"): Promise { + const directory = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-pi-catalog-")); + temporaryDirectories.push(directory); + process.env.PI_CODING_AGENT_SESSION_DIR = directory; + const entries = [ + { + type: "session", + version: 3, + id: "pi-session", + timestamp: "2026-07-13T10:00:00.000Z", + cwd: "/workspace", + }, + { + type: "message", + id: "user-1", + parentId: null, + timestamp: "2026-07-13T10:00:01.000Z", + message: { role: "user", content: "hello", timestamp: 1_783_938_001_000 }, + }, + { + type: "message", + id: "assistant-1", + parentId: "user-1", + timestamp: "2026-07-13T10:00:02.000Z", + message: { + role: "assistant", + provider: "anthropic", + model: "claude", + timestamp: 1_783_938_002_000, + content: [ + { type: "thinking", thinking: "thinking" }, + { type: "text", text: assistantText }, + { type: "toolCall", id: "call-1", name: "bash", arguments: { command: "pwd" } }, + ], + }, + }, + { + type: "message", + id: "tool-1", + parentId: "assistant-1", + timestamp: "2026-07-13T10:00:03.000Z", + message: { + role: "toolResult", + toolCallId: "call-1", + toolName: "bash", + timestamp: 1_783_938_003_000, + content: [{ type: "text", text: "/workspace" }], + }, + }, + { + type: "session_info", + id: "info-1", + parentId: "tool-1", + timestamp: "2026-07-13T10:00:04.000Z", + name: "Pi catalog session", + }, + ]; + await fs.writeFile( + path.join(directory, "session.jsonl"), + `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`, + ); + return directory; +} + +function registerPiNodeHostCommands(): Parameters< + OpenClawPluginApi["registerNodeHostCommand"] +>[0][] { + const commands: Parameters[0][] = []; + registerPiSessionCatalog({ + pluginConfig: {}, + registerSessionCatalog: vi.fn(), + registerNodeHostCommand: ( + command: Parameters[0], + ) => commands.push(command), + registerNodeInvokePolicy: vi.fn(), + } as unknown as OpenClawPluginApi); + return commands; +} + +afterEach(async () => { + if (originalSessionDir === undefined) { + delete process.env.PI_CODING_AGENT_SESSION_DIR; + } else { + process.env.PI_CODING_AGENT_SESSION_DIR = originalSessionDir; + } + if (originalAgentDir === undefined) { + delete process.env.PI_CODING_AGENT_DIR; + } else { + process.env.PI_CODING_AGENT_DIR = originalAgentDir; + } + if (originalHome === undefined) { + delete process.env.HOME; + } else { + process.env.HOME = originalHome; + } + if (originalUserProfile === undefined) { + delete process.env.USERPROFILE; + } else { + process.env.USERPROFILE = originalUserProfile; + } + await Promise.all( + temporaryDirectories.splice(0).map(async (directory) => { + await fs.rm(directory, { recursive: true, force: true }); + }), + ); +}); + +describe("Pi session catalog", () => { + it("rejects Windows drive-less rooted session paths", () => { + const originalPlatform = process.platform; + try { + Object.defineProperty(process, "platform", { configurable: true, value: "win32" }); + expect(() => piSessionStore({ PI_CODING_AGENT_SESSION_DIR: "\\sessions" })).toThrow( + "absolute or home-relative", + ); + expect(() => piSessionStore({ PI_CODING_AGENT_SESSION_DIR: "C:\\sessions" })).not.toThrow(); + expect(() => + piSessionStore({ PI_CODING_AGENT_SESSION_DIR: "\\\\server\\share\\sessions" }), + ).not.toThrow(); + } finally { + Object.defineProperty(process, "platform", { configurable: true, value: originalPlatform }); + } + }); + + it("trims the configured Pi agent directory", () => { + const agentDir = path.join(os.tmpdir(), "pi-agent"); + expect(piSessionStore({ PI_CODING_AGENT_DIR: ` ${agentDir} ` })).toEqual({ + root: path.join(agentDir, "sessions"), + flat: false, + }); + }); + + it("resolves relative project and global session directories like Pi", async () => { + const projectDirectory = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-pi-project-")); + const agentDirectory = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-pi-agent-")); + temporaryDirectories.push(projectDirectory, agentDirectory); + await fs.mkdir(path.join(projectDirectory, ".pi"), { recursive: true }); + await fs.writeFile( + path.join(projectDirectory, ".pi", "settings.json"), + `${JSON.stringify({ sessionDir: "sessions" })}\n`, + ); + const env = { + HOME: projectDirectory, + USERPROFILE: projectDirectory, + PI_CODING_AGENT_DIR: agentDirectory, + }; + + expect(piSessionStore(env, projectDirectory)).toEqual({ + root: path.join(projectDirectory, ".pi", "sessions"), + flat: true, + }); + + await fs.rm(path.join(projectDirectory, ".pi", "settings.json")); + await fs.writeFile( + path.join(agentDirectory, "settings.json"), + `${JSON.stringify({ sessionDir: "custom-sessions" })}\n`, + ); + expect(piSessionStore(env, projectDirectory)).toEqual({ + root: path.join(agentDirectory, "custom-sessions"), + flat: true, + }); + }); + + it("lists named sessions and reads the active JSONL branch", async () => { + await createPiStore(); + const listed = await listLocalPiSessionPage({ limit: 20 }); + expect(listed).toEqual({ + sessions: [ + expect.objectContaining({ + threadId: "pi-session", + name: "Pi catalog session", + cwd: "/workspace", + source: "pi-cli", + canContinue: false, + }), + ], + }); + + const transcript = await readLocalPiTranscriptPage({ threadId: "pi-session", limit: 20 }); + expect(transcript.items.map((item) => [item.type, item.text])).toEqual([ + ["userMessage", "hello"], + ["reasoning", "thinking"], + ["agentMessage", "hi"], + ["toolCall", 'bash\n{"command":"pwd"}'], + ["toolResult", "bash\n/workspace"], + ]); + const itemIds = transcript.items.flatMap((item) => (item.id ? [item.id] : [])); + expect(new Set(itemIds).size).toBe(itemIds.length); + + const latest = await readLocalPiTranscriptPage({ threadId: "pi-session", limit: 2 }); + expect(latest.items.map((item) => item.type)).toEqual(["toolCall", "toolResult"]); + expect(latest.nextCursor).toBeTruthy(); + const older = await readLocalPiTranscriptPage({ + threadId: "pi-session", + limit: 2, + cursor: latest.nextCursor, + }); + expect(older.items.map((item) => item.type)).toEqual(["reasoning", "agentMessage"]); + await expect(listLocalPiSessionPage({ cursor: " " })).rejects.toThrow("cursor is invalid"); + await expect( + readLocalPiTranscriptPage({ threadId: "pi-session", cursor: 123 }), + ).rejects.toThrow("cursor is invalid"); + + let provider: Parameters[0] | undefined; + registerPiSessionCatalog({ + pluginConfig: {}, + runtime: { nodes: { list: vi.fn().mockResolvedValue({ nodes: [] }) } }, + registerSessionCatalog: (value: NonNullable) => { + provider = value; + }, + registerNodeHostCommand: vi.fn(), + registerNodeInvokePolicy: vi.fn(), + } as unknown as OpenClawPluginApi); + await expect( + provider!.read({ hostId: "gateway", threadId: "pi-session", limit: 2 }), + ).resolves.toMatchObject({ threadId: "pi-session", items: expect.any(Array) }); + await expect(provider!.list({ search: " " })).resolves.toEqual([ + expect.objectContaining({ hostId: "gateway", sessions: [expect.any(Object)] }), + ]); + await expect(provider!.list({ search: "x".repeat(501) })).resolves.toEqual([ + expect.objectContaining({ hostId: "gateway", sessions: [] }), + ]); + }); + + it("summarizes and pages a large session within transport limits", async () => { + await createPiStore("x".repeat(600 * 1024)); + const listed = await listLocalPiSessionPage({ limit: 20 }); + expect(listed.sessions[0]?.name).toBe("Pi catalog session"); + const transcript = await readLocalPiTranscriptPage({ threadId: "pi-session", limit: 20 }); + const answer = transcript.items.find((item) => item.type === "agentMessage"); + expect(answer?.text?.endsWith("…")).toBe(true); + expect(Buffer.byteLength(JSON.stringify(transcript), "utf8")).toBeLessThan(20 * 1024 * 1024); + }); + + it("reads legacy linear sessions and visible extended messages", async () => { + const directory = await createPiStore(); + const entries = [ + { + type: "session", + version: 1, + id: "pi-session", + timestamp: "2026-07-13T10:00:00Z", + cwd: "/workspace", + }, + { + type: "message", + timestamp: "2026-07-13T10:00:01Z", + message: { role: "user", content: "legacy hello" }, + }, + { + type: "message", + timestamp: "2026-07-13T10:00:02Z", + message: { + role: "user", + content: [{ type: "image", mimeType: "image/png", data: "cG5n" }], + }, + }, + { + type: "message", + timestamp: "2026-07-13T10:00:03Z", + message: { role: "bashExecution", command: "pwd", output: "/workspace", exitCode: 0 }, + }, + { + type: "message", + timestamp: "2026-07-13T10:00:04Z", + message: { role: "custom", customType: "review", content: "visible note", display: true }, + }, + { + type: "message", + timestamp: "2026-07-13T10:00:05Z", + message: { + role: "hookMessage", + customType: "legacy-review", + content: "legacy visible note", + display: true, + }, + }, + ]; + await fs.writeFile( + path.join(directory, "session.jsonl"), + `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`, + ); + + const transcript = await readLocalPiTranscriptPage({ threadId: "pi-session", limit: 20 }); + expect(transcript.items.map((item) => [item.type, item.text])).toEqual([ + ["userMessage", "legacy hello"], + ["userMessage", "[image: image/png]"], + ["toolCall", "bash\npwd"], + ["toolResult", "/workspace"], + ["other", "review\nvisible note"], + ["other", "legacy-review\nlegacy visible note"], + ]); + }); + + it("resolves names anywhere in the file and honors a later clear", async () => { + const directory = await createPiStore(); + const file = path.join(directory, "session.jsonl"); + const entries = [ + { + type: "session", + version: 3, + id: "pi-session", + timestamp: "2026-07-13T10:00:00Z", + cwd: "/workspace", + }, + { + type: "message", + id: "u", + parentId: null, + timestamp: "2026-07-13T10:00:01Z", + message: { role: "user", content: "fallback title" }, + }, + { + type: "session_info", + id: "n", + parentId: "u", + timestamp: "2026-07-13T10:00:02Z", + name: "Assigned title", + }, + { + type: "message", + id: "a", + parentId: "n", + timestamp: "2026-07-13T10:00:03Z", + message: { role: "assistant", content: [{ type: "text", text: "x".repeat(40 * 1024) }] }, + }, + ]; + await fs.writeFile(file, `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`); + expect((await listLocalPiSessionPage({ limit: 20 })).sessions[0]?.name).toBe("Assigned title"); + + await fs.appendFile( + file, + `${JSON.stringify({ type: "session_info", id: "clear", parentId: "a", timestamp: "2026-07-13T10:00:04Z", name: "" })}\n`, + ); + expect((await listLocalPiSessionPage({ limit: 20 })).sessions[0]?.name).toBe("fallback title"); + }); + + it("indexes final records without a trailing newline", async () => { + const directory = await createPiStore(); + const file = path.join(directory, "session.jsonl"); + await fs.writeFile( + file, + JSON.stringify({ + type: "session", + version: 3, + id: "pi-session", + timestamp: "2026-07-13T10:00:00Z", + cwd: "/workspace", + }), + ); + expect((await listLocalPiSessionPage({ limit: 20 })).sessions[0]?.threadId).toBe("pi-session"); + + await fs.appendFile( + file, + `\n${JSON.stringify({ + type: "session_info", + id: "name", + parentId: null, + timestamp: "2026-07-13T10:00:01Z", + name: "No final newline", + })}`, + ); + expect((await listLocalPiSessionPage({ limit: 20 })).sessions[0]?.name).toBe( + "No final newline", + ); + }); + + it("rebuilds metadata after a same-file replacement grows", async () => { + const directory = await createPiStore("old session"); + const file = path.join(directory, "session.jsonl"); + await listLocalPiSessionPage({ limit: 20 }); + + const entries = [ + { + type: "session", + version: 3, + id: "pi-replaced-session", + timestamp: "2026-07-13T11:00:00Z", + cwd: "/workspace/replaced", + }, + { + type: "message", + id: "replacement-user", + parentId: null, + timestamp: "2026-07-13T11:00:01Z", + message: { role: "user", content: `replacement ${"x".repeat(4_096)}` }, + }, + { + type: "session_info", + id: "replacement-name", + parentId: "replacement-user", + timestamp: "2026-07-13T11:00:02Z", + name: "Replacement session", + }, + ]; + await fs.writeFile(file, `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`); + + await expect(listLocalPiSessionPage({ limit: 20 })).resolves.toMatchObject({ + sessions: [ + expect.objectContaining({ + threadId: "pi-replaced-session", + name: "Replacement session", + cwd: "/workspace/replaced", + }), + ], + }); + }); + + it("does not reuse transcript paths after the configured store changes", async () => { + await createPiStore("old store"); + await listLocalPiSessionPage({ limit: 20 }); + await expect( + readLocalPiTranscriptPage({ threadId: "pi-session", limit: 20 }), + ).resolves.toMatchObject({ + items: expect.arrayContaining([expect.objectContaining({ text: "old store" })]), + }); + + await createPiStore("new store"); + await expect( + readLocalPiTranscriptPage({ threadId: "pi-session", limit: 20 }), + ).resolves.toMatchObject({ + items: expect.arrayContaining([expect.objectContaining({ text: "new store" })]), + }); + }); + + it("paginates, searches, and reads beyond the newest summary batch", async () => { + const directory = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-pi-catalog-")); + temporaryDirectories.push(directory); + process.env.PI_CODING_AGENT_SESSION_DIR = directory; + const baseTime = Date.parse("2026-07-13T10:00:00Z") / 1_000; + await Promise.all( + Array.from({ length: 105 }, async (_, index) => { + const file = path.join(directory, `session-${String(index)}.jsonl`); + const entries = [ + { + type: "session", + version: 3, + id: `pi-session-${String(index)}`, + timestamp: "2026-07-13T10:00:00Z", + cwd: "/workspace", + }, + { + type: "session_info", + id: `info-${String(index)}`, + parentId: null, + timestamp: "2026-07-13T10:00:01Z", + name: `Pi history ${String(index)}`, + }, + ]; + await fs.writeFile(file, `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`); + await fs.utimes(file, baseTime + index, baseTime + index); + }), + ); + + const first = await listLocalPiSessionPage({ limit: 100 }); + expect(first.sessions).toHaveLength(100); + expect(first.nextCursor).toBeTruthy(); + const second = await listLocalPiSessionPage({ limit: 100, cursor: first.nextCursor }); + expect(second.sessions).toHaveLength(5); + expect(second.nextCursor).toBeUndefined(); + await expect( + listLocalPiSessionPage({ limit: 20, searchTerm: "Pi history 0" }), + ).resolves.toMatchObject({ + sessions: [expect.objectContaining({ threadId: "pi-session-0" })], + }); + await expect( + readLocalPiTranscriptPage({ threadId: "pi-session-0", limit: 20 }), + ).resolves.toMatchObject({ threadId: "pi-session-0" }); + }); + + it("uses the configured Pi session directory and lists oversized sessions", async () => { + const agentDirectory = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-pi-agent-")); + const homeDirectory = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-pi-home-")); + temporaryDirectories.push(agentDirectory, homeDirectory); + const sessionDirectory = path.join(homeDirectory, "custom-sessions"); + await fs.mkdir(sessionDirectory, { recursive: true }); + await fs.writeFile( + path.join(agentDirectory, "settings.json"), + `${JSON.stringify({ sessionDir: "~/custom-sessions" })}\n`, + ); + const file = path.join(sessionDirectory, "large.jsonl"); + await fs.writeFile( + file, + `${JSON.stringify({ + type: "session", + version: 3, + id: "pi-large-session", + timestamp: "2026-07-13T10:00:00Z", + cwd: "/workspace", + })}\n`, + ); + const middle = await fs.open(file, "r+"); + await middle.truncate(2 * 1024 * 1024); + await middle.close(); + await fs.appendFile( + file, + `\n${JSON.stringify({ + type: "session_info", + id: "large-info", + parentId: null, + timestamp: "2026-07-13T10:00:01Z", + name: "Pi large session", + })}\n`, + ); + const handle = await fs.open(file, "r+"); + await handle.truncate(33 * 1024 * 1024); + await handle.close(); + delete process.env.PI_CODING_AGENT_SESSION_DIR; + process.env.PI_CODING_AGENT_DIR = agentDirectory; + process.env.HOME = homeDirectory; + process.env.USERPROFILE = homeDirectory; + + await expect(listLocalPiSessionPage({ limit: 20 })).resolves.toMatchObject({ + sessions: [ + expect.objectContaining({ threadId: "pi-large-session", name: "Pi large session" }), + ], + }); + await expect( + readLocalPiTranscriptPage({ threadId: "pi-large-session", limit: 20 }), + ).rejects.toThrow("32 MiB read safety limit"); + + await fs.appendFile( + file, + `\n${JSON.stringify({ + type: "session_info", + id: "large-clear", + parentId: "large-info", + timestamp: "2026-07-13T10:00:02Z", + name: "", + })}\n`, + ); + expect((await listLocalPiSessionPage({ limit: 20 })).sessions[0]?.name).toBeUndefined(); + }); + + it("auto-detects the store and honors the node-local Web UI switch", async () => { + const directory = await createPiStore(); + const commands = registerPiNodeHostCommands(); + expect(commands.map((command) => command.command)).toEqual([ + PI_SESSIONS_LIST_COMMAND, + PI_SESSION_READ_COMMAND, + ]); + expect( + commands.every((command) => + command.isAvailable?.({ + config: {}, + env: { PI_CODING_AGENT_SESSION_DIR: directory }, + } as never), + ), + ).toBe(true); + expect( + commands.every((command) => + command.isAvailable?.({ + config: { + plugins: { entries: { acpx: { config: { piSessionCatalog: { enabled: false } } } } }, + }, + env: { PI_CODING_AGENT_SESSION_DIR: directory }, + } as never), + ), + ).toBe(false); + expect( + commands.every((command) => + command.isAvailable?.({ + config: {}, + env: { PI_CODING_AGENT_SESSION_DIR: path.join(directory, "missing") }, + } as never), + ), + ).toBe(false); + expect( + commands.every((command) => + command.isAvailable?.({ + config: {}, + env: { PI_CODING_AGENT_SESSION_DIR: ".pi/sessions" }, + } as never), + ), + ).toBe(false); + }); + + it("does not register the catalog when explicitly disabled", () => { + const registerSessionCatalog = vi.fn(); + const api = { + pluginConfig: { piSessionCatalog: { enabled: false } }, + registerSessionCatalog, + } as unknown as OpenClawPluginApi; + registerPiSessionCatalog(api); + expect(registerSessionCatalog).not.toHaveBeenCalled(); + }); + + it("bridges paired-node list and read requests without undefined transport fields", async () => { + let provider: Parameters[0] | undefined; + const invoke = vi + .fn() + .mockResolvedValueOnce({ + payloadJSON: JSON.stringify({ + sessions: [ + { + threadId: "pi-remote", + status: "stored", + source: "pi-cli", + archived: false, + canContinue: false, + canArchive: false, + }, + ], + }), + }) + .mockResolvedValueOnce({ + payloadJSON: JSON.stringify({ + threadId: "pi-remote", + items: [{ type: "agentMessage", text: "remote answer" }], + }), + }); + const api = { + pluginConfig: {}, + runtime: { + nodes: { + list: vi.fn().mockResolvedValue({ + nodes: [ + { + nodeId: "node-1", + displayName: "Remote", + connected: true, + commands: [PI_SESSIONS_LIST_COMMAND, PI_SESSION_READ_COMMAND], + }, + ], + }), + invoke, + }, + }, + registerSessionCatalog: (value: NonNullable) => { + provider = value; + }, + registerNodeHostCommand: vi.fn(), + registerNodeInvokePolicy: vi.fn(), + } as unknown as OpenClawPluginApi; + + registerPiSessionCatalog(api); + const catalog = provider; + expect(catalog).toBeDefined(); + await catalog!.list({ hostIds: ["node:node-1"], search: " " }); + await catalog!.read({ hostId: "node:node-1", threadId: "pi-remote" }); + + expect(invoke).toHaveBeenNthCalledWith(1, { + nodeId: "node-1", + command: PI_SESSIONS_LIST_COMMAND, + params: {}, + timeoutMs: 20_000, + scopes: ["operator.write"], + }); + expect(invoke).toHaveBeenNthCalledWith(2, { + nodeId: "node-1", + command: PI_SESSION_READ_COMMAND, + params: { threadId: "pi-remote" }, + timeoutMs: 20_000, + scopes: ["operator.write"], + }); + + invoke.mockResolvedValueOnce({ + payloadJSON: JSON.stringify({ + sessions: [ + { + threadId: 123, + status: "stored", + archived: false, + canContinue: false, + canArchive: false, + }, + ], + }), + }); + await expect(catalog!.list({ hostIds: ["node:node-1"] })).resolves.toEqual([ + expect.objectContaining({ + error: { code: "NODE_INVOKE_FAILED", message: expect.any(String) }, + }), + ]); + + invoke.mockResolvedValueOnce({ + payloadJSON: JSON.stringify({ + sessions: [ + { + threadId: "--help", + status: "stored", + archived: false, + canContinue: false, + canArchive: false, + }, + ], + }), + }); + await expect(catalog!.list({ hostIds: ["node:node-1"] })).resolves.toEqual([ + expect.objectContaining({ + error: { code: "NODE_INVOKE_FAILED", message: expect.any(String) }, + }), + ]); + + invoke.mockResolvedValueOnce({ + payloadJSON: JSON.stringify({ + threadId: "pi-remote", + items: [{ type: "invalid", text: "bad" }], + }), + }); + await expect(catalog!.read({ hostId: "node:node-1", threadId: "pi-remote" })).rejects.toThrow( + "invalid transcript page", + ); + }); +}); diff --git a/extensions/acpx/src/pi-session-catalog.ts b/extensions/acpx/src/pi-session-catalog.ts new file mode 100644 index 000000000000..5bf718ba193a --- /dev/null +++ b/extensions/acpx/src/pi-session-catalog.ts @@ -0,0 +1,373 @@ +import process from "node:process"; +import type { + SessionCatalogSession, + SessionCatalogTranscriptItem, + SessionsCatalogReadResult, +} from "openclaw/plugin-sdk/session-catalog"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { listPiSummaryPage, readPiSessionById } from "./pi-session-store.js"; + +const LOCAL_HOST_ID = "gateway"; +const DEFAULT_PAGE_LIMIT = 20; +const MAX_PAGE_LIMIT = 100; +const MAX_SEARCH_LENGTH = 500; +const MAX_CURSOR_LENGTH = 128; +const MAX_TRANSCRIPT_ITEM_BYTES = 512 * 1024; +const MAX_TRANSCRIPT_PAGE_BYTES = 20 * 1024 * 1024; +const SESSION_ID_PATTERN = /^(?!-)[A-Za-z0-9._:-]{1,256}$/u; + +export type PiSessionPage = { sessions: SessionCatalogSession[]; nextCursor?: string }; + +export function optionalPiString(value: unknown, maxLength: number): string | undefined { + if (typeof value !== "string") { + return undefined; + } + const trimmed = value.trim(); + return trimmed && trimmed.length <= maxLength ? trimmed : undefined; +} + +function boundedLimit(value: unknown, fallback = DEFAULT_PAGE_LIMIT): number { + if (value === undefined) { + return fallback; + } + if (!Number.isInteger(value) || Number(value) < 1 || Number(value) > MAX_PAGE_LIMIT) { + throw new Error(`limit must be an integer between 1 and ${String(MAX_PAGE_LIMIT)}`); + } + return Number(value); +} + +function encodeCursor(offset: number): string { + return Buffer.from(JSON.stringify({ offset }), "utf8").toString("base64url"); +} + +function decodeCursor(value: unknown): number { + if (value === undefined) { + return 0; + } + const cursor = optionalPiString(value, MAX_CURSOR_LENGTH); + if (!cursor) { + throw new Error("cursor is invalid"); + } + try { + const parsed = JSON.parse(Buffer.from(cursor, "base64url").toString("utf8")) as unknown; + if (!isRecord(parsed) || !Number.isInteger(parsed.offset) || Number(parsed.offset) < 0) { + throw new Error("invalid offset"); + } + return Number(parsed.offset); + } catch (error) { + throw new Error("cursor is invalid", { cause: error }); + } +} + +function truncateUtf8(text: string, maxBytes: number): string { + if (Buffer.byteLength(text, "utf8") <= maxBytes) { + return text; + } + let low = 0; + let high = text.length; + while (low < high) { + const middle = Math.ceil((low + high) / 2); + if (Buffer.byteLength(text.slice(0, middle), "utf8") <= maxBytes - 3) { + low = middle; + } else { + high = middle - 1; + } + } + const end = low > 0 && /[\uD800-\uDBFF]/u.test(text.charAt(low - 1)) ? low - 1 : low; + return `${text.slice(0, end)}…`; +} + +function transcriptPage( + items: SessionCatalogTranscriptItem[], + limit: number, + offset: number, +): { items: SessionCatalogTranscriptItem[]; nextCursor?: string } { + const end = Math.max(0, items.length - offset); + const start = Math.max(0, end - limit); + const page: SessionCatalogTranscriptItem[] = []; + let pageBytes = 2; + for (let index = end - 1; index >= start; index -= 1) { + const item = items[index]; + if (!item) { + continue; + } + const bounded: SessionCatalogTranscriptItem = { + ...item, + text: truncateUtf8(item.text ?? "", MAX_TRANSCRIPT_ITEM_BYTES), + }; + const itemBytes = Buffer.byteLength(JSON.stringify(bounded), "utf8") + 1; + if (page.length > 0 && pageBytes + itemBytes > MAX_TRANSCRIPT_PAGE_BYTES) { + break; + } + page.unshift(bounded); + pageBytes += itemBytes; + } + const consumed = offset + page.length; + return { + items: page, + ...(consumed < items.length ? { nextCursor: encodeCursor(consumed) } : {}), + }; +} + +function textFromContent(content: unknown): string { + if (typeof content === "string") { + return content; + } + if (!Array.isArray(content)) { + return ""; + } + return content + .flatMap((part) => { + if (!isRecord(part)) { + return []; + } + if (part.type === "text" && typeof part.text === "string") { + return [part.text]; + } + if (part.type === "image") { + const mimeType = optionalPiString(part.mimeType, 128); + return [mimeType ? `[image: ${mimeType}]` : "[image]"]; + } + return []; + }) + .join("\n"); +} + +function timestampMs(value: unknown): number | undefined { + if (typeof value === "number" && Number.isFinite(value)) { + return value; + } + if (typeof value === "string") { + const parsed = Date.parse(value); + return Number.isNaN(parsed) ? undefined : parsed; + } + return undefined; +} + +function parseListParams(value: unknown): { searchTerm?: string; limit: number; cursor?: string } { + if (value === undefined || value === null) { + return { limit: DEFAULT_PAGE_LIMIT }; + } + if (!isRecord(value)) { + throw new Error("Pi session list parameters must be an object"); + } + const unknown = Object.keys(value).find( + (key) => !["searchTerm", "limit", "cursor"].includes(key), + ); + if (unknown) { + throw new Error(`unknown Pi session list parameter: ${unknown}`); + } + const searchTerm = optionalPiString(value.searchTerm, MAX_SEARCH_LENGTH); + if (value.searchTerm !== undefined && !searchTerm) { + throw new Error("searchTerm is invalid"); + } + const cursor = optionalPiString(value.cursor, MAX_CURSOR_LENGTH); + if (value.cursor !== undefined && !cursor) { + throw new Error("cursor is invalid"); + } + return { + limit: boundedLimit(value.limit), + ...(searchTerm ? { searchTerm } : {}), + ...(cursor ? { cursor } : {}), + }; +} + +function parseReadParams(value: unknown): { threadId: string; limit: number; cursor?: string } { + if (!isRecord(value)) { + throw new Error("Pi session read parameters must be an object"); + } + const unknown = Object.keys(value).find((key) => !["threadId", "limit", "cursor"].includes(key)); + if (unknown) { + throw new Error(`unknown Pi session read parameter: ${unknown}`); + } + const threadId = optionalPiString(value.threadId, 256); + if (!threadId || !SESSION_ID_PATTERN.test(threadId)) { + throw new Error("threadId is invalid"); + } + const cursor = optionalPiString(value.cursor, MAX_CURSOR_LENGTH); + if (value.cursor !== undefined && !cursor) { + throw new Error("cursor is invalid"); + } + return { + threadId, + limit: boundedLimit(value.limit), + ...(cursor ? { cursor } : {}), + }; +} + +export async function listLocalPiSessionPage(value?: unknown): Promise { + const params = parseListParams(value); + const offset = decodeCursor(params.cursor); + const { summaries, hasMore } = await listPiSummaryPage(process.env, { + offset, + limit: params.limit, + ...(params.searchTerm ? { searchTerm: params.searchTerm } : {}), + }); + const page = summaries.map(({ file: _file, ...session }) => session); + return { + sessions: page, + ...(hasMore ? { nextCursor: encodeCursor(offset + page.length) } : {}), + }; +} + +function isoTimestamp( + message: Record, + entry: Record, +): string | undefined { + const value = timestampMs(message.timestamp) ?? timestampMs(entry.timestamp); + if (value === undefined) { + return undefined; + } + const date = new Date(value); + return Number.isNaN(date.getTime()) ? undefined : date.toISOString(); +} + +function jsonText(value: unknown, maxLength = 20_000): string | undefined { + try { + const text = JSON.stringify(value); + return text.length > maxLength ? `${text.slice(0, maxLength)}…` : text; + } catch { + return undefined; + } +} + +function activePiEntries(entries: Record[]): Record[] { + const header = entries[0]; + const version = + header?.type === "session" && typeof header.version === "number" ? header.version : 1; + if (version < 2) { + return entries.slice(1); + } + const body = entries.filter( + (entry) => entry.type !== "session" && optionalPiString(entry.id, 256), + ); + const byId = new Map(body.map((entry) => [String(entry.id), entry])); + const active: Record[] = []; + let current = body.at(-1); + const visited = new Set(); + while (current) { + const id = String(current.id); + if (visited.has(id)) { + break; + } + visited.add(id); + active.push(current); + const parentId = optionalPiString(current.parentId, 256); + current = parentId ? byId.get(parentId) : undefined; + } + return active.toReversed(); +} + +function piMessageItems(entry: Record): SessionCatalogTranscriptItem[] { + if (!isRecord(entry.message)) { + return []; + } + const message = entry.message; + const role = message.role; + const id = optionalPiString(entry.id, 256); + const timestamp = isoTimestamp(message, entry); + const model = optionalPiString(message.model, 256); + const provider = optionalPiString(message.provider, 256); + const modelRef = provider && model ? `${provider}/${model}` : model; + const common = { + ...(id ? { id } : {}), + ...(timestamp ? { timestamp } : {}), + ...(modelRef ? { model: modelRef } : {}), + }; + if (role === "user") { + const text = textFromContent(message.content); + return text ? [{ ...common, type: "userMessage", text }] : []; + } + if (role === "toolResult") { + const toolName = optionalPiString(message.toolName, 256); + const text = textFromContent(message.content); + return [{ ...common, type: "toolResult", text: toolName ? `${toolName}\n${text}` : text }]; + } + if (role === "bashExecution") { + const command = optionalPiString(message.command, 4_096) ?? "bash"; + const output = typeof message.output === "string" ? message.output : ""; + const status = + message.cancelled === true + ? "command cancelled" + : typeof message.exitCode === "number" && message.exitCode !== 0 + ? `command exited with code ${String(message.exitCode)}` + : ""; + return [ + { ...common, type: "toolCall", text: `bash\n${command}` }, + { + ...common, + ...(id ? { id: `${id}:result` } : {}), + type: "toolResult", + text: [output, status].filter(Boolean).join("\n\n"), + }, + ]; + } + if (role === "custom" || role === "hookMessage") { + if (message.display !== true) { + return []; + } + const customType = optionalPiString(message.customType, 256); + const text = textFromContent(message.content); + return text + ? [{ ...common, type: "other", text: customType ? `${customType}\n${text}` : text }] + : []; + } + if (role !== "assistant" || !Array.isArray(message.content)) { + return []; + } + return message.content.flatMap((part, index): SessionCatalogTranscriptItem[] => { + if (!isRecord(part)) { + return []; + } + const partCommon = { ...common, ...(id ? { id: `${id}:${String(index)}` } : {}) }; + if (part.type === "text" && typeof part.text === "string") { + return [{ ...partCommon, type: "agentMessage", text: part.text }]; + } + if (part.type === "thinking" && typeof part.thinking === "string") { + return [{ ...partCommon, type: "reasoning", text: part.thinking }]; + } + if (part.type === "toolCall") { + const name = optionalPiString(part.name, 256) ?? "tool"; + const args = jsonText(part.arguments); + return [{ ...partCommon, type: "toolCall", text: args ? `${name}\n${args}` : name }]; + } + return []; + }); +} + +function piTranscriptItems(entries: Record[]): SessionCatalogTranscriptItem[] { + return activePiEntries(entries).flatMap((entry): SessionCatalogTranscriptItem[] => { + if (entry.type === "message") { + return piMessageItems(entry); + } + const id = optionalPiString(entry.id, 256); + const timestamp = optionalPiString(entry.timestamp, 128); + const common = { ...(id ? { id } : {}), ...(timestamp ? { timestamp } : {}) }; + if (entry.type === "compaction" && typeof entry.summary === "string") { + return [{ ...common, type: "other", text: entry.summary }]; + } + if (entry.type === "branch_summary" && typeof entry.summary === "string") { + return [{ ...common, type: "other", text: entry.summary }]; + } + if (entry.type === "custom_message" && entry.display === true) { + const text = textFromContent(entry.content); + return text ? [{ ...common, type: "other", text }] : []; + } + return []; + }); +} + +export async function readLocalPiTranscriptPage( + value: unknown, +): Promise { + const params = parseReadParams(value); + const offset = decodeCursor(params.cursor); + const items = piTranscriptItems(await readPiSessionById(params.threadId, process.env)); + const page = transcriptPage(items, params.limit, offset); + return { + hostId: LOCAL_HOST_ID, + label: "Local Pi", + threadId: params.threadId, + ...page, + }; +} diff --git a/extensions/acpx/src/pi-session-paths.ts b/extensions/acpx/src/pi-session-paths.ts new file mode 100644 index 000000000000..d4bfff43df98 --- /dev/null +++ b/extensions/acpx/src/pi-session-paths.ts @@ -0,0 +1,97 @@ +import { readFileSync, statSync } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; + +function optionalString(value: unknown, maxLength: number): string | undefined { + if (typeof value !== "string") { + return undefined; + } + const trimmed = value.trim(); + return trimmed && trimmed.length <= maxLength ? trimmed : undefined; +} + +function piHome(env: NodeJS.ProcessEnv): string { + const configured = process.platform === "win32" ? env.USERPROFILE?.trim() : env.HOME?.trim(); + return configured || os.homedir(); +} + +function isPiSessionCatalogPathAbsolute( + value: string, + platform: NodeJS.Platform = process.platform, +): boolean { + if (platform !== "win32") { + return path.posix.isAbsolute(value); + } + const root = path.win32.parse(value).root; + return path.win32.isAbsolute(value) && root !== "\\" && root !== "/"; +} + +function resolveConfiguredPath( + value: string, + env: NodeJS.ProcessEnv, + relativeBase?: string, +): string { + const home = piHome(env); + let resolved = value; + if (value === "~") { + resolved = home; + } + if (value.startsWith("~/") || value.startsWith("~\\")) { + resolved = path.join(home, value.slice(2)); + } + if (!isPiSessionCatalogPathAbsolute(resolved)) { + if (relativeBase) { + return path.resolve(relativeBase, resolved); + } + throw new Error("Pi session catalog requires absolute or home-relative storage paths"); + } + return path.resolve(resolved); +} + +function settingsSessionDir(file: string): string | undefined { + try { + const value = JSON.parse(readFileSync(file, "utf8")) as unknown; + return isRecord(value) ? optionalString(value.sessionDir, 4_096) : undefined; + } catch { + return undefined; + } +} + +export function piSessionStore( + env: NodeJS.ProcessEnv, + cwd = process.cwd(), +): { root: string; flat: boolean } { + const customSessionDir = env.PI_CODING_AGENT_SESSION_DIR?.trim(); + if (customSessionDir) { + return { root: resolveConfiguredPath(customSessionDir, env), flat: true }; + } + const home = piHome(env); + const customAgentDir = env.PI_CODING_AGENT_DIR?.trim(); + const agentDir = customAgentDir + ? resolveConfiguredPath(customAgentDir, env) + : path.join(home, ".pi", "agent"); + const projectSessionDir = settingsSessionDir(path.join(cwd, ".pi", "settings.json")); + if (projectSessionDir) { + return { + root: resolveConfiguredPath(projectSessionDir, env, path.join(cwd, ".pi")), + flat: true, + }; + } + const globalSessionDir = settingsSessionDir(path.join(agentDir, "settings.json")); + if (globalSessionDir) { + return { root: resolveConfiguredPath(globalSessionDir, env, agentDir), flat: true }; + } + return { + root: path.join(agentDir, "sessions"), + flat: false, + }; +} + +export function piSessionStoreAvailable(env: NodeJS.ProcessEnv): boolean { + try { + return statSync(piSessionStore(env).root).isDirectory(); + } catch { + return false; + } +} diff --git a/extensions/acpx/src/pi-session-store.ts b/extensions/acpx/src/pi-session-store.ts new file mode 100644 index 000000000000..6d1f844808dc --- /dev/null +++ b/extensions/acpx/src/pi-session-store.ts @@ -0,0 +1,489 @@ +import { createReadStream } from "node:fs"; +import fs from "node:fs/promises"; +import path from "node:path"; +import type { SessionCatalogSession } from "openclaw/plugin-sdk/session-catalog"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { piSessionStore } from "./pi-session-paths.js"; + +const MAX_DISCOVERY_FILES = 10_000; +const SUMMARY_SCAN_BATCH_SIZE = 100; +const MAX_SUMMARY_CACHE_ENTRIES = 256; +const MAX_SESSION_BYTES = 32 * 1024 * 1024; +const MAX_SUMMARY_LINE_BYTES = 1024 * 1024; +const APPEND_PROOF_EDGE_BYTES = 64 * 1024; +const IO_CONCURRENCY = 8; +const SESSION_ID_PATTERN = /^(?!-)[A-Za-z0-9._:-]{1,256}$/u; + +type PiSessionSummary = SessionCatalogSession & { file: string }; + +type PiFileCandidate = { + file: string; + storeRoot: string; + identity: string; + mtimeMs: number; + size: number; +}; + +type PiSummaryScanState = { + header?: Record; + name?: string; + firstMessage?: string; + pending: Buffer; + discarding: boolean; + invalid: boolean; +}; + +type CachedSummary = PiFileCandidate & { + summary?: PiSessionSummary; + scanState: PiSummaryScanState; + appendProof: { head: Buffer; tail: Buffer }; +}; + +// Pi owns session-file mutation. The bounded cache resumes append-only metadata +// scans, avoiding a full reread every time an active transcript grows. +const summaryCache = new Map(); +const threadFileCache = new Map(); + +function threadCacheKey(storeRoot: string, threadId: string): string { + return `${storeRoot}\0${threadId}`; +} + +function forgetCachedSummary(file: string): void { + const cached = summaryCache.get(file); + const threadId = cached?.summary?.threadId; + if (cached && threadId) { + const key = threadCacheKey(cached.storeRoot, threadId); + if (threadFileCache.get(key) === file) { + threadFileCache.delete(key); + } + } + summaryCache.delete(file); +} + +function cacheSummary(file: string, value: CachedSummary): void { + forgetCachedSummary(file); + summaryCache.set(file, value); + while (summaryCache.size > MAX_SUMMARY_CACHE_ENTRIES) { + const oldest = summaryCache.keys().next().value; + if (typeof oldest !== "string") { + break; + } + forgetCachedSummary(oldest); + } +} + +function optionalString(value: unknown, maxLength: number): string | undefined { + if (typeof value !== "string") { + return undefined; + } + const trimmed = value.trim(); + return trimmed && trimmed.length <= maxLength ? trimmed : undefined; +} + +async function discoverPiSessionFiles( + env: NodeJS.ProcessEnv, +): Promise<{ root: string; files: string[] }> { + const store = piSessionStore(env); + let entries: Array; + try { + entries = await fs.readdir(store.root, { withFileTypes: true }); + } catch { + return { root: store.root, files: [] }; + } + if (store.flat) { + return { + root: store.root, + files: entries + .filter((entry) => entry.isFile() && entry.name.endsWith(".jsonl")) + .slice(0, MAX_DISCOVERY_FILES) + .map((entry) => path.join(store.root, entry.name)), + }; + } + const files: string[] = []; + for (const entry of entries) { + if (!entry.isDirectory() || files.length >= MAX_DISCOVERY_FILES) { + continue; + } + const directory = path.join(store.root, entry.name); + let children: Array; + try { + children = await fs.readdir(directory, { withFileTypes: true }); + } catch { + continue; + } + for (const child of children) { + if (child.isFile() && child.name.endsWith(".jsonl")) { + files.push(path.join(directory, child.name)); + if (files.length >= MAX_DISCOVERY_FILES) { + break; + } + } + } + } + return { root: store.root, files }; +} + +async function mapConcurrent( + values: T[], + limit: number, + mapper: (value: T) => Promise, +): Promise { + const results: R[] = []; + results.length = values.length; + let nextIndex = 0; + const workers = Array.from({ length: Math.min(limit, values.length) }, async () => { + while (nextIndex < values.length) { + const index = nextIndex++; + results[index] = await mapper(values[index]!); + } + }); + await Promise.all(workers); + return results; +} + +async function piFileCandidates(env: NodeJS.ProcessEnv): Promise { + const { root, files } = await discoverPiSessionFiles(env); + const candidates = await mapConcurrent(files, IO_CONCURRENCY, async (file) => { + try { + const stats = await fs.stat(file); + return stats.isFile() + ? { + file, + storeRoot: root, + identity: `${String(stats.dev)}:${String(stats.ino)}:${String(stats.birthtimeMs)}`, + mtimeMs: stats.mtimeMs, + size: stats.size, + } + : undefined; + } catch { + return undefined; + } + }); + return candidates + .filter((candidate): candidate is PiFileCandidate => candidate !== undefined) + .toSorted((left, right) => right.mtimeMs - left.mtimeMs); +} + +function parsePiJsonLines(content: string): Record[] { + return content.split(/\r?\n/u).flatMap((line) => { + if (!line.trim()) { + return []; + } + try { + const value = JSON.parse(line) as unknown; + return isRecord(value) ? [value] : []; + } catch { + return []; + } + }); +} + +function textFromContent(content: unknown): string { + if (typeof content === "string") { + return content; + } + if (!Array.isArray(content)) { + return ""; + } + return content + .flatMap((part) => + isRecord(part) && part.type === "text" && typeof part.text === "string" ? [part.text] : [], + ) + .join("\n"); +} + +function timestampMs(value: unknown): number | undefined { + if (typeof value === "number" && Number.isFinite(value)) { + return value; + } + if (typeof value === "string") { + const parsed = Date.parse(value); + return Number.isNaN(parsed) ? undefined : parsed; + } + return undefined; +} + +function processSummaryLine(state: PiSummaryScanState, line: Buffer): void { + const content = line.at(-1) === 0x0d ? line.subarray(0, -1) : line; + const entry = parsePiJsonLines(content.toString("utf8"))[0]; + if (!entry) { + return; + } + if (!state.header) { + if (entry.type !== "session") { + state.invalid = true; + return; + } + state.header = entry; + return; + } + if (entry.type === "session_info") { + // Latest metadata wins, including an explicit empty-name clear. + state.name = optionalString(entry.name, 1_000); + } else if ( + !state.firstMessage && + entry.type === "message" && + isRecord(entry.message) && + entry.message.role === "user" + ) { + state.firstMessage = optionalString(textFromContent(entry.message.content), 1_000); + } +} + +function appendSummaryBytes(state: PiSummaryScanState, bytes: Buffer): void { + if (state.discarding || bytes.length === 0) { + return; + } + if (state.pending.length + bytes.length > MAX_SUMMARY_LINE_BYTES) { + state.pending = Buffer.alloc(0); + state.discarding = true; + return; + } + state.pending = + state.pending.length === 0 ? Buffer.from(bytes) : Buffer.concat([state.pending, bytes]); +} + +async function scanSummaryAppend( + candidate: PiFileCandidate, + start: number, + state: PiSummaryScanState, +): Promise { + if (start >= candidate.size || state.invalid) { + return; + } + const stream = createReadStream(candidate.file, { start, end: candidate.size - 1 }); + for await (const value of stream) { + const chunk = Buffer.isBuffer(value) ? value : Buffer.from(value); + let offset = 0; + while (offset < chunk.length) { + const newline = chunk.indexOf(0x0a, offset); + const end = newline < 0 ? chunk.length : newline; + appendSummaryBytes(state, chunk.subarray(offset, end)); + if (newline < 0) { + break; + } + if (!state.discarding) { + processSummaryLine(state, state.pending); + } + state.pending = Buffer.alloc(0); + state.discarding = false; + if (state.invalid) { + return; + } + offset = newline + 1; + } + } +} + +async function readAppendProof( + file: string, + size: number, +): Promise<{ head: Buffer; tail: Buffer }> { + const length = Math.min(size, APPEND_PROOF_EDGE_BYTES); + if (length === 0) { + return { head: Buffer.alloc(0), tail: Buffer.alloc(0) }; + } + const handle = await fs.open(file, "r"); + try { + const head = Buffer.alloc(length); + const tail = Buffer.alloc(length); + const [headRead, tailRead] = await Promise.all([ + handle.read(head, 0, length, 0), + handle.read(tail, 0, length, size - length), + ]); + return { + head: head.subarray(0, headRead.bytesRead), + tail: tail.subarray(0, tailRead.bytesRead), + }; + } finally { + await handle.close(); + } +} + +async function cachedPrefixIsUnchanged(candidate: PiFileCandidate, cached: CachedSummary) { + if (cached.identity !== candidate.identity || cached.size >= candidate.size) { + return false; + } + // Pi persists established sessions with appendFileSync. Its in-place rewrite + // paths (notably version migration) rewrite the header, so the head proof + // rejects them; the tail proof rejects truncation before later growth. + const current = await readAppendProof(candidate.file, cached.size); + return ( + current.head.equals(cached.appendProof.head) && current.tail.equals(cached.appendProof.tail) + ); +} + +async function readPiSessionSummary( + candidate: PiFileCandidate, +): Promise { + const cached = summaryCache.get(candidate.file); + if (cached?.mtimeMs === candidate.mtimeMs && cached.size === candidate.size) { + summaryCache.delete(candidate.file); + summaryCache.set(candidate.file, cached); + return cached.summary; + } + let summary: PiSessionSummary | undefined; + let scanState: PiSummaryScanState; + let appendProof: CachedSummary["appendProof"]; + try { + // Pi normally appends JSONL. Resume only when bounded edge proofs show the + // previously indexed prefix survived; rewrites rebuild from byte zero. + const resumable = + cached && (await cachedPrefixIsUnchanged(candidate, cached)) ? cached : undefined; + scanState = resumable + ? { + ...resumable.scanState, + pending: Buffer.from(resumable.scanState.pending), + } + : { + pending: Buffer.alloc(0), + discarding: false, + invalid: false, + }; + await scanSummaryAppend(candidate, resumable?.size ?? 0, scanState); + appendProof = await readAppendProof(candidate.file, candidate.size); + // A complete final record is valid without a newline. Project it from a + // clone so later appends can still finish the cached pending line once. + const projectedState = { ...scanState, pending: Buffer.from(scanState.pending) }; + if (!projectedState.discarding && projectedState.pending.length > 0) { + processSummaryLine(projectedState, projectedState.pending); + } + const { header, name, firstMessage } = projectedState; + const threadId = header?.type === "session" ? optionalString(header.id, 256) : undefined; + if (header && threadId && SESSION_ID_PATTERN.test(threadId)) { + const cwd = optionalString(header.cwd, 4_096); + const createdAt = timestampMs(header.timestamp); + summary = { + file: candidate.file, + threadId, + ...(name || firstMessage ? { name: name ?? firstMessage } : {}), + ...(cwd ? { cwd } : {}), + status: "stored", + ...(createdAt !== undefined ? { createdAt } : {}), + updatedAt: candidate.mtimeMs, + recencyAt: candidate.mtimeMs, + source: "pi-cli", + modelProvider: "pi", + archived: false, + canContinue: false, + canArchive: false, + }; + } + } catch { + // Permissions and concurrent file replacement are retryable. Preserve the + // last good index instead of poisoning future append scans. + return cached?.summary; + } + if (cached?.summary?.threadId && cached.summary.threadId !== summary?.threadId) { + threadFileCache.delete(threadCacheKey(cached.storeRoot, cached.summary.threadId)); + } + cacheSummary(candidate.file, { ...candidate, summary, scanState, appendProof }); + if (summary) { + threadFileCache.set(threadCacheKey(candidate.storeRoot, summary.threadId), candidate.file); + } + return summary; +} + +function summaryMatches(summary: PiSessionSummary, needle?: string): boolean { + if (!needle) { + return true; + } + return [summary.threadId, summary.name, summary.cwd].some((field) => + field?.toLocaleLowerCase().includes(needle), + ); +} + +export async function listPiSummaryPage( + env: NodeJS.ProcessEnv, + params: { offset: number; limit: number; searchTerm?: string }, +): Promise<{ summaries: PiSessionSummary[]; hasMore: boolean }> { + const candidates = await piFileCandidates(env); + const activeFiles = new Set(candidates.map((candidate) => candidate.file)); + for (const file of summaryCache.keys()) { + if (!activeFiles.has(file)) { + forgetCachedSummary(file); + } + } + const target = params.offset + params.limit + 1; + const matches: PiSessionSummary[] = []; + const needle = params.searchTerm?.toLocaleLowerCase(); + for ( + let index = 0; + index < candidates.length && matches.length < target; + index += SUMMARY_SCAN_BATCH_SIZE + ) { + const batch = candidates.slice(index, index + SUMMARY_SCAN_BATCH_SIZE); + const summaries = await mapConcurrent(batch, IO_CONCURRENCY, readPiSessionSummary); + for (const summary of summaries) { + if (summary && summaryMatches(summary, needle)) { + matches.push(summary); + if (matches.length >= target) { + break; + } + } + } + } + return { + summaries: matches.slice(params.offset, params.offset + params.limit), + hasMore: matches.length > params.offset + params.limit, + }; +} + +async function findPiSummary( + threadId: string, + env: NodeJS.ProcessEnv, +): Promise { + const candidates = await piFileCandidates(env); + for (let index = 0; index < candidates.length; index += SUMMARY_SCAN_BATCH_SIZE) { + const summaries = await mapConcurrent( + candidates.slice(index, index + SUMMARY_SCAN_BATCH_SIZE), + IO_CONCURRENCY, + readPiSessionSummary, + ); + const match = summaries.find((summary) => summary?.threadId === threadId); + if (match) { + return match; + } + } + return undefined; +} + +export async function readPiSessionById( + threadId: string, + env: NodeJS.ProcessEnv, +): Promise[]> { + const cacheKey = threadCacheKey(piSessionStore(env).root, threadId); + let file = threadFileCache.get(cacheKey); + for (let attempt = 0; attempt < 2; attempt += 1) { + if (!file) { + file = (await findPiSummary(threadId, env))?.file; + } + if (!file) { + throw new Error("Pi session was not found"); + } + try { + const stats = await fs.stat(file); + if (!stats.isFile()) { + throw new Error("Pi session is not a file"); + } + if (stats.size > MAX_SESSION_BYTES) { + throw new RangeError("Pi session exceeds the 32 MiB read safety limit"); + } + const entries = parsePiJsonLines(await fs.readFile(file, "utf8")); + if (entries[0]?.type === "session" && entries[0].id === threadId) { + return entries; + } + } catch (error) { + if (error instanceof RangeError) { + throw error; + } + if (attempt > 0) { + throw new Error("Pi session is unavailable", { cause: error }); + } + } + // The cached path can disappear when Pi replaces or prunes a session file. + threadFileCache.delete(cacheKey); + file = undefined; + } + throw new Error("Pi session changed during read"); +} diff --git a/extensions/opencode/index.ts b/extensions/opencode/index.ts index 3f0486851d63..3ac57e38f527 100644 --- a/extensions/opencode/index.ts +++ b/extensions/opencode/index.ts @@ -16,6 +16,7 @@ import { normalizeOpencodeZenBaseUrl, resolveOpencodeZenModel, } from "./provider-catalog.js"; +import { registerOpenCodeSessionCatalog } from "./session-catalog-plugin.js"; const PROVIDER_ID = "opencode"; const MINIMAX_MODERN_MODEL_MATCHERS = ["minimax-m2.7"] as const; @@ -153,5 +154,6 @@ export default definePluginEntry({ resolveThinkingProfile: ({ modelId }) => resolveClaudeThinkingProfile(modelId), }); api.registerMediaUnderstandingProvider(opencodeMediaUnderstandingProvider); + registerOpenCodeSessionCatalog(api); }, }); diff --git a/extensions/opencode/openclaw.plugin.json b/extensions/opencode/openclaw.plugin.json index 0097ca14157f..07148ac48230 100644 --- a/extensions/opencode/openclaw.plugin.json +++ b/extensions/opencode/openclaw.plugin.json @@ -2,7 +2,7 @@ "id": "opencode", "icon": "https://cdn.simpleicons.org/opencode", "activation": { - "onStartup": false + "onStartup": true }, "providerCatalogEntry": "./provider-discovery.ts", "enabledByDefault": true, @@ -225,6 +225,23 @@ "configSchema": { "type": "object", "additionalProperties": false, - "properties": {} + "properties": { + "sessionCatalog": { + "type": "object", + "additionalProperties": false, + "properties": { + "enabled": { + "type": "boolean", + "default": true + } + } + } + } + }, + "uiHints": { + "sessionCatalog.enabled": { + "label": "OpenCode Session Catalog", + "help": "Auto-detect OpenCode sessions on the Gateway and paired nodes, then show them in the sessions sidebar." + } } } diff --git a/extensions/opencode/session-catalog-plugin.ts b/extensions/opencode/session-catalog-plugin.ts new file mode 100644 index 000000000000..92904aeaa5ba --- /dev/null +++ b/extensions/opencode/session-catalog-plugin.ts @@ -0,0 +1,388 @@ +import { accessSync, constants, statSync } from "node:fs"; +import path from "node:path"; +import process from "node:process"; +import type { + OpenClawPluginApi, + OpenClawPluginNodeHostCommand, + OpenClawPluginNodeInvokePolicy, +} from "openclaw/plugin-sdk/plugin-entry"; +import type { PluginRuntime } from "openclaw/plugin-sdk/plugin-runtime"; +import type { + SessionCatalogHost, + SessionCatalogProvider, + SessionCatalogSession, + SessionCatalogTranscriptItem, + SessionsCatalogReadResult, +} from "openclaw/plugin-sdk/session-catalog"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { + listLocalOpenCodeSessionPage, + optionalOpenCodeString, + readLocalOpenCodeTranscriptPage, + type OpenCodeSessionPage, +} from "./session-catalog.js"; + +export const OPENCODE_SESSIONS_LIST_COMMAND = "opencode.sessions.list.v1"; +export const OPENCODE_SESSION_READ_COMMAND = "opencode.sessions.read.v1"; + +const CAPABILITY = "opencode-sessions"; +const LOCAL_HOST_ID = "gateway"; +const MAX_PAGE_LIMIT = 100; +const MAX_HOSTS = 100; +const MAX_CURSOR_LENGTH = 128; +const MAX_SEARCH_LENGTH = 500; +const NODE_TIMEOUT_MS = 35_000; +const SESSION_ID_PATTERN = /^(?!-)[A-Za-z0-9._:-]{1,256}$/u; +const TRANSCRIPT_ITEM_TYPES = new Set([ + "userMessage", + "agentMessage", + "reasoning", + "toolCall", + "toolResult", + "other", +]); + +function isOptionalString(value: unknown): boolean { + return value === undefined || typeof value === "string"; +} + +function isOptionalNumber(value: unknown): boolean { + return value === undefined || typeof value === "number"; +} + +function isNodeSession(value: unknown): value is SessionCatalogSession { + return ( + isRecord(value) && + typeof value.threadId === "string" && + SESSION_ID_PATTERN.test(value.threadId) && + typeof value.status === "string" && + value.status.length > 0 && + typeof value.archived === "boolean" && + typeof value.canContinue === "boolean" && + typeof value.canArchive === "boolean" && + isOptionalString(value.name) && + isOptionalString(value.cwd) && + isOptionalString(value.source) && + isOptionalString(value.modelProvider) && + isOptionalString(value.cliVersion) && + isOptionalString(value.gitBranch) && + isOptionalString(value.openClawSessionKey) && + isOptionalNumber(value.createdAt) && + isOptionalNumber(value.updatedAt) && + isOptionalNumber(value.recencyAt) + ); +} + +function isNodeTranscriptItem(value: unknown): value is SessionCatalogTranscriptItem { + return ( + isRecord(value) && + typeof value.type === "string" && + TRANSCRIPT_ITEM_TYPES.has(value.type) && + isOptionalString(value.id) && + isOptionalString(value.text) && + isOptionalString(value.timestamp) && + isOptionalString(value.model) && + (value.truncated === undefined || typeof value.truncated === "boolean") + ); +} + +function executableOnPath(command: string, env: NodeJS.ProcessEnv): boolean { + const pathValue = env.PATH ?? env.Path ?? ""; + const delimiter = process.platform === "win32" ? ";" : path.delimiter; + const extensions = + process.platform === "win32" ? (env.PATHEXT ?? ".EXE;.CMD;.BAT;.COM").split(";") : [""]; + for (const directory of pathValue.split(delimiter)) { + for (const extension of extensions) { + if (!directory.trim()) { + continue; + } + const candidate = path.join(directory, `${command}${extension}`); + try { + if (!statSync(candidate).isFile()) { + continue; + } + if (process.platform !== "win32") { + accessSync(candidate, constants.X_OK); + } + return true; + } catch { + // Keep searching PATH. + } + } + } + return false; +} + +function parseNodeParams(paramsJSON?: string | null): unknown { + if (!paramsJSON) { + return undefined; + } + try { + return JSON.parse(paramsJSON) as unknown; + } catch (error) { + throw new Error("OpenCode session parameters must be valid JSON", { cause: error }); + } +} + +function fullConfigCatalogEnabled(config: unknown): boolean { + if (!isRecord(config) || !isRecord(config.plugins) || !isRecord(config.plugins.entries)) { + return true; + } + const entry = config.plugins.entries.opencode; + if (!isRecord(entry) || !isRecord(entry.config) || !isRecord(entry.config.sessionCatalog)) { + return true; + } + return entry.config.sessionCatalog.enabled !== false; +} + +export function isOpenCodeSessionCatalogEnabled(pluginConfig: unknown): boolean { + return ( + !isRecord(pluginConfig) || + !isRecord(pluginConfig.sessionCatalog) || + pluginConfig.sessionCatalog.enabled !== false + ); +} + +export function createOpenCodeSessionNodeHostCommands(): OpenClawPluginNodeHostCommand[] { + const available = ({ config, env }: { config: unknown; env: NodeJS.ProcessEnv }) => + fullConfigCatalogEnabled(config) && executableOnPath("opencode", env); + return [ + { + command: OPENCODE_SESSIONS_LIST_COMMAND, + cap: CAPABILITY, + dangerous: false, + isAvailable: available, + handle: async (paramsJSON) => + JSON.stringify(await listLocalOpenCodeSessionPage(parseNodeParams(paramsJSON))), + }, + { + command: OPENCODE_SESSION_READ_COMMAND, + cap: CAPABILITY, + dangerous: false, + isAvailable: available, + handle: async (paramsJSON) => + JSON.stringify(await readLocalOpenCodeTranscriptPage(parseNodeParams(paramsJSON))), + }, + ]; +} + +export function createOpenCodeSessionNodeInvokePolicies(): OpenClawPluginNodeInvokePolicy[] { + return [ + { + commands: [OPENCODE_SESSIONS_LIST_COMMAND, OPENCODE_SESSION_READ_COMMAND], + defaultPlatforms: ["macos", "linux", "windows"], + handle: (context) => context.invokeNode(), + }, + ]; +} + +function nodeLabel(node: { displayName?: string; remoteIp?: string; nodeId: string }): string { + return node.displayName?.trim() || node.remoteIp?.trim() || node.nodeId; +} + +function unwrapNodePayload(value: unknown): unknown { + return isRecord(value) && typeof value.payloadJSON === "string" + ? (JSON.parse(value.payloadJSON) as unknown) + : value; +} + +type CatalogNode = Awaited>["nodes"][number]; + +async function listOpenCodeNodeHost( + runtime: PluginRuntime, + query: Parameters[0], + node: CatalogNode, +): Promise { + const hostId = `node:${node.nodeId}`; + const common = { + hostId, + label: nodeLabel(node), + kind: "node" as const, + connected: node.connected === true, + nodeId: node.nodeId, + }; + if (node.connected !== true) { + return { + ...common, + sessions: [], + error: { code: "NODE_OFFLINE", message: "Paired node is offline" }, + }; + } + try { + const raw = await runtime.nodes.invoke({ + nodeId: node.nodeId, + command: OPENCODE_SESSIONS_LIST_COMMAND, + params: { + ...(query.limitPerHost ? { limit: query.limitPerHost } : {}), + ...(query.search?.trim() + ? { searchTerm: query.search.trim().slice(0, MAX_SEARCH_LENGTH) } + : {}), + ...(query.cursors?.[hostId] ? { cursor: query.cursors[hostId] } : {}), + }, + timeoutMs: NODE_TIMEOUT_MS, + scopes: ["operator.write"], + }); + return { ...common, ...parseNodeSessionPage(unwrapNodePayload(raw)) }; + } catch { + return { + ...common, + sessions: [], + error: { + code: "NODE_INVOKE_FAILED", + message: "Paired node OpenCode sessions are unavailable", + }, + }; + } +} + +function parseNodeSessionPage(value: unknown): OpenCodeSessionPage { + if ( + !isRecord(value) || + !Array.isArray(value.sessions) || + value.sessions.length > MAX_PAGE_LIMIT + ) { + throw new Error("OpenCode node returned an invalid session page"); + } + if (!value.sessions.every(isNodeSession)) { + throw new Error("OpenCode node returned an invalid session page"); + } + const sessions = value.sessions; + const nextCursor = optionalOpenCodeString(value.nextCursor, MAX_CURSOR_LENGTH); + if (value.nextCursor !== undefined && !nextCursor) { + throw new Error("OpenCode node returned an invalid cursor"); + } + return { sessions, ...(nextCursor ? { nextCursor } : {}) }; +} + +function parseNodeTranscriptPage(value: unknown, threadId: string): SessionsCatalogReadResult { + if ( + !isRecord(value) || + value.threadId !== threadId || + !Array.isArray(value.items) || + value.items.length > MAX_PAGE_LIMIT || + !value.items.every(isNodeTranscriptItem) + ) { + throw new Error("OpenCode node returned an invalid transcript page"); + } + const nextCursor = optionalOpenCodeString(value.nextCursor, MAX_CURSOR_LENGTH); + if (value.nextCursor !== undefined && !nextCursor) { + throw new Error("OpenCode node returned an invalid cursor"); + } + return { + hostId: LOCAL_HOST_ID, + threadId, + items: value.items, + ...(nextCursor ? { nextCursor } : {}), + }; +} + +async function listOpenCodeHosts( + runtime: PluginRuntime, + query: Parameters[0], +): Promise { + const requested = query.hostIds ? new Set(query.hostIds) : undefined; + const searchTerm = query.search?.trim().slice(0, MAX_SEARCH_LENGTH) || undefined; + const hosts: SessionCatalogHost[] = []; + if ((!requested || requested.has(LOCAL_HOST_ID)) && executableOnPath("opencode", process.env)) { + try { + hosts.push({ + hostId: LOCAL_HOST_ID, + label: "Local OpenCode", + kind: "gateway", + connected: true, + ...(await listLocalOpenCodeSessionPage({ + limit: query.limitPerHost, + ...(searchTerm ? { searchTerm } : {}), + cursor: query.cursors?.[LOCAL_HOST_ID], + })), + }); + } catch { + hosts.push({ + hostId: LOCAL_HOST_ID, + label: "Local OpenCode", + kind: "gateway", + connected: true, + sessions: [], + error: { code: "LOCAL_READ_FAILED", message: "Local OpenCode sessions are unavailable" }, + }); + } + } + let nodes: Awaited>["nodes"]; + try { + nodes = (await runtime.nodes.list()).nodes; + } catch { + return hosts; + } + const eligible = nodes + .filter( + (node) => + node.commands?.includes(OPENCODE_SESSIONS_LIST_COMMAND) && + (!requested || requested.has(`node:${node.nodeId}`)), + ) + .toSorted((left, right) => nodeLabel(left).localeCompare(nodeLabel(right))) + .slice(0, MAX_HOSTS - hosts.length); + const nodeHosts = await Promise.all( + eligible.map((node) => listOpenCodeNodeHost(runtime, query, node)), + ); + return [...hosts, ...nodeHosts]; +} + +async function readOpenCodeTranscript( + runtime: PluginRuntime, + request: Parameters[0], +): Promise { + if (request.hostId === LOCAL_HOST_ID) { + return await readLocalOpenCodeTranscriptPage({ + threadId: request.threadId, + ...(request.limit ? { limit: request.limit } : {}), + ...(request.cursor ? { cursor: request.cursor } : {}), + }); + } + if (!request.hostId.startsWith("node:")) { + throw new Error("hostId is invalid"); + } + const nodeId = request.hostId.slice("node:".length); + const node = (await runtime.nodes.list()).nodes.find( + (candidate) => + candidate.nodeId === nodeId && + candidate.connected === true && + candidate.commands?.includes(OPENCODE_SESSION_READ_COMMAND), + ); + if (!node) { + throw new Error("paired-node OpenCode session host is unavailable"); + } + const raw = await runtime.nodes.invoke({ + nodeId, + command: OPENCODE_SESSION_READ_COMMAND, + params: { + threadId: request.threadId, + ...(request.limit ? { limit: request.limit } : {}), + ...(request.cursor ? { cursor: request.cursor } : {}), + }, + timeoutMs: NODE_TIMEOUT_MS, + scopes: ["operator.write"], + }); + return { + ...parseNodeTranscriptPage(unwrapNodePayload(raw), request.threadId), + hostId: request.hostId, + label: nodeLabel(node), + }; +} + +export function registerOpenCodeSessionCatalog(api: OpenClawPluginApi): void { + if (!isOpenCodeSessionCatalogEnabled(api.pluginConfig)) { + return; + } + api.registerSessionCatalog({ + id: "opencode", + label: "OpenCode", + list: async (query) => await listOpenCodeHosts(api.runtime, query), + read: async (request) => await readOpenCodeTranscript(api.runtime, request), + }); + for (const command of createOpenCodeSessionNodeHostCommands()) { + api.registerNodeHostCommand(command); + } + for (const policy of createOpenCodeSessionNodeInvokePolicies()) { + api.registerNodeInvokePolicy(policy); + } +} diff --git a/extensions/opencode/session-catalog.test.ts b/extensions/opencode/session-catalog.test.ts new file mode 100644 index 000000000000..029a2bd65d12 --- /dev/null +++ b/extensions/opencode/session-catalog.test.ts @@ -0,0 +1,406 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + createOpenCodeSessionNodeHostCommands, + isOpenCodeSessionCatalogEnabled, + OPENCODE_SESSIONS_LIST_COMMAND, + OPENCODE_SESSION_READ_COMMAND, + registerOpenCodeSessionCatalog, +} from "./session-catalog-plugin.js"; +import { + listLocalOpenCodeSessionPage, + readLocalOpenCodeTranscriptPage, +} from "./session-catalog.js"; + +const temporaryDirectories: string[] = []; +const originalPath = process.env.PATH; +const originalUnrelatedEnv = process.env.CATALOG_UNRELATED_ENV; + +async function installFakeOpenCode(assistantText = "hi"): Promise { + const directory = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-opencode-catalog-")); + temporaryDirectories.push(directory); + const executable = path.join(directory, "opencode"); + const session = { + id: "ses_test", + title: "Catalog session", + created: 1_700_000_000_000, + updated: 1_700_000_001_000, + projectId: "project", + directory: "/workspace", + }; + const exported = { + info: session, + messages: [ + { + info: { + id: "msg_user", + role: "user", + time: { created: 1_700_000_000_000 }, + model: { providerID: "anthropic", modelID: "claude" }, + }, + parts: [{ id: "prt_user", type: "text", text: "hello" }], + }, + { + info: { + id: "msg_assistant", + role: "assistant", + time: { created: 1_700_000_001_000 }, + providerID: "anthropic", + modelID: "claude", + }, + parts: [ + { id: "prt_reason", type: "reasoning", text: "thinking" }, + { id: "prt_answer", type: "text", text: assistantText }, + { + id: "prt_tool", + type: "tool", + tool: "bash", + state: { status: "completed", input: { command: "pwd" }, output: "/workspace" }, + }, + ], + }, + ], + }; + await fs.writeFile( + executable, + `#!/usr/bin/env node +const args = process.argv.slice(2); +if (process.env.CATALOG_UNRELATED_ENV) process.exit(3); +if (args[0] === "--pure" && args[1] === "db" && args.includes("--format") && args.includes("json")) { + process.stdout.write(${JSON.stringify(JSON.stringify([session]))}); +} else if (args[0] === "--pure" && args[1] === "export" && args[2] === "ses_test") { + process.stdout.write(${JSON.stringify(JSON.stringify(exported))}); +} else { + process.exitCode = 2; +} +`, + ); + await fs.chmod(executable, 0o755); + process.env.PATH = `${directory}${path.delimiter}${originalPath ?? ""}`; + process.env.CATALOG_UNRELATED_ENV = "present"; + return directory; +} + +afterEach(async () => { + process.env.PATH = originalPath; + if (originalUnrelatedEnv === undefined) { + delete process.env.CATALOG_UNRELATED_ENV; + } else { + process.env.CATALOG_UNRELATED_ENV = originalUnrelatedEnv; + } + await Promise.all( + temporaryDirectories.splice(0).map(async (directory) => { + await fs.rm(directory, { recursive: true, force: true }); + }), + ); +}); + +describe("OpenCode session catalog", () => { + it.runIf(process.platform !== "win32")( + "lists and reads sessions through the official CLI JSON surfaces", + async () => { + await installFakeOpenCode(); + const listed = await listLocalOpenCodeSessionPage({ limit: 20 }); + expect(listed).toEqual({ + sessions: [ + expect.objectContaining({ + threadId: "ses_test", + name: "Catalog session", + cwd: "/workspace", + source: "opencode-cli", + canContinue: false, + }), + ], + }); + + const transcript = await readLocalOpenCodeTranscriptPage({ + threadId: "ses_test", + limit: 20, + }); + expect(transcript.items.map((item) => [item.type, item.text])).toEqual([ + ["userMessage", "hello"], + ["reasoning", "thinking"], + ["agentMessage", "hi"], + ["toolCall", 'bash\n{"command":"pwd"}'], + ["toolResult", "/workspace"], + ]); + const itemIds = transcript.items.flatMap((item) => (item.id ? [item.id] : [])); + expect(new Set(itemIds).size).toBe(itemIds.length); + + const latest = await readLocalOpenCodeTranscriptPage({ threadId: "ses_test", limit: 2 }); + expect(latest.items.map((item) => item.type)).toEqual(["toolCall", "toolResult"]); + expect(latest.nextCursor).toBeTruthy(); + const older = await readLocalOpenCodeTranscriptPage({ + threadId: "ses_test", + limit: 2, + cursor: latest.nextCursor, + }); + expect(older.items.map((item) => item.type)).toEqual(["reasoning", "agentMessage"]); + await expect(listLocalOpenCodeSessionPage({ cursor: " " })).rejects.toThrow( + "cursor is invalid", + ); + await expect( + readLocalOpenCodeTranscriptPage({ threadId: "ses_test", cursor: 123 }), + ).rejects.toThrow("cursor is invalid"); + await expect(readLocalOpenCodeTranscriptPage({ threadId: "--help" })).rejects.toThrow( + "threadId is invalid", + ); + + let provider: Parameters[0] | undefined; + registerOpenCodeSessionCatalog({ + pluginConfig: {}, + runtime: { nodes: { list: vi.fn().mockResolvedValue({ nodes: [] }) } }, + registerSessionCatalog: (value: NonNullable) => { + provider = value; + }, + registerNodeHostCommand: vi.fn(), + registerNodeInvokePolicy: vi.fn(), + } as unknown as OpenClawPluginApi); + await expect( + provider!.read({ hostId: "gateway", threadId: "ses_test", limit: 2 }), + ).resolves.toMatchObject({ threadId: "ses_test", items: expect.any(Array) }); + await expect(provider!.list({ search: " " })).resolves.toEqual([ + expect.objectContaining({ hostId: "gateway", sessions: [expect.any(Object)] }), + ]); + await expect(provider!.list({ search: "x".repeat(501) })).resolves.toEqual([ + expect.objectContaining({ hostId: "gateway", sessions: [] }), + ]); + }, + ); + + it.runIf(process.platform !== "win32")( + "keeps oversized transcript items below the node payload budget", + async () => { + await installFakeOpenCode("x".repeat(600 * 1024)); + const transcript = await readLocalOpenCodeTranscriptPage({ + threadId: "ses_test", + limit: 20, + }); + const answer = transcript.items.find((item) => item.type === "agentMessage"); + expect(answer?.text?.endsWith("…")).toBe(true); + expect(Buffer.byteLength(JSON.stringify(transcript), "utf8")).toBeLessThan(20 * 1024 * 1024); + }, + ); + + it.runIf(process.platform !== "win32")( + "auto-detects the CLI and honors the node-local Web UI switch", + async () => { + const directory = await installFakeOpenCode(); + const commands = createOpenCodeSessionNodeHostCommands(); + expect(commands.map((command) => command.command)).toEqual([ + OPENCODE_SESSIONS_LIST_COMMAND, + OPENCODE_SESSION_READ_COMMAND, + ]); + expect( + commands.every((command) => + command.isAvailable?.({ config: {}, env: { PATH: directory } } as never), + ), + ).toBe(true); + expect( + commands.every((command) => + command.isAvailable?.({ + config: { + plugins: { + entries: { opencode: { config: { sessionCatalog: { enabled: false } } } }, + }, + }, + env: { PATH: directory }, + } as never), + ), + ).toBe(false); + expect( + commands.every((command) => + command.isAvailable?.({ + config: {}, + env: { PATH: path.join(directory, "missing") }, + } as never), + ), + ).toBe(false); + }, + ); + + it("does not register the catalog when explicitly disabled", () => { + const registerSessionCatalog = vi.fn(); + const api = { + pluginConfig: { sessionCatalog: { enabled: false } }, + registerSessionCatalog, + } as unknown as OpenClawPluginApi; + registerOpenCodeSessionCatalog(api); + expect(isOpenCodeSessionCatalogEnabled(api.pluginConfig)).toBe(false); + expect(registerSessionCatalog).not.toHaveBeenCalled(); + }); + + it("bridges paired-node list and read requests without undefined transport fields", async () => { + let provider: Parameters[0] | undefined; + const invoke = vi + .fn() + .mockResolvedValueOnce({ + payloadJSON: JSON.stringify({ + sessions: [ + { + threadId: "ses_remote", + status: "stored", + source: "opencode-cli", + archived: false, + canContinue: false, + canArchive: false, + }, + ], + }), + }) + .mockResolvedValueOnce({ + payloadJSON: JSON.stringify({ + threadId: "ses_remote", + items: [{ type: "agentMessage", text: "remote answer" }], + }), + }); + const api = { + pluginConfig: {}, + runtime: { + nodes: { + list: vi.fn().mockResolvedValue({ + nodes: [ + { + nodeId: "node-1", + displayName: "Remote", + connected: true, + commands: [OPENCODE_SESSIONS_LIST_COMMAND, OPENCODE_SESSION_READ_COMMAND], + }, + ], + }), + invoke, + }, + }, + registerSessionCatalog: (value: NonNullable) => { + provider = value; + }, + registerNodeHostCommand: vi.fn(), + registerNodeInvokePolicy: vi.fn(), + } as unknown as OpenClawPluginApi; + + registerOpenCodeSessionCatalog(api); + const catalog = provider; + expect(catalog).toBeDefined(); + await catalog!.list({ hostIds: ["node:node-1"], search: " " }); + await catalog!.read({ hostId: "node:node-1", threadId: "ses_remote" }); + + expect(invoke).toHaveBeenNthCalledWith(1, { + nodeId: "node-1", + command: OPENCODE_SESSIONS_LIST_COMMAND, + params: {}, + timeoutMs: 35_000, + scopes: ["operator.write"], + }); + expect(invoke).toHaveBeenNthCalledWith(2, { + nodeId: "node-1", + command: OPENCODE_SESSION_READ_COMMAND, + params: { threadId: "ses_remote" }, + timeoutMs: 35_000, + scopes: ["operator.write"], + }); + + invoke.mockResolvedValueOnce({ + payloadJSON: JSON.stringify({ + sessions: [ + { + threadId: 123, + status: "stored", + archived: false, + canContinue: false, + canArchive: false, + }, + ], + }), + }); + await expect(catalog!.list({ hostIds: ["node:node-1"] })).resolves.toEqual([ + expect.objectContaining({ + error: { code: "NODE_INVOKE_FAILED", message: expect.any(String) }, + }), + ]); + + invoke.mockResolvedValueOnce({ + payloadJSON: JSON.stringify({ + sessions: [ + { + threadId: "--help", + status: "stored", + archived: false, + canContinue: false, + canArchive: false, + }, + ], + }), + }); + await expect(catalog!.list({ hostIds: ["node:node-1"] })).resolves.toEqual([ + expect.objectContaining({ + error: { code: "NODE_INVOKE_FAILED", message: expect.any(String) }, + }), + ]); + + invoke.mockResolvedValueOnce({ + payloadJSON: JSON.stringify({ + threadId: "ses_remote", + items: [{ type: "invalid", text: "bad" }], + }), + }); + await expect(catalog!.read({ hostId: "node:node-1", threadId: "ses_remote" })).rejects.toThrow( + "invalid transcript page", + ); + }); + + it("fans out paired-node listing instead of blocking later hosts", async () => { + let provider: Parameters[0] | undefined; + let releaseSlow: ((value: unknown) => void) | undefined; + const slow = new Promise((resolve) => { + releaseSlow = resolve; + }); + const page = (threadId: string) => ({ + payloadJSON: JSON.stringify({ + sessions: [ + { + threadId, + status: "stored", + archived: false, + canContinue: false, + canArchive: false, + }, + ], + }), + }); + const invoke = vi.fn(({ nodeId }: { nodeId: string }) => + nodeId === "node-a" ? slow : Promise.resolve(page("session-b")), + ); + const api = { + pluginConfig: {}, + runtime: { + nodes: { + list: vi.fn().mockResolvedValue({ + nodes: ["node-a", "node-b"].map((nodeId) => ({ + nodeId, + connected: true, + commands: [OPENCODE_SESSIONS_LIST_COMMAND], + })), + }), + invoke, + }, + }, + registerSessionCatalog: (value: NonNullable) => { + provider = value; + }, + registerNodeHostCommand: vi.fn(), + registerNodeInvokePolicy: vi.fn(), + } as unknown as OpenClawPluginApi; + registerOpenCodeSessionCatalog(api); + + const listing = provider!.list({}); + await vi.waitFor(() => expect(invoke).toHaveBeenCalledTimes(2)); + releaseSlow?.(page("session-a")); + await expect(listing).resolves.toEqual([ + expect.objectContaining({ nodeId: "node-a", sessions: [expect.any(Object)] }), + expect.objectContaining({ nodeId: "node-b", sessions: [expect.any(Object)] }), + ]); + }); +}); diff --git a/extensions/opencode/session-catalog.ts b/extensions/opencode/session-catalog.ts new file mode 100644 index 000000000000..4c7b0c6c4bcd --- /dev/null +++ b/extensions/opencode/session-catalog.ts @@ -0,0 +1,457 @@ +import { spawn } from "node:child_process"; +import process from "node:process"; +import type { + SessionCatalogSession, + SessionCatalogTranscriptItem, + SessionsCatalogReadResult, +} from "openclaw/plugin-sdk/session-catalog"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { + materializeWindowsSpawnProgram, + resolveWindowsSpawnProgram, +} from "openclaw/plugin-sdk/windows-spawn"; + +const LOCAL_HOST_ID = "gateway"; +const DEFAULT_PAGE_LIMIT = 20; +const MAX_PAGE_LIMIT = 100; +const MAX_SEARCH_LENGTH = 500; +const MAX_CURSOR_LENGTH = 128; +const MAX_CLI_LIST_SESSIONS = 10_000; +const MAX_CLI_OUTPUT_BYTES = 32 * 1024 * 1024; +const MAX_TRANSCRIPT_ITEM_BYTES = 512 * 1024; +const MAX_TRANSCRIPT_PAGE_BYTES = 20 * 1024 * 1024; +const CLI_TIMEOUT_MS = 30_000; +const SESSION_ID_PATTERN = /^(?!-)[A-Za-z0-9._:-]{1,256}$/u; +const SAFE_ENV_KEYS = [ + "APPDATA", + "COMSPEC", + "HOME", + "LANG", + "LC_ALL", + "LOCALAPPDATA", + "OPENCODE_DB", + "PATH", + "Path", + "PATHEXT", + "SYSTEMROOT", + "TEMP", + "TMP", + "TMPDIR", + "USERPROFILE", + "WINDIR", + "XDG_CACHE_HOME", + "XDG_CONFIG_HOME", + "XDG_DATA_HOME", + "XDG_STATE_HOME", +] as const; + +export type OpenCodeSessionPage = { + sessions: SessionCatalogSession[]; + nextCursor?: string; +}; + +type OpenCodeListParams = { + searchTerm?: string; + limit?: number; + cursor?: string; +}; + +type OpenCodeReadParams = { + threadId: string; + limit?: number; + cursor?: string; +}; + +export function optionalOpenCodeString(value: unknown, maxLength: number): string | undefined { + if (typeof value !== "string") { + return undefined; + } + const trimmed = value.trim(); + return trimmed && trimmed.length <= maxLength ? trimmed : undefined; +} + +function boundedLimit(value: unknown, fallback = DEFAULT_PAGE_LIMIT): number { + if (value === undefined) { + return fallback; + } + if (!Number.isInteger(value) || Number(value) < 1 || Number(value) > MAX_PAGE_LIMIT) { + throw new Error(`limit must be an integer between 1 and ${String(MAX_PAGE_LIMIT)}`); + } + return Number(value); +} + +function encodeCursor(offset: number): string { + return Buffer.from(JSON.stringify({ offset }), "utf8").toString("base64url"); +} + +function decodeCursor(value: unknown): number { + if (value === undefined) { + return 0; + } + const cursor = optionalOpenCodeString(value, MAX_CURSOR_LENGTH); + if (!cursor) { + throw new Error("cursor is invalid"); + } + try { + const parsed = JSON.parse(Buffer.from(cursor, "base64url").toString("utf8")) as unknown; + if (!isRecord(parsed) || !Number.isInteger(parsed.offset) || Number(parsed.offset) < 0) { + throw new Error("invalid offset"); + } + return Number(parsed.offset); + } catch (error) { + throw new Error("cursor is invalid", { cause: error }); + } +} + +function truncateUtf8(text: string, maxBytes: number): string { + if (Buffer.byteLength(text, "utf8") <= maxBytes) { + return text; + } + let low = 0; + let high = text.length; + while (low < high) { + const middle = Math.ceil((low + high) / 2); + if (Buffer.byteLength(text.slice(0, middle), "utf8") <= maxBytes - 3) { + low = middle; + } else { + high = middle - 1; + } + } + const end = low > 0 && /[\uD800-\uDBFF]/u.test(text.charAt(low - 1)) ? low - 1 : low; + return `${text.slice(0, end)}…`; +} + +function transcriptPage( + items: SessionCatalogTranscriptItem[], + limit: number, + offset: number, +): { items: SessionCatalogTranscriptItem[]; nextCursor?: string } { + const end = Math.max(0, items.length - offset); + const start = Math.max(0, end - limit); + const page: SessionCatalogTranscriptItem[] = []; + let pageBytes = 2; + for (let index = end - 1; index >= start; index -= 1) { + const item = items[index]; + if (!item) { + continue; + } + const bounded: SessionCatalogTranscriptItem = { + ...item, + text: truncateUtf8(item.text ?? "", MAX_TRANSCRIPT_ITEM_BYTES), + }; + const itemBytes = Buffer.byteLength(JSON.stringify(bounded), "utf8") + 1; + if (page.length > 0 && pageBytes + itemBytes > MAX_TRANSCRIPT_PAGE_BYTES) { + break; + } + page.unshift(bounded); + pageBytes += itemBytes; + } + const consumed = offset + page.length; + return { + items: page, + ...(consumed < items.length ? { nextCursor: encodeCursor(consumed) } : {}), + }; +} + +function parseListParams( + value: unknown, +): Required> & OpenCodeListParams { + if (value === undefined || value === null) { + return { limit: DEFAULT_PAGE_LIMIT }; + } + if (!isRecord(value)) { + throw new Error("OpenCode session list parameters must be an object"); + } + const unknown = Object.keys(value).find( + (key) => !["searchTerm", "limit", "cursor"].includes(key), + ); + if (unknown) { + throw new Error(`unknown OpenCode session list parameter: ${unknown}`); + } + const searchTerm = optionalOpenCodeString(value.searchTerm, MAX_SEARCH_LENGTH); + if (value.searchTerm !== undefined && !searchTerm) { + throw new Error("searchTerm is invalid"); + } + const cursor = optionalOpenCodeString(value.cursor, MAX_CURSOR_LENGTH); + if (value.cursor !== undefined && !cursor) { + throw new Error("cursor is invalid"); + } + return { + limit: boundedLimit(value.limit), + ...(searchTerm ? { searchTerm } : {}), + ...(cursor ? { cursor } : {}), + }; +} + +function parseReadParams( + value: unknown, +): Required> & OpenCodeReadParams { + if (!isRecord(value)) { + throw new Error("OpenCode session read parameters must be an object"); + } + const unknown = Object.keys(value).find((key) => !["threadId", "limit", "cursor"].includes(key)); + if (unknown) { + throw new Error(`unknown OpenCode session read parameter: ${unknown}`); + } + const threadId = optionalOpenCodeString(value.threadId, 256); + if (!threadId || !SESSION_ID_PATTERN.test(threadId)) { + throw new Error("threadId is invalid"); + } + const cursor = optionalOpenCodeString(value.cursor, MAX_CURSOR_LENGTH); + if (value.cursor !== undefined && !cursor) { + throw new Error("cursor is invalid"); + } + return { + threadId, + limit: boundedLimit(value.limit), + ...(cursor ? { cursor } : {}), + }; +} + +function resolveSpawnInvocation(args: string[]): { + command: string; + argv: string[]; + shell?: boolean; + windowsHide?: boolean; +} { + const program = resolveWindowsSpawnProgram({ + command: "opencode", + platform: process.platform, + env: process.env, + execPath: process.execPath, + packageName: "opencode-ai", + }); + return materializeWindowsSpawnProgram(program, args); +} + +async function runOpenCode(args: string[]): Promise { + const invocation = resolveSpawnInvocation(args); + const env: NodeJS.ProcessEnv = { OPENCODE_PURE: "1", NO_COLOR: "1" }; + for (const key of SAFE_ENV_KEYS) { + if (process.env[key] !== undefined) { + env[key] = process.env[key]; + } + } + const child = spawn(invocation.command, invocation.argv, { + env, + shell: invocation.shell, + windowsHide: invocation.windowsHide, + stdio: ["ignore", "pipe", "pipe"], + }); + const stdout: Buffer[] = []; + const stderr: Buffer[] = []; + let bytes = 0; + let overflow = false; + const timeout = setTimeout(() => child.kill("SIGKILL"), CLI_TIMEOUT_MS); + timeout.unref?.(); + const collect = (target: Buffer[], chunk: Buffer) => { + bytes += chunk.length; + if (bytes > MAX_CLI_OUTPUT_BYTES) { + overflow = true; + child.kill("SIGKILL"); + return; + } + target.push(chunk); + }; + child.stdout.on("data", (chunk: Buffer) => collect(stdout, chunk)); + child.stderr.on("data", (chunk: Buffer) => collect(stderr, chunk)); + const exitCode = await new Promise((resolve, reject) => { + child.on("error", reject); + child.on("close", resolve); + }).finally(() => clearTimeout(timeout)); + if (overflow) { + throw new Error("OpenCode session output exceeded the safety limit"); + } + if (exitCode !== 0) { + const detail = Buffer.concat(stderr).toString("utf8").trim(); + throw new Error(detail || `OpenCode exited with code ${String(exitCode)}`); + } + return Buffer.concat(stdout).toString("utf8"); +} + +function parseOpenCodeSession(value: unknown): SessionCatalogSession | undefined { + if (!isRecord(value)) { + return undefined; + } + const threadId = optionalOpenCodeString(value.id, 256); + if (!threadId || !SESSION_ID_PATTERN.test(threadId)) { + return undefined; + } + const name = optionalOpenCodeString(value.title, 1_000); + const cwd = optionalOpenCodeString(value.directory, 4_096); + const createdAt = + typeof value.created === "number" && Number.isFinite(value.created) ? value.created : undefined; + const updatedAt = + typeof value.updated === "number" && Number.isFinite(value.updated) ? value.updated : undefined; + return { + threadId, + ...(name ? { name } : {}), + ...(cwd ? { cwd } : {}), + status: "stored", + ...(createdAt !== undefined ? { createdAt } : {}), + ...(updatedAt !== undefined ? { updatedAt, recencyAt: updatedAt } : {}), + source: "opencode-cli", + modelProvider: "opencode", + archived: false, + canContinue: false, + canArchive: false, + }; +} + +export async function listLocalOpenCodeSessionPage(value?: unknown): Promise { + const params = parseListParams(value); + const offset = decodeCursor(params.cursor); + const requestedCount = params.searchTerm + ? MAX_CLI_LIST_SESSIONS + : Math.min(MAX_CLI_LIST_SESSIONS, offset + params.limit + 1); + const query = [ + "SELECT id, title, time_created AS created, time_updated AS updated,", + "project_id AS projectId, directory FROM session", + "WHERE parent_id IS NULL AND time_archived IS NULL", + `ORDER BY time_updated DESC, id DESC LIMIT ${String(requestedCount)}`, + ].join(" "); + const output = await runOpenCode(["--pure", "db", query, "--format", "json"]); + const parsed = output.trim() ? (JSON.parse(output) as unknown) : []; + if (!Array.isArray(parsed) || parsed.length > MAX_CLI_LIST_SESSIONS) { + throw new Error("OpenCode returned an invalid session list"); + } + const needle = params.searchTerm?.toLocaleLowerCase(); + const sessions = parsed + .flatMap((entry) => { + const session = parseOpenCodeSession(entry); + return session ? [session] : []; + }) + .filter((session) => { + if (!needle) { + return true; + } + return [session.threadId, session.name, session.cwd].some((field) => + field?.toLocaleLowerCase().includes(needle), + ); + }); + const page = sessions.slice(offset, offset + params.limit); + return { + sessions: page, + ...(offset + page.length < sessions.length + ? { nextCursor: encodeCursor(offset + page.length) } + : {}), + }; +} + +function jsonText(value: unknown, maxLength = 20_000): string | undefined { + try { + const text = JSON.stringify(value); + return text.length > maxLength ? `${text.slice(0, maxLength)}…` : text; + } catch { + return undefined; + } +} + +function timestampFromInfo(info: Record): string | undefined { + if (!isRecord(info.time) || typeof info.time.created !== "number") { + return undefined; + } + const date = new Date(info.time.created); + return Number.isNaN(date.getTime()) ? undefined : date.toISOString(); +} + +function openCodeTranscriptItems(value: unknown): SessionCatalogTranscriptItem[] { + if (!isRecord(value) || !Array.isArray(value.messages)) { + throw new Error("OpenCode returned an invalid session export"); + } + return value.messages.flatMap((message): SessionCatalogTranscriptItem[] => { + if (!isRecord(message) || !isRecord(message.info) || !Array.isArray(message.parts)) { + return []; + } + const info = message.info; + const role = info.role; + const messageId = optionalOpenCodeString(info.id, 256); + const timestamp = timestampFromInfo(info); + const modelId = + role === "assistant" + ? optionalOpenCodeString(info.modelID, 256) + : isRecord(info.model) + ? optionalOpenCodeString(info.model.modelID, 256) + : undefined; + const providerId = + role === "assistant" + ? optionalOpenCodeString(info.providerID, 256) + : isRecord(info.model) + ? optionalOpenCodeString(info.model.providerID, 256) + : undefined; + const model = providerId && modelId ? `${providerId}/${modelId}` : modelId; + return message.parts.flatMap((part, partIndex): SessionCatalogTranscriptItem[] => { + if (!isRecord(part)) { + return []; + } + const id = + optionalOpenCodeString(part.id, 256) ?? + (messageId ? `${messageId}:${String(partIndex)}` : undefined); + const common = { + ...(id ? { id } : {}), + ...(timestamp ? { timestamp } : {}), + ...(model ? { model } : {}), + }; + if (part.type === "text" && typeof part.text === "string") { + return [ + { ...common, type: role === "user" ? "userMessage" : "agentMessage", text: part.text }, + ]; + } + if (part.type === "reasoning" && typeof part.text === "string") { + return [{ ...common, type: "reasoning", text: part.text }]; + } + if (part.type === "tool") { + const tool = optionalOpenCodeString(part.tool, 256) ?? "tool"; + const state = isRecord(part.state) ? part.state : undefined; + const callText = state && "input" in state ? jsonText(state.input) : undefined; + const resultText = + state?.status === "completed" && typeof state.output === "string" + ? state.output + : state?.status === "error" && typeof state.error === "string" + ? state.error + : undefined; + return [ + { ...common, type: "toolCall", text: callText ? `${tool}\n${callText}` : tool }, + ...(resultText + ? [ + { + ...common, + ...(id ? { id: `${id}:result` } : {}), + type: "toolResult" as const, + text: resultText, + }, + ] + : []), + ]; + } + if (part.type === "file") { + const filename = optionalOpenCodeString(part.filename, 1_000); + const mime = optionalOpenCodeString(part.mime, 256); + return [ + { + ...common, + type: "other", + text: `[Attachment${filename ? `: ${filename}` : ""}${mime ? ` (${mime})` : ""}]`, + }, + ]; + } + return []; + }); + }); +} + +export async function readLocalOpenCodeTranscriptPage( + value: unknown, +): Promise { + const params = parseReadParams(value); + const offset = decodeCursor(params.cursor); + const output = await runOpenCode(["--pure", "export", params.threadId]); + const items = openCodeTranscriptItems(JSON.parse(output) as unknown); + const page = transcriptPage(items, params.limit, offset); + return { + hostId: LOCAL_HOST_ID, + label: "Local OpenCode", + threadId: params.threadId, + ...page, + }; +} diff --git a/src/gateway/server-plugins.test.ts b/src/gateway/server-plugins.test.ts index 0df01c877972..1f4c3116f265 100644 --- a/src/gateway/server-plugins.test.ts +++ b/src/gateway/server-plugins.test.ts @@ -1009,6 +1009,34 @@ describe("loadGatewayPlugins", () => { expect(getLastDispatchedClientInternal().pluginRuntimeOwnerId).toBe("google-meet"); }); + test("honors trusted plugin node scopes inside a narrower Gateway request", async () => { + loadOpenClawPlugins.mockReturnValue(addLoadedPlugin(createRegistry([]), { id: "opencode" })); + loadGatewayStartupPluginsForTest(); + const scope = { + context: createTestContext("nodes-invoke-read-caller"), + client: { + connect: { scopes: ["operator.read"] }, + } as GatewayRequestOptions["client"], + isWebchatConnect: () => false, + } satisfies PluginRuntimeGatewayRequestScope; + const runtime = runtimeModule.createPluginRuntime({ allowGatewaySubagentBinding: true }); + + await gatewayRequestScopeModule.withPluginRuntimeGatewayRequestScope(scope, () => + gatewayRequestScopeModule.withPluginRuntimePluginScope( + { pluginId: "opencode", pluginOrigin: "bundled" }, + () => + runtime.nodes.invoke({ + nodeId: "node-1", + command: "opencode.sessions.list.v1", + scopes: ["operator.write"], + }), + ), + ); + + expect(getLastDispatchedClientScopes()).toEqual(["operator.write"]); + expect(getLastDispatchedClientInternal().pluginRuntimeOwnerId).toBe("opencode"); + }); + test("dispatches gateway methods with the trusted plugin identity", async () => { loadOpenClawPlugins.mockReturnValue(addLoadedPlugin(createRegistry([]), { id: "google-meet" })); loadGatewayStartupPluginsForTest(); diff --git a/src/gateway/server-plugins.ts b/src/gateway/server-plugins.ts index a7b76171a816..84920d85d48f 100644 --- a/src/gateway/server-plugins.ts +++ b/src/gateway/server-plugins.ts @@ -254,13 +254,10 @@ function resolveRuntimeNodeInvokeSyntheticScopes(params: { pluginTrustedOfficialInstall?: boolean; requestedScopes?: OperatorScope[]; }): OperatorScope[] | undefined { - if (!params.requestedScopes) { - return undefined; - } - if (!canTrustedOfficialPluginRequestScopes(params)) { - return undefined; - } - return params.requestedScopes; + // Requested scopes may replace caller scopes, so only bundled or trusted official plugins qualify. + return params.requestedScopes && canTrustedOfficialPluginRequestScopes(params) + ? params.requestedScopes + : undefined; } function mergeGatewayClientInternal( @@ -710,7 +707,7 @@ export function createGatewayNodesRuntime(): PluginRuntime["nodes"] { }, { ...(pluginId ? { pluginRuntimeOwnerId: pluginId } : {}), - ...(syntheticScopes ? { syntheticScopes } : {}), + ...(syntheticScopes ? { forceSyntheticClient: true, syntheticScopes } : {}), }, ); return payload; diff --git a/ui/src/e2e/external-session-catalogs.e2e.test.ts b/ui/src/e2e/external-session-catalogs.e2e.test.ts new file mode 100644 index 000000000000..41122cd11a29 --- /dev/null +++ b/ui/src/e2e/external-session-catalogs.e2e.test.ts @@ -0,0 +1,136 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { chromium, type Browser } from "playwright"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + canRunPlaywrightChromium, + installMockGateway, + resolvePlaywrightChromiumExecutablePath, + startControlUiE2eServer, + type ControlUiE2eServer, +} from "../test-helpers/control-ui-e2e.ts"; + +const executablePath = resolvePlaywrightChromiumExecutablePath(chromium.executablePath()); +const available = canRunPlaywrightChromium(executablePath); +const allowMissing = process.env.OPENCLAW_UI_E2E_ALLOW_MISSING_CHROMIUM === "1"; +const suite = available || !allowMissing ? describe : describe.skip; + +let browser: Browser; +let server: ControlUiE2eServer; + +suite("OpenCode and Pi external session catalogs", () => { + beforeAll(async () => { + if (!available) { + throw new Error(`Playwright Chromium is unavailable at ${executablePath}`); + } + server = await startControlUiE2eServer(); + browser = await chromium.launch({ executablePath }); + }); + + afterAll(async () => { + await browser?.close(); + await server?.close(); + }); + + it("shows both paired-node catalogs and opens their view-only transcripts", async () => { + const page = await browser.newPage({ viewport: { width: 1440, height: 900 } }); + const gateway = await installMockGateway(page, { + featureMethods: [ + "chat.metadata", + "chat.startup", + "sessions.catalog.list", + "sessions.catalog.read", + ], + methodResponses: { + "sessions.catalog.list": { + catalogs: [ + { + id: "opencode", + label: "OpenCode", + capabilities: { continueSession: false, archive: false }, + hosts: [ + { + hostId: "node:devbox", + label: "Dev Box", + kind: "node", + connected: true, + nodeId: "devbox", + sessions: [ + { + threadId: "opencode-1", + name: "OpenCode release review", + status: "stored", + canContinue: false, + canArchive: false, + }, + ], + }, + ], + }, + { + id: "pi", + label: "Pi", + capabilities: { continueSession: false, archive: false }, + hosts: [ + { + hostId: "node:devbox", + label: "Dev Box", + kind: "node", + connected: true, + nodeId: "devbox", + sessions: [ + { + threadId: "pi-1", + name: "Pi architecture notes", + status: "stored", + canContinue: false, + canArchive: false, + }, + ], + }, + ], + }, + ], + }, + "sessions.catalog.read": { + cases: [ + { + match: { catalogId: "opencode", threadId: "opencode-1" }, + response: { + hostId: "node:devbox", + threadId: "opencode-1", + items: [{ type: "agentMessage", text: "OpenCode transcript loaded" }], + }, + }, + { + match: { catalogId: "pi", threadId: "pi-1" }, + response: { + hostId: "node:devbox", + threadId: "pi-1", + items: [{ type: "agentMessage", text: "Pi transcript loaded" }], + }, + }, + ], + }, + }, + }); + + await page.goto(`${server.baseUrl}chat`); + await page.getByText("OpenCode release review", { exact: true }).click(); + await expect.poll(() => page.getByText("OpenCode transcript loaded").count()).toBe(1); + await page.getByText("Pi architecture notes", { exact: true }).click(); + await expect.poll(() => page.getByText("Pi transcript loaded").count()).toBe(1); + expect(await page.locator(".agent-chat__composer-combobox > textarea").isDisabled()).toBe(true); + expect(await gateway.getRequests("sessions.catalog.read")).toHaveLength(2); + + const artifactDir = process.env.OPENCLAW_UI_E2E_ARTIFACT_DIR?.trim(); + if (artifactDir) { + await fs.mkdir(artifactDir, { recursive: true }); + await page.screenshot({ + path: path.join(artifactDir, "external-session-catalogs.png"), + fullPage: true, + }); + } + await page.close(); + }); +});