camino-package-runtime/test/runtime.test.mjs

import assert from "node:assert/strict";
import http from "node:http";
import test from "node:test";
import { referenceFromWire } from "../dist/references.js";
import { create } from "@bufbuild/protobuf";
import { createClient } from "@connectrpc/connect";
import { connectNodeAdapter, createConnectTransport } from "@connectrpc/connect-node";
import {
  createPackageRuntimeRoutes,
  createRuntimeContext,
  derived,
  jsToProtoValue,
  liveValue,
  protoValueToJs,
} from "../dist/index.js";
import {
  CaminoObjectSchema,
  CaminoService,
  CrdtValueSchema,
  StateValueSourceSchema,
  ValueSchema,
  ValueSourceSchema,
} from "../dist/camino/api_pb.js";
import { EdgeDependencySchema, InjectedDependencySchema, PackageExportRefSchema } from "../dist/quixos/refs_pb.js";
import { InvokeCapabilityResponseSchema, OrchestratorRuntime } from "../dist/quixos/orch_pb.js";
import { DerivedDependencySchema, PackageRuntime } from "../dist/quixos/runtime_pb.js";

const listen = async (routes) => {
  const server = http.createServer((request, response) => void connectNodeAdapter({ routes })(request, response));
  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
  const address = server.address();
  assert.ok(address && typeof address === "object");
  return {
    url: `http://127.0.0.1:${address.port}`,
    close: async () => {
      server.closeAllConnections();
      await new Promise((resolve) => server.close(resolve));
    },
  };
};

test("generic values preserve nested values and object references", () => {
  const value = jsToProtoValue({
    title: "A task",
    target: referenceFromWire("obj:target"),
    tags: ["one", "two"],
  });
  assert.deepEqual(protoValueToJs(value), {
    title: "A task",
    target: referenceFromWire("obj:target"),
    tags: ["one", "two"],
  });
});

test("live values preserve writable CRDT source identity through derivation", () => {
  const value = create(ValueSchema, {
    kind: { case: "stringValue", value: "notes" },
    source: create(ValueSourceSchema, {
      state: create(StateValueSourceSchema, {
        objectId: "obj:task",
        slotId: "slot:task:notes",
        revision: 7n,
        crdtSnapshot: create(CrdtValueSchema, {
          type: "quixos.automerge-document.v1",
          encoding: "automerge-snapshot-v1",
          payload: new Uint8Array([1, 2, 3]),
        }),
      }),
    }),
  });

  const forwarded = jsToProtoValue(liveValue(value));
  assert.equal(forwarded.source?.state?.objectId, "obj:task");
  assert.equal(forwarded.source?.state?.slotId, "slot:task:notes");
  assert.equal(forwarded.source?.state?.revision, 7n);
  assert.deepEqual(forwarded.source?.state?.crdtSnapshot?.payload, new Uint8Array([1, 2, 3]));
});

test("runtime context exposes only explicitly injected ports", async () => {
  const writes = [];
  const camino = {
    readState: async ({ objectId, slotId }) => ({
      value: jsToProtoValue(`${objectId}/${slotId}`),
    }),
    writeState: async (request) => {
      writes.push(request);
      return {};
    },
    resolveEdge: async () => ({ edges: [] }),
    connectEdge: async () => ({}),
  };
  const context = createRuntimeContext(
    camino,
    {},
    {
      objectId: "obj:task",
      input: {},
      dependencies: [
        create(InjectedDependencySchema, {
          portId: "port:title",
          binding: { case: "stateSlotId", value: "slot:task:title" },
        }),
        create(InjectedDependencySchema, {
          portId: "port:children",
          binding: {
            case: "edge",
            value: create(EdgeDependencySchema, {
              edgeTypeId: "edge:task:children",
              projectionId: "projection:task:children",
            }),
          },
        }),
      ],
    },
  );

  assert.equal(await context.state("port:title").get(), "obj:task/slot:task:title");
  await context.state("port:title").set("Changed");
  assert.equal(writes.length, 1);
  assert.equal("camino" in context, false);
  assert.equal("orch" in context, false);
  assert.throws(() => context.state("port:not-injected"), /Missing slotId dependency/);
  assert.throws(() => context.interface("port:title"), /Missing interfaceRevisionId dependency/);
  assert.deepEqual(await context.edge("port:children").resolve(), []);
});

