feat(examples): add alchemy-cloudflare chat session example

examples/alchemy-cloudflare/.gitignore

@@ -0,0 +1,2 @@
+.alchemy/
+.env

examples/alchemy-cloudflare/README.md

@@ -0,0 +1,35 @@
+# alchemy-cloudflare
+
+A minimal fold chat host on Cloudflare, deployed with [Alchemy v2](https://alchemy.run/cloudflare/). Inference runs on OpenAI (`gpt-5.6-terra`, medium reasoning) through fold's `openaiModel`. Each fold session is one Durable Object, named by its `SessionId`.
+
+- `src/ChatSession.ts` — the Durable Object. It opens the fold log in its SQLite on activation; the first message starts the session, later activations resume it.
+- `src/DurableObjectEventLog.ts` — the service that opens fold's `EventLogService` over the object's SQLite.
+- `src/Keepalive.ts` — a 30s alarm heartbeat that keeps the object alive while a turn runs.
+- `src/ChatSessions.ts` — the service the routes require; its `layer` reaches each session's Durable Object.
+- `src/Api.ts` — the `HttpRouter` routes, encoding responses with fold's log schemas.
+- `src/Worker.ts` — serves the routes, with `ChatSessions` as RPC to the Durable Objects.
+- `alchemy.run.ts` — the stack.
+
+## Run
+
+```sh
+echo 'OPENAI_API_KEY=sk-...' > .env   # read at deploy time, stored on the Worker as a secret
+bun alchemy profile edit --add Cloudflare
+bun run dev                       # local workerd
+bun run deploy                    # prints the Worker url
+```
+
+```sh
+ID=sess_...   # any new SessionId; the first message starts that session
+curl -X POST "$URL/sessions/$ID/messages" -d '{"text":"hi"}'                              # the agent-finished entry
+curl -X POST "$URL/sessions/$ID/messages" -d '{"text":"stop","whenRunning":"interrupt"}'  # queue (default) | steer | interrupt
+curl "$URL/sessions/$ID"                                                                  # all log entries
+```
+
+`bun run test` checks the routes against an in-memory `ChatSessions` built on fold's in-memory log.
+
+## Limits
+
+- A turn cut off by a crash or deploy continues when the object next wakes, through a synthetic "continue" user message (at most 3 per cut-off turn). A subagent cut off with it is left for the root model to resume by id.
+
+- Replies arrive whole. `FoldSession.events` can stream deltas once a route forwards them.

examples/alchemy-cloudflare/alchemy.run.ts

@@ -0,0 +1,14 @@
+import * as Alchemy from 'alchemy'
+import * as Cloudflare from 'alchemy/Cloudflare'
+import { Effect } from 'effect'
+
+import Worker from './src/Worker'
+
+export default Alchemy.Stack(
+	'FoldAlchemyCloudflare',
+	{ providers: Cloudflare.providers(), state: Cloudflare.state() },
+	Effect.gen(function* () {
+		const worker = yield* Worker
+		return { url: worker.url }
+	}),
+)

examples/alchemy-cloudflare/package.json

@@ -0,0 +1,28 @@
+{
+	"name": "@humanlayer/example-alchemy-cloudflare",
+	"version": "0.0.0",
+	"private": true,
+	"license": "MIT",
+	"type": "module",
+	"scripts": {
+		"dev": "alchemy dev",
+		"deploy": "alchemy deploy",
+		"destroy": "alchemy destroy",
+		"typecheck": "tsc --noEmit",
+		"test": "bun vitest run",
+		"test:watch": "bun vitest"
+	},
+	"dependencies": {
+		"@humanlayer/fold-core": "workspace:*",
+		"@effect/platform-bun": "catalog:",
+		"@effect/platform-node": "catalog:",
+		"alchemy": "catalog:",
+		"effect": "catalog:"
+	},
+	"devDependencies": {
+		"@effect/vitest": "catalog:",
+		"@humanlayer/fold-vitest-config": "workspace:*",
+		"typescript": "catalog:",
+		"vitest": "catalog:"
+	}
+}

examples/alchemy-cloudflare/src/Api.ts

@@ -0,0 +1,48 @@
+/**
+ * The chat HTTP routes over {@link ChatSessions}, encoding responses with fold's log schemas.
+ *
+ * - `POST /sessions/:sessionId/messages` with `{ "text": string, "whenRunning"?: "queue" | "steer" |
+ *   "interrupt" }` delivers one user message, starting the session if the id is new, and returns the
+ *   `agent-finished` entry of the run that consumed it. `whenRunning` defaults to `queue`.
+ * - `GET /sessions/:sessionId` returns the session's log entries.
+ */
+import { AgentFinishedLogEntry, LogEntry, SessionId } from '@humanlayer/fold-core'
+import { Effect, Schema } from 'effect'
+import { HttpRouter, HttpServerRequest, HttpServerResponse } from 'effect/unstable/http'
+
+import { ChatSessions, WhenRunning } from './ChatSessions'
+
+const SessionParams = Schema.Struct({ sessionId: SessionId })
+const MessageBody = Schema.Struct({
+	text: Schema.String,
+	whenRunning: WhenRunning.pipe(Schema.withDecodingDefaultKey(Effect.succeed('queue'))),
+})
+const entriesJson = HttpServerResponse.schemaJson(Schema.Array(LogEntry))
+const finishedJson = HttpServerResponse.schemaJson(AgentFinishedLogEntry)
+
+const badRequest = () => Effect.succeed(HttpServerResponse.text('Bad request', { status: 400 }))
+
+export const ChatRoutes = HttpRouter.use((router) =>
+	Effect.gen(function* () {
+		const sessions = yield* ChatSessions
+
+		yield* router.add(
+			'GET',
+			'/sessions/:sessionId',
+			Effect.gen(function* () {
+				const { sessionId } = yield* HttpRouter.schemaPathParams(SessionParams)
+				return yield* entriesJson(yield* sessions.entries(sessionId))
+			}).pipe(Effect.catchTag('SchemaError', badRequest)),
+		)
+
+		yield* router.add(
+			'POST',
+			'/sessions/:sessionId/messages',
+			Effect.gen(function* () {
+				const { sessionId } = yield* HttpRouter.schemaPathParams(SessionParams)
+				const { text, whenRunning } = yield* HttpServerRequest.schemaBodyJson(MessageBody)
+				return yield* finishedJson(yield* sessions.send(sessionId, text, whenRunning))
+			}).pipe(Effect.catchTag('SchemaError', badRequest)),
+		)
+	}),
+)

examples/alchemy-cloudflare/src/ChatSession.ts

@@ -0,0 +1,140 @@
+/**
+ * One fold session per Durable Object, named by its SessionId. The object's SQLite log is the session:
+ * addressing a new id creates the object with an empty log, and its first `send` starts the session
+ * under that id; a written log resumes on activation. Callers never create a session explicitly.
+ *
+ * Every turn runs under a {@link Keepalive} lease so the object is not evicted mid-run. A turn can still
+ * be cut off - a crash, a deploy - leaving the root's user message with no finished run after it. When a
+ * written log activates in that state, the object nudges the root to continue.
+ * Fold fills any tool call the cut left without a result, and the model can resume a cut-off subagent
+ * by id itself. A turn that keeps getting cut off stops being nudged after {@link MAX_RESTART_NUDGES}.
+ */
+import {
+	SessionId,
+	type AgentId,
+	type FoldSession,
+	type LogEntry,
+	type UserMessageLogEntry,
+	defineAgent,
+	eventLogSource,
+	openaiModel,
+	resumeSession,
+	startSession,
+} from '@humanlayer/fold-core'
+import * as Cloudflare from 'alchemy/Cloudflare'
+import { Config, Effect, Match, Option, Predicate, Result, Schema, Scope, Stream } from 'effect'
+
+import type { WhenRunning } from './ChatSessions'
+import { DurableObjectEventLog } from './DurableObjectEventLog'
+import { Keepalive } from './Keepalive'
+
+const MODEL = 'gpt-5.6-terra'
+
+const RESTART_NUDGE =
+	'<system-information>A restart cut you off before you finished. Continue where you left off.</system-information>'
+const MAX_RESTART_NUDGES = 3
+
+/** The root's user messages since its last finished run: any means a restart cut that run off. */
+const openRootMessages = (entries: ReadonlyArray<LogEntry>, rootAgentId: AgentId) => {
+	const lastFinishedSeq =
+		entries.findLast((entry) => Predicate.isTagged(entry, 'agent-finished') && entry.agentId === rootAgentId)
+			?.seq ?? -1
+	return entries.filter(
+		(entry): entry is UserMessageLogEntry =>
+			Predicate.isTagged(entry, 'user-message') && entry.agentId === rootAgentId && entry.seq > lastFinishedSeq,
+	)
+}
+
+const isRestartNudge = ({ message }: UserMessageLogEntry) =>
+	typeof message.content === 'string'
+		? message.content === RESTART_NUDGE
+		: message.content.some((part) => part.type === 'text' && part.text === RESTART_NUDGE)
+
+/** The root's next `agent-finished` at or after `fromSeq`: the end of the run a steered message joined. */
+const rootFinishedFrom = (session: FoldSession, fromSeq: number) =>
+	session.events(fromSeq).pipe(
+		Stream.filterMap((event) =>
+			event.kind === 'log' &&
+			Predicate.isTagged(event.entry, 'agent-finished') &&
+			event.entry.agentId === session.rootAgentId
+				? Result.succeed(event.entry)
+				: Result.fail(event),
+		),
+		Stream.runHead,
+		Effect.flatMap(Effect.fromOption),
+		Effect.orDie,
+	)
+
+/**
+ * Deliver one message to the root agent per {@link WhenRunning}. Idle roots just start a run. Only a
+ * subagent target can be missing, so fold's `SubagentNotFoundError` cannot happen here.
+ */
+const deliver = (session: FoldSession, text: string, whenRunning: WhenRunning) =>
+	Match.value(whenRunning).pipe(
+		Match.when('queue', () => session.send(text)),
+		Match.when('interrupt', () => Effect.andThen(session.interrupt(), session.send(text))),
+		Match.when('steer', () =>
+			Effect.gen(function* () {
+				const fromSeq = (yield* session.entries).length
+				yield* session.steer(text)
+				return yield* rootFinishedFrom(session, fromSeq)
+			}).pipe(Effect.catchTag('AgentNotRunningError', () => session.send(text))),
+		),
+		Match.exhaustive,
+		Effect.catchTag('SubagentNotFoundError', (error) => Effect.die(error)),
+	)
+
+export default class ChatSession extends Cloudflare.DurableObject<ChatSession>()(
+	'ChatSession',
+	Effect.gen(function* () {
+		const state = yield* Cloudflare.DurableObjectState
+		// Read from the deploy environment and bound to the Worker as a secret. Missing, the deploy fails.
+		const apiKey = yield* Config.redacted('OPENAI_API_KEY').pipe(Effect.orDie)
+		const eventLogs = yield* DurableObjectEventLog
+		const keepalive = yield* Keepalive
+
+		return Effect.gen(function* () {
+			const sessionId = yield* Schema.decodeUnknownEffect(SessionId)(state.id.name)
+			const eventLog = yield* eventLogs.open
+			const log = eventLogSource(Effect.succeed(eventLog))
+			const isEmpty = Option.isNone(yield* Stream.runHead(eventLog.entries()))
+
+			const agent = defineAgent({
+				name: 'alchemy-cloudflare-chat',
+				systemPrompt: 'You are a helpful, concise assistant.',
+				model: openaiModel({ apiKey, model: MODEL, reasoning: 'medium' }),
+			})
+
+			// Never closed: the session lives as long as the object stays in memory, past every call's own
+			// scope. Opened on the first send, so reading an unknown id writes nothing.
+			const scope = yield* Scope.make()
+			const session = yield* Effect.cached(
+				(isEmpty ? startSession({ agent, log, sessionId }) : resumeSession({ agent, log })).pipe(
+					Scope.provide(scope),
+				),
+			)
+
+			// A written log may hold a turn a restart cut off: nudge the root to continue it, in the background
+			// under a lease so the heartbeat keeps the object alive until it lands - even when that heartbeat's
+			// alarm is the only thing that woke the object.
+			if (!isEmpty) {
+				yield* Effect.forkIn(
+					Effect.gen(function* () {
+						const opened = yield* session
+						const open = openRootMessages(yield* opened.entries, opened.rootAgentId)
+						if (open.length === 0 || open.filter(isRestartNudge).length >= MAX_RESTART_NUDGES) return
+						yield* keepalive.whileRunning(deliver(opened, RESTART_NUDGE, 'queue'))
+					}),
+					scope,
+				)
+			}
+
+			return {
+				send: (text: string, whenRunning: WhenRunning) =>
+					keepalive.whileRunning(Effect.flatMap(session, (opened) => deliver(opened, text, whenRunning))),
+				entries: () => Stream.runCollect(eventLog.entries()).pipe(Effect.orDie),
+				alarm: () => keepalive.alarm,
+			}
+		}).pipe(Effect.orDie)
+	}).pipe(Effect.provide([DurableObjectEventLog.layer, Keepalive.layer])),
+) {}

examples/alchemy-cloudflare/src/ChatSessions.ts

@@ -0,0 +1,50 @@
+/**
+ * The fold chat sessions this host serves, each addressed by its SessionId. There is no create step:
+ * the first `send` to an id starts that session. The Worker implements this as RPC to the ChatSession
+ * Durable Object holding each session.
+ */
+import type { AgentFinishedLogEntry, LogEntry, SessionId } from '@humanlayer/fold-core'
+import { Context, Effect, Layer, Schema } from 'effect'
+
+import ChatSession from './ChatSession'
+
+/**
+ * What `send` does when the session's root agent is already running. When it is idle, every mode just
+ * starts a run.
+ *
+ * - `queue`: fold's follow-up - the message joins the running run when it would otherwise finish.
+ * - `steer`: fold's steering - the message lands between the running run's turns.
+ * - `interrupt`: interrupt the running run (fold writes its interrupted markers), then start a new one.
+ */
+export const WhenRunning = Schema.Literals(['queue', 'steer', 'interrupt'])
+export type WhenRunning = typeof WhenRunning.Type
+
+export class ChatSessions extends Context.Service<
+	ChatSessions,
+	{
+		/** Every durable log entry of the session so far; empty for an id nothing was sent to. */
+		readonly entries: (sessionId: SessionId) => Effect.Effect<ReadonlyArray<LogEntry>>
+		/**
+		 * Deliver one user message to the session's root agent, starting the session if it is new, and
+		 * resolve with the `agent-finished` entry of the run that consumed it.
+		 */
+		readonly send: (
+			sessionId: SessionId,
+			text: string,
+			whenRunning: WhenRunning,
+		) => Effect.Effect<AgentFinishedLogEntry>
+	}
+>()('alchemy-cloudflare/ChatSessions') {
+	/** Each session is the ChatSession Durable Object named by its id. Build it in a Worker's construction. */
+	static readonly layer = Layer.effect(
+		ChatSessions,
+		Effect.gen(function* () {
+			const sessions = yield* ChatSession
+
+			return ChatSessions.of({
+				entries: (sessionId) => sessions.getByName(sessionId).entries(),
+				send: (sessionId, text, whenRunning) => sessions.getByName(sessionId).send(text, whenRunning),
+			})
+		}),
+	)
+}