fix: agent-signals repopulate signals reliably; scope UI to LLM streams (#13348)

Companion to the o2-enterprise change. On a live instance the Agent
Behavior/Graph views were empty because the rollup never re-processed
the agent trace streams, and the stream pickers offered non-LLM streams.

service_graph/processor.rs — discovery gap: the rollup discovered
streams from a usage-windowed query, falling back to the schema cache
only when usage was ENTIRELY empty. So any stream that dropped out of
the current window's usage (producer paused, bursty ingest) was silently
skipped forever on a busy instance — exactly how the agent streams went
dark. Replace the all-or-nothing fallback with a union: usage-active
streams ∪ every schema-cache trace stream (deduped). Idle windows are
cheap no-ops, so this doesn't widen the worst case.

agent_signals/processor.rs — always-on + stream-level LLM gate: drop the
removed O2_AGENT_SIGNALS_ENABLED check, and add an early return that
skips streams whose schema has no gen_ai_operation_name (mirrors the
service-graph has_gen_ai gate) so non-LLM streams cost nothing. Thread
has_agent_id into the SQL builders for the schema-gated agent-id
predicate.

AgentGraphPage/AgentBehaviorPage.vue — filter the stream picker to LLM
streams (settings.is_llm_stream !== false), matching LLM Insights /
Sessions. A plain service trace stream has no agents or signals to show
here.

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());
     }

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<RecentIngestedTraceStream> = 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,

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];
     }

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)