qbit-scripting/worker/test/worker.test.mjs

/* Copyright (C) 2026 Kingsrook, LLC.
 * SPDX-License-Identifier: Apache-2.0
 */
import assert from "node:assert/strict";
import { Buffer } from "node:buffer";
import process from "node:process";
import { tmpdir } from "node:os";
import { spawn } from "node:child_process";
import { Readable } from "node:stream";
import test from "node:test";
import { compile } from "../dist/compiler.js";
import { sourceHash, PROTOCOL, TOOLCHAIN, hash } from "../dist/protocol.js";
import { frames } from "../dist/frames.js";
import { createArguments, qualify, relay } from "../dist/launcher.js";

/*******************************************************************************
 ** Synthetic compile fixtures only; user-supplied code is never run in test host.
 *******************************************************************************/
function request(content, extra = [])
{
   const files = [{ path: "main.ts", content }, ...extra];
   return { protocol: PROTOCOL, type: "compile", requestId: "fixture-1", strict: true, toolchainId: TOOLCHAIN, entryPoint: "main.ts", files, sourceSha256: sourceHash(files) };
}

test("strict checking returns no executable on a type error", () =>
{
   const result = compile(request("export function main(): string { return 12; }"));
   assert.equal(result.success, false);
   assert.equal(result.bundle, undefined);
   assert.ok(result.diagnostics.length > 0);
});

test("local ESM imports compile with stable bundles and source maps", () =>
{
   const source = request('import { value } from "./helper.js"; export function main() { return { answer: value }; }', [{ path: "helper.ts", content: "export const value: number = 42;" }]);
   const first = compile(source);
   assert.equal(first.success, true, JSON.stringify(first.diagnostics));
   assert.equal(first.bundle, compile(source).bundle);
   assert.equal(first.bundleSha256, hash(first.bundle));
   assert.equal(JSON.parse(first.bundle).files.length, 2);
   assert.match(first.sourceMap, /sourcesContent/);
});

test("SDK type imports support async queries without Node types", () =>
{
   const result = compile(request('import type {Context,Qqq,HookResult} from "@qqq/sdk"; export async function main(context: Context, qqq: Qqq): Promise<HookResult> { await qqq.query("orders", ["id"]); return {action:"ALLOW"}; }'));
   assert.equal(result.success, true, JSON.stringify(result.diagnostics));
   assert.equal(compile(request('import fs from "node:fs"; export const value = fs;')).success, false);
   assert.equal(compile(request("export const value = process.env;")).success, false);
   assert.equal(compile(request('export async function main() { return import("./helper.js"); }', [{ path: "helper.ts", content: "export const value = 1;" }])).success, false);
});

test("SDK exposes nine phases and refuses deleted-value assumptions", () =>
{
   const result = compile(request('import type {HookEvent,Context,CommittedEvent} from "@qqq/sdk"; const phases: HookEvent[] = ["PRE_INSERT","PRE_UPDATE","PRE_DELETE","POST_INSERT","POST_UPDATE","POST_DELETE","AFTER_INSERT","AFTER_UPDATE","AFTER_DELETE"]; export function main(context: Context) { const event: CommittedEvent | undefined = context.payload.event; if(event?.dataState === "CURRENT") { return {action:"ALLOW", value:event.record.id, phases}; } return {action:"ALLOW"}; }'));
   assert.equal(result.success, true, JSON.stringify(result.diagnostics));
   assert.equal(compile(request('import type {HookEvent} from "@qqq/sdk"; export const phase: HookEvent = "AFTER_ARBITRARY";')).success, false);
   assert.equal(compile(request('import type {CommittedEvent} from "@qqq/sdk"; export function main(event: CommittedEvent) { if(event.dataState === "DELETED_METADATA") { return event.record.id; } return null; }')).success, false);
});

