Extract LLM attributes from any OTEL convention, not just GenAI

## Summary

The span ingestion pipeline only recognized `gen_ai.*` attribute keys when extracting LLM-specific fields into promoted ClickHouse columns. Spans arriving via OpenInference, OpenLLMetry, Vercel AI SDK, or OpenAI Agents SDK had their LLM columns left empty despite carrying equivalent data under different keys and vocabularies.

This introduces a multi-convention extraction layer with two components: **scalar attribute resolvers** and **content payload parsers**.

### Scalar attribute resolvers (`resolvers.ts`)

Each promoted column is resolved from a priority-ordered list of convention-specific candidates. The first candidate that returns a value wins. Value translation is applied where conventions use different vocabularies.

| Column | GenAI current | GenAI deprecated / OpenLLMetry | OpenInference | Vercel AI SDK |
|---|---|---|---|---|
| `operation` | `gen_ai.operation.name` | `llm.request.type` (maps `completion`→`text_completion`, `embedding`→`embeddings`, etc.) | `openinference.span.kind` (maps `LLM`→`chat`, `EMBEDDING`→`embeddings`, `TOOL`→`execute_tool`, etc.) | `ai.operationId` (maps `ai.generateText`→`chat`, `ai.toolCall`→`execute_tool`, etc.) |
| `provider` | `gen_ai.provider.name` | `gen_ai.system` (aliases `bedrock`→`aws.bedrock`, `gemini`→`gcp.gemini`, `mistral`→`mistral_ai`, etc.) | `llm.system` (aliases `mistralai`→`mistral_ai`, `xai`→`x_ai`, `vertexai`→`gcp.vertex_ai`) | `ai.model.provider` (strips `.chat`/`.messages`/`.responses` suffixes, aliases `google.generative-ai`→`gcp.gemini`, `amazon-bedrock`→`aws.bedrock`) |
| `model` | `gen_ai.request.model` | same | `llm.model_name`, `embedding.model_name`, `reranker.model_name` | `ai.model.id` |
| `response_model` | `gen_ai.response.model` | same | `llm.model_name` (no request/response distinction) | `ai.response.model` |
| `tokens_input` | `gen_ai.usage.input_tokens` | `gen_ai.usage.prompt_tokens` | `llm.token_count.prompt` | `ai.usage.promptTokens` |
| `tokens_output` | `gen_ai.usage.output_tokens` | `gen_ai.usage.completion_tokens` | `llm.token_count.completion` | `ai.usage.completionTokens` |
| `tokens_cache_read` | `gen_ai.usage.cache_read.input_tokens` | same | `llm.token_count.prompt_details.cache_read` | — |
| `tokens_cache_create` | `gen_ai.usage.cache_creation.input_tokens` | same | `llm.token_count.prompt_details.cache_write` | — |
| `tokens_reasoning` | `gen_ai.usage.reasoning_tokens` | same | `llm.token_count.completion_details.reasoning` | — |
| `response_id` | `gen_ai.response.id` | same | — | `ai.response.id` |
| `finish_reasons` | `gen_ai.response.finish_reasons` (string[]) | same | — | `ai.response.finishReason` (singular string, wrapped to array; `tool-calls`→`tool_calls`, `content-filter`→`content_filter`) |
| `session_id` | `gen_ai.conversation.id` | same | `session.id` | — |
| `cost_*_microcents` | — | `gen_ai.usage.cost` (total only, USD float→microcents) | `llm.cost.prompt`, `llm.cost.completion`, `llm.cost.total` (USD float→microcents) | — |

OpenAI Agents SDK spans are handled implicitly — when bridged to OTEL via the official instrumentor, they emit GenAI convention attributes.

### Content payload parsers (`content/`)

LLM message payloads use fundamentally different storage structures across conventions, so each gets a dedicated parser with sentinel-based detection:

- **GenAI current** (sentinel: `gen_ai.input.messages` or `gen_ai.output.messages`): Parses structured/JSON messages already in GenAI parts-based format. Extracts `gen_ai.system_instructions` and `gen_ai.tool.definitions` as dedicated attributes.

- **GenAI deprecated / OpenLLMetry** (sentinel: `gen_ai.prompt` or `gen_ai.completion`): Parses flat JSON strings containing `{role, content}` message arrays. Translates to GenAI format via `rosetta-ai` auto-detection. Extracts `llm.request.functions` for tool definitions.

