char-slop/berry

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

Charlotte Somallow project reorderingb4fc6cf2026-10-29

import { assert, assertEquals, assertStringIncludes } from "@std/assert";
import { encodeBase64 } from "@std/encoding/base64";
import { applyImmutable, type Op } from "@earendil-works/chord/delta";
import { BACKGROUND_CONTEXT as context } from "@earendil-works/chord/context";
import { type ImageContent, type Message, Type } from "@earendil-works/pi-ai";
import { createModels } from "@earendil-works/pi-ai/models";
import {
  fauxAssistantMessage,
  fauxProvider,
  fauxToolCall,
} from "@earendil-works/pi-ai/providers/faux";
import { InMemoryCredentialStore } from "@earendil-works/pi-ai";
import { InboxDoc, LiveDoc, MemoryStorage } from "@earendil-works/pi-durable";
import { MAX_MESSAGE_BYTES } from "@berry/api/images.ts";
import { PNG, tempDir, until } from "@berry/util";
import { defineExtension } from "@berry/api";
import { type Host, openBerry } from "./berry.ts";
import { createHandler } from "./http.ts";
import { extensionLoader } from "./loader.ts";
import { extensionWeb } from "./web.ts";
import type { ServerMessage } from "./pubsub.ts";
import { blocking, gated } from "./fixture.ts";

const TOKEN = "test-token";

const dir = defineExtension<{ cwd: string }, { cwd: string }>({
  name: "dir",
  initial: () => ({ cwd: "/" }),
  apply: (_, event) => ({ cwd: event.cwd }),
  setup(ext) {
    ext.configuration({
      parameters: Type.Object({ cwd: Type.String({ default: "/" }) }),
      values: ({ cwd }) => ({ cwd }),
      events: ({ cwd }) => {
        if (!cwd.startsWith("/")) throw new Error(`not absolute: ${cwd}`);
        return [{ cwd }];
      },
    });
    ext.tool({
      name: "noop",
      description: "nothing",
      parameters: Type.Object({}),
      execute: () => undefined,
    });
    ext.channel({
      name: "echo",
      parameters: Type.Object({ prefix: Type.String({ default: "" }) }),
      open({ prefix }, channel, ctx) {
        if (prefix === "refuse") throw new Error("not today");
        void (async () => {
          for await (const message of channel.messages) {
            if (message === "bye") channel.close();
            else if (message instanceof Uint8Array) channel.send(message.reverse());
            else channel.send({ echo: message, prefix, cwd: ctx.state.cwd });
          }
        })();
      },
    });
    ext.handle({
      name: "where",
      parameters: Type.Object({ under: Type.String({ default: "" }) }),
      execute: ({ under }, ctx) => ctx.state.cwd + under,
    });
    return (conv) =>
      conv.command({
        name: "cd",
        description: "change directory",
        async execute(path, ctx) {
          if (!path.startsWith("/")) throw new Error(`not absolute: ${path}`);
          await ctx.emit({ cwd: path });
          return path;
        },
      });
  },
});

async function serve(host?: Host, supportsImages = true) {
  const faux = fauxProvider(
    supportsImages ? undefined : { models: [{ id: "text", input: ["text"] }] },
  );
  const credentials = new InMemoryCredentialStore();
  const models = createModels({ credentials });
  models.setProvider(faux.provider);
  // a provider whose login asks for a key, as most do
  models.setProvider({
    ...faux.provider,
    id: "keyed",
    name: "Keyed",
    auth: {
      apiKey: {
        name: "Keyed API key",
        login: async (interaction) => {
          const key = await interaction.prompt({ type: "secret", message: "your key?" });
          interaction.notify({ type: "info", message: "thanks" });
          return { type: "api_key", key };
        },
        resolve: ({ credential }) =>
          Promise.resolve(
            credential?.key
              ? { auth: { apiKey: credential.key }, source: "stored credential" }
              : undefined,
          ),
      },
    },
  });
  const berry = await openBerry(new MemoryStorage(), models);
  await berry.install(dir);
  await berry.enableModels([{ provider: "faux", modelId: faux.getModel().id }]);
  const web = extensionWeb();
  const server = Deno.serve(
    { port: 0, onListen() {} },
    createHandler({
      berry,
      models,
      credentials,
      web,
      loader: extensionLoader({ berry, web, root: await Deno.makeTempDir() }),
      host,
      token: TOKEN,
    }),
  );
  const base = `http://127.0.0.1:${server.addr.port}`;
  const api = async (method: string, path: string, body?: unknown) => {
    const response = await fetch(base + path, {
      method,
      headers: { authorization: `Bearer ${TOKEN}`, "content-type": "application/json" },
      ...(body === undefined ? {} : { body: JSON.stringify(body) }),
    });
    return { status: response.status, json: await response.json() };
  };
  return {
    berry,
    faux,
    base,
    api,
    model: { provider: "faux", modelId: faux.getModel().id },
    async close() {
      await server.shutdown();
      await berry.close();
    },
    [Symbol.asyncDispose]() {
      return this.close();
    },
  };
}

