runa/vscode/test/client.test.ts

import assert from "node:assert/strict";
import { once } from "node:events";
import test from "node:test";
import { ImageClient, parseEndpoint } from "../src/client";
import { MAX_MESSAGE_BYTES, ProtocolError, STEP_MODES } from "../src/protocol";
import { actor, fakeImage, json, notification, REF, response } from "./fixtures";

test("only numeric loopback endpoints and valid ports are accepted", () => {
  assert.deepEqual(parseEndpoint("127.0.0.1:3737"), { host: "127.0.0.1", port: 3737 });
  assert.deepEqual(parseEndpoint("127.2.3.4:1"), { host: "127.2.3.4", port: 1 });
  assert.deepEqual(parseEndpoint("[::1]:65535"), { host: "::1", port: 65535 });
  for (const input of [
    "localhost:3737",
    "0.0.0.0:3737",
    "192.168.1.1:3737",
    "[::]:3737",
    "[::ffff:127.0.0.1]:3737",
    "example.org:3737",
    "127.0.0.1:0",
    "127.0.0.1:65536",
    "127.0.0.1:-1",
    "127.0.0.1:1/path",
  ]) {
    assert.throws(() => parseEndpoint(input));
  }
});

test("image snapshot commands validate server-local paths and propagate failures", async (t) => {
  const operations: string[] = [];
  let firstSave = true;
  const address = await fakeImage(t, (request, socket) => {
    operations.push(request.command.operation);
    assert.equal(request.command.parameters?.file, "/server/image.json");
    if (request.command.operation === "save_image")
      assert.equal(typeof request.command.parameters?.overwrite, "boolean");
    if (
      request.command.operation === "save_image" &&
      !request.command.parameters?.overwrite &&
      firstSave
    ) {
      firstSave = false;
      socket.send(
        json({
          version: 5,
          request_id: request.request_id,
          result: { Err: { code: "conflict", message: "snapshot destination already exists" } },
        }),
      );
    } else if (request.command.operation === "restore_image")
      socket.send(
        json({
          version: 5,
          request_id: request.request_id,
          result: { Err: { code: "runtime_failure", message: "unfinished handler" } },
        }),
      );
    else socket.send(response(request, { kind: "acknowledged" }));
  });
  const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined));
  t.after(() => client.close());
  for (const file of ["", " ", "\0", "a".repeat(4097), "é".repeat(2049)]) {
    await assert.rejects(client.saveImage(file), /server-local path/);
    await assert.rejects(client.restoreImage(file), /server-local path/);
  }
  assert.deepEqual(operations, []);
  await assert.rejects(
    client.saveImage("/server/image.json"),
    /snapshot destination already exists/,
  );
  await client.saveImage("/server/image.json", true);
  await assert.rejects(client.restoreImage("/server/image.json"), /unfinished handler/);
  assert.deepEqual(operations, ["save_image", "save_image", "restore_image"]);
});

test("source step modes are sent explicitly and retain the default into behavior", async (t) => {
  const modes: unknown[] = [];
  const address = await fakeImage(t, (request, socket) => {
    assert.equal(request.command.operation, "step_source");
    modes.push(request.command.parameters?.mode);
    socket.send(response(request, { kind: "debug_step_completed", data: { reason: "step" } }));
  });
  const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined));
  t.after(() => client.close());
  assert.equal(await client.stepSource(REF, "impl"), "step");
  for (const mode of STEP_MODES)
    assert.equal(await client.stepSource(REF, "impl", 100, mode), "step");
  assert.deepEqual(modes, ["into", "into", "over", "out"]);
});

test("debugger arguments reject invalid budgets, identities and line lists before transport", async (t) => {
  let requests = 0;
  const address = await fakeImage(t, (request, socket) => {
    requests++;
    socket.send(response(request, { kind: "actors", data: [] }));
  });
  const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined));
  t.after(() => client.close());
  for (const budget of [0, -1, 10001, 1.5, NaN])
    await assert.rejects(client.stepSource(REF, "impl", budget), /Invalid actor/);
  for (const lines of [[-1], [0.5], [4294967296], Array(129).fill(0), [NaN]])
    await assert.rejects(client.setBreakpoints(REF, "impl", lines), /Invalid actor/);
  await assert.rejects(client.setBreakpoints(REF, "", []), /Invalid actor/);
  await assert.rejects(client.stepSource("bad", "impl"), /Invalid actor/);
  assert.equal(requests, 0);
  assert.deepEqual(await client.listActors(), []);
});

