diff --git a/apps/sim/app/api/emails/preview/route.ts b/apps/sim/app/api/emails/preview/route.ts
index c99e3ff9f6e..c4791882bb4 100644
--- a/apps/sim/app/api/emails/preview/route.ts
+++ b/apps/sim/app/api/emails/preview/route.ts
@@ -11,6 +11,8 @@ import {
   renderPasswordResetEmail,
   renderPaymentFailedEmail,
   renderPlanWelcomeEmail,
+  renderScheduleDisabledEmail,
+  renderUsageLimitReachedEmail,
   renderUsageThresholdEmail,
   renderWelcomeEmail,
   renderWorkspaceInvitationEmail,
@@ -93,6 +95,43 @@ const emailTemplates = {
       billingPortalUrl: 'https://sim.ai/settings/billing',
       failureReason: 'Card declined',
     }),
+  'usage-limit-reached': () =>
+    renderUsageLimitReachedEmail({
+      userName: 'John',
+      planName: 'Pro',
+      scope: 'user',
+      currentUsage: 20,
+      limit: 20,
+      ctaLink: 'https://sim.ai/settings/billing',
+    }),
+  'usage-limit-reached-org': () =>
+    renderUsageLimitReachedEmail({
+      userName: 'John',
+      planName: 'Team',
+      scope: 'organization',
+      currentUsage: 500,
+      limit: 500,
+      ctaLink: 'https://sim.ai/organization/org_123/settings/billing',
+    }),
+
+  // Operational notification emails
+  'schedule-disabled': () =>
+    renderScheduleDisabledEmail({
+      recipientName: 'John',
+      kind: 'workflow',
+      resourceName: 'Daily digest',
+      reason: 'consecutive_failures',
+      failedCount: 100,
+      manageLink: 'https://sim.ai/workspace/ws_123/w/wf_456',
+    }),
+  'schedule-disabled-auth': () =>
+    renderScheduleDisabledEmail({
+      recipientName: 'John',
+      kind: 'job',
+      resourceName: 'Weekly report',
+      reason: 'authentication_error',
+      manageLink: 'https://sim.ai/workspace/ws_123/scheduled-tasks',
+    }),
 } as const
 
 type EmailTemplate = keyof typeof emailTemplates
@@ -122,7 +161,10 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
         'plan-welcome-team',
         'credit-purchase',
         'payment-failed',
+        'usage-limit-reached',
+        'usage-limit-reached-org',
       ],
+      Notifications: ['schedule-disabled', 'schedule-disabled-auth'],
     }
 
     const categoryHtml = Object.entries(categories)
