fix(core): give each run its own external fallback trace id

Runs that carry no external trace context (schedules, task-to-task
triggers) fall back to a trace id generated once in the TracingSDK
constructor. With `experimental_processKeepAlive` the TracingSDK outlives
the run, so every run on a warm process was exported to the external OTLP
endpoint under that one id, merging unrelated runs into a single trace.
Across our production traces, 80.3% contained spans from more than one
run, worst case 25.

This is the same warm-start hazard c043c4a fixed for the external
context path, which read the context live but deliberately left the
fallback captured at construction.

Key the fallback off the internal trace id that every span and log record
of a run already carries, rather than off ambient state. Batch processors
drain asynchronously, so a run's records are routinely exported after the
next run has started; deciding the id at export time from whatever run is
current would stamp the earlier run's records with the later run's id.
Letting the record decide sidesteps the timing entirely, and makes a run's
spans and logs agree without coordinating.

The map is bounded, since a warm process serves unboundedly many runs and
only the in-flight ones can still have records to export. An empty
configured id still means external export is off.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: NERLOE

.changeset/external-trace-id-per-run.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Runs that don't continue an incoming trace are no longer merged into one trace when they execute on the same warm worker process. Each run now appears as its own trace in your external observability tool, so per-run cost and latency attribution works again.

packages/core/src/v3/otel/tracingSDK.ts

@@ -162,12 +162,13 @@ export class TracingSDK {
       )
     );
 
-    const externalTraceId = idGenerator.generateTraceId();
+    // Shared by every wrapper below so a run's spans and logs agree on the id.
+    const fallbackTraceIds = new FallbackExternalTraceIds(idGenerator.generateTraceId());
 
     for (const exporter of config.exporters ?? []) {
       spanProcessors.push(
         getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1"
-          ? new BatchSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId), {
+          ? new BatchSpanProcessor(new ExternalSpanExporterWrapper(exporter, fallbackTraceIds), {
               maxExportBatchSize: parseInt(
                 getEnvVar("TRIGGER_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE") ?? "64"
               ),
@@ -179,7 +180,7 @@ export class TracingSDK {
               ),
               maxQueueSize: parseInt(getEnvVar("TRIGGER_OTEL_SPAN_MAX_QUEUE_SIZE") ?? "512"),
             })
-          : new SimpleSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId))
+          : new SimpleSpanProcessor(new ExternalSpanExporterWrapper(exporter, fallbackTraceIds))
       );
     }
 
@@ -231,7 +232,7 @@ export class TracingSDK {
       logProcessors.push(
         getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1"
           ? new BatchLogRecordProcessor(
-              new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId),
+              new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceIds),
               {
                 maxExportBatchSize: parseInt(
                   getEnvVar("TRIGGER_OTEL_LOG_MAX_EXPORT_BATCH_SIZE") ?? "64"
@@ -246,7 +247,7 @@ export class TracingSDK {
               }
             )
           : new SimpleLogRecordProcessor(
-              new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId)
+              new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceIds)
             )
       );
     }
@@ -393,10 +394,77 @@ function setLogLevel(level: TracingDiagnosticLogLevel) {
   diag.setLogger(new DiagConsoleLogger(), diagLogLevel);
 }
 