Deno.test("the model list refreshes extension catalogs before returning available models", async () => {
  await using server = await serve();
  const faux = fauxProvider();
  let source = [{ ...faux.getModel(), provider: "dynamic", id: "first" }];
  let catalog = source;
  await server.berry.install(defineExtension({
    name: "provider",
    initial: () => null,
    setup(ext) {
      ext.provider({
        ...faux.provider,
        id: "dynamic",
        getModels: () => catalog,
        async refreshModels(ctx) {
          if (ctx.allowNetwork) {
            await ctx.publish({ update: () => void (catalog = source) });
          }
        },
      });
    },
  }));
  source = [{ ...source[0], id: "discovered" }];
  const result = await server.api("GET", "/api/models");
  assertEquals(result.status, 200);
  assertEquals(
    (result.json as { provider: string; id: string }[]).filter((model) =>
      model.provider === "dynamic"
    )
      .map((model) => model.id),
    ["discovered"],
  );
});

Deno.test("settings accept partial prompt edits and reject invalid fields without saving them", async () => {
  await using server = await serve();
  const original = (await server.api("GET", "/api/settings")).json;
  const saved = await server.api("PATCH", "/api/settings", {
    systemPrompt: { userName: "Jordan", generalGuidelines: "" },
  });
  assertEquals(saved.status, 200);
  assertEquals(saved.json.systemPrompt, {
    ...original.systemPrompt,
    userName: "Jordan",
    generalGuidelines: "",
    promptProse: {},
  });
  assertEquals(saved.json.models, original.models);
  const prose = await server.api("PATCH", "/api/settings", {
    systemPrompt: { promptProse: { code: "Keep it small.", other: "" } },
  });
  assertEquals(prose.status, 200);
  assertEquals(prose.json.systemPrompt.promptProse, { code: "Keep it small.", other: "" });
  assertEquals(prose.json.systemPrompt.userName, "Jordan");
  const kept = await server.api("PATCH", "/api/settings", {
    systemPrompt: { promptProse: { other: "Changed." } },
  });
  assertEquals(kept.json.systemPrompt.promptProse, { code: "Keep it small.", other: "Changed." });
  const invalid = await server.api("PATCH", "/api/settings", {
    systemPrompt: { promptProse: { code: ["not text"] } },
  });
  assertEquals(invalid.status, 400);
  assertEquals((await server.api("GET", "/api/settings")).json, kept.json);
});

Deno.test("settings can disable and re-enable local without replacing other settings", async () => {
  await using server = await serve();
  const original = (await server.api("GET", "/api/settings")).json;
  const saved = await server.api("PATCH", "/api/settings", { allowLocalEnvironment: false });
  assertEquals(saved.status, 200);
  assertEquals(saved.json, { ...original, allowLocalEnvironment: false });
  const invalid = await server.api("PATCH", "/api/settings", { allowLocalEnvironment: "false" });
  assertEquals(invalid.status, 400);
  assertEquals((await server.api("GET", "/api/settings")).json.allowLocalEnvironment, false);
  const restored = await server.api("PATCH", "/api/settings", { allowLocalEnvironment: true });
  assertEquals(restored.json.allowLocalEnvironment, true);
});

/** Replays every topic from its snapshot and ops, as the frontend will. */
async function subscriber(base: string) {
  const socket = new WebSocket(base.replace("http", "ws") + "/ws");
  const values = new Map<string, unknown>();
  const errors: string[] = [];
  let changed = () => {};
  socket.onmessage = (event) => {
    const message = JSON.parse(event.data) as ServerMessage;
    if (message.t === "snapshot") values.set(message.topic, message.value);
    if (message.t === "older") {
      const value = values.get(message.topic) as { entries: unknown[] };
      values.set(message.topic, { ...value, entries: [...message.entries, ...value.entries] });
    }
    if (message.t === "ops") {
      values.set(message.topic, applyImmutable(values.get(message.topic), message.ops as Op[]));
    }
    if (message.t === "error") errors.push(message.message);
    changed();
  };
  await new Promise((resolve) => (socket.onopen = resolve));
  socket.send(JSON.stringify({ t: "auth", token: TOKEN }));
  return {
    values,
    errors,
    sub: (topic: string) => socket.send(JSON.stringify({ t: "sub", topic })),
    async until<T>(topic: string, check: (value: T) => boolean): Promise<T> {
      while (!(values.has(topic) && check(values.get(topic) as T))) {
        await new Promise<void>((resolve, reject) => {
          const timeout = setTimeout(
            () => reject(new Error(`timed out waiting on ${topic}`)),
            5000,
          );
          changed = () => (clearTimeout(timeout), resolve());
        });
      }
      return values.get(topic) as T;
    },
    close: () => new Promise((resolve) => ((socket.onclose = resolve), socket.close())),
  };
}

