char-slop/berry

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

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

// What the list of conversations shows of each beyond its index entry: whether it is answering, and how it began.

import { BACKGROUND_CONTEXT as context } from "@earendil-works/chord/context";
import { homedir } from "node:os";
import { type Binding, ENVIRONMENT } from "./environment.ts";
import { EventEntry } from "./entries.ts";
import { imageSummary } from "@berry/api/images.ts";
import {
  type ConversationId,
  type Cursor,
  type EntryId,
  InboxDoc,
  LiveDoc,
} from "@earendil-works/pi-durable";
import { type Berry, type ConversationInfo, Conversations, headOf } from "./berry.ts";
import { type Feed, published } from "./pubsub.ts";
import { clip, textOf, userText } from "./tree.ts";
import { SettledDoc, settledThrough } from "./settled.ts";

export type Summary = {
  /** Whether it is generating or running tools. */
  busy: boolean;
  /** Its first message, on one line; empty before there is one. */
  preview: string;
  /** Its newest answer, on one line; empty before one has arrived. */
  reply: string;
  /** A settled turn whose newest message has not been viewed. */
  unread: boolean;
  /** Its newest user message or return of control to the user, or creation time before either. */
  updated: number;
  environment: string;
  cwd: string;
};

/** Every listed conversation's summary as one feed, kept up as commits land. */
export async function summaries(berry: Berry): Promise<{ feed(): Feed; stop(): void }> {
  const { harness } = berry;
  const value = published<Record<string, Summary>>({});
  // a branch's first message never changes once it has one
  const previews = new Map<ConversationId, string>();

  async function previewOf(id: ConversationId): Promise<string> {
    const known = previews.get(id);
    if (known !== undefined) return known;
    const conversation = await harness.conversation(id, context);
    let cursor: Cursor | undefined;
    do {
      const page = await conversation!.entries({ order: "ascending" }, 16, cursor, context);
      const first = page.items.find((entry) => entry.kind === "pi.user")?.model?.[0];
      if (first?.role === "user") {
        const text = clip(
          160,
          [userText(first.content), imageSummary(first.content)].filter(Boolean).join(" "),
        );
        previews.set(id, text);
        return text;
      }
      cursor = page.next;
    } while (cursor !== undefined);
    return "";
  }

  async function busyOf(head: ConversationId) {
    return (await harness.snapshot(LiveDoc, head, context))?.run !== undefined ||
      ((await harness.snapshot(InboxDoc, head, context))?.items.length ?? 0) > 0;
  }

  async function summarize(info: ConversationInfo): Promise<Summary> {
    await settledThrough(harness);
    const head = headOf(info);
    const settled = await harness.snapshot(SettledDoc, head, context);
    const busy = await busyOf(head);
    let unread = false;
    let updated = settled?.updated ?? info.created;
    const conversation = (await harness.conversation(head, context))!;
    // scanned on its own, since the loop below stops as soon as it has what it needs of the branch
    const replyOf = async () => {
      const page = await conversation.entries({ order: "descending" }, 16, undefined, context);
      for (const entry of page.items) {
        const message = entry.model?.[0];
        if (
          entry.kind === "pi.assistant" && message?.role === "assistant" &&
          message.stopReason !== "toolUse"
        ) {
          return clip(160, textOf(message.content));
        }
      }
      return "";
    };
    let cursor: Cursor | undefined;
    let boundary: EntryId | undefined;
    let found = false;
    let timeFound = settled !== undefined && head === info.id && info.parent === undefined;
    let userTime: number | undefined;
    let userFound = false;
    do {
      const page = await conversation.entries(
        { order: "descending" },
        16,
        cursor,
        context,
      );
      for (const entry of page.items) {
        // match the active transcript: a reset must not resurrect unread older messages
        boundary ??= entry.head;
        const message = entry.model?.[0];
        if (
          !timeFound && entry.kind === "pi.assistant" && message?.role === "assistant" &&
          message.stopReason !== "toolUse"
        ) {
          if (settled === undefined || entry.conversationId !== head) {
            // a fork at the answer precedes the separate commit that records its completion
            const task = entry.byTaskId === undefined
              ? undefined
              : await harness.getTask(entry.byTaskId, context);
            updated = Math.max(settled?.updated ?? 0, task?.endedAt ?? message.timestamp);
          }
          timeFound = true;
        }
        if (!userFound && entry.kind === "pi.user" && message?.role === "user") {
          userTime = message.timestamp;
          userFound = true;
        }
        if (!found && (entry.kind === "pi.assistant" || entry.kind === "pi.user")) {
          unread = !busy && (boundary === undefined || entry.id >= boundary) &&
            entry.id > (info.readThrough ?? 0);
          found = true;
        }
        if (found && timeFound && userFound) break;
      }
      cursor = page.next;
    } while ((!found || !timeFound || !userFound) && cursor !== undefined);
    if (userTime !== undefined) updated = Math.max(updated, userTime);
    const binding = await berry.state(await berry.tidOf(head), ENVIRONMENT) as Binding;
    return {
      busy,
      preview: await previewOf(head),
      reply: await replyOf(),
      unread,
      updated,
      environment: binding.environment,
      cwd: binding.cwd ?? homedir(),
    };
  }

  // what changed since the last refresh: some conversations' ids, or everything
  let dirty: Set<ConversationId> | "all" = "all";
  // each summary's index entry as last summarized; an index commit restales only those it changed, as a read receipt
  const summarized = new Map<string, string>();
  let reindexed = false;
  let refreshing = false;
  let stopped = false;
  async function refresh() {
    if (refreshing || stopped) return;
    refreshing = true;
    try {
      while (!stopped && (reindexed || dirty === "all" || dirty.size > 0)) {
        const changed = dirty;
        dirty = new Set();
        reindexed = false;
        const index = (await harness.snapshot(Conversations, context)) ?? {};
        // an owned conversation shows while it is busy, then leaves the feed: check only
        // whether it is working before doing its full summarize, so idle history cannot pile up
        const listed: [string, ConversationInfo][] = [];
        for (const [tid, info] of Object.entries(index)) {
          const head = headOf(info);
          const touched = value.value[tid] !== undefined || changed === "all" || reindexed ||
            changed.has(head);
          if (info.owner === undefined || (touched && await busyOf(head))) listed.push([tid, info]);
        }
        const next: Record<string, Summary> = {};
        for (const [tid, info] of listed) {
          const previous = value.value[tid];
          const entry = JSON.stringify(info);
          const stale = changed === "all" || previous === undefined ||
            summarized.get(tid) !== entry || changed.has(headOf(info));
          next[tid] = stale ? await summarize(info) : previous;
          summarized.set(tid, entry);
        }
        for (const tid of summarized.keys()) if (!(tid in next)) summarized.delete(tid);
        value.set(next);
      }
    } catch (error) {
      if (!stopped) console.error("[berry] summaries:", error);
    } finally {
      refreshing = false;
    }
  }

  const live = LiveDoc.definition.kind;
  const inbox = InboxDoc.definition.kind;
  const settled = SettledDoc.definition.kind;
  const index = Conversations.definition.kind;
  // commit listeners must not call Session APIs inline
  const unsubscribe = harness.subscribeCommits((publication) => {
    let touched = false;
    for (const change of publication.changes) {
      if (change.type === "document" && change.record.kind === index) reindexed = true;
      else if (
        change.type === "document" &&
        (change.record.kind === live || change.record.kind === inbox ||
          change.record.kind === settled) &&
        change.conversationId
      ) {
        if (dirty !== "all") dirty.add(change.conversationId);
      } else if (
        change.type === "entry" &&
        (change.value.kind === "pi.user" || change.value.kind === "pi.assistant" ||
          change.value.head !== undefined ||
          (EventEntry.is(change.value) &&
            (change.value.data.ext === ENVIRONMENT ||
              ((change.value.data.ext === "code" || change.value.data.ext === "coding") &&
                typeof change.value.data.event === "object" && change.value.data.event !== null &&
                !Array.isArray(change.value.data.event) && change.value.data.event.type === "cd"))))
      ) {
        if (dirty !== "all") dirty.add(change.value.conversationId);
      } else continue;
      touched = true;
    }
    if (touched) queueMicrotask(() => void refresh());
  });
  await refresh();
  return {
    feed: () => value.feed(),
    stop() {
      stopped = true;
      unsubscribe();
    },
  };
}