test("normal disconnect completes a code-1000 close handshake and rejects further work", async (t) => {
  let closed: Promise<unknown[]> | undefined;
  const address = await fakeImage(t, (request, socket) => {
    closed = once(socket, "close");
    socket.send(response(request, { kind: "actors", data: [] }));
  });
  let disconnects = 0;
  const client = await ImageClient.connect(address, (error) => {
    assert.equal(error, undefined);
    disconnects++;
  });
  await client.listActors();
  await client.close();
  assert.ok(closed);
  assert.equal((await closed)[0], 1000);
  await client.close();
  assert.equal(disconnects, 1);
  await assert.rejects(client.listActors(), /Not connected/);
});

test("fragmented UTF-8 replies and simultaneous reads preserve correlation and values", async (t) => {
  const operations: string[] = [];
  const address = await fakeImage(t, (request, socket) => {
    operations.push(request.command.operation);
    assert.equal(request.version, 5);
    const reply =
      request.command.operation === "list_actors"
        ? { kind: "actors", data: [actor()] }
        : { kind: "actor", data: actor() };
    const bytes = Buffer.from(response(request, reply));
    const split = bytes.indexOf(Buffer.from("雪")) + 1;
    socket.send(bytes.subarray(0, split), { binary: false, fin: false });
    setImmediate(() => socket.send(bytes.subarray(split), { binary: false, fin: true }));
  });
  const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined));
  t.after(() => client.close());
  const [actors, inspected] = await Promise.all([client.listActors(), client.inspectActor(REF)]);
  assert.deepEqual(actors, [actor()]);
  assert.deepEqual(inspected, actor());
  assert.deepEqual(operations, ["list_actors", "inspect_actor"]);
});

test("protocol errors preserve the connection and are never retried", async (t) => {
  let requests = 0;
  const address = await fakeImage(t, (request, socket) => {
    requests++;
    socket.send(
      requests === 1
        ? json({
            version: 5,
            request_id: request.request_id,
            result: { Err: { code: "not_found", message: "Missing actor" } },
          })
        : response(request, { kind: "actors", data: [] }),
    );
  });

  test("outgoing mutations enforce the exact UTF-8 envelope limit without closing the connection", async (t) => {
    const sizes: number[] = [];
    const address = await fakeImage(t, (request, socket) => {
      sizes.push(Buffer.byteLength(json(request)));
      socket.send(
        response(
          request,
          request.command.operation === "send"
            ? { kind: "acknowledged" }
            : { kind: "actors", data: [] },
        ),
      );
    });
    const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined));
    t.after(() => client.close());
    const overhead = Buffer.byteLength(
      json({
        version: 5,
        request_id: "runa-vscode-1",
        command: {
          operation: "send",
          parameters: { actor_ref: REF, message: { type: "string", data: "" } },
        },
      }),
    );
    const available = MAX_MESSAGE_BYTES - overhead;
    const data = "雪".repeat(Math.floor(available / 3)) + "x".repeat(available % 3);
    await client.send(REF, { type: "string", data });
    await assert.rejects(client.send(REF, { type: "string", data: data + "x" }), /Request exceeds/);
    await client.listActors();
    assert.equal(sizes[0], MAX_MESSAGE_BYTES);
    assert.equal(sizes.length, 2);
  });

  test("a lost mutation response rejects queued mutations without retrying", async (t) => {
    let requests = 0;
    const address = await fakeImage(t, (_request, socket) => {
      requests++;
      socket.terminate();
    });
    const client = await ImageClient.connect(address, () => undefined);
    t.after(() => client.close());
    const results = await Promise.allSettled([
      client.spawn("actors/counter", { type: "integer", data: 0n }),
      client.send(REF, { type: "null" }),
    ]);
    assert.ok(results.every((result) => result.status === "rejected"));
    assert.equal(requests, 1);
  });
  const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined));
  t.after(() => client.close());
  await assert.rejects(client.inspectActor(REF), ProtocolError);
  assert.deepEqual(await client.listActors(), []);
  assert.equal(requests, 2);
});

