import { timingSafeEqual } from "node:crypto"; import { decodeBase64, encodeBase64 } from "@std/encoding/base64"; import { MAX_IMAGE_BYTES, MAX_IMAGES, MAX_MESSAGE_BYTES, validateImage, } from "@berry/api/images.ts"; import type { JsonValue } from "@earendil-works/chord"; import { BACKGROUND_CONTEXT as context, withAbortSignal } from "@earendil-works/chord/context"; import { type CredentialStore, getSupportedThinkingLevels, type Models, type Static, type TSchema, Type, } from "@earendil-works/pi-ai"; import { ConversationBusy, type EntryId, InboxDoc, type InboxItem, type SubmissionId, } from "@earendil-works/pi-durable"; import { Value } from "typebox/value"; import { type Berry, BerryError, Conversations, type Host, Settings } from "./berry.ts"; import { ProjectOrder, Projects } from "./projects.ts"; import { SystemPromptPatch } from "./prompts.ts"; import { createLogins } from "./logins.ts"; import type { Loader } from "./loader.ts"; import type { ExtensionWeb } from "./web.ts"; import { type Feed, servePubsub } from "./pubsub.ts"; import { serveChannel } from "./channel.ts"; import { TID_PATTERN } from "./tid.ts"; import { summaries } from "./summaries.ts"; import { conversationTree } from "./tree.ts"; import { references } from "./references.ts"; class HttpError extends Error { constructor(readonly status: number, message: string) { super(message); } } async function body( request: Request, schema: T, maxBytes?: number, ): Promise> { let value: unknown; try { if (maxBytes === undefined) value = await request.json(); else { const chunks: Uint8Array[] = []; let size = 0; const reader = request.body?.getReader(); if (!reader) throw new HttpError(400, "expected a json body"); try { while (true) { const { value: chunk, done } = await reader.read(); if (done) break; size += chunk.length; if (size > maxBytes) { await reader.cancel(); throw new HttpError(413, "Message exceeds the 30 MiB request limit"); } chunks.push(chunk); } } finally { reader.releaseLock(); } const bytes = new Uint8Array(size); let offset = 0; for (const chunk of chunks) { bytes.set(chunk, offset); offset += chunk.length; } value = JSON.parse(new TextDecoder().decode(bytes)); } } catch (error) { if (error instanceof HttpError) throw error; throw new HttpError(400, "expected a json body"); } const [error] = Value.Errors(schema, value); if (error) throw new HttpError(400, `${error.instancePath || "body"}: ${error.message}`); return value as Static; } const ModelRef = Type.Object({ provider: Type.String(), modelId: Type.String() }); const ThinkingLevel = Type.Union([ Type.Literal("off"), Type.Literal("minimal"), Type.Literal("low"), Type.Literal("medium"), Type.Literal("high"), Type.Literal("xhigh"), Type.Literal("max"), ]); const UserContent = Type.Union([ Type.String(), Type.Array(Type.Union([ Type.Object({ type: Type.Literal("text"), text: Type.String() }), Type.Object({ type: Type.Literal("image"), data: Type.String(), mimeType: Type.String() }), ])), ]); const Json = Type.Unsafe(Type.Unknown()); const CreateBody = Type.Object({ /** Each selected extension's configuration values. */ extensions: Type.Record(Type.String(), Type.Record(Type.String(), Json)), model: Type.Optional(ModelRef), thinkingLevel: Type.Optional(ThinkingLevel), }); const ReadBody = Type.Object({ entryId: Type.Integer({ minimum: 1 }) }); const ProjectBody = Type.Object({ name: Type.String({ minLength: 1, maxLength: 200 }), roots: Type.Array( Type.Object({ environment: Type.String({ minLength: 1 }), cwd: Type.String({ maxLength: 4096 }), }), { minItems: 1, maxItems: 100 }, ), defaultRoot: Type.Integer({ minimum: 0 }), moveConversations: Type.Optional(Type.Boolean()), }); const MessageBody = Type.Object({ content: UserContent, whenBusy: Type.Optional( Type.Union([Type.Literal("steer"), Type.Literal("followUp"), Type.Literal("reject")]), ), requestId: Type.Optional(Type.String()), }); const EventBody = Type.Object({ ext: Type.String(), event: Json }); const QueuedInputBody = Type.Object({ text: Type.String() }, { additionalProperties: false }); const ConfigurationBody = Type.Record(Type.String(), Json); /** Without an entry, the fork continues from the newest: a copy of the conversation as it is. */ const ForkBody = Type.Object({ entryId: Type.Optional(Type.Integer()) }); /** Where in any of its branches a conversation continues from. */ const BranchBody = Type.Object({ entryId: Type.Integer() }); const CompactBody = Type.Object({ instructions: Type.Optional(Type.String()) }); const ResetBody = Type.Object({ handoff: Type.Optional(Type.String()) }); const CommandBody = Type.Object({ args: Type.Optional(Type.String()) }); const RequestBody = Type.Object({ args: Type.Optional(Json) }); const CHANNEL = new URLPattern({ pathname: "/api/conversations/:tid/extensions/:extension/:name" }); const ShellBody = Type.Object({ command: Type.String(), excludeFromContext: Type.Optional(Type.Boolean()), }, { additionalProperties: false }); const SettingsBody = Type.Object({ allowLocalEnvironment: Type.Optional(Type.Boolean()), systemPrompt: Type.Optional(SystemPromptPatch), models: Type.Optional(Type.Array(ModelRef)), environments: Type.Optional( Type.Record( Type.String(), Type.Object({ ssh: Type.String(), deno: Type.Optional(Type.String()) }, { additionalProperties: false, }), ), ), }); const LoginBody = Type.Object({ type: Type.Union([Type.Literal("api_key"), Type.Literal("oauth")]), }); const AnswerBody = Type.Object({ prompt: Type.Integer(), value: Type.String() }); const AgentBody = Type.Object({ model: Type.Optional(Type.Union([ModelRef, Type.Null()])), thinkingLevel: Type.Optional(Type.Union([ThinkingLevel, Type.Null()])), extensions: Type.Optional(Type.Array(Type.String())), }); const ConversationBody = Type.Object({ archived: Type.Optional(Type.Boolean()), title: Type.Optional(Type.Union([Type.String(), Type.Null()])), }); type Handler = (params: Record, request: Request) => unknown; export type HandlerOptions = { readonly berry: Berry; readonly models: Models; /** Where `models` keeps logins: a provider is logged out by deleting its credential. */ readonly credentials: CredentialStore; /** The extensions' built frontends. */ readonly web: ExtensionWeb; /** Where the extensions come from, and how their loads went. */ readonly loader: Loader; /** The process, for restarting it; absent where nothing can. */ readonly host?: Host; readonly token: string; /** Answers everything outside `/api` and `/ws`, such as the built frontend. */ readonly fallback?: (request: Request) => Promise; }; /** How long a response must be before it is worth compressing. */ const COMPRESS_AT = 1024; export function createHandler({ berry, models, credentials, web, loader, host, token, fallback = () => Promise.resolve(new Response("not found", { status: 404 })), }: HandlerOptions) { const logins = createLogins(models); const expected = new TextEncoder().encode(token); const authorize = (candidate: string) => { const given = new TextEncoder().encode(candidate); return given.length === expected.length && timingSafeEqual(given, expected); }; async function conversation(tid: string) { const found = TID_PATTERN.test(tid) ? await berry.conversation(tid) : undefined; if (found === undefined) throw new HttpError(404, `no conversation ${tid}`); return found; } /** An entry of any of `tid`'s branches, or without one the newest of the branch it continues on. */ async function entryOf(tid: string, id?: number) { const head = await conversation(tid); if (id !== undefined) { const found = await berry.locate(tid, id); if (found === undefined) throw new HttpError(400, `no entry ${id} in ${tid}`); return { found, at: id as EntryId }; } const [entry] = (await head.entries({ order: "descending" }, 1, undefined, context)).items; if (entry === undefined) throw new HttpError(400, `${tid} has no entries`); return { found: head, at: entry.id }; } async function changeQueuedInput(tid: string, rawId: string, text: string | null) { const id = Number(rawId); if (!/^[1-9]\d*$/.test(rawId) || !Number.isSafeInteger(id)) { throw new HttpError(400, "expected a positive safe integer submission id"); } if (!TID_PATTERN.test(tid)) throw new HttpError(404, `no conversation ${tid}`); const result = await berry.changeQueuedInput(tid, id as SubmissionId, text); switch (result) { case "ok": return {}; case "not_found": throw new HttpError(404, "no input submission in this conversation"); case "empty": throw new HttpError(400, "A message needs text or an image attachment"); case "too_large": throw new HttpError(413, "Message exceeds the 30 MiB limit including attachments"); default: throw new HttpError(409, "Input is no longer queued"); } } // kept up from the first subscriber on let summarized: ReturnType | undefined; async function resolveTopic(topic: string): Promise { const [kind, tid, ext] = topic.split(":"); if (topic === "conversations") return await berry.harness.watchDoc(Conversations, context); if (topic === "projects") return await berry.harness.watchDoc(Projects, context); if (topic === "project-order") return await berry.harness.watchDoc(ProjectOrder, context); if (topic === "settings") return await berry.harness.watchDoc(Settings, context); if (topic === "extensions") return loader.feed(); if (topic === "summaries") return (await (summarized ??= summaries(berry))).feed(); if (kind === "conversation" && tid) { return await berry.watch(tid); } if (kind === "state" && tid && ext) return await berry.watchState(tid, ext); if (kind === "shell" && tid) return await berry.watchShell(tid); if (kind === "configuration" && tid && ext) return await berry.watchConfiguration(tid, ext); if (kind === "login" && tid) return logins.feed(tid); return undefined; } const table: [method: string, path: string, handler: Handler][] = [ ["GET", "/api/projects", async () => await berry.harness.snapshot(Projects, context)], ["POST", "/api/projects", async (_, request) => ({ id: await berry.saveProject(undefined, await body(request, ProjectBody)), })], ["PUT", "/api/project-order", async (_, request) => { const { ids } = await body(request, Type.Object({ ids: Type.Array(Type.String()) })); await berry.harness.commit(async (tx) => { (await tx.doc(ProjectOrder)).ids = ids; }, context); return {}; }], ["PUT", "/api/projects/:id", async ({ id }, request) => ({ id: await berry.saveProject(id, await body(request, ProjectBody)), })], ["DELETE", "/api/projects/:id", async ({ id }) => { await berry.deleteProject(id); return {}; }], ["GET", "/api/folders", async (_, request) => { const params = new URL(request.url).searchParams; const environment = params.get("environment") ?? "local"; const prefix = params.get("prefix") ?? ""; const cwd = params.get("cwd"); if (prefix.length > 4096 || (cwd?.length ?? 0) > 4096) { throw new HttpError(400, "Directory path is too long"); } return await berry.folders( environment, cwd, prefix, withAbortSignal(request.signal, context), ); }], // a new conversation's composer completes references before the conversation exists ["GET", "/api/references", async (_, request) => { const params = new URL(request.url).searchParams; const prefix = params.get("prefix") ?? ""; const cwd = params.get("cwd"); if (prefix.length > 4096 || (cwd?.length ?? 0) > 4096) { throw new HttpError(400, "Path is too long"); } return await berry.references( params.get("environment") ?? "local", cwd, prefix, withAbortSignal(request.signal, context), ); }], ["GET", "/api/conversations", async () => await berry.harness.snapshot(Conversations, context)], [ "POST", "/api/conversations", async (_, request) => ({ tid: await berry.create(await body(request, CreateBody)) }), ], ["POST", "/api/conversations/:tid/read", async ({ tid }, request) => { await conversation(tid); const { entryId } = await body(request, ReadBody); await berry.markRead(tid, entryId); return {}; }], ["PATCH", "/api/conversations/:tid/inbox/:id", async ({ tid, id }, request) => { const { text } = await body(request, QueuedInputBody, MAX_MESSAGE_BYTES); return await changeQueuedInput(tid, id, text); }], [ "DELETE", "/api/conversations/:tid/inbox/:id", async ({ tid, id }) => await changeQueuedInput(tid, id, null), ], ["POST", "/api/conversations/:tid/messages", async ({ tid }, request) => { const { content, whenBusy, requestId } = await body(request, MessageBody, MAX_MESSAGE_BYTES); const current = await conversation(tid); const images = typeof content === "string" ? [] : content.filter((block) => block.type === "image"); if (images.length > MAX_IMAGES) { throw new HttpError(400, `A message can have at most ${MAX_IMAGES} images`); } for (const image of images) { if (image.data.length > Math.ceil(MAX_IMAGE_BYTES / 3) * 4) { throw new HttpError(413, "Images must be no larger than 5 MiB"); } try { const bytes = decodeBase64(image.data); if (bytes.length > MAX_IMAGE_BYTES) { throw new HttpError(413, "Images must be no larger than 5 MiB"); } if (encodeBase64(bytes) !== image.data) { throw new Error("Expected canonical base64 image data"); } validateImage(bytes, image.mimeType); } catch (error) { if (error instanceof HttpError) throw error; throw new HttpError(400, error instanceof Error ? error.message : "Invalid image"); } } if (images.length > 0) { const { model: ref } = await current.agent(context); const model = ref && models.getModel(ref.provider, ref.modelId); if (!model?.input.includes("image")) { throw new HttpError( 400, "Choose a model that supports image input before sending images", ); } } const submission = await current.submit({ type: "input", content, ...(whenBusy ? { whenBusy } : {}), ...(requestId ? { requestId } : {}), }, context); return { submission: submission.id }; }], ["POST", "/api/conversations/:tid/events", async ({ tid }, request) => { const { ext, event } = await body(request, EventBody); return { submission: (await berry.emit(tid, ext, event)).id }; }], ["POST", "/api/conversations/:tid/shell", async ({ tid }, request) => { await conversation(tid); const { command, excludeFromContext } = await body(request, ShellBody); return { task: await berry.shell(tid, command, excludeFromContext) }; }], ["POST", "/api/conversations/:tid/abort", async ({ tid }) => { const current = await conversation(tid); let items: readonly InboxItem[] | undefined; const inputs: InboxItem[] = []; const stop = berry.harness.subscribeCommits(({ changes }) => { const withdrawn = new Set(changes.flatMap((change) => { if (change.type !== "submission") return []; const record = change.value; return record.conversationId === current.id && record.type === "input" && record.status === "unanswered" && record.reason === "aborted" && record.entry === undefined ? [record.id] : []; })); // use the revision before withdrawal, including edits and admissions that raced the request for (const item of items ?? []) { if (item.mode !== "write" && withdrawn.has(item.id)) inputs.push(item); } for (const change of changes) { if ( change.type === "document" && change.conversationId === current.id && change.record.kind === InboxDoc.definition.kind ) items = (change.value?.items ?? []) as InboxItem[]; } }); try { const initial = await berry.harness.snapshot(InboxDoc, current.id, context); items ??= initial?.items ?? []; await current.abort(context); return { inputs }; } finally { stop(); } }], ["POST", "/api/conversations/:tid/compact", async ({ tid }, request) => { const { instructions } = await body(request, CompactBody); return { task: await (await conversation(tid)).compact(instructions, context) }; }], ["POST", "/api/conversations/:tid/reset", async ({ tid }, request) => { const { handoff } = await body(request, ResetBody); await (await conversation(tid)).reset(handoff, context); return {}; }], ["POST", "/api/conversations/:tid/fork", async ({ tid }, request) => { const { entryId } = await body(request, ForkBody); const { found, at } = await entryOf(tid, entryId); const fork = await found.fork(at, { ownership: { kind: "ownerless" } }, context); return { tid: await berry.tidOf(fork.id) }; }], ["POST", "/api/conversations/:tid/branch", async ({ tid }, request) => { const { entryId } = await body(request, BranchBody); await conversation(tid); await berry.branch(tid, entryId); return {}; }], ["POST", "/api/conversations/:tid/regenerate", async ({ tid }, request) => { const { entryId } = await body(request, BranchBody); await conversation(tid); await berry.regenerate(tid, entryId); return {}; }], ["GET", "/api/conversations/:tid/tree", async ({ tid }) => { await conversation(tid); return await conversationTree(berry, tid); }], // what continuing from `at` would start with ["GET", "/api/conversations/:tid/context", async ({ tid }, request) => { const at = Number(new URL(request.url).searchParams.get("at")); if (!Number.isInteger(at) || at <= 0) throw new HttpError(400, "expected ?at="); const { found } = await entryOf(tid, at); const view = await found.context(context, { at: at as EntryId }); return { entries: view.entries }; }], ["GET", "/api/conversations/:tid/references", async ({ tid }, request) => { const params = new URL(request.url).searchParams; const rawAt = params.get("at"); const at = rawAt === null ? undefined : Number(rawAt); if (at !== undefined && (!Number.isInteger(at) || at <= 0)) { throw new HttpError(400, "expected ?at="); } const target = at === undefined ? await conversation(tid) : (await entryOf(tid, at)).found; const agent = await target.agent(context); if (!agent.extensions.some(({ name }) => name === "code")) return []; const env = await berry.environment(tid, at as EntryId | undefined); return await references(env, params.get("prefix") ?? "", context); }], ["PATCH", "/api/conversations/:tid/configuration/:name", async ({ tid, name }, request) => { await conversation(tid); return await berry.configureExtension(tid, name, await body(request, ConfigurationBody)); }], ["GET", "/api/conversations/:tid/commands", async ({ tid }) => { await conversation(tid); return await berry.commands(tid); }], ["POST", "/api/conversations/:tid/commands/:name", async ({ tid, name }, request) => { await conversation(tid); const { args } = await body(request, CommandBody); return { result: await berry.command(tid, name, args ?? "") ?? null }; }], [ "POST", "/api/conversations/:tid/extensions/:extension/:name", async ({ tid, extension, name }, request) => { await conversation(tid); const { args } = await body(request, RequestBody); // the browser gives up on a request by closing it, which ends the handler's work too return { result: await berry.request(tid, extension, name, args, request.signal) }; }, ], ["PATCH", "/api/conversations/:tid", async ({ tid }, request) => { await conversation(tid); const { archived, title } = await body(request, ConversationBody); if (archived !== undefined) await berry.archive(tid, archived); if (title !== undefined) await berry.rename(tid, title); return {}; }], ["PATCH", "/api/conversations/:tid/agent", async ({ tid }, request) => { await conversation(tid); await berry.configure(tid, await body(request, AgentBody)); return {}; }], ["GET", "/api/providers", async () => { const stored = new Set((await credentials.list()).map((credential) => credential.providerId)); return await Promise.all( models.getProviders().map(async ({ id, name, auth }) => ({ id, name, apiKey: auth.apiKey?.login ? { name: auth.apiKey.name } : null, oauth: auth.oauth ? { name: auth.oauth.name, label: auth.oauth.loginLabel ?? null, subscription: auth.oauth.isSubscription ?? false, } : null, // how it is configured, if it is: a stored login, or something ambient such as an environment variable auth: await models.checkAuth(id).catch(() => undefined) ?? null, stored: stored.has(id), })), ); }], ["POST", "/api/providers/:id/login", async ({ id }, request) => { const { type } = await body(request, LoginBody); return { login: logins.start(id, type) }; }], ["POST", "/api/providers/:id/logout", async ({ id }) => { await models.logout(id); return {}; }], ["POST", "/api/logins/:id/answer", async ({ id }, request) => { const { prompt, value } = await body(request, AnswerBody); logins.answer(id, prompt, value); return {}; }], ["POST", "/api/logins/:id/cancel", ({ id }) => { logins.cancel(id); return {}; }], ["GET", "/api/settings", async () => await berry.harness.snapshot(Settings, context)], ["PATCH", "/api/settings", async (_, request) => { const { models, environments, allowLocalEnvironment, systemPrompt } = await body( request, SettingsBody, ); if (allowLocalEnvironment !== undefined) { await berry.setAllowLocalEnvironment(allowLocalEnvironment); } if (models) await berry.enableModels(models); if (environments) await berry.setEnvironments(environments); if (systemPrompt) await berry.setSystemPrompt(systemPrompt); return await berry.harness.snapshot(Settings, context); }], ["GET", "/api/extensions", () => loader.entries()], ["POST", "/api/extensions/:name/reload", async ({ name }) => await loader.reload(name)], ["POST", "/api/restart", async () => { if (host === undefined) throw new HttpError(404, "this berry cannot restart"); const failures = await host.check(); if (failures !== null) throw new HttpError(400, failures); // after this response is on its way setTimeout(() => host.restart(), 100); return {}; }], ["GET", "/api/models", async (_, request) => { const refreshed = await models.refresh({ signal: request.signal }); for (const [id, error] of refreshed.errors) { console.error(`[berry] refreshing ${id}:`, error); } return (await models.getAvailable(undefined, { signal: request.signal })).map((model) => ({ provider: model.provider, id: model.id, name: model.name, input: model.input, reasoning: model.reasoning, thinkingLevels: getSupportedThinkingLevels(model), contextWindow: model.contextWindow, cost: model.cost, })); }], ]; const routes = table.map(([method, path, handler]) => [method, new URLPattern({ pathname: path }), handler] as const ); const sockets = new Set(); /** Close every socket as a restart closes it; open sockets would hold a graceful shutdown. */ const closeSockets = () => { for (const socket of sockets) socket.close(1012, "restarting"); }; return Object.assign(async (request: Request): Promise => { const url = new URL(request.url); if (url.pathname === "/ws") { if (request.headers.get("upgrade") !== "websocket") { return new Response("expected a websocket", { status: 426 }); } const { socket, response } = Deno.upgradeWebSocket(request); sockets.add(socket); socket.addEventListener("close", () => sockets.delete(socket)); servePubsub(socket, { authorize, resolve: resolveTopic }); return response; } // frontend code, as public as the app's bundle; a dynamic import() could not send the token anyway if (url.pathname.startsWith("/ext/")) { return web.serve(url.pathname) ?? new Response("not found", { status: 404 }); } if (!url.pathname.startsWith("/api/")) return await fallback(request); // a browser cannot send a websocket its token as a header, so a channel authorizes with its first message const channel = CHANNEL.exec(url)?.pathname.groups; if (channel && request.headers.get("upgrade") === "websocket") { const { tid, extension, name } = channel as Record; const { socket, response } = Deno.upgradeWebSocket(request); sockets.add(socket); socket.addEventListener("close", () => sockets.delete(socket)); serveChannel(socket, { authorize, open: (args, opened, signal) => berry.channel(tid, extension, name, args, opened, signal), }); return response; } const bearer = request.headers.get("authorization")?.match(/^Bearer (.+)$/)?.[1]; if (bearer === undefined || !authorize(bearer)) { return Response.json({ error: "unauthorized" }, { status: 401 }); } for (const [method, pattern, handler] of routes) { const match = pattern.exec(url); if (match === null) continue; if (request.method !== method) continue; try { const body = JSON.stringify( await handler(match.pathname.groups as Record, request) ?? null, ); // a diff or a long history is mostly text, which a remote reader waits for byte by byte if ( body.length < COMPRESS_AT || !/\bgzip\b/.test(request.headers.get("accept-encoding") ?? "") ) { return new Response(body, { headers: { "content-type": "application/json" } }); } return new Response(new Blob([body]).stream().pipeThrough(new CompressionStream("gzip")), { headers: { "content-type": "application/json", "content-encoding": "gzip", vary: "accept-encoding", }, }); } catch (error) { if (error instanceof HttpError) { return Response.json({ error: error.message }, { status: error.status }); } if (error instanceof BerryError) { return Response.json({ error: error.message }, { status: 400 }); } if (error instanceof ConversationBusy) { return Response.json({ error: error.message }, { status: 409 }); } console.error(`[http] ${request.method} ${url.pathname}:`, error); return Response.json({ error: "internal error" }, { status: 500 }); } } return Response.json({ error: "not found" }, { status: 404 }); }, { closeSockets }); }