mur_chat/

adapter.rs

//! Chat Adapter — bridges SSE workflow events to chat platform messages.
//!
//! Subscribes to the EventBroadcaster and converts WorkflowEvents into
//! chat messages sent via the ChatPlatform trait. Each workflow execution
//! gets its own thread for organized conversation.

use crate::platform::{
    ChatPlatform, ProgressStatus, ProgressUpdate, WorkflowNotification,
};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::broadcast;
use tokio::sync::RwLock;
use tracing::{debug, error, info};

/// SSE event types that the adapter consumes.
/// Mirrors mur_daemon::sse::WorkflowEvent but decoupled to avoid circular deps.
#[derive(Debug, Clone)]
pub enum WorkflowEvent {
    Started {
        execution_id: String,
        workflow_id: String,
        total_steps: usize,
    },
    StepStarted {
        execution_id: String,
        step_index: usize,
        step_name: String,
    },
    StepCompleted {
        execution_id: String,
        step_index: usize,
        step_name: String,
        success: bool,
        output: String,
        duration_ms: u64,
    },
    BreakpointHit {
        execution_id: String,
        step_index: usize,
        step_name: String,
        message: String,
    },
    Completed {
        execution_id: String,
        success: bool,
        steps_completed: usize,
        total_steps: usize,
        duration_ms: u64,
        error: Option<String>,
    },
}

/// Tracks per-execution thread state.
#[derive(Debug, Clone)]
struct ExecutionThread {
    thread_id: String,
    channel_id: String,
    workflow_id: String,
    total_steps: usize,
}

/// Chat adapter that converts workflow events to chat messages.
pub struct ChatAdapter<P: ChatPlatform> {
    platform: Arc<P>,
    channel_id: String,
    /// Maps execution_id → thread state.
    threads: Arc<RwLock<HashMap<String, ExecutionThread>>>,
    shadow_default: bool,
}

