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 →
4.3 KiB129 linesraw
1import type { Context, JsonValue } from "@earendil-works/chord";
2import {
3  defineDoc,
4  defineEntry,
5  defineTask,
6  type Harness,
7  type TaskRuntime,
8} from "@earendil-works/pi-durable";
9import type { ShellExecution, ShellState } from "@berry/api/shell.ts";
10
11export const ShellDoc = defineDoc<ShellState>({
12  kind: "berry.shell",
13  version: 1,
14  scope: "conversation",
15  history: "latest",
16  fork: "initial",
17  initial: () => ({}),
18  checkpointWhen: () => true,
19});
20export const ShellEntry = defineEntry<ShellExecution>("berry.shell");
21
22type Input = { command: string; excludeFromContext: boolean };
23type State = { phase: "run" };
24type Runtime = TaskRuntime<Input, State, null, object>;
25
26/** Passive shell execution: its result is a write, never an input that starts a model run. */
27export function shellTask(
28  execute: (
29    command: string,
30    runtime: Runtime,
31    context: Context,
32    progress: (details: JsonValue) => void,
33  ) => Promise<{ value: JsonValue | undefined; text: string }>,
34  harness: () => Harness,
35) {
36  async function finish(runtime: Runtime, context: Context, run: ShellExecution, text: string) {
37    const conversation = (await harness().conversation(runtime.conversationId, context))!;
38    await conversation.submit({
39      type: "write",
40      requestId: `shell:${runtime.taskId}`,
41      entry: {
42        kind: ShellEntry.kind,
43        data: run,
44        ...(run.excludeFromContext ? {} : {
45          model: [{
46            role: "user" as const,
47            content: `Ran \`${run.command}\`\n\n${text}`,
48            timestamp: run.startedAt,
49          }],
50        }),
51      },
52    }, context);
53    await runtime.commit(async (tx) => {
54      delete (await tx.doc(ShellDoc, runtime.conversationId)).run;
55      return {
56        status: "terminal",
57        outcome: run.status === "cancelled"
58          ? { status: "aborted" }
59          : { status: "completed", result: null },
60      };
61    }, context);
62  }
63
64  return defineTask<Input, State, null>({
65    name: "berry.shell",
66    version: 1,
67    initial: () => ({ phase: "run" }),
68    phases: {
69      async run(_, runtime, context) {
70        const run = structuredClone(
71          (await runtime.snapshot(ShellDoc, runtime.conversationId, context))?.run,
72        );
73        if (!run) throw new Error("Shell execution has no live state");
74        // shell side effects cannot safely be replayed after a restart
75        const invocation = crypto.randomUUID();
76        const started = await runtime.memo("started", invocation, context);
77        let pending = Promise.resolve();
78        let text = "";
79        try {
80          if (started !== invocation) {
81            throw new Error("Command interrupted by a restart; not rerun.");
82          }
83          const result = await execute(run.command, runtime, context, (details) => {
84            run.details = details;
85            pending = pending.then(() =>
86              runtime.commit(async (tx) => {
87                const live = (await tx.doc(ShellDoc, runtime.conversationId)).run;
88                if (live) live.details = details;
89              }, context)
90            );
91            void pending.catch(() => {});
92          });
93          run.result = result.value ?? null;
94          run.status = "ok";
95          text = result.text;
96        } catch (error) {
97          if (runtime.signal.aborted) throw error;
98          run.status = "error";
99          run.error = error instanceof Error ? error.message : String(error);
100          text = `${output(run)}\n\n${run.error}`;
101        }
102        await pending;
103        run.durationMs = runtime.now() - run.startedAt;
104        await finish(runtime, context, run, text);
105      },
106    },
107    async abort(_, runtime, context) {
108      const run = structuredClone(
109        (await runtime.snapshot(ShellDoc, runtime.conversationId, context))?.run,
110      );
111      if (!run) {
112        await runtime.commit(
113          () => ({ status: "terminal", outcome: { status: "aborted" } }),
114          context,
115        );
116        return;
117      }
118      run.status = "cancelled";
119      run.error = "Command cancelled.";
120      run.durationMs = runtime.now() - run.startedAt;
121      await finish(runtime, context, run, `${output(run)}\n\n(command cancelled)`);
122    },
123  });
124}
125
126function output(run: ShellExecution): string {
127  const details = run.details as { rawOutput?: string } | null;
128  return details?.rawOutput ?? "(no output)";
129}