diff --git a/.changeset/chat-session-attributes.md b/.changeset/chat-session-attributes.md new file mode 100644 index 000000000..ec4c6a540 --- /dev/null +++ b/.changeset/chat-session-attributes.md @@ -0,0 +1,6 @@ +--- +"@trigger.dev/sdk": patch +"@trigger.dev/core": patch +--- + +Stamp `gen_ai.conversation.id` (the chat id) on every span and metric emitted from inside a `chat.task` or `chat.agent` run. Lets you filter dashboard spans, runs, and metrics by the chat conversation that produced them — independent of the run boundary, so multi-run chats correlate cleanly. No code changes required on the user side. diff --git a/packages/core/src/v3/semanticInternalAttributes.ts b/packages/core/src/v3/semanticInternalAttributes.ts index 2c715a03e..e6e016066 100644 --- a/packages/core/src/v3/semanticInternalAttributes.ts +++ b/packages/core/src/v3/semanticInternalAttributes.ts @@ -13,6 +13,7 @@ export const SemanticInternalAttributes = { RUN_ID: "ctx.run.id", RUN_IS_TEST: "ctx.run.isTest", RUN_IS_REPLAY: "ctx.run.isReplay", + GEN_AI_CONVERSATION_ID: "gen_ai.conversation.id", ORIGINAL_RUN_ID: "$original_run_id", BATCH_ID: "ctx.batch.id", TASK_SLUG: "ctx.task.id", diff --git a/packages/core/src/v3/taskContext/index.test.ts b/packages/core/src/v3/taskContext/index.test.ts new file mode 100644 index 000000000..34d169a17 --- /dev/null +++ b/packages/core/src/v3/taskContext/index.test.ts @@ -0,0 +1,86 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { unregisterGlobal } from "../utils/globals.js"; +import { SemanticInternalAttributes } from "../semanticInternalAttributes.js"; +import { TaskContextAPI } from "./index.js"; + +const FAKE_CTX = { + attempt: { id: "attempt_1", number: 1, startedAt: new Date(), status: "EXECUTING" as const }, + run: { + id: "run_1", + payload: undefined, + payloadType: "application/json", + context: undefined, + createdAt: new Date(), + tags: [], + isTest: false, + isReplay: false, + startedAt: new Date(), + durationMs: 0, + costInCents: 0, + baseCostInCents: 0, + }, + task: { id: "my-task", filePath: "src/trigger/task.ts", exportName: "myTask" }, + queue: { id: "queue_1", name: "default" }, + environment: { id: "env_1", slug: "dev", type: "DEVELOPMENT" as const }, + organization: { id: "org_1", slug: "acme", name: "Acme" }, + project: { id: "proj_1", ref: "proj_xyz", slug: "demo", name: "Demo" }, + machine: { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, +} as never; + +const FAKE_WORKER = { id: "worker_1", version: "1.0.0", contentHash: "abc" } as never; + +describe("TaskContextAPI conversation id", () => { + afterEach(() => { + unregisterGlobal("task-context"); + TaskContextAPI.getInstance().setConversationId(undefined); + }); + + it("returns no conversation attribute when setConversationId was never called", () => { + const api = TaskContextAPI.getInstance(); + api.setGlobalTaskContext({ ctx: FAKE_CTX, worker: FAKE_WORKER }); + + expect(api.attributes[SemanticInternalAttributes.GEN_AI_CONVERSATION_ID]).toBeUndefined(); + }); + + it("includes gen_ai.conversation.id after setConversationId", () => { + const api = TaskContextAPI.getInstance(); + api.setGlobalTaskContext({ ctx: FAKE_CTX, worker: FAKE_WORKER }); + + api.setConversationId("chat_123"); + + expect(api.attributes[SemanticInternalAttributes.GEN_AI_CONVERSATION_ID]).toBe("chat_123"); + }); + + it("clears the conversation attribute when called with undefined", () => { + const api = TaskContextAPI.getInstance(); + api.setGlobalTaskContext({ ctx: FAKE_CTX, worker: FAKE_WORKER }); + api.setConversationId("chat_123"); + + api.setConversationId(undefined); + + expect(api.attributes[SemanticInternalAttributes.GEN_AI_CONVERSATION_ID]).toBeUndefined(); + expect(api.conversationId).toBeUndefined(); + }); + + it("returns no attributes when there is no task context", () => { + const api = TaskContextAPI.getInstance(); + api.setConversationId("chat_123"); + + expect(api.attributes).toEqual({}); + }); + + it("clears conversation id when a new task context is registered (warm restart)", () => { + const api = TaskContextAPI.getInstance(); + api.setGlobalTaskContext({ ctx: FAKE_CTX, worker: FAKE_WORKER }); + api.setConversationId("chat_old"); + + api.setGlobalTaskContext({ ctx: FAKE_CTX, worker: FAKE_WORKER }); + + expect(api.attributes[SemanticInternalAttributes.GEN_AI_CONVERSATION_ID]).toBeUndefined(); + }); +}); diff --git a/packages/core/src/v3/taskContext/index.ts b/packages/core/src/v3/taskContext/index.ts index 92e0194cd..ecbfa184a 100644 --- a/packages/core/src/v3/taskContext/index.ts +++ b/packages/core/src/v3/taskContext/index.ts @@ -9,6 +9,7 @@ const API_NAME = "task-context"; export class TaskContextAPI { private static _instance?: TaskContextAPI; private _runDisabled = false; + private _conversationId?: string; private constructor() {} @@ -45,6 +46,7 @@ export class TaskContextAPI { return { ...this.contextAttributes, ...this.workerAttributes, + ...this.conversationAttributes, [SemanticInternalAttributes.WARM_START]: !!this.isWarmStart, }; } @@ -52,6 +54,19 @@ export class TaskContextAPI { return {}; } + get conversationAttributes(): Attributes { + if (!this._conversationId) return {}; + return { [SemanticInternalAttributes.GEN_AI_CONVERSATION_ID]: this._conversationId }; + } + + get conversationId(): string | undefined { + return this._conversationId; + } + + public setConversationId(conversationId: string | undefined): void { + this._conversationId = conversationId || undefined; + } + get resourceAttributes(): Attributes { if (this.ctx) { return { @@ -109,6 +124,11 @@ export class TaskContextAPI { public setGlobalTaskContext(taskContext: TaskContext): boolean { this._runDisabled = false; + // Each run boot re-registers the global; clear any conversation id + // left over from a previous run on this warm-restarted process so + // attributes don't bleed across runs that don't call + // `setConversationId` themselves. + this._conversationId = undefined; return registerGlobal(API_NAME, taskContext, true); } diff --git a/packages/core/src/v3/taskContext/otelProcessors.ts b/packages/core/src/v3/taskContext/otelProcessors.ts index 1c0958d65..fc30e9d11 100644 --- a/packages/core/src/v3/taskContext/otelProcessors.ts +++ b/packages/core/src/v3/taskContext/otelProcessors.ts @@ -36,6 +36,17 @@ export class TaskContextSpanProcessor implements SpanProcessor { if (!taskContext.isRunDisabled && taskContext.ctx.run.tags?.length) { span.setAttribute(SemanticInternalAttributes.RUN_TAGS, taskContext.ctx.run.tags); } + + // Stamp `gen_ai.conversation.id` (OTel GenAI semantic convention) + // directly on every span so it survives the OTLP ingest's `ctx.*` + // strip and lands in the stored attributes column without a schema + // migration. + if (taskContext.conversationId) { + span.setAttribute( + SemanticInternalAttributes.GEN_AI_CONVERSATION_ID, + taskContext.conversationId + ); + } } if (!isPartialSpan(span) && !skipPartialSpan(span)) { @@ -178,6 +189,11 @@ export class TaskContextMetricExporter implements PushMetricExporter { contextAttrs[SemanticInternalAttributes.RUN_TAGS] = ctx.run.tags; } + if (taskContext.conversationId) { + contextAttrs[SemanticInternalAttributes.GEN_AI_CONVERSATION_ID] = + taskContext.conversationId; + } + const modified: ResourceMetrics = { resource: metrics.resource, scopeMetrics: metrics.scopeMetrics.map((scope) => ({ diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index fe0576fbc..fb9514ac9 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -149,6 +149,20 @@ function getChatSession(): SessionHandle { return handle; } +/** + * Stamp `gen_ai.conversation.id` on the active span at chat-run boot. + * The run-level span is already alive when the run callback fires, so + * `TaskContextSpanProcessor.onStart` (which stamps subsequent spans + * automatically) won't catch it — set explicitly here. + */ +function stampConversationIdOnActiveSpan( + conversationId: string | undefined, + span = trace.getActiveSpan() +): void { + if (!span || !conversationId) return; + span.setAttribute(SemanticInternalAttributes.GEN_AI_CONVERSATION_ID, conversationId); +} + type ToolResultContent = Array< | { type: "text"; @@ -3697,6 +3711,8 @@ function chatCustomAgent< // No client-side upsert needed. locals.set(chatSessionHandleKey, sessions.open(payload.chatId)); locals.set(chatAgentRunContextKey, runOptions.ctx); + taskContext.setConversationId(payload.chatId); + stampConversationIdOnActiveSpan(payload.chatId); return userRun(payload, runOptions); }, }); @@ -3779,12 +3795,13 @@ function chatAgent< // `chat.createStartSessionAction` or browser-direct) before this // run is triggered — no client-side upsert needed here. locals.set(chatSessionHandleKey, sessions.open(payload.chatId)); + taskContext.setConversationId(payload.chatId); - // Set gen_ai.conversation.id on the run-level span for dashboard context + // Stamp `gen_ai.conversation.id` on the run-level span. Every + // nested span inherits the same attribute via + // `TaskContextSpanProcessor.onStart`. const activeSpan = trace.getActiveSpan(); - if (activeSpan) { - activeSpan.setAttribute("gen_ai.conversation.id", payload.chatId); - } + stampConversationIdOnActiveSpan(payload.chatId, activeSpan); // Store static UIMessageStream options in locals so resolveUIMessageStreamOptions() can read them if (uiMessageStreamOptions) {