char-slop/berry

multipurpose self-modifiable agentic harness

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

Charlotte Sommcp: add streaming fetch on conversation-local and remote environmentse5ec380

main
branchesmainall refs →
3.7 KiB115 linesraw
1// What berry and an environment's worker say to each other: one json value per line, with bytes and the
2// environment's errors spelled out so that they arrive as what they were.
3
4import { Buffer } from "node:buffer";
5import { ExecutionError, FileError } from "@earendil-works/pi-durable/env";
6
7/** Bumped whenever either side's messages change shape. */
8export const PROTOCOL = 3;
9
10/** What the worker says first: the machine it found itself on. */
11export type Hello = {
12  readonly protocol: number;
13  readonly platform: string;
14  readonly home: string;
15  readonly variables: Record<string, string>;
16};
17
18export type FetchOptions = {
19  method: string;
20  headers: [string, string][];
21  body?: Uint8Array;
22  redirect: RequestRedirect;
23  credentials: RequestCredentials;
24  cache: RequestCache;
25  referrer: string;
26  referrerPolicy: ReferrerPolicy;
27  integrity: string;
28};
29
30export type FetchResponse = {
31  status: number;
32  statusText: string;
33  headers: [string, string][];
34  url: string;
35  redirected: boolean;
36  handle?: number;
37};
38
39export type Request =
40  /** Call `m` with `a`, resolving paths against `cwd`. */
41  | {
42    readonly i: number;
43    readonly m: string;
44    readonly a: readonly unknown[];
45    readonly cwd?: string;
46  }
47  /** Abort the call `c`. */
48  | { readonly c: number };
49
50export type Response =
51  | { readonly hello: Hello }
52  /** The call `i` settled with `r`. */
53  | { readonly i: number; readonly r: unknown }
54  /** The call `i` threw, which the environment's methods never mean to. */
55  | { readonly i: number; readonly e: string }
56  /** Something call `i` streams, while it runs or for as long as what it opened stays open. */
57  | { readonly i: number; readonly o: unknown };
58
59export function encode(message: Request | Response): string {
60  return JSON.stringify(message, function (this: Record<string, unknown>, key, converted) {
61    // what toJSON() made of it, for buffers an array of numbers
62    const value = this[key];
63    if (value instanceof Uint8Array) {
64      return {
65        $bytes: Buffer.from(value.buffer, value.byteOffset, value.byteLength).toString("base64"),
66      };
67    }
68    if (value instanceof FileError) {
69      return { $file: { code: value.code, message: value.message, path: value.path } };
70    }
71    if (value instanceof ExecutionError) {
72      return { $exec: { code: value.code, message: value.message, spillPath: value.spillPath } };
73    }
74    return converted;
75  });
76}
77
78export function decode(line: string): Request | Response {
79  return JSON.parse(line, (_, value) => {
80    if (typeof value !== "object" || value === null) return value;
81    if (typeof value.$bytes === "string") {
82      return new Uint8Array(Buffer.from(value.$bytes, "base64"));
83    }
84    if (value.$file) return new FileError(value.$file.code, value.$file.message, value.$file.path);
85    if (value.$exec) {
86      const error = new ExecutionError(value.$exec.code, value.$exec.message);
87      if (value.$exec.spillPath !== undefined) error.spillPath = value.$exec.spillPath;
88      return error;
89    }
90    return value;
91  });
92}
93
94/** Lines of a byte stream, as they complete. */
95export async function* lines(
96  reader: ReadableStreamDefaultReader<Uint8Array>,
97  buffered: Uint8Array = new Uint8Array(),
98): AsyncGenerator<string> {
99  const decoder = new TextDecoder();
100  // chunks of the line in progress, which can be a whole file's worth
101  let parts: string[] = [];
102  let text = decoder.decode(buffered, { stream: true });
103  while (true) {
104    let start = 0;
105    for (let end; (end = text.indexOf("\n", start)) >= 0; start = end + 1) {
106      parts.push(text.slice(start, end));
107      yield parts.join("");
108      parts = [];
109    }
110    parts.push(text.slice(start));
111    const { value, done } = await reader.read();
112    if (done) return;
113    text = decoder.decode(value, { stream: true });
114  }
115}