Deno.test("api requests need the bearer token and a valid body, which is refused by its failing path", async () => {
  await using server = await serve();
  const { api, base } = server;
  assertEquals((await fetch(base + "/api/conversations")).status, 401);
  const wrong = await fetch(base + "/api/conversations", {
    headers: { authorization: "Bearer nope" },
  });
  assertEquals(wrong.status, 401);
  await wrong.body?.cancel();
  const response = await api("POST", "/api/conversations", { extensions: { dir: "nope" } });
  assertEquals(response.status, 400);
  assert(String(response.json.error).includes("/extensions/dir"));
  assertEquals(
    (await api("POST", "/api/conversations", { extensions: { missing: {} } })).status,
    400,
  );
});

Deno.test("restart refuses code that fails its check, and restarts after responding when it passes", async () => {
  let failures: string | null = "error: TS2304 nope";
  let restarts = 0;
  const restarted = Promise.withResolvers<void>();
  await using server = await serve({
    root: "/berry",
    check: () => Promise.resolve(failures),
    restart: () => {
      restarts++;
      restarted.resolve();
    },
    extensions: () => [],
    reload: () => Promise.reject(new Error("no extensions")),
  });
  const { api } = server;
  assertEquals(await api("POST", "/api/restart"), {
    status: 400,
    json: { error: failures },
  });
  assertEquals(restarts, 0);
  failures = null;
  assertEquals(await api("POST", "/api/restart"), { status: 200, json: {} });
  await restarted.promise;
  assertEquals(restarts, 1);
});

Deno.test("topics replay to the server's state over the websocket", async () => {
  await using server = await serve();
  const { api, base, faux, model } = server;
  const client = await subscriber(base);

  client.sub("conversations");
  const { json: { tid } } = await api("POST", "/api/conversations", {
    extensions: { dir: {} },
    model,
  });
  await client.until<Record<string, unknown>>("conversations", (index) => tid in index);

  client.sub(`conversation:${tid}`);
  client.sub(`state:${tid}:dir`);
  await client.until(`conversation:${tid}`, () => true);
  assertEquals(await client.until(`state:${tid}:dir`, () => true), { cwd: "/" });

  faux.setResponses([fauxAssistantMessage("hello there")]);
  await api("POST", `/api/conversations/${tid}/messages`, { content: "hi" });
  type View = {
    entries: { kind: string; model?: { content: { type: string; text?: string }[] }[] }[];
  };
  const view = await client.until<View>(
    `conversation:${tid}`,
    (view) => view.entries.some((entry) => entry.kind === "pi.assistant"),
  );
  const answer = view.entries.find((entry) => entry.kind === "pi.assistant")!.model![0].content[0];
  assertEquals(answer.text, "hello there");

  await api("POST", `/api/conversations/${tid}/events`, {
    ext: "dir",
    event: { cwd: "/elsewhere" },
  });
  await client.until<{ cwd: string }>(`state:${tid}:dir`, (state) => state.cwd === "/elsewhere");

  client.sub("nonsense");
  await client.until("conversations", () => client.errors.length > 0);
  assertEquals(client.errors, ["no topic nonsense"]);

  await client.close();
});

Deno.test("a conversation's configuration, commands, and handlers are served over http", async () => {
  await using server = await serve();
  const { base, api, model } = server;
  const client = await subscriber(base);
  try {
    const { tid } =
      (await api("POST", "/api/conversations", { extensions: { dir: {} }, model })).json;
    const conversation = `/api/conversations/${tid}`;
    const topic = `configuration:${tid}:dir`;
    const until = (cwd: string) =>
      client.until<{ cwd: string }>(topic, (values) => values.cwd === cwd);
    client.sub(topic);
    await until("/");
    const configuration = `${conversation}/configuration/dir`;
    assertEquals(await api("PATCH", configuration, { cwd: "/edited" }), {
      status: 200,
      json: { cwd: "/edited" },
    });
    await until("/edited");
    assertEquals((await api("PATCH", configuration, { cwd: 5 })).status, 400);
    assertEquals(await api("PATCH", configuration, { cwd: "relative" }), {
      status: 400,
      json: { error: "not absolute: relative" },
    });
    assertEquals((await api("PATCH", configuration, { hidden: true })).status, 400);

    assertEquals((await api("GET", `${conversation}/commands`)).json, [
      { name: "cd", extension: "dir", description: "change directory" },
    ]);
    assertEquals(await api("POST", `${conversation}/commands/cd`, { args: "/x" }), {
      status: 200,
      json: { result: "/x" },
    });
    await until("/x");
    assertEquals(await api("POST", `${conversation}/commands/cd`, { args: "x" }), {
      status: 400,
      json: { error: "not absolute: x" },
    });
    assertEquals((await api("POST", `${conversation}/commands/nope`, {})).status, 400);

    const where = (body: unknown) => api("POST", `${conversation}/extensions/dir/where`, body);
    assertEquals(await where({}), { status: 200, json: { result: "/x" } });
    assertEquals(await where({ args: { under: "/y" } }), { status: 200, json: { result: "/x/y" } });
    assertEquals(await where({ args: { under: 1 } }), {
      status: 400,
      json: { error: "under: must be string" },
    });
    assertEquals((await api("POST", `${conversation}/extensions/dir/nope`, {})).status, 400);
    // a long answer is gzipped for a client that accepts it
    const under = "/y".repeat(5000);
    const fetchWhere = (encoding: string) =>
      fetch(`${base}${conversation}/extensions/dir/where`, {
        method: "POST",
        headers: {
          authorization: `Bearer ${TOKEN}`,
          "content-type": "application/json",
          "accept-encoding": encoding,
        },
        body: JSON.stringify({ args: { under } }),
      });
    const gzipped = await fetchWhere("gzip, br");
    assertEquals(gzipped.headers.get("content-encoding"), "gzip");
    assertEquals(await gzipped.json(), { result: `/x${under}` });
    const plain = await fetchWhere("identity");
    assertEquals(plain.headers.get("content-encoding"), null);
    assertEquals(await plain.json(), { result: `/x${under}` });

    await api("PATCH", `${conversation}/agent`, { extensions: [] });
    assertEquals((await api("PATCH", configuration, { cwd: "/disabled" })).status, 400);
  } finally {
    await client.close();
  }
});