- **OpenInference** (sentinel: `llm.input_messages.*` prefix or `openinference.span.kind`): Reassembles flattened indexed span attributes (`llm.input_messages.{i}.message.role`, `.content`, `.tool_calls.{j}.tool_call.function.name`, etc.) by scanning, grouping by index, and sorting. Reconstructs `llm.tools.{i}.tool.json_schema` for tool definitions. Translates reassembled messages via `rosetta-ai`.

- **Vercel AI SDK** (sentinel: `ai.prompt` or `ai.prompt.messages`): Handles both top-level spans (`ai.prompt` JSON with `system` + `messages` fields) and call-level spans (`ai.prompt.messages` JSON array). Reconstructs output from split `ai.response.text` + `ai.response.toolCalls`. Parses `ai.prompt.tools` string array for tool definitions. Translates via `rosetta-ai` with explicit `Provider.VercelAI`.

All raw span attributes remain in the dynamic `attr_*` maps regardless of whether they were also extracted to promoted columns.

Author: csansoon

apps/ingest/src/routes/traces.ts

@@ -1,5 +1,6 @@
 import { OrganizationId, ProjectId, putInDisk } from "@domain/shared"
 import type { OtlpExportTraceServiceRequest } from "@domain/spans"
+import { validateOtlpCompliance } from "@domain/spans"
 import { Effect } from "effect"
 import type { Hono } from "hono"
 import { getSpanIngestionQueue, getStorageDisk } from "../clients.ts"
@@ -32,6 +33,11 @@ export const registerTracesRoute = ({ app }: TracesRouteContext) => {
       return c.json({})
     }
 
+    const validationError = validateOtlpCompliance(request)
+    if (validationError) {
+      return c.json({ error: `Non-compliant OTLP payload: ${validationError}` }, 400)
+    }
+
     const organizationId = c.get("organizationId")
     const projectId = c.get("projectId")
     const apiKeyId = c.get("apiKeyId")

packages/domain/spans/src/index.ts

@@ -5,4 +5,5 @@ export type { TraceListOptions, TraceRepository } from "./ports/trace-repository
 
 export type { TransformContext } from "./otlp/transform.ts"
 export { transformOtlpToSpans } from "./otlp/transform.ts"
-export type { OtlpExportTraceServiceRequest } from "./otlp/types.ts"
+export type { OtlpExportTraceServiceRequest, OtlpSpan } from "./otlp/types.ts"
+export { validateOtlpCompliance } from "./otlp/validate.ts"

packages/domain/spans/src/otlp/content/genai.ts

@@ -0,0 +1,44 @@
+/**
+ * Content parser for OTEL GenAI semantic convention v1.37+.
+ *
+ * Attributes:
+ *   gen_ai.input.messages  — structured object or JSON string (parts-based GenAI format)
+ *   gen_ai.output.messages — same
+ *   gen_ai.system_instructions — structured array of parts or JSON string
+ *   gen_ai.tool.definitions    — structured array or JSON string
+ */
+import type { GenAIMessage } from "rosetta-ai"
+import type { OtlpKeyValue } from "../types.ts"
+import type { ParsedContent } from "./index.ts"
+
+function extractJsonAttr(attrs: readonly OtlpKeyValue[], key: string): unknown {
+  const kv = attrs.find((a) => a.key === key)
+  if (!kv?.value) return undefined
+  if (kv.value.stringValue) {
+    try {
+      return JSON.parse(kv.value.stringValue)
+    } catch {
+      return undefined
+    }
+  }
+  return undefined
+}
+
+function parseMessages(attrs: readonly OtlpKeyValue[], key: string): GenAIMessage[] {
+  const raw = extractJsonAttr(attrs, key)
+  if (!Array.isArray(raw)) return []
+  return raw as GenAIMessage[]
+}
+
+export function parseGenAICurrent(attrs: readonly OtlpKeyValue[]): ParsedContent {
+  const inputMessages = parseMessages(attrs, "gen_ai.input.messages")
+  const outputMessages = parseMessages(attrs, "gen_ai.output.messages")
+
+  const systemRaw = extractJsonAttr(attrs, "gen_ai.system_instructions")
+  const systemInstructions = systemRaw ? JSON.stringify(systemRaw) : ""
+
+  const toolsRaw = extractJsonAttr(attrs, "gen_ai.tool.definitions")
+  const toolDefinitions = toolsRaw ? JSON.stringify(toolsRaw) : ""
+
+  return { inputMessages, outputMessages, systemInstructions, toolDefinitions }
+}

