diff --git a/apps/docs/content/docs/api-reference/(generated)/workflows/meta.json b/apps/docs/content/docs/api-reference/(generated)/workflows/meta.json
index 0416e441abe..4c6c32150a3 100644
--- a/apps/docs/content/docs/api-reference/(generated)/workflows/meta.json
+++ b/apps/docs/content/docs/api-reference/(generated)/workflows/meta.json
@@ -22,6 +22,8 @@
     "importWorkflow",
     "getWorkflowDeployment",
     "updateWorkflowPublicApi",
+    "listWorkflowTriggers",
+    "updateWorkflowTrigger",
     "deployWorkflow",
     "undeployWorkflow",
     "rollbackWorkflow",
diff --git a/apps/docs/content/docs/cli/reference.mdx b/apps/docs/content/docs/cli/reference.mdx
index 121b9b1557a..3b2b1a26de7 100644
--- a/apps/docs/content/docs/cli/reference.mdx
+++ b/apps/docs/content/docs/cli/reference.mdx
@@ -8228,6 +8228,57 @@ sim workflows list [options]
 
 API request and response schemas: [listWorkflows](/api-reference/workflows/listWorkflows).
 
+### sim workflows triggers list
+
+Show the health of every trigger the live deployment serves
+
+```bash
+sim workflows triggers list <workflowId>
+```
+
+**Arguments**
+
+<CommandTable>
+
+| Argument | Required | Description |
+| --- | --- | --- |
+| `workflowId` | Yes | Unique workflow identifier. |
+
+</CommandTable>
+
+API request and response schemas: [listWorkflowTriggers](/api-reference/workflows/listWorkflowTriggers).
+
+### sim workflows triggers update
+
+Enable or disable one trigger of the live deployment
+
+```bash
+sim workflows triggers update <workflowId> <blockId> [options]
+```
+
+**Arguments**
+
+<CommandTable>
+
+| Argument | Required | Description |
+| --- | --- | --- |
+| `workflowId` | Yes | Unique workflow identifier. |
+| `blockId` | Yes | Trigger block in the deployed workflow. |
+
+</CommandTable>
+
+**Options**
+
+<CommandTable>
+
+| Option | Required | Description |
+| --- | --- | --- |
+| `--enabled <true\|false>` | Yes | Whether the trigger should fire. A disable lasts until the next deploy or version activation, which re-enables every trigger. Accepted values: `true`, `false`. |
+
+</CommandTable>
+
+API request and response schemas: [updateWorkflowTrigger](/api-reference/workflows/updateWorkflowTrigger).
+
 ### sim workflows move
 
 Move workflows
diff --git a/apps/docs/content/docs/cli/workflows.mdx b/apps/docs/content/docs/cli/workflows.mdx
index 2a3c7e338d6..d0d274cc03c 100644
--- a/apps/docs/content/docs/cli/workflows.mdx
+++ b/apps/docs/content/docs/cli/workflows.mdx
@@ -431,6 +431,53 @@ sim workflows deployment update <workflowId> [options]
 
 API request and response schemas: [updateWorkflowPublicApi](/api-reference/workflows/updateWorkflowPublicApi).
 
