char-slop/berry

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

Charlotte Somadd background subagents (!!!)2d397472026-10-18

import { assert, assertEquals, assertObjectMatch } from "@std/assert";
import { BACKGROUND_CONTEXT as context } from "@earendil-works/chord/context";
import { fauxAssistantMessage, fauxToolCall } from "@earendil-works/pi-ai/providers/faux";
import type { Message } from "@earendil-works/pi-ai";
import { type Berry, Conversations } from "./berry.ts";
import { blocking, setup } from "./fixture.ts";
import { summaries, type Summary } from "./summaries.ts";
import { SettledDoc } from "./settled.ts";
import { until } from "@berry/util";

/** A conversation's summary, as a fresh feed first sees it. */
async function summary(berry: Berry, tid: string): Promise<Summary> {
  const watched = await summaries(berry);
  try {
    return (watched.feed().value as Record<string, Summary>)[tid];
  } finally {
    watched.stop();
  }
}

Deno.test("a summary says whether a conversation is answering, how it began, and when it was last active", async () => {
  await using harness = await setup();
  const { faux, berry, model, gate } = harness;
  const blocker = blocking();
  await berry.install(blocker.extension);
  const tid = await berry.create({ extensions: { gate: {}, blocking: {} }, model });
  const conversation = (await berry.conversation(tid))!;
  const created = (await berry.harness.snapshot(Conversations, context))![tid].created;
  assertObjectMatch(await summary(berry, tid), {
    busy: false,
    preview: "",
    reply: "",
    unread: false,
    updated: created,
  });

  // a queued follow-up keeps it working until the final answer
  const followUp = {
    entered: Promise.withResolvers<void>(),
    release: Promise.withResolvers<void>(),
  };
  faux.setResponses([
    fauxAssistantMessage(fauxToolCall("hold", {}), { stopReason: "toolUse" }),
    fauxAssistantMessage("first done"),
    async () => {
      followUp.entered.resolve();
      await followUp.release.promise;
      return fauxAssistantMessage("follow-up done");
    },
  ]);
  await until(() => Date.now() > created);
  const submitted = Date.now();
  await conversation.submit({ type: "input", content: "hello\n  there" }, context);
  await gate.entered.promise;
  await conversation.submit({ type: "input", content: "next", whenBusy: "followUp" }, context);
  const sending = await summary(berry, tid);
  assertObjectMatch(sending, { busy: true, preview: "hello there", unread: false, reply: "" });
  assert(sending.updated >= submitted);
  gate.release.resolve("released");
  await followUp.entered.promise;
  const queued = await summary(berry, tid);
  assertObjectMatch(queued, { busy: true, unread: false, reply: "first done" });
  assert(queued.updated >= sending.updated);
  let returned = Date.now();
  followUp.release.resolve();
  await conversation.waitForIdle(context);
  let current = await summary(berry, tid);
  assertObjectMatch(current, {
    busy: false,
    preview: "hello there",
    unread: true,
    reply: "follow-up done",
  });
  assert(current.updated >= returned);

  // interrupted tools return control without a new answer
  faux.setResponses([fauxAssistantMessage(fauxToolCall("block", {}), { stopReason: "toolUse" })]);
  await until(() => Date.now() > current.updated);
  const resubmitted = Date.now();
  await conversation.submit({ type: "input", content: "again" }, context);
  await until(() => blocker.running > 0);
  const interrupting = await summary(berry, tid);
  assert(interrupting.updated >= resubmitted);
  assert(interrupting.updated > current.updated);
  await until(() => Date.now() > interrupting.updated);
  returned = Date.now();
  await conversation.abort(context);
  current = await summary(berry, tid);
  assertObjectMatch(current, { busy: false });
  assert(current.updated >= returned);

  // as does a failure before generation
  await berry.configure(tid, { model: null });
  await until(() => Date.now() > current.updated);
  returned = Date.now();
  const failed = await (await conversation.submit({ type: "input", content: "go" }, context))
    .wait(context);
  assertEquals(failed.status, "unanswered");
  await conversation.waitForIdle(context);
  assert((await summary(berry, tid)).updated >= returned);
});

