char-slop/berry

multipurpose self-modifiable agentic harness

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

Charlotte Sombump pi to 1.4.0f07649d

main
branchesmainall refs →
2.6 KiB72 linesraw
1import { BACKGROUND_CONTEXT as context } from "@earendil-works/chord/context";
2import {
3  type ConversationId,
4  defineDoc,
5  GenerationTask,
6  type Harness,
7  InboxDoc,
8  LiveDoc,
9} from "@earendil-works/pi-durable";
10
11/** The last time the conversation handed control back to the user. */
12export const SettledDoc = defineDoc<{ updated?: number }>({
13  kind: "berry.settled",
14  version: 1,
15  scope: "conversation",
16  history: "rewindable",
17  fork: "asOf",
18  initial: () => ({}),
19  checkpointWhen: () => true,
20});
21
22const pending = new WeakMap<Harness, Promise<void>>();
23
24/** Record when a run yields control instead of handing it to another generation. */
25export function recordSettled(harness: Harness): () => Promise<void> {
26  const queued = new Map<ConversationId, number>();
27  const unsubscribe = harness.subscribeCommits((publication) => {
28    for (const change of publication.changes) {
29      if (
30        change.type === "document" && change.record.kind === InboxDoc.definition.kind &&
31        change.conversationId !== undefined && Array.isArray(change.value?.items)
32      ) queued.set(change.conversationId, change.value.items.length);
33    }
34    for (const change of publication.changes) {
35      if (
36        change.type !== "task" || change.value.kind !== GenerationTask.definition.name ||
37        change.value.state.status !== "terminal"
38      ) continue;
39      const task = change.value;
40      const live = publication.changes.find((change) =>
41        change.type === "document" && change.record.kind === LiveDoc.definition.kind &&
42        change.conversationId === task.conversationId
43      );
44      if (live?.type !== "document" || live.value === null || live.value.run !== undefined) {
45        continue;
46      }
47      const count = queued.get(task.conversationId);
48      const next = (pending.get(harness) ?? Promise.resolve()).then(async () => {
49        const remaining = count ??
50          (await harness.snapshot(InboxDoc, task.conversationId, context))?.items.length ?? 0;
51        if (remaining > 0) return;
52        await harness.commit(async (tx) => {
53          (await tx.doc(SettledDoc, task.conversationId)).updated = task.endedAt ?? Date.now();
54        }, context);
55      }).catch((error) => console.error("[berry] recording settled turn:", error));
56      pending.set(harness, next);
57    }
58  });
59  return async () => {
60    unsubscribe();
61    await settledThrough(harness);
62  };
63}
64
65/** Wait for the timestamps queued by committed run endings. */
66export async function settledThrough(harness: Harness): Promise<void> {
67  let next;
68  do {
69    next = pending.get(harness);
70    await next;
71  } while (next !== pending.get(harness));
72}