diff --git a/‎apps/sim/app/api/webhooks/trigger/[path]/route.test.ts‎
Lines changed: 91 additions & 0 deletions b/‎apps/sim/app/api/webhooks/trigger/[path]/route.test.ts‎
Lines changed: 91 additions & 0 deletions
diff --git a/‎apps/sim/app/api/webhooks/trigger/[path]/route.ts‎
Lines changed: 25 additions & 7 deletions b/‎apps/sim/app/api/webhooks/trigger/[path]/route.ts‎
Lines changed: 25 additions & 7 deletions
diff --git a/‎apps/sim/lib/billing/checkout-admission.test.ts‎
Lines changed: 7 additions & 21 deletions b/‎apps/sim/lib/billing/checkout-admission.test.ts‎
Lines changed: 7 additions & 21 deletions
diff --git a/‎apps/sim/lib/core/admission/rejection.ts‎
Lines changed: 27 additions & 0 deletions b/‎apps/sim/lib/core/admission/rejection.ts‎
Lines changed: 27 additions & 0 deletions
diff --git a/‎apps/sim/lib/execution/blocked-run-log.integration.ts‎
Lines changed: 92 additions & 0 deletions b/‎apps/sim/lib/execution/blocked-run-log.integration.ts‎
Lines changed: 92 additions & 0 deletions
diff --git a/‎apps/sim/lib/execution/blocked-run-log.ts‎
Lines changed: 43 additions & 0 deletions b/‎apps/sim/lib/execution/blocked-run-log.ts‎
Lines changed: 43 additions & 0 deletions
@@ -703,6 +703,97 @@ describe('Webhook Trigger API Route', () => {
     })
   })
 
