feat(otel): cxtx OTEL bootstrap, usage parser, LLM cost spans

Cargo.toml

@@ -3,5 +3,6 @@ resolver = "2"
 members = [
     "server",
     "clients/rust",
+    "cxdb-otel",
     "cxtx",
 ]

Dockerfile

@@ -37,21 +37,24 @@ COPY Cargo.toml Cargo.lock* ./
 COPY server/Cargo.toml ./server/
 COPY clients/rust/Cargo.toml ./clients/rust/
 COPY cxtx/Cargo.toml ./cxtx/
+COPY cxdb-otel/Cargo.toml ./cxdb-otel/
 
 # Create dummy sources to build dependencies
-RUN mkdir -p server/src clients/rust/src cxtx/src && \
+RUN mkdir -p server/src clients/rust/src cxtx/src cxdb-otel/src && \
     echo "fn main() {}" > server/src/main.rs && \
     echo "pub fn dummy() {}" > clients/rust/src/lib.rs && \
     echo "pub fn dummy() {}" > cxtx/src/lib.rs && \
     echo "fn main() {}" > cxtx/src/main.rs && \
+    echo "pub fn dummy() {}" > cxdb-otel/src/lib.rs && \
     cargo build --release --manifest-path server/Cargo.toml && \
-    rm -rf server/src clients/rust/src cxtx/src
+    rm -rf server/src clients/rust/src cxtx/src cxdb-otel/src
 
 # Copy actual source and build
 COPY server/ ./server/
 COPY clients/ ./clients/
 COPY cxtx/ ./cxtx/
-RUN find server/src clients/rust/src cxtx/src -type f -exec touch {} + && \
+COPY cxdb-otel/ ./cxdb-otel/
+RUN find server/src clients/rust/src cxtx/src cxdb-otel/src -type f -exec touch {} + && \
     cargo build --release --manifest-path server/Cargo.toml
 
 # ============================================

clients/go/types/conversation.go

@@ -299,6 +299,14 @@ type TurnMetrics struct {
 
 	// Model is the model used for this turn.
 	Model string `msgpack:"7" json:"model,omitempty"`
+
+	// UsageStatus tags non-happy-path usage parses for OTEL emit.
+	// Empty / missing on the happy "Reported" path; "not_reported" when
+	// the stream finished cleanly with no usage object;
+	// "error:<class>" when the finalize path classified the call as an
+	// upstream error. Additive-only field; pre-existing records leave
+	// it unset.
+	UsageStatus string `msgpack:"8" json:"usage_status,omitempty"`
 }
 
 // =============================================================================
@@ -428,6 +436,11 @@ type ContextMetadata struct {
 	// Custom contains arbitrary key-value metadata.
 	Custom map[string]string `msgpack:"4" json:"custom,omitempty"`
 
+	// Tenant labels the context for OTEL `app.tenant` cost attribution.
+	// Empty / missing means no tenant — the OTEL emit path MUST omit the
+	// attribute entirely (no sentinel, no empty-string stamp).
+	Tenant string `msgpack:"5" json:"tenant,omitempty"`
+
 	// Provenance captures the origin story of this context.
 	// Includes process identity, user identity, trace context, and more.
 	// See Provenance type for full documentation.

clients/rust/src/types/builders.rs

@@ -17,6 +17,7 @@ impl ConversationItem {
                 title: String::new(),
                 labels: Vec::new(),
                 custom: std::collections::HashMap::new(),
+                tenant: None,
                 provenance: None,
             });
         }
@@ -123,6 +124,7 @@ impl AssistantTurnBuilder {
                 reasoning_tokens: None,
                 duration_ms: None,
                 model: String::new(),
+                usage_status: None,
             });
         }
         self

clients/rust/src/types/conversation.rs

