diff --git a/CHANGELOG.md b/CHANGELOG.md index ad68004..afd3fdf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,21 +5,11 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [1.8.0] - 2026-08-03 - -### Added - -- **Postscript Failed Status**: After-hook failures (`after`, `afterEach`, `afterAll`) on otherwise successful scenarios are now reported as `postscript_failed` instead of being logged and ignored. Existing failure statuses (`failed` / `prescript_failed`) are never overwritten. `after` and `afterEach` always both attempt to run; `afterAll` failure marks only successfully completed scenarios. - -## [1.7.0] - 2026-07-27 +## [1.9.0] - Unreleased ### Added -- **Simulation Lifecycle Hooks**: Added prescript/postscript support for multi-turn simulations via `SimulationHooks` (`beforeAll`, `beforeEach`, `before`, `after`, `afterEach`, `afterAll`). Hooks can return setup context passed into `BaseTask.run`, and the run uses a two-phase initialize / first-turn flow so hooks execute before any LLM spend. Execution order is `beforeAll` → `beforeEach` → item-specific `before` → task → item-specific `after` → `afterEach` → `afterAll`. `beforeAll` failure aborts the run as `prescript_failed`; item `before` failure marks only that scenario; `after`, `afterEach`, and `afterAll` failures are logged and do not affect status. - -## [Unreleased] - -### Added +- **Time to First Token (TTFT) & Relative TTFT**: All LLM generation spans now record `gen_ai.performance.time_to_first_token` (seconds from call start to first content chunk), `gen_ai.performance.relative_time_to_first_token` (seconds from trace root span start to first content chunk), and `gen_ai.performance.time_to_first_token.timestamp` (absolute ISO 8601 UTC time of first token). Supported across OpenAI, Anthropic, Groq, Mistral, Google GenAI, and Google Generative AI for both streaming and non-streaming calls. RTTFT is silently skipped when no root span exists. - **Opt-in prompt caching** — `Netra.prompts.getPrompt()` accepts `useCache` and `cacheTtl`. When `useCache` is true, responses are served from an in-memory TTL cache (default TTL: `PROMPT_CACHE_TTL_SECONDS` = 60). Caching is off by default. - **Models API** — `Netra.models.getModelPricing()` fetches model pricing (optional `name` filter) with the same opt-in cache pattern (`useCache`, `cacheTtl`; default TTL: `MODEL_PRICING_CACHE_TTL_SECONDS` = 300). @@ -30,6 +20,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Prompt cache TTL** — Default TTL is the module constant `PROMPT_CACHE_TTL_SECONDS` (60). Override per call with `cacheTtl`. Removed unused `cacheTtlSeconds` init config and `NETRA_CACHE_TTL_SECONDS` env var. +## [1.8.0] - 2026-08-03 + +### Added + +- **Postscript Failed Status**: After-hook failures (`after`, `afterEach`, `afterAll`) on otherwise successful scenarios are now reported as `postscript_failed` instead of being logged and ignored. Existing failure statuses (`failed` / `prescript_failed`) are never overwritten. `after` and `afterEach` always both attempt to run; `afterAll` failure marks only successfully completed scenarios. + +## [1.7.0] - 2026-07-27 + +### Added + +- **Simulation Lifecycle Hooks**: Added prescript/postscript support for multi-turn simulations via `SimulationHooks` (`beforeAll`, `beforeEach`, `before`, `after`, `afterEach`, `afterAll`). Hooks can return setup context passed into `BaseTask.run`, and the run uses a two-phase initialize / first-turn flow so hooks execute before any LLM spend. Execution order is `beforeAll` → `beforeEach` → item-specific `before` → task → item-specific `after` → `afterEach` → `afterAll`. `beforeAll` failure aborts the run as `prescript_failed`; item `before` failure marks only that scenario; `after`, `afterEach`, and `afterAll` failures are logged and do not affect status. + ## [1.6.0] - 2026-07-17 ### Added diff --git a/package-lock.json b/package-lock.json index bc6fb8b..804b9ba 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "netra-sdk", - "version": "1.8.0", + "version": "1.9.0-dev.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "netra-sdk", - "version": "1.8.0", + "version": "1.9.0-dev.0", "license": "Apache-2.0", "dependencies": { "@opentelemetry/api": "^1.9.0", diff --git a/package.json b/package.json index 2e50dab..454e74b 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "netra-sdk", - "version": "1.8.0", + "version": "1.9.0-dev.0", "description": "A comprehensive TypeScript/JavaScript SDK for AI application observability built on top of OpenTelemetry and Traceloop", "type": "module", "main": "./dist/index.cjs", @@ -79,40 +79,46 @@ "openai": "^4.0.0 || ^5.0.0 || ^6.0.0" }, "peerDependenciesMeta": { - "@opentelemetry/instrumentation": { + "@anthropic-ai/sdk": { "optional": true }, - "@opentelemetry/instrumentation-http": { + "@google/genai": { "optional": true }, - "@opentelemetry/instrumentation-express": { + "@google/generative-ai": { "optional": true }, - "@opentelemetry/instrumentation-undici": { + "@langchain/langgraph": { "optional": true }, - "@prisma/instrumentation": { + "@langchain/ollama": { "optional": true }, - "openai": { + "@mistralai/mistralai": { "optional": true }, - "groq-sdk": { + "@openai/agents": { "optional": true }, - "@mistralai/mistralai": { + "@opentelemetry/instrumentation": { "optional": true }, - "@google/generative-ai": { + "@opentelemetry/instrumentation-express": { "optional": true }, - "@google/genai": { + "@opentelemetry/instrumentation-http": { "optional": true }, - "@anthropic-ai/sdk": { + "@opentelemetry/instrumentation-undici": { "optional": true }, - "@openai/agents": { + "@prisma/instrumentation": { + "optional": true + }, + "groq-sdk": { + "optional": true + }, + "openai": { "optional": true } }, diff --git a/src/index.ts b/src/index.ts index ea42421..2cfe47e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -235,23 +235,6 @@ export class Netra { } this._initialized = true; - Logger.info("Netra successfully initialized."); - - { - let pkgVersion = Config.LIBRARY_VERSION; - let pkgPath = "unknown"; - try { - const req = createRequire(import.meta.url); - pkgPath = req.resolve("../package.json"); - const pkg = req("../package.json"); - pkgVersion = pkg?.version || pkgVersion; - } catch { - // keep defaults - } - Logger.debug( - `SDK version=${pkgVersion} libraryVersion=${Config.LIBRARY_VERSION} build=langgraph-parenting-v3 packageJson=${pkgPath}`, - ); - } // Graceful shutdown logic const handleSignal = async (signal: string) => { @@ -287,6 +270,7 @@ export class Netra { // Wait for all async instrumentations to be ready await instrumentationsReady; + Logger.info("Netra successfully initialized."); } static async shutdown(): Promise { diff --git a/src/instrumentation/anthropic/utils.ts b/src/instrumentation/anthropic/utils.ts index b378226..0ff521f 100644 --- a/src/instrumentation/anthropic/utils.ts +++ b/src/instrumentation/anthropic/utils.ts @@ -1,5 +1,6 @@ import { Span, SpanStatusCode } from "@opentelemetry/api"; import { Logger } from "../../logger"; +import type { FirstTokenTracker } from "../../utils/span-timing"; import { setRequestAttributes as setBaseRequestAttributes, setResponseAttributes as setBaseResponseAttributes, @@ -13,6 +14,7 @@ export function processStreamChunk( completeResponse: Record, chunk: any, span: Span, + tokenTracker?: FirstTokenTracker, ): void { try { switch (chunk.type) { @@ -67,6 +69,7 @@ export function processStreamChunk( targetBlock.input += chunk.delta.partial_json ?? ""; } else if (chunk.delta?.text) { targetBlock.text += chunk.delta.text; + tokenTracker?.markFirstToken(); } break; } diff --git a/src/instrumentation/anthropic/wrappers.ts b/src/instrumentation/anthropic/wrappers.ts index 73267b0..41642aa 100644 --- a/src/instrumentation/anthropic/wrappers.ts +++ b/src/instrumentation/anthropic/wrappers.ts @@ -10,6 +10,10 @@ import { import { Logger } from "../../logger"; import { wrapResponse } from "../../utils/response-handler"; import { safeStringify } from "../../utils/serialization"; +import { + FirstTokenTracker, + recordNonStreamingTimingAttributes, +} from "../../utils/span-timing"; import { SpanAttributes } from "../span-attributes"; import { defineHidden, @@ -38,6 +42,8 @@ const WRAPPER_OWN_PROPS = new Set([ "spanFinalized", "completionPending", "listenerMap", + "tokenTracker", + "ttftListener", ]); const EVENT_EMITTER_METHODS = new Set([ @@ -152,6 +158,8 @@ class MessageStreamWrapper { private spanFinalized = false; private completionPending = false; private listenerMap = new WeakMap>(); + private tokenTracker!: FirstTokenTracker; + private ttftListener!: (data: any) => void; constructor( span: Span, @@ -165,6 +173,7 @@ class MessageStreamWrapper { defineHidden(this, "messageStream", messageStream); defineHidden(this, "startTime", startTime); defineHidden(this, "requestKwargs", requestKwargs); + defineHidden(this, "tokenTracker", new FirstTokenTracker(span, startTime)); defineHidden( this, "spanContext", @@ -238,7 +247,13 @@ class MessageStreamWrapper { if (prop === "removeAllListeners") { return function (event?: string) { target.listenerMap = new WeakMap(); - return method.call(target.messageStream, event); + const result = method.call(target.messageStream, event); + if (!event) { + target.attachSafetyNetListeners(); + } else if (event === "text") { + target.messageStream.on("text", target.ttftListener); + } + return result; }; } return method.bind(target.messageStream); @@ -254,12 +269,20 @@ class MessageStreamWrapper { const result = await method.call(target.messageStream, ...args); if (prop === "finalMessage" || prop === "done") { + if (result) { + const hasText = Array.isArray(result.content) && + result.content.some((b: any) => b.type === "text" && b.text); + if (hasText) { + target.tokenTracker.markFirstToken(); + } + } target.finalizeSpanFromMessage(result); } else if (prop === "finalText") { if (typeof result === "string" && result.length > 0) { target.completeResponse.content = [ { type: "text", text: result }, ]; + target.tokenTracker.markFirstToken(); } else { target.flushCurrentText(); } @@ -282,27 +305,35 @@ class MessageStreamWrapper { }); } + private attachSafetyNetListeners(): void { + this.ttftListener = (data: any) => { + if (data) this.tokenTracker.markFirstToken(); + }; + this.messageStream.on("text", this.ttftListener); + + this.messageStream.on("end", () => { + if (!this.completionPending) { + this.finalizeSpanOnce(SpanStatusCode.OK); + } + }); + this.messageStream.on("error", (err: any) => { + if (err && !this.spanFinalized) { + this.span.setStatus({ + code: SpanStatusCode.ERROR, + message: err instanceof Error ? err.message : String(err), + }); + this.span.recordException( + err instanceof Error ? err : new Error(String(err)), + ); + } + this.finalizeSpanOnce(SpanStatusCode.ERROR); + }); + } + private registerSafetyNetListeners(): void { try { if (typeof this.messageStream?.on !== "function") return; - - this.messageStream.on("end", () => { - if (!this.completionPending) { - this.finalizeSpanOnce(SpanStatusCode.OK); - } - }); - this.messageStream.on("error", (err: any) => { - if (err && !this.spanFinalized) { - this.span.setStatus({ - code: SpanStatusCode.ERROR, - message: err instanceof Error ? err.message : String(err), - }); - this.span.recordException( - err instanceof Error ? err : new Error(String(err)), - ); - } - this.finalizeSpanOnce(SpanStatusCode.ERROR); - }); + this.attachSafetyNetListeners(); } catch (e) { Logger.error( "netra.instrumentation.anthropic: safety net listener registration failed", @@ -319,7 +350,12 @@ class MessageStreamWrapper { let errorOccurred = false; try { for await (const chunk of this.messageStream) { - processStreamChunk(this.completeResponse, chunk, this.span); + processStreamChunk( + this.completeResponse, + chunk, + this.span, + this.tokenTracker, + ); yield chunk; } } catch (err) { @@ -350,6 +386,7 @@ class MessageStreamWrapper { this.completeResponse.currentText = ""; } this.completeResponse.currentText += data; + if (data) this.tokenTracker.markFirstToken(); break; case "contentBlock": @@ -444,13 +481,19 @@ function anthropicWrapper( model: "", usage: {}, }; + const tokenTracker = new FirstTokenTracker(span, startTime); return wrapResponse( response, { withContext: (fn) => context.with(spanContext, fn), onChunk: (chunk) => - processStreamChunk(completeResponse, chunk, span), + processStreamChunk( + completeResponse, + chunk, + span, + tokenTracker, + ), onError: (error) => { Logger.error("netra.instrumentation.anthropic:", error); span.setStatus({ @@ -468,6 +511,9 @@ function anthropicWrapper( "llm.response.duration", (endTime - startTime) / 1000, ); + if (requestType !== "batches") { + recordNonStreamingTimingAttributes(span, startTime, endTime); + } }, finalize: (status) => { const hasStreamData = diff --git a/src/instrumentation/google-genai/utils.ts b/src/instrumentation/google-genai/utils.ts index e2308ab..91840db 100644 --- a/src/instrumentation/google-genai/utils.ts +++ b/src/instrumentation/google-genai/utils.ts @@ -14,6 +14,7 @@ import { Span } from "@opentelemetry/api"; import { Logger } from "../../logger"; import { safeStringify } from "../../utils/serialization"; +import type { FirstTokenTracker } from "../../utils/span-timing"; import { SpanAttributes } from "../span-attributes"; import { TracedMessage, @@ -488,6 +489,7 @@ export function processStreamChunk( chunk: any, span: Span, startTime: number, + tokenTracker?: FirstTokenTracker, ): void { try { if (chunk.modelVersion) { @@ -502,10 +504,7 @@ export function processStreamChunk( if (chunkText && chunkText.length > 0) { if (!accumulated._text) { accumulated._text = chunkText; - span.setAttribute( - "gen_ai.performance.time_to_first_token", - (Date.now() - startTime) / 1000, - ); + tokenTracker?.markFirstToken(); } else { accumulated._text += chunkText; } diff --git a/src/instrumentation/google-genai/wrappers.ts b/src/instrumentation/google-genai/wrappers.ts index 724fc2e..3a5fd96 100644 --- a/src/instrumentation/google-genai/wrappers.ts +++ b/src/instrumentation/google-genai/wrappers.ts @@ -21,6 +21,10 @@ import { } from "@opentelemetry/api"; import { Logger } from "../../logger"; import { wrapResponse } from "../../utils/response-handler"; +import { + FirstTokenTracker, + recordNonStreamingTimingAttributes, +} from "../../utils/span-timing"; import { createSuppressedContext, modelAsDict, @@ -127,6 +131,13 @@ function genericWrapperFactory( SpanAttributes.LLM_RESPONSE_DURATION, (endTime - startTime) / 1000, ); + if (requestType !== "embedding") { + recordNonStreamingTimingAttributes( + span, + startTime, + endTime, + ); + } }, onError: (error) => { span.setStatus({ @@ -194,6 +205,7 @@ function streamWrapperFactory( (span: Span) => { const startTime = Date.now(); const accumulated: Record = {}; + const tokenTracker = new FirstTokenTracker(span, startTime); try { setRequestAttributes(span, params, requestType); @@ -209,7 +221,13 @@ function streamWrapperFactory( { withContext: (fn) => context.with(spanContext, fn), onChunk: (chunk) => { - processStreamChunk(accumulated, chunk, span, startTime); + processStreamChunk( + accumulated, + chunk, + span, + startTime, + tokenTracker, + ); }, onError: (error) => { span.setStatus({ diff --git a/src/instrumentation/google-generative-ai/wrappers.ts b/src/instrumentation/google-generative-ai/wrappers.ts index e60a8a3..7eb0811 100644 --- a/src/instrumentation/google-generative-ai/wrappers.ts +++ b/src/instrumentation/google-generative-ai/wrappers.ts @@ -6,6 +6,10 @@ import { context, } from "@opentelemetry/api"; import { Logger } from "../../logger"; +import { + FirstTokenTracker, + recordNonStreamingTimingAttributes, +} from "../../utils/span-timing"; import { isPromise, modelAsDict, @@ -99,12 +103,15 @@ function googleGenerativeAIWrapper( const endTime = Date.now(); const responseDict = modelAsDict(value); setResponseAttributes(span, responseDict); - const duration = (endTime - startTime) / 1000; - span.setAttribute("llm.response.duration", duration); + span.setAttribute( + "llm.response.duration", + (endTime - startTime) / 1000, + ); if (requestType !== "embedding") { - span.setAttribute( - "gen_ai.performance.time_to_first_token", - duration, + recordNonStreamingTimingAttributes( + span, + startTime, + endTime, ); } } catch (e) { @@ -133,6 +140,9 @@ function googleGenerativeAIWrapper( "llm.response.duration", (endTime - startTime) / 1000, ); + if (requestType !== "embedding") { + recordNonStreamingTimingAttributes(span, startTime, endTime); + } } catch (e) { Logger.error(`${LOG_PREFIX}:`, e); } @@ -196,6 +206,8 @@ function googleGenerativeAIStreamWrapper( return response; } + const tokenTracker = new FirstTokenTracker(span, startTime); + return (async () => { try { const streamResult: any = await response; @@ -210,11 +222,14 @@ function googleGenerativeAIStreamWrapper( const endTime = Date.now(); const responseDict = modelAsDict(streamResult); setResponseAttributes(span, responseDict); - const duration = (endTime - startTime) / 1000; - span.setAttribute("llm.response.duration", duration); span.setAttribute( - "gen_ai.performance.time_to_first_token", - duration, + "llm.response.duration", + (endTime - startTime) / 1000, + ); + recordNonStreamingTimingAttributes( + span, + startTime, + endTime, ); } catch (e) { Logger.error(`${LOG_PREFIX}:`, e); @@ -224,8 +239,6 @@ function googleGenerativeAIStreamWrapper( return streamResult; } - let firstTokenRecorded = false; - const wrappedStream: AsyncIterable = { [Symbol.asyncIterator]() { const iterator = originalStream[Symbol.asyncIterator](); @@ -256,10 +269,9 @@ function googleGenerativeAIStreamWrapper( setResponseAttributes(span, responseDict); } - const duration = (endTime - startTime) / 1000; span.setAttribute( "llm.response.duration", - duration, + (endTime - startTime) / 1000, ); } catch (e) { Logger.error(`${LOG_PREFIX}:`, e); @@ -276,14 +288,8 @@ function googleGenerativeAIStreamWrapper( typeof chunk?.text === "function" ? chunk.text() : chunk?.text; - if (typeof t === "string") { - if (t && !firstTokenRecorded) { - span.setAttribute( - "gen_ai.performance.time_to_first_token", - (Date.now() - startTime) / 1000, - ); - firstTokenRecorded = true; - } + if (typeof t === "string" && t) { + tokenTracker.markFirstToken(); } } catch { // ignore chunk parsing issues @@ -305,8 +311,10 @@ function googleGenerativeAIStreamWrapper( }, async return(value?: any) { const endTime = Date.now(); - const duration = (endTime - startTime) / 1000; - span.setAttribute("llm.response.duration", duration); + span.setAttribute( + "llm.response.duration", + (endTime - startTime) / 1000, + ); span.setStatus({ code: SpanStatusCode.OK }); span.end(); return iterator.return?.(value) ?? { value: undefined, done: true as const }; diff --git a/src/instrumentation/groq/wrappers.ts b/src/instrumentation/groq/wrappers.ts index 3743f32..adef38e 100644 --- a/src/instrumentation/groq/wrappers.ts +++ b/src/instrumentation/groq/wrappers.ts @@ -1,12 +1,16 @@ import { Tracer, Span, SpanKind, SpanStatusCode, context } from "@opentelemetry/api"; import { Logger } from "../../logger"; -import { setRequestAttributes, setResponseAttributes } from "./utils"; +import { + FirstTokenTracker, + recordNonStreamingTimingAttributes, +} from "../../utils/span-timing"; import { defineHidden, modelAsDict, isPromise, shouldSuppressInstrumentation, } from "../utils"; +import { setRequestAttributes, setResponseAttributes } from "./utils"; type GroqRequestType = "chat"; @@ -98,6 +102,7 @@ function groqWrapper( "llm.response.duration", (endTime - startTime) / 1000 ); + recordNonStreamingTimingAttributes(span, startTime, endTime); span.setStatus({ code: SpanStatusCode.OK }); span.end(); return value; @@ -121,6 +126,7 @@ function groqWrapper( "llm.response.duration", (endTime - startTime) / 1000 ); + recordNonStreamingTimingAttributes(span, startTime, endTime); span.setStatus({ code: SpanStatusCode.OK }); span.end(); return response; @@ -152,12 +158,14 @@ export class StreamingWrapper implements Iterable, Iterator { private response!: any; private startTime!: number; private requestKwargs!: Record; + private tokenTracker!: FirstTokenTracker; constructor(span: Span, response: any, startTime: number, requestKwargs: Record) { defineHidden(this, "span", span); defineHidden(this, "response", response); defineHidden(this, "startTime", startTime); defineHidden(this, "requestKwargs", requestKwargs); + defineHidden(this, "tokenTracker", new FirstTokenTracker(span, startTime)); } toJSON() { @@ -224,6 +232,7 @@ export class StreamingWrapper implements Iterable, Iterator { choices[index].message = { role: "assistant", content: "" }; } choices[index].message.content += String(delta.content); + this.tokenTracker.markFirstToken(); } if (choice.finish_reason) { @@ -235,16 +244,19 @@ export class StreamingWrapper implements Iterable, Iterator { if (chunkDict.usage) this.completeResponse.usage = chunkDict.usage; if (chunkDict.response?.status === "completed") { + let hasText = false; const outputs = chunkDict.response.output || []; outputs.forEach((output: any) => { const content = output.content || []; content.forEach((item: any) => { + if (item.text) hasText = true; choices.push({ message: { role: "assistant", content: item.text || "" }, }); }); }); this.completeResponse.usage = chunkDict.response.usage || {}; + if (hasText) this.tokenTracker.markFirstToken(); } this.span.addEvent("llm.content.completion.chunk"); @@ -281,12 +293,14 @@ export class AsyncStreamingWrapper private response!: any; private startTime!: number; private requestKwargs!: Record; + private tokenTracker!: FirstTokenTracker; constructor(span: Span, response: any, startTime: number, requestKwargs: Record) { defineHidden(this, "span", span); defineHidden(this, "response", response); defineHidden(this, "startTime", startTime); defineHidden(this, "requestKwargs", requestKwargs); + defineHidden(this, "tokenTracker", new FirstTokenTracker(span, startTime)); } toJSON() { @@ -367,6 +381,7 @@ export class AsyncStreamingWrapper } const message = choiceEntry.message as Record; message.content = String(message.content || "") + contentPiece; + this.tokenTracker.markFirstToken(); } if (choice.finish_reason) { @@ -379,6 +394,7 @@ export class AsyncStreamingWrapper // Response API if ((chunkDict.response as any)?.status === "completed") { + let hasText = false; const response = chunkDict.response as Record; const responseOutput = (response.output || []) as Array< Record @@ -388,7 +404,7 @@ export class AsyncStreamingWrapper if (content) { for (const contentItem of content) { const assistantText = contentItem.text || ""; - // Append to choices array instead of replacing + if (contentItem.text) hasText = true; ( this.completeResponse.choices as Array> ).push({ @@ -399,6 +415,7 @@ export class AsyncStreamingWrapper const usage = response.usage || {}; this.completeResponse.usage = usage; }); + if (hasText) this.tokenTracker.markFirstToken(); } this.span.addEvent("llm.content.completion.chunk"); } diff --git a/src/instrumentation/index.ts b/src/instrumentation/index.ts index 3142785..b560fce 100644 --- a/src/instrumentation/index.ts +++ b/src/instrumentation/index.ts @@ -125,7 +125,7 @@ function patchTraceloopLangchainCallbackHandler(): void { try { const mod = require("@traceloop/instrumentation-langchain"); - Logger.debug(`Loaded @traceloop/instrumentation-langchain via require from ${require.resolve("@traceloop/instrumentation-langchain")}`); + Logger.debug("Loaded @traceloop/instrumentation-langchain"); applyPatch(mod, "require"); return; } catch (e) { @@ -395,8 +395,6 @@ export function initInstrumentations( Logger.debug(` App Name: ${config.appName}`); Logger.debug(` OTLP Endpoint: ${config.otlpEndpoint || "(default - localhost:3002)"}`); Logger.debug(` API Key: ${config.apiKey ? "***" + config.apiKey.slice(-4) : "(not set)"}`); - Logger.debug(` Trace Content: ${config.traceContent}`); - Logger.debug(` Enable Scrubbing: ${config.enableScrubbing}`); // Initialize Traceloop SDK const traceloopOptions: InitializeOptions = { diff --git a/src/instrumentation/langgraph/index.ts b/src/instrumentation/langgraph/index.ts index 473053f..a662d49 100644 --- a/src/instrumentation/langgraph/index.ts +++ b/src/instrumentation/langgraph/index.ts @@ -32,48 +32,33 @@ function findModuleInCache(moduleName: string): any { } Logger.debug( `Module ${moduleName} not found in require.cache. Cache keys containing 'langgraph':`, - Object.keys(cache).filter(k => k.includes('langgraph')), + Object.keys(cache).filter((k) => k.includes('langgraph')), ); } return null; } async function resolveLanggraph(): Promise { + const moduleName = "@langchain/langgraph"; if (LanggraphClass) return LanggraphClass; try { // First, try to find the module in require.cache (already loaded by the app) // This ensures we patch the same module instance the app is using - let langgraphModule = findModuleInCache('@langchain/langgraph'); + let langgraphModule = findModuleInCache(moduleName); - if (langgraphModule) { - Logger.debug("Found @langchain/langgraph in require.cache (using app's module instance)"); - } else { + if (!langgraphModule) { // Fallback to dynamic import if not in cache - langgraphModule = await import("@langchain/langgraph"); - Logger.debug("Loaded @langchain/langgraph via dynamic import"); + langgraphModule = await import(moduleName); } - Logger.debug("LangGraph Module Exports:", Object.keys(langgraphModule)); - LanggraphClass = langgraphModule.CompiledStateGraph ?? langgraphModule.StateGraph; - Logger.debug("Resolved LanggraphClass:", !!LanggraphClass); - Logger.debug("LanggraphClass name:", LanggraphClass?.name); - Logger.debug( - "LanggraphClass.prototype keys:", - LanggraphClass?.prototype ? Object.getOwnPropertyNames(LanggraphClass.prototype) : "no prototype", - ); - Logger.debug("Has invoke on prototype:", !!LanggraphClass?.prototype?.invoke); - Logger.debug("Has stream on prototype:", !!LanggraphClass?.prototype?.stream); + LanggraphClass = + langgraphModule.CompiledStateGraph ?? langgraphModule.StateGraph; // Check prototype chain to find where invoke is defined - Logger.debug("Checking prototype chain for invoke location:"); let proto = LanggraphClass?.prototype; while (proto) { - const hasOwn = Object.getOwnPropertyNames(proto).includes('invoke'); - Logger.debug(` ${proto.constructor?.name}: hasOwnProperty('invoke')=${hasOwn}`); - if (hasOwn) { - Logger.debug(` -> invoke is defined on: ${proto.constructor?.name}`); - break; - } + const hasOwn = Object.getOwnPropertyNames(proto).includes("invoke"); + if (hasOwn) break; proto = Object.getPrototypeOf(proto); } @@ -160,8 +145,6 @@ export class NetraLanggraphInstrumentor { return; } - Logger.debug(`Found invoke on prototype: ${targetProto.constructor?.name}`); - const originalInvoke = targetProto.invoke; originalMethods.set("langgraph.graph.invoke", originalInvoke); // Store the target prototype for uninstrumentation @@ -188,9 +171,6 @@ export class NetraLanggraphInstrumentor { // Add marker to identify patched method (patchedInvoke as any).__netra_patched = true; targetProto.invoke = patchedInvoke; - - Logger.debug(`Successfully instrumented LangGraph invoke method on ${targetProto.constructor?.name}`); - Logger.debug(`Patched Pregel class identity:`, targetProto.constructor); } catch (error) { Logger.error(`Failed to instrument langgraph invoke: ${error}`); } @@ -211,8 +191,6 @@ export class NetraLanggraphInstrumentor { return; } - Logger.debug(`Found stream on prototype: ${targetProto.constructor?.name}`); - const originalStream = targetProto.stream; originalMethods.set("langgraph.graph.stream", originalStream); // Store the target prototype for uninstrumentation @@ -236,8 +214,6 @@ export class NetraLanggraphInstrumentor { ...rest, ); }; - - Logger.debug(`Successfully instrumented LangGraph stream method on ${targetProto.constructor?.name}`); } catch (error) { Logger.error(`Failed to instrument langgraph stream: ${error}`); } diff --git a/src/instrumentation/langgraph/wrappers.ts b/src/instrumentation/langgraph/wrappers.ts index f319dce..b434193 100644 --- a/src/instrumentation/langgraph/wrappers.ts +++ b/src/instrumentation/langgraph/wrappers.ts @@ -14,55 +14,12 @@ import { } from "@opentelemetry/api"; import { Logger } from "../../logger"; +import { recordNonStreamingTimingAttributes } from "../../utils/span-timing"; import { defineHidden, setResponseAttributes as setBaseResponseAttributes, shouldSuppressInstrumentation, } from "../utils"; - -// Context key to track if we're inside a LangGraph instrumented call -// This prevents double-instrumentation when invoke internally calls stream -const LANGGRAPH_INSTRUMENTATION_ACTIVE = createContextKey("netra.langgraph.active"); - -function getContextManager(): any { - try { - if ((context as any)._getContextManager) { - return (context as any)._getContextManager(); - } - - const globalSymbols = Object.getOwnPropertySymbols(global); - const otelSymbol = globalSymbols.find(s => - s.toString().includes("opentelemetry.js.api"), - ); - if (otelSymbol) { - const globalState = (global as any)[otelSymbol]; - if (globalState?.contextManager) { - return globalState.contextManager; - } - } - - return null; - } catch { - return null; - } -} - -function enterWithContext(newContext: Context): void { - const contextManager = getContextManager(); - if (!contextManager) return; - - if (typeof contextManager.enterWith === "function") { - contextManager.enterWith(newContext); - return; - } - - if ( - contextManager._asyncLocalStorage && - typeof contextManager._asyncLocalStorage.enterWith === "function" - ) { - contextManager._asyncLocalStorage.enterWith(newContext); - } -} import { NetraLanggraphAttributes, setChainInputAttributes, @@ -73,6 +30,10 @@ import { setToolAttributes, } from "./utils"; +// Context key to track if we're inside a LangGraph instrumented call +// This prevents double-instrumentation when invoke internally calls stream +const LANGGRAPH_INSTRUMENTATION_ACTIVE = createContextKey("netra.langgraph.active"); + type AnyFunc = (...args: any[]) => any; type AsyncIterableFunc = (...args: any[]) => Promise>; @@ -83,6 +44,7 @@ class NetraLanggraphCallbackHandler extends BaseCallbackHandler { private nodeAttributes: Map> = new Map(); private runStack: string[] = []; private inferredParents: Map = new Map(); + private streamedRuns: Set = new Set(); constructor( private tracer: Tracer, @@ -241,10 +203,19 @@ class NetraLanggraphCallbackHandler extends BaseCallbackHandler { metadata, prompts, extraParams, - parentRunId: effectiveParentRunId, // Store parent ID to link back + parentRunId: effectiveParentRunId, + startTimeMs: Date.now(), }); } + async handleLLMNewToken( + _token: string, + _idx: { prompt: number; completion: number }, + runId: string, + ) { + this.streamedRuns.add(runId); + } + async handleLLMEnd( output: LLMResult, runId: string, @@ -272,9 +243,15 @@ class NetraLanggraphCallbackHandler extends BaseCallbackHandler { attributes.extraParams, ); setBaseResponseAttributes(span, response); + // Streaming TTFT is captured by the underlying provider instrumentation + // (e.g. OpenAI, Anthropic). We only record timing for non-streaming calls. + if (attributes.startTimeMs && !this.streamedRuns.has(runId)) { + recordNonStreamingTimingAttributes(span, attributes.startTimeMs, Date.now()); + } span.end(); this.nodeAttributes.delete(runId); + this.streamedRuns.delete(runId); } async handleLLMError( @@ -302,6 +279,7 @@ class NetraLanggraphCallbackHandler extends BaseCallbackHandler { span.end(); this.nodeAttributes.delete(runId); + this.streamedRuns.delete(runId); } async handleToolStart( @@ -411,14 +389,19 @@ class LanggraphStreamingWrapper implements AsyncIterable { config?: RunnableConfig, ...rest: any[] ) { - this.iterable = await originalFunc.call(instance, input, config, ...rest); + const spanContext = trace.setSpan(this.rootContext, this.rootSpan); + this.iterable = await context.with(spanContext, () => + originalFunc.call(instance, input, config, ...rest), + ); return this; } async *[Symbol.asyncIterator]() { const spanContext = trace.setSpan(this.rootContext, this.rootSpan); try { - const iterator = await this.iterable[Symbol.asyncIterator](); + const iterator = await context.with(spanContext, () => + this.iterable[Symbol.asyncIterator](), + ); while (true) { let result: any; await context.with(spanContext, async () => { @@ -427,7 +410,7 @@ class LanggraphStreamingWrapper implements AsyncIterable { if (result.done) break; const value = result?.value ?? {}; this.output = { ...this.output, ...value }; - yield result; + yield value; } this.rootSpan.setAttribute( NetraLanggraphAttributes.entityOutput, @@ -502,7 +485,6 @@ export class LanggraphWrapper { // Set the active flag to prevent nested instrumentation (e.g., when invoke calls stream internally) const ctxWithSpan = trace.setSpan(context.active(), span); const ctxWithFlag = ctxWithSpan.setValue(LANGGRAPH_INSTRUMENTATION_ACTIVE, true); - enterWithContext(ctxWithFlag); return context.with(ctxWithFlag, async () => { { @@ -561,18 +543,20 @@ export class LanggraphWrapper { try { const ctxWithSpan = trace.setSpan(context.active(), span); const ctxWithFlag = ctxWithSpan.setValue(LANGGRAPH_INSTRUMENTATION_ACTIVE, true); - enterWithContext(ctxWithFlag); const streamingWrapper = new LanggraphStreamingWrapper(span, ctxWithFlag); - return streamingWrapper.startStream( - originalFunc, - instance, - input, - updatedConfig, - ...rest, + return await context.with(ctxWithFlag, () => + streamingWrapper.startStream( + originalFunc, + instance, + input, + updatedConfig, + ...rest, + ), ); } catch (error) { span.recordException(error as Error); span.end(); + throw error; } } } diff --git a/src/instrumentation/mistralai/wrappers.ts b/src/instrumentation/mistralai/wrappers.ts index c904c3d..6794104 100644 --- a/src/instrumentation/mistralai/wrappers.ts +++ b/src/instrumentation/mistralai/wrappers.ts @@ -10,6 +10,10 @@ import { context, } from "@opentelemetry/api"; import { Logger } from "../../logger"; +import { + FirstTokenTracker, + recordNonStreamingTimingAttributes, +} from "../../utils/span-timing"; import { defineHidden, isPromise } from "../utils"; import { modelAsDict, @@ -67,6 +71,9 @@ function mistralWrapper( "llm.response.duration", (endTime - startTime) / 1000 ); + if (requestType !== "embedding") { + recordNonStreamingTimingAttributes(span, startTime, endTime); + } span.setStatus({ code: SpanStatusCode.OK }); span.end(); return value; @@ -90,6 +97,9 @@ function mistralWrapper( "llm.response.duration", (endTime - startTime) / 1000 ); + if (requestType !== "embedding") { + recordNonStreamingTimingAttributes(span, startTime, endTime); + } span.setStatus({ code: SpanStatusCode.OK }); span.end(); return response; @@ -245,12 +255,14 @@ export class StreamingWrapper implements Iterable, Iterator { private response!: unknown; private startTime!: number; private requestKwargs!: Record; + private tokenTracker!: FirstTokenTracker; constructor(span: Span, response: unknown, startTime: number, requestKwargs: Record) { defineHidden(this, "span", span); defineHidden(this, "response", response); defineHidden(this, "startTime", startTime); defineHidden(this, "requestKwargs", requestKwargs); + defineHidden(this, "tokenTracker", new FirstTokenTracker(span, startTime)); } toJSON() { @@ -354,6 +366,7 @@ export class StreamingWrapper implements Iterable, Iterator { } else { choiceEntry.text = String(choiceEntry.text || "") + contentPiece; } + this.tokenTracker.markFirstToken(); } if (choice.finishReason) { @@ -389,6 +402,7 @@ export class StreamingWrapper implements Iterable, Iterator { } else { choiceEntry.text = String(choiceEntry.text || "") + contentPiece; } + this.tokenTracker.markFirstToken(); } if (choice.finishReason) { @@ -439,17 +453,19 @@ export class AsyncStreamingWrapper private startTime!: number; private requestKwargs!: Record; private completeResponse: Record; + private tokenTracker!: FirstTokenTracker; constructor( span: Span, response: unknown, startTime: number, - requestKwargs: Record + requestKwargs: Record, ) { defineHidden(this, "span", span); defineHidden(this, "response", response); defineHidden(this, "startTime", startTime); defineHidden(this, "requestKwargs", requestKwargs); + defineHidden(this, "tokenTracker", new FirstTokenTracker(span, startTime)); this.completeResponse = { choices: [], model: "" }; } @@ -572,6 +588,7 @@ export class AsyncStreamingWrapper } else { choiceEntry.text = String(choiceEntry.text || "") + contentPiece; } + this.tokenTracker.markFirstToken(); } if (choice.finishReason) { @@ -607,6 +624,7 @@ export class AsyncStreamingWrapper } else { choiceEntry.text = String(choiceEntry.text || "") + contentPiece; } + this.tokenTracker.markFirstToken(); } if (choice.finishReason) { diff --git a/src/instrumentation/openai/wrappers.ts b/src/instrumentation/openai/wrappers.ts index 5a9a455..ee3bfa2 100644 --- a/src/instrumentation/openai/wrappers.ts +++ b/src/instrumentation/openai/wrappers.ts @@ -7,6 +7,10 @@ import { context, trace, } from "@opentelemetry/api"; +import { + FirstTokenTracker, + recordNonStreamingTimingAttributes, +} from "../../utils/span-timing"; import { defineHidden, isPromise, @@ -15,6 +19,7 @@ import { } from "../utils"; import { setRequestAttributes, setResponseAttributes } from "./utils"; import { OpenAIRequestType, StreamResponse, WrapperFn } from "./types"; +import { Logger } from "../../logger"; const SPAN_NAMES: Record = { chat: "openai.chat", @@ -45,9 +50,14 @@ function finalizeSpanSuccess( span: Span, response: Record, startTime: number, + requestType: OpenAIRequestType, ): void { + const endTime = Date.now(); setResponseAttributes(span, response); - span.setAttribute("llm.response.duration", (Date.now() - startTime) / 1000); + span.setAttribute("llm.response.duration", (endTime - startTime) / 1000); + if (requestType !== "embedding") { + recordNonStreamingTimingAttributes(span, startTime, endTime); + } span.setStatus({ code: SpanStatusCode.OK }); span.end(); } @@ -58,6 +68,7 @@ abstract class BaseStreamHandler { protected span!: Span; protected startTime!: number; protected requestKwargs!: Record; + protected tokenTracker!: FirstTokenTracker; constructor( span: Span, @@ -67,6 +78,7 @@ abstract class BaseStreamHandler { defineHidden(this, "span", span); defineHidden(this, "startTime", startTime); defineHidden(this, "requestKwargs", requestKwargs); + defineHidden(this, "tokenTracker", new FirstTokenTracker(span, startTime)); } toJSON(): StreamResponse { @@ -93,6 +105,7 @@ abstract class BaseStreamHandler { } const msg = entry.message as Record; msg.content = String(msg.content ?? "") + String(delta.content); + this.tokenTracker.markFirstToken(); } if (choice.finish_reason) { this.completeResponse.choices[index].finish_reason = @@ -110,6 +123,7 @@ abstract class BaseStreamHandler { | Record | undefined; if (responseChunk?.status === "completed") { + let hasText = false; const outputs = (responseChunk.output ?? []) as Array< Record >; @@ -119,6 +133,7 @@ abstract class BaseStreamHandler { | undefined; if (Array.isArray(content)) { for (const item of content) { + if (item.text) hasText = true; this.completeResponse.choices.push({ message: { role: "assistant", content: String(item.text ?? "") }, }); @@ -126,6 +141,7 @@ abstract class BaseStreamHandler { } } this.completeResponse.usage = responseChunk.usage ?? {}; + if (hasText) this.tokenTracker.markFirstToken(); } this.span.addEvent("llm.content.completion.chunk"); @@ -141,20 +157,20 @@ abstract class BaseStreamHandler { } protected finalizeSpan(code: SpanStatusCode): void { - if (code === SpanStatusCode.OK) { - finalizeSpanSuccess( + try { + setResponseAttributes( this.span, this.completeResponse as Record, - this.startTime, - ); - } else { - this.span.setAttribute( - "llm.response.duration", - (Date.now() - this.startTime) / 1000, ); - this.span.setStatus({ code }); - this.span.end(); + } catch { + Logger.debug('Failed to set response attributes'); } + this.span.setAttribute( + "llm.response.duration", + (Date.now() - this.startTime) / 1000, + ); + this.span.setStatus({ code }); + this.span.end(); } } @@ -317,7 +333,7 @@ function executeNonStreaming( if (isPromise(result)) { return result.then( (value) => { - finalizeSpanSuccess(span, modelAsDict(value), startTime); + finalizeSpanSuccess(span, modelAsDict(value), startTime, requestType); return value; }, (error) => { @@ -327,7 +343,7 @@ function executeNonStreaming( ); } - finalizeSpanSuccess(span, modelAsDict(result), startTime); + finalizeSpanSuccess(span, modelAsDict(result), startTime, requestType); return result; } catch (error) { handleSpanError(span, error); diff --git a/src/instrumentation/span-attributes.ts b/src/instrumentation/span-attributes.ts index 578810d..5500cf8 100644 --- a/src/instrumentation/span-attributes.ts +++ b/src/instrumentation/span-attributes.ts @@ -33,4 +33,10 @@ export const SpanAttributes = { LLM_IS_STREAMING: "llm.is_streaming", LLM_COMPLETIONS: "gen_ai.completion", LLM_PROMPTS: "gen_ai.prompt", + + LLM_TIME_TO_FIRST_TOKEN: "gen_ai.performance.time_to_first_token", + LLM_RELATIVE_TIME_TO_FIRST_TOKEN: + "gen_ai.performance.relative_time_to_first_token", + LLM_TIME_TO_FIRST_TOKEN_TIMESTAMP: + "gen_ai.performance.time_to_first_token.timestamp", } as const; diff --git a/src/utils/span-timing.ts b/src/utils/span-timing.ts new file mode 100644 index 0000000..49eeda9 --- /dev/null +++ b/src/utils/span-timing.ts @@ -0,0 +1,81 @@ +import { Span } from "@opentelemetry/api"; +import type { ReadableSpan } from "@opentelemetry/sdk-trace-base"; +import { RootSpanProcessor } from "../processors/root-span-processor"; +import { SpanAttributes } from "../instrumentation/span-attributes"; +import { Logger } from "../logger"; + +function hrTimeToMs(hrTime: [number, number]): number { + return hrTime[0] * 1000 + hrTime[1] / 1e6; +} + +function recordTimeToFirstToken( + span: Span, + nowMs: number, + startTimeMs: number, +): void { + span.setAttribute( + SpanAttributes.LLM_TIME_TO_FIRST_TOKEN, + (nowMs - startTimeMs) / 1000, + ); + span.setAttribute( + SpanAttributes.LLM_TIME_TO_FIRST_TOKEN_TIMESTAMP, + new Date(nowMs).toISOString(), + ); +} + +function recordRelativeTimeToFirstToken(span: Span, nowMs: number): void { + try { + const rootSpan = RootSpanProcessor.getRootSpan(span); + if (!rootSpan) return; + const hrStart = (rootSpan as unknown as ReadableSpan).startTime; + if (!hrStart) return; + const rootStartMs = hrTimeToMs(hrStart); + span.setAttribute( + SpanAttributes.LLM_RELATIVE_TIME_TO_FIRST_TOKEN, + (nowMs - rootStartMs) / 1000, + ); + } catch (e) { + Logger.warn("span-timing: failed to compute RTTFT", e); + } +} + +/** + * Tracks the first content token in a streaming LLM response and records + * TTFT, RTTFT, and the absolute first-token timestamp on the span. + * + * `markFirstToken()` is idempotent — only the first call writes attributes. + */ +export class FirstTokenTracker { + private _recorded = false; + + constructor( + private readonly span: Span, + private readonly startTimeMs: number, + ) {} + + markFirstToken(): void { + if (this._recorded || !this.span.isRecording()) return; + this._recorded = true; + + const now = Date.now(); + + recordTimeToFirstToken(this.span, now, this.startTimeMs); + recordRelativeTimeToFirstToken(this.span, now); + } +} + +/** + * Record TTFT + RTTFT for non-streaming LLM calls. + * For non-streaming, "first token" = full response arrival, + * so TTFT equals response duration. + */ +export function recordNonStreamingTimingAttributes( + span: Span, + startTimeMs: number, + endTimeMs: number, +): void { + if (!span.isRecording()) return; + + recordTimeToFirstToken(span, endTimeMs, startTimeMs); + recordRelativeTimeToFirstToken(span, endTimeMs); +} diff --git a/src/version.ts b/src/version.ts index f27afec..9ae0113 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1 +1 @@ -export const SDK_VERSION = "1.8.0"; +export const SDK_VERSION = "1.9.0-dev.0";