Deno.test("an extension's channels carry json and bytes both ways once the first message authorizes them", async () => {
  await using server = await serve();
  const { base, api, model } = server;
  const { json: { tid } } = await api("POST", "/api/conversations", {
    extensions: { dir: { cwd: "/x" } },
    model,
  });
  const open = (name: string, first: unknown) => {
    const socket = new WebSocket(
      `${base.replace("http", "ws")}/api/conversations/${tid}/extensions/dir/${name}`,
    );
    socket.binaryType = "arraybuffer";
    const received: unknown[] = [];
    const closed = new Promise<{ code: number; reason: string }>((resolve) =>
      socket.onclose = ({ code, reason }) => resolve({ code, reason })
    );
    socket.onmessage = ({ data }) =>
      received.push(typeof data === "string" ? JSON.parse(data) : [...new Uint8Array(data)]);
    socket.onopen = () => socket.send(JSON.stringify(first));
    return { socket, received, closed };
  };

  const echo = open("echo", { token: TOKEN, args: { prefix: ">" } });
  await new Promise((resolve) => echo.socket.addEventListener("open", resolve));
  echo.socket.send(JSON.stringify({ hello: 1 }));
  echo.socket.send(new Uint8Array([1, 2, 3]));
  echo.socket.send(JSON.stringify("bye"));
  assertEquals(await echo.closed, { code: 1000, reason: "" });
  assertEquals(echo.received, [{ echo: { hello: 1 }, prefix: ">", cwd: "/x" }, [3, 2, 1]]);

  assertEquals((await open("echo", { token: "wrong" }).closed).code, 4001);
  assertEquals(await open("echo", { token: TOKEN, args: { prefix: "refuse" } }).closed, {
    code: 4400,
    reason: "not today",
  });
  assertEquals(
    (await open("nope", { token: TOKEN }).closed).reason,
    "dir has no channel named nope",
  );
  assertEquals((await open("echo", { token: TOKEN, args: { prefix: 1 } }).closed).code, 4400);
});

