diff --git a/TELEMETRY.md b/TELEMETRY.md index 928b77a6b5..444c5e362c 100644 --- a/TELEMETRY.md +++ b/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. diff --git a/packages/junior/src/chat/logging.ts b/packages/junior/src/chat/logging.ts index 8d45a22ec2..bb3def1943 100644 --- a/packages/junior/src/chat/logging.ts +++ b/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): void { const span = getActiveSpan(); diff --git a/packages/junior/src/chat/providers/slack/resume.ts b/packages/junior/src/chat/providers/slack/resume.ts index 3acdd15b1e..a7e8edadb9 100644 --- a/packages/junior/src/chat/providers/slack/resume.ts +++ b/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 { - return withLogContext(getResumeLogContext(args), () => - resumeSlackTurnInContext(args), + return await withTurnSpan( + getResumeLogContext(args), + args.traceContext && { traceContexts: [args.traceContext] }, + () => resumeSlackTurnInContext(args), ); } 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 32ebe0c8a9..61d26ad43d 100644 --- a/packages/junior/src/chat/providers/slack/turn-input-telemetry.ts +++ b/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 diff --git a/packages/junior/src/chat/task-execution/checkpoint.ts b/packages/junior/src/chat/task-execution/checkpoint.ts index 9d51740801..cf08ff13ad 100644 --- a/packages/junior/src/chat/task-execution/checkpoint.ts +++ b/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: diff --git a/packages/junior/src/chat/task-execution/paused-turn.ts b/packages/junior/src/chat/task-execution/paused-turn.ts index 0776557471..8e6e33cda4 100644 --- a/packages/junior/src/chat/task-execution/paused-turn.ts +++ b/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) => diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index c8bebf4bf0..4acb4a8d11 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 { 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(), }; } diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index a40718489f..3083c1b5ca 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/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; -export type TraceContext = NonNullable; +export type TraceContext = NonNullable>; /** Redis mailbox shape kept for deployed workers that still require this field. */ const storedInboundMessageSchema = inboundMessageSchema.extend({ diff --git a/packages/junior/src/chat/task-execution/turn-cursor.ts b/packages/junior/src/chat/task-execution/turn-cursor.ts index 82c5e8623e..42b14f6c28 100644 --- a/packages/junior/src/chat/task-execution/turn-cursor.ts +++ b/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, }), }), diff --git a/packages/junior/src/chat/task-execution/turn-wake.ts b/packages/junior/src/chat/task-execution/turn-wake.ts index 99637fc4a3..65b771652c 100644 --- a/packages/junior/src/chat/task-execution/turn-wake.ts +++ b/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, }; } 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 3499bcdd3c..f9dea1279b 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 @@ -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[] = []; @@ -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, + }), + ], + }); + }); });