+### Show the health of every trigger the live deployment serves
+
+```bash
+sim workflows triggers list <workflowId>
+```
+
+**Arguments**
+
+<CommandTable>
+
+| Argument | Required | Description |
+| --- | --- | --- |
+| `workflowId` | Yes | Unique workflow identifier. |
+
+</CommandTable>
+
+API request and response schemas: [listWorkflowTriggers](/api-reference/workflows/listWorkflowTriggers).
+
+### Enable or disable one trigger of the live deployment
+
+```bash
+sim workflows triggers update <workflowId> <blockId> [options]
+```
+
+**Arguments**
+
+<CommandTable>
+
+| Argument | Required | Description |
+| --- | --- | --- |
+| `workflowId` | Yes | Unique workflow identifier. |
+| `blockId` | Yes | Trigger block in the deployed workflow. |
+
+</CommandTable>
+
+**Options**
+
+<CommandTable>
+
+| Option | Required | Description |
+| --- | --- | --- |
+| `--enabled <true\|false>` | Yes | Whether the trigger should fire. A disable lasts until the next deploy or version activation, which re-enables every trigger. Accepted values: `true`, `false`. |
+
+</CommandTable>
+
+API request and response schemas: [updateWorkflowTrigger](/api-reference/workflows/updateWorkflowTrigger).
+
 ### List workflow versions
 
 ```bash
diff --git a/apps/docs/openapi-v2-workflows.json b/apps/docs/openapi-v2-workflows.json
index 404c5d208ce..b0d6a030674 100644
--- a/apps/docs/openapi-v2-workflows.json
+++ b/apps/docs/openapi-v2-workflows.json
@@ -2565,6 +2565,246 @@
         }
       }
     },
+    "/api/v2/workflows/{workflowId}/triggers": {
+      "get": {
+        "operationId": "listWorkflowTriggers",
+        "summary": "List workflow triggers",
+        "description": "List the schedules, polling triggers, internal-event triggers, and webhooks the live deployment serves, including disabled ones, with their failure counts. Requires workspace read access. Schedules and polling triggers disable themselves after 100 consecutive failures; a `webhook` trigger's failures happen at the sender, so its `failedCount` stays 0. Empty while the workflow is not deployed. Returns one page; `nextCursor` is always null.\n\nOAuth scope: `api:read`.",
+        "x-sim-operation": "workflows.triggers.read",
+        "x-oauth-scope": "api:read",
+        "tags": ["Workflows"],
+        "security": [
+          {
+            "apiKey": []
+          },
+          {
+            "oauthBearer": ["api:read"]
+          }
+        ],
+        "parameters": [
+          {
+            "name": "workflowId",
+            "in": "path",
+            "required": true,
+            "description": "Unique workflow identifier.",
+            "schema": {
+              "type": "string",
+              "minLength": 1,
+              "description": "Unique workflow identifier.",
+              "examples": ["3b1f7c92-8d4e-4a6b-9c0d-5e2f8a714b36"]
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "The live deployment's triggers.",
+            "headers": {
+              "X-RateLimit-Limit": {
+                "$ref": "#/components/headers/X-RateLimit-Limit"
+              },
+              "X-RateLimit-Remaining": {
+                "$ref": "#/components/headers/X-RateLimit-Remaining"
+              },
+              "X-RateLimit-Reset": {
+                "$ref": "#/components/headers/X-RateLimit-Reset"
+              }
+            },
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/WorkflowTriggerListResponse"
+                },
+                "examples": {
+                  "example1": {
+                    "summary": "Response example 1",
+                    "value": {
+                      "data": [
+                        {
+                          "blockId": "blk_01J8ZK3QW4M6X2R9T7B5C0V4",
+                          "kind": "schedule",
+                          "provider": null,
+                          "enabled": false,
+                          "status": "disabled",
+                          "failedCount": 100,
+                          "lastFailedAt": "2026-08-09T18:04:11.000Z",
+                          "nextRunAt": null,
+                          "lastRanAt": "2026-08-09T18:04:00.000Z",
+                          "cronExpression": "0 * * * *",
+                          "timezone": "UTC",
+                          "url": null
+                        }
+                      ],
+                      "nextCursor": null
+                    }
+                  }
+                }
+              }
+            }
+          },
+          "400": {
+            "$ref": "#/components/responses/BadRequest"
+          },
+          "401": {
+            "$ref": "#/components/responses/Unauthorized"
+          },
+          "403": {
+            "$ref": "#/components/responses/Forbidden"
+          },
+          "404": {
+            "$ref": "#/components/responses/NotFound"
+          },
+          "429": {
+            "$ref": "#/components/responses/RateLimited"
+          },
+          "500": {
+            "$ref": "#/components/responses/InternalError"
+          },
+          "503": {
+            "$ref": "#/components/responses/ServiceUnavailable"
+          }
+        }
+      }
+    },
+    "/api/v2/workflows/{workflowId}/triggers/{blockId}": {
+      "patch": {
+        "operationId": "updateWorkflowTrigger",
+        "summary": "Update workflow trigger",
+        "description": "Enable or disable one trigger of the live deployment without redeploying. Requires workspace write access. Enabling resets `failedCount`, schedules the next run, and clears any polling backoff; a schedule whose cron expression is invalid or has no future runs returns `400`. A permission group that blocks webhook triggers returns `403` when a member it governs enables a polling, internal-event, or webhook trigger; a workspace API key is not governed by permission groups, as with every other capability. Disabling a completed schedule marks it disabled. A block with several live rows switches them together. Requesting the current state changes nothing.\n\nDisabling only stops Sim from accepting the trigger; nothing is removed at the provider. A disable lasts until the next deploy or version activation, which re-enables every trigger, the same way it clears a trigger that disabled itself after repeated failures. Deliveries to a disabled webhook receive `404`, and some providers disable their subscription after repeated failures, which requires a redeploy to restore.\n\nOAuth scope: `api:write`.",
+        "x-sim-operation": "workflows.triggers.set_enabled",
+        "x-oauth-scope": "api:write",
+        "tags": ["Workflows"],
+        "security": [
+          {
+            "apiKey": []
+          },
+          {
+            "oauthBearer": ["api:write"]
+          }
+        ],
+        "parameters": [
+          {
+            "name": "workflowId",
+            "in": "path",
+            "required": true,
+            "description": "Unique workflow identifier.",
+            "schema": {
+              "type": "string",
+              "minLength": 1,
+              "description": "Unique workflow identifier.",
+              "examples": ["3b1f7c92-8d4e-4a6b-9c0d-5e2f8a714b36"]
+            }
+          },
+          {
+            "name": "blockId",
+            "in": "path",
+            "required": true,
+            "description": "Trigger block in the deployed workflow.",
+            "schema": {
+              "type": "string",
+              "minLength": 1,
+              "maxLength": 128,
+              "description": "Trigger block in the deployed workflow."
+            }
+          }
+        ],
+        "requestBody": {
+          "required": true,
+          "description": "Enable or disable one trigger of the live deployment.",
+          "content": {
+            "application/json": {
+              "schema": {
+                "$ref": "#/components/schemas/UpdateWorkflowTriggerRequest"
+              },
+              "examples": {
+                "example1": {
+                  "summary": "Request example 1",
+                  "value": {
+                    "enabled": true
+                  }
+                }
+              }
+            }
+          }
+        },
+        "responses": {
+          "200": {
+            "description": "The trigger after the update.",
+            "headers": {
+              "X-RateLimit-Limit": {
+                "$ref": "#/components/headers/X-RateLimit-Limit"
+              },
+              "X-RateLimit-Remaining": {
+                "$ref": "#/components/headers/X-RateLimit-Remaining"
+              },
+              "X-RateLimit-Reset": {
+                "$ref": "#/components/headers/X-RateLimit-Reset"
+              }
+            },
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/UpdateWorkflowTriggerResponse"
+                },
+                "examples": {
+                  "example1": {
+                    "summary": "Response example 1",
+                    "value": {
+                      "data": {
+                        "blockId": "blk_01J8ZK3QW4M6X2R9T7B5C0V4",
+                        "kind": "schedule",
+                        "provider": null,
+                        "enabled": true,
+                        "status": "active",
+                        "failedCount": 0,
+                        "lastFailedAt": "2026-08-09T18:04:11.000Z",
+                        "nextRunAt": "2026-08-10T09:00:00.000Z",
+                        "lastRanAt": "2026-08-09T18:04:00.000Z",
+                        "cronExpression": "0 * * * *",
+                        "timezone": "UTC",
+                        "url": null
+                      }
+                    }
+                  }
+                }
+              }
+            }
+          },
+          "400": {
+            "$ref": "#/components/responses/BadRequest"
+          },
+          "401": {
+            "$ref": "#/components/responses/Unauthorized"
+          },
+          "403": {
+            "$ref": "#/components/responses/Forbidden"
+          },
+          "404": {
+            "$ref": "#/components/responses/NotFound"
+          },
+          "409": {
+            "$ref": "#/components/responses/Conflict"
+          },
+          "413": {
+            "$ref": "#/components/responses/PayloadTooLarge"
+          },
+          "415": {
+            "$ref": "#/components/responses/UnsupportedMediaType"
+          },
+          "423": {
+            "$ref": "#/components/responses/Locked"
+          },
+          "429": {
+            "$ref": "#/components/responses/RateLimited"
+          },
+          "500": {
+            "$ref": "#/components/responses/InternalError"
+          },
+          "503": {
+            "$ref": "#/components/responses/ServiceUnavailable"
+          }
+        }
+      }
+    },
     "/api/v2/workflows/{workflowId}/deploy": {
       "post": {
         "operationId": "deployWorkflow",
@@ -12581,6 +12821,230 @@
           }
         ]
       },
+      "WorkflowTrigger": {
+        "type": "object",
+        "properties": {
+          "blockId": {
+            "type": "string",
+            "description": "Trigger block in the deployed workflow."
+          },
+          "kind": {
+            "type": "string",
+            "enum": ["schedule", "polling", "event", "webhook"],
+            "description": "How the trigger fires: `schedule` on a cron, `polling` when Sim polls the source, `event` on an event raised inside Sim, `webhook` when an external sender delivers to its URL."
+          },
+          "provider": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "Trigger provider, e.g. `gmail` or `github`; null for schedules."
+          },
+          "enabled": {
+            "type": "boolean",
+            "description": "Whether the trigger currently fires."
+          },
+          "status": {
+            "type": "string",
+            "enum": ["active", "disabled", "completed"],
+            "description": "`disabled` after a manual disable or an automatic one following repeated failures. `completed` applies only to schedules that have no runs left."
+          },
+          "failedCount": {
+            "type": "integer",
+            "minimum": 0,
+            "maximum": 9007199254740991,
+            "description": "Consecutive failures since the last success; the trigger disables itself at 100. Always 0 for `webhook` and `event` triggers."
+          },
+          "lastFailedAt": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "ISO 8601 timestamp of the most recent counted failure, or null.",
+            "format": "date-time"
+          },
+          "nextRunAt": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "ISO 8601 timestamp of the next scheduled run; null when disabled and for non-schedules.",
+            "format": "date-time"
+          },
+          "lastRanAt": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "ISO 8601 timestamp of the last scheduled run; null for non-schedules.",
+            "format": "date-time"
+          },
+          "cronExpression": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "Cron expression of a schedule; null for non-schedules."
+          },
+          "timezone": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "IANA timezone the cron expression runs in; null for non-schedules."
+          },
+          "url": {
+            "anyOf": [
+              {
+                "type": "string",
+                "format": "uri"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "Delivery URL of a `webhook` trigger; null for other kinds and for providers on a shared endpoint."
+          }
+        },
+        "required": [
+          "blockId",
+          "kind",
+          "provider",
+          "enabled",
+          "status",
+          "failedCount",
+          "lastFailedAt",
+          "nextRunAt",
+          "lastRanAt",
+          "cronExpression",
+          "timezone",
+          "url"
+        ],
+        "additionalProperties": false,
+        "title": "Workflow trigger",
+        "description": "Health and enablement of one trigger the live deployment serves."
+      },
+      "WorkflowTriggerListResponse": {
+        "type": "object",
+        "properties": {
+          "data": {
+            "type": "array",
+            "items": {
+              "$ref": "#/components/schemas/WorkflowTrigger"
+            },
+            "description": "Items in the current page."
+          },
+          "nextCursor": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "Always `null`. This list has no `cursor` or `limit` parameters; see the operation for inventory limits."
+          }
+        },
+        "required": ["data", "nextCursor"],
+        "additionalProperties": false,
+        "title": "Workflow trigger list response",
+        "description": "Every trigger the live deployment serves.",
+        "examples": [
+          {
+            "data": [
+              {
+                "blockId": "blk_01J8ZK3QW4M6X2R9T7B5C0V4",
+                "kind": "schedule",
+                "provider": null,
+                "enabled": false,
+                "status": "disabled",
+                "failedCount": 100,
+                "lastFailedAt": "2026-08-09T18:04:11.000Z",
+                "nextRunAt": null,
+                "lastRanAt": "2026-08-09T18:04:00.000Z",
+                "cronExpression": "0 * * * *",
+                "timezone": "UTC",
+                "url": null
+              }
+            ],
+            "nextCursor": null
+          }
+        ]
+      },
+      "UpdateWorkflowTriggerResponse": {
+        "type": "object",
+        "properties": {
+          "data": {
+            "description": "Response data.",
+            "$ref": "#/components/schemas/WorkflowTrigger"
+          }
+        },
+        "required": ["data"],
+        "additionalProperties": false,
+        "title": "Update workflow trigger response",
+        "description": "The trigger after the update.",
+        "examples": [
+          {
+            "data": {
+              "blockId": "blk_01J8ZK3QW4M6X2R9T7B5C0V4",
+              "kind": "schedule",
+              "provider": null,
+              "enabled": true,
+              "status": "active",
+              "failedCount": 0,
+              "lastFailedAt": "2026-08-09T18:04:11.000Z",
+              "nextRunAt": "2026-08-10T09:00:00.000Z",
+              "lastRanAt": "2026-08-09T18:04:00.000Z",
+              "cronExpression": "0 * * * *",
+              "timezone": "UTC",
+              "url": null
+            }
+          }
+        ]
+      },
+      "UpdateWorkflowTriggerRequest": {
+        "type": "object",
+        "properties": {
+          "enabled": {
+            "type": "boolean",
+            "description": "Whether the trigger should fire. A disable lasts until the next deploy or version activation, which re-enables every trigger."
+          }
+        },
+        "required": ["enabled"],
+        "additionalProperties": false,
+        "title": "Update workflow trigger request",
+        "description": "Enable or disable one trigger of the live deployment.",
+        "examples": [
+          {
+            "enabled": true
+          }
+        ]
+      },
       "DeployResult": {
         "type": "object",
         "properties": {
diff --git a/apps/docs/scripts/fixtures/content-baseline.json b/apps/docs/scripts/fixtures/content-baseline.json
index f6563814165..94df7334dd3 100644
--- a/apps/docs/scripts/fixtures/content-baseline.json
+++ b/apps/docs/scripts/fixtures/content-baseline.json
@@ -12986,6 +12986,20 @@
           "bytes": 28,
           "line": 8128
         },
+        {
+          "lang": "bash",
+          "meta": null,
+          "sha256": "7579bd6ac197afa0afa5f62d9b412a7e265725bd507c1c63b438a60fd72ce1d0",
+          "bytes": 40,
+          "line": 8235
+        },
+        {
+          "lang": "bash",
+          "meta": null,
+          "sha256": "63aac11606dc0eeb6cc452c20c405806840960425372cf2cf7c8b4b90a7de878",
+          "bytes": 62,
+          "line": 8255
+        },
         {
           "lang": "bash",
           "meta": null,
@@ -17173,6 +17187,20 @@
           "bytes": 54,
           "line": 404
         },
+        {
+          "lang": "bash",
+          "meta": null,
+          "sha256": "7579bd6ac197afa0afa5f62d9b412a7e265725bd507c1c63b438a60fd72ce1d0",
+          "bytes": 40,
+          "line": 436
+        },
+        {
+          "lang": "bash",
+          "meta": null,
+          "sha256": "63aac11606dc0eeb6cc452c20c405806840960425372cf2cf7c8b4b90a7de878",
+          "bytes": 62,
+          "line": 454
+        },
         {
           "lang": "bash",
           "meta": null,
diff --git a/apps/sim/app/api/schedules/[id]/route.test.ts b/apps/sim/app/api/schedules/[id]/route.test.ts
deleted file mode 100644
index 82b401a1f2c..00000000000
--- a/apps/sim/app/api/schedules/[id]/route.test.ts
+++ /dev/null
@@ -1,126 +0,0 @@
-/**
- * Tests for schedule reactivate PUT API route
- */
-import {
-  auditMock,
-  authMockFns,
-  databaseMock,
-  workflowAuthzMockFns,
-  workflowsUtilsMock,
-} from '@sim/testing'
-import { NextRequest } from 'next/server'
-import { beforeEach, describe, expect, it, vi } from 'vitest'
-
-vi.mock('@/lib/workflows/utils', () => workflowsUtilsMock)
-
-vi.mock('@sim/audit', () => auditMock)
-
-import { PUT } from './route'
-
-function createRequest(body: Record<string, unknown>): NextRequest {
-  return new NextRequest(new URL('http://test/api/schedules/sched-1'), {
-    method: 'PUT',
-    body: JSON.stringify(body),
-    headers: { 'Content-Type': 'application/json' },
-  })
-}
-
-function createParams(id: string): { params: Promise<{ id: string }> } {
-  return { params: Promise.resolve({ id }) }
-}
-
-const mockDbSelect = databaseMock.db.select as ReturnType<typeof vi.fn>
-const mockDbUpdate = databaseMock.db.update as ReturnType<typeof vi.fn>
-
-function mockDbChain(selectResults: unknown[][]) {
-  let selectCallIndex = 0
-  mockDbSelect.mockImplementation(() => ({
-    from: () => ({
-      where: () => ({
-        limit: () => selectResults[selectCallIndex++] || [],
-      }),
-    }),
-  }))
-
-  mockDbUpdate.mockImplementation(() => ({
-    set: () => ({
-      where: vi.fn().mockResolvedValue({}),
-    }),
-  }))
-}
-
-describe('Schedule PUT API (Reactivate)', () => {
-  beforeEach(() => {
-    authMockFns.mockGetSession.mockResolvedValue({ user: { id: 'user-1' } })
-    workflowAuthzMockFns.mockAuthorizeWorkflowByWorkspacePermission.mockResolvedValue({
-      allowed: true,
-      status: 200,
-      workflow: { id: 'wf-1', workspaceId: 'ws-1' },
-      workspacePermission: 'write',
-    })
-  })
-
-  describe('Authorization', () => {
-    it('returns 403 when user is not workflow owner', async () => {
-      workflowAuthzMockFns.mockAuthorizeWorkflowByWorkspacePermission.mockResolvedValue({
-        allowed: false,
-        status: 403,
-        workflow: { id: 'wf-1', workspaceId: null },
-        workspacePermission: null,
-        message:
-          'This workflow is not attached to a workspace. Personal workflows are deprecated and cannot be accessed.',
-      })
-      mockDbChain([
-        [{ id: 'sched-1', workflowId: 'wf-1', status: 'disabled' }],
-        [{ userId: 'other-user', workspaceId: null }],
-      ])
-
-      const res = await PUT(createRequest({ action: 'reactivate' }), createParams('sched-1'))
-
-      expect(res.status).toBe(403)
-      const data = await res.json()
-      expect(data.error).toContain('Personal workflows are deprecated')
-    })
-
-    it('returns 403 for workspace member with only read permission', async () => {
-      workflowAuthzMockFns.mockAuthorizeWorkflowByWorkspacePermission.mockResolvedValue({
-        allowed: false,
-        status: 403,
-        workflow: { id: 'wf-1', workspaceId: 'ws-1' },
-        workspacePermission: 'read',
-        message: 'Unauthorized: Access denied to write this workflow',
-      })
-      mockDbChain([
-        [{ id: 'sched-1', workflowId: 'wf-1', status: 'disabled' }],
-        [{ userId: 'other-user', workspaceId: 'ws-1' }],
-      ])
-
-      const res = await PUT(createRequest({ action: 'reactivate' }), createParams('sched-1'))
-
-      expect(res.status).toBe(403)
-    })
-  })
-
-  describe('Schedule State Handling', () => {
-    it('returns 400 when schedule has invalid cron expression', async () => {
-      mockDbChain([
-        [
-          {
-            id: 'sched-1',
-            workflowId: 'wf-1',
-            status: 'disabled',
-            cronExpression: 'invalid-cron',
-            timezone: 'UTC',
-          },
-        ],
-        [{ userId: 'user-1', workspaceId: null }],
-      ])
-
-      const res = await PUT(createRequest({ action: 'reactivate' }), createParams('sched-1'))
-
-      expect(res.status).toBe(400)
-      const data = await res.json()
-      expect(data.error).toBe('Schedule has invalid cron expression')
-    })
-  })
-})
diff --git a/apps/sim/app/api/schedules/[id]/route.ts b/apps/sim/app/api/schedules/[id]/route.ts
index 2aa958fe13d..6fc6bca6206 100644
--- a/apps/sim/app/api/schedules/[id]/route.ts
+++ b/apps/sim/app/api/schedules/[id]/route.ts
@@ -2,19 +2,26 @@ import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
 import { db } from '@sim/db'
 import { workflowSchedule } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import {
-  assertWorkflowMutable,
-  authorizeWorkflowByWorkspacePermission,
-  WorkflowLockedError,
-} from '@sim/platform-authz/workflow'
+import { authorizeWorkflowByWorkspacePermission } from '@sim/platform-authz/workflow'
 import { and, eq, isNull } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { getScheduleByIdContract, updateScheduleContract } from '@/lib/api/contracts/schedules'
 import { parseRequest } from '@/lib/api/server'
+import {
+  createInternalResourceConcealmentPolicy,
+  defineInternalJsonRoute,
+  extendInternalErrorPolicy,
+  internalErrorResponse,
+  internalOrchestrationErrorPolicy,
+  internalRateLimits,
+  internalSessionAuth,
+} from '@/lib/api/server/routes'
 import { getSession } from '@/lib/auth'
+import { InsufficientWorkspacePermissionsError } from '@/lib/core/application'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { validateCronExpression } from '@/lib/workflows/schedules/utils'
+import { workflowOperations } from '@/lib/workflows/application/operations'
+import { setWorkflowTriggerEnabled } from '@/lib/workflows/application/triggers'
 
 const logger = createLogger('ScheduleAPI')
 
@@ -97,124 +104,57 @@ export const GET = withRouteHandler(
   }
 )
 
-export const PUT = withRouteHandler(
-  async (request: NextRequest, context: { params: Promise<{ id: string }> }) => {
-    const requestId = generateRequestId()
-
-    try {
-      const session = await getSession()
-      if (!session?.user?.id) {
-        logger.warn('Unauthorized schedule update attempt')
-        return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
-      }
-
-      const parsed = await parseRequest(updateScheduleContract, request, context, {
-        validationErrorResponse: () =>
-          NextResponse.json({ error: 'Invalid request body' }, { status: 400 }),
-      })
-      if (!parsed.success) return parsed.response
-
-      const { id: scheduleId } = parsed.data.params
-      const validatedBody = parsed.data.body
-
-      const result = await fetchAndAuthorize(requestId, scheduleId, session.user.id, 'write')
-      if (result instanceof NextResponse) return result
-      const { schedule, workspaceId } = result
-      if (schedule.workflowId) {
-        await assertWorkflowMutable(schedule.workflowId)
+/**
+ * Disables or reactivates a workflow schedule through the same use case as
+ * `PATCH /api/v2/workflows/{workflowId}/triggers/{blockId}`. The id must name
+ * the live deployment's row: a schedule of an older version answers 404
+ * instead of being revived.
+ */
+export const PUT = defineInternalJsonRoute({
+  contract: updateScheduleContract,
+  auth: internalSessionAuth,
+  operation: workflowOperations.setTriggerEnabled,
+  rateLimit: internalRateLimits.none({
+    reason: 'Schedule enablement retains its existing session admission policy.',
+  }),
+  errorPolicy: createInternalResourceConcealmentPolicy({
+    notFoundMessage: 'Schedule not found',
+    base: extendInternalErrorPolicy(
+      {
+        project: internalOrchestrationErrorPolicy.project,
+        unhandled: () => internalErrorResponse(500, { error: 'Failed to update schedule' }),
+      },
+      (error) =>
+        error instanceof InsufficientWorkspacePermissionsError
+          ? internalErrorResponse(403, {
+              error: 'Unauthorized: Access denied to write this workflow',
+            })
+          : null
+    ),
+  }),
+  parseOptions: {
+    validationErrorResponse: () =>
+      NextResponse.json({ error: 'Invalid request body' }, { status: 400 }),
+  },
+  mapInput: ({ params, body }) => ({
+    target: { kind: 'row' as const, table: 'schedule' as const, id: params.id },
+    enabled: body.action === 'reactivate',
+  }),
+  useCase: setWorkflowTriggerEnabled,
+  present: ({ trigger, changes }, { input }) => {
+    const changed = changes.length > 0
+    if (!input.enabled) {
+      return {
+        message: changed ? 'Schedule disabled successfully' : 'Schedule is already disabled',
       }
-
-      const { action } = validatedBody
-
-      if (action === 'disable') {
-        if (schedule.status === 'disabled') {
-          return NextResponse.json({ message: 'Schedule is already disabled' })
-        }
-
-        await db
-          .update(workflowSchedule)
-          .set({ status: 'disabled', nextRunAt: null, lastQueuedAt: null, updatedAt: new Date() })
-          .where(and(eq(workflowSchedule.id, scheduleId), isNull(workflowSchedule.archivedAt)))
-
-        logger.info(`Disabled schedule: ${scheduleId}`)
-
-        recordAudit({
-          workspaceId,
-          actorId: session.user.id,
-          actorName: session.user.name,
-          actorEmail: session.user.email,
-          action: AuditAction.SCHEDULE_UPDATED,
-          resourceType: AuditResourceType.SCHEDULE,
-          resourceId: scheduleId,
-          resourceName: schedule.jobTitle ?? undefined,
-          description: `Disabled schedule "${schedule.jobTitle ?? scheduleId}"`,
-          metadata: {
-            operation: 'disable',
-            sourceType: schedule.sourceType,
-            previousStatus: schedule.status,
-          },
-          request,
-        })
-
-        return NextResponse.json({ message: 'Schedule disabled successfully' })
-      }
-
-      // reactivate
-      if (schedule.status === 'active') {
-        return NextResponse.json({ message: 'Schedule is already active' })
-      }
-
-      if (!schedule.cronExpression) {
-        logger.error(`Schedule has no cron expression: ${scheduleId}`)
-        return NextResponse.json({ error: 'Schedule has no cron expression' }, { status: 400 })
-      }
-
-      const cronResult = validateCronExpression(schedule.cronExpression, schedule.timezone || 'UTC')
-      if (!cronResult.isValid || !cronResult.nextRun) {
-        logger.error(`Invalid cron expression for schedule: ${scheduleId}`)
-        return NextResponse.json({ error: 'Schedule has invalid cron expression' }, { status: 400 })
-      }
-
-      const now = new Date()
-      const nextRunAt = cronResult.nextRun
-
-      await db
-        .update(workflowSchedule)
-        .set({ status: 'active', failedCount: 0, infraRetryCount: 0, updatedAt: now, nextRunAt })
-        .where(and(eq(workflowSchedule.id, scheduleId), isNull(workflowSchedule.archivedAt)))
-
-      logger.info(`Reactivated schedule: ${scheduleId}`)
-
-      recordAudit({
-        workspaceId,
-        actorId: session.user.id,
-        actorName: session.user.name,
-        actorEmail: session.user.email,
-        action: AuditAction.SCHEDULE_UPDATED,
-        resourceType: AuditResourceType.SCHEDULE,
-        resourceId: scheduleId,
-        resourceName: schedule.jobTitle ?? undefined,
-        description: `Reactivated schedule "${schedule.jobTitle ?? scheduleId}"`,
-        metadata: {
-          operation: 'reactivate',
-          sourceType: schedule.sourceType,
-          cronExpression: schedule.cronExpression,
-          timezone: schedule.timezone,
-        },
-        request,
-      })
-
-      return NextResponse.json({ message: 'Schedule activated successfully', nextRunAt })
-    } catch (error) {
-      if (error instanceof WorkflowLockedError) {
-        return NextResponse.json({ error: error.message }, { status: error.status })
-      }
-
-      logger.error('Error updating schedule', error)
-      return NextResponse.json({ error: 'Failed to update schedule' }, { status: 500 })
     }
-  }
-)
+    if (!changed) return { message: 'Schedule is already active' }
+    return {
+      message: 'Schedule activated successfully',
+      ...(trigger.nextRunAt ? { nextRunAt: trigger.nextRunAt.toISOString() } : {}),
+    }
+  },
+})
 
 export const DELETE = withRouteHandler(
   async (request: NextRequest, { params }: { params: Promise<{ id: string }> }) => {
diff --git a/apps/sim/app/api/v2/workflows/[workflowId]/triggers/[blockId]/route.ts b/apps/sim/app/api/v2/workflows/[workflowId]/triggers/[blockId]/route.ts
new file mode 100644
index 00000000000..00ca9b5c7d0
--- /dev/null
+++ b/apps/sim/app/api/v2/workflows/[workflowId]/triggers/[blockId]/route.ts
@@ -0,0 +1,26 @@
+import { v2UpdateWorkflowTriggerContract } from '@/lib/api/contracts/v2/workflows'
+import { defineV2JsonRoute, v2ApiKeyAuth, v2RateLimits } from '@/lib/api/server/routes'
+import { presentV2WorkflowTrigger, v2WorkflowErrorPolicies } from '@/lib/workflows/api'
+import { workflowOperations } from '@/lib/workflows/application/operations'
+import { setWorkflowTriggerEnabled } from '@/lib/workflows/application/triggers'
+
+export const dynamic = 'force-dynamic'
+export const revalidate = 0
+
+/**
+ * PATCH /api/v2/workflows/[workflowId]/triggers/[blockId] — enable or disable
+ * one trigger of the live deployment without redeploying.
+ */
+export const PATCH = defineV2JsonRoute({
+  contract: v2UpdateWorkflowTriggerContract,
+  auth: v2ApiKeyAuth,
+  operation: workflowOperations.setTriggerEnabled,
+  rateLimit: v2RateLimits.publicApi,
+  errorPolicy: v2WorkflowErrorPolicies.concealWorkflowAuthorization,
+  mapInput: ({ params, body }) => ({
+    target: { kind: 'block' as const, workflowId: params.workflowId, blockId: params.blockId },
+    enabled: body.enabled,
+  }),
+  useCase: setWorkflowTriggerEnabled,
+  present: ({ trigger }) => ({ data: presentV2WorkflowTrigger(trigger) }),
+})
diff --git a/apps/sim/app/api/v2/workflows/[workflowId]/triggers/route.ts b/apps/sim/app/api/v2/workflows/[workflowId]/triggers/route.ts
new file mode 100644
index 00000000000..4ed3513fd75
--- /dev/null
+++ b/apps/sim/app/api/v2/workflows/[workflowId]/triggers/route.ts
@@ -0,0 +1,30 @@
+import { v2ListWorkflowTriggersContract } from '@/lib/api/contracts/v2/workflows'
+import { defineV2JsonRoute, v2ApiKeyAuth, v2RateLimits } from '@/lib/api/server/routes'
+import { presentV2WorkflowTrigger, v2WorkflowErrorPolicies } from '@/lib/workflows/api'
+import { workflowOperations } from '@/lib/workflows/application/operations'
+import { listWorkflowTriggers } from '@/lib/workflows/application/triggers'
+
+export const dynamic = 'force-dynamic'
+export const revalidate = 0
+
+/**
+ * GET /api/v2/workflows/[workflowId]/triggers — health of every trigger the
+ * live deployment serves.
+ *
+ * Disabled triggers are listed, because finding the ones that switched
+ * themselves off after repeated failures is what this read is for; the
+ * deployment read publishes only the webhooks currently accepting deliveries.
+ */
+export const GET = defineV2JsonRoute({
+  contract: v2ListWorkflowTriggersContract,
+  auth: v2ApiKeyAuth,
+  operation: workflowOperations.readTriggers,
+  rateLimit: v2RateLimits.publicApi,
+  errorPolicy: v2WorkflowErrorPolicies.concealWorkflowAuthorization,
+  mapInput: ({ params }) => ({ workflowId: params.workflowId }),
+  useCase: listWorkflowTriggers,
+  present: ({ triggers }) => ({
+    data: triggers.map(presentV2WorkflowTrigger),
+    nextCursor: null,
+  }),
+})
diff --git a/apps/sim/app/api/webhooks/[id]/reactivation-gate.test.ts b/apps/sim/app/api/webhooks/[id]/reactivation-gate.test.ts
deleted file mode 100644
index 9703530e464..00000000000
--- a/apps/sim/app/api/webhooks/[id]/reactivation-gate.test.ts
+++ /dev/null
@@ -1,91 +0,0 @@
-import { webhook } from '@sim/db/schema'
-import {
-  auditMock,
-  createMockRequest,
-  hybridAuthMockFns,
-  permissionGroupScopeMock,
-  permissionGroupScopeMockFns,
-  posthogServerMock,
-  queueTableRows,
-  resetDbChainMock,
-  telemetryMock,
-  workflowAuthzMockFns,
-} from '@sim/testing'
-import { beforeEach, describe, expect, it, vi } from 'vitest'
-
-vi.mock('@sim/audit', () => auditMock)
-vi.mock('@/lib/permission-groups/config-scope.server', () => permissionGroupScopeMock)
-vi.mock('@/lib/core/telemetry', () => telemetryMock)
-vi.mock('@/lib/posthog/server', () => posthogServerMock)
-vi.mock('@/lib/webhooks/provider-subscriptions', () => ({ cleanupExternalWebhook: vi.fn() }))
-
-import { DEFAULT_PERMISSION_GROUP_CONFIG } from '@/lib/permission-groups/fields'
-import { PATCH } from '@/app/api/webhooks/[id]/route'
-
-const ACTOR_ID = 'actor-1'
-
-function reactivate() {
-  return PATCH(createMockRequest('PATCH', { isActive: true }), {
-    params: Promise.resolve({ id: 'webhook-1' }),
-  })
-}
-
-/** The single joined read the PATCH handler issues. */
-function queueDormantWebhook(): void {
-  queueTableRows(webhook, [
-    {
-      webhook: { id: 'webhook-1', isActive: false, failedCount: 0 },
-      workflow: { id: 'workflow-1', userId: ACTOR_ID, workspaceId: 'workspace-1' },
-    },
-  ])
-}
-
-/**
- * `triggers.webhook` is withheld from the actor's group. Flipping a dormant
- * webhook back on is the act the key names, so a session belonging to that
- * person must be refused — and an executor delegation carrying the same id
- * must not be, since it holds the actor's role and none of their capabilities.
- */
-describe('the subject the webhook reactivation gate is decided about', () => {
-  beforeEach(() => {
-    resetDbChainMock()
-    workflowAuthzMockFns.mockAuthorizeWorkflowByWorkspacePermission.mockResolvedValue({
-      allowed: true,
-      status: 200,
-      workflow: { id: 'workflow-1' },
-      workspacePermission: 'write',
-    })
-    workflowAuthzMockFns.mockAssertWorkflowMutable.mockResolvedValue(undefined)
-    permissionGroupScopeMockFns.mockResolvePermissionGroupConfig.mockResolvedValue({
-      ...DEFAULT_PERMISSION_GROUP_CONFIG,
-      disableWebhookTriggers: true,
-    })
-  })
-
-  it('refuses the actor’s own session', async () => {
-    hybridAuthMockFns.mockCheckSessionOrInternalAuth.mockResolvedValue({
-      success: true,
-      userId: ACTOR_ID,
-      authType: 'session',
-    })
-    queueDormantWebhook()
-
-    const response = await reactivate()
-
-    expect(response.status).toBe(403)
-  })
-
-  it('lets an internal executor JWT through without consulting that group', async () => {
-    hybridAuthMockFns.mockCheckSessionOrInternalAuth.mockResolvedValue({
-      success: true,
-      userId: ACTOR_ID,
-      authType: 'internal_jwt',
-    })
-    queueDormantWebhook()
-
-    const response = await reactivate()
-
-    expect(response.status).toBe(200)
-    expect(permissionGroupScopeMockFns.mockResolvePermissionGroupConfig).not.toHaveBeenCalled()
-  })
-})
diff --git a/apps/sim/app/api/webhooks/[id]/route.ts b/apps/sim/app/api/webhooks/[id]/route.ts
index 41803bcc8f3..68e03fa7939 100644
--- a/apps/sim/app/api/webhooks/[id]/route.ts
+++ b/apps/sim/app/api/webhooks/[id]/route.ts
@@ -7,6 +7,7 @@ import {
   authorizeWorkflowByWorkspacePermission,
   WorkflowLockedError,
 } from '@sim/platform-authz/workflow'
+import { isRecordLike } from '@sim/utils/object'
 import { and, eq, isNull } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import {
@@ -15,16 +16,24 @@ import {
   updateWebhookContract,
 } from '@/lib/api/contracts/webhooks'
 import { parseRequest } from '@/lib/api/server'
-import { capabilityGovernedAuthUserId, checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
+import {
+  createInternalResourceConcealmentPolicy,
+  defineInternalJsonRoute,
+  extendInternalErrorPolicy,
+  internalErrorResponse,
+  internalOrchestrationErrorPolicy,
+  internalRateLimits,
+} from '@/lib/api/server/routes'
+import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
+import { InsufficientWorkspacePermissionsError } from '@/lib/core/application'
 import { PlatformEvents } from '@/lib/core/telemetry'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import {
-  capabilityRefusal,
-  isWorkspaceCapabilityWithheld,
-} from '@/lib/permission-groups/capability-assertions'
 import { captureServerEvent } from '@/lib/posthog/server'
 import { cleanupExternalWebhook } from '@/lib/webhooks/provider-subscriptions'
+import { internalWorkflowSessionOrExecutorAuth } from '@/lib/workflows/api'
+import { workflowOperations } from '@/lib/workflows/application/operations'
+import { setWorkflowTriggerEnabled } from '@/lib/workflows/application/triggers'
 
 const logger = createLogger('WebhookAPI')
 
@@ -88,113 +97,47 @@ export const GET = withRouteHandler(
   }
 )
 
-export const PATCH = withRouteHandler(
-  async (request: NextRequest, context: { params: Promise<{ id: string }> }) => {
-    try {
-      const parsed = await parseRequest(updateWebhookContract, request, context)
-      if (!parsed.success) return parsed.response
-      const { id } = parsed.data.params
-
-      const auth = await checkSessionOrInternalAuth(request, { requireWorkflowId: false })
-      if (!auth.success || !auth.userId) {
-        logger.warn('Unauthorized webhook update attempt')
-        return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
-      }
-      const userId = auth.userId
-
-      const { isActive, failedCount } = parsed.data.body
-
-      const webhooks = await db
-        .select({
-          webhook: webhook,
-          workflow: {
-            id: workflow.id,
-            userId: workflow.userId,
-            workspaceId: workflow.workspaceId,
-          },
-        })
-        .from(webhook)
-        .innerJoin(workflow, eq(webhook.workflowId, workflow.id))
-        .where(and(eq(webhook.id, id), isNull(webhook.archivedAt)))
-        .limit(1)
-
-      if (webhooks.length === 0) {
-        logger.warn(`Webhook not found: ${id}`)
-        return NextResponse.json({ error: 'Webhook not found' }, { status: 404 })
-      }
-
-      const webhookData = webhooks[0]
-      const authorization = await authorizeWorkflowByWorkspacePermission({
-        workflowId: webhookData.workflow.id,
-        userId,
-        action: 'write',
-      })
-      const canModify = authorization.allowed
-
-      if (!canModify) {
-        logger.warn(`User ${userId} denied permission to modify webhook: ${id}`)
-        return NextResponse.json({ error: 'Access denied' }, { status: 403 })
-      }
-      await assertWorkflowMutable(webhookData.workflow.id)
-
-      const setClause: Partial<typeof webhook.$inferInsert> = {}
-      if (isActive !== undefined && isActive !== webhooks[0].webhook.isActive) {
-        /**
-         * permission-group-enforced: triggers.webhook — reactivating is the act
-         * the key names, "prevent making a workflow reachable from an inbound
-         * webhook", so gating creation alone left it reachable by flipping a
-         * dormant webhook back on. Only this direction: deactivating must stay
-         * open, or a policy change would strand a member with a live webhook
-         * they cannot turn off.
-         *
-         * Keyed to the governed subject rather than `auth.userId`: an internal
-         * executor JWT embeds the run's actor, and gating on it would apply that
-         * person's capabilities to a delegation that carries only their role.
-         */
-        const governedUserId = capabilityGovernedAuthUserId(auth)
-        if (isActive && governedUserId) {
-          const withheld = await isWorkspaceCapabilityWithheld(
-            governedUserId,
-            webhooks[0].workflow.workspaceId ?? '',
-            'triggers.webhook'
-          )
-          if (withheld) {
-            return NextResponse.json(
-              { error: capabilityRefusal('triggers.webhook') },
-              { status: 403 }
-            )
-          }
-        }
-        setClause.isActive = isActive
-      }
-      if (failedCount !== undefined && failedCount !== webhooks[0].webhook.failedCount) {
-        setClause.failedCount = failedCount
-      }
-
-      if (Object.keys(setClause).length === 0) {
-        logger.info(`No-op webhook PATCH (no field changes): ${id}`)
-        return NextResponse.json({ webhook: webhooks[0].webhook }, { status: 200 })
-      }
-
-      setClause.updatedAt = new Date()
-      const updatedWebhook = await db
-        .update(webhook)
-        .set(setClause)
-        .where(eq(webhook.id, id))
-        .returning()
-
-      logger.info(`Successfully updated webhook: ${id}`)
-      return NextResponse.json({ webhook: updatedWebhook[0] }, { status: 200 })
-    } catch (error) {
-      if (error instanceof WorkflowLockedError) {
-        return NextResponse.json({ error: error.message }, { status: error.status })
-      }
-
-      logger.error('Error updating webhook', error)
-      return NextResponse.json({ error: 'Internal server error' }, { status: 500 })
+/**
+ * Pauses or resumes a webhook trigger through the same use case as
+ * `PATCH /api/v2/workflows/{workflowId}/triggers/{blockId}`, which also owns the
+ * `triggers.webhook` gate on re-enabling. The id must name the live
+ * deployment's row: a candidate, retired, or older-version registration answers
+ * 404 instead of being revived.
+ * Admits a session or a workflow run's executor delegation; the executor is not
+ * judged against the triggering member's permission group.
+ */
+export const PATCH = defineInternalJsonRoute({
+  contract: updateWebhookContract,
+  auth: internalWorkflowSessionOrExecutorAuth,
+  operation: workflowOperations.setTriggerEnabled,
+  rateLimit: internalRateLimits.none({
+    reason: 'Webhook enablement retains its existing session admission policy.',
+  }),
+  errorPolicy: createInternalResourceConcealmentPolicy({
+    notFoundMessage: 'Webhook not found',
+    base: extendInternalErrorPolicy(internalOrchestrationErrorPolicy, (error) =>
+      error instanceof InsufficientWorkspacePermissionsError
+        ? internalErrorResponse(403, { error: 'Access denied' })
+        : null
+    ),
+  }),
+  mapInput: ({ params, body }) => ({
+    target: { kind: 'row' as const, table: 'webhook' as const, id: params.id },
+    enabled: body.isActive,
+  }),
+  useCase: setWorkflowTriggerEnabled,
+  present: ({ record }) => {
+    if (record.table !== 'webhook') throw new Error('Webhook update resolved a schedule row')
+    const { row } = record
+    return {
+      webhook: {
+        ...row,
+        path: row.path ?? undefined,
+        providerConfig: isRecordLike(row.providerConfig) ? row.providerConfig : undefined,
+      },
     }
-  }
-)
+  },
+})
 
 // Delete a webhook
 export const DELETE = withRouteHandler(
diff --git a/apps/sim/hooks/queries/webhooks.ts b/apps/sim/hooks/queries/webhooks.ts
index 5db9bc9422f..2e0ae031b50 100644
--- a/apps/sim/hooks/queries/webhooks.ts
+++ b/apps/sim/hooks/queries/webhooks.ts
@@ -69,7 +69,7 @@ export function useReactivateWebhook() {
     mutationFn: ({ webhookId }: ReactivateWebhookVariables) =>
       requestJson(updateWebhookContract, {
         params: { id: webhookId },
-        body: { isActive: true, failedCount: 0 },
+        body: { isActive: true },
       }),
     onSettled: (_data, _error, variables) =>
       queryClient.invalidateQueries({
diff --git a/apps/sim/lib/api/contracts/v2/list-pagination.test.ts b/apps/sim/lib/api/contracts/v2/list-pagination.test.ts
index f11a22ae1bf..eec048b12f4 100644
--- a/apps/sim/lib/api/contracts/v2/list-pagination.test.ts
+++ b/apps/sim/lib/api/contracts/v2/list-pagination.test.ts
@@ -137,6 +137,7 @@ const FULL_SET_LISTS = [
   'GET /api/v2/tables/[tableId]/views',
   'GET /api/v2/tables/folders',
   'GET /api/v2/workflows/folders',
+  'GET /api/v2/workflows/[workflowId]/triggers',
 ] as const
 
 /**
diff --git a/apps/sim/lib/api/contracts/v2/openapi/workflows.ts b/apps/sim/lib/api/contracts/v2/openapi/workflows.ts
index 5637d6fe7fd..0eb21e359f1 100644
--- a/apps/sim/lib/api/contracts/v2/openapi/workflows.ts
+++ b/apps/sim/lib/api/contracts/v2/openapi/workflows.ts
@@ -54,6 +54,7 @@ import {
   v2ListWorkflowFoldersContract,
   v2ListWorkflowRunsContract,
   v2ListWorkflowsContract,
+  v2ListWorkflowTriggersContract,
   v2ListWorkflowVersionsContract,
   v2MoveWorkflowsContract,
   v2RelocateWorkflowFolderContract,
@@ -67,6 +68,7 @@ import {
   v2UndeployWorkflowContract,
   v2UpdateWorkflowContract,
   v2UpdateWorkflowPublicApiContract,
+  v2UpdateWorkflowTriggerContract,
   v2UpdateWorkflowVersionContract,
 } from '@/lib/api/contracts/v2/workflows'
 import {
@@ -96,6 +98,21 @@ const WORKFLOW_EXAMPLE = {
   updatedAt: '2026-08-09T18:04:11.000Z',
 } as const
 
+const WORKFLOW_TRIGGER_EXAMPLE = {
+  blockId: 'blk_01J8ZK3QW4M6X2R9T7B5C0V4',
+  kind: 'schedule',
+  provider: null,
+  enabled: false,
+  status: 'disabled',
+  failedCount: 100,
+  lastFailedAt: '2026-08-09T18:04:11.000Z',
+  nextRunAt: null,
+  lastRanAt: '2026-08-09T18:04:00.000Z',
+  cronExpression: '0 * * * *',
+  timezone: 'UTC',
+  url: null,
+} as const
+
 const WORKFLOW_FOLDER_EXAMPLE = {
   name: 'Operations',
   path: '/Operations',
@@ -904,6 +921,61 @@ const declaredRoutes = [
       ),
     }
   ),
+  defineOpenApiRoute(
+    v2ListWorkflowTriggersContract,
+    workflowOperation({
+      applicationOperation: workflowOperations.readTriggers,
+      operationId: 'listWorkflowTriggers',
+      summary: 'List workflow triggers',
+      description: `List the schedules, polling triggers, internal-event triggers, and webhooks the live deployment serves, including disabled ones, with their failure counts. Requires workspace read access. Schedules and polling triggers disable themselves after 100 consecutive failures; a \`webhook\` trigger's failures happen at the sender, so its \`failedCount\` stays 0. Empty while the workflow is not deployed. ${FULL_SET_LIST}`,
+      errors: RESOURCE_ERRORS,
+      success: jsonSuccess("The live deployment's triggers."),
+    }),
+    {
+      query: v2ListWorkflowTriggersContract.query,
+      params: v2ListWorkflowTriggersContract.params,
+      response: documentedSchema(
+        v2ListWorkflowTriggersContract.response.schema,
+        'WorkflowTriggerListResponse',
+        'Workflow trigger list response',
+        'Every trigger the live deployment serves.',
+        [{ data: [WORKFLOW_TRIGGER_EXAMPLE], nextCursor: null }]
+      ),
+    }
+  ),
+  defineOpenApiRoute(
+    v2UpdateWorkflowTriggerContract,
+    workflowOperation({
+      applicationOperation: workflowOperations.setTriggerEnabled,
+      operationId: 'updateWorkflowTrigger',
+      summary: 'Update workflow trigger',
+      description: `Enable or disable one trigger of the live deployment without redeploying. Requires workspace write access. Enabling resets \`failedCount\`, schedules the next run, and clears any polling backoff; a schedule whose cron expression is invalid or has no future runs returns \`400\`. A permission group that blocks webhook triggers returns \`403\` when a member it governs enables a polling, internal-event, or webhook trigger; a workspace API key is not governed by permission groups, as with every other capability. Disabling a completed schedule marks it disabled. A block with several live rows switches them together. Requesting the current state changes nothing.\n\nDisabling only stops Sim from accepting the trigger; nothing is removed at the provider. A disable lasts until the next deploy or version activation, which re-enables every trigger, the same way it clears a trigger that disabled itself after repeated failures. Deliveries to a disabled webhook receive \`404\`, and some providers disable their subscription after repeated failures, which requires a redeploy to restore.`,
+      errors: [...RESOURCE_MUTATION_ERRORS, 'PayloadTooLarge'],
+      success: jsonSuccess('The trigger after the update.'),
+    }),
+    {
+      query: v2UpdateWorkflowTriggerContract.query,
+      params: v2UpdateWorkflowTriggerContract.params,
+      body: v2UpdateWorkflowTriggerContract.body,
+      response: documentedSchema(
+        v2UpdateWorkflowTriggerContract.response.schema,
+        'UpdateWorkflowTriggerResponse',
+        'Update workflow trigger response',
+        'The trigger after the update.',
+        [
+          {
+            data: {
+              ...WORKFLOW_TRIGGER_EXAMPLE,
+              enabled: true,
+              status: 'active',
+              failedCount: 0,
+              nextRunAt: '2026-08-10T09:00:00.000Z',
+            },
+          },
+        ]
+      ),
+    }
+  ),
   defineOpenApiRoute(
     v2DeployWorkflowContract,
     workflowOperation({
diff --git a/apps/sim/lib/api/contracts/v2/workflows.ts b/apps/sim/lib/api/contracts/v2/workflows.ts
index ecf991224de..d608b59c7a6 100644
--- a/apps/sim/lib/api/contracts/v2/workflows.ts
+++ b/apps/sim/lib/api/contracts/v2/workflows.ts
@@ -1298,6 +1298,127 @@ export const v2UpdateWorkflowPublicApiContract = defineRouteContract({
   },
 })
 
+const v2TriggerTimestampSchema = (description: string) =>
+  z.string().nullable().describe(description).meta({ format: 'date-time' })
+
+/**
+ * One trigger the live deployment serves. Schedule-only fields are `null` on
+ * every other kind, and `failedCount` only moves for schedules and polling
+ * triggers: a push webhook's failures happen at the sender, which Sim cannot
+ * count.
+ */
+const v2WorkflowTriggerSchema = z
+  .object({
+    blockId: z.string().describe('Trigger block in the deployed workflow.'),
+    kind: z
+      .enum(['schedule', 'polling', 'event', 'webhook'])
+      .describe(
+        'How the trigger fires: `schedule` on a cron, `polling` when Sim polls the source, `event` on an event raised inside Sim, `webhook` when an external sender delivers to its URL.'
+      ),
+    provider: z
+      .string()
+      .nullable()
+      .describe('Trigger provider, e.g. `gmail` or `github`; null for schedules.'),
+    enabled: z.boolean().describe('Whether the trigger currently fires.'),
+    status: z
+      .enum(['active', 'disabled', 'completed'])
+      .describe(
+        '`disabled` after a manual disable or an automatic one following repeated failures. `completed` applies only to schedules that have no runs left.'
+      ),
+    failedCount: z
+      .number()
+      .int()
+      .nonnegative()
+      .describe(
+        'Consecutive failures since the last success; the trigger disables itself at 100. Always 0 for `webhook` and `event` triggers.'
+      ),
+    lastFailedAt: v2TriggerTimestampSchema(
+      'ISO 8601 timestamp of the most recent counted failure, or null.'
+    ),
+    nextRunAt: v2TriggerTimestampSchema(
+      'ISO 8601 timestamp of the next scheduled run; null when disabled and for non-schedules.'
+    ),
+    lastRanAt: v2TriggerTimestampSchema(
+      'ISO 8601 timestamp of the last scheduled run; null for non-schedules.'
+    ),
+    cronExpression: z
+      .string()
+      .nullable()
+      .describe('Cron expression of a schedule; null for non-schedules.'),
+    timezone: z
+      .string()
+      .nullable()
+      .describe('IANA timezone the cron expression runs in; null for non-schedules.'),
+    url: z
+      .string()
+      .url()
+      .nullable()
+      .describe(
+        'Delivery URL of a `webhook` trigger; null for other kinds and for providers on a shared endpoint.'
+      ),
+  })
+  .strict()
+  .meta({
+    id: 'WorkflowTrigger',
+    title: 'Workflow trigger',
+    description: 'Health and enablement of one trigger the live deployment serves.',
+  })
+export type V2WorkflowTrigger = z.output<typeof v2WorkflowTriggerSchema>
+
+const v2WorkflowTriggerParamsSchema = z
+  .object({
+    workflowId: v2WorkflowIdParamsSchema.shape.workflowId,
+    blockId: z
+      .string({ error: missingFieldError('Block ID is required') })
+      .min(1, 'Block ID is required')
+      .max(MAX_ID_LENGTH, 'Block ID is too long')
+      .describe('Trigger block in the deployed workflow.'),
+  })
+  .meta({
+    id: 'WorkflowTriggerParams',
+    title: 'Workflow trigger path parameters',
+    description: 'Workflow and trigger block selected by the request path.',
+  })
+
+export const v2ListWorkflowTriggersContract = defineRouteContract({
+  method: 'GET',
+  path: '/api/v2/workflows/[workflowId]/triggers',
+  query: noInputSchema,
+  params: v2WorkflowIdParamsSchema,
+  response: {
+    mode: 'json',
+    schema: v2CursorListResponse(v2WorkflowTriggerSchema, { paged: false }),
+  },
+})
+
+const v2UpdateWorkflowTriggerBodySchema = z
+  .object({
+    enabled: z
+      .boolean({ error: 'enabled must be a boolean' })
+      .describe(
+        'Whether the trigger should fire. A disable lasts until the next deploy or version activation, which re-enables every trigger.'
+      ),
+  })
+  .strict()
+  .meta({
+    id: 'UpdateWorkflowTriggerRequest',
+    title: 'Update workflow trigger request',
+    description: 'Enable or disable one trigger of the live deployment.',
+    examples: [{ enabled: true }],
+  })
+
+export const v2UpdateWorkflowTriggerContract = defineRouteContract({
+  method: 'PATCH',
+  path: '/api/v2/workflows/[workflowId]/triggers/[blockId]',
+  query: noInputSchema,
+  params: v2WorkflowTriggerParamsSchema,
+  body: v2UpdateWorkflowTriggerBodySchema,
+  response: {
+    mode: 'json',
+    schema: v2DataResponse(v2WorkflowTriggerSchema),
+  },
+})
+
 /**
  * Structured execution error — mirrors `WorkflowExecutionErrorCode` in
  * `@/executor/utils/errors` (duplicated literally: contracts are
diff --git a/apps/sim/lib/api/contracts/webhooks.ts b/apps/sim/lib/api/contracts/webhooks.ts
index 25597f9281f..2c64885c4c7 100644
--- a/apps/sim/lib/api/contracts/webhooks.ts
+++ b/apps/sim/lib/api/contracts/webhooks.ts
@@ -52,15 +52,15 @@ export const webhookIdParamsSchema = z.object({
   id: z.string().min(1),
 })
 
+/**
+ * Pauses or resumes a webhook trigger. Re-enabling always clears the failure
+ * count, so the count is not caller-settable; `failedCount: 0` is still
+ * accepted and ignored so an editor loaded before that change keeps working.
+ */
 export const webhookPatchBodySchema = z
   .object({
-    isActive: z.boolean().optional(),
-    failedCount: z
-      .number({ error: 'failedCount must be a number' })
-      .finite('failedCount must be a valid number')
-      .int('failedCount must be an integer')
-      .min(0, 'failedCount must be at least 0')
-      .optional(),
+    isActive: z.boolean({ error: 'isActive must be a boolean' }),
+    failedCount: z.literal(0).optional(),
   })
   .strict()
 export type WebhookPatchBody = z.input<typeof webhookPatchBodySchema>
diff --git a/apps/sim/lib/api/mcp/generated/v2-operations.ts b/apps/sim/lib/api/mcp/generated/v2-operations.ts
index 20a0a4b34b8..1607efce7e0 100644
--- a/apps/sim/lib/api/mcp/generated/v2-operations.ts
+++ b/apps/sim/lib/api/mcp/generated/v2-operations.ts
@@ -316,6 +316,7 @@ import {
   v2ListWorkflowFoldersContract,
   v2ListWorkflowRunsContract,
   v2ListWorkflowsContract,
+  v2ListWorkflowTriggersContract,
   v2ListWorkflowVersionsContract,
   v2MoveWorkflowsContract,
   v2PreviewWorkflowImportContract,
@@ -328,6 +329,7 @@ import {
   v2UndeployWorkflowContract,
   v2UpdateWorkflowContract,
   v2UpdateWorkflowPublicApiContract,
+  v2UpdateWorkflowTriggerContract,
   v2UpdateWorkflowVersionContract,
 } from '@/lib/api/contracts/v2/workflows'
 import {
@@ -2198,6 +2200,14 @@ export const V2_MCP_OPERATIONS = {
       'List active workflows in a workspace. Requires workspace read access.\n\nUse `scope=archived` to find workflows available for restoration. Filter by folder, deployment, or name; sort and page with the returned cursor. Workspace folder trees exceeding 10,000 folders return `413`.\n\nOAuth scope: `api:read`.',
     handler: () => import('@/app/api/v2/workflows/route').then((route) => route.GET),
   },
+  listWorkflowTriggers: {
+    contract: v2ListWorkflowTriggersContract,
+    summary: 'List workflow triggers',
+    description:
+      "List the schedules, polling triggers, internal-event triggers, and webhooks the live deployment serves, including disabled ones, with their failure counts. Requires workspace read access. Schedules and polling triggers disable themselves after 100 consecutive failures; a `webhook` trigger's failures happen at the sender, so its `failedCount` stays 0. Empty while the workflow is not deployed. Returns one page; `nextCursor` is always null.\n\nOAuth scope: `api:read`.",
+    handler: () =>
+      import('@/app/api/v2/workflows/[workflowId]/triggers/route').then((route) => route.GET),
+  },
   listWorkflowVersions: {
     contract: v2ListWorkflowVersionsContract,
     summary: 'List workflow versions',
@@ -2989,6 +2999,16 @@ export const V2_MCP_OPERATIONS = {
     handler: () =>
       import('@/app/api/v2/workflows/[workflowId]/deployment/route').then((route) => route.PATCH),
   },
+  updateWorkflowTrigger: {
+    contract: v2UpdateWorkflowTriggerContract,
+    summary: 'Update workflow trigger',
+    description:
+      'Enable or disable one trigger of the live deployment without redeploying. Requires workspace write access. Enabling resets `failedCount`, schedules the next run, and clears any polling backoff; a schedule whose cron expression is invalid or has no future runs returns `400`. A permission group that blocks webhook triggers returns `403` when a member it governs enables a polling, internal-event, or webhook trigger; a workspace API key is not governed by permission groups, as with every other capability. Disabling a completed schedule marks it disabled. A block with several live rows switches them together. Requesting the current state changes nothing.\n\nDisabling only stops Sim from accepting the trigger; nothing is removed at the provider. A disable lasts until the next deploy or version activation, which re-enables every trigger, the same way it clears a trigger that disabled itself after repeated failures. Deliveries to a disabled webhook receive `404`, and some providers disable their subscription after repeated failures, which requires a redeploy to restore.\n\nOAuth scope: `api:write`.',
+    handler: () =>
+      import('@/app/api/v2/workflows/[workflowId]/triggers/[blockId]/route').then(
+        (route) => route.PATCH
+      ),
+  },
   updateWorkflowVersion: {
     contract: v2UpdateWorkflowVersionContract,
     summary: 'Update workflow version',
diff --git a/apps/sim/lib/api/server/routes/v2-route-table.generated.ts b/apps/sim/lib/api/server/routes/v2-route-table.generated.ts
index 8047e84737b..a7ad4c8e5ce 100644
--- a/apps/sim/lib/api/server/routes/v2-route-table.generated.ts
+++ b/apps/sim/lib/api/server/routes/v2-route-table.generated.ts
@@ -839,6 +839,14 @@ export const V2_ROUTES: readonly V2RouteEntry[] = [
     pattern: '/api/v2/workflows/{workflowId}/state',
     load: () => import('@/app/api/v2/workflows/[workflowId]/state/route'),
   },
+  {
+    pattern: '/api/v2/workflows/{workflowId}/triggers',
+    load: () => import('@/app/api/v2/workflows/[workflowId]/triggers/route'),
+  },
+  {
+    pattern: '/api/v2/workflows/{workflowId}/triggers/{blockId}',
+    load: () => import('@/app/api/v2/workflows/[workflowId]/triggers/[blockId]/route'),
+  },
   {
     pattern: '/api/v2/workflows/{workflowId}/variables',
     load: () => import('@/app/api/v2/workflows/[workflowId]/variables/route'),
diff --git a/apps/sim/lib/webhooks/polling/utils.ts b/apps/sim/lib/webhooks/polling/utils.ts
index 58d210971a1..3228a3a1bfb 100644
--- a/apps/sim/lib/webhooks/polling/utils.ts
+++ b/apps/sim/lib/webhooks/polling/utils.ts
@@ -28,8 +28,8 @@ const POLL_TICK_TOLERANCE_MS = 10_000
 /** Ceiling on a source's own `Retry-After`, so a hostile feed cannot park a trigger for days. */
 const POLL_RETRY_AFTER_MAX_MS = 24 * 60 * 60_000
 
-const POLL_BACKOFF_UNTIL_KEY = 'pollBackoffUntil'
-const POLL_SOURCE_FAILURES_KEY = 'pollSourceFailures'
+export const POLL_BACKOFF_UNTIL_KEY = 'pollBackoffUntil'
+export const POLL_SOURCE_FAILURES_KEY = 'pollSourceFailures'
 
 /**
  * Whether a webhook is inside the backoff window {@link recordPollSourceFailure}
@@ -145,29 +145,27 @@ export async function recordPollSourceFailure(
   await markWebhookFailed(webhookData.id, logger)
 }
 
-/** Increment the webhook's failure count. Auto-disables after MAX_CONSECUTIVE_FAILURES. */
+/**
+ * Increment the webhook's failure count, auto-disabling it once the count
+ * reaches MAX_CONSECUTIVE_FAILURES. One statement, so an enable that resets the
+ * count concurrently is never followed by a disable decided on the old count.
+ */
 export async function markWebhookFailed(webhookId: string, logger: Logger): Promise<void> {
   try {
-    const result = await db
+    const now = new Date()
+    const reachesLimit = sql`COALESCE(${webhook.failedCount}, 0) + 1 >= ${MAX_CONSECUTIVE_FAILURES}`
+    const [result] = await db
       .update(webhook)
       .set({
         failedCount: sql`COALESCE(${webhook.failedCount}, 0) + 1`,
-        lastFailedAt: new Date(),
-        updatedAt: new Date(),
+        lastFailedAt: now,
+        isActive: sql`CASE WHEN ${reachesLimit} THEN false ELSE ${webhook.isActive} END`,
+        updatedAt: now,
       })
       .where(eq(webhook.id, webhookId))
       .returning({ failedCount: webhook.failedCount })
 
-    const newFailedCount = result[0]?.failedCount || 0
-    if (newFailedCount >= MAX_CONSECUTIVE_FAILURES) {
-      await db
-        .update(webhook)
-        .set({
-          isActive: false,
-          updatedAt: new Date(),
-        })
-        .where(eq(webhook.id, webhookId))
-
+    if ((result?.failedCount ?? 0) >= MAX_CONSECUTIVE_FAILURES) {
       logger.warn(
         `Webhook ${webhookId} auto-disabled after ${MAX_CONSECUTIVE_FAILURES} consecutive failures`
       )
diff --git a/apps/sim/lib/workflows/api/index.ts b/apps/sim/lib/workflows/api/index.ts
index 44fde23abf1..7d40f29c6f4 100644
--- a/apps/sim/lib/workflows/api/index.ts
+++ b/apps/sim/lib/workflows/api/index.ts
@@ -7,3 +7,4 @@ export {
   v2WorkflowErrorPolicies,
   WORKFLOW_NOT_FOUND_MESSAGE,
 } from '@/lib/workflows/api/route-policies'
+export { presentV2WorkflowTrigger } from '@/lib/workflows/api/trigger-presenter'
diff --git a/apps/sim/lib/workflows/api/trigger-presenter.ts b/apps/sim/lib/workflows/api/trigger-presenter.ts
new file mode 100644
index 00000000000..f376817c59e
--- /dev/null
+++ b/apps/sim/lib/workflows/api/trigger-presenter.ts
@@ -0,0 +1,20 @@
+import type { V2WorkflowTrigger } from '@/lib/api/contracts/v2/workflows'
+import type { WorkflowTriggerHealth } from '@/lib/workflows/triggers/health'
+
+/** The v2 wire shape of one live trigger, shared by the list and the update. */
+export function presentV2WorkflowTrigger(trigger: WorkflowTriggerHealth): V2WorkflowTrigger {
+  return {
+    blockId: trigger.blockId,
+    kind: trigger.kind,
+    provider: trigger.provider,
+    enabled: trigger.enabled,
+    status: trigger.status,
+    failedCount: trigger.failedCount,
+    lastFailedAt: trigger.lastFailedAt?.toISOString() ?? null,
+    nextRunAt: trigger.nextRunAt?.toISOString() ?? null,
+    lastRanAt: trigger.lastRanAt?.toISOString() ?? null,
+    cronExpression: trigger.cronExpression,
+    timezone: trigger.timezone,
+    url: trigger.url,
+  }
+}
diff --git a/apps/sim/lib/workflows/application/operations.ts b/apps/sim/lib/workflows/application/operations.ts
index dec0ed036fd..e23b8bcba8f 100644
--- a/apps/sim/lib/workflows/application/operations.ts
+++ b/apps/sim/lib/workflows/application/operations.ts
@@ -22,6 +22,18 @@ const WORKFLOW_READ_PRINCIPAL_POLICY = {
   delegatedServices: ['copilot', 'executor'],
 } as const
 
+/** Every caller, including a workflow run's executor delegation. */
+const EXECUTOR_REACHABLE_WORKFLOW_PRINCIPAL_POLICY = {
+  principalKinds: [
+    'session',
+    'personal_api_key',
+    'oauth_access_token',
+    'workspace_api_key',
+    'delegated',
+  ],
+  delegatedServices: ['copilot', 'executor'],
+} as const
+
 const HUMAN_WORKFLOW_PRINCIPAL_POLICY = {
   principalKinds: ['session', 'personal_api_key', 'oauth_access_token', 'delegated'],
   delegatedServices: ['copilot'],
@@ -342,6 +354,34 @@ export const workflowOperations = {
     principalKinds: ['session', 'personal_api_key', 'oauth_access_token', 'delegated'],
     delegatedServices: ['copilot'],
   }),
+  // permission-group-exempt: trigger health is deployment state, governed by workspace role like the deployment read that already lists webhook URLs
+  readTriggers: defineWorkspaceOperation({
+    id: 'workflows.triggers.read',
+    oauthScope: 'api:read',
+    minimumRole: 'read',
+    workspaceApiKey: 'allow',
+    capability: 'none',
+    contentDelivery: 'none',
+    ...WORKFLOW_READ_PRINCIPAL_POLICY,
+  }),
+  /**
+   * Pausing or resuming one trigger of the live deployment. Not a deploy: the
+   * deployed graph and its provider registrations are untouched, so it is a
+   * write rather than an admin act, and a workspace key may perform it.
+   * A workflow run's executor delegation may too, as it could before this
+   * moved onto the use case.
+   *
+   * permission-group-exempt: `triggers.webhook` is asserted inside the use case and only for enabling a webhook-table row, because a group that withholds inbound webhooks must still let a member switch a live one off
+   */
+  setTriggerEnabled: defineWorkspaceOperation({
+    id: 'workflows.triggers.set_enabled',
+    oauthScope: 'api:write',
+    minimumRole: 'write',
+    workspaceApiKey: 'allow',
+    capability: 'none',
+    contentDelivery: 'none',
+    ...EXECUTOR_REACHABLE_WORKFLOW_PRINCIPAL_POLICY,
+  }),
   activateVersion: defineWorkspaceOperation({
     id: 'workflows.versions.activate',
     oauthScope: 'api:write',
diff --git a/apps/sim/lib/workflows/application/triggers.integration.ts b/apps/sim/lib/workflows/application/triggers.integration.ts
new file mode 100644
index 00000000000..320095009b3
--- /dev/null
+++ b/apps/sim/lib/workflows/application/triggers.integration.ts
@@ -0,0 +1,760 @@
+import { mkdir, writeFile } from 'node:fs/promises'
+import { dirname, resolve } from 'node:path'
+import type { Principal } from '@sim/auth/principal'
+import { db } from '@sim/db'
+import {
+  auditLog,
+  member,
+  organization,
+  permissionGroup,
+  permissionGroupWorkspace,
+  permissions,
+  user,
+  webhook,
+  workflow,
+  workflowDeploymentVersion,
+  workflowSchedule,
+  workspace,
+} from '@sim/db/schema'
+import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
+import { createLogger } from '@sim/logger'
+import {
+  createExecutorPrincipal,
+  createSessionPrincipal,
+  createWorkspaceApiKeyPrincipal,
+} from '@sim/testing/factories/principal.factory'
+import { featureFlagsMock } from '@sim/testing/mocks/feature-flags.mock'
+import { getErrorMessage } from '@sim/utils/errors'
+import { generateId } from '@sim/utils/id'
+import { and, eq, inArray } from 'drizzle-orm'
+import { afterAll, beforeAll, expect, it, vi } from 'vitest'
+import { markWebhookFailed } from '@/lib/webhooks/polling/utils'
+import { v2WorkflowErrorPolicies } from '@/lib/workflows/api'
+import { WORKFLOW_DELEGATION_AUDIENCE } from '@/lib/workflows/application/authorization'
+import {
+  listWorkflowTriggers,
+  setWorkflowTriggerEnabled,
+} from '@/lib/workflows/application/triggers'
+import { setLiveTriggersEnabled } from '@/lib/workflows/triggers/health'
+
+vi.mock('@/lib/core/config/feature-flags', () => featureFlagsMock)
+
+const ownerId = generateId()
+const writerId = generateId()
+const readerId = generateId()
+const outsiderId = generateId()
+const organizationId = generateId()
+const workspaceId = generateId()
+const otherWorkspaceId = generateId()
+const groupId = generateId()
+
+const workflowId = generateId()
+const lockedWorkflowId = generateId()
+const undeployedWorkflowId = generateId()
+const versionlessWorkflowId = generateId()
+const oldVersionId = generateId()
+const activeVersionId = generateId()
+const lockedVersionId = generateId()
+const undeployedVersionId = generateId()
+
+const liveScheduleId = generateId()
+const oldScheduleId = generateId()
+const completedScheduleId = generateId()
+const badCronScheduleId = generateId()
+const lockedScheduleId = generateId()
+const undeployedScheduleId = generateId()
+const unversionedScheduleId = generateId()
+const versionlessScheduleId = generateId()
+const liveWebhookId = generateId()
+const candidateWebhookId = generateId()
+const retiredWebhookId = generateId()
+const oldWebhookId = generateId()
+const pollingWebhookId = generateId()
+const eventWebhookId = generateId()
+const twinWebhookIds = [generateId(), generateId()]
+const versionlessWebhookId = generateId()
+const scalarConfigWebhookId = generateId()
+const presaveWebhookId = generateId()
+const executorWebhookId = generateId()
+const flakyWebhookId = generateId()
+const hookPath = `trigger-health-${generateId()}`
+
+const writer = createSessionPrincipal({ userId: writerId })
+const reader = createSessionPrincipal({ userId: readerId })
+const outsider = createSessionPrincipal({ userId: outsiderId })
+const workspaceKey = createWorkspaceApiKeyPrincipal({ workspaceId, keyId: generateId() })
+const foreignKey = createWorkspaceApiKeyPrincipal({
+  workspaceId: otherWorkspaceId,
+  keyId: generateId(),
+})
+
+const checks: { name: string; status: string; durationMs: number; error?: string }[] = []
+
+function check(name: string, run: () => Promise<void>) {
+  it(name, async () => {
+    const started = performance.now()
+    try {
+      await run()
+      checks.push({ name, status: 'passed', durationMs: performance.now() - started })
+    } catch (error) {
+      checks.push({
+        name,
+        status: 'failed',
+        durationMs: performance.now() - started,
+        error: getErrorMessage(error),
+      })
+      throw error
+    }
+  })
+}
+
+function setBlock(blockId: string, enabled: boolean, principal: Principal = writer) {
+  return setWorkflowTriggerEnabled.execute({
+    principal,
+    input: { target: { kind: 'block', workflowId, blockId }, enabled },
+  })
+}
+
+async function setWebhookTriggersWithheld(withheld: boolean) {
+  await db
+    .update(permissionGroup)
+    .set({ config: withheld ? { disableWebhookTriggers: true } : {} })
+    .where(eq(permissionGroup.id, groupId))
+}
+
+async function scheduleRow(id: string) {
+  const [row] = await db.select().from(workflowSchedule).where(eq(workflowSchedule.id, id))
+  return row
+}
+
+async function webhookRow(id: string) {
+  const [row] = await db.select().from(webhook).where(eq(webhook.id, id))
+  return row
+}
+
+function auditRows(resourceId: string) {
+  return db
+    .select()
+    .from(auditLog)
+    .where(and(eq(auditLog.workspaceId, workspaceId), eq(auditLog.resourceId, resourceId)))
+}
+
+async function surfaceStatus(run: () => Promise<unknown>): Promise<number | undefined> {
+  const error = await run().then(
+    () => null,
+    (caught: unknown) => caught
+  )
+  expect(error).not.toBeNull()
+  return v2WorkflowErrorPolicies.concealWorkflowAuthorization.render(error)?.status
+}
+
+const emptyState = { blocks: {}, edges: [], loops: {}, parallels: {} }
+
+beforeAll(async () => {
+  const now = new Date()
+  await db.insert(user).values(
+    [ownerId, writerId, readerId, outsiderId].map((id) => ({
+      id,
+      name: 'Trigger health fixture',
+      email: `${id}@trigger-health.invalid`,
+      emailVerified: true,
+      createdAt: now,
+      updatedAt: now,
+    }))
+  )
+  await db.insert(organization).values({
+    id: organizationId,
+    name: 'Trigger health',
+    slug: organizationId,
+    createdAt: now,
+  })
+  await db.insert(member).values([
+    { id: generateId(), organizationId, userId: ownerId, role: 'owner', createdAt: now },
+    { id: generateId(), organizationId, userId: writerId, role: 'member', createdAt: now },
+    { id: generateId(), organizationId, userId: readerId, role: 'member', createdAt: now },
+  ])
+  await insertWorkspaceFixture(db, [
+    {
+      id: workspaceId,
+      organizationId,
+      ownerId,
+      billedAccountUserId: ownerId,
+      workspaceMode: 'organization',
+      name: 'Trigger health',
+    },
+    {
+      id: otherWorkspaceId,
+      ownerId: outsiderId,
+      billedAccountUserId: outsiderId,
+      name: 'Trigger health elsewhere',
+    },
+  ])
+  await db.insert(permissions).values([
+    {
+      id: generateId(),
+      entityType: 'workspace',
+      entityId: workspaceId,
+      userId: writerId,
+      permissionType: 'write',
+    },
+    {
+      id: generateId(),
+      entityType: 'workspace',
+      entityId: workspaceId,
+      userId: readerId,
+      permissionType: 'read',
+    },
+  ])
+  await db.insert(permissionGroup).values({
+    id: groupId,
+    organizationId,
+    createdBy: ownerId,
+    name: 'Trigger health fixture',
+    config: {},
+    membershipMode: 'inherit',
+  })
+  await db.insert(permissionGroupWorkspace).values({
+    id: generateId(),
+    permissionGroupId: groupId,
+    workspaceId,
+    organizationId,
+  })
+
+  await db.insert(workflow).values(
+    [
+      { id: workflowId, locked: false, isDeployed: true },
+      { id: lockedWorkflowId, locked: true, isDeployed: true },
+      { id: undeployedWorkflowId, locked: false, isDeployed: false },
+      { id: versionlessWorkflowId, locked: false, isDeployed: true },
+    ].map((row, index) => ({
+      ...row,
+      userId: ownerId,
+      workspaceId,
+      name: `Trigger health fixture ${index}`,
+      lastSynced: now,
+      createdAt: now,
+      updatedAt: now,
+    }))
+  )
+  await db.insert(workflowDeploymentVersion).values([
+    { id: oldVersionId, workflowId, version: 1, state: emptyState, isActive: false },
+    { id: activeVersionId, workflowId, version: 2, state: emptyState, isActive: true },
+    {
+      id: lockedVersionId,
+      workflowId: lockedWorkflowId,
+      version: 1,
+      state: emptyState,
+      isActive: true,
+    },
+    {
+      id: undeployedVersionId,
+      workflowId: undeployedWorkflowId,
+      version: 1,
+      state: emptyState,
+      isActive: false,
+    },
+  ])
+
+  const disabledSchedule = {
+    triggerType: 'schedule',
+    cronExpression: '0 * * * *',
+    timezone: 'UTC',
+    status: 'disabled',
+    failedCount: 100,
+    infraRetryCount: 3,
+    lastFailedAt: now,
+    nextRunAt: null,
+  }
+  await db.insert(workflowSchedule).values([
+    {
+      ...disabledSchedule,
+      id: liveScheduleId,
+      workflowId,
+      deploymentVersionId: activeVersionId,
+      blockId: 'sched',
+    },
+    {
+      ...disabledSchedule,
+      id: oldScheduleId,
+      workflowId,
+      deploymentVersionId: oldVersionId,
+      blockId: 'sched',
+    },
+    {
+      ...disabledSchedule,
+      id: completedScheduleId,
+      workflowId,
+      deploymentVersionId: activeVersionId,
+      blockId: 'done',
+      cronExpression: '0 0 30 2 *',
+      status: 'completed',
+      failedCount: 0,
+    },
+    {
+      ...disabledSchedule,
+      id: badCronScheduleId,
+      workflowId,
+      deploymentVersionId: activeVersionId,
+      blockId: 'badcron',
+      cronExpression: 'not a cron',
+    },
+    {
+      ...disabledSchedule,
+      id: lockedScheduleId,
+      workflowId: lockedWorkflowId,
+      deploymentVersionId: lockedVersionId,
+      blockId: 'sched',
+    },
+    {
+      ...disabledSchedule,
+      id: unversionedScheduleId,
+      workflowId,
+      deploymentVersionId: null,
+      blockId: 'sched',
+    },
+    {
+      ...disabledSchedule,
+      id: versionlessScheduleId,
+      workflowId: versionlessWorkflowId,
+      deploymentVersionId: null,
+      blockId: 'sched',
+    },
+    {
+      ...disabledSchedule,
+      id: undeployedScheduleId,
+      workflowId: undeployedWorkflowId,
+      deploymentVersionId: undeployedVersionId,
+      blockId: 'sched',
+    },
+  ])
+
+  const dormantWebhook = {
+    workflowId,
+    deploymentVersionId: activeVersionId,
+    provider: 'generic',
+    isActive: false,
+    failedCount: 0,
+  }
+  await db.insert(webhook).values([
+    {
+      ...dormantWebhook,
+      id: liveWebhookId,
+      blockId: 'hook',
+      path: hookPath,
+      registrationStatus: 'active',
+    },
+    {
+      ...dormantWebhook,
+      id: candidateWebhookId,
+      blockId: 'hook',
+      path: `${hookPath}-candidate`,
+      registrationStatus: 'candidate',
+      archivedAt: now,
+    },
+    {
+      ...dormantWebhook,
+      id: retiredWebhookId,
+      blockId: 'hook',
+      path: `${hookPath}-retired`,
+      registrationStatus: 'retired',
+      archivedAt: now,
+    },
+    {
+      ...dormantWebhook,
+      id: oldWebhookId,
+      deploymentVersionId: oldVersionId,
+      blockId: 'hook',
+      path: hookPath,
+    },
+    {
+      ...dormantWebhook,
+      id: pollingWebhookId,
+      blockId: 'poll',
+      provider: 'gmail',
+      path: `${hookPath}-poll`,
+      failedCount: 100,
+      lastFailedAt: now,
+      providerConfig: {
+        labelIds: ['INBOX'],
+        pollBackoffUntil: new Date(now.getTime() + 3_600_000).toISOString(),
+        pollSourceFailures: 7,
+      },
+    },
+    {
+      ...dormantWebhook,
+      id: eventWebhookId,
+      blockId: 'evt',
+      provider: 'table',
+      path: `${hookPath}-evt`,
+      isActive: true,
+    },
+    ...twinWebhookIds.map((id, index) => ({
+      ...dormantWebhook,
+      id,
+      blockId: 'twin',
+      path: `${hookPath}-twin-${index}`,
+      failedCount: 100,
+      providerConfig: { pollSourceFailures: 1 },
+    })),
+    {
+      ...dormantWebhook,
+      id: scalarConfigWebhookId,
+      blockId: 'scalar',
+      provider: 'gmail',
+      path: `${hookPath}-scalar`,
+      providerConfig: 'legacy-scalar-config',
+    },
+    {
+      ...dormantWebhook,
+      id: presaveWebhookId,
+      workflowId: undeployedWorkflowId,
+      deploymentVersionId: null,
+      blockId: 'presave',
+      path: `${hookPath}-presave`,
+    },
+    {
+      ...dormantWebhook,
+      id: executorWebhookId,
+      blockId: 'exec',
+      path: `${hookPath}-exec`,
+    },
+    {
+      ...dormantWebhook,
+      id: flakyWebhookId,
+      blockId: 'flaky',
+      provider: 'gmail',
+      path: `${hookPath}-flaky`,
+      isActive: true,
+      failedCount: 98,
+    },
+    {
+      ...dormantWebhook,
+      id: versionlessWebhookId,
+      workflowId: versionlessWorkflowId,
+      deploymentVersionId: null,
+      blockId: 'hook',
+      path: `${hookPath}-versionless`,
+    },
+  ])
+})
+
+afterAll(async () => {
+  const reportPath =
+    process.env.TRIGGER_HEALTH_REPORT_PATH ?? resolve('test-results/trigger-health.json')
+  await mkdir(dirname(reportPath), { recursive: true })
+  await writeFile(reportPath, JSON.stringify({ checks }, null, 2))
+  await db.delete(auditLog).where(eq(auditLog.workspaceId, workspaceId))
+  await deleteWorkspaceFixture(db, inArray(workspace.id, [workspaceId, otherWorkspaceId]))
+  await db.delete(organization).where(eq(organization.id, organizationId))
+  await db.delete(user).where(inArray(user.id, [ownerId, writerId, readerId, outsiderId]))
+})
+
+check('lists only the live deployment’s rows, disabled ones included', async () => {
+  const { triggers } = await listWorkflowTriggers.execute({
+    principal: reader,
+    input: { workflowId },
+  })
+  expect(triggers.map((trigger) => [trigger.blockId, trigger.kind, trigger.status])).toEqual([
+    ['badcron', 'schedule', 'disabled'],
+    ['done', 'schedule', 'completed'],
+    ['evt', 'event', 'active'],
+    ['exec', 'webhook', 'disabled'],
+    ['flaky', 'polling', 'active'],
+    ['hook', 'webhook', 'disabled'],
+    ['poll', 'polling', 'disabled'],
+    ['scalar', 'polling', 'disabled'],
+    ['sched', 'schedule', 'disabled'],
+    ['twin', 'webhook', 'disabled'],
+    ['twin', 'webhook', 'disabled'],
+  ])
+  const hook = triggers.find((trigger) => trigger.blockId === 'hook')
+  expect(hook?.url).toContain(`/api/webhooks/trigger/${hookPath}`)
+  expect(triggers.find((trigger) => trigger.blockId === 'evt')?.url).toBeNull()
+  expect(triggers.find((trigger) => trigger.blockId === 'sched')).toMatchObject({
+    enabled: false,
+    failedCount: 100,
+    cronExpression: '0 * * * *',
+    timezone: 'UTC',
+  })
+
+  const undeployed = await listWorkflowTriggers.execute({
+    principal: reader,
+    input: { workflowId: undeployedWorkflowId },
+  })
+  expect(undeployed.triggers).toEqual([])
+})
+
+check('enabling a schedule resets its counters and arms the next run', async () => {
+  const before = Date.now()
+  const result = await setBlock('sched', true)
+  expect(result.changes).toHaveLength(1)
+  expect(result.trigger).toMatchObject({ enabled: true, status: 'active', failedCount: 0 })
+
+  const live = await scheduleRow(liveScheduleId)
+  expect(live).toMatchObject({ status: 'active', failedCount: 0, infraRetryCount: 0 })
+  expect(live.nextRunAt?.getTime()).toBeGreaterThan(before)
+
+  for (const id of [oldScheduleId, unversionedScheduleId]) {
+    expect(await scheduleRow(id)).toMatchObject({
+      status: 'disabled',
+      failedCount: 100,
+      infraRetryCount: 3,
+    })
+  }
+})
+
+check('a no-op write records no audit, a real change records one', async () => {
+  await expect.poll(async () => (await auditRows(liveScheduleId)).length).toBe(1)
+  const [enabled] = await auditRows(liveScheduleId)
+  expect(enabled).toMatchObject({ action: 'schedule.updated', actorId: writerId })
+
+  const noop = await setBlock('sched', true)
+  expect(noop.changes).toEqual([])
+
+  const disabled = await setBlock('sched', false)
+  expect(disabled.changes).toHaveLength(1)
+  expect(await scheduleRow(liveScheduleId)).toMatchObject({
+    status: 'disabled',
+    nextRunAt: null,
+    lastQueuedAt: null,
+  })
+  await expect.poll(async () => (await auditRows(liveScheduleId)).length).toBe(2)
+})
+
+check('legacy row ids never revive a row the live deployment does not serve', async () => {
+  await expect(
+    setWorkflowTriggerEnabled.execute({
+      principal: writer,
+      input: { target: { kind: 'row', table: 'schedule', id: oldScheduleId }, enabled: true },
+    })
+  ).rejects.toMatchObject({ code: 'not_found', message: 'Schedule not found' })
+  for (const webhookId of [candidateWebhookId, retiredWebhookId, oldWebhookId]) {
+    await expect(
+      setWorkflowTriggerEnabled.execute({
+        principal: writer,
+        input: { target: { kind: 'row', table: 'webhook', id: webhookId }, enabled: true },
+      })
+    ).rejects.toMatchObject({ code: 'not_found', message: 'Webhook not found' })
+  }
+  expect((await scheduleRow(oldScheduleId)).status).toBe('disabled')
+
+  const viaRow = await setWorkflowTriggerEnabled.execute({
+    principal: writer,
+    input: { target: { kind: 'row', table: 'schedule', id: liveScheduleId }, enabled: true },
+  })
+  expect(viaRow.changes).toHaveLength(1)
+  expect(viaRow.record.row.id).toBe(liveScheduleId)
+})
+
+check('a switch is all or nothing: one row no longer live rolls back the rest', async () => {
+  const updated = await setLiveTriggersEnabled({
+    workflowId,
+    enabled: true,
+    schedules: [
+      { id: badCronScheduleId, nextRunAt: new Date(Date.now() + 60_000) },
+      { id: oldScheduleId, nextRunAt: new Date(Date.now() + 60_000) },
+    ],
+    webhookIds: [twinWebhookIds[1], oldWebhookId, candidateWebhookId],
+    liveness: 'editor',
+    now: new Date(),
+  })
+  expect(updated).toBeNull()
+  expect((await scheduleRow(oldScheduleId)).status).toBe('disabled')
+  expect((await scheduleRow(badCronScheduleId)).status).toBe('disabled')
+  expect((await webhookRow(twinWebhookIds[1])).isActive).toBe(false)
+  expect((await webhookRow(oldWebhookId)).isActive).toBe(false)
+  expect((await webhookRow(candidateWebhookId)).isActive).toBe(false)
+})
+
+check('rows that never run are hidden from v2 but kept for the editor row id', async () => {
+  for (const id of [versionlessWorkflowId, undeployedWorkflowId]) {
+    const { triggers } = await listWorkflowTriggers.execute({
+      principal: reader,
+      input: { workflowId: id },
+    })
+    expect(triggers).toEqual([])
+  }
+  for (const [id, blockId] of [
+    [versionlessWorkflowId, 'sched'],
+    [versionlessWorkflowId, 'hook'],
+    [undeployedWorkflowId, 'presave'],
+  ]) {
+    await expect(
+      setWorkflowTriggerEnabled.execute({
+        principal: writer,
+        input: { target: { kind: 'block', workflowId: id, blockId }, enabled: true },
+      })
+    ).rejects.toMatchObject({ code: 'not_found' })
+  }
+  expect((await scheduleRow(versionlessScheduleId)).status).toBe('disabled')
+  expect((await webhookRow(presaveWebhookId)).isActive).toBe(false)
+
+  const presave = await setWorkflowTriggerEnabled.execute({
+    principal: writer,
+    input: { target: { kind: 'row', table: 'webhook', id: presaveWebhookId }, enabled: true },
+  })
+  expect(presave.changes).toHaveLength(1)
+
+  const schedule = await setWorkflowTriggerEnabled.execute({
+    principal: writer,
+    input: { target: { kind: 'row', table: 'schedule', id: versionlessScheduleId }, enabled: true },
+  })
+  expect(schedule.changes).toHaveLength(1)
+  const hook = await setWorkflowTriggerEnabled.execute({
+    principal: writer,
+    input: { target: { kind: 'row', table: 'webhook', id: versionlessWebhookId }, enabled: true },
+  })
+  expect(hook.changes).toHaveLength(1)
+  expect((await webhookRow(versionlessWebhookId)).isActive).toBe(true)
+})
+
+check('a block switches every live row together; a row id switches only that row', async () => {
+  const [first, second] = twinWebhookIds
+  const viaRow = await setWorkflowTriggerEnabled.execute({
+    principal: writer,
+    input: { target: { kind: 'row', table: 'webhook', id: first }, enabled: true },
+  })
+  expect(viaRow.changes.map((change) => change.record.row.id)).toEqual([first])
+  expect((await webhookRow(second)).isActive).toBe(false)
+
+  const viaBlock = await setBlock('twin', false)
+  expect(viaBlock.changes.map((change) => change.record.row.id)).toEqual([first])
+  const enabled = await setBlock('twin', true)
+  expect(enabled.changes.map((change) => change.record.row.id).sort()).toEqual(
+    [...twinWebhookIds].sort()
+  )
+  for (const id of twinWebhookIds) {
+    expect(await webhookRow(id)).toMatchObject({
+      isActive: true,
+      failedCount: 0,
+      providerConfig: { pollSourceFailures: 1 },
+    })
+  }
+  await expect.poll(async () => (await auditRows(second)).length).toBe(1)
+})
+
+check(
+  'an invalid or exhausted cron is refused; disabling a completed schedule sticks',
+  async () => {
+    await expect(setBlock('badcron', true)).rejects.toMatchObject({
+      code: 'validation',
+      message: 'Schedule has invalid cron expression',
+    })
+    await expect(setBlock('done', true)).rejects.toMatchObject({
+      code: 'validation',
+      message: 'Schedule has invalid cron expression',
+    })
+    expect((await scheduleRow(badCronScheduleId)).status).toBe('disabled')
+    expect((await scheduleRow(completedScheduleId)).status).toBe('completed')
+
+    const disabled = await setBlock('done', false)
+    expect(disabled.changes.map((change) => change.previousStatus)).toEqual(['completed'])
+    expect((await scheduleRow(completedScheduleId)).status).toBe('disabled')
+  }
+)
+
+check('a locked workflow refuses the change', async () => {
+  await expect(
+    setWorkflowTriggerEnabled.execute({
+      principal: writer,
+      input: {
+        target: { kind: 'block', workflowId: lockedWorkflowId, blockId: 'sched' },
+        enabled: true,
+      },
+    })
+  ).rejects.toMatchObject({ code: 'locked' })
+  expect((await scheduleRow(lockedScheduleId)).status).toBe('disabled')
+})
+
+check('read access lists but cannot write; other tenants see not found', async () => {
+  expect(await surfaceStatus(() => setBlock('sched', false, reader))).toBe(403)
+  expect(
+    await surfaceStatus(() =>
+      listWorkflowTriggers.execute({ principal: outsider, input: { workflowId } })
+    )
+  ).toBe(404)
+  expect(await surfaceStatus(() => setBlock('sched', false, outsider))).toBe(404)
+  expect(await surfaceStatus(() => setBlock('sched', false, foreignKey))).toBe(404)
+  expect((await scheduleRow(liveScheduleId)).status).toBe('active')
+})
+
+check('a workflow run’s executor delegation may enable, ungoverned by the group', async () => {
+  const executor = createExecutorPrincipal({
+    audience: WORKFLOW_DELEGATION_AUDIENCE,
+    workspaceId,
+    subjectUserId: writerId,
+    delegationContext: { kind: 'workflow_execution', workflowId },
+  })
+  await setWebhookTriggersWithheld(true)
+  try {
+    const enabled = await setBlock('exec', true, executor)
+    expect(enabled.changes).toHaveLength(1)
+    expect((await webhookRow(executorWebhookId)).isActive).toBe(true)
+  } finally {
+    await setWebhookTriggersWithheld(false)
+  }
+})
+
+check('the webhook capability gates enabling only, and only governed callers', async () => {
+  await setWebhookTriggersWithheld(true)
+  try {
+    await expect(setBlock('hook', true)).rejects.toMatchObject({
+      detailCode: 'PERMISSION_GROUP_CAPABILITY_BLOCKED',
+    })
+    expect((await webhookRow(liveWebhookId)).isActive).toBe(false)
+
+    const byKey = await setBlock('hook', true, workspaceKey)
+    expect(byKey.changes).toHaveLength(1)
+    expect((await webhookRow(liveWebhookId)).isActive).toBe(true)
+
+    const disabled = await setBlock('hook', false)
+    expect(disabled.changes).toHaveLength(1)
+    expect((await webhookRow(liveWebhookId)).isActive).toBe(false)
+  } finally {
+    await setWebhookTriggersWithheld(false)
+  }
+
+  const enabled = await setWorkflowTriggerEnabled.execute({
+    principal: writer,
+    input: { target: { kind: 'row', table: 'webhook', id: liveWebhookId }, enabled: true },
+  })
+  expect(enabled.trigger).toMatchObject({ kind: 'webhook', enabled: true })
+  for (const id of [candidateWebhookId, retiredWebhookId, oldWebhookId]) {
+    expect((await webhookRow(id)).isActive).toBe(false)
+  }
+  await expect.poll(async () => (await auditRows(liveWebhookId)).length).toBe(3)
+  expect((await auditRows(liveWebhookId)).every((row) => row.action === 'webhook.updated')).toBe(
+    true
+  )
+})
+
+check('enabling a poller clears its failures and backoff but keeps its config', async () => {
+  const result = await setBlock('poll', true)
+  expect(result.trigger).toMatchObject({ kind: 'polling', enabled: true, failedCount: 0 })
+  const row = await webhookRow(pollingWebhookId)
+  expect(row).toMatchObject({ isActive: true, failedCount: 0 })
+  expect(row.providerConfig).toEqual({ labelIds: ['INBOX'] })
+
+  const scalar = await setBlock('scalar', true)
+  expect(scalar.changes).toHaveLength(1)
+  expect((await webhookRow(scalarConfigWebhookId)).providerConfig).toBe('legacy-scalar-config')
+})
+
+check('a poll failure auto-disables at the threshold in the same write', async () => {
+  const logger = createLogger('TriggerHealthIntegration')
+  await markWebhookFailed(flakyWebhookId, logger)
+  expect(await webhookRow(flakyWebhookId)).toMatchObject({ isActive: true, failedCount: 99 })
+  await markWebhookFailed(flakyWebhookId, logger)
+  expect(await webhookRow(flakyWebhookId)).toMatchObject({ isActive: false, failedCount: 100 })
+
+  await setBlock('flaky', true)
+  await markWebhookFailed(flakyWebhookId, logger)
+  expect(await webhookRow(flakyWebhookId)).toMatchObject({ isActive: true, failedCount: 1 })
+})
+
+check('an unknown block is not found', async () => {
+  await expect(setBlock('missing', true)).rejects.toMatchObject({
+    code: 'not_found',
+    message: 'Trigger not found',
+  })
+})
diff --git a/apps/sim/lib/workflows/application/triggers.ts b/apps/sim/lib/workflows/application/triggers.ts
new file mode 100644
index 00000000000..1478e43e929
--- /dev/null
+++ b/apps/sim/lib/workflows/application/triggers.ts
@@ -0,0 +1,253 @@
+import { AuditAction, AuditResourceType } from '@sim/audit'
+import type { Principal } from '@sim/auth/principal'
+import { capabilityGovernedPrincipalUserId } from '@/lib/core/application'
+import { OrchestrationError } from '@/lib/core/orchestration/types'
+import { assertWorkspaceCapability } from '@/lib/permission-groups/capability-assertions'
+import { defineAuthorizedWorkflowUseCase } from '@/lib/workflows/application/authorized-workflow-use-case'
+import {
+  type ActiveWorkflowApplicationContext,
+  resolveActiveWorkflowApplicationContext,
+} from '@/lib/workflows/application/context'
+import { requireMutableWorkflow } from '@/lib/workflows/application/mutability'
+import { workflowOperations } from '@/lib/workflows/application/operations'
+import {
+  assertedWorkflowWorkspaceId,
+  resolvePrincipalWorkflowContext,
+} from '@/lib/workflows/application/principal-scope'
+import { validateCronExpression } from '@/lib/workflows/schedules/utils'
+import {
+  findLiveBlockTriggers,
+  findTriggerRowOwner,
+  isInRequestedState,
+  type LiveWorkflowTriggerRecord,
+  listLiveWorkflowTriggers,
+  setLiveTriggersEnabled,
+  type TriggerLiveness,
+  toWorkflowTriggerHealth,
+  type WorkflowTriggerHealth,
+  type WorkflowTriggerStatus,
+} from '@/lib/workflows/triggers/health'
+
+const TRIGGER_NOT_FOUND = 'Trigger not found'
+
+interface ListWorkflowTriggersInput {
+  workflowId: string
+  assertedWorkspaceId?: string
+}
+
+/** Every trigger the live deployment serves, disabled ones included. */
+export const listWorkflowTriggers = defineAuthorizedWorkflowUseCase({
+  operation: workflowOperations.readTriggers,
+  resolveContext: resolvePrincipalWorkflowContext<ListWorkflowTriggersInput>,
+  async execute({ context }) {
+    const records = await listLiveWorkflowTriggers(context.workflowId)
+    return { triggers: records.map(toWorkflowTriggerHealth) }
+  },
+})
+
+/**
+ * Which trigger to switch. Public callers name a block of the live deployment;
+ * the legacy editor routes still name a row id, which is resolved to its block
+ * and must itself be the live row — a row of an older version, or a registration
+ * that is not active, answers not found rather than being revived.
+ */
+type WorkflowTriggerTarget =
+  | { kind: 'block'; workflowId: string; blockId: string; assertedWorkspaceId?: string }
+  | { kind: 'row'; table: LiveWorkflowTriggerRecord['table']; id: string }
+
+interface SetWorkflowTriggerEnabledInput {
+  target: WorkflowTriggerTarget
+  enabled: boolean
+}
+
+interface TriggerTargetContext extends ActiveWorkflowApplicationContext {
+  blockId: string
+  row?: { table: LiveWorkflowTriggerRecord['table']; id: string }
+  notFoundMessage: string
+}
+
+interface SetWorkflowTriggerEnabledResult {
+  /** The target after the update; for a block with several live rows, the one the editor shows. */
+  trigger: WorkflowTriggerHealth
+  record: LiveWorkflowTriggerRecord
+  /** Every row this request switched; empty when all were already in the requested state. */
+  changes: { record: LiveWorkflowTriggerRecord; previousStatus: WorkflowTriggerStatus }[]
+}
+
+async function resolveTriggerTargetContext({
+  principal,
+  input,
+}: {
+  principal: Principal
+  input: SetWorkflowTriggerEnabledInput
+}): Promise<TriggerTargetContext> {
+  const { target } = input
+  if (target.kind === 'block') {
+    const workflowContext = await resolveActiveWorkflowApplicationContext({
+      workflowId: target.workflowId,
+      assertedWorkspaceId: assertedWorkflowWorkspaceId(principal, target.assertedWorkspaceId),
+    })
+    return { ...workflowContext, blockId: target.blockId, notFoundMessage: TRIGGER_NOT_FOUND }
+  }
+
+  const { table } = target
+  const notFoundMessage = table === 'schedule' ? 'Schedule not found' : 'Webhook not found'
+  const row = await findTriggerRowOwner(table, target.id)
+  // A row with no workflow or block is never switchable, and answering before the tenant check
+  // must not tell a caller anything beyond what a missing row would.
+  if (!row?.workflowId || !row.blockId) throw new OrchestrationError('not_found', notFoundMessage)
+  const workflowContext = await resolveActiveWorkflowApplicationContext({
+    workflowId: row.workflowId,
+  })
+  return {
+    ...workflowContext,
+    blockId: row.blockId,
+    row: { table, id: row.id },
+    notFoundMessage,
+  }
+}
+
+/** The run a re-enabled schedule is armed for, refused when its cron has none ahead. */
+function nextScheduledRun(
+  row: Extract<LiveWorkflowTriggerRecord, { table: 'schedule' }>['row']
+): Date {
+  if (!row.cronExpression) {
+    throw new OrchestrationError('validation', 'Schedule has no cron expression')
+  }
+  const cron = validateCronExpression(row.cronExpression, row.timezone || 'UTC')
+  if (!cron.isValid || !cron.nextRun) {
+    throw new OrchestrationError('validation', 'Schedule has invalid cron expression')
+  }
+  return cron.nextRun
+}
+
+/**
+ * permission-group-enforced: triggers.webhook — enabling a dormant webhook row
+ * makes the workflow reachable again exactly as creating it would, so it is
+ * gated here, where the row's table is known. Only this direction: disabling
+ * must stay open, or a policy change would strand a member with a live webhook
+ * they cannot switch off. Keyed to the governed subject, so a workspace key —
+ * which no group governs — is not judged against its creator's group.
+ */
+async function assertWebhookEnableAllowed(
+  principal: Principal,
+  context: TriggerTargetContext
+): Promise<void> {
+  const governedUserId = capabilityGovernedPrincipalUserId(principal)
+  if (!governedUserId) return
+  await assertWorkspaceCapability(
+    governedUserId,
+    context.workspaceId,
+    'triggers.webhook',
+    context.workspaceOrganizationId
+  )
+}
+
+/**
+ * Pauses or resumes one trigger of the live deployment without redeploying.
+ * A block target switches every live row of the block together; a legacy row
+ * target switches exactly that row.
+ *
+ * Enabling resets the failure counters, re-arms a schedule's next run, and
+ * clears a poller's backoff. Disabling only closes Sim's gate: nothing is torn
+ * down at the provider. A request for the state the trigger is already in
+ * changes nothing and records no audit. A disable lasts until the next deploy
+ * or version activation, which re-registers every trigger enabled, exactly as
+ * it already clears an automatic disable.
+ */
+export const setWorkflowTriggerEnabled = defineAuthorizedWorkflowUseCase({
+  operation: workflowOperations.setTriggerEnabled,
+  resolveContext: resolveTriggerTargetContext,
+  async execute({ principal, input, context }): Promise<SetWorkflowTriggerEnabledResult> {
+    // A legacy row id keeps the rows the editor shows; a block target sees only what runs.
+    const liveness: TriggerLiveness = context.row ? 'editor' : 'running'
+    const readLive = () => findLiveBlockTriggers({ ...context, liveness })
+    const [, current] = await Promise.all([requireMutableWorkflow(context.workflowId), readLive()])
+    if (current.length === 0) throw new OrchestrationError('not_found', context.notFoundMessage)
+    const pending = current.filter((record) => !isInRequestedState(record, input.enabled))
+    if (pending.length === 0) {
+      return { trigger: toWorkflowTriggerHealth(current[0]), record: current[0], changes: [] }
+    }
+
+    const schedules: { id: string; nextRunAt: Date | null }[] = []
+    const webhookIds: string[] = []
+    for (const record of pending) {
+      if (record.table === 'schedule') {
+        schedules.push({
+          id: record.row.id,
+          nextRunAt: input.enabled ? nextScheduledRun(record.row) : null,
+        })
+      } else {
+        webhookIds.push(record.row.id)
+      }
+    }
+    if (input.enabled && webhookIds.length > 0) await assertWebhookEnableAllowed(principal, context)
+
+    const updated = await setLiveTriggersEnabled({
+      workflowId: context.workflowId,
+      enabled: input.enabled,
+      schedules,
+      webhookIds,
+      liveness,
+      now: new Date(),
+    })
+    if (!updated) {
+      // Nothing was written. Succeed only if a concurrent request already reached the requested state.
+      const latest = await readLive()
+      if (latest.length === 0) throw new OrchestrationError('not_found', context.notFoundMessage)
+      if (latest.some((record) => !isInRequestedState(record, input.enabled))) {
+        throw new OrchestrationError('conflict', 'Trigger changed during the update; retry')
+      }
+      return { trigger: toWorkflowTriggerHealth(latest[0]), record: latest[0], changes: [] }
+    }
+
+    const previousStatusById = new Map(
+      pending.map((record) => [record.row.id, toWorkflowTriggerHealth(record).status])
+    )
+    const primary = updated.find((record) => record.row.id === current[0].row.id) ?? current[0]
+    return {
+      trigger: toWorkflowTriggerHealth(primary),
+      record: primary,
+      changes: updated.map((record) => ({
+        record,
+        previousStatus:
+          previousStatusById.get(record.row.id) ?? toWorkflowTriggerHealth(record).status,
+      })),
+    }
+  },
+  projectAudit: ({ input, context, result }) =>
+    result.changes.map(({ record, previousStatus }) => {
+      const verb = input.enabled ? 'Enabled' : 'Disabled'
+      const shared = {
+        workflowId: context.workflowId,
+        blockId: record.row.blockId,
+        enabled: input.enabled,
+        previousStatus,
+      }
+      if (record.table === 'schedule') {
+        const { row } = record
+        return {
+          action: AuditAction.SCHEDULE_UPDATED,
+          resourceType: AuditResourceType.SCHEDULE,
+          resourceId: row.id,
+          resourceName: row.jobTitle ?? undefined,
+          description: `${verb} schedule "${row.jobTitle ?? row.id}"`,
+          metadata: {
+            ...shared,
+            sourceType: row.sourceType,
+            cronExpression: row.cronExpression,
+            timezone: row.timezone,
+          },
+        }
+      }
+      const provider = record.row.provider || 'generic'
+      return {
+        action: AuditAction.WEBHOOK_UPDATED,
+        resourceType: AuditResourceType.WEBHOOK,
+        resourceId: record.row.id,
+        resourceName: provider,
+        description: `${verb} ${provider} webhook`,
+        metadata: { ...shared, provider },
+      }
+    }),
+})
diff --git a/apps/sim/lib/workflows/triggers/health.ts b/apps/sim/lib/workflows/triggers/health.ts
new file mode 100644
index 00000000000..219485c820d
--- /dev/null
+++ b/apps/sim/lib/workflows/triggers/health.ts
@@ -0,0 +1,381 @@
+import { db } from '@sim/db'
+import { webhook, workflow, workflowDeploymentVersion, workflowSchedule } from '@sim/db/schema'
+import { compareStrings } from '@sim/utils/string'
+import {
+  and,
+  asc,
+  desc,
+  eq,
+  exists,
+  inArray,
+  isNotNull,
+  isNull,
+  ne,
+  notExists,
+  or,
+  type SQL,
+  sql,
+} from 'drizzle-orm'
+import type { AnyPgColumn } from 'drizzle-orm/pg-core'
+import { resolveDeployedWebhookUrl } from '@/lib/webhooks/deployed-urls'
+import { POLL_BACKOFF_UNTIL_KEY, POLL_SOURCE_FAILURES_KEY } from '@/lib/webhooks/polling/utils'
+import {
+  isInternalTriggerProvider,
+  isPollingWebhookProvider,
+  POLLING_PROVIDERS,
+} from '@/triggers/constants'
+
+/**
+ * How a deployed trigger is fired: on a cron (`schedule`), by Sim polling a
+ * source (`polling`), by an event raised inside Sim (`event`), or by an external
+ * sender delivering to a URL (`webhook`).
+ */
+type WorkflowTriggerKind = 'schedule' | 'polling' | 'event' | 'webhook'
+
+/** `completed` is reachable only by schedules. */
+export type WorkflowTriggerStatus = 'active' | 'disabled' | 'completed'
+
+export interface WorkflowTriggerHealth {
+  blockId: string
+  kind: WorkflowTriggerKind
+  provider: string | null
+  enabled: boolean
+  status: WorkflowTriggerStatus
+  failedCount: number
+  lastFailedAt: Date | null
+  nextRunAt: Date | null
+  lastRanAt: Date | null
+  cronExpression: string | null
+  timezone: string | null
+  url: string | null
+}
+
+type ScheduleRow = typeof workflowSchedule.$inferSelect
+type WebhookRow = typeof webhook.$inferSelect
+
+/** One live trigger row, tagged with the table it lives in. */
+export type LiveWorkflowTriggerRecord =
+  | { table: 'schedule'; row: ScheduleRow & { blockId: string } }
+  | { table: 'webhook'; row: WebhookRow & { blockId: string } }
+
+function activeDeploymentVersionIds(workflowId: string) {
+  return db
+    .select({ id: workflowDeploymentVersion.id })
+    .from(workflowDeploymentVersion)
+    .where(
+      and(
+        eq(workflowDeploymentVersion.workflowId, workflowId),
+        eq(workflowDeploymentVersion.isActive, true)
+      )
+    )
+}
+
+/**
+ * Which rows count as live.
+ *
+ * `running` is what actually fires: a row of the active version of a deployed
+ * workflow. The schedule dispatcher and preprocessing's deployment check both
+ * require it, and webhook processing refuses a block missing from the active
+ * version, so a versionless row never runs. The v2 surface reports only these.
+ *
+ * `editor` is what the editor's reads show: the active version's rows, or,
+ * when the workflow has no active version, rows that name none (a deploy that
+ * predates versions, or a webhook saved before the first deploy). The legacy
+ * row-id routes keep switching those, as they always have.
+ */
+export type TriggerLiveness = 'running' | 'editor'
+
+function servedByDeployment(
+  deploymentVersionId: AnyPgColumn,
+  workflowId: string,
+  liveness: TriggerLiveness
+): SQL {
+  const predicate =
+    liveness === 'running'
+      ? and(
+          inArray(deploymentVersionId, activeDeploymentVersionIds(workflowId)),
+          exists(
+            db
+              .select({ id: workflow.id })
+              .from(workflow)
+              .where(and(eq(workflow.id, workflowId), eq(workflow.isDeployed, true)))
+          )
+        )
+      : or(
+          inArray(deploymentVersionId, activeDeploymentVersionIds(workflowId)),
+          and(isNull(deploymentVersionId), notExists(activeDeploymentVersionIds(workflowId)))
+        )
+  if (!predicate) throw new Error('Live deployment predicate is empty')
+  return predicate
+}
+
+/**
+ * The schedule rows the deployment serves. Job schedules share the table but
+ * belong to no workflow deployment, so they are excluded by source.
+ */
+function liveSchedulePredicate(workflowId: string, liveness: TriggerLiveness): SQL {
+  const predicate = and(
+    eq(workflowSchedule.workflowId, workflowId),
+    servedByDeployment(workflowSchedule.deploymentVersionId, workflowId, liveness),
+    isNull(workflowSchedule.archivedAt),
+    or(eq(workflowSchedule.sourceType, 'workflow'), isNull(workflowSchedule.sourceType)),
+    isNotNull(workflowSchedule.blockId)
+  )
+  if (!predicate) throw new Error('Live schedule predicate is empty')
+  return predicate
+}
+
+/**
+ * The webhook rows the deployment serves. `isActive` is the enablement gate,
+ * never the liveness test: a disabled row of the active version is still live,
+ * while a row of an older version, or a `candidate`, `retired`, or `orphaned`
+ * registration (always archived), is not, and must never be revived by an
+ * enable.
+ */
+function liveWebhookPredicate(workflowId: string, liveness: TriggerLiveness): SQL {
+  const predicate = and(
+    eq(webhook.workflowId, workflowId),
+    servedByDeployment(webhook.deploymentVersionId, workflowId, liveness),
+    isNull(webhook.archivedAt),
+    isNotNull(webhook.blockId)
+  )
+  if (!predicate) throw new Error('Live webhook predicate is empty')
+  return predicate
+}
+
+/**
+ * An unrecognized status reads as `active`, because the dispatcher fires every
+ * schedule that is neither `disabled` nor `completed` and the editor shows it
+ * as not disabled.
+ */
+function scheduleStatus(status: string): WorkflowTriggerStatus {
+  return status === 'disabled' || status === 'completed' ? status : 'active'
+}
+
+function webhookKind(provider: string | null): Exclude<WorkflowTriggerKind, 'schedule'> {
+  if (isPollingWebhookProvider(provider)) return 'polling'
+  if (isInternalTriggerProvider(provider)) return 'event'
+  return 'webhook'
+}
+
+/** Projects a live trigger row onto its health view. */
+export function toWorkflowTriggerHealth(record: LiveWorkflowTriggerRecord): WorkflowTriggerHealth {
+  if (record.table === 'schedule') {
+    const { row } = record
+    const status = scheduleStatus(row.status)
+    return {
+      blockId: row.blockId,
+      kind: 'schedule',
+      provider: null,
+      enabled: status === 'active',
+      status,
+      failedCount: row.failedCount,
+      lastFailedAt: row.lastFailedAt,
+      nextRunAt: row.nextRunAt,
+      lastRanAt: row.lastRanAt,
+      cronExpression: row.cronExpression,
+      timezone: row.timezone,
+      url: null,
+    }
+  }
+  const { row } = record
+  const kind = webhookKind(row.provider)
+  return {
+    blockId: row.blockId,
+    kind,
+    provider: row.provider,
+    enabled: row.isActive,
+    status: row.isActive ? 'active' : 'disabled',
+    failedCount: row.failedCount ?? 0,
+    lastFailedAt: row.lastFailedAt,
+    nextRunAt: null,
+    lastRanAt: null,
+    cronExpression: null,
+    timezone: null,
+    url: kind === 'webhook' ? resolveDeployedWebhookUrl(row) : null,
+  }
+}
+
+function withBlockId<R extends { blockId: string | null }>(row: R): R & { blockId: string } {
+  if (row.blockId === null) throw new Error('Live trigger row is missing its block')
+  return { ...row, blockId: row.blockId }
+}
+
+/** Every trigger the workflow's live deployment runs, ordered by block. */
+export async function listLiveWorkflowTriggers(
+  workflowId: string
+): Promise<LiveWorkflowTriggerRecord[]> {
+  const [schedules, webhooks] = await Promise.all([
+    db.select().from(workflowSchedule).where(liveSchedulePredicate(workflowId, 'running')),
+    db.select().from(webhook).where(liveWebhookPredicate(workflowId, 'running')),
+  ])
+  const records: LiveWorkflowTriggerRecord[] = [
+    ...schedules.map((row) => ({ table: 'schedule' as const, row: withBlockId(row) })),
+    ...webhooks.map((row) => ({ table: 'webhook' as const, row: withBlockId(row) })),
+  ]
+  return records.sort(
+    (a, b) => compareStrings(a.row.blockId, b.row.blockId) || compareStrings(a.row.id, b.row.id)
+  )
+}
+
+/**
+ * The workflow and block an unarchived trigger row belongs to, live or not.
+ * Liveness is decided afterwards, once the caller is authorized for the workflow.
+ */
+export async function findTriggerRowOwner(
+  table: LiveWorkflowTriggerRecord['table'],
+  rowId: string
+): Promise<{ id: string; workflowId: string | null; blockId: string | null } | null> {
+  const [row] =
+    table === 'schedule'
+      ? await db
+          .select({
+            id: workflowSchedule.id,
+            workflowId: workflowSchedule.workflowId,
+            blockId: workflowSchedule.blockId,
+          })
+          .from(workflowSchedule)
+          .where(and(eq(workflowSchedule.id, rowId), isNull(workflowSchedule.archivedAt)))
+          .limit(1)
+      : await db
+          .select({ id: webhook.id, workflowId: webhook.workflowId, blockId: webhook.blockId })
+          .from(webhook)
+          .where(and(eq(webhook.id, rowId), isNull(webhook.archivedAt)))
+          .limit(1)
+  return row ?? null
+}
+
+/**
+ * The live rows a block serves: its schedules, or else its webhook rows, newest
+ * first as the editor reads them. A block normally has one, but rows written
+ * before registrations were tracked can leave several. Pinned to `row`, only
+ * that row, and only while it is live: an older version, an archived row, or a
+ * non-active registration reads as absent.
+ */
+export async function findLiveBlockTriggers(input: {
+  workflowId: string
+  blockId: string
+  row?: { table: LiveWorkflowTriggerRecord['table']; id: string }
+  liveness: TriggerLiveness
+}): Promise<LiveWorkflowTriggerRecord[]> {
+  const { row } = input
+  const [schedules, webhooks] = await Promise.all([
+    row?.table === 'webhook'
+      ? []
+      : db
+          .select()
+          .from(workflowSchedule)
+          .where(
+            and(
+              liveSchedulePredicate(input.workflowId, input.liveness),
+              eq(workflowSchedule.blockId, input.blockId),
+              row ? eq(workflowSchedule.id, row.id) : undefined
+            )
+          )
+          .orderBy(asc(workflowSchedule.id)),
+    row?.table === 'schedule'
+      ? []
+      : db
+          .select()
+          .from(webhook)
+          .where(
+            and(
+              liveWebhookPredicate(input.workflowId, input.liveness),
+              eq(webhook.blockId, input.blockId),
+              row ? eq(webhook.id, row.id) : undefined
+            )
+          )
+          .orderBy(desc(webhook.updatedAt), asc(webhook.id)),
+  ])
+  if (schedules.length > 0) {
+    return schedules.map((schedule) => ({ table: 'schedule', row: withBlockId(schedule) }))
+  }
+  return webhooks.map((hook) => ({ table: 'webhook', row: withBlockId(hook) }))
+}
+
+/** Whether a live row already is in the requested state. Disabling moves a completed schedule to `disabled`. */
+export function isInRequestedState(record: LiveWorkflowTriggerRecord, enabled: boolean): boolean {
+  if (record.table === 'webhook') return record.row.isActive === enabled
+  return record.row.status === (enabled ? 'active' : 'disabled')
+}
+
+class TriggerSwitchConflict extends Error {}
+
+/**
+ * Switches live rows all or nothing. Each `UPDATE` re-evaluates liveness and
+ * the current state itself; if any row a redeploy retired, or another request
+ * switched, since the caller's read is not written, the whole transaction rolls
+ * back and the result is `null`, so a caller never commits part of a block.
+ *
+ * Enabling a schedule resets both failure counters and arms `nextRunAt`;
+ * disabling clears the pending run and any claim. Enabling a webhook resets the
+ * failure count and drops a poller's source backoff, so a re-enabled polling
+ * trigger is polled on the next tick instead of waiting out a window earned
+ * before it was disabled.
+ */
+export async function setLiveTriggersEnabled(input: {
+  workflowId: string
+  enabled: boolean
+  schedules: { id: string; nextRunAt: Date | null }[]
+  webhookIds: string[]
+  liveness: TriggerLiveness
+  now: Date
+}): Promise<LiveWorkflowTriggerRecord[] | null> {
+  const { workflowId, enabled, liveness, now } = input
+  try {
+    return await db.transaction(async (tx) => {
+      const updated: LiveWorkflowTriggerRecord[] = []
+      for (const schedule of input.schedules) {
+        const [row] = await tx
+          .update(workflowSchedule)
+          .set(
+            enabled
+              ? {
+                  status: 'active',
+                  failedCount: 0,
+                  infraRetryCount: 0,
+                  nextRunAt: schedule.nextRunAt,
+                  updatedAt: now,
+                }
+              : { status: 'disabled', nextRunAt: null, lastQueuedAt: null, updatedAt: now }
+          )
+          .where(
+            and(
+              eq(workflowSchedule.id, schedule.id),
+              liveSchedulePredicate(workflowId, liveness),
+              ne(workflowSchedule.status, enabled ? 'active' : 'disabled')
+            )
+          )
+          .returning()
+        if (!row) throw new TriggerSwitchConflict()
+        updated.push({ table: 'schedule', row: withBlockId(row) })
+      }
+
+      if (input.webhookIds.length > 0) {
+        /** Only a poller records backoff, and only an object config can hold it. */
+        const clearedBackoff = sql`CASE WHEN ${inArray(webhook.provider, [...POLLING_PROVIDERS])} AND jsonb_typeof(${webhook.providerConfig}::jsonb) = 'object' THEN (${webhook.providerConfig}::jsonb - ${POLL_BACKOFF_UNTIL_KEY}::text - ${POLL_SOURCE_FAILURES_KEY}::text)::json ELSE ${webhook.providerConfig} END`
+        const rows = await tx
+          .update(webhook)
+          .set(
+            enabled
+              ? { isActive: true, failedCount: 0, providerConfig: clearedBackoff, updatedAt: now }
+              : { isActive: false, updatedAt: now }
+          )
+          .where(
+            and(
+              inArray(webhook.id, input.webhookIds),
+              liveWebhookPredicate(workflowId, liveness),
+              eq(webhook.isActive, !enabled)
+            )
+          )
+          .returning()
+        if (rows.length !== input.webhookIds.length) throw new TriggerSwitchConflict()
+        for (const row of rows) updated.push({ table: 'webhook', row: withBlockId(row) })
+      }
+      return updated
+    })
+  } catch (error) {
+    if (error instanceof TriggerSwitchConflict) return null
+    throw error
+  }
+}
diff --git a/packages/audit/src/types.ts b/packages/audit/src/types.ts
index 2c897ceff76..d72dcf309df 100644
--- a/packages/audit/src/types.ts
+++ b/packages/audit/src/types.ts
@@ -234,6 +234,7 @@ export const AuditAction = {
 
   WEBHOOK_CREATED: 'webhook.created',
   WEBHOOK_DELETED: 'webhook.deleted',
+  WEBHOOK_UPDATED: 'webhook.updated',
 
   WORKFLOW_CREATED: 'workflow.created',
   WORKFLOW_UPDATED: 'workflow.updated',
diff --git a/packages/sim-cli/src/contract/commands.ts b/packages/sim-cli/src/contract/commands.ts
index 831ebd1c710..31e1b3f126f 100644
--- a/packages/sim-cli/src/contract/commands.ts
+++ b/packages/sim-cli/src/contract/commands.ts
@@ -355,6 +355,21 @@ export const CLI_CONTRACT: CliContract = {
     renamedFrom: ['workflows deployment list'],
     describe: 'Show a workflow’s current deployment',
   },
+  listWorkflowTriggers: {
+    describe: 'Show the health of every trigger the live deployment serves',
+    columns: [
+      { header: 'block', path: 'blockId' },
+      { header: 'kind' },
+      { header: 'provider' },
+      { header: 'status' },
+      { header: 'failures', path: 'failedCount' },
+      { header: 'last failed', path: 'lastFailedAt', format: 'timestamp' },
+      { header: 'next run', path: 'nextRunAt', format: 'timestamp' },
+    ],
+  },
+  updateWorkflowTrigger: {
+    describe: 'Enable or disable one trigger of the live deployment',
+  },
   setSecret: { hidden: true },
 
   // Soft deletes, all three: `tables restore`, `knowledge restore` and
diff --git a/packages/sim-cli/src/generated/v2-api.ts b/packages/sim-cli/src/generated/v2-api.ts
index 17621517fa6..7816e9eaf55 100644
--- a/packages/sim-cli/src/generated/v2-api.ts
+++ b/packages/sim-cli/src/generated/v2-api.ts
@@ -10855,6 +10855,33 @@ export type ListWorkflowsResponse = {
   nextCursor: string | null
 }
 
+/** `GET /api/v2/workflows/[workflowId]/triggers` */
+export type ListWorkflowTriggersParams = {
+  workflowId: string
+}
+
+export type ListWorkflowTriggersQuery = Record<string, unknown>
+
+type ListWorkflowTriggersResponseRef0 = {
+  blockId: string
+  kind: 'schedule' | 'polling' | 'event' | 'webhook'
+  provider: string | null
+  enabled: boolean
+  status: 'active' | 'disabled' | 'completed'
+  failedCount: number
+  lastFailedAt: string | null
+  nextRunAt: string | null
+  lastRanAt: string | null
+  cronExpression: string | null
+  timezone: string | null
+  url: string | null
+}
+
+export type ListWorkflowTriggersResponse = {
+  data: Array<ListWorkflowTriggersResponseRef0>
+  nextCursor: string | null
+}
+
 /** `GET /api/v2/workflows/[workflowId]/versions` */
 export type ListWorkflowVersionsParams = {
   workflowId: string
@@ -15792,6 +15819,37 @@ export type UpdateWorkflowPublicApiResponse = {
   data: UpdateWorkflowPublicApiResponseRef0
 }
 
+/** `PATCH /api/v2/workflows/[workflowId]/triggers/[blockId]` */
+export type UpdateWorkflowTriggerParams = {
+  workflowId: string
+  blockId: string
+}
+
+export type UpdateWorkflowTriggerQuery = Record<string, unknown>
+
+export type UpdateWorkflowTriggerBody = {
+  enabled: boolean
+}
+
+type UpdateWorkflowTriggerResponseRef0 = {
+  blockId: string
+  kind: 'schedule' | 'polling' | 'event' | 'webhook'
+  provider: string | null
+  enabled: boolean
+  status: 'active' | 'disabled' | 'completed'
+  failedCount: number
+  lastFailedAt: string | null
+  nextRunAt: string | null
+  lastRanAt: string | null
+  cronExpression: string | null
+  timezone: string | null
+  url: string | null
+}
+
+export type UpdateWorkflowTriggerResponse = {
+  data: UpdateWorkflowTriggerResponseRef0
+}
+
 /** `PATCH /api/v2/workflows/[workflowId]/versions/[version]` */
 export type UpdateWorkflowVersionParams = {
   version: number
@@ -22370,6 +22428,14 @@ export const V2_OPERATIONS = {
       },
     },
   },
+  listWorkflowTriggers: {
+    method: 'GET',
+    path: '/api/v2/workflows/[workflowId]/triggers',
+    pathParams: ['workflowId'] as const,
+    pathParamDocs: { workflowId: 'Unique workflow identifier.' },
+    responseMode: 'json',
+    summary: 'List workflow triggers',
+  },
   listWorkflowVersions: {
     method: 'GET',
     path: '/api/v2/workflows/[workflowId]/versions',
@@ -25041,6 +25107,25 @@ export const V2_OPERATIONS = {
       },
     },
   },
+  updateWorkflowTrigger: {
+    method: 'PATCH',
+    path: '/api/v2/workflows/[workflowId]/triggers/[blockId]',
+    pathParams: ['workflowId', 'blockId'] as const,
+    pathParamDocs: {
+      workflowId: 'Unique workflow identifier.',
+      blockId: 'Trigger block in the deployed workflow.',
+    },
+    responseMode: 'json',
+    summary: 'Update workflow trigger',
+    body: {
+      enabled: {
+        kind: 'boolean',
+        required: true,
+        describe:
+          'Whether the trigger should fire. A disable lasts until the next deploy or version activation, which re-enables every trigger.',
+      },
+    },
+  },
   updateWorkflowVersion: {
     method: 'PATCH',
     path: '/api/v2/workflows/[workflowId]/versions/[version]',
diff --git a/packages/sim-cli/src/http/client.test.ts b/packages/sim-cli/src/http/client.test.ts
index 3d61ffca49f..65c1af026f4 100644
--- a/packages/sim-cli/src/http/client.test.ts
+++ b/packages/sim-cli/src/http/client.test.ts
@@ -565,6 +565,7 @@ describe('destructive operations are gated', () => {
     'updateWorkflowGroup',
     'updateWorkflowMcpServer',
     'updateWorkflowPublicApi',
+    'updateWorkflowTrigger',
     'updateWorkflowVersion',
     'uploadKnowledgeDocument',
     'upsertFileShare',
diff --git a/packages/testing/src/mocks/audit.mock.ts b/packages/testing/src/mocks/audit.mock.ts
index 50af39f6484..0335df3d1ee 100644
--- a/packages/testing/src/mocks/audit.mock.ts
+++ b/packages/testing/src/mocks/audit.mock.ts
@@ -193,6 +193,7 @@ const AuditAction = {
   TABLE_EXPORTED: 'table.exported',
   WEBHOOK_CREATED: 'webhook.created',
   WEBHOOK_DELETED: 'webhook.deleted',
+  WEBHOOK_UPDATED: 'webhook.updated',
   WORKFLOW_CREATED: 'workflow.created',
   WORKFLOW_UPDATED: 'workflow.updated',
   WORKFLOW_DELETED: 'workflow.deleted',
diff --git a/scripts/check-source-conventions.baseline.json b/scripts/check-source-conventions.baseline.json
index de87ae606f5..5b00b8d034f 100644
--- a/scripts/check-source-conventions.baseline.json
+++ b/scripts/check-source-conventions.baseline.json
@@ -8,7 +8,6 @@
     "apps/sim/app/api/cron/renew-subscriptions/route.test.ts": 1,
     "apps/sim/app/api/desktop/tool/authorize/route.test.ts": 1,
     "apps/sim/app/api/mcp/oauth/start/route.test.ts": 1,
-    "apps/sim/app/api/schedules/[id]/route.test.ts": 1,
     "apps/sim/app/api/v2/workflows/[workflowId]/execute/route.test.ts": 1,
     "apps/sim/app/api/webhooks/poll/[provider]/route.test.ts": 1,
     "apps/sim/app/api/workflows/[id]/execute/route.test.ts": 1,
diff --git a/scripts/check-test-types.baseline.json b/scripts/check-test-types.baseline.json
index 9ae010b7853..b4defe1f4e3 100644
--- a/scripts/check-test-types.baseline.json
+++ b/scripts/check-test-types.baseline.json
@@ -401,10 +401,6 @@
   "apps/sim/app/api/v2/workflows/[workflowId]/versions/[version]/route.test.ts": {
     "TS2345": 1
   },
-  "apps/sim/app/api/webhooks/[id]/reactivation-gate.test.ts": {
-    "TS2322": 1,
-    "TS2345": 1
-  },
   "apps/sim/app/api/webhooks/agentmail/route.test.ts": {
     "TS2554": 6
   },
diff --git a/scripts/check-unused-exports.baseline.json b/scripts/check-unused-exports.baseline.json
index 999b30ba486..412b6fbc44f 100644
--- a/scripts/check-unused-exports.baseline.json
+++ b/scripts/check-unused-exports.baseline.json
@@ -5026,7 +5026,6 @@
   "apps/sim/lib/uploads/utils/validation.ts#SupportedVideoExtension",
   "apps/sim/lib/uploads/utils/validation.ts#isValidPng",
   "apps/sim/lib/url-state/index.ts#SortDefault",
-  "apps/sim/lib/webhooks/deployed-urls.ts#resolveDeployedWebhookUrl",
   "apps/sim/lib/webhooks/env-resolver.ts#normalizeWebhookProviderConfig",
   "apps/sim/lib/webhooks/execution-principal.ts#normalizeWebhookPrincipalProvider",
   "apps/sim/lib/webhooks/polling/types.ts#getProviderConfigRecord",
diff --git a/scripts/generate-cli-docs.ts b/scripts/generate-cli-docs.ts
index 9d442e36519..6b5320089ef 100644
--- a/scripts/generate-cli-docs.ts
+++ b/scripts/generate-cli-docs.ts
@@ -392,7 +392,16 @@ const GROUP_SECTIONS: Record<string, readonly CommandSection[]> = {
     { title: 'Draft state and edits', prefixes: ['state', 'operations', 'variables'] },
     {
       title: 'Deployments and versions',
-      prefixes: ['deploy', 'deployment', 'versions', 'activate', 'revert', 'rollback', 'undeploy'],
+      prefixes: [
+        'deploy',
+        'deployment',
+        'triggers',
+        'versions',
+        'activate',
+        'revert',
+        'rollback',
+        'undeploy',
+      ],
     },
     {
       title: 'Run and inspect',