v0.5.76: posthog improvements, readme updates

v0.5.75: search modal overhaul, helm chart updates, run from block, terminal and visual debugging improvements
v0.5.74: autolayout improvements, clerk integration, auth enforcements
2026-01-30 01:07:59 -05:00 · 2026-01-29 00:13:19 -08:00 · 2026-01-28 22:54:13 -08:00 · 2026-01-27 20:37:39 -08:00 · 2026-01-26 22:04:35 -08:00 · 2026-01-25 23:42:27 -08:00
33 changed files with 972 additions and 2293 deletions
--- a/apps/sim/app/api/auth/socket-token/route.ts
+++ b/apps/sim/app/api/auth/socket-token/route.ts
@@ -4,22 +4,22 @@ import { auth } from '@/lib/auth'
 import { isAuthDisabled } from '@/lib/core/config/feature-flags'

 export async function POST() {
-  if (isAuthDisabled) {
-    return NextResponse.json({ token: 'anonymous-socket-token' })
-  }
-
  try {
+    if (isAuthDisabled) {
+      return NextResponse.json({ token: 'anonymous-socket-token' })
+    }
+
    const hdrs = await headers()
    const response = await auth.api.generateOneTimeToken({
      headers: hdrs,
    })

-    if (!response?.token) {
-      return NextResponse.json({ error: 'Authentication required' }, { status: 401 })
+    if (!response) {
+      return NextResponse.json({ error: 'Failed to generate token' }, { status: 500 })
    }

    return NextResponse.json({ token: response.token })
-  } catch {
+  } catch (error) {
    return NextResponse.json({ error: 'Failed to generate token' }, { status: 500 })
  }
 }
--- a/apps/sim/app/api/workflows/[id]/deployments/[version]/revert/route.ts
+++ b/apps/sim/app/api/workflows/[id]/deployments/[version]/revert/route.ts
@@ -97,10 +97,7 @@ export async function POST(
      const socketServerUrl = env.SOCKET_SERVER_URL || 'http://localhost:3002'
      await fetch(`${socketServerUrl}/api/workflow-reverted`, {
        method: 'POST',
-        headers: {
-          'Content-Type': 'application/json',
-          'x-api-key': env.INTERNAL_API_SECRET,
-        },
+        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ workflowId: id, timestamp: Date.now() }),
      })
    } catch (e) {
--- a/apps/sim/app/api/workflows/[id]/route.ts
+++ b/apps/sim/app/api/workflows/[id]/route.ts
@@ -361,10 +361,7 @@ export async function DELETE(
      const socketUrl = env.SOCKET_SERVER_URL || 'http://localhost:3002'
      const socketResponse = await fetch(`${socketUrl}/api/workflow-deleted`, {
        method: 'POST',
-        headers: {
-          'Content-Type': 'application/json',
-          'x-api-key': env.INTERNAL_API_SECRET,
-        },
+        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ workflowId }),
      })

--- a/apps/sim/app/api/workflows/[id]/state/route.ts
+++ b/apps/sim/app/api/workflows/[id]/state/route.ts
@@ -254,10 +254,7 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
      const socketUrl = env.SOCKET_SERVER_URL || 'http://localhost:3002'
      const notifyResponse = await fetch(`${socketUrl}/api/workflow-updated`, {
        method: 'POST',
-        headers: {
-          'Content-Type': 'application/json',
-          'x-api-key': env.INTERNAL_API_SECRET,
-        },
+        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ workflowId }),
      })

--- a/apps/sim/app/workspace/providers/socket-provider.tsx
+++ b/apps/sim/app/workspace/providers/socket-provider.tsx
@@ -17,19 +17,6 @@ import { getEnv } from '@/lib/core/config/env'

 const logger = createLogger('SocketContext')

-const TAB_SESSION_ID_KEY = 'sim_tab_session_id'
-
-function getTabSessionId(): string {
-  if (typeof window === 'undefined') return ''
-
-  let tabSessionId = sessionStorage.getItem(TAB_SESSION_ID_KEY)
-  if (!tabSessionId) {
-    tabSessionId = crypto.randomUUID()
-    sessionStorage.setItem(TAB_SESSION_ID_KEY, tabSessionId)
-  }
-  return tabSessionId
-}
-
 interface User {
  id: string
  name?: string
@@ -49,13 +36,11 @@ interface SocketContextType {
  socket: Socket | null
  isConnected: boolean
  isConnecting: boolean
-  authFailed: boolean
  currentWorkflowId: string | null
  currentSocketId: string | null
  presenceUsers: PresenceUser[]
  joinWorkflow: (workflowId: string) => void
  leaveWorkflow: () => void
-  retryConnection: () => void
  emitWorkflowOperation: (
    operation: string,
    target: string,
@@ -78,6 +63,8 @@ interface SocketContextType {

  onCursorUpdate: (handler: (data: any) => void) => void
  onSelectionUpdate: (handler: (data: any) => void) => void
+  onUserJoined: (handler: (data: any) => void) => void
+  onUserLeft: (handler: (data: any) => void) => void
  onWorkflowDeleted: (handler: (data: any) => void) => void
  onWorkflowReverted: (handler: (data: any) => void) => void
  onOperationConfirmed: (handler: (data: any) => void) => void
@@ -88,13 +75,11 @@ const SocketContext = createContext<SocketContextType>({
  socket: null,
  isConnected: false,
  isConnecting: false,
-  authFailed: false,
  currentWorkflowId: null,
  currentSocketId: null,
  presenceUsers: [],
  joinWorkflow: () => {},
  leaveWorkflow: () => {},
-  retryConnection: () => {},
  emitWorkflowOperation: () => {},
  emitSubblockUpdate: () => {},
  emitVariableUpdate: () => {},
@@ -105,6 +90,8 @@ const SocketContext = createContext<SocketContextType>({
  onVariableUpdate: () => {},
  onCursorUpdate: () => {},
  onSelectionUpdate: () => {},
+  onUserJoined: () => {},
+  onUserLeft: () => {},
  onWorkflowDeleted: () => {},
  onWorkflowReverted: () => {},
  onOperationConfirmed: () => {},
@@ -125,43 +112,33 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
  const [currentWorkflowId, setCurrentWorkflowId] = useState<string | null>(null)
  const [currentSocketId, setCurrentSocketId] = useState<string | null>(null)
  const [presenceUsers, setPresenceUsers] = useState<PresenceUser[]>([])
-  const [authFailed, setAuthFailed] = useState(false)
  const initializedRef = useRef(false)
-  const socketRef = useRef<Socket | null>(null)

  const params = useParams()
  const urlWorkflowId = params?.workflowId as string | undefined
-  const urlWorkflowIdRef = useRef(urlWorkflowId)
-  urlWorkflowIdRef.current = urlWorkflowId

  const eventHandlers = useRef<{
    workflowOperation?: (data: any) => void
    subblockUpdate?: (data: any) => void
    variableUpdate?: (data: any) => void
+
    cursorUpdate?: (data: any) => void
    selectionUpdate?: (data: any) => void
+    userJoined?: (data: any) => void
+    userLeft?: (data: any) => void
    workflowDeleted?: (data: any) => void
    workflowReverted?: (data: any) => void
    operationConfirmed?: (data: any) => void
    operationFailed?: (data: any) => void
  }>({})

-  const positionUpdateTimeouts = useRef<Map<string, number>>(new Map())
-  const isRejoiningRef = useRef<boolean>(false)
-  const pendingPositionUpdates = useRef<Map<string, any>>(new Map())
-
  const generateSocketToken = async (): Promise<string> => {
    const res = await fetch('/api/auth/socket-token', {
      method: 'POST',
      credentials: 'include',
      headers: { 'cache-control': 'no-store' },
    })
-    if (!res.ok) {
-      if (res.status === 401) {
-        throw new Error('Authentication required')
-      }
-      throw new Error('Failed to generate socket token')
-    }
+    if (!res.ok) throw new Error('Failed to generate socket token')
    const body = await res.json().catch(() => ({}))
    const token = body?.token
    if (!token || typeof token !== 'string') throw new Error('Invalid socket token')
@@ -171,11 +148,6 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
  useEffect(() => {
    if (!user?.id) return

-    if (authFailed) {
-      logger.info('Socket initialization skipped - auth failed, waiting for retry')
-      return
-    }
-
    if (initializedRef.current || socket || isConnecting) {
      logger.info('Socket already exists or is connecting, skipping initialization')
      return
@@ -208,11 +180,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
              cb({ token: freshToken })
            } catch (error) {
              logger.error('Failed to generate fresh token for connection:', error)
-              if (error instanceof Error && error.message === 'Authentication required') {
-                // True auth failure - pass null token, server will reject with "Authentication required"
-                cb({ token: null })
-              }
-              // For server errors, don't call cb - connection will timeout and Socket.IO will retry
+              cb({ token: null })
            }
          },
        })
@@ -226,19 +194,26 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
            connected: socketInstance.connected,
            transport: socketInstance.io.engine?.transport?.name,
          })
-          // Note: join-workflow is handled by the useEffect watching isConnected
+
+          if (urlWorkflowId) {
+            logger.info(`Joining workflow room after connection: ${urlWorkflowId}`)
+            socketInstance.emit('join-workflow', {
+              workflowId: urlWorkflowId,
+            })
+            setCurrentWorkflowId(urlWorkflowId)
+          }
        })

        socketInstance.on('disconnect', (reason) => {
          setIsConnected(false)
          setIsConnecting(false)
          setCurrentSocketId(null)
-          setCurrentWorkflowId(null)
-          setPresenceUsers([])

          logger.info('Socket disconnected', {
            reason,
          })
+
+          setPresenceUsers([])
        })

        socketInstance.on('connect_error', (error: any) => {
@@ -251,34 +226,24 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
            transport: error.transport,
          })

-          // Check if this is an authentication failure
-          const isAuthError =
+          if (
            error.message?.includes('Token validation failed') ||
            error.message?.includes('Authentication failed') ||
            error.message?.includes('Authentication required')
-
-          if (isAuthError) {
+          ) {
            logger.warn(
-              'Authentication failed - stopping reconnection attempts. User may need to refresh/re-login.'
+              'Authentication failed - this could indicate session expiry or token generation issues'
            )
-            // Stop reconnection attempts to prevent infinite loop
-            socketInstance.disconnect()
-            // Reset state to allow re-initialization when session is restored
-            setSocket(null)
-            setAuthFailed(true)
-            initializedRef.current = false
          }
        })

        socketInstance.on('reconnect', (attemptNumber) => {
-          setIsConnected(true)
          setCurrentSocketId(socketInstance.id ?? null)
          logger.info('Socket reconnected successfully', {
            attemptNumber,
            socketId: socketInstance.id,
            transport: socketInstance.io.engine?.transport?.name,
          })
-          // Note: join-workflow is handled by the useEffect watching isConnected
        })

        socketInstance.on('reconnect_attempt', (attemptNumber) => {
@@ -319,26 +284,6 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
          })
        })

-        // Handle join workflow success - confirms room membership with presence list
-        socketInstance.on('join-workflow-success', ({ workflowId, presenceUsers }) => {
-          isRejoiningRef.current = false
-          // Ignore stale success responses from previous navigation
-          if (workflowId !== urlWorkflowIdRef.current) {
-            logger.debug(`Ignoring stale join-workflow-success for ${workflowId}`)
-            return
-          }
-          setCurrentWorkflowId(workflowId)
-          setPresenceUsers(presenceUsers || [])
-          logger.info(`Successfully joined workflow room: ${workflowId}`, {
-            presenceCount: presenceUsers?.length || 0,
-          })
-        })
-
-        socketInstance.on('join-workflow-error', ({ error }) => {
-          isRejoiningRef.current = false
-          logger.error('Failed to join workflow:', error)
-        })
-
        socketInstance.on('workflow-operation', (data) => {
          eventHandlers.current.workflowOperation?.(data)
        })
@@ -353,13 +298,10 @@ export function SocketProvider({ children, user }: SocketProviderProps) {

        socketInstance.on('workflow-deleted', (data) => {
          logger.warn(`Workflow ${data.workflowId} has been deleted`)
-          setCurrentWorkflowId((current) => {
-            if (current === data.workflowId) {
-              setPresenceUsers([])
-              return null
-            }
-            return current
-          })
+          if (currentWorkflowId === data.workflowId) {
+            setCurrentWorkflowId(null)
+            setPresenceUsers([])
+          }
          eventHandlers.current.workflowDeleted?.(data)
        })

@@ -502,35 +444,25 @@ export function SocketProvider({ children, user }: SocketProviderProps) {

        socketInstance.on('operation-forbidden', (error) => {
          logger.warn('Operation forbidden:', error)
+        })

-          if (error?.type === 'SESSION_ERROR') {
-            const workflowId = urlWorkflowIdRef.current
-
-            if (workflowId && !isRejoiningRef.current) {
-              isRejoiningRef.current = true
-              logger.info(`Session expired, rejoining workflow: ${workflowId}`)
-              socketInstance.emit('join-workflow', {
-                workflowId,
-                tabSessionId: getTabSessionId(),
-              })
-            }
-          }
+        socketInstance.on('operation-confirmed', (data) => {
+          logger.debug('Operation confirmed:', data)
        })

        socketInstance.on('workflow-state', async (workflowData) => {
          logger.info('Received workflow state from server')

          if (workflowData?.state) {
-            try {
-              await rehydrateWorkflowStores(workflowData.id, workflowData.state, 'workflow-state')
-            } catch (error) {
-              logger.error('Error rehydrating workflow state:', error)
-            }
+            await rehydrateWorkflowStores(workflowData.id, workflowData.state, 'workflow-state')
          }
        })

-        socketRef.current = socketInstance
        setSocket(socketInstance)
+
+        return () => {
+          socketInstance.close()
+        }
      } catch (error) {
        logger.error('Failed to initialize socket with token:', error)
        setIsConnecting(false)
@@ -545,20 +477,12 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
      })
      positionUpdateTimeouts.current.clear()
      pendingPositionUpdates.current.clear()
-
-      // Close socket on unmount
-      if (socketRef.current) {
-        logger.info('Closing socket connection on unmount')
-        socketRef.current.close()
-        socketRef.current = null
-      }
    }
-  }, [user?.id, authFailed])
+  }, [user?.id])

  useEffect(() => {
    if (!socket || !isConnected || !urlWorkflowId) return

-    // Skip if already in the correct room
    if (currentWorkflowId === urlWorkflowId) return

    logger.info(
@@ -573,10 +497,19 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
    logger.info(`Joining workflow room: ${urlWorkflowId}`)
    socket.emit('join-workflow', {
      workflowId: urlWorkflowId,
-      tabSessionId: getTabSessionId(),
    })
+    setCurrentWorkflowId(urlWorkflowId)
  }, [socket, isConnected, urlWorkflowId, currentWorkflowId])

+  useEffect(() => {
+    return () => {
+      if (socket) {
+        logger.info('Cleaning up socket connection on unmount')
+        socket.disconnect()
+      }
+    }
+  }, [])
+
  const joinWorkflow = useCallback(
    (workflowId: string) => {
      if (!socket || !user?.id) {
@@ -597,9 +530,8 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
      logger.info(`Joining workflow: ${workflowId}`)
      socket.emit('join-workflow', {
        workflowId,
-        tabSessionId: getTabSessionId(),
      })
-      // currentWorkflowId will be set by join-workflow-success handler
+      setCurrentWorkflowId(workflowId)
    },
    [socket, user, currentWorkflowId]
  )
@@ -607,13 +539,10 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
  const leaveWorkflow = useCallback(() => {
    if (socket && currentWorkflowId) {
      logger.info(`Leaving workflow: ${currentWorkflowId}`)
-      import('@/stores/operation-queue/store')
-        .then(({ useOperationQueueStore }) => {
-          useOperationQueueStore.getState().cancelOperationsForWorkflow(currentWorkflowId)
-        })
-        .catch((error) => {
-          logger.warn('Failed to cancel operations for workflow:', error)
-        })
+      try {
+        const { useOperationQueueStore } = require('@/stores/operation-queue/store')
+        useOperationQueueStore.getState().cancelOperationsForWorkflow(currentWorkflowId)
+      } catch {}
      socket.emit('leave-workflow')
      setCurrentWorkflowId(null)
      setPresenceUsers([])
@@ -626,20 +555,8 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
    }
  }, [socket, currentWorkflowId])

-  /**
-   * Retry socket connection after auth failure.
-   * Call this when user has re-authenticated (e.g., after login redirect).
-   */
-  const retryConnection = useCallback(() => {
-    if (!authFailed) {
-      logger.info('retryConnection called but no auth failure - ignoring')
-      return
-    }
-    logger.info('Retrying socket connection after auth failure')
-    setAuthFailed(false)
-    // initializedRef.current was already reset in connect_error handler
-    // Effect will re-run and attempt connection
-  }, [authFailed])
+  const positionUpdateTimeouts = useRef<Map<string, number>>(new Map())
+  const pendingPositionUpdates = useRef<Map<string, any>>(new Map())

  const emitWorkflowOperation = useCallback(
    (operation: string, target: string, payload: any, operationId?: string) => {
@@ -799,6 +716,14 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
    eventHandlers.current.selectionUpdate = handler
  }, [])

+  const onUserJoined = useCallback((handler: (data: any) => void) => {
+    eventHandlers.current.userJoined = handler
+  }, [])
+
+  const onUserLeft = useCallback((handler: (data: any) => void) => {
+    eventHandlers.current.userLeft = handler
+  }, [])
+
  const onWorkflowDeleted = useCallback((handler: (data: any) => void) => {
    eventHandlers.current.workflowDeleted = handler
  }, [])
@@ -820,13 +745,11 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
      socket,
      isConnected,
      isConnecting,
-      authFailed,
      currentWorkflowId,
      currentSocketId,
      presenceUsers,
      joinWorkflow,
      leaveWorkflow,
-      retryConnection,
      emitWorkflowOperation,
      emitSubblockUpdate,
      emitVariableUpdate,
@@ -837,6 +760,8 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
      onVariableUpdate,
      onCursorUpdate,
      onSelectionUpdate,
+      onUserJoined,
+      onUserLeft,
      onWorkflowDeleted,
      onWorkflowReverted,
      onOperationConfirmed,
@@ -846,13 +771,11 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
      socket,
      isConnected,
      isConnecting,
-      authFailed,
      currentWorkflowId,
      currentSocketId,
      presenceUsers,
      joinWorkflow,
      leaveWorkflow,
-      retryConnection,
      emitWorkflowOperation,
      emitSubblockUpdate,
      emitVariableUpdate,
@@ -863,6 +786,8 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
      onVariableUpdate,
      onCursorUpdate,
      onSelectionUpdate,
+      onUserJoined,
+      onUserLeft,
      onWorkflowDeleted,
      onWorkflowReverted,
      onOperationConfirmed,
--- a/apps/sim/executor/execution/edge-manager.test.ts
+++ b/apps/sim/executor/execution/edge-manager.test.ts
@@ -2417,177 +2417,4 @@ describe('EdgeManager', () => {
      expect(successReady).toContain(targetId)
    })
  })
