snapshot of https://github.com/tldraw/tldraw.git (main, 2026-09-23; blobs > 131072B omitted)

templates/agent/worker/do/AgentDurableObject.ts

@@ -0,0 +1,80 @@
+import { DurableObject } from 'cloudflare:workers'
+import { AutoRouter, error } from 'itty-router'
+import { AgentAction } from '../../shared/types/AgentAction'
+import { AgentPrompt } from '../../shared/types/AgentPrompt'
+import { Streaming } from '../../shared/types/Streaming'
+import { Environment } from '../environment'
+import { AgentService } from './AgentService'
+
+export class AgentDurableObject extends DurableObject<Environment> {
+	service: AgentService
+
+	constructor(ctx: DurableObjectState, env: Environment) {
+		super(ctx, env)
+		this.service = new AgentService(this.env) // swap this with your own service
+	}
+
+	private readonly router = AutoRouter({
+		catch: (e) => {
+			console.error(e)
+			return error(e)
+		},
+	}).post('/stream', (request) => this.stream(request))
+
+	// `fetch` is the entry point for all requests to the Durable Object
+	override fetch(request: Request): Response | Promise<Response> {
+		return this.router.fetch(request)
+	}
+
+	/**
+	 * Stream changes from the model.
+	 *
+	 * @param request - The request object containing the prompt.
+	 * @returns A Promise that resolves to a Response object containing the streamed changes.
+	 */
+	private async stream(request: Request): Promise<Response> {
+		const encoder = new TextEncoder()
+		const { readable, writable } = new TransformStream()
+		const writer = writable.getWriter()
+
+		const response: { changes: Streaming<AgentAction>[] } = { changes: [] }
+
+		;(async () => {
+			try {
+				const prompt = (await request.json()) as AgentPrompt
+
+				for await (const change of this.service.stream(prompt)) {
+					response.changes.push(change)
+					const data = `data: ${JSON.stringify(change)}\n\n`
+					await writer.write(encoder.encode(data))
+					await writer.ready
+				}
+				await writer.close()
+			} catch (error: any) {
+				console.error('Stream error:', error)
+
+				// Send error through the stream
+				const errorData = `data: ${JSON.stringify({ error: error.message })}\n\n`
+				try {
+					await writer.write(encoder.encode(errorData))
+					await writer.close()
+				} catch (writeError) {
+					await writer.abort(writeError)
+				}
+			}
+		})()
+
+		return new Response(readable, {
+			headers: {
+				'Content-Type': 'text/event-stream',
+				'Cache-Control': 'no-cache, no-transform',
+				Connection: 'keep-alive',
+				'X-Accel-Buffering': 'no',
+				'Transfer-Encoding': 'chunked',
+				'Access-Control-Allow-Origin': '*',
+				'Access-Control-Allow-Methods': 'POST, OPTIONS',
+				'Access-Control-Allow-Headers': 'Content-Type',
+			},
+		})
+	}
+}

templates/agent/worker/do/AgentService.ts