+  it('does not let an acknowledged admission refusal mask another target that must retry', async () => {
+    testData.webhooks.push(
+      {
+        id: 'refused-webhook',
+        provider: 'generic',
+        path: 'fan-out-path',
+        isActive: true,
+        providerConfig: {},
+        workflowId: 'test-workflow-id',
+      },
+      {
+        id: 'failing-webhook',
+        provider: 'generic',
+        path: 'fan-out-path',
+        isActive: true,
+        providerConfig: {},
+        workflowId: 'test-workflow-id',
+      }
+    )
+    dispatchResolvedWebhookTargetMock
+      .mockResolvedValueOnce({
+        outcome: 'ignored',
+        reason: 'admission-rejected',
+        response: new NextResponse(null, { status: 200 }),
+      })
+      .mockResolvedValueOnce({
+        outcome: 'failed',
+        reason: 'preprocessing',
+        response: new NextResponse(null, { status: 503 }),
+      })
+
+    const response = await POST(
+      createMockRequest('POST', { event: 'x' }),
+      createRouteContext({ path: 'fan-out-path' })
+    )
+
+    expect(response.status).toBe(503)
+  })
+
+  it('answers with a failing target rather than a missing block so the sender retries', async () => {
+    testData.webhooks.push(
+      {
+        id: 'missing-block-webhook',
+        provider: 'generic',
+        path: 'mixed-path',
+        isActive: true,
+        providerConfig: {},
+        workflowId: 'test-workflow-id',
+      },
+      {
+        id: 'failing-webhook',
+        provider: 'generic',
+        path: 'mixed-path',
+        isActive: true,
+        providerConfig: {},
+        workflowId: 'test-workflow-id',
+      }
+    )
+    dispatchResolvedWebhookTargetMock
+      .mockResolvedValueOnce({
+        outcome: 'ignored',
+        reason: 'block-missing',
+        response: new NextResponse('Trigger block not found in deployment', {
+          status: 404,
+          headers: { 'x-slack-no-retry': '1' },
+        }),
+      })
+      .mockResolvedValueOnce({
+        outcome: 'failed',
+        reason: 'queue-failed',
+        response: new NextResponse(null, { status: 500 }),
+      })
+
+    const response = await POST(
+      createMockRequest('POST', { event: 'x' }),
+      createRouteContext({ path: 'mixed-path' })
+    )
+
+    expect(response.status).toBe(500)
+    expect(response.headers.get('x-slack-no-retry')).toBeNull()
+  })
+
+  it('tells Slack not to redeliver a POST to a path with no webhook', async () => {
+    const req = createMockRequest('POST', { type: 'event_callback' })
+
+    const response = await POST(req, createRouteContext({ path: 'deleted-path' }))
+
+    expect(response.status).toBe(404)
+    expect(response.headers.get('x-slack-no-retry')).toBe('1')
+  })
+
   describe('PUT, PATCH and DELETE deliveries', () => {
     /**
      * Every non-POST rejection is the same 405, whether the path is unknown, holds only
 
@@ -11,6 +11,7 @@ import { parseRequest } from '@/lib/api/server'
 import { admissionRejectedResponse, tryAdmit } from '@/lib/core/admission/gate'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { isDroppedDispatch } from '@/lib/webhooks/dispatch-result'
 import {
   dispatchResolvedWebhookTarget,
   findAllWebhooksForPath,
@@ -126,10 +127,13 @@ function methodNotAllowedResponse(): NextResponse {
  * existing callers see no change; anything else answers 405 uniformly, whether the path is
  * unknown, holds only non-path triggers, or holds a trigger that has not opted into the method —
  * so a probe cannot tell those apart.
+ *
+ * Every `POST` 404 carries `x-slack-no-retry`, which tells Slack not to redeliver an event to a
+ * deleted trigger. It is sent unconditionally, so it reveals nothing the 404 itself does not.
  */
 function notDeliverableResponse(method: string): NextResponse {
   return method === 'POST'
-    ? new NextResponse('Not Found', { status: 404 })
+    ? new NextResponse('Not Found', { status: 404, headers: { 'x-slack-no-retry': '1' } })
     : methodNotAllowedResponse()
 }
 
@@ -201,7 +205,7 @@ async function handleWebhookDelivery(
       return verificationResponse
     }
 
-    logger.warn(`[${requestId}] Webhook or workflow not found for path: ${path}`)
+    logger.debug(`[${requestId}] Webhook or workflow not found for path: ${path}`)
     return notDeliverableResponse(request.method)
   }
 
@@ -261,14 +265,16 @@ async function handleWebhookDelivery(
    */
   const responses: NextResponse[] = []
   const failures: NextResponse[] = []
-  let hasPermanentlyIgnoredLegacyTarget = false
+  /** Kept apart so a missing block's no-retry 404 never stands in for a target that must retry. */
+  const blockMissingResponses: NextResponse[] = []
+  let hasDroppedTarget = false
   for (const dispatchResult of legacySlackDispatchResults) {
     if (dispatchResult.outcome === 'failed') {
       failures.push(getSlackDispatchFailureResponse(dispatchResult))
       continue
     }
-    if (dispatchResult.reason === 'block-missing') {
-      hasPermanentlyIgnoredLegacyTarget = true
+    if (isDroppedDispatch(dispatchResult)) {
+      hasDroppedTarget = true
       continue
     }
     responses.push(dispatchResult.response)
@@ -333,13 +339,22 @@ async function handleWebhookDelivery(
       continue
     }
 
+    if (dispatchResult.reason === 'admission-rejected') {
+      hasDroppedTarget = true
+      continue
+    }
+
     if (dispatchResult.outcome === 'failed' || dispatchResult.reason === 'block-missing') {
       if (dispatchTargetCount > 1) {
         logger.warn(
           `[${requestId}] Webhook dispatch failed for ${foundWebhook.id}, continuing to next`,
           { reason: dispatchResult.reason, status: dispatchResult.response.status }
         )
-        failures.push(dispatchResult.response)
+        if (dispatchResult.outcome === 'failed') {
+          failures.push(dispatchResult.response)
+        } else {
+          blockMissingResponses.push(dispatchResult.response)
+        }
         continue
       }
       return dispatchResult.response
@@ -352,7 +367,10 @@ async function handleWebhookDelivery(
     if (failures.length > 0) {
       return failures[0]
     }
-    if (hasPermanentlyIgnoredLegacyTarget) {
+    if (blockMissingResponses.length > 0) {
+      return blockMissingResponses[0]
+    }
+    if (hasDroppedTarget) {
       return new NextResponse(null, { status: 200 })
     }
     return new NextResponse('No webhooks processed successfully', { status: 500 })
 
@@ -1,33 +1,19 @@
+import {
+  idempotencyServiceMock,
+  idempotencyServiceMockFns,
+} from '@sim/testing/mocks/idempotency-service.mock'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockAtomicallyClaim, mockRelease, mockIdempotencyService } = vi.hoisted(() => ({
-  mockAtomicallyClaim: vi.fn(),
-  mockRelease: vi.fn(),
-  mockIdempotencyService: vi.fn(),
-}))
-
-vi.mock('@/lib/core/idempotency/service', () => ({
-  IdempotencyService: class MockIdempotencyService {
-    constructor(options: unknown) {
-      mockIdempotencyService(options)
-    }
-
-    atomicallyClaim(...args: unknown[]) {
-      return mockAtomicallyClaim(...args)
-    }
-
-    release(...args: unknown[]) {
-      return mockRelease(...args)
-    }
-  },
-}))
+vi.mock('@/lib/core/idempotency/service', () => idempotencyServiceMock)
 
 import {
   claimCheckoutAdmission,
   releaseCheckoutAdmission,
   resolveCheckoutReferenceId,
 } from '@/lib/billing/checkout-admission'
 
+const { mockAtomicallyClaim, mockRelease } = idempotencyServiceMockFns
+
 describe('checkout admission', () => {
   beforeEach(() => {
     mockAtomicallyClaim.mockReset()
 
@@ -0,0 +1,27 @@
+/**
+ * Codes for admission refusals that hold until a person changes billing or
+ * account state, carried on the preprocessing error next to
+ * `WORKFLOW_NOT_DEPLOYED_CODE`. Resending the same delivery cannot succeed, so
+ * an unattended sender that retries on a non-2xx only loops.
+ *
+ * Reservation headroom denials are deliberately absent: they clear as in-flight
+ * runs settle, so a retry can succeed. So is a usage ledger that could not be
+ * read, which fails closed without saying anything about the payer.
+ */
+export const ADMISSION_REJECTION_CODE = {
+  USAGE_LIMIT_EXCEEDED: 'USAGE_LIMIT_EXCEEDED',
+  ACCOUNT_SUSPENDED: 'ACCOUNT_SUSPENDED',
+} as const
+
+const DETERMINISTIC_ADMISSION_REJECTION_CODES: ReadonlySet<string> = new Set(
+  Object.values(ADMISSION_REJECTION_CODE)
+)
+
+/** The failure's code when it is a deterministic admission rejection, else `undefined`. */
+export function getDeterministicAdmissionRejectionCode(failure: {
+  code?: string
+}): string | undefined {
+  return failure.code && DETERMINISTIC_ADMISSION_REJECTION_CODES.has(failure.code)
+    ? failure.code
+    : undefined
+}
@@ -0,0 +1,92 @@
+/**
+ * The blocked-run log claim against a real Redis: concurrent refusals from several app
+ * instances must agree on exactly one row per workflow, gate, and window. Skipped without
+ * `TEST_REDIS_URL`. Each test claims a fresh workflow id, so no test sees another's key.
+ */
+
+import { readTestRedisUrl } from '@sim/db/testing/test-infrastructure'
+import { redisConfigMock, redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock'
+import { generateId } from '@sim/utils/id'
+import Redis from 'ioredis'
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const redisUrl = readTestRedisUrl()
+
+vi.mock('@/lib/core/config/redis', () => redisConfigMock)
+
+import { BLOCKED_RUN_LOG_WINDOW_SECONDS, claimBlockedRunLog } from '@/lib/execution/blocked-run-log'
+
+describe.runIf(Boolean(redisUrl))('blocked-run log claim', () => {
+  let redis: Redis
+  const workflowIds: string[] = []
+
+  const freshWorkflowId = () => {
+    const workflowId = `workflow-${generateId()}`
+    workflowIds.push(workflowId)
+    return workflowId
+  }
+
+  beforeAll(async () => {
+    if (!redisUrl) throw new Error('TEST_REDIS_URL is required for this suite')
+    redis = new Redis(redisUrl, { lazyConnect: true, maxRetriesPerRequest: 0 })
+    await redis.connect()
+  })
+
+  beforeEach(() => {
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(redis)
+  })
+
+  afterAll(async () => {
+    const keys = await Promise.all(
+      workflowIds.map((workflowId) => redis.keys(`blocked-run-log:v1:${workflowId}:*`))
+    )
+    const flat = keys.flat()
+    if (flat.length > 0) await redis.del(...flat)
+    await redis.quit()
+  })
+
+  it('grants exactly one of many concurrent refusals the row', async () => {
+    const workflowId = freshWorkflowId()
+
+    const claims = await Promise.all(
+      Array.from({ length: 25 }, () => claimBlockedRunLog(workflowId, 'USAGE_LIMIT_EXCEEDED'))
+    )
+
+    expect(claims.filter(Boolean)).toHaveLength(1)
+  })
+
+  it('grants a different gate its own row in the same window', async () => {
+    const workflowId = freshWorkflowId()
+
+    expect(await claimBlockedRunLog(workflowId, 'USAGE_LIMIT_EXCEEDED')).toBe(true)
+    expect(await claimBlockedRunLog(workflowId, 'ACCOUNT_SUSPENDED')).toBe(true)
+    expect(await claimBlockedRunLog(workflowId, 'USAGE_LIMIT_EXCEEDED')).toBe(false)
+  })
+
+  it('expires the claim at the end of the window', async () => {
+    const workflowId = freshWorkflowId()
+    await claimBlockedRunLog(workflowId, 'USAGE_LIMIT_EXCEEDED')
+
+    const ttl = await redis.ttl(`blocked-run-log:v1:${workflowId}:USAGE_LIMIT_EXCEEDED`)
+    expect(ttl).toBeGreaterThan(BLOCKED_RUN_LOG_WINDOW_SECONDS - 5)
+    expect(ttl).toBeLessThanOrEqual(BLOCKED_RUN_LOG_WINDOW_SECONDS)
+
+    await redis.expire(`blocked-run-log:v1:${workflowId}:USAGE_LIMIT_EXCEEDED`, 1)
+    await vi.waitFor(
+      async () => expect(await claimBlockedRunLog(workflowId, 'USAGE_LIMIT_EXCEEDED')).toBe(true),
+      { timeout: 3000, interval: 200 }
+    )
+  })
+
+  it('records the row when Redis fails', async () => {
+    const broken = new Redis('redis://127.0.0.1:1', {
+      lazyConnect: true,
+      maxRetriesPerRequest: 0,
+      enableOfflineQueue: false,
+    })
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(broken)
+
+    expect(await claimBlockedRunLog(freshWorkflowId(), 'USAGE_LIMIT_EXCEEDED')).toBe(true)
+    broken.disconnect()
+  })
+})
@@ -0,0 +1,43 @@
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
+import { getRedisClient } from '@/lib/core/config/redis'
+
+const logger = createLogger('BlockedRunLog')
+
+/**
+ * How long one blocked-run log row stands for every later refusal of the same
+ * workflow by the same gate. A sender that retries a refused delivery would
+ * otherwise write a fresh execution row, trace archive, and file-ownership row
+ * per attempt; one row per window still tells the owner their runs are blocked.
+ */
+export const BLOCKED_RUN_LOG_WINDOW_SECONDS = 15 * 60
+
+/**
+ * Claims the right to record this window's blocked-run log row for a workflow
+ * and gate. Returns false when another refusal already recorded one. Without
+ * Redis, or when Redis fails, it returns true: a duplicate row is better than
+ * hiding that runs are blocked. Usage-limit refusals only occur on hosted
+ * billing deployments, which always run Redis.
+ */
+export async function claimBlockedRunLog(workflowId: string, gate: string): Promise<boolean> {
+  const redis = getRedisClient()
+  if (!redis) return true
+
+  try {
+    const claimed = await redis.set(
+      `blocked-run-log:v1:${workflowId}:${gate}`,
+      '1',
+      'EX',
+      BLOCKED_RUN_LOG_WINDOW_SECONDS,
+      'NX'
+    )
+    return claimed === 'OK'
+  } catch (error) {
+    logger.debug('Blocked-run log claim failed; recording the row', {
+      workflowId,
+      gate,
+      error: getErrorMessage(error),
+    })
+    return true
+  }
+}