Deno.test("a conversation forks at an entry or at its newest, and branches in place with its topic following", async () => {
  await using server = await serve();
  const { api, base, faux, model } = server;
  const client = await subscriber(base);
  try {
    const { json: { tid } } = await api("POST", "/api/conversations", { extensions: {}, model });
    type View = { entries: { id: number; kind: string }[] };
    const topic = `conversation:${tid}`;
    const has = (kind: string) => (view: View) => view.entries.some((entry) => entry.kind === kind);
    client.sub(topic);
    faux.setResponses([fauxAssistantMessage("hi")]);
    await api("POST", `/api/conversations/${tid}/messages`, { content: "hello" });
    const { entries } = await client.until<View>(topic, has("pi.assistant"));
    const [start, user] = entries;

    const { json: { tid: copy } } = await api("POST", `/api/conversations/${tid}/fork`, {});
    const { json: { tid: edit } } = await api("POST", `/api/conversations/${tid}/fork`, {
      entryId: start.id,
    });
    const index = (await api("GET", "/api/conversations")).json;
    assertEquals(index[copy].parent, { tid, at: entries.at(-1)!.id });
    assertEquals(index[edit].parent, { tid, at: start.id });
    const forked = (await api("GET", `/api/conversations/${edit}/tree`)).json;
    assertEquals(forked.branches, [{
      id: index[edit].id,
      created: index[edit].created,
      tip: start.id,
    }]);
    // what a fork at an entry starts with: the entries through it
    const context = await api("GET", `/api/conversations/${tid}/context?at=${user.id}`);
    assertEquals(context.json.entries.map((entry: { id: number }) => entry.id), [
      start.id,
      user.id,
    ]);
    assertEquals((await api("GET", `/api/conversations/${tid}/context`)).status, 400);
    // the edit's history ends before the first message
    assertEquals(await api("GET", `/api/conversations/${edit}/context?at=${user.id}`), {
      status: 400,
      json: { error: `no entry ${user.id} in ${edit}` },
    });
    assertEquals(
      (await api("POST", `/api/conversations/${edit}/fork`, { entryId: user.id })).status,
      400,
    );

    // before the first message, then back to where it was
    assertEquals(await api("POST", `/api/conversations/${tid}/branch`, { entryId: start.id }), {
      status: 200,
      json: {},
    });
    await client.until<View>(topic, (view) => !has("pi.user")(view));
    const tree = (await api("GET", `/api/conversations/${tid}/tree`)).json;
    assertEquals(tree.branches.length, 2);
    assertEquals(tree.head, tree.branches[1].id);
    await api("POST", `/api/conversations/${tid}/branch`, { entryId: entries.at(-1)!.id });
    await client.until<View>(topic, has("pi.assistant"));
    assertEquals(
      (await api("GET", `/api/conversations/${tid}/tree`)).json.head,
      tree.branches[0].id,
    );
    assertEquals(
      (await api("POST", `/api/conversations/${tid}/branch`, { entryId: 1e9 })).status,
      400,
    );
  } finally {
    await client.close();
  }
});

Deno.test("logins ask through their topic and are answered over http", async () => {
  await using server = await serve();
  const { api, base } = server;
  const client = await subscriber(base);
  type Provider = {
    id: string;
    apiKey: { name: string } | null;
    auth: { type: string } | null;
    stored: boolean;
  };
  const keyed = async () =>
    ((await api("GET", "/api/providers")).json as Provider[]).find((provider) =>
      provider.id === "keyed"
    )!;
  assertEquals((await keyed()).stored, false);

  type Login = {
    status: string;
    prompt: { id: number; type: string; message: string } | null;
    events: { message?: string }[];
  };
  const { json: { login } } = await api("POST", "/api/providers/keyed/login", { type: "api_key" });
  client.sub(`login:${login}`);
  const asked = await client.until<Login>(`login:${login}`, (state) => state.prompt !== null);
  assertEquals(asked.prompt, { id: asked.prompt!.id, type: "secret", message: "your key?" });
  await api("POST", `/api/logins/${login}/answer`, { prompt: asked.prompt!.id, value: "sekrit" });
  const done = await client.until<Login>(`login:${login}`, (state) => state.status === "done");
  assertEquals(done.events.map((event) => event.message), ["thanks"]);
  assertEquals((await keyed()).stored, true);
  assertEquals((await keyed()).auth?.type, "api_key");
  // the question is gone once answered
  assertEquals(
    (await api("POST", `/api/logins/${login}/answer`, { prompt: asked.prompt!.id, value: "x" }))
      .status,
    400,
  );

  await api("POST", "/api/providers/keyed/logout");
  assertEquals((await keyed()).stored, false);
  assertEquals((await keyed()).auth, null);

  const { json: { login: again } } = await api("POST", "/api/providers/keyed/login", {
    type: "api_key",
  });
  client.sub(`login:${again}`);
  await client.until<Login>(`login:${again}`, (state) => state.prompt !== null);
  await api("POST", `/api/logins/${again}/cancel`);
  const cancelled = await client.until<Login>(
    `login:${again}`,
    (state) => state.status !== "running",
  );
  assertEquals([cancelled.status, cancelled.prompt], ["cancelled", null]);
  assertEquals((await keyed()).stored, false);

  assertEquals((await api("POST", "/api/providers/keyed/login", { type: "oauth" })).status, 400);
  await client.close();
});

