trace summaries triggers (#952)

* tmp

* refactor

* tmp

* tmp

* trigger span id

* tmp

Author: skull8888888

app-server/src/cache/keys.rs

@@ -6,5 +6,6 @@ pub const PROJECT_API_KEY_CACHE_KEY: &str = "project_api_key";
 pub const PROJECT_CACHE_KEY: &str = "project";
 pub const WORKSPACE_LIMITS_CACHE_KEY: &str = "workspace_limits";
 pub const PROJECT_EVALUATORS_BY_PATH_CACHE_KEY: &str = "project_evaluators_by_path";
+pub const SUMMARY_TRIGGER_SPANS_CACHE_KEY: &str = "summary_trigger_spans";
 
 pub const WORKSPACE_BYTES_USAGE_CACHE_KEY: &str = "workspace_bytes_usage";

app-server/src/db/mod.rs

@@ -13,6 +13,7 @@ pub mod projects;
 pub mod provider_api_keys;
 pub mod spans;
 pub mod stats;
+pub mod summary_trigger_spans;
 pub mod tags;
 pub mod trace;
 pub mod user_cookies;

app-server/src/db/summary_trigger_spans.rs

@@ -0,0 +1,86 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+/// Event definition with semantic analysis configuration
+#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
+pub struct EventDefinition {
+    pub id: Uuid,
+    pub name: String,
+    pub prompt: Option<String>,
+    pub structured_output: Option<Value>,
+}
+
+/// Summary trigger span with joined event definition
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SummaryTriggerSpanWithEvent {
+    pub span_name: String,
+    pub event_definition: Option<EventDefinition>,
+}
+
+#[derive(Debug, Clone, sqlx::FromRow)]
+struct DBTriggerSpanWithEvent {
+    span_name: String,
+    event_definition_id: Option<Uuid>,
+    event_definition_name: Option<String>,
+    event_definition_prompt: Option<String>,
+    event_definition_structured_output: Option<Value>,
+}
+
+/// Get summary trigger spans for a project with their associated semantic event definitions
+/// Returns all trigger spans for the project
+/// Only joins semantic event definitions (is_semantic = true) via the LEFT JOIN condition
+/// Triggers without event definitions will have event_definition = None
+pub async fn get_summary_trigger_spans_with_events(
+    pool: &PgPool,
+    project_id: Uuid,
+) -> Result<Vec<SummaryTriggerSpanWithEvent>, sqlx::Error> {
+    let results = sqlx::query_as::<_, DBTriggerSpanWithEvent>(
+        r#"
+        SELECT 
+            sts.span_name as span_name,
+            ed.id as event_definition_id,
+            ed.name as event_definition_name,
+            ed.prompt as event_definition_prompt,
+            ed.structured_output as event_definition_structured_output
+        FROM 
+            summary_trigger_spans sts
+        LEFT JOIN 
+            event_definitions ed 
+            ON sts.event_name = ed.name 
+            AND sts.project_id = ed.project_id
+            AND ed.is_semantic = true
+        WHERE 
+            sts.project_id = $1
+        "#,
+    )
+    .bind(project_id)
+    .fetch_all(pool)
+    .await?;
+
+    Ok(results
+        .into_iter()
+        .map(|db_trigger_span_with_event| {
+            let event_definition = if let Some(id) = db_trigger_span_with_event.event_definition_id
+            {
+                Some(EventDefinition {
+                    id,
+                    name: db_trigger_span_with_event
+                        .event_definition_name
+                        .unwrap_or_default(),
+                    prompt: db_trigger_span_with_event.event_definition_prompt,
+                    structured_output: db_trigger_span_with_event
+                        .event_definition_structured_output,
+                })
+            } else {
+                None
+            };
+
+            SummaryTriggerSpanWithEvent {
+                span_name: db_trigger_span_with_event.span_name,
+                event_definition,
+            }
+        })
+        .collect())
+}

app-server/src/traces/consumer.rs

@@ -12,6 +12,7 @@ use uuid::Uuid;
 use super::{
     OBSERVATIONS_EXCHANGE, OBSERVATIONS_QUEUE, OBSERVATIONS_ROUTING_KEY,
     summary::push_to_trace_summary_queue,
+    trigger::{check_span_trigger, get_summary_trigger_spans_cached},
 };
 use crate::{
     api::v1::traces::RabbitMqSpanMessage,
@@ -236,6 +237,13 @@ async fn process_batch(
     let mut span_usage_vec = Vec::new();
     let mut all_events = Vec::new();
 
+    // we get project id from the first span in the batch
+    // because all spans in the batch have the same project id
+    // batching is happening on the Otel SpanProcessor level
+    // project_id can never be None, because batch is never empty
+    // but we do unwrap_or_default to avoid Option<Uuid> in the rest of the code
+    let project_id = spans.first().map(|s| s.project_id).unwrap_or_default();
+
     for span in &mut spans {
         let span_usage =
             get_llm_usage_for_span(&mut span.attributes, db.clone(), cache.clone(), &span.name)
@@ -330,21 +338,9 @@ async fn process_batch(
         }
     }
 
-    // Check for completed traces (top-level spans) and push to trace summary queue
-    for span in &spans {
-        if span.parent_span_id.is_none() {
-            if let Err(e) =
-                push_to_trace_summary_queue(span.trace_id, span.project_id, queue.clone()).await
-            {
-                log::error!(
-                    "Failed to push trace completion to summary queue: trace_id={}, project_id={}, error={:?}",
-                    span.trace_id,
-                    span.project_id,
-                    e
-                );
-            }
-        }
-    }
+    // Check for spans matching trigger conditions and push to trace summary queue
+    check_and_push_trace_summaries(project_id, &spans, db.clone(), cache.clone(), queue.clone())
+        .await;
 
     // Send realtime messages directly to SSE connections after successful ClickHouse writes
     send_realtime_messages_to_sse(&spans, &sse_connections).await;
@@ -379,26 +375,21 @@ async fn process_batch(
         .sum::<usize>()
         + total_events_ingested_bytes;
 
-    // we get project id from the first span in the batch
-    // because all spans in the batch have the same project id
-    // batching is happening on the Otel SpanProcessor level
-    if let Some(project_id) = stripped_spans.first().map(|s| s.project_id) {
-        if is_feature_enabled(Feature::UsageLimit) {
-            if let Err(e) = update_workspace_limit_exceeded_by_project_id(
-                db.clone(),
-                clickhouse.clone(),
-                cache.clone(),
+    if is_feature_enabled(Feature::UsageLimit) {
+        if let Err(e) = update_workspace_limit_exceeded_by_project_id(
+            db.clone(),
+            clickhouse.clone(),
+            cache.clone(),
+            project_id,
+            total_ingested_bytes,
+        )
+        .await
+        {
+            log::error!(
+                "Failed to update workspace limit exceeded for project [{}]: {:?}",
                 project_id,
-                total_ingested_bytes,
-            )
-            .await
-            {
-                log::error!(
-                    "Failed to update workspace limit exceeded for project [{}]: {:?}",
-                    project_id,
-                    e
-                );
-            }
+                e
+            );
         }
     }
 
@@ -513,3 +504,51 @@ fn span_to_realtime_span(span: &Span) -> Value {
         // Note: input and output fields are intentionally excluded for performance
     })
 }
+
+/// Check spans against trigger conditions and push matching traces to summary queue
+/// This function groups spans by project to minimize database/cache queries
+async fn check_and_push_trace_summaries(
+    project_id: Uuid,
+    spans: &[Span],
+    db: Arc<DB>,
+    cache: Arc<Cache>,
+    queue: Arc<MessageQueue>,
+) {
+    match get_summary_trigger_spans_cached(db.clone(), cache.clone(), project_id).await {
+        Ok(trigger_spans) => {
+            // Check each span against its project's trigger spans
+            for span in spans {
+                // Check if this span name matches any trigger
+                let matching_triggers = check_span_trigger(&span.name, &trigger_spans);
+
+                // Send one message per matching trigger
+                for trigger in matching_triggers {
+                    if let Err(e) = push_to_trace_summary_queue(
+                        span.trace_id,
+                        span.project_id,
+                        span.span_id,
+                        trigger.event_definition,
+                        queue.clone(),
+                    )
+                    .await
+                    {
+                        log::error!(
+                            "Failed to push trace completion to summary queue: trace_id={}, project_id={}, span_name={}, error={:?}",
+                            span.trace_id,
+                            span.project_id,
+                            span.name,
+                            e
+                        );
+                    }
+                }
+            }
+        }
+        Err(e) => {
+            log::error!(
+                "Failed to get summary trigger spans for project {}: {:?}",
+                project_id,
+                e
+            );
+        }
+    }
+}

app-server/src/traces/eligibility.rs

@@ -15,7 +15,6 @@ use crate::{
 #[derive(Debug, Clone)]
 pub struct TraceEligibilityResult {
     pub is_eligible: bool,
-    pub reason: Option<String>,
     #[allow(dead_code)]
     pub tier_name: Option<String>,
     #[allow(dead_code)]
@@ -41,7 +40,6 @@ pub async fn check_trace_eligibility(
                 Err(_) => {
                     return Ok(TraceEligibilityResult {
                         is_eligible: false,
-                        reason: Some("project not found".to_string()),
                         tier_name: None,
                         has_trace_analysis: false,
                     });
@@ -58,7 +56,6 @@ pub async fn check_trace_eligibility(
     if !is_paid_tier {
         return Ok(TraceEligibilityResult {
             is_eligible: false,
-            reason: Some("workspace is on free tier".to_string()),
             tier_name,
             has_trace_analysis: false,
         });
@@ -70,15 +67,13 @@ pub async fn check_trace_eligibility(
     if !is_trace_analysis_enabled {
         return Ok(TraceEligibilityResult {
             is_eligible: false,
-            reason: Some("trace analysis not enabled in project settings".to_string()),
             tier_name,
             has_trace_analysis: false,
         });
     }
 
     Ok(TraceEligibilityResult {
         is_eligible: true,
-        reason: None,
         tier_name,
         has_trace_analysis: true,
     })

app-server/src/traces/mod.rs

@@ -8,6 +8,7 @@ pub mod provider;
 pub mod span_attributes;
 pub mod spans;
 pub mod summary;
+pub mod trigger;
 pub mod utils;
 
 pub const OBSERVATIONS_QUEUE: &str = "observations_queue";