-
-  describe('Condition with loop downstream - deactivation propagation', () => {
-    it('should deactivate nodes after loop when condition branch containing loop is deactivated', () => {
-      // Scenario: condition → (if) → sentinel_start → loopBody → sentinel_end → (loop_exit) → after_loop
-      //                     → (else) → other_branch
-      // When condition takes "else" path, the entire if-branch including nodes after the loop should be deactivated
-      const conditionId = 'condition'
-      const sentinelStartId = 'sentinel-start'
-      const loopBodyId = 'loop-body'
-      const sentinelEndId = 'sentinel-end'
-      const afterLoopId = 'after-loop'
-      const otherBranchId = 'other-branch'
-
-      const conditionNode = createMockNode(conditionId, [
-        { target: sentinelStartId, sourceHandle: 'condition-if' },
-        { target: otherBranchId, sourceHandle: 'condition-else' },
-      ])
-
-      const sentinelStartNode = createMockNode(
-        sentinelStartId,
-        [{ target: loopBodyId }],
-        [conditionId]
-      )
-
-      const loopBodyNode = createMockNode(
-        loopBodyId,
-        [{ target: sentinelEndId }],
-        [sentinelStartId]
-      )
-
-      const sentinelEndNode = createMockNode(
-        sentinelEndId,
-        [
-          { target: sentinelStartId, sourceHandle: 'loop_continue' },
-          { target: afterLoopId, sourceHandle: 'loop_exit' },
-        ],
-        [loopBodyId]
-      )
-
-      const afterLoopNode = createMockNode(afterLoopId, [], [sentinelEndId])
-      const otherBranchNode = createMockNode(otherBranchId, [], [conditionId])
-
-      const nodes = new Map<string, DAGNode>([
-        [conditionId, conditionNode],
-        [sentinelStartId, sentinelStartNode],
-        [loopBodyId, loopBodyNode],
-        [sentinelEndId, sentinelEndNode],
-        [afterLoopId, afterLoopNode],
-        [otherBranchId, otherBranchNode],
-      ])
-
-      const dag = createMockDAG(nodes)
-      const edgeManager = new EdgeManager(dag)
-
-      // Condition selects "else" branch, deactivating the "if" branch (which contains the loop)
-      const readyNodes = edgeManager.processOutgoingEdges(conditionNode, { selectedOption: 'else' })
-
-      // Only otherBranch should be ready
-      expect(readyNodes).toContain(otherBranchId)
-      expect(readyNodes).not.toContain(sentinelStartId)
-
-      // afterLoop should NOT be ready - its incoming edge from sentinel_end should be deactivated
-      expect(readyNodes).not.toContain(afterLoopId)
-
-      // Verify that countActiveIncomingEdges returns 0 for afterLoop
-      // (meaning the loop_exit edge was properly deactivated)
-      // Note: isNodeReady returns true when all edges are deactivated (no pending deps),
-      // but the node won't be in readyNodes since it wasn't reached via an active path
-      expect(edgeManager.isNodeReady(afterLoopNode)).toBe(true) // All edges deactivated = no blocking deps
-    })
-
-    it('should deactivate nodes after parallel when condition branch containing parallel is deactivated', () => {
-      // Similar scenario with parallel instead of loop
-      const conditionId = 'condition'
-      const parallelStartId = 'parallel-start'
-      const parallelBodyId = 'parallel-body'
-      const parallelEndId = 'parallel-end'
-      const afterParallelId = 'after-parallel'
-      const otherBranchId = 'other-branch'
-
-      const conditionNode = createMockNode(conditionId, [
-        { target: parallelStartId, sourceHandle: 'condition-if' },
-        { target: otherBranchId, sourceHandle: 'condition-else' },
-      ])
-
-      const parallelStartNode = createMockNode(
-        parallelStartId,
-        [{ target: parallelBodyId }],
-        [conditionId]
-      )
-
-      const parallelBodyNode = createMockNode(
-        parallelBodyId,
-        [{ target: parallelEndId }],
-        [parallelStartId]
-      )
-
-      const parallelEndNode = createMockNode(
-        parallelEndId,
-        [{ target: afterParallelId, sourceHandle: 'parallel_exit' }],
-        [parallelBodyId]
-      )
-
-      const afterParallelNode = createMockNode(afterParallelId, [], [parallelEndId])
-      const otherBranchNode = createMockNode(otherBranchId, [], [conditionId])
-
-      const nodes = new Map<string, DAGNode>([
-        [conditionId, conditionNode],
-        [parallelStartId, parallelStartNode],
-        [parallelBodyId, parallelBodyNode],
-        [parallelEndId, parallelEndNode],
-        [afterParallelId, afterParallelNode],
-        [otherBranchId, otherBranchNode],
-      ])
-
-      const dag = createMockDAG(nodes)
-      const edgeManager = new EdgeManager(dag)
-
-      // Condition selects "else" branch
-      const readyNodes = edgeManager.processOutgoingEdges(conditionNode, { selectedOption: 'else' })
-
-      expect(readyNodes).toContain(otherBranchId)
-      expect(readyNodes).not.toContain(parallelStartId)
-      expect(readyNodes).not.toContain(afterParallelId)
-      // isNodeReady returns true when all edges are deactivated (no pending deps)
-      expect(edgeManager.isNodeReady(afterParallelNode)).toBe(true)
-    })
-
-    it('should still correctly handle normal loop exit (not deactivate when loop runs)', () => {
-      // When a loop actually executes and exits normally, after_loop should become ready
-      const sentinelStartId = 'sentinel-start'
-      const loopBodyId = 'loop-body'
-      const sentinelEndId = 'sentinel-end'
-      const afterLoopId = 'after-loop'
-
-      const sentinelStartNode = createMockNode(sentinelStartId, [{ target: loopBodyId }])
-
-      const loopBodyNode = createMockNode(
-        loopBodyId,
-        [{ target: sentinelEndId }],
-        [sentinelStartId]
-      )
-
-      const sentinelEndNode = createMockNode(
-        sentinelEndId,
-        [
-          { target: sentinelStartId, sourceHandle: 'loop_continue' },
-          { target: afterLoopId, sourceHandle: 'loop_exit' },
-        ],
-        [loopBodyId]
-      )
-
-      const afterLoopNode = createMockNode(afterLoopId, [], [sentinelEndId])
-
-      const nodes = new Map<string, DAGNode>([
-        [sentinelStartId, sentinelStartNode],
-        [loopBodyId, loopBodyNode],
-        [sentinelEndId, sentinelEndNode],
-        [afterLoopId, afterLoopNode],
-      ])
-
-      const dag = createMockDAG(nodes)
-      const edgeManager = new EdgeManager(dag)
-
-      // Simulate sentinel_end completing with loop_exit (loop is done)
-      const readyNodes = edgeManager.processOutgoingEdges(sentinelEndNode, {
-        selectedRoute: 'loop_exit',
-      })
-
-      // afterLoop should be ready
-      expect(readyNodes).toContain(afterLoopId)
-    })
-  })
 })
