dirigent/crates/dirigent_langfuse/src/mapping.rs

//! BusEvent → Langfuse ingestion mapping.
//!
//! Maps the common BusEvent kinds to Langfuse ingestion items (traces,
//! generations, spans). Events without a `scroll_id` are dropped —
//! Langfuse requires a trace id up-front.

// The items below are only wired into the stream when the `server`
// feature is on; the default-feature build keeps them for symmetry but
// does not reference them, so allow dead-code warnings there.
#![cfg_attr(not(feature = "server"), allow(dead_code))]

use chrono::{DateTime, Utc};
use serde::Serialize;
use uuid::Uuid;

use dirigent_protocol::{streaming::BusEvent, Event};

/// A single Langfuse ingestion item.
///
/// Batched into `{ "batch": [...] }` in `LangfuseClient::ingest_batch`.
#[derive(Debug, Clone, Serialize)]
pub struct IngestItem {
    pub id: String, // UUIDv7
    pub timestamp: DateTime<Utc>,
    #[serde(rename = "type")]
    pub kind: IngestKind,
    pub body: serde_json::Value,
}

#[derive(Debug, Clone, Copy, Serialize)]
#[serde(rename_all = "kebab-case")]
#[allow(dead_code)] // SpanCreate/SpanUpdate reserved for future tool-call mapping
pub enum IngestKind {
    TraceCreate,
    GenerationCreate,
    GenerationUpdate,
    SpanCreate,
    SpanUpdate,
}

pub fn bus_event_to_items(bus_event: &BusEvent) -> Vec<IngestItem> {
    let Some(scroll_id) = bus_event.routing.scroll_id else {
        // No scroll_id binding yet — drop. Upstream callers may choose to
        // buffer pending events keyed by (connector_id, native_id) until
        // SessionRegistered arrives; Phase 4 scope: drop and log.
        return Vec::new();
    };

    let trace_id = scroll_id.to_string();
    let now = Utc::now();

    match &*bus_event.event {
        Event::SessionCreated { session, .. } => {
            // `session.title` is a `String`; fall back to the id if empty.
            let name = if session.title.is_empty() {
                session.id.clone()
            } else {
                session.title.clone()
            };
            vec![IngestItem {
                id: Uuid::now_v7().to_string(),
                timestamp: now,
                kind: IngestKind::TraceCreate,
                body: serde_json::json!({
                    "id": trace_id,
                    "name": name,
                }),
            }]
        }
        Event::MessageStarted { message, .. } => {
            vec![IngestItem {
                id: Uuid::now_v7().to_string(),
                timestamp: now,
                kind: IngestKind::GenerationCreate,
                body: serde_json::json!({
                    "id": message.id,
                    "traceId": trace_id,
                    "name": format!("{:?}", message.role),
                    "startTime": message.created_at,
                }),
            }]
        }
        Event::MessageCompleted { message, .. } => {
            vec![IngestItem {
                id: Uuid::now_v7().to_string(),
                timestamp: now,
                kind: IngestKind::GenerationUpdate,
                body: serde_json::json!({
                    "id": message.id,
                    "traceId": trace_id,
                    "endTime": now,
                    "output": serialize_content(&message.content),
                }),
            }]
        }
        Event::TurnComplete { .. } => Vec::new(), // captured by MessageCompleted
        // SessionUpdate::ToolCall* — would need a case-by-case mapping; out of
        // Phase 4 scope. Return empty for now.
        _ => Vec::new(),
    }
}

fn serialize_content(parts: &[dirigent_protocol::MessagePart]) -> serde_json::Value {
    serde_json::to_value(parts).unwrap_or(serde_json::Value::Null)
}

#[cfg(test)]
mod tests {
    use super::*;
    use dirigent_protocol::streaming::{BusEvent, EventKind, EventOrigin, EventRouting};
    use dirigent_protocol::{Event, Message, MessageRole, MessageStatus};
    use std::sync::Arc;

    fn make_bus_event_with_scroll(event: Event, scroll_id: Uuid) -> BusEvent {
        BusEvent {
            routing: EventRouting {
                scroll_id: Some(scroll_id),
                connector_uid: Some(Uuid::new_v4()),
                connector_id: Some("c".into()),
                native_session_id: Some("s".into()),
                kind: EventKind::Message,
            },
            origin: EventOrigin::Runtime,
            event: Arc::new(event),
        }
    }

    #[test]
    fn message_started_produces_generation_create() {
        let scroll_id = Uuid::new_v4();
        let msg = Message {
            id: "m1".into(),
            session_id: "s".into(),
            role: MessageRole::Assistant,
            created_at: chrono::Utc::now(),
            content: vec![],
            status: MessageStatus::Streaming,
            metadata: None,
        };
        let bus_event = make_bus_event_with_scroll(
            Event::MessageStarted {
                connector_id: "c".into(),
                message: msg,
            },
            scroll_id,
        );
        let items = bus_event_to_items(&bus_event);
        assert_eq!(items.len(), 1);
        assert!(matches!(items[0].kind, IngestKind::GenerationCreate));
    }

    #[test]
    fn no_scroll_id_drops_event() {
        let event = Event::Connected;
        let bus_event = BusEvent {
            routing: EventRouting::default(),
            origin: EventOrigin::Runtime,
            event: Arc::new(event),
        };
        let items = bus_event_to_items(&bus_event);
        assert_eq!(items.len(), 0);
    }

    #[test]
    fn unmapped_event_returns_empty() {
        // `Connected` is not one of our mapped variants even when a scroll_id
        // is bound → expect 0 items.
        let scroll_id = Uuid::new_v4();
        let bus_event = make_bus_event_with_scroll(Event::Connected, scroll_id);
        let items = bus_event_to_items(&bus_event);
        assert_eq!(items.len(), 0);
    }
}