feat: add eventbus-backed RPA tab commands

Author: lius-new

src/infrastructure/eventbus/manager.rs

@@ -1,14 +1,14 @@
 use super::connection::{BrowserConnection, MessageHandler};
 use super::error::{EventBusError, Result};
 use super::global::EnvConnectionPayload;
-use super::message::Message;
+use super::message::{Message, MessageType};
 use super::topics::Topic;
 use super::transport::PipeServer;
 use crate::infrastructure::diagnostics::{log_error, log_info, log_warn};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::sync::Arc;
-use tokio::sync::{RwLock, mpsc};
+use tokio::sync::{RwLock, mpsc, oneshot};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct FingerprintConfig {
@@ -110,6 +110,7 @@ type AuthInfoProvider = Arc<dyn Fn() -> AuthInfo + Send + Sync>;
 
 pub struct EventBusManager {
     connections: Arc<RwLock<HashMap<String, Arc<BrowserConnection>>>>,
+    pending_requests: Arc<RwLock<HashMap<String, HashMap<u32, oneshot::Sender<Message>>>>>,
     message_handler: Arc<RwLock<Option<MessageHandler>>>,
     disconnect_handler: Arc<RwLock<Option<DisconnectHandler>>>,
     connection_status_handler: Arc<RwLock<Option<ConnectionStatusHandler>>>,
@@ -122,6 +123,7 @@ impl EventBusManager {
     pub fn new() -> Self {
         Self {
             connections: Arc::new(RwLock::new(HashMap::new())),
+            pending_requests: Arc::new(RwLock::new(HashMap::new())),
             message_handler: Arc::new(RwLock::new(None)),
             disconnect_handler: Arc::new(RwLock::new(None)),
             connection_status_handler: Arc::new(RwLock::new(None)),
@@ -297,6 +299,7 @@ impl EventBusManager {
         let handler = self.message_handler.read().await.clone();
         let disconnect_handler = self.disconnect_handler.read().await.clone();
         let connections = self.connections.clone();
+        let pending_requests = self.pending_requests.clone();
         let status_handler = self.connection_status_handler.clone();
 
         if let Some(handler) = handler {
@@ -305,6 +308,25 @@ impl EventBusManager {
                 loop {
                     match browser_connection.recv().await {
                         Ok(msg) => {
+                            if msg.msg_type == MessageType::Response {
+                                let mut pending = pending_requests.write().await;
+                                if let Some(waiters) = pending.get_mut(&env_id_for_loop) {
+                                    if let Some(sender) = waiters.remove(&msg.msg_id) {
+                                        let _ = sender.send(msg);
+                                        continue;
+                                    }
+                                }
+
+                                log_warn(
+                                    "eventbus",
+                                    format!(
+                                        "[{}] Received unexpected response message: {}",
+                                        env_id_for_loop, msg.msg_id
+                                    ),
+                                );
+                                continue;
+                            }
+
                             if msg.topic == Topic::AuthRequest {
                                 if let Err(error) =
                                     Self::handle_auth_request(&browser_connection, msg).await
@@ -333,6 +355,11 @@ impl EventBusManager {
                                 guard.remove(&env_id_for_loop);
                             }
 
+                            {
+                                let mut pending = pending_requests.write().await;
+                                pending.remove(&env_id_for_loop);
+                            }
+
                             {
                                 let status = status_handler.read().await.clone();
                                 if let Some(status) = status {
@@ -393,7 +420,32 @@ impl EventBusManager {
             .clone();
         drop(connections);
 
-        connection.send_request(topic, data).await
+        let message = Message::request(topic, data);
+        let msg_id = message.msg_id;
+        let (tx, rx) = oneshot::channel();
+
+        {
+            let mut pending = self.pending_requests.write().await;
+            pending
+                .entry(env_id.to_string())
+                .or_default()
+                .insert(msg_id, tx);
+        }
+
+        if let Err(error) = connection.send(&message).await {
+            let mut pending = self.pending_requests.write().await;
+            if let Some(waiters) = pending.get_mut(env_id) {
+                waiters.remove(&msg_id);
+            }
+            return Err(error);
+        }
+
+        rx.await.map_err(|error| {
+            EventBusError::ReceiveFailed(format!(
+                "failed to receive response for env {} msg {}: {}",
+                env_id, msg_id, error
+            ))
+        })
     }
 
     pub async fn broadcast(&self, topic: Topic, data: Vec<u8>) -> Vec<(String, Result<()>)> {
@@ -450,9 +502,11 @@ impl EventBusManager {
 
     pub async fn disconnect_all(&self) {
         let mut connections = self.connections.write().await;
+        let mut pending_requests = self.pending_requests.write().await;
         for (env_id, connection) in connections.drain() {
             let message = Message::event(Topic::Disconnect, vec![]);
             let _ = connection.send(&message).await;
+            pending_requests.remove(&env_id);
             log_info("eventbus", format!("[{}] Disconnected", env_id));
         }
     }

src/services/environment/kernel/launcher.rs

@@ -1,7 +1,8 @@
 use super::cdp::CdpEndpointManager;
 use super::job::JobManager;
 use super::types::{
-    BatchLaunchResult, CdpEndpointResponse, EnvironmentStartRequest, WindowBoundsRequest,
+    BatchLaunchResult, CdpEndpointResponse, EnvironmentStartRequest, RpaTabCloseResult,
+    RpaTabSelection, RpaTabsSnapshot, WindowBoundsRequest,
 };
 use crate::app::{EventPublisher, Result, RuntimeError};
 use crate::infrastructure::diagnostics::log_info;
@@ -11,6 +12,13 @@ use crate::services::environment::{EnvironmentStatus, EnvironmentStatusManager};
 use std::path::Path;
 use std::sync::Arc;
 
+#[derive(serde::Serialize)]
+struct RpaCommandPayload<'a> {
+    action: &'a str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    position: Option<u32>,
+}
+
 pub async fn launch_browser(
     request: EnvironmentStartRequest,
     cdp_endpoint_manager: Arc<CdpEndpointManager>,
@@ -299,6 +307,59 @@ pub async fn get_cdp_endpoint(
         }))
 }
 
+pub async fn list_rpa_tabs(env_uuid: String) -> Result<RpaTabsSnapshot> {
+    let env_id = env_uuid.trim().to_string();
+    let manager = eventbus_manager();
+
+    if !manager.is_connected(&env_id).await {
+        return Err(RuntimeError::Internal(format!("环境 {} 未连接", env_id)));
+    }
+
+    let response = manager
+        .send_request(&env_id, Topic::RpaCommand, encode_rpa_command("list_tabs", None)?)
+        .await?;
+
+    decode_rpa_response::<RpaTabsSnapshot>(response)
+}
+
+pub async fn select_rpa_tab(env_uuid: String, position: u32) -> Result<RpaTabSelection> {
+    let env_id = env_uuid.trim().to_string();
+    let manager = eventbus_manager();
+
+    if !manager.is_connected(&env_id).await {
+        return Err(RuntimeError::Internal(format!("环境 {} 未连接", env_id)));
+    }
+
+    let response = manager
+        .send_request(
+            &env_id,
+            Topic::RpaCommand,
+            encode_rpa_command("select_tab", Some(position))?,
+        )
+        .await?;
+
+    decode_rpa_response::<RpaTabSelection>(response)
+}
+
+pub async fn close_rpa_tab(env_uuid: String, position: u32) -> Result<RpaTabCloseResult> {
+    let env_id = env_uuid.trim().to_string();
+    let manager = eventbus_manager();
+
+    if !manager.is_connected(&env_id).await {
+        return Err(RuntimeError::Internal(format!("环境 {} 未连接", env_id)));
+    }
+
+    let response = manager
+        .send_request(
+            &env_id,
+            Topic::RpaCommand,
+            encode_rpa_command("close_tab", Some(position))?,
+        )
+        .await?;
+
+    decode_rpa_response::<RpaTabCloseResult>(response)
+}
+
 pub async fn batch_launch_environments(
     requests: Vec<EnvironmentStartRequest>,
     cdp_endpoint_manager: Arc<CdpEndpointManager>,
@@ -361,6 +422,36 @@ fn encode_window_bounds_payload(x: i32, y: i32, width: i32, height: i32) -> Resu
     Ok(payload)
 }
 
+fn encode_rpa_command(action: &str, position: Option<u32>) -> Result<Vec<u8>> {
+    serde_json::to_vec(&RpaCommandPayload { action, position })
+        .map_err(|error| RuntimeError::Serialization(error.to_string()))
+}
+
+fn decode_rpa_response<T>(response: Message) -> Result<T>
+where
+    T: serde::de::DeserializeOwned,
+{
+    if response.error_code != 0 {
+        return Err(RuntimeError::Internal(read_rpa_error_message(&response.data)));
+    }
+
+    serde_json::from_slice(&response.data)
+        .map_err(|error| RuntimeError::Serialization(error.to_string()))
+}
+
+fn read_rpa_error_message(data: &[u8]) -> String {
+    #[derive(serde::Deserialize)]
+    struct ErrorPayload {
+        message: Option<String>,
+    }
+
+    serde_json::from_slice::<ErrorPayload>(data)
+        .ok()
+        .and_then(|payload| payload.message)
+        .filter(|message| !message.trim().is_empty())
+        .unwrap_or_else(|| "RPA_COMMAND_FAILED".to_string())
+}
+
 #[cfg(test)]
 mod tests {
     use super::encode_window_bounds_payload;

src/services/environment/kernel/mod.rs

@@ -9,7 +9,8 @@ use cdp::CdpEndpointManager;
 use job::JobManager;
 use launcher::{
     batch_launch_environments, batch_stop_environments, get_cdp_endpoint,
-    get_connected_environments, launch_browser, refresh_proxy, set_window_bounds, stop_environment,
+    get_connected_environments, launch_browser, list_rpa_tabs, refresh_proxy, close_rpa_tab,
+    select_rpa_tab, set_window_bounds, stop_environment,
 };
 use std::sync::Arc;
 use types::{EnvironmentCommandRequest, EnvironmentCommandResponse};
@@ -96,6 +97,18 @@ impl KernelRuntime {
                     get_cdp_endpoint(env_uuid, self.cdp_endpoint_manager.clone()).await?;
                 Ok(EnvironmentCommandResponse::CdpEndpoint { endpoint })
             }
+            EnvironmentCommandRequest::ListRpaTabs { env_uuid } => {
+                let snapshot = list_rpa_tabs(env_uuid).await?;
+                Ok(EnvironmentCommandResponse::RpaTabsSnapshot { snapshot })
+            }
+            EnvironmentCommandRequest::SelectRpaTab { env_uuid, position } => {
+                let selection = select_rpa_tab(env_uuid, position).await?;
+                Ok(EnvironmentCommandResponse::RpaTabSelected { selection })
+            }
+            EnvironmentCommandRequest::CloseRpaTab { env_uuid, position } => {
+                let result = close_rpa_tab(env_uuid, position).await?;
+                Ok(EnvironmentCommandResponse::RpaTabClosed { result })
+            }
             EnvironmentCommandRequest::GetEnvironmentStatus { env_uuid } => {
                 let status = self.status_manager.get_status(&env_uuid).await;
                 Ok(EnvironmentCommandResponse::Status { status })

src/services/environment/kernel/types.rs

@@ -62,6 +62,35 @@ pub struct WindowBoundsRequest {
     pub height: i32,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RpaTabInfo {
+    pub position: u32,
+    pub title: String,
+    pub url: String,
+    pub active: bool,
+    pub target_id: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RpaTabsSnapshot {
+    pub tabs: Vec<RpaTabInfo>,
+    pub active_position: Option<u32>,
+    pub total: u32,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RpaTabSelection {
+    pub position: u32,
+    pub target_id: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RpaTabCloseResult {
+    pub closed_position: u32,
+    pub active_position: u32,
+    pub target_id: String,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(tag = "command", rename_all = "snake_case")]
 pub enum EnvironmentCommandRequest {
@@ -88,6 +117,17 @@ pub enum EnvironmentCommandRequest {
     GetCdpEndpoint {
         env_uuid: String,
     },
+    ListRpaTabs {
+        env_uuid: String,
+    },
+    SelectRpaTab {
+        env_uuid: String,
+        position: u32,
+    },
+    CloseRpaTab {
+        env_uuid: String,
+        position: u32,
+    },
     GetEnvironmentStatus {
         env_uuid: String,
     },
@@ -107,6 +147,15 @@ pub enum EnvironmentCommandResponse {
     CdpEndpoint {
         endpoint: Option<CdpEndpointResponse>,
     },
+    RpaTabsSnapshot {
+        snapshot: RpaTabsSnapshot,
+    },
+    RpaTabSelected {
+        selection: RpaTabSelection,
+    },
+    RpaTabClosed {
+        result: RpaTabCloseResult,
+    },
     BatchLaunchResults {
         results: Vec<BatchLaunchResult>,
     },