diff --git a/apps/sim/app/api/schedules/execute/route.test.ts b/apps/sim/app/api/schedules/execute/route.test.ts
index 4de1d5aaebf..d738fb326a7 100644
--- a/apps/sim/app/api/schedules/execute/route.test.ts
+++ b/apps/sim/app/api/schedules/execute/route.test.ts
@@ -30,6 +30,8 @@ const {
   mockShouldExecuteInline,
   mockResolveSystemBillingAttribution,
   mockAssertBillingAttributionSnapshot,
+  mockApplyScheduleFailureUpdate,
+  mockNotifyScheduleAutoDisabled,
 } = vi.hoisted(() => ({
   mockVerifyCronAuth: vi.fn().mockReturnValue(null),
   mockExecuteScheduleJob: vi.fn().mockResolvedValue(undefined),
@@ -44,6 +46,8 @@ const {
   mockShouldExecuteInline: vi.fn().mockReturnValue(false),
   mockResolveSystemBillingAttribution: vi.fn(),
   mockAssertBillingAttributionSnapshot: vi.fn(),
+  mockApplyScheduleFailureUpdate: vi.fn().mockResolvedValue({ updated: true, disabled: false }),
+  mockNotifyScheduleAutoDisabled: vi.fn().mockResolvedValue(undefined),
 }))
 
 vi.mock('@/lib/auth/internal', () => ({
@@ -59,15 +63,11 @@ vi.mock('@/background/schedule-execution', () => ({
   executeScheduleJob: mockExecuteScheduleJob,
   executeJobInline: mockExecuteJobInline,
   releaseScheduleLock: mockReleaseScheduleLock,
-  buildScheduleFailureUpdate: (now: Date, nextRunAt: Date | null) => ({
-    updatedAt: now,
-    lastQueuedAt: null,
-    nextRunAt,
-    failedCount: { type: 'sql' },
-    lastFailedAt: now,
-    status: { type: 'sql' },
-    infraRetryCount: 0,
-  }),
+  applyScheduleFailureUpdate: mockApplyScheduleFailureUpdate,
+}))
+
+vi.mock('@/lib/workflows/schedules/disable-notifications', () => ({
+  notifyScheduleAutoDisabled: mockNotifyScheduleAutoDisabled,
 }))
 
 vi.mock('@/lib/core/async-jobs', () => ({
@@ -730,11 +730,11 @@ describe('Scheduled Workflow Execution API Route', () => {
         error: expect.stringContaining('exhausted retry attempts'),
       })
     )
-    expect(dbChainMockFns.set).toHaveBeenCalledWith(
+    expect(mockApplyScheduleFailureUpdate).toHaveBeenCalledWith(
       expect.objectContaining({
-        lastQueuedAt: null,
-        lastFailedAt: expect.any(Date),
-        nextRunAt: expect.any(Date),
+        scheduleId: 'schedule-1',
+        expectedLastQueuedAt: claimedAt,
+        executor: expect.anything(),
       })
     )
   })
@@ -779,12 +779,11 @@ describe('Scheduled Workflow Execution API Route', () => {
 
     await runScheduleTick('test-request-id')
     expect(mockEnqueue).not.toHaveBeenCalled()
-    expect(dbChainMockFns.set).toHaveBeenCalledWith(
+    expect(mockApplyScheduleFailureUpdate).toHaveBeenCalledWith(
       expect.objectContaining({
-        lastQueuedAt: null,
-        lastFailedAt: expect.any(Date),
+        scheduleId: schedule.id,
+        expectedLastQueuedAt: claimedAt,
         nextRunAt: expect.any(Date),
-        infraRetryCount: 0,
       })
     )
     expect(dbChainMockFns.set).not.toHaveBeenCalledWith(
@@ -794,6 +793,44 @@ describe('Scheduled Workflow Execution API Route', () => {
     )
   })
 
+  it('emails the schedule owners when a non-retryable setup failure disables the schedule', async () => {
+    const claimedAt = new Date('2025-01-01T00:00:00.000Z')
+    const schedule = {
+      ...SINGLE_SCHEDULE[0],
+      lastQueuedAt: claimedAt,
+    }
+    mockGetJob.mockRejectedValueOnce(new Error('bad setup invariant'))
+    mockApplyScheduleFailureUpdate.mockResolvedValueOnce({ updated: true, disabled: true })
+    dbChainMockFns.limit
+      .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS)
+      .mockResolvedValueOnce([])
+    dbChainMockFns.returning.mockReturnValueOnce([schedule]).mockReturnValueOnce([])
+
+    await runScheduleTick('test-request-id')
+
+    expect(mockNotifyScheduleAutoDisabled).toHaveBeenCalledWith(
+      expect.objectContaining({ scheduleId: schedule.id, reason: 'consecutive_failures' })
+    )
+  })
+
+  it('does not email when the failure update leaves the schedule active', async () => {
+    const claimedAt = new Date('2025-01-01T00:00:00.000Z')
+    const schedule = {
+      ...SINGLE_SCHEDULE[0],
+      lastQueuedAt: claimedAt,
+    }
+    mockGetJob.mockRejectedValueOnce(new Error('bad setup invariant'))
+    mockApplyScheduleFailureUpdate.mockResolvedValueOnce({ updated: true, disabled: false })
+    dbChainMockFns.limit
+      .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS)
+      .mockResolvedValueOnce([])
+    dbChainMockFns.returning.mockReturnValueOnce([schedule]).mockReturnValueOnce([])
+
+    await runScheduleTick('test-request-id')
+
+    expect(mockNotifyScheduleAutoDisabled).not.toHaveBeenCalled()
+  })
+
   it('uses one backend mode decision for slot accounting and schedule processing', async () => {
     mockShouldExecuteInline.mockReturnValue(true)
     dbChainMockFns.limit
diff --git a/apps/sim/app/api/schedules/execute/route.ts b/apps/sim/app/api/schedules/execute/route.ts
index 13f87274810..debf5680af0 100644
--- a/apps/sim/app/api/schedules/execute/route.ts
+++ b/apps/sim/app/api/schedules/execute/route.ts
@@ -22,6 +22,7 @@ import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 import { runDetached } from '@/lib/core/utils/background'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { notifyScheduleAutoDisabled } from '@/lib/workflows/schedules/disable-notifications'
 import {
   SCHEDULE_EXECUTION_CONCURRENCY_LIMIT,
   SCHEDULE_EXECUTION_QUEUE_NAME,
@@ -31,7 +32,7 @@ import {
 } from '@/lib/workflows/schedules/execution-limits'
 import { calculateScheduleInfraRetryDelayMs } from '@/lib/workflows/schedules/retry'
 import {
-  buildScheduleFailureUpdate,
+  applyScheduleFailureUpdate,
   executeJobInline,
   executeScheduleJob,
   releaseScheduleLock,
@@ -43,6 +44,12 @@ export const maxDuration = 3600
 
 const logger = createLogger('ScheduledExecuteAPI')
 const WORKFLOW_CHUNK_SIZE = 100
+/**
+ * Recovery sweeps a batch of up to `STALE_SCHEDULE_RECOVERY_BATCH_SIZE` schedules,
+ * each fanning out to every workspace admin. Cap the mail so one tick can't turn
+ * into hundreds of inline sends; the remainder is logged.
+ */
+const STALE_SCHEDULE_RECOVERY_NOTIFY_LIMIT = 25
 const JOB_CHUNK_SIZE = 100
 const MAX_TICK_DURATION_MS = 3 * 60 * 1000
 const STALE_SCHEDULE_CLAIM_MS = getMaxExecutionTimeout()
@@ -394,20 +401,22 @@ async function markClaimedScheduleFailed(
   context: string
 ): Promise<void> {
   const now = new Date()
-  await db
-    .update(workflowSchedule)
-    .set(buildScheduleFailureUpdate(now, getScheduleNextRunAt(schedule, now)))
-    .where(
-      and(
-        eq(workflowSchedule.id, schedule.id),
-        isNull(workflowSchedule.archivedAt),
-        eq(workflowSchedule.lastQueuedAt, expectedLastQueuedAt)
-      )
-    )
-    .catch((error) => {
-      logger.error(`[${requestId}] ${context}`, error)
-      throw error
+  const { disabled } = await applyScheduleFailureUpdate({
+    scheduleId: schedule.id,
+    now,
+    nextRunAt: getScheduleNextRunAt(schedule, now),
+    expectedLastQueuedAt,
+    requestId,
+    context,
+  })
+
+  if (disabled) {
+    await notifyScheduleAutoDisabled({
+      scheduleId: schedule.id,
+      reason: 'consecutive_failures',
+      requestId,
     })
+  }
 }
 
 async function deferClaimedScheduleAfterQueueFailure(
@@ -491,6 +500,13 @@ async function handleClaimedScheduleSetupFailure(
 async function recoverStaleDatabaseScheduleJobs(now: Date): Promise<void> {
   const staleStartedBefore = getStaleScheduleExecutionCutoff(now)
 
+  /**
+   * Collected inside the transaction, flushed after it commits. Emailing inside
+   * would both notify about writes a rollback discards and issue pooled-client
+   * reads while the transaction still holds row locks under the advisory lock.
+   */
+  const disabledScheduleIds: string[] = []
+
   await db.transaction(async (tx) => {
     const [lock] = await tx.execute<{ acquired: boolean }>(
       sql`SELECT pg_try_advisory_xact_lock(hashtextextended(${SCHEDULE_EXECUTION_QUEUE_NAME}, 0)) AS acquired`
@@ -539,16 +555,17 @@ async function recoverStaleDatabaseScheduleJobs(now: Date): Promise<void> {
       const claimedAt = getSchedulePayloadClaimedAt(payload)
       if (!payload || !claimedAt) continue
 
-      await tx
-        .update(workflowSchedule)
-        .set(buildScheduleFailureUpdate(now, getScheduleNextRunAt(payload, now)))
-        .where(
-          and(
-            eq(workflowSchedule.id, payload.scheduleId),
-            isNull(workflowSchedule.archivedAt),
-            eq(workflowSchedule.lastQueuedAt, claimedAt)
-          )
-        )
+      const { disabled } = await applyScheduleFailureUpdate({
+        scheduleId: payload.scheduleId,
+        now,
+        nextRunAt: getScheduleNextRunAt(payload, now),
+        expectedLastQueuedAt: claimedAt,
+        requestId: 'stale-schedule-recovery',
+        context: `Error updating schedule ${payload.scheduleId} after stale lease recovery`,
+        executor: tx,
+      })
+
+      if (disabled) disabledScheduleIds.push(payload.scheduleId)
     }
 
     if (retryableRows.length > 0) {
@@ -568,6 +585,22 @@ async function recoverStaleDatabaseScheduleJobs(now: Date): Promise<void> {
         )
     }
   })
+
+  const notifiable = disabledScheduleIds.slice(0, STALE_SCHEDULE_RECOVERY_NOTIFY_LIMIT)
+  if (disabledScheduleIds.length > notifiable.length) {
+    logger.warn('Capped schedule auto-disable notifications for stale recovery batch', {
+      disabled: disabledScheduleIds.length,
+      notified: notifiable.length,
+    })
+  }
+
+  for (const scheduleId of notifiable) {
+    await notifyScheduleAutoDisabled({
+      scheduleId,
+      reason: 'consecutive_failures',
+      requestId: 'stale-schedule-recovery',
+    })
+  }
 }
 
 function isStaleDatabaseScheduleJob(job: { status: string; startedAt?: Date }): boolean {
diff --git a/apps/sim/background/schedule-execution.test.ts b/apps/sim/background/schedule-execution.test.ts
new file mode 100644
index 00000000000..8ebb5cdd8c4
--- /dev/null
+++ b/apps/sim/background/schedule-execution.test.ts
@@ -0,0 +1,101 @@
+/**
+ * Covers the `active -> disabled` edge detection that drives the auto-disable
+ * email. `applyScheduleFailureUpdate` is the exported seam over the shared
+ * guarded UPDATE, so asserting on it also pins the behavior of the trigger.dev
+ * call sites that go through the same helper.
+ *
+ * @vitest-environment node
+ */
+import { databaseMock, dbChainMockFns, resetDbChainMock, schemaMock } from '@sim/testing'
+import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { notifyMock } = vi.hoisted(() => ({
+  notifyMock: vi.fn(() => Promise.resolve()),
+}))
+
+vi.mock('@/lib/workflows/schedules/disable-notifications', () => ({
+  notifyScheduleAutoDisabled: notifyMock,
+}))
+
+// This module imports its tables from `@sim/db` directly, which the global mock
+// does not re-export. Widen it rather than rewriting the source's imports.
+vi.mock('@sim/db', () => ({ ...databaseMock, ...schemaMock }))
+
+import { applyScheduleFailureUpdate, releaseScheduleLock } from '@/background/schedule-execution'
+
+const BASE = {
+  scheduleId: 'schedule-1',
+  now: new Date('2025-01-01T00:00:00.000Z'),
+  nextRunAt: new Date('2025-01-01T01:00:00.000Z'),
+  expectedLastQueuedAt: new Date('2024-12-31T23:00:00.000Z'),
+  requestId: 'req-1',
+  context: 'test context',
+}
+
+afterAll(() => {
+  resetDbChainMock()
+})
+
+describe('applyScheduleFailureUpdate', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('reports disabled when the write returns a disabled row', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1', status: 'disabled' }])
+
+    const result = await applyScheduleFailureUpdate(BASE)
+
+    expect(result).toEqual({ updated: true, disabled: true })
+  })
+
+  it('reports not disabled below the threshold (the 99-of-100 case)', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1', status: 'active' }])
+
+    const result = await applyScheduleFailureUpdate(BASE)
+
+    expect(result).toEqual({ updated: true, disabled: false })
+  })
+
+  it('reports neither updated nor disabled when the claim guard matches no row', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([])
+
+    const result = await applyScheduleFailureUpdate(BASE)
+
+    expect(result).toEqual({ updated: false, disabled: false })
+  })
+
+  it('defers the email so an in-transaction caller can send it after commit', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1', status: 'disabled' }])
+
+    await applyScheduleFailureUpdate(BASE)
+
+    expect(notifyMock).not.toHaveBeenCalled()
+  })
+
+  it('excludes already-disabled rows so a disable is reported at most once', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1', status: 'disabled' }])
+
+    await applyScheduleFailureUpdate(BASE)
+
+    const condition = dbChainMockFns.where.mock.calls.at(-1)?.[0]
+    expect(JSON.stringify(condition)).toContain('disabled')
+  })
+})
+
+describe('releaseScheduleLock', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('never notifies, even when the row it releases is already disabled', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1', status: 'disabled' }])
+
+    const released = await releaseScheduleLock('schedule-1', 'req-1', new Date(), 'release')
+
+    expect(released).toBe(true)
+    expect(notifyMock).not.toHaveBeenCalled()
+  })
+})
diff --git a/apps/sim/background/schedule-execution.ts b/apps/sim/background/schedule-execution.ts
index da02f0ebfaa..7195b6ad34e 100644
--- a/apps/sim/background/schedule-execution.ts
+++ b/apps/sim/background/schedule-execution.ts
@@ -11,7 +11,7 @@ import { describeError, toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { task } from '@trigger.dev/sdk'
 import { Cron } from 'croner'
-import { and, eq, isNull, type SQL, sql } from 'drizzle-orm'
+import { and, eq, isNull, ne, type SQL, sql } from 'drizzle-orm'
 import {
   assertBillingAttributionSnapshot,
   BILLING_ATTRIBUTION_HEADER,
@@ -30,6 +30,7 @@ import {
   getExecutionTimeout,
   getTimeoutErrorMessage,
 } from '@/lib/core/execution-limits'
+import type { DbOrTx } from '@/lib/db/types'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
@@ -40,6 +41,8 @@ import {
 } from '@/lib/workflows/executor/execution-core'
 import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
 import { loadDeployedWorkflowState } from '@/lib/workflows/persistence/utils'
+import { notifyScheduleAutoDisabled } from '@/lib/workflows/schedules/disable-notifications'
+import type { ScheduleDisableReason } from '@/lib/workflows/schedules/disable-reasons'
 import {
   SCHEDULE_EXECUTION_CONCURRENCY_LIMIT,
   SCHEDULE_EXECUTION_QUEUE_NAME,
@@ -70,6 +73,12 @@ type WorkflowScheduleUpdate = Partial<Omit<WorkflowScheduleInsert, 'failedCount'
 }
 type ExecutionCoreResult = Awaited<ReturnType<typeof executeWorkflowCore>>
 
+/** Result of a guarded schedule UPDATE. `status` is the row's value after the write. */
+type ScheduleUpdateOutcome = {
+  updated: boolean
+  status: string | null
+}
+
 function incrementScheduleFailedCount(): SQL {
   return sql`COALESCE(${workflowSchedule.failedCount}, 0) + 1`
 }
@@ -141,8 +150,22 @@ async function applyScheduleUpdate(
   updates: WorkflowScheduleUpdate,
   requestId: string,
   context: string,
-  options: { expectedLastQueuedAt?: Date | null } = {}
-): Promise<boolean> {
+  options: {
+    expectedLastQueuedAt?: Date | null
+    /**
+     * Set at call sites that can transition the row to `disabled`. Presence both
+     * opts the site into the auto-disable email and adds a `status <> 'disabled'`
+     * guard, so the transition fires exactly once per disable.
+     */
+    disableReason?: ScheduleDisableReason
+    /** Required inside a transaction, where mail must wait for commit. */
+    deferNotification?: boolean
+    /** Join a caller's transaction instead of using the pooled client. */
+    executor?: DbOrTx
+  } = {}
+): Promise<ScheduleUpdateOutcome> {
+  let outcome: ScheduleUpdateOutcome
+
   try {
     const claimGuard =
       options.expectedLastQueuedAt === undefined
@@ -151,19 +174,46 @@ async function applyScheduleUpdate(
           ? isNull(workflowSchedule.lastQueuedAt)
           : eq(workflowSchedule.lastQueuedAt, options.expectedLastQueuedAt)
 
-    const updatedRows = await db
+    /**
+     * `RETURNING` yields the NEW row, so `status === 'disabled'` alone only means
+     * "is disabled". Excluding rows that were already disabled makes a returned
+     * row a true `active -> disabled` edge. Scoped to disable-capable call sites
+     * so lock releases on already-disabled rows still work.
+     */
+    const notAlreadyDisabled = options.disableReason
+      ? ne(workflowSchedule.status, 'disabled')
+      : undefined
+
+    const updatedRows = await (options.executor ?? db)
       .update(workflowSchedule)
       .set(updates)
       .where(
-        and(eq(workflowSchedule.id, scheduleId), isNull(workflowSchedule.archivedAt), claimGuard)
+        and(
+          eq(workflowSchedule.id, scheduleId),
+          isNull(workflowSchedule.archivedAt),
+          claimGuard,
+          notAlreadyDisabled
+        )
       )
-      .returning({ id: workflowSchedule.id })
+      .returning({ id: workflowSchedule.id, status: workflowSchedule.status })
 
-    return updatedRows.length > 0
+    const row = updatedRows[0]
+    outcome = { updated: Boolean(row), status: row?.status ?? null }
   } catch (error) {
     logger.error(`[${requestId}] ${context}`, error, { cause: describeError(error) })
     throw error
   }
+
+  // Outside the try: a mail failure must never surface as a schedule-tick fault.
+  if (options.disableReason && !options.deferNotification && outcome.status === 'disabled') {
+    await notifyScheduleAutoDisabled({
+      scheduleId,
+      reason: options.disableReason,
+      requestId,
+    })
+  }
+
+  return outcome
 }
 
 export async function releaseScheduleLock(
@@ -183,7 +233,41 @@ export async function releaseScheduleLock(
     updates.nextRunAt = nextRunAt
   }
 
-  return applyScheduleUpdate(scheduleId, updates, requestId, context, options)
+  const outcome = await applyScheduleUpdate(scheduleId, updates, requestId, context, options)
+  return outcome.updated
+}
+
+/**
+ * Applies {@link buildScheduleFailureUpdate} through the same guarded write the
+ * trigger.dev path uses, and reports whether the row just transitioned to
+ * `disabled`. Callers own the notification so an in-transaction caller can defer
+ * it until after commit.
+ */
+export async function applyScheduleFailureUpdate(params: {
+  scheduleId: string
+  now: Date
+  nextRunAt: Date | null
+  expectedLastQueuedAt: Date
+  requestId: string
+  context: string
+  executor?: DbOrTx
+}): Promise<{ updated: boolean; disabled: boolean }> {
+  const { scheduleId, now, nextRunAt, expectedLastQueuedAt, requestId, context, executor } = params
+
+  const outcome = await applyScheduleUpdate(
+    scheduleId,
+    buildScheduleFailureUpdate(now, nextRunAt),
+    requestId,
+    context,
+    {
+      expectedLastQueuedAt,
+      disableReason: 'consecutive_failures',
+      deferNotification: true,
+      executor,
+    }
+  )
+
+  return { updated: outcome.updated, disabled: outcome.status === 'disabled' }
 }
 
 function getScheduleClaimedAt(payload: ScheduleExecutionPayload): Date | null {
@@ -223,7 +307,7 @@ async function retryScheduleAfterInfraFailure({
       buildScheduleFailureUpdate(now, nextRunAt),
       requestId,
       `Error updating schedule ${payload.scheduleId} after exhausted infrastructure retries`,
-      { expectedLastQueuedAt: claimedAt }
+      { expectedLastQueuedAt: claimedAt, disableReason: 'consecutive_failures' }
     )
     return
   }
@@ -648,10 +732,12 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
 
     const updateClaimedSchedule = (
       updates: WorkflowScheduleUpdate,
-      context: string
-    ): Promise<boolean> =>
+      context: string,
+      disableReason?: ScheduleDisableReason
+    ): Promise<ScheduleUpdateOutcome> =>
       applyScheduleUpdate(payload.scheduleId, updates, requestId, context, {
         expectedLastQueuedAt: claimedAt,
+        disableReason,
       })
 
     try {
@@ -777,7 +863,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
                 status: 'disabled',
                 ...resetScheduleInfraRetryCount(),
               },
-              `Failed to disable schedule ${payload.scheduleId} after authentication error`
+              `Failed to disable schedule ${payload.scheduleId} after authentication error`,
+              'authentication_error'
             )
             return
           }
@@ -794,7 +881,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
                 status: 'disabled',
                 ...resetScheduleInfraRetryCount(),
               },
-              `Failed to disable schedule ${payload.scheduleId} after authorization error`
+              `Failed to disable schedule ${payload.scheduleId} after authorization error`,
+              'authorization_error'
             )
             return
           }
@@ -808,7 +896,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
                 status: 'disabled',
                 ...resetScheduleInfraRetryCount(),
               },
-              `Failed to disable schedule ${payload.scheduleId} after missing workflow`
+              `Failed to disable schedule ${payload.scheduleId} after missing workflow`,
+              'workflow_not_found'
             )
             return
           }
@@ -847,7 +936,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
             })
             await updateClaimedSchedule(
               buildScheduleFailureUpdate(now, nextRunAt),
-              `Error updating schedule ${payload.scheduleId} after usage limit check`
+              `Error updating schedule ${payload.scheduleId} after usage limit check`,
+              'consecutive_failures'
             )
             return
           }