impl<P: ChatPlatform + 'static> ChatAdapter<P> {
    pub fn new(platform: Arc<P>, channel_id: String) -> Self {
        Self {
            platform,
            channel_id,
            threads: Arc::new(RwLock::new(HashMap::new())),
            shadow_default: false,
        }
    }

    pub fn with_shadow_default(mut self, shadow: bool) -> Self {
        self.shadow_default = shadow;
        self
    }

    /// Start listening to a broadcast channel of workflow events.
    /// Runs until the channel is closed or an error occurs.
    pub async fn run(&self, mut rx: broadcast::Receiver<WorkflowEvent>) {
        info!("ChatAdapter started, forwarding events to chat");

        loop {
            match rx.recv().await {
                Ok(event) => {
                    if let Err(e) = self.handle_event(event).await {
                        error!("ChatAdapter error: {}", e);
                    }
                }
                Err(broadcast::error::RecvError::Lagged(n)) => {
                    debug!("ChatAdapter lagged by {} events", n);
                }
                Err(broadcast::error::RecvError::Closed) => {
                    info!("ChatAdapter: event channel closed");
                    break;
                }
            }
        }
    }

    async fn handle_event(&self, event: WorkflowEvent) -> anyhow::Result<()> {
        match event {
            WorkflowEvent::Started {
                execution_id,
                workflow_id,
                total_steps,
            } => {
                let thread_id = self
                    .platform
                    .start_thread(
                        &self.channel_id,
                        &execution_id,
                        &workflow_id,
                        total_steps,
                        self.shadow_default,
                    )
                    .await?;

                let mut threads = self.threads.write().await;
                threads.insert(
                    execution_id,
                    ExecutionThread {
                        thread_id,
                        channel_id: self.channel_id.clone(),
                        workflow_id,
                        total_steps,
                    },
                );
            }

            WorkflowEvent::StepStarted {
                execution_id,
                step_index,
                step_name,
            } => {
                let threads = self.threads.read().await;
                if let Some(thread) = threads.get(&execution_id) {
                    let progress = ProgressUpdate {
                        execution_id: execution_id.clone(),
                        workflow_id: thread.workflow_id.clone(),
                        step_index,
                        total_steps: thread.total_steps,
                        step_name,
                        status: ProgressStatus::StepRunning,
                        output: None,
                        duration_ms: None,
                    };
                    self.platform
                        .send_progress(&thread.channel_id, &thread.thread_id, &progress)
                        .await?;
                }
            }

            WorkflowEvent::StepCompleted {
                execution_id,
                step_index,
                step_name,
                success,
                output,
                duration_ms,
            } => {
                let threads = self.threads.read().await;
                if let Some(thread) = threads.get(&execution_id) {
                    let status = if success {
                        ProgressStatus::StepDone
                    } else {
                        ProgressStatus::StepFailed
                    };
                    let trimmed = if output.len() > 200 {
                        format!("{}...", &output[..output.char_indices().take_while(|(i, _)| *i < 200).last().map(|(i, c)| i + c.len_utf8()).unwrap_or(0)])
                    } else {
                        output
                    };
                    let progress = ProgressUpdate {
                        execution_id: execution_id.clone(),
                        workflow_id: thread.workflow_id.clone(),
                        step_index,
                        total_steps: thread.total_steps,
                        step_name,
                        status,
                        output: Some(trimmed),
                        duration_ms: Some(duration_ms),
                    };
                    self.platform
                        .send_progress(&thread.channel_id, &thread.thread_id, &progress)
                        .await?;
                }
            }

            WorkflowEvent::BreakpointHit {
                execution_id,
                step_name,
                message,
                ..
            } => {
                let threads = self.threads.read().await;
                if let Some(thread) = threads.get(&execution_id) {
                    let approval = crate::platform::ApprovalRequest {
                        execution_id,
                        step_name,
                        description: message,
                        action: "breakpoint".into(),
                        allowed_approvers: Vec::new(),
                    };
                    self.platform
                        .send_approval(&thread.channel_id, &approval)
                        .await?;
                }
            }

            WorkflowEvent::Completed {
                execution_id,
                success,
                steps_completed,
                total_steps,
                duration_ms,
                error,
            } => {
                // Clone data while holding read lock, then drop it before network calls
                let thread_info = {
                    let threads = self.threads.read().await;
                    threads.get(&execution_id).cloned()
                };

                if let Some(thread) = thread_info {
                    let notification = WorkflowNotification {
                        execution_id: execution_id.clone(),
                        workflow_id: thread.workflow_id.clone(),
                        success,
                        steps_completed,
                        total_steps,
                        duration_ms,
                        error,
                    };
                    self.platform
                        .send_notification(
                            &thread.channel_id,
                            Some(&thread.thread_id),
                            &notification,
                        )
                        .await?;

                    self.platform
                        .send_notification(&thread.channel_id, None, &notification)
                        .await?;

                    // Clean up thread tracking
                    let mut threads = self.threads.write().await;
                    threads.remove(&execution_id);
                }
            }
        }
        Ok(())
    }

    /// Get the thread ID for an active execution.
    pub async fn thread_for_execution(&self, execution_id: &str) -> Option<String> {
        self.threads
            .read()
            .await
            .get(execution_id)
            .map(|t| t.thread_id.clone())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::platform::{ApprovalRequest, OutgoingMessage};
    use std::sync::Mutex;

    /// Mock chat platform for testing.
    struct MockPlatform {
        messages: Arc<Mutex<Vec<String>>>,
    }

    impl MockPlatform {
        fn new() -> Self {
            Self {
                messages: Arc::new(Mutex::new(Vec::new())),
            }
        }
    }

    impl ChatPlatform for MockPlatform {
        async fn send_message(&self, msg: &OutgoingMessage) -> anyhow::Result<String> {
            self.messages.lock().unwrap().push(msg.text.clone());
            Ok("mock-ts".into())
        }

        async fn send_approval(
            &self,
            _channel_id: &str,
            _request: &ApprovalRequest,
        ) -> anyhow::Result<String> {
            Ok("mock-ts".into())
        }

        async fn update_message(
            &self,
            _channel_id: &str,
            _message_id: &str,
            _text: &str,
        ) -> anyhow::Result<()> {
            Ok(())
        }

        async fn add_reaction(
            &self,
            _channel_id: &str,
            _message_id: &str,
            _emoji: &str,
        ) -> anyhow::Result<()> {
            Ok(())
        }

        async fn send_progress(
            &self,
            _channel_id: &str,
            _thread_id: &str,
            progress: &ProgressUpdate,
        ) -> anyhow::Result<String> {
            self.messages
                .lock()
                .unwrap()
                .push(format!("progress:{}", progress.step_name));
            Ok("mock-ts".into())
        }

        async fn send_notification(
            &self,
            _channel_id: &str,
            _thread_id: Option<&str>,
            notification: &WorkflowNotification,
        ) -> anyhow::Result<String> {
            self.messages.lock().unwrap().push(format!(
                "notification:{}:{}",
                notification.workflow_id, notification.success
            ));
            Ok("mock-ts".into())
        }

        async fn start_thread(
            &self,
            _channel_id: &str,
            _execution_id: &str,
            workflow_id: &str,
            _total_steps: usize,
            _shadow: bool,
        ) -> anyhow::Result<String> {
            self.messages
                .lock()
                .unwrap()
                .push(format!("thread:{}", workflow_id));
            Ok("mock-thread-ts".into())
        }
    }

    #[tokio::test]
    async fn test_adapter_full_lifecycle() {
        let platform = Arc::new(MockPlatform::new());
        let _adapter = ChatAdapter::new(platform.clone(), "#test".into());
        let (tx, rx) = broadcast::channel(16);

        // Run adapter in background
        let adapter_handle = tokio::spawn({
            let adapter = ChatAdapter::new(platform.clone(), "#test".into());
            async move { adapter.run(rx).await }
        });

        // Simulate workflow lifecycle
        tx.send(WorkflowEvent::Started {
            execution_id: "e1".into(),
            workflow_id: "deploy".into(),
            total_steps: 2,
        })
        .unwrap();

        tx.send(WorkflowEvent::StepStarted {
            execution_id: "e1".into(),
            step_index: 0,
            step_name: "build".into(),
        })
        .unwrap();

        tx.send(WorkflowEvent::StepCompleted {
            execution_id: "e1".into(),
            step_index: 0,
            step_name: "build".into(),
            success: true,
            output: "OK".into(),
            duration_ms: 100,
        })
        .unwrap();

        tx.send(WorkflowEvent::Completed {
            execution_id: "e1".into(),
            success: true,
            steps_completed: 2,
            total_steps: 2,
            duration_ms: 500,
            error: None,
        })
        .unwrap();

        // Give adapter time to process
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        // Drop sender to close channel
        drop(tx);
        let _ = tokio::time::timeout(
            std::time::Duration::from_millis(200),
            adapter_handle,
        )
        .await;

        let messages = platform.messages.lock().unwrap();
        assert!(
            messages.iter().any(|m| m.contains("thread:deploy")),
            "Should have started a thread"
        );
        assert!(
            messages.iter().any(|m| m.contains("progress:build")),
            "Should have sent progress"
        );
        assert!(
            messages.iter().any(|m| m.contains("notification:deploy:true")),
            "Should have sent completion notification"
        );
    }

    #[tokio::test]
    async fn test_adapter_thread_tracking() {
        let platform = Arc::new(MockPlatform::new());
        let adapter = ChatAdapter::new(platform.clone(), "#test".into());

        // No thread initially
        assert!(adapter.thread_for_execution("e1").await.is_none());
    }
}