diff --git a/apps/api/src/trigger/integration-platform/run-integration-checks-schedule.spec.ts b/apps/api/src/trigger/integration-platform/run-integration-checks-schedule.spec.ts
index a39980e8df..7340b87b41 100644
--- a/apps/api/src/trigger/integration-platform/run-integration-checks-schedule.spec.ts
+++ b/apps/api/src/trigger/integration-platform/run-integration-checks-schedule.spec.ts
@@ -3,9 +3,11 @@ import { db } from '@db';
 import { getManifest } from '@trycompai/integration-platform';
 import {
   filterDueTasks,
+  groupTasksByOrg,
   integrationChecksSchedule,
   resolveProviderChecks,
 } from './run-integration-checks-schedule';
+import { runOrgIntegrationChecks } from './run-org-integration-checks';
 
 // Mock @db at the module boundary so importing the orchestrator does not try
 // to connect to Postgres. We never call the scheduled `run` function itself
@@ -42,8 +44,11 @@ jest.mock('@trigger.dev/sdk', () => ({
   },
 }));
 
-jest.mock('./run-task-integration-checks', () => ({
-  runTaskIntegrationChecks: { batchTrigger: jest.fn() },
+// The orchestrator now dispatches ONE per-org runner instead of N per-task
+// runs. Mock the runner module so importing the orchestrator doesn't load the
+// real runner (which calls queue()/task() at module load).
+jest.mock('./run-org-integration-checks', () => ({
+  runOrgIntegrationChecks: { batchTrigger: jest.fn() },
 }));
 
 jest.mock('./run-device-sync', () => ({
@@ -181,6 +186,54 @@ describe('resolveProviderChecks (static vs dynamic)', () => {
   });
 });
 
+describe('groupTasksByOrg (bundling key)', () => {
+  const base = {
+    taskTitle: 'T',
+    connectionId: 'conn',
+    providerSlug: 'github',
+    checkIds: ['c1'],
+  };
+
+  it('groups tasks by organization and attaches the org name', () => {
+    const groups = groupTasksByOrg({
+      tasksToRun: [
+        { ...base, taskId: 't1', organizationId: 'org1' },
+        { ...base, taskId: 't2', organizationId: 'org1' },
+        { ...base, taskId: 't3', organizationId: 'org2' },
+      ],
+      orgNameById: new Map([
+        ['org1', 'Org One'],
+        ['org2', 'Org Two'],
+      ]),
+    });
+
+    expect(groups).toHaveLength(2);
+    const org1 = groups.find((g) => g.organizationId === 'org1');
+    expect(org1?.organizationName).toBe('Org One');
+    expect(org1?.tasks.map((t) => t.taskId)).toEqual(['t1', 't2']);
+    // organizationId is hoisted to the group; tasks carry the per-task shape.
+    expect(org1?.tasks[0]).not.toHaveProperty('organizationId');
+    expect(groups.find((g) => g.organizationId === 'org2')?.tasks).toHaveLength(
+      1,
+    );
+  });
+
+  it('falls back to "your organization" when the name is unknown', () => {
+    const groups = groupTasksByOrg({
+      tasksToRun: [{ ...base, taskId: 't1', organizationId: 'orgX' }],
+      orgNameById: new Map(),
+    });
+
+    expect(groups[0]?.organizationName).toBe('your organization');
+  });
+
+  it('returns [] for no tasks', () => {
+    expect(
+      groupTasksByOrg({ tasksToRun: [], orgNameById: new Map() }),
+    ).toEqual([]);
+  });
+});
+
 describe('orchestrator excludes MANUAL tasks from scheduled runs', () => {
   // Cast the db/getManifest mocks to their jest.Mock shape for setup.
   const taskFindMany = (db as unknown as { task: { findMany: jest.Mock } }).task
@@ -236,3 +289,94 @@ describe('orchestrator excludes MANUAL tasks from scheduled runs', () => {
     );
   });
 });
+
+describe('orchestrator dispatches ONE runner per org', () => {
+  const taskFindMany = (db as unknown as { task: { findMany: jest.Mock } }).task
+    .findMany;
+  const connectionFindMany = (
+    db as unknown as { integrationConnection: { findMany: jest.Mock } }
+  ).integrationConnection.findMany;
+  const dynamicFindMany = (
+    db as unknown as { dynamicIntegration: { findMany: jest.Mock } }
+  ).dynamicIntegration.findMany;
+  const orgFindMany = (
+    db as unknown as { organization: { findMany: jest.Mock } }
+  ).organization.findMany;
+  const getManifestMock = getManifest as jest.Mock;
+  const batchTrigger = (
+    runOrgIntegrationChecks as unknown as { batchTrigger: jest.Mock }
+  ).batchTrigger;
+
+  const runOrchestrator = (
+    integrationChecksSchedule as unknown as {
+      run: (p: { timestamp: Date; lastTimestamp?: Date }) => Promise<unknown>;
+    }
+  ).run;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    // Two orgs, each with one active connection mapped to the same template.
+    connectionFindMany.mockResolvedValue([
+      {
+        id: 'conn1',
+        provider: { slug: 'github' },
+        organizationId: 'org1',
+        organization: { id: 'org1', name: 'Org One' },
+        metadata: null,
+      },
+      {
+        id: 'conn2',
+        provider: { slug: 'github' },
+        organizationId: 'org2',
+        organization: { id: 'org2', name: 'Org Two' },
+        metadata: null,
+      },
+    ]);
+    dynamicFindMany.mockResolvedValue([]);
+    getManifestMock.mockReturnValue({
+      checks: [{ id: 'c1', taskMapping: 'tpl_a' }],
+    });
+    // One due (daily) task per connection.
+    taskFindMany
+      .mockResolvedValueOnce([
+        {
+          id: 'task_org1',
+          title: 'Org1 Task',
+          taskTemplateId: 'tpl_a',
+          integrationScheduleFrequency: 'daily',
+          integrationLastRunAt: null,
+        },
+      ])
+      .mockResolvedValueOnce([
+        {
+          id: 'task_org2',
+          title: 'Org2 Task',
+          taskTemplateId: 'tpl_a',
+          integrationScheduleFrequency: 'daily',
+          integrationLastRunAt: null,
+        },
+      ]);
+    orgFindMany.mockResolvedValue([]);
+  });
+
+  it('triggers the per-org runner once with one payload per org (not per task)', async () => {
+    await runOrchestrator({ timestamp: new Date(), lastTimestamp: new Date() });
+
+    expect(batchTrigger).toHaveBeenCalledTimes(1);
+    const payloads = batchTrigger.mock.calls[0][0] as Array<{
+      payload: {
+        organizationId: string;
+        organizationName: string;
+        tasks: Array<{ taskId: string }>;
+      };
+    }>;
+    expect(payloads).toHaveLength(2);
+
+    const org1 = payloads.find((p) => p.payload.organizationId === 'org1');
+    expect(org1?.payload.organizationName).toBe('Org One');
+    expect(org1?.payload.tasks.map((t) => t.taskId)).toEqual(['task_org1']);
+
+    const org2 = payloads.find((p) => p.payload.organizationId === 'org2');
+    expect(org2?.payload.tasks.map((t) => t.taskId)).toEqual(['task_org2']);
+  });
+});
diff --git a/apps/api/src/trigger/integration-platform/run-integration-checks-schedule.ts b/apps/api/src/trigger/integration-platform/run-integration-checks-schedule.ts
index f3e7a91dcf..cbaeb16f5b 100644
--- a/apps/api/src/trigger/integration-platform/run-integration-checks-schedule.ts
+++ b/apps/api/src/trigger/integration-platform/run-integration-checks-schedule.ts
@@ -1,7 +1,10 @@
 import { getManifest } from '@trycompai/integration-platform';
 import { db, TaskAutomationStatus, TaskFrequency } from '@db';
 import { logger, schedules } from '@trigger.dev/sdk';
-import { runTaskIntegrationChecks } from './run-task-integration-checks';
+import {
+  runOrgIntegrationChecks,
+  type OrgTaskCheck,
+} from './run-org-integration-checks';
 import { runDeviceSync } from './run-device-sync';
 import { isCheckDisabledForTask } from '../../integration-platform/utils/disabled-task-checks';
 import { isDueToday } from '../shared/is-due-today';
@@ -66,6 +69,59 @@ export function resolveProviderChecks({
   return dynamicChecks ?? [];
 }
 
+/** A flat scheduled-task entry, before grouping by org. */
+export interface ScheduledTask {
+  taskId: string;
+  taskTitle: string;
+  connectionId: string;
+  providerSlug: string;
+  organizationId: string;
+  checkIds: string[];
+}
+
+/** One org's worth of scheduled tasks, ready to hand to the per-org runner. */
+export interface OrgTaskGroup {
+  organizationId: string;
+  organizationName: string;
+  tasks: OrgTaskCheck[];
+}
+
+/**
+ * Group the flat task list into one entry per organization, attaching each
+ * org's display name. Pure + exported for unit testing. This is what makes the
+ * failure email bundle-able: one runner (and thus one email) per org instead of
+ * one independent run (and email) per task.
+ */
+export function groupTasksByOrg({
+  tasksToRun,
+  orgNameById,
+}: {
+  tasksToRun: ScheduledTask[];
+  orgNameById: Map<string, string>;
+}): OrgTaskGroup[] {
+  const byOrg = new Map<string, OrgTaskGroup>();
+  for (const t of tasksToRun) {
+    let group = byOrg.get(t.organizationId);
+    if (!group) {
+      group = {
+        organizationId: t.organizationId,
+        organizationName:
+          orgNameById.get(t.organizationId) ?? 'your organization',
+        tasks: [],
+      };
+      byOrg.set(t.organizationId, group);
+    }
+    group.tasks.push({
+      taskId: t.taskId,
+      taskTitle: t.taskTitle,
+      connectionId: t.connectionId,
+      providerSlug: t.providerSlug,
+      checkIds: t.checkIds,
+    });
+  }
+  return Array.from(byOrg.values());
+}
+
 /**
  * Daily scheduled task (orchestrator) that finds all tasks with integration checks
  * and triggers individual check runs for each.
@@ -95,6 +151,17 @@ export const integrationChecksSchedule = schedules.task({
 
     logger.info(`Found ${activeConnections.length} active connections`);
 
+    // Org display names, used by the per-org runner's bundled failure email.
+    const orgNameById = new Map<string, string>();
+    for (const connection of activeConnections) {
+      if (connection.organization?.name) {
+        orgNameById.set(
+          connection.organizationId,
+          connection.organization.name,
+        );
+      }
+    }
+
     // Dynamic (DB-backed) integrations are NOT in the Trigger.dev manifest
     // registry, so getManifest() returns undefined for them below. Load their
     // enabled check → task mappings straight from the DB so this orchestrator
@@ -118,14 +185,7 @@ export const integrationChecksSchedule = schedules.task({
     );
 
     // For each connection, find tasks that have checks mapped to them
-    const tasksToRun: Array<{
-      taskId: string;
-      taskTitle: string;
-      connectionId: string;
-      providerSlug: string;
-      organizationId: string;
-      checkIds: string[];
-    }> = [];
+    const tasksToRun: ScheduledTask[] = [];
 
     for (const connection of activeConnections) {
       // Static providers resolve from the code manifest; dynamic ones from the
@@ -202,35 +262,48 @@ export const integrationChecksSchedule = schedules.task({
       }
     }
 
-    // Trigger integration checks in batches
-    let totalTriggered = 0;
+    // Dispatch ONE runner per org (fire-and-forget). Each runner waits on its
+    // own tasks and sends a SINGLE bundled failure email — instead of every
+    // per-task run emailing independently (the old one-email-per-failure spam).
+    // Keeping the top level fire-and-forget means the orchestrator stays fast
+    // and a slow/failing org can't hold the whole daily run open.
+    let tasksTriggered = 0;
+    let orgsTriggered = 0;
+
+    const orgGroups = groupTasksByOrg({ tasksToRun, orgNameById });
 
-    if (tasksToRun.length === 0) {
+    if (orgGroups.length === 0) {
       logger.info('No tasks with mapped integration checks found');
     } else {
       logger.info(
-        `Found ${tasksToRun.length} tasks with integration checks to run`,
+        `Found ${tasksToRun.length} task(s) across ${orgGroups.length} org(s) to run`,
       );
 
-      const BATCH_SIZE = 500;
-      const triggerPayloads = tasksToRun.map((t) => ({ payload: t }));
+      const ORG_BATCH_SIZE = 100;
+      const triggerPayloads = orgGroups.map((g) => ({ payload: g }));
 
       try {
-        for (let i = 0; i < triggerPayloads.length; i += BATCH_SIZE) {
-          const batch = triggerPayloads.slice(i, i + BATCH_SIZE);
-          await runTaskIntegrationChecks.batchTrigger(batch);
-          totalTriggered += batch.length;
+        for (let i = 0; i < triggerPayloads.length; i += ORG_BATCH_SIZE) {
+          const batch = triggerPayloads.slice(i, i + ORG_BATCH_SIZE);
+          await runOrgIntegrationChecks.batchTrigger(batch);
+          orgsTriggered += batch.length;
+          tasksTriggered += batch.reduce(
+            (n, p) => n + p.payload.tasks.length,
+            0,
+          );
 
           logger.info(
-            `Triggered batch ${Math.floor(i / BATCH_SIZE) + 1}: ${batch.length} tasks`,
+            `Triggered org batch ${Math.floor(i / ORG_BATCH_SIZE) + 1}: ${batch.length} org(s)`,
           );
         }
 
-        logger.info(`Triggered ${totalTriggered} task integration check runs`);
+        logger.info(
+          `Triggered ${orgsTriggered} org runner(s) covering ${tasksTriggered} task(s)`,
+        );
       } catch (error) {
-        logger.error('Failed to trigger task integration checks', {
+        logger.error('Failed to trigger org integration check runs', {
           error: error instanceof Error ? error.message : String(error),
-          triggeredBeforeError: totalTriggered,
+          orgsTriggeredBeforeError: orgsTriggered,
         });
       }
     }
@@ -281,10 +354,13 @@ export const integrationChecksSchedule = schedules.task({
     logger.info(`Triggered ${deviceSyncsTriggered} device syncs`);
 
     return {
-      // Report failure when not every queued task batch was dispatched OR a
-      // device-sync dispatch threw, so partial/failed runs aren't masked.
-      success: totalTriggered === tasksToRun.length && deviceSyncFailures === 0,
-      tasksTriggered: totalTriggered,
+      // Report failure when not every queued task was dispatched (via its org
+      // runner) OR a device-sync dispatch threw, so partial/failed runs aren't
+      // masked.
+      success:
+        tasksTriggered === tasksToRun.length && deviceSyncFailures === 0,
+      tasksTriggered,
+      orgsTriggered,
       deviceSyncsTriggered,
     };
   },
diff --git a/apps/api/src/trigger/integration-platform/run-org-integration-checks.spec.ts b/apps/api/src/trigger/integration-platform/run-org-integration-checks.spec.ts
new file mode 100644
index 0000000000..740c92fde8
--- /dev/null
+++ b/apps/api/src/trigger/integration-platform/run-org-integration-checks.spec.ts
@@ -0,0 +1,200 @@
+const mockDb = {
+  task: { findMany: jest.fn() },
+  member: { findMany: jest.fn() },
+};
+
+jest.mock('@db', () => ({ db: mockDb }));
+
+const isUserUnsubscribedMock = jest.fn().mockResolvedValue(false);
+jest.mock('@trycompai/email', () => ({
+  isUserUnsubscribed: (...args: unknown[]) => isUserUnsubscribedMock(...args),
+}));
+
+const triggerEmailMock = jest.fn().mockResolvedValue({ id: 'email_1' });
+jest.mock('../../email/trigger-email', () => ({
+  triggerEmail: (...args: unknown[]) => triggerEmailMock(...args),
+}));
+
+// Return the props so tests can inspect the bundled task list passed to the
+// email template.
+jest.mock('../../email/templates/automation-bulk-failures', () => ({
+  AutomationBulkFailuresEmail: (props: unknown) => ({ __email: 'bulk', props }),
+}));
+
+// Importing the runner evaluates queue()/task() at module load — stub them.
+// runTaskIntegrationChecks is only referenced inside the task body (not load).
+jest.mock('@trigger.dev/sdk', () => ({
+  logger: { info: jest.fn(), warn: jest.fn(), error: jest.fn() },
+  queue: jest.fn(() => ({ name: 'q' })),
+  task: (config: unknown) => config,
+}));
+jest.mock('./run-task-integration-checks', () => ({
+  runTaskIntegrationChecks: { batchTriggerAndWait: jest.fn() },
+}));
+
+import {
+  collectFailedTasks,
+  sendBundledFailureEmails,
+  type FailedTaskSummary,
+} from './run-org-integration-checks';
+import type { TaskCheckRunResult } from './run-task-integration-checks';
+
+function recipientEmails(): string[] {
+  return triggerEmailMock.mock.calls.map((call) => call[0].to as string);
+}
+
+function emailedTaskTitles(callIndex = 0): string[] {
+  const react = triggerEmailMock.mock.calls[callIndex][0].react as {
+    props: { tasks: Array<{ title: string }> };
+  };
+  return react.props.tasks.map((t) => t.title);
+}
+
+const okRun = (
+  output: Partial<Extract<TaskCheckRunResult, { success: true }>> & {
+    taskId: string;
+  },
+): { ok: true; output: TaskCheckRunResult } => ({
+  ok: true,
+  output: {
+    success: true,
+    taskTitle: output.taskId,
+    checksRun: 1,
+    totalPassing: 0,
+    totalFindings: 0,
+    taskStatus: 'failed',
+    statusChangedToFailed: true,
+    failedCount: 1,
+    totalCount: 1,
+    ...output,
+  },
+});
+
+describe('collectFailedTasks', () => {
+  it('keeps only ok + success + freshly-transitioned-to-failed runs', () => {
+    const failed = collectFailedTasks([
+      okRun({ taskId: 't1', taskTitle: 'Task 1', failedCount: 2, totalCount: 9 }),
+      // success but NOT a fresh transition (already failed) → dropped
+      okRun({ taskId: 't2', statusChangedToFailed: false }),
+      // child run errored/crashed → dropped
+      { ok: false },
+      // child returned a failure result → dropped
+      { ok: true, output: { success: false, taskId: 't3', error: 'boom' } },
+    ]);
+
+    expect(failed).toEqual([
+      { taskId: 't1', taskTitle: 'Task 1', failedCount: 2, totalCount: 9 },
+    ]);
+  });
+
+  it('returns [] when nothing failed', () => {
+    expect(
+      collectFailedTasks([okRun({ taskId: 't1', statusChangedToFailed: false })]),
+    ).toEqual([]);
+  });
+});
+
+describe('sendBundledFailureEmails', () => {
+  const failedTasks: FailedTaskSummary[] = [
+    { taskId: 't1', taskTitle: 'Task 1', failedCount: 2, totalCount: 10 },
+    { taskId: 't2', taskTitle: 'Task 2', failedCount: 1, totalCount: 5 },
+  ];
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    isUserUnsubscribedMock.mockResolvedValue(false);
+    // Assignee of t1; t2 unassigned.
+    mockDb.task.findMany.mockResolvedValue([
+      { assignee: { user: { id: 'u_assignee', name: 'Ann', email: 'ann@x.com' } } },
+      { assignee: null },
+    ]);
+    // admin + owner + an employee (excluded) + a substring-only custom role
+    // (must be excluded by EXACT token matching) + the assignee again (deduped).
+    mockDb.member.findMany.mockResolvedValue([
+      { role: 'admin', user: { id: 'u_admin', name: 'Adam', email: 'adam@x.com' } },
+      { role: 'owner', user: { id: 'u_owner', name: 'Oli', email: 'oli@x.com' } },
+      { role: 'employee', user: { id: 'u_emp', name: 'Eve', email: 'eve@x.com' } },
+      { role: 'co-owner', user: { id: 'u_sub', name: 'Sub', email: 'sub@x.com' } },
+      {
+        role: 'admin,auditor',
+        user: { id: 'u_assignee', name: 'Ann', email: 'ann@x.com' },
+      },
+    ]);
+  });
+
+  it('sends ONE bundled email per recipient (assignees ∪ admins/owners, deduped)', async () => {
+    await sendBundledFailureEmails({
+      organizationId: 'org1',
+      organizationName: 'Acme',
+      failedTasks,
+    });
+
+    // assignee + admin + owner; employee + substring-only "co-owner" excluded;
+    // assignee not double-sent.
+    expect(recipientEmails().sort()).toEqual(
+      ['adam@x.com', 'ann@x.com', 'oli@x.com'].sort(),
+    );
+    // EXACT-token matching: "co-owner" must NOT be treated as owner.
+    expect(recipientEmails()).not.toContain('sub@x.com');
+    // Every recipient gets the FULL list of failed tasks.
+    expect(emailedTaskTitles(0)).toEqual(['Task 1', 'Task 2']);
+    expect(triggerEmailMock.mock.calls[0][0].subject).toBe(
+      '2 tasks failed automated checks in Acme',
+    );
+  });
+
+  it('queries members WITHOUT a platform user.role filter (notifies admin/owner regardless of platform role)', async () => {
+    await sendBundledFailureEmails({
+      organizationId: 'org1',
+      organizationName: 'Acme',
+      failedTasks,
+    });
+
+    expect(mockDb.member.findMany).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: { organizationId: 'org1', deactivated: false },
+      }),
+    );
+  });
+
+  it('skips unsubscribed recipients', async () => {
+    isUserUnsubscribedMock.mockImplementation(
+      (_db, email: string) => Promise.resolve(email === 'adam@x.com'),
+    );
+
+    await sendBundledFailureEmails({
+      organizationId: 'org1',
+      organizationName: 'Acme',
+      failedTasks,
+    });
+
+    expect(recipientEmails().sort()).toEqual(['ann@x.com', 'oli@x.com'].sort());
+  });
+
+  it('sends nothing when there are no failed tasks', async () => {
+    await sendBundledFailureEmails({
+      organizationId: 'org1',
+      organizationName: 'Acme',
+      failedTasks: [],
+    });
+
+    expect(triggerEmailMock).not.toHaveBeenCalled();
+    expect(mockDb.task.findMany).not.toHaveBeenCalled();
+  });
+
+  it('swallows a recipient-resolution failure (best-effort) instead of throwing', async () => {
+    // A DB blip here must NOT propagate out of the runner — otherwise the whole
+    // org run would fail and retry, and the email would be lost permanently.
+    mockDb.task.findMany.mockRejectedValue(new Error('db down'));
+
+    await expect(
+      sendBundledFailureEmails({
+        organizationId: 'org1',
+        organizationName: 'Acme',
+        failedTasks,
+      }),
+    ).resolves.toBeUndefined();
+
+    expect(triggerEmailMock).not.toHaveBeenCalled();
+  });
+});
diff --git a/apps/api/src/trigger/integration-platform/run-org-integration-checks.ts b/apps/api/src/trigger/integration-platform/run-org-integration-checks.ts
new file mode 100644
index 0000000000..815ab10cdb
--- /dev/null
+++ b/apps/api/src/trigger/integration-platform/run-org-integration-checks.ts
@@ -0,0 +1,295 @@
+import { db } from '@db';
+import { logger, queue, task } from '@trigger.dev/sdk';
+import { isUserUnsubscribed } from '@trycompai/email';
+import { triggerEmail } from '../../email/trigger-email';
+import { AutomationBulkFailuresEmail } from '../../email/templates/automation-bulk-failures';
+import {
+  runTaskIntegrationChecks,
+  type TaskCheckRunResult,
+} from './run-task-integration-checks';
+
+/** One task scheduled for an org, as handed down by the orchestrator. */
+export interface OrgTaskCheck {
+  taskId: string;
+  taskTitle: string;
+  connectionId: string;
+  providerSlug: string;
+  checkIds: string[];
+}
+
+/** A task that freshly transitioned into `failed` during this run. */
+export interface FailedTaskSummary {
+  taskId: string;
+  taskTitle: string;
+  failedCount: number;
+  totalCount: number;
+}
+
+// Bound how many per-org runners wait concurrently. The child checks run on the
+// default queue at full env concurrency; only the PARENTS are capped here, so a
+// day with many orgs can't pin the whole environment with runners that are
+// merely suspended waiting on their children.
+const orgRunnerQueue = queue({
+  name: 'integration-checks-org-runner',
+  concurrencyLimit: 50,
+});
+
+// Chunk large orgs: batchTriggerAndWait accepts a bounded number of items per
+// call (the orchestrator already self-limits its own triggers the same way).
+const CHILD_BATCH_SIZE = 100;
+
+/**
+ * Reduce the results of a batchTriggerAndWait over runTaskIntegrationChecks to
+ * the tasks that freshly failed. Pure + exported for unit testing.
+ *
+ * Errored/crashed child runs (`!ok`) and runs that didn't transition into
+ * `failed` are dropped — matching the prior behavior where an already-failed or
+ * errored task produced no email (it just retries on the next orchestrator tick).
+ */
+export function collectFailedTasks(
+  runs: Array<{ ok: boolean; output?: TaskCheckRunResult }>,
+): FailedTaskSummary[] {
+  const failed: FailedTaskSummary[] = [];
+  for (const run of runs) {
+    if (!run.ok || !run.output || run.output.success !== true) continue;
+    if (!run.output.statusChangedToFailed) continue;
+    failed.push({
+      taskId: run.output.taskId,
+      taskTitle: run.output.taskTitle,
+      failedCount: run.output.failedCount,
+      totalCount: run.output.totalCount,
+    });
+  }
+  return failed;
+}
+
+type Recipient = { id: string; name: string; email: string };
+
+function toRecipient(user: {
+  id: string;
+  name: string | null;
+  email: string;
+}): Recipient {
+  return {
+    id: user.id,
+    name: user.name?.trim() || user.email.trim() || 'User',
+    email: user.email,
+  };
+}
+
+/**
+ * Recipients for an org's bundled failure email: the assignees of the failed
+ * tasks UNION the org's admins/owners (by EXACT member-role token), deduped by
+ * user id. Mirrors the canonical getOwnerAdminRecipients resolver in
+ * task-notifier.service.ts.
+ *
+ * Note the deliberate product change: every recipient receives the FULL org
+ * digest (all tasks that failed this run), so a non-admin assignee now sees the
+ * org's other failed tasks too — not just their own. This is the intended
+ * "one bundled email per org" behavior (replacing one email per failing task).
+ */
+async function resolveRecipients(params: {
+  organizationId: string;
+  failedTaskIds: string[];
+}): Promise<Recipient[]> {
+  const { organizationId, failedTaskIds } = params;
+
+  const [tasks, allMembers] = await Promise.all([
+    db.task.findMany({
+      where: { id: { in: failedTaskIds }, organizationId },
+      select: {
+        assignee: {
+          select: { user: { select: { id: true, name: true, email: true } } },
+        },
+      },
+    }),
+    db.member.findMany({
+      where: {
+        organizationId,
+        deactivated: false,
+      },
+      select: {
+        role: true,
+        user: { select: { id: true, name: true, email: true } },
+      },
+    }),
+  ]);
+
+  const recipientMap = new Map<string, Recipient>();
+
+  // Assignees of the failed tasks.
+  for (const t of tasks) {
+    const user = t.assignee?.user;
+    if (user?.id && user.email) recipientMap.set(user.id, toRecipient(user));
+  }
+
+  // Org admins/owners. member.role is a comma-separated list (e.g.
+  // "admin,auditor"); match EXACT role tokens, not substrings, so a custom role
+  // like "co-owner" or "billing-admin" is not mistaken for owner/admin.
+  for (const member of allMembers) {
+    const roles = (member.role ?? '').split(',').map((r) => r.trim());
+    if (!roles.includes('admin') && !roles.includes('owner')) continue;
+    const user = member.user;
+    if (user?.id && user.email) recipientMap.set(user.id, toRecipient(user));
+  }
+
+  return Array.from(recipientMap.values());
+}
+
+/**
+ * Send ONE bundled email per recipient listing every task that failed this run,
+ * replacing the previous one-email-per-failing-task spam. Exported for testing.
+ */
+export async function sendBundledFailureEmails(params: {
+  organizationId: string;
+  organizationName: string;
+  failedTasks: FailedTaskSummary[];
+}): Promise<void> {
+  const { organizationId, organizationName, failedTasks } = params;
+  if (failedTasks.length === 0) return;
+
+  // The email is best-effort: a failure here (e.g. a transient DB blip while
+  // resolving recipients) must NOT throw out of the runner, which would fail it
+  // and retry the WHOLE org's checks — and since the tasks are already `failed`
+  // by then, the retry would report no transitions and the email would be lost
+  // forever. Mirrors the old per-task email's outer try/catch guard.
+  try {
+    const appUrl =
+      process.env.NEXT_PUBLIC_APP_URL ||
+      process.env.BETTER_AUTH_URL ||
+      'https://app.trycomp.ai';
+    const tasksUrl = `${appUrl}/${organizationId}/tasks`;
+
+    const recipients = await resolveRecipients({
+      organizationId,
+      failedTaskIds: failedTasks.map((t) => t.taskId),
+    });
+
+    const taskItems = failedTasks.map((t) => ({
+      title: t.taskTitle,
+      url: `${appUrl}/${organizationId}/tasks/${t.taskId}`,
+      failedCount: t.failedCount,
+      totalCount: t.totalCount,
+    }));
+
+    const count = failedTasks.length;
+    const taskText = count === 1 ? 'task' : 'tasks';
+
+    await Promise.allSettled(
+      recipients.map(async (recipient) => {
+        const isUnsubscribed = await isUserUnsubscribed(
+          db,
+          recipient.email,
+          'taskAssignments',
+          organizationId,
+        );
+        if (isUnsubscribed) {
+          logger.info(
+            `Skipping bundled failure email: ${recipient.email} is unsubscribed`,
+          );
+          return;
+        }
+
+        try {
+          await triggerEmail({
+            to: recipient.email,
+            subject: `${count} ${taskText} failed automated checks in ${organizationName}`,
+            react: AutomationBulkFailuresEmail({
+              toName: recipient.name,
+              toEmail: recipient.email,
+              organizationName,
+              tasksUrl,
+              tasks: taskItems,
+            }),
+            system: true,
+          });
+          logger.info(`Bundled failure email sent to ${recipient.email}`);
+        } catch (error) {
+          logger.error(
+            `Failed to send bundled failure email to ${recipient.email}`,
+            { error: error instanceof Error ? error.message : 'Unknown error' },
+          );
+        }
+      }),
+    );
+
+    logger.info(
+      `Sent bundled failure email for ${count} ${taskText} to ${recipients.length} recipient(s) in org ${organizationId}`,
+    );
+  } catch (error) {
+    logger.error('Failed to send bundled failure email(s)', {
+      organizationId,
+      error: error instanceof Error ? error.message : 'Unknown error',
+    });
+  }
+}
+
+/**
+ * Per-org runner. The daily orchestrator dispatches ONE of these per org
+ * (fire-and-forget). It runs that org's due integration checks in parallel via
+ * batchTriggerAndWait, then sends a SINGLE bundled email listing every task that
+ * failed this run — instead of one email per failing task.
+ *
+ * Mirrors the established in-repo nested fan-out pattern (e.g. onboarding's
+ * per-org runners that batchTriggerAndWait their work internally).
+ */
+export const runOrgIntegrationChecks = task({
+  id: 'run-org-integration-checks',
+  queue: orgRunnerQueue,
+  // maxDuration is max COMPUTE time in SECONDS (the suspended wait during
+  // batchTriggerAndWait is checkpointed and doesn't count against it). 1h is
+  // ample for the runner's own work (collect results + send emails) and matches
+  // the sibling batchTriggerAndWait orchestrator (process-knowledge-base-documents).
+  maxDuration: 60 * 60, // 1 hour (in seconds)
+  run: async (payload: {
+    organizationId: string;
+    organizationName: string;
+    tasks: OrgTaskCheck[];
+  }) => {
+    const { organizationId, organizationName, tasks } = payload;
+
+    logger.info(
+      `Running integration checks for org ${organizationId} (${tasks.length} task(s))`,
+    );
+
+    if (tasks.length === 0) {
+      return { organizationId, tasksRun: 0, failedTasks: 0, emailed: false };
+    }
+
+    const failedTasks: FailedTaskSummary[] = [];
+
+    for (let i = 0; i < tasks.length; i += CHILD_BATCH_SIZE) {
+      const batch = tasks.slice(i, i + CHILD_BATCH_SIZE);
+      const batchResult = await runTaskIntegrationChecks.batchTriggerAndWait(
+        batch.map((t) => ({
+          payload: {
+            taskId: t.taskId,
+            taskTitle: t.taskTitle,
+            connectionId: t.connectionId,
+            providerSlug: t.providerSlug,
+            organizationId,
+            checkIds: t.checkIds,
+          },
+        })),
+      );
+      failedTasks.push(...collectFailedTasks(batchResult.runs));
+    }
+
+    logger.info(
+      `Org ${organizationId}: ${failedTasks.length} of ${tasks.length} task(s) transitioned to failed`,
+    );
+
+    await sendBundledFailureEmails({
+      organizationId,
+      organizationName,
+      failedTasks,
+    });
+
+    return {
+      organizationId,
+      tasksRun: tasks.length,
+      failedTasks: failedTasks.length,
+      emailed: failedTasks.length > 0,
+    };
+  },
+});
diff --git a/apps/api/src/trigger/integration-platform/run-task-integration-checks.ts b/apps/api/src/trigger/integration-platform/run-task-integration-checks.ts
index 06dd592545..a5db15636d 100644
--- a/apps/api/src/trigger/integration-platform/run-task-integration-checks.ts
+++ b/apps/api/src/trigger/integration-platform/run-task-integration-checks.ts
@@ -1,9 +1,6 @@
 import { getManifest, runAllChecks } from '@trycompai/integration-platform';
 import { db } from '@db';
 import { logger, tags, task } from '@trigger.dev/sdk';
-import { triggerEmail } from '../../email/trigger-email';
-import { TaskStatusChangedEmail } from '../../email/templates/task-status-changed';
-import { isUserUnsubscribed } from '@trycompai/email';
 import { isCheckDisabledForTask } from '../../integration-platform/utils/disabled-task-checks';
 import {
   getAccessToken,
@@ -26,164 +23,34 @@ import {
 } from '../../integration-platform/utils/task-check-evaluation';
 
 /**
- * Send email notifications for task status change
+ * Result of one task's integration-check run. The per-org runner
+ * (run-org-integration-checks) reads the success branch to bundle every task
+ * that freshly transitioned into `failed` into a single email per recipient.
  */
-async function sendTaskStatusChangeEmails(params: {
-  organizationId: string;
-  taskId: string;
-  taskTitle: string;
-  oldStatus: string;
-  newStatus: 'done' | 'failed';
-}) {
-  const { organizationId, taskId, taskTitle, oldStatus, newStatus } = params;
-
-  try {
-    // Get organization, task assignee, and org owners
-    const [organization, task, allMembers] = await Promise.all([
-      db.organization.findUnique({
-        where: { id: organizationId },
-        select: { name: true },
-      }),
-      db.task.findUnique({
-        where: { id: taskId },
-        select: {
-          assignee: {
-            select: {
-              user: {
-                select: {
-                  id: true,
-                  name: true,
-                  email: true,
-                },
-              },
-            },
-          },
-        },
-      }),
-      db.member.findMany({
-        where: {
-          organizationId,
-          deactivated: false,
-          user: { role: { not: 'admin' } },
-        },
-        select: {
-          role: true,
-          user: {
-            select: {
-              id: true,
-              name: true,
-              email: true,
-            },
-          },
-        },
-      }),
-    ]);
-
-    const organizationName = organization?.name ?? 'your organization';
-    const appUrl =
-      process.env.NEXT_PUBLIC_APP_URL ||
-      process.env.BETTER_AUTH_URL ||
-      'https://app.trycomp.ai';
-    const taskUrl = `${appUrl}/${organizationId}/tasks/${taskId}`;
-
-    // Filter for admins/owners
-    const adminMembers = allMembers.filter(
-      (member) =>
-        member.role &&
-        (member.role.includes('admin') || member.role.includes('owner')),
-    );
-
-    // Build recipient list: assignee + admins
-    const recipientMap = new Map<
-      string,
-      { id: string; name: string; email: string }
-    >();
-
-    // Add assignee
-    if (task?.assignee?.user?.id && task.assignee.user.email) {
-      recipientMap.set(task.assignee.user.id, {
-        id: task.assignee.user.id,
-        name:
-          task.assignee.user.name?.trim() ||
-          task.assignee.user.email?.trim() ||
-          'User',
-        email: task.assignee.user.email,
-      });
+export type TaskCheckRunResult =
+  | {
+      success: true;
+      taskId: string;
+      taskTitle: string;
+      checksRun: number;
+      totalPassing: number;
+      totalFindings: number;
+      taskStatus: 'failed' | 'done' | null;
+      statusChangedToFailed: boolean;
+      failedCount: number;
+      totalCount: number;
     }
-
-    // Add admin members
-    for (const member of adminMembers) {
-      if (member.user?.id && member.user.email) {
-        recipientMap.set(member.user.id, {
-          id: member.user.id,
-          name: member.user.name?.trim() || member.user.email?.trim() || 'User',
-          email: member.user.email,
-        });
-      }
-    }
-
-    const recipients = Array.from(recipientMap.values());
-
-    // Send emails to each recipient
-    await Promise.allSettled(
-      recipients.map(async (recipient) => {
-        // Check if user is unsubscribed
-        const isUnsubscribed = await isUserUnsubscribed(
-          db,
-          recipient.email,
-          'taskAssignments',
-          organizationId,
-        );
-
-        if (isUnsubscribed) {
-          logger.info(
-            `Skipping notification: user ${recipient.email} is unsubscribed from task assignments`,
-          );
-          return;
-        }
-
-        try {
-          await triggerEmail({
-            to: recipient.email,
-            subject: `Task "${taskTitle}" status changed to ${newStatus}`,
-            react: TaskStatusChangedEmail({
-              toName: recipient.name,
-              toEmail: recipient.email,
-              taskTitle,
-              oldStatus: oldStatus.charAt(0).toUpperCase() + oldStatus.slice(1),
-              newStatus: newStatus === 'failed' ? 'Failed' : 'Done',
-              changedByName: 'Automation',
-              organizationName,
-              taskUrl,
-            }),
-            system: true,
-          });
-
-          logger.info(`Status change email sent to ${recipient.email}`);
-        } catch (error) {
-          logger.error(
-            `Failed to send status change email to ${recipient.email}`,
-            {
-              error: error instanceof Error ? error.message : 'Unknown error',
-            },
-          );
-        }
-      }),
-    );
-
-    logger.info(
-      `Sent ${recipients.length} status change notifications for task ${taskId} (status: ${newStatus})`,
-    );
-  } catch (error) {
-    logger.error('Failed to send task status change emails', {
-      error: error instanceof Error ? error.message : 'Unknown error',
-    });
-  }
-}
+  | { success: false; taskId?: string; error: string };
 
 /**
  * Worker task that runs integration checks for a single task.
  * Triggered by the orchestrator (integration-checks-schedule).
+ *
+ * This worker no longer sends a per-task status-change email. The orchestrator
+ * fans these out one run per task, so emailing here produced one email per
+ * failing task (spam). Instead the per-task run REPORTS the transition back to
+ * its per-org parent (run-org-integration-checks), which bundles every task
+ * that failed in the run into a single email per recipient.
  */
 export const runTaskIntegrationChecks = task({
   id: 'run-task-integration-checks',
@@ -195,7 +62,7 @@ export const runTaskIntegrationChecks = task({
     providerSlug: string;
     organizationId: string;
     checkIds: string[];
-  }) => {
+  }): Promise<TaskCheckRunResult> => {
     const {
       taskId,
       taskTitle,
@@ -560,6 +427,12 @@ export const runTaskIntegrationChecks = task({
         totalFindings,
       );
 
+      // Whether THIS run flipped the task into `failed` (e.g. todo/done →
+      // failed). The per-org runner bundles only these transitions into one
+      // email, so a task that was already failed isn't re-reported every run —
+      // preserving the previous "only notify on transition" behavior.
+      let statusChangedToFailed = false;
+
       if (newStatus === 'failed') {
         // Get current status before updating
         const taskBeforeUpdate = await db.task.findUnique({
@@ -576,18 +449,10 @@ export const runTaskIntegrationChecks = task({
           `Task ${taskId} marked as failed due to ${effectiveFailures} finding(s)`,
         );
 
-        // Only send email notifications if status actually changed
-        if (oldStatus !== 'failed') {
-          await sendTaskStatusChangeEmails({
-            organizationId,
-            taskId,
-            taskTitle,
-            oldStatus,
-            newStatus: 'failed',
-          });
-        } else {
+        statusChangedToFailed = oldStatus !== 'failed';
+        if (!statusChangedToFailed) {
           logger.info(
-            `Skipping notification: task ${taskId} was already in failed status`,
+            `Task ${taskId} was already in failed status; not reporting it for the bundled email`,
           );
         }
       } else if (newStatus === 'done') {
@@ -630,10 +495,17 @@ export const runTaskIntegrationChecks = task({
       return {
         success: true,
         taskId,
+        taskTitle,
         checksRun: effectiveCheckIds.length,
         totalPassing,
         totalFindings,
         taskStatus: newStatus,
+        // Consumed by the per-org bundled-failure email
+        // (run-org-integration-checks). `statusChangedToFailed` gates inclusion;
+        // failedCount/totalCount feed the "(X/Y failed)" line per task.
+        statusChangedToFailed,
+        failedCount: effectiveFailures,
+        totalCount: totalPassing + totalFindings,
       };
     } catch (error) {
       logger.error(`Failed to run checks for task ${taskId}`, {