@@ -165,6 +165,13 @@ pub struct TurnMetrics {
     pub duration_ms: Option<i64>,
     #[serde(rename = "7", skip_serializing_if = "String::is_empty")]
     pub model: String,
+    /// Diagnostic tag for non-happy-path usage parses. `None` for the
+    /// happy `Reported` path; `Some("not_reported")` when the stream
+    /// finished cleanly with no usage object; `Some("error:<class>")`
+    /// when the finalize path classified the call as an upstream error.
+    /// Additive-only field — serde defaults to `None` for pre-existing records.
+    #[serde(rename = "8", default, skip_serializing_if = "Option::is_none")]
+    pub usage_status: Option<String>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
@@ -247,6 +254,12 @@ pub struct ContextMetadata {
     pub labels: Vec<String>,
     #[serde(rename = "4", skip_serializing_if = "map_is_empty")]
     pub custom: std::collections::HashMap<String, String>,
+    /// Optional tenant label used for OTEL `app.tenant` attribution
+    /// (primary KPI = per-tenant LLM cost). When `None`, emit sites
+    /// MUST omit the attribute entirely — no sentinel, no empty string.
+    /// Wire tag 5 in both Rust and Go clients.
+    #[serde(default, rename = "5", skip_serializing_if = "Option::is_none")]
+    pub tenant: Option<String>,
     #[serde(rename = "10")]
     pub provenance: Option<super::provenance::Provenance>,
 }

clients/rust/src/types/tests.rs

@@ -31,6 +31,7 @@ fn fixture_conversation_item() -> ConversationItem {
         title: "Fixture Title".to_string(),
         labels: vec!["alpha".to_string(), "beta".to_string()],
         custom: std::collections::HashMap::from([("env".to_string(), "test".to_string())]),
+        tenant: None,
         provenance: None,
     });
     item

cxdb-otel/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "cxdb-otel"
+version = "0.1.0"
+edition = "2021"
+license = "Apache-2.0"
+description = "Narrow OpenTelemetry bootstrap shared by cxtx and cxdb-server binaries"
+authors = ["StrongDM <opensource@strongdm.com>"]
+repository = "https://github.com/strongdm/cxdb"
+homepage = "https://github.com/strongdm/cxdb"
+publish = false
+
+[dependencies]
+opentelemetry = "0.27"
+opentelemetry_sdk = { version = "0.27", features = ["rt-tokio"] }
+opentelemetry-otlp = { version = "0.27", features = ["grpc-tonic"] }
+opentelemetry-semantic-conventions = "0.27"
+tracing = "0.1"
+tracing-opentelemetry = "0.28"
+tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
+thiserror = "1"
+tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros"] }
+tiny_http = "0.12"
+reqwest = { version = "0.12", default-features = false, features = ["json"] }

cxdb-otel/src/gen_ai.rs

