feat(trigger): add org tags to all trigger job runs

apps/api/src/trigger/browser-automation/run-browser-automation.ts

@@ -1,5 +1,5 @@
 import { db } from '@db';
-import { logger, task } from '@trigger.dev/sdk';
+import { logger, tags, task } from '@trigger.dev/sdk';
 import { BrowserbaseService } from '../../browserbase/browserbase.service';
 import { triggerEmail } from '../../email/trigger-email';
 import { TaskStatusChangedEmail } from '../../email/templates/task-status-changed';
@@ -187,6 +187,8 @@ export const runBrowserAutomation = task({
   }) => {
     const { automationId, automationName, organizationId, taskId } = payload;
 
+    await tags.add([`org:${organizationId}`]);
+
     logger.info(`Running browser automation "${automationName}"`, {
       automationId,
       organizationId,

apps/api/src/trigger/cloud-security/run-cloud-security-scan.ts

@@ -1,5 +1,5 @@
 import { db } from '@db';
-import { logger, task } from '@trigger.dev/sdk';
+import { logger, tags, task } from '@trigger.dev/sdk';
 
 /**
  * Trigger task that runs a cloud security scan for a single connection.
@@ -20,6 +20,8 @@ export const runCloudSecurityScan = task({
     const { connectionId, organizationId, providerSlug, connectionName } =
       payload;
 
+    await tags.add([`org:${organizationId}`]);
+
     logger.info(
       `Starting cloud security scan for connection: ${connectionName}`,
       {

apps/api/src/trigger/integration-platform/run-connection-checks.ts

@@ -1,6 +1,6 @@
 import { getManifest, runAllChecks } from '@trycompai/integration-platform';
 import { db } from '@db';
-import { logger, task } from '@trigger.dev/sdk';
+import { logger, tags, task } from '@trigger.dev/sdk';
 
 /**
  * Trigger task that runs all checks for a connection.
@@ -22,6 +22,8 @@ export const runConnectionChecks = task({
   }) => {
     const { connectionId, organizationId, providerSlug } = payload;
 
+    await tags.add([`org:${organizationId}`]);
+
     logger.info(`Auto-running checks for connection ${connectionId}`, {
       provider: providerSlug,
       organizationId,

apps/api/src/trigger/integration-platform/run-task-integration-checks.ts

@@ -1,6 +1,6 @@
 import { getManifest, runAllChecks } from '@trycompai/integration-platform';
 import { db } from '@db';
-import { logger, task } from '@trigger.dev/sdk';
+import { logger, tags, task } from '@trigger.dev/sdk';
 import { triggerEmail } from '../../email/trigger-email';
 import { TaskStatusChangedEmail } from '../../email/templates/task-status-changed';
 import { isUserUnsubscribed } from '@trycompai/email';
@@ -185,6 +185,8 @@ export const runTaskIntegrationChecks = task({
       checkIds,
     } = payload;
 
+    await tags.add([`org:${organizationId}`]);
+
     logger.info(`Running integration checks for task "${taskTitle}"`, {
       taskId,
       connectionId,

apps/api/src/trigger/policies/update-policy.ts

@@ -1,4 +1,4 @@
-import { logger, metadata, queue, schemaTask } from '@trigger.dev/sdk';
+import { logger, metadata, queue, schemaTask, tags } from '@trigger.dev/sdk';
 import { z } from 'zod';
 import { processPolicyUpdate } from './update-policy-helpers';
 
@@ -29,6 +29,8 @@ export const updatePolicy = schemaTask({
     memberId: z.string().optional(),
   }),
   run: async (params) => {
+    await tags.add([`org:${params.organizationId}`]);
+
     try {
       logger.info(`Starting policy update for policy ${params.policyId}`);
 

apps/api/src/trigger/questionnaire/answer-question.ts

@@ -1,5 +1,5 @@
 import { syncOrganizationEmbeddings } from '@/vector-store/lib';
-import { logger, metadata, task } from '@trigger.dev/sdk';
+import { logger, metadata, tags, task } from '@trigger.dev/sdk';
 import { generateAnswerWithRAG } from './answer-question-helpers';
 
 export interface AnswerQuestionPayload {
@@ -177,6 +177,8 @@ export const answerQuestionTask = task({
     questionIndex: number;
     totalQuestions: number;
   }) => {
+    await tags.add([`org:${payload.organizationId}`]);
+
     return await answerQuestion(payload);
   },
 });

apps/api/src/trigger/questionnaire/parse-questionnaire.ts

@@ -1,7 +1,7 @@
 import { extractS3KeyFromUrl } from '@/app/s3';
 import { GetObjectCommand, S3Client } from '@aws-sdk/client-s3';
 import { db } from '@db';
-import { logger, task } from '@trigger.dev/sdk';
+import { logger, tags, task } from '@trigger.dev/sdk';
 
 // Import shared utilities
 import {
@@ -253,6 +253,8 @@ export const parseQuestionnaireTask = task({
   }) => {
     const taskStartTime = Date.now();
 
+    await tags.add([`org:${payload.organizationId}`]);
+
     logger.info('Starting parse questionnaire task', {
       inputType: payload.inputType,
       organizationId: payload.organizationId,

apps/api/src/trigger/tasks/onboarding/migrate-policies-for-org.ts

@@ -1,5 +1,5 @@
 import { db, PolicyStatus, type Prisma } from '@db';
-import { logger, schemaTask } from '@trigger.dev/sdk';
+import { logger, schemaTask, tags } from '@trigger.dev/sdk';
 import { z } from 'zod';
 
 const POLICY_BATCH_SIZE = 50;
@@ -10,6 +10,8 @@ export const migratePoliciesForOrg = schemaTask({
     organizationId: z.string(),
   }),
   run: async ({ organizationId }) => {
+    await tags.add([`org:${organizationId}`]);
+
     // Find policies without any versions
     const policiesWithoutVersions = await db.policy.findMany({
       where: {

apps/api/src/trigger/vector-store/delete-all-manual-answers-orchestrator.ts

@@ -1,4 +1,4 @@
-import { logger, metadata, task } from '@trigger.dev/sdk';
+import { logger, metadata, tags, task } from '@trigger.dev/sdk';
 import { db } from '@db';
 import { deleteManualAnswerTask } from './delete-manual-answer';
 
@@ -17,6 +17,8 @@ export const deleteAllManualAnswersOrchestratorTask = task({
     organizationId: string;
     manualAnswerIds?: string[]; // Optional: IDs passed directly to avoid race condition
   }) => {
+    await tags.add([`org:${payload.organizationId}`]);
+
     logger.info('Starting delete all manual answers from vector DB', {
       organizationId: payload.organizationId,
       manualAnswerIdsProvided: !!payload.manualAnswerIds,

apps/api/src/trigger/vector-store/delete-knowledge-base-document.ts

@@ -1,4 +1,4 @@
-import { logger, task } from '@trigger.dev/sdk';
+import { logger, tags, task } from '@trigger.dev/sdk';
 import { findEmbeddingsForSource } from '@/vector-store/lib/core/find-existing-embeddings';
 import { vectorIndex } from '@/vector-store/lib/core/client';
 import { db } from '@db';
@@ -12,6 +12,8 @@ export const deleteKnowledgeBaseDocumentTask = task({
     maxAttempts: 3,
   },
   run: async (payload: { documentId: string; organizationId: string }) => {
+    await tags.add([`org:${payload.organizationId}`]);
+
     logger.info('Deleting Knowledge Base document from vector DB', {
       documentId: payload.documentId,
       organizationId: payload.organizationId,

apps/api/src/trigger/vector-store/delete-manual-answer.ts

@@ -1,4 +1,4 @@
-import { logger, task } from '@trigger.dev/sdk';
+import { logger, tags, task } from '@trigger.dev/sdk';
 import { deleteManualAnswerFromVector } from '@/vector-store/lib/sync/sync-manual-answer';
 
 /**
@@ -11,6 +11,8 @@ export const deleteManualAnswerTask = task({
     maxAttempts: 3,
   },
   run: async (payload: { manualAnswerId: string; organizationId: string }) => {
+    await tags.add([`org:${payload.organizationId}`]);
+
     logger.info('Deleting manual answer from vector DB', {
       manualAnswerId: payload.manualAnswerId,
       organizationId: payload.organizationId,

apps/api/src/trigger/vector-store/process-knowledge-base-document.ts

@@ -1,4 +1,4 @@
-import { logger, task } from '@trigger.dev/sdk';
+import { logger, tags, task } from '@trigger.dev/sdk';
 import { GetObjectCommand, S3Client } from '@aws-sdk/client-s3';
 import { db } from '@db';
 import { batchUpsertEmbeddings } from '@/vector-store/lib/core/upsert-embedding';
@@ -86,6 +86,8 @@ export const processKnowledgeBaseDocumentTask = task({
   },
   maxDuration: 1000 * 60 * 30, // 30 minutes for large files
   run: async (payload: { documentId: string; organizationId: string }) => {
+    await tags.add([`org:${payload.organizationId}`]);
+
     logger.info('Processing Knowledge Base document', {
       documentId: payload.documentId,
       organizationId: payload.organizationId,

apps/api/src/trigger/vector-store/process-knowledge-base-documents-orchestrator.ts

@@ -1,4 +1,4 @@
-import { logger, metadata, task } from '@trigger.dev/sdk';
+import { logger, metadata, tags, task } from '@trigger.dev/sdk';
 import { processKnowledgeBaseDocumentTask } from './process-knowledge-base-document';
 
 const BATCH_SIZE = 10; // Process 10 documents at a time
@@ -14,6 +14,8 @@ export const processKnowledgeBaseDocumentsOrchestratorTask = task({
   },
   maxDuration: 60 * 60, // 1 hour (in seconds) for large document batches
   run: async (payload: { documentIds: string[]; organizationId: string }) => {
+    await tags.add([`org:${payload.organizationId}`]);
+
     logger.info('Starting Knowledge Base documents processing orchestrator', {
       organizationId: payload.organizationId,
       documentCount: payload.documentIds.length,

apps/api/src/trigger/vendor/vendor-risk-assessment-task.ts

@@ -10,7 +10,7 @@ import {
 import { openai } from '@ai-sdk/openai';
 import type { Prisma } from '@db';
 import type { Task } from '@trigger.dev/sdk';
-import { logger, metadata, queue, schemaTask } from '@trigger.dev/sdk';
+import { logger, metadata, queue, schemaTask, tags } from '@trigger.dev/sdk';
 import { generateObject } from 'ai';
 import { z } from 'zod';
 
@@ -478,6 +478,8 @@ export const vendorRiskAssessmentTask: Task<
   },
   maxDuration: 1000 * 60 * 10,
   run: async (payload) => {
+    await tags.add([`org:${payload.organizationId}`]);
+
     const vendor = await db.vendor.findFirst({
       where: {
         id: payload.vendorId,

apps/app/src/trigger/tasks/auditor/generate-auditor-content.ts

@@ -1,7 +1,7 @@
 import { getOrganizationContext } from '@/trigger/tasks/onboarding/onboard-organization-helpers';
 import { openai } from '@ai-sdk/openai';
 import { db } from '@db/server';
-import { logger, metadata, schemaTask } from '@trigger.dev/sdk';
+import { logger, metadata, schemaTask, tags } from '@trigger.dev/sdk';
 import { generateText } from 'ai';
 import { z } from 'zod';
 
@@ -299,6 +299,7 @@ export const generateAuditorContentTask = schemaTask({
   },
   run: async (payload) => {
     const { organizationId } = payload;
+    await tags.add([`org:${organizationId}`]);
 
     logger.info(`Starting auditor content generation for org ${organizationId}`);
 

apps/app/src/trigger/tasks/device/create-fleet-label-for-org.ts

@@ -1,7 +1,7 @@
 import { getFleetInstance } from '@/lib/fleet';
 import { db } from '@db/server';
 
-import { logger, queue, task } from '@trigger.dev/sdk';
+import { logger, queue, tags, task } from '@trigger.dev/sdk';
 import { AxiosError } from 'axios';
 // Optional: define a queue if we want to control concurrency in v4
 const fleetQueue = queue({ name: 'create-fleet-label-for-org', concurrencyLimit: 10 });
@@ -13,6 +13,8 @@ export const createFleetLabelForOrg = task({
     maxAttempts: 3,
   },
   run: async ({ organizationId }: { organizationId: string }) => {
+    await tags.add([`org:${organizationId}`]);
+
     const organization = await db.organization.findUnique({
       where: {
         id: organizationId,

apps/app/src/trigger/tasks/email/new-policy-email.ts

@@ -1,7 +1,7 @@
 import { db } from '@db/server';
 import { PolicyNotificationEmail } from '@trycompai/email';
 import { isUserUnsubscribed } from '@trycompai/email/lib/check-unsubscribe';
-import { logger, queue, task } from '@trigger.dev/sdk';
+import { logger, queue, tags, task } from '@trigger.dev/sdk';
 import { sendEmailViaApi } from '../../lib/send-email-via-api';
 
 const policyEmailQueue = queue({
@@ -26,6 +26,7 @@ export const sendNewPolicyEmail = task({
       email: payload.email,
       policyName: payload.policyName,
     });
+    await tags.add([`org:${payload.organizationId}`]);
 
     try {
       const unsubscribed = await isUserUnsubscribed(db, payload.email, 'policyNotifications', payload.organizationId);

apps/app/src/trigger/tasks/email/publish-all-policies-email.ts

@@ -1,7 +1,7 @@
 import { db } from '@db/server';
 import { AllPolicyNotificationEmail } from '@trycompai/email';
 import { isUserUnsubscribed } from '@trycompai/email/lib/check-unsubscribe';
-import { logger, queue, task } from '@trigger.dev/sdk';
+import { logger, queue, tags, task } from '@trigger.dev/sdk';
 import { sendEmailViaApi } from '../../lib/send-email-via-api';
 
 const allPolicyEmailQueue = queue({
@@ -24,6 +24,7 @@ export const sendPublishAllPoliciesEmail = task({
       email: payload.email,
       organizationName: payload.organizationName,
     });
+    await tags.add([`org:${payload.organizationId}`]);
 
     try {
       const unsubscribed = await isUserUnsubscribed(db, payload.email, 'policyNotifications', payload.organizationId);

apps/app/src/trigger/tasks/email/weekly-task-digest-email.ts

@@ -1,5 +1,5 @@
 import { db } from '@db/server';
-import { logger, queue, task } from '@trigger.dev/sdk';
+import { logger, queue, tags, task } from '@trigger.dev/sdk';
 import WeeklyTaskDigestEmail from '@trycompai/email/emails/reminders/weekly-task-digest';
 import { isUserUnsubscribed } from '@trycompai/email/lib/check-unsubscribe';
 import { sendEmailViaApi } from '../../lib/send-email-via-api';
@@ -34,6 +34,7 @@ export const sendWeeklyTaskDigestEmailTask = task({
       organizationName: payload.organizationName,
       taskCount: payload.tasks.length,
     });
+    await tags.add([`org:${payload.organizationId}`]);
 
     try {
       const unsubscribed = await isUserUnsubscribed(db, payload.email, 'weeklyTaskDigest', payload.organizationId);

apps/app/src/trigger/tasks/integration/run-integration-tests.ts

@@ -1,11 +1,12 @@
 import { db } from '@db/server';
-import { logger, task } from '@trigger.dev/sdk';
+import { logger, tags, task } from '@trigger.dev/sdk';
 import { sendIntegrationResults } from './integration-results';
 
 export const runIntegrationTests = task({
   id: 'run-integration-tests',
   run: async (payload: { organizationId: string; integrationId?: string }) => {
     const { organizationId, integrationId } = payload;
+    await tags.add([`org:${organizationId}`]);
 
     logger.info(
       integrationId

apps/app/src/trigger/tasks/onboarding/backfill-training-videos-for-org.ts

@@ -1,6 +1,6 @@
 import { trainingVideos } from '@/lib/data/training-videos';
 import { db } from '@db/server';
-import { logger, task } from '@trigger.dev/sdk';
+import { logger, tags, task } from '@trigger.dev/sdk';
 
 export const backfillTrainingVideosForOrg = task({
   id: 'backfill-training-videos-for-org',
@@ -9,6 +9,7 @@ export const backfillTrainingVideosForOrg = task({
   },
   run: async (payload: { organizationId: string }) => {
     logger.info(`Starting training video backfill for organization ${payload.organizationId}`);
+    await tags.add([`org:${payload.organizationId}`]);
 
     try {
       // Get all members for this organization

apps/app/src/trigger/tasks/onboarding/generate-full-policies.ts

@@ -1,5 +1,5 @@
 import { db } from '@db/server';
-import { logger, queue, task } from '@trigger.dev/sdk';
+import { logger, queue, tags, task } from '@trigger.dev/sdk';
 import { getOrganizationContext, triggerPolicyUpdates } from './onboard-organization-helpers';
 
 // v4 queues must be declared in advance
@@ -16,6 +16,7 @@ export const generateFullPolicies = task({
   },
   run: async (payload: { organizationId: string }) => {
     logger.info(`Starting full policy generation for organization ${payload.organizationId}`);
+    await tags.add([`org:${payload.organizationId}`]);
 
     try {
       // Get organization context

apps/app/src/trigger/tasks/onboarding/generate-risk-mitigation.ts

@@ -1,5 +1,5 @@
 import { RiskStatus, db } from '@db/server';
-import { logger, metadata, queue, task } from '@trigger.dev/sdk';
+import { logger, metadata, queue, tags, task } from '@trigger.dev/sdk';
 import axios from 'axios';
 import {
   createRiskMitigationComment,
@@ -24,6 +24,7 @@ export const generateRiskMitigation = task({
     policies: PolicyContext[];
   }) => {
     const { organizationId, riskId, authorId, policies } = payload;
+    await tags.add([`org:${organizationId}`]);
     logger.info(`Generating risk mitigation for risk ${riskId} in org ${organizationId}`);
 
     const risk = await db.risk.findFirst({ where: { id: riskId, organizationId } });
@@ -85,6 +86,7 @@ export const generateRiskMitigationsForOrg = task({
   queue: riskMitigationFanoutQueue,
   run: async (payload: { organizationId: string }) => {
     const { organizationId } = payload;
+    await tags.add([`org:${organizationId}`]);
     logger.info(`Fan-out risk mitigations for org ${organizationId}`);
 
     const [risks, policyRows, author] = await Promise.all([

apps/app/src/trigger/tasks/onboarding/generate-vendor-mitigation.ts

@@ -1,5 +1,5 @@
 import { VendorStatus, db } from '@db/server';
-import { logger, metadata, queue, task } from '@trigger.dev/sdk';
+import { logger, metadata, queue, tags, task } from '@trigger.dev/sdk';
 import axios from 'axios';
 import {
   createVendorRiskComment,
@@ -27,6 +27,7 @@ export const generateVendorMitigation = task({
     policies: PolicyContext[];
   }) => {
     const { organizationId, vendorId, authorId, policies } = payload;
+    await tags.add([`org:${organizationId}`]);
     logger.info(`Generating vendor mitigation for vendor ${vendorId} in org ${organizationId}`);
 
     const vendor = await db.vendor.findFirst({ where: { id: vendorId, organizationId } });
@@ -85,6 +86,7 @@ export const generateVendorMitigationsForOrg = task({
   queue: vendorMitigationFanoutQueue,
   run: async (payload: { organizationId: string }) => {
     const { organizationId } = payload;
+    await tags.add([`org:${organizationId}`]);
     logger.info(`Fan-out vendor mitigations for org ${organizationId}`);
 
     const [vendors, policyRows, author] = await Promise.all([