Deno.test("unread answers follow read receipts across feeds, forks, branches, and resets", async () => {
  await using harness = await setup();
  const { faux, berry, model } = harness;
  const tid = await berry.create({ extensions: {}, model });
  const watched = await summaries(berry);
  const feeds = [watched.feed(), watched.feed()];
  const seen = feeds.map((feed) => feed.value as Record<string, Summary>);
  feeds.forEach((feed, i) =>
    feed.start((value) => {
      seen[i] = value as Record<string, Summary>;
      return Promise.resolve();
    })
  );
  const unread = (expected: boolean) =>
    until(() => seen.every((value) => value[tid]?.unread === expected));
  let conversation = (await berry.conversation(tid))!;
  const answer = async () => {
    faux.setResponses([fauxAssistantMessage("done")]);
    await conversation.submit({ type: "input", content: "go" }, context);
    await conversation.waitForIdle(context);
    await unread(true);
    return (await conversation.context(context)).entries.at(-1)!.id;
  };
  try {
    const old = await answer();
    await berry.markRead(tid, old);
    await unread(false);
    const latest = await answer();
    // an older receipt cannot clear a newer answer
    await berry.markRead(tid, old);
    assertEquals((await summary(berry, tid)).unread, true);
    await berry.markRead(tid, latest);
    await unread(false);
    await berry.markRead(tid, old);
    assertEquals((await summary(berry, tid)).unread, false);

    const answered = (await summary(berry, tid)).updated;
    const fork = await conversation.fork(latest, { ownership: { kind: "ownerless" } }, context);
    assertObjectMatch(await summary(berry, await berry.tidOf(fork.id)), {
      unread: false,
      updated: answered,
    });

    const [start] = (await conversation.context(context)).entries;
    await berry.branch(tid, start.id);
    conversation = (await berry.conversation(tid))!;
    await answer();
    const branched = (await summary(berry, tid)).updated;
    await conversation.reset(undefined, context);
    assertObjectMatch(await summary(berry, tid), { unread: false, updated: branched });
    await berry.branch(tid, latest);
    assertEquals((await summary(berry, tid)).updated, answered);
  } finally {
    for (const feed of feeds) await feed.stop();
    watched.stop();
  }
});

Deno.test("legacy summaries find completed answers across history pages", async () => {
  await using harness = await setup();
  const { berry, model } = harness;
  const tid = await berry.create({ extensions: {}, model });
  const conversation = (await berry.conversation(tid))!;
  await conversation.commit((tx) => tx.retireDoc(SettledDoc, conversation.id), context);
  const created = (await berry.harness.snapshot(Conversations, context))![tid].created;
  const messages: Message[] = [
    { role: "user", content: "hello", timestamp: 1000 },
    fauxAssistantMessage("done", { timestamp: 2000 }),
  ];
  for (const message of messages) {
    await conversation.commit(async (tx) => {
      await tx.appendEntry(conversation.id, { kind: `pi.${message.role}`, model: [message] });
      for (let i = 0; i < 32; i++) {
        await tx.appendEntry(conversation.id, {
          kind: "berry.event",
          data: { ext: "dir", event: { type: "cd", cwd: "/elsewhere" } },
        });
      }
    }, context);
    assertEquals(
      (await summary(berry, tid)).updated,
      message.role === "assistant" ? message.timestamp : created,
    );
  }
});

Deno.test("renaming one conversation leaves the others' summaries as they were", async () => {
  await using harness = await setup();
  const { berry, model } = harness;
  const renamed = await berry.create({ extensions: {}, model });
  const other = await berry.create({ extensions: {}, model });
  const watched = await summaries(berry);
  try {
    const before = watched.feed().value as Record<string, Summary>;
    await berry.rename(renamed, "named");
    await berry.archive(renamed, true);
    await until(() => watched.feed().value !== before);
    const after = watched.feed().value as Record<string, Summary>;
    assert(after[other] === before[other]);
  } finally {
    watched.stop();
  }
});

Deno.test("a busy background subagent's conversation gets a summary that leaves the feed once it settles", async () => {
  await using harness = await setup();
  const { berry, faux, model } = harness;
  const blocker = blocking();
  await berry.install(blocker.extension);
  const tid = await berry.create({ extensions: { subagent: {}, blocking: {} }, model });
  const parent = (await berry.conversation(tid))!;
  const last = (messages: readonly Message[]) => messages.findLast((m) => m.role !== "system")!;
  const text = (message: Message) =>
    typeof message.content === "string"
      ? message.content
      : message.content.flatMap((block) => block.type === "text" ? [block.text] : []).join("\n");
  faux.setResponses(Array.from({ length: 20 }, () => (request) => {
    const message = last(request.messages);
    if (message.role === "user" && text(message) === "go") {
      return fauxAssistantMessage(
        fauxToolCall("codemode", {
          code: `return await tools.subagent({task:"child work", model:${
            JSON.stringify(model)
          }, background:true})`,
        }),
        { stopReason: "toolUse" },
      );
    }
    if (message.role === "user" && text(message) === "child work") {
      return fauxAssistantMessage(fauxToolCall("block", {}), { stopReason: "toolUse" });
    }
    if (message.role === "toolResult" && message.toolName === "block") {
      return fauxAssistantMessage(text(message));
    }
    return fauxAssistantMessage("done");
  }));
  await (await parent.submit({ type: "input", content: "go" }, context)).wait(context);
  await until(() => blocker.running > 0);
  const index = (await berry.harness.snapshot(Conversations, context))!;
  const [child] = Object.keys(index).filter((key) => index[key].owner === tid);
  assert(child);
  const busy = await summary(berry, child);
  assertObjectMatch(busy, { busy: true, preview: "child work" });
  blocker.release.resolve("child answer");
  for (let i = 0;; i++) {
    if (i > 250) throw new Error("the settled subagent never left the feed");
    if (await summary(berry, child) === undefined) break;
    await new Promise((resolve) => setTimeout(resolve, 20));
  }
});