feat(sdk): host seams for upload processing without OS processes

packages/sdk/src/bootstrap-image-resizer.test.ts

@@ -9,6 +9,7 @@ import { SessionFileStore } from './core/file-store/file-store.js'
 import type { ImageResizeOptions, ImageResizer } from './core/image/types.js'
 import { ModelId } from './core/llm/schema.js'
 import type { Platform, ProcessRunner } from './platform/index.js'
+import { fileUploadSource } from './plugins/uploads/source.js'
 import { createTestPreset } from './testing/preset-helpers.js'
 
 const cleanups: Array<() => Promise<void> | void> = []
@@ -63,7 +64,15 @@ async function classifyImage(host: Awaited<ReturnType<typeof boot>>) {
 	const classifier = host.services.preprocessorRegistry.getForMimeType('image/png')
 	if (!classifier) throw new Error('No image preprocessor registered')
 	const files = new SessionFileStore(host.dataPath, undefined, false, host.services.platform.fs, 'session')
-	await classifier.process(host.imagePath, 'image/png', { files: files.scoped('upload') })
+	const source = await fileUploadSource(host.services.platform.fs, host.imagePath)
+	if (!source.ok) throw source.error
+	await classifier.process(source.value, 'image/png', {
+		files: files.scoped('upload'),
+		uploadId: 'upload',
+		sessionId: 'session',
+		filename: 'screenshot.png',
+		attempt: 1,
+	})
 }
 
 describe('bootstrap image resizer', () => {

packages/sdk/src/bootstrap.ts

@@ -17,9 +17,11 @@ import type { LLMProvider } from '~/core/llm/provider.js'
 import { RoutingLLMProvider } from '~/core/llm/routing-provider.js'
 import type { RoutableLLMProvider } from '~/core/llm/routing-provider.js'
 import type { Preset } from '~/core/preset/index.js'
+import type { FileSystem } from '~/platform/fs.js'
 import type { Platform } from '~/platform/index.js'
-import { PreprocessorRegistry } from '~/plugins/uploads/preprocessor.js'
+import { type Preprocessor, PreprocessorRegistry } from '~/plugins/uploads/preprocessor.js'
 import { ImageClassifierPreprocessor, MarkitdownPreprocessor, PdfPreprocessor, ZipPreprocessor } from '~/plugins/uploads/preprocessors/index.js'
+import type { UploadBlobStore } from '~/plugins/uploads/source.js'
 import type { Config } from './config.js'
 import { SessionFileStore } from './core/file-store/file-store.js'
 import type { SessionManager } from './core/sessions/session-manager.js'
@@ -107,6 +109,26 @@ export type BuiltinMethodSchemas = AllMethodSchemas<typeof fullPlugins>
 /** Method schemas the `isolate` profile actually registers */
 export type IsolateMethodSchemas = AllMethodSchemas<typeof isolatePlugins>
 
+/** What a host's {@link BootstrapUploadOptions.createPreprocessors} builds its preprocessors from. */
+export interface UploadPreprocessorDeps {
+	llmProvider: LLMProvider
+	imageResizer: ImageResizer
+	logger: Logger
+	fs: FileSystem
+	/** The default list (image classifier, pdf, markitdown, zip) so a host can keep some of them. */
+	builtins: () => Preprocessor[]
+}
+
+/** How uploads are processed and where host-stored upload bytes live. */
+export interface BootstrapUploadOptions {
+	/** Replaces the built-in preprocessor list. Order = match priority (first match wins). */
+	createPreprocessors?: (deps: UploadPreprocessorDeps) => Preprocessor[]
+	/** Required for registerStored. */
+	blobStore?: UploadBlobStore
+	/** Bound on one processing run in ms; defaults to 120 s. Hosts with slow conversions raise it. */
+	processingTimeoutMs?: number
+}
+
 /** Options for {@link bootstrap}. */
 export interface BootstrapOptions<TProfile extends PluginProfile = PluginProfile> {
 	/** Built-in plugin set to register. Defaults to `full`. */
@@ -118,6 +140,7 @@ export interface BootstrapOptions<TProfile extends PluginProfile = PluginProfile
 	 * here, rather than having the store it wants patched in after the fact.
 	 */
 	eventStore?: EventStore
+	uploads?: BootstrapUploadOptions
 }
 
 /**
@@ -142,6 +165,10 @@ export interface Services<TProfile extends PluginProfile = 'full'> {
 	pidRegistry: ServicePidRegistry
 	/** Preprocessor registry for upload content extraction */
 	preprocessorRegistry: PreprocessorRegistry
+	/** Where the host keeps upload bytes it stored itself. */
+	uploadBlobStore?: UploadBlobStore
+	/** Bound on one upload processing run in ms; absent keeps the plugin default. */
+	uploadProcessingTimeoutMs?: number
 	/** Host-environment adapters (filesystem, process, scheduler). */
 	platform: Platform
 	/** Built-in plugin set chosen at bootstrap — selects what createSystemFromServices registers. */
@@ -190,15 +217,27 @@ export function bootstrap(config: Config, userConfig: RojConfig, platform: Platf
 
 	const preprocessorRegistry = new PreprocessorRegistry()
 	const imageClassifierGate = new Semaphore(config.imageClassifierConcurrency ?? 10)
-	// Shares the LLM provider's resizer; the classifier asks it for 1024px images per call.
-	preprocessorRegistry.register(new ImageClassifierPreprocessor({ llmProvider, logger, fs: platform.fs, gate: imageClassifierGate, imageResizer }))
-	// PdfPreprocessor must come before MarkitdownPreprocessor — both could
-	// match `application/pdf` in principle, but the registry uses first-hit
-	// and PdfPreprocessor's `pdftotext + pdfimages -all + streaming` pipeline
-	// is dramatically faster than markitdown's pdfminer.six backend.
-	preprocessorRegistry.register(new PdfPreprocessor({ registry: preprocessorRegistry, logger, fs: platform.fs, process: platform.process }))
-	preprocessorRegistry.register(new MarkitdownPreprocessor({ registry: preprocessorRegistry, logger, fs: platform.fs, process: platform.process }))
-	preprocessorRegistry.register(new ZipPreprocessor({ registry: preprocessorRegistry, logger, process: platform.process }))
+	// Built-ins look nested files up in this registry, so they see whatever list the host chose.
+	const builtins = (): Preprocessor[] => [
+		// Shares the LLM provider's resizer; the classifier asks it for 1024px images per call.
+		new ImageClassifierPreprocessor({ llmProvider, logger, fs: platform.fs, gate: imageClassifierGate, imageResizer, tmpDir: platform.tmpDir }),
+		// PdfPreprocessor must come before MarkitdownPreprocessor — both could
+		// match `application/pdf` in principle, but the registry uses first-hit
+		// and PdfPreprocessor's `pdftotext + pdfimages -all + streaming` pipeline
+		// is dramatically faster than markitdown's pdfminer.six backend.
+		new PdfPreprocessor({ registry: preprocessorRegistry, logger, fs: platform.fs, process: platform.process }),
+		new MarkitdownPreprocessor({ registry: preprocessorRegistry, logger, fs: platform.fs, process: platform.process }),
+		new ZipPreprocessor({ registry: preprocessorRegistry, logger, fs: platform.fs, process: platform.process }),
+	]
+	const uploadProcessingTimeoutMs = options?.uploads?.processingTimeoutMs
+	if (uploadProcessingTimeoutMs !== undefined && (!Number.isSafeInteger(uploadProcessingTimeoutMs) || uploadProcessingTimeoutMs <= 0)) {
+		throw new Error(`Invalid uploads.processingTimeoutMs: ${uploadProcessingTimeoutMs}`)
+	}
+	const createPreprocessors = options?.uploads?.createPreprocessors
+	const preprocessors = createPreprocessors
+		? createPreprocessors({ llmProvider, imageResizer, logger, fs: platform.fs, builtins })
+		: builtins()
+	for (const preprocessor of preprocessors) preprocessorRegistry.register(preprocessor)
 
 	logger.info('Bootstrap complete')
 
@@ -215,6 +254,8 @@ export function bootstrap(config: Config, userConfig: RojConfig, platform: Platf
 		portPool,
 		pidRegistry,
 		preprocessorRegistry,
+		uploadBlobStore: options?.uploads?.blobStore,
+		uploadProcessingTimeoutMs,
 		platform,
 		pluginProfile,
 	}
@@ -313,6 +354,8 @@ function buildSystem(services: Services<PluginProfile>, options?: CreateSystemFr
 		dataFileStore: services.dataFileStore,
 		onUserOutput: options?.onUserOutput,
 		preprocessorRegistry: services.preprocessorRegistry,
+		uploadBlobStore: services.uploadBlobStore,
+		uploadProcessingTimeoutMs: services.uploadProcessingTimeoutMs,
 		uploadArchiveLimits: services.config.uploadArchiveLimits,
 		uploadMaxFileSize: services.config.uploadMaxFileSize,
 		resourceArchiveLimits: services.config.resourceArchiveLimits,

packages/sdk/src/core/agents/agent.ts

@@ -16,6 +16,7 @@ import { agentEvents } from '~/core/agents/state.js'
 import { withSessionId } from '~/core/events/test-helpers.js'
 import type { FileStore } from '~/core/file-store/types.js'
 import { applyCacheBreakpoint, type CacheTtlConfig } from '~/core/llm/cache-breakpoints.js'
+import { storeDocuments, storeMessageDocuments } from '~/core/llm/document-content.js'
 import type { ToolResultContent } from '~/core/llm/llm-log-types.js'
 import type { InferenceRequest, InferenceResponse, LLMError, LLMMessage, LLMProvider } from '~/core/llm/provider.js'
 import { LLMCallId, ModelId } from '~/core/llm/schema.js'
@@ -682,13 +683,17 @@ export class Agent {
 		}
 
 		// 2. Build pending messages (tool results only)
-		const pendingMessages = this.buildPendingMessages(agentState)
+		const collectedMessages = this.buildPendingMessages(agentState)
 
 		// 2b. Append plugin dequeued messages (includes mailbox messages)
 		for (const dequeued of pluginDequeued) {
-			pendingMessages.push(...dequeued.messages)
+			collectedMessages.push(...dequeued.messages)
 		}
 
+		// 2c. Move inline documents to the session store: history keeps a reference the provider reads
+		// on every inference, and the event log never holds their bytes.
+		const pendingMessages = await Promise.all(collectedMessages.map((message) => storeMessageDocuments(message, this.fileStore)))
+
 		// 3. Inference start - emit with pending messages
 		await this.store.emit(withSessionId(
 			this.store.sessionId,
@@ -1107,9 +1112,9 @@ export class Agent {
 
 		const result = await this.toolExecutor.execute(tool, effectiveToolCall.input, context)
 
-		// Build initial tool result with ToolResultContent
+		// Build initial tool result with ToolResultContent; its documents go to the session store before any event records it
 		let toolResult: { isError: boolean; content: ToolResultContent } = result.ok
-			? { isError: false, content: result.value }
+			? { isError: false, content: await storeDocuments(result.value, this.fileStore) }
 			: { isError: true, content: result.error.message }
 
 		// afterToolCall handler - can modify result or pause.
@@ -1521,9 +1526,14 @@ export class Agent {
 					continue
 				}
 				switch (result.action) {
-					case 'modify':
-						await this.emitHandlerCompleted('afterToolCall', result)
-						return result
+					case 'modify': {
+						const modified: AfterToolCallResult = {
+							action: 'modify',
+							result: { ...result.result, content: await storeDocuments(result.result.content, this.fileStore) },
+						}
+						await this.emitHandlerCompleted('afterToolCall', modified)
+						return modified
+					}
 					case 'pause':
 						await this.emitHandlerCompleted('afterToolCall', result)
 						await this.emitHandlerPause(result.reason)

packages/sdk/src/core/agents/document-history.integration.test.ts

@@ -0,0 +1,186 @@
+import { describe, expect, it } from 'bun:test'
+import z from 'zod/v4'
+import { agentEvents } from '~/core/agents/state.js'
+import type { LLMMessage } from '~/core/agents/state.js'
+import type { ChatMessageContentItem } from '~/core/llm/llm-log-types.js'
+import { MockLLMProvider } from '~/core/llm/mock.js'
+import type { InferenceResponse } from '~/core/llm/provider.js'
+import { definePlugin } from '~/core/plugins/plugin-builder.js'
+import { createTool } from '~/core/tools/definition.js'
+import { ToolCallId } from '~/core/tools/schema.js'
+import { toolEvents } from '~/core/tools/state.js'
+import { Err, Ok } from '~/lib/utils/result.js'
+import { createTestPreset, TestHarness, TestSession } from '~/testing/index.js'
+
+const DOCUMENT_BYTES = Buffer.from('%PDF-1.4\nscanned pages\n')
+const DOCUMENT_DATA = DOCUMENT_BYTES.toString('base64')
+const DOCUMENT_SHA256 = new Bun.CryptoHasher('sha256').update(DOCUMENT_BYTES).digest('hex')
+const STORED_PATH = `.documents/${DOCUMENT_SHA256}.pdf`
+/** What `read_stored` answers when the session store holds the document's bytes. */
+const READ_BACK = `read ${DOCUMENT_BYTES.length} bytes, sha256 ${DOCUMENT_SHA256}`
+
+const inlineDocument = { type: 'document' as const, document: { mediaType: 'application/pdf' as const, data: DOCUMENT_DATA, filename: 'scan.pdf' } }
+const storedDocumentItem: ChatMessageContentItem = {
+	type: 'document',
+	document: { mediaType: 'application/pdf', path: STORED_PATH, byteLength: DOCUMENT_BYTES.length, filename: 'scan.pdf' },
+}
+
+/** Delivers one user message carrying a PDF, the way a host hands the agent a document. */
+const createDocumentPlugin = () => {
+	let delivered = false
+	return definePlugin('document-source')
+		.dequeue<true>({
+			hasPendingMessages: () => !delivered,
+			getPendingMessages: () => delivered
+				? null
+				: { messages: [{ role: 'user', content: [inlineDocument, { type: 'text', text: 'Transcribe it.' }] }], token: true },
+			markConsumed: async () => {
+				delivered = true
+			},
+		})
+		.build()
+}
+
+/** Reads a file through the session store, the way a provider reads a stored document back. */
+const readStoredPlugin = definePlugin('read-stored')
+	.tools(() => [
+		createTool({
+			name: 'read_stored',
+			description: 'Read a stored document back',
+			input: z.object({ path: z.string() }),
+			execute: async (input, context) => {
+				const read = await context.files.session.read(input.path, { type: 'buffer' })
+				if (!read.ok) return Err({ message: read.error, recoverable: true })
+				return Ok(`read ${read.value.length} bytes, sha256 ${new Bun.CryptoHasher('sha256').update(read.value).digest('hex')}`)
+			},
+		}),
+		createTool({
+			name: 'fetch_scan',
+			description: 'Return a scanned PDF',
+			input: z.object({}),
+			execute: async () => Ok([inlineDocument]),
+		}),
+		createTool({ name: 'noop', description: 'Do nothing', input: z.object({}), execute: async () => Ok('ok') }),
+	])
+	.build()
+
+/** Swaps every `noop` result for the PDF, as a plugin rewriting a tool's result would. */
+const attachScanPlugin = definePlugin('attach-scan')
+	.hook('afterToolCall', async (ctx) => ctx.toolCall.name === 'noop' ? { action: 'modify', result: { isError: false, content: [inlineDocument] } } : null)
+	.build()
+
+const callTool = (requests: LLMMessage[][], name: string, input: Record<string, unknown> = {}): InferenceResponse => ({
+	content: null,
+	toolCalls: [{ id: ToolCallId(`tc${requests.length}`), name, input }],
+	finishReason: 'tool_calls',
+	metrics: MockLLMProvider.defaultMetrics(),
+})
+
+const done = (): InferenceResponse => ({ content: 'Done', toolCalls: [], finishReason: 'stop', metrics: MockLLMProvider.defaultMetrics() })
+
+/** Answers each user turn by reading the stored document back, then finishes once the tool result is in. */
+const readBackEachTurn = (requests: LLMMessage[][]) => (request: { messages: LLMMessage[] }): InferenceResponse => {
+	requests.push(request.messages)
+	return request.messages[request.messages.length - 1]?.role === 'tool' ? done() : callTool(requests, 'read_stored', { path: STORED_PATH })
+}
+
+describe('a document in a pending message', () => {
+	it('stays in every inference of the turn and of later turns as a stored reference', async () => {
+		const requests: LLMMessage[][] = []
+		const harness = new TestHarness({
+			presets: [createTestPreset()],
+			systemPlugins: [createDocumentPlugin(), readStoredPlugin],
+			mockHandler: readBackEachTurn(requests),
+		})
+
+		const session = await harness.createSession('test')
+		await session.sendAndWaitForIdle('Here is a file.')
+		await session.sendAndWaitForIdle('Anything else?')
+
+		// Two turns of two inferences each; the second of each turn follows a tool call.
+		expect(requests).toHaveLength(4)
+		for (const request of requests) {
+			expect(request).toContainEqual(expect.objectContaining({ role: 'user', content: [storedDocumentItem, { type: 'text', text: 'Transcribe it.' }] }))
+		}
+
+		// The reference names the document's bytes in the session store.
+		const completed = await session.getEventsByType(toolEvents, 'tool_completed')
+		expect(completed.map((event) => event.result)).toEqual([READ_BACK, READ_BACK])
+		expect(JSON.stringify(await session.getEvents())).not.toContain(DOCUMENT_DATA)
+
+		await harness.shutdown()
+	})
+
+	it('stays readable in a fork, which gets its own copy of the stored documents', async () => {
+		const requests: LLMMessage[][] = []
+		const harness = new TestHarness({
+			presets: [createTestPreset()],
+			systemPlugins: [createDocumentPlugin(), readStoredPlugin],
+			mockHandler: readBackEachTurn(requests),
+		})
+
+		const source = await harness.createSession('test')
+		await source.sendAndWaitForIdle('Here is a file.')
+		const sourceEvents = await source.getEvents()
+		const forkResult = await harness.sessionManager.forkSession(source.sessionId, sourceEvents.length - 1)
+		if (!forkResult.ok) throw new Error('Expected fork')
+		const fork = new TestSession(forkResult.value, harness)
+
+		await fork.sendAndWaitForIdle('Read it again.')
+
+		const completed = await fork.getEventsByType(toolEvents, 'tool_completed')
+		expect(completed.map((event) => event.result)).toEqual([READ_BACK, READ_BACK])
+		expect(requests[requests.length - 1]).toContainEqual(expect.objectContaining({ role: 'user', content: [storedDocumentItem, { type: 'text', text: 'Transcribe it.' }] }))
+
+		await harness.shutdown()
+	})
+})
+
+describe('a document returned by a tool', () => {
+	it('is recorded in tool_completed as a stored reference and reaches the next inference', async () => {
+		const requests: LLMMessage[][] = []
+		const harness = new TestHarness({
+			presets: [createTestPreset()],
+			systemPlugins: [readStoredPlugin],
+			mockHandler: (request) => {
+				requests.push(request.messages)
+				return requests.length === 1 ? callTool(requests, 'fetch_scan') : done()
+			},
+		})
+
+		const session = await harness.createSession('test')
+		await session.sendAndWaitForIdle('Fetch the scan.')
+
+		const completed = await session.getEventsByType(toolEvents, 'tool_completed')
+		expect(completed.map((event) => event.result)).toEqual([[storedDocumentItem]])
+		expect(requests[1]).toContainEqual(expect.objectContaining({ role: 'tool', content: [storedDocumentItem] }))
+		expect(JSON.stringify(await session.getEvents())).not.toContain(DOCUMENT_DATA)
+
+		await harness.shutdown()
+	})
+
+	it('put there by an afterToolCall hook is recorded in handler_completed and tool_completed only as a reference', async () => {
+		const requests: LLMMessage[][] = []
+		const harness = new TestHarness({
+			presets: [createTestPreset()],
+			systemPlugins: [readStoredPlugin, attachScanPlugin],
+			mockHandler: (request) => {
+				requests.push(request.messages)
+				return requests.length === 1 ? callTool(requests, 'noop') : done()
+			},
+		})
+
+		const session = await harness.createSession('test')
+		await session.sendAndWaitForIdle('Run it.')
+
+		const handled = await session.getEventsByType(agentEvents, 'handler_completed')
+		expect(handled.filter((event) => event.handlerName === 'afterToolCall').map((event) => event.result)).toEqual([
+			{ action: 'modify', result: { isError: false, content: [storedDocumentItem] } },
+		])
+		const completed = await session.getEventsByType(toolEvents, 'tool_completed')
+		expect(completed.map((event) => event.result)).toEqual([[storedDocumentItem]])
+		expect(JSON.stringify(await session.getEvents())).not.toContain(DOCUMENT_DATA)
+
+		await harness.shutdown()
+	})
+})