@@ -459,7 +459,7 @@ export const env = createEnv({
     IVM_MAX_BROKERS_PER_EXECUTION:         z.string().optional().default('1000'),    // Max broker calls per sandbox task execution
     IVM_REMOTE_WORKER_URL:                 z.string().optional(),                    // ws(s)://<host>/v1/worker of a dedicated worker host; workers run there unless the isolated-vm-local-workers kill switch is on
     IVM_REMOTE_WORKER_TOKEN:               z.string().optional(),                    // Token the worker host expects (its IVM_HOST_TOKEN, >= 32 chars); required with IVM_REMOTE_WORKER_URL
-    IVM_REMOTE_WORKER_MAX_AGE_MS:          z.string().optional().default('120000'), // Age at which a worker host connection stops taking executions, so host deploys drain; keep below the host's deregistration delay
+    IVM_REMOTE_WORKER_MAX_AGE_MS:          z.coerce.number().int().positive().optional(), // (default 120000) Age at which a worker host connection stops taking executions, so host deploys drain; keep below the host's deregistration delay
 
     // Knowledge Base Processing Configuration - Shared across all processing methods
     KB_CONFIG_MAX_DURATION:                z.number().optional().default(600),     // Max processing duration in seconds (10 minutes)
 
@@ -195,24 +195,26 @@ server.on('upgrade', (req, socket, head) => {
 function authenticate(ws, socket) {
   pendingConnections++
   let settled = false
+  let authenticated = false
   let preAuthBytes = 0
   // Counts what arrives on the TCP socket, so an oversized frame is cut off as it streams in
   // rather than after ws has buffered all of it.
   const onPreAuthData = (chunk) => {
     // ws parses each chunk before this listener runs, so the chunk that authenticated still arrives.
-    if (settled) return
+    // A refused connection stays counted: its peer may ignore the close and keep sending.
+    if (authenticated) return
     preAuthBytes += chunk.length
     if (preAuthBytes <= MAX_PRE_AUTH_BYTES) return
-    settle()
-    log('warn', 'Rejected worker connection', { reason: 'too much data before authentication' })
+    if (settle()) {
+      log('warn', 'Rejected worker connection', { reason: 'too much data before authentication' })
+    }
     socket.destroy()
   }
   const settle = () => {
     if (settled) return false
     settled = true
     pendingConnections--
     clearTimeout(authTimeout)
-    socket.off('data', onPreAuthData)
     return true
   }
   const refuse = (reason) => {
@@ -237,6 +239,8 @@ function authenticate(ws, socket) {
       return
     }
     if (!settle()) return
+    authenticated = true
+    socket.off('data', onPreAuthData)
     if (draining) {
       ws.close(1012, 'draining')
       return
 
@@ -6,7 +6,7 @@ import { createLogger } from '@sim/logger'
 import { getErrorMessage, toError } from '@sim/utils/errors'
 import { filterUndefined } from '@sim/utils/object'
 import { randomFloat } from '@sim/utils/random'
-import { env } from '@/lib/core/config/env'
+import { env, envNumber } from '@/lib/core/config/env'
 import { getConfiguredCacheProvider } from '@/lib/core/config/env-capabilities.server'
 import { isFeatureEnabled } from '@/lib/core/config/feature-flags'
 import { getRedisClient } from '@/lib/core/config/redis'
@@ -154,7 +154,10 @@ const MAX_EXECUTIONS_PER_WORKER = Number.parseInt(env.IVM_MAX_EXECUTIONS_PER_WOR
  * delay and never tells the host, so this bound is what drains a host deploy or scale-in: infra
  * sets that delay to at least this age plus the longest execution a release keeps alive.
  */
-const REMOTE_WORKER_MAX_AGE_MS = Number.parseInt(env.IVM_REMOTE_WORKER_MAX_AGE_MS) || 120_000
+const REMOTE_WORKER_MAX_AGE_MS = envNumber(env.IVM_REMOTE_WORKER_MAX_AGE_MS, 120_000, {
+  min: 1000,
+  integer: true,
+})
 /** Spreads retirements so a pool's connections don't all reopen at once; only ever shortens the age. */
 const REMOTE_WORKER_MAX_AGE_JITTER = 0.2
 /** Times one execution moves to another worker because its connection closed before it was sent. */
@@ -1380,15 +1383,17 @@ function dispatchToWorker(
       workerInfo.retiring = true
       if (workerInfo.activeExecutions === 0) cleanupWorker(workerInfo.id)
       // The owner's state may have been released above; look it up again rather than queue on it.
-      queueMicrotask(() =>
+      queueMicrotask(() => {
+        // An abort in between already settled the caller and has no queue entry to remove.
+        if (state.cancelled) return
         enqueueExecution(
           getOrCreateOwnerState(ownerState.ownerKey, ownerState.weight),
           req,
           resolve,
           state,
           brokers
         )
-      )
+      })
       return
     }
     resolve({
@@ -1549,7 +1554,6 @@ export async function executeInIsolatedVM(
 ): Promise<IsolatedVMExecutionResult> {
   const ownerKey = normalizeOwnerKey(req.ownerKey)
   const ownerWeight = normalizeOwnerWeight(req.ownerWeight)
-  const ownerState = getOrCreateOwnerState(ownerKey, ownerWeight)
   const brokers = options?.brokers
   const signal = options?.signal
 
@@ -1642,6 +1646,9 @@ export async function executeInIsolatedVM(
   // An undetermined lease cannot reject the execution: the per-process pool and
   // the per-owner active/queued limits above still bound this work.
 
+  // Looked up only after the waits above: another execution finishing meanwhile can release
+  // this owner's state, and work queued on a released state is never dispatched.
+  const ownerState = getOrCreateOwnerState(ownerKey, ownerWeight)
   const state: ExecutionState = {
     cancelled: false,
     runInOutboundScope: captureOutboundScope(),
 
@@ -199,6 +199,60 @@ function upgradeStatus(host: Host, options: { path?: string } = {}): Promise<num
   })
 }
 
+/** A client-to-server text frame header, masked with a zero key so the payload goes as is. */
+function maskedTextFrameHeader(payloadBytes: number): Buffer {
+  if (payloadBytes < 126) return Buffer.from([0x81, 0x80 | payloadBytes, 0, 0, 0, 0])
+  const header = Buffer.alloc(14)
+  header[0] = 0x81
+  header[1] = 0x80 | 127
+  header.writeBigUInt64BE(BigInt(payloadBytes), 2)
+  return header
+}
+
+/**
+ * Upgrades a raw socket, optionally sends `firstFrame`, then declares a 32 MiB frame (under the
+ * host's message limit) and streams 1 MiB of it without ever authenticating. Resolves with
+ * whether the host closed the connection within a second.
+ */
+async function streamUnauthenticated(host: Host, firstFrame?: string): Promise<'closed' | 'open'> {
+  const socket = connect(host.port, '127.0.0.1')
+  socket.on('error', () => {})
+  const closed = new Promise<void>((resolve) => socket.once('close', () => resolve()))
+  await new Promise<void>((resolve) => socket.once('connect', () => resolve()))
+  const handshake = new Promise<string>((resolve) =>
+    socket.once('data', (data: Buffer) => resolve(data.toString()))
+  )
+  socket.write(
+    [
+      'GET /v1/worker HTTP/1.1',
+      `Host: 127.0.0.1:${host.port}`,
+      'Upgrade: websocket',
+      'Connection: Upgrade',
+      'Sec-WebSocket-Version: 13',
+      `Sec-WebSocket-Key: ${Buffer.from(generateRandomBytes(16)).toString('base64')}`,
+      '',
+      '',
+    ].join('\r\n')
+  )
+  const response = await handshake
+  assert.match(response, /^HTTP\/1\.1 101/, `Expected the upgrade to be accepted: ${response}`)
+  if (firstFrame) {
+    const payload = Buffer.from(firstFrame)
+    socket.write(Buffer.concat([maskedTextFrameHeader(payload.length), payload]))
+  }
+  socket.write(maskedTextFrameHeader(32 * 1024 * 1024))
+  const chunk = Buffer.alloc(64 * 1024)
+  for (let sent = 0; sent < 1024 * 1024 && !socket.destroyed; sent += chunk.length) {
+    socket.write(chunk)
+  }
+  const outcome = await Promise.race([
+    closed.then(() => 'closed' as const),
+    sleep(1000).then(() => 'open' as const),
+  ])
+  socket.destroy()
+  return outcome
+}
+
 /** Opens a connection and sends `firstFrame`; resolves with the socket once it opens. */
 function openSocket(host: Host, firstFrame: string): Promise<WebSocket> {
   return new Promise((resolve, reject) => {
@@ -312,9 +366,15 @@ async function run() {
   const { executeInIsolatedVM } = await import('@/lib/execution/isolated-vm')
 
   // The isolate has no timers; a slow upstream holds an execution open the way real ones run long.
+  // `/hold` answers only when the check releases it, so an inspection never races the execution.
+  let onHeld: (release: () => void) => void = () => {}
   const slowServer = createHttpServer((req, res) => {
-    const delay = Number(new URL(req.url ?? '/', 'http://slow').searchParams.get('ms'))
-    setTimeout(() => res.end('done'), delay)
+    const url = new URL(req.url ?? '/', 'http://slow')
+    if (url.pathname === '/hold') {
+      onHeld(() => res.end('done'))
+      return
+    }
+    setTimeout(() => res.end('done'), Number(url.searchParams.get('ms')))
   })
   await new Promise<void>((resolve) => slowServer.listen(0, '127.0.0.1', resolve))
   const slowAddress = slowServer.address()
@@ -345,12 +405,19 @@ async function run() {
 
   /** Inspects workers while an execution holds one open, since an idle worker can retire at any time. */
   const whileAWorkerRuns = async (inspect: () => void) => {
-    const running = execute(slowFetch(500))
-    await sleep(200)
+    const held = new Promise<() => void>((resolve) => {
+      onHeld = resolve
+    })
+    const running = execute(
+      `const r = await fetch('http://127.0.0.1:${slowAddress.port}/hold'); return await r.text()`
+    )
+    const release = await held
     try {
       inspect()
     } finally {
-      await running
+      release()
+      const result = await running
+      assert.equal(result.result, 'done', result.error?.message)
     }
   }
 
@@ -437,48 +504,16 @@ async function run() {
   })
 
   await check('drops a connection that streams a large frame before authenticating', async () => {
-    const socket = connect(probeHost.port, '127.0.0.1')
-    socket.on('error', () => {})
-    const closed = new Promise<void>((resolve) => socket.once('close', () => resolve()))
-    await new Promise<void>((resolve) => socket.once('connect', () => resolve()))
-    const handshake = new Promise<string>((resolve) =>
-      socket.once('data', (data: Buffer) => resolve(data.toString()))
-    )
-    socket.write(
-      [
-        'GET /v1/worker HTTP/1.1',
-        `Host: 127.0.0.1:${probeHost.port}`,
-        'Upgrade: websocket',
-        'Connection: Upgrade',
-        'Sec-WebSocket-Version: 13',
-        `Sec-WebSocket-Key: ${Buffer.from(generateRandomBytes(16)).toString('base64')}`,
-        '',
-        '',
-      ].join('\r\n')
-    )
-    const response = await handshake
-    assert.match(response, /^HTTP\/1\.1 101/, `Expected the upgrade to be accepted: ${response}`)
-
-    // A masked text frame that declares 32 MiB, under the host's message limit, and never authenticates.
-    const header = Buffer.alloc(14)
-    header[0] = 0x81
-    header[1] = 0x80 | 127
-    header.writeBigUInt64BE(BigInt(32 * 1024 * 1024), 2)
-    socket.write(header)
-    const chunk = Buffer.alloc(64 * 1024)
-    for (let sent = 0; sent < 1024 * 1024 && !socket.destroyed; sent += chunk.length) {
-      socket.write(chunk)
-    }
-    const outcome = await Promise.race([
-      closed.then(() => 'closed' as const),
-      sleep(1000).then(() => 'open' as const),
-    ])
-    socket.destroy()
-    assert.equal(outcome, 'closed', 'The host kept buffering an unauthenticated frame')
+    assert.equal(await streamUnauthenticated(probeHost), 'closed', 'The host kept buffering')
     const health = await fetch(`http://127.0.0.1:${probeHost.port}/health`)
     assert.equal(health.status, 200)
   })
 
+  await check('drops a refused connection that ignores the close and keeps sending', async () => {
+    const refusedAuth = authFrame(generateRandomHex(32))
+    assert.equal(await streamUnauthenticated(probeHost, refusedAuth), 'closed')
+  })
+
   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`)