fix(runtime): support external storage for large session blobs

Author: mhart

packages/cli/src/lib/build-plugin-cloudflare.ts

@@ -156,6 +156,7 @@ import {
   CLOUDFLARE_AGENT_INTERNAL_DISPATCH_PATH,
   CLOUDFLARE_WORKFLOW_INTERNAL_METADATA_PATH,
   createCloudflareAgentRuntime,
+  createR2SessionAttachmentStore,
   createSqlSessionStore,
    SqliteEventStreamStore,
   bashFactoryToSessionEnv,
@@ -206,6 +207,7 @@ if (!hasRegisteredProvider('cloudflare')) {
 const skills = {};
 const packagedSkills = ${packagedSkillsValue};
 const systemPrompt = '';
+const sessionAttachmentStore = createR2SessionAttachmentStore(env.FLUE_SESSION_ATTACHMENTS);
 
 ${builtModuleNormalizationSource}
 const agentModules = {
@@ -348,7 +350,7 @@ function createAgentContextForRequest(executionStore, id, payload, doInstance, r
 
 function createWorkflowContextForRequest(id, runId, payload, doInstance, req, initialEventIndex, dispatchId) {
   const storage = doInstance?.ctx?.storage;
-  const defaultStore = storage?.sql ? createSqlSessionStore(storage.sql, storage.transactionSync?.bind(storage)) : memoryWorkflowSessionStore;
+  const defaultStore = storage?.sql ? createSqlSessionStore(storage.sql, storage.transactionSync?.bind(storage), { attachmentStore: sessionAttachmentStore }) : memoryWorkflowSessionStore;
   return createContextForRequest(id, runId, payload, doInstance, req, defaultStore, initialEventIndex, dispatchId);
 }
 
@@ -403,6 +405,7 @@ function createEventStreamStoreForInstance(doInstance) {
 
 const cloudflareAgents = createCloudflareAgentRuntime({
   createdAgents,
+  sessionStoreOptions: { attachmentStore: sessionAttachmentStore },
   createContext: ({ executionStore, instance, payload, request, initialEventIndex, dispatchId }) =>
     createAgentContextForRequest(executionStore, instance.name, payload, instance, request, initialEventIndex, dispatchId),
   runWithInstanceContext: (instance, agentName, fn) => runWithInstanceContext(instance, agentRuntimeIdentity(agentName), fn),

packages/runtime/src/cloudflare/agent-coordinator.ts

@@ -1,4 +1,8 @@
-import type { AgentExecutionStore, AgentSubmission, AgentSubmissionStore } from '../agent-execution-store.ts';
+import type {
+	AgentExecutionStore,
+	AgentSubmission,
+	AgentSubmissionStore,
+} from '../agent-execution-store.ts';
 import type { FlueContextInternal } from '../client.ts';
 import {
 	createAgentSubmissionObserverRegistry,
@@ -22,6 +26,7 @@ const FLUE_AGENT_SUBMISSION_ATTEMPT_STALE_MS = 15 * 60 * 1000;
 const FLUE_AGENT_SUBMISSION_ATTEMPT_FIBER = 'flue:submission-attempt';
 
 import type { SqlStorage } from '../sql-storage.ts';
+import type { SqlSessionStoreOptions } from '../sql-agent-execution-store.ts';
 
 interface CloudflareAgentStorage {
 	sql?: SqlStorage;
@@ -73,7 +78,10 @@ interface CloudflareAgentRuntimeOptions {
 		agentName: string,
 		callback: () => T,
 	) => T;
-	readonly createEventStreamStore: (instance: CloudflareAgentInstance) => import('../runtime/event-stream-store.ts').EventStreamStore;
+	readonly createEventStreamStore: (
+		instance: CloudflareAgentInstance,
+	) => import('../runtime/event-stream-store.ts').EventStreamStore;
+	readonly sessionStoreOptions?: SqlSessionStoreOptions;
 }
 
 export interface CloudflareAgentRuntime {
@@ -96,7 +104,9 @@ export interface CloudflareAgentRuntime {
 	): Promise<unknown>;
 }
 
-export function createCloudflareAgentRuntime(options: CloudflareAgentRuntimeOptions): CloudflareAgentRuntime {
+export function createCloudflareAgentRuntime(
+	options: CloudflareAgentRuntimeOptions,
+): CloudflareAgentRuntime {
 	const coordinators = new WeakMap<CloudflareAgentInstance, CloudflareAgentCoordinator>();
 	const observers = createAgentSubmissionObserverRegistry();
 	const activeAttempts = new Set<string>();
@@ -113,7 +123,11 @@ export function createCloudflareAgentRuntime(options: CloudflareAgentRuntimeOpti
 		prepare({ storage, className, agentName }) {
 			return {
 				agentName,
-				executionStore: createSqlAgentExecutionStore(storage, className),
+				executionStore: createSqlAgentExecutionStore(
+					storage,
+					className,
+					options.sessionStoreOptions,
+				),
 			};
 		},
 		attach(instance, prepared) {
@@ -184,7 +198,8 @@ class CloudflareAgentCoordinator {
 				id: this.instance.name,
 				agentName: this.agentName,
 				eventStreamStore: this.eventStreamStore,
-				admitAttachedSubmission: (payload, onEvent, waitForResult) => this.admitAttachedSubmission(payload, onEvent, waitForResult),
+				admitAttachedSubmission: (payload, onEvent, waitForResult) =>
+					this.admitAttachedSubmission(payload, onEvent, waitForResult),
 			}),
 		);
 	}
@@ -242,7 +257,9 @@ class CloudflareAgentCoordinator {
 		}
 	}
 
-	private armSubmissionWake(options: { delaySeconds?: number; idempotent?: boolean } = {}): Promise<unknown> {
+	private armSubmissionWake(
+		options: { delaySeconds?: number; idempotent?: boolean } = {},
+	): Promise<unknown> {
 		this.assertAgentsDurabilityApi('schedule');
 		return this.instance.schedule(
 			options.delaySeconds ?? FLUE_AGENT_SUBMISSION_WAKE_SECONDS,
@@ -258,7 +275,9 @@ class CloudflareAgentCoordinator {
 		return true;
 	}
 
-	private async reconcileSubmissions(options: { driverAlreadyArmed?: boolean } = {}): Promise<boolean> {
+	private async reconcileSubmissions(
+		options: { driverAlreadyArmed?: boolean } = {},
+	): Promise<boolean> {
 		if (!(await this.submissions.hasUnsettledSubmissions())) return false;
 		if (!options.driverAlreadyArmed) await this.restoreSubmissionWake();
 		try {
@@ -277,16 +296,16 @@ class CloudflareAgentCoordinator {
 				}
 			}
 			for (const submission of await this.submissions.listRunnableSubmissions()) {
-			// Cloudflare DOs are single-threaded per instance — leases are
-			// advisory-only. Set to 0 so reconciliation never misidentifies
-			// an active submission as expired. The Node coordinator uses real
-			// lease expiry with heartbeat renewal for multi-process safety.
-			const claimed = await this.submissions.claimSubmission({
-				submissionId: submission.submissionId,
-				attemptId: crypto.randomUUID(),
-				ownerId: this.instance.ctx.id.toString(),
-				leaseExpiresAt: 0,
-			});
+				// Cloudflare DOs are single-threaded per instance — leases are
+				// advisory-only. Set to 0 so reconciliation never misidentifies
+				// an active submission as expired. The Node coordinator uses real
+				// lease expiry with heartbeat renewal for multi-process safety.
+				const claimed = await this.submissions.claimSubmission({
+					submissionId: submission.submissionId,
+					attemptId: crypto.randomUUID(),
+					ownerId: this.instance.ctx.id.toString(),
+					leaseExpiresAt: 0,
+				});
 				if (!claimed) continue;
 				try {
 					this.startSubmissionAttempt(claimed);
@@ -339,7 +358,12 @@ class CloudflareAgentCoordinator {
 				submission,
 				agent,
 				(payload, dispatchId) =>
-					this.createContext(payload, submissionSyntheticRequest(submission.input), undefined, dispatchId),
+					this.createContext(
+						payload,
+						submissionSyntheticRequest(submission.input),
+						undefined,
+						dispatchId,
+					),
 				{ ownerId: this.instance.ctx.id.toString(), leaseExpiresAt: 0 },
 			),
 		);
@@ -400,24 +424,32 @@ class CloudflareAgentCoordinator {
 		if (!rows) throw new Error('[flue] Cloudflare durable agent SQL storage is unavailable.');
 		for (const row of rows) {
 			if (typeof row.created_at !== 'number') {
-				console.warn('[flue:submission-reconciliation] Skipping attempt marker with non-numeric created_at.');
+				console.warn(
+					'[flue:submission-reconciliation] Skipping attempt marker with non-numeric created_at.',
+				);
 				continue;
 			}
 			if (Date.now() - row.created_at > FLUE_AGENT_SUBMISSION_ATTEMPT_STALE_MS) continue;
 			if (row.snapshot === null) continue;
 			if (typeof row.snapshot !== 'string') {
-				console.warn('[flue:submission-reconciliation] Skipping attempt marker with non-string snapshot.');
+				console.warn(
+					'[flue:submission-reconciliation] Skipping attempt marker with non-string snapshot.',
+				);
 				continue;
 			}
 			let snapshot: unknown;
 			try {
 				snapshot = JSON.parse(row.snapshot);
 			} catch {
-				console.warn('[flue:submission-reconciliation] Skipping attempt marker with unparseable snapshot.');
+				console.warn(
+					'[flue:submission-reconciliation] Skipping attempt marker with unparseable snapshot.',
+				);
 				continue;
 			}
 			if (!isAttemptMarkerSnapshot(snapshot)) {
-				console.warn('[flue:submission-reconciliation] Skipping attempt marker with invalid snapshot shape.');
+				console.warn(
+					'[flue:submission-reconciliation] Skipping attempt marker with invalid snapshot shape.',
+				);
 				continue;
 			}
 			keys.add(`${snapshot.submissionId}:${snapshot.attemptId}`);
@@ -439,7 +471,12 @@ class CloudflareAgentCoordinator {
 				return agent;
 			},
 			createContext: (payload, dispatchId) => {
-				const ctx = this.createContext(payload, submissionSyntheticRequest(submission.input), undefined, dispatchId);
+				const ctx = this.createContext(
+					payload,
+					submissionSyntheticRequest(submission.input),
+					undefined,
+					dispatchId,
+				);
 				const streamPath = agentStreamPath(this.agentName, this.instance.name);
 				ctx.subscribeEvent((event) => {
 					eventStreamStore.appendEvent(streamPath, event).catch((error) => {
@@ -472,7 +509,11 @@ class CloudflareAgentCoordinator {
 		onEvent?: (event: AttachedAgentEvent) => Promise<void> | void,
 		waitForResult = true,
 	): Promise<unknown> {
-		const input = createDirectAgentSubmissionInput({ agent: this.agentName, id: this.instance.name, payload });
+		const input = createDirectAgentSubmissionInput({
+			agent: this.agentName,
+			id: this.instance.name,
+			payload,
+		});
 		const attachment = this.observers.attach(input.submissionId, { onEvent });
 		try {
 			await this.armSubmissionWake();
@@ -512,12 +553,16 @@ class CloudflareAgentCoordinator {
 			return new Response('Conflicting internal dispatch replay.', { status: 409 });
 		}
 		await this.reconcileSubmissions({ driverAlreadyArmed: true });
-		return Response.json({ dispatchId: admission.submission.submissionId, acceptedAt: input.acceptedAt });
+		return Response.json({
+			dispatchId: admission.submission.submissionId,
+			acceptedAt: input.acceptedAt,
+		});
 	}
-
 }
 
-function isAttemptMarkerSnapshot(value: unknown): value is { submissionId: string; attemptId: string } {
+function isAttemptMarkerSnapshot(
+	value: unknown,
+): value is { submissionId: string; attemptId: string } {
 	if (!value || typeof value !== 'object') return false;
 	const snapshot = value as Record<string, unknown>;
 	return typeof snapshot.submissionId === 'string' && typeof snapshot.attemptId === 'string';
@@ -528,7 +573,8 @@ function submissionAttemptMarkerKey(submission: AgentSubmission): string {
 }
 
 function isInternalDispatchRequest(request: Request): boolean {
-	return request.method === 'POST' && new URL(request.url).pathname === CLOUDFLARE_AGENT_INTERNAL_DISPATCH_PATH;
+	return (
+		request.method === 'POST' &&
+		new URL(request.url).pathname === CLOUDFLARE_AGENT_INTERNAL_DISPATCH_PATH
+	);
 }
-
-

packages/runtime/src/cloudflare/agent-execution-store.ts

@@ -4,6 +4,8 @@ import {
 	createSqlAgentExecutionStoreFromSql,
 	ensureSqlAgentExecutionTables,
 	ensureSessionTable,
+	type SessionAttachmentStore,
+	type SqlSessionStoreOptions,
 	SqlSessionStore,
 } from '../sql-agent-execution-store.ts';
 import type { SessionStore } from '../types.ts';
@@ -13,17 +15,47 @@ interface DurableObjectStorage {
 	transactionSync?<T>(closure: () => T): T;
 }
 
+interface R2BucketLike {
+	put(key: string, value: string, options?: unknown): Promise<unknown>;
+	get(key: string): Promise<{ text(): Promise<string> } | null>;
+	delete(key: string): Promise<unknown>;
+}
+
 export function createSqlSessionStore(
 	sql: SqlStorage,
 	transactionSync?: <T>(closure: () => T) => T,
+	options: SqlSessionStoreOptions = {},
 ): SessionStore {
 	ensureSessionTable(sql);
-	return new SqlSessionStore(sql, transactionSync);
+	return new SqlSessionStore(sql, transactionSync, options);
+}
+
+export function createR2SessionAttachmentStore(
+	bucket: R2BucketLike | null | undefined,
+): SessionAttachmentStore | undefined {
+	if (!bucket) return undefined;
+	return {
+		async put(key: string, data: string): Promise<void> {
+			await bucket.put(key, data, {
+				httpMetadata: { contentType: 'text/plain; charset=utf-8' },
+				customMetadata: { flue: 'session-attachment' },
+			});
+		},
+		async get(key: string): Promise<string> {
+			const object = await bucket.get(key);
+			if (!object) throw new Error('[flue] Persisted session attachment object is missing.');
+			return object.text();
+		},
+		async delete(key: string): Promise<void> {
+			await bucket.delete(key);
+		},
+	};
 }
 
 export function createSqlAgentExecutionStore(
 	storage: DurableObjectStorage | undefined,
 	className: string,
+	options: SqlSessionStoreOptions = {},
 ): AgentExecutionStore {
 	const sql = storage?.sql;
 	const transactionSync = storage?.transactionSync;
@@ -38,7 +70,7 @@ export function createSqlAgentExecutionStore(
 	try {
 		ensureSqlAgentExecutionTables(sql);
 		const runTransaction = <T>(closure: () => T): T => transactionSync.call(storage, closure) as T;
-		return createSqlAgentExecutionStoreFromSql(sql, runTransaction);
+		return createSqlAgentExecutionStoreFromSql(sql, runTransaction, options);
 	} catch (cause) {
 		const detail = cause instanceof Error ? cause.message : String(cause);
 		throw new Error(

packages/runtime/src/internal.ts

@@ -26,7 +26,10 @@ export {
 	createCloudflareAgentRuntime,
 } from './cloudflare/agent-coordinator.ts';
 export { CLOUDFLARE_WORKFLOW_INTERNAL_METADATA_PATH } from './runtime/flue-app.ts';
-export { createSqlSessionStore } from './cloudflare/agent-execution-store.ts';
+export {
+	createR2SessionAttachmentStore,
+	createSqlSessionStore,
+} from './cloudflare/agent-execution-store.ts';
 export { createDurableRunStore } from './cloudflare/run-store.ts';
 export { createNodeAgentCoordinator, createNodeDispatchQueue } from './node/agent-coordinator.ts';
 export { InMemoryRunStore } from './node/run-store.ts';
@@ -95,7 +98,10 @@ export { SqliteEventStreamStore } from './runtime/event-stream-store.ts';
 export type { RunRecord, RunStatus, RunStore } from './runtime/run-store.ts';
 
 export { bashFactoryToSessionEnv } from './sandbox.ts';
-export type { DirectAgentSubmissionInput, DispatchAgentSubmissionInput } from './runtime/agent-submissions.ts';
+export type {
+	DirectAgentSubmissionInput,
+	DispatchAgentSubmissionInput,
+} from './runtime/agent-submissions.ts';
 export { InMemorySessionStore } from './session.ts';
 export { parseSkillMarkdown } from './skill-frontmatter.ts';