test("injected ports preserve an explicitly traversed object through PackageRuntime RPC", async () => {
  const reads = [];
  const caminoServer = await listen((router) =>
    router.service(CaminoService, {
      readState: (request) => {
        reads.push(request);
        return { value: jsToProtoValue("Project name") };
      },
    }),
  );
  const runtimeServer = await listen(
    createPackageRuntimeRoutes({
      packageRevisionId: "package:test@1",
      caminoUrl: caminoServer.url,
      exports: {
        "export:test:value": (context) => context.state("port:value").get(),
      },
    }),
  );
  const client = createClient(
    PackageRuntime,
    createConnectTransport({
      baseUrl: runtimeServer.url,
      httpVersion: "1.1",
    }),
  );
  try {
    const response = await client.invoke({
      export: create(PackageExportRefSchema, {
        packageRevisionId: "package:test@1",
        exportId: "export:test:value",
      }),
      objectId: "obj:component",
      dependencies: [
        create(InjectedDependencySchema, {
          portId: "port:value",
          objectId: "obj:project",
          binding: { case: "stateSlotId", value: "slot:project:name" },
        }),
      ],
    });
    assert.equal(response.ok, true);
    assert.equal(protoValueToJs(response.result), "Project name");
    assert.equal(reads.length, 1);
    assert.equal(reads[0]?.objectId, "obj:project");
  } finally {
    await runtimeServer.close();
    await caminoServer.close();
  }
});

test("interface views invoke a related object and propagate transitive live dependencies", async () => {
  const invocations = [];
  const sourceValue = create(ValueSchema, {
    kind: { case: "stringValue", value: "Project name" },
    source: create(ValueSourceSchema, {
      state: create(StateValueSourceSchema, {
        objectId: "obj:project",
        slotId: "slot:project:name",
        revision: 4n,
      }),
    }),
  });
  const orchServer = await listen((router) =>
    router.service(OrchestratorRuntime, {
      invokeCapability: (request) => {
        invocations.push(request);
        return create(InvokeCapabilityResponseSchema, {
          ok: true,
          result: sourceValue,
          dependencies: [
            create(DerivedDependencySchema, {
              kind: "state",
              objectId: "obj:project",
              attachmentId: "slot:project:name",
            }),
          ],
        });
      },
    }),
  );
  const runtimeServer = await listen(
    createPackageRuntimeRoutes({
      packageRevisionId: "package:test@1",
      orchUrl: orchServer.url,
      exports: {
        "export:test:value": derived((context) => context.interface("port:named").live("operation:named:name:get")),
      },
    }),
  );
  const client = createClient(
    PackageRuntime,
    createConnectTransport({
      baseUrl: runtimeServer.url,
      httpVersion: "1.1",
    }),
  );
  try {
    const response = await client.invoke({
      export: create(PackageExportRefSchema, {
        packageRevisionId: "package:test@1",
        exportId: "export:test:value",
      }),
      objectId: "obj:component",
      dependencies: [
        create(InjectedDependencySchema, {
          portId: "port:named",
          objectId: "obj:project",
          binding: { case: "interfaceRevisionId", value: "interface:named@1" },
        }),
      ],
    });
    assert.equal(response.ok, true);
    assert.equal(invocations[0]?.objectId, "obj:project");
    assert.equal(invocations[0]?.capability?.interfaceRevisionId, "interface:named@1");
    assert.equal(response.dependencies[0]?.objectId, "obj:project");
    assert.equal(response.dependencies[0]?.attachmentId, "slot:project:name");
    assert.equal(response.result?.source?.state?.revision, 4n);
  } finally {
    await runtimeServer.close();
    await orchServer.close();
  }
});