packages/domain/spans/src/otlp/content/genai_deprecated.ts

@@ -0,0 +1,59 @@
+/**
+ * Content parser for OTEL GenAI deprecated convention (pre-v1.37) and OpenLLMetry.
+ *
+ * Both use the same attribute keys:
+ *   gen_ai.prompt     — single JSON string with array of {role, content} messages
+ *   gen_ai.completion — single JSON string with array of {role, content} messages
+ *
+ * The JSON content is in provider-native format (typically OpenAI-style {role, content}).
+ * We use rosetta-ai's auto-detection to translate to GenAI format.
+ *
+ * OpenLLMetry additionally defines:
+ *   llm.request.functions — JSON string array of function/tool definitions
+ */
+import { safeTranslate } from "rosetta-ai"
+import type { GenAIMessage } from "rosetta-ai"
+import type { OtlpKeyValue } from "../types.ts"
+import type { ParsedContent } from "./index.ts"
+
+function parseJsonString(attrs: readonly OtlpKeyValue[], key: string): unknown {
+  const kv = attrs.find((a) => a.key === key)
+  if (!kv?.value?.stringValue) return undefined
+  try {
+    return JSON.parse(kv.value.stringValue)
+  } catch {
+    return undefined
+  }
+}
+
+function translateMessages(raw: unknown, direction: "input" | "output"): GenAIMessage[] {
+  if (!Array.isArray(raw) || raw.length === 0) return []
+  const result = safeTranslate(raw, { direction })
+  if (result.error) return []
+  return result.messages as GenAIMessage[]
+}
+
+export function parseGenAIDeprecated(attrs: readonly OtlpKeyValue[]): ParsedContent {
+  const promptRaw = parseJsonString(attrs, "gen_ai.prompt")
+  const completionRaw = parseJsonString(attrs, "gen_ai.completion")
+
+  let inputMessages: GenAIMessage[] = []
+  let systemInstructions = ""
+
+  if (Array.isArray(promptRaw) && promptRaw.length > 0) {
+    const result = safeTranslate(promptRaw, { direction: "input" })
+    if (!result.error) {
+      inputMessages = result.messages as GenAIMessage[]
+      if (result.system) {
+        systemInstructions = JSON.stringify(result.system)
+      }
+    }
+  }
+
+  const outputMessages = translateMessages(completionRaw, "output")
+
+  const functionsRaw = parseJsonString(attrs, "llm.request.functions")
+  const toolDefinitions = Array.isArray(functionsRaw) ? JSON.stringify(functionsRaw) : ""
+
+  return { inputMessages, outputMessages, systemInstructions, toolDefinitions }
+}

packages/domain/spans/src/otlp/content/index.ts

