char-slop/berry

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

Charlotte Somcut a bunch of code volumecb117252026-10-09

import {
  assert,
  assertEquals,
  assertGreater,
  assertObjectMatch,
  assertRejects,
  assertStrictEquals,
  assertStringIncludes,
} from "@std/assert";
import { BACKGROUND_CONTEXT as context, withAbortSignal } from "@earendil-works/chord/context";
import { fauxAssistantMessage, fauxToolCall } from "@earendil-works/pi-ai/providers/faux";
import { getOrThrow } from "@earendil-works/pi-durable/env";
import { registerEnvConformance } from "@earendil-works/pi-durable/testing";
import code from "../../extensions/code/mod.ts";
import { configuredEnvironments, type EnvironmentSettings } from "../environment.ts";
import { lastToolResult, sectionsSeen, setup, turn } from "../fixture.ts";
import { BOOTSTRAP, Connection, RemoteEnvironment } from "./connection.ts";

/** The worker on this machine, as ssh would start it on another, telling itself apart by `BERRY_MACHINE`. */
const here = (name: string) =>
  Connection.open(name, [
    "env",
    `BERRY_MACHINE=${name}`,
    Deno.execPath(),
    "eval",
    "--quiet",
    BOOTSTRAP,
  ]);

Deno.test("the bootstrap passes through any remote shell's double quotes as it is", () => {
  // sh, cmd, and powershell each give one of these a meaning inside double quotes
  for (const special of ['"', "$", "`", "\\", "%", "!", "\n"]) {
    assert(!BOOTSTRAP.includes(special), `the bootstrap holds ${JSON.stringify(special)}`);
  }
});

/** Cases a local NodeExecutionEnv fails as well when deno runs it, which a worker cannot do better at. */
const UPSTREAM = new Set(["watch follows a directory replaced at the same path"]);

{
  let connection: Connection | undefined;
  const cases: { name: string; test: () => Promise<void> }[] = [];
  registerEnvConformance(
    {
      describe(name, suite) {
        suite();
        // one worker for every case, as a conversation's calls share one
        Deno.test(name, async (t) => {
          connection = await here("conformance");
          try {
            for (const { name, test } of cases) {
              await t.step({ name, fn: test, ignore: UPSTREAM.has(name) });
            }
          } finally {
            await connection.close();
          }
        });
      },
      it: (name, test) => void cases.push({ name, test }),
      expect: (actual) => ({
        toBe: (expected: unknown) => assertStrictEquals(actual, expected),
        toBeGreaterThan: (expected: number) => assertGreater(actual as number, expected),
        toEqual: (expected: unknown) => assertEquals(actual, expected),
        toMatchObject: (expected: Record<PropertyKey, unknown>) =>
          assertObjectMatch(actual as Record<PropertyKey, unknown>, expected),
        toBeTruthy: () => assert(actual),
        rejects: {
          toThrow: (expected?: string) =>
            assertRejects(() => actual as Promise<unknown>, Error, expected),
        },
      }),
    },
    "a remote environment behaves as a local one",
    async (use) => {
      const root = new RemoteEnvironment("conformance", connection!, "/");
      const dir = getOrThrow(await root.createTempDir("berry-conformance-", context));
      try {
        await use(new RemoteEnvironment("conformance", connection!, dir));
      } finally {
        await root.remove(dir, { recursive: true, force: true }, context);
      }
    },
  );
}

Deno.test("each machine keeps one connection, until it is lost or its settings change", async () => {
  let settings: Record<string, EnvironmentSettings> = { far: { ssh: "far" } };
  const started: Connection[] = [];
  const environments = configuredEnvironments(
    () => Promise.resolve(settings),
    async (name) => {
      const connection = await here(name);
      started.push(connection);
      return connection;
    },
  );
  try {
    const [first, second] = await Promise.all([
      environments.open("far", null, context),
      environments.open("far", "/", context),
    ]);
    assertEquals(started.length, 1);
    assertEquals([first.cwd, second.cwd], [first.home, "/"]);
    assertEquals(first.variables.BERRY_MACHINE, "far");

    await started[0].close();
    const lost = await first.fileInfo("/", context);
    assert(!lost.ok);
    assertStringIncludes(lost.error.message, "far: lost");
    assert((await (await environments.open("far", null, context)).fileInfo("/", context)).ok);
    assertEquals(started.length, 2);

    settings = { far: { ssh: "far", deno: "deno" } };
    await environments.open("far", null, context);
    assertEquals(started.length, 3);

    settings = {};
    await assertRejects(
      () => environments.open("far", null, context),
      Error,
      "No environment named far",
    );
    assertEquals((await environments.open("local", null, context)).name, "local");
  } finally {
    await environments.close();
  }
});