Deno.test("a conversation's read receipt, name, and archive state change independently over http", async () => {
  await using server = await serve();
  const { api, berry, faux, model } = server;
  const { json: { tid } } = await api("POST", "/api/conversations", { extensions: {}, model });
  const path = `/api/conversations/${tid}`;
  const info = async () => (await api("GET", "/api/conversations")).json[tid];
  faux.setResponses([fauxAssistantMessage("done")]);
  await api("POST", `${path}/messages`, { content: "go" });
  const current = (await berry.conversation(tid))!;
  await current.waitForIdle(context);
  const { entries } = await current.context(context);
  const entryId = entries.findLast((entry) => entry.kind === "pi.assistant")!.id;
  const read = (target: string, entryId: number) =>
    api("POST", `/api/conversations/${target}/read`, { entryId });
  assertEquals((await read(tid, entryId)).status, 200);
  assertEquals((await info()).readThrough, entryId);
  assertEquals((await read(tid, 1.5)).status, 400);
  const { json: { tid: other } } = await api("POST", "/api/conversations", { extensions: {} });
  assertEquals((await read(other, entryId)).status, 400);
  assertEquals((await read("aaaaaaaaaaaaa", entryId)).status, 404);

  assertEquals((await api("PATCH", path, { title: "  planning 🦊  " })).status, 200);
  assertEquals((await info()).title, "planning 🦊");
  await api("PATCH", path, { archived: true });
  const archived = (await info()).archived;
  assertEquals(typeof archived, "number");
  await api("PATCH", path, { title: "another name" });
  const renamed = await info();
  assertEquals([renamed.title, renamed.archived, renamed.readThrough], [
    "another name",
    archived,
    entryId,
  ]);
  await api("PATCH", path, { title: "   ", archived: false });
  assertEquals([(await info()).title, (await info()).archived], [undefined, undefined]);
  assertEquals((await api("PATCH", path, { title: 5 })).status, 400);
  assertEquals(
    (await api("PATCH", "/api/conversations/aaaaaaaaaaaaa", { archived: true })).status,
    404,
  );
});

Deno.test("an image message reaches the model and its history whole, and invalid ones are refused", async () => {
  await using server = await serve();
  const { api, berry, faux, model } = server;
  const image: ImageContent = { type: "image", mimeType: "image/png", data: PNG };
  const seen: Message[][] = [];
  faux.setResponses([(request) => {
    seen.push([...request.messages]);
    return fauxAssistantMessage("I see the image");
  }]);
  const { json: { tid } } = await api("POST", "/api/conversations", { extensions: {}, model });
  const messages = `/api/conversations/${tid}/messages`;
  const refused = async (content: unknown) => (await api("POST", messages, { content })).status;
  assertEquals(await refused([{ ...image, data: "not base64!" }]), 400);
  assertEquals(await refused([{ ...image, mimeType: "image/svg+xml" }]), 400);
  assertEquals(await refused(Array.from({ length: 5 }, () => image)), 400);
  const large = encodeBase64(new Uint8Array(5 * 1024 * 1024 + 1));
  assertEquals(await refused([{ ...image, data: large }]), 413);
  assertEquals(await refused("x".repeat(31 * 1024 * 1024)), 413);

  const content = [{ type: "text" as const, text: "What is this?" }, image, image];
  await api("POST", messages, { content });
  const current = (await berry.conversation(tid))!;
  await current.waitForIdle(context);
  const users = (messages: readonly Message[]) =>
    messages.filter((message) => message.role === "user").map((message) => message.content);
  assertEquals(users(seen.at(-1)!), [content]);
  assertEquals(users((await current.context(context)).messages), [content]);
  const { json: tree } = await api("GET", `/api/conversations/${tid}/tree`);
  const user = tree.entries.find((entry: { kind: string }) => entry.kind === "pi.user");
  assertEquals([user.text, user.imageCount], ["What is this?", 2]);

  await using textOnly = await serve(undefined, false);
  const { json: { tid: text } } = await textOnly.api("POST", "/api/conversations", {
    extensions: {},
    model: textOnly.model,
  });
  const response = await textOnly.api("POST", `/api/conversations/${text}/messages`, {
    content: [image],
  });
  assertEquals(response.status, 400);
  assertStringIncludes(response.json.error, "supports image input");
});

Deno.test("abort returns edited queued inputs with attachments, not the input already in the transcript", async () => {
  await using server = await serve();
  const { api, berry, faux, model } = server;
  const blocker = blocking();
  await berry.install(blocker.extension);
  faux.setResponses([
    fauxAssistantMessage(fauxToolCall("block", {}), { stopReason: "toolUse" }),
  ]);
  const tid = await berry.create({ extensions: { blocking: {} }, model });
  const current = (await berry.conversation(tid))!;
  const base = `/api/conversations/${tid}`;
  try {
    await current.submit({ type: "input", content: "already placed" }, context);
    await until(() => blocker.running > 0);
    const image: ImageContent = { type: "image", mimeType: "image/png", data: PNG };
    const first = await api("POST", `${base}/messages`, {
      content: [{ type: "text", text: "old" }, image],
      whenBusy: "steer",
    });
    await api("PATCH", `${base}/inbox/${first.json.submission}`, { text: "latest" });
    await api("POST", `${base}/messages`, { content: "later", whenBusy: "followUp" });
    const queued = (await berry.harness.snapshot(InboxDoc, current.id, context))!;
    const response = await api("POST", `${base}/abort`);
    assertEquals(response, { status: 200, json: { inputs: queued.items } });
    assertEquals((await berry.harness.snapshot(InboxDoc, current.id, context))!.items, []);
    assertEquals((await api("POST", `${base}/abort`)).json, { inputs: [] });
  } finally {
    blocker.release.resolve("released");
  }
});