test("source identities and paths cannot be substituted", () =>
{
   assert.throws(() => compile({ ...request("export const value = 1;"), sourceSha256: "0".repeat(64) }));
   assert.throws(() => compile(request("export const value = 1;", [{ path: "../escape.ts", content: "" }])));
   assert.throws(() => compile(request("export const value = 1;", [{ path: "MAIN.ts", content: "" }])));
});

test("frames enforce limits before parsing and reject missing delimiters", async () =>
{
   const result = [];
   for await (const frame of frames(Readable.from([Buffer.from('{"a":'), Buffer.from("1}\n")]), 20))
   {
      result.push(frame);
   }
   assert.deepEqual(result, [{ a: 1 }]);
   await assert.rejects(async () =>
   {
      for await (const unused of frames(Readable.from([Buffer.alloc(21, 97)]), 20))
      {
         void unused;
      }
   });
   await assert.rejects(async () =>
   {
      for await (const unused of frames(Readable.from([Buffer.from("{}")]), 20))
      {
         void unused;
      }
   });
});

test("production requires runsc, matching evidence and an immutable image", () =>
{
   const config = { image: "qqq-worker@sha256:" + "a".repeat(64), qualificationId: "review-1", requiredQualificationId: "review-1", ownerId: "fixture", stateDirectory: tmpdir() };
   qualify(config, { OSType: "linux", Runtimes: { runsc: {} } });
   assert.throws(() => qualify(config, { OSType: "linux", Runtimes: { runc: {} } }));
   assert.throws(() => qualify({ ...config, qualificationId: "" }, { OSType: "linux", Runtimes: { runsc: {} } }));
   assert.throws(() => qualify({ ...config, image: "node:latest" }, { OSType: "linux", Runtimes: { runsc: {} } }));
   const args = createArguments(request("export const value=1;"), config, "fixture");
   for(const flag of ["--network=none", "--read-only", "--runtime=runsc", "--cap-drop=ALL", "--pids-limit=32", "--pull=never"])
   {
      assert.ok(args.includes(flag));
   }
   assert.equal(args.some(arg => arg.startsWith("--volume") || arg === "--privileged"), false);
});

test("container entry refuses direct host execution without the marker", async () =>
{
   const child = spawn(process.execPath, ["dist/entrypoint.js"], { env: { PATH: process.env.PATH }, stdio: ["pipe", "pipe", "pipe"] });
   child.stdin.end();
   const code = await new Promise(resolve => child.once("close", resolve));
   assert.equal(code, 1);
});

test("trusted synthetic compile and async broker fixture run in a separate process", async () =>
{
   // This fixture tests protocol mechanics only, never hostile-code containment.
   const built = compile(request('import type {Context,Qqq} from "@qqq/sdk"; export async function main(context:Context, qqq:Qqq) { const rows=await qqq.query("orders",["id"],[],1); return {action:"PATCH",patch:{count:rows.length}}; }'));
   assert.equal(built.success, true, JSON.stringify(built.diagnostics));
   const child = spawn(process.execPath, ["dist/entrypoint.js"], { env: { PATH: process.env.PATH, QQQ_WORKER_MODE: "container" }, stdio: ["pipe", "pipe", "pipe"] });
   const exit = new Promise(resolve => child.once("close", resolve));
   child.stdin.write(JSON.stringify({ protocol: PROTOCOL, type: "execute", requestId: "fixture-1", artifactId: "fixture", artifactSha256: built.bundleSha256, bundle: built.bundle, entryPoint: "main.ts", event: "PRE_INSERT", payload: {}, capabilityToken: "fixture-capability", limits: { brokerCalls: 1 } }) + "\n");
   let final;
   for await (const message of frames(child.stdout))
   {
      if(message.type === "broker")
      {
         assert.equal(message.operation, "query");
         assert.equal(message.token, "fixture-capability");
         child.stdin.write(JSON.stringify({ protocol: PROTOCOL, type: "brokerResult", ok: true, rows: [{ id: 1 }], truncated: false }) + "\n");
      }
      else
      {
         final = message;
      }
   }
   assert.equal(await exit, 0);
   assert.deepEqual(final.result, { action: "PATCH", patch: { count: 1 } });
});

