feat(ingest): AI span classifier

.github/workflows/ingest-rust-tests.yml

@@ -7,13 +7,19 @@ on:
             - "apps/ingest/**"
             - "packages/db/drizzle/**"
             - "packages/db/src/schema/**"
+            # The read path's slug mirror: `slug_set_mirrors_the_typescript_vendors`
+            # include_str!s it, so a TS-only edit must run cargo test too.
+            - "packages/domain/src/ai/vendors.ts"
             - ".github/scripts/format-ingest-benchmark-comment.py"
             - ".github/workflows/ingest-rust-tests.yml"
     pull_request:
         paths:
             - "apps/ingest/**"
             - "packages/db/drizzle/**"
             - "packages/db/src/schema/**"
+            # The read path's slug mirror: `slug_set_mirrors_the_typescript_vendors`
+            # include_str!s it, so a TS-only edit must run cargo test too.
+            - "packages/domain/src/ai/vendors.ts"
             - ".github/scripts/format-ingest-benchmark-comment.py"
             - ".github/workflows/ingest-rust-tests.yml"
     workflow_dispatch:

.oxfmtrc.jsonc

@@ -13,6 +13,9 @@
 	// Vue turns a newline between two inline elements into a rendered space, and
 	// oxfmt happily breaks `<span>NEW</span>{{ message }}` across lines. They are
 	// authored in the exact shape they compile to.
+	// apps/ingest/fixtures/classification holds the vendored classification fixture,
+	// also copied verbatim from trace-capture — its test asserts the files'
+	// sha256 against manifest.json, so any reformat fails CI.
 	"ignorePatterns": [
 		".context",
 		"deploy",
@@ -22,5 +25,6 @@
 		"lib/thinking-orbs",
 		"packages/email/emails",
 		"packages/email/components",
+		"apps/ingest/fixtures/classification",
 	],
 }

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"

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. Budget: ~50 ns/span mean, ~300 ns worst case on a
+/// 60-attribute AI span, out of ~500 ns total per span. 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<KeyValue> = [
+        ("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<KeyValue> = 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 a trace's worth of spans.
+    // 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);

apps/ingest/fixtures/classification/README.md

@@ -0,0 +1,80 @@
+# Vendored classification fixture
+
+Recorded OTLP spans that CI replays through the AI span classifier — the tests
+live in `src/ai_classification_fixture_test.rs`. The files are generated in the
+trace-capture repo by `bun run fixture` and copied here unchanged.
+
+| file | what it is |
+| ---- | ---------- |
+| `classification-fixture.jsonl` | one span per line — the classifier's inputs. Spans no vendor matches are deduplicated: one representative line with a weight instead of many near-identical lines. |
+| `expectations.json` | the expected classification results per capture, taken from hand-reviewed goldens |
+| `manifest.json` | line counts, sha256 hashes, and a record of exactly which trace-capture revision produced the files |
+
+## Updating the fixture
+
+Whenever trace-capture's captures or vendor seeds change, regenerate and
+re-vendor in the same change that edits `src/ai_vendors.rs`:
+
+```sh
+cd ../../../trace-capture   # or wherever the checkout lives
+bun run fixture
+cp fixture/classification-fixture.jsonl fixture/expectations.json fixture/manifest.json \
+   <maple>/apps/ingest/fixtures/classification/
+```
+
+The test recomputes the sha256s against `manifest.json`, so a hand-edited
+fixture, a partial copy, or mangled line endings fail CI loudly.
+
+The hashes only cover the two data files — `manifest.json` itself is not
+hashed. What keeps the manifest honest is its `source` block: it names the
+trace-capture commit that produced the files, and `dirty: false` means checking
+out that commit and running `bun run fixture` reproduces these exact bytes. If
+a re-vendor ever comes out `dirty: true`, don't ship it (the test rejects it):
+commit the trace-capture side first, then regenerate — otherwise the recorded
+commit can't reproduce the files and the provenance is worthless.
+
+One more manifest field is checked mechanically: `dedup.value_sensitive_keys`
+lists the attribute keys whose *values* affect classification, so
+deduplication never collapses spans that differ on one of them. A test
+re-derives the list from `ai_vendors.rs` and fails if it has fallen behind.
+That check matters: a missing key would merge spans that classify differently,
+making the false-positive numbers and histograms wrong rather than merely loose.
+
+## Limits of the format
+
+Two rule shapes can't be expressed in the fixture lines, and a test asserts
+nothing depends on them yet:
+
+- **Scope attributes** — the lines carry the scope name/version/schema-url but
+  not scope attributes. No rule reads one today; the first that does needs a
+  line-schema extension and `format_version` bump in trace-capture.
+- **Link contents** — only the *number* of links is recorded. The classifier
+  has no link accessor at all today, so the field is currently unread.
+
+Maple CI never reaches into trace-capture, deliberately. The flip side: a PR
+that touches vendor rules without a fixture update (or the reverse) is a
+review smell nothing automated will catch.
+
+## Where the expected results come from
+
+`expectations.json` was **not** produced by this classifier. The numbers come
+from running each vendor seed's own rules over its captures in trace-capture,
+followed by human review. A green replay therefore proves maple's Rust rules
+compute the same answers as the reviewed rules — an independent check, though
+not ground truth.
+
+There are marked exceptions: six golden fields across four captures carry a
+`v2_resolved` note, meaning the shipped rule can't be expressed in the seed
+rule language, so those numbers were predicted by hand instead. The replay
+test pins that exact set so it can't grow silently. Two fields have even
+weaker coverage:
+
+| field | gate |
+| --- | --- |
+| `pydantic_ai_agents.unsessioned_traces` | trace-level, and this fixture strips trace ids — only `ai_classifier_corpus_test.rs` checks it, which needs a local trace-capture checkout (`TRACE_CAPTURE_DIR`). No CI coverage. |
+| `pydantic_ai_agents.key_state_by_candidate` | none — not emitted into `expectations.json` at all. |
+
+If a golden looks wrong, re-review the seed in trace-capture (see its
+`frameworks/REVIEW_IMPLEMENTATION.md`). Never regenerate goldens from this
+classifier's output — the gate would then only check that the classifier
+agrees with itself, and nothing in either repo would record that it happened.