fix(network): preserve transport and background execution behavior

Author: waleedlatif1

apps/docs/content/docs/platform/enterprise/security.mdx

@@ -25,6 +25,8 @@ Dedicated routing supports public IPv4 destinations over HTTPS on port 443. It a
 
 Native database connections, AWS SDK integrations, remote sandbox traffic, and browser requests keep their existing network paths. Other provider SDKs and OAuth authorization or token refresh calls require separate coverage confirmation. Dedicated IPs do not change access permissions in connected services.
 
+Google Drive, Fireflies, Google Workspace user and group discovery, and Atlassian OAuth site discovery currently use their existing network paths.
+
 If dedicated routing is unavailable, affected requests fail instead of using shared IPs. Organizations without dedicated routing keep their existing behavior.
 
 ## Availability

apps/sim/background/enrichment-capability-subject.test.ts

@@ -5,6 +5,10 @@ import { resetDbChainMock } from '@sim/testing'
 import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const mocks = vi.hoisted(() => ({
+  loadWorkspaceApplicationContext: vi.fn(),
+  resolveOutboundRoute: vi.fn(async (_organizationId: string | null | undefined) => ({
+    kind: 'direct',
+  })),
   getTableById: vi.fn(),
   getRowById: vi.fn(),
   updateRow: vi.fn(),
@@ -22,6 +26,14 @@ const mocks = vi.hoisted(() => ({
   loadTableRowSecretProvenance: vi.fn(async () => ({ scope: null, entries: [] })),
 }))
 
+vi.mock('@/lib/core/network/config.server', () => ({
+  isOutboundRoutingEnabled: () => true,
+  resolveOutboundRoute: mocks.resolveOutboundRoute,
+}))
+vi.mock('@/lib/workspaces/application/workspace-context', () => ({
+  loadWorkspaceApplicationContext: mocks.loadWorkspaceApplicationContext,
+}))
+
 vi.mock('@/lib/table/service', () => ({ getTableById: mocks.getTableById }))
 vi.mock('@/lib/table/rows/service', () => ({
   getRowById: mocks.getRowById,
@@ -69,6 +81,7 @@ vi.mock('@/lib/core/rate-limiter/rate-limiter', () => ({
   },
 }))
 
+import { resolveCurrentOutboundRoute } from '@/lib/core/network/context.server'
 import { runRowCascadeLoop } from '@/background/workflow-column-execution'
 
 const GROUP = {
@@ -143,6 +156,7 @@ describe('enrichment cell capability subject', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     resetDbChainMock()
+    mocks.loadWorkspaceApplicationContext.mockResolvedValue({ workspaceOrganizationId: null })
     mocks.getTableById.mockResolvedValue(TABLE)
     mocks.getRowById.mockResolvedValue({
       id: 'row-1',
@@ -162,6 +176,24 @@ describe('enrichment cell capability subject', () => {
     mocks.runEnrichment.mockResolvedValue({ result: {}, cost: 0, detail: {} })
   })
 
+  it.each(['org_reserved', 'org_other', null])(
+    'restores the current workspace owner %s before running a queued enrichment',
+    async (organizationId) => {
+      mocks.loadWorkspaceApplicationContext.mockResolvedValue({
+        workspaceOrganizationId: organizationId,
+      })
+      mocks.runEnrichment.mockImplementationOnce(async () => {
+        await resolveCurrentOutboundRoute()
+        return { result: {}, cost: 0, detail: {} }
+      })
+
+      await runRowCascadeLoop(payload(null, 'billing-owner') as never)
+
+      expect(mocks.resolveOutboundRoute).toHaveBeenCalledExactlyOnceWith(organizationId)
+      expect(mocks.loadWorkspaceApplicationContext).toHaveBeenCalledWith('workspace-1', {})
+    }
+  )
+
   /**
    * A workspace-key write is actorless: nobody's permission group governs it,
    * and the billing owner beside it on the payload is a bystander. Handing that

apps/sim/background/webhook-execution.test.ts

@@ -19,6 +19,7 @@ import {
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const {
+  mockWithResourceOutboundScope,
   mockResolveWebhookRecordProviderConfig,
   mockExecuteWorkflowCore,
   mockWasExecutionFinalizedByCore,
@@ -32,6 +33,7 @@ const {
 } = vi.hoisted(() => {
   const mockEnqueue = vi.fn()
   return {
+    mockWithResourceOutboundScope: vi.fn(),
     mockResolveWebhookRecordProviderConfig: vi.fn(),
     mockExecuteWorkflowCore: vi.fn(),
     mockWasExecutionFinalizedByCore: vi.fn(),
@@ -62,6 +64,10 @@ const mockGetExecutionEnvironment = environmentUtilsMockFns.mockGetExecutionEnvi
 
 afterAll(resetEnvironmentUtilsMock)
 
+vi.mock('@/lib/core/network/resource-scope.server', () => ({
+  withResourceOutboundScope: mockWithResourceOutboundScope,
+}))
+
 vi.mock('@/lib/execution/preprocessing', () => executionPreprocessingMock)
 vi.mock('@/lib/logs/execution/logging-session', () => loggingSessionMock)
 
@@ -285,6 +291,7 @@ describe('executeWebhookJob fault vs error handling', () => {
         projectDiagnosticError: loggingSessionMockFns.mockProjectDiagnosticError,
       }
     })
+    mockWithResourceOutboundScope.mockReset().mockImplementation((_owner, run) => run())
     mockRefreshExecutionSlotExpiry.mockReset().mockResolvedValue(true)
     mockReleaseExecutionSlot.mockReset().mockResolvedValue(undefined)
     mockGetProviderHandler.mockReturnValue({})
@@ -856,6 +863,19 @@ describe('executeWebhookJob fault vs error handling', () => {
     expect(mockExecuteWorkflowCore).not.toHaveBeenCalled()
   })
 
+  it('requeues transient organization ownership failures before any workflow block starts', async () => {
+    mockWithResourceOutboundScope.mockRejectedValueOnce(
+      Object.assign(new Error('Connection terminated unexpectedly'), { code: 'ECONNRESET' })
+    )
+
+    await expect(executeWebhookJob(payload)).resolves.toMatchObject({
+      requeued: true,
+    })
+    expect(mockExecuteWorkflowCore).not.toHaveBeenCalled()
+    expect(mockEnqueue).toHaveBeenCalledOnce()
+    expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
+  })
+
   it('requeues on retryable infrastructure errors thrown by setup reads', async () => {
     dbChainMockFns.limit.mockRejectedValueOnce(
       Object.assign(new Error('write CONNECT_TIMEOUT'), { code: 'CONNECT_TIMEOUT' })

apps/sim/background/webhook-execution.ts

@@ -818,21 +818,21 @@ async function executeWebhookJobInternal(
     throw new Error(`Workflow ${payload.workflowId} has no associated workspace`)
   }
 
-  return withResourceOutboundScope({ workspaceId }, async () => {
-    const workflowVariables = (workflowRecord.variables as Record<string, unknown>) || {}
+  const workflowVariables = (workflowRecord.variables as Record<string, unknown>) || {}
 
-    let deploymentVersionId: string | undefined
-    /**
-     * Flipped immediately before `executeWorkflowCore` is invoked. While false,
-     * no block has run and no execution effect exists, so a retryable
-     * infrastructure error may be surfaced as a `RetryableSetupError` and the
-     * whole delivery safely re-attempted. Once true, errors are never
-     * reclassified as retryable — retrying after the executor started could
-     * double-run the workflow.
-     */
-    let workflowCoreStarted = false
+  let deploymentVersionId: string | undefined
+  /**
+   * Flipped immediately before `executeWorkflowCore` is invoked. While false,
+   * no block has run and no execution effect exists, so a retryable
+   * infrastructure error may be surfaced as a `RetryableSetupError` and the
+   * whole delivery safely re-attempted. Once true, errors are never
+   * reclassified as retryable — retrying after the executor started could
+   * double-run the workflow.
+   */
+  let workflowCoreStarted = false
 
-    try {
+  try {
+    return await withResourceOutboundScope({ workspaceId }, async () => {
       const workflowStatePromise = payload.deploymentVersionId
         ? loadWorkflowDeploymentVersionState(
             payload.workflowId,
@@ -1175,105 +1175,105 @@ async function executeWebhookJobInternal(
         executedAt: new Date().toISOString(),
         provider: payload.provider,
       }
-    } catch (error: unknown) {
-      const errorMessage = toError(error).message
-      const errorStack = error instanceof Error ? error.stack : undefined
+    })
+  } catch (error: unknown) {
+    const errorMessage = toError(error).message
+    const errorStack = error instanceof Error ? error.stack : undefined
 
-      /**
-       * Mirrors the schedule executor's setup boundary: an infrastructure error
-       * raised before the workflow core started left no execution effect, so it
-       * is surfaced as a `RetryableSetupError` — releasing the idempotency claim
-       * and, while attempts remain, requeueing without recording a terminal
-       * failed row for an attempt that will be retried. Exhausted retries fall
-       * through to normal failure handling but still throw typed so a provider
-       * redelivery is not rejected for a run that never happened.
-       */
-      const retryableSetupCause =
-        !workflowCoreStarted && isRetryableInfrastructureError(error)
-          ? describeRetryableInfrastructureError(error)
-          : undefined
-      if (retryableSetupCause && hasRemainingWebhookInfraRetry(payload)) {
-        logger.warn(`[${requestId}] Retryable setup failure before webhook workflow started`, {
-          workflowId: payload.workflowId,
-          provider: payload.provider,
-          cause: retryableSetupCause,
-        })
-        throw new RetryableSetupError(errorMessage, { cause: retryableSetupCause })
-      }
+    /**
+     * Mirrors the schedule executor's setup boundary: an infrastructure error
+     * raised before the workflow core started left no execution effect, so it
+     * is surfaced as a `RetryableSetupError` — releasing the idempotency claim
+     * and, while attempts remain, requeueing without recording a terminal
+     * failed row for an attempt that will be retried. Exhausted retries fall
+     * through to normal failure handling but still throw typed so a provider
+     * redelivery is not rejected for a run that never happened.
+     */
+    const retryableSetupCause =
+      !workflowCoreStarted && isRetryableInfrastructureError(error)
+        ? describeRetryableInfrastructureError(error)
+        : undefined
+    if (retryableSetupCause && hasRemainingWebhookInfraRetry(payload)) {
+      logger.warn(`[${requestId}] Retryable setup failure before webhook workflow started`, {
+        workflowId: payload.workflowId,
+        provider: payload.provider,
+        cause: retryableSetupCause,
+      })
+      throw new RetryableSetupError(errorMessage, { cause: retryableSetupCause })
+    }
 
-      logger.error(
-        `[${requestId}] Webhook execution failed`,
-        loggingSession.projectDiagnosticError(error, {
-          workflowId: payload.workflowId,
-          provider: payload.provider,
-        })
-      )
+    logger.error(
+      `[${requestId}] Webhook execution failed`,
+      loggingSession.projectDiagnosticError(error, {
+        workflowId: payload.workflowId,
+        provider: payload.provider,
+      })
+    )
 
-      // The finalized flag is set inside a fire-and-forget post-execution promise; await it so the
-      // signal is reliable and the failure is fully persisted before we decide fault vs error.
-      await loggingSession.waitForPostExecution()
+    // The finalized flag is set inside a fire-and-forget post-execution promise; await it so the
+    // signal is reliable and the failure is fully persisted before we decide fault vs error.
+    await loggingSession.waitForPostExecution()
 
-      // A failure inside workflow execution (block error, provider 4xx, missing required field, etc.)
-      // is finalized by core and already recorded in the execution logs. That is a user/workflow error,
-      // not a trigger.dev job fault — complete the run normally so we don't fire a false alert. Errors
-      // that were not finalized came from the webhook pipeline itself, so we re-throw to fault below.
-      if (wasExecutionFinalizedByCore(error, executionId)) {
-        return {
-          success: false,
-          workflowId: payload.workflowId,
-          executionId,
-          output: hasExecutionResult(error) ? error.executionResult.output : {},
-          executedAt: new Date().toISOString(),
-          provider: payload.provider,
-        }
+    // A failure inside workflow execution (block error, provider 4xx, missing required field, etc.)
+    // is finalized by core and already recorded in the execution logs. That is a user/workflow error,
+    // not a trigger.dev job fault — complete the run normally so we don't fire a false alert. Errors
+    // that were not finalized came from the webhook pipeline itself, so we re-throw to fault below.
+    if (wasExecutionFinalizedByCore(error, executionId)) {
+      return {
+        success: false,
+        workflowId: payload.workflowId,
+        executionId,
+        output: hasExecutionResult(error) ? error.executionResult.output : {},
+        executedAt: new Date().toISOString(),
+        provider: payload.provider,
       }
+    }
 
-      try {
-        await loggingSession.safeStart({
-          userId: actorUserId,
-          actorUserId,
-          billingAttribution,
-          workspaceId,
-          variables: {},
-          triggerData: {
-            isTest: false,
-            correlation,
-          },
-          deploymentVersionId,
-        })
-
-        const executionResult = hasExecutionResult(error)
-          ? error.executionResult
-          : {
-              success: false,
-              output: {},
-              logs: [],
-            }
-        const { traceSpans } = buildTraceSpans(executionResult)
+    try {
+      await loggingSession.safeStart({
+        userId: actorUserId,
+        actorUserId,
+        billingAttribution,
+        workspaceId,
+        variables: {},
+        triggerData: {
+          isTest: false,
+          correlation,
+        },
+        deploymentVersionId,
+      })
 
-        await loggingSession.safeCompleteWithError({
-          endedAt: new Date().toISOString(),
-          totalDurationMs: 0,
-          error: {
-            message: errorMessage || 'Webhook execution failed',
-            stackTrace: errorStack,
-          },
-          traceSpans,
-          executionState: executionResult.executionState,
-        })
-      } catch (loggingError) {
-        logger.error(
-          `[${requestId}] Failed to complete logging session`,
-          loggingSession.projectDiagnosticError(loggingError)
-        )
-      }
+      const executionResult = hasExecutionResult(error)
+        ? error.executionResult
+        : {
+            success: false,
+            output: {},
+            logs: [],
+          }
+      const { traceSpans } = buildTraceSpans(executionResult)
+
+      await loggingSession.safeCompleteWithError({
+        endedAt: new Date().toISOString(),
+        totalDurationMs: 0,
+        error: {
+          message: errorMessage || 'Webhook execution failed',
+          stackTrace: errorStack,
+        },
+        traceSpans,
+        executionState: executionResult.executionState,
+      })
+    } catch (loggingError) {
+      logger.error(
+        `[${requestId}] Failed to complete logging session`,
+        loggingSession.projectDiagnosticError(loggingError)
+      )
+    }
 
-      if (retryableSetupCause) {
-        throw new RetryableSetupError(errorMessage, { cause: retryableSetupCause })
-      }
-      throw error
+    if (retryableSetupCause) {
+      throw new RetryableSetupError(errorMessage, { cause: retryableSetupCause })
     }
-  })
+    throw error
+  }
 }
 
 export const webhookExecution = task({