@@ -869,7 +959,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
 
             await updateClaimedSchedule(
               buildScheduleFailureUpdate(now, nextRunAt),
-              `Error updating schedule ${payload.scheduleId} after preprocessing failure`
+              `Error updating schedule ${payload.scheduleId} after preprocessing failure`,
+              'consecutive_failures'
             )
             return
           }
@@ -937,7 +1028,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
               nextRunAt: null,
               ...resetScheduleInfraRetryCount(),
             },
-            `Failed to disable schedule ${payload.scheduleId} after skip`
+            `Failed to disable schedule ${payload.scheduleId} after skip`,
+            'invalid_schedule'
           )
           return
         }
@@ -967,7 +1059,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
 
         await updateClaimedSchedule(
           buildScheduleFailureUpdate(now, nextRunAt),
-          `Error updating schedule ${payload.scheduleId} after failure`
+          `Error updating schedule ${payload.scheduleId} after failure`,
+          'consecutive_failures'
         )
       } catch (error: unknown) {
         logger.error(
@@ -979,7 +1072,8 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
 
         await updateClaimedSchedule(
           buildScheduleFailureUpdate(now, nextRunAt),
-          `Error updating schedule ${payload.scheduleId} after execution error`
+          `Error updating schedule ${payload.scheduleId} after execution error`,
+          'consecutive_failures'
         )
       }
     } catch (error: unknown) {
@@ -1450,7 +1544,10 @@ export async function executeJobInline(payload: JobExecutionPayload) {
       },
       requestId,
       `Error updating job ${payload.scheduleId} after failure`,
-      { expectedLastQueuedAt: now }
+      {
+        expectedLastQueuedAt: now,
+        disableReason: shouldDisable ? 'consecutive_failures' : undefined,
+      }
     )
   }
 }
diff --git a/apps/sim/components/emails/billing/index.ts b/apps/sim/components/emails/billing/index.ts
index 9c8faaca3b0..eab1c008435 100644
--- a/apps/sim/components/emails/billing/index.ts
+++ b/apps/sim/components/emails/billing/index.ts
@@ -6,4 +6,5 @@ export { FreeTierUpgradeEmail } from './free-tier-upgrade-email'
 export { LimitThresholdEmail } from './limit-threshold-email'
 export { PaymentFailedEmail } from './payment-failed-email'
 export { PlanWelcomeEmail } from './plan-welcome-email'
+export { UsageLimitReachedEmail } from './usage-limit-reached-email'
 export { UsageThresholdEmail } from './usage-threshold-email'
