feat(sdk): stamp gen_ai.conversation.id on chat spans and metrics
Adds `taskContext.setConversationId()` to the core API so the chat.task and chat.agent run boots can flag a chat run with the OTel GenAI `gen_ai.conversation.id` semantic attribute. The TaskContextSpanProcessor stamps it on every span at start and TaskContextMetricExporter copies it into every metric data point — `ctx.*` is filtered by the OTLP ingest, but `gen_ai.*` survives to the stored attributes column without a schema migration. Lets dashboard span/metric views correlate by chat conversation across multiple runs. Closes TRI-9082.
This commit is contained in:
@@ -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.
|
||||
@@ -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",
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
|
||||
@@ -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) => ({
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user