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::collections::HashSet;
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>,
}

fn collect_citation_markdowns_from_value(value: &Value) -> Vec<Value> {
    fn add_citation_value(value: &Value, seen: &mut HashSet<String>, out: &mut Vec<Value>) {
        let Some(citation) = value.get("citationMarkdown").and_then(Value::as_str) else {
            return;
        };
        let citation = citation.trim();
        if citation.is_empty() || !seen.insert(citation.to_string()) {
            return;
        }
        out.push(json!({
            "schema": value.get("schema").cloned().unwrap_or_else(|| json!("mnote.knowledge_rag.citation.v1")),
            "citationMarkdown": citation,
            "citationId": value.get("citationId").cloned().unwrap_or(Value::Null),
            "citationLabel": value.get("citationLabel").cloned().unwrap_or(Value::Null),
            "sourceRootRelativePath": value.get("sourceRootRelativePath").cloned().unwrap_or(Value::Null),
            "sourcePath": value.get("sourcePath").cloned().unwrap_or(Value::Null),
            "filePath": value.get("filePath").or_else(|| value.get("lightRagFilePath")).cloned().unwrap_or(Value::Null),
            "headingPath": value.get("headingPath").cloned().unwrap_or_else(|| json!([])),
            "displayQuote": value.get("displayQuote").or_else(|| value.get("quote")).cloned().unwrap_or(Value::Null),
            "locatorEvidenceText": value.get("locatorEvidenceText").cloned().unwrap_or(Value::Null),
            "locatorPrecision": value.get("locatorPrecision").cloned().unwrap_or(Value::Null),
            "locatorDegraded": value.get("locatorDegraded").cloned().unwrap_or(Value::Bool(true)),
            "citationUrl": value.get("citationUrl").cloned().unwrap_or(Value::Null),
        }));
    }

    fn add_reference_citations(
        references: &[Value],
        seen: &mut HashSet<String>,
        out: &mut Vec<Value>,
    ) -> bool {
        let has_precise = references.iter().any(|reference| {
            reference
                .get("citationMarkdown")
                .and_then(Value::as_str)
                .is_some_and(|value| !value.trim().is_empty())
                && reference.get("locatorDegraded").and_then(Value::as_bool) != Some(true)
        });
        let mut added = false;
        for reference in references {
            if out.len() >= 8 {
                break;
            }
            let Some(_citation) = reference.get("citationMarkdown").and_then(Value::as_str) else {
                continue;
            };
            if has_precise
                && reference.get("locatorDegraded").and_then(Value::as_bool) == Some(true)
            {
                continue;
            }
            let before = out.len();
            add_citation_value(reference, seen, out);
            added = added || out.len() > before;
        }
        added
    }

    fn visit(value: &Value, seen: &mut HashSet<String>, out: &mut Vec<Value>) {
        if out.len() >= 8 {
            return;
        }
        match value {
            Value::String(text) => {
                let trimmed = text.trim();
                if (trimmed.starts_with('{') || trimmed.starts_with('['))
                    && trimmed.contains("citationMarkdown")
                {
                    if let Ok(parsed) = serde_json::from_str::<Value>(trimmed) {
                        visit(&parsed, seen, out);
                    } else if let Some(first_line) = trimmed.lines().next() {
                        if let Ok(parsed) = serde_json::from_str::<Value>(first_line.trim()) {
                            visit(&parsed, seen, out);
                        }
                    }
                }
            }
            Value::Array(items) => {
                for item in items {
                    visit(item, seen, out);
                    if out.len() >= 8 {
                        break;
                    }
                }
            }
            Value::Object(map) => {
                let has_filtered_references = map
                    .get("references")
                    .and_then(Value::as_array)
                    .is_some_and(|references| add_reference_citations(references, seen, out));
                if let Some(_citation) = map.get("citationMarkdown").and_then(Value::as_str) {
                    if !has_filtered_references {
                        add_citation_value(value, seen, out);
                    }
                }
                for (key, item) in map {
                    if has_filtered_references
                        && matches!(key.as_str(), "references" | "citations" | "uiCitations")
                    {
                        continue;
                    }
                    visit(item, seen, out);
                    if out.len() >= 8 {
                        break;
                    }
                }
            }
            _ => {}
        }
    }

    let mut seen = HashSet::new();
    let mut out = Vec::new();
    visit(value, &mut seen, &mut out);
    out
}

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: recycle/wolai-frontend bridge.ts HermesRunEvent type (historic)
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,
            status,
            raw_input,
            locations,
        } => Some(SseEvent {
            event: "tool.started".into(),
            data: json!({
                "tool": title,
                "toolCallId": tool_call_id,
                "kind": kind,
                "status": status,
                "input": raw_input,
                "locations": locations,
            }),
        }),
        AcpSessionEvent::ToolCallUpdate {
            tool_call_id,
            status,
            content,
        } => {
            let output = json!(content);
            let citation_markdowns = collect_citation_markdowns_from_value(&output);
            let error = status == crate::acp_types::ToolCallStatus::Failed;
            let event = if error {
                "tool.failed"
            } else if status == crate::acp_types::ToolCallStatus::Completed {
                "tool.completed"
            } else {
                "tool.started"
            };
            Some(SseEvent {
                event: event.into(),
                data: json!({
                    "toolCallId": tool_call_id,
                    "status": status,
                    "error": error,
                    "output": output,
                    "citationMarkdowns": citation_markdowns,
                }),
            })
        }
        AcpSessionEvent::UsageUpdate { used, size } => Some(SseEvent {
            event: "usage.updated".into(),
            data: json!({ "used": used, "size": size }),
        }),
        AcpSessionEvent::PermissionRequest {
            permission_id,
            tool_name,
            params,
            decision,
        } => {
            let event = match decision.as_str() {
                "allowed" => "permission.allowed",
                "requested" => "permission.requested",
                _ => "permission.denied",
            };
            Some(SseEvent {
                event: event.into(),
                data: json!({
                    "permissionId": permission_id,
                    "toolName": tool_name,
                    "params": params,
                    "decision": decision,
                }),
            })
        }
        AcpSessionEvent::SessionInfoUpdate { title } => Some(SseEvent {
            event: "session.info.updated".into(),
            data: json!({ "title": title }),
        }),
        AcpSessionEvent::ProviderConversationBound {
            provider,
            remote_conversation_id,
            remote_url,
            acp_session_id,
        } => Some(SseEvent {
            event: "provider.conversation.bound".into(),
            data: json!({
                "provider": provider,
                "remoteConversationId": remote_conversation_id,
                "remoteUrl": remote_url,
                "acpSessionId": acp_session_id,
            }),
        }),
        AcpSessionEvent::PlanUpdate { entries } => Some(SseEvent {
            event: "plan.updated".into(),
            data: json!({ "entries": entries }),
        }),
        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");
    }

    #[test]
    fn acp_permission_request_emits_frontend_decision_event() {
        let event = AcpSessionEvent::PermissionRequest {
            permission_id: "perm_1".into(),
            tool_name: "mnote.page.save".into(),
            params: json!({"documentId": "doc_1"}),
            decision: "denied".into(),
        };

        let sse = acp_event_to_sse(event).expect("permission decision should be forwarded");
        assert_eq!(sse.event, "permission.denied");
        assert_eq!(sse.data["permissionId"], "perm_1");
        assert_eq!(sse.data["toolName"], "mnote.page.save");
        assert_eq!(sse.data["params"]["documentId"], "doc_1");
        assert_eq!(sse.data["decision"], "denied");
    }

    #[test]
    fn acp_permission_requested_emits_permission_requested_event() {
        let event = AcpSessionEvent::PermissionRequest {
            permission_id: "perm_2".into(),
            tool_name: "mnote.page.get".into(),
            params: json!({"documentId": "doc_2"}),
            decision: "requested".into(),
        };

        let sse = acp_event_to_sse(event).expect("permission requested should be forwarded");
        assert_eq!(sse.event, "permission.requested");
        assert_eq!(sse.data["permissionId"], "perm_2");
        assert_eq!(sse.data["toolName"], "mnote.page.get");
        assert_eq!(sse.data["decision"], "requested");
    }

    #[test]
    fn acp_permission_allowed_emits_permission_allowed_event() {
        let event = AcpSessionEvent::PermissionRequest {
            permission_id: "perm_3".into(),
            tool_name: "mnote.page.save".into(),
            params: json!({"documentId": "doc_1"}),
            decision: "allowed".into(),
        };

        let sse = acp_event_to_sse(event).expect("permission allowed should be forwarded");
        assert_eq!(sse.event, "permission.allowed");
        assert_eq!(sse.data["permissionId"], "perm_3");
        assert_eq!(sse.data["decision"], "allowed");
    }

    #[test]
    fn acp_tool_events_keep_detail_for_collapsible_ui() {
        let started = acp_event_to_sse(AcpSessionEvent::ToolCall {
            tool_call_id: "tool_1".into(),
            title: "mnote.page.get".into(),
            kind: "read".into(),
            status: crate::acp_types::ToolCallStatus::InProgress,
            raw_input: Some(json!({"documentId": "doc_1", "includeBody": true})),
            locations: vec!["/mnt/Data1T/mnote/src/main.rs".into()],
        })
        .expect("tool start");
        assert_eq!(started.event, "tool.started");
        assert_eq!(started.data["tool"], "mnote.page.get");
        assert_eq!(started.data["status"], "in_progress");
        assert_eq!(started.data["input"]["documentId"], "doc_1");
        assert_eq!(
            started.data["locations"][0],
            "/mnt/Data1T/mnote/src/main.rs"
        );
        assert_eq!(started.data["locations"].as_array().unwrap().len(), 1);

        let completed = acp_event_to_sse(AcpSessionEvent::ToolCallUpdate {
            tool_call_id: "tool_1".into(),
            status: crate::acp_types::ToolCallStatus::Completed,
            content: Some(vec![crate::acp_types::ContentBlockWrapper {
                wrapper_type: "content".into(),
                content: crate::acp_types::TextContent {
                    content_type: "text".into(),
                    text: "读取完成".into(),
                },
            }]),
        })
        .expect("tool complete");
        assert_eq!(completed.event, "tool.completed");
        assert_eq!(completed.data["status"], "completed");
        assert_eq!(completed.data["output"][0]["content"]["text"], "读取完成");

        let running = acp_event_to_sse(AcpSessionEvent::ToolCallUpdate {
            tool_call_id: "tool_1".into(),
            status: crate::acp_types::ToolCallStatus::InProgress,
            content: None,
        })
        .expect("tool running");
        assert_eq!(running.event, "tool.started");
        assert_eq!(running.data["status"], "in_progress");
    }

    #[test]
    fn acp_tool_completed_extracts_precise_ui_citations_from_prefixed_text() {
        let prefix = json!({
            "schema": "mnote.acp.tool_result_ui_citations.v1",
            "references": [{
                "citationMarkdown": "[来源定位降级：a.md](/documents/a)",
                "locatorDegraded": true
            }, {
                "citationMarkdown": "[b.md · p.2](/documents/b?page=2)",
                "locatorDegraded": false
            }],
            "uiCitations": [{
                "citationMarkdown": "[来源定位降级：a.md](/documents/a)"
            }]
        })
        .to_string();
        let completed = acp_event_to_sse(AcpSessionEvent::ToolCallUpdate {
            tool_call_id: "tool_2".into(),
            status: crate::acp_types::ToolCallStatus::Completed,
            content: Some(vec![crate::acp_types::ContentBlockWrapper {
                wrapper_type: "content".into(),
                content: crate::acp_types::TextContent {
                    content_type: "text".into(),
                    text: format!("{prefix}\n工具正文"),
                },
            }]),
        })
        .expect("tool complete");

        assert_eq!(
            completed.data["citationMarkdowns"][0]["citationMarkdown"].as_str(),
            Some("[b.md · p.2](/documents/b?page=2)")
        );
        assert_eq!(
            completed.data["citationMarkdowns"]
                .as_array()
                .unwrap()
                .len(),
            1
        );
    }

    #[test]
    fn acp_session_info_update_emits_session_info_updated_sse() {
        let sse = acp_event_to_sse(AcpSessionEvent::SessionInfoUpdate {
            title: "我的新会话标题".into(),
        })
        .expect("session info update should be forwarded");
        assert_eq!(sse.event, "session.info.updated");
        assert_eq!(sse.data["title"], "我的新会话标题");
    }

    #[test]
    fn acp_plan_update_emits_plan_updated_sse() {
        let entries = vec![
            "步骤 1：读取文件".into(),
            "步骤 2：修改配置".into(),
            "步骤 3：验证更改".into(),
        ];
        let sse = acp_event_to_sse(AcpSessionEvent::PlanUpdate {
            entries: entries.clone(),
        })
        .expect("plan update should be forwarded");
        assert_eq!(sse.event, "plan.updated");
        let sse_entries: Vec<String> =
            serde_json::from_value(sse.data["entries"].clone()).unwrap_or_default();
        assert_eq!(sse_entries.len(), 3);
        assert_eq!(sse_entries[0], "步骤 1：读取文件");
        assert_eq!(sse_entries[1], "步骤 2：修改配置");
        assert_eq!(sse_entries[2], "步骤 3：验证更改");
    }
}