Deno.test("queued input edits keep identity, mode, order, attachments, and original receipts", async () => {
  const whenBusy = "followUp";
  await using server = await serve();
  const { api, berry, faux, model } = server;
  const { gate, extension } = gated();
  await berry.install(extension);
  const image: ImageContent = { type: "image", mimeType: "image/png", data: PNG };
  const seen: Message[][] = [];
  const record = (request: { messages: readonly Message[] }) => {
    seen.push([...request.messages]);
    return fauxAssistantMessage("done");
  };
  try {
    const tid = await berry.create({ extensions: { gate: {} }, model });
    const current = (await berry.conversation(tid))!;
    const base = `/api/conversations/${tid}`;
    faux.setResponses([
      fauxAssistantMessage(fauxToolCall("hold", {}), { stopReason: "toolUse" }),
      record,
      record,
      record,
    ]);
    const active = await current.submit({ type: "input", content: "wait" }, context);
    await gate.entered.promise;
    const contents = [
      [{ type: "text", text: "old" }, image, { type: "text", text: "also old" }, image],
      "retract me",
      "old tail",
    ];
    const receipts = [];
    for (const [index, content] of contents.entries()) {
      const response = await api("POST", `${base}/messages`, {
        content,
        whenBusy,
        requestId: `queued-${index}`,
      });
      assertEquals(response.status, 200);
      receipts.push((await berry.harness.submission(response.json.submission, context))!);
    }
    const before = await berry.harness.snapshot(InboxDoc, current.id, context);
    const live = await berry.harness.snapshot(LiveDoc, current.id, context);
    const original = await Promise.all(receipts.map((receipt) => receipt.status(context)));
    assertEquals(
      before!.items.map(({ id, mode }) => ({ id, mode })),
      receipts.map(({ id }) => ({ id, mode: whenBusy })),
    );
    const withdrawn = receipts[1].wait(context);
    assertEquals(await api("PATCH", `${base}/inbox/${receipts[0].id}`, { text: "edited" }), {
      status: 200,
      json: {},
    });
    assertEquals(
      (await api("PATCH", `${base}/inbox/${receipts[2].id}`, {
        text: "edited tail",
      })).status,
      200,
    );
    assertEquals(await api("DELETE", `${base}/inbox/${receipts[1].id}`), {
      status: 200,
      json: {},
    });
    const edited = [{ type: "text" as const, text: "edited" }, image, image];
    assert(before!.items[0].mode !== "write" && before!.items[2].mode !== "write");
    assertEquals(await berry.harness.snapshot(InboxDoc, current.id, context), {
      items: [
        { ...before!.items[0], content: edited },
        { ...before!.items[2], content: "edited tail" },
      ],
    });
    assert(original[1].status === "queued");
    assertEquals(await withdrawn, { ...original[1], status: "unanswered", reason: "aborted" });
    assertEquals(await receipts[0].status(context), original[0]);
    assertEquals(await receipts[2].status(context), original[2]);
    assertEquals(await berry.harness.snapshot(LiveDoc, current.id, context), live);
    assertEquals((await active.status(context)).status, "placed");
    const duplicate = await api("POST", `${base}/messages`, {
      content: "must not replace the edit",
      requestId: "queued-0",
    });
    assertEquals(duplicate.json.submission, receipts[0].id);
    gate.release.resolve("released");
    await current.waitForIdle(context);
    assertEquals(
      seen.at(-1)!.filter((message) => message.role === "user").map((message) => message.content),
      ["wait", edited, "edited tail"],
    );
    assertEquals((await active.wait(context)).status, "done");
    for (const receipt of [receipts[0], receipts[2]]) {
      const settled = await receipt.wait(context);
      assertEquals(settled.id, receipt.id);
      assertEquals(settled.status, "done");
    }
    const late = `${base}/inbox/${receipts[0].id}`;
    assertEquals((await api("PATCH", late, { text: "late" })).status, 409);
    assertEquals((await api("DELETE", late)).status, 409);
  } finally {
    gate.release.resolve("released");
  }
});

