improvement(triggers): add tags to all trigger.dev task invocations (#3878)

* improvement(triggers): add tags to all trigger.dev task invocations

* fix(triggers): prefix unused type param in buildTags

* fix(triggers): remove unused type param from buildTags

Author: waleedlatif1

apps/sim/app/api/schedules/execute/route.ts

@@ -146,7 +146,11 @@ export async function GET(request: NextRequest) {
           })
         } else {
           jobId = await jobQueue.enqueue('schedule-execution', payload, {
-            metadata: { workflowId: schedule.workflowId ?? undefined, correlation },
+            metadata: {
+              workflowId: schedule.workflowId ?? undefined,
+              workspaceId: resolvedWorkspaceId ?? undefined,
+              correlation,
+            },
           })
         }
         logger.info(

apps/sim/app/api/webhooks/agentmail/route.ts

@@ -162,7 +162,13 @@ export async function POST(req: Request) {
 
     if (isTriggerDevEnabled) {
       try {
-        const handle = await tasks.trigger('mothership-inbox-execution', { taskId })
+        const handle = await tasks.trigger(
+          'mothership-inbox-execution',
+          { taskId },
+          {
+            tags: [`workspaceId:${result.id}`, `taskId:${taskId}`],
+          }
+        )
         await db
           .update(mothershipInboxTask)
           .set({ triggerJobId: handle.id })

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -237,7 +237,7 @@ async function handleAsyncExecution(params: AsyncExecutionParams): Promise<NextR
           },
         })
       : await jobQueue!.enqueue('workflow-execution', payload, {
-          metadata: { workflowId, userId, correlation },
+          metadata: { workflowId, workspaceId, userId, correlation },
         })
 
     asyncLogger.info('Queued async workflow execution', { jobId })

apps/sim/app/workspace/[workspaceId]/knowledge/components/edit-knowledge-base-modal/edit-knowledge-base-modal.tsx

@@ -146,34 +146,34 @@ export const EditKnowledgeBaseModal = memo(function EditKnowledgeBaseModal({
                   <Label>Chunking Configuration</Label>
                   <div className='grid grid-cols-3 gap-2'>
                     <div className='rounded-sm border border-[var(--border-1)] bg-[var(--surface-2)] px-2.5 py-2'>
-                      <p className='text-[var(--text-tertiary)] text-[11px] leading-tight'>
+                      <p className='text-[11px] text-[var(--text-tertiary)] leading-tight'>
                         Max Size
                       </p>
                       <p className='font-medium text-[var(--text-primary)] text-sm'>
                         {chunkingConfig.maxSize.toLocaleString()}
-                        <span className='ml-0.5 font-normal text-[var(--text-tertiary)] text-[11px]'>
+                        <span className='ml-0.5 font-normal text-[11px] text-[var(--text-tertiary)]'>
                           tokens
                         </span>
                       </p>
                     </div>
                     <div className='rounded-sm border border-[var(--border-1)] bg-[var(--surface-2)] px-2.5 py-2'>
-                      <p className='text-[var(--text-tertiary)] text-[11px] leading-tight'>
+                      <p className='text-[11px] text-[var(--text-tertiary)] leading-tight'>
                         Min Size
                       </p>
                       <p className='font-medium text-[var(--text-primary)] text-sm'>
                         {chunkingConfig.minSize.toLocaleString()}
-                        <span className='ml-0.5 font-normal text-[var(--text-tertiary)] text-[11px]'>
+                        <span className='ml-0.5 font-normal text-[11px] text-[var(--text-tertiary)]'>
                           chars
                         </span>
                       </p>
                     </div>
                     <div className='rounded-sm border border-[var(--border-1)] bg-[var(--surface-2)] px-2.5 py-2'>
-                      <p className='text-[var(--text-tertiary)] text-[11px] leading-tight'>
+                      <p className='text-[11px] text-[var(--text-tertiary)] leading-tight'>
                         Overlap
                       </p>
                       <p className='font-medium text-[var(--text-primary)] text-sm'>
                         {chunkingConfig.overlap.toLocaleString()}
-                        <span className='ml-0.5 font-normal text-[var(--text-tertiary)] text-[11px]'>
+                        <span className='ml-0.5 font-normal text-[11px] text-[var(--text-tertiary)]'>
                           tokens
                         </span>
                       </p>

apps/sim/lib/a2a/push-notifications.ts

@@ -114,7 +114,12 @@ export async function notifyTaskStateChange(taskId: string, state: TaskState): P
       const { a2aPushNotificationTask } = await import(
         '@/background/a2a-push-notification-delivery'
       )
-      await a2aPushNotificationTask.trigger({ taskId, state })
+      await a2aPushNotificationTask.trigger(
+        { taskId, state },
+        {
+          tags: [`taskId:${taskId}`],
+        }
+      )
       logger.info('Push notification queued to trigger.dev', { taskId, state })
     } catch (error) {
       logger.warn('Failed to queue push notification, falling back to inline delivery', {

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

@@ -66,9 +66,10 @@ export class TriggerDevJobQueue implements JobQueueBackend {
         ? { ...payload, ...options.metadata }
         : payload
 
-    const handle = await tasks.trigger(taskId, enrichedPayload)
+    const tags = buildTags(options)
+    const handle = await tasks.trigger(taskId, enrichedPayload, tags.length > 0 ? { tags } : {})
 
-    logger.debug('Enqueued job via trigger.dev', { jobId: handle.id, type, taskId })
+    logger.debug('Enqueued job via trigger.dev', { jobId: handle.id, type, taskId, tags })
     return handle.id
   }
 
@@ -121,3 +122,33 @@ export class TriggerDevJobQueue implements JobQueueBackend {
 
   async markJobFailed(_jobId: string, _error: string): Promise<void> {}
 }
+
+/**
+ * Derives trigger.dev tags from job type, metadata, and explicit tags.
+ * Tags follow the `namespace:value` convention for consistent filtering.
+ * Max 10 tags per run, each max 128 chars.
+ */
+function buildTags(options?: EnqueueOptions): string[] {
+  const tags: string[] = []
+  const meta = options?.metadata
+
+  if (meta?.workspaceId) tags.push(`workspaceId:${meta.workspaceId}`)
+  if (meta?.workflowId) tags.push(`workflowId:${meta.workflowId}`)
+  if (meta?.userId) tags.push(`userId:${meta.userId}`)
+
+  if (meta?.correlation) {
+    const c = meta.correlation
+    tags.push(`source:${c.source}`)
+    if (c.webhookId) tags.push(`webhookId:${c.webhookId}`)
+    if (c.scheduleId) tags.push(`scheduleId:${c.scheduleId}`)
+    if (c.provider) tags.push(`provider:${c.provider}`)
+  }
+
+  if (options?.tags) {
+    for (const tag of options.tags) {
+      if (!tags.includes(tag)) tags.push(tag)
+    }
+  }
+
+  return tags.slice(0, 10)
+}

apps/sim/lib/core/async-jobs/types.ts

@@ -54,6 +54,7 @@ export interface Job<TPayload = unknown, TOutput = unknown> {
 
 export interface JobMetadata {
   workflowId?: string
+  workspaceId?: string
   userId?: string
   correlation?: AsyncExecutionCorrelation
   [key: string]: unknown
@@ -66,6 +67,7 @@ export interface EnqueueOptions {
   priority?: number
   name?: string
   delayMs?: number
+  tags?: string[]
 }
 
 /**

apps/sim/lib/knowledge/connectors/sync-engine.ts

@@ -152,13 +152,30 @@ export async function dispatchSync(
   const requestId = options?.requestId ?? crypto.randomUUID()
 
   if (isTriggerAvailable()) {
+    const connectorRows = await db
+      .select({
+        knowledgeBaseId: knowledgeConnector.knowledgeBaseId,
+        workspaceId: knowledgeBase.workspaceId,
+        userId: knowledgeBase.userId,
+      })
+      .from(knowledgeConnector)
+      .innerJoin(knowledgeBase, eq(knowledgeBase.id, knowledgeConnector.knowledgeBaseId))
+      .where(eq(knowledgeConnector.id, connectorId))
+      .limit(1)
+
+    const row = connectorRows[0]
+    const tags = [`connectorId:${connectorId}`]
+    if (row?.knowledgeBaseId) tags.push(`knowledgeBaseId:${row.knowledgeBaseId}`)
+    if (row?.workspaceId) tags.push(`workspaceId:${row.workspaceId}`)
+    if (row?.userId) tags.push(`userId:${row.userId}`)
+
     await knowledgeConnectorSync.trigger(
       {
         connectorId,
         fullSync: options?.fullSync,
         requestId,
       },
-      { tags: [`connector:${connectorId}`] }
+      { tags }
     )
     logger.info(`Dispatched connector sync to Trigger.dev`, { connectorId, requestId })
   } else if (isBullMQEnabled()) {

apps/sim/lib/knowledge/documents/service.ts

@@ -120,7 +120,9 @@ export interface DocumentJobData {
 
 export async function dispatchDocumentProcessingJob(payload: DocumentJobData): Promise<void> {
   if (isTriggerAvailable()) {
-    await tasks.trigger('knowledge-process-document', payload)
+    await tasks.trigger('knowledge-process-document', payload, {
+      tags: [`knowledgeBaseId:${payload.knowledgeBaseId}`, `documentId:${payload.documentId}`],
+    })
     return
   }
 
@@ -692,7 +694,7 @@ export async function processDocumentsWithTrigger(
           payload: doc,
           options: {
             idempotencyKey: `doc-process-${doc.documentId}-${requestId}`,
-            tags: [`kb:${doc.knowledgeBaseId}`, `doc:${doc.documentId}`],
+            tags: [`knowledgeBaseId:${doc.knowledgeBaseId}`, `documentId:${doc.documentId}`],
           },
         }))
       )

apps/sim/lib/logs/events.ts

@@ -139,7 +139,13 @@ export async function emitWorkflowExecutionCompleted(log: WorkflowExecutionLog):
       }
 
       if (isTriggerDevEnabled) {
-        await workspaceNotificationDeliveryTask.trigger(payload)
+        await workspaceNotificationDeliveryTask.trigger(payload, {
+          tags: [
+            `workspaceId:${workspaceId}`,
+            `workflowId:${log.workflowId}`,
+            `notificationType:${subscription.notificationType}`,
+          ],
+        })
         logger.info(
           `Enqueued ${subscription.notificationType} notification ${deliveryId} via Trigger.dev`
         )