Merge remote-tracking branch 'origin/staging' into feat/permission-aware-knowledge

Author: waleedlatif1

apps/sim/app/(landing)/integrations/data/landing-content.ts

@@ -38,6 +38,6 @@ export const INTEGRATION_LANDING_CONTENT: Record<string, IntegrationLandingConte
       href: '/privacy',
     },
     aiDisclaimer:
-      'Sim agents use AI models to generate messages and responses sent to Slack. AI-generated content can be inaccurate or incomplete, so review automated outputs before relying on them, especially for important communications.',
+      "A paid Slack plan is required to use Sim's AI agent in Slack's app container. Other Sim features continue to work on free Slack plans. Sim agents use AI models to generate messages and responses sent to Slack. AI-generated content can be inaccurate or incomplete, so review automated outputs before relying on them, especially for important communications.",
   },
 }

apps/sim/lib/admin/member-operation.test.ts

@@ -51,6 +51,11 @@ vi.mock('@/lib/workspaces/organization-workspaces', () => ({
   ownedAttachableWorkspacesWhere: vi.fn(() => undefined),
 }))
 vi.mock('@/lib/core/outbox/service', () => ({
+  continueOutboxHandler: (reason: string) => ({
+    outcome: 'deferred',
+    reason,
+    consumeAttempt: false,
+  }),
   deferOutboxHandler: (reason: string, _minimum?: number, consumeAttempt = true) => ({
     outcome: 'deferred',
     reason,

apps/sim/lib/admin/member-operation.ts

@@ -16,7 +16,7 @@ import {
 import { reconcileOrganizationSeats } from '@/lib/billing/organizations/seats'
 import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/utils'
 import {
-  deferOutboxHandler,
+  continueOutboxHandler,
   enqueueOutboxEvent,
   type OutboxHandler,
   outboxEventHasSourceOperationId,
@@ -688,7 +688,7 @@ export const processAdminMemberOperation: OutboxHandler<unknown> = async (rawPay
   }
 
   if (nextWorkspaceIndex < payload.request.workspaceIds.length) {
-    return deferOutboxHandler('Continuing bounded member workspace moves', undefined, false)
+    return continueOutboxHandler('Continuing bounded member workspace moves')
   }
 }
 

apps/sim/lib/billing/enterprise-provisioning.test.ts

@@ -61,6 +61,11 @@ vi.mock('@/lib/billing/webhooks/enterprise-reconciliation-lease', () => ({
   ),
 }))
 vi.mock('@/lib/core/outbox/service', () => ({
+  continueOutboxHandler: (reason: string) => ({
+    outcome: 'deferred',
+    reason,
+    consumeAttempt: false,
+  }),
   deferOutboxHandler: (reason: string, minimumBackoffMs?: number, consumeAttempt = true) => ({
     outcome: 'deferred',
     reason,

apps/sim/lib/billing/enterprise-provisioning.ts

@@ -79,6 +79,7 @@ import { withEnterpriseReconciliationLease } from '@/lib/billing/webhooks/enterp
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-handlers'
 import { env } from '@/lib/core/config/env'
 import {
+  continueOutboxHandler,
   deferOutboxHandler,
   enqueueOutboxEvent,
   type OutboxEventContext,
@@ -3108,7 +3109,7 @@ export const reconcileEnterpriseMembers: OutboxHandler<unknown> = async (rawPayl
 
   if (!nextCursor) return
   await context.checkpointPayload({ afterUserId: nextCursor })
-  return deferOutboxHandler('Continuing bounded Enterprise member reconciliation', undefined, false)
+  return continueOutboxHandler('Continuing bounded Enterprise member reconciliation')
 }
 
 export const enterpriseIssuanceOutboxHandlers = {

apps/sim/lib/core/outbox/service.test.ts

@@ -25,6 +25,7 @@ vi.mock('@sim/utils/id', () => ({
 }))
 
 import {
+  continueOutboxHandler,
   deferOutboxHandler,
   enqueueOrReschedulePendingOutboxEvent,
   enqueueOutboxEvent,
@@ -387,6 +388,20 @@ describe('processOutboxEvents — handler success and retry', () => {
     expect(deferredUpdate).toMatchObject({ attempts: 4, lastError: null, lockedAt: null })
   })
 
+  it('re-runs a continued handler without consuming its attempt budget', async () => {
+    const handler = vi.fn(async () => continueOutboxHandler('continuing bounded cleanup'))
+    queueTableRows(outboxEvent, [makePendingRow({ attempts: 4, maxAttempts: 5 })])
+    holdLease()
+
+    const result = await processOutboxEvents({ 'test.event': handler })
+
+    expect(result.retried).toBe(1)
+    const continuedUpdate = updateSets().find(
+      (set) => set.status === 'pending' && 'attempts' in set
+    )
+    expect(continuedUpdate).toMatchObject({ attempts: 4, lastError: null, lockedAt: null })
+  })
+
   it('dead-letters on failure when attempts reaches maxAttempts', async () => {
     const handler = vi.fn(async () => {
       throw new Error('permanent failure')

apps/sim/lib/core/outbox/service.ts

@@ -81,8 +81,9 @@ export interface DeferredOutboxHandlerResult {
   minimumBackoffMs?: number
   /**
    * Defaults to true for an external acknowledgement with a finite retry
-   * budget. Set false only for an internal dependency whose own outbox row
-   * independently reaches completed or dead-letter.
+   * budget. False is reserved for waits on an internal dependency whose own
+   * outbox row independently reaches completed or dead-letter, and for
+   * bounded continuation after durable progress (`continueOutboxHandler`).
    */
   consumeAttempt?: boolean
 }
@@ -100,6 +101,19 @@ export function deferOutboxHandler(
   }
 }
 
+/**
+ * Yields after durable progress so the worker re-runs the event without
+ * spending an attempt. For bounded batches whose remaining work shrinks on
+ * every run; a run that made no progress must throw or `deferOutboxHandler`
+ * instead, or the event never reaches a terminal state.
+ */
+export function continueOutboxHandler(
+  reason: string,
+  minimumBackoffMs?: number
+): DeferredOutboxHandlerResult {
+  return deferOutboxHandler(reason, minimumBackoffMs, false)
+}
+
 export type OutboxHandler<T = unknown> = (
   payload: T,
   context: OutboxEventContext

apps/sim/lib/webhooks/deploy.test.ts

@@ -1,9 +1,15 @@
 /**
  * @vitest-environment node
  */
-import { account, credential } from '@sim/db/schema'
-import { queueTableRows, resetDbChainMock, resetEnvFlagsMock, setEnvFlags } from '@sim/testing'
-import { eq } from 'drizzle-orm'
+import { account, credential, webhook, workflowDeploymentVersion } from '@sim/db/schema'
+import {
+  dbChainMockFns,
+  queueTableRows,
+  resetDbChainMock,
+  resetEnvFlagsMock,
+  setEnvFlags,
+} from '@sim/testing'
+import { eq, ne } from 'drizzle-orm'
 import { afterAll, beforeEach, describe, expect, it, type Mock, vi } from 'vitest'
 import type { SubBlockConfig } from '@/blocks/types'
 import type { BlockState } from '@/stores/workflows/workflow/types'
@@ -29,6 +35,14 @@ vi.mock('@/lib/webhooks/utils.server', () => ({
 vi.mock('@/lib/webhooks/pending-verification', () => ({
   PendingWebhookVerificationTracker: vi.fn(),
 }))
+const { mockIsDeploymentVersionActive, mockIsDeploymentVersionProtected } = vi.hoisted(() => ({
+  mockIsDeploymentVersionActive: vi.fn(),
+  mockIsDeploymentVersionProtected: vi.fn(),
+}))
+vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({
+  isDeploymentVersionActive: mockIsDeploymentVersionActive,
+  isDeploymentVersionProtectedByCurrentOperation: mockIsDeploymentVersionProtected,
+}))
 
 const {
   mockGetSlackBotCredential,
@@ -52,9 +66,11 @@ vi.mock('@/lib/webhooks/providers/slack', () => ({
 
 import {
   buildProviderConfig,
+  cleanupInactiveDeploymentWebhooks,
   resolveTriggerCredentialId,
   resolveWebhookConfigForBlock,
 } from '@/lib/webhooks/deploy'
+import { cleanupExternalWebhook } from '@/lib/webhooks/provider-subscriptions'
 import { getBlock } from '@/blocks'
 import { getTrigger } from '@/triggers'
 
@@ -639,3 +655,107 @@ describe('resolveWebhookConfigForBlock — TikTok routing', () => {
     expect(result?.error.message).toContain('Reconnect')
   })
 })
+
+describe('cleanupInactiveDeploymentWebhooks', () => {
+  const workflow = { id: 'workflow-1', userId: 'user-1', workspaceId: 'workspace-1' }
+  const input = {
+    workflowId: 'workflow-1',
+    workflow,
+    requestId: 'request-1',
+    protectedDeploymentVersionId: null,
+    limit: 5,
+  }
+
+  function staleWebhookRow(id: string) {
+    return {
+      id,
+      workflowId: 'workflow-1',
+      deploymentVersionId: 'version-1',
+      provider: 'github',
+      providerConfig: {},
+      archivedAt: null,
+      createdAt: new Date('2026-07-14T08:00:00.000Z'),
+    }
+  }
+
+  beforeEach(() => {
+    mockIsDeploymentVersionActive.mockResolvedValue(false)
+    mockIsDeploymentVersionProtected.mockResolvedValue(false)
+  })
+
+  it('retires one bounded batch of stale rows and reports the remainder', async () => {
+    queueTableRows(webhook, [
+      staleWebhookRow('wh-1'),
+      staleWebhookRow('wh-2'),
+      staleWebhookRow('wh-3'),
+    ])
+    queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }])
+    queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }])
+
+    await expect(cleanupInactiveDeploymentWebhooks({ ...input, limit: 2 })).resolves.toEqual({
+      hasMore: true,
+    })
+
+    expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledTimes(2)
+    expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledWith(
+      expect.objectContaining({ id: 'wh-1' }),
+      workflow,
+      'request-1',
+      { throwOnError: true }
+    )
+    expect(dbChainMockFns.delete).toHaveBeenCalledTimes(2)
+  })
+
+  it('reports completion once the batch drains every stale row', async () => {
+    queueTableRows(webhook, [staleWebhookRow('wh-1')])
+    queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }])
+
+    await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: false })
+
+    expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledTimes(1)
+    expect(dbChainMockFns.delete).toHaveBeenCalledTimes(1)
+  })
+
+  it('excludes the version the current operation is preparing from the batch', async () => {
+    queueTableRows(webhook, [])
+
+    await expect(
+      cleanupInactiveDeploymentWebhooks({ ...input, protectedDeploymentVersionId: 'version-3' })
+    ).resolves.toEqual({ hasMore: false })
+
+    expect(ne).toHaveBeenCalledWith(webhook.deploymentVersionId, 'version-3')
+  })
+
+  it('stops before any provider call once the fence reports a change', async () => {
+    queueTableRows(webhook, [staleWebhookRow('wh-1')])
+
+    await expect(
+      cleanupInactiveDeploymentWebhooks({ ...input, shouldContinue: async () => false })
+    ).resolves.toEqual({ hasMore: true })
+
+    expect(vi.mocked(cleanupExternalWebhook)).not.toHaveBeenCalled()
+    expect(dbChainMockFns.delete).not.toHaveBeenCalled()
+  })
+
+  it('leaves a row alone when its version was re-activated after the batch was selected', async () => {
+    queueTableRows(webhook, [staleWebhookRow('wh-1')])
+    mockIsDeploymentVersionActive.mockResolvedValue(true)
+
+    await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: true })
+
+    expect(mockIsDeploymentVersionActive).toHaveBeenCalledWith('workflow-1', 'version-1')
+    expect(vi.mocked(cleanupExternalWebhook)).not.toHaveBeenCalled()
+    expect(dbChainMockFns.delete).not.toHaveBeenCalled()
+  })
+
+  it('leaves a row alone when its version became the current candidate mid-batch', async () => {
+    queueTableRows(webhook, [staleWebhookRow('wh-1')])
+    mockIsDeploymentVersionProtected.mockResolvedValue(true)
+
+    await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: true })
+
+    expect(mockIsDeploymentVersionProtected).toHaveBeenCalledWith('workflow-1', 'version-1')
+    expect(vi.mocked(cleanupExternalWebhook)).not.toHaveBeenCalled()
+    expect(dbChainMockFns.delete).not.toHaveBeenCalled()
+  })
+})