/*******************************************************************************
 ** Exercise the external transport supervisor with a harmless synthetic peer.
 *******************************************************************************/
function peer(mode)
{
   return spawn(process.execPath, ["test/transport-fixture.mjs", mode], { env: { PATH: process.env.PATH }, stdio: ["pipe", "pipe", "pipe"] });
}

test("supervisor terminates stalled, oversized and cancelled transports", async () =>
{
   const run = { protocol: PROTOCOL, type: "execute", requestId: "fixture-1", capabilityToken: "expected" };
   const limits = { wallMillis: 150, cpuMillis: 1000, memoryBytes: 67108864, outputBytes: 1024, brokerCalls: 1 };
   const deny = async () =>
   {
      throw new Error("Unexpected broker invocation");
   };
   await assert.rejects(relay(peer("idle"), run, limits, deny), /WALL_LIMIT/);
   await assert.rejects(relay(peer("output"), run, { ...limits, wallMillis: 2000 }, deny), /LIMIT/);
   const controller = new globalThis.AbortController();
   const pending = relay(peer("idle"), run, { ...limits, wallMillis: 2000 }, deny, controller.signal);
   controller.abort();
   await assert.rejects(pending, /CANCELLED/);
});

test("supervisor rejects forged capability tokens and duplicate final frames", async () =>
{
   const run = { protocol: PROTOCOL, type: "execute", requestId: "fixture-1", capabilityToken: "expected" };
   const limits = { wallMillis: 2000, cpuMillis: 1000, memoryBytes: 67108864, outputBytes: 1024, brokerCalls: 1 };
   const deny = async () =>
   {
      throw new Error("Unexpected broker invocation");
   };
   await assert.rejects(relay(peer("broker"), run, limits, deny), /BROKER_LIMIT/);
   await assert.rejects(relay(peer("duplicate"), run, limits, deny), /INVALID_RESPONSE/);
});

test("published CONTEXT fixture returns its JSON object without hook coercion", async () =>
{
   const built = compile(request('import type {Context} from "@qqq/sdk"; export function main(context:Context) { return {document:context.payload.context}; }'));
   assert.equal(built.success, true, JSON.stringify(built.diagnostics));
   const child = spawn(process.execPath, ["dist/entrypoint.js"], { env: { PATH: process.env.PATH, QQQ_WORKER_MODE: "container" }, stdio: ["pipe", "pipe", "pipe"] });
   const limits = { wallMillis: 2000, cpuMillis: 1000, memoryBytes: 67108864, outputBytes: 131072, brokerCalls: 0 };
   const run = { protocol: PROTOCOL, type: "execute", requestId: "fixture-1", artifactSha256: built.bundleSha256, bundle: built.bundle, entryPoint: "main.ts", event: "CONTEXT", payload: { context: { order: { id: 17 } } }, capabilityToken: "fixture-capability", limits };
   const result = await relay(child, run, limits, async () =>
   {
      throw new Error("Unexpected broker invocation");
   });
   assert.deepEqual(result.result, { document: { order: { id: 17 } } });
});

test("launcher rejects unsupported and coerced resource limits before Docker", () =>
{
   const limits = { wallMillis: 2000, cpuMillis: 1000, memoryBytes: 67108864, outputBytes: 131072, brokerCalls: 0 };
   const config = { image: "qqq-worker@sha256:" + "a".repeat(64), qualificationId: "review-1", requiredQualificationId: "review-1", ownerId: "fixture", stateDirectory: tmpdir() };
   assert.throws(() => createArguments({ type: "execute", limits: { ...limits, cpuMillis: 500 } }, config, "fixture"), /UNSUPPORTED_LIMITS/);
   assert.throws(() => createArguments({ type: "execute", limits: { ...limits, wallMillis: "2000" } }, config, "fixture"), /UNSUPPORTED_LIMITS/);
});