test("bad correlation, malformed JSON/UTF-8, oversize, truncation, and wrong versions close the stream", async (t) => {
  for (const mode of [
    "correlation",
    "json",
    "utf8",
    "size",
    "truncated",
    "version",
    "kind",
    "binary",
    "event",
  ]) {
    await t.test(mode, async (t) => {
      let requests = 0;
      const failures: Error[] = [];
      const address = await fakeImage(t, (request, socket) => {
        requests++;
        if (mode === "json") socket.send("{broken}");
        else if (mode === "utf8") socket.send(Buffer.from([0xff]), { binary: false });
        else if (mode === "size") socket.send(" ".repeat(MAX_MESSAGE_BYTES + 1));
        else if (mode === "truncated") {
          socket.send('{"version":3', { fin: false });
          socket.terminate();
        } else if (mode === "binary") socket.send(Buffer.from("{}"));
        else if (mode === "event") socket.send(json({ version: 5, event: { event: "unknown" } }));
        else
          socket.send(
            json({
              version: mode === "version" ? 2 : 3,
              request_id: mode === "correlation" ? "wrong" : request.request_id,
              result: { Ok: { kind: mode === "kind" ? "acknowledged" : "actors", data: [] } },
            }),
          );
      });
      const client = await ImageClient.connect(address, (error) => {
        if (error) failures.push(error);
      });
      t.after(() => client.close());
      await assert.rejects(client.listActors());
      await assert.rejects(client.listActors(), /Not connected/);
      assert.equal(requests, 1);
      assert.equal(failures.length, 1);
    });
  }
});

test("request timeout rejects pending and queued reads without retry", async (t) => {
  let requests = 0;
  const address = await fakeImage(t, () => {
    requests++;
  });
  const client = await ImageClient.connect(
    address,
    () => {},
    () => {},
    { connect: 1000, request: 30 },
  );
  t.after(() => client.close());
  await Promise.all([
    assert.rejects(client.listActors(), /timed out/),
    assert.rejects(client.inspectActor(REF), /Not connected/),
  ]);
  assert.equal(requests, 1);
});

test("the 1 MiB response limit is inclusive and framing survives the next request", async (t) => {
  const address = await fakeImage(t, (request, socket) => {
    const line = response(request, { kind: "actors", data: [] });
    socket.send(line + " ".repeat(MAX_MESSAGE_BYTES - Buffer.byteLength(line)));
  });
  const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined));
  t.after(() => client.close());
  assert.deepEqual(await client.listActors(), []);
  assert.deepEqual(await client.listActors(), []);
});

test("unsolicited responses invalidate the connection", async (t) => {
  const failures: Error[] = [];
  const address = await fakeImage(t, (request, socket) => {
    socket.send(response(request, { kind: "actors", data: [] }));
    socket.send(response(request, { kind: "actors", data: [] }));
  });
  const client = await ImageClient.connect(address, (error) => {
    if (error) failures.push(error);
  });
  t.after(() => client.close());
  assert.deepEqual(await client.listActors(), []);
  await assert.rejects(client.listActors(), /Not connected/);
  assert.match(failures[0]?.message ?? "", /unsolicited/);
});

test("disconnect rejects pending reads and idle peer close is reported", async (t) => {
  const address = await fakeImage(t, (_request, socket) => socket.close());
  let disconnected = false;
  const client = await ImageClient.connect(address, (error) => {
    assert.match(error?.message ?? "", /closed/);
    disconnected = true;
  });
  t.after(() => client.close());
  await assert.rejects(client.listActors(), /closed/);
  assert.equal(disconnected, true);

  const idleAddress = await fakeImage(t, (request, socket) => {
    socket.send(response(request, { kind: "actors", data: [] }));
    setImmediate(() => socket.close());
  });
  let onClosed = () => {};
  const closed = new Promise<void>((resolve) => {
    onClosed = resolve;
  });
  const idle = await ImageClient.connect(idleAddress, () => onClosed());
  t.after(() => idle.close());
  await idle.listActors();
  await closed;
  await assert.rejects(idle.listActors(), /Not connected/);
});

test("requires the runa-debug WebSocket subprotocol", async (t) => {
  const address = await fakeImage(t, () => {}, "wrong-protocol");
  await assert.rejects(
    ImageClient.connect(address, () => {}),
    /subprotocol/,
  );
});

test("notifications interleave with replies and arrive when no request is pending", async (t) => {
  const revisions: bigint[] = [];
  let onIdle = () => {};
  const idle = new Promise<void>((resolve) => {
    onIdle = resolve;
  });
  const address = await fakeImage(t, (request, socket) => {
    socket.send(notification(1n));
    socket.send(response(request, { kind: "actors", data: [] }));
    setImmediate(() => socket.send(notification(5n)));
  });
  const client = await ImageClient.connect(
    address,
    (error) => assert.equal(error, undefined),
    (revision) => {
      revisions.push(revision);
      if (revision === 5n) onIdle();
    },
  );
  t.after(() => client.close());
  assert.deepEqual(await client.listActors(), []);
  await idle;
  assert.deepEqual(revisions, [1n, 5n]);
});

