feat(slack): Keep resumed Slack slices in the Turn trace

TELEMETRY.md

@@ -307,6 +307,15 @@ A Slack `chat.turn` span is in the trace of a Slack webhook request:
 - Without stored headers, `chat.turn` is a child of the `queue.process` span.
   Headers that are not valid start a new trace.
 
+A Turn that yields, times out, or waits for auth continues in a resumed slice.
+Each resumed slice has its own `chat.turn` span in the trace of the first slice:
+
+- The first save of the Turn cursor stores the active `sentry-trace` and
+  `baggage` headers. Later saves keep them.
+- A resumed `chat.turn` continues that trace. Its parent is the span that was
+  active at the first save. It has a span link to the `queue.process` span of
+  the worker.
+
 The timing attributes are absent when the value is negative. A negative value
 is clock skew between Slack and Junior hosts, not latency. Resumed `chat.turn`
 spans do not record these attributes.

packages/junior/src/chat/logging.ts

@@ -1692,6 +1692,14 @@ export function getTracePropagationHeaders(): TracePropagationHeaders {
   return headers;
 }
 
+/** Return the active trace as a durable trace context, if there is one. */
+export function getActiveTraceContext():
+  | { baggage?: string; sentryTrace: string }
+  | undefined {
+  const { "sentry-trace": sentryTrace, baggage } = getTracePropagationHeaders();
+  return sentryTrace ? { baggage, sentryTrace } : undefined;
+}
+
 /** Set attributes on the currently active Sentry span. */
 export function setSpanAttributes(attributes: Record<string, unknown>): void {
   const span = getActiveSpan();

packages/junior/src/chat/providers/slack/resume.ts

@@ -34,9 +34,10 @@ import {
   logException,
   setSpanAttributes,
   setTags,
-  withLogContext,
   type LogContext,
 } from "@/chat/logging";
+import { withTurnSpan } from "@/chat/providers/slack/turn-input-telemetry";
+import type { TraceContext } from "@/chat/task-execution/state";
 import {
   finalizeFailedTurnReplyWithEvent,
   getAgentTurnDiagnosticsAttributes,
@@ -153,6 +154,8 @@ interface ResumeSlackTurnArgs {
   messageText: string;
   conversationId: string;
   turnId: string;
+  /** Trace of the first slice of the Turn. */
+  traceContext?: TraceContext;
   /** Active durable execution slice being resumed. */
   sliceId?: number;
   channelId: string;
@@ -373,8 +376,10 @@ function buildResumedRun(
 export async function resumeSlackTurn(
   args: ResumeSlackTurnArgs,
 ): Promise<boolean> {
-  return withLogContext(getResumeLogContext(args), () =>
-    resumeSlackTurnInContext(args),
+  return await withTurnSpan(
+    getResumeLogContext(args),
+    args.traceContext && { traceContexts: [args.traceContext] },
+    () => resumeSlackTurnInContext(args),
   );
 }
 

packages/junior/src/chat/providers/slack/turn-input-telemetry.ts

@@ -29,7 +29,7 @@ function spanLinks({ sentryTrace }: TraceContext) {
 }
 
 /**
- * Run a `chat.turn` span in the ingress trace of the first trace context.
+ * Run a `chat.turn` span in the trace of the first trace context.
  *
  * The span links to the other trace contexts and to the span that was active,
  * such as the worker span. Without a trace context, the span is a child of the

packages/junior/src/chat/task-execution/checkpoint.ts

@@ -12,7 +12,7 @@
  */
 import type { ConversationPrivacy } from "@/chat/conversation-privacy";
 import type { Destination, Actor, Source } from "@sentry/junior-plugin-api";
-import { getActiveTraceId, logException } from "@/chat/logging";
+import { getActiveTraceContext, logException } from "@/chat/logging";
 import type { PiMessage } from "@/chat/pi/messages";
 import type { CredentialSubject } from "@/chat/credentials/context";
 import type { ConversationMessageProvenance } from "@/chat/conversations/provenance";
@@ -199,7 +199,7 @@ function sharedWrite(args: TurnCheckpointWrite, latest?: TurnRecord) {
       dispatchId: args.dispatchId ?? latest?.dispatchId,
       source: args.source,
       surface: args.surface ?? latest?.surface,
-      traceId: getActiveTraceId() ?? latest?.traceId,
+      traceContext: latest?.traceContext ?? getActiveTraceContext(),
       turnStartMessageIndex:
         args.turnStartMessageIndex ?? latest?.turnStartMessageIndex,
       cumulativeToolCallCount:

packages/junior/src/chat/task-execution/paused-turn.ts

@@ -399,6 +399,7 @@ async function runPausedTurnInContext(
     messageText: "",
     conversationId: payload.conversationId,
     turnId: payload.turnId,
+    traceContext: payload.traceContext,
     channelId: thread?.channelId ?? destination.channelId,
     ...(thread?.threadTs ? { threadTs: thread.threadTs } : undefined),
     executeTurn: async (run, saveResult) =>

packages/junior/src/chat/task-execution/slack-work.ts

@@ -9,7 +9,7 @@ import {
 } from "chat";
 import { z } from "zod";
 import { objectTypeSchema } from "@sentry/junior-plugin-api";
-import { getTracePropagationHeaders, logException } from "@/chat/logging";
+import { getActiveTraceContext, logException } from "@/chat/logging";
 import type {
   SlackTurnOptions,
   SteeringCandidateMessage,
@@ -849,7 +849,6 @@ export function buildSlackInboundMessage(args: {
     throw new Error("Slack inbound message requires destination context");
   }
   const hasInterruptOverride = hasSteeringOverride(args.message.text);
-  const { "sentry-trace": sentryTrace, baggage } = getTracePropagationHeaders();
   return {
     conversationId: args.conversationId,
     destination,
@@ -876,6 +875,6 @@ export function buildSlackInboundMessage(args: {
         message: args.message.toJSON(),
       } satisfies SlackConversationMessageMetadata,
     },
-    ...(sentryTrace && { traceContext: { sentryTrace, baggage } }),
+    traceContext: getActiveTraceContext(),
   };
 }

packages/junior/src/chat/task-execution/state.ts

@@ -103,8 +103,8 @@ export type InboundMessageDelivery = z.output<
   typeof inboundMessageDeliverySchema
 >;
 
-/** Sentry trace propagation headers captured when an inbound message arrives. */
-const inboundMessageTraceContextSchema = z
+/** Sentry trace propagation headers. A malformed value is dropped, not rejected. */
+export const traceContextSchema = z
   .object({
     baggage: z.string().optional(),
     sentryTrace: z.string().min(1),
@@ -125,13 +125,13 @@ export const inboundMessageSchema = z
     input: agentInputSchema,
     receivedAtMs: z.number().finite(),
     source: inboundMessageSourceSchema,
-    traceContext: inboundMessageTraceContextSchema,
+    traceContext: traceContextSchema,
   })
   .strict();
 
 export type InboundMessage = z.output<typeof inboundMessageSchema>;
 
-export type TraceContext = NonNullable<InboundMessage["traceContext"]>;
+export type TraceContext = NonNullable<z.output<typeof traceContextSchema>>;
 
 /** Redis mailbox shape kept for deployed workers that still require this field. */
 const storedInboundMessageSchema = inboundMessageSchema.extend({

packages/junior/src/chat/task-execution/turn-cursor.ts

@@ -41,6 +41,7 @@ import type {
   ConversationExecution,
   ConversationStore,
 } from "@/chat/conversations/store";
+import { traceContextSchema, type TraceContext } from "./state";
 import {
   turnCursorIndexKey,
   turnCursorKey,
@@ -152,7 +153,8 @@ export interface TurnRecord {
   startedAtMs: number;
   state: TurnStatus;
   surface?: AgentTurnSurface;
-  traceId?: string;
+  /** Trace of the first saved slice. Resumed slices continue it. */
+  traceContext?: TraceContext;
   turnStartMessageIndex?: number;
   updatedAtMs: number;
 }
@@ -270,7 +272,7 @@ const storedTurnRecordSchema = z
     startedAtMs: nonNegativeNumberSchema,
     state: turnStatusSchema,
     surface: turnSurfaceSchema.optional(),
-    traceId: z.string().optional(),
+    traceContext: traceContextSchema,
     updatedAtMs: nonNegativeNumberSchema,
     committedSeq: seqCursorSchema,
     historyVersion: z.number().int().nonnegative().optional(),
@@ -474,7 +476,7 @@ function materializeTurnRecord(
       resultMessageId: stored.resultMessageId,
       resumedFromSliceId: stored.resumedFromSliceId,
       surface: stored.surface,
-      traceId: stored.traceId,
+      traceContext: stored.traceContext,
       turnStartMessageIndex,
     }),
   };
@@ -579,7 +581,7 @@ function buildStoredRecord(args: {
   source?: Source;
   errorMessage?: string;
   resumedFromSliceId?: number;
-  traceId?: string;
+  traceContext?: TraceContext;
   cumulativeToolCallCount?: number;
   turnLimitWrappedUp?: boolean;
   lostWorkerResumeCount?: number;
@@ -613,7 +615,7 @@ function buildStoredRecord(args: {
       resultMessageId: args.resultMessageId,
       resumedFromSliceId: args.resumedFromSliceId,
       surface: args.surface,
-      traceId: args.traceId,
+      traceContext: args.traceContext,
       turnStartSeq: args.turnStartSeq,
     }),
   };
@@ -744,7 +746,7 @@ async function updateTurnState(args: {
         resultMessageId: args.resultMessageId ?? args.existing.resultMessageId,
         resumedFromSliceId: args.existing.resumedFromSliceId,
         surface: args.existing.surface,
-        traceId: args.existing.traceId,
+        traceContext: args.existing.traceContext,
         turnStartSeq: parsed.turnStartSeq,
       }),
     }),
@@ -781,7 +783,7 @@ export async function upsertTurnRecord(args: {
   resumeReason?: TurnPauseReason;
   errorMessage?: string;
   resumedFromSliceId?: number;
-  traceId?: string;
+  traceContext?: TraceContext;
   turnContexts?: PluginTurnContext[];
   turnStartMessageIndex?: number;
   ttlMs?: number;
@@ -928,7 +930,7 @@ async function upsertTurnRecordLocked(
         resumedFromSliceId: args.resumedFromSliceId,
         startedAtMs: existingRecord?.startedAtMs,
         surface: args.surface ?? existingRecord?.surface,
-        traceId: args.traceId ?? existingRecord?.traceId,
+        traceContext: args.traceContext ?? existingRecord?.traceContext,
         turnStartSeq,
       }),
     }),

packages/junior/src/chat/task-execution/turn-wake.ts

@@ -19,6 +19,7 @@ import {
   loadTurnCheckpoint,
 } from "@/chat/task-execution/checkpoint";
 import type { ConversationWorkQueue } from "@/chat/task-execution/queue";
+import type { TraceContext } from "@/chat/task-execution/state";
 import {
   ensureConversationWake,
   requestConversationWork,
@@ -29,6 +30,7 @@ export interface PausedTurnRequest {
   conversationId: string;
   destination: Destination;
   expectedVersion: number;
+  traceContext?: TraceContext;
   turnId: string;
 }
 
@@ -90,6 +92,7 @@ export async function getPausedTurnRequest(args: {
     destination: routing.destination,
     turnId: args.turnId,
     expectedVersion: turn.version,
+    traceContext: turn.traceContext,
   };
 }
 

packages/junior/tests/component/task-execution/conversation-work-telemetry.test.ts

@@ -8,11 +8,16 @@ import {
   it,
   vi,
 } from "vitest";
+import { createSlackSource } from "@sentry/junior-plugin-api";
 import * as Sentry from "@/chat/sentry";
 import { closeDb } from "@/chat/db";
 import { withLogContext } from "@/chat/logging";
 import { withTurnSpan } from "@/chat/providers/slack/turn-input-telemetry";
+import { completedAgentRun } from "@/chat/runtime/agent-run-outcome";
+import { persistThreadStateById } from "@/chat/runtime/thread-state";
 import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter";
+import { saveTurnCheckpoint } from "@/chat/task-execution/checkpoint";
+import { runNextPausedTurn } from "@/chat/task-execution/paused-turn";
 import { createSlackConversationWorker } from "@/chat/task-execution/slack-work";
 import { appendInboundMessage } from "@/chat/task-execution/store";
 import { consumeConversationQueueMessage } from "@/chat/task-execution/vercel-callback";
@@ -26,9 +31,11 @@ import {
   handleSlackWebhookAndFlush,
   inboundMessage,
   SLACK_BOT_USER_ID,
+  SLACK_DESTINATION,
   slackEnvelope,
   slackWebhookRequest,
 } from "../../fixtures/conversation-work";
+import { deliverAssistantMessagesForTest } from "../../fixtures/agent-runner";
 
 describe("conversation work telemetry", () => {
   const spans: ReturnType<typeof Sentry.spanToJSON>[] = [];
@@ -67,6 +74,9 @@ describe("conversation work telemetry", () => {
     await disconnectStateAdapter();
   });
 
+  const byOp = (op: string) =>
+    spans.filter((span) => span.attributes?.["sentry.op"] === op);
+
   afterEach(async () => {
     vi.useRealTimers();
     await closeDb();
@@ -172,8 +182,6 @@ describe("conversation work telemetry", () => {
       state,
     });
 
-    const byOp = (op: string) =>
-      spans.filter((span) => span.attributes?.["sentry.op"] === op);
     const [oldest, newest] = byOp("slack.message.ingress");
     const [worker] = byOp("queue.process");
     expect(oldest?.trace_id).not.toBe(newest?.trace_id);
@@ -195,4 +203,93 @@ describe("conversation work telemetry", () => {
       }),
     ]);
   });
+
+  it("puts a resumed slice in the trace of its first slice", async () => {
+    const conversationId = "slack:C123:1712345.0001";
+    const turnId = "turn_1712345_0001";
+    const checkpoint = {
+      conversationId,
+      turnId,
+      sliceId: 1,
+      destination: SLACK_DESTINATION,
+      source: createSlackSource({
+        ...SLACK_DESTINATION,
+        threadTs: "1712345.0001",
+        visibility: "private",
+      }),
+      messages: [
+        {
+          role: "user" as const,
+          content: [{ type: "text" as const, text: "original request" }],
+          timestamp: 1_000,
+        },
+      ],
+    };
+    await Sentry.startNewTrace(() =>
+      Sentry.startSpan({ name: "chat.turn", op: "chat.turn" }, () =>
+        saveTurnCheckpoint({ ...checkpoint, mode: "running" }),
+      ),
+    );
+    // A later save from another trace, such as lost-worker recovery.
+    await Sentry.startNewTrace(() =>
+      saveTurnCheckpoint({ ...checkpoint, mode: "paused", reason: "yield" }),
+    );
+    await persistThreadStateById(conversationId, {
+      conversation: {
+        schemaVersion: 1,
+        compactions: [],
+        messages: [
+          {
+            id: "1712345.0001",
+            role: "user",
+            text: "original request",
+            createdAtMs: 1_000,
+            author: { userId: "U123" },
+          },
+        ],
+        processing: { activeTurnId: turnId },
+        vision: { byFileId: {} },
+      },
+    });
+
+    await Sentry.startNewTrace(() =>
+      Sentry.startSpan(
+        { name: "conversation.work.process", op: "queue.process" },
+        () =>
+          runNextPausedTurn(conversationId, {
+            agentRunner: {
+              run: async (run) => {
+                await deliverAssistantMessagesForTest(run, [{ text: "done" }]);
+                return completedAgentRun({
+                  text: "done",
+                  diagnostics: {
+                    assistantMessageCount: 1,
+                    modelId: "fake-agent-model",
+                    outcome: "success",
+                    toolCalls: [],
+                    toolErrorCount: 0,
+                    toolResultCount: 0,
+                    usedPrimaryText: true,
+                  },
+                });
+              },
+            },
+          }),
+      ),
+    );
+
+    const [first, resumed] = byOp("chat.turn");
+    const [worker] = byOp("queue.process");
+    expect(worker?.trace_id).not.toBe(first?.trace_id);
+    expect(resumed).toMatchObject({
+      trace_id: first?.trace_id,
+      parent_span_id: first?.span_id,
+      links: [
+        expect.objectContaining({
+          trace_id: worker?.trace_id,
+          span_id: worker?.span_id,
+        }),
+      ],
+    });
+  });
 });