From e19841c725ff1a55f2620009abcaea1095268860 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 25 Jul 2026 08:35:47 -0700 Subject: [PATCH] feat: continue Pi and OpenCode sessions from the session catalog (#113718) * feat(plugins): add acpSessionBinding seam for ACP session adoption * feat(acpx,opencode): adopt existing Pi and OpenCode sessions from the catalog * fix(ci): satisfy dead-code and temp-path guards * docs(acpx): record why pi-acp session-root resolution stays strict --- docs/plugins/reference/acpx.md | 11 +- docs/plugins/reference/opencode.md | 15 +- docs/plugins/sdk-runtime.md | 4 +- .../acpx/src/pi-session-catalog-plugin.ts | 150 +++++++- .../src/pi-session-catalog.test-support.ts | 163 +++++++++ .../acpx/src/pi-session-catalog.test.ts | 339 ++++++++++-------- extensions/acpx/src/pi-session-paths.test.ts | 72 ++++ extensions/acpx/src/pi-session-paths.ts | 16 + extensions/acpx/src/pi-session-store.ts | 37 +- extensions/opencode/session-catalog-plugin.ts | 128 ++++++- extensions/opencode/session-catalog-shared.ts | 2 + extensions/opencode/session-catalog.test.ts | 143 +++++++- extensions/opencode/session-catalog.ts | 2 +- src/config/sessions/types.ts | 8 + src/plugin-sdk/acp-runtime.test.ts | 49 ++- src/plugin-sdk/acp-runtime.ts | 28 +- src/plugin-sdk/session-catalog.ts | 6 + .../test-helpers/plugin-runtime-mock.ts | 20 +- src/plugins/registry-runtime.ts | 21 +- src/plugins/registry.runtime-config.test.ts | 54 +++ .../runtime/runtime-agent.acp-binding.test.ts | 165 +++++++++ src/plugins/runtime/runtime-agent.ts | 139 ++++++- src/plugins/runtime/types-core.ts | 11 + src/plugins/session-catalog.ts | 73 ++++ src/plugins/session-entry-slot-keys.ts | 1 + 25 files changed, 1447 insertions(+), 210 deletions(-) create mode 100644 extensions/acpx/src/pi-session-catalog.test-support.ts create mode 100644 extensions/acpx/src/pi-session-paths.test.ts create mode 100644 src/plugins/runtime/runtime-agent.acp-binding.test.ts diff --git a/docs/plugins/reference/acpx.md b/docs/plugins/reference/acpx.md index 4fc8445accc8..20af452f9db1 100644 --- a/docs/plugins/reference/acpx.md +++ b/docs/plugins/reference/acpx.md @@ -24,8 +24,15 @@ skills 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 +transcript browsing from Pi's documented JSONL session format. Local rows also +offer **Continue**, which creates an OpenClaw session whose first turn resumes +the native Pi session through ACP. Pi retains the full model context from its +session file, and the catalog viewer continues to show that history. The new +OpenClaw transcript starts empty and records only subsequent turns. Paired-node +rows remain view-only. Custom session directories outside the store scanned by +`pi-acp` remain browse-only because the adapter cannot resume those files by id. + +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. diff --git a/docs/plugins/reference/opencode.md b/docs/plugins/reference/opencode.md index 331ab7290407..f3f4042c1f5a 100644 --- a/docs/plugins/reference/opencode.md +++ b/docs/plugins/reference/opencode.md @@ -23,11 +23,16 @@ providers: `opencode`; contracts: `mediaUnderstandingProviders` ## 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. +sessions then appear in the **OpenCode** sessions-sidebar group, with transcript +browsing through the official `opencode --pure db ... --format json` and +`opencode --pure export` commands. Local rows also offer **Continue**, which +creates an OpenClaw session whose first turn resumes the native OpenCode session +through ACP. OpenCode retains the full server-side model context, and the catalog +viewer continues to show that history. The new OpenClaw transcript starts empty +and records only subsequent turns. Paired-node rows remain view-only. + +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. diff --git a/docs/plugins/sdk-runtime.md b/docs/plugins/sdk-runtime.md index 69d6d02387fe..34769fc07aef 100644 --- a/docs/plugins/sdk-runtime.md +++ b/docs/plugins/sdk-runtime.md @@ -185,7 +185,9 @@ two-party event loops that do not go through the shared inbound reply runner. Prefer `getSessionEntry(...)`, `listSessionEntries(...)`, `patchSessionEntry(...)`, or `upsertSessionEntry(...)` for session workflows. These helpers address sessions by agent/session identity so plugins do not depend on the legacy `sessions.json` storage shape. Use `preserveActivity: true` for metadata-only patches that should not refresh session activity, and `replaceEntry: true` only when the callback returns a complete entry and deleted fields must stay deleted. Doctor and migration paths can combine `fallbackEntry`, `skipMaintenance`, and `requireWriteSuccess` for one atomic canonical-store repair. - `createSessionEntry(...)` creates a new canonical session row and transcript. Its trusted `initialEntry` surface is deliberately narrow: a non-empty `agentHarnessId`, optional `modelSelectionLocked: true`, and optional `pluginExtensions`. The injected runtime accepts only harness ids owned by the calling plugin through `registerAgentHarness(...)`; this is an ownership invariant, not a sandbox between in-process plugins. It rejects an existing row; `label` and `spawnedCwd` are separate creation fields rather than trusted-entry patches. + `createSessionEntry(...)` creates a new canonical session row and transcript. Its trusted `initialEntry` surface is deliberately narrow. A plugin may select an owned `agentHarnessId`; seed an owned CLI backend with `cliBackendId`, `model`, and `cliSessionBinding`; or seed a persistent ACP session with `acpBackendId` and `acpSessionBinding: { acpAgentId, agentSessionId }`. The ACP variant persists the supplied native agent session id through the canonical SQLite ACP metadata owner so the first turn resumes that external session. The injected runtime restricts plugin-owned CLI and ACP sessions to the calling plugin's `plugin::` namespace; harness ids must be owned through `registerAgentHarness(...)`. These are ownership invariants, not a sandbox between in-process plugins. Creation rejects an existing row; `label` and `spawnedCwd` are separate creation fields rather than trusted-entry patches. + + Before advertising an ACP-backed action, use `resolveAcpSessionAvailability(...)` from `openclaw/plugin-sdk/acp-runtime`. It applies the canonical enablement, dispatch, allowed-agent, registered-backend, and backend-health checks; recheck it immediately before creating the session. Creation holds the session lifecycle mutation fence through `afterCreate`, so new work waits for plugin-owned initialization to finish and pre-existing admitted work makes creation fail. The callback receives a clone of the created state. If it returns a patch, that patch may contain only `pluginExtensions`, and its value is the complete final `pluginExtensions` field. A callback or final-persistence failure rolls back the unchanged new row and transcript; guarded rollback preserves a row changed or claimed concurrently. `recoverMatchingInitialEntry: true` is only for retrying interrupted initialization when the persisted trusted fields match exactly, and recovery requires `afterCreate` to return a final patch. diff --git a/extensions/acpx/src/pi-session-catalog-plugin.ts b/extensions/acpx/src/pi-session-catalog-plugin.ts index 6b6b12768a6e..5e503c3cd455 100644 --- a/extensions/acpx/src/pi-session-catalog-plugin.ts +++ b/extensions/acpx/src/pi-session-catalog-plugin.ts @@ -1,4 +1,7 @@ import process from "node:process"; +import { resolveAcpSessionAvailability } from "openclaw/plugin-sdk/acp-runtime"; +import { resolveDefaultAgentId } from "openclaw/plugin-sdk/agent-runtime"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { decodeNodePtyResumeParams, resolveNodeHostExecutable, @@ -18,6 +21,12 @@ import type { SessionCatalogTranscriptItem, SessionsCatalogReadResult, } from "openclaw/plugin-sdk/session-catalog"; +import { + createSessionCatalogAdoptionCoordinator, + listAdoptedSessionCatalogSessions, + sessionCatalogAdoptedSessionKey, + sessionCatalogAdoptedSourceKey, +} from "openclaw/plugin-sdk/session-catalog"; import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; import { isExactPiSessionCursor, @@ -45,6 +54,13 @@ const TRANSCRIPT_ITEM_TYPES = new Set([ "toolResult", "other", ]); +const ACPX_BACKEND_ID = "acpx"; +const PI_ACP_AGENT_ID = "pi"; +const PI_ADOPTED_SESSION_KEY_PREFIX = "plugin:acpx:catalog-adopt:pi:"; + +class PiCatalogParamsError extends Error {} + +const continueAdoption = createSessionCatalogAdoptionCoordinator(); function validatePiThreadId(value: unknown): string { if (typeof value !== "string" || !SESSION_ID_PATTERN.test(value)) { @@ -215,9 +231,13 @@ function unwrapNodePayload(value: unknown): unknown { type CatalogNode = Awaited>["nodes"][number]; -function setTerminalCapability(page: PiSessionPage, canOpenTerminal: boolean): PiSessionPage { +function setCatalogCapabilities( + page: PiSessionPage, + capabilities: { canContinue: boolean; canOpenTerminal: boolean }, +): PiSessionPage { for (const session of page.sessions) { - session.canOpenTerminal = canOpenTerminal; + session.canContinue = capabilities.canContinue && session.canContinue; + session.canOpenTerminal = capabilities.canOpenTerminal; } return page; } @@ -263,7 +283,7 @@ async function listPiNodeHost( const canOpenTerminal = commands?.includes(PI_TERMINAL_RESUME_COMMAND) === true; return { ...common, - ...setTerminalCapability(page, canOpenTerminal), + ...setCatalogCapabilities(page, { canContinue: false, canOpenTerminal }), }; } catch { return { @@ -316,9 +336,11 @@ function parseNodeTranscriptPage(value: unknown, threadId: string): SessionsCata } async function listPiHosts( - runtime: PluginRuntime, + api: OpenClawPluginApi, query: Parameters[0], ): Promise { + const runtime = api.runtime; + const canContinue = resolvePiContinuationAvailability(api).available; const requested = query.hostIds ? new Set(query.hostIds) : undefined; const hosts: SessionCatalogHost[] = []; if ((!requested || requested.has(LOCAL_HOST_ID)) && piSessionStoreAvailable(process.env)) { @@ -333,14 +355,15 @@ async function listPiHosts( ...(query.search ? { searchTerm: query.search } : {}), cursor: query.cursors?.[LOCAL_HOST_ID], }).then((page) => - setTerminalCapability( - page, - resolveNodeHostExecutable("pi", { - env: process.env, - pathEnv: process.env.PATH ?? "", - strategy: "fallback", - }) !== undefined, - ), + setCatalogCapabilities(page, { + canContinue, + canOpenTerminal: + resolveNodeHostExecutable("pi", { + env: process.env, + pathEnv: process.env.PATH ?? "", + strategy: "fallback", + }) !== undefined, + }), )), }); } catch { @@ -381,6 +404,105 @@ async function requireLocalPiSession(threadId: string): Promise { + return listAdoptedSessionCatalogSessions({ + config: currentPiCatalogConfig(api), + pluginId: api.id, + runtime: api.runtime, + sourceFromEntry: (entry) => { + const acpx = isRecord(entry.pluginExtensions?.acpx) ? entry.pluginExtensions.acpx : undefined; + const marker = acpx && isRecord(acpx.piSessionCatalog) ? acpx.piSessionCatalog : undefined; + return marker && typeof marker.sourceThreadId === "string" + ? { hostId: LOCAL_HOST_ID, threadId: marker.sourceThreadId } + : undefined; + }, + }); +} + +async function continuePiSession( + api: OpenClawPluginApi, + hostId: string, + threadId: string, +): Promise<{ sessionKey: string }> { + if (hostId.startsWith("node:")) { + throw new PiCatalogParamsError("paired-node Pi session rows are view-only"); + } + if (hostId !== LOCAL_HOST_ID) { + throw new PiCatalogParamsError("Pi session catalog hostId is invalid"); + } + const availability = resolvePiContinuationAvailability(api); + if (!availability.available) { + throw new PiCatalogParamsError(availability.message); + } + const sourceKey = sessionCatalogAdoptedSourceKey(hostId, threadId); + return await continueAdoption({ + sourceKey, + findExisting: () => listAdoptedPiSessions(api).get(sourceKey), + create: async () => { + const record = await requireLocalPiSession(threadId).catch(() => undefined); + if (!record) { + throw new PiCatalogParamsError("Pi session is unavailable"); + } + if (!record.canContinue) { + throw new PiCatalogParamsError( + "Pi session is outside the session store supported by pi-acp", + ); + } + const currentAvailability = resolvePiContinuationAvailability(api); + if (!currentAvailability.available) { + throw new PiCatalogParamsError(currentAvailability.message); + } + const config = currentPiCatalogConfig(api); + const marker = { sourceThreadId: threadId }; + // ACPX consumes load replay before OpenClaw turn handlers attach, so the + // OpenClaw transcript starts empty while Pi resumes from its session file. + const created = await api.runtime.agent.session.createSessionEntry({ + cfg: config, + key: sessionCatalogAdoptedSessionKey(PI_ADOPTED_SESSION_KEY_PREFIX, threadId), + agentId: resolveDefaultAgentId(config), + recoverMatchingInitialEntry: true, + ...(record.name ? { label: record.name } : {}), + ...(record.cwd ? { spawnedCwd: record.cwd } : {}), + initialEntry: { + acpBackendId: ACPX_BACKEND_ID, + acpSessionBinding: { + acpAgentId: PI_ACP_AGENT_ID, + agentSessionId: threadId, + }, + pluginExtensions: { acpx: { piSessionCatalog: marker } }, + }, + afterCreate: async () => ({ + pluginExtensions: { acpx: { piSessionCatalog: marker } }, + }), + }); + return { sessionKey: created.key }; + }, + }); +} + async function resolveNodePiSession(params: { runtime: PluginRuntime; nodeId: string; @@ -509,8 +631,10 @@ export function registerPiSessionCatalog(api: OpenClawPluginApi): void { api.registerSessionCatalog({ id: "pi", label: "Pi", - list: async (query) => await listPiHosts(api.runtime, query), + list: async (query) => await listPiHosts(api, query), read: async (request) => await readPiTranscript(api.runtime, request), + continueSession: async (request) => + await continuePiSession(api, request.hostId, request.threadId), openTerminal: async (request) => await openPiTerminal({ runtime: api.runtime, ...request }), }); for (const command of createPiSessionNodeHostCommands()) { diff --git a/extensions/acpx/src/pi-session-catalog.test-support.ts b/extensions/acpx/src/pi-session-catalog.test-support.ts new file mode 100644 index 000000000000..fd00efa7a0d0 --- /dev/null +++ b/extensions/acpx/src/pi-session-catalog.test-support.ts @@ -0,0 +1,163 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; +import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path"; +import { vi } from "vitest"; +import { registerPiSessionCatalog } from "./pi-session-catalog-plugin.js"; + +export async function createPiStoreFixture( + temporaryDirectories: string[], + assistantText = "hi", + sessionName = "Pi catalog session", + toolArguments: unknown = { command: "pwd" }, + acpResolvable = false, +): Promise { + const root = await fs.mkdtemp( + path.join(resolvePreferredOpenClawTmpDir(), "openclaw-pi-catalog-"), + ); + temporaryDirectories.push(root); + const directory = acpResolvable ? path.join(root, "sessions", "project") : root; + if (acpResolvable) { + await fs.mkdir(directory, { recursive: true }); + delete process.env.PI_CODING_AGENT_SESSION_DIR; + process.env.PI_CODING_AGENT_DIR = root; + } else { + 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: toolArguments }, + ], + }, + }, + { + 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: sessionName, + }, + ]; + await fs.writeFile( + path.join(directory, "session.jsonl"), + `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`, + ); + return directory; +} + +export async function installFakePiFixture( + temporaryDirectories: string[], + originalPath: string | undefined, +): Promise { + const directory = await fs.mkdtemp( + path.join(resolvePreferredOpenClawTmpDir(), "openclaw-pi-cli-"), + ); + temporaryDirectories.push(directory); + const executable = path.join(directory, "pi"); + await fs.writeFile(executable, "#!/bin/sh\nexit 0\n"); + await fs.chmod(executable, 0o755); + process.env.PATH = `${directory}${path.delimiter}${originalPath ?? ""}`; + return directory; +} + +export 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; +} + +export function capturePiContinuationCatalog() { + let provider: Parameters[0] | undefined; + const entries: Array<{ sessionKey: string; entry: Record }> = []; + const createSessionEntry = vi.fn( + async ( + params: Parameters[0], + ) => { + const sessionKey = `agent:${params.agentId ?? "main"}:${params.key}`; + const entry = { + sessionId: "adopted-pi-session", + updatedAt: Date.now(), + pluginOwnerId: "acpx", + ...(params.label ? { label: params.label } : {}), + ...(params.spawnedCwd ? { spawnedCwd: params.spawnedCwd } : {}), + pluginExtensions: params.initialEntry.pluginExtensions, + }; + entries.push({ sessionKey, entry }); + return { + key: sessionKey, + agentId: params.agentId ?? "main", + sessionId: entry.sessionId, + entry, + }; + }, + ); + registerPiSessionCatalog({ + id: "acpx", + pluginConfig: {}, + config: {}, + runtime: { + config: { current: () => ({}) }, + nodes: { list: vi.fn().mockResolvedValue({ nodes: [] }) }, + agent: { + session: { + createSessionEntry, + listSessionEntries: vi.fn(() => entries), + }, + }, + }, + registerSessionCatalog: (value: NonNullable) => { + provider = value; + }, + registerNodeHostCommand: vi.fn(), + registerNodeInvokePolicy: vi.fn(), + } as unknown as OpenClawPluginApi); + return { createSessionEntry, entries, provider: provider! }; +} diff --git a/extensions/acpx/src/pi-session-catalog.test.ts b/extensions/acpx/src/pi-session-catalog.test.ts index 92cbfce1acc6..f61194fad3ce 100644 --- a/extensions/acpx/src/pi-session-catalog.test.ts +++ b/extensions/acpx/src/pi-session-catalog.test.ts @@ -4,9 +4,20 @@ import path from "node:path"; import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; import { afterEach, describe, expect, it, vi } from "vitest"; +type ResolveAcpSessionAvailability = + (typeof import("openclaw/plugin-sdk/acp-runtime"))["resolveAcpSessionAvailability"]; + const nodeHostMocks = vi.hoisted(() => ({ runNodePtyCommand: vi.fn(async () => ({ exitCode: 0 })), })); +const acpRuntimeMocks = vi.hoisted(() => ({ + resolveAcpSessionAvailability: vi.fn(() => ({ available: true })), +})); + +vi.mock("openclaw/plugin-sdk/acp-runtime", async (importOriginal) => ({ + ...(await importOriginal()), + resolveAcpSessionAvailability: acpRuntimeMocks.resolveAcpSessionAvailability, +})); vi.mock("openclaw/plugin-sdk/node-host", async (importOriginal) => { const actual = await importOriginal(); @@ -34,7 +45,13 @@ vi.mock("openclaw/plugin-sdk/node-host", async (importOriginal) => { import { registerPiSessionCatalog } from "./pi-session-catalog-plugin.js"; import { listLocalPiSessionPage, readLocalPiTranscriptPage } from "./pi-session-catalog.js"; -import { piSessionStore } from "./pi-session-paths.js"; +import { + capturePiContinuationCatalog, + createPiStoreFixture, + installFakePiFixture, + registerPiNodeHostCommands, +} from "./pi-session-catalog.test-support.js"; +import { listPiSummaryPage } from "./pi-session-store.js"; const PI_SESSIONS_LIST_COMMAND = "acpx.pi.sessions.list.v1"; const PI_SESSION_READ_COMMAND = "acpx.pi.sessions.read.v1"; @@ -47,100 +64,23 @@ const originalHome = process.env.HOME; const originalUserProfile = process.env.USERPROFILE; const originalPath = process.env.PATH; -async function createPiStore( +const createPiStore = ( assistantText = "hi", sessionName = "Pi catalog session", toolArguments: unknown = { command: "pwd" }, -): 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: toolArguments }, - ], - }, - }, - { - 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: sessionName, - }, - ]; - await fs.writeFile( - path.join(directory, "session.jsonl"), - `${entries.map((entry) => JSON.stringify(entry)).join("\n")}\n`, + acpResolvable = false, +) => + createPiStoreFixture( + temporaryDirectories, + assistantText, + sessionName, + toolArguments, + acpResolvable, ); - return directory; -} - -async function installFakePi(): Promise { - const directory = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-pi-cli-")); - temporaryDirectories.push(directory); - const executable = path.join(directory, "pi"); - await fs.writeFile(executable, "#!/bin/sh\nexit 0\n"); - await fs.chmod(executable, 0o755); - process.env.PATH = `${directory}${path.delimiter}${originalPath ?? ""}`; - 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; -} +const installFakePi = () => installFakePiFixture(temporaryDirectories, originalPath); afterEach(async () => { + acpRuntimeMocks.resolveAcpSessionAvailability.mockReset().mockReturnValue({ available: true }); nodeHostMocks.runNodePtyCommand.mockClear(); process.env.PATH = originalPath; if (originalSessionDir === undefined) { @@ -171,63 +111,9 @@ afterEach(async () => { }); 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(); + await createPiStore("hi", "Pi catalog session", { command: "pwd" }, true); + await installFakePi(); const listed = await listLocalPiSessionPage({ limit: 20 }); expect(listed).toEqual({ sessions: [ @@ -236,7 +122,7 @@ describe("Pi session catalog", () => { name: "Pi catalog session", cwd: "/workspace", source: "pi-cli", - canContinue: false, + canContinue: true, }), ], }); @@ -305,6 +191,62 @@ describe("Pi session catalog", () => { ]); }); + it("recognizes Pi sessions when the agent directory uses a symlinked path", async () => { + const sessionDirectory = await createPiStore(); + const agentDirectory = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-pi-agent-real-")); + const symlinkParent = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-pi-agent-link-")); + const linkedAgentDirectory = path.join(symlinkParent, "agent"); + temporaryDirectories.push(agentDirectory, symlinkParent); + await fs.mkdir(path.join(agentDirectory, "sessions"), { recursive: true }); + await fs.rename(sessionDirectory, path.join(agentDirectory, "sessions", "project")); + await fs.symlink( + agentDirectory, + linkedAgentDirectory, + process.platform === "win32" ? "junction" : "dir", + ); + process.env.PI_CODING_AGENT_SESSION_DIR = path.join(agentDirectory, "sessions", "project"); + process.env.PI_CODING_AGENT_DIR = linkedAgentDirectory; + + await expect(listLocalPiSessionPage({ limit: 20 })).resolves.toEqual({ + sessions: [expect.objectContaining({ threadId: "pi-session", canContinue: true })], + }); + }); + + it("refreshes cached continuation eligibility without mutating prior summaries", async () => { + const sessionDirectory = await createPiStore( + "hi", + "Pi catalog session", + { command: "pwd" }, + true, + ); + const agentDirectory = path.dirname(path.dirname(sessionDirectory)); + const unrelatedAgentDirectory = await fs.mkdtemp( + path.join(os.tmpdir(), "openclaw-pi-agent-unrelated-"), + ); + temporaryDirectories.push(unrelatedAgentDirectory); + const baseEnv = { + ...process.env, + PI_CODING_AGENT_SESSION_DIR: sessionDirectory, + PI_CODING_AGENT_DIR: agentDirectory, + }; + + const first = await listPiSummaryPage(baseEnv, { offset: 0, limit: 20 }); + const firstSummary = first.summaries[0]; + expect(firstSummary?.canContinue).toBe(true); + + const second = await listPiSummaryPage( + { ...baseEnv, PI_CODING_AGENT_DIR: unrelatedAgentDirectory }, + { offset: 0, limit: 20 }, + ); + expect(second.summaries[0]?.canContinue).toBe(false); + expect(second.summaries[0]).not.toBe(firstSummary); + expect(firstSummary?.canContinue).toBe(true); + + const third = await listPiSummaryPage(baseEnv, { offset: 0, limit: 20 }); + expect(third.summaries[0]?.canContinue).toBe(true); + expect(firstSummary?.canContinue).toBe(true); + }); + it("summarizes and pages a large session within transport limits", async () => { await createPiStore("x".repeat(600 * 1024)); const listed = await listLocalPiSessionPage({ limit: 20 }); @@ -737,6 +679,101 @@ describe("Pi session catalog", () => { }, ); + it("adopts local Pi sessions once with the native ACP resume binding", async () => { + await createPiStore("hi", "Pi catalog session", { command: "pwd" }, true); + await installFakePi(); + const { createSessionEntry, provider } = capturePiContinuationCatalog(); + + const [first, concurrent] = await Promise.all([ + provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }), + provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }), + ]); + const second = await provider.continueSession!({ + hostId: "gateway", + threadId: "pi-session", + }); + + expect(first).toEqual(concurrent); + expect(second).toEqual(first); + expect(createSessionEntry).toHaveBeenCalledTimes(1); + expect(createSessionEntry).toHaveBeenCalledWith( + expect.objectContaining({ + label: "Pi catalog session", + spawnedCwd: "/workspace", + initialEntry: { + acpBackendId: "acpx", + acpSessionBinding: { acpAgentId: "pi", agentSessionId: "pi-session" }, + pluginExtensions: { + acpx: { piSessionCatalog: { sourceThreadId: "pi-session" } }, + }, + }, + }), + ); + }); + + it("rejects paired-node and unknown Pi session continuation", async () => { + await createPiStore("hi", "Pi catalog session", { command: "pwd" }, true); + await installFakePi(); + const { createSessionEntry, provider } = capturePiContinuationCatalog(); + + await expect( + provider.continueSession!({ hostId: "node:remote", threadId: "pi-session" }), + ).rejects.toThrow("paired-node Pi session rows are view-only"); + await expect( + provider.continueSession!({ hostId: "gateway", threadId: "missing" }), + ).rejects.toThrow("Pi session is unavailable"); + expect(createSessionEntry).not.toHaveBeenCalled(); + }); + + it("hides and rejects Continue when ACP cannot resume Pi", async () => { + await createPiStore("hi", "Pi catalog session", { command: "pwd" }, true); + await installFakePi(); + acpRuntimeMocks.resolveAcpSessionAvailability.mockReturnValue({ + available: false, + message: "ACP is disabled by policy", + }); + const { provider } = capturePiContinuationCatalog(); + + await expect(provider.list({ hostIds: ["gateway"] })).resolves.toEqual([ + expect.objectContaining({ + sessions: [expect.objectContaining({ threadId: "pi-session", canContinue: false })], + }), + ]); + await expect( + provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }), + ).rejects.toThrow("ACP is disabled by policy"); + }); + + it("keeps custom Pi stores browse-only when pi-acp cannot resolve them", async () => { + await createPiStore(); + await installFakePi(); + const { provider } = capturePiContinuationCatalog(); + + await expect(provider.list({ hostIds: ["gateway"] })).resolves.toEqual([ + expect.objectContaining({ + sessions: [expect.objectContaining({ threadId: "pi-session", canContinue: false })], + }), + ]); + await expect( + provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }), + ).rejects.toThrow("outside the session store supported by pi-acp"); + }); + + it("hides and rejects Continue when the Pi CLI is unavailable", async () => { + await createPiStore("hi", "Pi catalog session", { command: "pwd" }, true); + process.env.PATH = ""; + const { provider } = capturePiContinuationCatalog(); + + await expect(provider.list({ hostIds: ["gateway"] })).resolves.toEqual([ + expect.objectContaining({ + sessions: [expect.objectContaining({ threadId: "pi-session", canContinue: false })], + }), + ]); + await expect( + provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }), + ).rejects.toThrow("Pi CLI is unavailable"); + }); + it("opens paired-node Pi sessions only through the advertised terminal command", async () => { let provider: Parameters[0] | undefined; const page = { @@ -747,7 +784,7 @@ describe("Pi session catalog", () => { cwd: "/remote/workspace", status: "stored", archived: false, - canContinue: false, + canContinue: true, canArchive: false, }, ], @@ -779,7 +816,13 @@ describe("Pi session catalog", () => { await expect(provider!.list({ hostIds: ["node:node-1"], search: "remote" })).resolves.toEqual([ expect.objectContaining({ - sessions: [expect.objectContaining({ threadId: "pi-remote", canOpenTerminal: true })], + sessions: [ + expect.objectContaining({ + threadId: "pi-remote", + canContinue: false, + canOpenTerminal: true, + }), + ], }), ]); expect(invoke).toHaveBeenNthCalledWith(1, { diff --git a/extensions/acpx/src/pi-session-paths.test.ts b/extensions/acpx/src/pi-session-paths.test.ts new file mode 100644 index 000000000000..30d5f243b6ef --- /dev/null +++ b/extensions/acpx/src/pi-session-paths.test.ts @@ -0,0 +1,72 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { piSessionStore } from "./pi-session-paths.js"; + +const temporaryDirectories: string[] = []; + +afterEach(async () => { + await Promise.all( + temporaryDirectories.splice(0).map(async (directory) => { + await fs.rm(directory, { recursive: true, force: true }); + }), + ); +}); + +describe("Pi session paths", () => { + 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, + }); + }); +}); diff --git a/extensions/acpx/src/pi-session-paths.ts b/extensions/acpx/src/pi-session-paths.ts index d4bfff43df98..18532f404781 100644 --- a/extensions/acpx/src/pi-session-paths.ts +++ b/extensions/acpx/src/pi-session-paths.ts @@ -88,6 +88,22 @@ export function piSessionStore( }; } +/** Store root scanned by pi-acp@0.0.26 when resolving a native session id. */ +export function piAcpSessionStoreRoot(env: NodeJS.ProcessEnv): string | undefined { + const configuredAgentDir = env.PI_CODING_AGENT_DIR?.trim(); + // Deliberately stricter than piSessionStore(): pi-acp's session lookup reads + // PI_CODING_AGENT_DIR raw, with no `~` expansion (getPiAgentDir, dist/index.js:1044, + // reached via findPiSessionFile -> listPiSessions). Expanding `~` here would advertise + // Continue on sessions pi-acp then fails to resolve with "Unknown sessionId". + if (configuredAgentDir && !isPiSessionCatalogPathAbsolute(configuredAgentDir)) { + return undefined; + } + const agentDir = configuredAgentDir + ? path.resolve(configuredAgentDir) + : path.join(piHome(env), ".pi", "agent"); + return path.join(agentDir, "sessions"); +} + export function piSessionStoreAvailable(env: NodeJS.ProcessEnv): boolean { try { return statSync(piSessionStore(env).root).isDirectory(); diff --git a/extensions/acpx/src/pi-session-store.ts b/extensions/acpx/src/pi-session-store.ts index 6d1f844808dc..cbcb317a91a8 100644 --- a/extensions/acpx/src/pi-session-store.ts +++ b/extensions/acpx/src/pi-session-store.ts @@ -3,7 +3,7 @@ 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"; +import { piAcpSessionStoreRoot, piSessionStore } from "./pi-session-paths.js"; const MAX_DISCOVERY_FILES = 10_000; const SUMMARY_SCAN_BATCH_SIZE = 100; @@ -22,6 +22,7 @@ type PiFileCandidate = { identity: string; mtimeMs: number; size: number; + resumable: boolean; }; type PiSummaryScanState = { @@ -84,9 +85,10 @@ async function discoverPiSessionFiles( env: NodeJS.ProcessEnv, ): Promise<{ root: string; files: string[] }> { const store = piSessionStore(env); + const resolvedRoot = await realpathOrResolve(store.root); let entries: Array; try { - entries = await fs.readdir(store.root, { withFileTypes: true }); + entries = await fs.readdir(resolvedRoot, { withFileTypes: true }); } catch { return { root: store.root, files: [] }; } @@ -96,7 +98,7 @@ async function discoverPiSessionFiles( files: entries .filter((entry) => entry.isFile() && entry.name.endsWith(".jsonl")) .slice(0, MAX_DISCOVERY_FILES) - .map((entry) => path.join(store.root, entry.name)), + .map((entry) => path.join(resolvedRoot, entry.name)), }; } const files: string[] = []; @@ -104,7 +106,7 @@ async function discoverPiSessionFiles( if (!entry.isDirectory() || files.length >= MAX_DISCOVERY_FILES) { continue; } - const directory = path.join(store.root, entry.name); + const directory = path.join(resolvedRoot, entry.name); let children: Array; try { children = await fs.readdir(directory, { withFileTypes: true }); @@ -123,6 +125,14 @@ async function discoverPiSessionFiles( return { root: store.root, files }; } +async function realpathOrResolve(value: string): Promise { + try { + return await fs.realpath(value); + } catch { + return path.resolve(value); + } +} + async function mapConcurrent( values: T[], limit: number, @@ -143,6 +153,8 @@ async function mapConcurrent( async function piFileCandidates(env: NodeJS.ProcessEnv): Promise { const { root, files } = await discoverPiSessionFiles(env); + const configuredAcpRoot = piAcpSessionStoreRoot(env); + const acpRoot = configuredAcpRoot ? await realpathOrResolve(configuredAcpRoot) : undefined; const candidates = await mapConcurrent(files, IO_CONCURRENCY, async (file) => { try { const stats = await fs.stat(file); @@ -153,6 +165,7 @@ async function piFileCandidates(env: NodeJS.ProcessEnv): Promise right.mtimeMs - left.mtimeMs); } +function pathIsWithin(root: string, candidate: string): boolean { + const relative = path.relative(root, candidate); + return ( + relative !== "" && + relative !== ".." && + !relative.startsWith(`..${path.sep}`) && + !path.isAbsolute(relative) + ); +} + function parsePiJsonLines(content: string): Record[] { return content.split(/\r?\n/u).flatMap((line) => { if (!line.trim()) { @@ -320,7 +343,9 @@ async function readPiSessionSummary( if (cached?.mtimeMs === candidate.mtimeMs && cached.size === candidate.size) { summaryCache.delete(candidate.file); summaryCache.set(candidate.file, cached); - return cached.summary; + return cached.summary + ? { ...cached.summary, canContinue: candidate.resumable } + : cached.summary; } let summary: PiSessionSummary | undefined; let scanState: PiSummaryScanState; @@ -365,7 +390,7 @@ async function readPiSessionSummary( source: "pi-cli", modelProvider: "pi", archived: false, - canContinue: false, + canContinue: candidate.resumable, canArchive: false, }; } diff --git a/extensions/opencode/session-catalog-plugin.ts b/extensions/opencode/session-catalog-plugin.ts index d73548d89bee..8382c1a64651 100644 --- a/extensions/opencode/session-catalog-plugin.ts +++ b/extensions/opencode/session-catalog-plugin.ts @@ -1,6 +1,9 @@ import { accessSync, constants, statSync } from "node:fs"; import path from "node:path"; import process from "node:process"; +import { resolveAcpSessionAvailability } from "openclaw/plugin-sdk/acp-runtime"; +import { resolveDefaultAgentId } from "openclaw/plugin-sdk/agent-runtime"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { resolveNodeHostExecutable } from "openclaw/plugin-sdk/node-host"; import type { OpenClawPluginApi, @@ -15,6 +18,12 @@ import type { SessionCatalogTranscriptItem, SessionsCatalogReadResult, } from "openclaw/plugin-sdk/session-catalog"; +import { + createSessionCatalogAdoptionCoordinator, + listAdoptedSessionCatalogSessions, + sessionCatalogAdoptedSessionKey, + sessionCatalogAdoptedSourceKey, +} from "openclaw/plugin-sdk/session-catalog"; import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; import { OPENCODE_LOCAL_SESSION_HOST_ID as LOCAL_HOST_ID, @@ -25,6 +34,7 @@ import { OPENCODE_SESSION_ID_PATTERN as SESSION_ID_PATTERN, OPENCODE_SESSION_READ_COMMAND, OPENCODE_TERMINAL_RESUME_COMMAND, + OpenCodeCatalogParamsError, } from "./session-catalog-shared.js"; import { createOpenCodeTerminalNodeHostCommand, @@ -46,6 +56,11 @@ const TRANSCRIPT_ITEM_TYPES = new Set([ "toolResult", "other", ]); +const ACPX_BACKEND_ID = "acpx"; +const OPENCODE_ACP_AGENT_ID = "opencode"; +const OPENCODE_ADOPTED_SESSION_KEY_PREFIX = "plugin:opencode:catalog-adopt:"; + +const continueAdoption = createSessionCatalogAdoptionCoordinator(); function isOptionalString(value: unknown): boolean { return value === undefined || typeof value === "string"; @@ -199,12 +214,13 @@ function unwrapNodePayload(value: unknown): unknown { type CatalogNode = Awaited>["nodes"][number]; -function setTerminalCapability( +function setCatalogCapabilities( page: OpenCodeSessionPage, - canOpenTerminal: boolean, + capabilities: { canContinue: boolean; canOpenTerminal: boolean }, ): OpenCodeSessionPage { for (const session of page.sessions) { - session.canOpenTerminal = canOpenTerminal; + session.canContinue = capabilities.canContinue; + session.canOpenTerminal = capabilities.canOpenTerminal; } return page; } @@ -250,7 +266,7 @@ async function listOpenCodeNodeHost( const canOpenTerminal = commands?.includes(OPENCODE_TERMINAL_RESUME_COMMAND) === true; return { ...common, - ...setTerminalCapability(page, canOpenTerminal), + ...setCatalogCapabilities(page, { canContinue: false, canOpenTerminal }), }; } catch { return { @@ -306,9 +322,15 @@ function parseNodeTranscriptPage(value: unknown, threadId: string): SessionsCata } async function listOpenCodeHosts( - runtime: PluginRuntime, + api: OpenClawPluginApi, query: Parameters[0], ): Promise { + const runtime = api.runtime; + const canContinue = resolveAcpSessionAvailability({ + config: currentOpenCodeCatalogConfig(api), + backendId: ACPX_BACKEND_ID, + agentId: OPENCODE_ACP_AGENT_ID, + }).available; const requested = query.hostIds ? new Set(query.hostIds) : undefined; const hosts: SessionCatalogHost[] = []; if ( @@ -329,7 +351,7 @@ async function listOpenCodeHosts( limit: query.limitPerHost, ...(query.search ? { searchTerm: query.search } : {}), cursor: query.cursors?.[LOCAL_HOST_ID], - }).then((page) => setTerminalCapability(page, true))), + }).then((page) => setCatalogCapabilities(page, { canContinue, canOpenTerminal: true }))), }); } catch { hosts.push({ @@ -408,6 +430,96 @@ async function readOpenCodeTranscript( }; } +function currentOpenCodeCatalogConfig(api: OpenClawPluginApi): OpenClawConfig { + return (api.runtime.config?.current?.() ?? api.config ?? {}) as OpenClawConfig; +} + +function listAdoptedOpenCodeSessions(api: OpenClawPluginApi): Map { + return listAdoptedSessionCatalogSessions({ + config: currentOpenCodeCatalogConfig(api), + pluginId: api.id, + runtime: api.runtime, + sourceFromEntry: (entry) => { + const opencode = isRecord(entry.pluginExtensions?.opencode) + ? entry.pluginExtensions.opencode + : undefined; + const marker = + opencode && isRecord(opencode.sessionCatalog) ? opencode.sessionCatalog : undefined; + return marker && typeof marker.sourceThreadId === "string" + ? { hostId: LOCAL_HOST_ID, threadId: marker.sourceThreadId } + : undefined; + }, + }); +} + +async function continueOpenCodeSession( + api: OpenClawPluginApi, + hostId: string, + threadId: string, +): Promise<{ sessionKey: string }> { + if (hostId.startsWith("node:")) { + throw new OpenCodeCatalogParamsError("paired-node OpenCode session rows are view-only"); + } + if (hostId !== LOCAL_HOST_ID) { + throw new OpenCodeCatalogParamsError("OpenCode session catalog hostId is invalid"); + } + const availability = resolveAcpSessionAvailability({ + config: currentOpenCodeCatalogConfig(api), + backendId: ACPX_BACKEND_ID, + agentId: OPENCODE_ACP_AGENT_ID, + }); + if (!availability.available) { + throw new OpenCodeCatalogParamsError(availability.message); + } + const sourceKey = sessionCatalogAdoptedSourceKey(hostId, threadId); + return await continueAdoption({ + sourceKey, + findExisting: () => listAdoptedOpenCodeSessions(api).get(sourceKey), + create: async () => { + const page = await listLocalOpenCodeSessionPage({ + searchTerm: threadId, + limit: MAX_PAGE_LIMIT, + }).catch(() => undefined); + const record = page?.sessions.find((session) => session.threadId === threadId); + if (!record) { + throw new OpenCodeCatalogParamsError("OpenCode session is unavailable"); + } + const config = currentOpenCodeCatalogConfig(api); + const currentAvailability = resolveAcpSessionAvailability({ + config, + backendId: ACPX_BACKEND_ID, + agentId: OPENCODE_ACP_AGENT_ID, + }); + if (!currentAvailability.available) { + throw new OpenCodeCatalogParamsError(currentAvailability.message); + } + const marker = { sourceThreadId: threadId }; + // ACPX binds the native session before OpenClaw turn handlers attach, so + // the OpenClaw transcript starts empty while OpenCode retains server context. + const created = await api.runtime.agent.session.createSessionEntry({ + cfg: config, + key: sessionCatalogAdoptedSessionKey(OPENCODE_ADOPTED_SESSION_KEY_PREFIX, threadId), + agentId: resolveDefaultAgentId(config), + recoverMatchingInitialEntry: true, + ...(record.name ? { label: record.name } : {}), + ...(record.cwd ? { spawnedCwd: record.cwd } : {}), + initialEntry: { + acpBackendId: ACPX_BACKEND_ID, + acpSessionBinding: { + acpAgentId: OPENCODE_ACP_AGENT_ID, + agentSessionId: threadId, + }, + pluginExtensions: { opencode: { sessionCatalog: marker } }, + }, + afterCreate: async () => ({ + pluginExtensions: { opencode: { sessionCatalog: marker } }, + }), + }); + return { sessionKey: created.key }; + }, + }); +} + export function registerOpenCodeSessionCatalog(api: OpenClawPluginApi): void { if (!isOpenCodeSessionCatalogEnabled(api.pluginConfig)) { return; @@ -415,8 +527,10 @@ export function registerOpenCodeSessionCatalog(api: OpenClawPluginApi): void { api.registerSessionCatalog({ id: "opencode", label: "OpenCode", - list: async (query) => await listOpenCodeHosts(api.runtime, query), + list: async (query) => await listOpenCodeHosts(api, query), read: async (request) => await readOpenCodeTranscript(api.runtime, request), + continueSession: async (request) => + await continueOpenCodeSession(api, request.hostId, request.threadId), openTerminal: async (request) => await openOpenCodeCatalogTerminal({ runtime: api.runtime, diff --git a/extensions/opencode/session-catalog-shared.ts b/extensions/opencode/session-catalog-shared.ts index 31591145a332..83f2441f2a03 100644 --- a/extensions/opencode/session-catalog-shared.ts +++ b/extensions/opencode/session-catalog-shared.ts @@ -7,3 +7,5 @@ export const OPENCODE_LOCAL_SESSION_HOST_ID = "gateway"; export const OPENCODE_SESSION_CATALOG_MAX_PAGE_LIMIT = 100; export const OPENCODE_NODE_INVOKE_TIMEOUT_MS = 35_000; export const OPENCODE_SESSION_ID_PATTERN = /^(?!-)[A-Za-z0-9._:-]{1,256}$/u; + +export class OpenCodeCatalogParamsError extends Error {} diff --git a/extensions/opencode/session-catalog.test.ts b/extensions/opencode/session-catalog.test.ts index 26edf2066cea..93340276dc7b 100644 --- a/extensions/opencode/session-catalog.test.ts +++ b/extensions/opencode/session-catalog.test.ts @@ -6,9 +6,15 @@ import path from "node:path"; import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; import { afterEach, describe, expect, it, vi } from "vitest"; +type ResolveAcpSessionAvailability = + (typeof import("openclaw/plugin-sdk/acp-runtime"))["resolveAcpSessionAvailability"]; + const nodeHostMocks = vi.hoisted(() => ({ runNodePtyCommand: vi.fn(async () => ({ exitCode: 0 })), })); +const acpRuntimeMocks = vi.hoisted(() => ({ + resolveAcpSessionAvailability: vi.fn(() => ({ available: true })), +})); const childProcessMocks = vi.hoisted(() => ({ children: [] as ChildProcess[], spawn: vi.fn(), @@ -24,6 +30,11 @@ vi.mock("node:child_process", async (importOriginal) => { return { ...actual, spawn: childProcessMocks.spawn }; }); +vi.mock("openclaw/plugin-sdk/acp-runtime", async (importOriginal) => ({ + ...(await importOriginal()), + resolveAcpSessionAvailability: acpRuntimeMocks.resolveAcpSessionAvailability, +})); + vi.mock("openclaw/plugin-sdk/node-host", async (importOriginal) => { const actual = await importOriginal(); return { @@ -83,6 +94,54 @@ function captureOpenCodeSessionRegistrations(pluginConfig: unknown = {}) { return { catalogs, commands, policies }; } +function captureOpenCodeContinuationCatalog() { + let provider: Parameters[0] | undefined; + const entries: Array<{ sessionKey: string; entry: Record }> = []; + const createSessionEntry = vi.fn( + async ( + params: Parameters[0], + ) => { + const sessionKey = `agent:${params.agentId ?? "main"}:${params.key}`; + const entry = { + sessionId: "adopted-opencode-session", + updatedAt: Date.now(), + pluginOwnerId: "opencode", + ...(params.label ? { label: params.label } : {}), + ...(params.spawnedCwd ? { spawnedCwd: params.spawnedCwd } : {}), + pluginExtensions: params.initialEntry.pluginExtensions, + }; + entries.push({ sessionKey, entry }); + return { + key: sessionKey, + agentId: params.agentId ?? "main", + sessionId: entry.sessionId, + entry, + }; + }, + ); + registerOpenCodeSessionCatalog({ + id: "opencode", + pluginConfig: {}, + config: {}, + runtime: { + config: { current: () => ({}) }, + nodes: { list: vi.fn().mockResolvedValue({ nodes: [] }) }, + agent: { + session: { + createSessionEntry, + listSessionEntries: vi.fn(() => entries), + }, + }, + }, + registerSessionCatalog: (value: NonNullable) => { + provider = value; + }, + registerNodeHostCommand: vi.fn(), + registerNodeInvokePolicy: vi.fn(), + } as unknown as OpenClawPluginApi); + return { createSessionEntry, entries, provider: provider! }; +} + async function installFakeOpenCode( assistantText = "hi", sessionTitle = "Catalog session", @@ -195,6 +254,7 @@ async function stopChild(child: ChildProcess | undefined): Promise { } afterEach(async () => { + acpRuntimeMocks.resolveAcpSessionAvailability.mockReset().mockReturnValue({ available: true }); nodeHostMocks.runNodePtyCommand.mockClear(); childProcessMocks.spawn.mockClear(); await Promise.all(childProcessMocks.children.splice(0).map((child) => stopChild(child))); @@ -229,7 +289,7 @@ describe("OpenCode session catalog", () => { name: "Catalog session", cwd: "/workspace", source: "opencode-cli", - canContinue: false, + canContinue: true, }), ], }); @@ -307,6 +367,27 @@ describe("OpenCode session catalog", () => { }, ); + it.runIf(process.platform !== "win32")( + "hides and rejects Continue when ACP cannot resume OpenCode", + async () => { + await installFakeOpenCode(); + acpRuntimeMocks.resolveAcpSessionAvailability.mockReturnValue({ + available: false, + message: "ACP runtime backend is unavailable", + }); + const { provider } = captureOpenCodeContinuationCatalog(); + + await expect(provider.list({ hostIds: ["gateway"] })).resolves.toEqual([ + expect.objectContaining({ + sessions: [expect.objectContaining({ threadId: "ses_test", canContinue: false })], + }), + ]); + await expect( + provider.continueSession!({ hostId: "gateway", threadId: "ses_test" }), + ).rejects.toThrow("ACP runtime backend is unavailable"); + }, + ); + it.runIf(process.platform !== "win32")( "keeps oversized transcript items below the node payload budget", async () => { @@ -321,6 +402,56 @@ describe("OpenCode session catalog", () => { }, ); + it.runIf(process.platform !== "win32")( + "adopts local OpenCode sessions once with the native ACP resume binding", + async () => { + await installFakeOpenCode(); + const { createSessionEntry, provider } = captureOpenCodeContinuationCatalog(); + + const [first, concurrent] = await Promise.all([ + provider.continueSession!({ hostId: "gateway", threadId: "ses_test" }), + provider.continueSession!({ hostId: "gateway", threadId: "ses_test" }), + ]); + const second = await provider.continueSession!({ + hostId: "gateway", + threadId: "ses_test", + }); + + expect(first).toEqual(concurrent); + expect(second).toEqual(first); + expect(createSessionEntry).toHaveBeenCalledTimes(1); + expect(createSessionEntry).toHaveBeenCalledWith( + expect.objectContaining({ + label: "Catalog session", + spawnedCwd: "/workspace", + initialEntry: { + acpBackendId: "acpx", + acpSessionBinding: { acpAgentId: "opencode", agentSessionId: "ses_test" }, + pluginExtensions: { + opencode: { sessionCatalog: { sourceThreadId: "ses_test" } }, + }, + }, + }), + ); + }, + ); + + it.runIf(process.platform !== "win32")( + "rejects paired-node and unknown OpenCode session continuation", + async () => { + await installFakeOpenCode(); + const { createSessionEntry, provider } = captureOpenCodeContinuationCatalog(); + + await expect( + provider.continueSession!({ hostId: "node:remote", threadId: "ses_test" }), + ).rejects.toThrow("paired-node OpenCode session rows are view-only"); + await expect( + provider.continueSession!({ hostId: "gateway", threadId: "missing" }), + ).rejects.toThrow("OpenCode session is unavailable"); + expect(createSessionEntry).not.toHaveBeenCalled(); + }, + ); + it.runIf(process.platform !== "win32")( "keeps truncated tool input on a valid UTF-16 boundary", async () => { @@ -469,7 +600,7 @@ describe("OpenCode session catalog", () => { cwd: "/remote/workspace", status: "stored", archived: false, - canContinue: false, + canContinue: true, canArchive: false, }, ], @@ -501,7 +632,13 @@ describe("OpenCode session catalog", () => { await expect(provider!.list({ hostIds: ["node:node-1"], search: "remote" })).resolves.toEqual([ expect.objectContaining({ - sessions: [expect.objectContaining({ threadId: "ses_remote", canOpenTerminal: true })], + sessions: [ + expect.objectContaining({ + threadId: "ses_remote", + canContinue: false, + canOpenTerminal: true, + }), + ], }), ]); expect(invoke).toHaveBeenNthCalledWith(1, { diff --git a/extensions/opencode/session-catalog.ts b/extensions/opencode/session-catalog.ts index 7cf7150d6ce7..e2049057cf40 100644 --- a/extensions/opencode/session-catalog.ts +++ b/extensions/opencode/session-catalog.ts @@ -326,7 +326,7 @@ function parseOpenCodeSession(value: unknown): SessionCatalogSession | undefined source: "opencode-cli", modelProvider: "opencode", archived: false, - canContinue: false, + canContinue: true, canArchive: false, }; } diff --git a/src/config/sessions/types.ts b/src/config/sessions/types.ts index da3a22dc83fa..9731b92653bd 100644 --- a/src/config/sessions/types.ts +++ b/src/config/sessions/types.ts @@ -80,6 +80,12 @@ export type CliSessionBinding = { reseedReceipt?: CliSessionReseedReceipt; }; +type AcpSessionBinding = { + acpBackendId: string; + acpAgentId: string; + agentSessionId: string; +}; + export type SessionCompactionCheckpointReason = | "manual" | "auto-threshold" @@ -520,6 +526,8 @@ export type SessionEntry = SessionRestartRecoveryState & memoryFlushLastFailureError?: string; cliSessionIds?: Record; cliSessionBindings?: Record; + /** Initialization fence for seeding canonical ACP metadata; cleared after creation. */ + acpSessionBinding?: AcpSessionBinding; claudeCliSessionId?: string; label?: string; /** User-defined organization bucket for session lists; unrelated to chat groupId/groupChannel. */ diff --git a/src/plugin-sdk/acp-runtime.test.ts b/src/plugin-sdk/acp-runtime.test.ts index 68b12eb18e70..f00fb2863cbb 100644 --- a/src/plugin-sdk/acp-runtime.test.ts +++ b/src/plugin-sdk/acp-runtime.test.ts @@ -1,5 +1,5 @@ // ACP runtime tests cover plugin-facing ACP runtime setup and gateway dispatch behavior. -import { beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { buildTestCtx } from "../auto-reply/reply/test-ctx.js"; import type { FinalizedMsgContext } from "../auto-reply/templating.js"; @@ -13,7 +13,12 @@ vi.mock("../auto-reply/reply/dispatch-acp.runtime.js", () => ({ tryDispatchAcpReply: dispatchMock, })); -import { tryDispatchAcpReplyHook } from "./acp-runtime.js"; +import { + registerAcpRuntimeBackend, + resolveAcpSessionAvailability, + testing, + tryDispatchAcpReplyHook, +} from "./acp-runtime.js"; const event = { ctx: buildTestCtx({ @@ -379,3 +384,43 @@ describe("tryDispatchAcpReplyHook", () => { }); }); }); + +describe("resolveAcpSessionAvailability", () => { + beforeEach(() => testing.resetAcpRuntimeBackendsForTests()); + afterEach(() => testing.resetAcpRuntimeBackendsForTests()); + + it("requires an allowed agent and a healthy registered backend", () => { + expect( + resolveAcpSessionAvailability({ config: {}, backendId: "acpx", agentId: "opencode" }), + ).toMatchObject({ available: false }); + registerAcpRuntimeBackend({ + id: "acpx", + runtime: { + ensureSession: vi.fn(), + async *runTurn() {}, + cancel: vi.fn(), + close: vi.fn(), + }, + }); + expect( + resolveAcpSessionAvailability({ config: {}, backendId: "acpx", agentId: "opencode" }), + ).toEqual({ available: true }); + expect( + resolveAcpSessionAvailability({ + config: { acp: { allowedAgents: ["pi"] } }, + backendId: "acpx", + agentId: "opencode", + }), + ).toMatchObject({ available: false, message: expect.stringContaining("not allowed") }); + }); + + it("honors the canonical ACP dispatch policy", () => { + expect( + resolveAcpSessionAvailability({ + config: { acp: { dispatch: { enabled: false } } }, + backendId: "acpx", + agentId: "pi", + }), + ).toMatchObject({ available: false, message: expect.stringContaining("dispatch is disabled") }); + }); +}); diff --git a/src/plugin-sdk/acp-runtime.ts b/src/plugin-sdk/acp-runtime.ts index 523c87d9bfc0..c80b9e5d1b52 100644 --- a/src/plugin-sdk/acp-runtime.ts +++ b/src/plugin-sdk/acp-runtime.ts @@ -1,7 +1,9 @@ // Public ACP runtime helpers for plugins that integrate with ACP control/session state. import { testing as managerTesting, getAcpSessionManager } from "../acp/control-plane/manager.js"; -import { testing as registryTesting } from "../acp/runtime/registry.js"; +import { resolveAcpAgentPolicyError, resolveAcpDispatchPolicyError } from "../acp/policy.js"; +import { testing as registryTesting, requireAcpRuntimeBackend } from "../acp/runtime/registry.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; export { getAcpSessionManager }; export { AcpRuntimeError, isAcpRuntimeError } from "../acp/runtime/errors.js"; @@ -9,9 +11,9 @@ export type { AcpRuntimeErrorCode } from "../acp/runtime/errors.js"; export { getAcpRuntimeBackend, registerAcpRuntimeBackend, - requireAcpRuntimeBackend, unregisterAcpRuntimeBackend, } from "../acp/runtime/registry.js"; +export { requireAcpRuntimeBackend }; export type { AcpRuntime, AcpRuntimeCapabilities, @@ -31,6 +33,28 @@ export { readAcpSessionEntry } from "../acp/runtime/session-meta.js"; export type { AcpSessionStoreEntry } from "../acp/runtime/session-meta.js"; export { tryDispatchAcpReplyHook } from "./acp-runtime-backend.js"; +export function resolveAcpSessionAvailability(params: { + config: OpenClawConfig; + backendId: string; + agentId: string; +}): { available: true } | { available: false; message: string } { + const policyError = + resolveAcpDispatchPolicyError(params.config) ?? + resolveAcpAgentPolicyError(params.config, params.agentId); + if (policyError) { + return { available: false, message: policyError.message }; + } + try { + requireAcpRuntimeBackend(params.backendId); + return { available: true }; + } catch (error) { + return { + available: false, + message: error instanceof Error ? error.message : "ACP runtime backend is unavailable.", + }; + } +} + // Keep test helpers off the hot init path. Eagerly merging them here can // create a back-edge through the bundled ACP runtime chunk before the imported // testing bindings finish initialization. diff --git a/src/plugin-sdk/session-catalog.ts b/src/plugin-sdk/session-catalog.ts index dfc5ca947efd..8f440f686e1d 100644 --- a/src/plugin-sdk/session-catalog.ts +++ b/src/plugin-sdk/session-catalog.ts @@ -11,6 +11,12 @@ export type { SessionUpstreamKind, SessionUpstreamProbe, } from "../plugins/session-catalog.js"; +export { + createSessionCatalogAdoptionCoordinator, + listAdoptedSessionCatalogSessions, + sessionCatalogAdoptedSessionKey, + sessionCatalogAdoptedSourceKey, +} from "../plugins/session-catalog.js"; export type { SessionCatalog, SessionCatalogCapabilities, diff --git a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts index 9609d70682cf..96b51592258f 100644 --- a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts +++ b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts @@ -533,12 +533,30 @@ export function createPluginRuntimeMock(overrides: DeepPartial = ) => { const sessionId = "plugin-runtime-mock-session"; const key = params.key; + const sessionInitialEntry = + "acpSessionBinding" in params.initialEntry + ? { + acpSessionBinding: { + acpBackendId: params.initialEntry.acpBackendId, + ...params.initialEntry.acpSessionBinding, + }, + ...(params.initialEntry.modelSelectionLocked + ? { modelSelectionLocked: true as const } + : {}), + ...(params.initialEntry.pluginExtensions + ? { pluginExtensions: structuredClone(params.initialEntry.pluginExtensions) } + : {}), + ...(params.initialEntry.pluginOwnerId + ? { pluginOwnerId: params.initialEntry.pluginOwnerId } + : {}), + } + : structuredClone(params.initialEntry); const initialEntry = { sessionId, updatedAt: Date.now(), ...(params.label !== undefined ? { label: params.label } : {}), ...(params.spawnedCwd !== undefined ? { spawnedCwd: params.spawnedCwd } : {}), - ...structuredClone(params.initialEntry), + ...sessionInitialEntry, ...(params.afterCreate ? { initializationPending: true as const } : {}), }; const initialized = { diff --git a/src/plugins/registry-runtime.ts b/src/plugins/registry-runtime.ts index 9e5aeedbb4af..ad2daf916d6f 100644 --- a/src/plugins/registry-runtime.ts +++ b/src/plugins/registry-runtime.ts @@ -614,10 +614,12 @@ export function createPluginRuntimeResolver(state: PluginRegistryState) { listSessionEntries: session.listSessionEntries, createSessionEntry: async (params) => await runWithPluginScope(async () => { - if ( - "agentHarnessId" in params.initialEntry === - "cliBackendId" in params.initialEntry - ) { + const runtimeOwnerCount = [ + "agentHarnessId" in params.initialEntry, + "cliBackendId" in params.initialEntry, + "acpSessionBinding" in params.initialEntry, + ].filter(Boolean).length; + if (runtimeOwnerCount !== 1) { throw new Error( `Plugin "${pluginId}" session creation requires exactly one runtime owner.`, ); @@ -629,6 +631,17 @@ export function createPluginRuntimeResolver(state: PluginRegistryState) { assertReservedSessionKeyOwned(params.key, "create"); return await session.createSessionEntry(params); } + if ("acpSessionBinding" in params.initialEntry) { + if (!params.key.startsWith(`plugin:${pluginId}:`)) { + throw new Error( + `Plugin "${pluginId}" session keys must start with "plugin:${pluginId}:".`, + ); + } + return await session.createSessionEntry({ + ...params, + initialEntry: { ...params.initialEntry, pluginOwnerId: pluginId }, + }); + } const cliInitial = params.initialEntry; const backend = registry.cliBackends.find( (entry) => entry.backend.id === cliInitial.cliBackendId, diff --git a/src/plugins/registry.runtime-config.test.ts b/src/plugins/registry.runtime-config.test.ts index 127c1233a686..76fcde5badf4 100644 --- a/src/plugins/registry.runtime-config.test.ts +++ b/src/plugins/registry.runtime-config.test.ts @@ -391,6 +391,60 @@ describe("plugin registry runtime config scope", () => { ).rejects.toThrow("requires exactly one runtime owner"); }); + it("limits ACP session creation to the calling plugin namespace", async () => { + const runtime = createPluginRuntime(); + const createSessionEntry = vi.fn(async (params) => ({ + key: params.key, + agentId: "main", + sessionId: "session-1", + entry: { sessionId: "session-1", updatedAt: 1 }, + })); + runtime.agent.session.createSessionEntry = createSessionEntry; + const pluginRegistry = createTestRegistry(runtime); + const record = createPluginRecord({ + id: "opencode", + source: "/plugins/opencode/index.js", + origin: "bundled", + enabled: true, + configSchema: false, + }); + const api = pluginRegistry.createApi(record, { config: {} as OpenClawConfig }); + const initialEntry = { + acpBackendId: "acpx", + acpSessionBinding: { + acpAgentId: "opencode", + agentSessionId: "source", + }, + }; + + await expect( + api.runtime.agent.session.createSessionEntry({ + cfg: {}, + key: "plugin:opencode:catalog-adopt:source", + initialEntry, + }), + ).resolves.toEqual(expect.objectContaining({ sessionId: "session-1" })); + expect(createSessionEntry).toHaveBeenCalledWith( + expect.objectContaining({ + initialEntry: expect.objectContaining({ pluginOwnerId: "opencode" }), + }), + ); + await expect( + api.runtime.agent.session.createSessionEntry({ + cfg: {}, + key: "agent:main:ordinary", + initialEntry, + }), + ).rejects.toThrow('must start with "plugin:opencode:"'); + await expect( + api.runtime.agent.session.createSessionEntry({ + cfg: {}, + key: "plugin:opencode:catalog-adopt:source", + initialEntry: { ...initialEntry, cliBackendId: "opencode" } as never, + }), + ).rejects.toThrow("requires exactly one runtime owner"); + }); + it("limits locked harness session mutation and execution to the harness owner", async () => { const reservedKey = "agent:main:harness:codex:thread-1"; const ordinaryKey = "agent:main:ordinary"; diff --git a/src/plugins/runtime/runtime-agent.acp-binding.test.ts b/src/plugins/runtime/runtime-agent.acp-binding.test.ts new file mode 100644 index 000000000000..64e83da52d7b --- /dev/null +++ b/src/plugins/runtime/runtime-agent.acp-binding.test.ts @@ -0,0 +1,165 @@ +import { describe, expect, it, vi } from "vitest"; +import { readAcpSessionMeta, upsertAcpSessionMeta } from "../../acp/runtime/session-meta.js"; +import { withOpenClawTestState } from "../../test-utils/openclaw-test-state.js"; +import { createRuntimeAgent } from "./runtime-agent.js"; + +describe("plugin runtime ACP session creation", () => { + it("persists a plugin-owned native resume binding", async () => { + await withOpenClawTestState({ label: "plugin-runtime-acp-session-create" }, async () => { + const runtime = createRuntimeAgent(); + const created = await runtime.session.createSessionEntry({ + cfg: {}, + key: "plugin:acpx:catalog-adopt:pi:source", + label: "Pi source", + spawnedCwd: "/workspace/pi", + initialEntry: { + acpBackendId: "acpx", + acpSessionBinding: { acpAgentId: "pi", agentSessionId: "pi-source" }, + pluginOwnerId: "acpx", + }, + }); + + expect(created.entry).toMatchObject({ + createdVia: "plugin", + createdActor: { type: "system", id: "acpx" }, + pluginOwnerId: "acpx", + label: "Pi source", + spawnedCwd: "/workspace/pi", + }); + expect(created.entry.initializationPending).toBeUndefined(); + expect(readAcpSessionMeta({ cfg: {}, sessionKey: created.key })).toMatchObject({ + backend: "acpx", + agent: "pi", + runtimeSessionName: created.key, + identity: { + state: "resolved", + agentSessionId: "pi-source", + source: "ensure", + }, + mode: "persistent", + cwd: "/workspace/pi", + state: "idle", + }); + }); + }); + + it("rejects recovery when the native resume binding differs", async () => { + await withOpenClawTestState({ label: "plugin-runtime-acp-recovery-binding" }, async () => { + const runtime = createRuntimeAgent(); + const key = "agent:main:plugin:opencode:catalog-adopt:source"; + const storePath = runtime.session.resolveStorePath(undefined, { agentId: "main" }); + await runtime.session.upsertSessionEntry({ + storePath, + sessionKey: key, + entry: { + sessionId: "interrupted-acp-initializer", + updatedAt: Date.now(), + delivery: { kind: "none" }, + initializationPending: true, + pluginOwnerId: "opencode", + spawnedCwd: "/workspace/opencode", + acpSessionBinding: { + acpBackendId: "acpx", + acpAgentId: "opencode", + agentSessionId: "expected-source", + }, + }, + }); + await upsertAcpSessionMeta({ + cfg: {}, + sessionKey: key, + mutate: () => ({ + backend: "acpx", + agent: "opencode", + runtimeSessionName: key, + identity: { + state: "resolved", + agentSessionId: "different-source", + source: "ensure", + lastUpdatedAt: Date.now(), + }, + mode: "persistent", + cwd: "/workspace/opencode", + state: "idle", + lastActivityAt: Date.now(), + }), + }); + const storedBeforeRecovery = runtime.session.getSessionEntry({ + sessionKey: key, + readConsistency: "latest", + }); + const afterCreate = vi.fn(async () => ({ pluginExtensions: {} })); + + await expect( + runtime.session.createSessionEntry({ + cfg: {}, + key, + spawnedCwd: "/workspace/opencode", + recoverMatchingInitialEntry: true, + initialEntry: { + acpBackendId: "acpx", + acpSessionBinding: { + acpAgentId: "opencode", + agentSessionId: "expected-source", + }, + pluginOwnerId: "opencode", + }, + afterCreate, + }), + ).rejects.toThrow("does not match its trusted recovery state"); + expect(afterCreate).not.toHaveBeenCalled(); + expect( + runtime.session.getSessionEntry({ sessionKey: key, readConsistency: "latest" }), + ).toEqual(storedBeforeRecovery); + }); + }); + + it("recovers an interrupted ACP initializer before metadata was seeded", async () => { + await withOpenClawTestState({ label: "plugin-runtime-acp-recovery-missing-meta" }, async () => { + const runtime = createRuntimeAgent(); + const key = "agent:main:plugin:acpx:catalog-adopt:pi:recovery"; + const storePath = runtime.session.resolveStorePath(undefined, { agentId: "main" }); + const marker = { acpx: { piSessionCatalog: { sourceThreadId: "pi-source" } } }; + await runtime.session.upsertSessionEntry({ + storePath, + sessionKey: key, + entry: { + sessionId: "interrupted-before-acp-meta", + updatedAt: Date.now(), + delivery: { kind: "none" }, + initializationPending: true, + pluginOwnerId: "acpx", + spawnedCwd: "/workspace/pi", + pluginExtensions: marker, + acpSessionBinding: { + acpBackendId: "acpx", + acpAgentId: "pi", + agentSessionId: "pi-source", + }, + }, + }); + + const recovered = await runtime.session.createSessionEntry({ + cfg: {}, + key, + spawnedCwd: "/workspace/pi", + recoverMatchingInitialEntry: true, + initialEntry: { + acpBackendId: "acpx", + acpSessionBinding: { acpAgentId: "pi", agentSessionId: "pi-source" }, + pluginOwnerId: "acpx", + pluginExtensions: marker, + }, + afterCreate: async () => ({ pluginExtensions: marker }), + }); + + expect(recovered.entry.initializationPending).toBeUndefined(); + expect(recovered.entry.acpSessionBinding).toBeUndefined(); + expect(readAcpSessionMeta({ cfg: {}, sessionKey: key })).toMatchObject({ + backend: "acpx", + agent: "pi", + identity: { agentSessionId: "pi-source" }, + }); + }); + }); +}); diff --git a/src/plugins/runtime/runtime-agent.ts b/src/plugins/runtime/runtime-agent.ts index d94b1ff5d313..4f202fd01d68 100644 --- a/src/plugins/runtime/runtime-agent.ts +++ b/src/plugins/runtime/runtime-agent.ts @@ -32,7 +32,7 @@ import { } from "../../config/sessions/session-accessor.js"; import { normalizeResolvedMaintenanceConfigInput } from "../../config/sessions/store-maintenance.js"; import type { ResolvedSessionMaintenanceConfigInput } from "../../config/sessions/store-maintenance.js"; -import type { SessionEntry } from "../../config/sessions/types.js"; +import type { SessionAcpMeta, SessionEntry } from "../../config/sessions/types.js"; import { beginSessionWorkAdmission, isSessionWorkAdmissionActive, @@ -170,9 +170,16 @@ async function createSessionEntry( ): Promise>> { // Session creation stays behind the canonical Gateway lifecycle boundary while // keeping that heavier runtime out of plugin discovery and cold startup. - const [{ createGatewaySession }, { resolveGatewaySessionStoreTarget }] = await Promise.all([ + const [ + { createGatewaySession }, + { resolveGatewaySessionStoreTarget }, + { readAcpSessionMetaForEntry, upsertAcpSessionMeta }, + ] = await Promise.all([ import("../../gateway/session-create-service.js"), import("../../gateway/session-utils.js"), + // session-meta rides the same lazy boundary: session-utils already pulls it + // in transitively, so a separate import here would only duplicate the edge. + import("../../acp/runtime/session-meta.js"), ]); type CreatedContext = Parameters< NonNullable[0]["afterCreate"]> @@ -183,7 +190,54 @@ async function createSessionEntry( ...(params.agentId !== undefined ? { agentId: params.agentId } : {}), }); const cliInitial = "cliBackendId" in params.initialEntry ? params.initialEntry : undefined; + const acpInitial = "acpSessionBinding" in params.initialEntry ? params.initialEntry : undefined; const harnessInitial = "agentHarnessId" in params.initialEntry ? params.initialEntry : undefined; + const pluginInitial = cliInitial ?? acpInitial; + const acpBackendId = acpInitial?.acpBackendId.trim(); + const acpAgentId = acpInitial?.acpSessionBinding.acpAgentId.trim(); + const agentSessionId = acpInitial?.acpSessionBinding.agentSessionId.trim(); + if (acpInitial && (!acpBackendId || !acpAgentId || !agentSessionId)) { + throw new Error("initial ACP session binding fields must be non-empty"); + } + const initialAcpMeta = (now: number): SessionAcpMeta | undefined => + acpInitial + ? { + backend: acpBackendId!, + agent: acpAgentId!, + runtimeSessionName: target.canonicalKey, + identity: { + state: "resolved", + agentSessionId: agentSessionId!, + source: "ensure", + lastUpdatedAt: now, + }, + mode: "persistent", + ...(params.spawnedCwd?.trim() ? { cwd: params.spawnedCwd.trim() } : {}), + state: "idle", + lastActivityAt: now, + } + : undefined; + const persistedAcpBinding = acpInitial + ? { acpBackendId: acpBackendId!, acpAgentId: acpAgentId!, agentSessionId: agentSessionId! } + : undefined; + const acpMetaMatches = (meta: SessionAcpMeta | undefined): boolean => { + return Boolean( + meta && + meta.backend === acpBackendId && + meta.agent === acpAgentId && + meta.runtimeSessionName === target.canonicalKey && + meta.identity?.state === "resolved" && + meta.identity.agentSessionId === agentSessionId && + meta.mode === "persistent" && + meta.cwd === (params.spawnedCwd?.trim() || undefined), + ); + }; + const initializesAfterCreate = Boolean(params.afterCreate || acpInitial); + const matchesExceptUpdatedAt = (left: SessionEntry, right: SessionEntry): boolean => { + const { updatedAt: _leftUpdatedAt, ...leftStable } = left; + const { updatedAt: _rightUpdatedAt, ...rightStable } = right; + return isDeepStrictEqual(leftStable, rightStable); + }; const identities = new Set([target.canonicalKey, ...target.storeKeys]); return await runExclusiveSessionLifecycleMutation({ scope: target.storePath, @@ -202,15 +256,35 @@ async function createSessionEntry( let rollbackExpectedEntry: SessionEntry | undefined; const runAfterCreate = async (context: CreatedContext): Promise => { callbackContext = context; - rollbackExpectedEntry = structuredClone(context.entry); + if (acpInitial) { + const meta = initialAcpMeta(Date.now()); + const persisted = await upsertAcpSessionMeta({ + cfg: params.cfg, + sessionKey: context.key, + mutate: () => meta, + }); + if (!persisted?.acp) { + throw new Error(`could not persist initial ACP binding for ${context.key}`); + } + const persistedEntry = getSessionEntry({ + sessionKey: context.key, + storePath: context.storePath, + readConsistency: "latest", + }); + if (!persistedEntry) { + throw new Error(`created ACP session ${context.key} disappeared during initialization`); + } + callbackContext = { ...context, entry: persistedEntry }; + } + rollbackExpectedEntry = structuredClone(callbackContext.entry); if (!afterCreate) { return; } const finalPatch = await afterCreate({ - key: context.key, - agentId: context.agentId, - sessionId: context.entry.sessionId, - entry: structuredClone(context.entry), + key: callbackContext.key, + agentId: callbackContext.agentId, + sessionId: callbackContext.entry.sessionId, + entry: structuredClone(callbackContext.entry), }); if (finalPatch === undefined) { return; @@ -238,10 +312,16 @@ async function createSessionEntry( const expectedSpawnedCwd = params.spawnedCwd?.trim() || undefined; const expectedExecNode = params.execNode?.trim() || undefined; const expectedExecCwd = params.execCwd?.trim() || undefined; + const matchingAcpMeta = acpInitial + ? readAcpSessionMetaForEntry({ + sessionKey: target.canonicalKey, + entry: matchingEntry, + }) + : undefined; const initialEntryMatches = matchingEntry.initializationPending === true && matchingEntry.agentHarnessId === harnessInitial?.agentHarnessId && - matchingEntry.pluginOwnerId === cliInitial?.pluginOwnerId && + matchingEntry.pluginOwnerId === pluginInitial?.pluginOwnerId && matchingEntry.modelSelectionLocked === params.initialEntry.modelSelectionLocked && (!cliInitial || (matchingEntry.providerOverride === cliInitial.cliBackendId && @@ -250,6 +330,9 @@ async function createSessionEntry( matchingEntry.cliSessionBindings?.[cliInitial.cliBackendId], cliInitial.cliSessionBinding, ))) && + (!acpInitial || + (isDeepStrictEqual(matchingEntry.acpSessionBinding, persistedAcpBinding) && + (matchingAcpMeta === undefined || acpMetaMatches(matchingAcpMeta)))) && matchingEntry.spawnedCwd === expectedSpawnedCwd && matchingEntry.execNode === expectedExecNode && matchingEntry.execCwd === expectedExecCwd && @@ -293,25 +376,33 @@ async function createSessionEntry( }, } : {}), + ...(acpInitial + ? { + pluginOwnerId: acpInitial.pluginOwnerId, + acpSessionBinding: persistedAcpBinding, + } + : {}), ...(params.initialEntry.modelSelectionLocked === true ? { modelSelectionLocked: true } : {}), ...(params.initialEntry.pluginExtensions ? { pluginExtensions: params.initialEntry.pluginExtensions } : {}), - ...(afterCreate ? { initializationPending: true } : {}), + ...(initializesAfterCreate ? { initializationPending: true } : {}), }, ...(harnessInitial ? { authorizedAgentHarnessId: harnessInitial.agentHarnessId } : {}), - ...(cliInitial?.pluginOwnerId ? { authorizedPluginId: cliInitial.pluginOwnerId } : {}), + ...(pluginInitial?.pluginOwnerId + ? { authorizedPluginId: pluginInitial.pluginOwnerId } + : {}), creation: { via: "plugin", actor: { type: "system", - ...(cliInitial?.pluginOwnerId ? { id: cliInitial.pluginOwnerId } : {}), + ...(pluginInitial?.pluginOwnerId ? { id: pluginInitial.pluginOwnerId } : {}), }, }, commandSource: "plugin-runtime", - ...(afterCreate ? { afterCreate: runAfterCreate } : {}), + ...(initializesAfterCreate ? { afterCreate: runAfterCreate } : {}), }); if (!result.ok) { throw new Error(result.error.message); @@ -322,10 +413,11 @@ async function createSessionEntry( throw new Error("session creation recovery requires a final patch"); } let finalEntry = created.entry; - if (afterCreate) { + if (initializesAfterCreate) { const patch: Partial = { ...finalEntryPatch, initializationPending: undefined, + ...(acpInitial ? { acpSessionBinding: undefined } : {}), }; const expectedEntry = rollbackExpectedEntry; if (!callbackContext || !expectedEntry) { @@ -373,7 +465,17 @@ async function createSessionEntry( try { // Delete only the untouched row created for this callback. A concurrent // claimant changes the snapshot and must survive failed initialization. - const expectedEntry = rollbackExpectedEntry ?? callbackContext.entry; + let expectedEntry = rollbackExpectedEntry ?? callbackContext.entry; + if (acpInitial && !rollbackExpectedEntry) { + const currentEntry = getSessionEntry({ + sessionKey: callbackContext.key, + storePath: callbackContext.storePath, + readConsistency: "latest", + }); + if (currentEntry && matchesExceptUpdatedAt(currentEntry, callbackContext.entry)) { + expectedEntry = currentEntry; + } + } const rollbackParams = { agentId: callbackContext.agentId, archiveTranscript: true, @@ -394,7 +496,7 @@ async function createSessionEntry( ? await rollbackAgentHarnessSessionEntryLifecycle(rollbackParams) : await rollbackPluginOwnedSessionEntryLifecycle({ ...rollbackParams, - expectedPluginOwnerId: cliInitial?.pluginOwnerId ?? "", + expectedPluginOwnerId: pluginInitial?.pluginOwnerId ?? "", }) : await deleteSessionEntryLifecycle(rollbackParams); if (!rolledBack.deleted) { @@ -402,6 +504,13 @@ async function createSessionEntry( cause: error, }); } + if (acpInitial) { + await upsertAcpSessionMeta({ + cfg: params.cfg, + sessionKey: callbackContext.key, + mutate: () => null, + }); + } } catch (rollbackError) { const aggregateError = new AggregateError( [error, rollbackError], diff --git a/src/plugins/runtime/types-core.ts b/src/plugins/runtime/types-core.ts index 233ddc7bf7e1..4c65bccbd7de 100644 --- a/src/plugins/runtime/types-core.ts +++ b/src/plugins/runtime/types-core.ts @@ -113,6 +113,17 @@ type RuntimeCreateSessionEntryBaseParams = { pluginExtensions?: RuntimeSessionPluginExtensions; /** Registry-injected owner; plugin callers cannot select another owner. */ pluginOwnerId?: string; + } + | { + acpBackendId: string; + acpSessionBinding: { + acpAgentId: string; + agentSessionId: string; + }; + modelSelectionLocked?: true; + pluginExtensions?: RuntimeSessionPluginExtensions; + /** Registry-injected owner; plugin callers cannot select another owner. */ + pluginOwnerId?: string; }; }; type RuntimeCreateSessionEntryParams = RuntimeCreateSessionEntryBaseParams & diff --git a/src/plugins/session-catalog.ts b/src/plugins/session-catalog.ts index 4a2475853aa5..11f92bc01448 100644 --- a/src/plugins/session-catalog.ts +++ b/src/plugins/session-catalog.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import type { SessionCatalogHost, SessionsCatalogArchiveParams, @@ -5,6 +6,9 @@ import type { SessionsCatalogReadParams, SessionsCatalogReadResult, } from "../../packages/gateway-protocol/src/schema/sessions-catalog.js"; +import { listAgentIds, resolveDefaultAgentId } from "../agents/agent-scope.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import type { PluginRuntime } from "./runtime/types.js"; export type SessionCatalogListProviderParams = { /** Trimmed, non-empty search capped at 500 UTF-16 code units by the gateway. */ @@ -123,3 +127,72 @@ export type SessionCatalogProvider = { threadId: string; }) => Promise; }; + +type SessionCatalogAdoptedSource = { hostId: string; threadId: string }; +type SessionCatalogEntry = ReturnType< + PluginRuntime["agent"]["session"]["listSessionEntries"] +>[number]["entry"]; + +export function sessionCatalogAdoptedSourceKey(hostId: string, threadId: string): string { + return `${hostId}\0${threadId}`; +} + +export function sessionCatalogAdoptedSessionKey(prefix: string, source: string): string { + return `${prefix}${createHash("sha256").update(source).digest("hex")}`; +} + +export function listAdoptedSessionCatalogSessions(params: { + config: OpenClawConfig; + pluginId: string; + runtime: PluginRuntime; + sourceFromEntry: (entry: SessionCatalogEntry) => SessionCatalogAdoptedSource | undefined; +}): Map { + const defaultAgentId = resolveDefaultAgentId(params.config); + const agentIds = [ + defaultAgentId, + ...listAgentIds(params.config).filter((agentId) => agentId !== defaultAgentId), + ]; + const adopted = new Map(); + for (const { sessionKey, entry } of agentIds.flatMap((agentId) => + params.runtime.agent.session.listSessionEntries({ agentId, readOnly: true }), + )) { + const source = params.sourceFromEntry(entry); + if (source && entry.pluginOwnerId === params.pluginId && entry.initializationPending !== true) { + adopted.set(sessionCatalogAdoptedSourceKey(source.hostId, source.threadId), sessionKey); + } + } + return adopted; +} + +export function createSessionCatalogAdoptionCoordinator() { + const operations = new Map>(); + return async (params: { + sourceKey: string; + findExisting: () => string | undefined; + create: () => Promise<{ sessionKey: string }>; + }): Promise<{ sessionKey: string }> => { + const existing = params.findExisting(); + if (existing) { + return { sessionKey: existing }; + } + const pending = operations.get(params.sourceKey); + if (pending) { + return await pending; + } + const operation = params.create().catch((error: unknown) => { + const raced = params.findExisting(); + if (raced) { + return { sessionKey: raced }; + } + throw error; + }); + operations.set(params.sourceKey, operation); + try { + return await operation; + } finally { + if (operations.get(params.sourceKey) === operation) { + operations.delete(params.sourceKey); + } + } + }; +} diff --git a/src/plugins/session-entry-slot-keys.ts b/src/plugins/session-entry-slot-keys.ts index e26439264819..f34c328be3de 100644 --- a/src/plugins/session-entry-slot-keys.ts +++ b/src/plugins/session-entry-slot-keys.ts @@ -156,6 +156,7 @@ const SESSION_ENTRY_RESERVED_SLOT_KEY_LIST = [ "memoryFlushLastFailureError", "cliSessionIds", "cliSessionBindings", + "acpSessionBinding", "claudeCliSessionId", "label", "category",