@@ -0,0 +1,228 @@
+import { AnthropicProvider, AnthropicProviderOptions, createAnthropic } from '@ai-sdk/anthropic'
+import {
+	createGoogleGenerativeAI,
+	GoogleGenerativeAIProvider,
+	GoogleGenerativeAIProviderOptions,
+} from '@ai-sdk/google'
+import { createOpenAI, OpenAIProvider, OpenAIResponsesProviderOptions } from '@ai-sdk/openai'
+import { LanguageModel, ModelMessage, streamText } from 'ai'
+import {
+	AgentModelDefinition,
+	AgentModelName,
+	getAgentModelDefinition,
+	isValidModelName,
+} from '../../shared/models'
+import { DebugPart } from '../../shared/schema/PromptPartDefinitions'
+import { AgentAction } from '../../shared/types/AgentAction'
+import { AgentPrompt } from '../../shared/types/AgentPrompt'
+import { Streaming } from '../../shared/types/Streaming'
+import { Environment } from '../environment'
+import { buildMessages } from '../prompt/buildMessages'
+import { buildSystemPrompt } from '../prompt/buildSystemPrompt'
+import { getModelName } from '../prompt/getModelName'
+import { closeAndParseJson } from './closeAndParseJson'
+
+export class AgentService {
+	openai: OpenAIProvider
+	anthropic: AnthropicProvider
+	google: GoogleGenerativeAIProvider
+
+	constructor(env: Environment) {
+		this.openai = createOpenAI({ apiKey: env.OPENAI_API_KEY })
+		this.anthropic = createAnthropic({ apiKey: env.ANTHROPIC_API_KEY })
+		this.google = createGoogleGenerativeAI({ apiKey: env.GOOGLE_API_KEY })
+	}
+
+	getModel(modelName: AgentModelName): LanguageModel {
+		const modelDefinition = getAgentModelDefinition(modelName)
+		const provider = modelDefinition.provider
+		return this[provider](modelDefinition.id)
+	}
+
+	async *stream(prompt: AgentPrompt): AsyncGenerator<Streaming<AgentAction>> {
+		try {
+			for await (const event of this.streamActions(prompt)) {
+				yield event
+			}
+		} catch (error: any) {
+			console.error('Stream error:', error)
+			throw error
+		}
+	}
+
+	private async *streamActions(prompt: AgentPrompt): AsyncGenerator<Streaming<AgentAction>> {
+		const modelName = getModelName(prompt)
+		const model = this.getModel(modelName)
+
+		if (typeof model === 'string') {
+			throw new Error('Model is a string, not a LanguageModel')
+		}
+
+		const { modelId, provider } = model
+		if (!isValidModelName(modelId)) {
+			throw new Error(`Model ${modelId} is not in AGENT_MODEL_DEFINITIONS`)
+		}
+
+		const modelDefinition = getAgentModelDefinition(modelId)
+		const systemPrompt = buildSystemPrompt(prompt)
+
+		// Build messages with provider-specific options
+		const messages: ModelMessage[] = []
+
+		// Add system prompt with Anthropic caching if applicable
+		if (provider === 'anthropic.messages') {
+			// Anthropic requires explicit cache breakpoints. We set one at the end of the
+			// system prompt to cache all system content (which generally changes together).
+			messages.push({
+				role: 'system',
+				content: systemPrompt,
+				providerOptions: {
+					anthropic: { cacheControl: { type: 'ephemeral' } },
+				},
+			})
+		} else {
+			messages.push({
+				role: 'system',
+				content: systemPrompt,
+			})
+		}
+
+		// Add prompt messages
+		const promptMessages = buildMessages(prompt)
+		messages.push(...promptMessages)
+
+		// Check for debug flags and log if enabled
+		const debugPart = prompt.debug as DebugPart | undefined
+		if (debugPart) {
+			if (debugPart.logSystemPrompt) {
+				const promptWithoutSchema = buildSystemPrompt(prompt, { withSchema: false })
+				console.log('[DEBUG] System Prompt (without schema):\n', promptWithoutSchema)
+			}
+			if (debugPart.logMessages) {
+				console.log('[DEBUG] Messages:\n', JSON.stringify(promptMessages, null, 2))
+			}
+		}
+
+		// Prefill the assistant turn to force the JSON start, where the model allows it.
+		// Opus 4.7+ and Sonnet 4.6 reject last-assistant-turn prefills (400), so skip it there.
+		if (modelDefinition.supportsPrefill) {
+			messages.push({
+				role: 'assistant',
+				content: '{"actions": [{"_type":',
+			})
+		}
+
+		try {
+			const { textStream } = streamText({
+				model,
+				messages,
+				maxOutputTokens: 8192,
+				// Opus 4.7+ removed `temperature` (and top_p/top_k); sending it returns a 400.
+				...(modelDefinition.supportsTemperature ? { temperature: 0 } : {}),
+				providerOptions: getProviderOptions(modelDefinition),
+				onAbort() {
+					console.warn('Stream actions aborted')
+				},
+				onError: (e) => {
+					console.error('Stream text error:', e)
+					throw e
+				},
+			})
+
+			const canForceResponseStart =
+				(provider === 'anthropic.messages' || provider === 'google.generative-ai') &&
+				modelDefinition.supportsPrefill
+			let buffer = canForceResponseStart ? '{"actions": [{"_type":' : ''
+			let cursor = 0
+			let maybeIncompleteAction: AgentAction | null = null
+
+			let startTime = Date.now()
+			for await (const text of textStream) {
+				buffer += text
+				const partialObject = closeAndParseJson(buffer)
+				if (!partialObject) continue
+
+				const actions = partialObject.actions
+				if (!Array.isArray(actions)) continue
+				if (actions.length === 0) continue
+
+				// If the events list is ahead of the cursor, we know we've completed the current event
+				// We can complete the event and move the cursor forward
+				if (actions.length > cursor) {
+					const action = actions[cursor - 1] as AgentAction
+					if (action) {
+						yield {
+							...action,
+							complete: true,
+							time: Date.now() - startTime,
+						}
+						maybeIncompleteAction = null
+					}
+					cursor++
+				}
+
+				// Now let's check the (potentially new) current event
+				// And let's yield it in its (potentially incomplete) state
+				const action = actions[cursor - 1] as AgentAction
+				if (action) {
+					// If we don't have an incomplete event yet, this is the start of a new one
+					if (!maybeIncompleteAction) {
+						startTime = Date.now()
+					}
+
+					maybeIncompleteAction = action
+
+					// Yield the potentially incomplete event
+					yield {
+						...action,
+						complete: false,
+						time: Date.now() - startTime,
+					}
+				}
+			}
+
+			// If we've finished receiving events, but there's still an incomplete event, we need to complete it
+			if (maybeIncompleteAction) {
+				yield {
+					...maybeIncompleteAction,
+					complete: true,
+					time: Date.now() - startTime,
+				}
+			}
+		} catch (error: any) {
+			console.error('streamActions error:', error)
+			throw error
+		}
+	}
+}
+
+type StreamTextProviderOptions = NonNullable<Parameters<typeof streamText>[0]['providerOptions']>
+
+/**
+ * Map a model definition's reasoning preferences to AI SDK provider options.
+ * Only the matching provider's options are set; the SDK ignores the rest.
+ */
+function getProviderOptions(definition: AgentModelDefinition): StreamTextProviderOptions {
+	switch (definition.provider) {
+		case 'anthropic':
+			return {
+				anthropic: {
+					thinking:
+						definition.thinking === 'adaptive' ? { type: 'adaptive' } : { type: 'disabled' },
+					...(definition.effort ? { effort: definition.effort } : {}),
+				} satisfies AnthropicProviderOptions,
+			}
+		case 'google':
+			return {
+				google: {
+					thinkingConfig: { thinkingLevel: definition.thinkingLevel },
+				} satisfies GoogleGenerativeAIProviderOptions,
+			}
+		case 'openai':
+			return {
+				openai: {
+					reasoningEffort: definition.reasoningEffort,
+				} satisfies OpenAIResponsesProviderOptions,
+			}
+	}
+}

