fix(execution): sweep a removed workspace's Trigger.dev runs of any age

The workspace sweep searched only runs created in the last 14 days, so an older
run still queued or waiting survived the removal. It now passes only an upper
time bound, which Trigger.dev treats as no lower bound, and falls back to the
retention window only if Trigger.dev refuses that search as too expensive
before returning a run. cancelByExecution is unchanged.

Author: waleedlatif1

apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts

@@ -948,3 +948,41 @@ describe('TriggerDevJobQueue workspace cancellation past refusing runs', () => {
     expect(mockCancel).toHaveBeenCalledWith('run-9')
   })
 })
+
+describe('TriggerDevJobQueue workspace cancellation of long-lived runs', () => {
+  beforeEach(() => {
+    mockList.mockReset()
+    mockCancel.mockReset().mockResolvedValue(undefined)
+  })
+
+  it('searches every active run of the workspace, however long ago it was created', async () => {
+    mockList.mockReturnValue(createListPage([{ id: 'run-old', tags: ['workspaceId:removed'] }]))
+    const queue = new TriggerDevJobQueue()
+
+    await queue.cancelByWorkspace('removed', { maxRuns: 100 })
+
+    const [filter] = mockList.mock.calls[0] as [Record<string, unknown>]
+    expect(filter.from).toBeUndefined()
+    expect(filter.period).toBeUndefined()
+    expect(filter.to).toBeInstanceOf(Date)
+    expect(mockCancel).toHaveBeenCalledWith('run-old')
+  })
+
+  it('falls back to the retention window when Trigger.dev refuses the unbounded search', async () => {
+    mockList
+      .mockImplementationOnce(() => {
+        throw new MockApiError(422, { message: 'narrow the time window' })
+      })
+      .mockReturnValueOnce(createListPage([{ id: 'run-recent', tags: ['workspaceId:removed'] }]))
+    const queue = new TriggerDevJobQueue()
+
+    await expect(queue.cancelByWorkspace('removed', { maxRuns: 100 })).resolves.toEqual({
+      cancelledJobs: 1,
+      complete: true,
+    })
+
+    const [fallback] = mockList.mock.calls[1] as [Record<string, unknown>]
+    expect(fallback.from).toBeInstanceOf(Date)
+    expect(mockCancel).toHaveBeenCalledWith('run-recent')
+  })
+})

apps/sim/lib/core/async-jobs/backends/trigger-dev.ts

@@ -252,6 +252,45 @@ async function scanAndCancelTriggerRuns(options: CancellationScanOptions): Promi
   }
 }
 
+/**
+ * Lists a workspace's active workflow runs however old they are: a removal must reach a run that
+ * has been queued or waiting for longer than any recent window. Trigger.dev searches only the
+ * last 7 days unless a time bound is given, and passing only `to` removes the lower bound. When
+ * Trigger.dev refuses that search as too expensive (422) before returning a run, the listing
+ * falls back to the queue-wait retention window rather than sweeping nothing.
+ */
+async function* listActiveWorkspaceRuns(
+  workspaceId: string,
+  workspaceTag: string
+): AsyncIterable<CancellationListRun> {
+  const filter = {
+    tag: workspaceTag,
+    taskIdentifier: [...WORKSPACE_EXECUTION_JOB_TYPES],
+    status: [...ACTIVE_TRIGGER_RUN_STATUSES],
+    limit: WORKSPACE_CANCELLATION_LIST_PAGE_SIZE,
+  }
+  let listed = false
+  try {
+    for await (const run of runs.list({ ...filter, to: new Date() })) {
+      listed = true
+      yield run
+    }
+    return
+  } catch (error) {
+    if (listed || !(error instanceof ApiError) || error.status !== 422) throw error
+    logger.warn(
+      'Trigger.dev refused an unbounded workspace run search; using the retention window',
+      {
+        workspaceId,
+      }
+    )
+  }
+  yield* runs.list({
+    ...filter,
+    from: new Date(Date.now() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000),
+  })
+}
+
 /**
  * Maps trigger.dev task IDs to our JobType
  */
@@ -679,14 +718,7 @@ export class TriggerDevJobQueue implements JobQueueBackend {
 
     await scanAndCancelTriggerRuns({
       cancelJob: (jobId) => this.cancelJob(jobId),
-      listRuns: () =>
-        runs.list({
-          tag: workspaceTag,
-          taskIdentifier: [...WORKSPACE_EXECUTION_JOB_TYPES],
-          status: [...ACTIVE_TRIGGER_RUN_STATUSES],
-          from: new Date(Date.now() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000),
-          limit: WORKSPACE_CANCELLATION_LIST_PAGE_SIZE,
-        }),
+      listRuns: () => listActiveWorkspaceRuns(workspaceId, workspaceTag),
       /**
        * Only cancelled runs count toward the cap, so runs that keep refusing cancellation cannot
        * hold every retry at the front of the listing; a separate bound on failures keeps one call