@@ -1,16 +1,21 @@
+/**
+ * @vitest-environment node
+ */
 import { describe, expect, it } from 'vitest'
-import { coldConnectionBudgetMs } from './retry'
+import { coldConnectionBudgetMs } from '@/lib/core/config/redis-budget'
 
 describe('coldConnectionBudgetMs', () => {
-  it('charges a dead handshake at two command deadlines when that exceeds the connect deadline', () => {
-    // Unauthenticated: SETNAME/SETINFO settle by timing out before INFO starts its own deadline.
+  it('charges an unanswered handshake at two command deadlines plus the half-close wait', () => {
+    // Unauthenticated: SETNAME/SETINFO time out, then INFO times out, then the
+    // half-closed socket waits for a FIN a wedged peer never sends.
     expect(
       coldConnectionBudgetMs({
         connectTimeoutMs: 1_000,
-        commandTimeoutMs: 2_000,
+        commandTimeoutMs: 5_000,
+        disconnectTimeoutMs: 2_000,
         reconnectDelayMs: 500,
       })
-    ).toBe(2 * 2_000 + 500 + 1_000)
+    ).toBe(2 * 5_000 + 2_000 + 500 + 1_000)
   })
 
   it('charges the connect deadline when a connection that never completes is the longer case', () => {
@@ -20,14 +25,20 @@ describe('coldConnectionBudgetMs', () => {
       coldConnectionBudgetMs({
         connectTimeoutMs: 10_000,
         commandTimeoutMs: 3_000,
+        disconnectTimeoutMs: 2_000,
         reconnectDelayMs: 500,
       })
     ).toBe(10_000 + 500 + 1_000)
   })
 
   it('always leaves room for the healthy attempt that follows the reconnect', () => {
     expect(
-      coldConnectionBudgetMs({ connectTimeoutMs: 0, commandTimeoutMs: 0, reconnectDelayMs: 0 })
+      coldConnectionBudgetMs({
+        connectTimeoutMs: 0,
+        commandTimeoutMs: 0,
+        disconnectTimeoutMs: 0,
+        reconnectDelayMs: 0,
+      })
     ).toBe(1_000)
   })
 })
@@ -0,0 +1,45 @@
+/** Generous room for the healthy handshake that follows a recovered stall; a real one takes tens of milliseconds. */
+const HEALTHY_HANDSHAKE_ALLOWANCE_MS = 1_000
+
+export interface ColdConnectionBudgetOptions {
+  /** The client's TCP connect deadline: bounds a connection that never completes at all. */
+  connectTimeoutMs: number
+  /** The client's per-command deadline: bounds a handshake command the server never answers. */
+  commandTimeoutMs: number
+  /** How long ioredis waits after half-closing a dead socket for the peer's FIN before destroying it. */
+  disconnectTimeoutMs: number
+  /** What the client's `retryStrategy` returns for its first reconnect. */
+  reconnectDelayMs: number
+}
+
+/**
+ * How long a wait for a cold ioredis connection must allow before giving up,
+ * if it is to survive one dead attempt and still see a healthy one land.
+ *
+ * A dead attempt is diagnosed by whichever deadline governs the phase it
+ * stalls in. A connect that never completes costs `connectTimeoutMs`, and the
+ * socket is destroyed outright. A connection that opens but whose handshake
+ * is never answered costs command deadlines — one with a password, because a
+ * timed-out `AUTH` is fatal; two without, because `CLIENT SETNAME`/`SETINFO`
+ * must settle, by timing out, before the `INFO` ready check starts its own —
+ * and then `disconnectTimeoutMs` more, because ioredis half-closes the socket
+ * and a peer that is wedged never answers with a FIN. The budget takes the
+ * larger phase so it holds for either URL shape without parsing it. Only then
+ * does `retryStrategy` run and a fresh attempt begin.
+ *
+ * A wait sized to a single deadline expires while the first attempt is still
+ * being diagnosed, so a configured retry can never be the thing that saves it.
+ *
+ * The guarantee is one dead attempt from a fresh or previously-ready client.
+ * ioredis feeds `retryStrategy` its running attempt count, so a wait that
+ * begins while the client is already deep in a reconnect loop faces larger
+ * delays this does not model. The handshake sequence is ioredis 5's; keep this
+ * beside the pinned client, not in a generic package.
+ */
+export function coldConnectionBudgetMs(options: ColdConnectionBudgetOptions): number {
+  const deadAttemptMs = Math.max(
+    options.connectTimeoutMs,
+    2 * options.commandTimeoutMs + options.disconnectTimeoutMs
+  )
+  return deadAttemptMs + options.reconnectDelayMs + HEALTHY_HANDSHAKE_ALLOWANCE_MS
+}
@@ -1,5 +1,4 @@
 import { createMockRedis } from '@sim/testing'
-import { coldConnectionBudgetMs } from '@sim/utils/retry'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const { mockEnv, MockRedisConstructor, mockLogger } = vi.hoisted(() => ({
@@ -48,15 +47,19 @@ vi.mock('ioredis', () => ({
 
 import {
   acquireLock,
+  CONNECT_TIMEOUT_MS,
   closeRedisConnection,
+  DISCONNECT_TIMEOUT_MS,
   describeRedisConnection,
   extendLock,
   getRedisClient,
   onRedisReconnect,
   resetForTesting,
+  SHARED_COMMAND_TIMEOUT_MS,
   sharedReconnectDelayMs,
   warmRedisConnection,
 } from '@/lib/core/config/redis'
+import { coldConnectionBudgetMs } from '@/lib/core/config/redis-budget'
 
 describe('redis config', () => {
   beforeEach(() => {
@@ -488,9 +491,12 @@ describe('redis config', () => {
       mockRedisInstance.status = 'connecting'
       const warm = warmRedisConnection()
 
-      // The dead attempt's own deadline, then the longest first reconnect
-      // delay: the moment a healthy second attempt can begin.
-      await vi.advanceTimersByTimeAsync(Math.max(10_000, 2 * 5_000) + sharedReconnectDelayMs(1, 1))
+      // The dead attempt's own diagnosis and half-close, then the longest first
+      // reconnect delay: the moment a healthy second attempt can begin.
+      await vi.advanceTimersByTimeAsync(
+        Math.max(CONNECT_TIMEOUT_MS, 2 * SHARED_COMMAND_TIMEOUT_MS + DISCONNECT_TIMEOUT_MS) +
+          sharedReconnectDelayMs(1, 1)
+      )
       const client = getRedisClient()
       Object.assign(client ?? {}, { status: 'ready' })
       client?.emit('ready')
@@ -504,8 +510,9 @@ describe('redis config', () => {
 
       await vi.advanceTimersByTimeAsync(
         coldConnectionBudgetMs({
-          connectTimeoutMs: 10_000,
-          commandTimeoutMs: 5_000,
+          connectTimeoutMs: CONNECT_TIMEOUT_MS,
+          commandTimeoutMs: SHARED_COMMAND_TIMEOUT_MS,
+          disconnectTimeoutMs: DISCONNECT_TIMEOUT_MS,
           reconnectDelayMs: sharedReconnectDelayMs(1, 1),
         })
       )
 
@@ -2,10 +2,10 @@ import { isIP } from 'node:net'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { randomFloat } from '@sim/utils/random'
-import { coldConnectionBudgetMs } from '@sim/utils/retry'
 import Redis from 'ioredis'
 import { env } from '@/lib/core/config/env'
 import { getConfiguredCacheProvider } from '@/lib/core/config/env-capabilities.server'
+import { coldConnectionBudgetMs } from '@/lib/core/config/redis-budget'
 
 const logger = createLogger('Redis')
 
@@ -46,17 +46,21 @@ function resolveRedisTlsOptions(url: string | undefined): { servername: string }
 export interface RedisConnectionDefaults {
   keepAlive: number
   connectTimeout: number
+  disconnectTimeout: number
   enableOfflineQueue: boolean
   tls?: { servername: string }
 }
 
-const CONNECT_TIMEOUT_MS = 10_000
+export const CONNECT_TIMEOUT_MS = 10_000
+/** ioredis's own default, stated so readiness budgets can be derived from it. */
+export const DISCONNECT_TIMEOUT_MS = 2_000
 
 export function getRedisConnectionDefaults(url: string | undefined): RedisConnectionDefaults {
   const tls = resolveRedisTlsOptions(url)
   return {
     keepAlive: 1000,
     connectTimeout: CONNECT_TIMEOUT_MS,
+    disconnectTimeout: DISCONNECT_TIMEOUT_MS,
     enableOfflineQueue: true,
     ...(tls ? { tls } : {}),
   }
@@ -211,7 +215,7 @@ export function describeRedisConnection(
 
 const PING_INTERVAL_MS = 15_000
 const MAX_PING_FAILURES = 2
-const SHARED_COMMAND_TIMEOUT_MS = 5_000
+export const SHARED_COMMAND_TIMEOUT_MS = 5_000
 const RECONNECT_BASE_MS = 1_000
 const RECONNECT_MAX_BASE_MS = 10_000
 const RECONNECT_JITTER_RATIO = 0.3
@@ -236,6 +240,7 @@ export function sharedReconnectDelayMs(times: number, jitter: number): number {
 const REDIS_WARMUP_TIMEOUT_MS = coldConnectionBudgetMs({
   connectTimeoutMs: CONNECT_TIMEOUT_MS,
   commandTimeoutMs: SHARED_COMMAND_TIMEOUT_MS,
+  disconnectTimeoutMs: DISCONNECT_TIMEOUT_MS,
   reconnectDelayMs: sharedReconnectDelayMs(1, 1),
 })
 
 
@@ -2,7 +2,6 @@
  * @vitest-environment node
  */
 import { EventEmitter } from 'node:events'
-import { coldConnectionBudgetMs } from '@sim/utils/retry'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 const { connection, mockRedisUrl, mockSubscribe, mockUnsubscribe } = vi.hoisted(() => ({
@@ -41,26 +40,30 @@ vi.mock('@/lib/core/config/redis', () => ({
     if (mockRedisUrl.error) throw mockRedisUrl.error
     return mockRedisUrl.value
   },
-  // Realistic defaults: the readiness budget has a connect-deadline term.
-  getRedisConnectionDefaults: () => ({ connectTimeout: 10_000 }),
+  // Realistic defaults: the readiness budget derives from these. The literals
+  // are inherent to mocking the module that exports the real constants.
+  getRedisConnectionDefaults: () => ({ connectTimeout: 10_000, disconnectTimeout: 2_000 }),
 }))
 
+import { coldConnectionBudgetMs } from '@/lib/core/config/redis-budget'
 import {
+  connectExecutionSignalHub,
   getExecutionSignalHub,
   publishLocalExecutionSignal,
-  warmExecutionSignalHub,
 } from '@/lib/execution/execution-signal'
 
 /** The readiness budget production derives from the options the subscriber was built with. */
 function readyBudgetMs(): number {
   const options = connection.options as {
     connectTimeout: number
     commandTimeout: number
+    disconnectTimeout: number
     retryStrategy: (attempt: number) => number
   }
   return coldConnectionBudgetMs({
     connectTimeoutMs: options.connectTimeout,
     commandTimeoutMs: options.commandTimeout,
+    disconnectTimeoutMs: options.disconnectTimeout,
     reconnectDelayMs: options.retryStrategy(1),
   })
 }
@@ -401,27 +404,33 @@ describe('ExecutionSignalHub', () => {
     }
   })
 
-  it('gives a subscribe that joins an in-flight warm-up its own full budget', async () => {
+  it('gives a subscribe that joins another in-flight wait its own full budget', async () => {
     vi.useFakeTimers()
     try {
       connection.status = 'connect'
-      const warm = warmExecutionSignalHub()
-      // Late joiner: the warm-up has almost spent its budget when this subscribe begins.
+      const hub = getExecutionSignalHub()
+      const first = hub.subscribe('execution-first', vi.fn())
+      const firstSettled = vi.fn()
+      void first.then(firstSettled, firstSettled)
+      // Late joiner: the first wait has almost spent its budget when this one begins.
       await vi.advanceTimersByTimeAsync(readyBudgetMs() - 1000)
-      const subscription = getExecutionSignalHub().subscribe('execution-1', vi.fn())
-      const settled = vi.fn()
-      void subscription.then(settled, settled)
+      const second = hub.subscribe('execution-second', vi.fn())
+      const secondSettled = vi.fn()
+      void second.then(secondSettled, secondSettled)
 
       await vi.advanceTimersByTimeAsync(1000)
-      await expect(warm).resolves.toBe(false)
-      // The warm-up's deadline was its own; the subscribe is still waiting.
-      expect(settled).not.toHaveBeenCalled()
+      await expect(first).rejects.toThrow('Timed out waiting for Redis subscriber readiness')
+      // The first deadline was its own; the second is still waiting on the shared signal.
+      expect(secondSettled).not.toHaveBeenCalled()
       expect(connection.client?.listenerCount('ready')).toBe(2)
 
       connection.status = 'ready'
       connection.client?.emit('ready')
-      await subscription
-      expect(mockSubscribe).toHaveBeenCalledOnce()
+      await second
+      expect(mockSubscribe).toHaveBeenCalledWith(
+        'execution:signal:execution-second',
+        'execution:cancel'
+      )
     } finally {
       vi.useRealTimers()
     }
@@ -472,42 +481,21 @@ describe('ExecutionSignalHub', () => {
     }
   })
 
-  it('warms to true once the subscriber becomes ready', async () => {
+  it('begins connecting when asked to connect ahead of a subscription', () => {
     connection.status = 'connecting'
-    const warm = warmExecutionSignalHub()
-
-    connection.status = 'ready'
-    connection.client?.emit('ready')
-
-    await expect(warm).resolves.toBe(true)
-  })
-
-  it('warms to false, not a rejection, when readiness never arrives', async () => {
-    vi.useFakeTimers()
-    try {
-      connection.status = 'connect'
-      const warm = warmExecutionSignalHub()
 
-      await vi.advanceTimersByTimeAsync(readyBudgetMs())
+    connectExecutionSignalHub()
 
-      await expect(warm).resolves.toBe(false)
-      expect(vi.getTimerCount()).toBe(0)
-    } finally {
-      vi.useRealTimers()
-    }
+    // Constructing the hub is what dials; the client exists before any subscribe.
+    expect(connection.client).toBeDefined()
+    expect(mockSubscribe).not.toHaveBeenCalled()
   })
 
-  it('reports not-warm instead of throwing when Redis is misconfigured', async () => {
-    // Start-up hooks call this; a throw there would fail the run attempt.
+  it('does not throw when Redis is misconfigured, leaving that to the first subscriber', () => {
     mockRedisUrl.error = new Error('Cache capability selected Redis but REDIS_URL is missing')
 
-    await expect(warmExecutionSignalHub()).resolves.toBe(false)
-  })
-
-  it('is trivially warm when signals are process-local', async () => {
-    mockRedisUrl.value = undefined
-
-    await expect(warmExecutionSignalHub()).resolves.toBe(true)
+    expect(() => connectExecutionSignalHub()).not.toThrow()
+    expect(connection.client).toBeUndefined()
   })
 
   it('uses a process-local signal hub when Redis is not configured', async () => {