fix(execution): address review on the dedicated worker host

- Host: reject an unparseable upgrade target instead of throwing; bound
  unauthenticated connections separately so they never hold a worker slot.
- Pool: re-enqueue an execution whose selected worker was retired before
  dispatch; forward TZ/LANG/LC_ALL with the limits so remote and local workers
  produce the same Date/Intl results.
- Keep the child_process import in its original form so the bundler does not
  trace the worker spawn; give ffmpeg's env NODE_ENV for ProcessEnv.
- E2E: stop the suite when a check cannot settle; assert the exact truncation
  bound; cover the unparseable target.

Author: waleedlatif1

.claude/rules/sim-sandbox.md

@@ -78,9 +78,9 @@ payload or `ivm.Reference` wrapper in the worker:
       allowlist in `isolated-vm.ts` **and** to the allowlist regression test in
       `isolated-vm.test.ts` — the worker does not inherit the app environment,
       so an un-allowlisted var is simply absent in the child. If the app owns
-      its value, also add it to `workerLimitsEnv()` and to
-      `CLIENT_WORKER_ENV_KEYS` in `isolated-vm-host.cjs`, so a remote worker
-      enforces the same value.
+      its value, put it in `appWorkerEnv()` and add it with its accepted shape
+      to `CLIENT_WORKER_ENV` in `isolated-vm-host.cjs`, so a remote worker
+      gets the same value.
 - [ ] Does the host send the worker a new message type? Add it to
       `ALLOWED_INBOUND_TYPES` in `isolated-vm-host.cjs`, or remote workers
       never receive it.

.cursor/rules/sim-sandbox.mdc

@@ -74,9 +74,9 @@ payload or `ivm.Reference` wrapper in the worker:
       allowlist in `isolated-vm.ts` **and** to the allowlist regression test in
       `isolated-vm.test.ts` — the worker does not inherit the app environment,
       so an un-allowlisted var is simply absent in the child. If the app owns
-      its value, also add it to `workerLimitsEnv()` and to
-      `CLIENT_WORKER_ENV_KEYS` in `isolated-vm-host.cjs`, so a remote worker
-      enforces the same value.
+      its value, put it in `appWorkerEnv()` and add it with its accepted shape
+      to `CLIENT_WORKER_ENV` in `isolated-vm-host.cjs`, so a remote worker
+      gets the same value.
 - [ ] Does the host send the worker a new message type? Add it to
       `ALLOWED_INBOUND_TYPES` in `isolated-vm-host.cjs`, or remote workers
       never receive it.

apps/sim/lib/execution/isolated-vm-host.cjs

@@ -10,7 +10,8 @@
  *
  * Wire contract (`/v1/worker`): one JSON object per text frame. The first
  * client frame is `{ type: 'auth', token, env }`, where `env` carries the
- * worker limits the app enforces; after it, frames are the objects the worker
+ * worker settings the app owns (limits, timezone, locale); after it, frames
+ * are the objects the worker
  * exchanges over IPC. Host-originated frames are `{ type: 'host:heartbeat' }`
  * every 30s and `{ type: 'host:draining' }` on SIGTERM, after which the app
  * retires the connection once its executions finish. A breaking change to the
@@ -40,8 +41,14 @@ const MAX_STARTUP_STDERR_CHARS = 64 * 1024
 const MIN_TOKEN_LENGTH = 32
 const UNAUTHORIZED_CLOSE_CODE = 4401
 const ALLOWED_INBOUND_TYPES = new Set(['execute', 'cancel', 'fetchResponse', 'brokerResponse'])
