diff --git a/packages/core/src/external-session.ts b/packages/core/src/external-session.ts index 528647efbbb..6556b1b4484 100644 --- a/packages/core/src/external-session.ts +++ b/packages/core/src/external-session.ts @@ -231,6 +231,8 @@ export interface ExternalMakaSession { cwd: string; }; messages: readonly StoredMessage[]; + /** Explicit source image parts, consumed host-side before publishing an import. */ + sourceImages?: readonly { readonly messageId: string; readonly path: string }[]; } export const EXTERNAL_SESSION_LIMIT_KINDS = [ 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 5ec8d5d3733..8f97302b939 100644 --- a/packages/runtime-host/src/__tests__/external-session-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/external-session-coordinator.test.ts @@ -1209,6 +1209,32 @@ test('retries a retained staged Session during a later recovery', async (t) => { assert.equal(fixture.readHeader('imported-1')?.transcriptLedgerVersion, 1); }); +test('prepares explicit imported images before publishing the destination Session', async () => { + const fixture = coordinatorFixture([ + adapterFixture({ + readSession: async (sourceSessionId) => ({ + sourceSessionId, + metadata: { name: 'Image-only import', cwd: '/external' }, + messages: [ + { type: 'user', id: 'image-user', turnId: 'source-turn', ts: 1, text: '[Image]' }, + ], + sourceImages: [{ messageId: 'image-user', path: '/missing-original.png' }], + }), + }), + ]); + const outcome = await fixture.coordinator.handlers['external-session.import']( + { adapterId: 'codex', sourceSessionId: 'source-0' }, + context, + ); + assert.equal(outcome.ok, true); + const user = fixture.creates[0]?.messages[0]; + assert.equal(user?.type, 'user'); + if (user?.type !== 'user') throw new Error('Missing imported user message'); + assert.match(user.text, /^\[Image\]/); + assert.match(user.text, /1 imported image attachment\(s\) unavailable/); + assert.equal(user.attachments, undefined); +}); + function coordinatorFixture( adapters: readonly ExternalSessionAdapter[], storeOverrides: Partial< @@ -1246,13 +1272,16 @@ function coordinatorFixture( externalOrigin, options, ) => { - options?.onCommitStarted?.(); sequence += 1; const header = { ...sessionHeader(`imported-${sequence}`, input.cwd, input.name ?? 'Imported'), transcriptLedgerVersion: 0 as const, }; - creates.push({ input, messages, externalOrigin }); + const preparedMessages = options?.prepareMessages + ? await options.prepareMessages(header.id, messages) + : messages; + options?.onCommitStarted?.(); + creates.push({ input, messages: preparedMessages, externalOrigin }); records.set(header.id, { header, revision: 1, diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 7f21ba8c096..f5252943382 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -3018,6 +3018,7 @@ export async function createExecutionRuntimeHostComposition( }); const externalSessions = new HostExternalSessionCoordinator({ adapters: createExternalSessionAdapterRegistry(), + artifacts: openedArtifactStore, admission: sessionAdmission, sessions: stores.sessionStore, workspaceResolver, @@ -3028,6 +3029,7 @@ export async function createExecutionRuntimeHostComposition( const outcomes = await Promise.allSettled([ stores.purgeConversationOperationalState(sessionId), sessionTodoStore.purgeSessionState(sessionId), + openedArtifactStore.purgeSessionArtifacts(sessionId), stores.sessionStore.remove(sessionId), ]); for (const outcome of outcomes) { diff --git a/packages/runtime-host/src/server/external-session-coordinator.ts b/packages/runtime-host/src/server/external-session-coordinator.ts index a97a78631dc..da0f0449f06 100644 --- a/packages/runtime-host/src/server/external-session-coordinator.ts +++ b/packages/runtime-host/src/server/external-session-coordinator.ts @@ -64,7 +64,14 @@ type ExternalSessionStore = { input: CreateSessionInput, messages: readonly StoredMessage[], externalOrigin: SessionExternalOrigin, - options?: { readonly onCommitStarted?: () => void }, + options?: { + readonly onCommitStarted?: () => void; + readonly prepareMessages?: ( + sessionId: string, + messages: readonly StoredMessage[], + ) => Promise; + readonly discardPreparedMessages?: (sessionId: string) => Promise; + }, ): Promise; lookupExternalSessionImports( adapterId: string, @@ -77,6 +84,7 @@ type ExternalSessionStore = { export interface HostExternalSessionCoordinatorOptions { readonly adapters: ExternalSessionAdapterRegistry; + readonly artifacts?: ConstructorParameters[2]; readonly admission: SessionAdmissionGate; readonly sessions: ExternalSessionStore; readonly workspaceResolver: Pick; @@ -95,6 +103,7 @@ export class HostExternalSessionCoordinator { }; readonly #adapters: ExternalSessionAdapterRegistry; + readonly #artifacts: HostExternalSessionCoordinatorOptions['artifacts']; readonly #admission: SessionAdmissionGate; readonly #sessions: ExternalSessionStore; readonly #workspaceResolver: Pick; @@ -132,6 +141,7 @@ export class HostExternalSessionCoordinator { constructor(options: HostExternalSessionCoordinatorOptions) { this.#adapters = options.adapters; + this.#artifacts = options.artifacts; this.#admission = options.admission; this.#sessions = options.sessions; this.#workspaceResolver = options.workspaceResolver; @@ -321,15 +331,20 @@ export class HostExternalSessionCoordinator { } let commitAttempted = false; - const importer = new ExternalSessionImporter(this.#adapters, { - createImportedSession: async (sessionInput, messages, externalOrigin) => { - return this.#sessions.createImportedSession(sessionInput, messages, externalOrigin, { - onCommitStarted: () => { - commitAttempted = true; - }, - }); + const importer = new ExternalSessionImporter( + this.#adapters, + { + createImportedSession: async (sessionInput, messages, externalOrigin, options) => { + return this.#sessions.createImportedSession(sessionInput, messages, externalOrigin, { + ...options, + onCommitStarted: () => { + commitAttempted = true; + }, + }); + }, }, - }); + this.#artifacts, + ); let header: SessionHeader; try { header = await importer.import({ diff --git a/packages/storage/src/__tests__/codex-session-adapter.test.ts b/packages/storage/src/__tests__/codex-session-adapter.test.ts index eab049a28ef..3001b942b90 100644 --- a/packages/storage/src/__tests__/codex-session-adapter.test.ts +++ b/packages/storage/src/__tests__/codex-session-adapter.test.ts @@ -103,6 +103,53 @@ async function listSessions( } describe('CodexSessionAdapter', () => { + test('keeps only explicit local image parts associated with their user message', async () => { + await withCodexHome(async (codexHome) => { + const id = 'codex-image-parts'; + const content = [ + JSON.stringify({ type: 'session_meta', payload: { id, cwd: '/workspace', source: 'cli' } }), + JSON.stringify({ + type: 'event_msg', + payload: { + type: 'item_completed', + item: { + type: 'UserMessage', + id: 'image-user', + content: [ + { type: 'text', text: 'Compare /ordinary/path.png' }, + { type: 'local_image', path: '/explicit/one.png' }, + { type: 'local_image', path: '/explicit/two.png' }, + ], + }, + }, + }), + JSON.stringify({ + type: 'event_msg', + payload: { + type: 'item_completed', + item: { + type: 'UserMessage', + id: 'image-only', + content: [{ type: 'local_image', path: '/explicit/three.png' }], + }, + }, + }), + '', + ].join('\n'); + await seedRawRollout(codexHome, id, content); + const session = await new CodexSessionAdapter({ codexHome }).readSession(id); + assert.deepEqual(session.sourceImages, [ + { messageId: 'image-user', path: '/explicit/one.png' }, + { messageId: 'image-user', path: '/explicit/two.png' }, + { messageId: 'image-only', path: '/explicit/three.png' }, + ]); + assert.deepEqual( + session.messages.filter((m) => m.type === 'user').map((m) => m.text), + ['Compare /ordinary/path.png', '[Image]'], + ); + }); + }); + test('lists active and archived root Sessions from the newest Codex state database', async () => { await withCodexHome(async (codexHome) => { const activePath = await seedFixtureRollout(codexHome, 'codex-session-1', false); diff --git a/packages/storage/src/__tests__/external-session-importer.test.ts b/packages/storage/src/__tests__/external-session-importer.test.ts index c8cffdc0dec..b91291e67de 100644 --- a/packages/storage/src/__tests__/external-session-importer.test.ts +++ b/packages/storage/src/__tests__/external-session-importer.test.ts @@ -18,7 +18,7 @@ */ import assert from 'node:assert/strict'; -import { mkdir, mkdtemp, realpath, rm, stat } from 'node:fs/promises'; +import { mkdir, mkdtemp, realpath, rm, stat, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { describe, test } from 'node:test'; @@ -32,8 +32,132 @@ import { type ExternalSessionImportTarget, } from '../external-session-importer.js'; import { createSessionStore, type SessionAuthorityStore } from '../session-store.js'; +import { createSqliteArtifactStoreWriteAuthority } from '../artifact-store.js'; +import { createAttachmentByteReader } from '../artifact-attachments.js'; +import { MAX_ATTACHMENT_BYTES } from '@maka/core/attachments'; describe('ExternalSessionImporter', () => { + test('persists explicit images in the destination Session and survives source removal', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-import-images-')); + const source = join(root, 'original.png'); + const png = Buffer.from( + 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg==', + 'base64', + ); + await writeFile(source, png); + const sessions = createSessionStore(root); + const authority = createSqliteArtifactStoreWriteAuthority(root); + try { + const adapter = fakeAdapter({ + metadata: { name: 'Images', cwd: root }, + messages: [ + { + type: 'user', + id: 'user-1', + turnId: 'turn-1', + ts: 1, + text: `Compare this image; ordinary path: ${join(root, 'not-an-attachment.png')}`, + }, + ], + sourceImages: [{ messageId: 'user-1', path: source }], + }); + const importer = new ExternalSessionImporter( + new ExternalSessionAdapterRegistry([adapter]), + sessions, + authority.store, + ); + const header = await importer.import({ + adapterId: 'fake', + sourceSessionId: 'source-1', + target: target(), + }); + await rm(source); + const messages = await sessions.readMessages(header.id); + const user = messages[0]; + assert.equal(user?.type, 'user'); + if (user?.type !== 'user') throw new Error('Missing user'); + assert.equal( + user.text, + `Compare this image; ordinary path: ${join(root, 'not-an-attachment.png')}`, + ); + assert.equal(user.attachments?.length, 1); + const attachment = user.attachments![0]!; + assert.equal(attachment.ref.kind, 'session_file'); + assert.equal(attachment.bytes, png.byteLength); + assert.deepEqual( + await createAttachmentByteReader({ artifactStore: authority.store, sessionId: header.id })( + attachment.ref, + ), + { ok: true, bytes: png }, + ); + assert.deepEqual( + await createAttachmentByteReader({ + artifactStore: authority.store, + sessionId: 'different-session', + })(attachment.ref), + { ok: false, reason: 'session_mismatch' }, + ); + assert.equal(JSON.stringify(messages).includes(source), false); + } finally { + await sessions.close?.(); + authority.close(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('bounds attempts even for missing images and reports unavailable originals', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-import-missing-images-')); + const invalid = join(root, 'invalid.png'); + const large = join(root, 'large.png'); + await writeFile(invalid, 'not an image'); + const beyondAttemptLimit = join(root, 'ninth.png'); + await writeFile( + beyondAttemptLimit, + Buffer.from( + 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg==', + 'base64', + ), + ); + const { open } = await import('node:fs/promises'); + const handle = await open(large, 'w'); + await handle.truncate(MAX_ATTACHMENT_BYTES + 1); + await handle.close(); + const sessions = createSessionStore(root); + const authority = createSqliteArtifactStoreWriteAuthority(root); + try { + const adapter = fakeAdapter({ + metadata: { name: 'Missing images', cwd: root }, + messages: [message()], + sourceImages: [ + join(root, 'missing.png'), + invalid, + large, + ...Array.from({ length: 5 }, (_, index) => join(root, `missing-${index}.png`)), + beyondAttemptLimit, + ].map((path) => ({ + messageId: 'user-1', + path, + })), + }); + const header = await new ExternalSessionImporter( + new ExternalSessionAdapterRegistry([adapter]), + sessions, + authority.store, + ).import({ adapterId: 'fake', sourceSessionId: 'source-1', target: target() }); + const user = (await sessions.readMessages(header.id))[0]; + assert.equal(user?.type, 'user'); + if (user?.type !== 'user') throw new Error('Missing user'); + assert.match(user.text, /^hello/); + assert.match(user.text, /9 imported image attachment\(s\) unavailable/); + assert.equal(user.attachments, undefined); + assert.equal((await authority.store.listPage(header.id, { offset: 0, limit: 10 })).total, 0); + } finally { + await sessions.close?.(); + authority.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({ @@ -353,7 +477,7 @@ function message(): StoredMessage { function fakeAdapter( session: Pick< Awaited>, - 'metadata' | 'messages' + 'metadata' | 'messages' | 'sourceImages' >, ): ExternalSessionAdapter { return { diff --git a/packages/storage/src/__tests__/session-store.test.ts b/packages/storage/src/__tests__/session-store.test.ts index c85f7554dd5..2f79faba798 100644 --- a/packages/storage/src/__tests__/session-store.test.ts +++ b/packages/storage/src/__tests__/session-store.test.ts @@ -599,17 +599,70 @@ describe('SQLite SessionStore', () => { const root = await mkdtemp(join(tmpdir(), 'maka-session-import-commit-boundary-')); const store = createSessionStore(root); let commitStarted = false; + let preparedId: string | undefined; + let discardedId: string | undefined; try { await assert.rejects( store.createImportedSession( makeInput(), [{ type: 'user' } as unknown as StoredMessage], { adapterId: 'fake', sourceSessionId: 'source-1' }, - { onCommitStarted: () => (commitStarted = true) }, + { + onCommitStarted: () => (commitStarted = true), + prepareMessages: async (sessionId, messages) => { + preparedId = sessionId; + return messages; + }, + discardPreparedMessages: async (sessionId) => { + discardedId = sessionId; + }, + }, ), /Invalid stored message schema/, ); assert.equal(commitStarted, false); + assert.ok(preparedId); + assert.equal(discardedId, preparedId); + assert.deepEqual(await store.listHeaders(), []); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('preserves the original import failure when prepared message cleanup fails', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-session-import-cleanup-failure-')); + const store = createSessionStore(root); + const preparationError = new Error('message preparation failed'); + const commitError = new Error('import commit callback failed'); + let cleanupAttempts = 0; + try { + for (const stage of ['prepare', 'decode', 'commit'] as const) { + await assert.rejects( + store.createImportedSession( + makeInput(), + stage === 'decode' ? [{ type: 'user' } as unknown as StoredMessage] : [], + { adapterId: 'fake', sourceSessionId: `source-${stage}` }, + { + prepareMessages: async (_sessionId, messages) => { + if (stage === 'prepare') throw preparationError; + return messages; + }, + onCommitStarted: () => { + if (stage === 'commit') throw commitError; + }, + discardPreparedMessages: async () => { + cleanupAttempts++; + throw new Error('artifact cleanup failed'); + }, + }, + ), + stage === 'decode' + ? /Invalid stored message schema/ + : (error) => error === (stage === 'prepare' ? preparationError : commitError), + ); + } + assert.equal(cleanupAttempts, 3); assert.deepEqual(await store.listHeaders(), []); } finally { await store.close?.(); diff --git a/packages/storage/src/codex-session-adapter.ts b/packages/storage/src/codex-session-adapter.ts index 47eb1c1f9f1..995ca5954e0 100644 --- a/packages/storage/src/codex-session-adapter.ts +++ b/packages/storage/src/codex-session-adapter.ts @@ -484,6 +484,7 @@ async function convertCodexRollout( class CodexRolloutConverter { private readonly messages: StoredMessage[] = []; + private readonly sourceImages: { messageId: string; path: string }[] = []; private readonly failedTurnIds = new Set(); private activeTurnId: string | undefined; private activeTurnIsExplicit = false; @@ -574,12 +575,30 @@ class CodexRolloutConverter { const text = codexCompletedItemText(item) || codexCompletedItemMediaText(item); if (text.length === 0) return; this.firstUserText ??= text; + const id = + stringField(item, 'client_id') ?? + stringField(item, 'id') ?? + generatedCodexId(this.expectedSessionId, 'user', record.line); + if (Array.isArray(item?.content)) { + for (const part of item.content) { + const image = asRecord(part); + if (stringField(image, 'type')?.toLowerCase() !== 'local_image') continue; + const path = stringField(image, 'path') ?? ''; + // Source descriptors share the converted-output budget with messages. + this.convertedBytes += Buffer.byteLength(JSON.stringify({ messageId: id, path })); + if (this.convertedBytes > this.limits.maxConvertedBytes) { + throw new ExternalSessionLimitError( + 'converted_bytes', + this.limits.maxConvertedBytes, + 'Codex image references exceed the converted Session limit', + ); + } + this.sourceImages.push({ messageId: id, path }); + } + } this.append({ type: 'user', - id: - stringField(item, 'client_id') ?? - stringField(item, 'id') ?? - generatedCodexId(this.expectedSessionId, 'user', record.line), + id, turnId: this.ensureTurnId(record.line), ts: this.timestampFor(record), ...codexUserContent(text), @@ -795,6 +814,7 @@ class CodexRolloutConverter { sourceSessionId: this.expectedSessionId, metadata: { name, cwd: this.metaCwd || this.fallbackCwd }, messages: this.messages, + ...(this.sourceImages.length > 0 ? { sourceImages: this.sourceImages } : {}), }; } diff --git a/packages/storage/src/external-session-images.ts b/packages/storage/src/external-session-images.ts new file mode 100644 index 00000000000..bdbe9c7e1bd --- /dev/null +++ b/packages/storage/src/external-session-images.ts @@ -0,0 +1,155 @@ +/* + * 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 { randomUUID } from 'node:crypto'; +import { constants } from 'node:fs'; +import { open } from 'node:fs/promises'; +import { basename, isAbsolute } from 'node:path'; +import { + MAX_ATTACHMENT_BYTES, + MAX_ATTACHMENT_COUNT, + sniffAttachmentMimeType, +} from '@maka/core/attachments'; +import type { AttachmentRef } from '@maka/core/events'; +import type { StoredMessage } from '@maka/core/session'; +import type { ExternalMakaSession } from '@maka/core/external-session'; +import type { ArtifactAuthorityStore } from './artifact-store.js'; + +// Bound disk growth and I/O independently of the transcript's text budget. +const MAX_IMPORT_IMAGE_BYTES = 256 * 1024 * 1024; + +type ImportArtifactStore = Pick; + +/** Source paths never become published references: only verified, durable bytes do. */ +export async function prepareExternalSessionImages( + sessionId: string, + messages: readonly StoredMessage[], + sources: NonNullable, + artifacts: ImportArtifactStore | undefined, +): Promise { + const byMessage = new Map(); + for (const source of sources) { + const paths = byMessage.get(source.messageId) ?? []; + paths.push(source.path); + byMessage.set(source.messageId, paths); + } + const prepared: StoredMessage[] = []; + let totalBytes = 0; + try { + for (const message of messages) { + const paths = message.type === 'user' ? byMessage.get(message.id) : undefined; + if (message.type !== 'user' || !paths?.length) { + prepared.push(message); + continue; + } + const attachments: AttachmentRef[] = [...(message.attachments ?? [])]; + const unavailable: number[] = []; + for (const [index, path] of paths.entries()) { + if ( + !artifacts || + index >= MAX_ATTACHMENT_COUNT || + attachments.length >= MAX_ATTACHMENT_COUNT + ) { + unavailable.push(index + 1); + continue; + } + const image = await readSourceImage(path, MAX_IMPORT_IMAGE_BYTES - totalBytes); + if (!image) { + unavailable.push(index + 1); + continue; + } + const id = `import_image_${randomUUID()}`; + const record = await artifacts.create({ + id, + sessionId, + // Match native uploads: the artifact owns an upload key, not a source Turn id. + turnId: id, + name: basename(path), + kind: 'image', + content: image.bytes, + mimeType: image.mimeType, + source: 'user_upload', + }); + totalBytes += image.bytes.byteLength; + attachments.push({ + kind: 'image', + name: record.name, + mimeType: image.mimeType, + bytes: record.sizeBytes, + ref: { kind: 'session_file', sessionId, relativePath: record.id }, + }); + } + const fallback = + unavailable.length > 0 + ? `\n\n[${unavailable.length} imported image attachment(s) unavailable (source positions: ${unavailable.join(', ')}): the original file could not be recovered within attachment limits.]` + : ''; + prepared.push({ + ...message, + // Keep all source text, including useful metadata for unavailable originals. + text: message.text + fallback, + ...(message.displayText !== undefined + ? { displayText: message.displayText + fallback } + : {}), + ...(attachments.length ? { attachments } : {}), + }); + } + return prepared; + } catch (error) { + // No Session metadata has been committed while this preparation callback runs. + await artifacts?.purgeSessionArtifacts(sessionId); + throw error; + } +} + +async function readSourceImage( + path: string, + remaining: number, +): Promise< + | { + bytes: Uint8Array; + mimeType: string; + } + | undefined +> { + if (!isAbsolute(path) || remaining <= 0) return undefined; + try { + const handle = await open(path, constants.O_RDONLY | constants.O_NONBLOCK); + try { + const metadata = await handle.stat(); + const limit = Math.min(MAX_ATTACHMENT_BYTES, remaining); + if (!metadata.isFile() || metadata.size <= 0 || metadata.size > limit) return undefined; + // One extra byte detects growth without ever buffering an unbounded source. + const bytes = Buffer.alloc(metadata.size + 1); + let total = 0; + while (total < bytes.length) { + const read = await handle.read(bytes, total, bytes.length - total, total); + if (read.bytesRead === 0) break; + total += read.bytesRead; + } + if (total !== metadata.size) return undefined; + const payload = bytes.subarray(0, total); + const mimeType = sniffAttachmentMimeType(payload); + return mimeType?.startsWith('image/') ? { bytes: payload, mimeType } : undefined; + } finally { + await handle.close(); + } + } catch { + return undefined; + } +} diff --git a/packages/storage/src/external-session-importer.ts b/packages/storage/src/external-session-importer.ts index c875c28f7b4..c5a1bcf5f04 100644 --- a/packages/storage/src/external-session-importer.ts +++ b/packages/storage/src/external-session-importer.ts @@ -27,6 +27,8 @@ import { type ExternalSessionAdapterRegistry, } from '@maka/core/external-session'; import type { SessionAuthorityStore } from './session-store.js'; +import type { ArtifactAuthorityStore } from './artifact-store.js'; +import { prepareExternalSessionImages } from './external-session-images.js'; export type ExternalSessionImportTarget = Omit & { /** Host-resolved destination; preserve its exact filesystem identity. */ @@ -48,6 +50,7 @@ export class ExternalSessionImporter { constructor( private readonly adapters: ExternalSessionAdapterRegistry, private readonly sessions: Pick, + private readonly artifacts?: Pick, ) {} async import(request: ExternalSessionImportRequest): Promise { @@ -88,6 +91,20 @@ export class ExternalSessionImporter { adapterId: request.adapterId, sourceSessionId: request.sourceSessionId, }, + external.sourceImages?.length + ? { + discardPreparedMessages: async (sessionId) => { + await this.artifacts?.purgeSessionArtifacts(sessionId); + }, + prepareMessages: (sessionId, sourceMessages) => + prepareExternalSessionImages( + sessionId, + sourceMessages, + external.sourceImages!, + this.artifacts, + ), + } + : undefined, ); } } diff --git a/packages/storage/src/session-store-contract.ts b/packages/storage/src/session-store-contract.ts index bd77786136d..82e95bcc862 100644 --- a/packages/storage/src/session-store-contract.ts +++ b/packages/storage/src/session-store-contract.ts @@ -463,7 +463,16 @@ export interface SessionAuthorityStore extends SessionStore, MessageAdmissionSto input: CreateSessionInput, messages: readonly StoredMessage[], externalOrigin: SessionExternalOrigin, - options?: { readonly onCommitStarted?: () => void }, + options?: { + readonly onCommitStarted?: () => void; + /** Discard prepared bytes only when no Session was committed. */ + readonly discardPreparedMessages?: (sessionId: string) => Promise; + /** Prepare Session-owned attachment references before any transcript publication. */ + readonly prepareMessages?: ( + sessionId: string, + messages: readonly StoredMessage[], + ) => Promise; + }, ): Promise; /** Look up live published imports for a bounded page of source Sessions. */ lookupExternalSessionImports( diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index e42014643fb..8bad42aab2f 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -54,6 +54,7 @@ import { type ArchiveRetentionCandidateCount, type ArchiveRetentionCandidateQuery, type ArchiveRetentionCandidateRow, + isSessionNotFoundError, } from './session-store-contract.js'; export { isSafeSessionId, @@ -200,28 +201,65 @@ class SqliteSessionStore implements SessionAuthorityStore { input: CreateSessionInput, messages: readonly StoredMessage[], externalOrigin: SessionExternalOrigin, - options: { readonly onCommitStarted?: () => void } = {}, + options: { + readonly onCommitStarted?: () => void; + readonly discardPreparedMessages?: (sessionId: string) => Promise; + readonly prepareMessages?: ( + sessionId: string, + messages: readonly StoredMessage[], + ) => Promise; + } = {}, ): Promise { await this.ensureReady(); assertNoConversationCopyMetadata(input); if (input.subagentSpawn) { throw new Error('Subagent spawn metadata requires createSubagent()'); } - const canonicalMessages = messages.map((message) => - decodeCanonicalMessage(JSON.parse(JSON.stringify(message)) as unknown), - ); const header: SessionHeader = { ...buildSessionHeader(this.workspaceRoot, input), externalOrigin, transcriptLedgerVersion: 0, }; - const catalogProjection = projectSessionCatalogMessages(canonicalMessages); - options.onCommitStarted?.(); - const outcome = await this.metadata.importSession(header, canonicalMessages, catalogProjection); - if (outcome !== 'imported') { - throw new Error(`Generated Session id already exists: ${header.id}`); + let commitStarted = false; + try { + const preparedMessages = options.prepareMessages + ? await options.prepareMessages(header.id, messages) + : messages; + const canonicalMessages = preparedMessages.map((message) => + decodeCanonicalMessage(JSON.parse(JSON.stringify(message)) as unknown), + ); + const catalogProjection = projectSessionCatalogMessages(canonicalMessages); + options.onCommitStarted?.(); + commitStarted = true; + const outcome = await this.metadata.importSession( + header, + canonicalMessages, + catalogProjection, + ); + if (outcome !== 'imported') { + throw new Error(`Generated Session id already exists: ${header.id}`); + } + return (await this.metadata.read(header.id)).header; + } catch (error) { + let uncommitted = !commitStarted; + if (commitStarted) { + try { + await this.metadata.read(header.id); + } catch (readError) { + // An uncertain commit must retain its bytes for recovery. Only an + // authoritative absence makes the prepared artifacts safe to discard. + uncommitted = isSessionNotFoundError(readError); + } + } + if (uncommitted) { + try { + await options.discardPreparedMessages?.(header.id); + } catch { + // Best-effort cleanup must not replace the original import failure. + } + } + throw error; } - return (await this.metadata.read(header.id)).header; } async lookupExternalSessionImports( diff --git a/packages/storage/src/test-only/memory-execution-session.ts b/packages/storage/src/test-only/memory-execution-session.ts index 3eeed2e194b..fd42c227316 100644 --- a/packages/storage/src/test-only/memory-execution-session.ts +++ b/packages/storage/src/test-only/memory-execution-session.ts @@ -469,12 +469,15 @@ export function createMemorySessionStore( write('session.createStable', (s) => stable(s, root, request, initial)), probeStableSessionCreate: async (id, fingerprint) => read((s) => probe(s, id, fingerprint)), createImportedSession: async (input, values, origin, options) => { - const canonicalValues = values.map((value) => decodeCanonicalMessage(copy(value))); const h = { ...buildSessionHeader(root, input), externalOrigin: copy(origin), transcriptLedgerVersion: 0 as const, }; + const preparedValues = options?.prepareMessages + ? await options.prepareMessages(h.id, values) + : values; + const canonicalValues = preparedValues.map((value) => decodeCanonicalMessage(copy(value))); const catalogProjection = projectSessionCatalogMessages(canonicalValues); options?.onCommitStarted?.(); return write('session.import', (s) => {