templates/agent/worker/do/closeAndParseJson.ts

@@ -0,0 +1,72 @@
+/**
+ * JSON helper. Given a potentially incomplete JSON string, return the parsed object.
+ * The string might be missing closing braces, brackets, or other characters like quotation marks.
+ * @param string - The string to parse.
+ * @returns The parsed object.
+ */
+export function closeAndParseJson(string: string) {
+	const stackOfOpenings = []
+
+	// Track openings and closings
+	let i = 0
+	while (i < string.length) {
+		const char = string[i]
+		const lastOpening = stackOfOpenings.at(-1)
+
+		if (char === '"') {
+			// Check if this quote is escaped
+			if (i > 0 && string[i - 1] === '\\') {
+				// This is an escaped quote, skip it
+				i++
+				continue
+			}
+
+			if (lastOpening === '"') {
+				stackOfOpenings.pop()
+			} else {
+				stackOfOpenings.push('"')
+			}
+		}
+
+		if (lastOpening === '"') {
+			i++
+			continue
+		}
+
+		if (char === '{' || char === '[') {
+			stackOfOpenings.push(char)
+		}
+
+		if (char === '}' && lastOpening === '{') {
+			stackOfOpenings.pop()
+		}
+
+		if (char === ']' && lastOpening === '[') {
+			stackOfOpenings.pop()
+		}
+
+		i++
+	}
+
+	// Now close all unclosed openings
+	for (let i = stackOfOpenings.length - 1; i >= 0; i--) {
+		const opening = stackOfOpenings[i]
+		if (opening === '{') {
+			string += '}'
+		}
+
+		if (opening === '[') {
+			string += ']'
+		}
+
+		if (opening === '"') {
+			string += '"'
+		}
+	}
+
+	try {
+		return JSON.parse(string)
+	} catch (_e) {
+		return null
+	}
+}