diff --git a/docs/plugins/reference/acpx.md b/docs/plugins/reference/acpx.md index 20af452f9db1..78d199c0a766 100644 --- a/docs/plugins/reference/acpx.md +++ b/docs/plugins/reference/acpx.md @@ -27,10 +27,12 @@ nodes. Stored sessions appear in the **Pi** sessions-sidebar group, with 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. +session file, and OpenClaw imports the recent native history into the adopted +session transcript. Very long transcripts import only their most recent 200 +items using a 512 KiB serialized-item budget. 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 diff --git a/docs/plugins/reference/opencode.md b/docs/plugins/reference/opencode.md index f3f4042c1f5a..9b37e98b6cd8 100644 --- a/docs/plugins/reference/opencode.md +++ b/docs/plugins/reference/opencode.md @@ -28,8 +28,10 @@ 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. +viewer continues to show that history. OpenClaw also imports the recent native +history into the adopted session transcript. Very long transcripts import only +their most recent 200 items using a 512 KiB serialized-item budget. Paired-node +rows remain view-only. The restricted environment and `--pure` mode prevent catalog browsing from loading project plugins or inheriting unrelated Gateway credentials. diff --git a/extensions/acpx/src/pi-session-catalog-continuation.test.ts b/extensions/acpx/src/pi-session-catalog-continuation.test.ts index ec4ee138400e..b14762f6d2ff 100644 --- a/extensions/acpx/src/pi-session-catalog-continuation.test.ts +++ b/extensions/acpx/src/pi-session-catalog-continuation.test.ts @@ -8,12 +8,54 @@ type ResolveAcpSessionAvailability = const acpRuntimeMocks = vi.hoisted(() => ({ resolveAcpSessionAvailability: vi.fn(() => ({ available: true })), })); +const transcriptMocks = vi.hoisted(() => ({ + messages: [] as Array>, + failAfter: undefined as number | undefined, +})); vi.mock("openclaw/plugin-sdk/acp-runtime", async (importOriginal) => ({ ...(await importOriginal()), resolveAcpSessionAvailability: acpRuntimeMocks.resolveAcpSessionAvailability, })); +vi.mock("openclaw/plugin-sdk/session-transcript-runtime", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + withSessionTranscriptWriteLock: async ( + _params: unknown, + run: (context: { + appendMessage: (params: { + message: Record; + idempotencyLookup?: string; + }) => Promise; + }) => Promise, + ) => { + const pending: Array> = []; + await run({ + appendMessage: async ({ message, idempotencyLookup }) => { + if (transcriptMocks.failAfter === pending.length + 1) { + throw new Error("transcript append failed"); + } + const key = message.idempotencyKey; + if ( + idempotencyLookup === "scan" && + typeof key === "string" && + [...transcriptMocks.messages, ...pending].some( + (candidate) => candidate.idempotencyKey === key, + ) + ) { + return; + } + pending.push(message); + }, + }); + transcriptMocks.messages.push(...pending); + }, + }; +}); + import { capturePiContinuationCatalog, createPiStoreFixture, @@ -28,6 +70,8 @@ const originalPath = process.env.PATH; afterEach(async () => { acpRuntimeMocks.resolveAcpSessionAvailability.mockReset().mockReturnValue({ available: true }); process.env.PATH = originalPath; + transcriptMocks.messages.length = 0; + transcriptMocks.failAfter = undefined; if (originalSessionDir === undefined) { delete process.env.PI_CODING_AGENT_SESSION_DIR; } else { @@ -84,6 +128,46 @@ describe("Pi session catalog continuation", () => { }, }), ); + expect( + transcriptMocks.messages.map((message) => + typeof message.content === "string" + ? message.content + : (message.content as Array<{ text: string }>)[0]?.text, + ), + ).toEqual([ + "hello", + "Thinking\n\nthinking", + "hi", + 'Tool call\n\nbash\n{"command":"pwd"}', + "Tool result\n\nbash\n/workspace", + ]); + expect(transcriptMocks.messages[0]?.["__openclaw"]).toEqual({ + mirrorOrigin: "pi-catalog-import", + }); + + const createParams = createSessionEntry.mock.calls[0]?.[0]; + const adopted = createSessionEntry.mock.results[0]?.value; + await createParams?.afterCreate?.(await adopted); + expect(transcriptMocks.messages).toHaveLength(5); + }); + + it("rolls adoption back when transcript import fails", async () => { + await createPiStoreFixture( + temporaryDirectories, + "hi", + "Pi catalog session", + { command: "pwd" }, + true, + ); + await installFakePiFixture(temporaryDirectories, originalPath); + transcriptMocks.failAfter = 2; + const { entries, provider } = capturePiContinuationCatalog(); + + await expect( + provider.continueSession!({ hostId: "gateway", threadId: "pi-session" }), + ).rejects.toThrow("transcript append failed"); + expect(entries).toEqual([]); + expect(transcriptMocks.messages).toEqual([]); }); it("rejects paired-node and unknown session continuation", async () => { diff --git a/extensions/acpx/src/pi-session-catalog-plugin.ts b/extensions/acpx/src/pi-session-catalog-plugin.ts index 0bd1da5ef801..17fd974b36c5 100644 --- a/extensions/acpx/src/pi-session-catalog-plugin.ts +++ b/extensions/acpx/src/pi-session-catalog-plugin.ts @@ -23,6 +23,7 @@ import type { } from "openclaw/plugin-sdk/session-catalog"; import { createSessionCatalogAdoptionCoordinator, + importSessionCatalogHistory, listAdoptedSessionCatalogSessions, sessionCatalogAdoptedSessionKey, sessionCatalogAdoptedSourceKey, @@ -479,8 +480,6 @@ async function continuePiSession( } 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), @@ -496,9 +495,25 @@ async function continuePiSession( }, pluginExtensions: { acpx: { piSessionCatalog: marker } }, }, - afterCreate: async () => ({ - pluginExtensions: { acpx: { piSessionCatalog: marker } }, - }), + afterCreate: async (entry) => { + await importSessionCatalogHistory({ + catalogId: "pi", + threadId, + read: async ({ cursor, limit }) => + await readPiTranscript(api.runtime, { + hostId, + threadId, + limit, + ...(cursor ? { cursor } : {}), + }), + sessionId: entry.sessionId, + sessionKey: entry.key, + agentId: entry.agentId, + ...(record.cwd ? { cwd: record.cwd } : {}), + config, + }); + return { pluginExtensions: { acpx: { piSessionCatalog: marker } } }; + }, }); return { sessionKey: created.key }; }, diff --git a/extensions/acpx/src/pi-session-catalog.test-support.ts b/extensions/acpx/src/pi-session-catalog.test-support.ts index fd00efa7a0d0..4586fb98140b 100644 --- a/extensions/acpx/src/pi-session-catalog.test-support.ts +++ b/extensions/acpx/src/pi-session-catalog.test-support.ts @@ -126,17 +126,30 @@ export function capturePiContinuationCatalog() { sessionId: "adopted-pi-session", updatedAt: Date.now(), pluginOwnerId: "acpx", + initializationPending: true as const, ...(params.label ? { label: params.label } : {}), ...(params.spawnedCwd ? { spawnedCwd: params.spawnedCwd } : {}), pluginExtensions: params.initialEntry.pluginExtensions, }; entries.push({ sessionKey, entry }); - return { + const created = { key: sessionKey, agentId: params.agentId ?? "main", sessionId: entry.sessionId, entry, }; + try { + const finalPatch = await params.afterCreate?.(created); + entry.pluginExtensions = finalPatch?.pluginExtensions ?? entry.pluginExtensions; + delete (entry as { initializationPending?: true }).initializationPending; + return created; + } catch (error) { + entries.splice( + entries.findIndex((candidate) => candidate.entry === entry), + 1, + ); + throw error; + } }, ); registerPiSessionCatalog({ diff --git a/extensions/opencode/session-catalog-plugin.ts b/extensions/opencode/session-catalog-plugin.ts index 3b2aa235c931..da155bc3a7a5 100644 --- a/extensions/opencode/session-catalog-plugin.ts +++ b/extensions/opencode/session-catalog-plugin.ts @@ -20,6 +20,7 @@ import type { } from "openclaw/plugin-sdk/session-catalog"; import { createSessionCatalogAdoptionCoordinator, + importSessionCatalogHistory, listAdoptedSessionCatalogSessions, sessionCatalogAdoptedSessionKey, sessionCatalogAdoptedSourceKey, @@ -501,8 +502,6 @@ async function continueOpenCodeSession( 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), @@ -518,9 +517,25 @@ async function continueOpenCodeSession( }, pluginExtensions: { opencode: { sessionCatalog: marker } }, }, - afterCreate: async () => ({ - pluginExtensions: { opencode: { sessionCatalog: marker } }, - }), + afterCreate: async (entry) => { + await importSessionCatalogHistory({ + catalogId: "opencode", + threadId, + read: async ({ cursor, limit }) => + await readOpenCodeTranscript(api.runtime, { + hostId, + threadId, + limit, + ...(cursor ? { cursor } : {}), + }), + sessionId: entry.sessionId, + sessionKey: entry.key, + agentId: entry.agentId, + ...(record.cwd ? { cwd: record.cwd } : {}), + config, + }); + return { pluginExtensions: { opencode: { sessionCatalog: marker } } }; + }, }); return { sessionKey: created.key }; }, diff --git a/extensions/opencode/session-catalog.test.ts b/extensions/opencode/session-catalog.test.ts index 37da0c7539ec..c0b446ac4e8c 100644 --- a/extensions/opencode/session-catalog.test.ts +++ b/extensions/opencode/session-catalog.test.ts @@ -19,6 +19,9 @@ const childProcessMocks = vi.hoisted(() => ({ children: [] as ChildProcess[], spawn: vi.fn(), })); +const transcriptMocks = vi.hoisted(() => ({ + messages: [] as Array>, +})); vi.mock("node:child_process", async (importOriginal) => { const actual = await importOriginal(); @@ -35,6 +38,37 @@ vi.mock("openclaw/plugin-sdk/acp-runtime", async (importOriginal) => ({ resolveAcpSessionAvailability: acpRuntimeMocks.resolveAcpSessionAvailability, })); +vi.mock("openclaw/plugin-sdk/session-transcript-runtime", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + withSessionTranscriptWriteLock: async ( + _params: unknown, + run: (context: { + appendMessage: (params: { + message: Record; + idempotencyLookup?: string; + }) => Promise; + }) => Promise, + ) => { + await run({ + appendMessage: async ({ message, idempotencyLookup }) => { + const key = message.idempotencyKey; + if ( + idempotencyLookup === "scan" && + typeof key === "string" && + transcriptMocks.messages.some((candidate) => candidate.idempotencyKey === key) + ) { + return; + } + transcriptMocks.messages.push(message); + }, + }); + }, + }; +}); + vi.mock("openclaw/plugin-sdk/node-host", async (importOriginal) => { const actual = await importOriginal(); return { @@ -106,17 +140,30 @@ function captureOpenCodeContinuationCatalog() { sessionId: "adopted-opencode-session", updatedAt: Date.now(), pluginOwnerId: "opencode", + initializationPending: true as const, ...(params.label ? { label: params.label } : {}), ...(params.spawnedCwd ? { spawnedCwd: params.spawnedCwd } : {}), pluginExtensions: params.initialEntry.pluginExtensions, }; entries.push({ sessionKey, entry }); - return { + const created = { key: sessionKey, agentId: params.agentId ?? "main", sessionId: entry.sessionId, entry, }; + try { + const finalPatch = await params.afterCreate?.(created); + entry.pluginExtensions = finalPatch?.pluginExtensions ?? entry.pluginExtensions; + delete (entry as { initializationPending?: true }).initializationPending; + return created; + } catch (error) { + entries.splice( + entries.findIndex((candidate) => candidate.entry === entry), + 1, + ); + throw error; + } }, ); registerOpenCodeSessionCatalog({ @@ -266,6 +313,7 @@ afterEach(async () => { acpRuntimeMocks.resolveAcpSessionAvailability.mockReset().mockReturnValue({ available: true }); nodeHostMocks.runNodePtyCommand.mockClear(); childProcessMocks.spawn.mockClear(); + transcriptMocks.messages.length = 0; await Promise.all(childProcessMocks.children.splice(0).map((child) => stopChild(child))); process.env.PATH = originalPath; if (originalPathExt === undefined) { @@ -450,6 +498,22 @@ describe("OpenCode session catalog", () => { }, }), ); + expect( + transcriptMocks.messages.map((message) => + typeof message.content === "string" + ? message.content + : (message.content as Array<{ text: string }>)[0]?.text, + ), + ).toEqual([ + "hello", + "Thinking\n\nthinking", + "hi", + 'Tool call\n\nbash\n{"command":"pwd"}', + "Tool result\n\n/workspace", + ]); + expect(transcriptMocks.messages[0]?.["__openclaw"]).toEqual({ + mirrorOrigin: "opencode-catalog-import", + }); }, ); diff --git a/src/plugin-sdk/session-catalog.ts b/src/plugin-sdk/session-catalog.ts index 954c1c8c1b2e..a109e2cc5de0 100644 --- a/src/plugin-sdk/session-catalog.ts +++ b/src/plugin-sdk/session-catalog.ts @@ -19,6 +19,7 @@ export { sessionCatalogAdoptedSessionKey, sessionCatalogAdoptedSourceKey, } from "../plugins/session-catalog.js"; +export { importSessionCatalogHistory } from "../plugins/session-catalog-history-import.js"; export type { SessionCatalog, SessionCatalogCapabilities, diff --git a/src/plugins/session-catalog-history-import.ts b/src/plugins/session-catalog-history-import.ts new file mode 100644 index 000000000000..bf4e154c4471 --- /dev/null +++ b/src/plugins/session-catalog-history-import.ts @@ -0,0 +1,190 @@ +import type { + SessionCatalogTranscriptItem, + SessionsCatalogReadResult, +} from "../../packages/gateway-protocol/src/schema/sessions-catalog.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import type { AgentMessage } from "../plugin-sdk/agent-core.js"; +import { withSessionTranscriptWriteLock } from "../plugin-sdk/session-transcript-runtime.js"; + +const SESSION_CATALOG_HISTORY_IMPORT_MAX_ITEMS = 200; +const SESSION_CATALOG_HISTORY_IMPORT_MAX_BYTES = 512 * 1024; +const SESSION_CATALOG_HISTORY_IMPORT_PAGE_LIMIT = 100; + +function importedSessionCatalogMessage(params: { + catalogId: string; + item: SessionCatalogTranscriptItem; + fallbackTimestamp: number; +}): AgentMessage | undefined { + const parsedTimestamp = params.item.timestamp ? Date.parse(params.item.timestamp) : Number.NaN; + const timestamp = Number.isFinite(parsedTimestamp) ? parsedTimestamp : params.fallbackTimestamp; + const importedText = params.item.text?.trim(); + if (!importedText && params.item.type === "reasoning") { + return undefined; + } + const text = importedText || "[Unsupported catalog transcript item]"; + if (params.item.type === "userMessage") { + // Imported native rows are not OpenClaw-authored; mirrorOrigin excludes them + // from self-echo provenance so a repeated external prompt stays observable. + return { + role: "user", + content: text, + timestamp, + __openclaw: { mirrorOrigin: `${params.catalogId}-catalog-import` }, + } as AgentMessage; + } + const prefix = + params.item.type === "reasoning" + ? "Thinking\n\n" + : params.item.type === "toolCall" + ? "Tool call\n\n" + : params.item.type === "toolResult" + ? "Tool result\n\n" + : params.item.type === "other" + ? "Other\n\n" + : ""; + return { + role: "assistant", + content: [{ type: "text", text: `${prefix}${text}` }], + timestamp, + api: "openai-responses", + provider: params.catalogId, + model: params.item.model ?? "native-history", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + } as AgentMessage; +} + +function fitSessionCatalogItemToBytes( + item: SessionCatalogTranscriptItem, + maxBytes: number, +): SessionCatalogTranscriptItem | undefined { + if (Buffer.byteLength(JSON.stringify(item), "utf8") <= maxBytes) { + return item; + } + const text = item.text; + if (typeof text !== "string") { + return undefined; + } + const candidate = (length: number): SessionCatalogTranscriptItem => { + const safeLength = + length > 0 && /[\uD800-\uDBFF]/u.test(text.charAt(length - 1)) ? length - 1 : length; + return { ...item, text: `${text.slice(0, safeLength)}…`, truncated: true }; + }; + let low = 0; + let high = text.length; + while (low < high) { + const middle = Math.ceil((low + high) / 2); + if (Buffer.byteLength(JSON.stringify(candidate(middle)), "utf8") <= maxBytes) { + low = middle; + } else { + high = middle - 1; + } + } + const bounded = candidate(low); + return Buffer.byteLength(JSON.stringify(bounded), "utf8") <= maxBytes ? bounded : undefined; +} + +function importableSessionCatalogItem( + item: SessionCatalogTranscriptItem, +): SessionCatalogTranscriptItem { + const { raw: _raw, ...importable } = item; + return importable; +} + +async function readBoundedSessionCatalogHistory(params: { + read: (params: { cursor?: string; limit: number }) => Promise; +}): Promise { + const pages: SessionCatalogTranscriptItem[][] = []; + let cursor: string | undefined; + let itemCount = 0; + let bytes = 0; + while (itemCount < SESSION_CATALOG_HISTORY_IMPORT_MAX_ITEMS) { + const page = await params.read({ + limit: Math.min( + SESSION_CATALOG_HISTORY_IMPORT_PAGE_LIMIT, + SESSION_CATALOG_HISTORY_IMPORT_MAX_ITEMS - itemCount, + ), + ...(cursor ? { cursor } : {}), + }); + const retained: SessionCatalogTranscriptItem[] = []; + // Catalog pages move newest-to-oldest while each page stays chronological. + // Walk backward for recent-window bounds, then prepend older retained pages. + for (let index = page.items.length - 1; index >= 0; index -= 1) { + const item = page.items[index]; + if (!item) { + continue; + } + const importableItem = importableSessionCatalogItem(item); + const itemBytes = Buffer.byteLength(JSON.stringify(importableItem), "utf8"); + const remainingBytes = SESSION_CATALOG_HISTORY_IMPORT_MAX_BYTES - bytes; + if (itemCount > 0 && itemBytes > remainingBytes) { + return [retained, ...pages.toReversed()].flat(); + } + const retainedItem = + itemBytes <= remainingBytes + ? importableItem + : fitSessionCatalogItemToBytes(importableItem, remainingBytes); + if (!retainedItem) { + continue; + } + const retainedItemBytes = Buffer.byteLength(JSON.stringify(retainedItem), "utf8"); + retained.unshift(retainedItem); + itemCount += 1; + bytes += retainedItemBytes; + if ( + itemCount === SESSION_CATALOG_HISTORY_IMPORT_MAX_ITEMS || + bytes === SESSION_CATALOG_HISTORY_IMPORT_MAX_BYTES + ) { + return [retained, ...pages.toReversed()].flat(); + } + } + pages.push(retained); + if (!page.nextCursor || page.nextCursor === cursor) { + break; + } + cursor = page.nextCursor; + } + return pages.toReversed().flat(); +} + +export async function importSessionCatalogHistory(params: { + catalogId: string; + threadId: string; + read: (params: { cursor?: string; limit: number }) => Promise; + sessionId: string; + sessionKey: string; + agentId: string; + cwd?: string; + config: OpenClawConfig; +}): Promise { + const items = await readBoundedSessionCatalogHistory({ read: params.read }); + const fallbackTimestamp = Date.now(); + await withSessionTranscriptWriteLock(params, async (transcript) => { + for (const [index, item] of items.entries()) { + const imported = importedSessionCatalogMessage({ + catalogId: params.catalogId, + item, + fallbackTimestamp: fallbackTimestamp + index, + }); + if (!imported) { + continue; + } + const message = { + ...(imported as unknown as Record), + idempotencyKey: `${params.catalogId}-catalog:${params.threadId}:${item.id ?? index}`, + } as unknown as AgentMessage; + await transcript.appendMessage({ + message, + idempotencyLookup: "scan", + cwd: params.cwd, + }); + } + }); +} diff --git a/src/plugins/session-catalog.test.ts b/src/plugins/session-catalog.test.ts new file mode 100644 index 000000000000..ef784f90a15c --- /dev/null +++ b/src/plugins/session-catalog.test.ts @@ -0,0 +1,221 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { importSessionCatalogHistory } from "./session-catalog-history-import.js"; + +const transcript = vi.hoisted(() => ({ + messages: [] as Array>, + lockCalls: 0, +})); + +vi.mock("../plugin-sdk/session-transcript-runtime.js", () => ({ + withSessionTranscriptWriteLock: async ( + _params: unknown, + run: (context: { + appendMessage: (params: { + message: Record; + idempotencyLookup?: string; + }) => Promise; + }) => Promise, + ) => { + transcript.lockCalls += 1; + await run({ + appendMessage: async ({ message, idempotencyLookup }) => { + const key = message.idempotencyKey; + if ( + idempotencyLookup === "scan" && + typeof key === "string" && + transcript.messages.some((candidate) => candidate.idempotencyKey === key) + ) { + return; + } + transcript.messages.push(message); + }, + }); + }, +})); + +type CatalogItem = Parameters[0]["read"]>[0]; +type TranscriptItem = Awaited< + ReturnType[0]["read"]> +>["items"][number]; + +function catalogReader(items: TranscriptItem[], maxPageSize = Number.POSITIVE_INFINITY) { + return vi.fn(async ({ cursor, limit }: CatalogItem) => { + const offset = cursor ? Number(cursor) : 0; + const pageLimit = Math.min(limit, maxPageSize); + const end = Math.max(0, items.length - offset); + const start = Math.max(0, end - pageLimit); + const page = items.slice(start, end); + const consumed = offset + page.length; + return { + hostId: "gateway", + threadId: "thread-1", + items: page, + ...(consumed < items.length ? { nextCursor: String(consumed) } : {}), + }; + }); +} + +function importHistory( + items: TranscriptItem[], + options: { maxPageSize?: number; read?: ReturnType } = {}, +) { + const read = options.read ?? catalogReader(items, options.maxPageSize); + return { + read, + result: importSessionCatalogHistory({ + catalogId: "pi", + threadId: "thread-1", + read, + sessionId: "session-1", + sessionKey: "agent:main:catalog-adopt", + agentId: "main", + config: {} as OpenClawConfig, + }), + }; +} + +function messageText(message: Record): string | undefined { + if (typeof message.content === "string") { + return message.content; + } + const content = Array.isArray(message.content) ? message.content[0] : undefined; + return content && typeof content === "object" && "text" in content + ? String(content.text) + : undefined; +} + +describe("importSessionCatalogHistory", () => { + beforeEach(() => { + transcript.messages.length = 0; + transcript.lockCalls = 0; + }); + + it("imports backward pages in chronological order with native provenance", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-25T12:00:00.000Z")); + try { + const { result } = importHistory( + [ + { id: "u-1", type: "userMessage", text: "repeat me", timestamp: "not-a-date" }, + { id: "r-empty", type: "reasoning" }, + { + id: "r-1", + type: "reasoning", + text: "careful", + timestamp: "1969-12-31T23:59:59.000Z", + }, + { id: "a-1", type: "agentMessage", text: "answer", model: "anthropic/claude" }, + { id: "t-1", type: "toolCall", text: "bash" }, + { id: "o-1", type: "other", text: "checkpoint" }, + ], + { maxPageSize: 2 }, + ); + await result; + } finally { + vi.useRealTimers(); + } + + expect(transcript.messages.map(messageText)).toEqual([ + "repeat me", + "Thinking\n\ncareful", + "answer", + "Tool call\n\nbash", + "Other\n\ncheckpoint", + ]); + expect(transcript.messages[0]?.["__openclaw"]).toEqual({ + mirrorOrigin: "pi-catalog-import", + }); + expect(transcript.messages[1]?.timestamp).toBe(-1_000); + expect(transcript.messages[2]?.model).toBe("anthropic/claude"); + expect(transcript.messages.map((message) => message.idempotencyKey)).toEqual([ + "pi-catalog:thread-1:u-1", + "pi-catalog:thread-1:r-1", + "pi-catalog:thread-1:a-1", + "pi-catalog:thread-1:t-1", + "pi-catalog:thread-1:o-1", + ]); + }); + + it("deduplicates a recovered import by scanning item idempotency keys", async () => { + const items: TranscriptItem[] = [ + { id: "u-1", type: "userMessage", text: "hello" }, + { id: "a-1", type: "agentMessage", text: "hi" }, + ]; + + await importHistory(items).result; + await importHistory(items).result; + + expect(transcript.lockCalls).toBe(2); + expect(transcript.messages).toHaveLength(2); + }); + + it("keeps only the most recent 200 items and returns them oldest-first", async () => { + const items: TranscriptItem[] = Array.from({ length: 205 }, (_, index) => ({ + id: `item-${String(index)}`, + type: "agentMessage", + text: `message-${String(index)}`, + })); + + const { read, result } = importHistory(items); + await result; + + expect(read).toHaveBeenCalledTimes(2); + expect(transcript.messages).toHaveLength(200); + expect(messageText(transcript.messages[0]!)).toBe("message-5"); + expect(messageText(transcript.messages.at(-1)!)).toBe("message-204"); + }); + + it("keeps the recent suffix within the 512 KiB serialized-item budget", async () => { + const items: TranscriptItem[] = Array.from({ length: 10 }, (_, index) => ({ + id: `item-${String(index)}`, + type: "agentMessage", + text: `${String(index)}${"x".repeat(100 * 1024)}`, + })); + + await importHistory(items, { maxPageSize: 2 }).result; + + expect(transcript.messages).toHaveLength(5); + expect(messageText(transcript.messages[0]!)?.startsWith("5")).toBe(true); + expect(messageText(transcript.messages.at(-1)!)?.startsWith("9")).toBe(true); + }); + + it("truncates a newest item that alone exceeds the byte budget", async () => { + await importHistory([{ id: "oversized", type: "toolResult", text: "x".repeat(600 * 1024) }]) + .result; + + const text = messageText(transcript.messages[0]!); + expect(text).toMatch(/^Tool result\n\n/u); + expect(text).toMatch(/…$/u); + expect(Buffer.byteLength(text ?? "", "utf8")).toBeLessThan(512 * 1024); + }); + + it("does not let an unused raw payload hide visible history", async () => { + await importHistory([ + { + id: "raw-heavy", + type: "agentMessage", + text: "visible answer", + raw: "x".repeat(600 * 1024), + }, + ]).result; + + expect(transcript.messages.map(messageText)).toEqual(["visible answer"]); + }); + + it("does not open the transcript write lock when a paged read fails", async () => { + const read = vi + .fn[0]["read"]>() + .mockResolvedValueOnce({ + hostId: "gateway", + threadId: "thread-1", + items: [{ id: "a-1", type: "agentMessage", text: "latest" }], + nextCursor: "older", + }) + .mockRejectedValueOnce(new Error("catalog read failed")); + + await expect(importHistory([], { read }).result).rejects.toThrow("catalog read failed"); + expect(transcript.lockCalls).toBe(0); + expect(transcript.messages).toEqual([]); + }); +}); diff --git a/src/sessions/session-upstream-monitor.test.ts b/src/sessions/session-upstream-monitor.test.ts index aa683ae3909d..3005a859956d 100644 --- a/src/sessions/session-upstream-monitor.test.ts +++ b/src/sessions/session-upstream-monitor.test.ts @@ -4,6 +4,7 @@ import { appendTranscriptMessage, upsertSessionEntry, } from "../config/sessions/session-accessor.js"; +import { importSessionCatalogHistory } from "../plugins/session-catalog-history-import.js"; import type { SessionCatalogProvider, SessionUpstreamProbe } from "../plugins/session-catalog.js"; import { closeOpenClawStateDatabaseForTest, @@ -647,6 +648,51 @@ describe("session upstream monitor", () => { expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([]); }); + it("reports a matching external prompt after catalog history import", async () => { + const database = createDatabaseOptions(); + const sessionKey = "agent:main:adopted:catalog-import"; + const sessionId = "session-catalog-import"; + await upsertSessionEntry( + { agentId: "main", sessionKey, env: database.env }, + { sessionId, updatedAt: 1 }, + ); + await importSessionCatalogHistory({ + catalogId: "pi", + threadId: "thread-pi", + read: async () => ({ + hostId: "gateway", + threadId: "thread-pi", + items: [{ id: "native-user-1", type: "userMessage", text: "repeat me" }], + }), + sessionId, + sessionKey, + agentId: "main", + config: {}, + }); + createLink(sessionKey, "pi", database); + const check = vi.fn(async (probes: SessionUpstreamProbe[]) => [ + { + kind: "activity" as const, + sessionKey, + occurredAt: 20_000, + humanTurns: probes[0]?.ownRecentUserTexts.includes("repeat me") ? 0 : 1, + nextMarker: { offset: 20 }, + dedupeId: "20", + }, + ]); + + await runSessionUpstreamMonitorTick({ + ...database, + providers: [provider("pi", check)], + isRunActive: () => false, + }); + + expect(check).toHaveBeenCalledWith([expect.objectContaining({ ownRecentUserTexts: [] })]); + expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([ + expect.objectContaining({ kind: "human_direct_message", summary: "human message via pi" }), + ]); + }); + it("records an external prompt five seconds after OpenClaw activity", async () => { const database = createDatabaseOptions(); const sessionKey = "agent:main:adopted:recent-external";