diff --git a/apps/sim/components/emails/billing/usage-limit-reached-email.tsx b/apps/sim/components/emails/billing/usage-limit-reached-email.tsx
new file mode 100644
index 00000000000..8ff73ab52d9
--- /dev/null
+++ b/apps/sim/components/emails/billing/usage-limit-reached-email.tsx
@@ -0,0 +1,81 @@
+import { Link, Section, Text } from '@react-email/components'
+import { baseStyles } from '@/components/emails/_styles'
+import { EmailLayout } from '@/components/emails/components'
+import { dollarsToCredits } from '@/lib/billing/credits/conversion'
+import { getBrandConfig } from '@/ee/whitelabeling'
+
+interface UsageLimitReachedEmailProps {
+  userName?: string
+  planName: string
+  /** Drives the remedy copy and CTA label — org admins raise the org limit. */
+  scope: 'user' | 'organization'
+  currentUsage: number
+  limit: number
+  ctaLink: string
+}
+
+/**
+ * Sent at 100% of the usage limit to paid and organization accounts. The
+ * free-tier equivalent is `CreditsExhaustedEmail`, whose remedy is an upgrade
+ * rather than a limit change.
+ */
+export function UsageLimitReachedEmail({
+  userName,
+  planName,
+  scope,
+  currentUsage,
+  limit,
+  ctaLink,
+}: UsageLimitReachedEmailProps) {
+  const brand = getBrandConfig()
+  const isOrganization = scope === 'organization'
+  const previewText = `${brand.name}: You've reached your ${planName} usage limit`
+
+  return (
+    <EmailLayout preview={previewText} showUnsubscribe={true}>
+      <Text style={{ ...baseStyles.paragraph, marginTop: 0 }}>
+        {userName ? `Hi ${userName},` : 'Hi,'}
+      </Text>
+
+      <Text style={baseStyles.paragraph}>
+        {isOrganization
+          ? `Your organization has reached its monthly usage limit on the ${planName} plan.`
+          : `You've reached your monthly usage limit on the ${planName} plan.`}{' '}
+        Workflow runs, knowledge base search, document uploads, and Chat are paused until the limit
+        goes up.
+      </Text>
+
+      <Section style={baseStyles.infoBox}>
+        <Text style={baseStyles.infoBoxTitle}>Usage</Text>
+        <Text style={baseStyles.infoBoxList}>
+          {dollarsToCredits(currentUsage).toLocaleString()} of{' '}
+          {dollarsToCredits(limit).toLocaleString()} credits used
+        </Text>
+      </Section>
+
+      {/* Divider */}
+      <div style={baseStyles.divider} />
+
+      <Text style={baseStyles.paragraph}>
+        {isOrganization
+          ? 'Raise the organization usage limit in billing settings to resume.'
+          : 'Raise your usage limit in billing settings, or upgrade your plan, to resume.'}
+      </Text>
+
+      <Link href={ctaLink} style={{ textDecoration: 'none' }}>
+        <Text style={baseStyles.button}>
+          {isOrganization ? 'Raise Organization Limit' : 'Raise Usage Limit'}
+        </Text>
+      </Link>
+
+      {/* Divider */}
+      <div style={baseStyles.divider} />
+
+      <Text style={{ ...baseStyles.footerText, textAlign: 'left' }}>
+        Sent to the people who manage billing for this account.
+      </Text>
+    </EmailLayout>
+  )
+}
+
+export default UsageLimitReachedEmail
diff --git a/apps/sim/components/emails/index.ts b/apps/sim/components/emails/index.ts
index ffbf4d11f22..a3c8bc50a93 100644
--- a/apps/sim/components/emails/index.ts
+++ b/apps/sim/components/emails/index.ts
@@ -8,6 +8,8 @@ export * from './billing'
 export * from './components'
 // Invitation emails
 export * from './invitations'
+// Operational notification emails
+export * from './notifications'
 // Render functions and subjects
 export * from './render'
 export * from './subjects'
diff --git a/apps/sim/components/emails/notifications/index.ts b/apps/sim/components/emails/notifications/index.ts
new file mode 100644
index 00000000000..c930a723ce9
--- /dev/null
+++ b/apps/sim/components/emails/notifications/index.ts
@@ -0,0 +1 @@
+export { ScheduleDisabledEmail } from './schedule-disabled-email'
diff --git a/apps/sim/components/emails/notifications/schedule-disabled-email.tsx b/apps/sim/components/emails/notifications/schedule-disabled-email.tsx
new file mode 100644
index 00000000000..a5116be4246
--- /dev/null
+++ b/apps/sim/components/emails/notifications/schedule-disabled-email.tsx
@@ -0,0 +1,83 @@
+import { Link, Section, Text } from '@react-email/components'
+import { baseStyles } from '@/components/emails/_styles'
+import { EmailLayout } from '@/components/emails/components'
+import {
+  SCHEDULE_DISABLE_REASON_COPY,
+  type ScheduleDisableReason,
+} from '@/lib/workflows/schedules/disable-reasons'
+import { getBrandConfig } from '@/ee/whitelabeling'
+
+interface ScheduleDisabledEmailProps {
+  recipientName?: string
+  /** Drives the noun and the CTA label. */
+  kind: 'workflow' | 'job'
+  /** Workflow name or job title. Absent when the source row could not be read. */
+  resourceName?: string
+  reason: ScheduleDisableReason
+  /** Consecutive failures at disable time. Rendered only for `consecutive_failures`. */
+  failedCount?: number
+  /** Deep link to the workflow or scheduled tasks. Absent when the workspace is unknown. */
+  manageLink?: string
+}
+
+/**
+ * Sent when Sim turns a schedule off on its own — either after repeated
+ * failures or on a terminal error that cannot resolve itself.
+ */
+export function ScheduleDisabledEmail({
+  recipientName,
+  kind,
+  resourceName,
+  reason,
+  failedCount,
+  manageLink,
+}: ScheduleDisabledEmailProps) {
+  const brand = getBrandConfig()
+  const resourceLabel =
+    resourceName ?? (kind === 'job' ? 'a scheduled task' : 'a scheduled workflow')
+  const reasonCopy =
+    reason === 'consecutive_failures' && failedCount
+      ? `It failed ${failedCount.toLocaleString()} times in a row.`
+      : SCHEDULE_DISABLE_REASON_COPY[reason]
+  const previewText = `${brand.name} turned off the schedule for ${resourceLabel}`
+
+  return (
+    <EmailLayout preview={previewText} showUnsubscribe={true}>
+      <Text style={{ ...baseStyles.paragraph, marginTop: 0 }}>
+        {recipientName ? `Hi ${recipientName},` : 'Hi,'}
+      </Text>
+
+      <Text style={baseStyles.paragraph}>
+        {brand.name} turned off the schedule for {resourceLabel}. It will not run again until you
+        turn it back on.
+      </Text>
+
+      <Section style={baseStyles.infoBox}>
+        <Text style={baseStyles.infoBoxTitle}>Reason</Text>
+        <Text style={baseStyles.infoBoxList}>{reasonCopy}</Text>
+      </Section>
+
+      {/* Divider */}
+      <div style={baseStyles.divider} />
+
+      <Text style={baseStyles.paragraph}>Fix the problem, then turn the schedule back on.</Text>
+
+      {manageLink ? (
+        <Link href={manageLink} style={{ textDecoration: 'none' }}>
+          <Text style={baseStyles.button}>
+            {kind === 'job' ? 'Open scheduled tasks' : 'Open workflow'}
+          </Text>
+        </Link>
+      ) : null}
+
+      {/* Divider */}
+      <div style={baseStyles.divider} />
+
+      <Text style={{ ...baseStyles.footerText, textAlign: 'left' }}>
+        Sent to workspace admins and the person who created this schedule.
+      </Text>
+    </EmailLayout>
+  )
+}
+
+export default ScheduleDisabledEmail
diff --git a/apps/sim/components/emails/render-notifications.test.ts b/apps/sim/components/emails/render-notifications.test.ts
new file mode 100644
index 00000000000..67d45e0736a
--- /dev/null
+++ b/apps/sim/components/emails/render-notifications.test.ts
@@ -0,0 +1,98 @@
+/**
+ * Executes the notification templates for real. Every other suite mocks the
+ * render functions, so without this nothing would catch a template that throws
+ * or silently drops its key copy.
+ *
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+import {
+  renderScheduleDisabledEmail,
+  renderUsageLimitReachedEmail,
+} from '@/components/emails/render'
+
+describe('renderScheduleDisabledEmail', () => {
+  it('renders the failure count for a threshold disable', async () => {
+    const html = await renderScheduleDisabledEmail({
+      recipientName: 'John',
+      kind: 'workflow',
+      resourceName: 'Daily digest',
+      reason: 'consecutive_failures',
+      failedCount: 100,
+      manageLink: 'https://sim.ai/workspace/ws_123/w/wf_456',
+    })
+
+    expect(html).toContain('Daily digest')
+    expect(html).toContain('It failed 100 times in a row.')
+    expect(html).toContain('Open workflow')
+    expect(html).toContain('https://sim.ai/workspace/ws_123/w/wf_456')
+  })
+
+  it('renders reason copy instead of a count for a single-strike disable', async () => {
+    const html = await renderScheduleDisabledEmail({
+      kind: 'job',
+      resourceName: 'Weekly report',
+      reason: 'authentication_error',
+      manageLink: 'https://sim.ai/workspace/ws_123/scheduled-tasks',
+    })
+
+    expect(html).toContain('could not be authenticated')
+    expect(html).not.toContain('times in a row')
+    expect(html).toContain('Open scheduled tasks')
+  })
+
+  it('falls back to a generic noun and omits the CTA when the source is unknown', async () => {
+    const html = await renderScheduleDisabledEmail({
+      kind: 'workflow',
+      reason: 'workflow_not_found',
+    })
+
+    expect(html).toContain('a scheduled workflow')
+    expect(html).not.toContain('Open workflow')
+  })
+})
+
+describe('renderUsageLimitReachedEmail', () => {
+  it('tells a personal payer to raise their limit, never to upgrade to Pro', async () => {
+    const html = await renderUsageLimitReachedEmail({
+      userName: 'John',
+      planName: 'Pro',
+      scope: 'user',
+      currentUsage: 20,
+      limit: 20,
+      ctaLink: 'https://sim.ai/account/settings/billing',
+    })
+
+    expect(html).toContain('Raise Usage Limit')
+    expect(html).toContain('upgrade your plan')
+    expect(html).not.toContain('Upgrade to Pro')
+    expect(html).not.toContain('free credits')
+  })
+
+  it('points an organization at the org limit', async () => {
+    const html = await renderUsageLimitReachedEmail({
+      planName: 'Team',
+      scope: 'organization',
+      currentUsage: 500,
+      limit: 500,
+      ctaLink: 'https://sim.ai/organization/org_1/settings/billing',
+    })
+
+    expect(html).toContain('Raise Organization Limit')
+    expect(html).toContain('organization usage limit')
+    expect(html).not.toContain('upgrade your plan')
+  })
+
+  it('renders credits, not dollars', async () => {
+    const html = await renderUsageLimitReachedEmail({
+      planName: 'Pro',
+      scope: 'user',
+      currentUsage: 20,
+      limit: 20,
+      ctaLink: 'https://sim.ai/account/settings/billing',
+    })
+
+    expect(html).toContain('credits used')
+    expect(html).not.toContain('$20')
+  })
+})
diff --git a/apps/sim/components/emails/render.ts b/apps/sim/components/emails/render.ts
index 652d6e7df2a..2510be6cbc6 100644
--- a/apps/sim/components/emails/render.ts
+++ b/apps/sim/components/emails/render.ts
@@ -15,6 +15,7 @@ import {
   LimitThresholdEmail,
   PaymentFailedEmail,
   PlanWelcomeEmail,
+  UsageLimitReachedEmail,
   UsageThresholdEmail,
 } from '@/components/emails/billing'
 import {
@@ -23,9 +24,11 @@ import {
   WorkspaceAddedEmail,
   WorkspaceInvitationEmail,
 } from '@/components/emails/invitations'
+import { ScheduleDisabledEmail } from '@/components/emails/notifications'
 import { HelpConfirmationEmail } from '@/components/emails/support'
 import type { UpgradeReason } from '@/lib/billing/upgrade-reasons'
 import { getBaseUrl } from '@/lib/core/utils/urls'
+import type { ScheduleDisableReason } from '@/lib/workflows/schedules/disable-reasons'
 
 export { getEmailSubject, getLimitEmailSubject } from './subjects'
 
@@ -136,6 +139,28 @@ export async function renderUsageThresholdEmail(params: {
   )
 }
 
+export async function renderUsageLimitReachedEmail(params: {
+  userName?: string
+  planName: string
+  scope: 'user' | 'organization'
+  currentUsage: number
+  limit: number
+  ctaLink: string
+}): Promise<string> {
+  return await render(UsageLimitReachedEmail(params))
+}
+
+export async function renderScheduleDisabledEmail(params: {
+  recipientName?: string
+  kind: 'workflow' | 'job'
+  resourceName?: string
+  reason: ScheduleDisableReason
+  failedCount?: number
+  manageLink?: string
+}): Promise<string> {
+  return await render(ScheduleDisabledEmail(params))
+}
+
 export async function renderFreeTierUpgradeEmail(params: {
   userName?: string
   percentUsed: number
diff --git a/apps/sim/components/emails/subjects.ts b/apps/sim/components/emails/subjects.ts
index 8eeb9b06fdf..b5a4dc7ac71 100644
--- a/apps/sim/components/emails/subjects.ts
+++ b/apps/sim/components/emails/subjects.ts
@@ -21,6 +21,7 @@ export type EmailSubjectType =
   | 'credit-purchase'
   | 'abandoned-checkout'
   | 'free-tier-exhausted'
+  | 'schedule-disabled'
   | 'onboarding-followup'
   | 'welcome'
 
@@ -69,6 +70,8 @@ export function getEmailSubject(type: EmailSubjectType): string {
       return `Quick question`
     case 'free-tier-exhausted':
       return `You've run out of free credits on ${brandName}`
+    case 'schedule-disabled':
+      return `A schedule was turned off on ${brandName}`
     case 'onboarding-followup':
       return `Quick question about ${brandName}`
     case 'welcome':
diff --git a/apps/sim/lib/billing/core/usage.test.ts b/apps/sim/lib/billing/core/usage.test.ts
index c855f49ea94..3d98e15e3c5 100644
--- a/apps/sim/lib/billing/core/usage.test.ts
+++ b/apps/sim/lib/billing/core/usage.test.ts
@@ -8,7 +8,14 @@
  *
  * @vitest-environment node
  */