@@ -0,0 +1,163 @@
+//! Minimal emit surface for the `gen_ai.*` metric family.
+//!
+//! Exactly three helpers:
+//!
+//! - `emit_token_usage` — histogram samples for `gen_ai.client.token.usage`
+//! - `emit_calls` — per-call counter `gen_ai.calls`
+//! - `emit_usage_missing` — breadcrumb counter `gen_ai.usage_missing`
+//!
+//! Domain logic (bucket derivation, finish-reason mapping) lives in
+//! `cxtx::otel::*`; this module only wraps the OpenTelemetry meter API
+//! with lazy `OnceLock` instrument creation and a uniform attribute
+//! builder.
+//!
+//! Cardinality view configuration (dropping `app.session_id` / `app.user`
+//! / `app.wrapper_version` on all three metric names) is NOT applied via
+//! otel views — this crate emits only the attributes it's asked to, and
+//! callers are responsible for not passing the dropped attributes. The
+//! per-metric attribute surface is documented per helper.
+
+use std::borrow::Cow;
+
+use opentelemetry::metrics::{Counter, Histogram};
+use opentelemetry::{global, KeyValue};
+
+/// A lightweight, clonable attribute builder used at every `gen_ai.*`
+/// emit site. Keeps per-call allocation minimal and presents a single
+/// call-shape across the three helpers.
+#[derive(Debug, Clone, Default)]
+pub struct Attrs {
+    pairs: Vec<(Cow<'static, str>, Cow<'static, str>)>,
+}
+
+impl Attrs {
+    pub fn new() -> Self {
+        Self { pairs: Vec::new() }
+    }
+
+    /// Append a `(key, value)` pair. Both key and value may be `&'static
+    /// str` (zero-copy) or owned `String` (copies into `Cow::Owned`).
+    pub fn with(
+        mut self,
+        key: impl Into<Cow<'static, str>>,
+        value: impl Into<Cow<'static, str>>,
+    ) -> Self {
+        self.pairs.push((key.into(), value.into()));
+        self
+    }
+
+    /// Snapshot into the OpenTelemetry `KeyValue` slice shape expected by
+    /// the meter API.
+    pub fn to_kvs(&self) -> Vec<KeyValue> {
+        self.pairs
+            .iter()
+            .map(|(k, v)| KeyValue::new(k.clone().into_owned(), v.clone().into_owned()))
+            .collect()
+    }
+
+    /// Accessor for tests + callers that need to inspect which attributes
+    /// are being emitted (e.g., the cardinality-view test).
+    pub fn keys(&self) -> impl Iterator<Item = &str> {
+        self.pairs.iter().map(|(k, _)| k.as_ref())
+    }
+}
+
+// Instruments are recreated per call. The meter provider's SDK
+// deduplicates by (name, kind, unit, description) so instrument identity
+// is preserved across calls; constructing a fresh handle here is an
+// O(HashMap lookup) op inside the SDK, cheap enough per LLM call while
+// remaining test-friendly (tests that swap the global meter provider
+// don't get stuck with a handle pinned to the prior NoopMeterProvider).
+fn token_usage() -> Histogram<u64> {
+    global::meter("cxdb")
+        .u64_histogram("gen_ai.client.token.usage")
+        .with_unit("{token}")
+        .with_description(
+            "LLM token usage per non-zero bucket (input/cached/output/reasoning/cache_write*).",
+        )
+        .build()
+}
+
+fn calls() -> Counter<u64> {
+    global::meter("cxdb")
+        .u64_counter("gen_ai.calls")
+        .with_unit("{call}")
+        .with_description("LLM calls finalized with reported usage (unsampled).")
+        .build()
+}
+
+fn usage_missing() -> Counter<u64> {
+    global::meter("cxdb")
+        .u64_counter("gen_ai.usage_missing")
+        .with_unit("1")
+        .with_description(
+            "LLM calls finalized without usable usage; reason=error|not_reported|invalid.",
+        )
+        .build()
+}
+
+/// Emit one histogram sample per non-zero bucket. `buckets` is expected
+/// to already be filtered (zero-valued entries skipped) by
+/// `cxtx::otel::buckets::derive_and_validate`.
+pub fn emit_token_usage(buckets: &[(impl TokenTypeName, u64)], attrs: &Attrs) {
+    let base = attrs.to_kvs();
+    let hist = token_usage();
+    for (token_type, value) in buckets {
+        if *value == 0 {
+            continue;
+        }
+        let mut kvs = base.clone();
+        kvs.push(KeyValue::new(
+            "gen_ai.token.type",
+            token_type.token_type_name().to_string(),
+        ));
+        hist.record(*value, &kvs);
+    }
+}
+
+/// Increment `gen_ai.calls` by 1.
+pub fn emit_calls(attrs: &Attrs) {
+    calls().add(1, &attrs.to_kvs());
+}
+
+/// Increment `gen_ai.usage_missing` by 1. The `reason` attribute MUST be
+/// present on `attrs` (the helper does not synthesize a default — callers
+/// explicitly stamp `error` / `not_reported` / `invalid` per the dispatch
+/// table).
+pub fn emit_usage_missing(attrs: &Attrs) {
+    usage_missing().add(1, &attrs.to_kvs());
+}
+
+/// Abstraction so callers in `cxtx::otel::buckets` can pass their
+/// `TokenType` enum directly without depending on this crate.
+pub trait TokenTypeName {
+    fn token_type_name(&self) -> &str;
+}
+
+impl TokenTypeName for &str {
+    fn token_type_name(&self) -> &str {
+        self
+    }
+}
+
+impl TokenTypeName for String {
+    fn token_type_name(&self) -> &str {
+        self.as_str()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn attrs_round_trip() {
+        let a = Attrs::new()
+            .with("gen_ai.system", "anthropic")
+            .with("app.client_tag", String::from("cxtx/claude"));
+        let kvs = a.to_kvs();
+        assert_eq!(kvs.len(), 2);
+        let keys: Vec<&str> = a.keys().collect();
+        assert_eq!(keys, vec!["gen_ai.system", "app.client_tag"]);
+    }
+}