test("derived interface views reread after their transitive subscriptions become live", async () => {
  let current = "before subscription";
  let invocationCount = 0;
  const caminoServer = await listen((router) =>
    router.service(CaminoService, {
      watchObject: async function* (request, context) {
        // Model a write racing the first nested capability read. Camino makes
        // the subscription live before yielding this snapshot.
        current = "after subscription";
        yield {
          objectId: request.objectId,
          snapshot: create(CaminoObjectSchema, {
            id: request.objectId,
            atomId: "atom:project",
            workspaceRevisionId: "workspace:test@1",
          }),
        };
        await new Promise((resolve) => context.signal.addEventListener("abort", resolve, { once: true }));
      },
    }),
  );
  const orchServer = await listen((router) =>
    router.service(OrchestratorRuntime, {
      invokeCapability: () => {
        invocationCount += 1;
        return create(InvokeCapabilityResponseSchema, {
          ok: true,
          result: jsToProtoValue(current),
          dependencies: [
            create(DerivedDependencySchema, {
              kind: "state",
              objectId: "obj:project",
              attachmentId: "slot:project:name",
            }),
          ],
        });
      },
    }),
  );
  const runtimeServer = await listen(
    createPackageRuntimeRoutes({
      packageRevisionId: "package:test@1",
      caminoUrl: caminoServer.url,
      orchUrl: orchServer.url,
      exports: {
        "export:test:value": derived((context) => context.interface("port:named").invoke("operation:named:name:get")),
      },
    }),
  );
  const client = createClient(
    PackageRuntime,
    createConnectTransport({
      baseUrl: runtimeServer.url,
      httpVersion: "1.1",
    }),
  );
  const controller = new AbortController();
  try {
    const stream = client
      .watch(
        {
          export: create(PackageExportRefSchema, {
            packageRevisionId: "package:test@1",
            exportId: "export:test:value",
          }),
          objectId: "obj:component",
          dependencies: [
            create(InjectedDependencySchema, {
              portId: "port:named",
              objectId: "obj:project",
              binding: { case: "interfaceRevisionId", value: "interface:named@1" },
            }),
          ],
        },
        { signal: controller.signal },
      )
      [Symbol.asyncIterator]();
    const initial = await stream.next();
    assert.equal(protoValueToJs(initial.value?.value), "after subscription");
    assert.equal(invocationCount, 2);
  } finally {
    controller.abort();
    await runtimeServer.close();
    await orchServer.close();
    await caminoServer.close();
  }
});

test("derived watches subscribe before reading and emit only real changes", async () => {
  let current = "before";
  let subscribed = false;
  const changes = [];
  const caminoServer = await listen((router) =>
    router.service(CaminoService, {
      readState: () => {
        assert.equal(subscribed, true, "the dependency stream must be live before the state read");
        return { value: jsToProtoValue(current) };
      },
      watchObject: async function* (request, context) {
        subscribed = true;
        yield {
          objectId: request.objectId,
          snapshot: create(CaminoObjectSchema, {
            id: request.objectId,
            atomId: "atom:test",
            workspaceRevisionId: "workspace:test@1",
          }),
        };
        while (!context.signal.aborted) {
          const changed = await new Promise((resolve) => {
            const finish = () => resolve(true);
            changes.push(finish);
            context.signal.addEventListener("abort", () => resolve(false), { once: true });
          });
          if (!changed) return;
          yield { objectId: request.objectId };
        }
      },
    }),
  );
  const runtimeServer = await listen(
    createPackageRuntimeRoutes({
      packageRevisionId: "package:test@1",
      caminoUrl: caminoServer.url,
      exports: {
        "export:test:value": derived((context) => context.state("port:value").get()),
      },
    }),
  );
  const client = createClient(
    PackageRuntime,
    createConnectTransport({
      baseUrl: runtimeServer.url,
      httpVersion: "1.1",
    }),
  );
  const controller = new AbortController();
  try {
    const stream = client
      .watch(
        {
          export: create(PackageExportRefSchema, {
            packageRevisionId: "package:test@1",
            exportId: "export:test:value",
          }),
          objectId: "obj:test",
          dependencies: [
            create(InjectedDependencySchema, {
              portId: "port:value",
              binding: { case: "stateSlotId", value: "slot:test:value" },
            }),
          ],
        },
        { signal: controller.signal },
      )
      [Symbol.asyncIterator]();
    const initial = await stream.next();
    assert.equal(protoValueToJs(initial.value?.value), "before");
    assert.equal(initial.value?.initial, true);

    current = "after";
    changes.shift()?.();
    const updated = await stream.next();
    assert.equal(protoValueToJs(updated.value?.value), "after");
    assert.equal(updated.value?.initial, false);
  } finally {
    controller.abort();
    await runtimeServer.close();
    await caminoServer.close();
  }
});