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 941b9d72e4..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 @@ -117,6 +117,121 @@ 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(), + rejectCreation: async () => undefined, + 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 07fb18e7c1..4e8dc02129 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 @@ -21,6 +21,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 { @@ -169,6 +170,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( @@ -549,6 +699,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 9d67cf7ac5..7c0b165bbe 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -728,6 +728,7 @@ runtimeHostManager = await startRuntimeHostDesktopManager( 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 37396022b1..529bfd7308 100644 --- a/apps/desktop/src/main/runtime-host-desktop-candidate.ts +++ b/apps/desktop/src/main/runtime-host-desktop-candidate.ts @@ -599,6 +599,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 e812492c62..ee37ac6524 100644 --- a/apps/desktop/src/main/runtime-host-native-capabilities.ts +++ b/apps/desktop/src/main/runtime-host-native-capabilities.ts @@ -19,12 +19,17 @@ 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, @@ -46,6 +51,7 @@ export interface DesktopCapabilityGroup { readonly label: string; readonly description: string; readonly tools: readonly MakaTool[]; + readonly invalidToolPolicy?: "reject" | "omit"; } interface NativeToolBinding { @@ -95,6 +101,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; readonly targetScope?: DesktopTargetScope; @@ -105,8 +112,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)), ); @@ -328,8 +339,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(); @@ -398,14 +408,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( @@ -415,13 +475,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-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts index 67180218e0..4724a95a71 100644 --- a/packages/runtime-host/src/__tests__/execution-composition.test.ts +++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts @@ -67,6 +67,35 @@ 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, + 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 never discovers PATH Git when no runtime is admitted', async () => { await new Promise((resolve, reject) => { execFile('git', ['--version'], (error) => (error ? reject(error) : resolve())); 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 6a7826628e..f833263654 100644 --- a/packages/runtime-host/src/__tests__/external-session-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/external-session-coordinator.test.ts @@ -23,7 +23,7 @@ import { ExternalSessionAdapterRegistry, type ExternalSessionAdapter, } from '@maka/core/external-session'; -import { type SessionHeader } from '@maka/core/session'; +import type { SessionExternalOrigin, SessionHeader, StoredMessage } from '@maka/core/session'; import { headerToSummary } from '@maka/runtime/session-manager'; import type { SessionCatalogRecord } from '@maka/storage/execution-stores'; import { @@ -305,6 +305,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); diff --git a/packages/runtime/package.json b/packages/runtime/package.json index 693b3d88f6..3a9242d2cc 100644 --- a/packages/runtime/package.json +++ b/packages/runtime/package.json @@ -109,6 +109,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 48bb09af68..88673730ae 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -123,12 +123,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 { AdmissionLimiter } from './admission-limiter.js'; import { type CodeModeExecutionResult, @@ -568,16 +566,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?: ( @@ -609,29 +597,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 { @@ -651,17 +621,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..10c3278619 --- /dev/null +++ b/packages/runtime/src/json-schema-validation.ts @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +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); +} 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 e9d92a7951..7bd97d5e41 100644 --- a/packages/storage/src/__tests__/codex-session-adapter.test.ts +++ b/packages/storage/src/__tests__/codex-session-adapter.test.ts @@ -18,7 +18,7 @@ */ 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'; @@ -317,6 +317,92 @@ describe('CodexSessionAdapter', () => { }); }); + 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( @@ -346,6 +432,166 @@ 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('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'; + 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 f74d0386bb..00f6e84e58 100644 --- a/packages/storage/src/__tests__/external-session-importer.test.ts +++ b/packages/storage/src/__tests__/external-session-importer.test.ts @@ -18,10 +18,11 @@ */ 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 { SessionExternalOrigin, SessionHeader, StoredMessage } from '@maka/core/session'; import { ExternalSessionAdapterRegistry, @@ -31,9 +32,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. Use OpenCV.js.'); + } + assert.equal(importedMessages[2]?.type, 'assistant'); + if (importedMessages[2]?.type === 'assistant') { + assert.equal(importedMessages[2].text, 'Use canvas. Then process the pixels.'); + } + } finally { + await sessions.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + test('forwards the exact external Session origin to imported persistence', async () => { const calls: SessionExternalOrigin[] = []; const adapter = fakeAdapter({ diff --git a/packages/storage/src/__tests__/foreign-session-store.test.ts b/packages/storage/src/__tests__/foreign-session-store.test.ts index 81e268ead4..a316f47d91 100644 --- a/packages/storage/src/__tests__/foreign-session-store.test.ts +++ b/packages/storage/src/__tests__/foreign-session-store.test.ts @@ -570,7 +570,7 @@ describe('foreign session store — digest', () => { assert.ok(!flat.includes('rm -rf'), flat); }); - it('refuses a transcript path replaced by an out-of-root symlink', async () => { + it('refuses a transcript path replaced by an out-of-root symlink', async (t) => { const home = await tempHome(); const path = await seedClaudeSession(home, { id: 'sym', cwd: '/repo' }); const store = createForeignSessionStore({ homeDir: home, env: {} }); @@ -581,7 +581,15 @@ describe('foreign session store — digest', () => { await writeFile(secret, 'not yours', 'utf8'); const { rm } = await import('node:fs/promises'); await rm(path); - await symlink(secret, path); + try { + await symlink(secret, path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'EPERM') { + t.skip('symlink creation is not permitted in this Windows test environment'); + return; + } + throw error; + } await assert.rejects(() => store.readDigest(session as ForeignSessionSummary), /escaped/); }); }); diff --git a/packages/storage/src/claude-code-session-adapter.ts b/packages/storage/src/claude-code-session-adapter.ts index 5a491c35e3..7162ee775d 100644 --- a/packages/storage/src/claude-code-session-adapter.ts +++ b/packages/storage/src/claude-code-session-adapter.ts @@ -30,15 +30,23 @@ // between `/Users/a/b` and `/Users/a-b`. Every record carries its own `cwd`, // and that is what a project-scoped query reads. import { existsSync } from 'node:fs'; -import { readFile, readdir, stat } from 'node:fs/promises'; +import { readFile, readdir, realpath, stat } from 'node:fs/promises'; import { homedir } from 'node:os'; -import { join, resolve } from 'node:path'; +import { basename, join, resolve, sep } from 'node:path'; import { + FOREIGN_SESSION_DIGEST_MAX_READ_BYTES, claudeAssistantText, + claudeToolFilePaths, claudeUserAuthoredText, + createDigestAccumulator, + finishDigest, isSyntheticClaudeUserText, + pushDigestFile, + pushDigestMessage, pickClaudeTitle, sanitizeForeignTitle, + type ForeignSessionDigest, + type ForeignSessionSummary, } from '@maka/core/foreign-session'; import { externalSessionMatchesQuery } from '@maka/core/external-session'; import type { @@ -52,6 +60,12 @@ import { resolveTranscriptLineage, type TranscriptRecord, } from './claude-code-transcript-lineage.js'; +import { + matchesSourceCatalogQuery, + readUtf8Tail, + type ExternalSourceCatalogEntry, + type ExternalSourceCatalogQuery, +} from './external-source-catalog.js'; export const CLAUDE_CODE_SESSION_ADAPTER_ID = 'claude-code'; @@ -60,10 +74,9 @@ export const CLAUDE_CODE_SESSION_ADAPTER_ID = 'claude-code'; * to exhaust the Host's memory during an import the user asked for. */ export const CLAUDE_TRANSCRIPT_MAX_BYTES = 64 * 1024 * 1024; -/** Session ids are the transcript's filename stem, and reach the filesystem. - * A uuid is what Claude Code writes; anything else is refused rather than - * joined onto a path. */ -const SESSION_ID_PATTERN = /^[0-9a-fA-F-]{1,128}$/u; +/** Session ids are transcript filename stems and reach the filesystem. Keep + * the source's opaque id safe without requiring a UUID in test/legacy data. */ +const SESSION_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/u; export interface ClaudeCodeSessionAdapterOptions { /** Overrides `~/.claude`. */ @@ -74,12 +87,22 @@ export interface ClaudeCodeSessionAdapterOptions { interface ParsedTranscript { readonly records: readonly TranscriptRecord[]; readonly cwd: string; + readonly gitBranch?: string; readonly title: string; readonly createdAt?: number; readonly updatedAt?: number; readonly isSidechain: boolean; } +interface ClaudeCodeCatalogEntry extends ExternalSourceCatalogEntry { + readonly source: 'claude-code'; +} + +interface ClaudeCodeSummary extends Omit { + readonly updatedAt: number; + readonly gitBranch?: string; +} + export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { readonly id = CLAUDE_CODE_SESSION_ADAPTER_ID; readonly #home: string; @@ -99,7 +122,7 @@ export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { */ readonly #summaries = new Map< string, - { mtimeMs: number; size: number; summary?: ExternalSessionSummary } + { mtimeMs: number; size: number; summary?: ClaudeCodeSummary } >(); constructor(options: ClaudeCodeSessionAdapterOptions = {}) { @@ -139,7 +162,7 @@ export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { * `undefined` is cached too: a sidechain transcript or an unreadable one is * a stable answer, and re-deriving it every list would defeat the point. */ - async #summaryOf(path: string, sessionId: string): Promise { + async #summaryOf(path: string, sessionId: string): Promise { let mtimeMs: number; let size: number; try { @@ -163,8 +186,13 @@ export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { id: sessionId, name: parsed.title || sessionId, cwd: parsed.cwd, + ...(parsed.gitBranch !== undefined ? { gitBranch: parsed.gitBranch } : {}), ...(parsed.createdAt !== undefined ? { createdAt: parsed.createdAt } : {}), - ...(parsed.updatedAt !== undefined ? { updatedAt: parsed.updatedAt } : {}), + // Some legacy/minimal transcripts carry no timestamp at all. The + // file mtime is still a source-owned freshness signal and keeps + // the handoff staleness policy from treating a readable session as + // the Unix epoch. + updatedAt: parsed.updatedAt ?? mtimeMs, } : undefined; this.#summaries.set(path, { mtimeMs, size, ...(summary ? { summary } : {}) }); @@ -177,7 +205,8 @@ export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { (candidate) => candidate.sessionId === sessionId, ); if (!file) throw new Error(`Claude Code transcript not found: ${sessionId}`); - const parsed = await this.#parse(file.path, sessionId); + const path = await this.#resolveTranscriptPath(file.path, sessionId); + const parsed = await this.#parse(path, sessionId); if (!parsed) throw new Error(`Claude Code transcript could not be read: ${sessionId}`); if (parsed.isSidechain) { throw new Error(`Claude Code transcript is a sub-agent sidechain: ${sessionId}`); @@ -189,10 +218,111 @@ export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { }; } + async listCatalogEntries( + query: ExternalSourceCatalogQuery = {}, + ): Promise { + if (query.limit !== undefined && query.limit <= 0) return []; + const entries: ClaudeCodeCatalogEntry[] = []; + for (const file of await this.#transcriptFiles()) { + const summary = await this.#summaryOf(file.path, file.sessionId); + if (!summary) continue; + const entry: ClaudeCodeCatalogEntry = { + source: 'claude-code', + id: summary.id, + title: summary.name, + cwd: summary.cwd, + ...(summary.gitBranch !== undefined ? { gitBranch: summary.gitBranch } : {}), + ...(summary.createdAt !== undefined ? { createdAtMs: summary.createdAt } : {}), + updatedAtMs: summary.updatedAt, + transcriptPath: file.path, + }; + if (matchesSourceCatalogQuery(entry, query)) entries.push(entry); + } + entries.sort( + (left, right) => + right.updatedAtMs - left.updatedAtMs || + left.transcriptPath.localeCompare(right.transcriptPath), + ); + return query.limit === undefined ? entries : entries.slice(0, query.limit); + } + + async readDigest(summary: ForeignSessionSummary): Promise { + if (summary.source !== 'claude-code') { + throw new Error('Claude Code 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 records: TranscriptRecord[] = []; + let malformedLines = 0; + for (const line of text.split('\n')) { + const trimmed = line.trim(); + if (!trimmed) continue; + try { + const value = JSON.parse(trimmed) as unknown; + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + throw new Error('record is not an object'); + } + records.push(value as TranscriptRecord); + } catch { + malformedLines += 1; + } + } + const lineage = resolveTranscriptLineage(records); + const acc = createDigestAccumulator(); + for (const record of lineage.records) { + if (record.isSidechain === true) continue; + if (record.type === 'user') { + const textValue = claudeUserAuthoredText(record); + if (textValue !== undefined) pushDigestMessage(acc, 'user', textValue); + } else if (record.type === 'assistant') { + const textValue = claudeAssistantText(record); + if (textValue !== undefined) pushDigestMessage(acc, 'assistant', textValue); + for (const path of claudeToolFilePaths(record)) pushDigestFile(acc, path); + } + } + if (truncated) { + acc.warnings.push( + `transcript exceeded ${FOREIGN_SESSION_DIGEST_MAX_READ_BYTES} bytes; only its tail was read`, + ); + acc.warnings.push( + 'Claude transcript lineage may be incomplete because rewind and compaction ancestors are outside the available transcript window', + ); + } + if (lineage.abandoned > 0) { + acc.warnings.push(`${lineage.abandoned} records from withdrawn Claude prompts were skipped`); + } + if (lineage.duplicates > 0) { + acc.warnings.push(`${lineage.duplicates} duplicate Claude transcript records were skipped`); + } + if (malformedLines > 0) { + acc.warnings.push(`${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, + }); + } + #projectsRoot(): string { return join(this.#home, 'projects'); } + async #resolveTranscriptPath(candidatePath: string, expectedId: string): Promise { + const root = await realpath(this.#projectsRoot()); + const candidate = await realpath(resolve(candidatePath)); + if (candidate !== root && !candidate.startsWith(root + sep)) { + throw new Error('Foreign transcript escaped its source root'); + } + if (basename(candidate, '.jsonl') !== expectedId || !(await stat(candidate)).isFile()) { + throw new Error(`Claude Code transcript not found: ${expectedId}`); + } + return candidate; + } + async #transcriptFiles(): Promise> { const root = this.#projectsRoot(); let projects: string[]; @@ -264,7 +394,9 @@ export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { const records: TranscriptRecord[] = []; let cwd = ''; - let isSidechain = false; + let gitBranch: string | undefined; + let hasConversationRecord = false; + let allConversationRecordsAreSidechain = true; let createdAt: number | undefined; let updatedAt: number | undefined; const titles: { @@ -291,8 +423,14 @@ export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { const typed = record as TranscriptRecord; records.push(typed); - if (typed.isSidechain === true) isSidechain = true; + if (typed.type === 'user' || typed.type === 'assistant') { + hasConversationRecord = true; + if (typed.isSidechain !== true) allConversationRecordsAreSidechain = false; + } if (typeof typed.cwd === 'string' && typed.cwd && !cwd) cwd = typed.cwd; + if (typeof typed.gitBranch === 'string' && typed.gitBranch && gitBranch === undefined) { + gitBranch = typed.gitBranch; + } const ts = timestampMs(typed); if (ts !== undefined) { createdAt ??= ts; @@ -305,14 +443,15 @@ export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { } } - if (records.length === 0) return undefined; + if (records.length === 0 || cwd.trim().length === 0) return undefined; return { records, cwd, + ...(gitBranch !== undefined ? { gitBranch } : {}), title: pickClaudeTitle(titles), ...(createdAt !== undefined ? { createdAt } : {}), ...(updatedAt !== undefined ? { updatedAt } : {}), - isSidechain, + isSidechain: hasConversationRecord && allConversationRecordsAreSidechain, }; } } diff --git a/packages/storage/src/codex-session-adapter.ts b/packages/storage/src/codex-session-adapter.ts index be3c48c4bc..215f4cb4d7 100644 --- a/packages/storage/src/codex-session-adapter.ts +++ b/packages/storage/src/codex-session-adapter.ts @@ -18,27 +18,52 @@ */ 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 { externalSessionMatchesQuery } from '@maka/core/external-session'; +import { + FOREIGN_SESSION_DIGEST_MAX_READ_BYTES, + FOREIGN_SESSION_SCAN_MAX_SESSIONS, + 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, + 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; 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]/; 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`. */ @@ -47,8 +72,8 @@ export interface CodexSessionAdapterOptions { maxRolloutBytes?: number; } -interface CodexCatalogEntry extends ExternalSessionSummary { - rolloutPath: string; +interface CodexCatalogEntry extends ExternalSourceCatalogEntry { + source: 'codex'; } interface CodexThreadRow { @@ -105,7 +130,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 { @@ -113,10 +151,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, @@ -125,15 +163,52 @@ 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 { + 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) => matchesQuery(entry, query)) - .sort(compareCatalogEntries); + const entries = await this.readCatalogFromDatabase(dbPath, query); + if (entries === undefined) continue; + return entries; } return this.scanRolloutCatalog(query); @@ -141,18 +216,60 @@ export class CodexSessionAdapter implements ExternalSessionAdapter { private async findCatalogEntry(sessionId: string): Promise { for (const dbPath of await codexStateDbsNewestFirst(this.codexHome)) { - const rows = await readCodexThreadRows(dbPath, { includeArchived: true }, sessionId); + const rows = await readCodexThreadRows( + dbPath, + { includeArchived: true }, + { exactId: sessionId }, + ); if (rows === undefined) continue; - for (const row of rows) { - const entry = await this.entryFromRow(row); - if (entry?.id === sessionId) return entry; - } - break; + 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 { if (!isSafeCodexSessionId(row.id)) return undefined; if (typeof row.rollout_path !== 'string' || row.rollout_path.length === 0) return undefined; @@ -166,55 +283,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 }); - } - 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 }; - } + 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 undefined; + entries.sort(compareCatalogEntries); + return query.limit === undefined ? entries : entries.slice(0, query.limit); } private async resolveRolloutPath( @@ -240,6 +345,12 @@ interface RolloutCandidate { archived: boolean; } +interface CodexThreadReadOptions { + exactId?: string; + limit?: number; + offset?: number; +} + function convertCodexRollout( text: string, expectedSessionId: string, @@ -247,6 +358,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'); @@ -254,7 +366,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; @@ -285,130 +396,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 === 'item_completed') { - const item = asRecord(payload.item); - const itemType = stringField(item, 'type')?.toLowerCase(); - const eventTurnId = stringField(payload, 'turn_id'); - if (eventTurnId) { - activeTurnId = eventTurnId; - activeTurnIsExplicit = true; - } - - if (itemType === 'usermessage') { - if (!activeTurnIsExplicit) { - activeTurnId = generatedCodexId(expectedSessionId, 'turn', record.line); - } - const text = codexCompletedItemText(item) || codexCompletedItemMediaText(item); - if (text.length === 0) continue; - firstUserText ??= text; - messages.push({ - type: 'user', - id: - stringField(item, 'client_id') ?? - stringField(item, 'id') ?? - generatedCodexId(expectedSessionId, 'user', record.line), - turnId: ensureTurnId(record.line), - ts: timestampFor(record), - text, - }); - continue; - } - - if (itemType === 'agentmessage') { - const text = codexCompletedItemText(item); - if (text.length === 0) continue; - messages.push({ - type: 'assistant', - id: - stringField(item, 'id') ?? - generatedCodexId(expectedSessionId, 'assistant', record.line), - turnId: ensureTurnId(record.line), - ts: timestampFor(record), - text, - modelId: activeModel, - contentOrder: ['text'], - }); - continue; - } - - if (itemType === 'reasoning') { - const reasoning = codexCompletedReasoningText(item); - if (reasoning.length === 0) continue; - messages.push({ - type: 'assistant', - id: - stringField(item, 'id') ?? - generatedCodexId(expectedSessionId, 'reasoning', record.line), - turnId: ensureTurnId(record.line), - ts: timestampFor(record), - text: '', - thinking: { text: reasoning }, - contentOrder: ['thinking'], - modelId: activeModel, - }); - continue; - } - } - - 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; } @@ -523,7 +563,7 @@ function convertCodexRollout( sanitizeForeignTitle(fallbackName) || sanitizeForeignTitle(firstUserText) || expectedSessionId; return { sourceSessionId: expectedSessionId, - metadata: { name, cwd: metaCwd || fallbackCwd }, + metadata: { name, cwd: fallbackCwd }, messages, }; } @@ -533,6 +573,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'); @@ -557,12 +696,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 { @@ -580,34 +720,40 @@ function catalogEntryFromRolloutHead( cwd = safeCodexCwd(payload.cwd) || cwd; createdAt = normalizeEpochMs(record.timestamp) ?? normalizeEpochMs(payload.timestamp) ?? createdAt; - } else if (record.type === 'event_msg' && firstUserText === undefined) { - if (payload.type === 'user_message') { + } else if (firstUserText === undefined) { + if (record.type === 'event_msg' && payload.type === 'user_message') { firstUserText = stringField(payload, 'message'); - } else if (payload.type === 'item_completed') { + } 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, + options: CodexThreadReadOptions = {}, ): Promise { try { const sqlite = await import('node:sqlite'); @@ -636,13 +782,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); + } // No cwd clause. `cwd IN (...)` enumerated spelling variants of the // query, but SQLite compares them exactly: a row stored `C:\\Repo\\App` // was discarded before `matchesQuery` could see that `c:/repo/app` names @@ -652,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(); @@ -690,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[]; @@ -706,7 +868,8 @@ 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; } @@ -828,15 +950,16 @@ function normalizeEpochMs(value: unknown): number | undefined { return undefined; } -function matchesQuery(entry: ExternalSessionSummary, query: ExternalSessionQuery): boolean { - // The single authority on whether a row answers a query, shared with every - // other adapter. The local path helpers this file used to keep were only - // reachable from the SQL prefilter that has been removed. - return externalSessionMatchesQuery(entry, query); +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 { diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index ea8e49bcda..46d4f3b837 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -25,7 +25,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, diff --git a/packages/storage/src/external-source-catalog.ts b/packages/storage/src/external-source-catalog.ts new file mode 100644 index 0000000000..56f39d7f1d --- /dev/null +++ b/packages/storage/src/external-source-catalog.ts @@ -0,0 +1,147 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { open } from 'node:fs/promises'; +import { + externalSessionMatchesQuery, + 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 ( + !externalSessionMatchesQuery( + { + id: entry.id, + name: entry.title, + cwd: entry.cwd, + ...(entry.createdAtMs !== undefined ? { createdAt: entry.createdAtMs } : {}), + updatedAt: entry.updatedAtMs, + ...(entry.archived !== undefined ? { archived: entry.archived } : {}), + }, + query, + ) + ) + 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 d130110634..111cae144d 100644 --- a/packages/storage/src/foreign-session-store.ts +++ b/packages/storage/src/foreign-session-store.ts @@ -18,68 +18,34 @@ */ /** - * 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 { ClaudeCodeSessionAdapter } from './claude-code-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; } @@ -116,596 +82,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 ClaudeCodeSessionAdapter({ claudeHome: join(homeDir, '.claude') }), + 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('\\', '/'),