diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index 6fd7e70d00b..eeb65923e27 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -10,7 +10,6 @@ extensions/amazon-bedrock/stream.runtime.ts extensions/anthropic/index.test.ts extensions/anthropic/register.runtime.ts extensions/anthropic/session-catalog.test.ts -extensions/anthropic/session-catalog.ts extensions/browser/src/browser-tool.test.ts extensions/browser/src/browser-tool.ts extensions/browser/src/browser/cdp.ts diff --git a/extensions/anthropic/session-catalog-continue.ts b/extensions/anthropic/session-catalog-continue.ts new file mode 100644 index 00000000000..400640c95ec --- /dev/null +++ b/extensions/anthropic/session-catalog-continue.ts @@ -0,0 +1,172 @@ +import fs from "node:fs/promises"; +import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; +import { createSessionCatalogAdoptionCoordinator } from "openclaw/plugin-sdk/session-catalog"; +import { CLAUDE_CLI_BACKEND_ID, CLAUDE_CLI_DEFAULT_MODEL_REF } from "./cli-constants.js"; +import { + adoptedSessionKey, + adoptedSourceKey, + CLAUDE_LOCAL_SESSION_HOST_ID, +} from "./session-catalog-adoption.js"; +import { type CatalogRecord, listClaudeSessions } from "./session-catalog-discovery.js"; +import { importClaudeHistory } from "./session-catalog-history.js"; +import { + readBoundedClaudeHistory, + readClaudeSessionTranscript, + resolveNodeClaudeRecord, +} from "./session-catalog-listing.js"; +import { + currentClaudeSessionCatalogConfig, + listBoundClaudeSessions, + resolveClaudeCliRoutedModelId, +} from "./session-catalog-runtime.js"; +import { currentHomeDir, gatewayClaudeScanOptions } from "./session-catalog-scan.js"; +import { + CLAUDE_CLI_NODE_RUN_COMMAND, + CLAUDE_SESSION_READ_COMMAND, + CLAUDE_SESSIONS_LIST_COMMAND, + ClaudeCatalogParamsError, + isResumableClaudeSource, +} from "./session-catalog-shared.js"; +import type { ClaudeTranscriptItem } from "./session-catalog-transcript.js"; +import type { ClaudeSessionCatalogSession } from "./session-catalog-types.js"; +import * as upstream from "./session-upstream-activity.js"; + +const continueClaudeAdoption = + createSessionCatalogAdoptionCoordinator>>(); + +export async function continueClaudeSession( + api: OpenClawPluginApi, + agentId: string, + hostId: string, + threadId: string, + allowProcessHomeFallback?: boolean, +): Promise<{ sessionKey: string }> { + const scanOptions = gatewayClaudeScanOptions(allowProcessHomeFallback); + const sourceKey = adoptedSourceKey(hostId, threadId); + const operationKey = `${agentId}\0${sourceKey}`; + const linkSession = async (sessionKey: string, history?: ClaudeTranscriptItem[]) => + await upstream.linkContinued({ + sessionKey, + hostId, + threadId, + ...(history ? { history } : {}), + listLocalSessions: () => listClaudeSessions(currentHomeDir(), scanOptions), + readRemote: async () => + ( + await readClaudeSessionTranscript({ + runtime: api.runtime, + hostId, + threadId, + limit: 1, + allowProcessHomeFallback, + }) + ).items, + }); + const existing = listBoundClaudeSessions(api, agentId).get(sourceKey); + if (existing) { + return await linkSession(existing); + } + let history: ClaudeTranscriptItem[] | undefined; + return await continueClaudeAdoption({ + sourceKey: operationKey, + findExisting: () => listBoundClaudeSessions(api, agentId).get(sourceKey), + create: async () => { + let nodeId: string | undefined; + let record: ClaudeSessionCatalogSession | undefined; + if (hostId === CLAUDE_LOCAL_SESSION_HOST_ID) { + record = (await listClaudeSessions(currentHomeDir(), scanOptions)).find( + (candidate) => candidate.threadId === threadId, + ); + if (!record || !isResumableClaudeSource(record.source)) { + throw new ClaudeCatalogParamsError("only local Claude Code sessions can be continued"); + } + } else if (hostId.startsWith("node:")) { + nodeId = hostId.slice("node:".length); + const node = (await api.runtime.nodes.list()).nodes.find( + (candidate) => + candidate.nodeId === nodeId && + candidate.connected === true && + candidate.commands?.includes(CLAUDE_SESSIONS_LIST_COMMAND) && + candidate.commands.includes(CLAUDE_SESSION_READ_COMMAND) && + candidate.commands.includes(CLAUDE_CLI_NODE_RUN_COMMAND) && + candidate.invocableCommands?.includes(CLAUDE_SESSIONS_LIST_COMMAND) === true && + candidate.invocableCommands.includes(CLAUDE_SESSION_READ_COMMAND) && + candidate.invocableCommands.includes(CLAUDE_CLI_NODE_RUN_COMMAND), + ); + if (!node) { + throw new ClaudeCatalogParamsError( + "paired node does not permit Claude CLI session continuation", + ); + } + // Node rows stay CLI-only: desktop transcripts on nodes have no + // node-side run command and remain view-only. + record = await resolveNodeClaudeRecord({ runtime: api.runtime, nodeId, threadId }); + if (!record || record.source !== "claude-cli") { + throw new ClaudeCatalogParamsError("only Claude CLI sessions can be continued"); + } + } else { + throw new ClaudeCatalogParamsError("hostId is invalid"); + } + if (hostId === CLAUDE_LOCAL_SESSION_HOST_ID) { + const source = await fs.stat((record as CatalogRecord).filePath).catch(() => undefined); + if (!source?.isFile()) { + throw new ClaudeCatalogParamsError("Claude session transcript is unavailable"); + } + } + // Narrowed local: afterCreate below needs a definite array, while the outer + // `history` stays optional so complete() can link without a reread on races. + const loadedHistory = await readBoundedClaudeHistory({ + runtime: api.runtime, + hostId, + threadId, + allowProcessHomeFallback, + }); + history = loadedHistory; + const config = currentClaudeSessionCatalogConfig(api); + const adoptingAgentId = agentId; + // Adopt onto the model this agent actually routes to the CLI backend; the + // packaged default may not be routed or allowed in an existing config. + const model = + resolveClaudeCliRoutedModelId(config, adoptingAgentId) ?? + CLAUDE_CLI_DEFAULT_MODEL_REF.slice(`${CLAUDE_CLI_BACKEND_ID}/`.length); + const marker = { + sourceThreadId: threadId, + ...(hostId !== CLAUDE_LOCAL_SESSION_HOST_ID ? { sourceHostId: hostId } : {}), + }; + const created = await api.runtime.agent.session.createSessionEntry({ + cfg: config, + key: adoptedSessionKey(hostId, threadId), + agentId: adoptingAgentId, + recoverMatchingInitialEntry: true, + ...(record.name ? { label: record.name } : {}), + ...(record.cwd ? { spawnedCwd: record.cwd } : {}), + ...(nodeId ? { execNode: nodeId, ...(record.cwd ? { execCwd: record.cwd } : {}) } : {}), + initialEntry: { + cliBackendId: CLAUDE_CLI_BACKEND_ID, + model, + modelSelectionLocked: true, + pluginOwnerId: api.id, + cliSessionBinding: { sessionId: threadId, forceReuse: true, forkNextResume: true }, + pluginExtensions: { anthropic: { sessionCatalog: marker } }, + }, + afterCreate: async (entry) => { + await importClaudeHistory({ + items: loadedHistory, + threadId, + sessionId: entry.sessionId, + sessionKey: entry.key, + agentId: entry.agentId, + storePath: api.runtime.agent.session.resolveStorePath(config.session?.store, { + agentId: entry.agentId, + }), + ...(record.cwd ? { cwd: record.cwd } : {}), + config, + }); + return { pluginExtensions: { anthropic: { sessionCatalog: marker } } }; + }, + }); + return { sessionKey: created.key }; + }, + complete: async (continued) => await linkSession(continued.sessionKey, history), + }); +} diff --git a/extensions/anthropic/session-catalog-discovery.ts b/extensions/anthropic/session-catalog-discovery.ts new file mode 100644 index 00000000000..bc206f17f3e --- /dev/null +++ b/extensions/anthropic/session-catalog-discovery.ts @@ -0,0 +1,728 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { parseDateFirstTimestampMs } from "openclaw/plugin-sdk/number-runtime"; +import type { SessionCatalogPullRequestSummary } from "openclaw/plugin-sdk/session-catalog"; +import { + asPositiveSafeInteger as pullRequestNumber, + isRecord, + normalizeBoundedOptionalString as readBoundedString, +} from "openclaw/plugin-sdk/string-coerce-runtime"; +import { readClaudeDesktopCustomGroups } from "./claude-desktop-groups.js"; +import { + CLAUDE_CATALOG_IO_CONCURRENCY, + childDirectories, + currentHomeDir, + desktopSessionStoreAvailable, + desktopSessionsDir, + type ClaudeProjectsTreeSnapshot, + type ClaudeSessionScanContext, + mapConcurrent, + projectsDir, + readJsonFile, + readProjectsTreeSnapshot, + safeSessionFileForScan, + setBoundedCache, +} from "./session-catalog-scan.js"; +import { collectTranscriptText } from "./session-catalog-transcript.js"; +import type { ClaudeSessionCatalogSession } from "./session-catalog-types.js"; + +export const MAX_STRING_LENGTH = 4096; +const MAX_SESSION_PULL_REQUESTS = 20; +const MAX_CATALOG_DISCOVERY_FILES = 10_000; +const MAX_CATALOG_DISCOVERY_CACHE_ENTRIES = 20_000; +const MAX_CLAUDE_SESSION_SCAN_CACHE_ENTRIES = 8; +const CLAUDE_SESSION_SCAN_HARD_TTL_MS = 5 * 60_000; +const CLAUDE_PARTIAL_SCAN_TTL_MS = 15_000; +const CLAUDE_DESKTOP_SCAN_TTL_MS = 60_000; +const CLAUDE_METADATA_PREFIX_BYTES = 1024 * 1024; +const CLAUDE_METADATA_READ_CHUNK_BYTES = 16 * 1024; +const MAX_CATALOG_METADATA_SCAN_BYTES = 64 * 1024 * 1024; +const CLI_ENTRYPOINTS = new Set(["cli", "sdk-cli"]); + +type CatalogDiscoveryCacheEntry = { + // The module-global cache is keyed by canonical transcript path, so an entry must also record the + // discovery context it was built in. `root` is the logical (unresolved) projects root: it scopes + // the entry to its homeDir even when the root itself is a symlink, so a different homeDir scan + // cannot reuse it and eviction can find it without re-resolving a now-missing root. mtime+size+ino + // detect any content change or atomic replacement; sessionId guards against a canonical path being + // reached under a different filename-derived id (e.g. an aliased/renamed symlink). + root: string; + mtimeMs: number; + size: number; + ino: number; + sessionId: string; + // Bytes this file charged against the scan budget when first scanned. Cache hits re-charge it so + // byte-budget-limited discovery stops at the same frontier whether or not the cache is warm, + // keeping pagination deterministic across repeated identical calls. + scannedBytes: number; + record: CatalogRecord | null; + sidechain: boolean; +}; + +type ClaudeSessionScanCacheEntry = { + treeStamp: string; + hardExpiresAt: number; + desktopStoreAvailable: boolean; + desktopExpiresAt: number; + records: Promise; +}; + +// Transcript discoveries stay valid only for the same root/id/inode/mtime/size and are LRU-bounded; +// a false hit would corrupt pagination, so warm scans re-charge the original deterministic byte cost. +const catalogDiscoveryCache = new Map(); +// Whole scans are root-scoped and bounded; tree/Desktop/hard expiries below own invalidation, avoiding +// an unbounded home map while preserving the exact resolved records promise for concurrent callers. +const claudeSessionScanCache = new Map(); + +function cacheCatalogDiscovery(filePath: string, entry: CatalogDiscoveryCacheEntry): void { + setBoundedCache(catalogDiscoveryCache, filePath, entry, MAX_CATALOG_DISCOVERY_CACHE_ENTRIES); +} + +type SessionIndexEntry = { + sessionId?: unknown; + fullPath?: unknown; + fileMtime?: unknown; + firstPrompt?: unknown; + summary?: unknown; + messageCount?: unknown; + created?: unknown; + modified?: unknown; + gitBranch?: unknown; + projectPath?: unknown; + isSidechain?: unknown; +}; + +type DesktopSessionMetadata = { + sessionId?: unknown; + cliSessionId?: unknown; + cwd?: unknown; + originCwd?: unknown; + createdAt?: unknown; + lastActivityAt?: unknown; + model?: unknown; + isArchived?: unknown; + title?: unknown; + customGroup?: unknown; + prNumber?: unknown; + prState?: unknown; + prs?: unknown; +}; + +type DesktopPullRequestMetadata = { + prNumber?: unknown; + state?: unknown; + dismissed?: unknown; +}; + +export type CatalogRecord = ClaudeSessionCatalogSession & { + filePath: string; +}; + +function pullRequestState(value: unknown): SessionCatalogPullRequestSummary["state"] | undefined { + if (typeof value !== "string") { + return undefined; + } + switch (value.trim().toLowerCase()) { + case "open": + case "draft": + case "merged": + case "closed": + return value.trim().toLowerCase() as SessionCatalogPullRequestSummary["state"]; + default: + return undefined; + } +} + +// Desktop retains historical PRs in order and marks hidden ones as dismissed; +// the top-level pair identifies the current PR whose state labels the row. +function desktopPullRequestSummary( + metadata: DesktopSessionMetadata, +): SessionCatalogPullRequestSummary | undefined { + const visibleByNumber = new Map(); + const dismissed = new Set(); + if (Array.isArray(metadata.prs)) { + for (const value of metadata.prs) { + if (!isRecord(value)) { + continue; + } + const entry = value as DesktopPullRequestMetadata; + const number = pullRequestNumber(entry.prNumber); + if (!number) { + continue; + } + if (entry.dismissed === true) { + dismissed.add(number); + visibleByNumber.delete(number); + continue; + } + if (!dismissed.has(number) && !visibleByNumber.has(number)) { + visibleByNumber.set(number, pullRequestState(entry.state)); + } + } + } + const currentNumber = pullRequestNumber(metadata.prNumber); + let currentState = currentNumber ? visibleByNumber.get(currentNumber) : undefined; + if (currentNumber && !dismissed.has(currentNumber)) { + currentState = pullRequestState(metadata.prState) ?? currentState; + // Reinsert the current PR at the tail so truncation always retains it. + visibleByNumber.delete(currentNumber); + visibleByNumber.set(currentNumber, currentState); + } + const visible = [...visibleByNumber].map(([number, state]) => ({ number, state })); + if (visible.length === 0) { + return undefined; + } + const state = currentState ?? visible.at(-1)?.state; + if (!state) { + return undefined; + } + return { + numbers: visible.slice(-MAX_SESSION_PULL_REQUESTS).map((entry) => entry.number), + state, + }; +} + +export function parsePullRequestSummary( + value: unknown, +): SessionCatalogPullRequestSummary | undefined { + if (value === undefined) { + return undefined; + } + if (!isRecord(value) || !Array.isArray(value.numbers)) { + throw new Error("Claude node returned an invalid pull request summary"); + } + const numbers = value.numbers.map(pullRequestNumber); + const state = pullRequestState(value.state); + if ( + numbers.length === 0 || + numbers.length > MAX_SESSION_PULL_REQUESTS || + numbers.some((number) => number === undefined) || + new Set(numbers).size !== numbers.length || + !state + ) { + throw new Error("Claude node returned an invalid pull request summary"); + } + return { numbers: numbers as number[], state }; +} + +function isCliEntrypoint(value: unknown): value is string { + return typeof value === "string" && CLI_ENTRYPOINTS.has(value); +} + +// Claude's persisted string timestamps are date expressions, including numeric-looking years. +// Numeric fields are already millisecond values, so preserve that distinct mixed-input contract. +function parseClaudeCatalogTimestampMs(value: unknown): number | undefined { + return parseDateFirstTimestampMs(value); +} + +async function readDesktopMetadata(homeDir: string): Promise<{ + active: Map; + archived: Set; +}> { + const active = new Map(); + const archived = new Set(); + const customGroups = await readClaudeDesktopCustomGroups(homeDir); + for (const accountDir of await childDirectories(desktopSessionsDir(homeDir))) { + for (const workspaceDir of await childDirectories(accountDir)) { + let entries: string[]; + try { + entries = await fs.readdir(workspaceDir); + } catch { + continue; + } + for (const name of entries) { + if (!name.startsWith("local_") || !name.endsWith(".json")) { + continue; + } + const raw = await readJsonFile(path.join(workspaceDir, name)); + if (!isRecord(raw)) { + continue; + } + const metadata = raw as DesktopSessionMetadata; + const cliSessionId = readBoundedString(metadata.cliSessionId, 256); + if (!cliSessionId) { + continue; + } + if (metadata.isArchived === true) { + archived.add(cliSessionId); + active.delete(cliSessionId); + continue; + } + if (!archived.has(cliSessionId)) { + const localSessionId = readBoundedString(metadata.sessionId, 256); + const customGroup = localSessionId ? customGroups.get(localSessionId) : undefined; + active.set(cliSessionId, customGroup ? { ...metadata, customGroup } : metadata); + } + } + } + } + return { active, archived }; +} + +async function readIndexRecords(context: ClaudeSessionScanContext): Promise<{ + records: Map; + sidechainIds: Set; +}> { + const records = new Map(); + const sidechainIds = new Set(); + if (!context.resolvedRoot) { + return { records, sidechainIds }; + } + const indexes = await mapConcurrent( + context.projectDirectories, + CLAUDE_CATALOG_IO_CONCURRENCY, + async ({ directory, childNames }) => ({ + directory, + raw: childNames.includes("sessions-index.json") + ? await readJsonFile(path.join(directory, "sessions-index.json"), { + onIoFailure: () => { + context.complete = false; + }, + }) + : undefined, + }), + ); + const candidates: Array<{ + directory: string; + entry: SessionIndexEntry; + sessionId: string; + }> = []; + for (const { directory, raw } of indexes) { + if (!isRecord(raw) || !Array.isArray(raw.entries)) { + continue; + } + for (const candidate of raw.entries) { + if (!isRecord(candidate)) { + continue; + } + const entry = candidate as SessionIndexEntry; + const sessionId = readBoundedString(entry.sessionId, 256); + if (!sessionId) { + continue; + } + candidates.push({ directory, entry, sessionId }); + } + } + const safeFiles = await mapConcurrent( + candidates, + CLAUDE_CATALOG_IO_CONCURRENCY, + async ({ directory, entry, sessionId }) => { + if (entry.isSidechain === true) { + return undefined; + } + const indexedPath = readBoundedString(entry.fullPath, MAX_STRING_LENGTH); + return await safeSessionFileForScan( + context, + indexedPath ?? path.join(directory, `${sessionId}.jsonl`), + sessionId, + ); + }, + ); + for (const [index, candidate] of candidates.entries()) { + const { entry, sessionId } = candidate; + if (entry.isSidechain === true) { + sidechainIds.add(sessionId); + records.delete(sessionId); + continue; + } + const safeFile = safeFiles[index]; + if (!safeFile) { + continue; + } + const createdAt = parseClaudeCatalogTimestampMs(entry.created); + const updatedAt = + parseClaudeCatalogTimestampMs(entry.modified) ?? + parseClaudeCatalogTimestampMs(entry.fileMtime); + const summary = readBoundedString(entry.summary, 500); + const firstPrompt = readBoundedString(entry.firstPrompt, 500); + records.set(sessionId, { + threadId: sessionId, + name: summary ?? firstPrompt ?? null, + cwd: readBoundedString(entry.projectPath, MAX_STRING_LENGTH), + status: "stored", + ...(createdAt !== undefined ? { createdAt } : {}), + ...(updatedAt !== undefined ? { updatedAt, recencyAt: updatedAt } : {}), + source: "claude-cli", + modelProvider: "anthropic", + ...(readBoundedString(entry.gitBranch, 500) + ? { gitBranch: readBoundedString(entry.gitBranch, 500) } + : {}), + archived: false, + filePath: safeFile.filePath, + }); + } + return { records, sidechainIds }; +} + +async function locateSessionFile( + context: ClaudeSessionScanContext, + sessionId: string, +): Promise { + const fileName = `${sessionId}.jsonl`; + for (const { directory, childNames } of context.projectDirectories) { + if (!childNames.includes(fileName)) { + continue; + } + const candidate = path.join(directory, fileName); + const safeFile = await safeSessionFileForScan(context, candidate, sessionId); + if (safeFile) { + return safeFile.filePath; + } + } + return undefined; +} + +async function discoverCliRecords( + context: ClaudeSessionScanContext, + records: Map, + sidechainIds: Set, +): Promise { + const { root } = context; + if (!context.resolvedRoot) { + // The root (or a parent) is gone. Entries are tagged with the logical root, so evict by that + // rather than a lexical containment test the canonical cache keys would never satisfy. + for (const [cachedPath, entry] of catalogDiscoveryCache) { + if (entry.root === root) { + catalogDiscoveryCache.delete(cachedPath); + } + } + return; + } + let discoveredFiles = 0; + let scannedBytes = 0; + let truncated = false; + const seenFilePaths = new Set(); + const candidates: Array<{ directory: string; name: string; sessionId: string }> = []; + collect: for (const { directory, childNames } of context.projectDirectories) { + for (const name of childNames) { + if (!name.endsWith(".jsonl")) { + continue; + } + if (discoveredFiles >= MAX_CATALOG_DISCOVERY_FILES) { + truncated = true; + break collect; + } + discoveredFiles += 1; + const sessionId = name.slice(0, -".jsonl".length); + if (sessionId) { + candidates.push({ directory, name, sessionId }); + } + } + } + const safeFiles = await mapConcurrent( + candidates, + CLAUDE_CATALOG_IO_CONCURRENCY, + async ({ directory, name, sessionId }) => + records.has(sessionId) || sidechainIds.has(sessionId) + ? undefined + : await safeSessionFileForScan(context, path.join(directory, name), sessionId), + ); + for (const [candidateIndex, candidate] of candidates.entries()) { + const { sessionId } = candidate; + // Resolve metadata concurrently, but make every semantic decision in the original directory + // order. In particular, duplicates and the global byte frontier must match a serial cold scan. + if (records.has(sessionId) || sidechainIds.has(sessionId)) { + continue; + } + const safeFile = safeFiles[candidateIndex]; + if (!safeFile) { + continue; + } + const { filePath, stat: fileStat } = safeFile; + seenFilePaths.add(filePath); + const cached = catalogDiscoveryCache.get(filePath); + // Claude transcripts only append while active, then stay static, so mtime+size+ino identify + // the parsed content (ino also rejects an atomic replacement that reused the same mtime/size), + // and sessionId ensures the record is served only under the filename-derived id it was built + // for. These files are owner-owned and append-only; a mid-scan read-permission revocation is + // not a state the Claude CLI produces, so a hit intentionally skips the open() re-check. + if ( + cached && + cached.root === root && + cached.mtimeMs === fileStat.mtimeMs && + cached.size === fileStat.size && + cached.ino === fileStat.ino && + cached.sessionId === sessionId && + // Only replay the cached record if a cold scan would also reach its metadata under the + // current remaining byte budget. Once earlier files grow, replaying a record whose original + // scan cost now crosses the frontier would surface a record a cold scan stops before; fall + // through to a bounded rescan instead so warm and cold discovery (and pagination) match. + scannedBytes + cached.scannedBytes <= MAX_CATALOG_METADATA_SCAN_BYTES + ) { + if (cached.sidechain) { + sidechainIds.add(sessionId); + } + if (cached.record) { + records.set(sessionId, cached.record); + } + // Cache hits read no transcript bytes, but they still charge the file's original scan cost + // so the byte-budget cutoff matches a cold scan; otherwise repeated calls would free budget + // and progressively discover more files. + scannedBytes += cached.scannedBytes; + if (scannedBytes >= MAX_CATALOG_METADATA_SCAN_BYTES) { + truncated = true; + break; + } + continue; + } + const handle = await fs.open(filePath, "r").catch(() => { + context.complete = false; + return undefined; + }); + if (!handle) { + continue; + } + let cacheable = false; + let fileScannedBytes = 0; + try { + const stat = await handle.stat(); + let aiTitle: string | undefined; + let pending = Buffer.alloc(0); + let fileOffset = 0; + let stopFile = false; + const inspectLine = (line: Buffer): boolean => { + let raw: unknown; + try { + raw = JSON.parse(line.toString("utf8")) as unknown; + } catch { + return false; + } + if (!isRecord(raw) || raw.sessionId !== sessionId) { + return false; + } + if (raw.type === "ai-title") { + aiTitle = readBoundedString(raw.aiTitle, 500) ?? aiTitle; + return false; + } + if (typeof raw.entrypoint === "string" && !isCliEntrypoint(raw.entrypoint)) { + return true; + } + if (isCliEntrypoint(raw.entrypoint) && raw.isSidechain === true) { + sidechainIds.add(sessionId); + return true; + } + if ( + !isCliEntrypoint(raw.entrypoint) || + raw.type !== "user" || + !isRecord(raw.message) || + raw.message.role !== "user" + ) { + return false; + } + const fragments: string[] = []; + collectTranscriptText(raw.message.content, fragments); + const firstPrompt = readBoundedString(fragments[0], 500); + const createdAt = parseClaudeCatalogTimestampMs(raw.timestamp); + records.set(sessionId, { + threadId: sessionId, + name: aiTitle ?? firstPrompt ?? null, + cwd: readBoundedString(raw.cwd, MAX_STRING_LENGTH), + status: "stored", + ...(createdAt !== undefined ? { createdAt } : {}), + updatedAt: stat.mtimeMs, + recencyAt: stat.mtimeMs, + source: "claude-cli", + modelProvider: "anthropic", + ...(readBoundedString(raw.version, 256) + ? { cliVersion: readBoundedString(raw.version, 256) } + : {}), + ...(readBoundedString(raw.gitBranch, 500) + ? { gitBranch: readBoundedString(raw.gitBranch, 500) } + : {}), + archived: false, + filePath, + }); + return true; + }; + while ( + !stopFile && + fileOffset < stat.size && + fileOffset < CLAUDE_METADATA_PREFIX_BYTES && + scannedBytes < MAX_CATALOG_METADATA_SCAN_BYTES + ) { + const size = Math.min( + CLAUDE_METADATA_READ_CHUNK_BYTES, + stat.size - fileOffset, + CLAUDE_METADATA_PREFIX_BYTES - fileOffset, + MAX_CATALOG_METADATA_SCAN_BYTES - scannedBytes, + ); + const chunk = Buffer.allocUnsafe(size); + const { bytesRead } = await handle.read(chunk, 0, size, fileOffset); + if (bytesRead === 0) { + break; + } + fileOffset += bytesRead; + scannedBytes += bytesRead; + pending = pending.length + ? Buffer.concat([pending, chunk.subarray(0, bytesRead)]) + : chunk.subarray(0, bytesRead); + let newline: number; + while (!stopFile && (newline = pending.indexOf(0x0a)) >= 0) { + stopFile = inspectLine(pending.subarray(0, newline)); + pending = pending.subarray(newline + 1); + } + } + if (!stopFile && fileOffset >= stat.size && pending.length > 0) { + inspectLine(pending); + } + // A read whose chunk was capped by the remaining global budget stops on a smaller boundary + // than a cold scan would, so its fileOffset undercounts the true unconstrained scan cost. + // Don't cache such an entry: replaying its low cost later (with more budget free) would let + // the warm scan cross the frontier and surface sessions a cold scan omits. + const budgetConstrained = scannedBytes >= MAX_CATALOG_METADATA_SCAN_BYTES; + cacheable = + !budgetConstrained && + (stopFile || fileOffset >= stat.size || fileOffset >= CLAUDE_METADATA_PREFIX_BYTES); + fileScannedBytes = fileOffset; + } finally { + await handle.close(); + } + // Negative and sidechain-only results are cached too; unchanged files should not be reparsed. + if (cacheable) { + cacheCatalogDiscovery(filePath, { + root, + mtimeMs: fileStat.mtimeMs, + size: fileStat.size, + ino: fileStat.ino, + sessionId, + scannedBytes: fileScannedBytes, + record: records.get(sessionId) ?? null, + sidechain: sidechainIds.has(sessionId), + }); + } + if (scannedBytes >= MAX_CATALOG_METADATA_SCAN_BYTES) { + truncated = true; + break; + } + } + if (!truncated) { + // A complete scan is authoritative for this root: drop any of its entries not seen this pass. + for (const [cachedPath, entry] of catalogDiscoveryCache) { + if (entry.root === root && !seenFilePaths.has(cachedPath)) { + catalogDiscoveryCache.delete(cachedPath); + } + } + } +} + +async function scanClaudeSessions( + homeDir: string, + snapshot: ClaudeProjectsTreeSnapshot, + includeDesktop: boolean, +): Promise<{ records: CatalogRecord[]; complete: boolean }> { + const context: ClaudeSessionScanContext = { ...snapshot, complete: true, safeFiles: new Map() }; + const [indexed, desktop] = await Promise.all([ + readIndexRecords(context), + includeDesktop + ? readDesktopMetadata(homeDir) + : Promise.resolve({ active: new Map(), archived: new Set() }), + ]); + const records = indexed.records; + await discoverCliRecords(context, records, indexed.sidechainIds); + for (const sessionId of desktop.archived) { + records.delete(sessionId); + } + for (const [sessionId, metadata] of desktop.active) { + if (indexed.sidechainIds.has(sessionId)) { + continue; + } + const existing = records.get(sessionId); + const filePath = existing?.filePath ?? (await locateSessionFile(context, sessionId)); + if (!filePath) { + continue; + } + const createdAt = parseClaudeCatalogTimestampMs(metadata.createdAt) ?? existing?.createdAt; + const updatedAt = parseClaudeCatalogTimestampMs(metadata.lastActivityAt) ?? existing?.updatedAt; + const customGroup = readBoundedString(metadata.customGroup, 500); + const pullRequest = desktopPullRequestSummary(metadata); + records.set(sessionId, { + ...(existing ?? { + threadId: sessionId, + status: "stored" as const, + modelProvider: "anthropic" as const, + archived: false as const, + }), + name: readBoundedString(metadata.title, 500) ?? existing?.name ?? null, + cwd: + readBoundedString(metadata.cwd, MAX_STRING_LENGTH) ?? + readBoundedString(metadata.originCwd, MAX_STRING_LENGTH) ?? + existing?.cwd, + ...(createdAt !== undefined ? { createdAt } : {}), + ...(updatedAt !== undefined ? { updatedAt, recencyAt: updatedAt } : {}), + ...(customGroup ? { customGroup } : {}), + ...(pullRequest ? { pullRequest } : {}), + source: "claude-desktop", + filePath, + }); + } + return { + records: [...records.values()].toSorted((left, right) => { + const recency = + (right.recencyAt ?? right.updatedAt ?? 0) - (left.recencyAt ?? left.updatedAt ?? 0); + return recency || left.threadId.localeCompare(right.threadId); + }), + complete: context.complete, + }; +} + +export async function listClaudeSessions( + homeDir = currentHomeDir(), + options: { forceRefresh?: boolean; configDir?: string; includeDesktop?: boolean } = {}, +): Promise { + const root = projectsDir(homeDir, options.configDir); + const includeDesktop = options.includeDesktop !== false; + const cacheKey = `${root}\0${includeDesktop ? "desktop" : "cli"}`; + const [treeSnapshot, desktopStoreAvailable] = await Promise.all([ + readProjectsTreeSnapshot(root), + includeDesktop ? desktopSessionStoreAvailable(homeDir) : Promise.resolve(false), + ]); + const now = Date.now(); + const cached = claudeSessionScanCache.get(cacheKey); + // Child membership + file mtime/size signatures invalidate CLI rows on the next poll; five minutes + // backstops metadata anomalies. Desktop has a 60s bound when its macOS store exists; Linux skips it. + // Specific-thread force refresh bypasses both, or a stale page could hide a just-created session. + if ( + options.forceRefresh !== true && + cached && + cached.treeStamp === treeSnapshot.treeStamp && + cached.hardExpiresAt > now && + cached.desktopStoreAvailable === desktopStoreAvailable && + (!desktopStoreAvailable || cached.desktopExpiresAt > now) + ) { + setBoundedCache( + claudeSessionScanCache, + cacheKey, + cached, + MAX_CLAUDE_SESSION_SCAN_CACHE_ENTRIES, + ); + return await cached.records; + } + const scan = scanClaudeSessions(homeDir, treeSnapshot, includeDesktop); + let scanComplete = true; + const records = scan.then((result) => { + scanComplete = result.complete; + return result.records; + }); + const entry = { + treeStamp: treeSnapshot.treeStamp, + hardExpiresAt: now + CLAUDE_SESSION_SCAN_HARD_TTL_MS, + desktopStoreAvailable, + desktopExpiresAt: now + CLAUDE_DESKTOP_SCAN_TTL_MS, + records, + }; + setBoundedCache(claudeSessionScanCache, cacheKey, entry, MAX_CLAUDE_SESSION_SCAN_CACHE_ENTRIES); + try { + const result = await records; + if (!scanComplete && claudeSessionScanCache.get(cacheKey) === entry) { + // Partial results still serve this caller, but retry within 15s so transient per-file I/O + // cannot hide recovered sessions behind the five-minute unchanged-tree backstop. + entry.hardExpiresAt = Date.now() + CLAUDE_PARTIAL_SCAN_TTL_MS; + } + return result; + } catch (error) { + if (claudeSessionScanCache.get(cacheKey) === entry) { + claudeSessionScanCache.delete(cacheKey); + } + throw error; + } +} diff --git a/extensions/anthropic/session-catalog-listing.ts b/extensions/anthropic/session-catalog-listing.ts new file mode 100644 index 00000000000..a1361837ca4 --- /dev/null +++ b/extensions/anthropic/session-catalog-listing.ts @@ -0,0 +1,507 @@ +import fs from "node:fs/promises"; +import type { PluginRuntime } from "openclaw/plugin-sdk/plugin-runtime"; +import { withTimeout } from "openclaw/plugin-sdk/security-runtime"; +import type { SessionCatalogProvider } from "openclaw/plugin-sdk/session-catalog"; +import { + isRecord, + normalizeBoundedOptionalString as readBoundedString, +} from "openclaw/plugin-sdk/string-coerce-runtime"; +import { CLAUDE_LOCAL_SESSION_HOST_ID } from "./session-catalog-adoption.js"; +import { listClaudeSessions } from "./session-catalog-discovery.js"; +import { createNodeListFailedError, resolveNodeLabel } from "./session-catalog-node-helpers.js"; +import { + decodeOffset, + encodeOffset, + MAX_HOSTS, + MAX_PAGE_LIMIT, + MAX_TRANSCRIPT_LIMIT, + parseCatalogPage, + parseGatewayQuery, + readListParams, + readNodePageCursor, + readOptionalCursor, + readTranscriptParams, + unwrapNodePayload, +} from "./session-catalog-parsing.js"; +import { + configuredClaudeConfigDir, + currentHomeDir, + gatewayClaudeScanOptions, +} from "./session-catalog-scan.js"; +import { + CLAUDE_CLI_NODE_RUN_COMMAND, + CLAUDE_SESSION_READ_COMMAND, + CLAUDE_SESSIONS_LIST_COMMAND, + ClaudeCatalogParamsError, +} from "./session-catalog-shared.js"; +import * as catalogTerminal from "./session-catalog-terminal.js"; +import { parseTranscriptLine, type ClaudeTranscriptItem } from "./session-catalog-transcript.js"; +import type { + ClaudeSessionCatalogHost, + ClaudeSessionCatalogPage, + ClaudeSessionCatalogResult, + ClaudeSessionCatalogSession, + ClaudeSessionTranscriptPage, +} from "./session-catalog-types.js"; + +const TRANSCRIPT_READ_CHUNK_BYTES = 128 * 1024; +const MAX_TRANSCRIPT_SCAN_BYTES = 64 * 1024 * 1024; +const MAX_TRANSCRIPT_PAGE_BYTES = 20 * 1024 * 1024; +const NODE_INVOKE_TIMEOUT_MS = 30_000; +// Catalog refresh is fail-soft: one unhealthy machine must not hold the whole sidebar. +// The node invoke keeps running so cold native discovery can warm the next poll. +const NODE_CATALOG_LIST_RESPONSE_TIMEOUT_MS = 8_000; +const CLAUDE_HISTORY_IMPORT_MAX_ITEMS = 200; +const CLAUDE_HISTORY_IMPORT_MAX_BYTES = 512 * 1024; + +export async function listLocalClaudeSessionPage( + value: unknown, + homeDir?: string, + scanOptions?: { configDir?: string; includeDesktop?: boolean }, +): Promise { + const resolvedHome = homeDir ?? currentHomeDir(); + const resolvedScanOptions = + scanOptions ?? (homeDir === undefined ? gatewayClaudeScanOptions(true) : {}); + const params = readListParams(value); + const offset = decodeOffset(params.cursor, "catalog"); + const search = params.searchTerm?.toLocaleLowerCase(); + const records = (await listClaudeSessions(resolvedHome, resolvedScanOptions)).filter((record) => { + if (!search) { + return true; + } + return [record.name, record.cwd, record.gitBranch, record.threadId].some((candidate) => + candidate?.toLocaleLowerCase().includes(search), + ); + }); + const page = records + .slice(offset, offset + params.limit) + .map(({ filePath: _filePath, ...record }) => record); + const nextOffset = offset + page.length; + return { + sessions: page, + ...(nextOffset < records.length ? { nextCursor: encodeOffset(nextOffset) } : {}), + }; +} + +export async function readLocalClaudeTranscriptPage( + value: unknown, + homeDir?: string, + scanOptions?: { configDir?: string; includeDesktop?: boolean }, +): Promise> { + const resolvedHome = homeDir ?? currentHomeDir(); + const resolvedScanOptions = + scanOptions ?? (homeDir === undefined ? gatewayClaudeScanOptions(true) : {}); + const params = readTranscriptParams(value); + let filePath = (await listClaudeSessions(resolvedHome, resolvedScanOptions)).find( + (record) => record.threadId === params.threadId, + )?.filePath; + if (!filePath) { + // A just-created session can race the stamp snapshot. Specific reads must retry against disk so + // opening a new thread never fails only because the assembled catalog is still warm. + filePath = ( + await listClaudeSessions(resolvedHome, { ...resolvedScanOptions, forceRefresh: true }) + ).find((record) => record.threadId === params.threadId)?.filePath; + } + if (!filePath) { + throw new ClaudeCatalogParamsError("Claude session is unavailable"); + } + const handle = await fs.open(filePath, "r"); + try { + const stat = await handle.stat(); + const requestedEnd = params.cursor ? decodeOffset(params.cursor, "transcript") : stat.size; + if (requestedEnd > stat.size) { + throw new ClaudeCatalogParamsError("transcript cursor is invalid"); + } + let position = requestedEnd; + let scanned = 0; + let fragments: Buffer[] = []; + const found: Array<{ item: ClaudeTranscriptItem; start: number }> = []; + while (position > 0 && scanned < MAX_TRANSCRIPT_SCAN_BYTES && found.length <= params.limit) { + const size = Math.min( + TRANSCRIPT_READ_CHUNK_BYTES, + position, + MAX_TRANSCRIPT_SCAN_BYTES - scanned, + ); + position -= size; + const chunk = Buffer.allocUnsafe(size); + // Positional reads may return short, so complete the bounded window. + // A zero-byte read before it fills means the file changed after stat. + let filled = 0; + while (filled < size) { + const { bytesRead } = await handle.read(chunk, filled, size - filled, position + filled); + if (bytesRead === 0) { + throw new Error("Claude transcript changed while it was being read"); + } + filled += bytesRead; + } + scanned += filled; + let right = filled; + for (let index = filled - 1; index >= 0; index -= 1) { + if (chunk[index] !== 0x0a) { + continue; + } + const segment = chunk.subarray(index + 1, right); + if (segment.length > 0 || fragments.length > 0) { + const line = Buffer.concat([segment, ...fragments.toReversed()]); + const item = parseTranscriptLine(line, readBoundedString); + fragments = []; + if (item) { + found.push({ item, start: position + index + 1 }); + if (found.length > params.limit) { + break; + } + } + } + right = index; + } + if (found.length > params.limit) { + break; + } + const prefix = chunk.subarray(0, right); + if (position === 0) { + if (prefix.length > 0 || fragments.length > 0) { + const line = Buffer.concat([prefix, ...fragments.toReversed()]); + const item = parseTranscriptLine(line, readBoundedString); + if (item) { + found.push({ item, start: 0 }); + } + } + fragments = []; + } else if (prefix.length > 0) { + fragments.push(prefix); + } + } + if (position > 0 && found.length < params.limit) { + throw new Error("Claude transcript page exceeded the safe scan limit"); + } + const requested = found.slice(0, params.limit); + const selected: typeof requested = []; + let selectedBytes = 0; + for (const entry of requested) { + const itemBytes = Buffer.byteLength(JSON.stringify(entry.item), "utf8"); + if ( + selected.length > 0 && + selectedBytes + itemBytes > MAX_TRANSCRIPT_PAGE_BYTES - 64 * 1024 + ) { + break; + } + selected.push(entry); + selectedBytes += itemBytes; + } + const earliestStart = selected.at(-1)?.start; + const hasEarlierItems = selected.length < found.length || position > 0; + return { + threadId: params.threadId, + // Match the Codex session-page contract: newest item first on the wire; + // the shared UI prepends each page after restoring chronological order. + items: selected.map((entry) => entry.item), + ...(hasEarlierItems && earliestStart !== undefined && earliestStart > 0 + ? { nextCursor: encodeOffset(earliestStart) } + : {}), + }; + } finally { + await handle.close(); + } +} + +export async function listClaudeSessionCatalog(params: { + runtime: PluginRuntime; + query?: unknown; + allowProcessHomeFallback?: boolean; + listNodes?: Parameters[0]["listNodes"]; + onHost?: (host: ClaudeSessionCatalogHost) => void; +}): Promise { + const query = parseGatewayQuery(params.query); + const requested = query.hostIds ? new Set(query.hostIds) : undefined; + const scanOptions = gatewayClaudeScanOptions(params.allowProcessHomeFallback); + const localHosts: Promise[] = + (params.allowProcessHomeFallback !== false || scanOptions.configDir !== undefined) && + (!requested || requested.has(CLAUDE_LOCAL_SESSION_HOST_ID)) + ? [ + (async () => { + try { + return { + hostId: CLAUDE_LOCAL_SESSION_HOST_ID, + label: "Local Claude", + kind: "gateway", + connected: true, + ...(await listLocalClaudeSessionPage( + { + limit: query.limitPerHost, + ...(query.search ? { searchTerm: query.search } : {}), + ...(query.cursors?.[CLAUDE_LOCAL_SESSION_HOST_ID] !== undefined + ? { cursor: query.cursors[CLAUDE_LOCAL_SESSION_HOST_ID] } + : {}), + }, + currentHomeDir(), + scanOptions, + )), + }; + } catch { + return { + hostId: CLAUDE_LOCAL_SESSION_HOST_ID, + label: "Local Claude", + kind: "gateway", + connected: true, + sessions: [], + error: { + code: "LOCAL_READ_FAILED", + message: "Local Claude sessions are unavailable", + }, + }; + } + })(), + ] + : []; + for (const host of localHosts) { + if (params.onHost) { + void host.then(params.onHost).catch(() => undefined); + } + } + const wantsNodes = !requested || query.hostIds?.some((hostId) => hostId.startsWith("node:")); + if (!wantsNodes) { + return { hosts: await Promise.all(localHosts) }; + } + let nodes: Awaited>["nodes"]; + try { + nodes = (await (params.listNodes?.() ?? params.runtime.nodes.list())).nodes; + } catch (error) { + const registryHost: ClaudeSessionCatalogHost = { + hostId: "node:registry", + label: "Paired nodes", + kind: "node", + connected: false, + sessions: [], + error: createNodeListFailedError(error), + }; + params.onHost?.(registryHost); + return { + hosts: [...(await Promise.all(localHosts)), registryHost], + }; + } + const eligible = nodes + .filter( + (node) => + node.gatewayLocal !== true && + node.commands?.includes(CLAUDE_SESSIONS_LIST_COMMAND) && + (!requested || requested.has(`node:${node.nodeId}`)), + ) + .slice(0, MAX_HOSTS - localHosts.length) + .toSorted((left, right) => resolveNodeLabel(left).localeCompare(resolveNodeLabel(right))); + const nodeHosts = await Promise.all( + eligible.map(async (node): Promise => { + const hostId = `node:${node.nodeId}`; + const common = { + hostId, + label: resolveNodeLabel(node), + kind: "node" as const, + connected: node.connected === true, + nodeId: node.nodeId, + canContinueClaude: + node.commands?.includes(CLAUDE_SESSION_READ_COMMAND) === true && + node.commands.includes(CLAUDE_CLI_NODE_RUN_COMMAND) && + node.invocableCommands?.includes(CLAUDE_SESSIONS_LIST_COMMAND) === true && + node.invocableCommands.includes(CLAUDE_SESSION_READ_COMMAND) && + node.invocableCommands.includes(CLAUDE_CLI_NODE_RUN_COMMAND), + ...catalogTerminal.claudeNodeTerminalCapability(node), + }; + if (node.connected !== true) { + const host: ClaudeSessionCatalogHost = Object.assign({}, common, { + sessions: [], + error: { code: "NODE_OFFLINE", message: "Paired node is offline" }, + }); + params.onHost?.(host); + return host; + } + const eventualHost = Promise.resolve() + .then(async () => { + const raw = await params.runtime.nodes.invoke({ + nodeId: node.nodeId, + command: CLAUDE_SESSIONS_LIST_COMMAND, + params: { + limit: query.limitPerHost, + ...(query.search ? { searchTerm: query.search } : {}), + ...(query.cursors?.[hostId] !== undefined ? { cursor: query.cursors[hostId] } : {}), + }, + timeoutMs: NODE_INVOKE_TIMEOUT_MS, + scopes: ["operator.write"], + }); + return Object.assign({}, common, parseCatalogPage(unwrapNodePayload(raw))); + }) + .catch( + (): ClaudeSessionCatalogHost => + Object.assign({}, common, { + sessions: [], + error: { + code: "NODE_INVOKE_FAILED", + message: "Paired node Claude sessions are unavailable", + }, + }), + ); + if (params.onHost) { + // The fail-soft response can finish first; the original node invoke still + // publishes its authoritative host page whenever cold discovery completes. + void eventualHost.then(params.onHost).catch(() => undefined); + } + try { + return await withTimeout(eventualHost, NODE_CATALOG_LIST_RESPONSE_TIMEOUT_MS, { + message: "paired node Claude session catalog timed out", + }); + } catch { + return Object.assign({}, common, { + sessions: [], + error: { + code: "NODE_INVOKE_FAILED", + message: "Paired node Claude sessions are unavailable", + }, + }); + } + }), + ); + return { hosts: [...(await Promise.all(localHosts)), ...nodeHosts] }; +} + +export async function readClaudeSessionTranscript(params: { + runtime: PluginRuntime; + hostId: string; + threadId: string; + cursor?: string; + limit: number; + allowProcessHomeFallback?: boolean; +}): Promise { + const cursor = readOptionalCursor(params.cursor, "transcript"); + if (params.hostId === CLAUDE_LOCAL_SESSION_HOST_ID) { + assertClaudeLocalAccess(params.hostId, params.allowProcessHomeFallback); + return { + hostId: params.hostId, + label: "Local Claude", + ...(await readLocalClaudeTranscriptPage( + { + threadId: params.threadId, + limit: params.limit, + ...(cursor !== undefined ? { cursor } : {}), + }, + currentHomeDir(), + gatewayClaudeScanOptions(params.allowProcessHomeFallback), + )), + }; + } + if (!params.hostId.startsWith("node:")) { + throw new ClaudeCatalogParamsError("hostId is invalid"); + } + const nodeId = params.hostId.slice("node:".length); + const node = (await params.runtime.nodes.list()).nodes.find( + (candidate) => + candidate.nodeId === nodeId && + candidate.connected === true && + candidate.commands?.includes(CLAUDE_SESSION_READ_COMMAND), + ); + if (!node) { + throw new ClaudeCatalogParamsError("paired-node Claude session host is unavailable"); + } + const raw = await params.runtime.nodes.invoke({ + nodeId, + command: CLAUDE_SESSION_READ_COMMAND, + params: { + threadId: params.threadId, + limit: params.limit, + ...(cursor !== undefined ? { cursor } : {}), + }, + timeoutMs: NODE_INVOKE_TIMEOUT_MS, + scopes: ["operator.write"], + }); + const page = unwrapNodePayload(raw); + if ( + !isRecord(page) || + !Array.isArray(page.items) || + page.items.length > MAX_TRANSCRIPT_LIMIT || + page.items.some((item) => !isRecord(item) || typeof item.type !== "string") || + page.threadId !== params.threadId || + Buffer.byteLength(JSON.stringify(page), "utf8") > MAX_TRANSCRIPT_PAGE_BYTES + ) { + throw new Error("Claude node returned an invalid transcript page"); + } + const nextCursor = readNodePageCursor(page, "Claude node returned an invalid transcript page"); + return { + hostId: params.hostId, + label: resolveNodeLabel(node), + threadId: params.threadId, + items: page.items as ClaudeTranscriptItem[], + ...(nextCursor !== undefined ? { nextCursor } : {}), + }; +} + +export function assertClaudeLocalAccess(hostId: string, allowProcessHomeFallback?: boolean): void { + if ( + hostId === CLAUDE_LOCAL_SESSION_HOST_ID && + allowProcessHomeFallback === false && + configuredClaudeConfigDir() === undefined + ) { + throw new ClaudeCatalogParamsError("local Claude sessions are unavailable in isolated state"); + } +} + +export async function readBoundedClaudeHistory(params: { + runtime: PluginRuntime; + hostId: string; + threadId: string; + allowProcessHomeFallback?: boolean; +}): Promise { + const items: ClaudeTranscriptItem[] = []; + let cursor: string | undefined; + let bytes = 0; + while (items.length < CLAUDE_HISTORY_IMPORT_MAX_ITEMS) { + const page = await readClaudeSessionTranscript({ + runtime: params.runtime, + hostId: params.hostId, + threadId: params.threadId, + limit: Math.min(MAX_TRANSCRIPT_LIMIT, CLAUDE_HISTORY_IMPORT_MAX_ITEMS - items.length), + allowProcessHomeFallback: params.allowProcessHomeFallback, + ...(cursor ? { cursor } : {}), + }); + for (const item of page.items) { + const itemBytes = Buffer.byteLength(JSON.stringify(item), "utf8"); + if (items.length > 0 && bytes + itemBytes > CLAUDE_HISTORY_IMPORT_MAX_BYTES) { + return items; + } + items.push(item); + bytes += itemBytes; + } + if (!page.nextCursor || page.nextCursor === cursor) { + break; + } + cursor = page.nextCursor; + } + return items; +} + +export async function resolveNodeClaudeRecord(params: { + runtime: PluginRuntime; + nodeId: string; + threadId: string; +}): Promise { + let cursor: string | undefined; + for (let pageIndex = 0; pageIndex < 100; pageIndex += 1) { + const raw = await params.runtime.nodes.invoke({ + nodeId: params.nodeId, + command: CLAUDE_SESSIONS_LIST_COMMAND, + params: { + limit: MAX_PAGE_LIMIT, + searchTerm: params.threadId, + ...(cursor ? { cursor } : {}), + }, + timeoutMs: NODE_INVOKE_TIMEOUT_MS, + scopes: ["operator.write"], + }); + const page = parseCatalogPage(unwrapNodePayload(raw)); + const record = page.sessions.find((candidate) => candidate.threadId === params.threadId); + if (record) { + return record; + } + if (!page.nextCursor || page.nextCursor === cursor) { + break; + } + cursor = page.nextCursor; + } + throw new ClaudeCatalogParamsError("Claude session is unavailable on the paired node"); +} diff --git a/extensions/anthropic/session-catalog-parsing.ts b/extensions/anthropic/session-catalog-parsing.ts new file mode 100644 index 00000000000..1cffddc113a --- /dev/null +++ b/extensions/anthropic/session-catalog-parsing.ts @@ -0,0 +1,279 @@ +import { + isRecord, + normalizeBoundedOptionalString as readBoundedString, +} from "openclaw/plugin-sdk/string-coerce-runtime"; +import { CLAUDE_LOCAL_SESSION_HOST_ID } from "./session-catalog-adoption.js"; +import { isExactClaudeSessionCursor } from "./session-catalog-cursor.js"; +import { MAX_STRING_LENGTH, parsePullRequestSummary } from "./session-catalog-discovery.js"; +import { ClaudeCatalogParamsError } from "./session-catalog-shared.js"; +import type { + ClaudeSessionCatalogPage, + ClaudeSessionCatalogSession, +} from "./session-catalog-types.js"; + +const DEFAULT_PAGE_LIMIT = 50; +export const MAX_PAGE_LIMIT = 100; +export const DEFAULT_TRANSCRIPT_LIMIT = 20; +export const MAX_TRANSCRIPT_LIMIT = 50; +export const MAX_HOSTS = 100; +const MAX_SEARCH_LENGTH = 500; + +export function encodeOffset(offset: number): string { + return Buffer.from(JSON.stringify({ offset }), "utf8").toString("base64url"); +} + +export function decodeOffset(cursor: string | undefined, label: string): number { + if (cursor === undefined) { + return 0; + } + if (!isExactClaudeSessionCursor(cursor)) { + throw new ClaudeCatalogParamsError(`${label} cursor is invalid`); + } + try { + const parsed = JSON.parse(Buffer.from(cursor, "base64url").toString("utf8")) as unknown; + if ( + !isRecord(parsed) || + !Number.isSafeInteger(parsed.offset) || + (parsed.offset as number) < 0 + ) { + throw new Error("invalid offset"); + } + return parsed.offset as number; + } catch (error) { + throw new ClaudeCatalogParamsError(`${label} cursor is invalid`, { cause: error }); + } +} + +function readLimit(value: unknown, fallback: number, max: number): number { + if (value === undefined) { + return fallback; + } + if (!Number.isInteger(value) || (value as number) < 1 || (value as number) > max) { + throw new ClaudeCatalogParamsError(`limit must be an integer from 1 to ${max}`); + } + return value as number; +} + +function readRequiredCursor(value: unknown, message: string): string { + if (!isExactClaudeSessionCursor(value)) { + throw new ClaudeCatalogParamsError(message); + } + return value; +} + +export function readOptionalCursor(value: unknown, label: string): string | undefined { + if (value === undefined) { + return undefined; + } + return readRequiredCursor(value, `${label} cursor is invalid`); +} + +export function readListParams(value: unknown): { + cursor?: string; + limit: number; + searchTerm?: string; +} { + if (value === undefined || value === null) { + return { limit: DEFAULT_PAGE_LIMIT }; + } + if (!isRecord(value)) { + throw new ClaudeCatalogParamsError("Claude session catalog parameters must be an object"); + } + const allowed = new Set(["cursor", "limit", "searchTerm"]); + const unknown = Object.keys(value).find((key) => !allowed.has(key)); + if (unknown) { + throw new ClaudeCatalogParamsError(`unknown Claude session catalog parameter: ${unknown}`); + } + const cursor = readOptionalCursor(value.cursor, "catalog"); + const searchTerm = readBoundedString(value.searchTerm, MAX_SEARCH_LENGTH); + return { + limit: readLimit(value.limit, DEFAULT_PAGE_LIMIT, MAX_PAGE_LIMIT), + ...(cursor ? { cursor } : {}), + ...(searchTerm ? { searchTerm } : {}), + }; +} + +export function readTranscriptParams( + value: unknown, + options: { includeHostId?: boolean } = {}, +): { threadId: string; cursor?: string; limit: number } { + if (!isRecord(value)) { + throw new ClaudeCatalogParamsError("Claude session read parameters must be an object"); + } + const allowed = new Set([ + "threadId", + "cursor", + "limit", + ...(options.includeHostId ? ["hostId"] : []), + ]); + const unknown = Object.keys(value).find((key) => !allowed.has(key)); + if (unknown) { + throw new ClaudeCatalogParamsError(`unknown Claude session read parameter: ${unknown}`); + } + const threadId = readBoundedString(value.threadId, 256); + if (!threadId || !/^[A-Za-z0-9._:-]+$/.test(threadId)) { + throw new ClaudeCatalogParamsError("threadId is invalid"); + } + const cursor = readOptionalCursor(value.cursor, "transcript"); + return { + threadId, + limit: readLimit(value.limit, DEFAULT_TRANSCRIPT_LIMIT, MAX_TRANSCRIPT_LIMIT), + ...(cursor ? { cursor } : {}), + }; +} + +export function readNodePageCursor( + value: Record, + invalidPageMessage: string, +): string | undefined { + if (!("nextCursor" in value)) { + return undefined; + } + if (!isExactClaudeSessionCursor(value.nextCursor)) { + throw new Error(invalidPageMessage); + } + return value.nextCursor; +} + +export function parseCatalogPage(value: unknown): ClaudeSessionCatalogPage { + if ( + !isRecord(value) || + !Array.isArray(value.sessions) || + value.sessions.length > MAX_PAGE_LIMIT + ) { + throw new Error("Claude node returned an invalid session page"); + } + const sessions = value.sessions.map((candidate): ClaudeSessionCatalogSession => { + if (!isRecord(candidate)) { + throw new Error("Claude node returned an invalid session"); + } + const threadId = readBoundedString(candidate.threadId, 256); + const source = candidate.source; + if ( + !threadId || + candidate.archived !== false || + candidate.status !== "stored" || + (source !== "claude-cli" && source !== "claude-desktop") || + candidate.modelProvider !== "anthropic" + ) { + throw new Error("Claude node returned an invalid session"); + } + const parseStringField = (key: string, maxLength = MAX_STRING_LENGTH): string | undefined => { + if (!(key in candidate)) { + return undefined; + } + const parsed = readBoundedString(candidate[key], maxLength); + if (!parsed) { + throw new Error("Claude node returned an invalid session"); + } + return parsed; + }; + const parseNumberField = (key: string, nullable = false): number | null | undefined => { + if (!(key in candidate)) { + return undefined; + } + if (nullable && candidate[key] === null) { + return null; + } + const parsed = candidate[key]; + if (typeof parsed !== "number" || !Number.isFinite(parsed)) { + throw new Error("Claude node returned an invalid session"); + } + return parsed; + }; + let name: string | null | undefined; + if (candidate.name === null) { + name = null; + } else { + name = parseStringField("name", 500); + } + const cwd = parseStringField("cwd"); + const createdAt = parseNumberField("createdAt") as number | undefined; + const updatedAt = parseNumberField("updatedAt") as number | undefined; + const recencyAt = parseNumberField("recencyAt", true); + const cliVersion = parseStringField("cliVersion", 256); + const gitBranch = parseStringField("gitBranch", 500); + const pullRequest = parsePullRequestSummary(candidate.pullRequest); + return { + threadId, + status: "stored", + source, + modelProvider: "anthropic", + archived: false, + ...(name !== undefined ? { name } : {}), + ...(cwd ? { cwd } : {}), + ...(createdAt !== undefined ? { createdAt } : {}), + ...(updatedAt !== undefined ? { updatedAt } : {}), + ...(recencyAt !== undefined ? { recencyAt } : {}), + ...(cliVersion ? { cliVersion } : {}), + ...(gitBranch ? { gitBranch } : {}), + ...(pullRequest ? { pullRequest } : {}), + }; + }); + const nextCursor = readNodePageCursor(value, "Claude node returned an invalid session page"); + return { sessions, ...(nextCursor ? { nextCursor } : {}) }; +} + +export function unwrapNodePayload(value: unknown): unknown { + if (isRecord(value) && typeof value.payloadJSON === "string") { + return JSON.parse(value.payloadJSON) as unknown; + } + return value; +} + +export function parseGatewayQuery(value: unknown): { + search?: string; + limitPerHost: number; + hostIds?: string[]; + cursors?: Record; +} { + if (value === undefined || value === null) { + return { limitPerHost: DEFAULT_PAGE_LIMIT }; + } + if (!isRecord(value)) { + throw new ClaudeCatalogParamsError("Claude session catalog parameters must be an object"); + } + const allowed = new Set(["search", "limitPerHost", "hostIds", "cursors"]); + const unknown = Object.keys(value).find((key) => !allowed.has(key)); + if (unknown) { + throw new ClaudeCatalogParamsError(`unknown Claude session catalog parameter: ${unknown}`); + } + const search = readBoundedString(value.search, MAX_SEARCH_LENGTH); + let hostIds: string[] | undefined; + if (value.hostIds !== undefined) { + if (!Array.isArray(value.hostIds) || value.hostIds.length > MAX_HOSTS) { + throw new ClaudeCatalogParamsError("hostIds must be a bounded array"); + } + hostIds = [ + ...new Set( + value.hostIds.map((hostId) => { + const normalized = readBoundedString(hostId, 256); + if ( + !normalized || + (normalized !== CLAUDE_LOCAL_SESSION_HOST_ID && !normalized.startsWith("node:")) + ) { + throw new ClaudeCatalogParamsError("hostId is invalid"); + } + return normalized; + }), + ), + ]; + } + let cursors: Record | undefined; + if (value.cursors !== undefined) { + if (!isRecord(value.cursors) || Object.keys(value.cursors).length > MAX_HOSTS) { + throw new ClaudeCatalogParamsError("cursors must be a bounded object"); + } + cursors = Object.fromEntries( + Object.entries(value.cursors).map(([hostId, cursor]) => { + return [hostId, readRequiredCursor(cursor, `cursor for ${hostId} is invalid`)]; + }), + ); + } + return { + limitPerHost: readLimit(value.limitPerHost, DEFAULT_PAGE_LIMIT, MAX_PAGE_LIMIT), + ...(search ? { search } : {}), + ...(hostIds ? { hostIds } : {}), + ...(cursors ? { cursors } : {}), + }; +} diff --git a/extensions/anthropic/session-catalog-scan.ts b/extensions/anthropic/session-catalog-scan.ts new file mode 100644 index 00000000000..dbb838f375a --- /dev/null +++ b/extensions/anthropic/session-catalog-scan.ts @@ -0,0 +1,285 @@ +import type { Dirent, Stats } from "node:fs"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; + +const MAX_CATALOG_JSON_CACHE_ENTRIES = 4_000; +export const CLAUDE_CATALOG_IO_CONCURRENCY = 32; + +type CatalogJsonCacheEntry = { + mtimeMs: number; + size: number; + value: unknown; +}; + +type SafeSessionFile = { filePath: string; stat: Stats } | undefined; + +type ClaudeProjectDirectorySnapshot = { + directory: string; + childNames: string[]; +}; + +type ClaudeChildFileSignature = readonly [name: string, mtimeMs: number, size: number, ino: number]; + +export type ClaudeProjectsTreeSnapshot = { + root: string; + resolvedRoot?: string; + projectDirectories: ClaudeProjectDirectorySnapshot[]; + treeStamp: string; +}; + +export type ClaudeSessionScanContext = ClaudeProjectsTreeSnapshot & { + complete: boolean; + safeFiles: Map>; +}; + +// Parsed index/Desktop JSON stays valid for one path+mtime+size and is LRU-bounded; read failures are +// never cached, so transient metadata I/O cannot hide a later successful read. +const catalogJsonCache = new Map(); + +export async function mapConcurrent( + values: T[], + limit: number, + mapper: (value: T) => Promise, +): Promise { + const results: R[] = []; + results.length = values.length; + let nextIndex = 0; + const workers = Array.from({ length: Math.min(limit, values.length) }, async () => { + while (nextIndex < values.length) { + const index = nextIndex++; + results[index] = await mapper(values[index]!); + } + }); + await Promise.all(workers); + return results; +} + +export function setBoundedCache( + cache: Map, + key: K, + value: V, + maxEntries: number, +): void { + cache.delete(key); + cache.set(key, value); + while (cache.size > maxEntries) { + const oldest = cache.keys().next(); + if (oldest.done) { + break; + } + cache.delete(oldest.value); + } +} + +function isWithin(root: string, candidate: string): boolean { + const relative = path.relative(path.resolve(root), path.resolve(candidate)); + return relative === "" || (!relative.startsWith("..") && !path.isAbsolute(relative)); +} + +async function safeSessionFile( + root: string, + resolvedRoot: string, + candidate: string, + sessionId: string, +): Promise { + if (!isWithin(root, candidate) || path.basename(candidate) !== `${sessionId}.jsonl`) { + return undefined; + } + try { + const resolvedCandidate = await fs.realpath(candidate); + if (!isWithin(resolvedRoot, resolvedCandidate)) { + return undefined; + } + const stat = await fs.stat(resolvedCandidate); + return stat.isFile() ? { filePath: resolvedCandidate, stat } : undefined; + } catch (error) { + const code = error && typeof error === "object" && "code" in error ? error.code : undefined; + if (code === "ENOENT" || code === "ENOTDIR") { + return undefined; + } + throw new Error("Claude session file validation failed", { cause: error }); + } +} + +export function safeSessionFileForScan( + context: ClaudeSessionScanContext, + candidate: string, + sessionId: string, +): Promise { + if (!context.resolvedRoot) { + return Promise.resolve(undefined); + } + const key = `${sessionId}\0${path.resolve(candidate)}`; + let pending = context.safeFiles.get(key); + if (!pending) { + // Canonical path + stat are valid only for this assembled scan. Sharing the promise prevents + // index fallback and discovery from serially resolving the same file twice. + const request = safeSessionFile(context.root, context.resolvedRoot, candidate, sessionId); + pending = request.catch(() => { + context.complete = false; + if (context.safeFiles.get(key) === pending) { + context.safeFiles.delete(key); + } + return undefined; + }); + context.safeFiles.set(key, pending); + } + return pending; +} + +export async function readJsonFile( + filePath: string, + options: { onIoFailure?: () => void } = {}, +): Promise { + const stat = await fs.stat(filePath).catch(() => { + options.onIoFailure?.(); + return undefined; + }); + if (!stat?.isFile()) { + catalogJsonCache.delete(filePath); + return undefined; + } + const cached = catalogJsonCache.get(filePath); + if (cached && cached.mtimeMs === stat.mtimeMs && cached.size === stat.size) { + setBoundedCache(catalogJsonCache, filePath, cached, MAX_CATALOG_JSON_CACHE_ENTRIES); + return cached.value; + } + let content: string; + try { + content = await fs.readFile(filePath, "utf8"); + } catch { + options.onIoFailure?.(); + return undefined; + } + try { + const value = JSON.parse(content) as unknown; + setBoundedCache( + catalogJsonCache, + filePath, + { mtimeMs: stat.mtimeMs, size: stat.size, value }, + MAX_CATALOG_JSON_CACHE_ENTRIES, + ); + return value; + } catch { + return undefined; + } +} + +export async function childDirectories(root: string): Promise { + try { + return (await fs.readdir(root, { withFileTypes: true })) + .filter((entry) => entry.isDirectory()) + .map((entry) => path.join(root, entry.name)); + } catch { + return []; + } +} + +export function projectsDir(homeDir: string, configDir?: string): string { + return path.join(configDir ?? path.join(homeDir, ".claude"), "projects"); +} + +export async function readProjectsTreeSnapshot(root: string): Promise { + let entries: Dirent[]; + try { + entries = await fs.readdir(root, { withFileTypes: true }); + } catch { + return { root, projectDirectories: [], treeStamp: "unavailable" }; + } + const directoryEntries = entries.filter((entry) => entry.isDirectory()); + const [resolvedRoot, directories] = await Promise.all([ + fs.realpath(root).catch(() => undefined), + mapConcurrent(directoryEntries, CLAUDE_CATALOG_IO_CONCURRENCY, async (entry) => { + const directory = path.join(root, entry.name); + const [stat, children] = await Promise.all([ + fs.stat(directory).catch(() => undefined), + fs.readdir(directory, { withFileTypes: true }).catch(() => undefined), + ]); + return { entry, directory, stat, children }; + }), + ]); + const childTargets = directories.flatMap(({ directory, children }, directoryIndex) => + (children ?? []).map((child) => ({ directoryIndex, directory, child })), + ); + const childSignatures = await mapConcurrent( + childTargets, + CLAUDE_CATALOG_IO_CONCURRENCY, + async ({ directoryIndex, directory, child }) => { + const childStat = await fs.stat(path.join(directory, child.name)).catch(() => undefined); + const signature = childStat?.isFile() + ? ([child.name, childStat.mtimeMs, childStat.size, childStat.ino] as const) + : undefined; + return { directoryIndex, signature }; + }, + ); + const signaturesByDirectory = Array.from( + { length: directories.length }, + (): ClaudeChildFileSignature[] => [], + ); + for (const { directoryIndex, signature } of childSignatures) { + if (signature) { + signaturesByDirectory[directoryIndex]?.push(signature); + } + } + const directorySnapshots = directories.map(({ entry, directory, stat, children }, index) => { + const fileSignatures = signaturesByDirectory[index] ?? []; + const maxChildMtime = fileSignatures.reduce( + (maximum, [, mtime]) => Math.max(maximum ?? mtime, mtime), + null, + ); + return { + directory, + childNames: children?.map((child) => child.name) ?? [], + stamp: [ + entry.name, + stat?.isDirectory() === true ? stat.mtimeMs : null, + children?.map((child) => child.name) ?? null, + maxChildMtime ?? null, + fileSignatures, + ] as const, + }; + }); + return { + root, + ...(resolvedRoot ? { resolvedRoot } : {}), + projectDirectories: directorySnapshots.map(({ directory, childNames }) => ({ + directory, + childNames, + })), + treeStamp: JSON.stringify([resolvedRoot ?? null, directorySnapshots.map(({ stamp }) => stamp)]), + }; +} + +export async function desktopSessionStoreAvailable(homeDir: string): Promise { + const stat = await fs.stat(desktopSessionsDir(homeDir)).catch(() => undefined); + return stat?.isDirectory() === true; +} + +export function desktopSessionsDir(homeDir: string): string { + return path.join(homeDir, "Library", "Application Support", "Claude", "claude-code-sessions"); +} + +export function currentHomeDir(env: NodeJS.ProcessEnv = process.env): string { + return env.HOME?.trim() || env.USERPROFILE?.trim() || os.homedir(); +} + +export function configuredClaudeConfigDir( + env: NodeJS.ProcessEnv = process.env, +): string | undefined { + const configured = env.CLAUDE_CONFIG_DIR?.trim(); + return configured ? path.resolve(configured) : undefined; +} + +export function gatewayClaudeScanOptions(allowProcessHomeFallback?: boolean): { + configDir?: string; + includeDesktop: boolean; +} { + const configDir = configuredClaudeConfigDir(); + // Upstream Claude Code's "Respect CLAUDE_CONFIG_DIR everywhere" convention replaces ~/.claude. + // Claude Desktop stays HOME/Library-scoped, so isolated scans exclude its metadata. + return { + ...(configDir ? { configDir } : {}), + includeDesktop: allowProcessHomeFallback !== false, + }; +} diff --git a/extensions/anthropic/session-catalog.ts b/extensions/anthropic/session-catalog.ts index e64e5fb59de..6fd66192f78 100644 --- a/extensions/anthropic/session-catalog.ts +++ b/extensions/anthropic/session-catalog.ts @@ -1,1900 +1,36 @@ -import type { Dirent, Stats } from "node:fs"; -import fs from "node:fs/promises"; -import os from "node:os"; -import path from "node:path"; import { resolveSessionAgentIds } from "openclaw/plugin-sdk/agent-runtime"; -import { parseDateFirstTimestampMs } from "openclaw/plugin-sdk/number-runtime"; import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; -import type { PluginRuntime } from "openclaw/plugin-sdk/plugin-runtime"; -import { withTimeout } from "openclaw/plugin-sdk/security-runtime"; import type { SessionCatalogHost, SessionCatalogProvider, - SessionCatalogPullRequestSummary, SessionCatalogTranscriptItem, } from "openclaw/plugin-sdk/session-catalog"; -import { createSessionCatalogAdoptionCoordinator } from "openclaw/plugin-sdk/session-catalog"; +import { adoptedSourceKey, CLAUDE_LOCAL_SESSION_HOST_ID } from "./session-catalog-adoption.js"; +import { continueClaudeSession } from "./session-catalog-continue.js"; +import { listClaudeSessions } from "./session-catalog-discovery.js"; import { - asPositiveSafeInteger as pullRequestNumber, - isRecord, - normalizeBoundedOptionalString as readBoundedString, -} from "openclaw/plugin-sdk/string-coerce-runtime"; -import { readClaudeDesktopCustomGroups } from "./claude-desktop-groups.js"; -import { CLAUDE_CLI_BACKEND_ID, CLAUDE_CLI_DEFAULT_MODEL_REF } from "./cli-constants.js"; + assertClaudeLocalAccess, + listClaudeSessionCatalog, + readClaudeSessionTranscript, + resolveNodeClaudeRecord, +} from "./session-catalog-listing.js"; +import { DEFAULT_TRANSCRIPT_LIMIT, MAX_TRANSCRIPT_LIMIT } from "./session-catalog-parsing.js"; +import { listBoundClaudeSessions } from "./session-catalog-runtime.js"; import { - adoptedSessionKey, - adoptedSourceKey, - CLAUDE_LOCAL_SESSION_HOST_ID, -} from "./session-catalog-adoption.js"; -import { isExactClaudeSessionCursor } from "./session-catalog-cursor.js"; -import { importClaudeHistory } from "./session-catalog-history.js"; -import { createNodeListFailedError, resolveNodeLabel } from "./session-catalog-node-helpers.js"; -import { - currentClaudeSessionCatalogConfig, - listBoundClaudeSessions, - resolveClaudeCliRoutedModelId, -} from "./session-catalog-runtime.js"; -import { - CLAUDE_CLI_NODE_RUN_COMMAND, - CLAUDE_SESSION_READ_COMMAND, - CLAUDE_SESSIONS_LIST_COMMAND, - ClaudeCatalogParamsError, - isResumableClaudeSource, -} from "./session-catalog-shared.js"; + configuredClaudeConfigDir, + currentHomeDir, + gatewayClaudeScanOptions, +} from "./session-catalog-scan.js"; import * as catalogTerminal from "./session-catalog-terminal.js"; -import { - collectTranscriptText, - parseTranscriptLine, - type ClaudeTranscriptItem, -} from "./session-catalog-transcript.js"; -import type { - ClaudeSessionCatalogHost, - ClaudeSessionCatalogPage, - ClaudeSessionCatalogResult, - ClaudeSessionCatalogSession, - ClaudeSessionTranscriptPage, -} from "./session-catalog-types.js"; +import type { ClaudeTranscriptItem } from "./session-catalog-transcript.js"; +import type { ClaudeSessionCatalogHost } from "./session-catalog-types.js"; import * as upstream from "./session-upstream-activity.js"; export * from "./session-catalog-shared.js"; - -const DEFAULT_PAGE_LIMIT = 50; -const MAX_PAGE_LIMIT = 100; -const DEFAULT_TRANSCRIPT_LIMIT = 20; -const MAX_TRANSCRIPT_LIMIT = 50; -const MAX_HOSTS = 100; -const MAX_STRING_LENGTH = 4096; -const MAX_SEARCH_LENGTH = 500; -const MAX_SESSION_PULL_REQUESTS = 20; -const MAX_CATALOG_DISCOVERY_FILES = 10_000; -const MAX_CATALOG_DISCOVERY_CACHE_ENTRIES = 20_000; -const MAX_CATALOG_JSON_CACHE_ENTRIES = 4_000; -const MAX_CLAUDE_SESSION_SCAN_CACHE_ENTRIES = 8; -const CLAUDE_SESSION_SCAN_HARD_TTL_MS = 5 * 60_000; -const CLAUDE_PARTIAL_SCAN_TTL_MS = 15_000; -const CLAUDE_DESKTOP_SCAN_TTL_MS = 60_000; -const CLAUDE_CATALOG_IO_CONCURRENCY = 32; -const CLAUDE_METADATA_PREFIX_BYTES = 1024 * 1024; -const CLAUDE_METADATA_READ_CHUNK_BYTES = 16 * 1024; -const MAX_CATALOG_METADATA_SCAN_BYTES = 64 * 1024 * 1024; -const TRANSCRIPT_READ_CHUNK_BYTES = 128 * 1024; -const MAX_TRANSCRIPT_SCAN_BYTES = 64 * 1024 * 1024; -const MAX_TRANSCRIPT_PAGE_BYTES = 20 * 1024 * 1024; -const CLI_ENTRYPOINTS = new Set(["cli", "sdk-cli"]); - -const NODE_INVOKE_TIMEOUT_MS = 30_000; -// Catalog refresh is fail-soft: one unhealthy machine must not hold the whole sidebar. -// The node invoke keeps running so cold native discovery can warm the next poll. -const NODE_CATALOG_LIST_RESPONSE_TIMEOUT_MS = 8_000; -const CLAUDE_HISTORY_IMPORT_MAX_ITEMS = 200; -const CLAUDE_HISTORY_IMPORT_MAX_BYTES = 512 * 1024; -const continueClaudeAdoption = - createSessionCatalogAdoptionCoordinator>>(); - -type SessionIndexEntry = { - sessionId?: unknown; - fullPath?: unknown; - fileMtime?: unknown; - firstPrompt?: unknown; - summary?: unknown; - messageCount?: unknown; - created?: unknown; - modified?: unknown; - gitBranch?: unknown; - projectPath?: unknown; - isSidechain?: unknown; -}; - -type DesktopSessionMetadata = { - sessionId?: unknown; - cliSessionId?: unknown; - cwd?: unknown; - originCwd?: unknown; - createdAt?: unknown; - lastActivityAt?: unknown; - model?: unknown; - isArchived?: unknown; - title?: unknown; - customGroup?: unknown; - prNumber?: unknown; - prState?: unknown; - prs?: unknown; -}; - -type DesktopPullRequestMetadata = { - prNumber?: unknown; - state?: unknown; - dismissed?: unknown; -}; - -type CatalogRecord = ClaudeSessionCatalogSession & { - filePath: string; -}; - -type CatalogDiscoveryCacheEntry = { - // The module-global cache is keyed by canonical transcript path, so an entry must also record the - // discovery context it was built in. `root` is the logical (unresolved) projects root: it scopes - // the entry to its homeDir even when the root itself is a symlink, so a different homeDir scan - // cannot reuse it and eviction can find it without re-resolving a now-missing root. mtime+size+ino - // detect any content change or atomic replacement; sessionId guards against a canonical path being - // reached under a different filename-derived id (e.g. an aliased/renamed symlink). - root: string; - mtimeMs: number; - size: number; - ino: number; - sessionId: string; - // Bytes this file charged against the scan budget when first scanned. Cache hits re-charge it so - // byte-budget-limited discovery stops at the same frontier whether or not the cache is warm, - // keeping pagination deterministic across repeated identical calls. - scannedBytes: number; - record: CatalogRecord | null; - sidechain: boolean; -}; - -type CatalogJsonCacheEntry = { - mtimeMs: number; - size: number; - value: unknown; -}; - -type ClaudeSessionScanCacheEntry = { - treeStamp: string; - hardExpiresAt: number; - desktopStoreAvailable: boolean; - desktopExpiresAt: number; - records: Promise; -}; - -type SafeSessionFile = { filePath: string; stat: Stats } | undefined; - -type ClaudeProjectDirectorySnapshot = { - directory: string; - childNames: string[]; -}; - -type ClaudeChildFileSignature = readonly [name: string, mtimeMs: number, size: number, ino: number]; - -type ClaudeProjectsTreeSnapshot = { - root: string; - resolvedRoot?: string; - projectDirectories: ClaudeProjectDirectorySnapshot[]; - treeStamp: string; -}; - -type ClaudeSessionScanContext = ClaudeProjectsTreeSnapshot & { - complete: boolean; - safeFiles: Map>; -}; - -// Transcript discoveries stay valid only for the same root/id/inode/mtime/size and are LRU-bounded; -// a false hit would corrupt pagination, so warm scans re-charge the original deterministic byte cost. -const catalogDiscoveryCache = new Map(); -// Parsed index/Desktop JSON stays valid for one path+mtime+size and is LRU-bounded; read failures are -// never cached, so transient metadata I/O cannot hide a later successful read. -const catalogJsonCache = new Map(); -// Whole scans are root-scoped and bounded; tree/Desktop/hard expiries below own invalidation, avoiding -// an unbounded home map while preserving the exact resolved records promise for concurrent callers. -const claudeSessionScanCache = new Map(); - -async function mapConcurrent( - values: T[], - limit: number, - mapper: (value: T) => Promise, -): Promise { - const results: R[] = []; - results.length = values.length; - let nextIndex = 0; - const workers = Array.from({ length: Math.min(limit, values.length) }, async () => { - while (nextIndex < values.length) { - const index = nextIndex++; - results[index] = await mapper(values[index]!); - } - }); - await Promise.all(workers); - return results; -} - -function setBoundedCache(cache: Map, key: K, value: V, maxEntries: number): void { - cache.delete(key); - cache.set(key, value); - while (cache.size > maxEntries) { - const oldest = cache.keys().next(); - if (oldest.done) { - break; - } - cache.delete(oldest.value); - } -} - -function cacheCatalogDiscovery(filePath: string, entry: CatalogDiscoveryCacheEntry): void { - setBoundedCache(catalogDiscoveryCache, filePath, entry, MAX_CATALOG_DISCOVERY_CACHE_ENTRIES); -} - -function pullRequestState(value: unknown): SessionCatalogPullRequestSummary["state"] | undefined { - if (typeof value !== "string") { - return undefined; - } - switch (value.trim().toLowerCase()) { - case "open": - case "draft": - case "merged": - case "closed": - return value.trim().toLowerCase() as SessionCatalogPullRequestSummary["state"]; - default: - return undefined; - } -} - -// Desktop retains historical PRs in order and marks hidden ones as dismissed; -// the top-level pair identifies the current PR whose state labels the row. -function desktopPullRequestSummary( - metadata: DesktopSessionMetadata, -): SessionCatalogPullRequestSummary | undefined { - const visibleByNumber = new Map(); - const dismissed = new Set(); - if (Array.isArray(metadata.prs)) { - for (const value of metadata.prs) { - if (!isRecord(value)) { - continue; - } - const entry = value as DesktopPullRequestMetadata; - const number = pullRequestNumber(entry.prNumber); - if (!number) { - continue; - } - if (entry.dismissed === true) { - dismissed.add(number); - visibleByNumber.delete(number); - continue; - } - if (!dismissed.has(number) && !visibleByNumber.has(number)) { - visibleByNumber.set(number, pullRequestState(entry.state)); - } - } - } - const currentNumber = pullRequestNumber(metadata.prNumber); - let currentState = currentNumber ? visibleByNumber.get(currentNumber) : undefined; - if (currentNumber && !dismissed.has(currentNumber)) { - currentState = pullRequestState(metadata.prState) ?? currentState; - // Reinsert the current PR at the tail so truncation always retains it. - visibleByNumber.delete(currentNumber); - visibleByNumber.set(currentNumber, currentState); - } - const visible = [...visibleByNumber].map(([number, state]) => ({ number, state })); - if (visible.length === 0) { - return undefined; - } - const state = currentState ?? visible.at(-1)?.state; - if (!state) { - return undefined; - } - return { - numbers: visible.slice(-MAX_SESSION_PULL_REQUESTS).map((entry) => entry.number), - state, - }; -} - -function parsePullRequestSummary(value: unknown): SessionCatalogPullRequestSummary | undefined { - if (value === undefined) { - return undefined; - } - if (!isRecord(value) || !Array.isArray(value.numbers)) { - throw new Error("Claude node returned an invalid pull request summary"); - } - const numbers = value.numbers.map(pullRequestNumber); - const state = pullRequestState(value.state); - if ( - numbers.length === 0 || - numbers.length > MAX_SESSION_PULL_REQUESTS || - numbers.some((number) => number === undefined) || - new Set(numbers).size !== numbers.length || - !state - ) { - throw new Error("Claude node returned an invalid pull request summary"); - } - return { numbers: numbers as number[], state }; -} - -function isCliEntrypoint(value: unknown): value is string { - return typeof value === "string" && CLI_ENTRYPOINTS.has(value); -} - -// Claude's persisted string timestamps are date expressions, including numeric-looking years. -// Numeric fields are already millisecond values, so preserve that distinct mixed-input contract. -function parseClaudeCatalogTimestampMs(value: unknown): number | undefined { - return parseDateFirstTimestampMs(value); -} - -function isWithin(root: string, candidate: string): boolean { - const relative = path.relative(path.resolve(root), path.resolve(candidate)); - return relative === "" || (!relative.startsWith("..") && !path.isAbsolute(relative)); -} - -async function safeSessionFile( - root: string, - resolvedRoot: string, - candidate: string, - sessionId: string, -): Promise { - if (!isWithin(root, candidate) || path.basename(candidate) !== `${sessionId}.jsonl`) { - return undefined; - } - try { - const resolvedCandidate = await fs.realpath(candidate); - if (!isWithin(resolvedRoot, resolvedCandidate)) { - return undefined; - } - const stat = await fs.stat(resolvedCandidate); - return stat.isFile() ? { filePath: resolvedCandidate, stat } : undefined; - } catch (error) { - const code = error && typeof error === "object" && "code" in error ? error.code : undefined; - if (code === "ENOENT" || code === "ENOTDIR") { - return undefined; - } - throw new Error("Claude session file validation failed", { cause: error }); - } -} - -function safeSessionFileForScan( - context: ClaudeSessionScanContext, - candidate: string, - sessionId: string, -): Promise { - if (!context.resolvedRoot) { - return Promise.resolve(undefined); - } - const key = `${sessionId}\0${path.resolve(candidate)}`; - let pending = context.safeFiles.get(key); - if (!pending) { - // Canonical path + stat are valid only for this assembled scan. Sharing the promise prevents - // index fallback and discovery from serially resolving the same file twice. - const request = safeSessionFile(context.root, context.resolvedRoot, candidate, sessionId); - pending = request.catch(() => { - context.complete = false; - if (context.safeFiles.get(key) === pending) { - context.safeFiles.delete(key); - } - return undefined; - }); - context.safeFiles.set(key, pending); - } - return pending; -} - -async function readJsonFile( - filePath: string, - options: { onIoFailure?: () => void } = {}, -): Promise { - const stat = await fs.stat(filePath).catch(() => { - options.onIoFailure?.(); - return undefined; - }); - if (!stat?.isFile()) { - catalogJsonCache.delete(filePath); - return undefined; - } - const cached = catalogJsonCache.get(filePath); - if (cached && cached.mtimeMs === stat.mtimeMs && cached.size === stat.size) { - setBoundedCache(catalogJsonCache, filePath, cached, MAX_CATALOG_JSON_CACHE_ENTRIES); - return cached.value; - } - let content: string; - try { - content = await fs.readFile(filePath, "utf8"); - } catch { - options.onIoFailure?.(); - return undefined; - } - try { - const value = JSON.parse(content) as unknown; - setBoundedCache( - catalogJsonCache, - filePath, - { mtimeMs: stat.mtimeMs, size: stat.size, value }, - MAX_CATALOG_JSON_CACHE_ENTRIES, - ); - return value; - } catch { - return undefined; - } -} - -async function childDirectories(root: string): Promise { - try { - return (await fs.readdir(root, { withFileTypes: true })) - .filter((entry) => entry.isDirectory()) - .map((entry) => path.join(root, entry.name)); - } catch { - return []; - } -} - -function projectsDir(homeDir: string, configDir?: string): string { - return path.join(configDir ?? path.join(homeDir, ".claude"), "projects"); -} - -async function readProjectsTreeSnapshot(root: string): Promise { - let entries: Dirent[]; - try { - entries = await fs.readdir(root, { withFileTypes: true }); - } catch { - return { root, projectDirectories: [], treeStamp: "unavailable" }; - } - const directoryEntries = entries.filter((entry) => entry.isDirectory()); - const [resolvedRoot, directories] = await Promise.all([ - fs.realpath(root).catch(() => undefined), - mapConcurrent(directoryEntries, CLAUDE_CATALOG_IO_CONCURRENCY, async (entry) => { - const directory = path.join(root, entry.name); - const [stat, children] = await Promise.all([ - fs.stat(directory).catch(() => undefined), - fs.readdir(directory, { withFileTypes: true }).catch(() => undefined), - ]); - return { entry, directory, stat, children }; - }), - ]); - const childTargets = directories.flatMap(({ directory, children }, directoryIndex) => - (children ?? []).map((child) => ({ directoryIndex, directory, child })), - ); - const childSignatures = await mapConcurrent( - childTargets, - CLAUDE_CATALOG_IO_CONCURRENCY, - async ({ directoryIndex, directory, child }) => { - const childStat = await fs.stat(path.join(directory, child.name)).catch(() => undefined); - const signature = childStat?.isFile() - ? ([child.name, childStat.mtimeMs, childStat.size, childStat.ino] as const) - : undefined; - return { directoryIndex, signature }; - }, - ); - const signaturesByDirectory = Array.from( - { length: directories.length }, - (): ClaudeChildFileSignature[] => [], - ); - for (const { directoryIndex, signature } of childSignatures) { - if (signature) { - signaturesByDirectory[directoryIndex]?.push(signature); - } - } - const directorySnapshots = directories.map(({ entry, directory, stat, children }, index) => { - const fileSignatures = signaturesByDirectory[index] ?? []; - const maxChildMtime = fileSignatures.reduce( - (maximum, [, mtime]) => Math.max(maximum ?? mtime, mtime), - null, - ); - return { - directory, - childNames: children?.map((child) => child.name) ?? [], - stamp: [ - entry.name, - stat?.isDirectory() === true ? stat.mtimeMs : null, - children?.map((child) => child.name) ?? null, - maxChildMtime ?? null, - fileSignatures, - ] as const, - }; - }); - return { - root, - ...(resolvedRoot ? { resolvedRoot } : {}), - projectDirectories: directorySnapshots.map(({ directory, childNames }) => ({ - directory, - childNames, - })), - treeStamp: JSON.stringify([resolvedRoot ?? null, directorySnapshots.map(({ stamp }) => stamp)]), - }; -} - -async function desktopSessionStoreAvailable(homeDir: string): Promise { - const stat = await fs.stat(desktopSessionsDir(homeDir)).catch(() => undefined); - return stat?.isDirectory() === true; -} - -function desktopSessionsDir(homeDir: string): string { - return path.join(homeDir, "Library", "Application Support", "Claude", "claude-code-sessions"); -} - -function currentHomeDir(env: NodeJS.ProcessEnv = process.env): string { - return env.HOME?.trim() || env.USERPROFILE?.trim() || os.homedir(); -} - -function configuredClaudeConfigDir(env: NodeJS.ProcessEnv = process.env): string | undefined { - const configured = env.CLAUDE_CONFIG_DIR?.trim(); - return configured ? path.resolve(configured) : undefined; -} - -function gatewayClaudeScanOptions(allowProcessHomeFallback?: boolean): { - configDir?: string; - includeDesktop: boolean; -} { - const configDir = configuredClaudeConfigDir(); - // Upstream Claude Code's "Respect CLAUDE_CONFIG_DIR everywhere" convention replaces ~/.claude. - // Claude Desktop stays HOME/Library-scoped, so isolated scans exclude its metadata. - return { - ...(configDir ? { configDir } : {}), - includeDesktop: allowProcessHomeFallback !== false, - }; -} - -async function readDesktopMetadata(homeDir: string): Promise<{ - active: Map; - archived: Set; -}> { - const active = new Map(); - const archived = new Set(); - const customGroups = await readClaudeDesktopCustomGroups(homeDir); - for (const accountDir of await childDirectories(desktopSessionsDir(homeDir))) { - for (const workspaceDir of await childDirectories(accountDir)) { - let entries: string[]; - try { - entries = await fs.readdir(workspaceDir); - } catch { - continue; - } - for (const name of entries) { - if (!name.startsWith("local_") || !name.endsWith(".json")) { - continue; - } - const raw = await readJsonFile(path.join(workspaceDir, name)); - if (!isRecord(raw)) { - continue; - } - const metadata = raw as DesktopSessionMetadata; - const cliSessionId = readBoundedString(metadata.cliSessionId, 256); - if (!cliSessionId) { - continue; - } - if (metadata.isArchived === true) { - archived.add(cliSessionId); - active.delete(cliSessionId); - continue; - } - if (!archived.has(cliSessionId)) { - const localSessionId = readBoundedString(metadata.sessionId, 256); - const customGroup = localSessionId ? customGroups.get(localSessionId) : undefined; - active.set(cliSessionId, customGroup ? { ...metadata, customGroup } : metadata); - } - } - } - } - return { active, archived }; -} - -async function readIndexRecords(context: ClaudeSessionScanContext): Promise<{ - records: Map; - sidechainIds: Set; -}> { - const records = new Map(); - const sidechainIds = new Set(); - if (!context.resolvedRoot) { - return { records, sidechainIds }; - } - const indexes = await mapConcurrent( - context.projectDirectories, - CLAUDE_CATALOG_IO_CONCURRENCY, - async ({ directory, childNames }) => ({ - directory, - raw: childNames.includes("sessions-index.json") - ? await readJsonFile(path.join(directory, "sessions-index.json"), { - onIoFailure: () => { - context.complete = false; - }, - }) - : undefined, - }), - ); - const candidates: Array<{ - directory: string; - entry: SessionIndexEntry; - sessionId: string; - }> = []; - for (const { directory, raw } of indexes) { - if (!isRecord(raw) || !Array.isArray(raw.entries)) { - continue; - } - for (const candidate of raw.entries) { - if (!isRecord(candidate)) { - continue; - } - const entry = candidate as SessionIndexEntry; - const sessionId = readBoundedString(entry.sessionId, 256); - if (!sessionId) { - continue; - } - candidates.push({ directory, entry, sessionId }); - } - } - const safeFiles = await mapConcurrent( - candidates, - CLAUDE_CATALOG_IO_CONCURRENCY, - async ({ directory, entry, sessionId }) => { - if (entry.isSidechain === true) { - return undefined; - } - const indexedPath = readBoundedString(entry.fullPath, MAX_STRING_LENGTH); - return await safeSessionFileForScan( - context, - indexedPath ?? path.join(directory, `${sessionId}.jsonl`), - sessionId, - ); - }, - ); - for (const [index, candidate] of candidates.entries()) { - const { entry, sessionId } = candidate; - if (entry.isSidechain === true) { - sidechainIds.add(sessionId); - records.delete(sessionId); - continue; - } - const safeFile = safeFiles[index]; - if (!safeFile) { - continue; - } - const createdAt = parseClaudeCatalogTimestampMs(entry.created); - const updatedAt = - parseClaudeCatalogTimestampMs(entry.modified) ?? - parseClaudeCatalogTimestampMs(entry.fileMtime); - const summary = readBoundedString(entry.summary, 500); - const firstPrompt = readBoundedString(entry.firstPrompt, 500); - records.set(sessionId, { - threadId: sessionId, - name: summary ?? firstPrompt ?? null, - cwd: readBoundedString(entry.projectPath, MAX_STRING_LENGTH), - status: "stored", - ...(createdAt !== undefined ? { createdAt } : {}), - ...(updatedAt !== undefined ? { updatedAt, recencyAt: updatedAt } : {}), - source: "claude-cli", - modelProvider: "anthropic", - ...(readBoundedString(entry.gitBranch, 500) - ? { gitBranch: readBoundedString(entry.gitBranch, 500) } - : {}), - archived: false, - filePath: safeFile.filePath, - }); - } - return { records, sidechainIds }; -} - -async function locateSessionFile( - context: ClaudeSessionScanContext, - sessionId: string, -): Promise { - const fileName = `${sessionId}.jsonl`; - for (const { directory, childNames } of context.projectDirectories) { - if (!childNames.includes(fileName)) { - continue; - } - const candidate = path.join(directory, fileName); - const safeFile = await safeSessionFileForScan(context, candidate, sessionId); - if (safeFile) { - return safeFile.filePath; - } - } - return undefined; -} - -async function discoverCliRecords( - context: ClaudeSessionScanContext, - records: Map, - sidechainIds: Set, -): Promise { - const { root } = context; - if (!context.resolvedRoot) { - // The root (or a parent) is gone. Entries are tagged with the logical root, so evict by that - // rather than a lexical containment test the canonical cache keys would never satisfy. - for (const [cachedPath, entry] of catalogDiscoveryCache) { - if (entry.root === root) { - catalogDiscoveryCache.delete(cachedPath); - } - } - return; - } - let discoveredFiles = 0; - let scannedBytes = 0; - let truncated = false; - const seenFilePaths = new Set(); - const candidates: Array<{ directory: string; name: string; sessionId: string }> = []; - collect: for (const { directory, childNames } of context.projectDirectories) { - for (const name of childNames) { - if (!name.endsWith(".jsonl")) { - continue; - } - if (discoveredFiles >= MAX_CATALOG_DISCOVERY_FILES) { - truncated = true; - break collect; - } - discoveredFiles += 1; - const sessionId = name.slice(0, -".jsonl".length); - if (sessionId) { - candidates.push({ directory, name, sessionId }); - } - } - } - const safeFiles = await mapConcurrent( - candidates, - CLAUDE_CATALOG_IO_CONCURRENCY, - async ({ directory, name, sessionId }) => - records.has(sessionId) || sidechainIds.has(sessionId) - ? undefined - : await safeSessionFileForScan(context, path.join(directory, name), sessionId), - ); - for (const [candidateIndex, candidate] of candidates.entries()) { - const { sessionId } = candidate; - // Resolve metadata concurrently, but make every semantic decision in the original directory - // order. In particular, duplicates and the global byte frontier must match a serial cold scan. - if (records.has(sessionId) || sidechainIds.has(sessionId)) { - continue; - } - const safeFile = safeFiles[candidateIndex]; - if (!safeFile) { - continue; - } - const { filePath, stat: fileStat } = safeFile; - seenFilePaths.add(filePath); - const cached = catalogDiscoveryCache.get(filePath); - // Claude transcripts only append while active, then stay static, so mtime+size+ino identify - // the parsed content (ino also rejects an atomic replacement that reused the same mtime/size), - // and sessionId ensures the record is served only under the filename-derived id it was built - // for. These files are owner-owned and append-only; a mid-scan read-permission revocation is - // not a state the Claude CLI produces, so a hit intentionally skips the open() re-check. - if ( - cached && - cached.root === root && - cached.mtimeMs === fileStat.mtimeMs && - cached.size === fileStat.size && - cached.ino === fileStat.ino && - cached.sessionId === sessionId && - // Only replay the cached record if a cold scan would also reach its metadata under the - // current remaining byte budget. Once earlier files grow, replaying a record whose original - // scan cost now crosses the frontier would surface a record a cold scan stops before; fall - // through to a bounded rescan instead so warm and cold discovery (and pagination) match. - scannedBytes + cached.scannedBytes <= MAX_CATALOG_METADATA_SCAN_BYTES - ) { - if (cached.sidechain) { - sidechainIds.add(sessionId); - } - if (cached.record) { - records.set(sessionId, cached.record); - } - // Cache hits read no transcript bytes, but they still charge the file's original scan cost - // so the byte-budget cutoff matches a cold scan; otherwise repeated calls would free budget - // and progressively discover more files. - scannedBytes += cached.scannedBytes; - if (scannedBytes >= MAX_CATALOG_METADATA_SCAN_BYTES) { - truncated = true; - break; - } - continue; - } - const handle = await fs.open(filePath, "r").catch(() => { - context.complete = false; - return undefined; - }); - if (!handle) { - continue; - } - let cacheable = false; - let fileScannedBytes = 0; - try { - const stat = await handle.stat(); - let aiTitle: string | undefined; - let pending = Buffer.alloc(0); - let fileOffset = 0; - let stopFile = false; - const inspectLine = (line: Buffer): boolean => { - let raw: unknown; - try { - raw = JSON.parse(line.toString("utf8")) as unknown; - } catch { - return false; - } - if (!isRecord(raw) || raw.sessionId !== sessionId) { - return false; - } - if (raw.type === "ai-title") { - aiTitle = readBoundedString(raw.aiTitle, 500) ?? aiTitle; - return false; - } - if (typeof raw.entrypoint === "string" && !isCliEntrypoint(raw.entrypoint)) { - return true; - } - if (isCliEntrypoint(raw.entrypoint) && raw.isSidechain === true) { - sidechainIds.add(sessionId); - return true; - } - if ( - !isCliEntrypoint(raw.entrypoint) || - raw.type !== "user" || - !isRecord(raw.message) || - raw.message.role !== "user" - ) { - return false; - } - const fragments: string[] = []; - collectTranscriptText(raw.message.content, fragments); - const firstPrompt = readBoundedString(fragments[0], 500); - const createdAt = parseClaudeCatalogTimestampMs(raw.timestamp); - records.set(sessionId, { - threadId: sessionId, - name: aiTitle ?? firstPrompt ?? null, - cwd: readBoundedString(raw.cwd, MAX_STRING_LENGTH), - status: "stored", - ...(createdAt !== undefined ? { createdAt } : {}), - updatedAt: stat.mtimeMs, - recencyAt: stat.mtimeMs, - source: "claude-cli", - modelProvider: "anthropic", - ...(readBoundedString(raw.version, 256) - ? { cliVersion: readBoundedString(raw.version, 256) } - : {}), - ...(readBoundedString(raw.gitBranch, 500) - ? { gitBranch: readBoundedString(raw.gitBranch, 500) } - : {}), - archived: false, - filePath, - }); - return true; - }; - while ( - !stopFile && - fileOffset < stat.size && - fileOffset < CLAUDE_METADATA_PREFIX_BYTES && - scannedBytes < MAX_CATALOG_METADATA_SCAN_BYTES - ) { - const size = Math.min( - CLAUDE_METADATA_READ_CHUNK_BYTES, - stat.size - fileOffset, - CLAUDE_METADATA_PREFIX_BYTES - fileOffset, - MAX_CATALOG_METADATA_SCAN_BYTES - scannedBytes, - ); - const chunk = Buffer.allocUnsafe(size); - const { bytesRead } = await handle.read(chunk, 0, size, fileOffset); - if (bytesRead === 0) { - break; - } - fileOffset += bytesRead; - scannedBytes += bytesRead; - pending = pending.length - ? Buffer.concat([pending, chunk.subarray(0, bytesRead)]) - : chunk.subarray(0, bytesRead); - let newline: number; - while (!stopFile && (newline = pending.indexOf(0x0a)) >= 0) { - stopFile = inspectLine(pending.subarray(0, newline)); - pending = pending.subarray(newline + 1); - } - } - if (!stopFile && fileOffset >= stat.size && pending.length > 0) { - inspectLine(pending); - } - // A read whose chunk was capped by the remaining global budget stops on a smaller boundary - // than a cold scan would, so its fileOffset undercounts the true unconstrained scan cost. - // Don't cache such an entry: replaying its low cost later (with more budget free) would let - // the warm scan cross the frontier and surface sessions a cold scan omits. - const budgetConstrained = scannedBytes >= MAX_CATALOG_METADATA_SCAN_BYTES; - cacheable = - !budgetConstrained && - (stopFile || fileOffset >= stat.size || fileOffset >= CLAUDE_METADATA_PREFIX_BYTES); - fileScannedBytes = fileOffset; - } finally { - await handle.close(); - } - // Negative and sidechain-only results are cached too; unchanged files should not be reparsed. - if (cacheable) { - cacheCatalogDiscovery(filePath, { - root, - mtimeMs: fileStat.mtimeMs, - size: fileStat.size, - ino: fileStat.ino, - sessionId, - scannedBytes: fileScannedBytes, - record: records.get(sessionId) ?? null, - sidechain: sidechainIds.has(sessionId), - }); - } - if (scannedBytes >= MAX_CATALOG_METADATA_SCAN_BYTES) { - truncated = true; - break; - } - } - if (!truncated) { - // A complete scan is authoritative for this root: drop any of its entries not seen this pass. - for (const [cachedPath, entry] of catalogDiscoveryCache) { - if (entry.root === root && !seenFilePaths.has(cachedPath)) { - catalogDiscoveryCache.delete(cachedPath); - } - } - } -} - -async function scanClaudeSessions( - homeDir: string, - snapshot: ClaudeProjectsTreeSnapshot, - includeDesktop: boolean, -): Promise<{ records: CatalogRecord[]; complete: boolean }> { - const context: ClaudeSessionScanContext = { ...snapshot, complete: true, safeFiles: new Map() }; - const [indexed, desktop] = await Promise.all([ - readIndexRecords(context), - includeDesktop - ? readDesktopMetadata(homeDir) - : Promise.resolve({ active: new Map(), archived: new Set() }), - ]); - const records = indexed.records; - await discoverCliRecords(context, records, indexed.sidechainIds); - for (const sessionId of desktop.archived) { - records.delete(sessionId); - } - for (const [sessionId, metadata] of desktop.active) { - if (indexed.sidechainIds.has(sessionId)) { - continue; - } - const existing = records.get(sessionId); - const filePath = existing?.filePath ?? (await locateSessionFile(context, sessionId)); - if (!filePath) { - continue; - } - const createdAt = parseClaudeCatalogTimestampMs(metadata.createdAt) ?? existing?.createdAt; - const updatedAt = parseClaudeCatalogTimestampMs(metadata.lastActivityAt) ?? existing?.updatedAt; - const customGroup = readBoundedString(metadata.customGroup, 500); - const pullRequest = desktopPullRequestSummary(metadata); - records.set(sessionId, { - ...(existing ?? { - threadId: sessionId, - status: "stored" as const, - modelProvider: "anthropic" as const, - archived: false as const, - }), - name: readBoundedString(metadata.title, 500) ?? existing?.name ?? null, - cwd: - readBoundedString(metadata.cwd, MAX_STRING_LENGTH) ?? - readBoundedString(metadata.originCwd, MAX_STRING_LENGTH) ?? - existing?.cwd, - ...(createdAt !== undefined ? { createdAt } : {}), - ...(updatedAt !== undefined ? { updatedAt, recencyAt: updatedAt } : {}), - ...(customGroup ? { customGroup } : {}), - ...(pullRequest ? { pullRequest } : {}), - source: "claude-desktop", - filePath, - }); - } - return { - records: [...records.values()].toSorted((left, right) => { - const recency = - (right.recencyAt ?? right.updatedAt ?? 0) - (left.recencyAt ?? left.updatedAt ?? 0); - return recency || left.threadId.localeCompare(right.threadId); - }), - complete: context.complete, - }; -} - -async function listClaudeSessions( - homeDir = currentHomeDir(), - options: { forceRefresh?: boolean; configDir?: string; includeDesktop?: boolean } = {}, -): Promise { - const root = projectsDir(homeDir, options.configDir); - const includeDesktop = options.includeDesktop !== false; - const cacheKey = `${root}\0${includeDesktop ? "desktop" : "cli"}`; - const [treeSnapshot, desktopStoreAvailable] = await Promise.all([ - readProjectsTreeSnapshot(root), - includeDesktop ? desktopSessionStoreAvailable(homeDir) : Promise.resolve(false), - ]); - const now = Date.now(); - const cached = claudeSessionScanCache.get(cacheKey); - // Child membership + file mtime/size signatures invalidate CLI rows on the next poll; five minutes - // backstops metadata anomalies. Desktop has a 60s bound when its macOS store exists; Linux skips it. - // Specific-thread force refresh bypasses both, or a stale page could hide a just-created session. - if ( - options.forceRefresh !== true && - cached && - cached.treeStamp === treeSnapshot.treeStamp && - cached.hardExpiresAt > now && - cached.desktopStoreAvailable === desktopStoreAvailable && - (!desktopStoreAvailable || cached.desktopExpiresAt > now) - ) { - setBoundedCache( - claudeSessionScanCache, - cacheKey, - cached, - MAX_CLAUDE_SESSION_SCAN_CACHE_ENTRIES, - ); - return await cached.records; - } - const scan = scanClaudeSessions(homeDir, treeSnapshot, includeDesktop); - let scanComplete = true; - const records = scan.then((result) => { - scanComplete = result.complete; - return result.records; - }); - const entry = { - treeStamp: treeSnapshot.treeStamp, - hardExpiresAt: now + CLAUDE_SESSION_SCAN_HARD_TTL_MS, - desktopStoreAvailable, - desktopExpiresAt: now + CLAUDE_DESKTOP_SCAN_TTL_MS, - records, - }; - setBoundedCache(claudeSessionScanCache, cacheKey, entry, MAX_CLAUDE_SESSION_SCAN_CACHE_ENTRIES); - try { - const result = await records; - if (!scanComplete && claudeSessionScanCache.get(cacheKey) === entry) { - // Partial results still serve this caller, but retry within 15s so transient per-file I/O - // cannot hide recovered sessions behind the five-minute unchanged-tree backstop. - entry.hardExpiresAt = Date.now() + CLAUDE_PARTIAL_SCAN_TTL_MS; - } - return result; - } catch (error) { - if (claudeSessionScanCache.get(cacheKey) === entry) { - claudeSessionScanCache.delete(cacheKey); - } - throw error; - } -} - -function encodeOffset(offset: number): string { - return Buffer.from(JSON.stringify({ offset }), "utf8").toString("base64url"); -} - -function decodeOffset(cursor: string | undefined, label: string): number { - if (cursor === undefined) { - return 0; - } - if (!isExactClaudeSessionCursor(cursor)) { - throw new ClaudeCatalogParamsError(`${label} cursor is invalid`); - } - try { - const parsed = JSON.parse(Buffer.from(cursor, "base64url").toString("utf8")) as unknown; - if ( - !isRecord(parsed) || - !Number.isSafeInteger(parsed.offset) || - (parsed.offset as number) < 0 - ) { - throw new Error("invalid offset"); - } - return parsed.offset as number; - } catch (error) { - throw new ClaudeCatalogParamsError(`${label} cursor is invalid`, { cause: error }); - } -} - -function readLimit(value: unknown, fallback: number, max: number): number { - if (value === undefined) { - return fallback; - } - if (!Number.isInteger(value) || (value as number) < 1 || (value as number) > max) { - throw new ClaudeCatalogParamsError(`limit must be an integer from 1 to ${max}`); - } - return value as number; -} - -function readRequiredCursor(value: unknown, message: string): string { - if (!isExactClaudeSessionCursor(value)) { - throw new ClaudeCatalogParamsError(message); - } - return value; -} - -function readOptionalCursor(value: unknown, label: string): string | undefined { - if (value === undefined) { - return undefined; - } - return readRequiredCursor(value, `${label} cursor is invalid`); -} - -function readListParams(value: unknown): { - cursor?: string; - limit: number; - searchTerm?: string; -} { - if (value === undefined || value === null) { - return { limit: DEFAULT_PAGE_LIMIT }; - } - if (!isRecord(value)) { - throw new ClaudeCatalogParamsError("Claude session catalog parameters must be an object"); - } - const allowed = new Set(["cursor", "limit", "searchTerm"]); - const unknown = Object.keys(value).find((key) => !allowed.has(key)); - if (unknown) { - throw new ClaudeCatalogParamsError(`unknown Claude session catalog parameter: ${unknown}`); - } - const cursor = readOptionalCursor(value.cursor, "catalog"); - const searchTerm = readBoundedString(value.searchTerm, MAX_SEARCH_LENGTH); - return { - limit: readLimit(value.limit, DEFAULT_PAGE_LIMIT, MAX_PAGE_LIMIT), - ...(cursor ? { cursor } : {}), - ...(searchTerm ? { searchTerm } : {}), - }; -} - -export async function listLocalClaudeSessionPage( - value: unknown, - homeDir?: string, - scanOptions?: { configDir?: string; includeDesktop?: boolean }, -): Promise { - const resolvedHome = homeDir ?? currentHomeDir(); - const resolvedScanOptions = - scanOptions ?? (homeDir === undefined ? gatewayClaudeScanOptions(true) : {}); - const params = readListParams(value); - const offset = decodeOffset(params.cursor, "catalog"); - const search = params.searchTerm?.toLocaleLowerCase(); - const records = (await listClaudeSessions(resolvedHome, resolvedScanOptions)).filter((record) => { - if (!search) { - return true; - } - return [record.name, record.cwd, record.gitBranch, record.threadId].some((candidate) => - candidate?.toLocaleLowerCase().includes(search), - ); - }); - const page = records - .slice(offset, offset + params.limit) - .map(({ filePath: _filePath, ...record }) => record); - const nextOffset = offset + page.length; - return { - sessions: page, - ...(nextOffset < records.length ? { nextCursor: encodeOffset(nextOffset) } : {}), - }; -} - -function readTranscriptParams( - value: unknown, - options: { includeHostId?: boolean } = {}, -): { threadId: string; cursor?: string; limit: number } { - if (!isRecord(value)) { - throw new ClaudeCatalogParamsError("Claude session read parameters must be an object"); - } - const allowed = new Set([ - "threadId", - "cursor", - "limit", - ...(options.includeHostId ? ["hostId"] : []), - ]); - const unknown = Object.keys(value).find((key) => !allowed.has(key)); - if (unknown) { - throw new ClaudeCatalogParamsError(`unknown Claude session read parameter: ${unknown}`); - } - const threadId = readBoundedString(value.threadId, 256); - if (!threadId || !/^[A-Za-z0-9._:-]+$/.test(threadId)) { - throw new ClaudeCatalogParamsError("threadId is invalid"); - } - const cursor = readOptionalCursor(value.cursor, "transcript"); - return { - threadId, - limit: readLimit(value.limit, DEFAULT_TRANSCRIPT_LIMIT, MAX_TRANSCRIPT_LIMIT), - ...(cursor ? { cursor } : {}), - }; -} - -export async function readLocalClaudeTranscriptPage( - value: unknown, - homeDir?: string, - scanOptions?: { configDir?: string; includeDesktop?: boolean }, -): Promise> { - const resolvedHome = homeDir ?? currentHomeDir(); - const resolvedScanOptions = - scanOptions ?? (homeDir === undefined ? gatewayClaudeScanOptions(true) : {}); - const params = readTranscriptParams(value); - let filePath = (await listClaudeSessions(resolvedHome, resolvedScanOptions)).find( - (record) => record.threadId === params.threadId, - )?.filePath; - if (!filePath) { - // A just-created session can race the stamp snapshot. Specific reads must retry against disk so - // opening a new thread never fails only because the assembled catalog is still warm. - filePath = ( - await listClaudeSessions(resolvedHome, { ...resolvedScanOptions, forceRefresh: true }) - ).find((record) => record.threadId === params.threadId)?.filePath; - } - if (!filePath) { - throw new ClaudeCatalogParamsError("Claude session is unavailable"); - } - const handle = await fs.open(filePath, "r"); - try { - const stat = await handle.stat(); - const requestedEnd = params.cursor ? decodeOffset(params.cursor, "transcript") : stat.size; - if (requestedEnd > stat.size) { - throw new ClaudeCatalogParamsError("transcript cursor is invalid"); - } - let position = requestedEnd; - let scanned = 0; - let fragments: Buffer[] = []; - const found: Array<{ item: ClaudeTranscriptItem; start: number }> = []; - while (position > 0 && scanned < MAX_TRANSCRIPT_SCAN_BYTES && found.length <= params.limit) { - const size = Math.min( - TRANSCRIPT_READ_CHUNK_BYTES, - position, - MAX_TRANSCRIPT_SCAN_BYTES - scanned, - ); - position -= size; - const chunk = Buffer.allocUnsafe(size); - // Positional reads may return short, so complete the bounded window. - // A zero-byte read before it fills means the file changed after stat. - let filled = 0; - while (filled < size) { - const { bytesRead } = await handle.read(chunk, filled, size - filled, position + filled); - if (bytesRead === 0) { - throw new Error("Claude transcript changed while it was being read"); - } - filled += bytesRead; - } - scanned += filled; - let right = filled; - for (let index = filled - 1; index >= 0; index -= 1) { - if (chunk[index] !== 0x0a) { - continue; - } - const segment = chunk.subarray(index + 1, right); - if (segment.length > 0 || fragments.length > 0) { - const line = Buffer.concat([segment, ...fragments.toReversed()]); - const item = parseTranscriptLine(line, readBoundedString); - fragments = []; - if (item) { - found.push({ item, start: position + index + 1 }); - if (found.length > params.limit) { - break; - } - } - } - right = index; - } - if (found.length > params.limit) { - break; - } - const prefix = chunk.subarray(0, right); - if (position === 0) { - if (prefix.length > 0 || fragments.length > 0) { - const line = Buffer.concat([prefix, ...fragments.toReversed()]); - const item = parseTranscriptLine(line, readBoundedString); - if (item) { - found.push({ item, start: 0 }); - } - } - fragments = []; - } else if (prefix.length > 0) { - fragments.push(prefix); - } - } - if (position > 0 && found.length < params.limit) { - throw new Error("Claude transcript page exceeded the safe scan limit"); - } - const requested = found.slice(0, params.limit); - const selected: typeof requested = []; - let selectedBytes = 0; - for (const entry of requested) { - const itemBytes = Buffer.byteLength(JSON.stringify(entry.item), "utf8"); - if ( - selected.length > 0 && - selectedBytes + itemBytes > MAX_TRANSCRIPT_PAGE_BYTES - 64 * 1024 - ) { - break; - } - selected.push(entry); - selectedBytes += itemBytes; - } - const earliestStart = selected.at(-1)?.start; - const hasEarlierItems = selected.length < found.length || position > 0; - return { - threadId: params.threadId, - // Match the Codex session-page contract: newest item first on the wire; - // the shared UI prepends each page after restoring chronological order. - items: selected.map((entry) => entry.item), - ...(hasEarlierItems && earliestStart !== undefined && earliestStart > 0 - ? { nextCursor: encodeOffset(earliestStart) } - : {}), - }; - } finally { - await handle.close(); - } -} - -function readNodePageCursor( - value: Record, - invalidPageMessage: string, -): string | undefined { - if (!("nextCursor" in value)) { - return undefined; - } - if (!isExactClaudeSessionCursor(value.nextCursor)) { - throw new Error(invalidPageMessage); - } - return value.nextCursor; -} - -function parseCatalogPage(value: unknown): ClaudeSessionCatalogPage { - if ( - !isRecord(value) || - !Array.isArray(value.sessions) || - value.sessions.length > MAX_PAGE_LIMIT - ) { - throw new Error("Claude node returned an invalid session page"); - } - const sessions = value.sessions.map((candidate): ClaudeSessionCatalogSession => { - if (!isRecord(candidate)) { - throw new Error("Claude node returned an invalid session"); - } - const threadId = readBoundedString(candidate.threadId, 256); - const source = candidate.source; - if ( - !threadId || - candidate.archived !== false || - candidate.status !== "stored" || - (source !== "claude-cli" && source !== "claude-desktop") || - candidate.modelProvider !== "anthropic" - ) { - throw new Error("Claude node returned an invalid session"); - } - const parseStringField = (key: string, maxLength = MAX_STRING_LENGTH): string | undefined => { - if (!(key in candidate)) { - return undefined; - } - const parsed = readBoundedString(candidate[key], maxLength); - if (!parsed) { - throw new Error("Claude node returned an invalid session"); - } - return parsed; - }; - const parseNumberField = (key: string, nullable = false): number | null | undefined => { - if (!(key in candidate)) { - return undefined; - } - if (nullable && candidate[key] === null) { - return null; - } - const parsed = candidate[key]; - if (typeof parsed !== "number" || !Number.isFinite(parsed)) { - throw new Error("Claude node returned an invalid session"); - } - return parsed; - }; - let name: string | null | undefined; - if (candidate.name === null) { - name = null; - } else { - name = parseStringField("name", 500); - } - const cwd = parseStringField("cwd"); - const createdAt = parseNumberField("createdAt") as number | undefined; - const updatedAt = parseNumberField("updatedAt") as number | undefined; - const recencyAt = parseNumberField("recencyAt", true); - const cliVersion = parseStringField("cliVersion", 256); - const gitBranch = parseStringField("gitBranch", 500); - const pullRequest = parsePullRequestSummary(candidate.pullRequest); - return { - threadId, - status: "stored", - source, - modelProvider: "anthropic", - archived: false, - ...(name !== undefined ? { name } : {}), - ...(cwd ? { cwd } : {}), - ...(createdAt !== undefined ? { createdAt } : {}), - ...(updatedAt !== undefined ? { updatedAt } : {}), - ...(recencyAt !== undefined ? { recencyAt } : {}), - ...(cliVersion ? { cliVersion } : {}), - ...(gitBranch ? { gitBranch } : {}), - ...(pullRequest ? { pullRequest } : {}), - }; - }); - const nextCursor = readNodePageCursor(value, "Claude node returned an invalid session page"); - return { sessions, ...(nextCursor ? { nextCursor } : {}) }; -} - -function unwrapNodePayload(value: unknown): unknown { - if (isRecord(value) && typeof value.payloadJSON === "string") { - return JSON.parse(value.payloadJSON) as unknown; - } - return value; -} - -function parseGatewayQuery(value: unknown): { - search?: string; - limitPerHost: number; - hostIds?: string[]; - cursors?: Record; -} { - if (value === undefined || value === null) { - return { limitPerHost: DEFAULT_PAGE_LIMIT }; - } - if (!isRecord(value)) { - throw new ClaudeCatalogParamsError("Claude session catalog parameters must be an object"); - } - const allowed = new Set(["search", "limitPerHost", "hostIds", "cursors"]); - const unknown = Object.keys(value).find((key) => !allowed.has(key)); - if (unknown) { - throw new ClaudeCatalogParamsError(`unknown Claude session catalog parameter: ${unknown}`); - } - const search = readBoundedString(value.search, MAX_SEARCH_LENGTH); - let hostIds: string[] | undefined; - if (value.hostIds !== undefined) { - if (!Array.isArray(value.hostIds) || value.hostIds.length > MAX_HOSTS) { - throw new ClaudeCatalogParamsError("hostIds must be a bounded array"); - } - hostIds = [ - ...new Set( - value.hostIds.map((hostId) => { - const normalized = readBoundedString(hostId, 256); - if ( - !normalized || - (normalized !== CLAUDE_LOCAL_SESSION_HOST_ID && !normalized.startsWith("node:")) - ) { - throw new ClaudeCatalogParamsError("hostId is invalid"); - } - return normalized; - }), - ), - ]; - } - let cursors: Record | undefined; - if (value.cursors !== undefined) { - if (!isRecord(value.cursors) || Object.keys(value.cursors).length > MAX_HOSTS) { - throw new ClaudeCatalogParamsError("cursors must be a bounded object"); - } - cursors = Object.fromEntries( - Object.entries(value.cursors).map(([hostId, cursor]) => { - return [hostId, readRequiredCursor(cursor, `cursor for ${hostId} is invalid`)]; - }), - ); - } - return { - limitPerHost: readLimit(value.limitPerHost, DEFAULT_PAGE_LIMIT, MAX_PAGE_LIMIT), - ...(search ? { search } : {}), - ...(hostIds ? { hostIds } : {}), - ...(cursors ? { cursors } : {}), - }; -} - -async function listClaudeSessionCatalog(params: { - runtime: PluginRuntime; - query?: unknown; - allowProcessHomeFallback?: boolean; - listNodes?: Parameters[0]["listNodes"]; - onHost?: (host: ClaudeSessionCatalogHost) => void; -}): Promise { - const query = parseGatewayQuery(params.query); - const requested = query.hostIds ? new Set(query.hostIds) : undefined; - const scanOptions = gatewayClaudeScanOptions(params.allowProcessHomeFallback); - const localHosts: Promise[] = - (params.allowProcessHomeFallback !== false || scanOptions.configDir !== undefined) && - (!requested || requested.has(CLAUDE_LOCAL_SESSION_HOST_ID)) - ? [ - (async () => { - try { - return { - hostId: CLAUDE_LOCAL_SESSION_HOST_ID, - label: "Local Claude", - kind: "gateway", - connected: true, - ...(await listLocalClaudeSessionPage( - { - limit: query.limitPerHost, - ...(query.search ? { searchTerm: query.search } : {}), - ...(query.cursors?.[CLAUDE_LOCAL_SESSION_HOST_ID] !== undefined - ? { cursor: query.cursors[CLAUDE_LOCAL_SESSION_HOST_ID] } - : {}), - }, - currentHomeDir(), - scanOptions, - )), - }; - } catch { - return { - hostId: CLAUDE_LOCAL_SESSION_HOST_ID, - label: "Local Claude", - kind: "gateway", - connected: true, - sessions: [], - error: { - code: "LOCAL_READ_FAILED", - message: "Local Claude sessions are unavailable", - }, - }; - } - })(), - ] - : []; - for (const host of localHosts) { - if (params.onHost) { - void host.then(params.onHost).catch(() => undefined); - } - } - const wantsNodes = !requested || query.hostIds?.some((hostId) => hostId.startsWith("node:")); - if (!wantsNodes) { - return { hosts: await Promise.all(localHosts) }; - } - let nodes: Awaited>["nodes"]; - try { - nodes = (await (params.listNodes?.() ?? params.runtime.nodes.list())).nodes; - } catch (error) { - const registryHost: ClaudeSessionCatalogHost = { - hostId: "node:registry", - label: "Paired nodes", - kind: "node", - connected: false, - sessions: [], - error: createNodeListFailedError(error), - }; - params.onHost?.(registryHost); - return { - hosts: [...(await Promise.all(localHosts)), registryHost], - }; - } - const eligible = nodes - .filter( - (node) => - node.gatewayLocal !== true && - node.commands?.includes(CLAUDE_SESSIONS_LIST_COMMAND) && - (!requested || requested.has(`node:${node.nodeId}`)), - ) - .slice(0, MAX_HOSTS - localHosts.length) - .toSorted((left, right) => resolveNodeLabel(left).localeCompare(resolveNodeLabel(right))); - const nodeHosts = await Promise.all( - eligible.map(async (node): Promise => { - const hostId = `node:${node.nodeId}`; - const common = { - hostId, - label: resolveNodeLabel(node), - kind: "node" as const, - connected: node.connected === true, - nodeId: node.nodeId, - canContinueClaude: - node.commands?.includes(CLAUDE_SESSION_READ_COMMAND) === true && - node.commands.includes(CLAUDE_CLI_NODE_RUN_COMMAND) && - node.invocableCommands?.includes(CLAUDE_SESSIONS_LIST_COMMAND) === true && - node.invocableCommands.includes(CLAUDE_SESSION_READ_COMMAND) && - node.invocableCommands.includes(CLAUDE_CLI_NODE_RUN_COMMAND), - ...catalogTerminal.claudeNodeTerminalCapability(node), - }; - if (node.connected !== true) { - const host: ClaudeSessionCatalogHost = Object.assign({}, common, { - sessions: [], - error: { code: "NODE_OFFLINE", message: "Paired node is offline" }, - }); - params.onHost?.(host); - return host; - } - const eventualHost = Promise.resolve() - .then(async () => { - const raw = await params.runtime.nodes.invoke({ - nodeId: node.nodeId, - command: CLAUDE_SESSIONS_LIST_COMMAND, - params: { - limit: query.limitPerHost, - ...(query.search ? { searchTerm: query.search } : {}), - ...(query.cursors?.[hostId] !== undefined ? { cursor: query.cursors[hostId] } : {}), - }, - timeoutMs: NODE_INVOKE_TIMEOUT_MS, - scopes: ["operator.write"], - }); - return Object.assign({}, common, parseCatalogPage(unwrapNodePayload(raw))); - }) - .catch( - (): ClaudeSessionCatalogHost => - Object.assign({}, common, { - sessions: [], - error: { - code: "NODE_INVOKE_FAILED", - message: "Paired node Claude sessions are unavailable", - }, - }), - ); - if (params.onHost) { - // The fail-soft response can finish first; the original node invoke still - // publishes its authoritative host page whenever cold discovery completes. - void eventualHost.then(params.onHost).catch(() => undefined); - } - try { - return await withTimeout(eventualHost, NODE_CATALOG_LIST_RESPONSE_TIMEOUT_MS, { - message: "paired node Claude session catalog timed out", - }); - } catch { - return Object.assign({}, common, { - sessions: [], - error: { - code: "NODE_INVOKE_FAILED", - message: "Paired node Claude sessions are unavailable", - }, - }); - } - }), - ); - return { hosts: [...(await Promise.all(localHosts)), ...nodeHosts] }; -} - -async function readClaudeSessionTranscript(params: { - runtime: PluginRuntime; - hostId: string; - threadId: string; - cursor?: string; - limit: number; - allowProcessHomeFallback?: boolean; -}): Promise { - const cursor = readOptionalCursor(params.cursor, "transcript"); - if (params.hostId === CLAUDE_LOCAL_SESSION_HOST_ID) { - assertClaudeLocalAccess(params.hostId, params.allowProcessHomeFallback); - return { - hostId: params.hostId, - label: "Local Claude", - ...(await readLocalClaudeTranscriptPage( - { - threadId: params.threadId, - limit: params.limit, - ...(cursor !== undefined ? { cursor } : {}), - }, - currentHomeDir(), - gatewayClaudeScanOptions(params.allowProcessHomeFallback), - )), - }; - } - if (!params.hostId.startsWith("node:")) { - throw new ClaudeCatalogParamsError("hostId is invalid"); - } - const nodeId = params.hostId.slice("node:".length); - const node = (await params.runtime.nodes.list()).nodes.find( - (candidate) => - candidate.nodeId === nodeId && - candidate.connected === true && - candidate.commands?.includes(CLAUDE_SESSION_READ_COMMAND), - ); - if (!node) { - throw new ClaudeCatalogParamsError("paired-node Claude session host is unavailable"); - } - const raw = await params.runtime.nodes.invoke({ - nodeId, - command: CLAUDE_SESSION_READ_COMMAND, - params: { - threadId: params.threadId, - limit: params.limit, - ...(cursor !== undefined ? { cursor } : {}), - }, - timeoutMs: NODE_INVOKE_TIMEOUT_MS, - scopes: ["operator.write"], - }); - const page = unwrapNodePayload(raw); - if ( - !isRecord(page) || - !Array.isArray(page.items) || - page.items.length > MAX_TRANSCRIPT_LIMIT || - page.items.some((item) => !isRecord(item) || typeof item.type !== "string") || - page.threadId !== params.threadId || - Buffer.byteLength(JSON.stringify(page), "utf8") > MAX_TRANSCRIPT_PAGE_BYTES - ) { - throw new Error("Claude node returned an invalid transcript page"); - } - const nextCursor = readNodePageCursor(page, "Claude node returned an invalid transcript page"); - return { - hostId: params.hostId, - label: resolveNodeLabel(node), - threadId: params.threadId, - items: page.items as ClaudeTranscriptItem[], - ...(nextCursor !== undefined ? { nextCursor } : {}), - }; -} - -function assertClaudeLocalAccess(hostId: string, allowProcessHomeFallback?: boolean): void { - if ( - hostId === CLAUDE_LOCAL_SESSION_HOST_ID && - allowProcessHomeFallback === false && - configuredClaudeConfigDir() === undefined - ) { - throw new ClaudeCatalogParamsError("local Claude sessions are unavailable in isolated state"); - } -} - -async function readBoundedClaudeHistory(params: { - runtime: PluginRuntime; - hostId: string; - threadId: string; - allowProcessHomeFallback?: boolean; -}): Promise { - const items: ClaudeTranscriptItem[] = []; - let cursor: string | undefined; - let bytes = 0; - while (items.length < CLAUDE_HISTORY_IMPORT_MAX_ITEMS) { - const page = await readClaudeSessionTranscript({ - runtime: params.runtime, - hostId: params.hostId, - threadId: params.threadId, - limit: Math.min(MAX_TRANSCRIPT_LIMIT, CLAUDE_HISTORY_IMPORT_MAX_ITEMS - items.length), - allowProcessHomeFallback: params.allowProcessHomeFallback, - ...(cursor ? { cursor } : {}), - }); - for (const item of page.items) { - const itemBytes = Buffer.byteLength(JSON.stringify(item), "utf8"); - if (items.length > 0 && bytes + itemBytes > CLAUDE_HISTORY_IMPORT_MAX_BYTES) { - return items; - } - items.push(item); - bytes += itemBytes; - } - if (!page.nextCursor || page.nextCursor === cursor) { - break; - } - cursor = page.nextCursor; - } - return items; -} - -async function resolveNodeClaudeRecord(params: { - runtime: PluginRuntime; - nodeId: string; - threadId: string; -}): Promise { - let cursor: string | undefined; - for (let pageIndex = 0; pageIndex < 100; pageIndex += 1) { - const raw = await params.runtime.nodes.invoke({ - nodeId: params.nodeId, - command: CLAUDE_SESSIONS_LIST_COMMAND, - params: { - limit: MAX_PAGE_LIMIT, - searchTerm: params.threadId, - ...(cursor ? { cursor } : {}), - }, - timeoutMs: NODE_INVOKE_TIMEOUT_MS, - scopes: ["operator.write"], - }); - const page = parseCatalogPage(unwrapNodePayload(raw)); - const record = page.sessions.find((candidate) => candidate.threadId === params.threadId); - if (record) { - return record; - } - if (!page.nextCursor || page.nextCursor === cursor) { - break; - } - cursor = page.nextCursor; - } - throw new ClaudeCatalogParamsError("Claude session is unavailable on the paired node"); -} - -async function continueClaudeSession( - api: OpenClawPluginApi, - agentId: string, - hostId: string, - threadId: string, - allowProcessHomeFallback?: boolean, -): Promise<{ sessionKey: string }> { - const scanOptions = gatewayClaudeScanOptions(allowProcessHomeFallback); - const sourceKey = adoptedSourceKey(hostId, threadId); - const operationKey = `${agentId}\0${sourceKey}`; - const linkSession = async (sessionKey: string, history?: ClaudeTranscriptItem[]) => - await upstream.linkContinued({ - sessionKey, - hostId, - threadId, - ...(history ? { history } : {}), - listLocalSessions: () => listClaudeSessions(currentHomeDir(), scanOptions), - readRemote: async () => - ( - await readClaudeSessionTranscript({ - runtime: api.runtime, - hostId, - threadId, - limit: 1, - allowProcessHomeFallback, - }) - ).items, - }); - const existing = listBoundClaudeSessions(api, agentId).get(sourceKey); - if (existing) { - return await linkSession(existing); - } - let history: ClaudeTranscriptItem[] | undefined; - return await continueClaudeAdoption({ - sourceKey: operationKey, - findExisting: () => listBoundClaudeSessions(api, agentId).get(sourceKey), - create: async () => { - let nodeId: string | undefined; - let record: ClaudeSessionCatalogSession | undefined; - if (hostId === CLAUDE_LOCAL_SESSION_HOST_ID) { - record = (await listClaudeSessions(currentHomeDir(), scanOptions)).find( - (candidate) => candidate.threadId === threadId, - ); - if (!record || !isResumableClaudeSource(record.source)) { - throw new ClaudeCatalogParamsError("only local Claude Code sessions can be continued"); - } - } else if (hostId.startsWith("node:")) { - nodeId = hostId.slice("node:".length); - const node = (await api.runtime.nodes.list()).nodes.find( - (candidate) => - candidate.nodeId === nodeId && - candidate.connected === true && - candidate.commands?.includes(CLAUDE_SESSIONS_LIST_COMMAND) && - candidate.commands.includes(CLAUDE_SESSION_READ_COMMAND) && - candidate.commands.includes(CLAUDE_CLI_NODE_RUN_COMMAND) && - candidate.invocableCommands?.includes(CLAUDE_SESSIONS_LIST_COMMAND) === true && - candidate.invocableCommands.includes(CLAUDE_SESSION_READ_COMMAND) && - candidate.invocableCommands.includes(CLAUDE_CLI_NODE_RUN_COMMAND), - ); - if (!node) { - throw new ClaudeCatalogParamsError( - "paired node does not permit Claude CLI session continuation", - ); - } - // Node rows stay CLI-only: desktop transcripts on nodes have no - // node-side run command and remain view-only. - record = await resolveNodeClaudeRecord({ runtime: api.runtime, nodeId, threadId }); - if (!record || record.source !== "claude-cli") { - throw new ClaudeCatalogParamsError("only Claude CLI sessions can be continued"); - } - } else { - throw new ClaudeCatalogParamsError("hostId is invalid"); - } - if (hostId === CLAUDE_LOCAL_SESSION_HOST_ID) { - const source = await fs.stat((record as CatalogRecord).filePath).catch(() => undefined); - if (!source?.isFile()) { - throw new ClaudeCatalogParamsError("Claude session transcript is unavailable"); - } - } - // Narrowed local: afterCreate below needs a definite array, while the outer - // `history` stays optional so complete() can link without a reread on races. - const loadedHistory = await readBoundedClaudeHistory({ - runtime: api.runtime, - hostId, - threadId, - allowProcessHomeFallback, - }); - history = loadedHistory; - const config = currentClaudeSessionCatalogConfig(api); - const adoptingAgentId = agentId; - // Adopt onto the model this agent actually routes to the CLI backend; the - // packaged default may not be routed or allowed in an existing config. - const model = - resolveClaudeCliRoutedModelId(config, adoptingAgentId) ?? - CLAUDE_CLI_DEFAULT_MODEL_REF.slice(`${CLAUDE_CLI_BACKEND_ID}/`.length); - const marker = { - sourceThreadId: threadId, - ...(hostId !== CLAUDE_LOCAL_SESSION_HOST_ID ? { sourceHostId: hostId } : {}), - }; - const created = await api.runtime.agent.session.createSessionEntry({ - cfg: config, - key: adoptedSessionKey(hostId, threadId), - agentId: adoptingAgentId, - recoverMatchingInitialEntry: true, - ...(record.name ? { label: record.name } : {}), - ...(record.cwd ? { spawnedCwd: record.cwd } : {}), - ...(nodeId ? { execNode: nodeId, ...(record.cwd ? { execCwd: record.cwd } : {}) } : {}), - initialEntry: { - cliBackendId: CLAUDE_CLI_BACKEND_ID, - model, - modelSelectionLocked: true, - pluginOwnerId: api.id, - cliSessionBinding: { sessionId: threadId, forceReuse: true, forkNextResume: true }, - pluginExtensions: { anthropic: { sessionCatalog: marker } }, - }, - afterCreate: async (entry) => { - await importClaudeHistory({ - items: loadedHistory, - threadId, - sessionId: entry.sessionId, - sessionKey: entry.key, - agentId: entry.agentId, - storePath: api.runtime.agent.session.resolveStorePath(config.session?.store, { - agentId: entry.agentId, - }), - ...(record.cwd ? { cwd: record.cwd } : {}), - config, - }); - return { pluginExtensions: { anthropic: { sessionCatalog: marker } } }; - }, - }); - return { sessionKey: created.key }; - }, - complete: async (continued) => await linkSession(continued.sessionKey, history), - }); -} +export { + listLocalClaudeSessionPage, + readLocalClaudeTranscriptPage, +} from "./session-catalog-listing.js"; function toGenericClaudeItem(item: ClaudeTranscriptItem): SessionCatalogTranscriptItem { const allowed = new Set([ @@ -2072,4 +208,3 @@ export function createClaudeSessionCatalogRuntime( }, }; } -/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */