char-slop/berry

multipurpose self-modifiable agentic harness

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

Charlotte Somadd terminal extension6334b24

main
branchesmainall refs →
2.5 KiB68 linesraw
1// A channel over a websocket: the first message authorizes it and carries its arguments; after that, text frames are
2// JSON and binary frames bytes, each way.
3
4import type { Channel, ChannelMessage, JsonValue } from "@berry/api";
5
6const AUTH_TIMEOUT_MS = 10_000;
7
8/** Why the server closed a channel it could not open, for the frontend's `messages` to fail with. */
9export const REFUSED = 4400;
10const UNAUTHORIZED = 4001;
11
12export type ChannelOptions = {
13  authorize(token: string): boolean;
14  /** Hand the channel to whoever opens it; a rejection closes it with the message. */
15  open(args: JsonValue | undefined, channel: Channel, signal: AbortSignal): Promise<void>;
16};
17
18export function serveChannel(socket: WebSocket, { authorize, open }: ChannelOptions): void {
19  socket.binaryType = "arraybuffer";
20  const closed = new AbortController();
21  let incoming!: ReadableStreamDefaultController<ChannelMessage>;
22  const messages = new ReadableStream<ChannelMessage>({
23    start: (controller) => void (incoming = controller),
24  });
25  const channel: Channel = {
26    send(message) {
27      if (socket.readyState !== WebSocket.OPEN) return;
28      socket.send(message instanceof Uint8Array ? message : JSON.stringify(message));
29    },
30    messages,
31    close: () => socket.close(1000),
32  };
33  let authorized = false;
34  const timeout = setTimeout(() => socket.close(UNAUTHORIZED, "unauthorized"), AUTH_TIMEOUT_MS);
35  socket.addEventListener("message", (event) => {
36    if (closed.signal.aborted) return;
37    let message: ChannelMessage;
38    try {
39      message = typeof event.data === "string"
40        ? JSON.parse(event.data)
41        : new Uint8Array(event.data as ArrayBuffer);
42    } catch {
43      return socket.close(REFUSED, "malformed message");
44    }
45    if (authorized) return incoming.enqueue(message);
46    clearTimeout(timeout);
47    const { token, args } = (message ?? {}) as { token?: unknown; args?: JsonValue };
48    if (typeof token !== "string" || !authorize(token)) {
49      return socket.close(UNAUTHORIZED, "unauthorized");
50    }
51    authorized = true;
52    open(args, channel, closed.signal).catch((error) => {
53      const reason = error instanceof Error ? error.message : String(error);
54      // a close reason is at most 123 bytes
55      const short = new TextDecoder().decode(new TextEncoder().encode(reason).slice(0, 120));
56      socket.close(REFUSED, short.replace(/\uFFFD$/, ""));
57    });
58  });
59  socket.addEventListener("close", () => {
60    clearTimeout(timeout);
61    closed.abort();
62    try {
63      incoming.close();
64    } catch {
65      // already ended
66    }
67  });
68}