feat: run annotations

.changeset/tame-oranges-change.md

@@ -0,0 +1,8 @@
+---
+"@trigger.dev/redis-worker": patch
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Adapted the CLI API client to propagate the trigger source via http headers.

apps/webapp/app/routes/api.v1.runs.$runParam.replay.ts

@@ -5,6 +5,7 @@ import { prisma } from "~/db.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
+import { sanitizeTriggerSource } from "~/utils/triggerSource";
 
 const ParamsSchema = z.object({
   /* This is the run friendly ID */
@@ -41,8 +42,11 @@ export async function action({ request, params }: ActionFunctionArgs) {
       return json({ error: "Run not found" }, { status: 404 });
     }
 
+    const triggerSource =
+      sanitizeTriggerSource(request.headers.get("x-trigger-source")) ?? "api";
+
     const service = new ReplayTaskRunService();
-    const newRun = await service.call(taskRun);
+    const newRun = await service.call(taskRun, { triggerSource });
 
     if (!newRun) {
       return json({ error: "Failed to create new run" }, { status: 400 });

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -19,6 +19,7 @@ import {
   handleRequestIdempotency,
   saveRequestIdempotency,
 } from "~/utils/requestIdempotency.server";
+import { sanitizeTriggerSource } from "~/utils/triggerSource";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { OutOfEntitlementError, TriggerTaskService } from "~/v3/services/triggerTask.server";
 
@@ -36,6 +37,7 @@ export const HeadersSchema = z.object({
   "x-trigger-engine-version": RunEngineVersionSchema.nullish(),
   "x-trigger-request-idempotency-key": z.string().nullish(),
   "x-trigger-realtime-streams-version": z.string().nullish(),
+  "x-trigger-source": z.string().nullish(),
   traceparent: z.string().optional(),
   tracestate: z.string().optional(),
 });
@@ -67,6 +69,7 @@ const { action, loader } = createActionApiRoute(
       "x-trigger-engine-version": engineVersion,
       "x-trigger-request-idempotency-key": requestIdempotencyKey,
       "x-trigger-realtime-streams-version": realtimeStreamsVersion,
+      "x-trigger-source": triggerSourceHeader,
     } = headers;
 
     const cachedResponse = await handleRequestIdempotency(requestIdempotencyKey, {
@@ -119,6 +122,8 @@ const { action, loader } = createActionApiRoute(
           realtimeStreamsVersion: determineRealtimeStreamsVersion(
             realtimeStreamsVersion ?? undefined
           ),
+          triggerSource: isFromWorker ? "sdk" : sanitizeTriggerSource(triggerSourceHeader) ?? "api",
+          triggerAction: "trigger",
         },
         engineVersion ?? undefined
       );

apps/webapp/app/routes/api.v1.tasks.batch.ts

@@ -15,6 +15,7 @@ import {
   BatchTriggerV3Service,
 } from "~/v3/services/batchTriggerV3.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
+import { sanitizeTriggerSource } from "~/utils/triggerSource";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
 import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 import { extractJwtSigningSecretKey } from "~/services/realtime/jwtAuth.server";
@@ -72,6 +73,7 @@ const { action, loader } = createActionApiRoute(
       "x-trigger-engine-version": engineVersion,
       "batch-processing-strategy": batchProcessingStrategy,
       "x-trigger-realtime-streams-version": realtimeStreamsVersion,
+      "x-trigger-source": triggerSourceHeader,
       traceparent,
       tracestate,
     } = headers;
@@ -113,6 +115,8 @@ const { action, loader } = createActionApiRoute(
         realtimeStreamsVersion: determineRealtimeStreamsVersion(
           realtimeStreamsVersion ?? undefined
         ),
+        triggerSource: isFromWorker ? "sdk" : sanitizeTriggerSource(triggerSourceHeader) ?? "api",
+        triggerAction: "trigger",
       });
 
       const $responseHeaders = await responseHeaders(

apps/webapp/app/routes/api.v2.tasks.batch.ts

@@ -17,6 +17,7 @@ import {
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { BatchProcessingStrategy } from "~/v3/services/batchTriggerV3.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
+import { sanitizeTriggerSource } from "~/utils/triggerSource";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
 import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 import { extractJwtSigningSecretKey } from "~/services/realtime/jwtAuth.server";
@@ -62,6 +63,7 @@ const { action, loader } = createActionApiRoute(
       "batch-processing-strategy": batchProcessingStrategy,
       "x-trigger-request-idempotency-key": requestIdempotencyKey,
       "x-trigger-realtime-streams-version": realtimeStreamsVersion,
+      "x-trigger-source": triggerSourceHeader,
       traceparent,
       tracestate,
     } = headers;
@@ -127,6 +129,8 @@ const { action, loader } = createActionApiRoute(
         realtimeStreamsVersion: determineRealtimeStreamsVersion(
           realtimeStreamsVersion ?? undefined
         ),
+        triggerSource: isFromWorker ? "sdk" : sanitizeTriggerSource(triggerSourceHeader) ?? "api",
+        triggerAction: "trigger",
       });
 
       const $responseHeaders = await responseHeaders(

apps/webapp/app/routes/api.v3.batches.ts

@@ -13,6 +13,7 @@ import {
 } from "~/utils/requestIdempotency.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
+import { sanitizeTriggerSource } from "~/utils/triggerSource";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
 import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
 import { extractJwtSigningSecretKey } from "~/services/realtime/jwtAuth.server";
@@ -65,6 +66,7 @@ const { action, loader } = createActionApiRoute(
       "x-trigger-worker": isFromWorker,
       "x-trigger-client": triggerClient,
       "x-trigger-realtime-streams-version": realtimeStreamsVersion,
+      "x-trigger-source": triggerSourceHeader,
       traceparent,
       tracestate,
     } = headers;
@@ -132,6 +134,7 @@ const { action, loader } = createActionApiRoute(
         realtimeStreamsVersion: determineRealtimeStreamsVersion(
           realtimeStreamsVersion ?? undefined
         ),
+        triggerSource: isFromWorker ? "sdk" : sanitizeTriggerSource(triggerSourceHeader) ?? "api",
       });
 
       const $responseHeaders = await responseHeaders(

apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts

@@ -214,6 +214,7 @@ export const action: ActionFunction = async ({ request, params }) => {
       ttlSeconds: submission.value.ttlSeconds,
       version: submission.value.version,
       prioritySeconds: submission.value.prioritySeconds,
+      triggerSource: "dashboard",
     });
 
     if (!newRun) {

apps/webapp/app/runEngine/services/batchTrigger.server.ts

@@ -48,6 +48,8 @@ export type BatchTriggerTaskServiceOptions = {
   spanParentAsLink?: boolean;
   oneTimeUseToken?: string;
   realtimeStreamsVersion?: "v1" | "v2";
+  triggerSource?: string;
+  triggerAction?: string;
 };
 
 /**
@@ -678,6 +680,8 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
         batchId: batch.id,
         batchIndex: currentIndex,
         realtimeStreamsVersion: options?.realtimeStreamsVersion,
+        triggerSource: options?.triggerSource ?? "api",
+        triggerAction: options?.triggerAction ?? "trigger",
       },
       "V2"
     );

apps/webapp/app/runEngine/services/createBatch.server.ts

@@ -17,6 +17,7 @@ export type CreateBatchServiceOptions = {
   spanParentAsLink?: boolean;
   oneTimeUseToken?: string;
   realtimeStreamsVersion?: "v1" | "v2";
+  triggerSource?: string;
 };
 
 /**
@@ -143,6 +144,7 @@ export class CreateBatchService extends WithRunEngine {
             idempotencyKey: body.idempotencyKey,
             processingConcurrency: config.processingConcurrency,
             planType,
+            triggerSource: options.triggerSource,
           };
 
           await this._engine.initializeBatch(initOptions);

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -6,6 +6,7 @@ import {
 import { Tracer } from "@opentelemetry/api";
 import { tryCatch } from "@trigger.dev/core/utils";
 import {
+  RunAnnotations,
   TaskRunError,
   taskRunErrorEnhancer,
   taskRunErrorToString,
@@ -289,6 +290,17 @@ export class RunEngineTriggerTaskService {
 
       const workerQueue = await this.queueConcern.getWorkerQueue(environment, body.options?.region);
 
+      // Build annotations for this run
+      const triggerSource = options.triggerSource ?? "api";
+      const triggerAction = options.triggerAction ?? "trigger";
+      const parentAnnotations = RunAnnotations.safeParse(parentRun?.annotations).data;
+      const annotations = {
+        triggerSource,
+        triggerAction,
+        rootTriggerSource: parentAnnotations?.rootTriggerSource ?? triggerSource,
+        rootScheduleId: parentAnnotations?.rootScheduleId || options.scheduleId || undefined,
+      };
+
       try {
         return await this.traceEventConcern.traceRun(
           triggerRequest,
@@ -369,6 +381,7 @@ export class RunEngineTriggerTaskService {
                 planType,
                 realtimeStreamsVersion: options.realtimeStreamsVersion,
                 debounce: body.options?.debounce,
+                annotations,
                 // When debouncing with triggerAndWait, create a span for the debounced trigger
                 onDebounced:
                   body.options?.debounce && body.options?.resumeParentOnCompletion

apps/webapp/app/utils/triggerSource.ts

@@ -0,0 +1,9 @@
+const ALLOWED_TRIGGER_SOURCES = new Set(["sdk", "cli", "mcp"]);
+
+/** Validates a client-provided trigger source header against the allowlist. */
+export function sanitizeTriggerSource(value: string | null | undefined): string | undefined {
+  if (value && ALLOWED_TRIGGER_SOURCES.has(value)) {
+    return value;
+  }
+  return undefined;
+}

apps/webapp/app/v3/runEngineHandlers.server.ts

@@ -750,6 +750,8 @@ export function setupBatchQueueCallbacks() {
               batchIndex: itemIndex,
               realtimeStreamsVersion: meta.realtimeStreamsVersion,
               planType: meta.planType,
+              triggerSource: meta.parentRunId ? "sdk" : meta.triggerSource ?? "api",
+              triggerAction: "trigger",
             },
             "V2"
           );

apps/webapp/app/v3/scheduleEngine.server.ts

@@ -106,6 +106,8 @@ function createScheduleEngine() {
             scheduleInstanceId,
             queueTimestamp: exactScheduleTime,
             overrideCreatedAt: exactScheduleTime,
+            triggerSource: "schedule",
+            triggerAction: "trigger",
           }
         );
 

apps/webapp/app/v3/services/batchTriggerV3.server.ts

@@ -57,6 +57,8 @@ export type BatchTriggerTaskServiceOptions = {
   spanParentAsLink?: boolean;
   oneTimeUseToken?: string;
   realtimeStreamsVersion?: "v1" | "v2";
+  triggerSource?: string;
+  triggerAction?: string;
 };
 
 type RunItemData = {
@@ -853,6 +855,8 @@ export class BatchTriggerV3Service extends BaseService {
         skipChecks: true,
         runFriendlyId: task.runId,
         realtimeStreamsVersion: options?.realtimeStreamsVersion,
+        triggerSource: options?.triggerSource ?? "api",
+        triggerAction: options?.triggerAction ?? "trigger",
       }
     );
 

apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts

@@ -242,6 +242,7 @@ export class BulkActionService extends BaseService {
             const [error, result] = await tryCatch(
               replayService.call(run, {
                 bulkActionId: bulkActionId,
+                triggerSource: "dashboard",
               })
             );
             if (error) {

apps/webapp/app/v3/services/bulk/performBulkAction.server.ts

@@ -27,7 +27,7 @@ export class PerformBulkActionService extends BaseService {
     switch (item.group.type) {
       case "REPLAY": {
         const service = new ReplayTaskRunService(this._prisma);
-        const result = await service.call(item.sourceRun);
+        const result = await service.call(item.sourceRun, { triggerSource: "dashboard" });
 
         await this._prisma.bulkActionItem.update({
           where: { id: item.id },

apps/webapp/app/v3/services/replayTaskRun.server.ts

@@ -18,6 +18,7 @@ type OverrideOptions = {
   payload?: string;
   metadata?: unknown;
   bulkActionId?: string;
+  triggerSource?: string;
 } & RunOptionsData;
 
 export class ReplayTaskRunService extends BaseService {
@@ -123,6 +124,8 @@ export class ReplayTaskRunService extends BaseService {
           realtimeStreamsVersion: determineRealtimeStreamsVersion(
             existingTaskRun.realtimeStreamsVersion
           ),
+          triggerSource: overrideOptions.triggerSource ?? "api",
+          triggerAction: "replay",
         }
       );
 

apps/webapp/app/v3/services/testTask.server.ts

@@ -11,28 +11,35 @@ export class TestTaskService extends BaseService {
 
     switch (triggerSource) {
       case "STANDARD": {
-        const result = await triggerTaskService.call(data.taskIdentifier, environment, {
-          payload: data.payload,
-          options: {
-            test: true,
-            metadata: data.metadata,
-            delay: data.delaySeconds ? new Date(Date.now() + data.delaySeconds * 1000) : undefined,
-            ttl: data.ttlSeconds,
-            idempotencyKey: data.idempotencyKey,
-            idempotencyKeyTTL: data.idempotencyKeyTTLSeconds
-              ? `${data.idempotencyKeyTTLSeconds}s`
-              : undefined,
-            queue: data.queue ? { name: data.queue } : undefined,
-            concurrencyKey: data.concurrencyKey,
-            maxAttempts: data.maxAttempts,
-            maxDuration: data.maxDurationSeconds,
-            tags: data.tags,
-            machine: data.machine,
-            region: data.region,
-            lockToVersion: data.version === "latest" ? undefined : data.version,
-            priority: data.prioritySeconds,
+        const result = await triggerTaskService.call(
+          data.taskIdentifier,
+          environment,
+          {
+            payload: data.payload,
+            options: {
+              test: true,
+              metadata: data.metadata,
+              delay: data.delaySeconds
+                ? new Date(Date.now() + data.delaySeconds * 1000)
+                : undefined,
+              ttl: data.ttlSeconds,
+              idempotencyKey: data.idempotencyKey,
+              idempotencyKeyTTL: data.idempotencyKeyTTLSeconds
+                ? `${data.idempotencyKeyTTLSeconds}s`
+                : undefined,
+              queue: data.queue ? { name: data.queue } : undefined,
+              concurrencyKey: data.concurrencyKey,
+              maxAttempts: data.maxAttempts,
+              maxDuration: data.maxDurationSeconds,
+              tags: data.tags,
+              machine: data.machine,
+              region: data.region,
+              lockToVersion: data.version === "latest" ? undefined : data.version,
+              priority: data.prioritySeconds,
+            },
           },
-        });
+          { triggerSource: "dashboard", triggerAction: "test" }
+        );
 
         return result?.run;
       }
@@ -72,7 +79,7 @@ export class TestTaskService extends BaseService {
               priority: data.prioritySeconds,
             },
           },
-          { customIcon: "scheduled" }
+          { customIcon: "scheduled", triggerSource: "dashboard", triggerAction: "test" }
         );
 
         return result?.run;

apps/webapp/app/v3/services/triggerTask.server.ts

@@ -33,6 +33,8 @@ export type TriggerTaskServiceOptions = {
   replayedFromTaskRunFriendlyId?: string;
   planType?: string;
   realtimeStreamsVersion?: "v1" | "v2";
+  triggerSource?: string;
+  triggerAction?: string;
 };
 
 export class OutOfEntitlementError extends Error {

internal-packages/database/prisma/migrations/20260323085243_add_task_run_annotations/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "public"."TaskRun" ADD COLUMN     "annotations" JSONB;

internal-packages/database/prisma/schema.prisma

@@ -837,6 +837,9 @@ model TaskRun {
   metadataType    String  @default("application/json")
   metadataVersion Int     @default(1)
 
+  /// Structured annotations: triggerSource, triggerAction, rootTriggerSource, rootScheduleId
+  annotations Json?
+
   /// Run output
   output     String?
   outputType String  @default("application/json")

internal-packages/run-engine/src/batch-queue/index.ts

@@ -296,6 +296,7 @@ export class BatchQueue {
       realtimeStreamsVersion: options.realtimeStreamsVersion,
       idempotencyKey: options.idempotencyKey,
       processingConcurrency: options.processingConcurrency,
+      triggerSource: options.triggerSource,
     };
 
     // Store metadata in completion tracker