From 671278827cf7484c3e7c7d703d306b83a61914b5 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Fri, 30 Oct 2026 03:03:47 +0000 Subject: [PATCH] feat(slack): Put Slack turns in the ingress trace Part of #2269 Co-Authored-By: Thomas Jiang <20501235+thomasjiangcy@users.noreply.github.com> --- TELEMETRY.md | 20 ++++- .../src/chat/providers/slack/runtime.ts | 12 +-- .../providers/slack/turn-input-telemetry.ts | 50 +++++++++++ .../src/chat/task-execution/slack-work.ts | 7 +- .../junior/src/chat/task-execution/state.ts | 2 + .../conversation-work-telemetry.test.ts | 86 ++++++++++++++++++- .../slack-conversation-work.test.ts | 1 + .../slack/slack-turn-input-telemetry.test.ts | 9 +- 8 files changed, 170 insertions(+), 17 deletions(-) diff --git a/TELEMETRY.md b/TELEMETRY.md index bf57b00468..928b77a6b5 100644 --- a/TELEMETRY.md +++ b/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. diff --git a/packages/junior/src/chat/providers/slack/runtime.ts b/packages/junior/src/chat/providers/slack/runtime.ts index 1a63f9b03e..a14420113d 100644 --- a/packages/junior/src/chat/providers/slack/runtime.ts +++ b/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)); diff --git a/packages/junior/src/chat/providers/slack/turn-input-telemetry.ts b/packages/junior/src/chat/providers/slack/turn-input-telemetry.ts index c3cc49a2e3..32ebe0c8a9 100644 --- a/packages/junior/src/chat/providers/slack/turn-input-telemetry.ts +++ b/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( + context: LogContext, + input: Pick | undefined, + callback: (setSpanAttributes: SetSpanAttributes) => Promise, +): Promise { + 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); + }), + ); } /** diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index ac9d794e7a..c8bebf4bf0 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/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 } }), }; } diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index adfef2a2b1..a40718489f 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -131,6 +131,8 @@ export const inboundMessageSchema = z export type InboundMessage = z.output; +export type TraceContext = NonNullable; + /** Redis mailbox shape kept for deployed workers that still require this field. */ const storedInboundMessageSchema = inboundMessageSchema.extend({ publishExternally: z.boolean(), diff --git a/packages/junior/tests/component/task-execution/conversation-work-telemetry.test.ts b/packages/junior/tests/component/task-execution/conversation-work-telemetry.test.ts index f9edea2cbf..3499bcdd3c 100644 --- a/packages/junior/tests/component/task-execution/conversation-work-telemetry.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work-telemetry.test.ts @@ -11,7 +11,9 @@ 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"; @@ -19,7 +21,13 @@ 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, + }), + ], + }), + ]); + }); }); diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index c420b86445..ed253a4b27 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/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); diff --git a/packages/junior/tests/unit/slack/slack-turn-input-telemetry.test.ts b/packages/junior/tests/unit/slack/slack-turn-input-telemetry.test.ts index 079b0f4079..a4c5944208 100644 --- a/packages/junior/tests/unit/slack/slack-turn-input-telemetry.test.ts +++ b/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: [], }, });