-import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
+import {
+  dbChainMockFns,
+  queueTableRows,
+  resetDbChainMock,
+  resetEnvFlagsMock,
+  schemaMock,
+  setEnvFlags,
+} from '@sim/testing'
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
 afterAll(() => {
@@ -56,22 +63,52 @@ vi.mock('@/lib/billing/credits/daily-refresh', () => ({
   getOrgMemberRefreshBounds: vi.fn(),
 }))
 
+const {
+  mockGetEmailSubject,
+  mockGetLimitEmailSubject,
+  mockRenderCreditsExhausted,
+  mockRenderFreeTierUpgrade,
+  mockRenderUsageLimitReached,
+  mockRenderUsageThreshold,
+  mockSendEmail,
+  mockGetEmailPreferences,
+  mockIsOrgAdminRole,
+} = vi.hoisted(() => ({
+  mockGetEmailSubject: vi.fn(() => 'Subject'),
+  mockGetLimitEmailSubject: vi.fn(() => 'Limit subject'),
+  mockRenderCreditsExhausted: vi.fn(() => Promise.resolve('<html>free</html>')),
+  mockRenderFreeTierUpgrade: vi.fn(() => Promise.resolve('<html>nudge</html>')),
+  mockRenderUsageLimitReached: vi.fn(() => Promise.resolve('<html>reached</html>')),
+  mockRenderUsageThreshold: vi.fn(() => Promise.resolve('<html>warning</html>')),
+  mockSendEmail: vi.fn(() => Promise.resolve({ success: true })),
+  mockGetEmailPreferences: vi.fn(() => Promise.resolve(null as unknown)),
+  mockIsOrgAdminRole: vi.fn(() => true),
+}))
+
 vi.mock('@/components/emails', () => ({
-  getEmailSubject: vi.fn(),
-  renderCreditsExhaustedEmail: vi.fn(),
-  renderFreeTierUpgradeEmail: vi.fn(),
-  renderUsageThresholdEmail: vi.fn(),
+  getEmailSubject: mockGetEmailSubject,
+  getLimitEmailSubject: mockGetLimitEmailSubject,
+  renderCreditsExhaustedEmail: mockRenderCreditsExhausted,
+  renderFreeTierUpgradeEmail: mockRenderFreeTierUpgrade,
+  renderUsageLimitReachedEmail: mockRenderUsageLimitReached,
+  renderUsageThresholdEmail: mockRenderUsageThreshold,
 }))
 
 vi.mock('@/lib/messaging/email/mailer', () => ({
-  sendEmail: vi.fn(),
+  sendEmail: mockSendEmail,
 }))
 
 vi.mock('@/lib/messaging/email/unsubscribe', () => ({
-  getEmailPreferences: vi.fn(),
+  getEmailPreferences: mockGetEmailPreferences,
 }))
 
-import { getUserUsageLimit, syncUsageLimitsFromSubscription } from '@/lib/billing/core/usage'
+vi.mock('@sim/platform-authz/workspace', () => ({ isOrgAdminRole: mockIsOrgAdminRole }))
+
+import {
+  getUserUsageLimit,
+  maybeSendUsageThresholdEmail,
+  syncUsageLimitsFromSubscription,
+} from '@/lib/billing/core/usage'
 
 const PRO_SUBSCRIPTION = {
   id: 'sub-1',
@@ -232,3 +269,159 @@ describe('syncUsageLimitsFromSubscription', () => {
     expect(expression).toContain('creditBalance')
   })
 })
+
+describe('maybeSendUsageThresholdEmail', () => {
+  const paidUser = {
+    scope: 'user' as const,
+    planName: 'Pro',
+    userId: 'user-1',
+    userEmail: 'user-1@example.com',
+    userName: 'Ada',
+    workspaceId: 'ws-1',
+    limit: 20,
+  }
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+    setEnvFlags({ isBillingEnabled: true })
+    mockGetEmailPreferences.mockResolvedValue(null)
+    mockIsOrgAdminRole.mockReturnValue(true)
+  })
+
+  afterAll(() => {
+    resetEnvFlagsMock()
+  })
+
+  it('emails a paid personal account at 100% with the raise-your-limit template', async () => {
+    await maybeSendUsageThresholdEmail({
+      ...paidUser,
+      percentBefore: 90,
+      percentAfter: 100,
+      currentUsageAfter: 20,
+    })
+
+    expect(mockRenderUsageLimitReached).toHaveBeenCalledWith(
+      expect.objectContaining({ scope: 'user', planName: 'Pro' })
+    )
+    expect(mockRenderCreditsExhausted).not.toHaveBeenCalled()
+    expect(mockGetLimitEmailSubject).toHaveBeenCalledWith('credits', 'reached')
+    expect(mockSendEmail).toHaveBeenCalledTimes(1)
+  })
+
+  it('links a paid personal account to billing settings, not the upgrade page', async () => {
+    await maybeSendUsageThresholdEmail({
+      ...paidUser,
+      percentBefore: 90,
+      percentAfter: 100,
+      currentUsageAfter: 20,
+    })
+
+    const props = mockRenderUsageLimitReached.mock.calls[0]?.[0] as { ctaLink: string }
+    expect(props.ctaLink).toContain('/account/settings/billing')
+  })
+
+  it('fans out to org admins at 100% and skips non-admin members', async () => {
+    mockIsOrgAdminRole.mockImplementation((role: unknown) => role === 'admin')
+    queueTableRows(schemaMock.member, [
+      { email: 'admin@example.com', name: 'Admin', enabled: null, role: 'admin' },
+      { email: 'member@example.com', name: 'Member', enabled: null, role: 'member' },
+    ])
+
+    await maybeSendUsageThresholdEmail({
+      scope: 'organization',
+      planName: 'Team',
+      organizationId: 'org-1',
+      workspaceId: 'ws-1',
+      percentBefore: 95,
+      percentAfter: 100,
+      currentUsageAfter: 500,
+      limit: 500,
+    })
+
+    expect(mockSendEmail).toHaveBeenCalledTimes(1)
+    expect(mockSendEmail).toHaveBeenCalledWith(
+      expect.objectContaining({ to: 'admin@example.com', emailType: 'notifications' })
+    )
+    expect(mockRenderUsageLimitReached).toHaveBeenCalledWith(
+      expect.objectContaining({ scope: 'organization' })
+    )
+  })
+
+  it('still sends the free-tier template to a free personal account at 100%', async () => {
+    await maybeSendUsageThresholdEmail({
+      ...paidUser,
+      planName: 'Free',
+      percentBefore: 90,
+      percentAfter: 100,
+      currentUsageAfter: 10,
+      limit: 10,
+    })
+
+    expect(mockRenderCreditsExhausted).toHaveBeenCalledTimes(1)
+    expect(mockRenderUsageLimitReached).not.toHaveBeenCalled()
+    expect(mockGetEmailSubject).toHaveBeenCalledWith('free-tier-exhausted')
+  })
+
+  it('sends only the reached email when one execution crosses 80 and 100 together', async () => {
+    await maybeSendUsageThresholdEmail({
+      ...paidUser,
+      percentBefore: 70,
+      percentAfter: 100,
+      currentUsageAfter: 20,
+    })
+
+    expect(mockRenderUsageThreshold).not.toHaveBeenCalled()
+    expect(mockRenderUsageLimitReached).toHaveBeenCalledTimes(1)
+    expect(mockSendEmail).toHaveBeenCalledTimes(1)
+  })
+
+  it('still sends the 80% warning to a paid account that has not reached 100%', async () => {
+    await maybeSendUsageThresholdEmail({
+      ...paidUser,
+      percentBefore: 70,
+      percentAfter: 85,
+      currentUsageAfter: 17,
+    })
+
+    expect(mockRenderUsageThreshold).toHaveBeenCalledTimes(1)
+    expect(mockRenderUsageLimitReached).not.toHaveBeenCalled()
+  })
+
+  it('does not send when the recipient unsubscribed from notifications', async () => {
+    mockGetEmailPreferences.mockResolvedValue({ unsubscribeNotifications: true })
+
+    await maybeSendUsageThresholdEmail({
+      ...paidUser,
+      percentBefore: 90,
+      percentAfter: 100,
+      currentUsageAfter: 20,
+    })
+
+    expect(mockSendEmail).not.toHaveBeenCalled()
+  })
+
+  it('does not send when the per-user billing toggle is off', async () => {
+    queueTableRows(schemaMock.settings, [{ enabled: false }])
+
+    await maybeSendUsageThresholdEmail({
+      ...paidUser,
+      percentBefore: 90,
+      percentAfter: 100,
+      currentUsageAfter: 20,
+    })
+
+    expect(mockSendEmail).not.toHaveBeenCalled()
+  })
+
+  it('does nothing when no threshold is crossed', async () => {
+    await maybeSendUsageThresholdEmail({
+      ...paidUser,
+      percentBefore: 50,
+      percentAfter: 60,
+      currentUsageAfter: 12,
+    })
+
+    expect(mockSendEmail).not.toHaveBeenCalled()
+  })
+})
diff --git a/apps/sim/lib/billing/core/usage.ts b/apps/sim/lib/billing/core/usage.ts
index 78b96fe8267..b68d1982623 100644
--- a/apps/sim/lib/billing/core/usage.ts
+++ b/apps/sim/lib/billing/core/usage.ts
@@ -6,8 +6,10 @@ import { generateId } from '@sim/utils/id'
 import { and, eq, isNull, sql } from 'drizzle-orm'
 import {
   getEmailSubject,
+  getLimitEmailSubject,
   renderCreditsExhaustedEmail,
   renderFreeTierUpgradeEmail,
+  renderUsageLimitReachedEmail,
   renderUsageThresholdEmail,
 } from '@/components/emails'
 import { getEffectiveBillingStatus } from '@/lib/billing/core/access'