test("repeated or backwards revisions close the connection", async (t) => {
  const failures: Error[] = [];
  const address = await fakeImage(t, (_request, socket) => {
    socket.send(notification(5n));
    socket.send(notification(5n));
  });
  const client = await ImageClient.connect(address, (error) => {
    if (error) failures.push(error);
  });
  t.after(() => client.close());
  await assert.rejects(client.listActors(), /Non-increasing/);
  assert.equal(failures.length, 1);
});
test("failed connection reports the error", async () => {
  const { createServer } = await import("node:net");
  const server = createServer();
  server.listen(0, "127.0.0.1");
  await once(server, "listening");
  const address = server.address();
  if (!address || typeof address === "string") throw new Error("Missing address");
  await new Promise<void>((resolve) => server.close(() => resolve()));
  await assert.rejects(
    ImageClient.connect(`127.0.0.1:${address.port}`, () => {}),
    /ECONNREFUSED/,
  );
});

test("source and system commands use the exact v5 wire contract", async (t) => {
  const requests: unknown[] = [];
  const file = { path: "actors/counter", source_id: "source:2", source: "pub fn handle() {}\n" };
  const address = await fakeImage(t, (request, socket) => {
    requests.push(request.command);
    const replies: Record<string, unknown> = {
      list_sources: { kind: "sources", data: [file] },
      read_source: { kind: "source", data: file },
      write_source: { kind: "source", data: file },
      rollback_source: { kind: "source", data: file },
      commit_source: { kind: "source_committed", data: { revision: "r1", source_id: "source:2" } },
      source_history: { kind: "source_history", data: [{ revision: "r1", source_id: "source:2" }] },
      revision_source: { kind: "behavior_source", data: { source: file.source } },
      system: { kind: "acknowledged" },
      spawn: { kind: "actor_spawned", data: { actor_ref: REF } },
    };
    socket.send(response(request, replies[request.command.operation]));
  });
  const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined));
  t.after(() => client.close());
  assert.deepEqual(await client.listSources(), [file]);
  assert.deepEqual(await client.readSource(file.path), file);
  assert.deepEqual(await client.writeSource(file.path, null, file.source), file);
  assert.deepEqual(await client.writeSource(file.path, "source:1", file.source), file);
  assert.deepEqual(await client.commitSource(file.path), { revision: "r1", source_id: "source:2" });
  assert.equal((await client.sourceHistory(file.path)).length, 1);
  assert.equal(await client.revisionSource(file.path, "r1"), file.source);
  assert.deepEqual(await client.rollbackSource(file.path, "r1"), file);
  await client.system(REF, { kind: "stop" });
  await client.system(REF, { kind: "become", path: "actors/other" });
  assert.equal(await client.spawn(file.path, { type: "null" }), REF);
  for (const invalid of [
    () => client.readSource("actors/counter.rn"),
    () => client.readSource("source/actors/counter"),
    () => client.writeSource(file.path, "", "x"),
    () => client.revisionSource(file.path, "r.1"),
    () => client.rollbackSource(file.path, ""),
    () => client.spawn("../counter", { type: "null" }),
    () => client.system(REF, { kind: "become", path: "actors/x.rn" }),
    () => client.system(REF, { kind: "explode" } as never),
    () => client.system("bad", { kind: "stop" }),
  ])
    await assert.rejects(invalid());
  assert.deepEqual(requests, [
    { operation: "list_sources" },
    { operation: "read_source", parameters: { path: file.path } },
    {
      operation: "write_source",
      parameters: { path: file.path, expected_source_id: null, source: file.source },
    },
    {
      operation: "write_source",
      parameters: { path: file.path, expected_source_id: "source:1", source: file.source },
    },
    { operation: "commit_source", parameters: { path: file.path } },
    { operation: "source_history", parameters: { path: file.path } },
    { operation: "revision_source", parameters: { path: file.path, revision: "r1" } },
    { operation: "rollback_source", parameters: { path: file.path, revision: "r1" } },
    { operation: "system", parameters: { actor_ref: REF, message: { kind: "stop" } } },
    {
      operation: "system",
      parameters: { actor_ref: REF, message: { kind: "become", path: "actors/other" } },
    },
    {
      operation: "spawn",
      parameters: { path: file.path, state: { type: "null" }, mailbox_capacity: 64 },
    },
  ]);
});