-/** Worker limits a client may set, each a decimal integer. */
-const CLIENT_WORKER_ENV_KEYS = new Set(['IVM_MAX_STDOUT_CHARS', 'IVM_MAX_FETCH_OPTIONS_JSON_CHARS'])
+/** Worker settings a client may set, each with the only shape it accepts. */
+const CLIENT_WORKER_ENV = {
+  IVM_MAX_STDOUT_CHARS: /^\d{1,12}$/,
+  IVM_MAX_FETCH_OPTIONS_JSON_CHARS: /^\d{1,12}$/,
+  TZ: /^:?[A-Za-z0-9_+\-/]{1,64}$/,
+  LANG: /^[A-Za-z0-9_.@-]{1,64}$/,
+  LC_ALL: /^[A-Za-z0-9_.@-]{1,64}$/,
+}
 
 function readPositiveInt(name, fallback) {
   const parsed = Number.parseInt(process.env[name] || '', 10)
@@ -62,6 +69,8 @@ if (process.versions.bun) {
 
 const PORT = readPositiveInt('IVM_HOST_PORT', 3100)
 const MAX_WORKERS = readPositiveInt('IVM_HOST_MAX_WORKERS', 16)
+/** Unauthenticated connections are bounded on their own, so they never hold a worker slot. */
+const MAX_PENDING_AUTH = MAX_WORKERS * 4
 const MAX_MESSAGE_BYTES = readPositiveInt('IVM_HOST_MAX_MESSAGE_BYTES', 64 * 1024 * 1024)
 const DRAIN_TIMEOUT_MS = readPositiveInt('IVM_HOST_DRAIN_TIMEOUT_MS', 110000)
 
@@ -87,22 +96,18 @@ function isAuthorized(presented) {
 /**
  * Mirrors `buildWorkerEnv()` in `isolated-vm.ts`: the worker runs untrusted
  * code and receives an explicit allowlist, never this process's environment.
- * The limits come from the client so a remote worker enforces the app's values.
+ * Settings the app owns come from the client, so a remote worker behaves like
+ * a local one.
  */
 function buildWorkerEnv(clientEnv) {
   const workerEnv = {}
-  for (const key of ['PATH', 'NODE_ENV', 'TZ', 'LANG', 'LC_ALL']) {
+  for (const key of ['PATH', 'NODE_ENV']) {
     if (process.env[key] !== undefined) workerEnv[key] = process.env[key]
   }
   if (clientEnv && typeof clientEnv === 'object') {
-    for (const [key, value] of Object.entries(clientEnv)) {
-      if (
-        CLIENT_WORKER_ENV_KEYS.has(key) &&
-        typeof value === 'string' &&
-        /^\d{1,12}$/.test(value)
-      ) {
-        workerEnv[key] = value
-      }
+    for (const [key, pattern] of Object.entries(CLIENT_WORKER_ENV)) {
+      const value = clientEnv[key]
+      if (typeof value === 'string' && pattern.test(value)) workerEnv[key] = value
     }
   }
   return workerEnv
@@ -154,23 +159,30 @@ const wss = new WebSocketServer({
 
 server.on('upgrade', (req, socket, head) => {
   socket.on('error', () => socket.destroy())
-  if (new URL(req.url || '/', 'http://host').pathname !== WORKER_PATH) {
+  let pathname
+  try {
+    pathname = new URL(req.url || '/', 'http://host').pathname
+  } catch {
+    rejectUpgrade(socket, 400, 'Bad Request')
+    return
+  }
+  if (pathname !== WORKER_PATH) {
     rejectUpgrade(socket, 404, 'Not Found')
     return
   }
   if (draining) {
     rejectUpgrade(socket, 503, 'Service Unavailable')
     return
   }
-  if (workers.size + pendingConnections >= MAX_WORKERS) {
+  if (workers.size >= MAX_WORKERS || pendingConnections >= MAX_PENDING_AUTH) {
     log('warn', 'Rejected worker connection at capacity', { maxWorkers: MAX_WORKERS })
     rejectUpgrade(socket, 503, 'Service Unavailable')
     return
   }
   wss.handleUpgrade(req, socket, head, (ws) => authenticate(ws))
 })
 
-/** Holds a capacity slot until the client's first frame proves it holds the token. */
+/** Admits the connection to a worker slot only once its first frame proves it holds the token. */
 function authenticate(ws) {
   pendingConnections++
   let settled = false
@@ -206,6 +218,11 @@ function authenticate(ws) {
       ws.close(1012, 'draining')
       return
     }
+    if (workers.size >= MAX_WORKERS) {
+      log('warn', 'Rejected worker connection at capacity', { maxWorkers: MAX_WORKERS })
+      ws.close(1013, 'at capacity')
+      return
+    }
     ws.off('error', onPreAuthError)
     attachWorker(ws, buildWorkerEnv(message.env))
   })

apps/sim/lib/execution/isolated-vm-remote.ts

@@ -27,7 +27,7 @@ export interface IsolatedVMWorkerProcess {
 export interface RemoteWorkerOptions {
   url: string
   token: string
-  /** Limits the worker reads from its environment, so a remote worker enforces the app's values. */
+  /** Worker settings the app owns (limits, timezone, locale), so a remote worker matches a local one. */
   workerEnv: Record<string, string | undefined>
 }
 

apps/sim/lib/execution/isolated-vm.ts

@@ -1013,10 +1013,7 @@ function resetWorkerIdleTimeout(workerId: number) {
 function buildWorkerEnv(): NodeJS.ProcessEnv {
   const allowed: Record<string, string | undefined> = {
     PATH: process.env.PATH,
-    ...workerLimitsEnv(),
-    TZ: process.env.TZ,
-    LANG: process.env.LANG,
-    LC_ALL: process.env.LC_ALL,
+    ...appWorkerEnv(),
     SYSTEMROOT: process.env.SYSTEMROOT,
     WINDIR: process.env.WINDIR,
     COMSPEC: process.env.COMSPEC,
@@ -1027,11 +1024,18 @@ function buildWorkerEnv(): NodeJS.ProcessEnv {
   return { ...filterUndefined(allowed), NODE_ENV: process.env.NODE_ENV }
 }
 
-/** The limits the worker reads from its environment; a remote worker receives them on connect. */
-function workerLimitsEnv(): Record<string, string | undefined> {
+/**
+ * Worker settings whose value the app owns: its limits, and the timezone and locale that shape
+ * `Date` and `Intl` inside isolates. A remote worker receives them on connect, so moving workers
+ * to the host changes no results.
+ */
+function appWorkerEnv(): Record<string, string | undefined> {
   return {
     IVM_MAX_STDOUT_CHARS: env.IVM_MAX_STDOUT_CHARS,
     IVM_MAX_FETCH_OPTIONS_JSON_CHARS: env.IVM_MAX_FETCH_OPTIONS_JSON_CHARS,
+    TZ: process.env.TZ,
+    LANG: process.env.LANG,
+    LC_ALL: process.env.LC_ALL,
   }
 }
 
@@ -1047,7 +1051,7 @@ async function openWorkerProcess(target: WorkerTarget): Promise<IsolatedVMWorker
     return connectRemoteWorker({
       url: target.url,
       token: target.token,
-      workerEnv: workerLimitsEnv(),
+      workerEnv: appWorkerEnv(),
     })
   }
 
@@ -1070,13 +1074,14 @@ async function openWorkerProcess(target: WorkerTarget): Promise<IsolatedVMWorker
     throw new Error(`Worker file not found at any of: ${candidatePaths.join(', ')}`)
   }
 
-  const { spawn } = await import('node:child_process')
-  // Required for isolated-vm on Node.js 20+ (issue #377)
-  return spawn('node', ['--no-node-snapshot', workerPath], {
-    stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
-    serialization: 'json',
-    env: buildWorkerEnv(),
-  })
+  return import('node:child_process').then(({ spawn }) =>
+    // Required for isolated-vm on Node.js 20+ (issue #377)
+    spawn('node', ['--no-node-snapshot', workerPath], {
+      stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
+      serialization: 'json',
+      env: buildWorkerEnv(),
+    })
+  )
 }
 
 function spawnWorker(): Promise<WorkerInfo> {
@@ -1259,6 +1264,12 @@ function dispatchToWorker(
     return
   }
 
+  // A target change can retire and remove the worker between its selection and this dispatch.
+  if (workers.get(workerInfo.id) !== workerInfo) {
+    enqueueExecution(ownerState, req, resolve, state, brokers)
+    return
+  }
+
   const execId = ++executionIdCounter
   state.workerId = workerInfo.id
   state.execId = execId

apps/sim/lib/media/ffmpeg-process.ts

@@ -70,13 +70,14 @@ export function resolveExecutable(binary: string): string | null {
  * and fonts, never the variables this process was configured with.
  */
 function executableEnv(): NodeJS.ProcessEnv {
-  return filterUndefined({
+  const allowed = filterUndefined({
     PATH: process.env.PATH,
     HOME: process.env.HOME,
     FONTCONFIG_FILE: process.env.FONTCONFIG_FILE,
     FONTCONFIG_PATH: process.env.FONTCONFIG_PATH,
     SYSTEMROOT: process.env.SYSTEMROOT,
   })
+  return { ...allowed, NODE_ENV: process.env.NODE_ENV }
 }
 
 /**

apps/sim/scripts/test-isolated-vm-remote-e2e.ts

@@ -205,14 +205,15 @@ const CHECK_DEADLINE_MS = 30_000
 async function check(name: string, run: () => Promise<void>) {
   const started = performance.now()
   let deadline: ReturnType<typeof setTimeout> | undefined
+  let timedOut = false
   try {
     await Promise.race([
       run(),
       new Promise<never>((_, reject) => {
-        deadline = setTimeout(
-          () => reject(new Error(`Did not settle within ${CHECK_DEADLINE_MS}ms`)),
-          CHECK_DEADLINE_MS
-        )
+        deadline = setTimeout(() => {
+          timedOut = true
+          reject(new Error(`Did not settle within ${CHECK_DEADLINE_MS}ms`))
+        }, CHECK_DEADLINE_MS)
       }),
     ])
     checks.push({ name, status: 'passed', durationMs: Math.round(performance.now() - started) })
@@ -226,6 +227,9 @@ async function check(name: string, run: () => Promise<void>) {
     })
     logger.error(`FAIL ${name}`, { error: getErrorMessage(error) })
     process.exitCode = 1
+    // A check that never settled may still be running against the shared hosts and pool, so
+    // later checks would not be independent of it.
+    if (timedOut) throw new Error(`Aborting: "${name}" did not settle`)
   } finally {
     clearTimeout(deadline)
   }
@@ -271,8 +275,9 @@ async function run() {
   await check("enforces the app's limits on the remote worker", async () => {
     const result = await execute("console.log('y'.repeat(5000)); return 1")
     assert.equal(result.result, 1, result.error?.message)
+    assert(result.stdout.endsWith('[stdout truncated]\n'), 'stdout was not truncated')
     assert(
-      result.stdout.length < 2 * STDOUT_LIMIT_CHARS,
+      result.stdout.length <= STDOUT_LIMIT_CHARS + '\n[stdout truncated]\n'.length,
       `stdout was ${result.stdout.length} chars; the app limit is ${STDOUT_LIMIT_CHARS}`
     )
   })
@@ -340,6 +345,12 @@ async function run() {
     assert.equal(await closeCode(ws), 4401)
   })
 
+  await check('rejects an unparseable request target and keeps serving', async () => {
+    assert.equal(await upgradeStatus(probeHost, { path: '//[' }), 400)
+    const health = await fetch(`http://127.0.0.1:${probeHost.port}/health`)
+    assert.equal(health.status, 200)
+  })
+
   await check('rejects paths other than the worker endpoint', async () => {
     assert.equal(await upgradeStatus(probeHost, { path: '/v0/worker' }), 404)
   })