@@ -863,36 +865,18 @@ export async function maybeSendUsageThresholdEmail(params: {
 
     // Check for 80% threshold crossing — used for paid users (budget warning) and free users (upgrade nudge)
     const crosses80 = params.percentBefore < 80 && params.percentAfter >= 80
-    // Check for 100% threshold (free users only — credits exhausted)
+    // Check for 100% threshold — every plan and scope (usage limit reached)
     const crosses100 = params.percentBefore < 100 && params.percentAfter >= 100
 
     // Skip if no thresholds crossed
     if (!crosses80 && !crosses100) return
 
-    // For 80% threshold email (paid users only)
-    if (crosses80 && !isFreeUser) {
-      const ctaLink = billingSettingsLink
-      const sendTo = async (email: string, name?: string) => {
-        const prefs = await getEmailPreferences(email)
-        if (prefs?.unsubscribeAll || prefs?.unsubscribeNotifications) return
-
-        const html = await renderUsageThresholdEmail({
-          userName: name,
-          planName: params.planName,
-          percentUsed: Math.min(100, Math.round(params.percentAfter)),
-          currentUsage: params.currentUsageAfter,
-          limit: params.limit,
-          ctaLink,
-        })
-
-        await sendEmail({
-          to: email,
-          subject: getEmailSubject('usage-threshold'),
-          html,
-          emailType: 'notifications',
-        })
-      }
-
+    /**
+     * Delivers to the account's notification recipients: the payer for personal
+     * scope, every org admin/owner for organization scope. Honors the per-user
+     * billing-notification toggle in both.
+     */
+    const deliverToScope = async (send: (email: string, name?: string) => Promise<void>) => {
       if (params.scope === 'user' && params.userId && params.userEmail) {
         const rows = await db
           .select({ enabled: settings.billingUsageNotificationsEnabled })
@@ -900,8 +884,11 @@ export async function maybeSendUsageThresholdEmail(params: {
           .where(eq(settings.userId, params.userId))
           .limit(1)
         if (rows.length > 0 && rows[0].enabled === false) return
-        await sendTo(params.userEmail, params.userName)
-      } else if (params.scope === 'organization' && params.organizationId) {
+        await send(params.userEmail, params.userName)
+        return
+      }
+
+      if (params.scope === 'organization' && params.organizationId) {
         const admins = await db
           .select({
             email: user.email,
@@ -915,19 +902,43 @@ export async function maybeSendUsageThresholdEmail(params: {
           .where(eq(member.organizationId, params.organizationId))
 
         for (const a of admins) {
-          const isAdmin = isOrgAdminRole(a.role)
-          if (!isAdmin) continue
+          if (!isOrgAdminRole(a.role)) continue
           if (a.enabled === false) continue
           if (!a.email) continue
-          await sendTo(a.email, a.name || undefined)
+          await send(a.email, a.name || undefined)
         }
       }
     }
 
+    // !crosses100: one "reached" email, not a "nearing" and a "reached" in the same moment
+    if (crosses80 && !isFreeUser && !crosses100) {
+      const ctaLink = billingSettingsLink
+      await deliverToScope(async (email, name) => {
+        const prefs = await getEmailPreferences(email)
+        if (prefs?.unsubscribeAll || prefs?.unsubscribeNotifications) return
+
+        const html = await renderUsageThresholdEmail({
+          userName: name,
+          planName: params.planName,
+          percentUsed: Math.min(100, Math.round(params.percentAfter)),
+          currentUsage: params.currentUsageAfter,
+          limit: params.limit,
+          ctaLink,
+        })
+
+        await sendEmail({
+          to: email,
+          subject: getEmailSubject('usage-threshold'),
+          html,
+          emailType: 'notifications',
+        })
+      })
+    }
+
     // For 80% threshold email (free users only — skip if they also crossed 100% in same call)
     if (crosses80 && isFreeUser && !crosses100) {
       const upgradeLink = upgradeCreditsLink
-      const sendFreeTierEmail = async (email: string, name?: string) => {
+      await deliverToScope(async (email, name) => {
         const prefs = await getEmailPreferences(email)
         if (prefs?.unsubscribeAll || prefs?.unsubscribeNotifications) return
 
@@ -952,56 +963,49 @@ export async function maybeSendUsageThresholdEmail(params: {
           currentUsage: params.currentUsageAfter,
           limit: params.limit,
         })
-      }
-
-      // Free users are always individual scope (not organization)
-      if (params.scope === 'user' && params.userId && params.userEmail) {
-        const rows = await db
-          .select({ enabled: settings.billingUsageNotificationsEnabled })
-          .from(settings)
-          .where(eq(settings.userId, params.userId))
-          .limit(1)
-        if (rows.length > 0 && rows[0].enabled === false) return
-        await sendFreeTierEmail(params.userEmail, params.userName)
-      }
+      })
     }
 
-    // For 100% threshold email (free users only — credits exhausted)
-    if (crosses100 && isFreeUser) {
-      const upgradeLink = upgradeCreditsLink
-      const sendExhaustedEmail = async (email: string, name?: string) => {
+    // Paid and org accounts get raise-your-limit copy — upgrading is not their remedy
+    if (crosses100) {
+      const useFreeCopy = isFreeUser && params.scope === 'user'
+
+      await deliverToScope(async (email, name) => {
         const prefs = await getEmailPreferences(email)
         if (prefs?.unsubscribeAll || prefs?.unsubscribeNotifications) return
 
-        const html = await renderCreditsExhaustedEmail({
-          userName: name,
-          limit: params.limit,
-          upgradeLink,
-        })
+        const html = useFreeCopy
+          ? await renderCreditsExhaustedEmail({
+              userName: name,
+              limit: params.limit,
+              upgradeLink: upgradeCreditsLink,
+            })
+          : await renderUsageLimitReachedEmail({
+              userName: name,
+              planName: params.planName,
+              scope: params.scope,
+              currentUsage: params.currentUsageAfter,
+              limit: params.limit,
+              ctaLink: billingSettingsLink,
+            })
 
         await sendEmail({
           to: email,
-          subject: getEmailSubject('free-tier-exhausted'),
+          subject: useFreeCopy
+            ? getEmailSubject('free-tier-exhausted')
+            : getLimitEmailSubject('credits', 'reached'),
           html,
           emailType: 'notifications',
         })
 
-        logger.info('Free tier credits exhausted email sent', {
+        logger.info('Usage limit reached email sent', {
           email,
+          scope: params.scope,
+          planName: params.planName,
           currentUsage: params.currentUsageAfter,
           limit: params.limit,
         })
-      }
-
-      if (params.scope === 'user' && params.userId && params.userEmail) {
-        const rows = await db
-          .select({ enabled: settings.billingUsageNotificationsEnabled })
-          .from(settings)
-          .where(eq(settings.userId, params.userId))
-          .limit(1)
-        if (rows.length > 0 && rows[0].enabled === false) return
-        await sendExhaustedEmail(params.userEmail, params.userName)
-      }
+      })
     }
   } catch (error) {
     logger.error('Failed to send usage threshold email', {
diff --git a/apps/sim/lib/workflows/schedules/disable-notifications.test.ts b/apps/sim/lib/workflows/schedules/disable-notifications.test.ts
new file mode 100644
index 00000000000..020105c8ccb
--- /dev/null
+++ b/apps/sim/lib/workflows/schedules/disable-notifications.test.ts
@@ -0,0 +1,227 @@
+/**
+ * @vitest-environment node
+ */
+import {
+  dbChainMockFns,
+  queueTableRows,
+  resetDbChainMock,
+  resetUrlsMock,
+  schemaMock,
+  urlsMockFns,
+} from '@sim/testing'
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { sendEmailSpy, renderMock, subjectMock, getUsersWithPermissionsMock } = vi.hoisted(() => ({
+  sendEmailSpy: vi.fn(() => Promise.resolve({ success: true })),
+  renderMock: vi.fn(() => Promise.resolve('<html></html>')),
+  subjectMock: vi.fn(() => 'A schedule was turned off'),
+  getUsersWithPermissionsMock: vi.fn(() => Promise.resolve([] as unknown[])),
+}))
+
+vi.mock('@/lib/messaging/email/mailer', () => ({ sendEmail: sendEmailSpy }))
+vi.mock('@/components/emails', () => ({
+  renderScheduleDisabledEmail: renderMock,
+  getEmailSubject: subjectMock,
+}))
+vi.mock('@/lib/workspaces/permissions/utils', () => ({
+  getUsersWithPermissions: getUsersWithPermissionsMock,
+}))
+
+import { notifyScheduleAutoDisabled } from '@/lib/workflows/schedules/disable-notifications'
+
+const WORKFLOW_SCHEDULE_ROW = {
+  sourceType: 'workflow',
+  jobTitle: null,
+  failedCount: 100,
+  sourceUserId: null,
+  sourceWorkspaceId: null,
+  workflowId: 'wf-1',
+  workflowName: 'Daily digest',
+  workflowUserId: 'creator-1',
+  workflowWorkspaceId: 'ws-1',
+}
+
+const CREATOR = { email: 'creator@example.com', name: 'Ada' }
+
+function admin(email: string, name = 'Admin') {
+  return { userId: `u-${email}`, email, name, permissionType: 'admin' }
+}
+
+beforeAll(() => {
+  urlsMockFns.mockGetBaseUrl.mockReturnValue('https://app.sim.ai')
+})
+
+afterAll(() => {
+  resetDbChainMock()
+  resetUrlsMock()
+})
+
+describe('notifyScheduleAutoDisabled', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+    getUsersWithPermissionsMock.mockResolvedValue([])
+  })
+
+  it('emails the creator and the workspace admins, one send per address', async () => {
+    queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
+    queueTableRows(schemaMock.user, [CREATOR])
+    getUsersWithPermissionsMock.mockResolvedValue([
+      admin('admin-a@example.com'),
+      admin('admin-b@example.com'),
+    ])
+
+    await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
+
+    expect(sendEmailSpy).toHaveBeenCalledTimes(3)
+    for (const call of sendEmailSpy.mock.calls) {
+      // Never a `to` array — prepare.ts only checks to[0] for unsubscribe.
+      expect(typeof (call[0] as { to: unknown }).to).toBe('string')
+      expect(call[0]).toMatchObject({ emailType: 'notifications' })
+    }
+  })
+
+  it('sends once when the creator is also a workspace admin', async () => {
+    queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
+    queueTableRows(schemaMock.user, [CREATOR])
+    getUsersWithPermissionsMock.mockResolvedValue([admin('CREATOR@example.com')])
+
+    await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
+
+    expect(sendEmailSpy).toHaveBeenCalledTimes(1)
+  })
+
+  it('skips non-admin workspace members', async () => {
+    queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
+    queueTableRows(schemaMock.user, [CREATOR])
+    getUsersWithPermissionsMock.mockResolvedValue([
+      { userId: 'u-2', email: 'writer@example.com', name: 'W', permissionType: 'write' },
+    ])
+
+    await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
+
+    expect(sendEmailSpy).toHaveBeenCalledTimes(1)
+    expect(sendEmailSpy).toHaveBeenCalledWith(
+      expect.objectContaining({ to: 'creator@example.com' })
+    )
+  })
+
+  it('resolves a job schedule through sourceUserId and links to scheduled tasks', async () => {
+    queueTableRows(schemaMock.workflowSchedule, [
+      {
+        ...WORKFLOW_SCHEDULE_ROW,
+        sourceType: 'job',
+        jobTitle: 'Weekly report',
+        sourceUserId: 'job-owner',
+        sourceWorkspaceId: 'ws-9',
+        workflowName: null,
+        workflowUserId: null,
+        workflowWorkspaceId: null,
+      },
+    ])
+    queueTableRows(schemaMock.user, [CREATOR])
+
+    await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'authentication_error' })
+
+    expect(renderMock).toHaveBeenCalledWith(
+      expect.objectContaining({
+        kind: 'job',
+        resourceName: 'Weekly report',
+        reason: 'authentication_error',
+        manageLink: 'https://app.sim.ai/workspace/ws-9/scheduled-tasks',
+      })
+    )
+  })
+
+  it('falls back to the creator alone when the workflow has no workspace', async () => {
+    queueTableRows(schemaMock.workflowSchedule, [
+      { ...WORKFLOW_SCHEDULE_ROW, workflowWorkspaceId: null },
+    ])
+    queueTableRows(schemaMock.user, [CREATOR])
+
+    await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
+
+    expect(getUsersWithPermissionsMock).not.toHaveBeenCalled()
+    expect(sendEmailSpy).toHaveBeenCalledTimes(1)
+    expect(renderMock).toHaveBeenCalledWith(expect.objectContaining({ manageLink: undefined }))
+  })
+
+  it('sends nothing when the workflow row is gone (404 disable)', async () => {
+    queueTableRows(schemaMock.workflowSchedule, [
+      {
+        ...WORKFLOW_SCHEDULE_ROW,
+        workflowName: null,
+        workflowUserId: null,
+        workflowWorkspaceId: null,
+      },
+    ])
+
+    await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'workflow_not_found' })
+
+    expect(sendEmailSpy).not.toHaveBeenCalled()
+  })
+
+  it('sends nothing when the schedule row cannot be read', async () => {
+    queueTableRows(schemaMock.workflowSchedule, [])
+
+    await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
+
+    expect(sendEmailSpy).not.toHaveBeenCalled()
+  })
+
+  it('caps the fan-out at 20 recipients', async () => {
+    queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
+    queueTableRows(schemaMock.user, [CREATOR])
+    getUsersWithPermissionsMock.mockResolvedValue(
+      Array.from({ length: 40 }, (_, i) => admin(`admin-${i}@example.com`))
+    )
+
+    await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
+
+    expect(sendEmailSpy).toHaveBeenCalledTimes(20)
+    expect(sendEmailSpy).toHaveBeenCalledWith(
+      expect.objectContaining({ to: 'creator@example.com' })
+    )
+  })
+
+  it('keeps sending after one recipient fails', async () => {
+    queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
+    queueTableRows(schemaMock.user, [CREATOR])
+    getUsersWithPermissionsMock.mockResolvedValue([admin('admin-a@example.com')])
+    sendEmailSpy.mockRejectedValueOnce(new Error('smtp down'))
+
+    await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
+
+    expect(sendEmailSpy).toHaveBeenCalledTimes(2)
+  })
+
+  it('still emails the creator when the admin lookup fails', async () => {
+    queueTableRows(schemaMock.workflowSchedule, [WORKFLOW_SCHEDULE_ROW])
+    queueTableRows(schemaMock.user, [CREATOR])
+    getUsersWithPermissionsMock.mockRejectedValue(new Error('db down'))
+
+    await expect(
+      notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
+    ).resolves.toBeUndefined()
+
+    expect(sendEmailSpy).toHaveBeenCalledTimes(1)
+    expect(sendEmailSpy).toHaveBeenCalledWith(
+      expect.objectContaining({ to: 'creator@example.com' })
+    )
+  })
+
+  it('still emails the admins when the creator lookup fails', async () => {
+    // First .limit() is the schedule row, second is the creator lookup.
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([WORKFLOW_SCHEDULE_ROW])
+      .mockRejectedValueOnce(new Error('db down'))
+    getUsersWithPermissionsMock.mockResolvedValue([admin('admin-a@example.com')])
+
+    await notifyScheduleAutoDisabled({ scheduleId: 's-1', reason: 'consecutive_failures' })
+
+    expect(sendEmailSpy).toHaveBeenCalledTimes(1)
+    expect(sendEmailSpy).toHaveBeenCalledWith(
+      expect.objectContaining({ to: 'admin-a@example.com' })
+    )
+  })
+})
diff --git a/apps/sim/lib/workflows/schedules/disable-notifications.ts b/apps/sim/lib/workflows/schedules/disable-notifications.ts
new file mode 100644
index 00000000000..2bebe8bfd80
--- /dev/null
+++ b/apps/sim/lib/workflows/schedules/disable-notifications.ts
@@ -0,0 +1,175 @@
+import { db } from '@sim/db'
+import { user, workflow, workflowSchedule } from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { eq } from 'drizzle-orm'
+import { getEmailSubject, renderScheduleDisabledEmail } from '@/components/emails'
+import { getBaseUrl } from '@/lib/core/utils/urls'
+import { sendEmail } from '@/lib/messaging/email/mailer'
+import type { ScheduleDisableReason } from '@/lib/workflows/schedules/disable-reasons'
+import { getUsersWithPermissions } from '@/lib/workspaces/permissions/utils'
+
+const logger = createLogger('ScheduleDisableNotifications')
+
+/**
+ * Fan-out cap so one tick over a large workspace can't turn into hundreds of
+ * inline sends. The schedule's creator is always kept.
+ */
+const MAX_SCHEDULE_DISABLE_RECIPIENTS = 20
+
+interface Recipient {
+  email: string
+  name: string | null
+}
+
+/**
+ * Emails the schedule's creator and the workspace admins after Sim turns a
+ * schedule off on its own.
+ *
+ * Best-effort by contract: this never throws, because a mail failure must not
+ * fault the schedule tick that called it.
+ */
+export async function notifyScheduleAutoDisabled(params: {
+  scheduleId: string
+  reason: ScheduleDisableReason
+  requestId?: string
+}): Promise<void> {
+  const { scheduleId, reason, requestId } = params
+
+  try {
+    const rows = await db
+      .select({
+        sourceType: workflowSchedule.sourceType,
+        jobTitle: workflowSchedule.jobTitle,
+        failedCount: workflowSchedule.failedCount,
+        sourceUserId: workflowSchedule.sourceUserId,
+        sourceWorkspaceId: workflowSchedule.sourceWorkspaceId,
+        workflowId: workflowSchedule.workflowId,
+        workflowName: workflow.name,
+        workflowUserId: workflow.userId,
+        workflowWorkspaceId: workflow.workspaceId,
+      })
+      .from(workflowSchedule)
+      .leftJoin(workflow, eq(workflow.id, workflowSchedule.workflowId))
+      .where(eq(workflowSchedule.id, scheduleId))
+      .limit(1)
+
+    const row = rows[0]
+    if (!row) {
+      logger.warn('Schedule auto-disabled but the row could not be read', { scheduleId, reason })
+      return
+    }
+
+    const isJob = row.sourceType === 'job'
+    const kind = isJob ? 'job' : 'workflow'
+    const ownerUserId = isJob ? row.sourceUserId : row.workflowUserId
+    const workspaceId = isJob ? row.sourceWorkspaceId : row.workflowWorkspaceId
+    const resourceName = (isJob ? row.jobTitle : row.workflowName) ?? undefined
+
+    const recipients = await resolveRecipients(ownerUserId, workspaceId)
+    if (recipients.length === 0) {
+      logger.warn('Schedule auto-disabled but no recipients could be resolved', {
+        scheduleId,
+        reason,
+        workspaceId,
+      })
+      return
+    }
+
+    const manageLink = buildManageLink(workspaceId, isJob ? null : row.workflowId)
+    const subject = getEmailSubject('schedule-disabled')
+
+    for (const recipient of recipients) {
+      try {
+        const html = await renderScheduleDisabledEmail({
+          recipientName: recipient.name ?? undefined,
+          kind,
+          resourceName,
+          reason,
+          failedCount: row.failedCount,
+          manageLink,
+        })
+
+        await sendEmail({
+          to: recipient.email,
+          subject,
+          html,
+          emailType: 'notifications',
+        })
+      } catch (error) {
+        logger.error('Failed to send schedule-disabled email', error, {
+          scheduleId,
+          requestId,
+        })
+      }
+    }
+
+    logger.info('Sent schedule-disabled notification', {
+      scheduleId,
+      reason,
+      recipientCount: recipients.length,
+      requestId,
+    })
+  } catch (error) {
+    logger.error('Failed to notify schedule auto-disable', error, { scheduleId, reason, requestId })
+  }
+}
+
+/**
+ * Creator first, then workspace admins. Deduped on lowercased email so someone
+ * who is both only receives one message.
+ *
+ * The two lookups are independent and each failure is contained: losing one
+ * must not zero out the other, or a blip in either turns an auto-disable back
+ * into the silent event this notification exists to prevent.
+ */
+async function resolveRecipients(
+  ownerUserId: string | null,
+  workspaceId: string | null
+): Promise<Recipient[]> {
+  const recipients: Recipient[] = []
+  const seen = new Set<string>()
+
+  const add = (email: string | null, name: string | null) => {
+    if (!email) return
+    const key = email.toLowerCase()
+    if (seen.has(key)) return
+    seen.add(key)
+    recipients.push({ email, name })
+  }
+
+  if (ownerUserId) {
+    try {
+      const ownerRows = await db
+        .select({ email: user.email, name: user.name })
+        .from(user)
+        .where(eq(user.id, ownerUserId))
+        .limit(1)
+      add(ownerRows[0]?.email ?? null, ownerRows[0]?.name ?? null)
+    } catch (error) {
+      logger.error('Failed to resolve schedule creator for disable email', error, { ownerUserId })
+    }
+  }
+
+  if (workspaceId) {
+    try {
+      const members = await getUsersWithPermissions(workspaceId)
+      for (const member of members) {
+        if (member.permissionType !== 'admin') continue
+        add(member.email, member.name)
+      }
+    } catch (error) {
+      logger.error('Failed to resolve workspace admins for disable email', error, { workspaceId })
+    }
+  }
+
+  return recipients.slice(0, MAX_SCHEDULE_DISABLE_RECIPIENTS)
+}
+
+function buildManageLink(
+  workspaceId: string | null,
+  workflowId: string | null
+): string | undefined {
+  if (!workspaceId) return undefined
+  const base = `${getBaseUrl()}/workspace/${workspaceId}`
+  return workflowId ? `${base}/w/${workflowId}` : `${base}/scheduled-tasks`
+}
diff --git a/apps/sim/lib/workflows/schedules/disable-reasons.ts b/apps/sim/lib/workflows/schedules/disable-reasons.ts
new file mode 100644
index 00000000000..d6d20e67591
--- /dev/null
+++ b/apps/sim/lib/workflows/schedules/disable-reasons.ts
@@ -0,0 +1,23 @@
+/**
+ * Why Sim turned a schedule off. Threaded from the disable call site into the
+ * notification, because `workflow_schedule` has no column that records it.
+ */
+export type ScheduleDisableReason =
+  | 'consecutive_failures'
+  | 'authentication_error'
+  | 'authorization_error'
+  | 'workflow_not_found'
+  | 'invalid_schedule'
+
+/**
+ * User-facing sentence for each reason. Fixed copy rather than the underlying
+ * error string: the threshold paths carry arbitrary workflow-execution errors
+ * that can contain credentials or customer data.
+ */
+export const SCHEDULE_DISABLE_REASON_COPY: Record<ScheduleDisableReason, string> = {
+  consecutive_failures: 'It failed too many times in a row.',
+  authentication_error: 'The run could not be authenticated.',
+  authorization_error: 'The run was not authorized.',
+  workflow_not_found: 'The workflow could not be found.',
+  invalid_schedule: 'The schedule is no longer valid.',
+}