Deno.test("queued input routes refuse invalid edits and other conversations' inputs and writes, and bound image edits", async () => {
  await using server = await serve();
  const { api, berry, faux, model } = server;
  const { gate, extension } = gated();
  await berry.install(extension);
  const image: ImageContent = { type: "image", mimeType: "image/png", data: PNG };
  try {
    const tid = await berry.create({ extensions: { gate: {}, dir: {} }, model });
    const other = await berry.create({ extensions: {}, model });
    const current = (await berry.conversation(tid))!;
    const base = `/api/conversations/${tid}/inbox`;
    faux.setResponses([
      fauxAssistantMessage(fauxToolCall("hold", {}), { stopReason: "toolUse" }),
      fauxAssistantMessage("done"),
      fauxAssistantMessage("done"),
    ]);
    const active = await current.submit({ type: "input", content: "wait" }, context);
    await gate.entered.promise;
    const queued = await current.submit({ type: "input", content: "original" }, context);
    const pictured = await current.submit({ type: "input", content: [image] }, context);
    const write = await berry.emit(tid, "dir", { cwd: "/kept" });
    const before = await berry.harness.snapshot(InboxDoc, current.id, context);
    const change = (path: string, text = "change") => api("PATCH", path, { text });
    assertEquals((await change(`${base}/1junk`)).status, 400);
    assertEquals((await change(`${base}/${queued.id}`, " \n\t")).status, 400);
    assertEquals((await change(`${base}/${write.id}`)).status, 404);
    assertEquals((await change(`/api/conversations/${other}/inbox/${queued.id}`)).status, 404);
    assertEquals((await change(`${base}/${active.id}`)).status, 409);
    // two bytes a character, so only the image takes it over the limit
    const text = "é".repeat((MAX_MESSAGE_BYTES - 100) / 2);
    assertEquals((await change(`${base}/${pictured.id}`, text)).status, 413);
    assertEquals(await berry.harness.snapshot(InboxDoc, current.id, context), before);
    // an image needs no text
    assertEquals((await change(`${base}/${pictured.id}`, "")).status, 200);
    const [, picture] = (await berry.harness.snapshot(InboxDoc, current.id, context))!.items;
    assertEquals(picture, { id: pictured.id, mode: "followUp", content: [image] });

    gate.release.resolve("released");
    await current.waitForIdle(context);
    const start = (await current.context(context)).entries[0].id;
    assertEquals(
      (await api("POST", `/api/conversations/${tid}/branch`, { entryId: start })).status,
      200,
    );
    assertEquals((await change(`${base}/${queued.id}`)).status, 404);
  } finally {
    gate.release.resolve("released");
  }
});

Deno.test("references and the shell follow the conversation's directory while code is selected", async () => {
  const { default: code } = await import("../extensions/code/mod.ts");
  await using rootDir = await tempDir();
  const root = rootDir.path;
  await using s = await serve();
  await s.berry.install(code);
  await Deno.mkdir(`${root}/other`);
  await Deno.writeTextFile(`${root}/first.ts`, "");
  await Deno.writeTextFile(`${root}/other/second.ts`, "");
  // before a conversation works there, as a new one's composer asks
  assertEquals(
    await s.api("GET", `/api/references?cwd=${encodeURIComponent(root)}&prefix=f`),
    { status: 200, json: ["first.ts"] },
  );
  const created = await s.api("POST", "/api/conversations", {
    extensions: { environment: { cwd: root }, code: {} },
    model: s.model,
  });
  const tid = created.json.tid;
  const base = `/api/conversations/${tid}`;
  const conv = (await s.berry.conversation(tid))!;
  const entries = await conv.entries({ order: "descending" }, 1, undefined, context);
  const at = entries.items[0].id;
  assertEquals(await s.api("GET", `${base}/references?prefix=f`), {
    status: 200,
    json: ["first.ts"],
  });
  await s.api("PATCH", `${base}/configuration/environment`, { cwd: `${root}/other` });
  assertEquals(await s.api("GET", `${base}/references?prefix=s`), {
    status: 200,
    json: ["second.ts"],
  });
  assertEquals(await s.api("GET", `${base}/references?prefix=f&at=${at}`), {
    status: 200,
    json: ["first.ts"],
  });
  const { json: { task } } = await s.api("POST", `${base}/shell`, { command: "touch made" });
  await s.berry.harness.waitForTask(task, context);
  await Deno.stat(`${root}/other/made`);

  await s.api("PATCH", `${base}/agent`, { extensions: [] });
  assertEquals(await s.api("GET", `${base}/references`), { status: 200, json: [] });
});

Deno.test("projects publish their edits, and folders are suggested over http", async () => {
  await using rootDir = await tempDir();
  const root = rootDir.path;
  await using server = await serve();
  const { api, base } = server;
  await Deno.mkdir(`${root}/workspace`);
  const socket = await subscriber(base);
  try {
    const folders = `/api/folders?cwd=${encodeURIComponent(root)}&prefix=work`;
    assertEquals(await api("GET", folders), { status: 200, json: ["workspace/"] });
    socket.sub("projects");
    const { json: { id } } = await api("POST", "/api/projects", {
      name: "Berry",
      defaultRoot: 0,
      roots: [{ environment: "local", cwd: root }],
    });
    type Index = Record<string, { name: string }>;
    await socket.until<Index>("projects", (value) => value[id]?.name === "Berry");
    socket.sub("project-order");
    await api("PUT", "/api/project-order", { ids: ["", id] });
    await socket.until<{ ids: string[] }>("project-order", (value) => value.ids[0] === "");
    await api("DELETE", `/api/projects/${id}`);
    await socket.until<Index>("projects", (value) => !(id in value));
  } finally {
    await socket.close();
  }
});