feat(slack): Put Slack turns in the ingress trace

TELEMETRY.md

@@ -239,8 +239,8 @@ Attributes: `gen_ai.conversation.id`, `messaging.system`,
 These spans use the [Sentry queue span conventions](https://develop.sentry.dev/sdk/telemetry/traces/modules/queues/),
 so they show in the Sentry Queues view. Each conversation work queue send gets
 one `queue.publish` span. Each queue message that the worker processes gets one
-`queue.process` span. The process span is not in the publish trace or the
-`slack.message.ingress` trace and has no link to them yet.
+`queue.process` span. The process span starts a new trace. A Slack `chat.turn`
+span links to it. See Slack Delivery.
 
 On these spans, the `messaging.*` attributes describe the queue, not Slack:
 
@@ -295,6 +295,18 @@ A Slack `chat.turn` span records mailbox timing for its input messages:
 For the oldest input message, delivery lag, wait, and the `chat.turn` duration
 are contiguous.
 
+A Slack `chat.turn` span is in the trace of a Slack webhook request:
+
+- Ingress stores the `sentry-trace` and `baggage` headers of its
+  `slack.message.ingress` span on the mailbox entry.
+- `chat.turn` continues the trace of the oldest input message that has these
+  headers. Its parent is that `slack.message.ingress` span. The trace keeps the
+  sampling decision of the ingress request.
+- `chat.turn` has a span link to the `slack.message.ingress` span of each other
+  input message, and a span link to the `queue.process` span of the worker.
+- Without stored headers, `chat.turn` is a child of the `queue.process` span.
+  Headers that are not valid start a new trace.
+
 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.
@@ -309,8 +321,8 @@ First feedback is the earliest `http.client` span for
 `POST slack.com/api/reactions.add` or
 `POST slack.com/api/assistant.threads.setStatus` in the turn's trace. For a
 mention, ingress adds the processing reaction before the worker runs. Use the
-ingress `reactions.add` span for a mention. It is in the same trace after
-#2269 PR 5b joins the ingress and turn traces.
+ingress `reactions.add` span for a mention. It is in the same trace as the
+turn.
 
 A mention turn also calls `reactions.add` again, and Slack returns an
 `already_reacted` error. This is a known double reaction. #2318 tracks it.

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

@@ -23,12 +23,7 @@ import {
   isTurnInputDeferredError,
   isTurnInputCommitLostError,
 } from "@/chat/runtime/turn";
-import {
-  logException,
-  logWarn,
-  withSpan,
-  withLogContext,
-} from "@/chat/logging";
+import { logException, logWarn, withLogContext } from "@/chat/logging";
 import { buildTurnErrorResponse } from "@/chat/services/turn-limit";
 import { getSlackErrorObservabilityAttributes } from "@/chat/slack/errors";
 import type {
@@ -46,6 +41,7 @@ import {
 } from "@/chat/providers/slack/processing-reaction";
 import {
   getTurnInputSpanAttributes,
+  withTurnSpan,
   type SlackTurnInputTelemetry,
 } from "@/chat/providers/slack/turn-input-telemetry";
 import {
@@ -736,7 +732,7 @@ export function createSlackTurnRuntime<
           hooks,
         );
 
-        await withSpan("chat.turn", "chat.turn", context, async (set) => {
+        await withTurnSpan(context, hooks.telemetry, async (set) => {
           const turnStartedAtMs = deps.now();
           if (hooks.telemetry) {
             set(getTurnInputSpanAttributes(hooks.telemetry, turnStartedAtMs));
@@ -920,7 +916,7 @@ export function createSlackTurnRuntime<
           channelId,
           runId,
         });
-        await withSpan("chat.turn", "chat.turn", turnContext, async (set) => {
+        await withTurnSpan(turnContext, hooks.telemetry, async (set) => {
           const turnStartedAtMs = deps.now();
           if (hooks.telemetry) {
             set(getTurnInputSpanAttributes(hooks.telemetry, turnStartedAtMs));

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

@@ -1,3 +1,11 @@
+import {
+  withSpan,
+  type LogContext,
+  type SetSpanAttributes,
+} from "@/chat/logging";
+import { continueTrace, getActiveSpan } from "@/chat/sentry";
+import type { TraceContext } from "@/chat/task-execution/state";
+
 /** Mailbox timing facts for the input messages that start a Slack turn. */
 export interface SlackTurnInputTelemetry {
   /** Input messages in the batch that starts the turn. */
@@ -6,6 +14,48 @@ export interface SlackTurnInputTelemetry {
   oldestCreatedAtMs: number;
   /** Webhook receipt time of the oldest input message. */
   oldestReceivedAtMs: number;
+  /** Ingress trace contexts of the input messages, oldest first. */
+  traceContexts: TraceContext[];
+}
+
+const SENTRY_TRACE_PATTERN = /^([0-9a-f]{32})-([0-9a-f]{16})(?:-([01]))?$/;
+
+function spanLinks({ sentryTrace }: TraceContext) {
+  const [, traceId, spanId, sampled] =
+    SENTRY_TRACE_PATTERN.exec(sentryTrace) ?? [];
+  return traceId && spanId
+    ? [{ context: { traceId, spanId, traceFlags: sampled === "1" ? 1 : 0 } }]
+    : [];
+}
+
+/**
+ * Run a `chat.turn` span in the ingress 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
+ * active span. A malformed trace context starts a new trace.
+ */
+export async function withTurnSpan<T>(
+  context: LogContext,
+  input: Pick<SlackTurnInputTelemetry, "traceContexts"> | undefined,
+  callback: (setSpanAttributes: SetSpanAttributes) => Promise<T>,
+): Promise<T> {
+  const [parent, ...others] = input?.traceContexts ?? [];
+  if (!parent) {
+    return await withSpan("chat.turn", "chat.turn", context, callback);
+  }
+  const activeSpan = getActiveSpan();
+  const links = [
+    ...others.flatMap(spanLinks),
+    ...(activeSpan ? [{ context: activeSpan.spanContext() }] : []),
+  ];
+  const { baggage, sentryTrace } = parent;
+  return await continueTrace({ baggage, sentryTrace }, () =>
+    withSpan("chat.turn", "chat.turn", context, async (setSpanAttributes) => {
+      getActiveSpan()?.addLinks(links);
+      return await callback(setSpanAttributes);
+    }),
+  );
 }
 
 /**

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 { logException } from "@/chat/logging";
+import { getTracePropagationHeaders, logException } from "@/chat/logging";
 import type {
   SlackTurnOptions,
   SteeringCandidateMessage,
@@ -678,6 +678,7 @@ export function createSlackConversationWorker(
       count: records.length,
       oldestCreatedAtMs: oldestRecord.createdAtMs,
       oldestReceivedAtMs: oldestRecord.receivedAtMs,
+      traceContexts: records.flatMap((record) => record.traceContext ?? []),
     };
     const latestMetadata = parseSlackMetadata(latestRecord.input.metadata);
     if (!latestMetadata) {
@@ -830,7 +831,7 @@ export function createSlackConversationWorker(
   };
 }
 
-/** Serialize a Slack message into the generic durable conversation mailbox. */
+/** Serialize a Slack message and the active trace into the durable mailbox. */
 export function buildSlackInboundMessage(args: {
   conversationId: string;
   installation?: SlackInstallationContext;
@@ -848,6 +849,7 @@ 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,
@@ -874,5 +876,6 @@ export function buildSlackInboundMessage(args: {
         message: args.message.toJSON(),
       } satisfies SlackConversationMessageMetadata,
     },
+    ...(sentryTrace && { traceContext: { sentryTrace, baggage } }),
   };
 }

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

@@ -131,6 +131,8 @@ export const inboundMessageSchema = z
 
 export type InboundMessage = z.output<typeof inboundMessageSchema>;
 
+export type TraceContext = NonNullable<InboundMessage["traceContext"]>;
+
 /** Redis mailbox shape kept for deployed workers that still require this field. */
 const storedInboundMessageSchema = inboundMessageSchema.extend({
   publishExternally: z.boolean(),

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

@@ -11,15 +11,23 @@ import {
 import * as Sentry from "@/chat/sentry";
 import { closeDb } from "@/chat/db";
 import { withLogContext } from "@/chat/logging";
-import { disconnectStateAdapter } from "@/chat/state/adapter";
+import { withTurnSpan } from "@/chat/providers/slack/turn-input-telemetry";
+import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter";
+import { createSlackConversationWorker } from "@/chat/task-execution/slack-work";
 import { appendInboundMessage } from "@/chat/task-execution/store";
 import { consumeConversationQueueMessage } from "@/chat/task-execution/vercel-callback";
 import { createVercelConversationWorkQueue } from "@/chat/task-execution/vercel-queue";
 import {
   CONVERSATION_ID,
   conversationQueueMessage,
   createConversationWorkQueueTestAdapter,
+  createNoopSlackWebhookRuntime,
+  createSlackAdapterFixture,
+  handleSlackWebhookAndFlush,
   inboundMessage,
+  SLACK_BOT_USER_ID,
+  slackEnvelope,
+  slackWebhookRequest,
 } from "../../fixtures/conversation-work";
 
 describe("conversation work telemetry", () => {
@@ -39,7 +47,12 @@ describe("conversation work telemetry", () => {
     client?.on("spanEnd", (span) => {
       const json = Sentry.spanToJSON(span);
       const op = json.attributes?.["sentry.op"];
-      if (op === "queue.publish" || op === "queue.process") {
+      if (
+        op === "queue.publish" ||
+        op === "queue.process" ||
+        op === "slack.message.ingress" ||
+        op === "chat.turn"
+      ) {
         spans.push(json);
       }
     });
@@ -113,4 +126,73 @@ describe("conversation work telemetry", () => {
       }),
     ]);
   });
+
+  it("puts a Slack turn in the ingress trace of its oldest input", async () => {
+    const state = getStateAdapter();
+    await state.connect();
+    const queue = createConversationWorkQueueTestAdapter();
+    const adapter = createSlackAdapterFixture();
+    // Each Slack webhook request starts its own trace.
+    for (const ts of ["1712345.0001", "1712345.0002"]) {
+      await Sentry.startNewTrace(() =>
+        handleSlackWebhookAndFlush({
+          request: slackWebhookRequest(
+            slackEnvelope({
+              text: `<@${SLACK_BOT_USER_ID}> check`,
+              threadTs: "1712345.0001",
+              ts,
+            }),
+          ),
+          services: {
+            getSlackAdapter: () => adapter,
+            queue,
+            runtime: createNoopSlackWebhookRuntime(),
+            state,
+          },
+        }),
+      );
+    }
+    await consumeConversationQueueMessage(queue.takeMessage(), {
+      messageId: "msg_1",
+      queue,
+      run: createSlackConversationWorker({
+        getSlackAdapter: () => adapter,
+        runNextPausedTurn: async () => false,
+        runtime: {
+          handleNewMention: async (_thread, _message, hooks) =>
+            await withTurnSpan({}, hooks.telemetry, async () => {
+              await hooks.ack?.();
+            }),
+          handleSubscribedMessage: async () => {
+            throw new Error("unexpected subscribed route");
+          },
+        },
+        state,
+      }),
+      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);
+    expect(worker?.trace_id).not.toBe(oldest?.trace_id);
+    expect(byOp("chat.turn")).toEqual([
+      expect.objectContaining({
+        trace_id: oldest?.trace_id,
+        parent_span_id: oldest?.span_id,
+        links: [
+          expect.objectContaining({
+            trace_id: newest?.trace_id,
+            span_id: newest?.span_id,
+          }),
+          expect.objectContaining({
+            trace_id: worker?.trace_id,
+            span_id: worker?.span_id,
+          }),
+        ],
+      }),
+    ]);
+  });
 });

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

@@ -516,6 +516,7 @@ describe("Slack conversation work execution", () => {
       count: 2,
       oldestCreatedAtMs: oldestInput?.createdAtMs,
       oldestReceivedAtMs: oldestInput?.receivedAtMs,
+      traceContexts: [],
     });
     expect(calls[0]?.destination).toEqual(SLACK_DESTINATION);
     expect(calls[0]?.thread.id).toBe(CONVERSATION_ID);

packages/junior/tests/unit/slack/slack-turn-input-telemetry.test.ts

@@ -67,6 +67,7 @@ describe("Slack turn input timing", () => {
           count: 3,
           oldestCreatedAtMs: 1_000.4,
           oldestReceivedAtMs: 1_250,
+          traceContexts: [],
         },
         1_900,
       ),
@@ -77,7 +78,12 @@ describe("Slack turn input timing", () => {
     });
     expect(
       getTurnInputSpanAttributes(
-        { count: 1, oldestCreatedAtMs: 2_000, oldestReceivedAtMs: 1_500 },
+        {
+          count: 1,
+          oldestCreatedAtMs: 2_000,
+          oldestReceivedAtMs: 1_500,
+          traceContexts: [],
+        },
         1_000,
       ),
     ).toEqual({ "app.turn.input_count": 1 });
@@ -100,6 +106,7 @@ describe("Slack turn input timing", () => {
           count: 2,
           oldestCreatedAtMs: TURN_STARTED_AT_MS - 4_000,
           oldestReceivedAtMs: TURN_STARTED_AT_MS - 3_000,
+          traceContexts: [],
         },
       });