+/** Only the current run and the tail of recently ended ones can still export. */
+export const MAX_TRACKED_INTERNAL_TRACES = 64;
+
+/**
+ * External trace ids for runs that carry no external trace context, one per run
+ * — with `processKeepAlive` the `TracingSDK` outlives the run, so an id
+ * captured at construction merges every run on the process into one trace.
+ *
+ * A record's id comes from its own internal trace id rather than from whatever
+ * run is current when the exporter is called. Batch processors drain
+ * asynchronously, so a run's records are routinely exported after the next run
+ * has started, and reading ambient state then would stamp them with the wrong
+ * run's id. It also makes a run's spans and logs agree without coordinating.
+ */
+export class FallbackExternalTraceIds {
+  private readonly byInternalTrace = new Map<string, string>();
+
+  constructor(
+    private seed: string,
+    private traceIdGenerator: Pick<RandomIdGenerator, "generateTraceId"> = idGenerator
+  ) {}
+
+  /** False when no external trace id was configured, i.e. external export is off. */
+  get enabled(): boolean {
+    return !!this.seed;
+  }
+
+  forInternalTrace(internalTraceId: string): string {
+    // An empty seed means external export is disabled — leave it that way
+    // rather than minting an id and switching the feature on.
+    if (!this.seed) {
+      return this.seed;
+    }
+
+    const known = this.byInternalTrace.get(internalTraceId);
+
+    if (known) {
+      // Re-insert so the map is ordered by last use rather than first. A run
+      // that is still exporting keeps its id even if enough unrelated traces
+      // appear alongside it to fill the map, which would otherwise split it
+      // across two external traces.
+      this.byInternalTrace.delete(internalTraceId);
+      this.byInternalTrace.set(internalTraceId, known);
+
+      return known;
+    }
+
+    // The first run reuses the id generated at construction, so the configured
+    // seed is not thrown away.
+    const traceId =
+      this.byInternalTrace.size === 0 ? this.seed : this.traceIdGenerator.generateTraceId();
+
+    this.byInternalTrace.set(internalTraceId, traceId);
+
+    if (this.byInternalTrace.size > MAX_TRACKED_INTERNAL_TRACES) {
+      // Map iterates in insertion order, so this drops the least recently used.
+      const stalest = this.byInternalTrace.keys().next().value;
+
+      if (stalest !== undefined) {
+        this.byInternalTrace.delete(stalest);
+      }
+    }
+
+    return traceId;
+  }
+}
+
 export class ExternalSpanExporterWrapper {
   constructor(
     private underlyingExporter: SpanExporter,
-    private externalTraceId: string
+    private fallback: FallbackExternalTraceIds
   ) {}
 
   private transformSpan(span: ReadableSpan): ReadableSpan | undefined {
@@ -407,7 +475,7 @@ export class ExternalSpanExporterWrapper {
 
     const isExternallySampled = externalTraceContext
       ? isTraceFlagSampled(externalTraceContext.traceFlags)
-      : !!this.externalTraceId;
+      : this.fallback.enabled;
 
     if (!isExternallySampled) {
       return;
@@ -419,7 +487,7 @@ export class ExternalSpanExporterWrapper {
 
     const externalTraceId = externalTraceContext
       ? externalTraceContext.traceId
-      : this.externalTraceId;
+      : this.fallback.forInternalTrace(span.spanContext().traceId);
 
     const isAttemptSpan = span.attributes[SemanticInternalAttributes.SPAN_ATTEMPT];
 
@@ -477,18 +545,18 @@ export class ExternalSpanExporterWrapper {
   }
 }
 
-class ExternalLogRecordExporterWrapper {
+export class ExternalLogRecordExporterWrapper {
   constructor(
     private underlyingExporter: LogRecordExporter,
-    private externalTraceId: string
+    private fallback: FallbackExternalTraceIds
   ) {}
 
   export(logs: any[], resultCallback: (result: any) => void): void {
     const externalTraceContext = traceContext.getExternalTraceContext();
 
     const isExternallySampled = externalTraceContext
       ? isTraceFlagSampled(externalTraceContext.traceFlags)
-      : !!this.externalTraceId;
+      : this.fallback.enabled;
 
     if (!isExternallySampled) {
       this.underlyingExporter.export([], resultCallback);
@@ -519,14 +587,20 @@ class ExternalLogRecordExporterWrapper {
       | { traceId: string; spanId: string; tracestate?: string; traceFlags: number }
       | undefined
   ): ReadableLogRecord {
-    // Capture externalTraceId for use within the proxy's scope.
-    // Use externalTraceContext.traceId if available, otherwise fall back to generated externalTraceId
+    // Without a spanContext there is no internal trace id to key the fallback
+    // on, and nothing to rewrite.
+    if (!logRecord.spanContext) {
+      return logRecord;
+    }
+
+    // Capture externalTraceId for use within the proxy's scope. Use
+    // externalTraceContext.traceId if available, otherwise the id belonging to
+    // the run this record came from.
     const externalTraceId = externalTraceContext
       ? externalTraceContext.traceId
-      : this.externalTraceId;
+      : this.fallback.forInternalTrace(logRecord.spanContext.traceId);
 
-    // If there's no spanContext, or if the externalTraceId is not set, return the original logRecord.
-    if (!logRecord.spanContext || !externalTraceId) {
+    if (!externalTraceId) {
       return logRecord;
     }