FE-1429: Run trial replicates through the shared experiment runtime

.changeset/trial-replicates-parallel.md

@@ -0,0 +1,6 @@
+---
+"@hashintel/petrinaut-core": patch
+"@hashintel/petrinaut-cli": patch
+---
+
+Optimization trials run their seeded replicates in parallel as one sharded experiment. The Monte Carlo worker protocol attaches to any thread runtime, and the CLI's `--threads <n>` bounds the workers, defaulting to one per core minus one.

.changeset/webgpu-experiment-backend.md

@@ -0,0 +1,6 @@
+---
+"@hashintel/petrinaut-core": patch
+"@hashintel/petrinaut": patch
+---
+
+Add an experimental WebGPU compute backend for experiments, chosen per experiment behind a user setting. It runs the net's lowered HIR on the device, declines nets it cannot run so they fall back to the CPU, and agrees with the CPU in distribution rather than seed for seed. A Compilation panel, also behind a setting, shows what the compiler made of each condition, kernel and equation.

libs/@hashintel/petrinaut-cli/src/cli.ts

@@ -2,6 +2,7 @@ import { parseArgs } from "node:util";
 
 import { serve } from "./commands/serve";
 import { serveStdio } from "./commands/stdio";
+import { resolveSimulationThreads } from "./runtime/simulation-threads";
 
 function printUsage(): void {
   process.stderr.write(`Usage:
@@ -16,6 +17,7 @@ Model sources:
   --model-stdin           Read a legacy model JSON object from the first stdin line
   --optimization <path>  Load an optimization manifest from a YAML or JSON file (stdio only)
   --optimization-stdin   Read an optimization manifest from the first stdin line (stdio only)
+  --threads <n>          Threads for seeded optimization replicates (default: cores - 1; 1 = no worker threads)
 
 Methods:
   healthz
@@ -43,6 +45,7 @@ async function main(): Promise<void> {
       "optimization-stdin": { type: "boolean" },
       socket: { type: "string" },
       stdio: { type: "boolean" },
+      threads: { type: "string" },
       help: { type: "boolean", short: "h" },
     },
     allowPositionals: false,
@@ -95,14 +98,17 @@ async function main(): Promise<void> {
       modelPath,
       socketPath: parsed.values.socket,
     });
-  } else if (modelStdin) {
-    await serveStdio({ modelStdin: true });
-  } else if (modelPath) {
-    await serveStdio({ modelPath });
-  } else if (optimizationStdin) {
-    await serveStdio({ optimizationStdin: true });
-  } else if (optimizationPath) {
-    await serveStdio({ optimizationPath });
+  } else {
+    const simulationThreads = resolveSimulationThreads(parsed.values.threads);
+    if (modelStdin) {
+      await serveStdio({ modelStdin: true, simulationThreads });
+    } else if (modelPath) {
+      await serveStdio({ modelPath, simulationThreads });
+    } else if (optimizationStdin) {
+      await serveStdio({ optimizationStdin: true, simulationThreads });
+    } else if (optimizationPath) {
+      await serveStdio({ optimizationPath, simulationThreads });
+    }
   }
 }
 

libs/@hashintel/petrinaut-cli/src/commands/stdio.ts

@@ -4,6 +4,7 @@ import { createInterface } from "node:readline";
 import { compilePetrinautModel } from "@hashintel/petrinaut-core/compiled-model";
 
 import { loadSdcpnModel, parseSdcpnModel } from "../runtime/load-model";
+import { createNodeSimulationWorkerFactory } from "../runtime/node-simulation-worker";
 import {
   createOptimizationProtocol,
   loadOptimizationManifest,
@@ -51,6 +52,11 @@ type ServeStdioOptions = (
   input?: Readable;
   output?: Writable;
   errorOutput?: Writable;
+  /**
+   * Threads for seeded optimization replicates. 1 simulates on the calling
+   * thread with no worker spawned; absent behaves as 1.
+   */
+  simulationThreads?: number;
 };
 
 function writeResponse(output: Writable, value: unknown): void {
@@ -117,8 +123,20 @@ export async function serveStdio(options: ServeStdioOptions): Promise<void> {
   }
 
   const model = compilePetrinautModel({ sdcpn });
+  const simulationThreads = options.simulationThreads ?? 1;
   const optimization = optimizationManifest
-    ? createOptimizationProtocol({ manifest: optimizationManifest, model })
+    ? createOptimizationProtocol({
+        manifest: optimizationManifest,
+        model,
+        // One thread means no worker at all: the in-process worker runs the
+        // same protocol on the calling thread without a spawn.
+        ...(simulationThreads > 1
+          ? {
+              createWorker: createNodeSimulationWorkerFactory({ errorOutput }),
+              shardCount: simulationThreads,
+            }
+          : {}),
+      })
     : undefined;
 
   errorOutput.write(

libs/@hashintel/petrinaut-cli/src/runtime/node-simulation-worker.ts

@@ -0,0 +1,63 @@
+/**
+ * Spawning simulation workers under Node.
+ *
+ * The core's experiment runtime asks for a {@link WorkerFactory} and speaks a
+ * structural `WorkerLike` — `postMessage`, `addEventListener` receiving a
+ * `{ data }` envelope, `terminate` — so Node's `worker_threads.Worker`, which
+ * emits bare values through `.on("message", ...)`, is adapted here.
+ *
+ * The worker entry is a sibling bundle of `cli.js`, so it exists when the CLI
+ * runs from its build output. Running from source (`tsx src/cli.ts`) has no
+ * such file: the factory then falls back to the in-process worker, which runs
+ * the same protocol on the calling thread, and says so once on stderr so a
+ * sequential dev run is never mistaken for a parallel one.
+ */
+import { existsSync } from "node:fs";
+import { Worker } from "node:worker_threads";
+
+import { createInProcessMonteCarloWorker } from "@hashintel/petrinaut-core/workers/monte-carlo";
+
+import type { WorkerFactory } from "@hashintel/petrinaut-core";
+
+const workerEntryUrl = new URL("./simulation-worker.js", import.meta.url);
+
+let warnedAboutFallback = false;
+
+export function createNodeSimulationWorkerFactory(options?: {
+  /** Overrides the entry bundle, e.g. for tests. */
+  entryUrl?: URL;
+  errorOutput?: { write: (chunk: string) => void };
+}): WorkerFactory {
+  const entryUrl = options?.entryUrl ?? workerEntryUrl;
+
+  if (!existsSync(entryUrl)) {
+    if (!warnedAboutFallback) {
+      warnedAboutFallback = true;
+      options?.errorOutput?.write(
+        `Simulation worker bundle not found at ${entryUrl.pathname}; runs execute in-process\n`,
+      );
+    }
+    return createInProcessMonteCarloWorker;
+  }
+
+  return () => {
+    // The worker stays referenced: the trial that asked for it disposes the
+    // experiment in a `finally`, which terminates every worker, so the process
+    // neither exits before a reply is written nor outlives the request.
+    const worker = new Worker(entryUrl);
+
+    return {
+      postMessage: (message) => {
+        worker.postMessage(message);
+      },
+      addEventListener: (_type, listener) => {
+        worker.on("message", (data: unknown) => {
+          listener({ data });
+        });
+      },
+      terminate: () => {
+        void worker.terminate();
+      },
+    };
+  };
+}