mnote/rust/crates/mnote-web/src/acp_bridge.rs

/// ACP ↔ SSE bridge for Hermes route integration.
///
/// Transforms ACP session events into the SSE event format expected by the
/// frontend (HermesRunEvent), and manages the background prompt lifecycle.
///
/// Reference: `hermes-vscode-main/src/sessionManager.ts` handleUpdate()
use crate::acp_runtime::AcpRuntimeManager;
use crate::acp_session_manager::{AcpSessionEvent, AcpSessionManager};
use crate::acp_types::ContentBlock;
use axum::body::Body;
use axum::http::{header, StatusCode};
use axum::response::Response;
use serde_json::{json, Value};
use std::sync::Arc;
use tokio::sync::broadcast;
use tracing::{info, warn};

/// Errors from the ACP bridge.
#[derive(Debug)]
pub enum AcpBridgeError {
    NoActiveRuntime,
    SessionError(String),
    StreamError(String),
}

impl std::fmt::Display for AcpBridgeError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            AcpBridgeError::NoActiveRuntime => write!(f, "no active ACP runtime"),
            AcpBridgeError::SessionError(msg) => write!(f, "ACP session error: {msg}"),
            AcpBridgeError::StreamError(msg) => write!(f, "ACP stream error: {msg}"),
        }
    }
}

/// SSE event types sent to the frontend.
/// Mirrors HermesRunEvent from bridge.ts.
#[derive(Debug, Clone)]
pub struct SseEvent {
    pub event: String,
    pub data: Value,
}

/// Bridge state for one run: holds the broadcast channel for SSE events.
pub struct AcpRunBridge {
    session_id: String,
    event_tx: broadcast::Sender<SseEvent>,
}

impl AcpRunBridge {
    /// Create a new ACP run: create session + start prompt in background.
    ///
    /// Returns a bridge with a broadcast receiver that the SSE endpoint can use.
    pub async fn start(
        runtime_mgr: &AcpRuntimeManager,
        runtime_name: &str,
        prompt_blocks: Vec<ContentBlock>,
    ) -> Result<Self, AcpBridgeError> {
        // Get or activate the runtime
        let client = if runtime_mgr.is_active().await {
            runtime_mgr
                .active_client()
                .await
                .ok_or(AcpBridgeError::NoActiveRuntime)?
        } else {
            runtime_mgr
                .switch_to(runtime_name)
                .await
                .map_err(|e| AcpBridgeError::SessionError(e.to_string()))?
        };

        // Create session manager
        let mgr = Arc::new(AcpSessionManager::new(client));

        // Create event channel (256 buffered, enough for SSE streaming)
        let (event_tx, _) = broadcast::channel(256);
        let event_tx_clone = event_tx.clone();

        // Set up event handler
        mgr.on_event(move |event| {
            if let Some(sse) = acp_event_to_sse(event) {
                let _ = event_tx_clone.send(sse);
            }
        });

        // Create session
        let sid = mgr
            .create_session(None, None)
            .await
            .map_err(|e| AcpBridgeError::SessionError(e.to_string()))?;

        // Start prompt in background
        let mgr_clone = mgr.clone();
        let event_tx_prompt = event_tx.clone();
        tokio::spawn(async move {
            match mgr_clone.run_prompt(prompt_blocks).await {
                Ok(result) => {
                    info!("ACP prompt completed: stop_reason={:?}", result.stop_reason);
                    let _ = event_tx_prompt.send(SseEvent {
                        event: "run.completed".into(),
                        data: json!({
                            "stopReason": format!("{:?}", result.stop_reason),
                        }),
                    });
                }
                Err(e) => {
                    warn!("ACP prompt failed: {e}");
                    let _ = event_tx_prompt.send(SseEvent {
                        event: "run.failed".into(),
                        data: json!({ "error": e.to_string() }),
                    });
                }
            }
        });

        info!("ACP run started: session={}", sid);
        Ok(Self {
            session_id: sid,
            event_tx,
        })
    }

    /// Cancel the current run.
    pub async fn abort(&self) {
        // Cancellation is sent via the session manager.
        // For now, we just drop the bridge — the background task will detect this
        // via the broadcast channel being closed.
        info!("ACP run aborted: session={}", self.session_id);
    }