@@ -0,0 +1,66 @@
+import type { GenAIMessage } from "rosetta-ai"
+import { stringAttr } from "../resolvers.ts"
+import type { OtlpKeyValue } from "../types.ts"
+import { parseGenAICurrent } from "./genai.ts"
+import { parseGenAIDeprecated } from "./genai_deprecated.ts"
+import { parseOpenInference } from "./openinference.ts"
+import { parseVercel } from "./vercel.ts"
+
+export interface ParsedContent {
+  readonly inputMessages: readonly GenAIMessage[]
+  readonly outputMessages: readonly GenAIMessage[]
+  readonly systemInstructions: string
+  readonly toolDefinitions: string
+}
+
+const EMPTY_CONTENT: ParsedContent = {
+  inputMessages: [],
+  outputMessages: [],
+  systemInstructions: "",
+  toolDefinitions: "",
+}
+
+interface ContentParser {
+  canHandle(attrs: readonly OtlpKeyValue[]): boolean
+  parse(attrs: readonly OtlpKeyValue[]): ParsedContent
+}
+
+function hasKey(attrs: readonly OtlpKeyValue[], key: string): boolean {
+  return attrs.some((a) => a.key === key)
+}
+
+function hasKeyPrefix(attrs: readonly OtlpKeyValue[], prefix: string): boolean {
+  return attrs.some((a) => a.key.startsWith(prefix))
+}
+
+const PARSERS: readonly ContentParser[] = [
+  {
+    canHandle: (attrs) => hasKey(attrs, "gen_ai.input.messages") || hasKey(attrs, "gen_ai.output.messages"),
+    parse: parseGenAICurrent,
+  },
+  {
+    canHandle: (attrs) =>
+      hasKeyPrefix(attrs, "llm.input_messages.") ||
+      hasKeyPrefix(attrs, "llm.output_messages.") ||
+      (stringAttr(attrs, "openinference.span.kind") !== undefined && hasKeyPrefix(attrs, "llm.")),
+    parse: parseOpenInference,
+  },
+  {
+    canHandle: (attrs) => hasKey(attrs, "ai.prompt") || hasKey(attrs, "ai.prompt.messages"),
+    parse: parseVercel,
+  },
+  // GenAI deprecated / OpenLLMetry is the broadest fallback (gen_ai.prompt is common)
+  {
+    canHandle: (attrs) => hasKey(attrs, "gen_ai.prompt") || hasKey(attrs, "gen_ai.completion"),
+    parse: parseGenAIDeprecated,
+  },
+]
+
+export function parseContent(attrs: readonly OtlpKeyValue[]): ParsedContent {
+  for (const parser of PARSERS) {
+    if (parser.canHandle(attrs)) {
+      return parser.parse(attrs)
+    }
+  }
+  return EMPTY_CONTENT
+}

packages/domain/spans/src/otlp/content/openinference.ts

