From 9dfb2a131818aa308def4fff7e10e5ecb9c48d5f Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 26 Jul 2026 02:09:48 -0400 Subject: [PATCH] fix: show native history in adopted Pi and OpenCode sessions (#114040) * fix(plugins): import adopted catalog history into the OpenClaw transcript Import native Pi and OpenCode history when a catalog session is adopted so the OpenClaw transcript reflects the full conversation immediately. Mark imported user rows with mirrorOrigin because they are transcript mirrors, not new external turns. This preserves watcher deduplication and prevents adopted history from being reported as fresh upstream input. * refactor(plugins): split catalog history importer Keep the session catalog contract module leaf-like by moving transcript mutation into a dedicated runtime module. This breaks the registry-to-transcript import cycle while preserving the Plugin SDK entrypoint and importer behavior. --- docs/plugins/reference/acpx.md | 10 +- docs/plugins/reference/opencode.md | 6 +- .../pi-session-catalog-continuation.test.ts | 84 +++++++ .../acpx/src/pi-session-catalog-plugin.ts | 25 +- .../src/pi-session-catalog.test-support.ts | 15 +- extensions/opencode/session-catalog-plugin.ts | 25 +- extensions/opencode/session-catalog.test.ts | 66 +++++- src/plugin-sdk/session-catalog.ts | 1 + src/plugins/session-catalog-history-import.ts | 190 +++++++++++++++ src/plugins/session-catalog.test.ts | 221 ++++++++++++++++++ src/sessions/session-upstream-monitor.test.ts | 46 ++++ 11 files changed, 671 insertions(+), 18 deletions(-) create mode 100644 src/plugins/session-catalog-history-import.ts create mode 100644 src/plugins/session-catalog.test.ts 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";