    /// Create an SSE response body from the event broadcast receiver.
    pub fn into_sse_response(self) -> Response {
        use tokio::sync::mpsc;
        let (tx, rx) = mpsc::channel::<Result<axum::body::Bytes, std::convert::Infallible>>(256);
        let mut broadcast_rx = self.event_tx.subscribe();

        // Forward events from broadcast to mpsc
        tokio::spawn(async move {
            loop {
                match broadcast_rx.recv().await {
                    Ok(event) => {
                        let json =
                            serde_json::to_string(&event.data).unwrap_or_else(|_| "{}".into());
                        let bytes = axum::body::Bytes::from(format!(
                            "event: {}\ndata: {}\n\n",
                            event.event, json
                        ));
                        if tx.send(Ok(bytes)).await.is_err() {
                            break; // receiver dropped
                        }
                    }
                    Err(broadcast::error::RecvError::Lagged(n)) => {
                        warn!("ACP SSE lagged: {n} events dropped");
                    }
                    Err(broadcast::error::RecvError::Closed) => {
                        break; // stream ended
                    }
                }
            }
        });

        let stream = tokio_stream::wrappers::ReceiverStream::new(rx);

        Response::builder()
            .status(StatusCode::OK)
            .header(header::CONTENT_TYPE, "text/event-stream; charset=utf-8")
            .header(header::CACHE_CONTROL, "no-cache, no-transform")
            .header("x-accel-buffering", "no")
            .body(Body::from_stream(stream))
            .unwrap_or_else(|_| {
                Response::builder()
                    .status(StatusCode::INTERNAL_SERVER_ERROR)
                    .body(Body::empty())
                    .unwrap()
            })
    }
}

/// Map an AcpSessionEvent to an SSE event for the frontend.
///
/// Reference: hermes-vscode-main protocol.ts extractTextContent/parseToolCall/parseToolCallUpdate
/// Reference: wolai-frontend bridge.ts HermesRunEvent type
pub fn acp_event_to_sse(event: AcpSessionEvent) -> Option<SseEvent> {
    match event {
        AcpSessionEvent::TextDelta { text } => Some(SseEvent {
            event: "message.delta".into(),
            data: json!({ "delta": text }),
        }),
        AcpSessionEvent::ThoughtDelta { text } => Some(SseEvent {
            event: "thought.delta".into(),
            data: json!({ "delta": text }),
        }),
        AcpSessionEvent::ToolCall {
            tool_call_id,
            title,
            kind,
            ..
        } => Some(SseEvent {
            event: "tool.started".into(),
            data: json!({
                "tool": title,
                "toolCallId": tool_call_id,
                "kind": kind,
            }),
        }),
        AcpSessionEvent::ToolCallUpdate {
            tool_call_id,
            status,
        } => {
            let error = status == crate::acp_types::ToolCallStatus::Failed;
            Some(SseEvent {
                event: "tool.completed".into(),
                data: json!({
                    "toolCallId": tool_call_id,
                    "error": error,
                }),
            })
        }
        AcpSessionEvent::UsageUpdate { used, size } => Some(SseEvent {
            event: "usage.updated".into(),
            data: json!({ "used": used, "size": size }),
        }),
        AcpSessionEvent::SessionInfoUpdate { .. } => {
            None // Not forwarded to frontend
        }
        AcpSessionEvent::PlanUpdate { .. } => {
            None // Not forwarded (Phase C)
        }
        AcpSessionEvent::Disconnected { reason } if reason == "session closed" => None,
        AcpSessionEvent::Disconnected { reason } => Some(SseEvent {
            event: "run.failed".into(),
            data: json!({ "error": reason }),
        }),
    }
}

/// Helper: get the runtime name from a profile.
/// For now, we use "hermes" or "reasonix" directly.
/// In Step 12, this will come from the profile config.
pub fn runtime_name_for_profile(profile: &str) -> &str {
    match profile {
        "reasonix" => "reasonix",
        _ => "hermes",
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn acp_normal_session_close_does_not_emit_failure() {
        let event = AcpSessionEvent::Disconnected {
            reason: "session closed".into(),
        };

        assert!(acp_event_to_sse(event).is_none());
    }

    #[test]
    fn acp_unexpected_disconnect_emits_failure() {
        let event = AcpSessionEvent::Disconnected {
            reason: "transport lost".into(),
        };

        let sse = acp_event_to_sse(event).expect("unexpected disconnect should be forwarded");
        assert_eq!(sse.event, "run.failed");
        assert_eq!(sse.data["error"], "transport lost");
    }

    #[test]
    fn acp_thought_delta_does_not_emit_message_delta() {
        let event = AcpSessionEvent::ThoughtDelta {
            text: "internal reasoning".into(),
        };

        let sse = acp_event_to_sse(event).expect("thought delta should be forwarded separately");
        assert_eq!(sse.event, "thought.delta");
        assert_eq!(sse.data["delta"], "internal reasoning");
    }
}