@@ -0,0 +1,172 @@
+/**
+ * Content parser for OpenInference (Arize Phoenix).
+ *
+ * OpenInference explodes messages into flattened indexed span attributes:
+ *   llm.input_messages.{i}.message.role
+ *   llm.input_messages.{i}.message.content
+ *   llm.input_messages.{i}.message.tool_calls.{j}.tool_call.function.name
+ *   llm.input_messages.{i}.message.tool_calls.{j}.tool_call.function.arguments
+ *   llm.input_messages.{i}.message.contents.{j}.message_content.type
+ *   llm.input_messages.{i}.message.contents.{j}.message_content.text
+ *   llm.input_messages.{i}.message.contents.{j}.message_content.image.image.url
+ *
+ * Output messages follow the same pattern with llm.output_messages.{i}.
+ *
+ * Tool definitions use:
+ *   llm.tools.{i}.tool.json_schema — JSON string of tool schema
+ *
+ * We reassemble these into message arrays, then translate via rosetta-ai.
+ */
+import { safeTranslate } from "rosetta-ai"
+import type { GenAIMessage } from "rosetta-ai"
+import type { OtlpKeyValue } from "../types.ts"
+import type { ParsedContent } from "./index.ts"
+
+interface ToolCallData {
+  name: string
+  arguments: string
+}
+
+interface ReassembledMessage {
+  role: string
+  content: string
+  tool_calls?: { id: string; type: string; function: { name: string; arguments: string } }[]
+}
+
+const INPUT_PREFIX = "llm.input_messages."
+const OUTPUT_PREFIX = "llm.output_messages."
+const TOOLS_PREFIX = "llm.tools."
+
+function reassembleMessages(attrs: readonly OtlpKeyValue[], prefix: string): ReassembledMessage[] {
+  const byIndex = new Map<number, Map<string, string>>()
+  const toolCalls = new Map<number, Map<number, ToolCallData>>()
+
+  for (const attr of attrs) {
+    if (!attr.key.startsWith(prefix)) continue
+    const rest = attr.key.slice(prefix.length)
+
+    const dotIdx = rest.indexOf(".")
+    if (dotIdx === -1) continue
+
+    const index = Number.parseInt(rest.slice(0, dotIdx), 10)
+    if (Number.isNaN(index)) continue
+
+    const field = rest.slice(dotIdx + 1)
+    const value = attr.value?.stringValue ?? ""
+
+    if (field.startsWith("message.tool_calls.")) {
+      const tcRest = field.slice("message.tool_calls.".length)
+      const tcDotIdx = tcRest.indexOf(".")
+      if (tcDotIdx === -1) continue
+      const tcIndex = Number.parseInt(tcRest.slice(0, tcDotIdx), 10)
+      if (Number.isNaN(tcIndex)) continue
+      const tcField = tcRest.slice(tcDotIdx + 1)
+
+      let msgToolCalls = toolCalls.get(index)
+      if (!msgToolCalls) {
+        msgToolCalls = new Map()
+        toolCalls.set(index, msgToolCalls)
+      }
+      let tc = msgToolCalls.get(tcIndex)
+      if (!tc) {
+        tc = { name: "", arguments: "" }
+        msgToolCalls.set(tcIndex, tc)
+      }
+      if (tcField === "tool_call.function.name") tc.name = value
+      else if (tcField === "tool_call.function.arguments") tc.arguments = value
+    } else if (field.startsWith("message.")) {
+      const msgField = field.slice("message.".length)
+      let fields = byIndex.get(index)
+      if (!fields) {
+        fields = new Map()
+        byIndex.set(index, fields)
+      }
+      fields.set(msgField, value)
+    }
+  }
+
+  const maxIndex = Math.max(...byIndex.keys(), ...toolCalls.keys(), -1)
+  if (maxIndex === -1) return []
+
+  const messages: ReassembledMessage[] = []
+  for (let i = 0; i <= maxIndex; i++) {
+    const fields = byIndex.get(i)
+    const role = fields?.get("role") ?? "user"
+    const content = fields?.get("content") ?? ""
+
+    const msg: ReassembledMessage = { role, content }
+
+    const msgToolCalls = toolCalls.get(i)
+    if (msgToolCalls && msgToolCalls.size > 0) {
+      const sorted = [...msgToolCalls.entries()].sort(([a], [b]) => a - b)
+      msg.tool_calls = sorted.map(([j, tc]) => ({
+        id: `call_${i}_${j}`,
+        type: "function" as const,
+        function: { name: tc.name, arguments: tc.arguments },
+      }))
+    }
+
+    messages.push(msg)
+  }
+
+  return messages
+}
+
+function reassembleToolDefinitions(attrs: readonly OtlpKeyValue[]): string {
+  const tools = new Map<number, string>()
+
+  for (const attr of attrs) {
+    if (!attr.key.startsWith(TOOLS_PREFIX)) continue
+    const rest = attr.key.slice(TOOLS_PREFIX.length)
+    const dotIdx = rest.indexOf(".")
+    if (dotIdx === -1) continue
+    const index = Number.parseInt(rest.slice(0, dotIdx), 10)
+    if (Number.isNaN(index)) continue
+    const field = rest.slice(dotIdx + 1)
+    if (field === "tool.json_schema" && attr.value?.stringValue) {
+      tools.set(index, attr.value.stringValue)
+    }
+  }
+
+  if (tools.size === 0) return ""
+
+  const sorted = [...tools.entries()].sort(([a], [b]) => a - b)
+  const parsed = sorted.map(([, json]) => {
+    try {
+      return JSON.parse(json)
+    } catch {
+      return json
+    }
+  })
+  return JSON.stringify(parsed)
+}
+
+function translateReassembled(messages: ReassembledMessage[], direction: "input" | "output"): GenAIMessage[] {
+  if (messages.length === 0) return []
+  const result = safeTranslate(messages, { direction })
+  if (result.error) return []
+  return result.messages as GenAIMessage[]
+}
+
+export function parseOpenInference(attrs: readonly OtlpKeyValue[]): ParsedContent {
+  const inputRaw = reassembleMessages(attrs, INPUT_PREFIX)
+  const outputRaw = reassembleMessages(attrs, OUTPUT_PREFIX)
+
+  let inputMessages: GenAIMessage[] = []
+  let systemInstructions = ""
+
+  if (inputRaw.length > 0) {
+    const result = safeTranslate(inputRaw, { direction: "input" })
+    if (!result.error) {
+      inputMessages = result.messages as GenAIMessage[]
+      if (result.system) {
+        systemInstructions = JSON.stringify(result.system)
+      }
+    }
+  }
+
+  const outputMessages = translateReassembled(outputRaw, "output")
+  const toolDefinitions = reassembleToolDefinitions(attrs)
+
+  return { inputMessages, outputMessages, systemInstructions, toolDefinitions }
+}