From 4b4fcffd3e875e81a47c3acb916208d67663578e Mon Sep 17 00:00:00 2001 From: Hsin <2129830748@qq.com> Date: Sat, 22 Aug 2026 17:31:26 +0800 Subject: [PATCH 1/4] fix(desktop): isolate invalid optional MCP tools --- .../__tests__/runtime-host-client-uds.test.ts | 114 ++++++++++++ .../runtime-host-native-capabilities.test.ts | 175 ++++++++++++++++++ apps/desktop/src/main/runtime-host-boot.ts | 1 + .../main/runtime-host-desktop-candidate.ts | 1 + .../main/runtime-host-native-capabilities.ts | 147 +++++++++++++-- packages/runtime/package.json | 1 + packages/runtime/src/ai-sdk-backend.ts | 54 +----- .../runtime/src/json-schema-validation.ts | 52 ++++++ 8 files changed, 485 insertions(+), 60 deletions(-) create mode 100644 packages/runtime/src/json-schema-validation.ts diff --git a/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts b/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts index 77954d09d6..bb08d781bd 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts @@ -99,6 +99,120 @@ test('drives Desktop Session operations through a real Runtime Host connection', } }); +test('keeps the Desktop candidate usable when an optional MCP tool has an invalid schema', async () => { + const base = await mkdtemp(join(tmpdir(), 'maka-desktop-invalid-mcp-')); + let host: RuntimeHostKernel | undefined; + try { + const capability = await resolveStorageRoot({ path: base, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + const projected = session('session-invalid-mcp'); + host = await RuntimeHostKernel.start({ + owner, + idleGraceMs: 10_000, + composition: defineInteractiveRuntimeHostComposition(async () => ({ + handlers: handlers({ + 'client.capability.replace': async (input) => { + const mcp = input.offers.find((offer) => offer.offerId === 'desktop_mcp'); + assert.deepEqual(mcp?.tools.map(({ name }) => name), ['mcp_valid']); + return { + ok: true, + result: { registrationId: input.registrationId, revision: 1 }, + }; + }, + 'client.capability.unregister': async (input) => ({ + ok: true, + result: { registrationId: input.registrationId, revision: 2 }, + }), + 'session.catalog.query': async (input) => ({ + ok: true, + result: + input.kind === 'get' + ? { kind: 'session', session: input.sessionId === projected.id ? projected : null } + : { + kind: 'page', + revision: catalogRevision('1'), + sessions: [projected], + nextCursor: null, + }, + }), + }), + beginDrain() {}, + async recover() {}, + async close() {}, + })), + }); + const ipc = ipcHarness(); + const diagnostics: unknown[] = []; + const invalidTool = { + ...nativeTool(), + name: 'mcp_invalid', + parameters: { jsonSchema: { type: 'string' } }, + } as unknown as MakaTool; + const validTool = { + ...nativeTool(), + name: 'mcp_valid', + parameters: { jsonSchema: { type: 'object', properties: {} } }, + } as unknown as MakaTool; + const started = await startDesktopRuntimeHostCandidate({ + rootPath: base, + candidateEntrypoint: new URL('file:///unused-runtime-host-candidate.js'), + ipcMain: ipc, + workspaceRoot: base, + attachmentApprovals: createAttachmentApprovalRegistry(), + stat: async () => ({ size: 0 }), + resizeImage: async (bytes) => bytes, + nativeCapabilities: { + browserTools: [], + releaseBrowserSession() {}, + computerUseTools: Object.assign([], { + clearSession() {}, + }) as unknown as ComputerUseToolSet, + releaseComputerUseSession() {}, + additionalGroups: () => [ + { + offerId: 'desktop_mcp', + label: 'MCP', + description: 'MCP tools', + invalidToolPolicy: 'omit', + tools: [invalidTool, validTool], + }, + ], + }, + botRegistry: {} as BotRegistry, + resolveBotCreateTarget: async () => ({ + workspace: { kind: 'host_path', path: base }, + }), + resolveSessionCreateProject: async () => ({ kind: 'host_path', path: base }), + emitSessionsChanged() {}, + completeComputerUseTurn() {}, + onError: (error) => diagnostics.push(error), + createSessionCopyCleanup: () => ({ + ownCreation: (_creation, operation) => operation(), + cleanup: async () => undefined, + schedule: async () => undefined, + abandonOwner: async () => undefined, + recover: async () => ({ removed: [], failed: [] }), + }), + }); + assert.equal(started.kind, 'ready'); + if (started.kind !== 'ready') throw new Error('Desktop candidate did not start'); + const { candidate } = started; + ipc.setHost(candidate.client.hostId, ipc.epoch); + + assert.deepEqual( + ((await ipc.invoke('sessions:list')) as SessionCatalogProjection[]).map(({ id }) => id), + [projected.id], + ); + assert.equal(diagnostics.length, 1); + assert.match(String(diagnostics[0]), /desktop_mcp\/mcp_invalid/); + await candidate.close(); + } finally { + await host?.close().catch(() => undefined); + await rm(base, { recursive: true, force: true }); + } +}); + test('drives the renderer Session catalog facade through real UDS framing', async () => { const base = await mkdtemp(join(tmpdir(), 'maka-desktop-host-ipc-')); let host: RuntimeHostKernel | undefined; diff --git a/apps/desktop/src/main/__tests__/runtime-host-native-capabilities.test.ts b/apps/desktop/src/main/__tests__/runtime-host-native-capabilities.test.ts index df0a08d376..ef2c19a68f 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-native-capabilities.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-native-capabilities.test.ts @@ -2,6 +2,7 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import { buildComputerUseTools, type ComputerUseToolSet } from '@maka/runtime/computer-use-tools'; import { type CuDispatchBackend } from '@maka/runtime/computer-use-types'; +import { buildMcpTools, type McpToolProvider } from '@maka/runtime/mcp-tools'; import { type MakaTool, type MakaToolContext } from '@maka/runtime/tool-runtime'; import type { ClientCapabilityProvider } from '@maka/runtime-host/client'; import { @@ -147,6 +148,155 @@ test('publishes every production Desktop-owned tool schema through the protocol' ); }); +test('publishes and invokes an MCP tool backed by an AI SDK JSON Schema', async () => { + let invocation: { args: Record; cwd: string } | undefined; + let accepted = false; + const mcpProvider: McpToolProvider = { + toolSnapshot: () => ({ + revision: 1, + tools: [ + { + binding: 'fixture-binding' as never, + descriptor: { + serverId: 'fixture', + name: 'lookup', + inputSchema: { + type: 'object', + properties: { query: { type: 'string' } }, + required: ['query'], + additionalProperties: false, + }, + }, + }, + ], + }), + async callTool(_binding, args, options) { + invocation = { args, cwd: options.context.cwd }; + return { content: [{ type: 'text', text: 'found' }] }; + }, + }; + const [mcpTool] = buildMcpTools(mcpProvider, { executionLocation: 'remote' }); + assert.ok(mcpTool); + const provider = createDesktopNativeCapabilityProvider({ + browserTools: [], + releaseBrowserSession() {}, + computerUseTools: computerTools(), + releaseComputerUseSession() {}, + additionalGroups: () => [ + { + offerId: 'desktop_mcp', + label: 'MCP', + description: 'MCP tools', + invalidToolPolicy: 'omit', + tools: [mcpTool], + }, + ], + }); + + assert.doesNotThrow(() => + decodeClientCapabilityReplaceInput({ + registrationId: 'registration-1', + offers: provider.offers(), + }), + ); + await assert.rejects( + () => + call( + provider, + capabilityFrame({ + offerId: 'desktop_mcp', + serverId: 'desktop_mcp', + toolName: mcpTool.name, + arguments: {}, + }), + () => { + accepted = true; + }, + ), + /Invalid arguments for tool/u, + ); + assert.equal(accepted, false); + assert.equal(invocation, undefined); + assert.deepEqual( + await call( + provider, + capabilityFrame({ + offerId: 'desktop_mcp', + serverId: 'desktop_mcp', + toolName: mcpTool.name, + arguments: { query: 'maka' }, + }), + () => { + accepted = true; + }, + ), + { content: [{ type: 'text', text: 'found' }] }, + ); + assert.deepEqual(invocation, { + args: { query: 'maka' }, + cwd: '/workspace', + }); + assert.equal(accepted, true); +}); + +test('omits optional MCP tools that would exceed one offer\'s tool limit', () => { + const diagnostics: Error[] = []; + const provider = createDesktopNativeCapabilityProvider( + { + browserTools: [], + releaseBrowserSession() {}, + computerUseTools: computerTools(), + releaseComputerUseSession() {}, + additionalGroups: () => [ + { + offerId: 'desktop_mcp', + label: 'MCP', + description: 'MCP tools', + invalidToolPolicy: 'omit', + tools: Array.from({ length: 65 }, (_, index) => + tool(`mcp_tool_${index + 1}`, z.object({}), async () => 'ok'), + ), + }, + ], + }, + { onInvalidTool: (error) => diagnostics.push(error) }, + ); + + assert.deepEqual( + provider.offers()[0]?.tools.map(({ name }) => name), + Array.from({ length: 64 }, (_, index) => `mcp_tool_${index + 1}`), + ); + assert.equal(diagnostics.length, 1); + assert.match(diagnostics[0]?.message ?? '', /desktop_mcp\/mcp_tool_65/u); + assert.doesNotThrow(() => + decodeClientCapabilityReplaceInput({ registrationId: 'registration-1', offers: provider.offers() }), + ); +}); + +test('omits optional MCP tools that would exceed the complete manifest byte limit', () => { + const diagnostics: Error[] = []; + const provider = createDesktopNativeCapabilityProvider( + { + browserTools: [], + releaseBrowserSession() {}, + computerUseTools: computerTools(), + releaseComputerUseSession() {}, + additionalGroups: () => [ + optionalMcpGroup('desktop_mcp_first', 'mcp_first', 28 * 1024), + optionalMcpGroup('desktop_mcp_second', 'mcp_second', 28 * 1024), + ], + }, + { onInvalidTool: (error) => diagnostics.push(error) }, + ); + + assert.deepEqual(provider.offers().map(({ offerId }) => offerId), ['desktop_mcp_first']); + assert.equal(diagnostics.length, 1); + assert.match(diagnostics[0]?.message ?? '', /desktop_mcp_second\/mcp_second/u); + assert.doesNotThrow(() => + decodeClientCapabilityReplaceInput({ registrationId: 'registration-1', offers: provider.offers() }), + ); +}); + test('publishes and admits additional Desktop native-effect services', async () => { let admitted = false; const provider = createDesktopNativeCapabilityProvider({ @@ -524,6 +674,31 @@ function tool( }; } +function optionalMcpGroup(offerId: string, name: string, schemaDescriptionLength: number) { + return { + offerId, + label: 'MCP', + description: 'MCP tools', + invalidToolPolicy: 'omit' as const, + tools: [ + { + name, + displayName: name, + description: `${name} description`, + parameters: { + jsonSchema: { + type: 'object', + description: 'x'.repeat(schemaDescriptionLength), + }, + }, + async impl() { + return 'ok'; + }, + } as MakaTool, + ], + }; +} + function serviceFrame(): ClientCapabilityServiceCallFrame { return { kind: 'client.capability.service_call', diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index cf058b5278..1a86f3ee2c 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -536,6 +536,7 @@ owner = await startRuntimeHostDesktopOwner( label: "MCP", description: "Use MCP tools connected by this Desktop client.", + invalidToolPolicy: "omit" as const, tools: mcpTools, }, ]), diff --git a/apps/desktop/src/main/runtime-host-desktop-candidate.ts b/apps/desktop/src/main/runtime-host-desktop-candidate.ts index e0ef9da523..2ff9ae9bd9 100644 --- a/apps/desktop/src/main/runtime-host-desktop-candidate.ts +++ b/apps/desktop/src/main/runtime-host-desktop-candidate.ts @@ -522,6 +522,7 @@ export async function createDesktopRuntimeHostCandidate( desktopSessionResourceKey({ ...scope, sessionId }), onSessionUsed: (sessionId) => nativeSessionIds.add(sessionId), onComputerUseTurnUsed: watchComputerUseTurn, + onInvalidTool: reportError, isTargetValid: deps.isTargetValid, onClosed: () => providers.delete(provider), }, diff --git a/apps/desktop/src/main/runtime-host-native-capabilities.ts b/apps/desktop/src/main/runtime-host-native-capabilities.ts index 9b988a4ea1..ad800629de 100644 --- a/apps/desktop/src/main/runtime-host-native-capabilities.ts +++ b/apps/desktop/src/main/runtime-host-native-capabilities.ts @@ -1,11 +1,16 @@ import { Buffer } from "node:buffer"; import type { ComputerUseToolSet } from '@maka/runtime/computer-use-tools'; +import { + jsonSchemaErrorSummary, + validateJsonSchemaInput, +} from '@maka/runtime/json-schema-validation'; import type { MakaTool } from '@maka/runtime/tool-runtime'; import { createOAuthPresentationClientProvider, type ClientCapabilityProvider, type OAuthPresentationBackend, } from "@maka/runtime-host/client"; +import { decodeClientCapabilityReplaceInput } from "@maka/runtime-host/protocol"; import type { ClientCapabilityCallFrame, ClientCapabilityCallResult, @@ -26,6 +31,7 @@ export interface DesktopCapabilityGroup { readonly label: string; readonly description: string; readonly tools: readonly MakaTool[]; + readonly invalidToolPolicy?: "reject" | "omit"; } interface NativeToolBinding { @@ -73,6 +79,7 @@ interface DesktopNativeCapabilityProviderOptions { readonly isTargetValid?: () => boolean; readonly onSessionUsed?: (sessionId: string) => void; readonly onComputerUseTurnUsed?: (sessionId: string, turnId: string) => void; + readonly onInvalidTool?: (error: Error) => void; readonly onClosed?: () => void; readonly nativeSessionId?: (sessionId: string) => string; } @@ -82,8 +89,12 @@ export function createDesktopNativeCapabilityProvider( input: DesktopNativeCapabilityProviderInput, providerOptions: DesktopNativeCapabilityProviderOptions = {}, ): DesktopNativeCapabilityProvider { - const groups = capabilityGroups(input); const hostPathAccess = providerOptions.hostPathAccess ?? "cwd"; + const groups = prepareCapabilityGroups( + capabilityGroups(input), + hostPathAccess, + providerOptions.onInvalidTool, + ); const offers = Object.freeze( groups.map((group) => capabilityOffer(group, hostPathAccess)), ); @@ -297,8 +308,7 @@ async function invokeNativeTool( } const signal = AbortSignal.any([options.signal, invocation.signal]); signal.throwIfAborted(); - const parameters = requireZodSchema(binding.tool); - const args = await parameters.parseAsync(frame.arguments); + const args = await parseToolArguments(binding.tool, frame.arguments); signal.throwIfAborted(); await options.accept(); signal.throwIfAborted(); @@ -365,14 +375,64 @@ function capabilityOffer( }); } +function prepareCapabilityGroups( + groups: readonly DesktopCapabilityGroup[], + hostPathAccess: ClientCapabilityHostPathAccess, + onInvalidTool: ((error: Error) => void) | undefined, +): DesktopCapabilityGroup[] { + const selectedOptionalTools = new Map(); + const selectedGroups = (): DesktopCapabilityGroup[] => + groups.flatMap((group) => { + if (group.invalidToolPolicy !== "omit") return [group]; + const tools = selectedOptionalTools.get(group) ?? []; + return tools.length === 0 ? [] : [{ ...group, tools }]; + }); + const validateSelectedGroups = () => { + const selected = selectedGroups(); + if (selected.length === 0) return; + decodeClientCapabilityReplaceInput({ + registrationId: "desktop_capability_validation", + offers: selected.map((group) => capabilityOffer(group, hostPathAccess)), + }); + }; + + // Required Desktop capabilities remain fail-closed, even when optional MCP + // capabilities are eligible for omission. + validateSelectedGroups(); + for (const group of groups) { + if (group.invalidToolPolicy !== "omit") continue; + const tools: MakaTool[] = []; + for (const tool of group.tools) { + selectedOptionalTools.set(group, [...tools, tool]); + try { + validateSelectedGroups(); + tools.push(tool); + } catch (cause) { + selectedOptionalTools.set(group, tools); + onInvalidTool?.( + new Error( + `Invalid optional Desktop capability tool omitted: ${group.offerId}/${tool.name}`, + { cause }, + ), + ); + } + } + selectedOptionalTools.set(group, tools); + } + return selectedGroups(); +} + function toolInputSchema(tool: MakaTool): Record { - const schema = toJSONSchema(requireZodSchema(tool), { - io: "input", - target: "draft-07", - unrepresentable: "any", - cycles: "ref", - reused: "inline", - }); + const schema = + tool.parameters instanceof z.ZodType + ? toJSONSchema(tool.parameters, { + io: "input", + target: "draft-07", + unrepresentable: "any", + cycles: "ref", + reused: "inline", + }) + : providerToolInputSchema(tool); delete schema.$schema; if (schema.type !== "object") { throw new Error( @@ -382,13 +442,74 @@ function toolInputSchema(tool: MakaTool): Record { return Object.freeze(schema); } -function requireZodSchema(tool: MakaTool): z.ZodType { - if (!(tool.parameters instanceof z.ZodType)) { +function providerToolInputSchema(tool: MakaTool): Record { + if (!tool.parameters || typeof tool.parameters !== "object") { + throw new Error( + `Desktop native capability tool has an invalid schema: ${tool.name}`, + ); + } + const schema = (tool.parameters as { jsonSchema?: unknown }).jsonSchema; + if (isPromiseLike(schema)) { + throw new Error( + `Desktop native capability tool requires a synchronous schema: ${tool.name}`, + ); + } + if (!schema || typeof schema !== "object" || Array.isArray(schema)) { throw new Error( `Desktop native capability tool has an invalid schema: ${tool.name}`, ); } - return tool.parameters; + return structuredClone(schema) as Record; +} + +function isPromiseLike(value: unknown): value is PromiseLike { + return ( + typeof value === "object" && + value !== null && + "then" in value && + typeof (value as { then?: unknown }).then === "function" + ); +} + +async function parseToolArguments(tool: MakaTool, value: unknown): Promise { + const parameters = tool.parameters as { + parseAsync?: (input: unknown) => Promise; + validate?: ( + input: unknown, + ) => + | { success: true; value: unknown } + | { success: false; error: unknown } + | PromiseLike< + | { success: true; value: unknown } + | { success: false; error: unknown } + >; + jsonSchema?: unknown; + }; + if (typeof parameters?.parseAsync === "function") { + return parameters.parseAsync(value); + } + if (typeof parameters?.validate === "function") { + const result = await parameters.validate(value); + if (result.success) return result.value; + throw result.error; + } + if ( + parameters?.jsonSchema && + typeof parameters.jsonSchema === "object" && + !Array.isArray(parameters.jsonSchema) + ) { + try { + return validateJsonSchemaInput(await parameters.jsonSchema, value); + } catch (error) { + throw new Error( + `Invalid arguments for tool "${tool.name}": ${jsonSchemaErrorSummary(error)}`, + { cause: error }, + ); + } + } + throw new Error( + `Desktop native capability tool has an invalid schema: ${tool.name}`, + ); } function indexBindings( diff --git a/packages/runtime/package.json b/packages/runtime/package.json index 527ceb5b66..f279944559 100644 --- a/packages/runtime/package.json +++ b/packages/runtime/package.json @@ -112,6 +112,7 @@ "./tool-result-archive-capability": "./dist/tool-result-archive-capability.js", "./tool-result-archive-resource": "./dist/tool-result-archive-resource.js", "./tool-runtime": "./dist/tool-runtime.js", + "./json-schema-validation": "./dist/json-schema-validation.js", "./web-fetch-tool": "./dist/web-fetch-tool.js", "./web-search-tool": "./dist/web-search-tool.js", "./xai-oauth-enrollment": "./dist/xai-oauth-enrollment.js" diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index 9a121de770..fbfcb34d46 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -104,12 +104,10 @@ import type { UserContent, } from './model-protocol.js'; import type { ModelCallCommit } from '@maka/core/agent-run'; -import Ajv, { type AnySchema, type ErrorObject, type ValidateFunction } from 'ajv'; -import Ajv2019 from 'ajv/dist/2019.js'; -import Ajv2020 from 'ajv/dist/2020.js'; import { z } from 'zod'; import { AsyncEventQueue } from './async-queue.js'; +import { jsonSchemaErrorSummary, validateJsonSchemaInput } from './json-schema-validation.js'; import { StreamWatchdog, formatStreamWatchdogError, @@ -556,16 +554,6 @@ function nestableToolSnapshot( ); } -const codeModeJsonSchemaOptions = { - allErrors: true, - strict: false, - validateFormats: false, -} as const; -const codeModeDraft7Validator = new Ajv(codeModeJsonSchemaOptions); -const codeModeDraft2019Validator = new Ajv2019(codeModeJsonSchemaOptions); -const codeModeDraft2020Validator = new Ajv2020(codeModeJsonSchemaOptions); -const codeModeCompiledSchemas = new WeakMap(); - async function validateCodeModeToolInput(tool: MakaTool, input: unknown): Promise { const parameters = tool.parameters as { safeParseAsync?: ( @@ -597,29 +585,11 @@ async function validateCodeModeToolInput(tool: MakaTool, input: unknown): Promis } const schema = await parameters.jsonSchema; - const validator = compileCodeModeJsonSchema(schema ?? tool.parameters); - if (!validator || validator(input)) return input; - throw invalidCodeModeToolArguments(tool.name, validator.errors); -} - -function compileCodeModeJsonSchema(schema: unknown): ValidateFunction | undefined { - if (typeof schema === 'boolean') return codeModeDraft2020Validator.compile(schema); - if (typeof schema !== 'object' || schema === null || Array.isArray(schema)) return undefined; - const cached = codeModeCompiledSchemas.get(schema); - if (cached) return cached; - const declaredDialect = (schema as { readonly $schema?: unknown }).$schema; - const dialect = typeof declaredDialect === 'string' ? declaredDialect : ''; - const validator = dialect.includes('draft-07') - ? codeModeDraft7Validator - : dialect.includes('2019-09') - ? codeModeDraft2019Validator - : codeModeDraft2020Validator; - const schemaForCompile = dialect.startsWith('https://json-schema.org/draft-07/schema') - ? { ...schema, $schema: dialect.replace('https://', 'http://') } - : schema; - const compiled = validator.compile(schemaForCompile as AnySchema); - codeModeCompiledSchemas.set(schema, compiled); - return compiled; + try { + return validateJsonSchemaInput(schema ?? tool.parameters, input); + } catch (error) { + throw invalidCodeModeToolArguments(tool.name, error); + } } function invalidCodeModeToolArguments(toolName: string, error: unknown): Error { @@ -639,17 +609,7 @@ function schemaErrorSummary(error: unknown): string { .join('; ') .slice(0, 1000); } - if (Array.isArray(error)) { - return (error as ErrorObject[]) - .slice(0, 5) - .map((issue) => { - const path = issue.instancePath || issue.schemaPath; - return `${path || 'input'} ${issue.message ?? 'is invalid'}`; - }) - .join('; ') - .slice(0, 1000); - } - return 'input does not match the declared schema'; + return jsonSchemaErrorSummary(error); } function joinPromptFragments(fragments: readonly (string | undefined)[]): string | undefined { diff --git a/packages/runtime/src/json-schema-validation.ts b/packages/runtime/src/json-schema-validation.ts new file mode 100644 index 0000000000..ef8ad67a14 --- /dev/null +++ b/packages/runtime/src/json-schema-validation.ts @@ -0,0 +1,52 @@ +import Ajv, { type AnySchema, type ErrorObject, type ValidateFunction } from 'ajv'; +import Ajv2019 from 'ajv/dist/2019.js'; +import Ajv2020 from 'ajv/dist/2020.js'; + +const jsonSchemaOptions = { + allErrors: true, + strict: false, + validateFormats: false, +} as const; +const draft7Validator = new Ajv(jsonSchemaOptions); +const draft2019Validator = new Ajv2019(jsonSchemaOptions); +const draft2020Validator = new Ajv2020(jsonSchemaOptions); +const compiledSchemas = new WeakMap(); + +/** Validate an input against a provider JSON Schema when the schema is compilable. */ +export function validateJsonSchemaInput(schema: unknown, input: unknown): unknown { + const validator = compileJsonSchema(schema); + if (!validator || validator(input)) return input; + throw validator.errors; +} + +function compileJsonSchema(schema: unknown): ValidateFunction | undefined { + if (typeof schema === 'boolean') return draft2020Validator.compile(schema); + if (typeof schema !== 'object' || schema === null || Array.isArray(schema)) return undefined; + const cached = compiledSchemas.get(schema); + if (cached) return cached; + const declaredDialect = (schema as { readonly $schema?: unknown }).$schema; + const dialect = typeof declaredDialect === 'string' ? declaredDialect : ''; + const validator = dialect.includes('draft-07') + ? draft7Validator + : dialect.includes('2019-09') + ? draft2019Validator + : draft2020Validator; + const schemaForCompile = dialect.startsWith('https://json-schema.org/draft-07/schema') + ? { ...schema, $schema: dialect.replace('https://', 'http://') } + : schema; + const compiled = validator.compile(schemaForCompile as AnySchema); + compiledSchemas.set(schema, compiled); + return compiled; +} + +export function jsonSchemaErrorSummary(error: unknown): string { + if (!Array.isArray(error)) return 'input does not match the declared schema'; + return (error as ErrorObject[]) + .slice(0, 5) + .map((issue) => { + const path = issue.instancePath || issue.schemaPath; + return `${path || 'input'} ${issue.message ?? 'is invalid'}`; + }) + .join('; ') + .slice(0, 1000); +} From 0deebe652298ace60dbf6b33ee5bb0edc7b120cb Mon Sep 17 00:00:00 2001 From: Hsin <2129830748@qq.com> Date: Mon, 24 Aug 2026 09:04:38 +0800 Subject: [PATCH 2/4] refactor(storage): consolidate external session source parsing Generated-by: Codex --- packages/core/src/session.ts | 7 + .../__tests__/execution-composition.test.ts | 30 + .../external-session-coordinator.test.ts | 20 +- .../server/external-session-coordinator.ts | 7 +- .../__tests__/claude-session-adapter.test.ts | 265 +++++++ .../__tests__/codex-session-adapter.test.ts | 259 ++++++- .../external-session-importer.test.ts | 61 +- .../codex-rollout-v0.149-item-completed.jsonl | 10 + .../storage/src/claude-session-adapter.ts | 693 ++++++++++++++++++ packages/storage/src/codex-session-adapter.ts | 498 ++++++++----- packages/storage/src/execution-stores.ts | 12 +- .../storage/src/external-session-adapters.ts | 10 +- .../storage/src/external-session-importer.ts | 4 + .../storage/src/external-source-catalog.ts | 118 +++ packages/storage/src/foreign-session-store.ts | 683 ++--------------- packages/storage/src/session-store.ts | 17 + 16 files changed, 1906 insertions(+), 788 deletions(-) create mode 100644 packages/storage/src/__tests__/claude-session-adapter.test.ts create mode 100644 packages/storage/src/__tests__/fixtures/codex-rollout-v0.149-item-completed.jsonl create mode 100644 packages/storage/src/claude-session-adapter.ts create mode 100644 packages/storage/src/external-source-catalog.ts diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index e9dc928dbd..b0696361b4 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -195,6 +195,11 @@ export function isSessionToolProfile(value: unknown): value is SessionToolProfil return typeof value === 'string' && (SESSION_TOOL_PROFILES as readonly string[]).includes(value); } +export interface SessionExternalOrigin { + readonly adapterId: string; + readonly sourceSessionId: string; +} + export interface SessionHeader { // Identity id: string; @@ -232,6 +237,8 @@ export interface SessionHeader { subagentWorkspace?: SubagentWorkspaceBinding; /** Immutable Host publication identity for a cross-Session conversation copy. */ conversationCopy?: SessionConversationCopy; + /** Immutable identity of the external Session imported into this Session. */ + readonly externalOrigin?: SessionExternalOrigin; /** Stable root id for an edit-and-resend version family. */ revisionRootSessionId?: string; /** Immediate previous version in the same conversation slot. */ diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index 354675e1ad..e3923f955f 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -41,6 +41,36 @@ test('filesystem worker follows the candidate executable runtime', () => { assert.equal(runtimeHostFilesystemWorkerRuntime({}), 'node'); }); +test('execution-store facade preserves external origin on imported Sessions', async () => { + await withCompositionRoot(async ({ root, owner }) => { + const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + try { + const header = await stores.sessionStore.createImportedSession( + { + cwd: root, + backend: 'fake', + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask', + }, + [], + { adapterId: 'codex', sourceSessionId: 'codex-session-1' }, + ); + + assert.deepEqual(header.externalOrigin, { + adapterId: 'codex', + sourceSessionId: 'codex-session-1', + }); + assert.deepEqual( + (await stores.sessionStore.readHeader(header.id)).externalOrigin, + header.externalOrigin, + ); + } finally { + await stores.sessionStore.close?.(); + } + }); +}); + test('production composition owns the long-term memory database lifecycle', async () => { await withCompositionRoot(async ({ root, owner }) => { const databasePath = join(root, LONG_TERM_MEMORY_DATABASE_NAME); diff --git a/packages/runtime-host/src/__tests__/external-session-coordinator.test.ts b/packages/runtime-host/src/__tests__/external-session-coordinator.test.ts index 6fb46bd51d..0a4baa96c3 100644 --- a/packages/runtime-host/src/__tests__/external-session-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/external-session-coordinator.test.ts @@ -4,7 +4,11 @@ import { ExternalSessionAdapterRegistry, type ExternalSessionAdapter, } from '@maka/core/external-session'; -import { type SessionHeader, type StoredMessage } from '@maka/core/session'; +import { + type SessionExternalOrigin, + type SessionHeader, + type StoredMessage, +} from '@maka/core/session'; import { headerToSummary } from '@maka/runtime/session-manager'; import type { SessionCatalogRecord } from '@maka/storage/execution-stores'; import { EXTERNAL_SESSION_RESULT_MAX_BYTES } from '../protocol/index.js'; @@ -154,6 +158,10 @@ test('coalesces a repeat import issued while the first is still running', async // Same task, and only one of them was ever created. Both callers are told // about it, so the one that clicked twice still gets taken to the result. assert.equal(first.result.session.id, second.result.session.id); + assert.deepEqual(fixture.creates[0]?.externalOrigin, { + adapterId: 'codex', + sourceSessionId: 'source-0', + }); assert.equal(fixture.creates.length, 1); assert.equal(fixture.drainRequests(), 0); @@ -317,6 +325,7 @@ function coordinatorFixture( createImportedSession( input: Parameters[0], messages: readonly StoredMessage[], + externalOrigin?: SessionExternalOrigin, ): Promise; prepareImportedSessionHistory(sessionId: string): Promise; discardImportedSession(sessionId: string): Promise; @@ -329,14 +338,19 @@ function coordinatorFixture( const creates: Array<{ input: Parameters[0]; messages: readonly StoredMessage[]; + externalOrigin?: SessionExternalOrigin; }> = []; - const defaultCreate: HostStore['createImportedSession'] = async (input, messages) => { + const defaultCreate: HostStore['createImportedSession'] = async ( + input, + messages, + externalOrigin, + ) => { sequence += 1; const header = { ...sessionHeader(`imported-${sequence}`, input.cwd, input.name ?? 'Imported'), transcriptLedgerVersion: 0 as const, }; - creates.push({ input, messages }); + creates.push({ input, messages, externalOrigin }); records.set(header.id, { header, revision: 1, diff --git a/packages/runtime-host/src/server/external-session-coordinator.ts b/packages/runtime-host/src/server/external-session-coordinator.ts index c42eb3939d..7cb5d3d265 100644 --- a/packages/runtime-host/src/server/external-session-coordinator.ts +++ b/packages/runtime-host/src/server/external-session-coordinator.ts @@ -4,7 +4,7 @@ import type { ExternalSessionSummary, } from '@maka/core/external-session'; import type { CreateSessionInput } from '@maka/core/runtime-inputs'; -import type { SessionHeader, StoredMessage } from '@maka/core/session'; +import type { SessionExternalOrigin, SessionHeader, StoredMessage } from '@maka/core/session'; import type { SessionCatalogRecord } from '@maka/storage/execution-stores'; import { ExternalSessionImporter } from '@maka/storage/external-sessions'; import { @@ -32,6 +32,7 @@ type ExternalSessionStore = { createImportedSession( input: CreateSessionInput, messages: readonly StoredMessage[], + externalOrigin?: SessionExternalOrigin, ): Promise; listHeaders(): Promise; readCatalogRecord(sessionId: string): Promise; @@ -204,9 +205,9 @@ export class HostExternalSessionCoordinator { let commitAttempted = false; const importer = new ExternalSessionImporter(this.#adapters, { - createImportedSession: async (sessionInput, messages) => { + createImportedSession: async (sessionInput, messages, externalOrigin) => { commitAttempted = true; - return this.#sessions.createImportedSession(sessionInput, messages); + return this.#sessions.createImportedSession(sessionInput, messages, externalOrigin); }, }); let header: SessionHeader; diff --git a/packages/storage/src/__tests__/claude-session-adapter.test.ts b/packages/storage/src/__tests__/claude-session-adapter.test.ts new file mode 100644 index 0000000000..32c3cbf60b --- /dev/null +++ b/packages/storage/src/__tests__/claude-session-adapter.test.ts @@ -0,0 +1,265 @@ +import assert from 'node:assert/strict'; +import { mkdir, mkdtemp, rm, utimes, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, test } from 'node:test'; +import { decodeStoredMessage } from '@maka/core/session'; +import { ClaudeSessionAdapter } from '../claude-session-adapter.js'; +import { createExternalSessionAdapterRegistry } from '../external-session-adapters.js'; +import { createForeignSessionStore } from '../foreign-session-store.js'; + +describe('ClaudeSessionAdapter', () => { + test('shares catalog identity while projecting full history and safe handoff separately', async () => { + const home = await mkdtemp(join(tmpdir(), 'maka-claude-adapter-')); + const id = 'claude-session-1'; + const directory = join(home, '.claude', 'projects', '-repo'); + const path = join(directory, `${id}.jsonl`); + await mkdir(directory, { recursive: true }); + await writeFile( + path, + [ + record({ + type: 'user', + uuid: 'u1', + cwd: '/repo', + gitBranch: 'main', + message: { content: 'Fix the parser' }, + }), + record({ + type: 'assistant', + uuid: 'a1', + model: 'claude-test', + message: { + model: 'claude-test', + content: [ + { type: 'thinking', thinking: 'Inspect the parser.' }, + { type: 'text', text: 'I found the failing branch.' }, + { + type: 'tool_use', + id: 'tool-1', + name: 'Edit', + input: { file_path: '/repo/parser.ts' }, + }, + ], + }, + }), + record({ + type: 'user', + uuid: 'r1', + message: { + content: [{ type: 'tool_result', tool_use_id: 'tool-1', content: 'patched' }], + }, + }), + record({ type: 'summary', uuid: 'compact-1', summary: 'Prior context was compacted.' }), + record({ type: 'system', uuid: 'rewind-1', subtype: 'rewind', reason: 'user_requested' }), + record({ + type: 'assistant', + uuid: 'a1', + model: 'claude-test', + message: { content: [{ type: 'text', text: 'duplicate record must not be imported' }] }, + }), + record({ type: 'user', uuid: 'u2', message: { content: 'Continue after rewind' } }), + record({ + type: 'assistant', + uuid: 'a2', + model: 'claude-test', + message: { content: [{ type: 'text', text: 'Done.' }] }, + }), + 'not json', + ].join('\n') + '\n', + 'utf8', + ); + + try { + const adapter = new ClaudeSessionAdapter({ homeDir: home }); + const listed = await adapter.listSessions(); + assert.equal(listed.length, 1); + assert.deepEqual(listed[0], { + id, + name: 'Prior context was compacted.', + cwd: '/repo', + updatedAt: listed[0]?.updatedAt, + }); + const imported = await adapter.readSession(id); + assert.deepEqual(imported.metadata, { name: 'Prior context was compacted.', cwd: '/repo' }); + assert.deepEqual( + imported.messages.map((message) => message.type), + [ + 'user', + 'assistant', + 'tool_call', + 'tool_result', + 'system_note', + 'system_note', + 'turn_state', + 'user', + 'assistant', + 'turn_state', + ], + ); + for (const message of imported.messages) { + assert.deepEqual(decodeStoredMessage(message), message); + } + assert.equal(imported.messages.filter((message) => message.type === 'assistant').length, 2); + const toolResult = imported.messages.find((message) => message.type === 'tool_result'); + assert.equal(toolResult?.type, 'tool_result'); + if (toolResult?.type === 'tool_result') assert.equal(toolResult.toolUseId, 'tool-1'); + + const store = createForeignSessionStore({ homeDir: home, env: {} }); + const [summary] = await store.listSessions(); + assert.ok(summary); + const digest = await store.readDigest(summary); + assert.deepEqual(digest.userMessages, ['Fix the parser', 'Continue after rewind']); + assert.deepEqual(digest.assistantTexts, ['I found the failing branch.', 'Done.']); + assert.deepEqual(digest.filesTouched, ['/repo/parser.ts']); + assert.ok(digest.warnings.some((warning) => warning.includes('malformed'))); + assert.ok(!JSON.stringify(digest).includes('duplicate record')); + } finally { + await rm(home, { recursive: true, force: true }); + } + }); + + test('is registered as a native full-import source', async () => { + const registry = createExternalSessionAdapterRegistry({ claude: { homeDir: 'C:\\missing' } }); + assert.equal(registry.require('claude-code').id, 'claude-code'); + }); + + test('warns when a bounded handoff tail cannot prove Claude lineage continuity', async () => { + const home = await mkdtemp(join(tmpdir(), 'maka-claude-bounded-handoff-')); + const id = 'claude-bounded-lineage'; + const directory = join(home, '.claude', 'projects', '-repo'); + const path = join(directory, `${id}.jsonl`); + await mkdir(directory, { recursive: true }); + await writeFile( + path, + [ + record({ type: 'user', uuid: 'u1', cwd: '/repo', message: { content: 'Start' } }), + record({ + type: 'assistant', + uuid: 'large', + parentUuid: 'u1', + message: { content: [{ type: 'text', text: 'x'.repeat(2 * 1024 * 1024) }] }, + }), + record({ + type: 'user', + uuid: 'u2', + parentUuid: 'large', + message: { content: 'Continue' }, + }), + ].join('\n') + '\n', + 'utf8', + ); + + try { + const adapter = new ClaudeSessionAdapter({ homeDir: home }); + const [entry] = await adapter.listCatalogEntries(); + assert.ok(entry); + const digest = await adapter.readDigest(entry); + assert.ok(digest.warnings.some((warning) => warning.includes('lineage may be incomplete'))); + assert.deepEqual(digest.userMessages, ['Continue']); + } finally { + await rm(home, { recursive: true, force: true }); + } + }); + + test('projects the active Claude parent lineage after a rewind', async () => { + const home = await mkdtemp(join(tmpdir(), 'maka-claude-lineage-')); + const id = 'claude-lineage-rewind'; + const directory = join(home, '.claude', 'projects', '-repo'); + const path = join(directory, `${id}.jsonl`); + await mkdir(directory, { recursive: true }); + await writeFile( + path, + [ + record({ type: 'user', uuid: 'u1', cwd: '/repo', message: { content: 'Start' } }), + record({ + type: 'assistant', + uuid: 'old-a1', + parentUuid: 'u1', + message: { content: [{ type: 'text', text: 'Old branch answer' }] }, + }), + record({ + type: 'system', + uuid: 'rewind-1', + parentUuid: 'old-a1', + subtype: 'rewind', + }), + record({ + type: 'user', + uuid: 'u2', + parentUuid: 'u1', + message: { content: 'Continue from the rewind' }, + }), + record({ + type: 'assistant', + uuid: 'a2', + parentUuid: 'u2', + message: { content: [{ type: 'text', text: 'Current branch answer' }] }, + }), + ].join('\n') + '\n', + 'utf8', + ); + + try { + const adapter = new ClaudeSessionAdapter({ homeDir: home }); + const imported = await adapter.readSession(id); + assert.deepEqual( + imported.messages + .filter((message) => message.type === 'assistant') + .map((message) => message.text), + ['Current branch answer'], + ); + + const [entry] = await adapter.listCatalogEntries(); + assert.ok(entry); + const digest = await adapter.readDigest(entry); + assert.deepEqual(digest.assistantTexts, ['Current branch answer']); + } finally { + await rm(home, { recursive: true, force: true }); + } + }); + + test('uses the same newest transcript when a Claude id exists in two projects', async () => { + const home = await mkdtemp(join(tmpdir(), 'maka-claude-duplicate-id-')); + const id = 'claude-duplicate-id'; + const oldDirectory = join(home, '.claude', 'projects', '-old'); + const newDirectory = join(home, '.claude', 'projects', '-new'); + const oldPath = join(oldDirectory, `${id}.jsonl`); + const newPath = join(newDirectory, `${id}.jsonl`); + await mkdir(oldDirectory, { recursive: true }); + await mkdir(newDirectory, { recursive: true }); + await writeFile( + oldPath, + record({ type: 'user', uuid: 'old-u', cwd: '/old', message: { content: 'Old copy' } }), + 'utf8', + ); + await writeFile( + newPath, + record({ type: 'user', uuid: 'new-u', cwd: '/new', message: { content: 'New copy' } }), + 'utf8', + ); + await utimes(oldPath, 1, 1); + await utimes(newPath, 2, 2); + + try { + const adapter = new ClaudeSessionAdapter({ homeDir: home }); + assert.deepEqual(await adapter.listSessions(), [ + { + id, + name: 'New copy', + cwd: '/new', + updatedAt: Date.parse('2026-08-23T00:00:00.000Z'), + }, + ]); + const imported = await adapter.readSession(id); + assert.equal(imported.metadata.cwd, '/new'); + assert.equal(imported.messages.find((message) => message.type === 'user')?.text, 'New copy'); + } finally { + await rm(home, { recursive: true, force: true }); + } + }); +}); + +function record(value: Record): string { + return JSON.stringify({ timestamp: '2026-08-23T00:00:00.000Z', ...value }); +} diff --git a/packages/storage/src/__tests__/codex-session-adapter.test.ts b/packages/storage/src/__tests__/codex-session-adapter.test.ts index 720178b2a9..682ccf5b78 100644 --- a/packages/storage/src/__tests__/codex-session-adapter.test.ts +++ b/packages/storage/src/__tests__/codex-session-adapter.test.ts @@ -1,5 +1,5 @@ import assert from 'node:assert/strict'; -import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { mkdir, mkdtemp, readFile, rm, utimes, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { describe, test } from 'node:test'; @@ -9,6 +9,7 @@ import { CodexSessionAdapter } from '../codex-session-adapter.js'; import { createExternalSessionAdapterRegistry } from '../external-session-adapters.js'; const CURRENT_FIXTURE = fixturePath('codex-rollout-v0.144.jsonl'); +const ITEM_COMPLETED_FIXTURE = fixturePath('codex-rollout-v0.149-item-completed.jsonl'); describe('CodexSessionAdapter', () => { test('lists active and archived root Sessions from the newest Codex state database', async () => { @@ -179,6 +180,147 @@ describe('CodexSessionAdapter', () => { }); }); + test('converts Codex Desktop completed items without importing response mirrors', async () => { + await withCodexHome(async (codexHome) => { + const sessionId = 'codex-item-completed'; + await seedRawRollout(codexHome, sessionId, await readFile(ITEM_COMPLETED_FIXTURE, 'utf8')); + + const adapter = new CodexSessionAdapter({ codexHome }); + assert.deepEqual( + (await adapter.listSessions()).map(({ id, name }) => ({ id, name })), + [{ id: sessionId, name: 'Analyze the image. Use OpenCV.js.' }], + ); + const session = await adapter.readSession(sessionId); + + assert.deepEqual(session.metadata, { + name: 'Analyze the image. Use OpenCV.js.', + cwd: '/workspace/opencv', + }); + assert.equal(session.messages.length, 4); + assert.deepEqual( + session.messages.map((message) => message.type), + ['user', 'assistant', 'assistant', 'turn_state'], + ); + for (const message of session.messages) { + assert.deepEqual(decodeStoredMessage(message), message); + } + assert.deepEqual(session.messages[0], { + type: 'user', + id: 'user-client-1', + turnId: 'codex-turn-item-completed', + ts: Date.parse('2026-08-22T00:00:02.100Z'), + text: 'Analyze the image.\nUse OpenCV.js.', + }); + assert.deepEqual(session.messages[1], { + type: 'assistant', + id: 'reasoning-item-1', + turnId: 'codex-turn-item-completed', + ts: Date.parse('2026-08-22T00:00:03.000Z'), + text: '', + thinking: { text: 'Inspect the pixels.\nDraft the solution.' }, + contentOrder: ['thinking'], + modelId: 'gpt-codex-item-test', + }); + assert.deepEqual(session.messages[2], { + type: 'assistant', + id: 'assistant-item-1', + turnId: 'codex-turn-item-completed', + ts: Date.parse('2026-08-22T00:00:04.000Z'), + text: 'Use canvas.\nThen process the pixels.', + modelId: 'gpt-codex-item-test', + contentOrder: ['text'], + }); + assert.equal(session.messages[3]?.type, 'turn_state'); + assert.equal(session.messages[3]?.status, 'completed'); + }); + }); + + test('builds the handoff from presentation events when response mirrors are absent', async () => { + await withCodexHome(async (codexHome) => { + const sessionId = 'codex-item-completed-no-mirrors'; + const fixture = await readFile(ITEM_COMPLETED_FIXTURE, 'utf8'); + const presentationOnly = fixture + .split('\n') + .filter((line) => { + if (line.trim().length === 0) return true; + const record = JSON.parse(line) as { type?: unknown }; + return record.type !== 'response_item'; + }) + .join('\n'); + await seedRawRollout( + codexHome, + sessionId, + presentationOnly.replaceAll('codex-item-completed', sessionId), + ); + + const adapter = new CodexSessionAdapter({ codexHome }); + const [entry] = await adapter.listCatalogEntries(); + assert.ok(entry); + const digest = await adapter.readDigest({ + source: 'codex', + id: entry.id, + title: entry.title, + cwd: entry.cwd, + updatedAtMs: entry.updatedAtMs, + transcriptPath: entry.transcriptPath, + }); + + assert.deepEqual(digest.userMessages, ['Analyze the image. Use OpenCV.js.']); + assert.deepEqual(digest.assistantTexts, ['Use canvas. Then process the pixels.']); + }); + }); + + test('falls back to response-item messages when presentation events are absent', async () => { + await withCodexHome(async (codexHome) => { + const sessionId = 'codex-response-item-fallback'; + const fixture = await readFile(CURRENT_FIXTURE, 'utf8'); + const responseOnly = fixture + .split('\n') + .filter((line) => { + if (line.trim().length === 0) return true; + const record = JSON.parse(line) as { type?: unknown }; + return record.type !== 'event_msg'; + }) + .join('\n') + .replaceAll('codex-session-1', sessionId); + await seedRawRollout(codexHome, sessionId, responseOnly); + + const adapter = new CodexSessionAdapter({ codexHome }); + const session = await adapter.readSession(sessionId); + assert.deepEqual( + session.messages + .filter((message) => message.type === 'user' || message.type === 'assistant') + .map((message) => message.text), + ['synthetic model input', 'I found the issue.'], + ); + }); + }); + + test('merges response-only presentation records without duplicating event mirrors', async () => { + await withCodexHome(async (codexHome) => { + const sessionId = 'codex-mixed-presentation'; + const fixture = await readFile(CURRENT_FIXTURE, 'utf8'); + const mixed = fixture + .split('\n') + .filter((line) => { + if (line.trim().length === 0) return true; + const record = JSON.parse(line) as { type?: unknown; payload?: { type?: unknown } }; + return !(record.type === 'event_msg' && record.payload?.type === 'user_message'); + }) + .join('\n') + .replaceAll('codex-session-1', sessionId); + await seedRawRollout(codexHome, sessionId, mixed); + + const session = await new CodexSessionAdapter({ codexHome }).readSession(sessionId); + assert.deepEqual( + session.messages + .filter((message) => message.type === 'user' || message.type === 'assistant') + .map((message) => message.text), + ['synthetic model input', '', 'I found the issue.'], + ); + }); + }); + test('filesystem fallback excludes internal subagent rollouts', async () => { await withCodexHome(async (codexHome) => { await seedMinimalRollout( @@ -208,6 +350,121 @@ describe('CodexSessionAdapter', () => { }); }); + test('filters internal sources before applying the catalog limit', async () => { + await withCodexHome(async (codexHome) => { + const rootId = 'codex-root-before-limit'; + const rootPath = await seedMinimalRollout( + codexHome, + rootId, + false, + '/workspace/root', + 'Root', + ); + const subagentRows = []; + for (const [index, updatedAtMs] of [4_000, 3_000, 2_000].entries()) { + const id = `codex-subagent-before-limit-${index}`; + const rolloutPath = await seedMinimalRollout( + codexHome, + id, + false, + '/workspace/root', + `Internal ${index}`, + ); + subagentRows.push({ + id, + rolloutPath, + cwd: '/workspace/root', + name: `Internal ${index}`, + createdAtMs: updatedAtMs, + updatedAtMs, + archived: false, + source: '{"subagent":{"thread_spawn":{}}}', + }); + } + await seedStateDatabase(codexHome, [ + ...subagentRows, + { + id: rootId, + rolloutPath: rootPath, + cwd: '/workspace/root', + name: 'Root task', + createdAtMs: 1_000, + updatedAtMs: 1_000, + archived: false, + source: 'cli', + }, + ]); + + const [entry] = await new CodexSessionAdapter({ codexHome }).listCatalogEntries({ limit: 1 }); + assert.equal(entry?.id, rootId); + }); + }); + + test('uses the same newest rollout when a Codex id has duplicate candidates', async () => { + await withCodexHome(async (codexHome) => { + const sessionId = 'codex-duplicate-id'; + const oldDirectory = join(codexHome, 'sessions', '2026', '08', '07'); + const newDirectory = join(codexHome, 'sessions', '2026', '08', '08'); + const oldPath = join(oldDirectory, `rollout-old-${sessionId}.jsonl`); + const newPath = join(newDirectory, `rollout-new-${sessionId}.jsonl`); + await mkdir(oldDirectory, { recursive: true }); + await mkdir(newDirectory, { recursive: true }); + await writeFile(oldPath, minimalRollout(sessionId, '/old', 'Old copy')); + await writeFile(newPath, minimalRollout(sessionId, '/new', 'New copy')); + await utimes(oldPath, 1, 1); + await utimes(newPath, 2, 2); + + const adapter = new CodexSessionAdapter({ codexHome }); + assert.deepEqual(await adapter.listSessions(), [ + { + id: sessionId, + name: 'New copy', + cwd: '/new', + createdAt: Date.parse('2026-08-08T00:00:00.000Z'), + updatedAt: 2_000, + archived: false, + }, + ]); + const imported = await adapter.readSession(sessionId); + assert.equal(imported.metadata.cwd, '/new'); + assert.equal(imported.messages.find((message) => message.type === 'user')?.text, 'New copy'); + }); + }); + + test('does not import a filesystem-only rollout when a state database is authoritative', async () => { + await withCodexHome(async (codexHome) => { + const visibleId = 'codex-state-visible'; + const hiddenId = 'codex-state-hidden'; + const visiblePath = await seedMinimalRollout( + codexHome, + visibleId, + false, + '/workspace/root', + 'Visible', + ); + await seedMinimalRollout(codexHome, hiddenId, false, '/workspace/root', 'Hidden'); + await seedStateDatabase(codexHome, [ + { + id: visibleId, + rolloutPath: visiblePath, + cwd: '/workspace/root', + name: 'Visible', + createdAtMs: 1_000, + updatedAtMs: 1_000, + archived: false, + source: 'cli', + }, + ]); + + const adapter = new CodexSessionAdapter({ codexHome }); + assert.deepEqual( + (await adapter.listSessions()).map((session) => session.id), + [visibleId], + ); + await assert.rejects(adapter.readSession(hiddenId), /not found/); + }); + }); + test('rejects corrupt interior records, tolerates a torn tail, and bounds full reads', async () => { await withCodexHome(async (codexHome) => { const fixture = await readFile(CURRENT_FIXTURE, 'utf8'); diff --git a/packages/storage/src/__tests__/external-session-importer.test.ts b/packages/storage/src/__tests__/external-session-importer.test.ts index 541feec199..483be814f3 100644 --- a/packages/storage/src/__tests__/external-session-importer.test.ts +++ b/packages/storage/src/__tests__/external-session-importer.test.ts @@ -1,8 +1,9 @@ import assert from 'node:assert/strict'; -import { mkdtemp, rm } from 'node:fs/promises'; +import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { describe, test } from 'node:test'; +import { fileURLToPath } from 'node:url'; import type { StoredMessage } from '@maka/core/session'; import { ExternalSessionAdapterRegistry, @@ -12,9 +13,63 @@ import { ExternalSessionImporter, type ExternalSessionImportTarget, } from '../external-session-importer.js'; +import { createExternalSessionAdapterRegistry } from '../external-session-adapters.js'; import { createSessionStore } from '../session-store.js'; describe('ExternalSessionImporter', () => { + test('imports a current Codex rollout through the real import route', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-codex-import-route-')); + const codexHome = join(root, '.codex'); + const sessionId = 'codex-item-completed'; + const rolloutDirectory = join(codexHome, 'sessions', '2026', '08', '22'); + const rolloutPath = join(rolloutDirectory, `rollout-2026-08-22T00-00-00-${sessionId}.jsonl`); + const sessions = createSessionStore(join(root, 'maka')); + const importer = new ExternalSessionImporter( + createExternalSessionAdapterRegistry({ codex: { codexHome } }), + sessions, + ); + + try { + await mkdir(rolloutDirectory, { recursive: true }); + const fixturePath = fileURLToPath( + new URL( + '../../src/__tests__/fixtures/codex-rollout-v0.149-item-completed.jsonl', + import.meta.url, + ), + ); + await writeFile(rolloutPath, await readFile(fixturePath)); + + const header = await importer.import({ + adapterId: 'codex', + sourceSessionId: sessionId, + target: target(), + }); + const importedMessages = await sessions.readMessages(header.id); + + assert.deepEqual(header.externalOrigin, { + adapterId: 'codex', + sourceSessionId: sessionId, + }); + assert.equal(header.name, 'Analyze the image. Use OpenCV.js.'); + assert.equal(header.cwd, '/workspace/opencv'); + assert.deepEqual( + importedMessages.map((message) => message.type), + ['user', 'assistant', 'assistant', 'turn_state'], + ); + assert.equal(importedMessages[0]?.type, 'user'); + if (importedMessages[0]?.type === 'user') { + assert.equal(importedMessages[0].text, 'Analyze the image.\nUse OpenCV.js.'); + } + assert.equal(importedMessages[2]?.type, 'assistant'); + if (importedMessages[2]?.type === 'assistant') { + assert.equal(importedMessages[2].text, 'Use canvas.\nThen process the pixels.'); + } + } finally { + await sessions.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + test('persists adapter output as native Maka StoredMessages', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-external-session-import-')); const sessions = createSessionStore(root); @@ -49,6 +104,10 @@ describe('ExternalSessionImporter', () => { assert.equal(header.cwd, '/external/repo'); assert.equal(header.model, 'maka-model'); assert.equal(header.connectionLocked, true); + assert.deepEqual(header.externalOrigin, { + adapterId: 'fake', + sourceSessionId: 'source-1', + }); assert.deepEqual(await sessions.readMessages(header.id), messages); } finally { await sessions.close?.(); diff --git a/packages/storage/src/__tests__/fixtures/codex-rollout-v0.149-item-completed.jsonl b/packages/storage/src/__tests__/fixtures/codex-rollout-v0.149-item-completed.jsonl new file mode 100644 index 0000000000..b30e247612 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/codex-rollout-v0.149-item-completed.jsonl @@ -0,0 +1,10 @@ +{"timestamp":"2026-08-22T00:00:00.000Z","ordinal":0,"type":"session_meta","payload":{"session_id":"codex-item-completed","id":"codex-item-completed","timestamp":"2026-08-22T00:00:00.000Z","cwd":"/workspace/opencv","originator":"Codex Desktop","cli_version":"0.149.0-alpha.4.1","source":"vscode","model_provider":"openai"}} +{"timestamp":"2026-08-22T00:00:01.000Z","ordinal":1,"type":"event_msg","payload":{"type":"task_started","turn_id":"codex-turn-item-completed","started_at":"2026-08-22T00:00:01.000Z"}} +{"timestamp":"2026-08-22T00:00:01.100Z","ordinal":2,"type":"turn_context","payload":{"turn_id":"codex-turn-item-completed","cwd":"/workspace/opencv","model":"gpt-codex-item-test"}} +{"timestamp":"2026-08-22T00:00:02.000Z","ordinal":3,"type":"response_item","payload":{"type":"message","id":"provider-user-mirror","role":"user","content":[{"type":"input_text","text":"Analyze the image.\nUse OpenCV.js."}]}} +{"timestamp":"2026-08-22T00:00:02.100Z","ordinal":4,"type":"event_msg","payload":{"type":"item_completed","thread_id":"codex-item-completed","turn_id":"codex-turn-item-completed","item":{"type":"UserMessage","id":"user-item-1","client_id":"user-client-1","content":[{"type":"text","text":"Analyze the image.","text_elements":[]},{"type":"local_image","path":"/tmp/input.png"},{"type":"text","text":"Use OpenCV.js.","text_elements":[]}]},"started_at_ms":1787356802000,"completed_at_ms":1787356802100}} +{"timestamp":"2026-08-22T00:00:03.000Z","ordinal":5,"type":"event_msg","payload":{"type":"item_completed","thread_id":"codex-item-completed","turn_id":"codex-turn-item-completed","item":{"type":"Reasoning","id":"reasoning-item-1","summary_text":["Inspect the pixels.","Draft the solution."],"raw_content":[]},"started_at_ms":1787356803000,"completed_at_ms":1787356803100}} +{"timestamp":"2026-08-22T00:00:03.100Z","ordinal":6,"type":"response_item","payload":{"type":"reasoning","id":"reasoning-item-1","summary":[{"type":"summary_text","text":"Inspect the pixels."},{"type":"summary_text","text":"Draft the solution."}],"encrypted_content":"opaque"}} +{"timestamp":"2026-08-22T00:00:04.000Z","ordinal":7,"type":"event_msg","payload":{"type":"item_completed","thread_id":"codex-item-completed","turn_id":"codex-turn-item-completed","item":{"type":"AgentMessage","id":"assistant-item-1","content":[{"type":"Text","text":"Use canvas."},{"type":"Text","text":"Then process the pixels."}],"phase":"final_answer"},"started_at_ms":1787356804000,"completed_at_ms":1787356804100}} +{"timestamp":"2026-08-22T00:00:04.100Z","ordinal":8,"type":"response_item","payload":{"type":"message","id":"assistant-item-1","role":"assistant","content":[{"type":"output_text","text":"Use canvas.\nThen process the pixels."}],"phase":"final_answer"}} +{"timestamp":"2026-08-22T00:00:05.000Z","ordinal":9,"type":"event_msg","payload":{"type":"task_complete","turn_id":"codex-turn-item-completed","last_agent_message":"Use canvas.\nThen process the pixels.","completed_at":"2026-08-22T00:00:05.000Z"}} diff --git a/packages/storage/src/claude-session-adapter.ts b/packages/storage/src/claude-session-adapter.ts new file mode 100644 index 0000000000..6f2a9d3c81 --- /dev/null +++ b/packages/storage/src/claude-session-adapter.ts @@ -0,0 +1,693 @@ +import { readdir, realpath, stat } from 'node:fs/promises'; +import { homedir } from 'node:os'; +import { basename, join, resolve, sep } from 'node:path'; +import type { ToolResultContent } from '@maka/core/events'; +import type { StoredMessage } from '@maka/core/session'; +import { + FOREIGN_SESSION_DIGEST_MAX_READ_BYTES, + FOREIGN_SESSION_HEAD_BYTES, + FOREIGN_SESSION_TITLE_WINDOW_BYTES, + claudeAssistantText, + claudeToolFilePaths, + claudeUserAuthoredText, + collectClaudeMeta, + collectClaudeTitle, + createDigestAccumulator, + finishDigest, + isSafeForeignId, + parseForeignJsonLine, + pickClaudeTitle, + pushDigestFile, + pushDigestMessage, + type ClaudeTitleCandidates, + type ClaudeTranscriptMeta, + type ForeignSessionDigest, + type ForeignSessionSummary, +} from '@maka/core/foreign-session'; +import type { + ExternalMakaSession, + ExternalSessionAdapter, + ExternalSessionQuery, + ExternalSessionSummary, +} from '@maka/core/external-session'; +import { + matchesSourceCatalogQuery, + readBoundedUtf8File, + readUtf8Prefix, + readUtf8Tail, + type ExternalSourceCatalogEntry, + type ExternalSourceCatalogQuery, +} from './external-source-catalog.js'; + +export const CLAUDE_SESSION_ADAPTER_ID = 'claude-code'; +const CLAUDE_HEAD_GROWTH_BYTES = 64 * 1024; +const CLAUDE_HEAD_MAX_BYTES = 4 * 1024 * 1024; +const CLAUDE_RECORD_ID_MAX_CHARS = 256; +const CLAUDE_DEFAULT_MAX_BYTES = 64 * 1024 * 1024; + +type JsonRecord = Record; + +interface ClaudeNativeRecord { + line: number; + value: JsonRecord; +} + +interface ClaudeNativeTranscript { + id: string; + records: readonly ClaudeNativeRecord[]; + malformedLines: number; + lineageComplete: boolean; + metadata: { + name: string; + cwd: string; + gitBranch?: string; + updatedAtMs: number; + }; +} + +interface ClaudeCatalogEntry extends ExternalSourceCatalogEntry { + source: 'claude-code'; +} + +export interface ClaudeSessionAdapterOptions { + /** Claude's home directory. Defaults to the user's home directory. */ + homeDir?: string; + /** Test/host override for the bounded full transcript read. */ + maxTranscriptBytes?: number; +} + +/** Catalog plus native parser for Claude Code transcripts. */ +export class ClaudeSessionAdapter implements ExternalSessionAdapter { + readonly id = CLAUDE_SESSION_ADAPTER_ID; + + private readonly homeDir: string; + private readonly maxTranscriptBytes: number; + + constructor(options: ClaudeSessionAdapterOptions = {}) { + this.homeDir = resolve(options.homeDir ?? homedir()); + this.maxTranscriptBytes = options.maxTranscriptBytes ?? CLAUDE_DEFAULT_MAX_BYTES; + if (!Number.isSafeInteger(this.maxTranscriptBytes) || this.maxTranscriptBytes <= 0) { + throw new Error('Claude transcript byte limit must be a positive safe integer'); + } + } + + async detect(): Promise { + return isDirectory(this.claudeRoot); + } + + async listSessions(query: ExternalSessionQuery = {}): Promise { + const entries = await this.listCatalogEntries(query); + return entries.map((entry) => ({ + id: entry.id, + name: entry.title, + cwd: entry.cwd, + ...(entry.createdAtMs !== undefined ? { createdAt: entry.createdAtMs } : {}), + updatedAt: entry.updatedAtMs, + ...(entry.archived !== undefined ? { archived: entry.archived } : {}), + })); + } + + async readSession(sessionId: string): Promise { + const entry = await this.findCatalogEntry(sessionId); + if (!entry) throw new Error(`Claude Session not found: ${sessionId}`); + const text = await readBoundedUtf8File( + await this.resolveTranscriptPath(entry.transcriptPath, sessionId), + this.maxTranscriptBytes, + ); + const transcript = parseClaudeTranscript(text, entry); + return projectClaudeSession(transcript, entry); + } + + async listCatalogEntries( + query: ExternalSourceCatalogQuery = {}, + ): Promise { + if (query.limit !== undefined && query.limit <= 0) return []; + const candidates = await this.listTranscriptCandidates(); + const entries: ClaudeCatalogEntry[] = []; + const seenIds = new Set(); + for (const candidate of candidates) { + const id = basename(candidate.path, '.jsonl'); + if (seenIds.has(id)) continue; + seenIds.add(id); + const entry = await catalogEntryFromTranscript(candidate.path, candidate.mtimeMs); + if (!entry || !matchesSourceCatalogQuery(entry, query)) continue; + entries.push(entry); + } + entries.sort(compareClaudeCatalogEntries); + return query.limit === undefined ? entries : entries.slice(0, query.limit); + } + + async readDigest(summary: ForeignSessionSummary): Promise { + if (summary.source !== 'claude-code') throw new Error('Claude adapter received another source'); + const path = await this.resolveTranscriptPath(summary.transcriptPath, summary.id); + const { text, truncated } = await readUtf8Tail(path, FOREIGN_SESSION_DIGEST_MAX_READ_BYTES); + const entry: ClaudeCatalogEntry = { + source: 'claude-code', + id: summary.id, + title: summary.title, + cwd: summary.cwd, + updatedAtMs: summary.updatedAtMs, + gitBranch: summary.gitBranch, + transcriptPath: path, + }; + const transcript = parseClaudeTranscript(text, entry); + const acc = createDigestAccumulator(); + for (const record of transcript.records) { + const value = record.value; + if (value.isSidechain === true) continue; + if (value.type === 'user') { + const userText = claudeUserAuthoredText(value); + if (userText !== undefined) pushDigestMessage(acc, 'user', userText); + } else if (value.type === 'assistant') { + const assistantText = claudeAssistantText(value); + if (assistantText !== undefined) pushDigestMessage(acc, 'assistant', assistantText); + for (const filePath of claudeToolFilePaths(value)) pushDigestFile(acc, filePath); + } + } + if (truncated) { + acc.warnings.push( + `transcript exceeded ${FOREIGN_SESSION_DIGEST_MAX_READ_BYTES} bytes; only its tail was read`, + ); + } + if (!transcript.lineageComplete) { + acc.warnings.push( + 'Claude transcript lineage may be incomplete because rewind and compaction ancestors are outside the available transcript window', + ); + } + if (transcript.malformedLines > 0) { + acc.warnings.push(`${transcript.malformedLines} malformed transcript lines were skipped`); + } + return finishDigest(acc, { + source: summary.source, + id: summary.id, + title: summary.title, + cwd: summary.cwd, + gitBranch: summary.gitBranch, + updatedAtMs: summary.updatedAtMs, + }); + } + + private get claudeRoot(): string { + return join(this.homeDir, '.claude', 'projects'); + } + + private async findCatalogEntry(sessionId: string): Promise { + if (!isSafeForeignId(sessionId)) return undefined; + return (await this.listCatalogEntries({ includeArchived: true })).find( + (entry) => entry.id === sessionId, + ); + } + + private async listTranscriptCandidates(): Promise<{ path: string; mtimeMs: number }[]> { + const candidates: { path: string; mtimeMs: number }[] = []; + for (const projectDir of await listSubdirectories(this.claudeRoot)) { + for (const candidate of await listJsonlFiles(projectDir)) candidates.push(candidate); + } + return candidates.sort((a, b) => b.mtimeMs - a.mtimeMs || a.path.localeCompare(b.path)); + } + + private async resolveTranscriptPath(path: string, expectedId: string): Promise { + const root = await realpath(this.claudeRoot); + const real = await realpath(resolve(path)); + if (real !== root && !real.startsWith(root + sep)) { + throw new Error('Foreign transcript escaped its source root'); + } + if (basename(real, '.jsonl') !== expectedId || !(await isFile(real))) { + throw new Error(`Claude transcript is unavailable: ${expectedId}`); + } + return real; + } +} + +async function catalogEntryFromTranscript( + path: string, + mtimeMs: number, +): Promise { + const id = basename(path, '.jsonl'); + if (!isSafeForeignId(id)) return undefined; + const meta: ClaudeTranscriptMeta = {}; + const titles: ClaudeTitleCandidates = {}; + for (const window of [ + await readClaudeHead(path), + await readTranscriptWindow(path, 'tail', CLAUDE_HEAD_GROWTH_BYTES), + ]) { + if (window === undefined) continue; + for (const line of window.split('\n')) { + const record = parseForeignJsonLine(line); + if (!record) continue; + collectClaudeMeta(record, meta); + collectClaudeTitle(record, titles); + } + } + if (meta.isSidechain === true || meta.cwd === undefined) return undefined; + return { + source: 'claude-code', + id, + title: pickClaudeTitle(titles) || id, + cwd: meta.cwd, + updatedAtMs: meta.timestampMs ?? mtimeMs, + ...(meta.gitBranch !== undefined ? { gitBranch: meta.gitBranch } : {}), + transcriptPath: path, + }; +} + +function parseClaudeTranscript( + text: string, + entry: Pick, +): ClaudeNativeTranscript { + const records: ClaudeNativeRecord[] = []; + const seenIds = new Set(); + let malformedLines = 0; + let updatedAtMs = entry.updatedAtMs; + let cwd = entry.cwd; + let gitBranch = entry.gitBranch; + const titles: ClaudeTitleCandidates = {}; + const lines = text.split('\n'); + if (text.endsWith('\n')) lines.pop(); + for (let index = 0; index < lines.length; index += 1) { + const line = lines[index]!; + if (line.trim().length === 0) continue; + const value = parseForeignJsonLine(line); + if (!value) { + malformedLines += 1; + continue; + } + const sourceId = stringField(value, 'uuid'); + if (sourceId !== undefined) { + if (sourceId.length > CLAUDE_RECORD_ID_MAX_CHARS || seenIds.has(sourceId)) continue; + seenIds.add(sourceId); + } + const meta: ClaudeTranscriptMeta = { cwd, gitBranch, timestampMs: updatedAtMs }; + collectClaudeMeta(value, meta); + cwd = meta.cwd ?? cwd; + gitBranch = meta.gitBranch ?? gitBranch; + updatedAtMs = Math.max(updatedAtMs, meta.timestampMs ?? updatedAtMs); + collectClaudeTitle(value, titles); + records.push({ line: index + 1, value }); + } + if (!isSafeForeignId(entry.id)) throw new Error(`Invalid Claude Session id: ${entry.id}`); + const lineage = resolveClaudeLineage(records); + return { + id: entry.id, + records: lineage.records, + malformedLines, + lineageComplete: lineage.complete, + metadata: { + name: pickClaudeTitle(titles) || entry.title, + cwd, + ...(gitBranch !== undefined ? { gitBranch } : {}), + updatedAtMs, + }, + }; +} + +function resolveClaudeLineage(records: readonly ClaudeNativeRecord[]): { + records: readonly ClaudeNativeRecord[]; + complete: boolean; +} { + const byId = new Map(); + const childIds = new Set(); + let hasParentLinks = false; + for (const record of records) { + const id = recordId(record.value); + if (id.length > 0) byId.set(id, record); + if (parentId(record.value) !== undefined) hasParentLinks = true; + const parent = parentId(record.value); + if (parent !== undefined) childIds.add(parent); + } + if (!hasParentLinks) return { records, complete: true }; + + const leaf = + [...records].reverse().find((record) => { + const id = recordId(record.value); + return id.length > 0 && !childIds.has(id) && !isClaudeLineageBoundary(record.value); + }) ?? [...records].reverse().find((record) => recordId(record.value).length > 0); + if (!leaf) return { records, complete: false }; + + const activeIds = new Set(); + let complete = true; + let currentId: string | undefined = recordId(leaf.value); + while (currentId !== undefined) { + if (activeIds.has(currentId)) { + complete = false; + break; + } + activeIds.add(currentId); + const record = byId.get(currentId); + if (!record) { + complete = false; + break; + } + currentId = parentId(record.value); + } + + return { + records: records.filter((record) => { + const value = record.value; + return activeIds.has(recordId(value)) || isClaudeLineageBoundary(value); + }), + complete, + }; +} + +function parentId(record: JsonRecord): string | undefined { + return ( + stringField(record, 'parentUuid') ?? + stringField(record, 'parent_uuid') ?? + stringField(record, 'logicalParentUuid') ?? + stringField(record, 'logical_parent_uuid') + ); +} + +function isClaudeLineageBoundary(record: JsonRecord): boolean { + const type = stringField(record, 'type')?.toLowerCase(); + const subtype = stringField(record, 'subtype')?.toLowerCase(); + return ( + type === 'summary' || + record.isCompactSummary === true || + record.isRewind === true || + type === 'rewind' || + subtype?.includes('rewind') === true || + subtype?.includes('compact') === true + ); +} + +function projectClaudeSession( + transcript: ClaudeNativeTranscript, + entry: Pick, +): ExternalMakaSession { + const messages: StoredMessage[] = []; + const usedIds = new Set(); + const toolIds = new Map(); + const resolvedToolIds = new Set(); + const turns = new Set(); + let activeTurnId: string | undefined; + let lastTimestamp = 0; + let turnCounter = 0; + + const timestampFor = (record: ClaudeNativeRecord): number => { + const parsed = parseTimestampMs(record.value.timestamp); + if (parsed !== undefined) lastTimestamp = Math.max(lastTimestamp, parsed); + else lastTimestamp += 1; + return parsed ?? lastTimestamp; + }; + const uniqueId = (base: string, line: number): string => { + const safeBase = base.length > 0 ? base : `claude-${transcript.id}-${line}`; + if (!usedIds.has(safeBase)) { + usedIds.add(safeBase); + return safeBase; + } + let candidate = `${safeBase}-${line}`; + let suffix = 2; + while (usedIds.has(candidate)) candidate = `${safeBase}-${line}-${suffix++}`; + usedIds.add(candidate); + return candidate; + }; + const ensureTurn = (line: number): string => { + activeTurnId ??= `claude-${transcript.id}-turn-${++turnCounter}-${line}`; + turns.add(activeTurnId); + return activeTurnId; + }; + const closeTurn = (turnId: string, ts: number): void => { + messages.push({ + type: 'turn_state', + id: uniqueId(`${turnId}-state`, ts), + turnId, + ts, + status: 'completed', + partialOutputRetained: true, + }); + }; + + for (const record of transcript.records) { + const value = record.value; + if (value.isSidechain === true) continue; + const ts = timestampFor(record); + if (value.type === 'user') { + const resultBlocks = claudeToolResultBlocks(value); + for (const block of resultBlocks) { + const originalToolId = + stringField(block, 'tool_use_id') ?? `${recordId(value) || 'claude'}-orphan-tool`; + const toolUseId = resolveToolId(originalToolId, toolIds, resolvedToolIds); + messages.push({ + type: 'tool_result', + id: uniqueId(`${recordId(value) || 'claude-result'}-result`, record.line), + turnId: ensureTurn(record.line), + ts, + toolUseId, + isError: block.is_error === true, + content: claudeToolResultContent(block.content), + }); + resolvedToolIds.add(toolUseId); + } + const userText = claudeUserAuthoredText(value); + if (userText !== undefined) { + if (activeTurnId !== undefined) closeTurn(activeTurnId, ts); + activeTurnId = `claude-${transcript.id}-turn-${++turnCounter}-${record.line}`; + turns.add(activeTurnId); + messages.push({ + type: 'user', + id: uniqueId(recordId(value) || `claude-user-${record.line}`, record.line), + turnId: activeTurnId, + ts, + text: userText, + }); + } + continue; + } + if (value.type === 'assistant') { + const turnId = ensureTurn(record.line); + const blocks = claudeContentBlocks(value); + const textParts: string[] = []; + const thinkingParts: { text: string; signature?: string }[] = []; + const contentOrder: ('thinking' | 'text' | 'tools')[] = []; + for (const block of blocks) { + const type = stringField(block, 'type'); + if (type === 'text' && typeof block.text === 'string') { + textParts.push(block.text); + if (!contentOrder.includes('text')) contentOrder.push('text'); + } else if (type === 'thinking' && typeof block.thinking === 'string') { + thinkingParts.push({ + text: block.thinking, + ...(typeof block.signature === 'string' ? { signature: block.signature } : {}), + }); + if (!contentOrder.includes('thinking')) contentOrder.push('thinking'); + } else if (type === 'tool_use') { + if (!contentOrder.includes('tools')) contentOrder.push('tools'); + } + } + if (textParts.length > 0 || thinkingParts.length > 0) { + const thinking = + thinkingParts.length > 0 + ? { + text: thinkingParts.map((part) => part.text).join(''), + ...(thinkingParts.length > 1 ? { parts: thinkingParts } : {}), + } + : undefined; + messages.push({ + type: 'assistant', + id: uniqueId(recordId(value) || `claude-assistant-${record.line}`, record.line), + turnId, + ts, + text: textParts.join('\n'), + ...(thinking ? { thinking } : {}), + contentOrder, + modelId: + stringField(asRecord(value.message), 'model') ?? + stringField(value, 'model') ?? + 'claude', + }); + } + for (const block of blocks.filter((item) => item.type === 'tool_use')) { + const rawToolId = stringField(block, 'id') ?? `claude-tool-${record.line}`; + const toolId = uniqueId(rawToolId, record.line); + const list = toolIds.get(rawToolId) ?? []; + list.push(toolId); + toolIds.set(rawToolId, list); + messages.push({ + type: 'tool_call', + id: toolId, + turnId, + ts, + toolName: stringField(block, 'name') ?? 'unknown', + args: block.input ?? {}, + }); + } + continue; + } + if (value.type === 'summary' || value.isCompactSummary === true) { + messages.push({ + type: 'system_note', + id: uniqueId(`${recordId(value) || 'claude-summary'}-compact`, record.line), + turnId: activeTurnId, + ts, + kind: 'context_compacted', + data: typeof value.summary === 'string' ? value.summary : undefined, + }); + continue; + } + if (isClaudeRewindRecord(value)) { + messages.push({ + type: 'system_note', + id: uniqueId(`${recordId(value) || 'claude-rewind'}-rewind`, record.line), + turnId: activeTurnId, + ts, + kind: 'session_resume', + data: { source: 'claude', record: JSON.parse(JSON.stringify(value)) as unknown }, + }); + } + } + if (activeTurnId !== undefined && turns.has(activeTurnId)) closeTurn(activeTurnId, lastTimestamp); + return { + sourceSessionId: transcript.id, + metadata: { + name: entry.title, + cwd: entry.cwd, + }, + messages, + }; +} + +function claudeContentBlocks(record: JsonRecord): JsonRecord[] { + const message = asRecord(record.message); + const content = message?.content; + if (!Array.isArray(content)) return []; + return content.filter(isRecord); +} + +function claudeToolResultBlocks(record: JsonRecord): JsonRecord[] { + return claudeContentBlocks(record).filter((block) => block.type === 'tool_result'); +} + +function claudeToolResultContent(value: unknown): ToolResultContent { + if (typeof value === 'string') return { kind: 'text', text: value }; + if (Array.isArray(value)) { + const text = value + .filter(isRecord) + .filter((block) => block.type === 'text' && typeof block.text === 'string') + .map((block) => block.text as string) + .join('\n'); + if (text.length > 0) return { kind: 'text', text }; + } + return { kind: 'json', value: value ?? null }; +} + +function resolveToolId( + originalId: string, + toolIds: Map, + resolvedToolIds: Set, +): string { + const candidates = toolIds.get(originalId) ?? []; + const unresolved = candidates.find((id) => !resolvedToolIds.has(id)); + return unresolved ?? originalId; +} + +function recordId(record: JsonRecord): string { + return stringField(record, 'uuid') ?? stringField(record, 'id') ?? ''; +} + +function isClaudeRewindRecord(record: JsonRecord): boolean { + const type = stringField(record, 'type')?.toLowerCase(); + const subtype = stringField(record, 'subtype')?.toLowerCase(); + return record.isRewind === true || type === 'rewind' || subtype?.includes('rewind') === true; +} + +function compareClaudeCatalogEntries(a: ClaudeCatalogEntry, b: ClaudeCatalogEntry): number { + return b.updatedAtMs - a.updatedAtMs || a.transcriptPath.localeCompare(b.transcriptPath); +} + +function asRecord(value: unknown): JsonRecord | undefined { + return isRecord(value) ? value : undefined; +} + +function isRecord(value: unknown): value is JsonRecord { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function stringField(record: JsonRecord | undefined, field: string): string | undefined { + const value = record?.[field]; + return typeof value === 'string' && value.length > 0 ? value : undefined; +} + +function parseTimestampMs(value: unknown): number | undefined { + if (typeof value === 'number' && Number.isFinite(value)) { + return value < 1_000_000_000_000 ? value * 1000 : value; + } + if (typeof value === 'string') { + const numeric = Number(value); + if (Number.isFinite(numeric) && numeric > 0) return parseTimestampMs(numeric); + const parsed = Date.parse(value); + return Number.isFinite(parsed) ? parsed : undefined; + } + return undefined; +} + +async function readClaudeHead(path: string): Promise { + for (const size of [ + FOREIGN_SESSION_HEAD_BYTES, + CLAUDE_HEAD_GROWTH_BYTES, + CLAUDE_HEAD_GROWTH_BYTES * 4, + CLAUDE_HEAD_MAX_BYTES, + ]) { + const text = await readUtf8Prefix(path, size).catch(() => undefined); + if (text === undefined) return undefined; + const meta: ClaudeTranscriptMeta = {}; + for (const line of text.split('\n')) { + const record = parseForeignJsonLine(line); + if (record) collectClaudeMeta(record, meta); + } + if (meta.cwd !== undefined && meta.isSidechain !== undefined) return text; + } + return readUtf8Prefix(path, CLAUDE_HEAD_MAX_BYTES).catch(() => undefined); +} + +async function readTranscriptWindow( + path: string, + side: 'head' | 'tail', + maxBytes: number, +): Promise { + if (side === 'head') return readUtf8Prefix(path, maxBytes).catch(() => undefined); + return readUtf8Tail(path, maxBytes) + .then(({ text }) => text) + .catch(() => undefined); +} + +async function listSubdirectories(root: string): Promise { + try { + return (await readdir(root, { withFileTypes: true })) + .filter((entry) => entry.isDirectory()) + .map((entry) => join(root, entry.name)); + } catch { + return []; + } +} + +async function listJsonlFiles(dir: string): Promise<{ path: string; mtimeMs: number }[]> { + const files: { path: string; mtimeMs: number }[] = []; + try { + for (const entry of await readdir(dir, { withFileTypes: true })) { + if (!entry.isFile() || !entry.name.endsWith('.jsonl')) continue; + const path = join(dir, entry.name); + files.push({ path, mtimeMs: (await stat(path)).mtimeMs }); + } + } catch { + // Foreign stores can change while they are being scanned. + } + return files; +} + +async function isDirectory(path: string): Promise { + try { + return (await stat(path)).isDirectory(); + } catch { + return false; + } +} + +async function isFile(path: string): Promise { + try { + return (await stat(path)).isFile(); + } catch { + return false; + } +} diff --git a/packages/storage/src/codex-session-adapter.ts b/packages/storage/src/codex-session-adapter.ts index 74ec33a0bb..06dadf8ceb 100644 --- a/packages/storage/src/codex-session-adapter.ts +++ b/packages/storage/src/codex-session-adapter.ts @@ -1,15 +1,33 @@ import type { Dirent } from 'node:fs'; -import { open, readdir, realpath, stat } from 'node:fs/promises'; +import { readdir, realpath, stat } from 'node:fs/promises'; import { homedir } from 'node:os'; import { basename, join, resolve, sep } from 'node:path'; import type { StoredMessage } from '@maka/core/session'; -import { sanitizeForeignTitle } from '@maka/core/foreign-session'; +import { + FOREIGN_SESSION_DIGEST_MAX_READ_BYTES, + codexRolloutMessage, + createDigestAccumulator, + finishDigest, + pushDigestMessage, + sanitizeForeignTitle, + type ForeignSessionDigest, + type ForeignSessionSummary, +} from '@maka/core/foreign-session'; import type { ExternalMakaSession, ExternalSessionAdapter, ExternalSessionQuery, ExternalSessionSummary, } from '@maka/core/external-session'; +import { + matchesSourceCatalogQuery, + normalizeSourcePath, + readBoundedUtf8File, + readUtf8Prefix, + readUtf8Tail, + type ExternalSourceCatalogEntry, + type ExternalSourceCatalogQuery, +} from './external-source-catalog.js'; export const CODEX_SESSION_ADAPTER_ID = 'codex'; export const CODEX_ROLLOUT_MAX_BYTES = 64 * 1024 * 1024; @@ -19,6 +37,13 @@ const CODEX_SESSION_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/; const CODEX_UNSAFE_PATH_CHARS = /[\u0000-\u001F\u007F\u0080-\u009F\u061C\u200B-\u200F\u202A-\u202E\u2060-\u2064\u2066-\u2069\uFEFF]/; const CODEX_ROOT_SOURCE_TOKENS = new Set(['cli', 'exec', 'vscode']); +const CODEX_ROOT_SOURCE_SQL_VALUES = [ + 'cli', + 'exec', + 'vscode', + '{"custom":"atlas"}', + '{"custom":"chatgpt"}', +]; export interface CodexSessionAdapterOptions { /** Codex's state root. Defaults to `$CODEX_HOME`, then `~/.codex`. */ @@ -27,8 +52,8 @@ export interface CodexSessionAdapterOptions { maxRolloutBytes?: number; } -interface CodexCatalogEntry extends ExternalSessionSummary { - rolloutPath: string; +interface CodexCatalogEntry extends ExternalSourceCatalogEntry { + source: 'codex'; } interface CodexThreadRow { @@ -84,7 +109,20 @@ export class CodexSessionAdapter implements ExternalSessionAdapter { async listSessions(query: ExternalSessionQuery = {}): Promise { const entries = await this.listCatalog(query); - return entries.map(({ rolloutPath: _rolloutPath, ...summary }) => summary); + return entries.map((entry) => ({ + id: entry.id, + name: entry.title, + cwd: entry.cwd, + ...(entry.createdAtMs !== undefined ? { createdAt: entry.createdAtMs } : {}), + updatedAt: entry.updatedAtMs, + archived: entry.archived, + })); + } + + async listCatalogEntries( + query: ExternalSourceCatalogQuery = {}, + ): Promise { + return this.listCatalog(query); } async readSession(sessionId: string): Promise { @@ -92,10 +130,10 @@ export class CodexSessionAdapter implements ExternalSessionAdapter { const catalogEntry = await this.findCatalogEntry(sessionId); if (!catalogEntry) throw new Error(`Codex Session not found: ${sessionId}`); - const rolloutPath = await this.resolveRolloutPath(catalogEntry.rolloutPath, sessionId); + const rolloutPath = await this.resolveRolloutPath(catalogEntry.transcriptPath, sessionId); if (!rolloutPath) throw new Error(`Codex rollout is unavailable: ${sessionId}`); const text = await readBoundedUtf8File(rolloutPath, this.maxRolloutBytes); - const converted = convertCodexRollout(text, sessionId, catalogEntry.name, catalogEntry.cwd); + const converted = convertCodexRollout(text, sessionId, catalogEntry.title, catalogEntry.cwd); return { sourceSessionId: sessionId, @@ -104,32 +142,65 @@ export class CodexSessionAdapter implements ExternalSessionAdapter { }; } - private async listCatalog(query: ExternalSessionQuery): Promise { + async readDigest(summary: ForeignSessionSummary): Promise { + if (summary.source !== 'codex') throw new Error('Codex adapter received another source'); + const rolloutPath = await this.resolveRolloutPath(summary.transcriptPath, summary.id); + if (!rolloutPath) throw new Error(`Codex rollout is unavailable: ${summary.id}`); + const { text, truncated } = await readUtf8Tail( + rolloutPath, + FOREIGN_SESSION_DIGEST_MAX_READ_BYTES, + ); + const acc = createDigestAccumulator(); + let dropped = 0; + const records: ParsedRolloutRecord[] = []; + for (const [index, line] of text.split('\n').entries()) { + if (line.trim().length === 0) continue; + try { + const value = JSON.parse(line) as unknown; + if (!isRecord(value)) throw new Error('record is not an object'); + records.push({ line: index + 1, value }); + } catch { + dropped += 1; + } + } + for (const message of codexPresentationRecords(records).values()) { + if (message.kind !== 'reasoning') pushDigestMessage(acc, message.kind, message.text); + } + if (truncated) { + acc.warnings.push( + `transcript exceeded ${FOREIGN_SESSION_DIGEST_MAX_READ_BYTES} bytes; only its tail was read`, + ); + } + if (dropped > 0) acc.warnings.push(`${dropped} malformed transcript lines were skipped`); + return finishDigest(acc, { + source: summary.source, + id: summary.id, + title: summary.title, + cwd: summary.cwd, + gitBranch: summary.gitBranch, + updatedAtMs: summary.updatedAtMs, + }); + } + + private async listCatalog(query: ExternalSourceCatalogQuery): Promise { for (const dbPath of await codexStateDbsNewestFirst(this.codexHome)) { const rows = await readCodexThreadRows(dbPath, query); if (rows === undefined) continue; const entries = await Promise.all(rows.map((row) => this.entryFromRow(row))); return entries .filter((entry): entry is CodexCatalogEntry => entry !== undefined) - .filter((entry) => matchesQuery(entry, query)) - .sort(compareCatalogEntries); + .filter((entry) => matchesSourceCatalogQuery(entry, query)) + .sort(compareCatalogEntries) + .slice(0, query.limit); } return this.scanRolloutCatalog(query); } private async findCatalogEntry(sessionId: string): Promise { - for (const dbPath of await codexStateDbsNewestFirst(this.codexHome)) { - const rows = await readCodexThreadRows(dbPath, { includeArchived: true }, sessionId); - if (rows === undefined) continue; - for (const row of rows) { - const entry = await this.entryFromRow(row); - if (entry?.id === sessionId) return entry; - } - break; - } - - return this.findRolloutEntry(sessionId); + return (await this.listCatalog({ includeArchived: true })).find( + (entry) => entry.id === sessionId, + ); } private async entryFromRow(row: CodexThreadRow): Promise { @@ -145,55 +216,43 @@ export class CodexSessionAdapter implements ExternalSessionAdapter { const updatedAt = normalizeEpochMs(row.updated_at_ms) ?? normalizeEpochMs(row.updated_at); return { + source: 'codex', id: row.id, - name, + title: name, cwd: safeCodexCwd(row.cwd), - ...(createdAt !== undefined ? { createdAt } : {}), - ...(updatedAt !== undefined ? { updatedAt } : {}), + ...(createdAt !== undefined ? { createdAtMs: createdAt } : {}), + updatedAtMs: updatedAt ?? 0, archived: row.archived === true || row.archived === 1, - rolloutPath, + transcriptPath: rolloutPath, }; } - private async scanRolloutCatalog(query: ExternalSessionQuery): Promise { + private async scanRolloutCatalog( + query: ExternalSourceCatalogQuery, + ): Promise { + if (query.limit !== undefined && query.limit <= 0) return []; const candidates = [ ...(await walkRolloutFiles(join(this.codexHome, 'sessions'), false)), ...(query.includeArchived ? await walkRolloutFiles(join(this.codexHome, 'archived_sessions'), true) : []), - ].sort((a, b) => b.mtimeMs - a.mtimeMs); + ].sort(compareRolloutCandidates); const entries: CodexCatalogEntry[] = []; + const seenIds = new Set(); for (const candidate of candidates) { const head = await readUtf8Prefix(candidate.path, CODEX_ROLLOUT_HEAD_BYTES).catch( () => undefined, ); if (head === undefined) continue; const entry = catalogEntryFromRolloutHead(head, candidate); - if (!entry || !matchesQuery(entry, query)) continue; - const rolloutPath = await this.resolveRolloutPath(candidate.path, entry.id); - if (rolloutPath) entries.push({ ...entry, rolloutPath }); + if (!entry || !matchesSourceCatalogQuery(entry, query)) continue; + if (seenIds.has(entry.id)) continue; + seenIds.add(entry.id); + const transcriptPath = await this.resolveRolloutPath(entry.transcriptPath, entry.id); + if (transcriptPath) entries.push({ ...entry, transcriptPath }); } - return entries.sort(compareCatalogEntries); - } - - private async findRolloutEntry(sessionId: string): Promise { - for (const [root, archived] of [ - [join(this.codexHome, 'sessions'), false], - [join(this.codexHome, 'archived_sessions'), true], - ] as const) { - for (const candidate of await walkRolloutFiles(root, archived)) { - if (!rolloutFilenameMatchesId(basename(candidate.path), sessionId)) continue; - const head = await readUtf8Prefix(candidate.path, CODEX_ROLLOUT_HEAD_BYTES).catch( - () => undefined, - ); - if (head === undefined) continue; - const entry = catalogEntryFromRolloutHead(head, candidate); - if (entry?.id !== sessionId) continue; - const rolloutPath = await this.resolveRolloutPath(candidate.path, sessionId); - if (rolloutPath) return { ...entry, rolloutPath }; - } - } - return undefined; + entries.sort(compareCatalogEntries); + return query.limit === undefined ? entries : entries.slice(0, query.limit); } private async resolveRolloutPath( @@ -226,6 +285,7 @@ function convertCodexRollout( fallbackCwd: string, ): ExternalMakaSession { const records = parseRolloutRecords(text, expectedSessionId); + const presentationByLine = codexPresentationRecords(records); const sessionMeta = records.find((record) => record.value.type === 'session_meta')?.value; const metaPayload = asRecord(sessionMeta?.payload); const actualSessionId = stringField(metaPayload, 'session_id') ?? stringField(metaPayload, 'id'); @@ -233,7 +293,6 @@ function convertCodexRollout( throw new Error(`Codex rollout Session id mismatch: expected ${expectedSessionId}`); } - const metaCwd = safeCodexCwd(metaPayload?.cwd); const messages: StoredMessage[] = []; let activeTurnId: string | undefined; let activeTurnIsExplicit = false; @@ -264,65 +323,59 @@ function convertCodexRollout( continue; } - if (envelope.type === 'event_msg') { - const eventType = stringField(payload, 'type'); - if (eventType === 'task_started' || eventType === 'turn_started') { - const turnId = stringField(payload, 'turn_id'); - if (turnId) { - activeTurnId = turnId; - activeTurnIsExplicit = true; - } - continue; + const presentation = presentationByLine.get(record.line); + if (presentation) { + const eventTurnId = stringField(payload, 'turn_id'); + if (eventTurnId) { + activeTurnId = eventTurnId; + activeTurnIsExplicit = true; + } else if (presentation.kind === 'user' && !activeTurnIsExplicit) { + activeTurnId = generatedCodexId(expectedSessionId, 'turn', record.line); } - - if (eventType === 'user_message') { - if (!activeTurnIsExplicit) { - activeTurnId = generatedCodexId(expectedSessionId, 'turn', record.line); - } - const text = stringField(payload, 'message') ?? mediaOnlyUserText(payload); - if (text.length === 0) continue; - firstUserText ??= text; - const turnId = ensureTurnId(record.line); + const turnId = ensureTurnId(record.line); + const ts = timestampFor(record); + if (presentation.kind === 'user') { + firstUserText ??= presentation.text; messages.push({ type: 'user', - id: - stringField(payload, 'client_id') ?? - generatedCodexId(expectedSessionId, 'user', record.line), + id: presentation.id ?? generatedCodexId(expectedSessionId, 'user', record.line), turnId, - ts: timestampFor(record), - text, + ts, + text: presentation.text, }); - continue; - } - - if (eventType === 'agent_message') { - const text = stringField(payload, 'message'); - if (!text) continue; + } else if (presentation.kind === 'assistant') { messages.push({ type: 'assistant', - id: generatedCodexId(expectedSessionId, 'assistant', record.line), - turnId: ensureTurnId(record.line), - ts: timestampFor(record), - text, + id: presentation.id ?? generatedCodexId(expectedSessionId, 'assistant', record.line), + turnId, + ts, + text: presentation.text, modelId: activeModel, contentOrder: ['text'], }); - continue; - } - - if (eventType === 'agent_reasoning') { - const reasoning = stringField(payload, 'text'); - if (!reasoning) continue; + } else { messages.push({ type: 'assistant', - id: generatedCodexId(expectedSessionId, 'reasoning', record.line), - turnId: ensureTurnId(record.line), - ts: timestampFor(record), + id: presentation.id ?? generatedCodexId(expectedSessionId, 'reasoning', record.line), + turnId, + ts, text: '', - thinking: { text: reasoning }, + thinking: { text: presentation.text }, contentOrder: ['thinking'], modelId: activeModel, }); + } + continue; + } + + if (envelope.type === 'event_msg') { + const eventType = stringField(payload, 'type'); + if (eventType === 'task_started' || eventType === 'turn_started') { + const turnId = stringField(payload, 'turn_id'); + if (turnId) { + activeTurnId = turnId; + activeTurnIsExplicit = true; + } continue; } @@ -437,7 +490,7 @@ function convertCodexRollout( sanitizeForeignTitle(fallbackName) || sanitizeForeignTitle(firstUserText) || expectedSessionId; return { sourceSessionId: expectedSessionId, - metadata: { name, cwd: metaCwd || fallbackCwd }, + metadata: { name, cwd: fallbackCwd }, messages, }; } @@ -447,6 +500,105 @@ interface ParsedRolloutRecord { value: JsonRecord; } +interface CodexPresentationMessage { + kind: 'user' | 'assistant' | 'reasoning'; + text: string; + id?: string; +} + +function codexPresentationRecords( + records: readonly ParsedRolloutRecord[], +): Map { + const eventMessages = new Map(); + const responseMessages = new Map(); + const seenCompletedItemIds = new Set(); + for (const record of records) { + const envelope = record.value; + if (envelope.type === 'response_item') { + const message = codexRolloutMessage(envelope); + if (message) { + responseMessages.set(record.line, { + kind: message.role, + text: message.text, + }); + } + continue; + } + if (envelope.type !== 'event_msg') continue; + const payload = asRecord(envelope.payload); + const eventType = stringField(payload, 'type'); + if (!payload || !eventType) continue; + + if (eventType === 'user_message') { + const text = stringField(payload, 'message') ?? mediaOnlyUserText(payload); + if (text.length > 0) { + eventMessages.set(record.line, { + kind: 'user', + text, + ...(stringField(payload, 'client_id') ? { id: stringField(payload, 'client_id') } : {}), + }); + } + continue; + } + if (eventType === 'agent_message') { + const text = stringField(payload, 'message'); + if (text) eventMessages.set(record.line, { kind: 'assistant', text }); + continue; + } + if (eventType === 'agent_reasoning') { + const text = stringField(payload, 'text'); + if (text) eventMessages.set(record.line, { kind: 'reasoning', text }); + continue; + } + if (eventType !== 'item_completed') continue; + + const item = asRecord(payload.item); + const itemType = stringField(item, 'type')?.toLowerCase(); + const itemId = stringField(item, 'id') ?? stringField(item, 'client_id'); + if (itemId !== undefined) { + if (seenCompletedItemIds.has(itemId)) continue; + seenCompletedItemIds.add(itemId); + } + if (itemType === 'usermessage') { + const text = codexCompletedItemText(item) || codexCompletedItemMediaText(item); + if (text.length > 0) { + eventMessages.set(record.line, { + kind: 'user', + text, + ...((stringField(item, 'client_id') ?? stringField(item, 'id')) + ? { id: stringField(item, 'client_id') ?? stringField(item, 'id') } + : {}), + }); + } + } else if (itemType === 'agentmessage') { + const text = codexCompletedItemText(item); + if (text.length > 0) { + eventMessages.set(record.line, { + kind: 'assistant', + text, + ...(stringField(item, 'id') ? { id: stringField(item, 'id') } : {}), + }); + } + } else if (itemType === 'reasoning') { + const text = codexCompletedReasoningText(item); + if (text.length > 0) { + eventMessages.set(record.line, { + kind: 'reasoning', + text, + ...(stringField(item, 'id') ? { id: stringField(item, 'id') } : {}), + }); + } + } + } + if (eventMessages.size === 0) return responseMessages; + const merged = new Map(eventMessages); + const eventKinds = new Set([...eventMessages.values()].map((message) => message.kind)); + for (const [line, responseMessage] of responseMessages) { + if (!eventKinds.has(responseMessage.kind)) merged.set(line, responseMessage); + } + return merged; +} + function parseRolloutRecords(text: string, sessionId: string): ParsedRolloutRecord[] { const endsWithNewline = text.endsWith('\n'); const lines = text.split('\n'); @@ -471,12 +623,13 @@ function parseRolloutRecords(text: string, sessionId: string): ParsedRolloutReco function catalogEntryFromRolloutHead( text: string, candidate: RolloutCandidate, -): Omit | undefined { +): CodexCatalogEntry | undefined { const lines = text.split('\n'); let id: string | undefined; let cwd = ''; let createdAt: number | undefined; let firstUserText: string | undefined; + let responseUserText: string | undefined; for (const line of lines) { let record: JsonRecord; try { @@ -494,31 +647,39 @@ function catalogEntryFromRolloutHead( cwd = safeCodexCwd(payload.cwd) || cwd; createdAt = normalizeEpochMs(record.timestamp) ?? normalizeEpochMs(payload.timestamp) ?? createdAt; - } else if ( - record.type === 'event_msg' && - payload.type === 'user_message' && - firstUserText === undefined - ) { - firstUserText = stringField(payload, 'message'); + } else if (firstUserText === undefined) { + if (record.type === 'event_msg' && payload.type === 'user_message') { + firstUserText = stringField(payload, 'message'); + } else if (record.type === 'event_msg' && payload.type === 'item_completed') { + const item = asRecord(payload.item); + if (stringField(item, 'type')?.toLowerCase() === 'usermessage') { + firstUserText = codexCompletedItemText(item) || codexCompletedItemMediaText(item); + } + } else { + const message = codexRolloutMessage(record); + if (message?.role === 'user') responseUserText ??= message.text; + } } if (id && firstUserText !== undefined) break; } + firstUserText ??= responseUserText; if (!isSafeCodexSessionId(id)) return undefined; if (!rolloutFilenameMatchesId(basename(candidate.path), id)) return undefined; return { + source: 'codex', id, - name: sanitizeForeignTitle(firstUserText) || id, + title: sanitizeForeignTitle(firstUserText) || id, cwd, - ...(createdAt !== undefined ? { createdAt } : {}), - updatedAt: candidate.mtimeMs, + ...(createdAt !== undefined ? { createdAtMs: createdAt } : {}), + updatedAtMs: candidate.mtimeMs, archived: candidate.archived, + transcriptPath: candidate.path, }; } async function readCodexThreadRows( dbPath: string, - query: ExternalSessionQuery, - exactId?: string, + query: ExternalSourceCatalogQuery, ): Promise { try { const sqlite = await import('node:sqlite'); @@ -547,13 +708,15 @@ async function readCodexThreadRows( ].filter((column) => columns.has(column)); const where: string[] = []; const params: Array = []; - if (exactId !== undefined) { - where.push('id = ?'); - params.push(exactId); - } if (!query.includeArchived && columns.has('archived')) { where.push('(archived IS NULL OR archived = 0)'); } + if (columns.has('source')) { + where.push( + `(source IS NULL OR source IN (${CODEX_ROOT_SOURCE_SQL_VALUES.map(() => '?').join(', ')}))`, + ); + params.push(...CODEX_ROOT_SOURCE_SQL_VALUES); + } if (query.cwd !== undefined && columns.has('cwd')) { const variants = cwdSqlVariants(query.cwd); where.push(`cwd IN (${variants.map(() => '?').join(', ')})`); @@ -567,7 +730,11 @@ async function readCodexThreadRows( const sql = `SELECT ${wanted.join(', ')} FROM threads` + (where.length > 0 ? ` WHERE ${where.join(' AND ')}` : '') + - ` ORDER BY ${orderColumn} DESC`; + ` ORDER BY ${orderColumn} DESC` + + (query.limit !== undefined ? ' LIMIT ?' : ''); + if (query.limit !== undefined) { + params.push(Math.max(0, Math.floor(query.limit * 2))); + } return db.prepare(sql).all(...params) as CodexThreadRow[]; } finally { db.close(); @@ -625,47 +792,6 @@ async function walkRolloutFiles(root: string, archived: boolean): Promise { - const handle = await open(path, 'r'); - try { - const metadata = await handle.stat(); - if (!metadata.isFile()) throw new Error('Codex rollout is not a regular file'); - if (metadata.size > maxBytes) throw new Error(`Codex rollout exceeds ${maxBytes} bytes`); - const chunks: Buffer[] = []; - let total = 0; - for (;;) { - const buffer = Buffer.allocUnsafe(Math.min(64 * 1024, maxBytes + 1 - total)); - const { bytesRead } = await handle.read(buffer, 0, buffer.length, total); - if (bytesRead === 0) break; - total += bytesRead; - if (total > maxBytes) throw new Error(`Codex rollout exceeds ${maxBytes} bytes`); - chunks.push(buffer.subarray(0, bytesRead)); - if (total === maxBytes) { - const probe = Buffer.allocUnsafe(1); - if ((await handle.read(probe, 0, 1, total)).bytesRead > 0) { - throw new Error(`Codex rollout exceeds ${maxBytes} bytes`); - } - break; - } - } - return Buffer.concat(chunks, total).toString('utf8'); - } finally { - await handle.close(); - } -} - -async function readUtf8Prefix(path: string, maxBytes: number): Promise { - const handle = await open(path, 'r'); - try { - if (!(await handle.stat()).isFile()) throw new Error('Codex rollout is not a regular file'); - const buffer = Buffer.allocUnsafe(maxBytes); - const { bytesRead } = await handle.read(buffer, 0, maxBytes, 0); - return buffer.subarray(0, bytesRead).toString('utf8'); - } finally { - await handle.close(); - } -} - function asRecord(value: unknown): JsonRecord | undefined { return isRecord(value) ? value : undefined; } @@ -733,13 +859,8 @@ function normalizeEpochMs(value: unknown): number | undefined { return undefined; } -function normalizePath(value: string): string { - const normalized = value.replaceAll('\\', '/').replace(/\/+$/, ''); - return /^[A-Za-z]:\//.test(normalized) ? normalized.toLowerCase() : normalized; -} - function cwdSqlVariants(cwd: string): string[] { - const normalized = normalizePath(cwd); + const normalized = normalizeSourcePath(cwd); const variants = new Set([cwd, normalized]); if (/^[A-Za-z]:\//.test(normalized)) variants.add(normalized.replaceAll('/', '\\')); if (normalized !== '/') { @@ -749,13 +870,16 @@ function cwdSqlVariants(cwd: string): string[] { return [...variants]; } -function matchesQuery(entry: ExternalSessionSummary, query: ExternalSessionQuery): boolean { - if (!query.includeArchived && entry.archived) return false; - return query.cwd === undefined || normalizePath(entry.cwd) === normalizePath(query.cwd); +function compareCatalogEntries(a: CodexCatalogEntry, b: CodexCatalogEntry): number { + return ( + b.updatedAtMs - a.updatedAtMs || + (b.createdAtMs ?? 0) - (a.createdAtMs ?? 0) || + a.transcriptPath.localeCompare(b.transcriptPath) + ); } -function compareCatalogEntries(a: CodexCatalogEntry, b: CodexCatalogEntry): number { - return (b.updatedAt ?? b.createdAt ?? 0) - (a.updatedAt ?? a.createdAt ?? 0); +function compareRolloutCandidates(a: RolloutCandidate, b: RolloutCandidate): number { + return b.mtimeMs - a.mtimeMs || a.path.localeCompare(b.path); } function stateGeneration(path: string): number { @@ -802,6 +926,50 @@ function codexToolOutputText(value: unknown): string { } } +function codexCompletedItemText(item: JsonRecord | undefined): string { + if (!item) return ''; + const direct = stringField(item, 'content'); + if (direct) return direct; + if (!Array.isArray(item.content)) return ''; + return item.content + .flatMap((part) => { + const record = asRecord(part); + const type = stringField(record, 'type')?.toLowerCase(); + return type === 'text' || type === 'input_text' || type === 'output_text' + ? [stringField(record, 'text') ?? ''] + : []; + }) + .filter((text) => text.length > 0) + .join('\n'); +} + +function codexCompletedReasoningText(item: JsonRecord | undefined): string { + if (!item) return ''; + const summary = codexTextFragments(item.summary_text); + if (summary.length > 0) return summary.join('\n'); + return codexCompletedItemText(item); +} + +function codexTextFragments(value: unknown): string[] { + if (typeof value === 'string') return value.length > 0 ? [value] : []; + if (!Array.isArray(value)) return []; + return value.flatMap((part) => { + if (typeof part === 'string') return part.length > 0 ? [part] : []; + const text = stringField(asRecord(part), 'text'); + return text ? [text] : []; + }); +} + +function codexCompletedItemMediaText(item: JsonRecord | undefined): string { + if (!item || !Array.isArray(item.content)) return ''; + const contentTypes = item.content.flatMap((part) => { + const type = stringField(asRecord(part), 'type')?.toLowerCase(); + return type ? [type] : []; + }); + if (contentTypes.some((type) => type.includes('image'))) return '[Image]'; + return contentTypes.some((type) => type.includes('audio')) ? '[Audio]' : ''; +} + function mediaOnlyUserText(payload: JsonRecord): string { const images = Array.isArray(payload.images) ? payload.images : []; const localImages = Array.isArray(payload.local_images) ? payload.local_images : []; diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index ff2dec15de..fe09ddc5ef 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -6,7 +6,13 @@ import type { } from '@maka/core/agent-run'; import type { RuntimeEvent, ToolBoundaryProtocol } from '@maka/core/runtime-event'; import type { RuntimeContinuationAuthorityStore } from '@maka/core/runtime-event-store'; -import type { SessionHeader, SessionSummary, StoredMessage, TurnRecord } from '@maka/core/session'; +import type { + SessionExternalOrigin, + SessionHeader, + SessionSummary, + StoredMessage, + TurnRecord, +} from '@maka/core/session'; import type { SessionListFilter } from '@maka/core/runtime-inputs'; import { createSqliteAgentRunStore, @@ -321,8 +327,8 @@ async function createExecutionStoresForWrite run(() => sessionStore.ready()), create: (input, initialBoundary) => run(() => sessionStore.create(input, initialBoundary)), - createImportedSession: (input, messages) => - run(() => sessionStore.createImportedSession(input, messages)), + createImportedSession: (input, messages, externalOrigin) => + run(() => sessionStore.createImportedSession(input, messages, externalOrigin)), probeStableSessionCreate: (sessionId, requestFingerprint) => run(() => sessionStore.probeStableSessionCreate(sessionId, requestFingerprint)), createStableSession: (request, initialBoundary) => diff --git a/packages/storage/src/external-session-adapters.ts b/packages/storage/src/external-session-adapters.ts index 51773e7e5e..f3c9baddc6 100644 --- a/packages/storage/src/external-session-adapters.ts +++ b/packages/storage/src/external-session-adapters.ts @@ -1,7 +1,12 @@ import { ExternalSessionAdapterRegistry } from '@maka/core/external-session'; +import { + ClaudeSessionAdapter, + type ClaudeSessionAdapterOptions, +} from './claude-session-adapter.js'; import { CodexSessionAdapter, type CodexSessionAdapterOptions } from './codex-session-adapter.js'; export interface ExternalSessionAdapterOptions { + claude?: ClaudeSessionAdapterOptions; codex?: CodexSessionAdapterOptions; } @@ -9,5 +14,8 @@ export interface ExternalSessionAdapterOptions { export function createExternalSessionAdapterRegistry( options: ExternalSessionAdapterOptions = {}, ): ExternalSessionAdapterRegistry { - return new ExternalSessionAdapterRegistry([new CodexSessionAdapter(options.codex)]); + return new ExternalSessionAdapterRegistry([ + new ClaudeSessionAdapter(options.claude), + new CodexSessionAdapter(options.codex), + ]); } diff --git a/packages/storage/src/external-session-importer.ts b/packages/storage/src/external-session-importer.ts index 0c517a614f..692c969dcd 100644 --- a/packages/storage/src/external-session-importer.ts +++ b/packages/storage/src/external-session-importer.ts @@ -35,6 +35,10 @@ export class ExternalSessionImporter { name: request.target.name ?? external.metadata.name, }, external.messages, + { + adapterId: request.adapterId, + sourceSessionId: request.sourceSessionId, + }, ); } } diff --git a/packages/storage/src/external-source-catalog.ts b/packages/storage/src/external-source-catalog.ts new file mode 100644 index 0000000000..4c15e22d99 --- /dev/null +++ b/packages/storage/src/external-source-catalog.ts @@ -0,0 +1,118 @@ +import { open } from 'node:fs/promises'; +import type { ExternalSessionQuery } from '@maka/core/external-session'; +import type { ForeignSessionSource } from '@maka/core/foreign-session'; + +/** Internal catalog row shared by source-specific catalogs and projections. */ +export interface ExternalSourceCatalogEntry { + source: ForeignSessionSource; + id: string; + title: string; + cwd: string; + updatedAtMs: number; + createdAtMs?: number; + gitBranch?: string; + archived?: boolean; + transcriptPath: string; +} + +export interface ExternalSourceCatalogQuery extends ExternalSessionQuery { + /** Optional handoff retention window. Full import leaves this unset. */ + maxAgeMs?: number; + nowMs?: number; + limit?: number; +} + +export function matchesSourceCatalogQuery( + entry: ExternalSourceCatalogEntry, + query: ExternalSourceCatalogQuery, +): boolean { + if (!query.includeArchived && entry.archived) return false; + if ( + query.cwd !== undefined && + normalizeSourcePath(entry.cwd) !== normalizeSourcePath(query.cwd) + ) { + return false; + } + if ( + query.maxAgeMs !== undefined && + query.nowMs !== undefined && + query.nowMs - entry.updatedAtMs > query.maxAgeMs + ) { + return false; + } + return true; +} + +export function normalizeSourcePath(value: string): string { + const normalized = value.replaceAll('\\', '/').replace(/\/+$/, ''); + return /^[A-Za-z]:\//.test(normalized) ? normalized.toLowerCase() : normalized; +} + +export async function readBoundedUtf8File(path: string, maxBytes: number): Promise { + const handle = await open(path, 'r'); + try { + const metadata = await handle.stat(); + if (!metadata.isFile()) throw new Error('External transcript is not a regular file'); + if (metadata.size > maxBytes) throw new Error(`External transcript exceeds ${maxBytes} bytes`); + const chunks: Buffer[] = []; + let total = 0; + for (;;) { + const buffer = Buffer.allocUnsafe(Math.min(64 * 1024, maxBytes + 1 - total)); + const { bytesRead } = await handle.read(buffer, 0, buffer.length, total); + if (bytesRead === 0) break; + total += bytesRead; + if (total > maxBytes) throw new Error(`External transcript exceeds ${maxBytes} bytes`); + chunks.push(buffer.subarray(0, bytesRead)); + if (total === maxBytes) { + const probe = Buffer.allocUnsafe(1); + if ((await handle.read(probe, 0, 1, total)).bytesRead > 0) { + throw new Error(`External transcript exceeds ${maxBytes} bytes`); + } + break; + } + } + return Buffer.concat(chunks, total).toString('utf8'); + } finally { + await handle.close(); + } +} + +export async function readUtf8Prefix(path: string, maxBytes: number): Promise { + const handle = await open(path, 'r'); + try { + if (!(await handle.stat()).isFile()) + throw new Error('External transcript is not a regular file'); + const buffer = Buffer.allocUnsafe(maxBytes); + const { bytesRead } = await handle.read(buffer, 0, maxBytes, 0); + return buffer.subarray(0, bytesRead).toString('utf8'); + } finally { + await handle.close(); + } +} + +export async function readUtf8Tail( + path: string, + maxBytes: number, +): Promise<{ text: string; truncated: boolean }> { + const handle = await open(path, 'r'); + try { + const metadata = await handle.stat(); + if (!metadata.isFile()) throw new Error('External transcript is not a regular file'); + if (metadata.size <= maxBytes) { + const buffer = Buffer.alloc(metadata.size); + await handle.read(buffer, 0, metadata.size, 0); + return { text: buffer.toString('utf8'), truncated: false }; + } + const start = metadata.size - maxBytes; + const buffer = Buffer.alloc(maxBytes); + await handle.read(buffer, 0, buffer.length, start); + const text = buffer.toString('utf8'); + const firstNewline = text.indexOf('\n'); + return { + text: firstNewline === -1 ? '' : text.slice(firstNewline + 1), + truncated: true, + }; + } finally { + await handle.close(); + } +} diff --git a/packages/storage/src/foreign-session-store.ts b/packages/storage/src/foreign-session-store.ts index ccebb769e8..93967481cc 100644 --- a/packages/storage/src/foreign-session-store.ts +++ b/packages/storage/src/foreign-session-store.ts @@ -1,66 +1,32 @@ /** - * Read-only scanner + digest reader over foreign agent session stores - * (#1057): Claude Code (~/.claude/projects) and Codex (~/.codex). + * Handoff projection over the shared external-session source catalog. * - * Boundary rules, in order of importance: - * - * 1. READ-ONLY. This store never writes, renames, locks, or truncates - * anything. It deliberately does NOT take the root-authority - * capability — that contract exists for Maka's own workspace; foreign - * stores belong to other tools and must stay byte-identical. - * 2. SCOPED. All reads resolve under the configured home directory's - * known subtrees (`.claude/projects`, `.codex`). Paths obtained from - * foreign metadata (Codex `rollout_path`) are realpath-checked to - * still live inside the source root — a hostile row cannot point the - * reader at ~/.ssh. - * 3. BOUNDED. Byte caps from @maka/core/foreign-session apply to every - * read (head window for metadata, head+tail window for titles, hard - * cap for digests); scan results cap at 50 sessions / 30 days. - * 4. UNTRUSTED. All extracted text passes the core sanitize/redact gate; - * malformed lines and unreadable files are skipped, never fatal. - * - * Codex is read SQLite-first (node:sqlite, readOnly; column availability - * introspected via PRAGMA so version drift degrades gracefully) with a - * rollout-file directory walk as fallback. + * Source discovery, identity selection, bounded reads, and native parsing live + * in the Claude and Codex adapters. This module owns only the untrusted digest + * projection and its source enable flags. */ -import { open, readdir, realpath, stat, type FileHandle } from 'node:fs/promises'; import { homedir } from 'node:os'; -import { basename, join, resolve, sep } from 'node:path'; +import { join } from 'node:path'; import { - FOREIGN_SESSION_DIGEST_MAX_READ_BYTES, - FOREIGN_SESSION_HEAD_BYTES, FOREIGN_SESSION_SCAN_MAX_AGE_MS, FOREIGN_SESSION_SCAN_MAX_SESSIONS, - FOREIGN_SESSION_TITLE_WINDOW_BYTES, - claudeAssistantText, - claudeToolFilePaths, - claudeUserAuthoredText, - codexRolloutMessage, - codexRolloutSessionMeta, - collectClaudeMeta, - collectClaudeTitle, - createDigestAccumulator, - finishDigest, - isSafeForeignId, - normalizeCodexThreadRow, - parseForeignJsonLine, - pickClaudeTitle, - pushDigestFile, - pushDigestMessage, sanitizeForeignMessage, sanitizeForeignTitle, - type ClaudeTitleCandidates, - type ClaudeTranscriptMeta, - type CodexThreadRow, type ForeignSessionDigest, type ForeignSessionSource, type ForeignSessionSummary, } from '@maka/core/foreign-session'; +import { ClaudeSessionAdapter } from './claude-session-adapter.js'; +import { CodexSessionAdapter } from './codex-session-adapter.js'; +import { + normalizeSourcePath, + type ExternalSourceCatalogEntry, + type ExternalSourceCatalogQuery, +} from './external-source-catalog.js'; export interface ForeignSessionScanOptions { - /** Only sessions whose recorded cwd equals this path (after realpath-free - * string normalization). Empty/undefined lists across all cwds. */ + /** Only sessions whose recorded cwd matches this path. */ cwd?: string; } @@ -97,596 +63,91 @@ export function createForeignSessionStore( return new FileForeignSessionStore(options.homeDir ?? homedir(), options.env ?? process.env); } +interface CatalogReader { + readonly id: ForeignSessionSource; + detect(): Promise; + listCatalogEntries( + query: ExternalSourceCatalogQuery, + ): Promise; + readDigest(summary: ForeignSessionSummary): Promise; +} + class FileForeignSessionStore implements ForeignSessionStore { + private readonly readers: readonly CatalogReader[]; + constructor( private readonly homeDir: string, private readonly env: Record, - ) {} - - private get claudeRoot(): string { - return join(this.homeDir, '.claude', 'projects'); - } - - private get codexRoot(): string { - return join(this.homeDir, '.codex'); + ) { + this.readers = [ + new ClaudeSessionAdapter({ homeDir }), + new CodexSessionAdapter({ codexHome: join(homeDir, '.codex') }), + ]; } async availableSources(): Promise { - const sources: ForeignSessionSource[] = []; - if (isClaudeCodeImportEnabled(this.env) && (await isDirectory(this.claudeRoot))) { - sources.push('claude-code'); - } - if (isCodexImportEnabled(this.env) && (await isDirectory(this.codexRoot))) { - sources.push('codex'); + const available: ForeignSessionSource[] = []; + for (const reader of this.readers) { + if (!sourceEnabled(reader.id, this.env)) continue; + if (await reader.detect()) available.push(reader.id); } - return sources; + return available; } async listSessions(options: ForeignSessionScanOptions = {}): Promise { - const sources = await this.availableSources(); - const now = Date.now(); + const nowMs = Date.now(); + const query: ExternalSourceCatalogQuery = { + cwd: options.cwd, + maxAgeMs: FOREIGN_SESSION_SCAN_MAX_AGE_MS, + nowMs, + limit: FOREIGN_SESSION_SCAN_MAX_SESSIONS, + }; const results: ForeignSessionSummary[] = []; - if (sources.includes('claude-code')) { - results.push(...(await this.listClaudeSessions(options, now))); - } - if (sources.includes('codex')) { - results.push(...(await this.listCodexSessions(options, now))); + for (const reader of this.readers) { + if (!sourceEnabled(reader.id, this.env) || !(await reader.detect())) continue; + const entries = await reader.listCatalogEntries(query); + results.push(...entries.map(toForeignSummary)); } results.sort((a, b) => b.updatedAtMs - a.updatedAtMs); - // Sanitize + redact display metadata at the single return choke point. - // cwd matching upstream used the raw values, so it is safe to scrub the - // returned cwd/gitBranch here — a TUI consumer must never receive terminal - // control characters, bidi overrides, or secrets in these fields. (title - // and id are already gated at their source; transcriptPath stays raw as an - // internal lookup key confined to the source roots.) - return results.slice(0, FOREIGN_SESSION_SCAN_MAX_SESSIONS).map((s) => ({ - ...s, - cwd: sanitizeForeignMessage(s.cwd), - ...(s.gitBranch !== undefined ? { gitBranch: sanitizeForeignTitle(s.gitBranch) } : {}), + return results.slice(0, FOREIGN_SESSION_SCAN_MAX_SESSIONS).map((summary) => ({ + ...summary, + cwd: sanitizeForeignMessage(summary.cwd), + ...(summary.gitBranch !== undefined + ? { gitBranch: sanitizeForeignTitle(summary.gitBranch) } + : {}), })); } - /* ------------------------------ Claude ------------------------------ */ - - private async listClaudeSessions( - options: ForeignSessionScanOptions, - now: number, - ): Promise { - const projectDirs = await listSubdirectories(this.claudeRoot); - const candidates: { path: string; mtimeMs: number }[] = []; - for (const dir of projectDirs) { - for (const entry of await listFilesWithSuffix(dir, '.jsonl')) { - candidates.push(entry); - } - } - // Newest transcripts first so the per-source cap keeps the useful ones - // and old files never get opened at all. - candidates.sort((a, b) => b.mtimeMs - a.mtimeMs); - - const results: ForeignSessionSummary[] = []; - for (const candidate of candidates) { - if (results.length >= FOREIGN_SESSION_SCAN_MAX_SESSIONS) break; - if (now - candidate.mtimeMs > FOREIGN_SESSION_SCAN_MAX_AGE_MS) break; - const summary = await this.scanClaudeTranscript( - candidate.path, - candidate.mtimeMs, - options.cwd, - ); - if (summary) results.push(summary); - } - return results; - } - - private async scanClaudeTranscript( - path: string, - mtimeMs: number, - cwdFilter: string | undefined, - ): Promise { - const id = basename(path, '.jsonl'); - if (!isSafeForeignId(id)) return undefined; - - // cwd and isSidechain both live in the first `user`/`assistant` record, - // but a continued session can open with a run of `summary`/`mode` lines - // or a huge first message, so a fixed 4KB head silently misses them and - // drops the session. Grow the head window (64KB → 4MB) until cwd is seen. - // isSidechain is a per-file property (every record in the file carries the - // same value), so first-defined wins — no need to scan the whole file. - const meta: ClaudeTranscriptMeta = {}; - for (const record of await readClaudeHeadRecords(path)) { - collectClaudeMeta(record, meta); - if (meta.cwd !== undefined && meta.isSidechain !== undefined) break; - } - if (meta.isSidechain === true) return undefined; - if (meta.cwd === undefined) return undefined; - if (cwdFilter !== undefined && normalizePath(meta.cwd) !== normalizePath(cwdFilter)) - return undefined; - - // Title fields use last-wins (freshest title in the tail beats an older - // one); firstUserMessage uses first-wins (opening request). Feed the head - // window first, then the tail, so both semantics fall out of iteration - // order (see collectClaudeTitle). - const titles: ClaudeTitleCandidates = {}; - const titleHead = await readWindow(path, 'head', FOREIGN_SESSION_TITLE_WINDOW_BYTES); - const titleTail = await readWindow(path, 'tail', FOREIGN_SESSION_TITLE_WINDOW_BYTES); - for (const window of [titleHead, titleTail]) { - if (window === undefined) continue; - for (const line of window.split('\n')) { - const record = parseForeignJsonLine(line); - if (record) { - collectClaudeTitle(record, titles); - collectClaudeMeta(record, meta); - } - } - } - return { - source: 'claude-code', - id, - title: pickClaudeTitle(titles) || id, - cwd: meta.cwd, - updatedAtMs: meta.timestampMs ?? mtimeMs, - gitBranch: meta.gitBranch, - transcriptPath: path, - }; - } - - /* ------------------------------ Codex ------------------------------- */ - - private async listCodexSessions( - options: ForeignSessionScanOptions, - now: number, - ): Promise { - // Try state DBs newest-generation first. A DB that cannot be opened or - // lacks the threads schema (rows === undefined) is skipped so a freshly - // created generation missing the schema doesn't shadow an older usable - // one. The FIRST usable DB is authoritative — its result is returned even - // when empty. Descending past it on an empty result would resurface stale - // rows from an older generation (e.g. a session archived in the newest DB - // reappearing active in an older one), and would send every no-match-cwd - // listing down the expensive rollout walk. - for (const dbPath of await codexStateDbsNewestFirst(this.codexRoot)) { - const rows = await readCodexThreadRows(dbPath, options.cwd); - if (rows === undefined) continue; - return this.codexRowsToSummaries(rows, options, now); - } - // No usable state DB at all → fall back to the rollout directory walk. - return this.listCodexSessionsFromRollouts(options, now); - } - - private async codexRowsToSummaries( - rows: CodexThreadRow[], - options: ForeignSessionScanOptions, - now: number, - ): Promise { - const results: ForeignSessionSummary[] = []; - for (const row of rows) { - if (results.length >= FOREIGN_SESSION_SCAN_MAX_SESSIONS) break; - const normalized = normalizeCodexThreadRow(row); - if (!normalized) continue; - if (now - normalized.updatedAtMs > FOREIGN_SESSION_SCAN_MAX_AGE_MS) continue; - if (options.cwd !== undefined && normalizePath(normalized.cwd) !== normalizePath(options.cwd)) - continue; - const transcriptPath = await this.resolveCodexRolloutPath( - normalized.rolloutPath, - normalized.id, - ); - if (transcriptPath === undefined) continue; - results.push({ - source: normalized.source, - id: normalized.id, - title: normalized.title, - cwd: normalized.cwd, - updatedAtMs: normalized.updatedAtMs, - gitBranch: normalized.gitBranch, - transcriptPath, - }); - } - return results; - } - - private async listCodexSessionsFromRollouts( - options: ForeignSessionScanOptions, - now: number, - ): Promise { - const sessionsRoot = join(this.codexRoot, 'sessions'); - const files = await walkRolloutFiles(sessionsRoot, now); - const results: ForeignSessionSummary[] = []; - for (const file of files) { - if (results.length >= FOREIGN_SESSION_SCAN_MAX_SESSIONS) break; - const head = await readWindow(file.path, 'head', FOREIGN_SESSION_HEAD_BYTES); - if (head === undefined) continue; - let meta: ReturnType; - let firstUserText: string | undefined; - for (const line of head.split('\n')) { - const record = parseForeignJsonLine(line); - if (!record) continue; - meta ??= codexRolloutSessionMeta(record); - if (firstUserText === undefined) { - const message = codexRolloutMessage(record); - if (message?.role === 'user') firstUserText = message.text; - } - if (meta && firstUserText !== undefined) break; - } - if (!meta?.id || meta.cwd === undefined) continue; - if (!isSafeForeignId(meta.id)) continue; - // The transcript filename must belong to this session (defends against - // renamed / planted rollout files, as in the DB path). - if (!rolloutFilenameMatchesId(basename(file.path), meta.id)) continue; - if (options.cwd !== undefined && normalizePath(meta.cwd) !== normalizePath(options.cwd)) - continue; - results.push({ - source: 'codex', - id: meta.id, - // session_meta has no title; the first user message in the head - // window is the best available label (Grok Build does the same). - title: sanitizeForeignTitle(firstUserText) || meta.id, - cwd: meta.cwd, - updatedAtMs: meta.timestampMs ?? file.mtimeMs, - gitBranch: meta.gitBranch, - transcriptPath: file.path, - }); - } - return results; - } - - /** - * Realpath-confine a rollout path from the (untrusted) DB to ~/.codex, and - * require the transcript filename to belong to this thread — the id (a uuid) - * is the trailing component of `rollout--.jsonl`, so a - * mismatch means the row points at some other session's transcript (orphan - * row or a forged path) and is dropped. The timestamp format varies across - * Codex versions (ISO datetime or epoch), so match by the id suffix rather - * than parsing the timestamp. - */ - private async resolveCodexRolloutPath( - rolloutPath: string, - expectedId: string, - ): Promise { - try { - const real = await realpath(resolve(rolloutPath)); - const root = await realpath(this.codexRoot); - if (real !== root && !real.startsWith(root + sep)) return undefined; - if (!(await stat(real)).isFile()) return undefined; - if (!rolloutFilenameMatchesId(basename(real), expectedId)) return undefined; - return real; - } catch { - return undefined; - } - } - - /* ------------------------------ Digest ------------------------------ */ - async readDigest(summary: ForeignSessionSummary): Promise { - // The transcript path was produced by our own scan, but re-confine it - // anyway: digests can be requested long after the scan, and the file - // may have been swapped for a symlink in between. - const root = summary.source === 'claude-code' ? this.claudeRoot : this.codexRoot; - const real = await realpath(resolve(summary.transcriptPath)); - const realRoot = await realpath(root); - if (real !== realRoot && !real.startsWith(realRoot + sep)) { - throw new Error('Foreign transcript escaped its source root'); - } - - const acc = createDigestAccumulator(); - // Open ONCE and read through the single fd: a stat-then-readFile pair has - // a TOCTOU window (the regular file could be swapped for a FIFO — which - // would block readFile forever — or grown past the cap between the two - // calls). fstat on the held fd, reject anything but a regular file, and - // never read more than the cap regardless of the size we observe. - let handle: Awaited> | undefined; - let text: string; - try { - handle = await open(real, 'r'); - const st = await handle.stat(); - if (!st.isFile()) throw new Error('Foreign transcript is not a regular file'); - if (st.size > FOREIGN_SESSION_DIGEST_MAX_READ_BYTES) { - text = await readHandleTailWindow(handle, st.size, FOREIGN_SESSION_DIGEST_MAX_READ_BYTES); - acc.warnings.push( - `transcript is ${st.size} bytes; only the trailing ${FOREIGN_SESSION_DIGEST_MAX_READ_BYTES} bytes were read`, - ); - } else { - const buffer = Buffer.alloc(st.size); - await handle.read(buffer, 0, st.size, 0); - text = buffer.toString('utf8'); - } - } finally { - await handle?.close(); - } - - let dropped = 0; - for (const line of text.split('\n')) { - if (line.trim().length === 0) continue; - const record = parseForeignJsonLine(line); - if (!record) { - dropped += 1; - continue; - } - if (summary.source === 'claude-code') { - // Sidechain records are a sub-agent's own conversation interleaved - // into the main transcript; they belong to neither role of the main - // session and must not enter its handoff (drop them for BOTH the user - // and assistant branches, not just the user one). - if (record.isSidechain === true) continue; - if (record.type === 'user') { - // claudeUserAuthoredText drops isMeta / isCompactSummary records so - // Claude's own injected context and generated compaction summaries - // never enter the handoff as user-authored text. - const text = claudeUserAuthoredText(record); - if (text !== undefined) pushDigestMessage(acc, 'user', text); - } else if (record.type === 'assistant') { - const text = claudeAssistantText(record); - if (text !== undefined) pushDigestMessage(acc, 'assistant', text); - for (const path of claudeToolFilePaths(record)) pushDigestFile(acc, path); - } - } else { - const message = codexRolloutMessage(record); - if (message) pushDigestMessage(acc, message.role, message.text); - } - } - if (dropped > 0) acc.warnings.push(`${dropped} malformed transcript lines were skipped`); - - return finishDigest(acc, { - source: summary.source, - id: summary.id, - title: summary.title, - cwd: summary.cwd, - gitBranch: summary.gitBranch, - updatedAtMs: summary.updatedAtMs, - }); - } -} - -/* ------------------------------ fs helpers ------------------------------ */ - -async function isDirectory(path: string): Promise { - try { - return (await stat(path)).isDirectory(); - } catch { - return false; - } -} - -async function listSubdirectories(root: string): Promise { - try { - const entries = await readdir(root, { withFileTypes: true }); - return entries.filter((e) => e.isDirectory()).map((e) => join(root, e.name)); - } catch { - return []; - } -} - -async function listFilesWithSuffix( - dir: string, - suffix: string, -): Promise<{ path: string; mtimeMs: number }[]> { - const out: { path: string; mtimeMs: number }[] = []; - try { - const entries = await readdir(dir, { withFileTypes: true }); - for (const entry of entries) { - if (!entry.isFile() || !entry.name.endsWith(suffix)) continue; - const path = join(dir, entry.name); - try { - out.push({ path, mtimeMs: (await stat(path)).mtimeMs }); - } catch { - // Deleted mid-scan; skip. - } - } - } catch { - // Unreadable project dir; skip. - } - return out; -} - -/** Codex sessions/YYYY/MM/DD/rollout-*.jsonl walk, newest days first. */ -async function walkRolloutFiles( - root: string, - now: number, -): Promise<{ path: string; mtimeMs: number }[]> { - const out: { path: string; mtimeMs: number }[] = []; - const years = (await listSubdirectories(root)).sort().reverse(); - for (const year of years) { - const months = (await listSubdirectories(year)).sort().reverse(); - for (const month of months) { - const days = (await listSubdirectories(month)).sort().reverse(); - for (const day of days) { - for (const file of await listFilesWithSuffix(day, '.jsonl')) { - if (!basename(file.path).startsWith('rollout-')) continue; - if (now - file.mtimeMs > FOREIGN_SESSION_SCAN_MAX_AGE_MS) continue; - out.push(file); - } - // Enough candidates for the cap even after per-file drops. - if (out.length >= FOREIGN_SESSION_SCAN_MAX_SESSIONS * 2) { - out.sort((a, b) => b.mtimeMs - a.mtimeMs); - return out; - } - } - } - } - out.sort((a, b) => b.mtimeMs - a.mtimeMs); - return out; -} - -/** All ~/.codex/state_N.sqlite paths, newest generation first. */ -async function codexStateDbsNewestFirst(codexRoot: string): Promise { - try { - const entries = await readdir(codexRoot); - return entries - .filter((name) => /^state_\d+\.sqlite$/.test(name)) - .sort((a, b) => Number(b.match(/\d+/)?.[0] ?? 0) - Number(a.match(/\d+/)?.[0] ?? 0)) - .map((name) => join(codexRoot, name)); - } catch { - return []; - } -} - -/** - * Codex source tokens as stored in the DB — bare for cli/vscode, JSON-wrapped - * for the `custom` variants. Used as bound `source IN (…)` params so archived - * / foreign-source rows are excluded IN SQL (before LIMIT), not after. - */ -const CODEX_SOURCE_SQL_VALUES = ['cli', 'vscode', '{"custom":"atlas"}', '{"custom":"chatgpt"}']; - -/** - * Read candidate thread rows from one state DB, filtered and ordered in SQL. - * undefined = DB unusable (cannot open, or lacks the id/rollout_path columns) - * so the caller descends to an older generation. An empty array is a real - * "this DB has no matching threads". - */ -async function readCodexThreadRows( - dbPath: string, - cwdFilter?: string, -): Promise { - try { - const sqlite = await import('node:sqlite'); - const db = new sqlite.DatabaseSync(dbPath, { readOnly: true }); - try { - const columns = new Set( - (db.prepare('PRAGMA table_info(threads)').all() as { name?: unknown }[]) - .map((c) => (typeof c.name === 'string' ? c.name : '')) - .filter((n) => n.length > 0), - ); - if (!columns.has('id') || !columns.has('rollout_path')) return undefined; - // Every identifier below is drawn from this fixed allowlist, never from - // the DB, so interpolation is injection-safe; values are bound params. - const wanted = [ - 'id', - 'rollout_path', - 'cwd', - 'title', - 'first_user_message', - 'updated_at_ms', - 'updated_at', - 'git_branch', - 'archived', - 'source', - ].filter((c) => columns.has(c)); - const where: string[] = []; - const params: string[] = []; - if (columns.has('archived')) where.push('(archived IS NULL OR archived = 0)'); - if (columns.has('source')) { - where.push(`source IN (${CODEX_SOURCE_SQL_VALUES.map(() => '?').join(', ')})`); - params.push(...CODEX_SOURCE_SQL_VALUES); - } - // Filter cwd IN SQL, before LIMIT: otherwise a multi-project store with - // many newer threads from other directories fills the LIMIT window and - // the target project's older thread never reaches the JS-side filter. - // This is a COARSE pre-filter across source-native and host-normalized - // separator forms. The authoritative two-sided normalizePath() - // comparison still runs in codexRowsToSummaries(). - if (cwdFilter !== undefined && columns.has('cwd')) { - const variants = codexCwdSqlVariants(cwdFilter); - where.push(`cwd IN (${variants.map(() => '?').join(', ')})`); - params.push(...variants); - } - const orderColumn = columns.has('updated_at_ms') - ? 'updated_at_ms' - : columns.has('updated_at') - ? 'updated_at' - : 'id'; - const sql = - `SELECT ${wanted.join(', ')} FROM threads` + - (where.length > 0 ? ` WHERE ${where.join(' AND ')}` : '') + - ` ORDER BY ${orderColumn} DESC LIMIT ${FOREIGN_SESSION_SCAN_MAX_SESSIONS * 2}`; - return db.prepare(sql).all(...params) as CodexThreadRow[]; - } finally { - db.close(); - } - } catch { - return undefined; - } -} - -/** - * Parsed records from the head of a Claude transcript, growing the read - * window (64KB → 4MB) so a session that opens with a run of summary lines or - * a very large first message still yields its cwd record. Stops early once a - * record carrying `cwd` is seen. - */ -async function readClaudeHeadRecords(path: string): Promise[]> { - for (let bytes = 64 * 1024; ; bytes *= 4) { - const capped = Math.min(bytes, CLAUDE_HEAD_MAX_BYTES); - const window = await readWindow(path, 'head', capped); - if (window === undefined) return []; - const records: Record[] = []; - let sawCwd = false; - for (const line of window.split('\n')) { - const record = parseForeignJsonLine(line); - if (!record) continue; - records.push(record); - if (typeof record.cwd === 'string') sawCwd = true; - } - if (sawCwd || capped >= CLAUDE_HEAD_MAX_BYTES || capped >= (await fileSize(path))) - return records; - } -} - -const CLAUDE_HEAD_MAX_BYTES = 4 * 1024 * 1024; - -async function fileSize(path: string): Promise { - try { - return (await stat(path)).size; - } catch { - return 0; - } -} - -/** Read the trailing `bytes` of an open handle, dropping the partial first line. */ -async function readHandleTailWindow( - handle: FileHandle, - size: number, - bytes: number, -): Promise { - const length = Math.min(bytes, size); - const buffer = Buffer.alloc(length); - await handle.read(buffer, 0, length, size - length); - const text = buffer.toString('utf8'); - if (length >= size) return text; // whole file — no partial first line - const nl = text.indexOf('\n'); - return nl === -1 ? '' : text.slice(nl + 1); -} - -/** - * Bounded read of a file's head or tail window; undefined on any error. A - * tail window drops its partial first line so a mid-line cut isn't parsed as - * a malformed record (and isn't reported as one). - */ -async function readWindow( - path: string, - where: 'head' | 'tail', - bytes: number, -): Promise { - let handle: FileHandle | undefined; - try { - handle = await open(path, 'r'); - const size = (await handle.stat()).size; - if (where === 'tail') return await readHandleTailWindow(handle, size, bytes); - const length = Math.min(bytes, size); - const buffer = Buffer.alloc(length); - await handle.read(buffer, 0, length, 0); - return buffer.toString('utf8'); - } catch { - return undefined; - } finally { - await handle?.close(); + const reader = this.readers.find((candidate) => candidate.id === summary.source); + if (!reader) throw new Error(`Unsupported foreign session source: ${summary.source}`); + return reader.readDigest(summary); } } -/** - * A Codex rollout file `rollout--.jsonl` belongs to thread - * `id` when the basename opens with `rollout-` and ends with `-.jsonl`. - * Timestamp-format-agnostic: the id (a uuid) is always the trailing segment. - */ -function rolloutFilenameMatchesId(base: string, id: string): boolean { - return base.startsWith('rollout-') && base.endsWith(`-${id}.jsonl`); +function toForeignSummary(entry: ExternalSourceCatalogEntry): ForeignSessionSummary { + return { + source: entry.source, + id: entry.id, + title: entry.title, + cwd: entry.cwd, + updatedAtMs: entry.updatedAtMs, + ...(entry.gitBranch !== undefined ? { gitBranch: entry.gitBranch } : {}), + transcriptPath: entry.transcriptPath, + }; } -function normalizePath(path: string): string { - const resolved = resolve(path); - return resolved.endsWith(sep) && resolved !== sep ? resolved.slice(0, -1) : resolved; +function sourceEnabled( + source: ForeignSessionSource, + env: Record, +): boolean { + return source === 'claude-code' ? isClaudeCodeImportEnabled(env) : isCodexImportEnabled(env); } -export function codexCwdSqlVariants(path: string): string[] { +/** SQL pre-filter variants for native Windows paths stored by Codex. */ +export function codexCwdSqlVariants(cwd: string): string[] { const variants = new Set(); - for (const candidate of [path, normalizePath(path)]) { + for (const candidate of [cwd, normalizeSourcePath(cwd)]) { for (const separatorForm of [ candidate, candidate.replaceAll('\\', '/'), diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index 02eb0ec807..c3dec5bc91 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -52,6 +52,7 @@ import { isSessionToolProfile, type SessionHeader, type SessionConversationCopy, + type SessionExternalOrigin, type SessionSummary, type StoredMessage, type TurnRecord, @@ -283,6 +284,7 @@ export interface SessionAuthorityStore extends SessionStore { createImportedSession( input: CreateSessionInput, messages: readonly StoredMessage[], + externalOrigin?: SessionExternalOrigin, ): Promise; createSubagent( input: CreateSessionInput, @@ -401,6 +403,7 @@ class SqliteSessionStore implements SessionAuthorityStore { async createImportedSession( input: CreateSessionInput, messages: readonly StoredMessage[], + externalOrigin?: SessionExternalOrigin, ): Promise { await this.ensureReady(); assertNoConversationCopyMetadata(input); @@ -412,6 +415,7 @@ class SqliteSessionStore implements SessionAuthorityStore { ); const header: SessionHeader = { ...buildSessionHeader(this.workspaceRoot, input), + ...(externalOrigin !== undefined ? { externalOrigin } : {}), transcriptLedgerVersion: 0, }; const outcome = await this.metadata.importSession( @@ -1036,6 +1040,7 @@ export function normalizeSessionHeader( isValidConversationCopyLineage(header) && isValidRevisionLineage(header) && isValidSubagentSessionLineage(header) && + isValidSessionExternalOrigin(header.externalOrigin) && (header.lastReadMessageId === undefined || typeof header.lastReadMessageId === 'string') && typeof header.hasUnread === 'boolean' && isBackendKind(header.backend) && @@ -1061,6 +1066,18 @@ export function normalizeSessionHeader( return { ...header, name: normalizedName }; } +function isValidSessionExternalOrigin(origin: SessionHeader['externalOrigin']): boolean { + if (origin === undefined) return true; + return ( + typeof origin === 'object' && + origin !== null && + typeof origin.adapterId === 'string' && + origin.adapterId.length > 0 && + typeof origin.sourceSessionId === 'string' && + origin.sourceSessionId.length > 0 + ); +} + function isValidRevisionLineage(header: SessionHeader): boolean { const values = [ header.revisionRootSessionId, From c801205300c5dd919c4aa4ff644dd888e3236ccd Mon Sep 17 00:00:00 2001 From: Hsin <2129830748@qq.com> Date: Mon, 24 Aug 2026 20:50:31 +0800 Subject: [PATCH 3/4] fix(storage): harden external session lookup Page Codex catalog reads before applying limits and use exact database lookups for import. Reject Claude transcripts without a usable cwd. Generated-by: Codex --- .../claude-code-session-adapter.test.ts | 35 ++++++ .../__tests__/codex-session-adapter.test.ts | 45 ++++++++ .../src/claude-code-session-adapter.ts | 2 +- packages/storage/src/codex-session-adapter.ts | 102 +++++++++++++++--- 4 files changed, 167 insertions(+), 17 deletions(-) diff --git a/packages/storage/src/__tests__/claude-code-session-adapter.test.ts b/packages/storage/src/__tests__/claude-code-session-adapter.test.ts index 717ec80559..8b2d8d7507 100644 --- a/packages/storage/src/__tests__/claude-code-session-adapter.test.ts +++ b/packages/storage/src/__tests__/claude-code-session-adapter.test.ts @@ -275,6 +275,41 @@ describe('ClaudeCodeSessionAdapter', () => { }); }); + test('drops transcripts without a non-empty cwd from catalog and import', async () => { + await withClaudeHome(async (home) => { + const directory = join(home, 'projects', '-workspace-project'); + await mkdir(directory, { recursive: true }); + const missingCwdId = 'aaaaaaaa-0000-4000-8000-000000000040'; + const wrongTypeCwdId = 'aaaaaaaa-0000-4000-8000-000000000041'; + const records = (cwd: unknown) => + [ + JSON.stringify({ + type: 'user', + ...(cwd === undefined ? {} : { cwd }), + message: { role: 'user', content: 'unsafe identity' }, + }), + JSON.stringify({ + type: 'assistant', + ...(cwd === undefined ? {} : { cwd }), + message: { + role: 'assistant', + id: 'msg_missing_cwd', + model: 'claude-opus-5', + content: [{ type: 'text', text: 'reply' }], + stop_reason: 'end_turn', + }, + }), + ].join('\n') + '\n'; + await writeFile(join(directory, `${missingCwdId}.jsonl`), records(undefined)); + await writeFile(join(directory, `${wrongTypeCwdId}.jsonl`), records({ value: 1 })); + + const adapter = new ClaudeCodeSessionAdapter({ claudeHome: home }); + assert.deepEqual(await adapter.listSessions(), []); + await assert.rejects(() => adapter.readSession(missingCwdId), /could not be read/u); + await assert.rejects(() => adapter.readSession(wrongTypeCwdId), /could not be read/u); + }); + }); + test('a text query filters the source, before paging', async () => { // The catalog pages 16 at a time over a source with 1128 sessions here, so // the term has to reach the adapter. A filter applied to an assembled page diff --git a/packages/storage/src/__tests__/codex-session-adapter.test.ts b/packages/storage/src/__tests__/codex-session-adapter.test.ts index 07c6ff1bd3..7bd97d5e41 100644 --- a/packages/storage/src/__tests__/codex-session-adapter.test.ts +++ b/packages/storage/src/__tests__/codex-session-adapter.test.ts @@ -482,6 +482,51 @@ describe('CodexSessionAdapter', () => { }); }); + test('pages past invalid database rows before applying the catalog limit', async () => { + await withCodexHome(async (codexHome) => { + const targetId = 'codex-paged-target'; + const targetPath = await seedMinimalRollout( + codexHome, + targetId, + false, + '/workspace/target', + 'Target task', + ); + const invalidRows = Array.from({ length: 101 }, (_, index) => ({ + id: `codex-invalid-${index}`, + rolloutPath: join(codexHome, 'sessions', `missing-${index}.jsonl`), + cwd: '/workspace/other', + name: `Invalid ${index}`, + createdAtMs: 10_000 + index, + updatedAtMs: 10_000 + index, + archived: false, + source: 'cli', + })); + await seedStateDatabase(codexHome, [ + ...invalidRows, + { + id: targetId, + rolloutPath: targetPath, + cwd: '/workspace/target', + name: 'Target task', + createdAtMs: 1, + updatedAtMs: 1, + archived: false, + source: 'cli', + }, + ]); + + const adapter = new CodexSessionAdapter({ codexHome }); + assert.deepEqual( + (await adapter.listCatalogEntries({ limit: 1 })).map((entry) => entry.id), + [targetId], + ); + const imported = await adapter.readSession(targetId); + assert.equal(imported.sourceSessionId, targetId); + assert.equal(imported.metadata.cwd, '/workspace/target'); + }); + }); + test('uses the same newest rollout when a Codex id has duplicate candidates', async () => { await withCodexHome(async (codexHome) => { const sessionId = 'codex-duplicate-id'; diff --git a/packages/storage/src/claude-code-session-adapter.ts b/packages/storage/src/claude-code-session-adapter.ts index 84cc64675b..7162ee775d 100644 --- a/packages/storage/src/claude-code-session-adapter.ts +++ b/packages/storage/src/claude-code-session-adapter.ts @@ -443,7 +443,7 @@ export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { } } - if (records.length === 0) return undefined; + if (records.length === 0 || cwd.trim().length === 0) return undefined; return { records, cwd, diff --git a/packages/storage/src/codex-session-adapter.ts b/packages/storage/src/codex-session-adapter.ts index a0e31b7daf..215f4cb4d7 100644 --- a/packages/storage/src/codex-session-adapter.ts +++ b/packages/storage/src/codex-session-adapter.ts @@ -24,6 +24,7 @@ import { basename, join, resolve, sep } from 'node:path'; import type { StoredMessage } from '@maka/core/session'; import { FOREIGN_SESSION_DIGEST_MAX_READ_BYTES, + FOREIGN_SESSION_SCAN_MAX_SESSIONS, codexRolloutMessage, createDigestAccumulator, finishDigest, @@ -51,6 +52,7 @@ export const CODEX_SESSION_ADAPTER_ID = 'codex'; export const CODEX_ROLLOUT_MAX_BYTES = 64 * 1024 * 1024; const CODEX_ROLLOUT_HEAD_BYTES = 512 * 1024; +const CODEX_CATALOG_PAGE_SIZE = FOREIGN_SESSION_SCAN_MAX_SESSIONS * 2; const CODEX_SESSION_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/; const CODEX_UNSAFE_PATH_CHARS = /[\u0000-\u001F\u007F\u0080-\u009F\u061C\u200B-\u200F\u202A-\u202E\u2060-\u2064\u2066-\u2069\uFEFF]/; @@ -202,24 +204,70 @@ export class CodexSessionAdapter implements ExternalSessionAdapter { } private async listCatalog(query: ExternalSourceCatalogQuery): Promise { + if (query.limit !== undefined && query.limit <= 0) return []; for (const dbPath of await codexStateDbsNewestFirst(this.codexHome)) { - const rows = await readCodexThreadRows(dbPath, query); - if (rows === undefined) continue; - const entries = await Promise.all(rows.map((row) => this.entryFromRow(row))); - return entries - .filter((entry): entry is CodexCatalogEntry => entry !== undefined) - .filter((entry) => matchesSourceCatalogQuery(entry, query)) - .sort(compareCatalogEntries) - .slice(0, query.limit); + const entries = await this.readCatalogFromDatabase(dbPath, query); + if (entries === undefined) continue; + return entries; } return this.scanRolloutCatalog(query); } private async findCatalogEntry(sessionId: string): Promise { - return (await this.listCatalog({ includeArchived: true })).find( - (entry) => entry.id === sessionId, - ); + for (const dbPath of await codexStateDbsNewestFirst(this.codexHome)) { + const rows = await readCodexThreadRows( + dbPath, + { includeArchived: true }, + { exactId: sessionId }, + ); + if (rows === undefined) continue; + const entry = rows[0] ? await this.entryFromRow(rows[0]) : undefined; + return entry?.id === sessionId ? entry : undefined; + } + + return this.findRolloutEntry(sessionId); + } + + private async readCatalogFromDatabase( + dbPath: string, + query: ExternalSourceCatalogQuery, + ): Promise { + const pageSize = CODEX_CATALOG_PAGE_SIZE; + const entries: CodexCatalogEntry[] = []; + let offset = 0; + for (;;) { + const rows = await readCodexThreadRows(dbPath, query, { limit: pageSize, offset }); + if (rows === undefined) return undefined; + const pageEntries = await Promise.all(rows.map((row) => this.entryFromRow(row))); + for (const entry of pageEntries) { + if (entry && matchesSourceCatalogQuery(entry, query)) entries.push(entry); + } + if (query.limit !== undefined && entries.length >= query.limit) break; + if (rows.length < pageSize) break; + offset += rows.length; + } + entries.sort(compareCatalogEntries); + return query.limit === undefined ? entries : entries.slice(0, query.limit); + } + + private async findRolloutEntry(sessionId: string): Promise { + const candidates = [ + ...(await walkRolloutFiles(join(this.codexHome, 'sessions'), false, sessionId)), + ...(await walkRolloutFiles(join(this.codexHome, 'archived_sessions'), true, sessionId)), + ].sort(compareRolloutCandidates); + for (const candidate of candidates) { + const head = await readUtf8Prefix(candidate.path, CODEX_ROLLOUT_HEAD_BYTES).catch( + () => undefined, + ); + if (head === undefined) continue; + const entry = catalogEntryFromRolloutHead(head, candidate); + if (entry?.id === sessionId) { + const transcriptPath = await this.resolveRolloutPath(entry.transcriptPath, sessionId); + if (transcriptPath) return { ...entry, transcriptPath }; + } + } + return undefined; } private async entryFromRow(row: CodexThreadRow): Promise { @@ -297,6 +345,12 @@ interface RolloutCandidate { archived: boolean; } +interface CodexThreadReadOptions { + exactId?: string; + limit?: number; + offset?: number; +} + function convertCodexRollout( text: string, expectedSessionId: string, @@ -699,6 +753,7 @@ function catalogEntryFromRolloutHead( async function readCodexThreadRows( dbPath: string, query: ExternalSourceCatalogQuery, + options: CodexThreadReadOptions = {}, ): Promise { try { const sqlite = await import('node:sqlite'); @@ -745,17 +800,27 @@ async function readCodexThreadRows( // belongs to. The archived clause stays: that one is an exact boolean // and agrees with the matcher by construction. // - // The statement has no LIMIT, so dropping the clause widens the read - // rather than truncating it. const orderColumn = columns.has('updated_at_ms') ? 'updated_at_ms' : columns.has('updated_at') ? 'updated_at' : 'id'; + if (options?.exactId !== undefined) { + where.push('id = ?'); + params.push(options.exactId); + } const sql = `SELECT ${wanted.join(', ')} FROM threads` + (where.length > 0 ? ` WHERE ${where.join(' AND ')}` : '') + - ` ORDER BY ${orderColumn} DESC`; + ` ORDER BY ${orderColumn} DESC` + + (options?.limit !== undefined + ? ' LIMIT ? OFFSET ?' + : options?.exactId !== undefined + ? ' LIMIT 1' + : ''); + if (options?.limit !== undefined) { + params.push(options.limit, options.offset ?? 0); + } return db.prepare(sql).all(...params) as CodexThreadRow[]; } finally { db.close(); @@ -783,7 +848,11 @@ async function codexStateDbsNewestFirst(codexHome: string): Promise { } } -async function walkRolloutFiles(root: string, archived: boolean): Promise { +async function walkRolloutFiles( + root: string, + archived: boolean, + expectedId?: string, +): Promise { const files: RolloutCandidate[] = []; const visit = async (directory: string): Promise => { let entries: Dirent[]; @@ -799,7 +868,8 @@ async function walkRolloutFiles(root: string, archived: boolean): Promise Date: Tue, 25 Aug 2026 07:46:03 +0800 Subject: [PATCH 4/4] test: update session cleanup mock --- apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts b/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts index 2302bbe8c1..71c4521a21 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts @@ -207,6 +207,7 @@ test('keeps the Desktop candidate usable when an optional MCP tool has an invali onError: (error) => diagnostics.push(error), createSessionCopyCleanup: () => ({ ownCreation: (_creation, operation) => operation(), + rejectCreation: async () => undefined, cleanup: async () => undefined, schedule: async () => undefined, abandonOwner: async () => undefined,