Deno.test("a worker that cannot start says why", async () => {
  await assertRejects(
    () => Connection.open("nowhere", ["sh", "-c", "echo no deno here >&2; exit 127"]),
    Error,
    "no deno here",
  );
});

Deno.test("aborting a call stops the command on the other machine", async () => {
  const connection = await here("far");
  try {
    const env = new RemoteEnvironment("far", connection, "/");
    const controller = new AbortController();
    const started = performance.now();
    setTimeout(() => controller.abort(), 100);
    const result = await env.exec(
      "sleep 30",
      undefined,
      withAbortSignal(controller.signal, context),
    );
    assert(!result.ok);
    assertEquals(result.error.code, "aborted");
    assert(performance.now() - started < 10_000);
  } finally {
    await connection.close();
  }
});

Deno.test("a conversation bound to a configured machine works there", async () => {
  const environments = configuredEnvironments(
    () => Promise.resolve({ far: { ssh: "far" } }),
    (name) => here(name),
  );
  await using harness = await setup(environments);
  const { berry, model } = harness;
  await berry.install(code);
  const tid = await berry.create({
    extensions: { environment: { environment: "far" }, code: {} },
    model,
  });
  const home = (await berry.environment(tid)).home;
  const seen = await turn(harness, tid, [
    fauxAssistantMessage(
      fauxToolCall("bash", { command: "echo $BERRY_MACHINE; pwd", timeout: 10 }),
      {
        stopReason: "toolUse",
      },
    ),
  ]);
  assertStringIncludes(lastToolResult(seen[1]).text, `far\n${home}`);
  const sections = sectionsSeen(seen[0])[0];
  assertStringIncludes(sections.code!, `current working directory: ${home}`);
  assertStringIncludes(sections.environment!, "files and commands are on far");
});

Deno.test("programs and sockets on the other machine speak both ways, and end with it", async () => {
  const connection = await here("far");
  const env = new RemoteEnvironment("far", connection, "/");
  const echo = Deno.listen({ hostname: "127.0.0.1", port: 0 });
  const echoed = (async () => {
    const accepted = await echo.accept();
    await accepted.readable.pipeTo(accepted.writable);
  })();
  try {
    const cat = await env.spawn(["cat"], {}, context);
    const writer = cat.stdin.getWriter();
    await writer.write(new TextEncoder().encode("hello"));
    await writer.close();
    assertEquals(await new Response(cat.stdout).text(), "hello");
    assertEquals(await cat.exited, 0);

    const told = await env.spawn(["sh", "-c", "echo $BERRY_MACHINE $THERE; exit 3"], {
      env: { THERE: "too" },
    }, context);
    assertEquals(await new Response(told.stdout).text(), "far too\n");
    assertEquals(await told.exited, 3);

    await assertRejects(() => env.spawn(["berry-no-such-program"], {}, context));
    const sleeping = await env.spawn(["sleep", "30"], {}, context);
    sleeping.kill();
    assertEquals(await sleeping.exited, null);

    // the worker's loopback, which here is this machine's
    const socket = await env.connect("127.0.0.1", (echo.addr as Deno.NetAddr).port, context);
    const socketWriter = socket.writable.getWriter();
    await socketWriter.write(new TextEncoder().encode("ping"));
    const { value } = await socket.readable.getReader().read();
    assertEquals(new TextDecoder().decode(value), "ping");
    socket.close();
    await echoed;

    const orphan = await env.spawn(["sleep", "30"], {}, context);
    await connection.close();
    assertEquals(await orphan.exited, null);
    await assertRejects(() => new Response(orphan.stdout).text(), Error, "far: lost");
  } finally {
    echo.close();
    await connection.close();
  }
});