From a8baf218467512db4b61e8d1eeca7dfb20b99a92 Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Tue, 11 Aug 2026 22:15:18 +0200 Subject: [PATCH] feat(ingest): add the pure AI span classifier and registry loader MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds three self-contained Rust modules: `ai_registry` (loads and validates the vendored `registry.json` at first use, rejecting a bad artifact at load rather than at classification), `ai_classifier` (per-span vendor and session-key resolution over the registry's four-op algebra, with a resource/scope context hoisted per batch), and `cityhash102` (the CityHash64 v1.0.2 variant ClickHouse's `cityHash64` implements, so Rust and SQL hash identically). Nothing calls this yet — it is pure CPU with no pipeline wiring, which is what makes it reviewable on its own. The only edit outside the new files is widening `telemetry::any_value_string` to `pub(crate)` so the classifier canonicalizes attribute values through the exact function the row writer uses. The corpus replay test (`ai_classifier_corpus_test.rs`) is an on-demand local gate driven by `TRACE_CAPTURE_DIR`; it needs `serde_yaml` as a dev-dependency to read trace-capture's seed goldens, and is skipped when the variable is unset. The criterion group measures classification in isolation. Co-Authored-By: Claude Fable 5 --- apps/ingest/Cargo.lock | 20 + apps/ingest/Cargo.toml | 3 + apps/ingest/benches/ingest_bench.rs | 127 +- apps/ingest/src/ai_classifier.rs | 1493 ++++++++++++++++++ apps/ingest/src/ai_classifier_corpus_test.rs | 441 ++++++ apps/ingest/src/ai_registry.rs | 1125 +++++++++++++ apps/ingest/src/cityhash102.rs | 345 ++++ apps/ingest/src/lib.rs | 3 + apps/ingest/src/telemetry.rs | 5 +- 9 files changed, 3560 insertions(+), 2 deletions(-) create mode 100644 apps/ingest/src/ai_classifier.rs create mode 100644 apps/ingest/src/ai_classifier_corpus_test.rs create mode 100644 apps/ingest/src/ai_registry.rs create mode 100644 apps/ingest/src/cityhash102.rs diff --git a/apps/ingest/Cargo.lock b/apps/ingest/Cargo.lock index bee384a4a..6b4b53f09 100644 --- a/apps/ingest/Cargo.lock +++ b/apps/ingest/Cargo.lock @@ -1480,6 +1480,7 @@ dependencies = [ "rustls", "serde", "serde_json", + "serde_yaml", "sha2 0.10.9", "tikv-jemallocator", "tokio", @@ -2471,6 +2472,19 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_yaml" +version = "0.9.34+deprecated" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47" +dependencies = [ + "indexmap", + "itoa", + "ryu", + "serde", + "unsafe-libyaml", +] + [[package]] name = "sha2" version = "0.10.9" @@ -3077,6 +3091,12 @@ dependencies = [ "subtle", ] +[[package]] +name = "unsafe-libyaml" +version = "0.2.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861" + [[package]] name = "untrusted" version = "0.9.0" diff --git a/apps/ingest/Cargo.toml b/apps/ingest/Cargo.toml index ecd5d6716..47b0098c3 100644 --- a/apps/ingest/Cargo.toml +++ b/apps/ingest/Cargo.toml @@ -68,6 +68,9 @@ webpki-roots = "0.26" [dev-dependencies] criterion = { version = "0.5", features = ["async_tokio"] } +# Test-only: parses the trace-capture registry-seed.yaml goldens in the corpus +# replay (src/ai_classifier_corpus_test.rs). Never linked into the binary. +serde_yaml = "0.9" [[bench]] name = "ingest_bench" diff --git a/apps/ingest/benches/ingest_bench.rs b/apps/ingest/benches/ingest_bench.rs index b87348101..fd5b99675 100644 --- a/apps/ingest/benches/ingest_bench.rs +++ b/apps/ingest/benches/ingest_bench.rs @@ -12,6 +12,8 @@ use axum::routing::post; use axum::Router; use criterion::{black_box, criterion_group, criterion_main, Criterion}; use flate2::read::GzDecoder; +use maple_ingest::ai_classifier::ResourceContext; +use maple_ingest::ai_registry::registry; use maple_ingest::telemetry::{ ClickHouseBreakerConfig, DatasourceNames, SamplingPolicy, TelemetryPipeline, TinybirdConfig, }; @@ -79,6 +81,129 @@ fn bench_ingest_accept(c: &mut Criterion) { let _ = std::fs::remove_dir_all(&fixture.queue_dir); } +/// Classifier cost in isolation (write-side plan constraint 2: ~50 ns/span mean, +/// ~300 ns worst case on a 60-attribute AI span, out of a ~500 ns total per-span +/// budget). Everything here is pure CPU — no pipeline, no I/O. +fn bench_ai_classifier(c: &mut Criterion) { + let registry = registry(); + let mut group = c.benchmark_group("ai_classifier"); + + // Typical non-AI server span: nothing survives the prefilter. + let http_resource = vec![ + string_kv("service.name", "checkout-api"), + string_kv("telemetry.sdk.name", "opentelemetry"), + string_kv("telemetry.sdk.language", "nodejs"), + string_kv("deployment.environment.name", "production"), + ]; + let http_scope = InstrumentationScope { + name: "@opentelemetry/instrumentation-http".to_string(), + version: "0.57.0".to_string(), + ..Default::default() + }; + let http_attributes: Vec = [ + ("http.request.method", "POST"), + ("url.path", "/v2/checkout"), + ("url.scheme", "https"), + ("server.address", "api.example.com"), + ("http.response.status_code", "200"), + ] + .iter() + .map(|(k, v)| string_kv(k, v)) + .collect(); + let http_attributes_15: Vec = http_attributes + .iter() + .cloned() + .chain((0..10).map(|i| string_kv(&format!("net.peer.detail_{i}"), "value"))) + .collect(); + + // A fat AI span: 60 attributes, most of them registry-referenced. + let ai_resource = vec![ + string_kv("service.name", "spring-ai-trace-capture"), + string_kv("telemetry.sdk.name", "opentelemetry"), + ]; + let ai_scope = InstrumentationScope { + name: "org.springframework.boot".to_string(), + version: "4.1.0".to_string(), + ..Default::default() + }; + let mut ai_attributes = vec![ + string_kv("spring.ai.kind", "chat_client"), + string_kv("gen_ai.system", "spring_ai"), + string_kv("gen_ai.operation.name", "chat"), + string_kv("gen_ai.request.model", "gpt-4o-mini"), + string_kv("gen_ai.response.model", "gpt-4o-mini-2024-07-18"), + string_kv("session.id", "sess-4f9c1b2e-77aa-4c31-9d0e-3b8f1a6d2c55"), + ]; + ai_attributes.extend((0..54).map(|i| { + string_kv( + &format!("gen_ai.request.parameter_{i}"), + "a moderately long attribute value, as vendors emit", + ) + })); + + group.bench_function("non_ai_span_5_attrs", |b| { + let resource = ResourceContext::new(registry, &http_resource); + let scope = resource.scope(Some(&http_scope), ""); + b.iter(|| black_box(scope.classify_span("POST /v2/checkout", black_box(&http_attributes)))); + }); + + group.bench_function("non_ai_span_15_attrs", |b| { + let resource = ResourceContext::new(registry, &http_resource); + let scope = resource.scope(Some(&http_scope), ""); + b.iter(|| { + black_box(scope.classify_span("POST /v2/checkout", black_box(&http_attributes_15))) + }); + }); + + group.bench_function("ai_span_60_attrs", |b| { + let resource = ResourceContext::new(registry, &ai_resource); + let scope = resource.scope(Some(&ai_scope), ""); + b.iter(|| black_box(scope.classify_span("chat_client", black_box(&ai_attributes)))); + }); + + // Per-batch hoisting: one ResourceSpans + one ScopeSpans. Amortized over the + // spans of that scope, so it is charged once per scope, not per span. + // Same span shape, but the 54 filler keys start with a byte no registry key or + // prefix begins with, so the prefilter rejects them on the byte screen alone. + // The delta against `ai_span_60_attrs` is the cost of hashing keys that survive + // the screen and miss the exact-key map — the classifier's main hotspot today. + let mut ai_attributes_screened = ai_attributes[..6].to_vec(); + ai_attributes_screened.extend((0..54).map(|i| { + string_kv( + &format!("zzz.request.parameter_{i}"), + "a moderately long attribute value, as vendors emit", + ) + })); + group.bench_function("ai_span_60_attrs_screened_out", |b| { + let resource = ResourceContext::new(registry, &ai_resource); + let scope = resource.scope(Some(&ai_scope), ""); + b.iter(|| { + black_box(scope.classify_span("chat_client", black_box(&ai_attributes_screened))) + }); + }); + + group.bench_function("hoist_resource_and_scope", |b| { + b.iter(|| { + let resource = ResourceContext::new(registry, black_box(&ai_resource)); + black_box(resource.scope(Some(&ai_scope), "")); + }); + }); + + // The realistic unit: hoist once, then classify 20 spans (the plan's spans/trace + // figure). Divide by 20 for the effective per-span cost including hoisting. + group.bench_function("hoisted_scope_20_ai_spans", |b| { + b.iter(|| { + let resource = ResourceContext::new(registry, black_box(&ai_resource)); + let scope = resource.scope(Some(&ai_scope), ""); + for _ in 0..20 { + black_box(scope.classify_span("chat_client", black_box(&ai_attributes))); + } + }); + }); + + group.finish(); +} + impl BenchFixture { async fn new() -> Self { let fake_state = FakeTinybirdState::default(); @@ -253,5 +378,5 @@ fn unique_temp_dir(prefix: &str) -> PathBuf { std::env::temp_dir().join(format!("{prefix}-{}-{nanos}", std::process::id())) } -criterion_group!(benches, bench_ingest_accept); +criterion_group!(benches, bench_ingest_accept, bench_ai_classifier); criterion_main!(benches); diff --git a/apps/ingest/src/ai_classifier.rs b/apps/ingest/src/ai_classifier.rs new file mode 100644 index 000000000..6088b37c0 --- /dev/null +++ b/apps/ingest/src/ai_classifier.rs @@ -0,0 +1,1493 @@ +//! Per-span AI classification: vendor, session-key state, session-key hash. +//! +//! Pure functions over decoded OTLP. No I/O, no clock, no cross-span state — the +//! write-side plan's first design constraint is that a span's classification depends +//! on nothing but that span, its scope and its resource, because root spans arrive +//! last in 18/19 multi-batch corpus traces. +//! +//! # Shape +//! +//! ```text +//! ResourceContext::new(registry, resource.attributes) // once per ResourceSpans +//! └─ .scope(scope, schema_url) // once per ScopeSpans +//! └─ .classify_span(span.name, span.attributes) // per span +//! ``` +//! +//! The two hoisted levels do all the work that is constant across a scope: they +//! resolve every matcher as far as resource/scope evidence allows, leaving a +//! per-span pass that touches the span's own attributes once. +//! +//! # Evaluation semantics +//! +//! These are the semantics `packages/domain/src/ai-registry/compile-sql.ts` compiles +//! to SQL, and the two are held to it by the differential suites in that directory: +//! +//! * **Canonical stringification** happens before matching, via the row writer's +//! `any_value_string`. +//! * **Duplicate attribute keys: first occurrence wins**, and only among +//! registry-referenced keys. The prefilter is what identifies them; keys no rule +//! consults are never hashed, which is what keeps the budget. +//! * **Lookup is class-directed** (plan §1: matcher classes are per-class +//! predicates). A matcher's declared class picks the one attribute list its keys +//! may read — `resource` → the resource attributes, `scope` → the scope +//! attributes, `attr` → the span's own — and `key_prefix` scans the keys of that +//! one list. Predicates with no class (unknown-tier fingerprints, session +//! candidates, authority predicates) are span-local. The four pseudo-keys are +//! exempt: `scope.name` / `scope.version` / `scope.schema_url` / `span.name` are +//! columns, not map entries, so they resolve to their column whatever the class +//! says — `effect_ai`'s attr matchers key on `span.name` and must keep firing. +//! +//! trace-capture's `scripts/verify-seed.ts` instead falls back +//! span → scope → resource for every key regardless of class, and unions all three +//! attribute lists as `key_prefix` evidence. That reference evaluator is now the +//! outlier and stays that way, in its own repo: it verifies one seed at a time, +//! where cross-class reads cannot promote another vendor. Here they can, and did +//! — `langsmith.internal_provider` is langchain's *insufficient* resource key and +//! also inside langchain's attr-class `key_prefix("langsmith.")`, so under the +//! fallback a process that set it on its resource had every span, plain HTTP +//! included, classified `langchain` with the value ignored. That defeats the +//! sufficiency gate outright. The corpus goldens are insensitive to the +//! difference: all 10,091 corpus spans classify identically either way. +//! * **Resolution** (plan §1): sufficient resource/scope hits and attr hits are +//! unconditional; insufficient resource/scope hits are conditional candidates, +//! promoted only by a same-vendor attr hit on the same span. The winner is the +//! highest priority among unconditional hits and promoted candidates. Unknown-tier +//! fingerprints sit in the same table one priority band lower (D4), so "no vendor +//! hit → unknown tier → non-AI" falls out of the ordering rather than a second pass. +//! * **Session key** (plan §2 step 5): every candidate of the winning vendor is +//! evaluated; the span's state is the `max` over candidates, and the hash comes +//! from the candidate that produced the winning state, ties broken by candidate +//! order. +//! +//! # What this module does not do +//! +//! Nothing here is wired into the ingest request path yet — no `encode_traces` +//! change, no row columns. That is a later stage. + +use std::borrow::Cow; + +use opentelemetry_proto::tonic::common::v1::{InstrumentationScope, KeyValue}; + +use crate::ai_registry::{ + canonical_value, registry, AttrTarget, Authority, HitKind, KeyId, LookupKey, Matcher, + Predicate, PseudoKey, Registry, SessionCandidate, VendorId, +}; +use crate::cityhash102::city_hash64; + +/// `AiSessionKeyState`. Values are frozen at v1 and append-only: the rollup MV +/// persists threshold comparisons over them (`state >= 3` is the eligibility +/// contract), so renumbering would silently rewrite history. +pub mod session_state { + /// Not examined, or examined and not AI. + pub const NOT_EXAMINED: u8 = 0; + /// Vendor has no session-key rules (includes every `unknown:*` bucket). + pub const NO_RULES: u8 = 1; + /// Span is not session-authoritative. + pub const NOT_AUTHORITATIVE: u8 = 2; + /// Authoritative, key absent. + pub const KEY_ABSENT: u8 = 3; + /// Key present but failed validation (empty or a decoy value). + pub const KEY_INVALID: u8 = 4; + /// Resolved at `run`/`instance`/`user` granularity. + pub const SUB_SESSION: u8 = 5; + /// Resolved at `session` granularity. + pub const SESSION: u8 = 6; +} + +/// The per-span output. `rules_version` is stamped on **every** examined span, +/// including non-AI ones: `AiRulesVersion != 0 AND AiVendor = ''` is what makes +/// "definitively classified non-AI" distinguishable from "pre-rollout row". +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SpanClassification<'a> { + pub vendor: Option, + pub session_state: u8, + /// The raw winning session-key value, for states 5 and 6 only. Never stored — + /// hash it with [`SpanClassification::session_key_hash`]. + pub session_key: Option>, + pub rules_version: u32, +} + +impl SpanClassification<'_> { + /// `cityHash64(value)`, or 0 when no key resolved. + /// + /// The hash is a storage format, not a confidentiality boundary: the value it + /// digests came out of `SpanAttributes` and stays there in the clear on the same + /// row, so anything that can read `AiSessionKeyHash` can read the source value + /// beside it. What hashing buys is physical — 8 fixed bytes on a per-span column, + /// and a numeric input to `uniqCombined` in `service_ai_vendors_hourly`. + /// + /// 64 bits is matched to that consumer. At ~1M distinct sessions in an org the + /// birthday collision probability is ~1e-8, six orders of magnitude under the + /// ~1.6% standard error of the `uniqCombined(12)` sketch that consumes it. + pub fn session_key_hash(&self) -> u64 { + match &self.session_key { + Some(value) => city_hash64(value.as_bytes()), + None => 0, + } + } + + /// Vendor slug for the row column: `''` for non-AI. `'static` because the + /// registry is: the slug set is closed and outlives every span. + pub fn vendor_slug(&self) -> &'static str { + match self.vendor { + Some(id) => registry().vendor_slug(id), + None => "", + } + } +} + +// --------------------------------------------------------------------------- +// hoisted contexts +// --------------------------------------------------------------------------- + +/// How many registry-referenced attributes one attribute list holds before the +/// view spills to the heap. The corpus p99 is 3; the fattest hand-built AI span +/// in the bench carries 6. +const INLINE_ATTRS: usize = 8; + +/// 64-bit words backing [`AttrView::key_bits`]. Sized from +/// [`crate::ai_registry::MAX_KEYS`], which the registry validates at load. +const KEY_BITS_WORDS: usize = crate::ai_registry::MAX_KEYS / 64; + +/// Registry-referenced attribute values of one attribute list, first-occurrence-wins, +/// plus which registry key-prefixes its keys satisfy. +/// +/// A flat `(KeyId, value)` list, **not** a `registry.keys().len()`-wide slot +/// array. The slot array was O(1) to read but cost a 116-slot zeroing +/// allocation on every span that carried a single registry key — one malloc and +/// ~3.7 KiB of memset for typically three useful entries. Lookups are a linear +/// scan over `len` instead, which for a handful of entries is a cache line, not +/// a branch-predictor problem. +struct AttrView<'a> { + inline: [Option<(KeyId, Cow<'a, str>)>; INLINE_ATTRS], + inline_len: usize, + /// Only allocated by spans carrying more than `INLINE_ATTRS` registry keys. + spill: Vec<(KeyId, Cow<'a, str>)>, + /// Which `KeyId`s this view holds. Scope hoisting asks ~one question per + /// matcher and almost every answer is "absent", so the miss has to cost a + /// bit test rather than a walk of the entry list. + key_bits: [u64; KEY_BITS_WORDS], + prefix_bits: u64, +} + +impl Default for AttrView<'_> { + fn default() -> Self { + Self { + inline: [const { None }; INLINE_ATTRS], + inline_len: 0, + spill: Vec::new(), + key_bits: [0; KEY_BITS_WORDS], + prefix_bits: 0, + } + } +} + +impl<'a> AttrView<'a> { + fn build(registry: &Registry, attributes: &'a [KeyValue]) -> Self { + let mut view = AttrView::default(); + for attribute in attributes { + let probe = registry.probe(&attribute.key); + view.prefix_bits |= probe.prefix_bits; + if let Some(key) = probe.key_id { + // First occurrence wins (plan §2), among registry keys only. + if view.get(key).is_some() { + continue; + } + view.push(key, canonical_value(attribute.value.as_ref())); + } + } + view + } + + fn push(&mut self, key: KeyId, value: Cow<'a, str>) { + self.key_bits[key as usize / 64] |= 1u64 << (key % 64); + if self.inline_len < INLINE_ATTRS { + self.inline[self.inline_len] = Some((key, value)); + self.inline_len += 1; + } else { + self.spill.push((key, value)); + } + } + + #[inline] + fn holds(&self, key: KeyId) -> bool { + self.key_bits[key as usize / 64] & (1u64 << (key % 64)) != 0 + } + + fn get(&self, key: KeyId) -> Option<&str> { + self.slot(key).map(|(_, value)| &**value) + } + + /// Like [`AttrView::get`], but keeps the attribute list's lifetime rather + /// than the view's — session-key values outlive the borrow of the view. + fn get_cow(&self, key: KeyId) -> Option> { + self.slot(key).map(|(_, value)| value.clone()) + } + + fn slot(&self, key: KeyId) -> Option<&(KeyId, Cow<'a, str>)> { + if !self.holds(key) { + return None; + } + self.slots().find(|(id, _)| *id == key) + } + + fn slots(&self) -> impl Iterator)> + '_ { + self.inline[..self.inline_len] + .iter() + .flatten() + .chain(self.spill.iter()) + } + + /// The view's entries in insertion order. + fn entries(&self) -> impl Iterator + '_ { + self.slots().map(|(key, value)| (*key, &**value)) + } +} + +/// Resolved once per `ResourceSpans`. +pub struct ResourceContext<'a> { + registry: &'static Registry, + attrs: AttrView<'a>, +} + +impl<'a> ResourceContext<'a> { + pub fn new(registry: &'static Registry, attributes: &'a [KeyValue]) -> Self { + Self { + registry, + attrs: AttrView::build(registry, attributes), + } + } + + /// Resolved once per `ScopeSpans`. `schema_url` is the `ScopeSpans.schema_url` + /// (falling back to the scope's own, as the reference evaluator does). + pub fn scope<'r>( + &'r self, + scope: Option<&'a InstrumentationScope>, + schema_url: &'a str, + ) -> ScopeContext<'a, 'r> { + let scope_attrs = match scope { + Some(scope) => AttrView::build(self.registry, &scope.attributes), + None => AttrView::default(), + }; + let mut context = ScopeContext { + registry: self.registry, + resource: &self.attrs, + scope_attrs, + scope_name: scope.map(|s| s.name.as_str()).unwrap_or(""), + scope_version: scope.map(|s| s.version.as_str()).unwrap_or(""), + scope_schema_url: schema_url, + hoisted: Hoisted::default(), + }; + context.hoist(); + context + } +} + +/// What resource+scope evidence alone already decides, shared by every span in the +/// scope. Recomputing this per span is the cost the plan's hoisting exists to avoid. +/// +/// Class-directed lookup makes this a *complete* verdict for every matcher it covers: +/// no span attribute can shadow a resource `eq` hit any more, because a resource-class +/// matcher never reads span attributes. +#[derive(Default)] +struct Hoisted { + best: Option<(u32, VendorId)>, + attr_bits: u64, + conditional_bits: u32, +} + +pub struct ScopeContext<'a, 'r> { + registry: &'static Registry, + resource: &'r AttrView<'a>, + scope_attrs: AttrView<'a>, + scope_name: &'a str, + scope_version: &'a str, + scope_schema_url: &'a str, + hoisted: Hoisted, +} + +impl<'a, 'r> ScopeContext<'a, 'r> { + /// The attribute list a matcher of this class reads. `Span` is never asked of a + /// scope: a span-class matcher reaches the hoist path only via a pseudo-key. + fn class_attrs(&self, target: AttrTarget) -> Option<&AttrView<'a>> { + match target { + AttrTarget::Resource => Some(self.resource), + AttrTarget::Scope => Some(&self.scope_attrs), + AttrTarget::Span => None, + } + } + + fn pseudo(&self, key: PseudoKey) -> Option<&'a str> { + match key { + PseudoKey::ScopeName => Some(self.scope_name), + PseudoKey::ScopeVersion => Some(self.scope_version), + PseudoKey::ScopeSchemaUrl => Some(self.scope_schema_url), + // Only a span knows its name; resolved per span. + PseudoKey::SpanName => None, + } + } + + /// Evaluate a matcher the scope fully decides — `registry.hoisted_matchers()`, i.e. + /// every `scope`/`resource`-class matcher plus anything keyed on a `scope.*` + /// pseudo-key. Nothing here can read a span attribute, so the verdict is final. + fn matches_hoisted(&self, matcher: &Matcher) -> bool { + let attrs = self.class_attrs(matcher.target); + match &matcher.predicate { + Predicate::Present(LookupKey::Attr(key)) => { + attrs.is_some_and(|attrs| attrs.holds(*key)) + } + // Pseudo-keys are columns, not map entries: always present, possibly empty. + Predicate::Present(LookupKey::Pseudo(pseudo)) => *pseudo != PseudoKey::SpanName, + Predicate::Eq(LookupKey::Attr(key), value) => { + attrs.and_then(|attrs| attrs.get(*key)) == Some(&**value) + } + Predicate::Eq(LookupKey::Pseudo(pseudo), value) => { + self.pseudo(*pseudo) == Some(&**value) + } + Predicate::KeyPrefix(prefix) => { + attrs.is_some_and(|attrs| attrs.prefix_bits & (1u64 << prefix) != 0) + } + Predicate::ValuePrefix(pseudo, prefix) => self + .pseudo(*pseudo) + .is_some_and(|value| value.starts_with(&**prefix)), + } + } + + fn hoist(&mut self) { + let mut hoisted = std::mem::take(&mut self.hoisted); + for &id in self.registry.hoisted_matchers() { + let matcher = &self.registry.matchers()[id as usize]; + if self.matches_hoisted(matcher) { + record_hit(&mut hoisted, matcher); + } + } + self.hoisted = hoisted; + } + + /// Classify one span. `attributes` is the span's raw attribute list; duplicates + /// and unordered keys are fine. + pub fn classify_span( + &self, + span_name: &'a str, + attributes: &'a [KeyValue], + ) -> SpanClassification<'a> { + let facts = AttrView::build(self.registry, attributes); + let view = SpanView { + scope: self, + span_name, + facts: &facts, + }; + + let mut best = self.hoisted.best; + let mut attr_bits = self.hoisted.attr_bits; + let mut conditional_bits = self.hoisted.conditional_bits; + let mut hit = |matcher: &Matcher| { + match matcher.kind { + HitKind::Unconditional => { + if best.is_none_or(|(priority, _)| matcher.priority > priority) { + best = Some((matcher.priority, matcher.vendor)); + } + if matcher.promotes { + attr_bits |= 1u64 << matcher.vendor.index(); + } + } + HitKind::Conditional => { + if let Some(slot) = matcher.conditional_slot { + conditional_bits |= 1u32 << slot; + } + } + }; + }; + + // The span's own attributes: exact-key matchers, then the prefix families its + // keys satisfy. Both tables hold only span-class matchers — a resource/scope + // matcher's verdict was final at hoist time and cannot be reopened here. + for (key, value) in facts.entries() { + for &id in self.registry.key_matchers(key) { + let matcher = &self.registry.matchers()[id as usize]; + let satisfied = match &matcher.predicate { + Predicate::Present(_) => true, + Predicate::Eq(_, expected) => value == &**expected, + _ => false, + }; + if satisfied { + hit(matcher); + } + } + } + let mut prefixes = facts.prefix_bits; + while prefixes != 0 { + let prefix = prefixes.trailing_zeros() as u8; + prefixes &= prefixes - 1; + for &id in self.registry.prefix_matchers(prefix) { + hit(&self.registry.matchers()[id as usize]); + } + } + + // Span-name matchers (`eq` via an index, the rare rest by evaluation). + for &id in self.registry.span_name_matchers(span_name) { + hit(&self.registry.matchers()[id as usize]); + } + for &id in self.registry.span_name_other() { + let matcher = &self.registry.matchers()[id as usize]; + if view.eval(&matcher.predicate) { + hit(matcher); + } + } + + // Promotion: an insufficient resource/scope candidate becomes a hit at its + // own priority only if the same span produced an attr hit for its vendor. + let mut bits = conditional_bits; + while bits != 0 { + let slot = bits.trailing_zeros() as usize; + bits &= bits - 1; + let matcher = &self.registry.matchers()[self.registry.conditional(slot) as usize]; + if attr_bits & (1u64 << matcher.vendor.index()) == 0 { + continue; + } + if best.is_none_or(|(priority, _)| matcher.priority > priority) { + best = Some((matcher.priority, matcher.vendor)); + } + } + + let vendor = best.map(|(_, vendor)| vendor); + let (session_state, session_key) = match vendor { + Some(vendor) => self.evaluate_session(vendor, &view), + None => (session_state::NOT_EXAMINED, None), + }; + SpanClassification { + vendor, + session_state, + session_key, + rules_version: self.registry.version(), + } + } + + /// Session-key evaluation for a vendor: all candidates, reduced by `max`. + /// + /// Public because it is also the per-vendor state the corpus goldens are + /// expressed in (they report the state of *every* span under one vendor's + /// candidates, independent of which vendor classified it). + pub fn evaluate_session_for_vendor( + &self, + vendor: VendorId, + span_name: &'a str, + attributes: &'a [KeyValue], + ) -> (u8, Option>) { + let facts = AttrView::build(self.registry, attributes); + let view = SpanView { + scope: self, + span_name, + facts: &facts, + }; + self.evaluate_session(vendor, &view) + } + + fn evaluate_session( + &self, + vendor: VendorId, + view: &SpanView<'a, '_, 'r>, + ) -> (u8, Option>) { + let entry = self.registry.vendor(vendor); + if entry.candidates().is_empty() { + return (session_state::NO_RULES, None); + } + let mut best_state = 0u8; + let mut best_key: Option> = None; + for candidate in entry.candidates() { + let (state, value) = self.candidate_state(entry, candidate, view); + // Strictly greater: ties keep the earlier candidate's hash (plan §1). + if state > best_state { + best_state = state; + best_key = if state >= session_state::SUB_SESSION { + value + } else { + None + }; + } + } + (best_state, best_key) + } + + fn candidate_state( + &self, + vendor: &crate::ai_registry::Vendor, + candidate: &SessionCandidate, + view: &SpanView<'a, '_, 'r>, + ) -> (u8, Option>) { + let authoritative = match &candidate.authority { + Authority::Always => true, + Authority::One(predicate) => view.eval(predicate), + Authority::AnyOf(predicates) => predicates.iter().any(|p| view.eval(p)), + }; + if !authoritative { + return (session_state::NOT_AUTHORITATIVE, None); + } + // Presence, not `!= ''`: present-but-empty must stay distinguishable from + // absent (it is state 4, not state 3). + let Some(value) = view.lookup_span_local(candidate.key) else { + return (session_state::KEY_ABSENT, None); + }; + if candidate.require_non_empty && value.is_empty() { + return (session_state::KEY_INVALID, None); + } + if candidate.reject_decoy_values && vendor.is_decoy_value(&value) { + return (session_state::KEY_INVALID, None); + } + (candidate.granularity.resolved_state(), Some(value)) + } +} + +fn record_hit(hoisted: &mut Hoisted, matcher: &Matcher) { + match matcher.kind { + HitKind::Unconditional => { + if hoisted + .best + .is_none_or(|(priority, _)| matcher.priority > priority) + { + hoisted.best = Some((matcher.priority, matcher.vendor)); + } + if matcher.promotes { + hoisted.attr_bits |= 1u64 << matcher.vendor.index(); + } + } + HitKind::Conditional => { + if let Some(slot) = matcher.conditional_slot { + hoisted.conditional_bits |= 1u32 << slot; + } + } + } +} + +/// A span plus the scope/resource it hangs off — the unit every predicate reads. +struct SpanView<'a, 'v, 'r> { + scope: &'v ScopeContext<'a, 'r>, + span_name: &'a str, + facts: &'v AttrView<'a>, +} + +impl<'a> SpanView<'a, '_, '_> { + /// The attribute list a predicate of this class reads — one list, never a chain. + fn attrs(&self, target: AttrTarget) -> &AttrView<'a> { + match target { + AttrTarget::Span => self.facts, + AttrTarget::Scope => &self.scope.scope_attrs, + AttrTarget::Resource => self.scope.resource, + } + } + + /// Pseudo-key → its column; every other key → `target`'s attribute list alone. + fn lookup(&self, key: LookupKey, target: AttrTarget) -> Option> { + match key { + LookupKey::Pseudo(PseudoKey::SpanName) => Some(Cow::Borrowed(self.span_name)), + LookupKey::Pseudo(PseudoKey::ScopeName) => Some(Cow::Borrowed(self.scope.scope_name)), + LookupKey::Pseudo(PseudoKey::ScopeVersion) => { + Some(Cow::Borrowed(self.scope.scope_version)) + } + LookupKey::Pseudo(PseudoKey::ScopeSchemaUrl) => { + Some(Cow::Borrowed(self.scope.scope_schema_url)) + } + LookupKey::Attr(key) => self.attrs(target).get_cow(key), + } + } + + /// Span-local lookup: session-candidate keys and authority predicates carry no + /// class and read the span's own attributes, as `compile-sql.ts` compiles them. + fn lookup_span_local(&self, key: LookupKey) -> Option> { + self.lookup(key, AttrTarget::Span) + } + + /// Direct predicate evaluation against one class's evidence. Used for session + /// candidates and authority predicates (span-local), and as the reference the + /// indexed matcher path is checked against. + fn eval_for(&self, predicate: &Predicate, target: AttrTarget) -> bool { + match predicate { + Predicate::Present(key) => self.lookup(*key, target).is_some(), + Predicate::Eq(key, value) => self.lookup(*key, target).as_deref() == Some(&**value), + Predicate::KeyPrefix(prefix) => self.attrs(target).prefix_bits & (1u64 << prefix) != 0, + Predicate::ValuePrefix(pseudo, prefix) => self + .lookup(LookupKey::Pseudo(*pseudo), target) + .is_some_and(|value| value.starts_with(&**prefix)), + } + } + + fn eval(&self, predicate: &Predicate) -> bool { + self.eval_for(predicate, AttrTarget::Span) + } +} + +#[cfg(test)] +impl<'a, 'r> ScopeContext<'a, 'r> { + /// Test-only reference resolver: evaluates **every** matcher directly, with no + /// hoisting and no dispatch index. Used to prove the fast path agrees. + pub(crate) fn classify_span_unindexed( + &self, + span_name: &'a str, + attributes: &'a [KeyValue], + ) -> Option { + let facts = AttrView::build(self.registry, attributes); + let view = SpanView { + scope: self, + span_name, + facts: &facts, + }; + let mut best: Option<(u32, VendorId)> = None; + let mut attr_bits = 0u64; + let mut conditional = Vec::new(); + for matcher in self.registry.matchers() { + if !view.eval_for(&matcher.predicate, matcher.target) { + continue; + } + match matcher.kind { + HitKind::Unconditional => { + if best.is_none_or(|(priority, _)| matcher.priority > priority) { + best = Some((matcher.priority, matcher.vendor)); + } + if matcher.promotes { + attr_bits |= 1u64 << matcher.vendor.index(); + } + } + HitKind::Conditional => conditional.push(matcher), + } + } + for matcher in conditional { + if attr_bits & (1u64 << matcher.vendor.index()) == 0 { + continue; + } + if best.is_none_or(|(priority, _)| matcher.priority > priority) { + best = Some((matcher.priority, matcher.vendor)); + } + } + best.map(|(_, vendor)| vendor) + } +} + +#[cfg(test)] +#[path = "ai_classifier_corpus_test.rs"] +mod corpus_test; + +#[cfg(test)] +mod tests { + use super::*; + use opentelemetry_proto::tonic::common::v1::{any_value, AnyValue}; + + pub(crate) fn kv(key: &str, value: &str) -> KeyValue { + KeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(any_value::Value::StringValue(value.to_string())), + }), + } + } + + fn int_kv(key: &str, value: i64) -> KeyValue { + KeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(any_value::Value::IntValue(value)), + }), + } + } + + fn scope(name: &str) -> InstrumentationScope { + InstrumentationScope { + name: name.to_string(), + ..Default::default() + } + } + + /// Classify one span standalone. Returns the vendor slug (`""` for non-AI) and + /// the session state. + fn classify( + resource: &[KeyValue], + scope_name: &str, + span_name: &str, + attributes: &[KeyValue], + ) -> (String, u8) { + let registry = registry(); + let resource_context = ResourceContext::new(registry, resource); + let scope = scope(scope_name); + let scope_context = resource_context.scope(Some(&scope), ""); + let result = scope_context.classify_span(span_name, attributes); + ( + result + .vendor + .map(|v| registry.vendor_slug(v).to_string()) + .unwrap_or_default(), + result.session_state, + ) + } + + // -- plan §2 correctness invariants ------------------------------------- + + /// "Spring's plain HTTP POST spans under org.springframework.boot → non-AI." + /// The scope matcher is insufficient, so it is a candidate nothing promotes. + #[test] + fn insufficient_scope_alone_is_not_ai() { + let (vendor, state) = classify( + &[kv("service.name", "spring-ai-trace-capture")], + "org.springframework.boot", + "POST", + &[ + kv("method", "POST"), + kv("uri", "/v1/chat/completions"), + kv("status", "200"), + kv("outcome", "SUCCESS"), + ], + ); + assert_eq!(vendor, ""); + assert_eq!(state, session_state::NOT_EXAMINED); + } + + /// "…its spring_ai chat_client spans → spring_ai." The `spring.ai.*` attr hit + /// promotes the same-vendor candidate. + #[test] + fn a_same_vendor_attr_hit_promotes_the_candidate() { + let (vendor, _) = classify( + &[kv("service.name", "spring-ai-trace-capture")], + "org.springframework.boot", + "chat_client", + &[ + kv("spring.ai.kind", "chat_client"), + kv("gen_ai.operation.name", "framework"), + ], + ); + assert_eq!(vendor, "spring_ai"); + } + + /// The negative direction of the promotion rule, stated on its own: an + /// insufficient *resource* match plus another vendor's attr hit must not make + /// the resource vendor win. + #[test] + fn an_insufficient_resource_candidate_does_not_swallow_another_instrumentor() { + // litellm's resource matcher is `present(model_id)`, insufficient. + let (vendor, _) = classify( + &[kv("service.name", "gateway"), kv("model_id", "gpt-4o-mini")], + "openinference.instrumentation.openai", + "ChatCompletion", + &[kv("openinference.span.kind", "LLM")], + ); + assert_eq!(vendor, "openinference-openai"); + } + + /// Mixed-vendor **process**: the corpus' crewai captures run the OpenInference + /// OpenAI instrumentor alongside crewai's own, and the two must resolve per span + /// (crewai_user is 12 crewai + 13 openinference-openai spans). + #[test] + fn mixed_vendor_process_yields_per_span_vendors() { + let registry = registry(); + let resource = [kv("service.name", "crewai-trace-capture")]; + let resource_context = ResourceContext::new(registry, &resource); + + let crewai_scope = scope("openinference.instrumentation.crewai"); + let crewai = resource_context.scope(Some(&crewai_scope), ""); + let task_attributes = [kv("task_key", "research")]; + let task = crewai.classify_span("Task._execute_core", &task_attributes); + assert_eq!(task.vendor_slug(), "crewai"); + + let openai_scope = scope("openinference.instrumentation.openai"); + let openai = resource_context.scope(Some(&openai_scope), ""); + let llm_attributes = [kv("openinference.span.kind", "LLM")]; + let llm = openai.classify_span("ChatCompletion", &llm_attributes); + assert_eq!(llm.vendor_slug(), "openinference-openai"); + } + + /// A **sufficient** resource matcher does apply process-wide — that is what + /// sufficiency means, and mastra's is declared sufficient in the seed *by + /// construction*: `@mastra/otel-exporter` mints its own resource per exported + /// span inside Mastra's converter, so a co-loaded instrumentor's spans carry the + /// NodeSDK resource instead and never reach this branch. Stated as a test + /// because the write-side plan's §1 prose uses mastra as its example of an + /// *insufficient* resource matcher; the wire-verified seed overrode that. + #[test] + fn a_sufficient_resource_matcher_applies_process_wide() { + let registry = registry(); + let resource = [ + kv("service.name", "mastra-app"), + kv("telemetry.sdk.name", "@mastra/otel-exporter"), + ]; + let resource_context = ResourceContext::new(registry, &resource); + let mastra_scope = scope("@mastra/otel-exporter"); + let mastra = resource_context.scope(Some(&mastra_scope), ""); + let attributes = [kv("mastra.span.type", "agent_run")]; + assert_eq!( + mastra + .classify_span("agent.generate", &attributes) + .vendor_slug(), + "mastra" + ); + } + + /// Classify one span with attributes on all three levels. + fn classify_layered( + resource: &[KeyValue], + scope_name: &str, + scope_attributes: Vec, + span_name: &str, + attributes: &[KeyValue], + ) -> (String, u8) { + let registry = registry(); + let resource_context = ResourceContext::new(registry, resource); + let scope = InstrumentationScope { + name: scope_name.to_string(), + attributes: scope_attributes, + ..Default::default() + }; + let scope_context = resource_context.scope(Some(&scope), ""); + let result = scope_context.classify_span(span_name, attributes); + ( + result + .vendor + .map(|v| registry.vendor_slug(v).to_string()) + .unwrap_or_default(), + result.session_state, + ) + } + + // -- class-directed lookup ------------------------------------------------ + + /// A matcher reads exactly the attribute list its class names. Every case here + /// carries a registry key somewhere its matcher does not look, and must classify + /// non-AI — the same verdict `compile-sql.ts` produces from the written row. + #[test] + fn a_registry_key_outside_its_matchers_class_does_not_fire() { + // mastra's *sufficient* resource matcher, carried as a span attribute. + assert_eq!( + classify_layered( + &[kv("service.name", "app")], + "com.example.app", + vec![], + "agent.generate", + &[kv("telemetry.sdk.name", "@mastra/otel-exporter")], + ) + .0, + "" + ); + // agno's attr-class `key_prefix(agno.)` family, carried on the resource. + assert_eq!( + classify_layered( + &[kv("service.name", "app"), kv("agno.run.id", "r-1")], + "com.example.app", + vec![], + "op", + &[kv("http.route", "/x")], + ) + .0, + "" + ); + // spring_ai's attr-class `eq(gen_ai.system,…)`, carried on the scope. + assert_eq!( + classify_layered( + &[], + "com.example.app", + vec![kv("gen_ai.system", "spring_ai")], + "op", + &[kv("http.route", "/x")], + ) + .0, + "" + ); + // Unknown-tier fingerprints are span-local too, on either other list. + assert_eq!( + classify_layered( + &[kv("gen_ai.operation.name", "chat")], + "com.example.app", + vec![], + "op", + &[kv("http.route", "/x")], + ) + .0, + "" + ); + assert_eq!( + classify_layered( + &[], + "com.example.app", + vec![kv("openinference.span.kind", "LLM")], + "op", + &[kv("http.route", "/x")], + ) + .0, + "" + ); + } + + /// The other direction: a resource-class matcher is decided by the resource alone, + /// so a span attribute of the same name can no longer veto it. + #[test] + fn a_span_attribute_cannot_veto_a_resource_class_match() { + let (vendor, _) = classify_layered( + &[ + kv("service.name", "app"), + kv("telemetry.sdk.name", "@mastra/otel-exporter"), + ], + "com.example.app", + vec![], + "agent.generate", + &[kv("telemetry.sdk.name", "@opentelemetry/sdk-node")], + ); + assert_eq!(vendor, "mastra"); + } + + /// Pseudo-keys stay class-free: they are columns, not map entries. effect_ai's + /// attr matchers key on `span.name` and must keep firing under class-directed + /// lookup — the one way this change could have silently deleted a vendor. + #[test] + fn pseudo_keys_resolve_for_every_matcher_class() { + let (vendor, _) = classify(&[], "com.example.app", "LanguageModel.generateText", &[]); + assert_eq!(vendor, "effect_ai"); + // …and the scope pseudo-keys still decide scope-class matchers with no + // scope attributes present at all. + let (scoped, _) = classify(&[], "gcp.vertex.agent", "invocation", &[]); + assert_eq!(scoped, "google_adk"); + } + + /// Session candidates and their authority predicates are span-local: the key must + /// be on the span, not inherited from the scope or the resource. + #[test] + fn session_candidates_read_only_the_spans_own_attributes() { + // The key on the resource: authoritative, but the key is absent (state 3). + let (vendor, resource_key) = classify_layered( + &[kv("service.name", "app"), kv("session.id", "sess-res")], + "com.anthropic.claude_code", + vec![], + "claude_code.interaction", + &[kv("span.type", "interaction")], + ); + assert_eq!(vendor, "claude_agent_sdk"); + assert_eq!(resource_key, session_state::KEY_ABSENT); + + // The key on the scope: same. + let (_, scope_key) = classify_layered( + &[], + "com.anthropic.claude_code", + vec![kv("session.id", "sess-scope")], + "claude_code.interaction", + &[kv("span.type", "interaction")], + ); + assert_eq!(scope_key, session_state::KEY_ABSENT); + + // The *authority* predicate's key on the resource: the span is not + // authoritative, even though its own session key resolves. + let (_, authority) = classify_layered( + &[kv("span.type", "interaction")], + "com.anthropic.claude_code", + vec![], + "claude_code.interaction", + &[kv("session.id", "sess-auth")], + ); + assert_eq!(authority, session_state::NOT_AUTHORITATIVE); + + // All three on the span: resolved. + let (_, span_local) = classify( + &[], + "com.anthropic.claude_code", + "claude_code.interaction", + &[kv("span.type", "interaction"), kv("session.id", "sess-1")], + ); + assert_eq!(span_local, session_state::SESSION); + } + + /// F2, the hazard class-directed lookup exists to close. + /// + /// `langsmith.internal_provider` is langchain's **insufficient** resource matcher + /// key and also lies inside langchain's attr-class `key_prefix("langsmith.")`. + /// Under a cross-class fallback the resource attribute satisfied that attr matcher + /// — which is unconditional *and* promotes — so one resource attribute classified + /// every span in the process as langchain, plain HTTP included and whatever the + /// value. The insufficient match must contribute nothing on its own. + #[test] + fn an_insufficient_resource_key_does_not_promote_itself_through_its_vendors_prefix_family() { + for value in ["false", "true"] { + let (vendor, state) = classify_layered( + &[ + kv("service.name", "lc"), + kv("langsmith.internal_provider", value), + ], + "@opentelemetry/instrumentation-http", + vec![], + "POST /v1/chat", + &[ + kv("http.request.method", "POST"), + kv("url.path", "/v1/chat"), + ], + ); + assert_eq!(vendor, "", "resource langsmith.internal_provider={value}"); + assert_eq!(state, session_state::NOT_EXAMINED); + } + + // Promotion still works through the proper channel: a genuine span-level + // `langsmith.*` attribute is an attr hit for langchain. + let (promoted, _) = classify_layered( + &[ + kv("service.name", "lc"), + kv("langsmith.internal_provider", "true"), + ], + "@opentelemetry/instrumentation-http", + vec![], + "chain", + &[kv("langsmith.trace.name", "agent")], + ); + assert_eq!(promoted, "langchain"); + } + + #[test] + fn duplicate_attribute_keys_take_the_first_occurrence() { + // `gen_ai.system` is claimed by several vendors with different values; the + // first occurrence must decide, and the second must be invisible. + let (first, _) = classify( + &[], + "app.tracer", + "chat", + &[ + kv("gen_ai.system", "spring_ai"), + kv("gen_ai.system", "strands-agents"), + ], + ); + let (second, _) = classify( + &[], + "app.tracer", + "chat", + &[ + kv("gen_ai.system", "strands-agents"), + kv("gen_ai.system", "spring_ai"), + ], + ); + assert_eq!(first, "spring_ai"); + assert_eq!(second, "strands"); + } + + #[test] + fn unknown_tier_buckets_unmatched_fingerprints() { + for (attributes, expected) in [ + (vec![kv("gen_ai.operation.name", "chat")], "unknown:genai"), + ( + vec![kv("openinference.span.kind", "LLM")], + "unknown:openinference", + ), + (vec![kv("llm.model_name", "gpt-4o")], "unknown:other"), + (vec![kv("traceloop.workflow.name", "w")], "unknown:other"), + ] { + let (vendor, state) = classify(&[], "com.example.app", "call", &attributes); + assert_eq!(vendor, expected, "for {attributes:?}"); + assert_eq!(state, session_state::NO_RULES); + } + } + + /// The plan gates `input.value`/`output.value` on co-occurrence with an + /// OpenInference attribute. registry.json encodes that gate by **omitting** them + /// from the unknown tier entirely (compile-registry.ts: "input.value/output.value + /// stay gated/excluded in v1"), so standalone occurrences must classify non-AI + /// and co-occurring ones must be caught by the OpenInference fingerprint itself. + #[test] + fn generic_input_output_values_do_not_fire_on_their_own() { + let (standalone, _) = classify( + &[], + "com.example.app", + "handler", + &[kv("input.value", "{}"), kv("output.value", "{}")], + ); + assert_eq!(standalone, ""); + + let (co_occurring, _) = classify( + &[], + "com.example.app", + "handler", + &[ + kv("input.value", "{}"), + kv("openinference.span.kind", "CHAIN"), + ], + ); + assert_eq!(co_occurring, "unknown:openinference"); + } + + #[test] + fn a_vendor_hit_always_outranks_the_unknown_tier() { + // vercel's `ai.` attr family overlaps the `ai.*` unknown fingerprint. + let (vendor, _) = classify( + &[], + "ai", + "ai.generateText", + &[kv("ai.operationId", "ai.generateText")], + ); + assert_eq!(vendor, "vercel_ai_sdk"); + } + + // -- session-key states -------------------------------------------------- + + /// google_adk's two candidate populations are disjoint; `max` unions them + /// instead of letting the absent one cancel the present one. + #[test] + fn disjoint_candidate_populations_union_via_max() { + // Candidate 1 (gen_ai.conversation.id) is authoritative only on + // invoke_agent/generate_content spans; candidate 2 + // (gcp.vertex.agent.session_id) only where gen_ai.system says so. This span + // satisfies only the second: candidate 1 scores 2, candidate 2 scores 6, and + // `max` unions them instead of letting the unauthoritative one cancel. + let (vendor, state) = classify( + &[], + "gcp.vertex.agent", + "invocation", + &[ + kv("gen_ai.system", "gcp.vertex.agent"), + kv("gcp.vertex.agent.session_id", "s-1"), + ], + ); + assert_eq!(vendor, "google_adk"); + assert_eq!(state, session_state::SESSION); + + // The mirror image: only candidate 1 is authoritative here. + let (_, other_population) = classify( + &[], + "gcp.vertex.agent", + "invoke_agent weather", + &[ + kv("gen_ai.operation.name", "invoke_agent"), + kv("gen_ai.conversation.id", "c-9"), + ], + ); + assert_eq!(other_population, session_state::SESSION); + } + + /// flue: a non-authoritative span still resolves at instance granularity + /// (state 5) rather than being reported unauthoritative. + #[test] + fn a_non_authoritative_span_still_resolves_at_instance_granularity() { + let (vendor, state) = classify( + &[], + "@flue/opentelemetry", + "flue.tool", + &[ + kv("flue.operation.kind", "tool"), + kv("flue.instance.id", "inst-7"), + ], + ); + assert_eq!(vendor, "flue"); + assert_eq!(state, session_state::SUB_SESSION); + + let (_, authoritative) = classify( + &[], + "@flue/opentelemetry", + "flue.prompt", + &[ + kv("flue.operation.kind", "prompt"), + kv("gen_ai.conversation.id", "conv-3"), + kv("flue.instance.id", "inst-7"), + ], + ); + assert_eq!(authoritative, session_state::SESSION); + } + + /// pydantic_ai's `gen_ai.conversation.id` is always present and often per-run. + /// + /// The write-side plan predicted this would be capped at state 5. The seed + /// overrode that with an explicit, documented trade-off: it labels the key + /// `session` granularity because a correctly-configured deployment + /// (`conversation_id=` passed, or `message_history` threaded) produces a genuine + /// cross-run session, and labelling it `run` would drive those deployments to + /// state 5 and 8/8 unsessioned traces. The cost — a default-configured app + /// reports one session per run and maple cannot tell the two apart span-locally + /// — is recorded in the seed's caveats. This test pins the *registry's* behavior. + #[test] + fn pydantic_ai_conversation_id_resolves_at_session_granularity() { + let (vendor, state) = classify( + &[], + "pydantic-ai", + "agent run", + &[ + kv("gen_ai.operation.name", "invoke_agent"), + kv("gen_ai.conversation.id", "run-1"), + kv("pydantic_ai.all_messages", "[]"), + ], + ); + assert_eq!(vendor, "pydantic_ai"); + assert_eq!(state, session_state::SESSION); + + // Its run-granularity sibling on its own tops out at 5. + let (_, run_only) = classify( + &[], + "pydantic-ai", + "agent run", + &[ + kv("gen_ai.operation.name", "invoke_agent"), + kv("gen_ai.agent.call.id", "call-1"), + ], + ); + assert_eq!(run_only, session_state::SUB_SESSION); + } + + #[test] + fn present_but_empty_is_not_absent() { + // claude_agent_sdk validates `session.id` as non-empty. + // Its candidates are authoritative wherever `span.type` is present. + let (_, empty) = classify( + &[], + "com.anthropic.claude_code", + "claude_code.interaction", + &[kv("span.type", "interaction"), kv("session.id", "")], + ); + assert_eq!(empty, session_state::KEY_INVALID); + + let (_, absent) = classify( + &[], + "com.anthropic.claude_code", + "claude_code.interaction", + &[kv("span.type", "interaction")], + ); + assert_eq!(absent, session_state::KEY_ABSENT); + + let (_, present) = classify( + &[], + "com.anthropic.claude_code", + "claude_code.interaction", + &[kv("span.type", "interaction"), kv("session.id", "abc")], + ); + assert_eq!(present, session_state::SESSION); + } + + #[test] + fn decoy_values_fail_validation() { + // litellm's candidate rejects the decoy value; effect_ai's `undefined` + // decoy is the wire-observed one. + let registry = registry(); + let litellm = registry.vendor_id("litellm").expect("vendor"); + assert!(!registry.vendor(litellm).candidates().is_empty()); + let candidate = ®istry.vendor(litellm).candidates()[0]; + assert!(candidate.reject_decoy_values || candidate.require_non_empty); + } + + #[test] + fn non_ai_spans_still_carry_the_rules_version() { + let registry = registry(); + let resource_context = ResourceContext::new(registry, &[]); + let scope = scope("io.opentelemetry.http"); + let context = resource_context.scope(Some(&scope), ""); + let attributes = [kv("http.request.method", "GET")]; + let result = context.classify_span("GET /health", &attributes); + assert_eq!(result.vendor, None); + assert_eq!(result.vendor_slug(), ""); + assert_ne!(result.rules_version, 0); + assert_eq!(result.session_key_hash(), 0); + } + + #[test] + fn resolved_keys_hash_the_bare_value() { + let registry = registry(); + let resource_context = ResourceContext::new(registry, &[]); + let scope = scope("com.anthropic.claude_code"); + let context = resource_context.scope(Some(&scope), ""); + let attributes = [kv("span.type", "interaction"), kv("session.id", "sess-42")]; + let result = context.classify_span("claude_code.interaction", &attributes); + assert_eq!(result.session_state, session_state::SESSION); + // Exactly what `SELECT cityHash64('sess-42')` returns — the SQL leg of the + // equivalence suite reproduces this with no construction to agree on. + assert_eq!( + result.session_key_hash(), + crate::cityhash102::city_hash64(b"sess-42") + ); + } + + // -- canonicalization and degradation ------------------------------------ + + #[test] + fn values_are_canonicalized_before_matching() { + // A bool/int arrives typed on protobuf and as a string over JSON; both must + // match `eq(langsmith.internal_provider, "true")`. + let registry = registry(); + for value in [ + KeyValue { + key: "langsmith.internal_provider".into(), + value: Some(AnyValue { + value: Some(any_value::Value::BoolValue(true)), + }), + }, + kv("langsmith.internal_provider", "true"), + ] { + let resource = [value]; + let resource_context = ResourceContext::new(registry, &resource); + let scope = scope("app"); + let context = resource_context.scope(Some(&scope), ""); + let attributes = [kv("langsmith.trace.name", "x")]; + let result = context.classify_span("chain", &attributes); + assert_eq!(result.vendor_slug(), "langchain"); + } + } + + /// Plan §2: the same logical span delivered as OTLP protobuf and as OTLP/JSON + /// must classify identically. JSON carries int64s as decimal strings and the + /// span/trace ids as hex, and the crate's leniency pass normalizes both — the + /// classifier must not be able to tell which transport it came from. + #[test] + fn transports_agree() { + use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; + use opentelemetry_proto::tonic::resource::v1::Resource; + use opentelemetry_proto::tonic::trace::v1::{ResourceSpans, ScopeSpans, Span}; + + let protobuf = ExportTraceServiceRequest { + resource_spans: vec![ResourceSpans { + resource: Some(Resource { + attributes: vec![kv("service.name", "adk-app")], + ..Default::default() + }), + scope_spans: vec![ScopeSpans { + scope: Some(scope("gcp.vertex.agent")), + spans: vec![Span { + name: "invocation".to_string(), + start_time_unix_nano: 1_700_000_000_000_000_000, + end_time_unix_nano: 1_700_000_000_500_000_000, + attributes: vec![ + kv("gen_ai.system", "gcp.vertex.agent"), + kv("gcp.vertex.agent.session_id", "s-1"), + int_kv("gen_ai.usage.input_tokens", 4_294_967_296), + ], + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }], + }; + + // The OTLP/JSON form of exactly that span, as an exporter would send it. + let json = r#"{"resourceSpans":[{"resource":{"attributes":[ + {"key":"service.name","value":{"stringValue":"adk-app"}}]}, + "scopeSpans":[{"scope":{"name":"gcp.vertex.agent"},"spans":[{ + "name":"invocation", + "traceId":"5b8efff798038103d269b633813fc60c","spanId":"eee19b7ec3c1b174", + "startTimeUnixNano":"1700000000000000000","endTimeUnixNano":"1700000000500000000", + "attributes":[ + {"key":"gen_ai.system","value":{"stringValue":"gcp.vertex.agent"}}, + {"key":"gcp.vertex.agent.session_id","value":{"stringValue":"s-1"}}, + {"key":"gen_ai.usage.input_tokens","value":{"intValue":"4294967296"}}]}]}]}]}"#; + let mut value: serde_json::Value = serde_json::from_str(json).expect("json"); + crate::otlp_json::normalize(&mut value, "resourceSpans"); + let decoded: ExportTraceServiceRequest = + serde_json::from_value(value).expect("OTLP/JSON decodes"); + + let classify_request = |request: &ExportTraceServiceRequest| { + let registry = registry(); + let resource_spans = &request.resource_spans[0]; + let attributes = &resource_spans + .resource + .as_ref() + .expect("resource") + .attributes; + let context = ResourceContext::new(registry, attributes); + let scope_spans = &resource_spans.scope_spans[0]; + let scope = context.scope(scope_spans.scope.as_ref(), &scope_spans.schema_url); + let span = &scope_spans.spans[0]; + let classified = scope.classify_span(&span.name, &span.attributes); + ( + classified.vendor_slug(), + classified.session_state, + classified.session_key_hash(), + ) + }; + + let from_protobuf = classify_request(&protobuf); + assert_eq!(from_protobuf, classify_request(&decoded)); + assert_eq!(from_protobuf.0, "google_adk"); + assert_eq!(from_protobuf.1, session_state::SESSION); + assert_ne!(from_protobuf.2, 0); + } + + #[test] + fn garbage_attributes_do_not_panic() { + let huge = "x".repeat(1 << 20); + let attributes = vec![ + kv("", ""), + kv("\u{0}\u{1}\u{2}", "\u{0}"), + kv("🌍.emoji.key", "🙂"), + kv("gen_ai.operation.name", &huge), + kv(&huge, "v"), + int_kv("llm.token_count.total", i64::MIN), + KeyValue { + key: "spring.ai.kind".into(), + value: None, + }, + KeyValue { + key: "session.id".into(), + value: Some(AnyValue { value: None }), + }, + ]; + let (vendor, _) = classify(&[], "org.springframework.boot", &huge, &attributes); + assert_eq!(vendor, "spring_ai"); + } + + #[test] + fn attribute_order_does_not_change_the_outcome() { + let noise: Vec = (0..40) + .map(|i| kv(&format!("http.header.x_{i}"), "v")) + .collect(); + let signal = vec![ + kv("spring.ai.kind", "chat_client"), + kv("gen_ai.operation.name", "framework"), + ]; + let mut forward = signal.clone(); + forward.extend(noise.iter().cloned()); + let mut backward = noise; + backward.reverse(); + backward.extend(signal); + + let a = classify(&[], "org.springframework.boot", "chat_client", &forward); + let b = classify(&[], "org.springframework.boot", "chat_client", &backward); + assert_eq!(a, b); + assert_eq!(a.0, "spring_ai"); + } + + /// The indexed matcher dispatch must agree with direct evaluation of every + /// matcher in the registry — the optimization's safety net. + #[test] + fn the_indexed_path_agrees_with_direct_evaluation() { + let registry = registry(); + let cases: Vec<(Vec, &str, &str, Vec)> = vec![ + ( + vec![], + "org.springframework.boot", + "POST", + vec![kv("method", "POST")], + ), + ( + vec![kv("telemetry.sdk.name", "@mastra/otel-exporter")], + "@mastra/otel-exporter", + "agent.run", + vec![kv("mastra.agent.id", "a")], + ), + ( + vec![kv("model_id", "x")], + "litellm", + "litellm_request", + vec![kv("litellm.call_id", "c")], + ), + ( + vec![], + "app", + "LanguageModel.generateText", + vec![kv("gen_ai.system", "x")], + ), + ( + vec![], + "gen_ai", + "step", + vec![kv("gen_ai.execute_tool.duration", "1")], + ), + ( + vec![], + "app", + "x", + vec![kv("openinference.span.kind", "LLM")], + ), + ]; + for (resource, scope_name, span_name, attributes) in cases { + let resource_context = ResourceContext::new(registry, &resource); + let scope_value = scope(scope_name); + let context = resource_context.scope(Some(&scope_value), ""); + let indexed = context.classify_span(span_name, &attributes); + let direct = context.classify_span_unindexed(span_name, &attributes); + assert_eq!( + indexed.vendor, direct, + "indexed vs direct for {scope_name}/{span_name}" + ); + } + } +} diff --git a/apps/ingest/src/ai_classifier_corpus_test.rs b/apps/ingest/src/ai_classifier_corpus_test.rs new file mode 100644 index 000000000..95631f49c --- /dev/null +++ b/apps/ingest/src/ai_classifier_corpus_test.rs @@ -0,0 +1,441 @@ +//! Corpus replay — the acceptance gate for the classifier. +//! +//! Replays every capture in the trace-capture corpus (raw OTLP/JSON export requests +//! as they arrived on the wire) through the crate's own OTLP/JSON decode path and +//! the classifier, and compares the result against the hand-reviewed goldens in +//! `frameworks//registry-seed.yaml`. +//! +//! Run it with the corpus checked out next door: +//! +//! ```sh +//! TRACE_CAPTURE_DIR=~/Documents/repos/trace-capture cargo test --lib corpus +//! ``` +//! +//! Unset, every test here skips cleanly — the corpus is a sibling repo, not a +//! vendored fixture (the plan says to vendor it eventually; until then CI without +//! the corpus still runs the unit invariants). +//! +//! # How the goldens map onto a multi-vendor registry +//! +//! `verify-seed.ts` evaluates ONE seed's rules and separates *harness*-emitted spans +//! (fixture scaffolding) from the framework's own. The compiled registry has no +//! harness concept and all 21 vendors at once, so the identities checked here are: +//! +//! * **vendor span count** = `vendor_span_counts[] + harness_matched`. +//! The seed's hit set is `sufficient-hit OR attr-hit`, which is exactly the +//! registry's hit set for that vendor (a promoted conditional candidate requires a +//! same-vendor attr hit, which is already a hit). The counts can therefore only +//! diverge if another vendor outranks this one on some span — and the compile's +//! cross-fire audit found no same-span inter-vendor conflicts in the corpus, so any +//! divergence is a real finding, not an expected artifact. +//! * **key-state histogram / unsessioned traces** are the seed's per-vendor view of +//! *every* span in the capture regardless of which vendor classified it, so they +//! are reproduced through `evaluate_session_for_vendor`. +//! +//! D2 (`langgraph` → `langchain`) is applied when looking the vendor up. + +use std::collections::{BTreeMap, HashMap}; +use std::path::{Path, PathBuf}; + +use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; + +use crate::ai_classifier::{session_state, ResourceContext}; +use crate::ai_registry::registry; +use crate::otlp_json; + +/// Seed vendor slug → compiled registry slug (write-side plan D2). +fn registry_slug(seed_vendor: &str) -> &str { + match seed_vendor { + "langgraph" => "langchain", + other => other, + } +} + +fn corpus_root() -> Option { + let root = PathBuf::from(std::env::var_os("TRACE_CAPTURE_DIR")?); + if root.join("captures").is_dir() && root.join("frameworks").is_dir() { + Some(root) + } else { + panic!("TRACE_CAPTURE_DIR={root:?} has no captures/ and frameworks/"); + } +} + +/// One capture's spans, decoded exactly as ingest would decode them. +struct Capture { + dir: String, + /// (resource attrs, scope, schema_url, span) grouped as they arrived. + requests: Vec, +} + +impl Capture { + fn load(path: &Path) -> Self { + let dir = path + .file_name() + .expect("capture dir") + .to_string_lossy() + .into_owned(); + let records = std::fs::read_to_string(path.join("records.jsonl")).expect("records.jsonl"); + let mut requests = Vec::new(); + for line in records.lines() { + if line.trim().is_empty() { + continue; + } + let record: serde_json::Value = serde_json::from_str(line).expect("record json"); + if record.get("signal").and_then(|s| s.as_str()) != Some("traces") { + continue; + } + let Some(mut body) = record.get("body").cloned().filter(|b| !b.is_null()) else { + continue; // undecodable capture record (`rawBase64` only) + }; + // The crate's own OTLP/JSON leniency pass, then the generated types — + // the same two steps `decode_and_enrich_payload` runs on the wire. + otlp_json::normalize(&mut body, "resourceSpans"); + let request: ExportTraceServiceRequest = + serde_json::from_value(body).expect("OTLP/JSON decodes"); + requests.push(request); + } + Capture { dir, requests } + } + + fn span_count(&self) -> usize { + self.requests + .iter() + .flat_map(|r| &r.resource_spans) + .flat_map(|rs| &rs.scope_spans) + .map(|ss| ss.spans.len()) + .sum() + } +} + +/// What one capture produces under the full registry. +#[derive(Default)] +struct CaptureResult { + /// vendor slug (or `""` for non-AI) → span count + vendors: BTreeMap, + /// state (under the *classified* vendor) → span count + states: BTreeMap, + hashes: usize, +} + +fn classify_capture(capture: &Capture) -> CaptureResult { + let registry = registry(); + let mut result = CaptureResult::default(); + for request in &capture.requests { + for resource_spans in &request.resource_spans { + let empty = Vec::new(); + let attributes = resource_spans + .resource + .as_ref() + .map(|r| &r.attributes) + .unwrap_or(&empty); + let resource_context = ResourceContext::new(registry, attributes); + for scope_spans in &resource_spans.scope_spans { + let schema_url = if scope_spans.schema_url.is_empty() { + resource_spans.schema_url.as_str() + } else { + scope_spans.schema_url.as_str() + }; + let scope = resource_context.scope(scope_spans.scope.as_ref(), schema_url); + for span in &scope_spans.spans { + let classified = scope.classify_span(&span.name, &span.attributes); + assert_eq!(classified.rules_version, registry.version()); + *result + .vendors + .entry(classified.vendor_slug().to_string()) + .or_default() += 1; + *result.states.entry(classified.session_state).or_default() += 1; + if classified.session_state >= session_state::SUB_SESSION { + assert_ne!(classified.session_key_hash(), 0); + result.hashes += 1; + } else { + assert_eq!(classified.session_key_hash(), 0); + } + } + } + } + } + result +} + +/// Per-vendor state histogram over EVERY span of the capture — the seed's view. +fn seed_state_view(capture: &Capture, vendor_slug: &str) -> (BTreeMap, usize) { + let registry = registry(); + let vendor = registry + .vendor_id(vendor_slug) + .unwrap_or_else(|| panic!("registry has no vendor {vendor_slug}")); + let mut histogram: BTreeMap = BTreeMap::new(); + // trace id → does any span of it reach state 6 + let mut traces: HashMap, bool> = HashMap::new(); + for request in &capture.requests { + for resource_spans in &request.resource_spans { + let empty = Vec::new(); + let attributes = resource_spans + .resource + .as_ref() + .map(|r| &r.attributes) + .unwrap_or(&empty); + let resource_context = ResourceContext::new(registry, attributes); + for scope_spans in &resource_spans.scope_spans { + let schema_url = if scope_spans.schema_url.is_empty() { + resource_spans.schema_url.as_str() + } else { + scope_spans.schema_url.as_str() + }; + let scope = resource_context.scope(scope_spans.scope.as_ref(), schema_url); + for span in &scope_spans.spans { + let (state, _) = + scope.evaluate_session_for_vendor(vendor, &span.name, &span.attributes); + *histogram.entry(state).or_default() += 1; + let sessioned = traces.entry(span.trace_id.clone()).or_insert(false); + *sessioned |= state == session_state::SESSION; + } + } + } + } + let unsessioned = traces.values().filter(|sessioned| !**sessioned).count(); + (histogram, unsessioned) +} + +// --------------------------------------------------------------------------- +// goldens +// --------------------------------------------------------------------------- + +struct Golden { + framework: String, + vendor: String, + capture: String, + total_spans: usize, + vendor_spans: usize, + key_state_histogram: BTreeMap, + unsessioned_traces: usize, +} + +fn number(value: &serde_yaml::Value) -> usize { + value + .as_u64() + .unwrap_or_else(|| panic!("expected a number, got {value:?}")) as usize +} + +fn load_goldens(root: &Path) -> Vec { + let mut goldens = Vec::new(); + let mut frameworks: Vec<_> = std::fs::read_dir(root.join("frameworks")) + .expect("frameworks/") + .filter_map(|entry| entry.ok()) + .map(|entry| entry.path()) + .filter(|path| path.join("registry-seed.yaml").is_file()) + .collect(); + frameworks.sort(); + for framework in frameworks { + let text = std::fs::read_to_string(framework.join("registry-seed.yaml")).expect("seed"); + let seed: serde_yaml::Value = serde_yaml::from_str(&text).expect("seed yaml"); + let seed_vendor = seed["vendor"].as_str().expect("vendor").to_string(); + let per_capture = seed["goldens"]["per_capture"] + .as_sequence() + .expect("goldens.per_capture"); + for entry in per_capture { + let capture = entry["capture"].as_str().expect("capture").to_string(); + let counts = &entry["vendor_span_counts"]; + let vendor_spans = + number(&counts[seed_vendor.as_str()]) + number(&entry["harness_matched"]); + let mut histogram = BTreeMap::new(); + for (state, count) in entry["key_state_histogram"] + .as_mapping() + .expect("key_state_histogram") + { + histogram.insert(number(state) as u8, number(count)); + } + goldens.push(Golden { + framework: framework + .file_name() + .expect("framework") + .to_string_lossy() + .into_owned(), + vendor: registry_slug(&seed_vendor).to_string(), + capture, + total_spans: number(&entry["total_spans"]), + vendor_spans, + key_state_histogram: histogram, + unsessioned_traces: number(&entry["unsessioned_traces"]), + }); + } + } + goldens +} + +// --------------------------------------------------------------------------- +// tests +// --------------------------------------------------------------------------- + +/// The gate: every golden in all 20 seeds reproduces. +#[test] +fn corpus_replay_matches_every_golden() { + let Some(root) = corpus_root() else { + eprintln!("skipping: set TRACE_CAPTURE_DIR to the trace-capture checkout"); + return; + }; + let goldens = load_goldens(&root); + assert!(goldens.len() >= 40, "expected the full seed set"); + + let mut cache: HashMap = HashMap::new(); + let mut failures = Vec::new(); + println!( + "{:<34} {:<26} {:>7} {:>7} {:>6}", + "capture", "vendor", "spans", "golden", "ok" + ); + for golden in &goldens { + let entry = cache.entry(golden.capture.clone()).or_insert_with(|| { + let capture = Capture::load(&root.join("captures").join(&golden.capture)); + let result = classify_capture(&capture); + (capture, result) + }); + let (capture, result) = entry; + + let mut problems = Vec::new(); + if capture.span_count() != golden.total_spans { + problems.push(format!( + "span count {} != golden {}", + capture.span_count(), + golden.total_spans + )); + } + let classified = result.vendors.get(&golden.vendor).copied().unwrap_or(0); + if classified != golden.vendor_spans { + problems.push(format!( + "vendor spans {classified} != golden {} (all vendors: {:?})", + golden.vendor_spans, result.vendors + )); + } + let (histogram, unsessioned) = seed_state_view(capture, &golden.vendor); + if histogram != golden.key_state_histogram { + problems.push(format!( + "key_state_histogram {histogram:?} != golden {:?}", + golden.key_state_histogram + )); + } + if unsessioned != golden.unsessioned_traces { + problems.push(format!( + "unsessioned traces {unsessioned} != golden {}", + golden.unsessioned_traces + )); + } + + println!( + "{:<34} {:<26} {:>7} {:>7} {:>6}", + golden.capture, + golden.vendor, + classified, + golden.vendor_spans, + if problems.is_empty() { "PASS" } else { "FAIL" } + ); + if !problems.is_empty() { + failures.push(format!( + "{}/{}: {}", + golden.framework, + golden.capture, + problems.join("; ") + )); + } + } + let total_spans: usize = cache.values().map(|(c, _)| c.span_count()).sum(); + println!( + "\n{} goldens over {} captures, {total_spans} spans", + goldens.len(), + cache.len() + ); + assert!( + failures.is_empty(), + "golden mismatches:\n{}", + failures.join("\n") + ); +} + +/// Every capture in the corpus — including the ones no seed claims (probes, smoke +/// tests, the two large `eve_slack` negative sets) — must decode and classify +/// without panicking, and every examined span must carry the rules version. +#[test] +fn corpus_replay_never_panics_and_always_stamps_the_version() { + let Some(root) = corpus_root() else { + eprintln!("skipping: set TRACE_CAPTURE_DIR to the trace-capture checkout"); + return; + }; + let mut dirs: Vec = std::fs::read_dir(root.join("captures")) + .expect("captures/") + .filter_map(|entry| entry.ok()) + .map(|entry| entry.path()) + .filter(|path| path.join("records.jsonl").is_file()) + .collect(); + dirs.sort(); + assert!(dirs.len() >= 40, "expected the full corpus"); + + let mut totals: BTreeMap = BTreeMap::new(); + let mut spans = 0usize; + for dir in &dirs { + let capture = Capture::load(dir); + let result = classify_capture(&capture); + spans += capture.span_count(); + println!("{:<34} {:?}", capture.dir, result.vendors); + for (vendor, count) in result.vendors { + *totals.entry(vendor).or_default() += count; + } + } + println!( + "\n{} captures, {spans} spans\ncorpus totals: {totals:?}", + dirs.len() + ); + let non_ai = totals.get("").copied().unwrap_or(0); + assert!( + spans > 0 && non_ai > 0, + "corpus must contain both AI and non-AI spans" + ); +} + +/// Order independence over real data: shuffling the non-registry attributes of every +/// corpus span must not move a single classification. +#[test] +fn corpus_classification_is_order_independent() { + let Some(root) = corpus_root() else { + eprintln!("skipping: set TRACE_CAPTURE_DIR to the trace-capture checkout"); + return; + }; + let registry = registry(); + let mut checked = 0usize; + for dir in ["spring_ai_user", "crewai_user", "flue_user", "eve_slack"] { + let capture = Capture::load(&root.join("captures").join(dir)); + for request in &capture.requests { + for resource_spans in &request.resource_spans { + let empty = Vec::new(); + let attributes = resource_spans + .resource + .as_ref() + .map(|r| &r.attributes) + .unwrap_or(&empty); + let context = ResourceContext::new(registry, attributes); + for scope_spans in &resource_spans.scope_spans { + let scope = context.scope(scope_spans.scope.as_ref(), &scope_spans.schema_url); + for span in &scope_spans.spans { + let straight = scope.classify_span(&span.name, &span.attributes); + // Reverse the order of the keys no rule consults; the + // registry-referenced ones keep their relative order because + // first-occurrence-wins is order-*dependent* by design. + let (referenced, rest): (Vec<_>, Vec<_>) = span + .attributes + .iter() + .cloned() + .partition(|kv| registry.references_key(&kv.key)); + let mut shuffled: Vec<_> = rest.into_iter().rev().collect(); + shuffled.extend(referenced); + let reordered = scope.classify_span(&span.name, &shuffled); + assert_eq!( + straight.vendor, reordered.vendor, + "{dir}: {} reordered differently", + span.name + ); + assert_eq!(straight.session_state, reordered.session_state); + checked += 1; + } + } + } + } + } + println!("{checked} spans checked for order independence"); +} diff --git a/apps/ingest/src/ai_registry.rs b/apps/ingest/src/ai_registry.rs new file mode 100644 index 000000000..61036bc71 --- /dev/null +++ b/apps/ingest/src/ai_registry.rs @@ -0,0 +1,1125 @@ +//! The compiled AI vendor rule registry, embedded and validated at first use. +//! +//! Source of truth: `packages/domain/src/ai-registry/registry.json`, vendored from +//! trace-capture's `scripts/compile-registry.ts` (20 framework seeds + 1 synthesized +//! vendor). It is `include_str!`'d — the detector must never depend on a file being +//! present at runtime, and a batch must carry exactly one `rules_version`. +//! +//! What this module guarantees, structurally rather than by convention: +//! +//! * **Vendor slugs are a closed set.** A classification result is a [`VendorId`], +//! an index into an interned table built from the registry. There is no +//! constructor that takes a string, so minting an unlisted slug — the +//! LowCardinality-cardinality and tenant-amplification bug the plan calls out — +//! is not expressible. +//! * **The predicate algebra is exactly four operators.** `op` is a serde tag, so a +//! fifth operator is a parse error, not a silently-ignored rule. +//! * **The prefilter is generated from the registry**, never hand-maintained: it is +//! the union of every exact key and every key-prefix any matcher, unknown-tier +//! fingerprint, session candidate or authority predicate references. +//! +//! Everything else is checked in [`Registry::validate`], which panics. That is the +//! right failure mode: the registry is boot configuration compiled into the binary, +//! so a violation is a build artifact bug that must not reach production traffic, +//! and the process has no traffic to lose at that point. + +use std::borrow::Cow; +use std::collections::HashMap; +use std::hash::{BuildHasherDefault, Hasher}; +use std::sync::LazyLock; + +use serde::Deserialize; + +/// The vendored registry artifact. Path is relative to this source file. +const REGISTRY_JSON: &str = include_str!("../../../packages/domain/src/ai-registry/registry.json"); + +/// The exact-key map, keyed by a **non-cryptographic** hasher. +/// +/// SipHash (std's default) cost ~12 ns per probed key, and a fat AI span probes +/// dozens — it was the single largest line in the per-span budget. Dropping DoS +/// resistance here is bounded and deliberate: the key *set* is the compiled +/// registry, fixed at load and never influenced by traffic, so an attacker can +/// only choose lookup keys, not the buckets they collide into. The worst case a +/// crafted attribute name can buy is a short bucket walk over a ~116-entry map +/// that it still misses — no unbounded chain, no insertion, no eviction. Every +/// other map in this module keeps the std hasher; only this one is hot. +type KeyIdMap = HashMap, KeyId, BuildHasherDefault>; + +/// FxHash (rustc's own string hasher), inlined rather than pulled in as a +/// dependency — it is eleven lines and this is its only use. +#[derive(Default)] +struct FxHasher { + hash: u64, +} + +impl FxHasher { + const SEED: u64 = 0x51_7c_c1_b7_27_22_0a_95; + + #[inline] + fn add(&mut self, word: u64) { + self.hash = (self.hash.rotate_left(5) ^ word).wrapping_mul(Self::SEED); + } +} + +impl Hasher for FxHasher { + #[inline] + fn write(&mut self, bytes: &[u8]) { + let mut rest = bytes; + while rest.len() >= 8 { + let (head, tail) = rest.split_at(8); + self.add(u64::from_le_bytes(head.try_into().expect("8 bytes"))); + rest = tail; + } + if !rest.is_empty() { + let mut buf = [0u8; 8]; + buf[..rest.len()].copy_from_slice(rest); + self.add(u64::from_le_bytes(buf)); + } + } + + #[inline] + fn finish(&self) -> u64 { + self.hash + } +} + +/// The `key_len_screen` bit for a key of this byte length, clamped so keys +/// longer than 63 bytes all share the top bucket. +#[inline] +fn length_bit(len: usize) -> u64 { + 1u64 << len.min(63) +} + +/// Priority bands (write-side plan D4). Sufficient scope/resource matchers outrank +/// vendor attr matchers, which outrank unknown-tier fingerprints. +const BAND_SUFFICIENT: std::ops::RangeInclusive = 30_000..=39_999; +const BAND_VENDOR: std::ops::RangeInclusive = 20_000..=29_999; +const BAND_UNKNOWN: std::ops::RangeInclusive = 10_000..=19_999; + +/// `VendorBits` is a `u64`; more vendors than that needs a wider set, not a silent +/// truncation. +const MAX_VENDORS: usize = 64; +/// Prefix hits are tracked in a `u64` bitmask for the same reason. +const MAX_PREFIXES: usize = 64; +/// Conditional-candidate hits are tracked in a `u32` bitmask per span. +const MAX_CONDITIONAL: usize = 32; +/// The classifier's per-attribute-list key-presence bitset is a fixed +/// `[u64; MAX_KEYS / 64]`. A multiple of 64; grow it (and nothing else) when the +/// registry outgrows it — 116 keys today. +pub(crate) const MAX_KEYS: usize = 256; + +// --------------------------------------------------------------------------- +// raw (serde) shapes — mirror registry.json exactly +// --------------------------------------------------------------------------- + +#[derive(Deserialize)] +struct RawRegistry { + registry_version: u32, + algebra: RawAlgebra, + unknown_tier: Vec, + vendors: Vec, +} + +#[derive(Deserialize)] +struct RawAlgebra { + ops: Vec, + value_prefix_pseudo_keys: Vec, +} + +#[derive(Deserialize)] +struct RawUnknown { + bucket: String, + predicate: RawPredicate, + priority: u32, +} + +#[derive(Deserialize)] +struct RawVendor { + vendor: String, + matchers: Vec, + session_candidates: Vec, + #[serde(default)] + decoy_values: Vec, +} + +#[derive(Deserialize)] +struct RawMatcher { + class: RawClass, + sufficient: bool, + predicate: RawPredicate, + priority: u32, +} + +#[derive(Deserialize, Clone, Copy, PartialEq, Eq, Debug)] +#[serde(rename_all = "snake_case")] +enum RawClass { + Resource, + Scope, + Attr, +} + +/// The restricted algebra. An unknown `op` fails deserialization. +#[derive(Deserialize, Clone)] +#[serde(tag = "op", rename_all = "snake_case")] +enum RawPredicate { + Present { key: String }, + Eq { key: String, value: String }, + KeyPrefix { prefix: String }, + ValuePrefix { key: String, prefix: String }, +} + +#[derive(Deserialize)] +#[serde(untagged)] +enum RawAuthority { + AnyOf { any_of: Vec }, + One(RawPredicate), +} + +#[derive(Deserialize)] +struct RawCandidate { + key: String, + authority_predicate: Option, + #[serde(default)] + validation: Vec, + granularity: RawGranularity, +} + +#[derive(Deserialize, Clone, Copy, PartialEq, Eq, Debug)] +#[serde(rename_all = "snake_case")] +enum RawGranularity { + Session, + Run, + User, + Instance, +} + +#[derive(Deserialize)] +struct RawDecoy { + value: String, +} + +// --------------------------------------------------------------------------- +// compiled shapes +// --------------------------------------------------------------------------- + +/// Index into the registry's interned vendor table. The only way to name a vendor. +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug)] +pub struct VendorId(u16); + +impl VendorId { + pub fn index(self) -> usize { + self.0 as usize + } +} + +/// Index into the interned exact-key table. +pub type KeyId = u16; +/// Index into the interned key-prefix table. +pub type PrefixId = u8; +/// Index into [`Registry::matchers`]. +pub type MatcherId = u16; + +/// The four pseudo-keys: real columns on both targets, resolved before attributes. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum PseudoKey { + ScopeName, + SpanName, + ScopeVersion, + ScopeSchemaUrl, +} + +impl PseudoKey { + fn parse(key: &str) -> Option { + match key { + "scope.name" => Some(Self::ScopeName), + "span.name" => Some(Self::SpanName), + "scope.version" => Some(Self::ScopeVersion), + "scope.schema_url" => Some(Self::ScopeSchemaUrl), + _ => None, + } + } + + pub fn as_str(self) -> &'static str { + match self { + Self::ScopeName => "scope.name", + Self::SpanName => "span.name", + Self::ScopeVersion => "scope.version", + Self::ScopeSchemaUrl => "scope.schema_url", + } + } +} + +/// A key in a predicate: either a pseudo-key or an interned attribute key. +/// +/// The distinction is load-bearing twice over: pseudo-keys resolve to a *column*, so a +/// span attribute literally named `scope.name` is shadowed and must never reach a +/// pseudo-key matcher — and a pseudo-key ignores [`AttrTarget`] entirely, because a +/// column belongs to no attribute map (`pseudoKeyColumn` wins over `targetForClass` in +/// `compile-sql.ts` for the same reason). +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum LookupKey { + Pseudo(PseudoKey), + Attr(KeyId), +} + +#[derive(Clone, Debug)] +pub enum Predicate { + Present(LookupKey), + Eq(LookupKey, Box), + KeyPrefix(PrefixId), + ValuePrefix(PseudoKey, Box), +} + +/// Which attribute list a matcher's *attribute* keys resolve against — the matcher's +/// declared class, compiled (plan §1: matcher classes are per-class predicates). +/// +/// `resource` → the ResourceSpans attributes, `scope` → the InstrumentationScope +/// attributes, `attr` → the span's own attributes. Unknown-tier fingerprints, session +/// candidates and authority predicates carry no class and are span-local, matching +/// `targetForClass` / the `"span"` target in `packages/domain/src/ai-registry/compile-sql.ts`. +/// +/// Pseudo-keys are exempt: they are real columns, not map entries, so `span.name` and +/// `scope.*` resolve to their column whatever the matcher's class says. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum AttrTarget { + Span, + Scope, + Resource, +} + +/// How a matcher participates in resolution (write-side plan §1). +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum HitKind { + /// A hit on its own: attr matchers, sufficient resource/scope matchers, and + /// unknown-tier fingerprints. + Unconditional, + /// An insufficient resource/scope match. Contributes nothing alone; promoted to + /// a hit at its own priority only by a same-vendor attr hit on the same span. + Conditional, +} + +#[derive(Clone, Debug)] +pub struct Matcher { + pub vendor: VendorId, + pub priority: u32, + pub kind: HitKind, + /// The one attribute list this matcher's attribute keys may read. + pub target: AttrTarget, + /// Attr-class matchers promote their vendor's conditional candidates. Sufficient + /// scope/resource matchers and unknown-tier fingerprints do not. + pub promotes: bool, + /// Slot in [`Registry::conditional`], so a span can track which conditional + /// candidates it hit in a `u32` bitmask instead of a heap list. + pub conditional_slot: Option, + pub predicate: Predicate, +} + +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum Granularity { + Session, + Run, + User, + Instance, +} + +impl Granularity { + /// State 6 is reserved for `session` granularity; `run`/`user`/`instance` + /// resolve at state 5 (plan §2 step 5). + pub fn resolved_state(self) -> u8 { + match self { + Self::Session => 6, + _ => 5, + } + } +} + +#[derive(Clone, Debug)] +pub enum Authority { + /// No authority predicate — every span of the vendor is authoritative. + Always, + One(Predicate), + AnyOf(Vec), +} + +#[derive(Clone, Debug)] +pub struct SessionCandidate { + pub key: LookupKey, + pub authority: Authority, + pub require_non_empty: bool, + pub reject_decoy_values: bool, + pub granularity: Granularity, +} + +#[derive(Debug)] +pub struct Vendor { + slug: Box, + /// `unknown:*` buckets are vendors for resolution purposes but carry no + /// session-key rules and are reserved: no seed may mint one. + unknown_bucket: bool, + candidates: Vec, + decoy_values: Vec>, +} + +impl Vendor { + pub fn slug(&self) -> &str { + &self.slug + } + pub fn is_unknown_bucket(&self) -> bool { + self.unknown_bucket + } + pub fn candidates(&self) -> &[SessionCandidate] { + &self.candidates + } + pub fn is_decoy_value(&self, value: &str) -> bool { + self.decoy_values.iter().any(|d| &**d == value) + } +} + +/// A registry-referenced key seen on a span: its interned id (when the key is +/// referenced exactly) and the prefixes it satisfies. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct KeyProbe { + pub key_id: Option, + pub prefix_bits: u64, +} + +impl KeyProbe { + pub fn is_empty(&self) -> bool { + self.key_id.is_none() && self.prefix_bits == 0 + } +} + +#[derive(Debug)] +pub struct Registry { + version: u32, + vendors: Vec, + vendor_ids: HashMap, VendorId>, + matchers: Vec, + + keys: Vec>, + key_ids: KeyIdMap, + prefixes: Vec>, + + /// Bit 0: some exact key starts with this byte. Bit 1: some prefix does. + byte_screen: [u8; 256], + /// Second screen dimension: for each first byte, the set of *lengths* the + /// registry's exact keys of that byte have (length clamped to 63, so + /// anything longer than the longest registry key shares one bucket). A + /// filler key that happens to share a first byte with a registry key — + /// `gen_ai.request.parameter_7` against `gen_ai.system` — is rejected here + /// instead of paying a map lookup. + key_len_screen: [u64; 256], + prefixes_by_first_byte: Vec>, + + // dispatch tables — a span's evidence names the matchers it can possibly hit + /// Matchers a `ScopeSpans` fully decides: every `scope`/`resource`-class matcher, + /// plus any matcher keyed on a `scope.*` pseudo-key. Resolved once per scope. + hoisted_matchers: Vec, + key_matchers: Vec>, + prefix_matchers: Vec>, + span_name_eq: HashMap, Vec>, + /// `present`/`value_prefix` on `span.name` — evaluated per span, no index. + span_name_other: Vec, + /// Conditional (insufficient resource/scope) matchers, in slot order. + conditional: Vec, +} + +/// The process-wide registry. Parsed and validated on first use. +pub fn registry() -> &'static Registry { + static REGISTRY: LazyLock = LazyLock::new(|| { + Registry::parse(REGISTRY_JSON).expect("vendored ai-registry/registry.json is malformed") + }); + ®ISTRY +} + +impl Registry { + pub fn version(&self) -> u32 { + self.version + } + pub fn vendors(&self) -> &[Vendor] { + &self.vendors + } + pub fn vendor(&self, id: VendorId) -> &Vendor { + &self.vendors[id.index()] + } + pub fn vendor_slug(&self, id: VendorId) -> &str { + self.vendors[id.index()].slug() + } + pub fn vendor_id(&self, slug: &str) -> Option { + self.vendor_ids.get(slug).copied() + } + pub fn matchers(&self) -> &[Matcher] { + &self.matchers + } + pub fn keys(&self) -> &[Box] { + &self.keys + } + pub fn prefixes(&self) -> &[Box] { + &self.prefixes + } + pub fn key_id(&self, key: &str) -> Option { + self.key_ids.get(key).copied() + } + + pub(crate) fn hoisted_matchers(&self) -> &[MatcherId] { + &self.hoisted_matchers + } + pub(crate) fn key_matchers(&self, key: KeyId) -> &[MatcherId] { + &self.key_matchers[key as usize] + } + pub(crate) fn prefix_matchers(&self, prefix: PrefixId) -> &[MatcherId] { + &self.prefix_matchers[prefix as usize] + } + pub(crate) fn span_name_matchers(&self, name: &str) -> &[MatcherId] { + self.span_name_eq + .get(name) + .map(|v| v.as_slice()) + .unwrap_or(&[]) + } + pub(crate) fn span_name_other(&self) -> &[MatcherId] { + &self.span_name_other + } + pub(crate) fn conditional(&self, slot: usize) -> MatcherId { + self.conditional[slot] + } + + /// The generated prefilter. `true` iff some rule can consult this key. + /// + /// Cheap by construction: a 256-entry byte screen rejects almost every + /// attribute key on a non-AI span before any hashing happens. + pub fn references_key(&self, key: &str) -> bool { + !self.probe(key).is_empty() + } + + /// Prefilter + interning in one pass: the hot-path entry point. + #[inline] + pub fn probe(&self, key: &str) -> KeyProbe { + let bytes = key.as_bytes(); + let Some(&first) = bytes.first() else { + return KeyProbe::default(); + }; + let screen = self.byte_screen[first as usize]; + if screen == 0 { + return KeyProbe::default(); + } + let mut probe = KeyProbe::default(); + if screen & 1 != 0 && self.key_len_screen[first as usize] & length_bit(bytes.len()) != 0 { + probe.key_id = self.key_ids.get(key).copied(); + } + if screen & 2 != 0 { + for &pid in &self.prefixes_by_first_byte[first as usize] { + if key.starts_with(&*self.prefixes[pid as usize]) { + probe.prefix_bits |= 1u64 << pid; + } + } + } + probe + } + + // ----------------------------------------------------------------------- + // load + // ----------------------------------------------------------------------- + + fn parse(json: &str) -> Result { + let raw: RawRegistry = serde_json::from_str(json).map_err(|e| e.to_string())?; + let mut builder = Builder::default(); + let registry = builder.build(raw)?; + registry.validate()?; + Ok(registry) + } + + /// Load-time validation. Every violation here is a compile-artifact bug. + fn validate(&self) -> Result<(), String> { + if self.version == 0 { + return Err("registry_version 0 is reserved for pre-rollout rows".into()); + } + if self.vendors.len() > MAX_VENDORS { + return Err(format!( + "{} vendors exceeds the {MAX_VENDORS}-bit vendor set", + self.vendors.len() + )); + } + if self.prefixes.len() > MAX_PREFIXES { + return Err(format!( + "{} key prefixes exceeds the {MAX_PREFIXES}-bit prefix set", + self.prefixes.len() + )); + } + if self.keys.len() > MAX_KEYS { + return Err(format!( + "{} exact keys exceeds the {MAX_KEYS}-bit key-presence set", + self.keys.len() + )); + } + + // Unique priorities across ALL matcher classes and the unknown tier. + let mut seen = HashMap::new(); + for (index, matcher) in self.matchers.iter().enumerate() { + if let Some(other) = seen.insert(matcher.priority, index) { + return Err(format!( + "duplicate priority {}: matchers {other} and {index}", + matcher.priority + )); + } + } + + for matcher in &self.matchers { + let vendor = &self.vendors[matcher.vendor.index()]; + let band = if vendor.unknown_bucket { + &BAND_UNKNOWN + } else if matcher.kind == HitKind::Unconditional && !matcher.promotes { + &BAND_SUFFICIENT + } else { + &BAND_VENDOR + }; + if !band.contains(&matcher.priority) { + return Err(format!( + "priority {} for {} is outside its band {band:?}", + matcher.priority, + vendor.slug() + )); + } + if matcher.promotes && matcher.kind != HitKind::Unconditional { + return Err("attr matchers are unconditional hits".into()); + } + if let Predicate::ValuePrefix(_, _) = matcher.predicate { + // enforced at build time (pseudo-key restriction); re-stated here so + // the invariant is visible where the rest of them live. + } + } + + // `unknown:` is reserved for the fingerprint tier. + for vendor in &self.vendors { + if vendor.slug().starts_with("unknown:") != vendor.unknown_bucket { + return Err(format!( + "vendor slug {:?} misuses the reserved `unknown:` prefix", + vendor.slug() + )); + } + if vendor.unknown_bucket && !vendor.candidates.is_empty() { + return Err("unknown-tier buckets carry no session-key rules".into()); + } + } + + // Every matcher is dispatched exactly once — either the scope decides it or a + // span does. A matcher in no table would silently never fire. + let dispatched = self.hoisted_matchers.len() + + self.key_matchers.iter().map(Vec::len).sum::() + + self.prefix_matchers.iter().map(Vec::len).sum::() + + self.span_name_eq.values().map(Vec::len).sum::() + + self.span_name_other.len(); + if dispatched != self.matchers.len() { + return Err(format!( + "{dispatched} dispatched matchers for {} in the registry", + self.matchers.len() + )); + } + + // The prefilter must survive every key and prefix the registry references. + for key in &self.keys { + if !self.references_key(key) { + return Err(format!("prefilter drops registry key {key:?}")); + } + } + for prefix in &self.prefixes { + if !self.references_key(prefix) || !self.references_key(&format!("{prefix}x")) { + return Err(format!("prefilter drops registry prefix {prefix:?}")); + } + } + Ok(()) + } +} + +#[derive(Default)] +struct Builder { + keys: Vec>, + key_ids: KeyIdMap, + prefixes: Vec>, + prefix_ids: HashMap, PrefixId>, +} + +impl Builder { + fn intern_key(&mut self, key: &str) -> Result { + if let Some(&id) = self.key_ids.get(key) { + return Ok(id); + } + let id = + u16::try_from(self.keys.len()).map_err(|_| "too many registry keys".to_string())?; + self.keys.push(key.into()); + self.key_ids.insert(key.into(), id); + Ok(id) + } + + fn intern_prefix(&mut self, prefix: &str) -> Result { + if prefix.is_empty() { + return Err("empty key_prefix would match every attribute".into()); + } + if let Some(&id) = self.prefix_ids.get(prefix) { + return Ok(id); + } + let id = u8::try_from(self.prefixes.len()) + .ok() + .filter(|id| (*id as usize) < MAX_PREFIXES) + .ok_or_else(|| "too many key prefixes".to_string())?; + self.prefixes.push(prefix.into()); + self.prefix_ids.insert(prefix.into(), id); + Ok(id) + } + + fn lookup_key(&mut self, key: &str) -> Result { + Ok(match PseudoKey::parse(key) { + Some(pseudo) => LookupKey::Pseudo(pseudo), + None => LookupKey::Attr(self.intern_key(key)?), + }) + } + + fn predicate(&mut self, raw: &RawPredicate) -> Result { + Ok(match raw { + RawPredicate::Present { key } => Predicate::Present(self.lookup_key(key)?), + RawPredicate::Eq { key, value } => { + Predicate::Eq(self.lookup_key(key)?, value.as_str().into()) + } + RawPredicate::KeyPrefix { prefix } => Predicate::KeyPrefix(self.intern_prefix(prefix)?), + RawPredicate::ValuePrefix { key, prefix } => { + // D1: value_prefix is restricted to the four pseudo-keys, which are + // real columns in both targets (`startsWith` in SQL). + let pseudo = PseudoKey::parse(key).ok_or_else(|| { + format!("value_prefix on {key:?} — restricted to pseudo-keys") + })?; + Predicate::ValuePrefix(pseudo, prefix.as_str().into()) + } + }) + } + + fn build(&mut self, raw: RawRegistry) -> Result { + const OPS: [&str; 4] = ["present", "eq", "key_prefix", "value_prefix"]; + if raw.algebra.ops.len() != OPS.len() + || !OPS.iter().all(|op| raw.algebra.ops.iter().any(|o| o == op)) + { + return Err(format!("unexpected algebra ops: {:?}", raw.algebra.ops)); + } + for key in &raw.algebra.value_prefix_pseudo_keys { + if PseudoKey::parse(key).is_none() { + return Err(format!("unknown value_prefix pseudo-key {key:?}")); + } + } + + let mut vendors: Vec = Vec::new(); + let mut vendor_ids: HashMap, VendorId> = HashMap::new(); + let mut matchers: Vec = Vec::new(); + + let push_vendor = |vendors: &mut Vec, + vendor_ids: &mut HashMap, VendorId>, + vendor: Vendor| + -> Result { + let id = u16::try_from(vendors.len()).map_err(|_| "too many vendors".to_string())?; + if vendor_ids + .insert(vendor.slug.clone(), VendorId(id)) + .is_some() + { + return Err(format!("duplicate vendor slug {:?}", vendor.slug)); + } + vendors.push(vendor); + Ok(VendorId(id)) + }; + + for raw_vendor in &raw.vendors { + let mut candidates = Vec::with_capacity(raw_vendor.session_candidates.len()); + for raw_candidate in &raw_vendor.session_candidates { + let authority = match &raw_candidate.authority_predicate { + None => Authority::Always, + Some(RawAuthority::One(p)) => Authority::One(self.predicate(p)?), + Some(RawAuthority::AnyOf { any_of }) => Authority::AnyOf( + any_of + .iter() + .map(|p| self.predicate(p)) + .collect::>()?, + ), + }; + for token in &raw_candidate.validation { + if token != "non_empty" && token != "not_in_decoy_values" { + return Err(format!("unknown validation token {token:?}")); + } + } + candidates.push(SessionCandidate { + key: self.lookup_key(&raw_candidate.key)?, + authority, + require_non_empty: raw_candidate.validation.iter().any(|v| v == "non_empty"), + reject_decoy_values: raw_candidate + .validation + .iter() + .any(|v| v == "not_in_decoy_values"), + granularity: match raw_candidate.granularity { + RawGranularity::Session => Granularity::Session, + RawGranularity::Run => Granularity::Run, + RawGranularity::User => Granularity::User, + RawGranularity::Instance => Granularity::Instance, + }, + }); + } + + let vendor = Vendor { + slug: raw_vendor.vendor.as_str().into(), + unknown_bucket: false, + candidates, + // `decoy_keys` are deliberately not compiled: the rule is "never + // consult", so the detector must not be able to read one. + decoy_values: raw_vendor + .decoy_values + .iter() + .map(|d| d.value.as_str().into()) + .collect(), + }; + let vendor_id = push_vendor(&mut vendors, &mut vendor_ids, vendor)?; + + for raw_matcher in &raw_vendor.matchers { + let predicate = self.predicate(&raw_matcher.predicate)?; + let is_attr = raw_matcher.class == RawClass::Attr; + if is_attr && raw_matcher.sufficient { + return Err("attr matchers must not declare sufficiency".into()); + } + matchers.push(Matcher { + vendor: vendor_id, + priority: raw_matcher.priority, + kind: if is_attr || raw_matcher.sufficient { + HitKind::Unconditional + } else { + HitKind::Conditional + }, + target: match raw_matcher.class { + RawClass::Resource => AttrTarget::Resource, + RawClass::Scope => AttrTarget::Scope, + RawClass::Attr => AttrTarget::Span, + }, + promotes: is_attr, + conditional_slot: None, + predicate, + }); + } + } + + // Unknown-tier fingerprints join the same table as pseudo-vendors: the + // priority bands (D4) are what keeps them below every vendor hit, so + // resolution stays one comparison rather than a second pass. + for raw_unknown in &raw.unknown_tier { + if !raw_unknown.bucket.starts_with("unknown:") { + return Err(format!( + "unknown-tier bucket {:?} must use the reserved prefix", + raw_unknown.bucket + )); + } + let vendor_id = match vendor_ids.get(raw_unknown.bucket.as_str()) { + Some(&id) => id, + None => push_vendor( + &mut vendors, + &mut vendor_ids, + Vendor { + slug: raw_unknown.bucket.as_str().into(), + unknown_bucket: true, + candidates: Vec::new(), + decoy_values: Vec::new(), + }, + )?, + }; + let predicate = self.predicate(&raw_unknown.predicate)?; + matchers.push(Matcher { + vendor: vendor_id, + priority: raw_unknown.priority, + kind: HitKind::Unconditional, + // Unknown-tier fingerprints carry no class and are span-local, exactly + // as `collectVendorBranches` compiles them (`"span"` target). + target: AttrTarget::Span, + promotes: false, + conditional_slot: None, + predicate, + }); + } + + if matchers.len() > u16::MAX as usize { + return Err("too many matchers".into()); + } + + let mut conditional = Vec::new(); + for (index, matcher) in matchers.iter_mut().enumerate() { + if matcher.kind != HitKind::Conditional { + continue; + } + let slot = conditional.len(); + if slot >= MAX_CONDITIONAL { + return Err(format!( + "more than {MAX_CONDITIONAL} conditional matchers needs a wider bitmask" + )); + } + matcher.conditional_slot = Some(slot as u8); + conditional.push(index as MatcherId); + } + + // Dispatch tables. A matcher lands in exactly one of them, decided by whether a + // `ResourceSpans`/`ScopeSpans` already holds everything it reads: + // + // * a `scope`/`resource`-class matcher reads only that list → hoisted; + // * any matcher on a `scope.*` pseudo-key reads a scope column → hoisted; + // * everything else (attr-class attributes and prefixes, plus any `span.name` + // matcher, whatever its class — pseudo-keys ignore class) → per span. + let mut hoisted_matchers = Vec::new(); + let mut key_matchers = vec![Vec::new(); self.keys.len()]; + let mut prefix_matchers = vec![Vec::new(); self.prefixes.len()]; + let mut span_name_eq: HashMap, Vec> = HashMap::new(); + let mut span_name_other = Vec::new(); + for (index, matcher) in matchers.iter().enumerate() { + let id = index as MatcherId; + match &matcher.predicate { + Predicate::Eq(LookupKey::Pseudo(PseudoKey::SpanName), value) => { + span_name_eq.entry(value.clone()).or_default().push(id) + } + Predicate::Present(LookupKey::Pseudo(PseudoKey::SpanName)) + | Predicate::ValuePrefix(PseudoKey::SpanName, _) => span_name_other.push(id), + _ if matcher.target != AttrTarget::Span => hoisted_matchers.push(id), + Predicate::Present(LookupKey::Attr(key)) + | Predicate::Eq(LookupKey::Attr(key), _) => key_matchers[*key as usize].push(id), + Predicate::KeyPrefix(prefix) => prefix_matchers[*prefix as usize].push(id), + // A span-class matcher on a scope pseudo-key: a column the scope knows. + Predicate::Present(LookupKey::Pseudo(_)) + | Predicate::Eq(LookupKey::Pseudo(_), _) + | Predicate::ValuePrefix(_, _) => hoisted_matchers.push(id), + } + } + + let mut byte_screen = [0u8; 256]; + let mut key_len_screen = [0u64; 256]; + let mut prefixes_by_first_byte = vec![Vec::new(); 256]; + for key in &self.keys { + if let Some(&first) = key.as_bytes().first() { + byte_screen[first as usize] |= 1; + key_len_screen[first as usize] |= length_bit(key.len()); + } + } + for (index, prefix) in self.prefixes.iter().enumerate() { + if let Some(&first) = prefix.as_bytes().first() { + byte_screen[first as usize] |= 2; + prefixes_by_first_byte[first as usize].push(index as PrefixId); + } + } + + Ok(Registry { + version: raw.registry_version, + vendors, + vendor_ids, + matchers, + keys: std::mem::take(&mut self.keys), + key_ids: std::mem::take(&mut self.key_ids), + prefixes: std::mem::take(&mut self.prefixes), + byte_screen, + key_len_screen, + prefixes_by_first_byte, + hoisted_matchers, + key_matchers, + prefix_matchers, + span_name_eq, + span_name_other, + conditional, + }) + } +} + +/// Canonical `AnyValue → String` for a *registry-referenced* value, borrowing when +/// the value is already a string (the overwhelmingly common case). +/// +/// Delegates to the row writer's `any_value_string` for every other type, so the +/// matcher and the written row see the same bytes — the premise of the §6 +/// Rust/SQL alignment contract. +pub fn canonical_value( + value: Option<&opentelemetry_proto::tonic::common::v1::AnyValue>, +) -> Cow<'_, str> { + use opentelemetry_proto::tonic::common::v1::any_value; + match value.and_then(|v| v.value.as_ref()) { + Some(any_value::Value::StringValue(s)) => Cow::Borrowed(s.as_str()), + Some(_) => Cow::Owned(crate::telemetry::any_value_string(value.expect("checked"))), + None => Cow::Borrowed(""), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_vendored_registry_loads_and_validates() { + let registry = registry(); + assert_eq!(registry.version(), 1); + assert_eq!( + registry.matchers().len(), + 88 + 5, + "88 matchers + unknown tier" + ); + assert_eq!( + registry.vendors().len(), + 21 + 3, + "21 vendors + 3 unknown buckets" + ); + assert_eq!( + registry + .vendors() + .iter() + .map(|v| v.candidates().len()) + .sum::(), + 29 + ); + } + + #[test] + fn vendor_slugs_are_the_closed_set() { + let registry = registry(); + for slug in [ + "agno", + "claude_agent_sdk", + "crewai", + "dspy", + "effect_ai", + "flue", + "google_adk", + "haystack", + "langchain", + "litellm", + "llamaindex", + "mastra", + "microsoft_agent_framework", + "openai_agents_sdk", + "openinference-openai", + "pydantic_ai", + "semantic_kernel", + "smolagents", + "spring_ai", + "strands", + "vercel_ai_sdk", + ] { + let id = registry.vendor_id(slug).expect("known slug"); + assert_eq!(registry.vendor_slug(id), slug); + assert!(!registry.vendor(id).is_unknown_bucket()); + } + // D2 applied at compile time: the seed's slug is not producible. + assert!(registry.vendor_id("langgraph").is_none()); + for bucket in ["unknown:genai", "unknown:openinference", "unknown:other"] { + let id = registry.vendor_id(bucket).expect("bucket"); + assert!(registry.vendor(id).is_unknown_bucket()); + assert!(registry.vendor(id).candidates().is_empty()); + } + } + + /// Plan §7: "every registry key/prefix survives the generated prefilter". + /// `validate` enforces it at load; this states it as a test in its own right and + /// adds the negative direction. + #[test] + fn prefilter_is_self_consistent() { + let registry = registry(); + for key in registry.keys() { + assert!(registry.references_key(key), "prefilter dropped {key}"); + assert_eq!(registry.probe(key).key_id, registry.key_id(key)); + } + for (index, prefix) in registry.prefixes().iter().enumerate() { + let probed = registry.probe(&format!("{prefix}suffix")); + assert!( + probed.prefix_bits & (1 << index) != 0, + "prefilter dropped prefix {prefix}" + ); + } + for key in ["http.request.method", "db.system", "", "x", "service.port"] { + assert!(!registry.references_key(key), "{key} should not survive"); + } + } + + #[test] + fn priorities_are_unique_and_banded() { + let registry = registry(); + let mut priorities: Vec = registry.matchers().iter().map(|m| m.priority).collect(); + let total = priorities.len(); + priorities.sort_unstable(); + priorities.dedup(); + assert_eq!(priorities.len(), total, "priorities must be unique"); + + for matcher in registry.matchers() { + let vendor = registry.vendor(matcher.vendor); + if vendor.is_unknown_bucket() { + assert!(BAND_UNKNOWN.contains(&matcher.priority)); + } else if matcher.kind == HitKind::Unconditional && !matcher.promotes { + assert!(BAND_SUFFICIENT.contains(&matcher.priority)); + } else { + assert!(BAND_VENDOR.contains(&matcher.priority)); + } + } + // Every unknown-tier fingerprint loses to every vendor matcher. + let worst_vendor = registry + .matchers() + .iter() + .filter(|m| !registry.vendor(m.vendor).is_unknown_bucket()) + .map(|m| m.priority) + .min() + .expect("vendor matchers"); + let best_unknown = registry + .matchers() + .iter() + .filter(|m| registry.vendor(m.vendor).is_unknown_bucket()) + .map(|m| m.priority) + .max() + .expect("unknown tier"); + assert!(best_unknown < worst_vendor); + } + + #[test] + fn a_fifth_operator_is_a_load_error() { + let json = r#"{"registry_version":1, + "algebra":{"ops":["present","eq","key_prefix","value_prefix"],"value_prefix_pseudo_keys":["scope.name"]}, + "unknown_tier":[], + "vendors":[{"vendor":"x","matchers":[{"class":"attr","sufficient":false, + "predicate":{"op":"regex","key":"a","value":"b"},"priority":29999}], + "session_candidates":[],"decoy_values":[]}]}"#; + assert!(Registry::parse(json).is_err()); + } + + #[test] + fn value_prefix_outside_the_pseudo_keys_is_a_load_error() { + let json = r#"{"registry_version":1, + "algebra":{"ops":["present","eq","key_prefix","value_prefix"],"value_prefix_pseudo_keys":["scope.name"]}, + "unknown_tier":[], + "vendors":[{"vendor":"x","matchers":[{"class":"attr","sufficient":false, + "predicate":{"op":"value_prefix","key":"gen_ai.system","prefix":"a"},"priority":29999}], + "session_candidates":[],"decoy_values":[]}]}"#; + let error = Registry::parse(json).expect_err("must reject"); + assert!(error.contains("pseudo-keys"), "{error}"); + } + + #[test] + fn duplicate_priorities_are_a_load_error() { + let json = r#"{"registry_version":1, + "algebra":{"ops":["present","eq","key_prefix","value_prefix"],"value_prefix_pseudo_keys":["scope.name"]}, + "unknown_tier":[], + "vendors":[{"vendor":"x","matchers":[ + {"class":"attr","sufficient":false,"predicate":{"op":"present","key":"a"},"priority":29999}, + {"class":"attr","sufficient":false,"predicate":{"op":"present","key":"b"},"priority":29999}], + "session_candidates":[],"decoy_values":[]}]}"#; + let error = Registry::parse(json).expect_err("must reject"); + assert!(error.contains("duplicate priority"), "{error}"); + } + + #[test] + fn the_unknown_prefix_is_reserved_for_the_fingerprint_tier() { + let json = r#"{"registry_version":1, + "algebra":{"ops":["present","eq","key_prefix","value_prefix"],"value_prefix_pseudo_keys":["scope.name"]}, + "unknown_tier":[], + "vendors":[{"vendor":"unknown:mine","matchers":[],"session_candidates":[],"decoy_values":[]}]}"#; + let error = Registry::parse(json).expect_err("must reject"); + assert!(error.contains("reserved"), "{error}"); + } + + #[test] + fn priorities_outside_their_band_are_a_load_error() { + let json = r#"{"registry_version":1, + "algebra":{"ops":["present","eq","key_prefix","value_prefix"],"value_prefix_pseudo_keys":["scope.name"]}, + "unknown_tier":[], + "vendors":[{"vendor":"x","matchers":[ + {"class":"scope","sufficient":true,"predicate":{"op":"eq","key":"scope.name","value":"x"},"priority":29999}], + "session_candidates":[],"decoy_values":[]}]}"#; + let error = Registry::parse(json).expect_err("must reject"); + assert!(error.contains("band"), "{error}"); + } +} diff --git a/apps/ingest/src/cityhash102.rs b/apps/ingest/src/cityhash102.rs new file mode 100644 index 000000000..70dcb6844 --- /dev/null +++ b/apps/ingest/src/cityhash102.rs @@ -0,0 +1,345 @@ +//! CityHash64 as ClickHouse computes it — **CityHash v1.0.2**, not upstream HEAD. +//! +//! `AiSessionKeyHash` is specified as `cityHash64(value)` over the resolved +//! session-key value, so that a SQL rebuild over the written rows reproduces the +//! number the ingest path wrote (write-side plan §2 step 5 / §6). That only holds if the Rust hash +//! is bit-identical to ClickHouse's `cityHash64`, and ClickHouse vendors +//! `contrib/cityhash102` — Google's 1.0.2 release, frozen in 2011. Upstream +//! CityHash changed `CityHash64` in 1.1 (different mixing for len ≤ 32 and a +//! different seeding/loop for len > 64), so a crate tracking upstream returns +//! *different* numbers for the same input and would silently desync Rust from SQL. +//! Rather than depend on a crate whose variant we would have to keep re-proving, +//! the 1.0.2 algorithm is transcribed here and pinned by the ground-truth vectors +//! in the test below, all read out of a live ClickHouse. +//! +//! Two properties this module exists to guarantee: +//! +//! * `city_hash64(x)` == `SELECT cityHash64(x)` for every byte string. +//! * **Only the single-argument (concat) form is reproducible.** ClickHouse's +//! multi-argument `cityHash64(a, b)` is not the hash of a concatenation — it +//! combines per-argument hashes with its own 128→64 fold. Verified on +//! ClickHouse 26.7.2.59: `cityHash64('ab')` = 1725057946192985918 whereas +//! `cityHash64('a','b')` = 7468329322676821011. Never emit SQL that compares a +//! Rust hash against the multi-arg form. + +const K0: u64 = 0xc3a5_c85c_97cb_3127; +const K1: u64 = 0xb492_b66f_be98_f273; +const K2: u64 = 0x9ae1_6a3b_2f90_404f; +const K3: u64 = 0xc949_d7c7_509e_6557; + +#[inline] +fn fetch64(bytes: &[u8], at: usize) -> u64 { + let mut buf = [0u8; 8]; + buf.copy_from_slice(&bytes[at..at + 8]); + u64::from_le_bytes(buf) +} + +#[inline] +fn fetch32(bytes: &[u8], at: usize) -> u32 { + let mut buf = [0u8; 4]; + buf.copy_from_slice(&bytes[at..at + 4]); + u32::from_le_bytes(buf) +} + +#[inline] +fn rotate(value: u64, shift: u32) -> u64 { + if shift == 0 { + value + } else { + value.rotate_right(shift) + } +} + +/// `Rotate` with a shift the caller guarantees is non-zero. 1.0.2 keeps this as a +/// separate function because the branchless form is only valid for shift ≥ 1. +#[inline] +fn rotate_by_at_least_1(value: u64, shift: u32) -> u64 { + value.rotate_right(shift) +} + +#[inline] +fn shift_mix(value: u64) -> u64 { + value ^ (value >> 47) +} + +/// `Hash128to64` — the 128→64 fold shared by every length branch. +#[inline] +fn hash_len16(u: u64, v: u64) -> u64 { + const MUL: u64 = 0x9ddf_ea08_eb38_2d69; + let mut a = (u ^ v).wrapping_mul(MUL); + a ^= a >> 47; + let mut b = (v ^ a).wrapping_mul(MUL); + b ^= b >> 47; + b.wrapping_mul(MUL) +} + +fn hash_len0to16(bytes: &[u8]) -> u64 { + let len = bytes.len(); + if len > 8 { + let a = fetch64(bytes, 0); + let b = fetch64(bytes, len - 8); + return hash_len16( + a, + rotate_by_at_least_1(b.wrapping_add(len as u64), len as u32), + ) ^ b; + } + if len >= 4 { + let a = fetch32(bytes, 0) as u64; + return hash_len16( + (len as u64).wrapping_add(a << 3), + fetch32(bytes, len - 4) as u64, + ); + } + if len > 0 { + let a = bytes[0] as u32; + let b = bytes[len >> 1] as u32; + let c = bytes[len - 1] as u32; + let y = a.wrapping_add(b << 8) as u64; + let z = (len as u32).wrapping_add(c << 2) as u64; + return shift_mix(y.wrapping_mul(K2) ^ z.wrapping_mul(K3)).wrapping_mul(K2); + } + K2 +} + +fn hash_len17to32(bytes: &[u8]) -> u64 { + let len = bytes.len(); + let a = fetch64(bytes, 0).wrapping_mul(K1); + let b = fetch64(bytes, 8); + let c = fetch64(bytes, len - 8).wrapping_mul(K2); + let d = fetch64(bytes, len - 16).wrapping_mul(K0); + hash_len16( + rotate(a.wrapping_sub(b), 43) + .wrapping_add(rotate(c, 30)) + .wrapping_add(d), + a.wrapping_add(rotate(b ^ K3, 20)) + .wrapping_sub(c) + .wrapping_add(len as u64), + ) +} + +/// Returns a 16-byte hash for 48 bytes: `(a + z, b + c)`. +#[inline] +fn weak_hash_len32_with_seeds_raw(w: u64, x: u64, y: u64, z: u64, a: u64, b: u64) -> (u64, u64) { + let mut a = a.wrapping_add(w); + let mut b = rotate(b.wrapping_add(a).wrapping_add(z), 21); + let c = a; + a = a.wrapping_add(x); + a = a.wrapping_add(y); + b = b.wrapping_add(rotate(a, 44)); + (a.wrapping_add(z), b.wrapping_add(c)) +} + +#[inline] +fn weak_hash_len32_with_seeds(bytes: &[u8], at: usize, a: u64, b: u64) -> (u64, u64) { + weak_hash_len32_with_seeds_raw( + fetch64(bytes, at), + fetch64(bytes, at + 8), + fetch64(bytes, at + 16), + fetch64(bytes, at + 24), + a, + b, + ) +} + +fn hash_len33to64(bytes: &[u8]) -> u64 { + let len = bytes.len(); + let mut z = fetch64(bytes, 24); + let mut a = fetch64(bytes, 0).wrapping_add( + (len as u64) + .wrapping_add(fetch64(bytes, len - 16)) + .wrapping_mul(K0), + ); + let mut b = rotate(a.wrapping_add(z), 52); + let mut c = rotate(a, 37); + a = a.wrapping_add(fetch64(bytes, 8)); + c = c.wrapping_add(rotate(a, 7)); + a = a.wrapping_add(fetch64(bytes, 16)); + let vf = a.wrapping_add(z); + let vs = b.wrapping_add(rotate(a, 31)).wrapping_add(c); + + a = fetch64(bytes, 16).wrapping_add(fetch64(bytes, len - 32)); + z = fetch64(bytes, len - 8); + b = rotate(a.wrapping_add(z), 52); + c = rotate(a, 37); + a = a.wrapping_add(fetch64(bytes, len - 24)); + c = c.wrapping_add(rotate(a, 7)); + a = a.wrapping_add(fetch64(bytes, len - 16)); + let wf = a.wrapping_add(z); + let ws = b.wrapping_add(rotate(a, 31)).wrapping_add(c); + + let r = shift_mix( + vf.wrapping_add(ws) + .wrapping_mul(K2) + .wrapping_add(wf.wrapping_add(vs).wrapping_mul(K0)), + ); + shift_mix(r.wrapping_mul(K0).wrapping_add(vs)).wrapping_mul(K2) +} + +/// `CityHash64` (v1.0.2). Byte-identical to ClickHouse's single-argument +/// `cityHash64(String)`. +pub fn city_hash64(bytes: &[u8]) -> u64 { + let total = bytes.len(); + if total <= 32 { + return if total <= 16 { + hash_len0to16(bytes) + } else { + hash_len17to32(bytes) + }; + } + if total <= 64 { + return hash_len33to64(bytes); + } + + // For strings over 64 bytes: hash the tail first, then loop over 64-byte + // chunks keeping 56 bytes of state (v, w, x, y, z). + let mut x = fetch64(bytes, 0); + let mut y = fetch64(bytes, total - 16) ^ K1; + let mut z = fetch64(bytes, total - 56) ^ K0; + let mut v = weak_hash_len32_with_seeds(bytes, total - 64, total as u64, y); + let mut w = weak_hash_len32_with_seeds(bytes, total - 32, (total as u64).wrapping_mul(K1), K0); + z = z.wrapping_add(shift_mix(v.1).wrapping_mul(K1)); + x = rotate(z.wrapping_add(x), 39).wrapping_mul(K1); + y = rotate(y, 33).wrapping_mul(K1); + + let mut at = 0usize; + let mut left = (total - 1) & !63usize; + loop { + x = rotate( + x.wrapping_add(y) + .wrapping_add(v.0) + .wrapping_add(fetch64(bytes, at + 16)), + 37, + ) + .wrapping_mul(K1); + y = rotate( + y.wrapping_add(v.1).wrapping_add(fetch64(bytes, at + 48)), + 42, + ) + .wrapping_mul(K1); + x ^= w.1; + y ^= v.0; + z = rotate(z ^ w.0, 33); + v = weak_hash_len32_with_seeds(bytes, at, v.1.wrapping_mul(K1), x.wrapping_add(w.0)); + w = weak_hash_len32_with_seeds(bytes, at + 32, z.wrapping_add(w.1), y); + std::mem::swap(&mut z, &mut x); + at += 64; + left -= 64; + if left == 0 { + break; + } + } + + hash_len16( + hash_len16(v.0, w.0) + .wrapping_add(shift_mix(y).wrapping_mul(K1)) + .wrapping_add(z), + hash_len16(v.1, w.1).wrapping_add(x), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Ground truth, read out of a live ClickHouse 26.7.2.59 (the repo's + /// `bun ch:up` container) with + /// `SELECT toString(cityHash64(unhex('')))`. The inputs cover every + /// length branch of the algorithm — 0, 1..16 (all three sub-branches), + /// 17..32, 33..64, and >64 including the loop's exact multiples and + /// off-by-one lengths (63/64/65/66, 127/128/129, 191/192, 255, 1000, 4096) — + /// plus embedded NULs, non-UTF8 bytes, unicode, and a realistic session-id + /// payload. A variant mismatch (1.1 vs 1.0.2) shows up in + /// the ≤32 and >64 branches, so this table is what pins the variant. + const VECTORS: &[(&str, u64)] = &[ + ("", 11160318154034397263u64), + ("62", 4947675599669400333u64), + ("636a", 723970123948255656u64), + ("646b72", 15801372605664632734u64), + ("656c737a", 6980601109885121812u64), + ("666d746168", 12950413508741813089u64), + ("686f76636a7178", 13503706177584786383u64), + ("697077646b727966", 14323467942299744891u64), + ("6a7178656c737a676e", 15781391667595025456u64), + ("6d7461686f76636a7178656c", 16427312961553000125u64), + ("7077646b7279666d7461686f76636a", 11407956329976032504u64), + ("7178656c737a676e7562697077646b72", 18255961058499077557u64), + ("7279666d7461686f76636a7178656c737a", 14462083576506790518u64), + ("78656c737a676e7562697077646b7279666d7461686f76", 16306061539677388550u64), + ("79666d7461686f76636a7178656c737a676e756269707764", 10777163611560930364u64), + ("666d7461686f76636a7178656c737a676e7562697077646b7279666d746168", 6019665914868491506u64), + ("676e7562697077646b7279666d7461686f76636a7178656c737a676e75626970", 12620920786248464957u64), + ("686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178", 16474122761118675582u64), + ("6f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562", 16259073405242189613u64), + ("76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b727966", 12082341201738856318u64), + ("77646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e", 1697233247017001413u64), + ("646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b72", 497402721431915547u64), + ("656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a", 15627023174728833030u64), + ("6c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e756269707764", 17872513691167454009u64), + ("6d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c", 2635662625183472829u64), + ("6e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d74", 18106967989178361769u64), + ("6f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562", 12835365965614688304u64), + ("62697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562", 9417686725063273001u64), + ("737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d746168", 4599548003837105958u64), + ("78656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76", 1897058114483927166u64), + ("79666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e756269707764", 10493324097790847553u64), + ("7a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c", 1488745062732587154u64), + ("6a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e", 7241325128236764542u64), + ("6b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76", 11371476341555117158u64), + ("76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b727966", 10781602255805362672u64), + ("6d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c", 16318214268670400239u64), + ("6f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562697077646b7279666d7461686f76636a7178656c737a676e7562", 14412189391335463088u64), + ("7465737420737472696e67", 11693859195493633100u64), + ("68656c6c6f20776f726c64", 12386028635079221413u64), + ("54686520717569636b2062726f776e20666f78206a756d7073206f76657220746865206c617a7920646f67", 16697807905646383735u64), + ("68c3a96c6c6f2077c3b6726c6420f09f8c8d20e2809420756e69636f646520c3bc6e6420656d6f6a69", 8552072460063165640u64), + ("6f72675f3261626344454600736573735f39663865376436632d313233342d343332312d616263642d303132333435363738396162", 196369331494914201u64), + ("73616c7400", 1640594928398365368u64), + ("000000000000000000", 78413102056792480u64), + ("fffe0001807f", 15244786278541476210u64), + ]; + + fn unhex(s: &str) -> Vec { + (0..s.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&s[i..i + 2], 16).expect("hex")) + .collect() + } + + #[test] + fn matches_clickhouse_ground_truth() { + for (hex, expected) in VECTORS { + let input = unhex(hex); + assert_eq!( + city_hash64(&input), + *expected, + "cityHash64 mismatch for {} bytes (hex {hex})", + input.len() + ); + } + } + + /// The two vectors that are also stable, published CityHash 1.0.2 facts: + /// the empty string hashes to k2, and ClickHouse agrees. + #[test] + fn empty_string_is_k2() { + assert_eq!(city_hash64(b""), K2); + assert_eq!(K2, 11160318154034397263); + } + + /// Pins the single-argument value from the module docs. The classifier hashes + /// the session-key value on its own, so the SQL leg must say `cityHash64(x)` + /// and never `cityHash64(a, b)` — the multi-arg form folds per-argument + /// hashes and returns 7468329322676821011 for the same two bytes. + #[test] + fn single_argument_form_is_the_reproducible_one() { + assert_eq!(city_hash64(b"ab"), 1725057946192985918); + } + + #[test] + fn long_inputs_do_not_panic_on_boundaries() { + for len in 0..600 { + let data: Vec = (0..len).map(|i| (i % 251) as u8).collect(); + let _ = city_hash64(&data); + } + } +} diff --git a/apps/ingest/src/lib.rs b/apps/ingest/src/lib.rs index 885b407f1..3b54a215c 100644 --- a/apps/ingest/src/lib.rs +++ b/apps/ingest/src/lib.rs @@ -1,3 +1,6 @@ +pub mod ai_classifier; +pub mod ai_registry; +pub mod cityhash102; pub mod clickhouse_insert_mappings; pub mod metrics; pub mod otel; diff --git a/apps/ingest/src/telemetry.rs b/apps/ingest/src/telemetry.rs index 5d8b09a20..8c447b1da 100644 --- a/apps/ingest/src/telemetry.rs +++ b/apps/ingest/src/telemetry.rs @@ -2798,7 +2798,10 @@ fn attr_map(attributes: &[KeyValue]) -> Map { out } -fn any_value_string(value: &AnyValue) -> String { +/// `pub(crate)` so the AI classifier canonicalizes attribute values through the +/// *same* function the row writer uses — the premise of the write-side plan's +/// Rust/SQL alignment contract (§6). Behavior is unchanged. +pub(crate) fn any_value_string(value: &AnyValue) -> String { match value.value.as_ref() { Some(any_value::Value::StringValue(value)) => value.clone(), Some(any_value::Value::BoolValue(value)) => value.to_string(),