--- a/apps/sim/executor/execution/edge-manager.ts
+++ b/apps/sim/executor/execution/edge-manager.ts
@@ -243,7 +243,7 @@ export class EdgeManager {
    }

    for (const [, outgoingEdge] of targetNode.outgoingEdges) {
-      if (!this.isBackwardsEdge(outgoingEdge.sourceHandle)) {
+      if (!this.isControlEdge(outgoingEdge.sourceHandle)) {
        this.deactivateEdgeAndDescendants(
          targetId,
          outgoingEdge.target,
--- a/apps/sim/hooks/use-collaborative-workflow.ts
+++ b/apps/sim/hooks/use-collaborative-workflow.ts
@@ -119,6 +119,8 @@ export function useCollaborativeWorkflow() {
    onWorkflowOperation,
    onSubblockUpdate,
    onVariableUpdate,
+    onUserJoined,
+    onUserLeft,
    onWorkflowDeleted,
    onWorkflowReverted,
    onOperationConfirmed,
@@ -482,6 +484,14 @@ export function useCollaborativeWorkflow() {
      }
    }

+    const handleUserJoined = (data: any) => {
+      logger.info(`User joined: ${data.userName}`)
+    }
+
+    const handleUserLeft = (data: any) => {
+      logger.info(`User left: ${data.userId}`)
+    }
+
    const handleWorkflowDeleted = (data: any) => {
      const { workflowId } = data
      logger.warn(`Workflow ${workflowId} has been deleted`)
@@ -590,17 +600,26 @@ export function useCollaborativeWorkflow() {
      failOperation(operationId, retryable)
    }

+    // Register event handlers
    onWorkflowOperation(handleWorkflowOperation)
    onSubblockUpdate(handleSubblockUpdate)
    onVariableUpdate(handleVariableUpdate)
+    onUserJoined(handleUserJoined)
+    onUserLeft(handleUserLeft)
    onWorkflowDeleted(handleWorkflowDeleted)
    onWorkflowReverted(handleWorkflowReverted)
    onOperationConfirmed(handleOperationConfirmed)
    onOperationFailed(handleOperationFailed)
+
+    return () => {
+      // Cleanup handled by socket context
+    }
  }, [
    onWorkflowOperation,
    onSubblockUpdate,
    onVariableUpdate,
+    onUserJoined,
+    onUserLeft,
    onWorkflowDeleted,
    onWorkflowReverted,
    onOperationConfirmed,
--- a/apps/sim/lib/copilot/tools/client/knowledge/knowledge-base.ts
+++ b/apps/sim/lib/copilot/tools/client/knowledge/knowledge-base.ts
@@ -10,7 +10,6 @@ import {
  type KnowledgeBaseArgs,
 } from '@/lib/copilot/tools/shared/schemas'
 import { useCopilotStore } from '@/stores/panel/copilot/store'
-import { useWorkflowRegistry } from '@/stores/workflows/registry/store'

 /**
 * Client tool for knowledge base operations
@@ -103,19 +102,7 @@ export class KnowledgeBaseClientTool extends BaseClientTool {
    const logger = createLogger('KnowledgeBaseClientTool')
    try {
      this.setState(ClientToolCallState.executing)
-
-      // Get the workspace ID from the workflow registry hydration state
-      const { hydration } = useWorkflowRegistry.getState()
-      const workspaceId = hydration.workspaceId
-
-      // Build payload with workspace ID included in args
-      const payload: KnowledgeBaseArgs = {
-        ...(args || { operation: 'list' }),
-        args: {
-          ...(args?.args || {}),
-          workspaceId: workspaceId || undefined,
-        },
-      }
+      const payload: KnowledgeBaseArgs = { ...(args || { operation: 'list' }) }

      const res = await fetch('/api/copilot/execute-copilot-server-tool', {
        method: 'POST',
--- a/apps/sim/lib/copilot/tools/server/workflow/edit-workflow.ts
+++ b/apps/sim/lib/copilot/tools/server/workflow/edit-workflow.ts
@@ -2508,10 +2508,6 @@ async function validateWorkflowSelectorIds(
    for (const subBlockConfig of blockConfig.subBlocks) {
      if (!SELECTOR_TYPES.has(subBlockConfig.type)) continue

-      // Skip oauth-input - credentials are pre-validated before edit application
-      // This allows existing collaborator credentials to remain untouched
-      if (subBlockConfig.type === 'oauth-input') continue
-
      const subBlockValue = blockData.subBlocks?.[subBlockConfig.id]?.value
      if (!subBlockValue) continue

@@ -2577,295 +2573,6 @@ async function validateWorkflowSelectorIds(
  return errors
 }

-/**
- * Pre-validates credential and apiKey inputs in operations before they are applied.
- * - Validates oauth-input (credential) IDs belong to the user
- * - Filters out apiKey inputs for hosted models when isHosted is true
- * - Also validates credentials and apiKeys in nestedNodes (blocks inside loop/parallel)
- * Returns validation errors for any removed inputs.
- */
-async function preValidateCredentialInputs(
-  operations: EditWorkflowOperation[],
-  context: { userId: string },
-  workflowState?: Record<string, unknown>
-): Promise<{ filteredOperations: EditWorkflowOperation[]; errors: ValidationError[] }> {
-  const { isHosted } = await import('@/lib/core/config/feature-flags')
-  const { getHostedModels } = await import('@/providers/utils')
-
-  const logger = createLogger('PreValidateCredentials')
-  const errors: ValidationError[] = []
-
-  // Collect credential and apiKey inputs that need validation/filtering
-  const credentialInputs: Array<{
-    operationIndex: number
-    blockId: string
-    blockType: string
-    fieldName: string
-    value: string
-    nestedBlockId?: string
-  }> = []
-
-  const hostedApiKeyInputs: Array<{
-    operationIndex: number
-    blockId: string
-    blockType: string
-    model: string
-    nestedBlockId?: string
-  }> = []
-
-  const hostedModelsLower = isHosted ? new Set(getHostedModels().map((m) => m.toLowerCase())) : null
-
-  /**
-   * Collect credential inputs from a block's inputs based on its block config
-   */
-  function collectCredentialInputs(
-    blockConfig: ReturnType<typeof getBlock>,
-    inputs: Record<string, unknown>,
-    opIndex: number,
-    blockId: string,
-    blockType: string,
-    nestedBlockId?: string
-  ) {
-    if (!blockConfig) return
-
-    for (const subBlockConfig of blockConfig.subBlocks) {
-      if (subBlockConfig.type !== 'oauth-input') continue
-
-      const inputValue = inputs[subBlockConfig.id]
-      if (!inputValue || typeof inputValue !== 'string' || inputValue.trim() === '') continue
-
-      credentialInputs.push({
-        operationIndex: opIndex,
-        blockId,
-        blockType,
-        fieldName: subBlockConfig.id,
-        value: inputValue,
-        nestedBlockId,
-      })
-    }
-  }
-
-  /**
-   * Check if apiKey should be filtered for a block with the given model
-   */
-  function collectHostedApiKeyInput(
-    inputs: Record<string, unknown>,
-    modelValue: string | undefined,
-    opIndex: number,
-    blockId: string,
-    blockType: string,
-    nestedBlockId?: string
-  ) {
-    if (!hostedModelsLower || !inputs.apiKey) return
-    if (!modelValue || typeof modelValue !== 'string') return
-
-    if (hostedModelsLower.has(modelValue.toLowerCase())) {
-      hostedApiKeyInputs.push({
-        operationIndex: opIndex,
-        blockId,
-        blockType,
-        model: modelValue,
-        nestedBlockId,
-      })
-    }
-  }
-
-  operations.forEach((op, opIndex) => {
-    // Process main block inputs
-    if (op.params?.inputs && op.params?.type) {
-      const blockConfig = getBlock(op.params.type)
-      if (blockConfig) {
-        // Collect credentials from main block
-        collectCredentialInputs(
-          blockConfig,
-          op.params.inputs as Record<string, unknown>,
-          opIndex,
-          op.block_id,
-          op.params.type
-        )
-
-        // Check for apiKey inputs on hosted models
-        let modelValue = (op.params.inputs as Record<string, unknown>).model as string | undefined
-
-        // For edit operations, if model is not being changed, check existing block's model
-        if (
-          !modelValue &&
-          op.operation_type === 'edit' &&
-          (op.params.inputs as Record<string, unknown>).apiKey &&
-          workflowState
-        ) {
-          const existingBlock = (workflowState.blocks as Record<string, unknown>)?.[op.block_id] as
-            | Record<string, unknown>
-            | undefined
-          const existingSubBlocks = existingBlock?.subBlocks as Record<string, unknown> | undefined
-          const existingModelSubBlock = existingSubBlocks?.model as
-            | Record<string, unknown>
-            | undefined
-          modelValue = existingModelSubBlock?.value as string | undefined
-        }
-
-        collectHostedApiKeyInput(
-          op.params.inputs as Record<string, unknown>,
-          modelValue,
-          opIndex,
-          op.block_id,
-          op.params.type
-        )
-      }
-    }
-
-    // Process nested nodes (blocks inside loop/parallel containers)
-    const nestedNodes = op.params?.nestedNodes as
-      | Record<string, Record<string, unknown>>
-      | undefined
-    if (nestedNodes) {
-      Object.entries(nestedNodes).forEach(([childId, childBlock]) => {
-        const childType = childBlock.type as string | undefined
-        const childInputs = childBlock.inputs as Record<string, unknown> | undefined
-        if (!childType || !childInputs) return
-
-        const childBlockConfig = getBlock(childType)
-        if (!childBlockConfig) return
-
-        // Collect credentials from nested block
-        collectCredentialInputs(
-          childBlockConfig,
-          childInputs,
-          opIndex,
-          op.block_id,
-          childType,
-          childId
-        )
-
-        // Check for apiKey inputs on hosted models in nested block
-        const modelValue = childInputs.model as string | undefined
-        collectHostedApiKeyInput(childInputs, modelValue, opIndex, op.block_id, childType, childId)
-      })
-    }
-  })
-
-  const hasCredentialsToValidate = credentialInputs.length > 0
-  const hasHostedApiKeysToFilter = hostedApiKeyInputs.length > 0
-
-  if (!hasCredentialsToValidate && !hasHostedApiKeysToFilter) {
-    return { filteredOperations: operations, errors }
-  }
-
-  // Deep clone operations so we can modify them
-  const filteredOperations = structuredClone(operations)
-
-  // Filter out apiKey inputs for hosted models and add validation errors
-  if (hasHostedApiKeysToFilter) {
-    logger.info('Filtering apiKey inputs for hosted models', { count: hostedApiKeyInputs.length })
-
-    for (const apiKeyInput of hostedApiKeyInputs) {
-      const op = filteredOperations[apiKeyInput.operationIndex]
-
-      // Handle nested block apiKey filtering
-      if (apiKeyInput.nestedBlockId) {
-        const nestedNodes = op.params?.nestedNodes as
-          | Record<string, Record<string, unknown>>
-          | undefined
-        const nestedBlock = nestedNodes?.[apiKeyInput.nestedBlockId]
-        const nestedInputs = nestedBlock?.inputs as Record<string, unknown> | undefined
-        if (nestedInputs?.apiKey) {
-          nestedInputs.apiKey = undefined
-          logger.debug('Filtered apiKey for hosted model in nested block', {
-            parentBlockId: apiKeyInput.blockId,
-            nestedBlockId: apiKeyInput.nestedBlockId,
-            model: apiKeyInput.model,
-          })
-
-          errors.push({
-            blockId: apiKeyInput.nestedBlockId,
-            blockType: apiKeyInput.blockType,
-            field: 'apiKey',
-            value: '[redacted]',
-            error: `Cannot set API key for hosted model "${apiKeyInput.model}" - API keys are managed by the platform when using hosted models`,
-          })
-        }
-      } else if (op.params?.inputs?.apiKey) {
-        // Handle main block apiKey filtering
-        op.params.inputs.apiKey = undefined
-        logger.debug('Filtered apiKey for hosted model', {
-          blockId: apiKeyInput.blockId,
-          model: apiKeyInput.model,
-        })
-
-        errors.push({
-          blockId: apiKeyInput.blockId,
-          blockType: apiKeyInput.blockType,
-          field: 'apiKey',
-          value: '[redacted]',
-          error: `Cannot set API key for hosted model "${apiKeyInput.model}" - API keys are managed by the platform when using hosted models`,
-        })
-      }
-    }
-  }
-
-  // Validate credential inputs
-  if (hasCredentialsToValidate) {
-    logger.info('Pre-validating credential inputs', {
-      credentialCount: credentialInputs.length,
-      userId: context.userId,
-    })
-
-    const allCredentialIds = credentialInputs.map((c) => c.value)
-    const validationResult = await validateSelectorIds('oauth-input', allCredentialIds, context)
-    const invalidSet = new Set(validationResult.invalid)
-
-    if (invalidSet.size > 0) {
-      for (const credInput of credentialInputs) {
-        if (!invalidSet.has(credInput.value)) continue
-
-        const op = filteredOperations[credInput.operationIndex]
-
-        // Handle nested block credential removal
-        if (credInput.nestedBlockId) {
-          const nestedNodes = op.params?.nestedNodes as
-            | Record<string, Record<string, unknown>>
-            | undefined
-          const nestedBlock = nestedNodes?.[credInput.nestedBlockId]
-          const nestedInputs = nestedBlock?.inputs as Record<string, unknown> | undefined
-          if (nestedInputs?.[credInput.fieldName]) {
-            delete nestedInputs[credInput.fieldName]
-            logger.info('Removed invalid credential from nested block', {
-              parentBlockId: credInput.blockId,
-              nestedBlockId: credInput.nestedBlockId,
-              field: credInput.fieldName,
-              invalidValue: credInput.value,
-            })
-          }
-        } else if (op.params?.inputs?.[credInput.fieldName]) {
-          // Handle main block credential removal
-          delete op.params.inputs[credInput.fieldName]
-          logger.info('Removed invalid credential from operation', {
-            blockId: credInput.blockId,
-            field: credInput.fieldName,
-            invalidValue: credInput.value,
-          })
-        }
-
-        const warningInfo = validationResult.warning ? `. ${validationResult.warning}` : ''
-        const errorBlockId = credInput.nestedBlockId ?? credInput.blockId
-        errors.push({
-          blockId: errorBlockId,
-          blockType: credInput.blockType,
-          field: credInput.fieldName,
-          value: credInput.value,
-          error: `Invalid credential ID "${credInput.value}" - credential does not exist or user doesn't have access${warningInfo}`,
-        })
-      }
-
-      logger.warn('Filtered out invalid credentials', {
-        invalidCount: invalidSet.size,
-      })
-    }
-  }
-
-  return { filteredOperations, errors }
-}
-
 async function getCurrentWorkflowStateFromDb(
  workflowId: string
 ): Promise<{ workflowState: any; subBlockValues: Record<string, Record<string, any>> }> {
@@ -2950,29 +2657,12 @@ export const editWorkflowServerTool: BaseServerTool<EditWorkflowParams, any> = {
    // Get permission config for the user
    const permissionConfig = context?.userId ? await getUserPermissionConfig(context.userId) : null

-    // Pre-validate credential and apiKey inputs before applying operations
-    // This filters out invalid credentials and apiKeys for hosted models
-    let operationsToApply = operations
-    const credentialErrors: ValidationError[] = []
-    if (context?.userId) {
-      const { filteredOperations, errors: credErrors } = await preValidateCredentialInputs(
-        operations,
-        { userId: context.userId },
-        workflowState
-      )
-      operationsToApply = filteredOperations
-      credentialErrors.push(...credErrors)
-    }
-
    // Apply operations directly to the workflow state
    const {
      state: modifiedWorkflowState,
      validationErrors,
      skippedItems,
-    } = applyOperationsToWorkflowState(workflowState, operationsToApply, permissionConfig)
-
-    // Add credential validation errors
-    validationErrors.push(...credentialErrors)
+    } = applyOperationsToWorkflowState(workflowState, operations, permissionConfig)

    // Get workspaceId for selector validation
    let workspaceId: string | undefined
--- a/apps/sim/package.json
+++ b/apps/sim/package.json
@@ -74,7 +74,6 @@
    "@react-email/components": "^0.0.34",
    "@react-email/render": "2.0.0",
    "@sim/logger": "workspace:*",
-    "@socket.io/redis-adapter": "8.3.0",
    "@t3-oss/env-nextjs": "0.13.4",
    "@tanstack/react-query": "5.90.8",
    "@tanstack/react-query-devtools": "5.90.2",
@@ -145,7 +144,6 @@
    "react-simple-code-editor": "^0.14.1",
    "react-window": "2.2.3",
    "reactflow": "^11.11.4",
-    "redis": "5.10.0",
    "rehype-autolink-headings": "^7.1.0",
    "rehype-slug": "^6.0.0",
    "remark-gfm": "4.0.1",
--- a/apps/sim/socket/config/socket.ts
+++ b/apps/sim/socket/config/socket.ts
@@ -1,7 +1,5 @@
 import type { Server as HttpServer } from 'http'
 import { createLogger } from '@sim/logger'
-import { createAdapter } from '@socket.io/redis-adapter'
-import { createClient, type RedisClientType } from 'redis'
 import { Server } from 'socket.io'
 import { env } from '@/lib/core/config/env'
 import { isProd } from '@/lib/core/config/feature-flags'
@@ -9,16 +7,9 @@ import { getBaseUrl } from '@/lib/core/utils/urls'

 const logger = createLogger('SocketIOConfig')

-/** Socket.IO ping timeout - how long to wait for pong before considering connection dead */
-const PING_TIMEOUT_MS = 60000
-/** Socket.IO ping interval - how often to send ping packets */
-const PING_INTERVAL_MS = 25000
-/** Maximum HTTP buffer size for Socket.IO messages */
-const MAX_HTTP_BUFFER_SIZE = 1e6
-
-let adapterPubClient: RedisClientType | null = null
-let adapterSubClient: RedisClientType | null = null
-
+/**
+ * Get allowed origins for Socket.IO CORS configuration
+ */
 function getAllowedOrigins(): string[] {
  const allowedOrigins = [
    getBaseUrl(),
@@ -33,10 +24,11 @@ function getAllowedOrigins(): string[] {
 }

 /**
- * Create and configure a Socket.IO server instance.
- * If REDIS_URL is configured, adds Redis adapter for cross-pod broadcasting.
+ * Create and configure a Socket.IO server instance
+ * @param httpServer - The HTTP server instance to attach Socket.IO to
+ * @returns Configured Socket.IO server instance
 */
-export async function createSocketIOServer(httpServer: HttpServer): Promise<Server> {
+export function createSocketIOServer(httpServer: HttpServer): Server {
  const allowedOrigins = getAllowedOrigins()

  const io = new Server(httpServer, {
@@ -44,110 +36,31 @@ export async function createSocketIOServer(httpServer: HttpServer): Promise<Serv
      origin: allowedOrigins,
      methods: ['GET', 'POST', 'OPTIONS'],
      allowedHeaders: ['Content-Type', 'Authorization', 'Cookie', 'socket.io'],
-      credentials: true,
+      credentials: true, // Enable credentials to accept cookies
    },
-    transports: ['websocket', 'polling'],
-    allowEIO3: true,
-    pingTimeout: PING_TIMEOUT_MS,
-    pingInterval: PING_INTERVAL_MS,
-    maxHttpBufferSize: MAX_HTTP_BUFFER_SIZE,
+    transports: ['websocket', 'polling'], // WebSocket first, polling as fallback
+    allowEIO3: true, // Keep legacy support for compatibility
+    pingTimeout: 60000, // Back to original conservative setting
+    pingInterval: 25000, // Back to original interval
+    maxHttpBufferSize: 1e6,
    cookie: {
      name: 'io',
      path: '/',
      httpOnly: true,
-      sameSite: 'none',
-      secure: isProd,
+      sameSite: 'none', // Required for cross-origin cookies
+      secure: isProd, // HTTPS in production
    },
  })

-  if (env.REDIS_URL) {
-    logger.info('Configuring Socket.IO Redis adapter...')
-
-    const redisOptions = {
-      url: env.REDIS_URL,
-      socket: {
-        reconnectStrategy: (retries: number) => {
-          if (retries > 10) {
-            logger.error('Redis adapter reconnection failed after 10 attempts')
-            return new Error('Redis adapter reconnection failed')
-          }
-          const delay = Math.min(retries * 100, 3000)
-          logger.warn(`Redis adapter reconnecting in ${delay}ms (attempt ${retries})`)
-          return delay
-        },
-      },
-    }
-
-    // Create separate clients for pub and sub (recommended for reliability)
-    adapterPubClient = createClient(redisOptions)
-    adapterSubClient = createClient(redisOptions)
-
-    adapterPubClient.on('error', (err) => {
-      logger.error('Redis adapter pub client error:', err)
-    })
-
-    adapterSubClient.on('error', (err) => {
-      logger.error('Redis adapter sub client error:', err)
-    })
-
-    adapterPubClient.on('ready', () => {
-      logger.info('Redis adapter pub client ready')
-    })
-
-    adapterSubClient.on('ready', () => {
-      logger.info('Redis adapter sub client ready')
-    })
-
-    await Promise.all([adapterPubClient.connect(), adapterSubClient.connect()])
-
-    io.adapter(createAdapter(adapterPubClient, adapterSubClient))
-
-    logger.info('Socket.IO Redis adapter connected - cross-pod broadcasting enabled')
-  } else {
-    logger.warn('REDIS_URL not configured - running in single-pod mode')
-  }
-
  logger.info('Socket.IO server configured with:', {
    allowedOrigins: allowedOrigins.length,
    transports: ['websocket', 'polling'],
-    pingTimeout: PING_TIMEOUT_MS,
-    pingInterval: PING_INTERVAL_MS,
-    maxHttpBufferSize: MAX_HTTP_BUFFER_SIZE,
+    pingTimeout: 60000,
+    pingInterval: 25000,
+    maxHttpBufferSize: 1e6,
    cookieSecure: isProd,
    corsCredentials: true,
-    redisAdapter: !!env.REDIS_URL,
  })

  return io
 }
-
-/**
- * Clean up Redis adapter connections.
- * Call this during graceful shutdown.
- */
-export async function shutdownSocketIOAdapter(): Promise<void> {
-  const closePromises: Promise<void>[] = []
-
-  if (adapterPubClient) {
-    closePromises.push(
-      adapterPubClient.quit().then(() => {
-        logger.info('Redis adapter pub client closed')
-        adapterPubClient = null
-      })
-    )
-  }
-
-  if (adapterSubClient) {
-    closePromises.push(
-      adapterSubClient.quit().then(() => {
-        logger.info('Redis adapter sub client closed')
-        adapterSubClient = null
-      })
-    )
-  }
-
-  if (closePromises.length > 0) {
-    await Promise.all(closePromises)
-    logger.info('Socket.IO Redis adapter shutdown complete')
-  }
-}
--- a/apps/sim/socket/handlers/connection.ts
+++ b/apps/sim/socket/handlers/connection.ts
@@ -1,12 +1,17 @@
 import { createLogger } from '@sim/logger'
-import { cleanupPendingSubblocksForSocket } from '@/socket/handlers/subblocks'
-import { cleanupPendingVariablesForSocket } from '@/socket/handlers/variables'
+import type { HandlerDependencies } from '@/socket/handlers/workflow'
 import type { AuthenticatedSocket } from '@/socket/middleware/auth'
-import type { IRoomManager } from '@/socket/rooms'
+import type { RoomManager } from '@/socket/rooms/manager'

 const logger = createLogger('ConnectionHandlers')

-export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
+export function setupConnectionHandlers(
+  socket: AuthenticatedSocket,
+  deps: HandlerDependencies | RoomManager
+) {
+  const roomManager =
+    deps instanceof Object && 'roomManager' in deps ? deps.roomManager : (deps as RoomManager)
+
  socket.on('error', (error) => {
    logger.error(`Socket ${socket.id} error:`, error)
  })
@@ -15,22 +20,13 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager
    logger.error(`Socket ${socket.id} connection error:`, error)
  })

-  socket.on('disconnect', async (reason) => {
-    try {
-      // Clean up pending debounce entries for this socket to prevent memory leaks
-      cleanupPendingSubblocksForSocket(socket.id)
-      cleanupPendingVariablesForSocket(socket.id)
+  socket.on('disconnect', (reason) => {
+    const workflowId = roomManager.getWorkflowIdForSocket(socket.id)
+    const session = roomManager.getUserSession(socket.id)

-      const workflowId = await roomManager.removeUserFromRoom(socket.id)
-
-      if (workflowId) {
-        await roomManager.broadcastPresenceUpdate(workflowId)
-        logger.info(
-          `Socket ${socket.id} disconnected from workflow ${workflowId} (reason: ${reason})`
-        )
-      }
-    } catch (error) {
-      logger.error(`Error handling disconnect for socket ${socket.id}:`, error)
+    if (workflowId && session) {
+      roomManager.cleanupUserFromRoom(socket.id, workflowId)
+      roomManager.broadcastPresenceUpdate(workflowId)
    }
  })
 }
--- a/apps/sim/socket/handlers/index.ts
+++ b/apps/sim/socket/handlers/index.ts
@@ -5,9 +5,16 @@ import { setupSubblocksHandlers } from '@/socket/handlers/subblocks'
 import { setupVariablesHandlers } from '@/socket/handlers/variables'
 import { setupWorkflowHandlers } from '@/socket/handlers/workflow'
 import type { AuthenticatedSocket } from '@/socket/middleware/auth'
-import type { IRoomManager } from '@/socket/rooms'
+import type { RoomManager, UserPresence, WorkflowRoom } from '@/socket/rooms/manager'

-export function setupAllHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
+export type { UserPresence, WorkflowRoom }
+
+/**
+ * Sets up all socket event handlers for an authenticated socket connection
+ * @param socket - The authenticated socket instance
+ * @param roomManager - Room manager instance for state management
+ */
+export function setupAllHandlers(socket: AuthenticatedSocket, roomManager: RoomManager) {
  setupWorkflowHandlers(socket, roomManager)
  setupOperationsHandlers(socket, roomManager)
  setupSubblocksHandlers(socket, roomManager)
@@ -15,3 +22,12 @@ export function setupAllHandlers(socket: AuthenticatedSocket, roomManager: IRoom
  setupPresenceHandlers(socket, roomManager)
  setupConnectionHandlers(socket, roomManager)
 }
+
+export {
+  setupWorkflowHandlers,
+  setupOperationsHandlers,
+  setupSubblocksHandlers,
+  setupVariablesHandlers,
+  setupPresenceHandlers,
+  setupConnectionHandlers,
+}
--- a/apps/sim/socket/handlers/operations.ts
+++ b/apps/sim/socket/handlers/operations.ts
@@ -10,41 +10,38 @@ import {
  WORKFLOW_OPERATIONS,
 } from '@/socket/constants'
 import { persistWorkflowOperation } from '@/socket/database/operations'
+import type { HandlerDependencies } from '@/socket/handlers/workflow'
 import type { AuthenticatedSocket } from '@/socket/middleware/auth'
 import { checkRolePermission } from '@/socket/middleware/permissions'
-import type { IRoomManager } from '@/socket/rooms'
+import type { RoomManager } from '@/socket/rooms/manager'
 import { WorkflowOperationSchema } from '@/socket/validation/schemas'

 const logger = createLogger('OperationsHandlers')

-export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
+export function setupOperationsHandlers(
+  socket: AuthenticatedSocket,
+  deps: HandlerDependencies | RoomManager
+) {
+  const roomManager =
+    deps instanceof Object && 'roomManager' in deps ? deps.roomManager : (deps as RoomManager)
  socket.on('workflow-operation', async (data) => {
-    const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
-    const session = await roomManager.getUserSession(socket.id)
+    const workflowId = roomManager.getWorkflowIdForSocket(socket.id)
+    const session = roomManager.getUserSession(socket.id)

    if (!workflowId || !session) {
-      socket.emit('operation-forbidden', {
-        type: 'SESSION_ERROR',
-        message: 'Session expired, please rejoin workflow',
+      socket.emit('error', {
+        type: 'NOT_JOINED',
+        message: 'Not joined to any workflow',
      })
-      if (data?.operationId) {
-        socket.emit('operation-failed', { operationId: data.operationId, error: 'Session expired' })
-      }
      return
    }

-    const hasRoom = await roomManager.hasWorkflowRoom(workflowId)
-    if (!hasRoom) {
-      socket.emit('operation-forbidden', {
+    const room = roomManager.getWorkflowRoom(workflowId)
+    if (!room) {
+      socket.emit('error', {
        type: 'ROOM_NOT_FOUND',
        message: 'Workflow room not found',
      })
-      if (data?.operationId) {
-        socket.emit('operation-failed', {
-          operationId: data.operationId,
-          error: 'Workflow room not found',
-        })
-      }
      return
    }

@@ -63,18 +60,16 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
        isPositionUpdate && 'commit' in payload ? payload.commit === true : false
      const operationTimestamp = isPositionUpdate ? timestamp : Date.now()

-      // Get user presence for permission checking
-      const users = await roomManager.getWorkflowUsers(workflowId)
-      const userPresence = users.find((u) => u.socketId === socket.id)
-
      // Skip permission checks for non-committed position updates (broadcasts only, no persistence)
      if (isPositionUpdate && !commitPositionUpdate) {
        // Update last activity
+        const userPresence = room.users.get(socket.id)
        if (userPresence) {
-          await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() })
+          userPresence.lastActivity = Date.now()
        }
      } else {
        // Check permissions from cached role for all other operations
+        const userPresence = room.users.get(socket.id)
        if (!userPresence) {
          logger.warn(`User presence not found for socket ${socket.id}`)
          socket.emit('operation-forbidden', {
@@ -83,13 +78,10 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
            operation,
            target,
          })
-          if (operationId) {
-            socket.emit('operation-failed', { operationId, error: 'User session not found' })
-          }
          return
        }

-        await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() })
+        userPresence.lastActivity = Date.now()

        // Check permissions using cached role (no DB query)
        const permissionCheck = checkRolePermission(userPresence.role, operation)
@@ -140,7 +132,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
            timestamp: operationTimestamp,
            userId: session.userId,
          })
-          await roomManager.updateRoomLastModified(workflowId)
+          room.lastModified = Date.now()

          if (operationId) {
            socket.emit('operation-confirmed', {
@@ -186,7 +178,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
            timestamp: operationTimestamp,
            userId: session.userId,
          })
-          await roomManager.updateRoomLastModified(workflowId)
+          room.lastModified = Date.now()

          if (operationId) {
            socket.emit('operation-confirmed', { operationId, serverTimestamp: Date.now() })
@@ -219,7 +211,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
          userId: session.userId,
        })

-        await roomManager.updateRoomLastModified(workflowId)
+        room.lastModified = Date.now()

        const broadcastData = {
          operation,
@@ -259,7 +251,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
          userId: session.userId,
        })

-        await roomManager.updateRoomLastModified(workflowId)
+        room.lastModified = Date.now()

        const broadcastData = {
          operation,
@@ -296,7 +288,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
          userId: session.userId,
        })

-        await roomManager.updateRoomLastModified(workflowId)
+        room.lastModified = Date.now()

        socket.to(workflowId).emit('workflow-operation', {
          operation,
@@ -328,7 +320,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
          userId: session.userId,
        })

-        await roomManager.updateRoomLastModified(workflowId)
+        room.lastModified = Date.now()

        socket.to(workflowId).emit('workflow-operation', {
          operation,
@@ -357,7 +349,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
          userId: session.userId,
        })

-        await roomManager.updateRoomLastModified(workflowId)
+        room.lastModified = Date.now()

        socket.to(workflowId).emit('workflow-operation', {
          operation,
@@ -389,7 +381,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
          userId: session.userId,
        })

-        await roomManager.updateRoomLastModified(workflowId)
+        room.lastModified = Date.now()

        socket.to(workflowId).emit('workflow-operation', {
          operation,
@@ -421,7 +413,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
          userId: session.userId,
        })

-        await roomManager.updateRoomLastModified(workflowId)
+        room.lastModified = Date.now()

        socket.to(workflowId).emit('workflow-operation', {
          operation,
@@ -453,7 +445,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
          userId: session.userId,
        })

-        await roomManager.updateRoomLastModified(workflowId)
+        room.lastModified = Date.now()

        socket.to(workflowId).emit('workflow-operation', {
          operation,
@@ -482,7 +474,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
          userId: session.userId,
        })

-        await roomManager.updateRoomLastModified(workflowId)
+        room.lastModified = Date.now()

        socket.to(workflowId).emit('workflow-operation', {
          operation,
@@ -511,24 +503,27 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
        userId: session.userId,
      })

-      await roomManager.updateRoomLastModified(workflowId)
+      room.lastModified = Date.now()

      const broadcastData = {
        operation,
        target,
        payload,
-        timestamp: operationTimestamp,
+        timestamp: operationTimestamp, // Preserve client timestamp for position updates
        senderId: socket.id,
        userId: session.userId,
        userName: session.userName,
+        // Add operation metadata for better client handling
        metadata: {
          workflowId,
          operationId: crypto.randomUUID(),
+          isPositionUpdate, // Flag to help clients handle position updates specially
        },
      }

      socket.to(workflowId).emit('workflow-operation', broadcastData)

+      // Emit confirmation if operationId is provided
      if (operationId) {
        socket.emit('operation-confirmed', {
          operationId,
@@ -538,14 +533,16 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
    } catch (error) {
      const errorMessage = error instanceof Error ? error.message : 'Unknown error occurred'

+      // Emit operation-failed for queue-tracked operations
      if (operationId) {
        socket.emit('operation-failed', {
          operationId,
          error: errorMessage,
-          retryable: !(error instanceof ZodError),
+          retryable: !(error instanceof ZodError), // Don't retry validation errors
        })
      }

+      // Also emit legacy operation-error for backward compatibility
      if (error instanceof ZodError) {
        socket.emit('operation-error', {
          type: 'VALIDATION_ERROR',
@@ -556,6 +553,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager
        })
        logger.warn(`Validation error for operation from ${session.userId}:`, error.errors)
      } else if (error instanceof Error) {
+        // Handle specific database errors
        if (error.message.includes('not found')) {
          socket.emit('operation-error', {
            type: 'RESOURCE_NOT_FOUND',
--- a/apps/sim/socket/handlers/presence.ts
+++ b/apps/sim/socket/handlers/presence.ts
@@ -1,53 +1,62 @@
 import { createLogger } from '@sim/logger'
+import type { HandlerDependencies } from '@/socket/handlers/workflow'
 import type { AuthenticatedSocket } from '@/socket/middleware/auth'
-import type { IRoomManager } from '@/socket/rooms'
+import type { RoomManager } from '@/socket/rooms/manager'

 const logger = createLogger('PresenceHandlers')

-export function setupPresenceHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
-  socket.on('cursor-update', async ({ cursor }) => {
-    try {
-      const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
-      const session = await roomManager.getUserSession(socket.id)
+export function setupPresenceHandlers(
+  socket: AuthenticatedSocket,
+  deps: HandlerDependencies | RoomManager
+) {
+  const roomManager =
+    deps instanceof Object && 'roomManager' in deps ? deps.roomManager : (deps as RoomManager)
+  socket.on('cursor-update', ({ cursor }) => {
+    const workflowId = roomManager.getWorkflowIdForSocket(socket.id)
+    const session = roomManager.getUserSession(socket.id)

-      if (!workflowId || !session) return
+    if (!workflowId || !session) return

-      // Update cursor in room state
-      await roomManager.updateUserActivity(workflowId, socket.id, { cursor })
+    const room = roomManager.getWorkflowRoom(workflowId)
+    if (!room) return

-      // Broadcast to other users in the room
-      socket.to(workflowId).emit('cursor-update', {
-        socketId: socket.id,
-        userId: session.userId,
-        userName: session.userName,
-        avatarUrl: session.avatarUrl,
-        cursor,
-      })
-    } catch (error) {
-      logger.error(`Error handling cursor update for socket ${socket.id}:`, error)
+    const userPresence = room.users.get(socket.id)
+    if (userPresence) {
+      userPresence.cursor = cursor
+      userPresence.lastActivity = Date.now()
    }
+
+    socket.to(workflowId).emit('cursor-update', {
+      socketId: socket.id,
+      userId: session.userId,
+      userName: session.userName,
+      avatarUrl: session.avatarUrl,
+      cursor,
+    })
  })

-  socket.on('selection-update', async ({ selection }) => {
-    try {
-      const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
-      const session = await roomManager.getUserSession(socket.id)
+  // Handle user selection (for showing what block/element a user has selected)
+  socket.on('selection-update', ({ selection }) => {
+    const workflowId = roomManager.getWorkflowIdForSocket(socket.id)
+    const session = roomManager.getUserSession(socket.id)

-      if (!workflowId || !session) return
+    if (!workflowId || !session) return

-      // Update selection in room state
-      await roomManager.updateUserActivity(workflowId, socket.id, { selection })
+    const room = roomManager.getWorkflowRoom(workflowId)
+    if (!room) return

-      // Broadcast to other users in the room
-      socket.to(workflowId).emit('selection-update', {
-        socketId: socket.id,
-        userId: session.userId,
-        userName: session.userName,
-        avatarUrl: session.avatarUrl,
-        selection,
-      })
-    } catch (error) {
-      logger.error(`Error handling selection update for socket ${socket.id}:`, error)
+    const userPresence = room.users.get(socket.id)
+    if (userPresence) {
+      userPresence.selection = selection
+      userPresence.lastActivity = Date.now()
    }
+
+    socket.to(workflowId).emit('selection-update', {
+      socketId: socket.id,
+      userId: session.userId,
+      userName: session.userName,
+      avatarUrl: session.avatarUrl,
+      selection,
+    })
  })
 }
--- a/apps/sim/socket/handlers/subblocks.ts
+++ b/apps/sim/socket/handlers/subblocks.ts
@@ -2,14 +2,12 @@ import { db } from '@sim/db'
 import { workflow, workflowBlocks } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { and, eq } from 'drizzle-orm'
+import type { HandlerDependencies } from '@/socket/handlers/workflow'
 import type { AuthenticatedSocket } from '@/socket/middleware/auth'
-import type { IRoomManager } from '@/socket/rooms'
+import type { RoomManager } from '@/socket/rooms/manager'

 const logger = createLogger('SubblocksHandlers')

-/** Debounce interval for coalescing rapid subblock updates before persisting */
-const DEBOUNCE_INTERVAL_MS = 25
-
 type PendingSubblock = {
  latest: { blockId: string; subblockId: string; value: any; timestamp: number }
  timeout: NodeJS.Timeout
@@ -20,61 +18,44 @@ type PendingSubblock = {
 // Keyed by `${workflowId}:${blockId}:${subblockId}`
 const pendingSubblockUpdates = new Map<string, PendingSubblock>()

-/**
- * Cleans up pending updates for a disconnected socket.
- * Removes the socket's operationIds from pending updates to prevent memory leaks.
- */
-export function cleanupPendingSubblocksForSocket(socketId: string): void {
-  for (const [, pending] of pendingSubblockUpdates.entries()) {
-    // Remove this socket's operation entries
-    for (const [opId, sid] of pending.opToSocket.entries()) {
-      if (sid === socketId) {
-        pending.opToSocket.delete(opId)
-      }
-    }
-    // If no more operations are waiting, the timeout will still fire and flush
-    // This is fine - the update will still persist, just no confirmation to send
-  }
-}
-
-export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
+export function setupSubblocksHandlers(
+  socket: AuthenticatedSocket,
+  deps: HandlerDependencies | RoomManager
+) {
+  const roomManager =
+    deps instanceof Object && 'roomManager' in deps ? deps.roomManager : (deps as RoomManager)
  socket.on('subblock-update', async (data) => {
+    const workflowId = roomManager.getWorkflowIdForSocket(socket.id)
+    const session = roomManager.getUserSession(socket.id)
+
+    if (!workflowId || !session) {
+      logger.debug(`Ignoring subblock update: socket not connected to any workflow room`, {
+        socketId: socket.id,
+        hasWorkflowId: !!workflowId,
+        hasSession: !!session,
+      })
+      return
+    }
+
    const { blockId, subblockId, value, timestamp, operationId } = data
+    const room = roomManager.getWorkflowRoom(workflowId)
+
+    if (!room) {
+      logger.debug(`Ignoring subblock update: workflow room not found`, {
+        socketId: socket.id,
+        workflowId,
+        blockId,
+        subblockId,
+      })
+      return
+    }

    try {
-      const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
-      const session = await roomManager.getUserSession(socket.id)
-
-      if (!workflowId || !session) {
-        logger.debug(`Ignoring subblock update: socket not connected to any workflow room`, {
-          socketId: socket.id,
-          hasWorkflowId: !!workflowId,
-          hasSession: !!session,
-        })
-        socket.emit('operation-forbidden', {
-          type: 'SESSION_ERROR',
-          message: 'Session expired, please rejoin workflow',
-        })
-        if (operationId) {
-          socket.emit('operation-failed', { operationId, error: 'Session expired' })
-        }
-        return
+      const userPresence = room.users.get(socket.id)
+      if (userPresence) {
+        userPresence.lastActivity = Date.now()
      }

-      const hasRoom = await roomManager.hasWorkflowRoom(workflowId)
-      if (!hasRoom) {
-        logger.debug(`Ignoring subblock update: workflow room not found`, {
-          socketId: socket.id,
-          workflowId,
-          blockId,
-          subblockId,
-        })
-        return
-      }
-
-      // Update user activity
-      await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() })
-
      // Server-side debounce/coalesce by workflowId+blockId+subblockId
      const debouncedKey = `${workflowId}:${blockId}:${subblockId}`
      const existing = pendingSubblockUpdates.get(debouncedKey)
@@ -85,7 +66,7 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager:
        existing.timeout = setTimeout(async () => {
          await flushSubblockUpdate(workflowId, existing, roomManager)
          pendingSubblockUpdates.delete(debouncedKey)
-        }, DEBOUNCE_INTERVAL_MS)
+        }, 25)
      } else {
        const opToSocket = new Map<string, string>()
        if (operationId) opToSocket.set(operationId, socket.id)
@@ -95,7 +76,7 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager:
            await flushSubblockUpdate(workflowId, pending, roomManager)
            pendingSubblockUpdates.delete(debouncedKey)
          }
-        }, DEBOUNCE_INTERVAL_MS)
+        }, 25)
        pendingSubblockUpdates.set(debouncedKey, {
          latest: { blockId, subblockId, value, timestamp },
          timeout,
@@ -107,6 +88,7 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager:

      const errorMessage = error instanceof Error ? error.message : 'Unknown error'

+      // Best-effort failure for the single operation if provided
      if (operationId) {
        socket.emit('operation-failed', {
          operationId,
@@ -115,6 +97,7 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager:
        })
      }

+      // Also emit legacy operation-error for backward compatibility
      socket.emit('operation-error', {
        type: 'SUBBLOCK_UPDATE_FAILED',
        message: `Failed to update subblock ${blockId}.${subblockId}: ${errorMessage}`,
@@ -128,11 +111,9 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager:
 async function flushSubblockUpdate(
  workflowId: string,
  pending: PendingSubblock,
-  roomManager: IRoomManager
+  roomManager: RoomManager
 ) {
  const { blockId, subblockId, value, timestamp } = pending.latest
-  const io = roomManager.io
-
  try {
    // Verify workflow still exists
    const workflowExists = await db
@@ -143,11 +124,14 @@ async function flushSubblockUpdate(

    if (workflowExists.length === 0) {
      pending.opToSocket.forEach((socketId, opId) => {
-        io.to(socketId).emit('operation-failed', {
-          operationId: opId,
-          error: 'Workflow not found',
-          retryable: false,
-        })
+        const sock = (roomManager as any).io?.sockets?.sockets?.get(socketId)
+        if (sock) {
+          sock.emit('operation-failed', {
+            operationId: opId,
+            error: 'Workflow not found',
+            retryable: false,
+          })
+        }
      })
      return
    }
@@ -180,48 +164,60 @@ async function flushSubblockUpdate(
    })

    if (updateSuccessful) {
-      // Broadcast to room excluding all senders (works cross-pod via Redis adapter)
-      const senderSocketIds = [...pending.opToSocket.values()]
-      if (senderSocketIds.length > 0) {
-        io.to(workflowId).except(senderSocketIds).emit('subblock-update', {
-          blockId,
-          subblockId,
-          value,
-          timestamp,
-        })
-      } else {
-        io.to(workflowId).emit('subblock-update', {
-          blockId,
-          subblockId,
-          value,
-          timestamp,
-        })
+      // Broadcast to other clients (exclude senders to avoid overwriting their local state)
+      const senderSocketIds = new Set(pending.opToSocket.values())
+      const io = (roomManager as any).io
+      if (io) {
+        // Get all sockets in the room
+        const roomSockets = io.sockets.adapter.rooms.get(workflowId)
+        if (roomSockets) {
+          roomSockets.forEach((socketId: string) => {
+            // Only emit to sockets that didn't send any of the coalesced ops
+            if (!senderSocketIds.has(socketId)) {
+              const sock = io.sockets.sockets.get(socketId)
+              if (sock) {
+                sock.emit('subblock-update', {
+                  blockId,
+                  subblockId,
+                  value,
+                  timestamp,
+                })
+              }
+            }
+          })
+        }
      }

-      // Confirm all coalesced operationIds (io.to(socketId) works cross-pod)
+      // Confirm all coalesced operationIds
      pending.opToSocket.forEach((socketId, opId) => {
-        io.to(socketId).emit('operation-confirmed', {
-          operationId: opId,
-          serverTimestamp: Date.now(),
-        })
+        const sock = (roomManager as any).io?.sockets?.sockets?.get(socketId)
+        if (sock) {
+          sock.emit('operation-confirmed', { operationId: opId, serverTimestamp: Date.now() })
+        }
      })
    } else {
      pending.opToSocket.forEach((socketId, opId) => {
-        io.to(socketId).emit('operation-failed', {
-          operationId: opId,
-          error: 'Block no longer exists',
-          retryable: false,
-        })
+        const sock = (roomManager as any).io?.sockets?.sockets?.get(socketId)
+        if (sock) {
+          sock.emit('operation-failed', {
+            operationId: opId,
+            error: 'Block no longer exists',
+            retryable: false,
+          })
+        }
      })
    }
  } catch (error) {
    logger.error('Error flushing subblock update:', error)
    pending.opToSocket.forEach((socketId, opId) => {
-      io.to(socketId).emit('operation-failed', {
-        operationId: opId,
-        error: error instanceof Error ? error.message : 'Unknown error',
-        retryable: true,
-      })
+      const sock = (roomManager as any).io?.sockets?.sockets?.get(socketId)
+      if (sock) {
+        sock.emit('operation-failed', {
+          operationId: opId,
+          error: error instanceof Error ? error.message : 'Unknown error',
+          retryable: true,
+        })
+      }
    })
  }
 }
--- a/apps/sim/socket/handlers/variables.ts
+++ b/apps/sim/socket/handlers/variables.ts
@@ -2,14 +2,12 @@ import { db } from '@sim/db'
 import { workflow } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { eq } from 'drizzle-orm'
+import type { HandlerDependencies } from '@/socket/handlers/workflow'
 import type { AuthenticatedSocket } from '@/socket/middleware/auth'
-import type { IRoomManager } from '@/socket/rooms'
+import type { RoomManager } from '@/socket/rooms/manager'

 const logger = createLogger('VariablesHandlers')

-/** Debounce interval for coalescing rapid variable updates before persisting */
-const DEBOUNCE_INTERVAL_MS = 25
-
 type PendingVariable = {
  latest: { variableId: string; field: string; value: any; timestamp: number }
  timeout: NodeJS.Timeout
@@ -19,58 +17,45 @@ type PendingVariable = {
 // Keyed by `${workflowId}:${variableId}:${field}`
 const pendingVariableUpdates = new Map<string, PendingVariable>()

-/**
- * Cleans up pending updates for a disconnected socket.
- * Removes the socket's operationIds from pending updates to prevent memory leaks.
- */
-export function cleanupPendingVariablesForSocket(socketId: string): void {
-  for (const [, pending] of pendingVariableUpdates.entries()) {
-    for (const [opId, sid] of pending.opToSocket.entries()) {
-      if (sid === socketId) {
-        pending.opToSocket.delete(opId)
-      }
-    }
-  }
-}
+export function setupVariablesHandlers(
+  socket: AuthenticatedSocket,
+  deps: HandlerDependencies | RoomManager
+) {
+  const roomManager =
+    deps instanceof Object && 'roomManager' in deps ? deps.roomManager : (deps as RoomManager)

-export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
  socket.on('variable-update', async (data) => {
+    const workflowId = roomManager.getWorkflowIdForSocket(socket.id)
+    const session = roomManager.getUserSession(socket.id)
+
+    if (!workflowId || !session) {
+      logger.debug(`Ignoring variable update: socket not connected to any workflow room`, {
+        socketId: socket.id,
+        hasWorkflowId: !!workflowId,
+        hasSession: !!session,
+      })
+      return
+    }
+
    const { variableId, field, value, timestamp, operationId } = data
+    const room = roomManager.getWorkflowRoom(workflowId)
+
+    if (!room) {
+      logger.debug(`Ignoring variable update: workflow room not found`, {
+        socketId: socket.id,
+        workflowId,
+        variableId,
+        field,
+      })
+      return
+    }

    try {
-      const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
-      const session = await roomManager.getUserSession(socket.id)
-
-      if (!workflowId || !session) {
-        logger.debug(`Ignoring variable update: socket not connected to any workflow room`, {
-          socketId: socket.id,
-          hasWorkflowId: !!workflowId,
-          hasSession: !!session,
-        })
-        socket.emit('operation-forbidden', {
-          type: 'SESSION_ERROR',
-          message: 'Session expired, please rejoin workflow',
-        })
-        if (operationId) {
-          socket.emit('operation-failed', { operationId, error: 'Session expired' })
-        }
-        return
+      const userPresence = room.users.get(socket.id)
+      if (userPresence) {
+        userPresence.lastActivity = Date.now()
      }

-      const hasRoom = await roomManager.hasWorkflowRoom(workflowId)
-      if (!hasRoom) {
-        logger.debug(`Ignoring variable update: workflow room not found`, {
-          socketId: socket.id,
-          workflowId,
-          variableId,
-          field,
-        })
-        return
-      }
-
-      // Update user activity
-      await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() })
-
      const debouncedKey = `${workflowId}:${variableId}:${field}`
      const existing = pendingVariableUpdates.get(debouncedKey)
      if (existing) {
@@ -80,7 +65,7 @@ export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager:
        existing.timeout = setTimeout(async () => {
          await flushVariableUpdate(workflowId, existing, roomManager)
          pendingVariableUpdates.delete(debouncedKey)
-        }, DEBOUNCE_INTERVAL_MS)
+        }, 25)
      } else {
        const opToSocket = new Map<string, string>()
        if (operationId) opToSocket.set(operationId, socket.id)
@@ -90,7 +75,7 @@ export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager:
            await flushVariableUpdate(workflowId, pending, roomManager)
            pendingVariableUpdates.delete(debouncedKey)
          }
-        }, DEBOUNCE_INTERVAL_MS)
+        }, 25)
        pendingVariableUpdates.set(debouncedKey, {
          latest: { variableId, field, value, timestamp },
          timeout,
@@ -123,11 +108,9 @@ export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager:
 async function flushVariableUpdate(
  workflowId: string,
  pending: PendingVariable,
-  roomManager: IRoomManager
+  roomManager: RoomManager
 ) {
  const { variableId, field, value, timestamp } = pending.latest
-  const io = roomManager.io
-
  try {
    const workflowExists = await db
      .select({ id: workflow.id })
@@ -137,11 +120,14 @@ async function flushVariableUpdate(

    if (workflowExists.length === 0) {
      pending.opToSocket.forEach((socketId, opId) => {
-        io.to(socketId).emit('operation-failed', {
-          operationId: opId,
-          error: 'Workflow not found',
-          retryable: false,
-        })
+        const sock = (roomManager as any).io?.sockets?.sockets?.get(socketId)
+        if (sock) {
+          sock.emit('operation-failed', {
+            operationId: opId,
+            error: 'Workflow not found',
+            retryable: false,
+          })
+        }
      })
      return
    }
@@ -177,50 +163,59 @@ async function flushVariableUpdate(
    })

    if (updateSuccessful) {
-      // Broadcast to room excluding all senders (works cross-pod via Redis adapter)
-      const senderSocketIds = [...pending.opToSocket.values()]
-      if (senderSocketIds.length > 0) {
-        io.to(workflowId).except(senderSocketIds).emit('variable-update', {
-          variableId,
-          field,
-          value,
-          timestamp,
-        })
-      } else {
-        io.to(workflowId).emit('variable-update', {
-          variableId,
-          field,
-          value,
-          timestamp,
-        })
+      // Broadcast to other clients (exclude senders to avoid overwriting their local state)
+      const senderSocketIds = new Set(pending.opToSocket.values())
+      const io = (roomManager as any).io
+      if (io) {
+        const roomSockets = io.sockets.adapter.rooms.get(workflowId)
+        if (roomSockets) {
+          roomSockets.forEach((socketId: string) => {
+            if (!senderSocketIds.has(socketId)) {
+              const sock = io.sockets.sockets.get(socketId)
+              if (sock) {
+                sock.emit('variable-update', {
+                  variableId,
+                  field,
+                  value,
+                  timestamp,
+                })
+              }
+            }
+          })
+        }
      }

-      // Confirm all coalesced operationIds (io.to(socketId) works cross-pod)
      pending.opToSocket.forEach((socketId, opId) => {
-        io.to(socketId).emit('operation-confirmed', {
-          operationId: opId,
-          serverTimestamp: Date.now(),
-        })
+        const sock = (roomManager as any).io?.sockets?.sockets?.get(socketId)
+        if (sock) {
+          sock.emit('operation-confirmed', { operationId: opId, serverTimestamp: Date.now() })
+        }
      })

      logger.debug(`Flushed variable update ${workflowId}: ${variableId}.${field}`)
    } else {
      pending.opToSocket.forEach((socketId, opId) => {
-        io.to(socketId).emit('operation-failed', {
-          operationId: opId,
-          error: 'Variable no longer exists',
-          retryable: false,
-        })
+        const sock = (roomManager as any).io?.sockets?.sockets?.get(socketId)
+        if (sock) {
+          sock.emit('operation-failed', {
+            operationId: opId,
+            error: 'Variable no longer exists',
+            retryable: false,
+          })
+        }
      })
    }
  } catch (error) {
    logger.error('Error flushing variable update:', error)
    pending.opToSocket.forEach((socketId, opId) => {
-      io.to(socketId).emit('operation-failed', {
-        operationId: opId,
-        error: error instanceof Error ? error.message : 'Unknown error',
-        retryable: true,
-      })
+      const sock = (roomManager as any).io?.sockets?.sockets?.get(socketId)
+      if (sock) {
+        sock.emit('operation-failed', {
+          operationId: opId,
+          error: error instanceof Error ? error.message : 'Unknown error',
+          retryable: true,
+        })
+      }
    })
  }
 }
--- a/apps/sim/socket/handlers/workflow.ts
+++ b/apps/sim/socket/handlers/workflow.ts
@@ -4,12 +4,38 @@ import { eq } from 'drizzle-orm'
 import { getWorkflowState } from '@/socket/database/operations'
 import type { AuthenticatedSocket } from '@/socket/middleware/auth'
 import { verifyWorkflowAccess } from '@/socket/middleware/permissions'
-import type { IRoomManager, UserPresence } from '@/socket/rooms'
+import type { RoomManager, UserPresence, WorkflowRoom } from '@/socket/rooms/manager'

 const logger = createLogger('WorkflowHandlers')

-export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
-  socket.on('join-workflow', async ({ workflowId, tabSessionId }) => {
+export type { UserPresence, WorkflowRoom }
+
+export interface HandlerDependencies {
+  roomManager: RoomManager
+}
+
+export const createWorkflowRoom = (workflowId: string): WorkflowRoom => ({
+  workflowId,
+  users: new Map(),
+  lastModified: Date.now(),
+  activeConnections: 0,
+})
+
+export const cleanupUserFromRoom = (
+  socketId: string,
+  workflowId: string,
+  roomManager: RoomManager
+) => {
+  roomManager.cleanupUserFromRoom(socketId, workflowId)
+}
+
+export function setupWorkflowHandlers(
+  socket: AuthenticatedSocket,
+  deps: HandlerDependencies | RoomManager
+) {
+  const roomManager =
+    deps instanceof Object && 'roomManager' in deps ? deps.roomManager : (deps as RoomManager)
+  socket.on('join-workflow', async ({ workflowId }) => {
    try {
      const userId = socket.userId
      const userName = socket.userName
@@ -22,7 +48,6 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager:

      logger.info(`Join workflow request from ${userId} (${userName}) for workflow ${workflowId}`)

-      // Verify workflow access
      let userRole: string
      try {
        const accessInfo = await verifyWorkflowAccess(userId, workflowId)
@@ -38,37 +63,23 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager:
        return
      }

-      // Leave current room if in one
-      const currentWorkflowId = await roomManager.getWorkflowIdForSocket(socket.id)
+      const currentWorkflowId = roomManager.getWorkflowIdForSocket(socket.id)
      if (currentWorkflowId) {
        socket.leave(currentWorkflowId)
-        await roomManager.removeUserFromRoom(socket.id)
-        await roomManager.broadcastPresenceUpdate(currentWorkflowId)
+        roomManager.cleanupUserFromRoom(socket.id, currentWorkflowId)
+
+        roomManager.broadcastPresenceUpdate(currentWorkflowId)
      }

-      const STALE_THRESHOLD_MS = 60_000
-      const now = Date.now()
-      const existingUsers = await roomManager.getWorkflowUsers(workflowId)
-      for (const existingUser of existingUsers) {
-        if (existingUser.userId === userId && existingUser.socketId !== socket.id) {
-          const isSameTab = tabSessionId && existingUser.tabSessionId === tabSessionId
-          const isStale =
-            now - (existingUser.lastActivity || existingUser.joinedAt || 0) > STALE_THRESHOLD_MS
-
-          if (isSameTab || isStale) {
-            logger.info(
-              `Cleaning up socket ${existingUser.socketId} for user ${userId} (${isSameTab ? 'same tab' : 'stale'})`
-            )
-            await roomManager.removeUserFromRoom(existingUser.socketId)
-            roomManager.io.in(existingUser.socketId).socketsLeave(workflowId)
-          }
-        }
-      }
-
-      // Join the new room
      socket.join(workflowId)

-      // Get avatar URL
+      if (!roomManager.hasWorkflowRoom(workflowId)) {
+        roomManager.setWorkflowRoom(workflowId, roomManager.createWorkflowRoom(workflowId))
+      }
+
+      const room = roomManager.getWorkflowRoom(workflowId)!
+      room.activeConnections++
+
      let avatarUrl = socket.userImage || null
      if (!avatarUrl) {
        try {
@@ -84,68 +95,54 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager:
        }
      }

-      // Create presence entry
      const userPresence: UserPresence = {
        userId,
        workflowId,
        userName,
        socketId: socket.id,
-        tabSessionId,
        joinedAt: Date.now(),
        lastActivity: Date.now(),
        role: userRole,
        avatarUrl,
      }

-      // Add user to room
-      await roomManager.addUserToRoom(workflowId, socket.id, userPresence)
-
-      // Get current presence list for the join acknowledgment
-      const presenceUsers = await roomManager.getWorkflowUsers(workflowId)
-
-      // Get workflow state
-      const workflowState = await getWorkflowState(workflowId)
-
-      // Send join success with presence list (client waits for this to confirm join)
-      socket.emit('join-workflow-success', {
-        workflowId,
-        socketId: socket.id,
-        presenceUsers,
+      room.users.set(socket.id, userPresence)
+      roomManager.setWorkflowForSocket(socket.id, workflowId)
+      roomManager.setUserSession(socket.id, {
+        userId,
+        userName,
+        avatarUrl,
      })

-      // Send workflow state
+      const workflowState = await getWorkflowState(workflowId)
      socket.emit('workflow-state', workflowState)

-      // Broadcast presence update to all users in the room
-      await roomManager.broadcastPresenceUpdate(workflowId)
+      roomManager.broadcastPresenceUpdate(workflowId)

-      const uniqueUserCount = await roomManager.getUniqueUserCount(workflowId)
+      const uniqueUserCount = roomManager.getUniqueUserCount(workflowId)
      logger.info(
-        `User ${userId} (${userName}) joined workflow ${workflowId}. Room now has ${uniqueUserCount} unique users.`
+        `User ${userId} (${userName}) joined workflow ${workflowId}. Room now has ${uniqueUserCount} unique users (${room.activeConnections} connections).`
      )
    } catch (error) {
      logger.error('Error joining workflow:', error)
-      // Undo socket.join and room manager entry if any operation failed
-      socket.leave(workflowId)
-      await roomManager.removeUserFromRoom(socket.id)
-      socket.emit('join-workflow-error', { error: 'Failed to join workflow' })
+      socket.emit('error', {
+        type: 'JOIN_ERROR',
+        message: 'Failed to join workflow',
+      })
    }
  })

-  socket.on('leave-workflow', async () => {
-    try {
-      const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
-      const session = await roomManager.getUserSession(socket.id)
+  socket.on('leave-workflow', () => {
+    const workflowId = roomManager.getWorkflowIdForSocket(socket.id)
+    const session = roomManager.getUserSession(socket.id)

-      if (workflowId && session) {
-        socket.leave(workflowId)
-        await roomManager.removeUserFromRoom(socket.id)
-        await roomManager.broadcastPresenceUpdate(workflowId)
+    if (workflowId && session) {
+      socket.leave(workflowId)
+      roomManager.cleanupUserFromRoom(socket.id, workflowId)

-        logger.info(`User ${session.userId} (${session.userName}) left workflow ${workflowId}`)
-      }
-    } catch (error) {
-      logger.error('Error leaving workflow:', error)
+      roomManager.broadcastPresenceUpdate(workflowId)
+
+      logger.info(`User ${session.userId} (${session.userName}) left workflow ${workflowId}`)
    }
  })
 }
--- a/apps/sim/socket/index.test.ts
+++ b/apps/sim/socket/index.test.ts
@@ -7,7 +7,7 @@ import { createServer, request as httpRequest } from 'http'
 import { createMockLogger, databaseMock } from '@sim/testing'
 import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 import { createSocketIOServer } from '@/socket/config/socket'
-import { MemoryRoomManager } from '@/socket/rooms'
+import { RoomManager } from '@/socket/rooms/manager'
 import { createHttpHandler } from '@/socket/routes/http'

 vi.mock('@/lib/auth', () => ({
@@ -20,30 +20,6 @@ vi.mock('@/lib/auth', () => ({

 vi.mock('@sim/db', () => databaseMock)

-// Mock redis package to prevent actual Redis connections
-vi.mock('redis', () => ({
-  createClient: vi.fn(() => ({
-    on: vi.fn(),
-    connect: vi.fn().mockResolvedValue(undefined),
-    quit: vi.fn().mockResolvedValue(undefined),
-    duplicate: vi.fn().mockReturnThis(),
-  })),
-}))
-
-// Mock env to not have REDIS_URL (use importOriginal to get helper functions)
-vi.mock('@/lib/core/config/env', async (importOriginal) => {
-  const actual = await importOriginal<typeof import('@/lib/core/config/env')>()
-  return {
-    ...actual,
-    env: {
-      ...actual.env,
-      DATABASE_URL: 'postgres://localhost/test',
-      NODE_ENV: 'test',
-      REDIS_URL: undefined,
-    },
-  }
-})
-
 vi.mock('@/socket/middleware/auth', () => ({
  authenticateSocket: vi.fn((socket, next) => {
    socket.userId = 'test-user-id'
@@ -75,7 +51,7 @@ vi.mock('@/socket/database/operations', () => ({
 describe('Socket Server Index Integration', () => {
  let httpServer: any
  let io: any
-  let roomManager: MemoryRoomManager
+  let roomManager: RoomManager
  let logger: ReturnType<typeof createMockLogger>
  let PORT: number

@@ -88,10 +64,9 @@ describe('Socket Server Index Integration', () => {

    httpServer = createServer()

-    io = await createSocketIOServer(httpServer)
+    io = createSocketIOServer(httpServer)

-    roomManager = new MemoryRoomManager(io)
-    await roomManager.initialize()
+    roomManager = new RoomManager(io)

    const httpHandler = createHttpHandler(roomManager, logger)
    httpServer.on('request', httpHandler)
@@ -123,9 +98,6 @@ describe('Socket Server Index Integration', () => {
  }, 20000)

  afterEach(async () => {
-    if (roomManager) {
-      await roomManager.shutdown()
-    }
    if (io) {
      await new Promise<void>((resolve) => {
        io.close(() => resolve())
@@ -205,60 +177,43 @@ describe('Socket Server Index Integration', () => {
  })

  describe('Room Manager Integration', () => {
-    it('should create room manager successfully', async () => {
+    it('should create room manager successfully', () => {
      expect(roomManager).toBeDefined()
-      expect(await roomManager.getTotalActiveConnections()).toBe(0)
+      expect(roomManager.getTotalActiveConnections()).toBe(0)
    })

-    it('should add and get users from workflow rooms', async () => {
+    it('should create workflow rooms', () => {
      const workflowId = 'test-workflow-123'
-      const socketId = 'test-socket-123'
+      const room = roomManager.createWorkflowRoom(workflowId)
+      roomManager.setWorkflowRoom(workflowId, room)

-      const presence = {
-        userId: 'user-123',
-        workflowId,
-        userName: 'Test User',
-        socketId,
-        joinedAt: Date.now(),
-        lastActivity: Date.now(),
-        role: 'admin',
-      }
-
-      await roomManager.addUserToRoom(workflowId, socketId, presence)
-
-      expect(await roomManager.hasWorkflowRoom(workflowId)).toBe(true)
-      const users = await roomManager.getWorkflowUsers(workflowId)
-      expect(users).toHaveLength(1)
-      expect(users[0].socketId).toBe(socketId)
+      expect(roomManager.hasWorkflowRoom(workflowId)).toBe(true)
+      const retrievedRoom = roomManager.getWorkflowRoom(workflowId)
+      expect(retrievedRoom).toBeDefined()
+      expect(retrievedRoom?.workflowId).toBe(workflowId)
    })

-    it('should manage user sessions', async () => {
+    it('should manage user sessions', () => {
      const socketId = 'test-socket-123'
      const workflowId = 'test-workflow-456'
+      const session = { userId: 'user-123', userName: 'Test User' }

-      const presence = {
-        userId: 'user-123',
-        workflowId,
-        userName: 'Test User',
-        socketId,
-        joinedAt: Date.now(),
-        lastActivity: Date.now(),
-        role: 'admin',
-      }
+      roomManager.setWorkflowForSocket(socketId, workflowId)
+      roomManager.setUserSession(socketId, session)

-      await roomManager.addUserToRoom(workflowId, socketId, presence)
-
-      expect(await roomManager.getWorkflowIdForSocket(socketId)).toBe(workflowId)
-      const session = await roomManager.getUserSession(socketId)
-      expect(session).toBeDefined()
-      expect(session?.userId).toBe('user-123')
+      expect(roomManager.getWorkflowIdForSocket(socketId)).toBe(workflowId)
+      expect(roomManager.getUserSession(socketId)).toEqual(session)
    })

-    it('should clean up rooms properly', async () => {
+    it('should clean up rooms properly', () => {
      const workflowId = 'test-workflow-789'
      const socketId = 'test-socket-789'

-      const presence = {
+      const room = roomManager.createWorkflowRoom(workflowId)
+      roomManager.setWorkflowRoom(workflowId, room)
+
+      // Add user to room
+      room.users.set(socketId, {
        userId: 'user-789',
        workflowId,
        userName: 'Test User',
@@ -266,18 +221,16 @@ describe('Socket Server Index Integration', () => {
        joinedAt: Date.now(),
        lastActivity: Date.now(),
        role: 'admin',
-      }
+      })
+      room.activeConnections = 1

-      await roomManager.addUserToRoom(workflowId, socketId, presence)
+      roomManager.setWorkflowForSocket(socketId, workflowId)

-      expect(await roomManager.hasWorkflowRoom(workflowId)).toBe(true)
+      // Clean up user
+      roomManager.cleanupUserFromRoom(socketId, workflowId)

-      // Remove user
-      await roomManager.removeUserFromRoom(socketId)
-
-      // Room should be cleaned up since it's now empty
-      expect(await roomManager.hasWorkflowRoom(workflowId)).toBe(false)
-      expect(await roomManager.getWorkflowIdForSocket(socketId)).toBeNull()
+      expect(roomManager.hasWorkflowRoom(workflowId)).toBe(false)
+      expect(roomManager.getWorkflowIdForSocket(socketId)).toBeUndefined()
    })
  })

@@ -285,7 +238,7 @@ describe('Socket Server Index Integration', () => {
    it.concurrent('should properly import all extracted modules', async () => {
      const { createSocketIOServer } = await import('@/socket/config/socket')
      const { createHttpHandler } = await import('@/socket/routes/http')
-      const { MemoryRoomManager, RedisRoomManager } = await import('@/socket/rooms')
+      const { RoomManager } = await import('@/socket/rooms/manager')
      const { authenticateSocket } = await import('@/socket/middleware/auth')
      const { verifyWorkflowAccess } = await import('@/socket/middleware/permissions')
      const { getWorkflowState } = await import('@/socket/database/operations')
@@ -293,23 +246,22 @@ describe('Socket Server Index Integration', () => {

      expect(createSocketIOServer).toBeTypeOf('function')
      expect(createHttpHandler).toBeTypeOf('function')
-      expect(MemoryRoomManager).toBeTypeOf('function')
-      expect(RedisRoomManager).toBeTypeOf('function')
+      expect(RoomManager).toBeTypeOf('function')
      expect(authenticateSocket).toBeTypeOf('function')
      expect(verifyWorkflowAccess).toBeTypeOf('function')
      expect(getWorkflowState).toBeTypeOf('function')
      expect(WorkflowOperationSchema).toBeDefined()
    })

-    it.concurrent('should maintain all original functionality after refactoring', async () => {
+    it.concurrent('should maintain all original functionality after refactoring', () => {
      expect(httpServer).toBeDefined()
      expect(io).toBeDefined()
      expect(roomManager).toBeDefined()

-      expect(typeof roomManager.addUserToRoom).toBe('function')
-      expect(typeof roomManager.removeUserFromRoom).toBe('function')
+      expect(typeof roomManager.createWorkflowRoom).toBe('function')
+      expect(typeof roomManager.cleanupUserFromRoom).toBe('function')
      expect(typeof roomManager.handleWorkflowDeletion).toBe('function')
-      expect(typeof roomManager.broadcastPresenceUpdate).toBe('function')
+      expect(typeof roomManager.validateWorkflowConsistency).toBe('function')
    })
  })

@@ -334,7 +286,6 @@ describe('Socket Server Index Integration', () => {
    it('should have shutdown capability', () => {
      expect(typeof httpServer.close).toBe('function')
      expect(typeof io.close).toBe('function')
-      expect(typeof roomManager.shutdown).toBe('function')
    })
  })

--- a/apps/sim/socket/index.ts
+++ b/apps/sim/socket/index.ts
@@ -1,125 +1,112 @@
 import { createServer } from 'http'
 import { createLogger } from '@sim/logger'
-import type { Server as SocketIOServer } from 'socket.io'
 import { env } from '@/lib/core/config/env'
-import { createSocketIOServer, shutdownSocketIOAdapter } from '@/socket/config/socket'
+import { createSocketIOServer } from '@/socket/config/socket'
 import { setupAllHandlers } from '@/socket/handlers'
 import { type AuthenticatedSocket, authenticateSocket } from '@/socket/middleware/auth'
-import { type IRoomManager, MemoryRoomManager, RedisRoomManager } from '@/socket/rooms'
+import { RoomManager } from '@/socket/rooms/manager'
 import { createHttpHandler } from '@/socket/routes/http'

 const logger = createLogger('CollaborativeSocketServer')

-/** Maximum time to wait for graceful shutdown before forcing exit */
-const SHUTDOWN_TIMEOUT_MS = 10000
+// Enhanced server configuration - HTTP server will be configured with handler after all dependencies are set up
+const httpServer = createServer()

-async function createRoomManager(io: SocketIOServer): Promise<IRoomManager> {
-  if (env.REDIS_URL) {
-    logger.info('Initializing Redis-backed RoomManager for multi-pod support')
-    const manager = new RedisRoomManager(io, env.REDIS_URL)
-    await manager.initialize()
-    return manager
-  }
+const io = createSocketIOServer(httpServer)

-  logger.warn('No REDIS_URL configured - using in-memory RoomManager (single-pod only)')
-  const manager = new MemoryRoomManager(io)
-  await manager.initialize()
-  return manager
-}
+// Initialize room manager after io is created
+const roomManager = new RoomManager(io)

-async function main() {
-  const httpServer = createServer()
-  const PORT = Number(env.PORT || env.SOCKET_PORT || 3002)
+io.use(authenticateSocket)

-  logger.info('Starting Socket.IO server...', {
-    port: PORT,
-    nodeEnv: env.NODE_ENV,
-    hasDatabase: !!env.DATABASE_URL,
-    hasAuth: !!env.BETTER_AUTH_SECRET,
-    hasRedis: !!env.REDIS_URL,
+const httpHandler = createHttpHandler(roomManager, logger)
+httpServer.on('request', httpHandler)
+
+process.on('uncaughtException', (error) => {
+  logger.error('Uncaught Exception:', error)
+  // Don't exit in production, just log
+})
+
+process.on('unhandledRejection', (reason, promise) => {
+  logger.error('Unhandled Rejection at:', promise, 'reason:', reason)
+})
+
+httpServer.on('error', (error) => {
+  logger.error('HTTP server error:', error)
+})
+
+io.engine.on('connection_error', (err) => {
+  logger.error('Socket.IO connection error:', {
+    req: err.req?.url,
+    code: err.code,
+    message: err.message,
+    context: err.context,
  })
+})

-  // Create Socket.IO server with Redis adapter if configured
-  const io = await createSocketIOServer(httpServer)
+io.on('connection', (socket: AuthenticatedSocket) => {
+  logger.info(`New socket connection: ${socket.id}`)

-  // Initialize room manager (Redis or in-memory based on config)
-  const roomManager = await createRoomManager(io)
+  setupAllHandlers(socket, roomManager)
+})

-  // Set up authentication middleware
-  io.use(authenticateSocket)
-
-  // Set up HTTP handler for health checks and internal APIs
-  const httpHandler = createHttpHandler(roomManager, logger)
-  httpServer.on('request', httpHandler)
-
-  // Global error handlers
-  process.on('uncaughtException', (error) => {
-    logger.error('Uncaught Exception:', error)
+httpServer.on('request', (req, res) => {
+  logger.info(`🌐 HTTP Request: ${req.method} ${req.url}`, {
+    method: req.method,
+    url: req.url,
+    userAgent: req.headers['user-agent'],
+    origin: req.headers.origin,
+    host: req.headers.host,
+    timestamp: new Date().toISOString(),
  })
+})

-  process.on('unhandledRejection', (reason, promise) => {
-    logger.error('Unhandled Rejection at:', promise, 'reason:', reason)
+io.engine.on('connection_error', (err) => {
+  logger.error('❌ Engine.IO Connection error:', {
+    code: err.code,
+    message: err.message,
+    context: err.context,
+    req: err.req
+      ? {
+          url: err.req.url,
+          method: err.req.method,
+          headers: err.req.headers,
+        }
+      : 'No request object',
  })
+})

-  httpServer.on('error', (error: NodeJS.ErrnoException) => {
-    logger.error('HTTP server error:', error)
-    if (error.code === 'EADDRINUSE' || error.code === 'EACCES') {
-      process.exit(1)
-    }
-  })
+const PORT = Number(env.PORT || env.SOCKET_PORT || 3002)

-  io.engine.on('connection_error', (err) => {
-    logger.error('Socket.IO connection error:', {
-      req: err.req?.url,
-      code: err.code,
-      message: err.message,
-      context: err.context,
-    })
-  })
+logger.info('Starting Socket.IO server...', {
+  port: PORT,
+  nodeEnv: env.NODE_ENV,
+  hasDatabase: !!env.DATABASE_URL,
+  hasAuth: !!env.BETTER_AUTH_SECRET,
+})

-  io.on('connection', (socket: AuthenticatedSocket) => {
-    logger.info(`New socket connection: ${socket.id}`)
-    setupAllHandlers(socket, roomManager)
-  })
+httpServer.listen(PORT, '0.0.0.0', () => {
+  logger.info(`Socket.IO server running on port ${PORT}`)
+  logger.info(`🏥 Health check available at: http://localhost:${PORT}/health`)
+})

-  httpServer.listen(PORT, '0.0.0.0', () => {
-    logger.info(`Socket.IO server running on port ${PORT}`)
-    logger.info(`Health check available at: http://localhost:${PORT}/health`)
-  })
-
-  const shutdown = async () => {
-    logger.info('Shutting down Socket.IO server...')
-
-    try {
-      await roomManager.shutdown()
-      logger.info('RoomManager shutdown complete')
-    } catch (error) {
-      logger.error('Error during RoomManager shutdown:', error)
-    }
-
-    try {
-      await shutdownSocketIOAdapter()
-    } catch (error) {
-      logger.error('Error during Socket.IO adapter shutdown:', error)
-    }
-
-    httpServer.close(() => {
-      logger.info('Socket.IO server closed')
-      process.exit(0)
-    })
-
-    setTimeout(() => {
-      logger.error('Forced shutdown after timeout')
-      process.exit(1)
-    }, SHUTDOWN_TIMEOUT_MS)
-  }
-
-  process.on('SIGINT', shutdown)
-  process.on('SIGTERM', shutdown)
-}
-
-// Start the server
-main().catch((error) => {
-  logger.error('Failed to start server:', error)
+httpServer.on('error', (error) => {
+  logger.error('❌ Server failed to start:', error)
  process.exit(1)
 })
+
+process.on('SIGINT', () => {
+  logger.info('Shutting down Socket.IO server...')
+  httpServer.close(() => {
+    logger.info('Socket.IO server closed')
+    process.exit(0)
+  })
+})
+
+process.on('SIGTERM', () => {
+  logger.info('Shutting down Socket.IO server...')
+  httpServer.close(() => {
+    logger.info('Socket.IO server closed')
+    process.exit(0)
+  })
+})
--- a/apps/sim/socket/middleware/auth.ts
+++ b/apps/sim/socket/middleware/auth.ts
@@ -21,7 +21,7 @@ export interface AuthenticatedSocket extends Socket {
 * Socket.IO authentication middleware.
 * Handles both anonymous mode (DISABLE_AUTH=true) and normal token-based auth.
 */
-export async function authenticateSocket(socket: AuthenticatedSocket, next: (err?: Error) => void) {
+export async function authenticateSocket(socket: AuthenticatedSocket, next: any) {
  try {
    if (isAuthDisabled) {
      socket.userId = ANONYMOUS_USER_ID
--- a/apps/sim/socket/middleware/permissions.ts
+++ b/apps/sim/socket/middleware/permissions.ts
@@ -73,7 +73,7 @@ export function checkRolePermission(
  return { allowed: true }
 }

-async function verifyWorkspaceMembership(
+export async function verifyWorkspaceMembership(
  userId: string,
  workspaceId: string
 ): Promise<string | null> {
--- a/apps/sim/socket/rooms/index.ts
+++ b/apps/sim/socket/rooms/index.ts
@@ -1,3 +0,0 @@
-export { MemoryRoomManager } from '@/socket/rooms/memory-manager'
-export { RedisRoomManager } from '@/socket/rooms/redis-manager'
-export type { IRoomManager, UserPresence, UserSession, WorkflowRoom } from '@/socket/rooms/types'
--- a/apps/sim/socket/rooms/manager.ts
+++ b/apps/sim/socket/rooms/manager.ts
@@ -0,0 +1,291 @@
+import * as schema from '@sim/db/schema'
+import { workflowBlocks, workflowEdges } from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { and, eq, isNull } from 'drizzle-orm'
+import { drizzle } from 'drizzle-orm/postgres-js'
+import postgres from 'postgres'
+import type { Server } from 'socket.io'
+import { env } from '@/lib/core/config/env'
+
+const connectionString = env.DATABASE_URL
+const db = drizzle(
+  postgres(connectionString, {
+    prepare: false,
+    idle_timeout: 15,
+    connect_timeout: 20,
+    max: 3,
+    onnotice: () => {},
+  }),
+  { schema }
+)
+
+const logger = createLogger('RoomManager')
+
+export interface UserPresence {
+  userId: string
+  workflowId: string
+  userName: string
+  socketId: string
+  joinedAt: number
+  lastActivity: number
+  role: string
+  cursor?: { x: number; y: number }
+  selection?: { type: 'block' | 'edge' | 'none'; id?: string }
+  avatarUrl?: string | null
+}
+
+export interface WorkflowRoom {
+  workflowId: string
+  users: Map<string, UserPresence> // socketId -> UserPresence
+  lastModified: number
+  activeConnections: number
+}
+
+export class RoomManager {
+  private workflowRooms = new Map<string, WorkflowRoom>()
+  private socketToWorkflow = new Map<string, string>()
+  private userSessions = new Map<
+    string,
+    { userId: string; userName: string; avatarUrl?: string | null }
+  >()
+  private io: Server
+
+  constructor(io: Server) {
+    this.io = io
+  }
+
+  createWorkflowRoom(workflowId: string): WorkflowRoom {
+    return {
+      workflowId,
+      users: new Map(),
+      lastModified: Date.now(),
+      activeConnections: 0,
+    }
+  }
+
+  cleanupUserFromRoom(socketId: string, workflowId: string) {
+    const room = this.workflowRooms.get(workflowId)
+    if (room) {
+      room.users.delete(socketId)
+      room.activeConnections = Math.max(0, room.activeConnections - 1)
+
+      if (room.activeConnections === 0) {
+        this.workflowRooms.delete(workflowId)
+        logger.info(`Cleaned up empty workflow room: ${workflowId}`)
+      }
+    }
+
+    this.socketToWorkflow.delete(socketId)
+    this.userSessions.delete(socketId)
+  }
+
+  handleWorkflowDeletion(workflowId: string) {
+    logger.info(`Handling workflow deletion notification for ${workflowId}`)
+
+    const room = this.workflowRooms.get(workflowId)
+    if (!room) {
+      logger.debug(`No active room found for deleted workflow ${workflowId}`)
+      return
+    }
+
+    this.io.to(workflowId).emit('workflow-deleted', {
+      workflowId,
+      message: 'This workflow has been deleted',
+      timestamp: Date.now(),
+    })
+
+    const socketsToDisconnect: string[] = []
+    room.users.forEach((_presence, socketId) => {
+      socketsToDisconnect.push(socketId)
+    })
+
+    socketsToDisconnect.forEach((socketId) => {
+      const socket = this.io.sockets.sockets.get(socketId)
+      if (socket) {
+        socket.leave(workflowId)
+        logger.debug(`Disconnected socket ${socketId} from deleted workflow ${workflowId}`)
+      }
+      this.cleanupUserFromRoom(socketId, workflowId)
+    })
+
+    this.workflowRooms.delete(workflowId)
+    logger.info(
+      `Cleaned up workflow room ${workflowId} after deletion (${socketsToDisconnect.length} users disconnected)`
+    )
+  }
+
+  handleWorkflowRevert(workflowId: string, timestamp: number) {
+    logger.info(`Handling workflow revert notification for ${workflowId}`)
+
+    const room = this.workflowRooms.get(workflowId)
+    if (!room) {
+      logger.debug(`No active room found for reverted workflow ${workflowId}`)
+      return
+    }
+
+    this.io.to(workflowId).emit('workflow-reverted', {
+      workflowId,
+      message: 'Workflow has been reverted to deployed state',
+      timestamp,
+    })
+
+    room.lastModified = timestamp
+
+    logger.info(`Notified ${room.users.size} users about workflow revert: ${workflowId}`)
+  }
+
+  handleWorkflowUpdate(workflowId: string) {
+    logger.info(`Handling workflow update notification for ${workflowId}`)
+
+    const room = this.workflowRooms.get(workflowId)
+    if (!room) {
+      logger.debug(`No active room found for updated workflow ${workflowId}`)
+      return
+    }
+
+    const timestamp = Date.now()
+
+    // Notify all clients in the workflow room that the workflow has been updated
+    // This will trigger them to refresh their local state
+    this.io.to(workflowId).emit('workflow-updated', {
+      workflowId,
+      message: 'Workflow has been updated externally',
+      timestamp,
+    })
+
+    room.lastModified = timestamp
+
+    logger.info(`Notified ${room.users.size} users about workflow update: ${workflowId}`)
+  }
+
+  handleCopilotWorkflowEdit(workflowId: string, description?: string) {
+    logger.info(`Handling copilot workflow edit notification for ${workflowId}`)
+
+    const room = this.workflowRooms.get(workflowId)
+    if (!room) {
+      logger.debug(`No active room found for copilot workflow edit ${workflowId}`)
+      return
+    }
+
+    const timestamp = Date.now()
+
+    // Emit special event for copilot edits that tells clients to rehydrate from database
+    this.io.to(workflowId).emit('copilot-workflow-edit', {
+      workflowId,
+      description,
+      message: 'Copilot has edited the workflow - rehydrating from database',
+      timestamp,
+    })
+
+    room.lastModified = timestamp
+
+    logger.info(`Notified ${room.users.size} users about copilot workflow edit: ${workflowId}`)
+  }
+
+  async validateWorkflowConsistency(
+    workflowId: string
+  ): Promise<{ valid: boolean; issues: string[] }> {
+    try {
+      const issues: string[] = []
+
+      const orphanedEdges = await db
+        .select({
+          id: workflowEdges.id,
+          sourceBlockId: workflowEdges.sourceBlockId,
+          targetBlockId: workflowEdges.targetBlockId,
+        })
+        .from(workflowEdges)
+        .leftJoin(workflowBlocks, eq(workflowEdges.sourceBlockId, workflowBlocks.id))
+        .where(and(eq(workflowEdges.workflowId, workflowId), isNull(workflowBlocks.id)))
+
+      if (orphanedEdges.length > 0) {
+        issues.push(`Found ${orphanedEdges.length} orphaned edges with missing source blocks`)
+      }
+
+      return { valid: issues.length === 0, issues }
+    } catch (error) {
+      logger.error('Error validating workflow consistency:', error)
+      return { valid: false, issues: ['Consistency check failed'] }
+    }
+  }
+
+  getWorkflowRooms(): ReadonlyMap<string, WorkflowRoom> {
+    return this.workflowRooms
+  }
+
+  getSocketToWorkflow(): ReadonlyMap<string, string> {
+    return this.socketToWorkflow
+  }
+
+  getUserSessions(): ReadonlyMap<string, { userId: string; userName: string }> {
+    return this.userSessions
+  }
+
+  hasWorkflowRoom(workflowId: string): boolean {
+    return this.workflowRooms.has(workflowId)
+  }
+
+  getWorkflowRoom(workflowId: string): WorkflowRoom | undefined {
+    return this.workflowRooms.get(workflowId)
+  }
+
+  setWorkflowRoom(workflowId: string, room: WorkflowRoom): void {
+    this.workflowRooms.set(workflowId, room)
+  }
+
+  getWorkflowIdForSocket(socketId: string): string | undefined {
+    return this.socketToWorkflow.get(socketId)
+  }
+
+  setWorkflowForSocket(socketId: string, workflowId: string): void {
+    this.socketToWorkflow.set(socketId, workflowId)
+  }
+
+  getUserSession(
+    socketId: string
+  ): { userId: string; userName: string; avatarUrl?: string | null } | undefined {
+    return this.userSessions.get(socketId)
+  }
+
+  setUserSession(
+    socketId: string,
+    session: { userId: string; userName: string; avatarUrl?: string | null }
+  ): void {
+    this.userSessions.set(socketId, session)
+  }
+
+  getTotalActiveConnections(): number {
+    return Array.from(this.workflowRooms.values()).reduce(
+      (total, room) => total + room.activeConnections,
+      0
+    )
+  }
+
+  broadcastPresenceUpdate(workflowId: string): void {
+    const room = this.workflowRooms.get(workflowId)
+    if (room) {
+      const roomPresence = Array.from(room.users.values())
+      this.io.to(workflowId).emit('presence-update', roomPresence)
+    }
+  }
+
+  emitToWorkflow<T = unknown>(workflowId: string, event: string, payload: T): void {
+    this.io.to(workflowId).emit(event, payload)
+  }
+
+  /**
+   * Get the number of unique users in a workflow room
+   * (not the number of socket connections)
+   */
+  getUniqueUserCount(workflowId: string): number {
+    const room = this.workflowRooms.get(workflowId)
+    if (!room) return 0
+
+    const uniqueUsers = new Set<string>()
+    room.users.forEach((presence) => {
+      uniqueUsers.add(presence.userId)
+    })
+
+    return uniqueUsers.size
+  }
+}
--- a/apps/sim/socket/rooms/memory-manager.ts
+++ b/apps/sim/socket/rooms/memory-manager.ts
@@ -1,260 +0,0 @@
-import { createLogger } from '@sim/logger'
-import type { Server } from 'socket.io'
-import type { IRoomManager, UserPresence, UserSession, WorkflowRoom } from '@/socket/rooms/types'
-
-const logger = createLogger('MemoryRoomManager')
-
-/**
- * In-memory room manager for single-pod deployments
- * Used as fallback when REDIS_URL is not configured
- */
-export class MemoryRoomManager implements IRoomManager {
-  private workflowRooms = new Map<string, WorkflowRoom>()
-  private socketToWorkflow = new Map<string, string>()
-  private userSessions = new Map<string, UserSession>()
-  private _io: Server
-
-  constructor(io: Server) {
-    this._io = io
-  }
-
-  get io(): Server {
-    return this._io
-  }
-
-  async initialize(): Promise<void> {
-    logger.info('MemoryRoomManager initialized (single-pod mode)')
-  }
-
-  async shutdown(): Promise<void> {
-    this.workflowRooms.clear()
-    this.socketToWorkflow.clear()
-    this.userSessions.clear()
-    logger.info('MemoryRoomManager shutdown complete')
-  }
-
-  async addUserToRoom(workflowId: string, socketId: string, presence: UserPresence): Promise<void> {
-    // Create room if it doesn't exist
-    if (!this.workflowRooms.has(workflowId)) {
-      this.workflowRooms.set(workflowId, {
-        workflowId,
-        users: new Map(),
-        lastModified: Date.now(),
-        activeConnections: 0,
-      })
-    }
-
-    const room = this.workflowRooms.get(workflowId)!
-    room.users.set(socketId, presence)
-    room.activeConnections++
-    room.lastModified = Date.now()
-
-    // Map socket to workflow
-    this.socketToWorkflow.set(socketId, workflowId)
-
-    // Store session
-    this.userSessions.set(socketId, {
-      userId: presence.userId,
-      userName: presence.userName,
-      avatarUrl: presence.avatarUrl,
-    })
-
-    logger.debug(`Added user ${presence.userId} to workflow ${workflowId} (socket: ${socketId})`)
-  }
-
-  async removeUserFromRoom(socketId: string): Promise<string | null> {
-    const workflowId = this.socketToWorkflow.get(socketId)
-
-    if (!workflowId) {
-      return null
-    }
-
-    const room = this.workflowRooms.get(workflowId)
-    if (room) {
-      room.users.delete(socketId)
-      room.activeConnections = Math.max(0, room.activeConnections - 1)
-
-      // Clean up empty rooms
-      if (room.activeConnections === 0) {
-        this.workflowRooms.delete(workflowId)
-        logger.info(`Cleaned up empty workflow room: ${workflowId}`)
-      }
-    }
-
-    this.socketToWorkflow.delete(socketId)
-    this.userSessions.delete(socketId)
-
-    logger.debug(`Removed socket ${socketId} from workflow ${workflowId}`)
-    return workflowId
-  }
-
-  async getWorkflowIdForSocket(socketId: string): Promise<string | null> {
-    return this.socketToWorkflow.get(socketId) ?? null
-  }
-
-  async getUserSession(socketId: string): Promise<UserSession | null> {
-    return this.userSessions.get(socketId) ?? null
-  }
-
-  async getWorkflowUsers(workflowId: string): Promise<UserPresence[]> {
-    const room = this.workflowRooms.get(workflowId)
-    if (!room) return []
-    return Array.from(room.users.values())
-  }
-
-  async hasWorkflowRoom(workflowId: string): Promise<boolean> {
-    return this.workflowRooms.has(workflowId)
-  }
-
-  async updateUserActivity(
-    workflowId: string,
-    socketId: string,
-    updates: Partial<Pick<UserPresence, 'cursor' | 'selection' | 'lastActivity'>>
-  ): Promise<void> {
-    const room = this.workflowRooms.get(workflowId)
-    if (!room) return
-
-    const presence = room.users.get(socketId)
-    if (presence) {
-      if (updates.cursor !== undefined) presence.cursor = updates.cursor
-      if (updates.selection !== undefined) presence.selection = updates.selection
-      presence.lastActivity = updates.lastActivity ?? Date.now()
-    }
-  }
-
-  async updateRoomLastModified(workflowId: string): Promise<void> {
-    const room = this.workflowRooms.get(workflowId)
-    if (room) {
-      room.lastModified = Date.now()
-    }
-  }
-
-  async broadcastPresenceUpdate(workflowId: string): Promise<void> {
-    const users = await this.getWorkflowUsers(workflowId)
-    this._io.to(workflowId).emit('presence-update', users)
-  }
-
-  emitToWorkflow<T = unknown>(workflowId: string, event: string, payload: T): void {
-    this._io.to(workflowId).emit(event, payload)
-  }
-
-  async getUniqueUserCount(workflowId: string): Promise<number> {
-    const room = this.workflowRooms.get(workflowId)
-    if (!room) return 0
-
-    const uniqueUsers = new Set<string>()
-    room.users.forEach((presence) => {
-      uniqueUsers.add(presence.userId)
-    })
-
-    return uniqueUsers.size
-  }
-
-  async getTotalActiveConnections(): Promise<number> {
-    let total = 0
-    for (const room of this.workflowRooms.values()) {
-      total += room.activeConnections
-    }
-    return total
-  }
-
-  async handleWorkflowDeletion(workflowId: string): Promise<void> {
-    logger.info(`Handling workflow deletion notification for ${workflowId}`)
-
-    const room = this.workflowRooms.get(workflowId)
-    if (!room) {
-      logger.debug(`No active room found for deleted workflow ${workflowId}`)
-      return
-    }
-
-    this._io.to(workflowId).emit('workflow-deleted', {
-      workflowId,
-      message: 'This workflow has been deleted',
-      timestamp: Date.now(),
-    })
-
-    const socketsToDisconnect: string[] = []
-    room.users.forEach((_presence, socketId) => {
-      socketsToDisconnect.push(socketId)
-    })
-
-    for (const socketId of socketsToDisconnect) {
-      const socket = this._io.sockets.sockets.get(socketId)
-      if (socket) {
-        socket.leave(workflowId)
-        logger.debug(`Disconnected socket ${socketId} from deleted workflow ${workflowId}`)
-      }
-      await this.removeUserFromRoom(socketId)
-    }
-
-    this.workflowRooms.delete(workflowId)
-    logger.info(
-      `Cleaned up workflow room ${workflowId} after deletion (${socketsToDisconnect.length} users disconnected)`
-    )
-  }
-
-  async handleWorkflowRevert(workflowId: string, timestamp: number): Promise<void> {
-    logger.info(`Handling workflow revert notification for ${workflowId}`)
-
-    const room = this.workflowRooms.get(workflowId)
-    if (!room) {
-      logger.debug(`No active room found for reverted workflow ${workflowId}`)
-      return
-    }
-
-    this._io.to(workflowId).emit('workflow-reverted', {
-      workflowId,
-      message: 'Workflow has been reverted to deployed state',
-      timestamp,
-    })
-
-    room.lastModified = timestamp
-
-    logger.info(`Notified ${room.users.size} users about workflow revert: ${workflowId}`)
-  }
-
-  async handleWorkflowUpdate(workflowId: string): Promise<void> {
-    logger.info(`Handling workflow update notification for ${workflowId}`)
-
-    const room = this.workflowRooms.get(workflowId)
-    if (!room) {
-      logger.debug(`No active room found for updated workflow ${workflowId}`)
-      return
-    }
-
-    const timestamp = Date.now()
-
-    this._io.to(workflowId).emit('workflow-updated', {
-      workflowId,
-      message: 'Workflow has been updated externally',
-      timestamp,
-    })
-
-    room.lastModified = timestamp
-
-    logger.info(`Notified ${room.users.size} users about workflow update: ${workflowId}`)
-  }
-
-  async handleCopilotWorkflowEdit(workflowId: string, description?: string): Promise<void> {
-    logger.info(`Handling copilot workflow edit notification for ${workflowId}`)
-
-    const room = this.workflowRooms.get(workflowId)
-    if (!room) {
-      logger.debug(`No active room found for copilot workflow edit ${workflowId}`)
-      return
-    }
-
-    const timestamp = Date.now()
-
-    this._io.to(workflowId).emit('copilot-workflow-edit', {
-      workflowId,
-      description,
-      message: 'Copilot has edited the workflow - rehydrating from database',
-      timestamp,
-    })
-
-    room.lastModified = timestamp
-
-    logger.info(`Notified ${room.users.size} users about copilot workflow edit: ${workflowId}`)
-  }
-}
--- a/apps/sim/socket/rooms/redis-manager.ts
+++ b/apps/sim/socket/rooms/redis-manager.ts
@@ -1,434 +0,0 @@
-import { createLogger } from '@sim/logger'
-import { createClient, type RedisClientType } from 'redis'
-import type { Server } from 'socket.io'
-import type { IRoomManager, UserPresence, UserSession } from '@/socket/rooms/types'
-
-const logger = createLogger('RedisRoomManager')
-
-const KEYS = {
-  workflowUsers: (wfId: string) => `workflow:${wfId}:users`,
-  workflowMeta: (wfId: string) => `workflow:${wfId}:meta`,
-  socketWorkflow: (socketId: string) => `socket:${socketId}:workflow`,
-  socketSession: (socketId: string) => `socket:${socketId}:session`,
-} as const
-
-const SOCKET_KEY_TTL = 3600
-
-/**
- * Lua script for atomic user removal from room.
- * Returns workflowId if user was removed, null otherwise.
- * Handles room cleanup atomically to prevent race conditions.
- */
-const REMOVE_USER_SCRIPT = `
-local socketWorkflowKey = KEYS[1]
-local socketSessionKey = KEYS[2]
-local workflowUsersPrefix = ARGV[1]
-local workflowMetaPrefix = ARGV[2]
-local socketId = ARGV[3]
-
-local workflowId = redis.call('GET', socketWorkflowKey)
-if not workflowId then
-  return nil
-end
-
-local workflowUsersKey = workflowUsersPrefix .. workflowId .. ':users'
-local workflowMetaKey = workflowMetaPrefix .. workflowId .. ':meta'
-
-redis.call('HDEL', workflowUsersKey, socketId)
-redis.call('DEL', socketWorkflowKey, socketSessionKey)
-
-local remaining = redis.call('HLEN', workflowUsersKey)
-if remaining == 0 then
-  redis.call('DEL', workflowUsersKey, workflowMetaKey)
-end
-
-return workflowId
-`
-
-/**
- * Lua script for atomic user activity update.
- * Performs read-modify-write atomically to prevent lost updates.
- * Also refreshes TTL on socket keys to prevent expiry during long sessions.
- */
-const UPDATE_ACTIVITY_SCRIPT = `
-local workflowUsersKey = KEYS[1]
-local socketWorkflowKey = KEYS[2]
-local socketSessionKey = KEYS[3]
-local socketId = ARGV[1]
-local cursorJson = ARGV[2]
-local selectionJson = ARGV[3]
-local lastActivity = ARGV[4]
-local ttl = tonumber(ARGV[5])
-
-local existingJson = redis.call('HGET', workflowUsersKey, socketId)
-if not existingJson then
-  return 0
-end
-
-local existing = cjson.decode(existingJson)
-
-if cursorJson ~= '' then
-  existing.cursor = cjson.decode(cursorJson)
-end
-if selectionJson ~= '' then
-  existing.selection = cjson.decode(selectionJson)
-end
-existing.lastActivity = tonumber(lastActivity)
-
-redis.call('HSET', workflowUsersKey, socketId, cjson.encode(existing))
-redis.call('EXPIRE', socketWorkflowKey, ttl)
-redis.call('EXPIRE', socketSessionKey, ttl)
-return 1
-`
-
-/**
- * Redis-backed room manager for multi-pod deployments.
- * Uses Lua scripts for atomic operations to prevent race conditions.
- */
-export class RedisRoomManager implements IRoomManager {
-  private redis: RedisClientType
-  private _io: Server
-  private isConnected = false
-  private removeUserScriptSha: string | null = null
-  private updateActivityScriptSha: string | null = null
-
-  constructor(io: Server, redisUrl: string) {
-    this._io = io
-    this.redis = createClient({
-      url: redisUrl,
-      socket: {
-        reconnectStrategy: (retries) => {
-          if (retries > 10) {
-            logger.error('Redis reconnection failed after 10 attempts')
-            return new Error('Redis reconnection failed')
-          }
-          const delay = Math.min(retries * 100, 3000)
-          logger.warn(`Redis reconnecting in ${delay}ms (attempt ${retries})`)
-          return delay
-        },
-      },
-    })
-
-    this.redis.on('error', (err) => {
-      logger.error('Redis client error:', err)
-    })
-
-    this.redis.on('reconnecting', () => {
-      logger.warn('Redis client reconnecting...')
-      this.isConnected = false
-    })
-
-    this.redis.on('ready', () => {
-      logger.info('Redis client ready')
-      this.isConnected = true
-    })
-  }
-
-  get io(): Server {
-    return this._io
-  }
-
-  async initialize(): Promise<void> {
-    if (this.isConnected) return
-
-    try {
-      await this.redis.connect()
-      this.isConnected = true
-
-      // Pre-load Lua scripts for better performance
-      this.removeUserScriptSha = await this.redis.scriptLoad(REMOVE_USER_SCRIPT)
-      this.updateActivityScriptSha = await this.redis.scriptLoad(UPDATE_ACTIVITY_SCRIPT)
-
-      logger.info('RedisRoomManager connected to Redis and scripts loaded')
-    } catch (error) {
-      logger.error('Failed to connect to Redis:', error)
-      throw error
-    }
-  }
-
-  async shutdown(): Promise<void> {
-    if (!this.isConnected) return
-
-    try {
-      await this.redis.quit()
-      this.isConnected = false
-      logger.info('RedisRoomManager disconnected from Redis')
-    } catch (error) {
-      logger.error('Error during Redis shutdown:', error)
-    }
-  }
-
-  async addUserToRoom(workflowId: string, socketId: string, presence: UserPresence): Promise<void> {
-    try {
-      const pipeline = this.redis.multi()
-
-      pipeline.hSet(KEYS.workflowUsers(workflowId), socketId, JSON.stringify(presence))
-      pipeline.hSet(KEYS.workflowMeta(workflowId), 'lastModified', Date.now().toString())
-      pipeline.set(KEYS.socketWorkflow(socketId), workflowId)
-      pipeline.expire(KEYS.socketWorkflow(socketId), SOCKET_KEY_TTL)
-      pipeline.hSet(KEYS.socketSession(socketId), {
-        userId: presence.userId,
-        userName: presence.userName,
-        avatarUrl: presence.avatarUrl || '',
-      })
-      pipeline.expire(KEYS.socketSession(socketId), SOCKET_KEY_TTL)
-
-      const results = await pipeline.exec()
-
-      // Check if any command failed
-      const failed = results.some((result) => result instanceof Error)
-      if (failed) {
-        logger.error(`Pipeline partially failed when adding user to room`, { workflowId, socketId })
-        throw new Error('Failed to store user session data in Redis')
-      }
-
-      logger.debug(`Added user ${presence.userId} to workflow ${workflowId} (socket: ${socketId})`)
-    } catch (error) {
-      logger.error(`Failed to add user to room: ${socketId} -> ${workflowId}`, error)
-      throw error
-    }
-  }
-
-  async removeUserFromRoom(socketId: string, retried = false): Promise<string | null> {
-    if (!this.removeUserScriptSha) {
-      logger.error('removeUserFromRoom called before initialize()')
-      return null
-    }
-
-    try {
-      const workflowId = await this.redis.evalSha(this.removeUserScriptSha, {
-        keys: [KEYS.socketWorkflow(socketId), KEYS.socketSession(socketId)],
-        arguments: ['workflow:', 'workflow:', socketId],
-      })
-
-      if (workflowId) {
-        logger.debug(`Removed socket ${socketId} from workflow ${workflowId}`)
-      }
-      return workflowId as string | null
-    } catch (error) {
-      if ((error as Error).message?.includes('NOSCRIPT') && !retried) {
-        logger.warn('Lua script not found, reloading...')
-        this.removeUserScriptSha = await this.redis.scriptLoad(REMOVE_USER_SCRIPT)
-        return this.removeUserFromRoom(socketId, true)
-      }
-      logger.error(`Failed to remove user from room: ${socketId}`, error)
-      return null
-    }
-  }
-
-  async getWorkflowIdForSocket(socketId: string): Promise<string | null> {
-    return this.redis.get(KEYS.socketWorkflow(socketId))
-  }
-
-  async getUserSession(socketId: string): Promise<UserSession | null> {
-    try {
-      const session = await this.redis.hGetAll(KEYS.socketSession(socketId))
-
-      if (!session.userId) {
-        return null
-      }
-
-      return {
-        userId: session.userId,
-        userName: session.userName,
-        avatarUrl: session.avatarUrl || undefined,
-      }
-    } catch (error) {
-      logger.error(`Failed to get user session for ${socketId}:`, error)
-      return null
-    }
-  }
-
-  async getWorkflowUsers(workflowId: string): Promise<UserPresence[]> {
-    try {
-      const users = await this.redis.hGetAll(KEYS.workflowUsers(workflowId))
-      return Object.entries(users)
-        .map(([socketId, json]) => {
-          try {
-            return JSON.parse(json) as UserPresence
-          } catch {
-            logger.warn(`Corrupted user data for socket ${socketId}, skipping`)
-            return null
-          }
-        })
-        .filter((u): u is UserPresence => u !== null)
-    } catch (error) {
-      logger.error(`Failed to get workflow users for ${workflowId}:`, error)
-      return []
-    }
-  }
-
-  async hasWorkflowRoom(workflowId: string): Promise<boolean> {
-    const exists = await this.redis.exists(KEYS.workflowUsers(workflowId))
-    return exists > 0
-  }
-
-  async updateUserActivity(
-    workflowId: string,
-    socketId: string,
-    updates: Partial<Pick<UserPresence, 'cursor' | 'selection' | 'lastActivity'>>,
-    retried = false
-  ): Promise<void> {
-    if (!this.updateActivityScriptSha) {
-      logger.error('updateUserActivity called before initialize()')
-      return
-    }
-
-    try {
-      await this.redis.evalSha(this.updateActivityScriptSha, {
-        keys: [
-          KEYS.workflowUsers(workflowId),
-          KEYS.socketWorkflow(socketId),
-          KEYS.socketSession(socketId),
-        ],
-        arguments: [
-          socketId,
-          updates.cursor !== undefined ? JSON.stringify(updates.cursor) : '',
-          updates.selection !== undefined ? JSON.stringify(updates.selection) : '',
-          (updates.lastActivity ?? Date.now()).toString(),
-          SOCKET_KEY_TTL.toString(),
-        ],
-      })
-    } catch (error) {
-      if ((error as Error).message?.includes('NOSCRIPT') && !retried) {
-        logger.warn('Lua script not found, reloading...')
-        this.updateActivityScriptSha = await this.redis.scriptLoad(UPDATE_ACTIVITY_SCRIPT)
-        return this.updateUserActivity(workflowId, socketId, updates, true)
-      }
-      logger.error(`Failed to update user activity: ${socketId}`, error)
-    }
-  }
-
-  async updateRoomLastModified(workflowId: string): Promise<void> {
-    await this.redis.hSet(KEYS.workflowMeta(workflowId), 'lastModified', Date.now().toString())
-  }
-
-  async broadcastPresenceUpdate(workflowId: string): Promise<void> {
-    const users = await this.getWorkflowUsers(workflowId)
-    // io.to() with Redis adapter broadcasts to all pods
-    this._io.to(workflowId).emit('presence-update', users)
-  }
-
-  emitToWorkflow<T = unknown>(workflowId: string, event: string, payload: T): void {
-    this._io.to(workflowId).emit(event, payload)
-  }
-
-  async getUniqueUserCount(workflowId: string): Promise<number> {
-    const users = await this.getWorkflowUsers(workflowId)
-    const uniqueUserIds = new Set(users.map((u) => u.userId))
-    return uniqueUserIds.size
-  }
-
-  async getTotalActiveConnections(): Promise<number> {
-    // This is more complex with Redis - we'd need to scan all workflow:*:users keys
-    // For now, just count sockets in this server instance
-    // The true count would require aggregating across all pods
-    return this._io.sockets.sockets.size
-  }
-
-  async handleWorkflowDeletion(workflowId: string): Promise<void> {
-    logger.info(`Handling workflow deletion notification for ${workflowId}`)
-
-    try {
-      const users = await this.getWorkflowUsers(workflowId)
-      if (users.length === 0) {
-        logger.debug(`No active users found for deleted workflow ${workflowId}`)
-        return
-      }
-
-      // Notify all clients across all pods via Redis adapter
-      this._io.to(workflowId).emit('workflow-deleted', {
-        workflowId,
-        message: 'This workflow has been deleted',
-        timestamp: Date.now(),
-      })
-
-      // Use Socket.IO's cross-pod socketsLeave() to remove all sockets from the room
-      // This works across all pods when using the Redis adapter
-      await this._io.in(workflowId).socketsLeave(workflowId)
-      logger.debug(`All sockets left workflow room ${workflowId} via socketsLeave()`)
-
-      // Remove all users from Redis state
-      for (const user of users) {
-        await this.removeUserFromRoom(user.socketId)
-      }
-
-      // Clean up room data
-      await this.redis.del([KEYS.workflowUsers(workflowId), KEYS.workflowMeta(workflowId)])
-
-      logger.info(
-        `Cleaned up workflow room ${workflowId} after deletion (${users.length} users disconnected)`
-      )
-    } catch (error) {
-      logger.error(`Failed to handle workflow deletion for ${workflowId}:`, error)
-    }
-  }
-
-  async handleWorkflowRevert(workflowId: string, timestamp: number): Promise<void> {
-    logger.info(`Handling workflow revert notification for ${workflowId}`)
-
-    const hasRoom = await this.hasWorkflowRoom(workflowId)
-    if (!hasRoom) {
-      logger.debug(`No active room found for reverted workflow ${workflowId}`)
-      return
-    }
-
-    this._io.to(workflowId).emit('workflow-reverted', {
-      workflowId,
-      message: 'Workflow has been reverted to deployed state',
-      timestamp,
-    })
-
-    await this.updateRoomLastModified(workflowId)
-
-    const userCount = await this.getUniqueUserCount(workflowId)
-    logger.info(`Notified ${userCount} users about workflow revert: ${workflowId}`)
-  }
-
-  async handleWorkflowUpdate(workflowId: string): Promise<void> {
-    logger.info(`Handling workflow update notification for ${workflowId}`)
-
-    const hasRoom = await this.hasWorkflowRoom(workflowId)
-    if (!hasRoom) {
-      logger.debug(`No active room found for updated workflow ${workflowId}`)
-      return
-    }
-
-    const timestamp = Date.now()
-
-    this._io.to(workflowId).emit('workflow-updated', {
-      workflowId,
-      message: 'Workflow has been updated externally',
-      timestamp,
-    })
-
-    await this.updateRoomLastModified(workflowId)
-
-    const userCount = await this.getUniqueUserCount(workflowId)
-    logger.info(`Notified ${userCount} users about workflow update: ${workflowId}`)
-  }
-
-  async handleCopilotWorkflowEdit(workflowId: string, description?: string): Promise<void> {
-    logger.info(`Handling copilot workflow edit notification for ${workflowId}`)
-
-    const hasRoom = await this.hasWorkflowRoom(workflowId)
-    if (!hasRoom) {
-      logger.debug(`No active room found for copilot workflow edit ${workflowId}`)
-      return
-    }
-
-    const timestamp = Date.now()
-
-    this._io.to(workflowId).emit('copilot-workflow-edit', {
-      workflowId,
-      description,
-      message: 'Copilot has edited the workflow - rehydrating from database',
-      timestamp,
-    })
-
-    await this.updateRoomLastModified(workflowId)
-
-    const userCount = await this.getUniqueUserCount(workflowId)
-    logger.info(`Notified ${userCount} users about copilot workflow edit: ${workflowId}`)
-  }
-}
--- a/apps/sim/socket/rooms/types.ts
+++ b/apps/sim/socket/rooms/types.ts
@@ -1,140 +0,0 @@
-import type { Server } from 'socket.io'
-
-/**
- * User presence data stored in room state
- */
-export interface UserPresence {
-  userId: string
-  workflowId: string
-  userName: string
-  socketId: string
-  tabSessionId?: string
-  joinedAt: number
-  lastActivity: number
-  role: string
-  cursor?: { x: number; y: number }
-  selection?: { type: 'block' | 'edge' | 'none'; id?: string }
-  avatarUrl?: string | null
-}
-
-/**
- * User session data (minimal info for quick lookups)
- */
-export interface UserSession {
-  userId: string
-  userName: string
-  avatarUrl?: string | null
-}
-
-/**
- * Workflow room state
- */
-export interface WorkflowRoom {
-  workflowId: string
-  users: Map<string, UserPresence>
-  lastModified: number
-  activeConnections: number
-}
-
-/**
- * Common interface for room managers (in-memory and Redis)
- * All methods that access state are async to support Redis operations
- */
-export interface IRoomManager {
-  readonly io: Server
-
-  /**
-   * Initialize the room manager (connect to Redis, etc.)
-   */
-  initialize(): Promise<void>
-
-  /**
-   * Clean shutdown
-   */
-  shutdown(): Promise<void>
-
-  /**
-   * Add a user to a workflow room
-   */
-  addUserToRoom(workflowId: string, socketId: string, presence: UserPresence): Promise<void>
-
-  /**
-   * Remove a user from their current room
-   * Returns the workflowId they were in, or null if not in any room
-   */
-  removeUserFromRoom(socketId: string): Promise<string | null>
-
-  /**
-   * Get the workflow ID for a socket
-   */
-  getWorkflowIdForSocket(socketId: string): Promise<string | null>
-
-  /**
-   * Get user session data for a socket
-   */
-  getUserSession(socketId: string): Promise<UserSession | null>
-
-  /**
-   * Get all users in a workflow room
-   */
-  getWorkflowUsers(workflowId: string): Promise<UserPresence[]>
-
-  /**
-   * Check if a workflow room exists
-   */
-  hasWorkflowRoom(workflowId: string): Promise<boolean>
-
-  /**
-   * Update user activity (cursor, selection, lastActivity)
-   */
-  updateUserActivity(
-    workflowId: string,
-    socketId: string,
-    updates: Partial<Pick<UserPresence, 'cursor' | 'selection' | 'lastActivity'>>
-  ): Promise<void>
-
-  /**
-   * Update room's lastModified timestamp
-   */
-  updateRoomLastModified(workflowId: string): Promise<void>
-
-  /**
-   * Broadcast presence update to all clients in a workflow room
-   */
-  broadcastPresenceUpdate(workflowId: string): Promise<void>
-
-  /**
-   * Emit an event to all clients in a workflow room
-   */
-  emitToWorkflow<T = unknown>(workflowId: string, event: string, payload: T): void
-
-  /**
-   * Get the number of unique users in a workflow room
-   */
-  getUniqueUserCount(workflowId: string): Promise<number>
-
-  /**
-   * Get total active connections across all rooms
-   */
-  getTotalActiveConnections(): Promise<number>
-
-  /**
-   * Handle workflow deletion - notify users and clean up room
-   */
-  handleWorkflowDeletion(workflowId: string): Promise<void>
-
-  /**
-   * Handle workflow revert - notify users
-   */
-  handleWorkflowRevert(workflowId: string, timestamp: number): Promise<void>
-
-  /**
-   * Handle workflow update - notify users
-   */
-  handleWorkflowUpdate(workflowId: string): Promise<void>
-
-  /**
-   * Handle copilot workflow edit - notify users to rehydrate
-   */
-  handleCopilotWorkflowEdit(workflowId: string, description?: string): Promise<void>
-}
--- a/apps/sim/socket/routes/http.ts
+++ b/apps/sim/socket/routes/http.ts
@@ -1,52 +1,11 @@
 import type { IncomingMessage, ServerResponse } from 'http'
-import { env } from '@/lib/core/config/env'
-import type { IRoomManager } from '@/socket/rooms'
+import type { RoomManager } from '@/socket/rooms/manager'

 interface Logger {
-  info: (message: string, ...args: unknown[]) => void
-  error: (message: string, ...args: unknown[]) => void
-  debug: (message: string, ...args: unknown[]) => void
-  warn: (message: string, ...args: unknown[]) => void
-}
-
-function checkInternalApiKey(req: IncomingMessage): { success: boolean; error?: string } {
-  const apiKey = req.headers['x-api-key']
-  const expectedApiKey = env.INTERNAL_API_SECRET
-
-  if (!expectedApiKey) {
-    return { success: false, error: 'Internal API key not configured' }
-  }
-
-  if (!apiKey) {
-    return { success: false, error: 'API key required' }
-  }
-
-  if (apiKey !== expectedApiKey) {
-    return { success: false, error: 'Invalid API key' }
-  }
-
-  return { success: true }
-}
-
-function readRequestBody(req: IncomingMessage): Promise<string> {
-  return new Promise((resolve, reject) => {
-    let body = ''
-    req.on('data', (chunk) => {
-      body += chunk.toString()
-    })
-    req.on('end', () => resolve(body))
-    req.on('error', reject)
-  })
-}
-
-function sendSuccess(res: ServerResponse): void {
-  res.writeHead(200, { 'Content-Type': 'application/json' })
-  res.end(JSON.stringify({ success: true }))
-}
-
-function sendError(res: ServerResponse, message: string, status = 500): void {
-  res.writeHead(status, { 'Content-Type': 'application/json' })
-  res.end(JSON.stringify({ error: message }))
+  info: (message: string, ...args: any[]) => void
+  error: (message: string, ...args: any[]) => void
+  debug: (message: string, ...args: any[]) => void
+  warn: (message: string, ...args: any[]) => void
 }

 /**
@@ -55,91 +14,101 @@ function sendError(res: ServerResponse, message: string, status = 500): void {
 * @param logger - Logger instance for logging requests and errors
 * @returns HTTP request handler function
 */
-export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
-  return async (req: IncomingMessage, res: ServerResponse) => {
-    // Health check doesn't require auth
+export function createHttpHandler(roomManager: RoomManager, logger: Logger) {
+  return (req: IncomingMessage, res: ServerResponse) => {
    if (req.method === 'GET' && req.url === '/health') {
-      try {
-        const connections = await roomManager.getTotalActiveConnections()
-        res.writeHead(200, { 'Content-Type': 'application/json' })
-        res.end(
-          JSON.stringify({
-            status: 'ok',
-            timestamp: new Date().toISOString(),
-            connections,
-          })
-        )
-      } catch (error) {
-        logger.error('Error in health check:', error)
-        res.writeHead(503, { 'Content-Type': 'application/json' })
-        res.end(JSON.stringify({ status: 'error', message: 'Health check failed' }))
-      }
+      res.writeHead(200, { 'Content-Type': 'application/json' })
+      res.end(
+        JSON.stringify({
+          status: 'ok',
+          timestamp: new Date().toISOString(),
+          connections: roomManager.getTotalActiveConnections(),
+        })
+      )
      return
    }

-    // All POST endpoints require internal API key authentication
-    if (req.method === 'POST') {
-      const authResult = checkInternalApiKey(req)
-      if (!authResult.success) {
-        res.writeHead(401, { 'Content-Type': 'application/json' })
-        res.end(JSON.stringify({ error: authResult.error }))
-        return
-      }
-    }
-
    // Handle workflow deletion notifications from the main API
    if (req.method === 'POST' && req.url === '/api/workflow-deleted') {
-      try {
-        const body = await readRequestBody(req)
-        const { workflowId } = JSON.parse(body)
-        await roomManager.handleWorkflowDeletion(workflowId)
-        sendSuccess(res)
-      } catch (error) {
-        logger.error('Error handling workflow deletion notification:', error)
-        sendError(res, 'Failed to process deletion notification')
-      }
+      let body = ''
+      req.on('data', (chunk) => {
+        body += chunk.toString()
+      })
+      req.on('end', () => {
+        try {
+          const { workflowId } = JSON.parse(body)
+          roomManager.handleWorkflowDeletion(workflowId)
+          res.writeHead(200, { 'Content-Type': 'application/json' })
+          res.end(JSON.stringify({ success: true }))
+        } catch (error) {
+          logger.error('Error handling workflow deletion notification:', error)
+          res.writeHead(500, { 'Content-Type': 'application/json' })
+          res.end(JSON.stringify({ error: 'Failed to process deletion notification' }))
+        }
+      })
      return
    }

    // Handle workflow update notifications from the main API
    if (req.method === 'POST' && req.url === '/api/workflow-updated') {
-      try {
-        const body = await readRequestBody(req)
-        const { workflowId } = JSON.parse(body)
-        await roomManager.handleWorkflowUpdate(workflowId)
-        sendSuccess(res)
-      } catch (error) {
-        logger.error('Error handling workflow update notification:', error)
-        sendError(res, 'Failed to process update notification')
-      }
+      let body = ''
+      req.on('data', (chunk) => {
+        body += chunk.toString()
+      })
+      req.on('end', () => {
+        try {
+          const { workflowId } = JSON.parse(body)
+          roomManager.handleWorkflowUpdate(workflowId)
+          res.writeHead(200, { 'Content-Type': 'application/json' })
+          res.end(JSON.stringify({ success: true }))
+        } catch (error) {
+          logger.error('Error handling workflow update notification:', error)
+          res.writeHead(500, { 'Content-Type': 'application/json' })
+          res.end(JSON.stringify({ error: 'Failed to process update notification' }))
+        }
+      })
      return
    }

    // Handle copilot workflow edit notifications from the main API
    if (req.method === 'POST' && req.url === '/api/copilot-workflow-edit') {
-      try {
-        const body = await readRequestBody(req)
-        const { workflowId, description } = JSON.parse(body)
-        await roomManager.handleCopilotWorkflowEdit(workflowId, description)
-        sendSuccess(res)
-      } catch (error) {
-        logger.error('Error handling copilot workflow edit notification:', error)
-        sendError(res, 'Failed to process copilot edit notification')
-      }
+      let body = ''
+      req.on('data', (chunk) => {
+        body += chunk.toString()
+      })
+      req.on('end', () => {
+        try {
+          const { workflowId, description } = JSON.parse(body)
+          roomManager.handleCopilotWorkflowEdit(workflowId, description)
+          res.writeHead(200, { 'Content-Type': 'application/json' })
+          res.end(JSON.stringify({ success: true }))
+        } catch (error) {
+          logger.error('Error handling copilot workflow edit notification:', error)
+          res.writeHead(500, { 'Content-Type': 'application/json' })
+          res.end(JSON.stringify({ error: 'Failed to process copilot edit notification' }))
+        }
+      })
      return
    }

    // Handle workflow revert notifications from the main API
    if (req.method === 'POST' && req.url === '/api/workflow-reverted') {
-      try {
-        const body = await readRequestBody(req)
-        const { workflowId, timestamp } = JSON.parse(body)
-        await roomManager.handleWorkflowRevert(workflowId, timestamp)
-        sendSuccess(res)
-      } catch (error) {
-        logger.error('Error handling workflow revert notification:', error)
-        sendError(res, 'Failed to process revert notification')
-      }
+      let body = ''
+      req.on('data', (chunk) => {
+        body += chunk.toString()
+      })
+      req.on('end', () => {
+        try {
+          const { workflowId, timestamp } = JSON.parse(body)
+          roomManager.handleWorkflowRevert(workflowId, timestamp)
+          res.writeHead(200, { 'Content-Type': 'application/json' })
+          res.end(JSON.stringify({ success: true }))
+        } catch (error) {
+          logger.error('Error handling workflow revert notification:', error)
+          res.writeHead(500, { 'Content-Type': 'application/json' })
+          res.end(JSON.stringify({ error: 'Failed to process revert notification' }))
+        }
+      })
      return
    }

--- a/apps/sim/socket/validation/schemas.ts
+++ b/apps/sim/socket/validation/schemas.ts
@@ -239,3 +239,5 @@ export const WorkflowOperationSchema = z.union([
  VariableOperationSchema,
  WorkflowStateOperationSchema,
 ])
+
+export { PositionSchema, AutoConnectEdgeSchema }
--- a/apps/sim/stores/operation-queue/store.ts
+++ b/apps/sim/stores/operation-queue/store.ts
@@ -4,19 +4,6 @@ import type { OperationQueueState, QueuedOperation } from './types'

 const logger = createLogger('OperationQueue')

-/** Timeout for subblock/variable operations before considering them failed */
-const SUBBLOCK_VARIABLE_TIMEOUT_MS = 15000
-/** Timeout for structural operations before considering them failed */
-const STRUCTURAL_TIMEOUT_MS = 5000
-/** Maximum retry attempts for subblock/variable operations */
-const SUBBLOCK_VARIABLE_MAX_RETRIES = 5
-/** Maximum retry attempts for structural operations */
-const STRUCTURAL_MAX_RETRIES = 3
-/** Maximum retry delay cap for subblock/variable operations */
-const SUBBLOCK_VARIABLE_MAX_RETRY_DELAY_MS = 3000
-/** Base retry delay multiplier (1s, 2s, 3s for linear) */
-const RETRY_DELAY_BASE_MS = 1000
-
 const retryTimeouts = new Map<string, NodeJS.Timeout>()
 const operationTimeouts = new Map<string, NodeJS.Timeout>()

@@ -213,14 +200,14 @@ export const useOperationQueueStore = create<OperationQueueState>((set, get) =>
      (operation.operation.operation === 'variable-update' &&
        operation.operation.target === 'variable')

-    const maxRetries = isSubblockOrVariable ? SUBBLOCK_VARIABLE_MAX_RETRIES : STRUCTURAL_MAX_RETRIES
+    const maxRetries = isSubblockOrVariable ? 5 : 3 // 5 retries for text, 3 for structural

    if (operation.retryCount < maxRetries) {
      const newRetryCount = operation.retryCount + 1
      // Faster retries for subblock/variable, exponential for structural
      const delay = isSubblockOrVariable
-        ? Math.min(RETRY_DELAY_BASE_MS * newRetryCount, SUBBLOCK_VARIABLE_MAX_RETRY_DELAY_MS)
-        : 2 ** newRetryCount * RETRY_DELAY_BASE_MS
+        ? Math.min(1000 * newRetryCount, 3000) // 1s, 2s, 3s, 3s, 3s (cap at 3s)
+        : 2 ** newRetryCount * 1000 // 2s, 4s, 8s (exponential for structural)

      logger.warn(
        `Operation failed, retrying in ${delay}ms (attempt ${newRetryCount}/${maxRetries})`,
@@ -322,9 +309,7 @@ export const useOperationQueueStore = create<OperationQueueState>((set, get) =>
        nextOperation.operation.target === 'subblock') ||
      (nextOperation.operation.operation === 'variable-update' &&
        nextOperation.operation.target === 'variable')
-    const timeoutDuration = isSubblockOrVariable
-      ? SUBBLOCK_VARIABLE_TIMEOUT_MS
-      : STRUCTURAL_TIMEOUT_MS
+    const timeoutDuration = isSubblockOrVariable ? 15000 : 5000 // 15s for text edits, 5s for structural ops

    const timeoutId = setTimeout(() => {
      logger.warn(`Operation timeout - no server response after ${timeoutDuration}ms`, {
--- a/bun.lock
+++ b/bun.lock
@@ -104,7 +104,6 @@
        "@react-email/components": "^0.0.34",
        "@react-email/render": "2.0.0",
        "@sim/logger": "workspace:*",
-        "@socket.io/redis-adapter": "8.3.0",
        "@t3-oss/env-nextjs": "0.13.4",
        "@tanstack/react-query": "5.90.8",
        "@tanstack/react-query-devtools": "5.90.2",
@@ -175,7 +174,6 @@
        "react-simple-code-editor": "^0.14.1",
        "react-window": "2.2.3",
        "reactflow": "^11.11.4",
-        "redis": "5.10.0",
        "rehype-autolink-headings": "^7.1.0",
        "rehype-slug": "^6.0.0",
        "remark-gfm": "4.0.1",
@@ -1148,16 +1146,6 @@

    "@reactflow/node-toolbar": ["@reactflow/node-toolbar@1.3.14", "", { "dependencies": { "@reactflow/core": "11.11.4", "classcat": "^5.0.3", "zustand": "^4.4.1" }, "peerDependencies": { "react": ">=17", "react-dom": ">=17" } }, "sha512-rbynXQnH/xFNu4P9H+hVqlEUafDCkEoCy0Dg9mG22Sg+rY/0ck6KkrAQrYrTgXusd+cEJOMK0uOOFCK2/5rSGQ=="],

-    "@redis/bloom": ["@redis/bloom@5.10.0", "", { "peerDependencies": { "@redis/client": "^5.10.0" } }, "sha512-doIF37ob+l47n0rkpRNgU8n4iacBlKM9xLiP1LtTZTvz8TloJB8qx/MgvhMhKdYG+CvCY2aPBnN2706izFn/4A=="],
-
-    "@redis/client": ["@redis/client@5.10.0", "", { "dependencies": { "cluster-key-slot": "1.1.2" } }, "sha512-JXmM4XCoso6C75Mr3lhKA3eNxSzkYi3nCzxDIKY+YOszYsJjuKbFgVtguVPbLMOttN4iu2fXoc2BGhdnYhIOxA=="],
-
-    "@redis/json": ["@redis/json@5.10.0", "", { "peerDependencies": { "@redis/client": "^5.10.0" } }, "sha512-B2G8XlOmTPUuZtD44EMGbtoepQG34RCDXLZbjrtON1Djet0t5Ri7/YPXvL9aomXqP8lLTreaprtyLKF4tmXEEA=="],
-
-    "@redis/search": ["@redis/search@5.10.0", "", { "peerDependencies": { "@redis/client": "^5.10.0" } }, "sha512-3SVcPswoSfp2HnmWbAGUzlbUPn7fOohVu2weUQ0S+EMiQi8jwjL+aN2p6V3TI65eNfVsJ8vyPvqWklm6H6esmg=="],
-
-    "@redis/time-series": ["@redis/time-series@5.10.0", "", { "peerDependencies": { "@redis/client": "^5.10.0" } }, "sha512-cPkpddXH5kc/SdRhF0YG0qtjL+noqFT0AcHbQ6axhsPsO7iqPi1cjxgdkE9TNeKiBUUdCaU1DbqkR/LzbzPBhg=="],
-
    "@resvg/resvg-wasm": ["@resvg/resvg-wasm@2.4.0", "", {}, "sha512-C7c51Nn4yTxXFKvgh2txJFNweaVcfUPQxwEUFw4aWsCmfiBDJsTSwviIF8EcwjQ6k8bPyMWCl1vw4BdxE569Cg=="],

    "@rolldown/pluginutils": ["@rolldown/pluginutils@1.0.0-beta.27", "", {}, "sha512-+d0F4MKMCbeVUJwG96uQ4SgAznZNSq93I3V+9NHA4OpvqG8mRCpGdKmK8l/dl02h2CCDHwW2FqilnTyDcAnqjA=="],
@@ -1352,8 +1340,6 @@

    "@socket.io/component-emitter": ["@socket.io/component-emitter@3.1.2", "", {}, "sha512-9BCxFwvbGg/RsZK9tjXd8s4UcwR0MWeFQ1XEKIQVVvAGJyINdrqKMcTRyLoK8Rse1GjzLV9cwjWV1olXRWEXVA=="],

-    "@socket.io/redis-adapter": ["@socket.io/redis-adapter@8.3.0", "", { "dependencies": { "debug": "~4.3.1", "notepack.io": "~3.0.1", "uid2": "1.0.0" }, "peerDependencies": { "socket.io-adapter": "^2.5.4" } }, "sha512-ly0cra+48hDmChxmIpnESKrc94LjRL80TEmZVscuQ/WWkRP81nNj8W8cCGMqbI4L6NCuAaPRSzZF1a9GlAxxnA=="],
-
    "@standard-schema/spec": ["@standard-schema/spec@1.1.0", "", {}, "sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w=="],

    "@standard-schema/utils": ["@standard-schema/utils@0.3.0", "", {}, "sha512-e7Mew686owMaPJVNNLs55PUvgz371nKgwsc4vxE49zsODpJEnxgxRo2y/OKrqueavXgZNMDVj3DdHFlaSAeU8g=="],
@@ -2816,8 +2802,6 @@

    "normalize-range": ["normalize-range@0.1.2", "", {}, "sha512-bdok/XvKII3nUpklnV6P2hxtMNrCboOjAcyBuQnWEhO665FwrSNRxU+AqpsyvO6LgGYPspN+lu5CLtw4jPRKNA=="],

-    "notepack.io": ["notepack.io@3.0.1", "", {}, "sha512-TKC/8zH5pXIAMVQio2TvVDTtPRX+DJPHDqjRbxogtFiByHyzKmy96RA0JtCQJ+WouyyL4A10xomQzgbUT+1jCg=="],
-
    "npm-run-path": ["npm-run-path@5.3.0", "", { "dependencies": { "path-key": "^4.0.0" } }, "sha512-ppwTtiJZq0O/ai0z7yfudtBpWIoxM8yE6nHi1X47eFR2EWORqfbu6CnPlNsjeN683eT0qG6H/Pyf9fCcvjnnnQ=="],

    "npm-to-yarn": ["npm-to-yarn@3.0.1", "", {}, "sha512-tt6PvKu4WyzPwWUzy/hvPFqn+uwXO0K1ZHka8az3NnrhWJDmSqI8ncWq0fkL0k/lmmi5tAC11FXwXuh0rFbt1A=="],
@@ -3088,8 +3072,6 @@

    "redent": ["redent@3.0.0", "", { "dependencies": { "indent-string": "^4.0.0", "strip-indent": "^3.0.0" } }, "sha512-6tDA8g98We0zd0GvVeMT9arEOnTw9qM03L9cJXaCjrip1OO764RDBLBfrB4cwzNGDj5OA5ioymC9GkizgWJDUg=="],

-    "redis": ["redis@5.10.0", "", { "dependencies": { "@redis/bloom": "5.10.0", "@redis/client": "5.10.0", "@redis/json": "5.10.0", "@redis/search": "5.10.0", "@redis/time-series": "5.10.0" } }, "sha512-0/Y+7IEiTgVGPrLFKy8oAEArSyEJkU0zvgV5xyi9NzNQ+SLZmyFbUsWIbgPcd4UdUh00opXGKlXJwMmsis5Byw=="],
-
    "redis-errors": ["redis-errors@1.2.0", "", {}, "sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w=="],

    "redis-parser": ["redis-parser@3.0.0", "", { "dependencies": { "redis-errors": "^1.0.0" } }, "sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A=="],
@@ -3452,8 +3434,6 @@

    "ufo": ["ufo@1.6.3", "", {}, "sha512-yDJTmhydvl5lJzBmy/hyOAA0d+aqCBuwl818haVdYCRrWV84o7YyeVm4QlVHStqNrrJSTb6jKuFAVqAFsr+K3Q=="],

-    "uid2": ["uid2@1.0.0", "", {}, "sha512-+I6aJUv63YAcY9n4mQreLUt0d4lvwkkopDNmpomkAUz0fAkEMV9pRWxN0EjhW1YfRhcuyHg2v3mwddCDW1+LFQ=="],
-
    "ulid": ["ulid@2.4.0", "", { "bin": { "ulid": "bin/cli.js" } }, "sha512-fIRiVTJNcSRmXKPZtGzFQv9WRrZ3M9eoptl/teFJvjOzmpU+/K/JH6HZ8deBfb5vMEpicJcLn7JmvdknlMq7Zg=="],

    "unbzip2-stream": ["unbzip2-stream@1.4.3", "", { "dependencies": { "buffer": "^5.2.1", "through": "^2.3.8" } }, "sha512-mlExGW4w71ebDJviH16lQLtZS32VKqsSfk80GCfUlwT/4/hNRFsoscrF/c++9xinkMzECL1uL9DDwXqFWkruPg=="],
@@ -3872,8 +3852,6 @@

    "@shuding/opentype.js/fflate": ["fflate@0.7.4", "", {}, "sha512-5u2V/CDW15QM1XbbgS+0DfPxVB+jUKhWEKuuFuHncbk3tEEqzmoXL+2KyOFuKGqOnmdIy0/davWF1CkuwtibCw=="],

-    "@socket.io/redis-adapter/debug": ["debug@4.3.7", "", { "dependencies": { "ms": "^2.1.3" } }, "sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ=="],
-
    "@tailwindcss/node/jiti": ["jiti@2.6.1", "", { "bin": { "jiti": "lib/jiti-cli.mjs" } }, "sha512-ekilCSN1jwRvIbgeg/57YFh8qQDNbwDb9xT/qu2DAHbFFZUicIl4ygVaAvzveMhMVr3LnpSKTNnwt8PoOfmKhQ=="],

    "@tailwindcss/oxide-wasm32-wasi/@emnapi/core": ["@emnapi/core@1.8.1", "", { "dependencies": { "@emnapi/wasi-threads": "1.1.0", "tslib": "^2.4.0" }, "bundled": true }, "sha512-AvT9QFpxK0Zd8J0jopedNm+w/2fIzvtPKPjqyw9jwvBaReTTqPBk9Hixaz7KbjimP+QNz605/XnjFcDAL2pqBg=="],
--- a/helm/sim/templates/deployment-realtime.yaml
+++ b/helm/sim/templates/deployment-realtime.yaml
@@ -44,10 +44,6 @@ spec:
          env:
            - name: DATABASE_URL
              value: {{ include "sim.databaseUrl" . | quote }}
-            {{- if .Values.app.env.REDIS_URL }}
-            - name: REDIS_URL
-              value: {{ .Values.app.env.REDIS_URL | quote }}
-            {{- end }}
            {{- range $key, $value := .Values.realtime.env }}
            - name: {{ $key }}
              value: {{ $value | quote }}