Release: 3.4.0

.env.example

@@ -70,3 +70,9 @@ LANGFUSE_BASE_URL=https://cloud.langfuse.com
 # ZILLIZ_URI =  https://in01-XXXXXXXXXXXXX.aws-us-west-2.vectordb.zillizcloud.com:XXXXX
 # ZILLIZ_TOKEN =db_admin:password (or ApiKey)
 
+# Turn this on to use the fork server mechanism
+# APOLLO_FORK_SERVER=true
+
+# macOS only: required when APOLLO_FORK_SERVER=true
+# OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES
+

CHANGELOG.md

@@ -1,5 +1,17 @@
 # apollo
 
+## 3.4.0
+
+### Minor Changes
+
+- 02eaf61: global_chat: opt-in web search and fetch for the planner
+
+### Patch Changes
+
+- 0347252: Update dependencies, including the Anthropic Python SDK to 1.x, the
+  OpenAI SDK to 3.x and pandas to 3.x
+- 8ff4d37: fork python processes to reduce startup time
+
 ## 3.3.1
 
 ### Patch Changes

package.json

@@ -1,7 +1,7 @@
 {
   "name": "apollo",
   "module": "platform/index.ts",
-  "version": "3.3.1",
+  "version": "3.4.0",
   "type": "module",
   "scripts": {
     "start": "NODE_ENV=production bun platform/src/index.ts",
@@ -23,7 +23,7 @@
     "@elysiajs/html": "^1.4.0",
     "@openfn/adaptor-apis": "^0.3.0",
     "@sentry/bun": "^10.60.0",
-    "elysia": "1.4.27",
+    "elysia": "1.4.29",
     "jsdoc-babel": "^0.5.0"
   },
   "nodemonConfig": {

platform/src/bridge.fork.ts

@@ -0,0 +1,183 @@
+import readline from "node:readline";
+import net from "node:net";
+import path from "node:path";
+import { spawn, type ChildProcess } from "node:child_process";
+import { rm } from "node:fs/promises";
+import { getInternalToken } from "./auth/internal-token";
+import {
+  emptyResult,
+  malformedResult,
+  subprocessCancelled,
+  subprocessFailed,
+  subprocessKilled,
+  subprocessSpawnFailed,
+} from "./util/errors";
+import { LOG_LINE } from "./util/log-line";
+import pkg from "../../package.json";
+
+export const USE_FORK_SERVER =
+  (process.env.APOLLO_FORK_SERVER ?? "").toLowerCase() === "true";
+
+const onLines = (stream: NodeJS.ReadableStream, fn: (line: string) => void) =>
+  readline.createInterface({ input: stream, crlfDelay: Infinity }).on("line", fn);
+
+// --- the warm master: spawned once, forks a child per request ----------------
+
+type Master = { proc: ChildProcess; socketPath: string };
+let master: Promise<Master> | null = null;
+let masterProc: ChildProcess | null = null;
+
+// Singleton; a failed start clears the slot so the next request retries.
+function forkMaster(): Promise<Master> {
+  return (master ??= startMaster().catch((err) => {
+    master = null;
+    throw err;
+  }));
+}
+
+function startMaster(): Promise<Master> {
+  const socketPath = path.resolve(`tmp/fork_server-${process.pid}.sock`);
+  const proc = spawn(
+    "poetry",
+    ["run", "python", "services/entry.fork.py", "--socket", socketPath],
+    {
+      env: {
+        ...process.env,
+        APOLLO_INTERNAL_TOKEN: getInternalToken(),
+        APOLLO_VERSION: pkg.version,
+      },
+    }
+  );
+  masterProc = proc;
+  onLines(proc.stderr!, (line) => console.error(`fork-server: ${line}`));
+
+  return new Promise<Master>((resolve, reject) => {
+    const fail = (err: Error) => (clearTimeout(timer), reject(subprocessSpawnFailed("fork_server", err)));
+    const timer = setTimeout(() => fail(new Error("not READY in time")), 60_000);
+
+    onLines(proc.stdout!, (line) => {
+      if (line.startsWith("READY")) {
+        clearTimeout(timer);
+        resolve({ proc, socketPath });
+      }
+    });
+    proc.on("error", fail);
+    proc.on("exit", (code, sig) => {
+      console.error(`fork-server exited (code=${code} signal=${sig})`);
+      master = masterProc = null; // next request starts a fresh one
+      rm(socketPath).catch(() => {});
+      fail(new Error(`fork-server exited before ready (code=${code} signal=${sig})`));
+    });
+  });
+}
+
+// Don't leave an orphaned master when the server process goes.
+process.once("exit", () => masterProc?.kill("SIGTERM"));
+
+// --- per-request: ask the master to fork a child, stream it back -------------
+
+export async function runForked(
+  scriptName: string,
+  port: number,
+  inputPath: string,
+  outputPath: string,
+  onLog?: (str: string) => void,
+  onEvent?: (type: string, payload: any) => void,
+  signal?: AbortSignal
+): Promise<JSON | null> {
+  const cleanup = () =>
+    Promise.all([rm(inputPath).catch(() => {}), rm(outputPath).catch(() => {})]);
+
+  let active: Master;
+  try {
+    active = await forkMaster();
+  } catch (err) {
+    await cleanup();
+    throw err;
+  }
+
+  return new Promise<JSON | null>((resolve, reject) => {
+    const conn = net.createConnection({ path: active.socketPath });
+    let childPid: number | null = null;
+    let exitCode: number | null = null;
+    let cancelled = false;
+    let hardKill: ReturnType<typeof setTimeout> | undefined;
+
+    const killChild = () => {
+      if (childPid == null) return; // killed once the PID: line arrives
+      try {
+        process.kill(childPid, "SIGTERM");
+      } catch {}
+      hardKill = setTimeout(() => {
+        try {
+          process.kill(childPid!, "SIGKILL");
+        } catch {}
+      }, 5_000);
+      hardKill.unref?.();
+    };
+
+    const onAbort = () => {
+      cancelled = true;
+      console.warn(`cancelling ${scriptName}: client went away`);
+      killChild();
+      conn.destroy();
+    };
+    if (signal) {
+      if (signal.aborted) onAbort();
+      else signal.addEventListener("abort", onAbort, { once: true });
+    }
+
+    conn.on("connect", () =>
+      conn.write(JSON.stringify({ service: scriptName, input: inputPath, output: outputPath, port }) + "\n")
+    );
+
+    // The child's stream: PID:/EXIT: control lines around the service's own stdout.
+    onLines(conn, (line) => {
+      if (line.startsWith("PID:")) {
+        childPid = Number(line.slice(4)) || null;
+        if (cancelled) killChild();
+      } else if (line.startsWith("EXIT:")) {
+        exitCode = Number(line.slice(5));
+      } else if (LOG_LINE.test(line)) {
+        console.log(line);
+        onLog?.(line);
+      } else if (line.startsWith("EVENT:")) {
+        const [, type, ...rest] = line.split(":");
+        let payload: any = rest.join(":");
+        try {
+          payload = JSON.parse(payload);
+        } catch {
+          // not json, forward the raw string
+        }
+        onEvent?.(type, payload);
+      }
+    });
+
+    const settle = async (finish: () => void) => {
+      if (hardKill) clearTimeout(hardKill);
+      signal?.removeEventListener("abort", onAbort);
+      await cleanup();
+      finish();
+    };
+
+    conn.on("error", (err) =>
+      settle(() => reject(subprocessSpawnFailed(scriptName, err)))
+    );
+
+    conn.on("close", async () => {
+      const text = await Bun.file(outputPath).text().catch(() => "");
+      await settle(() => {
+        if (cancelled) return reject(subprocessCancelled(scriptName, "SIGTERM"));
+        // No EXIT line => the child died mid-flight (crash/OOM/signal).
+        if (exitCode === null) return reject(subprocessKilled(scriptName, "UNKNOWN"));
+        if (exitCode !== 0) return reject(subprocessFailed(scriptName, exitCode));
+        if (!text) return reject(emptyResult(scriptName));
+        try {
+          resolve(JSON.parse(text));
+        } catch {
+          reject(malformedResult(scriptName));
+        }
+      });
+    });
+  });
+}

platform/src/bridge.ts

@@ -12,10 +12,8 @@ import {
   subprocessSpawnFailed,
 } from "./util/errors";
 import pkg from "../../package.json";
-
-// A line a service logged on purpose, as opposed to whatever else lands on a
-// stream. Only these are forwarded to the caller.
-const LOG_LINE = /^(INFO|DEBUG|ERROR|WARNING):/;
+import { USE_FORK_SERVER, runForked } from "./bridge.fork";
+import { LOG_LINE } from "./util/log-line";
 
 /**
   Run a python script
@@ -61,6 +59,10 @@ export const run = async (
     throw subprocessSpawnFailed(scriptName, error);
   }
 
+  if (USE_FORK_SERVER) {
+    return runForked(scriptName, port, inputPath, outputPath, onLog, onEvent, signal);
+  }
+
   return new Promise<JSON | null>((resolve, reject) => {
     const proc = spawn(
       "poetry",

platform/src/util/log-line.ts

@@ -0,0 +1,4 @@
+// A line a service logged on purpose, as opposed to whatever else lands on a
+// stream. Only these are forwarded to the caller. Shared by bridge.ts and
+// bridge.fork.ts.
+export const LOG_LINE = /^(INFO|DEBUG|ERROR|WARNING):/;