char-slop/berry

git clone https://git.t4t.associates/char-slop/berry

Charlotte Somallow project reorderingb4fc6cf2026-10-29

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<T extends TSchema>(
  request: Request,
  schema: T,
  maxBytes?: number,
): Promise<Static<T>> {
  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<T>;
}

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<JsonValue>(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<string, string>, 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<Response>;
};

/** 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<typeof summaries> | undefined;

  async function resolveTopic(topic: string): Promise<Feed | undefined> {
    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=<entry id>");
      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=<entry id>");
      }
      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<WebSocket>();
  /** 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<Response> => {
    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<string, string>;
      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<string, string>, 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 });
}