diff --git a/.changeset/ts-gen-ai-openai.md b/.changeset/ts-gen-ai-openai.md new file mode 100644 index 0000000..07b187c --- /dev/null +++ b/.changeset/ts-gen-ai-openai.md @@ -0,0 +1,11 @@ +--- +'@smooai/observability': minor +--- + +Instrument the OpenAI Node SDK, and stop leaking prompt content into GenAI span events. + +`wrapOpenAI(client, options)` returns a proxy of an OpenAI client whose `chat.completions.create` emits OTel GenAI semantic-convention spans — request model / sampling params / tool names on the way out, response model, id, finish reason, and token usage on the way back. Streaming is handled: the span stays open until the stream drains (or the consumer breaks out early), and picks up the usage chunk emitted under `stream_options.include_usage`. The client is duck-typed, so this adds no dependency on `openai` and works against Groq, Together, Fireworks, DeepSeek, Azure OpenAI, and any OpenAI-compatible gateway via `{ system: 'groq' }`. + +Cost has a seam now: nothing in the platform computes an LLM price on its own, which is why the dashboard's cost column is empty. Pass `costUsd({ requestModel, responseModel, inputTokens, outputTokens, cachedTokens })` and it lands on `gen_ai.usage.cost_usd`. + +`recordGenAIMessage` now routes content through the SDK's PII scrub before it leaves the process. Prompts are the most PII-dense payload this SDK can touch, and `wrapOpenAI` keeps content recording **off** by default (`{ recordContent: true }` opts in). diff --git a/README.md b/README.md index 2d7b789..1de2021 100644 --- a/README.md +++ b/README.md @@ -158,6 +158,46 @@ The same ingest contract (`POST /webhooks/observability/{org_id}/{token}` with ` - 🐹 **Go** — `github.com/smooai/observability-go` (tracked in SMOODEV-1067 follow-ups) - 💠 **.NET** — `SmooAI.Observability` on NuGet (tracked in SMOODEV-1067 follow-ups) +### 🤖 GenAI telemetry (`gen_ai.*`) + +LLM and agent spans carry the [OTel GenAI semantic conventions](https://opentelemetry.io/docs/specs/semconv/gen-ai/), so any semconv-aware backend reads them — Smoo's LLM dashboard routes on `gen_ai.system` alone. + +```ts +import OpenAI from 'openai'; +import { wrapOpenAI } from '@smooai/observability'; + +// Instruments chat.completions.create — the original client is untouched. +const openai = wrapOpenAI(new OpenAI(), { + conversationId: conversation.id, + // Providers don't return a price. Supply one and the cost column fills in. + costUsd: ({ inputTokens = 0, outputTokens = 0 }) => inputTokens * 2.5e-6 + outputTokens * 1e-5, +}); +``` + +The same wrapper covers Groq, Together, Fireworks, DeepSeek, Azure OpenAI, and any OpenAI-compatible gateway — pass `{ system: 'groq' }` so spans attribute to the real provider. Prompt and completion **content is off by default**; `{ recordContent: true }` records it as `gen_ai.*.message` span events, PII-scrubbed on the way out. + +For hand-rolled calls, set the attributes directly: + +```ts +import { setGenAIAttributes, recordGenAIMessage } from '@smooai/observability'; + +setGenAIAttributes(span, { system: 'anthropic', operationName: 'chat', requestModel: 'claude-opus-4-7', usageInputTokens: 812, usageOutputTokens: 96 }); +``` + +> `gen_ai.operation.name` is a straight passthrough on ingest with **no fallback** — leave it unset and the operation column lands `NULL`. Always set it. + +**Parity across the five SDKs:** + +| SDK | Attribute helper | Message events | Content PII-scrubbed | Framework integration | +| -------------- | ----------------------------- | ----------------------------- | -------------------- | --------------------------------------------------- | +| **TypeScript** | `setGenAIAttributes` | `recordGenAIMessage` | ✅ | ✅ `wrapOpenAI` — OpenAI Node SDK + compatible APIs | +| **Rust** | `set_gen_ai_attributes` | `record_gen_ai_message` | ❌ | — | +| **Python** | `set_gen_ai_attributes` | `record_gen_ai_message` | ❌ | ✅ `SmooAICallbackHandler` — LangChain / LangGraph | +| **Go** | `SetGenAIAttributes` | `RecordGenAIMessage` | ❌ | — | +| **.NET** | `GenAIActivity.SetAttributes` | `GenAIActivity.RecordMessage` | ❌ | — | + +Known divergences: TypeScript, Python, Go and .NET emit `gen_ai.tool.names` as a **string array**; Rust emits a comma-joined string. Only TypeScript scrubs recorded message content today. + ## 📖 Architecture The SDK is intentionally thin. It captures, batches, redacts PII, and POSTs to a Smoo ingest endpoint. All of the heavy lifting — fingerprint grouping, source-map symbolication, dashboards, alerts, retention — lives in the Smoo platform. diff --git a/packages/core/README.md b/packages/core/README.md index ee67256..25c6687 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -103,6 +103,30 @@ deterministic, stable for a page's lifetime, and reproduced byte-identically by the Rust / Python / Go / .NET SDKs against [`parity/sampling-corpus.json`](../../parity/README.md). +## GenAI spans + +```ts +import OpenAI from 'openai'; +import { wrapOpenAI, setGenAIAttributes } from '@smooai/observability'; + +const openai = wrapOpenAI(new OpenAI(), { conversationId: convo.id }); +await openai.chat.completions.create({ model: 'gpt-4o', messages }); +``` + +`wrapOpenAI` proxies `chat.completions.create` (streaming included — the span +stays open until the stream drains) and emits the OTel +[GenAI semconv](https://opentelemetry.io/docs/specs/semconv/gen-ai/) attributes. +It needs no dependency on `openai`; the client is duck-typed, so the same +wrapper covers Groq / DeepSeek / Azure / any OpenAI-compatible gateway via +`{ system: 'groq' }`. + +- **Cost**: nothing computes a price on its own. Pass `costUsd(...)` to fill + `gen_ai.usage.cost_usd`. +- **Content**: prompts and completions are **not** recorded unless you pass + `{ recordContent: true }`, and are PII-scrubbed when you do. +- **Hand-rolled calls**: `setGenAIAttributes(span, attrs)` / + `recordGenAIMessage(span, role, content)`. + ## What it does NOT do - Does not capture `console.log` / `console.info` / `console.warn` diff --git a/packages/core/src/__tests__/gen-ai-attributes.test.ts b/packages/core/src/__tests__/gen-ai-attributes.test.ts new file mode 100644 index 0000000..059dbb5 --- /dev/null +++ b/packages/core/src/__tests__/gen-ai-attributes.test.ts @@ -0,0 +1,200 @@ +import { trace } from '@opentelemetry/api'; +import { BasicTracerProvider, InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base'; +import { beforeEach, describe, expect, it } from 'vitest'; +import { recordGenAIMessage, setGenAIAttributes, type GenAIAttributes } from '../gen-ai-attributes'; + +const exporter = new InMemorySpanExporter(); +trace.setGlobalTracerProvider(new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] })); +const tracer = trace.getTracer('test'); + +/** Run `fn` against a fresh span and hand back the exported span. */ +function onSpan(fn: (span: ReturnType) => void) { + const span = tracer.startSpan('llm'); + fn(span); + span.end(); + const finished = exporter.getFinishedSpans(); + return finished[finished.length - 1]!; +} + +/** + * Every attribute `setGenAIAttributes` can emit, and the key it MUST emit it + * under. A typo in any of these produces a span the platform cannot route, so + * the keys are asserted literally rather than derived from the source. + */ +const EXPECTED_KEYS: Record = { + system: 'gen_ai.system', + operationName: 'gen_ai.operation.name', + requestModel: 'gen_ai.request.model', + responseModel: 'gen_ai.response.model', + responseId: 'gen_ai.response.id', + temperature: 'gen_ai.request.temperature', + topP: 'gen_ai.request.top_p', + topK: 'gen_ai.request.top_k', + maxTokens: 'gen_ai.request.max_tokens', + seed: 'gen_ai.request.seed', + usageInputTokens: 'gen_ai.usage.input_tokens', + usageOutputTokens: 'gen_ai.usage.output_tokens', + usageCachedTokens: 'gen_ai.usage.cached_tokens', + usageCostUsd: 'gen_ai.usage.cost_usd', + toolNames: 'gen_ai.tool.names', + truncated: 'gen_ai.response.truncated', + finishReason: 'gen_ai.response.finish_reason', + endUserId: 'gen_ai.end_user.id', + conversationId: 'gen_ai.conversation.id', +}; + +const FULL_ATTRS: Required = { + system: 'anthropic', + operationName: 'chat', + requestModel: 'claude-opus-4-7', + responseModel: 'claude-opus-4-7-20260101', + responseId: 'msg_123', + temperature: 0.7, + topP: 0.9, + topK: 40, + maxTokens: 1024, + seed: 42, + usageInputTokens: 100, + usageOutputTokens: 250, + usageCachedTokens: 80, + usageCostUsd: 0.0123, + toolNames: ['search', 'calendar'], + truncated: false, + finishReason: 'stop', + endUserId: 'user_abc', + conversationId: 'conv_xyz', +}; + +describe('setGenAIAttributes', () => { + beforeEach(() => exporter.reset()); + + it('emits every attribute under its exact semantic-convention key', () => { + const span = onSpan((s) => setGenAIAttributes(s, FULL_ATTRS)); + + expect(span.attributes['gen_ai.system']).toBe('anthropic'); + expect(span.attributes['gen_ai.operation.name']).toBe('chat'); + expect(span.attributes['gen_ai.request.model']).toBe('claude-opus-4-7'); + expect(span.attributes['gen_ai.response.model']).toBe('claude-opus-4-7-20260101'); + expect(span.attributes['gen_ai.response.id']).toBe('msg_123'); + expect(span.attributes['gen_ai.request.temperature']).toBe(0.7); + expect(span.attributes['gen_ai.request.top_p']).toBe(0.9); + expect(span.attributes['gen_ai.request.top_k']).toBe(40); + expect(span.attributes['gen_ai.request.max_tokens']).toBe(1024); + expect(span.attributes['gen_ai.request.seed']).toBe(42); + expect(span.attributes['gen_ai.usage.input_tokens']).toBe(100); + expect(span.attributes['gen_ai.usage.output_tokens']).toBe(250); + expect(span.attributes['gen_ai.usage.cached_tokens']).toBe(80); + expect(span.attributes['gen_ai.usage.cost_usd']).toBe(0.0123); + expect(span.attributes['gen_ai.tool.names']).toEqual(['search', 'calendar']); + expect(span.attributes['gen_ai.response.truncated']).toBe(false); + expect(span.attributes['gen_ai.response.finish_reason']).toBe('stop'); + expect(span.attributes['gen_ai.end_user.id']).toBe('user_abc'); + expect(span.attributes['gen_ai.conversation.id']).toBe('conv_xyz'); + }); + + it('emits nothing outside the gen_ai.* vocabulary, and covers every field', () => { + const span = onSpan((s) => setGenAIAttributes(s, FULL_ATTRS)); + expect(new Set(Object.keys(span.attributes))).toEqual(new Set(Object.values(EXPECTED_KEYS))); + }); + + it('skips undefined fields so partial calls are additive', () => { + const span = onSpan((s) => { + setGenAIAttributes(s, { system: 'openai', operationName: 'chat', requestModel: 'gpt-4o' }); + setGenAIAttributes(s, { usageInputTokens: 5, usageOutputTokens: 9 }); + }); + expect(Object.keys(span.attributes).sort()).toEqual([ + 'gen_ai.operation.name', + 'gen_ai.request.model', + 'gen_ai.system', + 'gen_ai.usage.input_tokens', + 'gen_ai.usage.output_tokens', + ]); + }); + + it('omits gen_ai.tool.names for an empty tool list', () => { + const span = onSpan((s) => setGenAIAttributes(s, { toolNames: [] })); + expect(span.attributes['gen_ai.tool.names']).toBeUndefined(); + }); + + it('accepts an arbitrary system string for providers outside the known set', () => { + const span = onSpan((s) => setGenAIAttributes(s, { system: 'my-private-gateway' })); + expect(span.attributes['gen_ai.system']).toBe('my-private-gateway'); + }); +}); + +/** + * Routing contract with `rust/api-prime/src/handlers/observability/ingest_traces.rs`. + * That handler forwards a span to `gen_ai_events` iff it carries `gen_ai.system`, + * then reads exactly these columns off the merged attribute map. Every one is a + * straight passthrough — `gen_ai.operation.name` in particular has NO fallback, + * so a caller that leaves it unset writes a NULL operation column. + */ +const INGEST_READS = [ + 'gen_ai.system', + 'gen_ai.operation.name', + 'gen_ai.request.model', + 'gen_ai.request.temperature', + 'gen_ai.request.top_p', + 'gen_ai.request.max_tokens', + 'gen_ai.response.model', + 'gen_ai.response.id', + 'gen_ai.response.finish_reason', + 'gen_ai.usage.input_tokens', + 'gen_ai.usage.output_tokens', + 'gen_ai.usage.cached_tokens', + 'gen_ai.usage.cost_usd', + 'gen_ai.end_user.id', + 'gen_ai.conversation.id', +] as const; + +describe('ingest routing contract', () => { + beforeEach(() => exporter.reset()); + + it('produces every column ingest_traces.rs reads', () => { + const span = onSpan((s) => setGenAIAttributes(s, FULL_ATTRS)); + for (const key of INGEST_READS) { + expect(span.attributes[key], `ingest reads ${key} but the SDK never emitted it`).toBeDefined(); + } + }); + + it('carries the gen_ai.system trigger that routes the span at all', () => { + // Without this exact key the span lands in ClickHouse traces only and + // never reaches gen_ai_events — the whole LLM dashboard goes dark. + const span = onSpan((s) => setGenAIAttributes(s, { system: 'openai' })); + expect(Object.keys(span.attributes)).toContain('gen_ai.system'); + }); +}); + +describe('recordGenAIMessage', () => { + beforeEach(() => exporter.reset()); + + it('names the event gen_ai.{role}.message and keys content exactly', () => { + const span = onSpan((s) => recordGenAIMessage(s, 'assistant', 'hello there')); + expect(span.events[0]!.name).toBe('gen_ai.assistant.message'); + expect(span.events[0]!.attributes!['gen_ai.message.content']).toBe('hello there'); + }); + + it.each(['user', 'assistant', 'system', 'tool'] as const)('supports the %s role', (role) => { + const span = onSpan((s) => recordGenAIMessage(s, role, 'x')); + expect(span.events[0]!.name).toBe(`gen_ai.${role}.message`); + }); + + it('attaches tool linkage under the spec keys', () => { + const span = onSpan((s) => recordGenAIMessage(s, 'tool', 'result', { toolCallId: 'call_1', toolName: 'search' })); + expect(span.events[0]!.attributes!['gen_ai.tool_call.id']).toBe('call_1'); + expect(span.events[0]!.attributes!['gen_ai.tool.name']).toBe('search'); + }); + + it('omits tool linkage keys when not supplied', () => { + const span = onSpan((s) => recordGenAIMessage(s, 'user', 'hi')); + expect(Object.keys(span.events[0]!.attributes!)).toEqual(['gen_ai.message.content']); + }); + + it('PII-scrubs message content before it leaves the process', () => { + const span = onSpan((s) => recordGenAIMessage(s, 'user', 'call the api with Bearer abc123token and password=hunter2')); + const content = span.events[0]!.attributes!['gen_ai.message.content'] as string; + expect(content).not.toContain('abc123token'); + expect(content).not.toContain('hunter2'); + expect(content).toContain('[redacted]'); + }); +}); diff --git a/packages/core/src/__tests__/gen-ai-openai.test.ts b/packages/core/src/__tests__/gen-ai-openai.test.ts new file mode 100644 index 0000000..1555887 --- /dev/null +++ b/packages/core/src/__tests__/gen-ai-openai.test.ts @@ -0,0 +1,254 @@ +import { context, SpanStatusCode, trace } from '@opentelemetry/api'; +import { AsyncHooksContextManager } from '@opentelemetry/context-async-hooks'; +import { BasicTracerProvider, InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { wrapOpenAI } from '../gen-ai-openai'; + +const exporter = new InMemorySpanExporter(); +trace.setGlobalTracerProvider(new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] })); +const cm = new AsyncHooksContextManager(); +cm.enable(); +context.setGlobalContextManager(cm); + +/** Stand-in for the OpenAI Node client — same `chat.completions.create` shape. */ +function fakeClient(create: (body: unknown) => unknown) { + return { chat: { completions: { create: vi.fn(create) } }, apiKey: 'sk-not-real', baseURL: 'https://api.openai.com/v1' }; +} + +const RESPONSE = { + id: 'chatcmpl-abc', + model: 'gpt-4o-2024-11-20', + choices: [{ finish_reason: 'stop', message: { role: 'assistant', content: 'four' } }], + usage: { prompt_tokens: 11, completion_tokens: 3, prompt_tokens_details: { cached_tokens: 8 } }, +}; + +const REQUEST = { + model: 'gpt-4o', + messages: [{ role: 'user', content: 'what is 2+2' }], + temperature: 0.2, + top_p: 0.95, + max_tokens: 64, + seed: 7, + user: 'user_from_body', + tools: [{ type: 'function', function: { name: 'calculator' } }], +}; + +function lastSpan() { + const spans = exporter.getFinishedSpans(); + return spans[spans.length - 1]!; +} + +describe('wrapOpenAI — non-streaming', () => { + beforeEach(() => exporter.reset()); + + it('emits the request-side keys off the call body', async () => { + const client = wrapOpenAI(fakeClient(() => RESPONSE)); + await client.chat.completions.create(REQUEST); + + const { attributes } = lastSpan(); + expect(attributes['gen_ai.system']).toBe('openai'); + expect(attributes['gen_ai.request.model']).toBe('gpt-4o'); + expect(attributes['gen_ai.request.temperature']).toBe(0.2); + expect(attributes['gen_ai.request.top_p']).toBe(0.95); + expect(attributes['gen_ai.request.max_tokens']).toBe(64); + expect(attributes['gen_ai.request.seed']).toBe(7); + expect(attributes['gen_ai.tool.names']).toEqual(['calculator']); + expect(attributes['gen_ai.end_user.id']).toBe('user_from_body'); + }); + + it('always sets gen_ai.operation.name — ingest has no fallback for it', async () => { + const client = wrapOpenAI(fakeClient(() => RESPONSE)); + await client.chat.completions.create({ model: 'gpt-4o', messages: [] }); + expect(lastSpan().attributes['gen_ai.operation.name']).toBe('chat'); + }); + + it('emits the response-side keys off the provider response', async () => { + const client = wrapOpenAI(fakeClient(() => RESPONSE)); + await client.chat.completions.create(REQUEST); + + const { attributes } = lastSpan(); + expect(attributes['gen_ai.response.model']).toBe('gpt-4o-2024-11-20'); + expect(attributes['gen_ai.response.id']).toBe('chatcmpl-abc'); + expect(attributes['gen_ai.response.finish_reason']).toBe('stop'); + expect(attributes['gen_ai.usage.input_tokens']).toBe(11); + expect(attributes['gen_ai.usage.output_tokens']).toBe(3); + expect(attributes['gen_ai.usage.cached_tokens']).toBe(8); + expect(attributes['gen_ai.response.truncated']).toBe(false); + }); + + it('marks truncated when the provider stopped on length', async () => { + const client = wrapOpenAI(fakeClient(() => ({ ...RESPONSE, choices: [{ finish_reason: 'length' }] }))); + await client.chat.completions.create(REQUEST); + expect(lastSpan().attributes['gen_ai.response.truncated']).toBe(true); + }); + + it('names the span "chat {model}" per semconv', async () => { + const client = wrapOpenAI(fakeClient(() => RESPONSE)); + await client.chat.completions.create(REQUEST); + expect(lastSpan().name).toBe('chat gpt-4o'); + }); + + it('returns the provider response untouched', async () => { + const client = wrapOpenAI(fakeClient(() => RESPONSE)); + await expect(client.chat.completions.create(REQUEST)).resolves.toBe(RESPONSE); + }); + + it('passes every other client property straight through', () => { + const client = wrapOpenAI(fakeClient(() => RESPONSE)); + expect(client.baseURL).toBe('https://api.openai.com/v1'); + }); + + it('overrides the system for an OpenAI-compatible gateway', async () => { + const client = wrapOpenAI( + fakeClient(() => RESPONSE), + { system: 'groq' }, + ); + await client.chat.completions.create(REQUEST); + expect(lastSpan().attributes['gen_ai.system']).toBe('groq'); + }); + + it('stamps the conversation and end-user ids from options', async () => { + const client = wrapOpenAI( + fakeClient(() => RESPONSE), + { conversationId: 'conv_1', endUserId: 'user_1' }, + ); + await client.chat.completions.create(REQUEST); + expect(lastSpan().attributes['gen_ai.conversation.id']).toBe('conv_1'); + expect(lastSpan().attributes['gen_ai.end_user.id']).toBe('user_1'); + }); +}); + +describe('wrapOpenAI — cost seam', () => { + beforeEach(() => exporter.reset()); + + it('emits gen_ai.usage.cost_usd from the caller-supplied pricer', async () => { + const costUsd = vi.fn(({ inputTokens, outputTokens }) => (inputTokens ?? 0) * 0.001 + (outputTokens ?? 0) * 0.002); + const client = wrapOpenAI( + fakeClient(() => RESPONSE), + { costUsd }, + ); + await client.chat.completions.create(REQUEST); + + expect(costUsd).toHaveBeenCalledWith({ requestModel: 'gpt-4o', responseModel: 'gpt-4o-2024-11-20', inputTokens: 11, outputTokens: 3, cachedTokens: 8 }); + expect(lastSpan().attributes['gen_ai.usage.cost_usd']).toBeCloseTo(0.017); + }); + + it('leaves the cost attribute unset when no pricer is supplied', async () => { + const client = wrapOpenAI(fakeClient(() => RESPONSE)); + await client.chat.completions.create(REQUEST); + expect(lastSpan().attributes['gen_ai.usage.cost_usd']).toBeUndefined(); + }); +}); + +describe('wrapOpenAI — content recording', () => { + beforeEach(() => exporter.reset()); + + it('records no prompt or completion content by default', async () => { + const client = wrapOpenAI(fakeClient(() => RESPONSE)); + await client.chat.completions.create(REQUEST); + expect(lastSpan().events).toHaveLength(0); + }); + + it('records prompt and completion as gen_ai.*.message events when opted in', async () => { + const client = wrapOpenAI( + fakeClient(() => RESPONSE), + { recordContent: true }, + ); + await client.chat.completions.create(REQUEST); + + const events = lastSpan().events; + expect(events.map((e) => e.name)).toEqual(['gen_ai.user.message', 'gen_ai.assistant.message']); + expect(events[1]!.attributes!['gen_ai.message.content']).toBe('four'); + }); + + it('scrubs credentials out of recorded prompt content', async () => { + const client = wrapOpenAI( + fakeClient(() => RESPONSE), + { recordContent: true }, + ); + await client.chat.completions.create({ model: 'gpt-4o', messages: [{ role: 'user', content: 'my key is sk-abcdefghijklmnopqrstuvwxyz' }] }); + expect(lastSpan().events[0]!.attributes!['gen_ai.message.content']).not.toContain('abcdefghijklmnopqrstuvwxyz'); + }); +}); + +describe('wrapOpenAI — errors', () => { + beforeEach(() => exporter.reset()); + + it('marks the span ERROR and rethrows the provider failure', async () => { + const client = wrapOpenAI( + fakeClient(() => { + throw new Error('429 rate limited'); + }), + ); + await expect(client.chat.completions.create(REQUEST)).rejects.toThrow('429 rate limited'); + + const span = lastSpan(); + expect(span.status.code).toBe(SpanStatusCode.ERROR); + expect(span.events.map((e) => e.name)).toContain('exception'); + // Request attributes still land, so a failed call is still attributable. + expect(span.attributes['gen_ai.request.model']).toBe('gpt-4o'); + }); +}); + +/** Minimal stand-in for the OpenAI `Stream` object: async-iterable, plus extras. */ +function fakeStream(chunks: unknown[]) { + return { + controller: { abort: () => {} }, + async *[Symbol.asyncIterator]() { + for (const c of chunks) yield c; + }, + }; +} + +const STREAM_CHUNKS = [ + { id: 'chatcmpl-s', model: 'gpt-4o-2024-11-20', choices: [{ delta: { content: 'fo' } }] }, + { id: 'chatcmpl-s', model: 'gpt-4o-2024-11-20', choices: [{ delta: { content: 'ur' }, finish_reason: 'stop' }] }, + { id: 'chatcmpl-s', model: 'gpt-4o-2024-11-20', choices: [], usage: { prompt_tokens: 11, completion_tokens: 3 } }, +]; + +describe('wrapOpenAI — streaming', () => { + beforeEach(() => exporter.reset()); + + it('keeps the span open until the stream drains, then records usage', async () => { + const client = wrapOpenAI(fakeClient(() => fakeStream(STREAM_CHUNKS))); + const stream = await client.chat.completions.create({ ...REQUEST, stream: true }); + + expect(exporter.getFinishedSpans()).toHaveLength(0); + const seen = []; + for await (const chunk of stream as AsyncIterable) seen.push(chunk); + + expect(seen).toHaveLength(3); + const { attributes } = lastSpan(); + expect(attributes['gen_ai.response.id']).toBe('chatcmpl-s'); + expect(attributes['gen_ai.response.model']).toBe('gpt-4o-2024-11-20'); + expect(attributes['gen_ai.response.finish_reason']).toBe('stop'); + expect(attributes['gen_ai.usage.input_tokens']).toBe(11); + expect(attributes['gen_ai.usage.output_tokens']).toBe(3); + }); + + it('ends the span when the consumer breaks out early', async () => { + const client = wrapOpenAI(fakeClient(() => fakeStream(STREAM_CHUNKS))); + const stream = await client.chat.completions.create({ ...REQUEST, stream: true }); + for await (const _chunk of stream as AsyncIterable) break; + expect(exporter.getFinishedSpans()).toHaveLength(1); + }); + + it('preserves non-iterator members of the Stream object', async () => { + const client = wrapOpenAI(fakeClient(() => fakeStream(STREAM_CHUNKS))); + const stream = (await client.chat.completions.create({ ...REQUEST, stream: true })) as { controller: unknown }; + expect(stream.controller).toBeDefined(); + }); + + it('assembles streamed deltas into one assistant message when opted in', async () => { + const client = wrapOpenAI( + fakeClient(() => fakeStream(STREAM_CHUNKS)), + { recordContent: true }, + ); + const stream = await client.chat.completions.create({ ...REQUEST, stream: true }); + for await (const _chunk of stream as AsyncIterable) { + /* drain */ + } + const assistant = lastSpan().events.find((e) => e.name === 'gen_ai.assistant.message'); + expect(assistant!.attributes!['gen_ai.message.content']).toBe('four'); + }); +}); diff --git a/packages/core/src/gen-ai-attributes.ts b/packages/core/src/gen-ai-attributes.ts index 69515f1..3a8feaa 100644 --- a/packages/core/src/gen-ai-attributes.ts +++ b/packages/core/src/gen-ai-attributes.ts @@ -14,6 +14,7 @@ * `GenAIAttributes` as new attributes stabilize. */ import type { Span } from '@opentelemetry/api'; +import { scrubString } from './pii'; export type GenAIOperationName = 'chat' | 'text_completion' | 'embeddings' | 'tool' | 'agent' | 'rerank'; @@ -102,6 +103,16 @@ export function setGenAIAttributes(span: Span, attrs: GenAIAttributes): void { * Emit a `gen_ai.user.message` / `gen_ai.assistant.message` / `gen_ai.system.message` * span event so the dashboard's prompt/completion side-by-side view can render * the actual content of the LLM call. Use sparingly — these are size-heavy. + * + * Content is PII-scrubbed via {@link scrubString} before it leaves the process — + * prompts and tool arguments are the single most PII-dense payload this SDK can + * touch, so raw content never reaches the wire. + * + * ponytail: `scrubString` in TS is credentials-only today (Bearer tokens, api + * keys, `password=`). Keyed per-org hashing of names / emails / phones exists in + * Rust (`rust/observability/src/pii.rs`) and is being ported to TS in a parallel + * PR — when it lands, this call site inherits it for free because it already + * routes through the SDK's one scrub entry point. Do not scrub inline here. */ export function recordGenAIMessage( span: Span, @@ -111,7 +122,7 @@ export function recordGenAIMessage( ): void { const eventName = `gen_ai.${role}.message`; span.addEvent(eventName, { - 'gen_ai.message.content': content, + 'gen_ai.message.content': scrubString(content), ...(extra?.toolCallId !== undefined && { 'gen_ai.tool_call.id': extra.toolCallId }), ...(extra?.toolName !== undefined && { 'gen_ai.tool.name': extra.toolName }), }); diff --git a/packages/core/src/gen-ai-openai.ts b/packages/core/src/gen-ai-openai.ts new file mode 100644 index 0000000..56b327d --- /dev/null +++ b/packages/core/src/gen-ai-openai.ts @@ -0,0 +1,271 @@ +/** + * OpenAI Node SDK instrumentation — one wrapper that emits OTel GenAI + * semantic-convention spans for every `chat.completions.create` call. + * + * Why the OpenAI client and not the Vercel AI SDK: the OpenAI *wire shape* is + * the lingua franca. Groq, Together, Fireworks, DeepSeek, Azure OpenAI, and our + * own LiteLLM gateway at `llm.smoo.ai` all speak it through this same client, so + * one wrapper instruments all of them. It also needs no new dependency — the + * client is duck-typed structurally below, so `@smooai/observability` never + * imports `openai`. + * + * ```ts + * import OpenAI from 'openai'; + * import { wrapOpenAI } from '@smooai/observability'; + * + * const client = wrapOpenAI(new OpenAI(), { conversationId: convo.id }); + * await client.chat.completions.create({ model: 'gpt-4o', messages }); + * ``` + * + * Spec: https://opentelemetry.io/docs/specs/semconv/gen-ai/ + */ +import { SpanKind, SpanStatusCode, trace, type Span } from '@opentelemetry/api'; +import { recordGenAIMessage, setGenAIAttributes, type GenAIAttributes, type GenAISystem } from './gen-ai-attributes'; + +const TRACER_NAME = '@smooai/observability/gen-ai'; + +/** Token counts pulled off the provider response, handed to {@link WrapOpenAIOptions.costUsd}. */ +export interface GenAICostInput { + requestModel?: string; + responseModel?: string; + inputTokens?: number; + outputTokens?: number; + cachedTokens?: number; +} + +export interface WrapOpenAIOptions { + /** + * `gen_ai.system` value. Defaults to `'openai'`. Set it when pointing the + * client at an OpenAI-compatible gateway (`'groq'`, `'deepseek'`, …) so the + * spans attribute to the real provider. + */ + system?: GenAISystem; + /** `gen_ai.conversation.id` — stamped on every span from this client. */ + conversationId?: string; + /** `gen_ai.end_user.id`. Falls back to the request body's `user` field. */ + endUserId?: string; + /** + * Cost seam. Nothing in the platform emits `gen_ai.usage.cost_usd` on its + * own — providers don't return a price — so the dashboard's cost column + * stays empty until a caller supplies one. Return `undefined` to leave the + * attribute unset. + */ + costUsd?: (input: GenAICostInput) => number | undefined; + /** + * Record prompt + completion text as `gen_ai.*.message` span events. + * **Off by default**: this is the SDK's most PII-dense payload, and events + * are size-heavy. Content still goes through the SDK's PII scrub when on. + */ + recordContent?: boolean; +} + +/** Minimal structural view of the OpenAI client — avoids depending on `openai`. */ +type AnyRecord = Record; + +function num(value: unknown): number | undefined { + return typeof value === 'number' && Number.isFinite(value) ? value : undefined; +} + +function str(value: unknown): string | undefined { + return typeof value === 'string' && value.length > 0 ? value : undefined; +} + +function asRecord(value: unknown): AnyRecord | undefined { + return typeof value === 'object' && value !== null ? (value as AnyRecord) : undefined; +} + +function isAsyncIterable(value: unknown): value is AsyncIterable { + return typeof (value as AsyncIterable | undefined)?.[Symbol.asyncIterator] === 'function'; +} + +/** Tool names off an OpenAI `tools: [{ type: 'function', function: { name } }]` array. */ +function toolNames(body: AnyRecord): string[] | undefined { + const tools = body.tools; + if (!Array.isArray(tools)) return undefined; + const names = tools.map((t) => str(asRecord(asRecord(t)?.function)?.name) ?? str(asRecord(t)?.name)).filter((n): n is string => n !== undefined); + return names.length > 0 ? names : undefined; +} + +/** Request-side attributes, all of which are known before the call goes out. */ +function requestAttributes(body: AnyRecord, options: WrapOpenAIOptions): GenAIAttributes { + return { + system: options.system ?? 'openai', + // Straight passthrough on ingest with NO fallback — unset here means a + // NULL operation column in `gen_ai_events`, so always set it. + operationName: 'chat', + requestModel: str(body.model), + temperature: num(body.temperature), + topP: num(body.top_p), + maxTokens: num(body.max_tokens) ?? num(body.max_completion_tokens), + seed: num(body.seed), + toolNames: toolNames(body), + endUserId: options.endUserId ?? str(body.user), + conversationId: options.conversationId, + }; +} + +/** Usage block, tolerating the chat-completions and cached-token shapes. */ +function usageAttributes(usage: AnyRecord | undefined): Pick { + if (!usage) return {}; + return { + usageInputTokens: num(usage.prompt_tokens) ?? num(usage.input_tokens), + usageOutputTokens: num(usage.completion_tokens) ?? num(usage.output_tokens), + usageCachedTokens: num(asRecord(usage.prompt_tokens_details)?.cached_tokens) ?? num(asRecord(usage.input_tokens_details)?.cached_tokens), + }; +} + +/** Apply everything the provider told us after the call resolved. */ +function applyResponse(span: Span, requestModel: string | undefined, response: AnyRecord, options: WrapOpenAIOptions): void { + const usage = usageAttributes(asRecord(response.usage)); + const finishReason = str(asRecord(asRecord(response.choices)?.[0] as unknown)?.finish_reason); + const responseModel = str(response.model); + + setGenAIAttributes(span, { + ...usage, + responseModel, + responseId: str(response.id), + finishReason, + truncated: finishReason === undefined ? undefined : finishReason === 'length', + usageCostUsd: options.costUsd?.({ + requestModel, + responseModel, + inputTokens: usage.usageInputTokens, + outputTokens: usage.usageOutputTokens, + cachedTokens: usage.usageCachedTokens, + }), + }); + + if (options.recordContent) { + const message = asRecord(asRecord(asRecord(response.choices)?.[0] as unknown)?.message); + const content = str(message?.content); + if (content !== undefined) recordGenAIMessage(span, 'assistant', content); + } +} + +/** Record the prompt messages the caller sent, when content recording is on. */ +function recordPrompt(span: Span, body: AnyRecord): void { + const messages = body.messages; + if (!Array.isArray(messages)) return; + for (const raw of messages) { + const message = asRecord(raw); + const content = str(message?.content); + const role = str(message?.role); + if (content === undefined || role === undefined) continue; + if (role !== 'user' && role !== 'assistant' && role !== 'system' && role !== 'tool') continue; + recordGenAIMessage(span, role, content, { toolCallId: str(message?.tool_call_id) }); + } +} + +function failSpan(span: Span, error: unknown): void { + span.recordException(error instanceof Error ? error : new Error(String(error))); + span.setStatus({ code: SpanStatusCode.ERROR, message: error instanceof Error ? error.message : String(error) }); +} + +/** + * Wrap a streaming response so the span stays open for the whole stream and + * closes when the consumer finishes or breaks out early. Proxied rather than + * replaced so `stream.controller` / `stream.tee()` keep working. + */ +function instrumentStream(stream: AsyncIterable, span: Span, requestModel: string | undefined, options: WrapOpenAIOptions): AsyncIterable { + return new Proxy(stream as object, { + get(target, prop, receiver) { + if (prop !== Symbol.asyncIterator) { + const value = Reflect.get(target, prop, receiver); + return typeof value === 'function' ? value.bind(target) : value; + } + return function instrumentedIterator() { + return (async function* () { + // Chunks carry the response identity in pieces; the last one + // with `usage` only appears under `stream_options.include_usage`. + const merged: AnyRecord = {}; + let content = ''; + try { + for await (const raw of stream) { + const chunk = asRecord(raw); + if (chunk) { + if (str(chunk.id) !== undefined) merged.id = chunk.id; + if (str(chunk.model) !== undefined) merged.model = chunk.model; + if (asRecord(chunk.usage) !== undefined) merged.usage = chunk.usage; + const choice = asRecord(asRecord(chunk.choices)?.[0] as unknown); + if (str(choice?.finish_reason) !== undefined) { + merged.choices = [{ finish_reason: choice?.finish_reason }]; + } + if (options.recordContent) content += str(asRecord(choice?.delta)?.content) ?? ''; + } + yield raw; + } + if (options.recordContent && content.length > 0) { + merged.choices = [{ ...(asRecord(asRecord(merged.choices)?.[0] as unknown) ?? {}), message: { content } }]; + } + applyResponse(span, requestModel, merged, options); + } catch (error) { + failSpan(span, error); + throw error; + } finally { + span.end(); + } + })(); + }; + }, + }) as AsyncIterable; +} + +function instrumentedCreate(create: (...args: unknown[]) => unknown, self: unknown, options: WrapOpenAIOptions) { + return function wrappedCreate(this: unknown, ...args: unknown[]) { + const body = asRecord(args[0]) ?? {}; + const attrs = requestAttributes(body, options); + const spanName = attrs.requestModel !== undefined ? `chat ${attrs.requestModel}` : 'chat'; + + return trace.getTracer(TRACER_NAME).startActiveSpan(spanName, { kind: SpanKind.CLIENT }, async (span) => { + setGenAIAttributes(span, attrs); + if (options.recordContent) recordPrompt(span, body); + + let streaming = false; + try { + const result = await create.apply(self ?? this, args); + // A streaming call resolves to an async iterable; the span has to + // outlive it, so ownership of `span.end()` moves to the iterator. + if (body.stream === true && isAsyncIterable(result)) { + streaming = true; + return instrumentStream(result as AsyncIterable, span, attrs.requestModel, options); + } + const response = asRecord(result); + if (response) applyResponse(span, attrs.requestModel, response, options); + return result; + } catch (error) { + failSpan(span, error); + throw error; + } finally { + if (!streaming) span.end(); + } + }); + }; +} + +/** + * Return a proxy of `client` whose `chat.completions.create` is traced. The + * original client is untouched, and every other property passes through, so the + * wrapper is safe to apply once at construction. + * + * ponytail: only `chat.completions.create` is instrumented. `embeddings`, + * `responses`, and the Assistants API each need their own response shape — + * add them when something actually calls them. + */ +export function wrapOpenAI(client: T, options: WrapOpenAIOptions = {}): T { + return proxyPath(client, ['chat', 'completions', 'create'], (fn, self) => instrumentedCreate(fn, self, options)); +} + +/** Proxy just one property path on an object graph, leaving everything else alone. */ +function proxyPath(target: T, path: readonly string[], wrap: (fn: (...args: unknown[]) => unknown, self: unknown) => unknown): T { + const [head, ...rest] = path; + return new Proxy(target, { + get(obj, prop, receiver) { + const value = Reflect.get(obj, prop, receiver); + if (prop !== head) return value; + if (rest.length === 0) { + return typeof value === 'function' ? wrap(value as (...args: unknown[]) => unknown, obj) : value; + } + return typeof value === 'object' && value !== null ? proxyPath(value as object, rest, wrap) : value; + }, + }); +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index c4b8b94..7ac0175 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -9,6 +9,10 @@ export { Client } from './client'; // attributes to LLM/agent spans for the Smoo LLM dashboard + any // GenAI-semconv-aware OTel backend (Datadog, Honeycomb, Phoenix, …). export { setGenAIAttributes, recordGenAIMessage, type GenAIAttributes, type GenAIOperationName, type GenAISystem } from './gen-ai-attributes'; +// SMOODEV-1155 follow-up: OpenAI Node SDK instrumentation. The OpenAI wire +// shape is what Groq / Together / DeepSeek / Azure / our LiteLLM gateway all +// speak, so one wrapper covers every provider reachable through that client. +export { wrapOpenAI, type WrapOpenAIOptions, type GenAICostInput } from './gen-ai-openai'; // ADR-097: session-scoped sampling, config-served telemetry settings, and W3C // traceparent. Parity across the five SDKs is enforced by // `parity/sampling-corpus.json` — see `parity/README.md`.