triggerdotdev · myftija · Mar 23, 2026 · Mar 23, 2026 · Mar 23, 2026 · Mar 23, 2026
diff --git a/.changeset/tame-oranges-change.md b/.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.
diff --git a/apps/webapp/app/routes/api.v1.runs.$runParam.replay.ts b/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 });

diff --git a/apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts b/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
      );

diff --git a/apps/webapp/app/routes/api.v1.tasks.batch.ts b/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(

diff --git a/apps/webapp/app/routes/api.v2.tasks.batch.ts b/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(

diff --git a/apps/webapp/app/routes/api.v3.batches.ts b/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(

diff --git a/apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts b/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) {

diff --git a/apps/webapp/app/runEngine/services/batchTrigger.server.ts b/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"
    );

diff --git a/apps/webapp/app/runEngine/services/createBatch.server.ts b/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);

diff --git a/apps/webapp/app/runEngine/services/triggerTask.server.ts b/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

diff --git a/apps/webapp/app/utils/triggerSource.ts b/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;
+}
diff --git a/apps/webapp/app/v3/runEngineHandlers.server.ts b/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"
          );

diff --git a/apps/webapp/app/v3/scheduleEngine.server.ts b/apps/webapp/app/v3/scheduleEngine.server.ts
@@ -106,6 +106,8 @@ function createScheduleEngine() {
            scheduleInstanceId,
            queueTimestamp: exactScheduleTime,
            overrideCreatedAt: exactScheduleTime,
+            triggerSource: "schedule",
+            triggerAction: "trigger",
          }
        );


diff --git a/apps/webapp/app/v3/services/batchTriggerV3.server.ts b/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",
      }
    );


diff --git a/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts b/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) {

diff --git a/apps/webapp/app/v3/services/bulk/performBulkAction.server.ts b/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 },

diff --git a/apps/webapp/app/v3/services/replayTaskRun.server.ts b/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",
        }
      );


diff --git a/apps/webapp/app/v3/services/testTask.server.ts b/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;

diff --git a/apps/webapp/app/v3/services/triggerTask.server.ts b/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 {

diff --git a/...packages/database/prisma/migrations/20260323085243_add_task_run_annotations/migration.sql b/...packages/database/prisma/migrations/20260323085243_add_task_run_annotations/migration.sql
@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "public"."TaskRun" ADD COLUMN     "annotations" JSONB;
diff --git a/internal-packages/database/prisma/schema.prisma b/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")

diff --git a/internal-packages/run-engine/src/batch-queue/index.ts b/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
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,2 @@
		-- AlterTable
		ALTER TABLE "public"."TaskRun" ADD COLUMN "annotations" JSONB;