diff --git a/src/core/src/traces/agent_signals/processor.rs b/src/core/src/traces/agent_signals/processor.rs index 897866863a..057b3168fe 100644 --- a/src/core/src/traces/agent_signals/processor.rs +++ b/src/core/src/traces/agent_signals/processor.rs @@ -23,25 +23,22 @@ #![cfg(feature = "enterprise")] -use o2_enterprise::enterprise::{ - agent_signals::{ - build_cost_sql, build_failure_sql, build_loop_ratio_sql, map_cost_hits, map_failure_hits, - map_loop_hits, - }, - common::config::get_config as get_o2_config, +use o2_enterprise::enterprise::agent_signals::{ + build_cost_sql, build_failure_sql, build_loop_ratio_sql, map_cost_hits, map_failure_hits, + map_loop_hits, }; /// Run the three bounded passes for one stream+window and self-ingest the results. -/// Early-returns Ok when the feature is disabled (no query issued). +/// +/// Agent signals are always on (no enable flag); the work self-limits via the +/// stream-level LLM gate below (skip non-gen_ai streams) and the per-span activity +/// gate in the SQL, so a non-LLM instance issues no meaningful queries. pub async fn process_agent_signals_stream( org_id: &str, stream_name: &str, start_time: i64, end_time: i64, ) -> Result<(), anyhow::Error> { - if !get_o2_config().agent_signals.enabled { - return Ok(()); - } let ts = end_time; // stamp records at window end // Schema-gate each pass: streams from different frameworks carry different @@ -59,6 +56,20 @@ pub async fn process_agent_signals_stream( .map(|s| s.field_with_name(name).is_ok()) .unwrap_or(false) }; + + // Stream-level LLM gate: skip streams whose schema carries no gen_ai columns at + // all — they are plain service/HTTP trace streams (e.g. an OTel demo app), not + // agent telemetry, so none of the three passes can produce a meaningful signal. + // Mirrors the service-graph's `has_gen_ai` gate (keyed on `gen_ai_operation_name` + // presence), keeping the two rollups' notion of "is this an LLM stream" in sync + // and avoiding three wasted searches per window on non-agent streams. This is a + // coarse schema-level filter; the per-span `gen_ai_activity` predicate in the + // failure/cost SQL is what removes non-agent spans on a MIXED stream (where the + // column exists but most spans are infra traffic). + if !has_field("gen_ai_operation_name") { + return Ok(()); + } + // Failure classification reads the resolved taxonomy (org override → repo file // → embedded fallback), not a hardcoded set. Keep ONLY the configured columns // that actually exist on this stream — referencing a missing column is a hard @@ -73,6 +84,11 @@ pub async fn process_agent_signals_stream( let has_failure_cols = !detail_fields.is_empty(); let has_tool = has_field("gen_ai_tool_name"); let has_cost = has_field("gen_ai_usage_cost"); + // The failure/cost passes gate out non-agent (infra) spans via the gen_ai + // activity predicate, which references gen_ai_agent_id. Frameworks that never + // emit an agent id don't carry that column, so schema-gate it here — a missing + // column in the predicate would be a hard search error. + let has_agent_id = has_field("gen_ai_agent_id"); let mut records = Vec::new(); @@ -85,6 +101,7 @@ pub async fn process_agent_signals_stream( end_time, &detail_fields, &taxonomy.failure_rules, + has_agent_id, ); match crate::service::traces::service_graph::run_graph_search( org_id, sql, start_time, end_time, @@ -111,7 +128,7 @@ pub async fn process_agent_signals_stream( } // R4 cost/diagnosis — needs gen_ai_usage_cost. if has_cost { - let sql = build_cost_sql(stream_name, start_time, end_time); + let sql = build_cost_sql(stream_name, start_time, end_time, has_agent_id); match crate::service::traces::service_graph::run_graph_search( org_id, sql, start_time, end_time, ) @@ -127,10 +144,12 @@ pub async fn process_agent_signals_stream( #[cfg(all(test, feature = "enterprise"))] mod test { - // process_agent_signals_stream must early-return Ok when disabled — no panic, no query. + // A stream that doesn't exist (no schema, so no gen_ai columns) must hit the + // stream-level LLM gate and return Ok without attempting any search — no panic, + // no query. Agent signals are always on, so the gate (not an enable flag) is + // what makes this a no-op. #[tokio::test] - async fn test_process_stream_noop_when_disabled() { - // enabled defaults to false; the fn must return Ok without attempting a search. + async fn test_process_stream_noop_for_non_llm_stream() { let r = super::process_agent_signals_stream("default", "nostream", 0, 1).await; assert!(r.is_ok()); } diff --git a/src/core/src/traces/service_graph/processor.rs b/src/core/src/traces/service_graph/processor.rs index 6b71c2965a..287dba97b9 100644 --- a/src/core/src/traces/service_graph/processor.rs +++ b/src/core/src/traces/service_graph/processor.rs @@ -107,41 +107,56 @@ pub async fn process_service_graph() -> Result<(), anyhow::Error> { usage_results.len() ); - // Fallback: the usage stream is the primary discovery source, but if it is - // empty (self-reporting stalled/disabled, or a fresh instance) the service - // graph would silently go dark. When usage yields nothing, discover trace - // streams directly from the schema cache across all orgs so the graph keeps - // working regardless of the usage pipeline's health. - let discovered: Vec = if usage_results.is_empty() { - let mut fallback = Vec::new(); - match crate::service::organization::list_all_orgs(None).await { - Ok(orgs) => { - for org in orgs { - for stream_name in crate::service::db::schema::list_streams_from_cache( - &org.identifier, - StreamType::Traces, - ) - .await - { - fallback.push(RecentIngestedTraceStream { + // Discovery = usage-active streams UNION every trace stream in the schema cache. + // + // Usage-windowed discovery alone is not enough: it only surfaces streams that + // logged an `Ingestion` event *inside this window*. A stream whose producer + // paused, or ingests in bursts sparser than the window, silently drops out and + // stops getting a service graph / agent-signals rollup — even though it still + // holds queryable spans. (This is exactly how the agent-behavior streams went + // dark: they fell out of recent `usage` and were never re-processed.) The old + // code only fell back to the schema cache when usage was *entirely* empty, so + // on any busy instance an individually-stale stream was invisible forever. + // + // Unioning fixes that: every known trace stream is processed each window. This + // is the same total set the empty-usage fallback already processed, so it does + // not widen the worst case. Per-stream work self-bounds — an idle window's SQL + // returns no rows (no service-graph edges, no agent signals), so re-visiting a + // quiet stream is cheap. Dedup by (org_id, stream_name) so a stream present in + // both sources is processed once. + let mut discovered = usage_results; + let mut seen: std::collections::HashSet<(String, String)> = discovered + .iter() + .map(|s| (s.org_id.clone(), s.stream_name.clone())) + .collect(); + match crate::service::organization::list_all_orgs(None).await { + Ok(orgs) => { + for org in orgs { + for stream_name in crate::service::db::schema::list_streams_from_cache( + &org.identifier, + StreamType::Traces, + ) + .await + { + if seen.insert((org.identifier.clone(), stream_name.clone())) { + discovered.push(RecentIngestedTraceStream { org_id: org.identifier.clone(), stream_name, }); } } } - Err(e) => log::warn!( - "[ServiceGraph] usage empty and org list failed, no fallback discovery: {e}" - ), } - log::info!( - "[ServiceGraph] Usage empty; discovered {} trace streams via schema cache", - fallback.len() - ); - fallback - } else { - usage_results - }; + // Non-fatal: fall back to whatever usage discovered. A busy instance still + // gets its active streams; only the stale-but-active ones are missed. + Err(e) => log::warn!( + "[ServiceGraph] org list failed; processing usage-discovered streams only: {e}" + ), + } + log::info!( + "[ServiceGraph] Processing {} trace streams (usage ∪ schema cache)", + discovered.len() + ); for RecentIngestedTraceStream { org_id, diff --git a/web/src/enterprise/views/AIObservability/AgentBehaviorPage.vue b/web/src/enterprise/views/AIObservability/AgentBehaviorPage.vue index 82e9b908a3..7ba2651e41 100644 --- a/web/src/enterprise/views/AIObservability/AgentBehaviorPage.vue +++ b/web/src/enterprise/views/AIObservability/AgentBehaviorPage.vue @@ -286,9 +286,15 @@ onMounted(async () => { try { const res = (await getStreams("traces", false, false)) as { - list?: { name: string }[]; + list?: { name: string; settings?: { is_llm_stream?: boolean } }[]; }; - availableStreams.value = (res?.list ?? []).map((s) => s.name); + // Only LLM trace streams belong here — Agent Behaviour has no signals for a + // plain service/HTTP trace stream. `is_llm_stream` is the backend-maintained + // flag (auto-detected at ingest from gen_ai_* columns). Exclude only streams + // explicitly flagged non-LLM, matching LLM Insights / Sessions / Agent Graph. + availableStreams.value = (res?.list ?? []) + .filter((s) => s.settings?.is_llm_stream !== false) + .map((s) => s.name); if (availableStreams.value.length && !activeStream.value) { activeStream.value = availableStreams.value[0]; } diff --git a/web/src/enterprise/views/AIObservability/AgentGraphPage.vue b/web/src/enterprise/views/AIObservability/AgentGraphPage.vue index 3cdfc72c5c..1c62816400 100644 --- a/web/src/enterprise/views/AIObservability/AgentGraphPage.vue +++ b/web/src/enterprise/views/AIObservability/AgentGraphPage.vue @@ -242,9 +242,17 @@ function effectiveWindow() { async function loadStreams() { try { const res = (await getStreams("traces", false, false)) as { - list?: { name: string }[]; + list?: { name: string; settings?: { is_llm_stream?: boolean } }[]; }; - availableStreams.value = (res?.list ?? []).map((s) => s.name); + // Only LLM trace streams belong in the Agent Graph picker — a plain + // service/HTTP trace stream has no agents or gen_ai edges to graph, so + // offering it would just resolve to an empty graph. `is_llm_stream` is the + // backend-maintained flag (auto-detected at ingest from gen_ai_* columns). + // Exclude only streams explicitly flagged non-LLM, matching the same filter + // used by LLM Insights and Sessions. + availableStreams.value = (res?.list ?? []) + .filter((s) => s.settings?.is_llm_stream !== false) + .map((s) => s.name); if ( availableStreams.value.length && !availableStreams.value.includes(activeStream.value)