//! Event Bridge for ACP Server
//!
//! This module provides the `EventBridge` which subscribes to a source event stream
//! (e.g., from `dirigent_core`) and forwards translated events to connected ACP clients
//! via the SSE notifier.
//!
//! ## Architecture
//!
//! The `EventBridge` runs as a background task that:
//! 1. Subscribes to a source event stream
//! 2. For each event, looks up the session mapping to find the client
//! 3. Translates the event to an ACP notification
//! 4. Broadcasts the notification to the appropriate client via SSE
//!
//! ## Example
//!
//! ```rust,ignore
//! use dirigent_acp_api::event_bridge::EventBridge;
//! use dirigent_acp_api::session_manager::SessionManager;
//! use dirigent_acp_api::sse::SseNotifier;
//! use tokio_stream::wrappers::BroadcastStream;
//!
//! // Create components
//! let session_manager = SessionManager::new();
//! let sse_notifier = SseNotifier::new();
//!
//! // Create event bridge
//! let agent_request_tracker = Arc::new(AgentRequestTracker::new());
//! let bridge = EventBridge::new(session_manager, sse_notifier, agent_request_tracker);
//!
//! // Start with an event stream (e.g., from CoreHandle::subscribe())
//! bridge.start(event_stream).await;
//!
//! // Later, stop the bridge
//! bridge.stop();
//! ```

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tracing::{debug, info, trace, warn};

use dirigent_protocol::Event;

use crate::rpc::ConnectorOperations;
use crate::session_manager::SessionManager;
use crate::sse::{translate_event, SessionUpdateParams, SseNotifier};

// ============================================================================
// EventBridge (T032-T033)
// ============================================================================

/// Configuration for the event bridge
#[derive(Debug, Clone)]
pub struct EventBridgeConfig {
    /// Whether to broadcast system-wide errors to all clients
    pub broadcast_system_errors: bool,

    /// Whether to log events that don't have a session mapping
    pub log_unmapped_events: bool,
}

impl Default for EventBridgeConfig {
    fn default() -> Self {
        Self {
            broadcast_system_errors: true,
            log_unmapped_events: true,
        }
    }
}

/// Internal state for the event bridge
struct EventBridgeInner {
    /// Session manager for looking up session mappings
    session_manager: SessionManager,

    /// SSE notifier for broadcasting to clients
    sse_notifier: SseNotifier,

    /// Agent request tracker for bidirectional request/response
    agent_request_tracker: Arc<crate::agent_requests::AgentRequestTracker>,

    /// Configuration
    config: EventBridgeConfig,

    /// Flag to signal shutdown
    shutdown: AtomicBool,

    /// Optional connector operations handle for querying connectors
    /// Used for finding Gateway connectors during fallback scenarios and sending connector commands
    core_handle: Option<Arc<dyn ConnectorOperations>>,
}

/// Event Bridge for forwarding events from a source stream to ACP clients
///
/// The `EventBridge` subscribes to an event stream (typically from `CoreHandle`)
/// and forwards relevant events to connected ACP clients via SSE.
///
/// ## Thread Safety
///
/// The `EventBridge` is designed to be cloned and shared. The internal state
/// is wrapped in `Arc` for thread-safe access.
///
/// ## Lifecycle
///
/// 1. Create with `new()` or `with_config()`
/// 2. Start the background task with `start()`
/// 3. The bridge runs until `stop()` is called or the source stream ends
/// 4. Use `is_running()` to check status
#[derive(Clone)]
pub struct EventBridge {
    inner: Arc<EventBridgeInner>,

    /// Sender for the shutdown command
    shutdown_tx: Option<mpsc::Sender<()>>,
}

impl EventBridge {
    /// Create a new event bridge with default configuration
    ///
    /// # Parameters
    ///
    /// - `session_manager`: The session manager for looking up mappings
    /// - `sse_notifier`: The SSE notifier for broadcasting
    /// - `agent_request_tracker`: The agent request tracker for bidirectional requests
    pub fn new(
        session_manager: SessionManager,
        sse_notifier: SseNotifier,
        agent_request_tracker: Arc<crate::agent_requests::AgentRequestTracker>,
    ) -> Self {
        Self::with_config(
            session_manager,
            sse_notifier,
            agent_request_tracker,
            EventBridgeConfig::default(),
        )
    }

    /// Create a new event bridge with custom configuration
    ///
    /// # Parameters
    ///
    /// - `session_manager`: The session manager for looking up mappings
    /// - `sse_notifier`: The SSE notifier for broadcasting
    /// - `agent_request_tracker`: The agent request tracker for bidirectional requests
    /// - `config`: The bridge configuration
    pub fn with_config(
        session_manager: SessionManager,
        sse_notifier: SseNotifier,
        agent_request_tracker: Arc<crate::agent_requests::AgentRequestTracker>,
        config: EventBridgeConfig,
    ) -> Self {
        Self {
            inner: Arc::new(EventBridgeInner {
                session_manager,
                sse_notifier,
                agent_request_tracker,
                config,
                shutdown: AtomicBool::new(false),
                core_handle: None,
            }),
            shutdown_tx: None,
        }
    }

    /// Create a new event bridge with core handle for connector operations
    ///
    /// # Parameters
    ///
    /// - `session_manager`: The session manager for looking up mappings
    /// - `sse_notifier`: The SSE notifier for broadcasting
    /// - `agent_request_tracker`: The agent request tracker for bidirectional requests
    /// - `config`: The bridge configuration
    /// - `core_handle`: Optional connector operations handle for querying connectors and sending commands
    pub fn with_core_handle(
        session_manager: SessionManager,
        sse_notifier: SseNotifier,
        agent_request_tracker: Arc<crate::agent_requests::AgentRequestTracker>,
        config: EventBridgeConfig,
        core_handle: Option<Arc<dyn ConnectorOperations>>,
    ) -> Self {
        Self {
            inner: Arc::new(EventBridgeInner {
                session_manager,
                sse_notifier,
                agent_request_tracker,
                config,
                shutdown: AtomicBool::new(false),
                core_handle,
            }),
            shutdown_tx: None,
        }
    }

    /// Start the event bridge background task
    ///
    /// This method spawns a background task that reads from the provided event
    /// stream and forwards translated events to the appropriate ACP clients.
    ///
    /// # Type Parameters
    ///
    /// - `S`: The event stream type (must implement `Stream<Item = Event>`)
    ///
    /// # Parameters
    ///
    /// - `event_stream`: The source event stream to subscribe to
    ///
    /// # Returns
    ///
    /// A `JoinHandle` for the background task. The task will run until:
    /// - `stop()` is called
    /// - The source stream ends
    /// - An unrecoverable error occurs
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let agent_request_tracker = Arc::new(AgentRequestTracker::new());
    /// let bridge = EventBridge::new(session_manager, sse_notifier, agent_request_tracker);
    /// let handle = bridge.start(event_stream);
    ///
    /// // Later...
    /// bridge.stop();
    /// handle.await; // Wait for clean shutdown
    /// ```
    pub fn start<S>(&mut self, event_stream: S) -> JoinHandle<()>
    where
        S: tokio_stream::Stream<Item = Event> + Send + Unpin + 'static,
    {
        use tokio_stream::StreamExt;

        // Reset shutdown flag
        self.inner.shutdown.store(false, Ordering::SeqCst);

        // Create shutdown channel
        let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<()>(1);
        self.shutdown_tx = Some(shutdown_tx);

        let inner = self.inner.clone();

        tokio::spawn(async move {
            info!("Event bridge started");

            let mut stream = Box::pin(event_stream);
            let mut event_count = 0u64;

            loop {
                tokio::select! {
                    // Check for shutdown signal
                    _ = shutdown_rx.recv() => {
                        info!("Event bridge received shutdown signal");
                        break;
                    }

                    // Check for shutdown flag
                    _ = tokio::time::sleep(tokio::time::Duration::from_millis(100)), if inner.shutdown.load(Ordering::SeqCst) => {
                        info!("Event bridge shutdown flag set");
                        break;
                    }

                    // Process next event from stream
                    event = stream.next() => {
                        match event {
                            Some(evt) => {
                                event_count += 1;
                                Self::process_event(&inner, evt).await;
                            }
                            None => {
                                info!("Event bridge source stream ended");
                                break;
                            }
                        }
                    }
                }
            }

            info!(
                "Event bridge stopped after processing {} events",
                event_count
            );
        })
    }

    /// Process a single event
    async fn process_event(inner: &EventBridgeInner, event: Event) {
        debug!("EventBridge processing event: {:?}", event);

        // Handle special events that don't need translation
        match &event {
            Event::SessionTransferred {
                from_connector,
                from_session,
                to_connector,
                to_session,
                is_new_session,
                models,
                modes,
            } => {
                Self::handle_session_transferred_internal(
                    inner,
                    from_connector,
                    from_session,
                    to_connector,
                    to_session,
                    *is_new_session,
                    models.clone(),
                    modes.clone(),
                )
                .await;
                return;
            }
            Event::ForwardingPanic {
                connector_id,
                session_id,
                reason,
                fallback_gateway_session,
            } => {
                Self::handle_forwarding_panic_internal(
                    inner,
                    connector_id,
                    session_id,
                    reason,
                    fallback_gateway_session.as_deref(),
                )
                .await;
                return;
            }
            Event::AgentRequest {
                connector_id,
                session_id,
                request_id,
                method,
                params,
                is_forwarded: _, // Ignored: handled by routing logic before this point
            } => {
                Self::handle_agent_request_internal(
                    inner,
                    connector_id,
                    session_id,
                    request_id.clone(),
                    method,
                    params.clone(),
                )
                .await;
                return;
            }
            _ => {}
        }

        // Try to translate the event to session update params
        let update_params_list = translate_event(&event);

        if update_params_list.is_empty() {
            trace!("Event filtered (not forwarded to clients): {:?}", event);
            return;
        }

        // Process each update (most events emit 1, SessionMetadataReceived may emit 2)
        for mut update_params in update_params_list {
            debug!(
                "Translated to SessionUpdateParams: session_id={}, variant={}",
                update_params.session_id,
                update_params.update.variant_name()
            );

            // Determine which client(s) to send to based on the event
            // Also translate internal session ID to client session ID
            match Self::get_client_for_event_and_translate(inner, &event, &mut update_params) {
                EventTarget::Client(client_id) => {
                    // Send to specific client (session_id already translated)
                    match inner.sse_notifier.broadcast(&client_id, update_params) {
                        Ok(n) => {
                            debug!("Forwarded event to client {}: {} receivers", client_id, n);
                        }
                        Err(()) => {
                            if inner.config.log_unmapped_events {
                                warn!("Client {} is not subscribed to SSE", client_id);
                            }
                        }
                    }
                }
                EventTarget::AllClients => {
                    // Broadcast to all clients (e.g., system errors)
                    let count = inner.sse_notifier.broadcast_all(update_params);
                    debug!("Broadcast event to all clients: {} receivers", count);
                }
                EventTarget::NoTarget => {
                    // No target found
                    if inner.config.log_unmapped_events {
                        debug!("No target found for event, dropping");
                    }
                }
            }
        }
    }

    /// Determine which client(s) should receive the event AND translate session ID
    ///
    /// This function:
    /// 1. Looks up the session mapping (internal -> client session ID)
    /// 2. Translates the session_id in update_params from internal to client ID
    /// 3. Returns which client should receive the event
    fn get_client_for_event_and_translate(
        inner: &EventBridgeInner,
        event: &Event,
        update_params: &mut SessionUpdateParams,
    ) -> EventTarget {
        // Extract session_id from the update params (this is the internal session ID)
        let session_id = &update_params.session_id;

        // Try to find the session mapping by client_session_id first
        if let Some(mapping) = inner.session_manager.get_mapping(session_id) {
            // Already using client session ID
            return EventTarget::Client(mapping.client_id);
        }

        // Try to find by internal_session_id (events from connectors use internal IDs)
        // We need to extract the internal session ID from the original event
        if let Some(internal_id) = Self::extract_internal_session_id(event) {
            debug!(
                "Looking up mapping for internal session ID: {}",
                internal_id
            );
            if let Some(mapping) = inner
                .session_manager
                .get_mapping_by_internal_id(&internal_id)
            {
                // TRANSLATE: Replace internal session ID with client session ID
                update_params.session_id = mapping.client_session_id.clone();
                debug!(
                    "Translated session ID: internal {} -> client {}, client_id: {}",
                    internal_id, mapping.client_session_id, mapping.client_id
                );
                return EventTarget::Client(mapping.client_id);
            } else {
                warn!(
                    "No session mapping found for internal session ID: {}",
                    internal_id
                );
            }
        } else {
            warn!(
                "Could not extract internal session ID from event: {:?}",
                event
            );
        }

        debug!(
            "No target found for event with session_id: {}",
            update_params.session_id
        );
        EventTarget::NoTarget
    }

    /// Determine which client(s) should receive the event (deprecated - use get_client_for_event_and_translate)
    #[allow(dead_code)]
    fn get_client_for_event(
        inner: &EventBridgeInner,
        event: &Event,
        update_params: &SessionUpdateParams,
    ) -> EventTarget {
        // Extract session_id from the update params
        let session_id = &update_params.session_id;

        // Try to find the session mapping by client_session_id first
        if let Some(mapping) = inner.session_manager.get_mapping(session_id) {
            return EventTarget::Client(mapping.client_id);
        }

        // Try to find by internal_session_id (events from connectors use internal IDs)
        // We need to extract the internal session ID from the original event
        if let Some(internal_id) = Self::extract_internal_session_id(event) {
            if let Some(mapping) = inner
                .session_manager
                .get_mapping_by_internal_id(&internal_id)
            {
                return EventTarget::Client(mapping.client_id);
            }
        }

        EventTarget::NoTarget
    }

    /// Extract the internal session ID from an event
    fn extract_internal_session_id(event: &Event) -> Option<String> {
        match event {
            Event::SessionUpdate { session_id, .. } => Some(session_id.clone()),
            Event::MessageCompleted { message, .. } => Some(message.session_id.clone()),
            Event::SessionIdle { session_id, .. } => Some(session_id.clone()),
            Event::SessionMetadataReceived { session_id, .. } => Some(session_id.clone()),
            Event::MessageFailed { .. } => {
                // MessageFailed uses message_id, not session_id
                // We can't directly map this, so return None
                None
            }
            Event::Error { .. } => None,
            _ => None,
        }
    }

    /// Internal handler for session transfer (static method for use in process_event)
    ///
    /// The `models` and `modes` parameters are now passed directly from the `SessionTransferred`
    /// event, which carries the new connector's session initialization data. This avoids needing
    /// to query the connector for metadata after the transfer.
    async fn handle_session_transferred_internal(
        inner: &EventBridgeInner,
        from_connector: &str,
        from_session: &str,
        to_connector: &str,
        to_session: &str,
        is_new_session: bool,
        models: Option<dirigent_protocol::SessionModelState>,
        modes: Option<dirigent_protocol::SessionModeState>,
    ) {
        tracing::info!(
            from_connector = %from_connector,
            from_session = %from_session,
            to_connector = %to_connector,
            to_session = %to_session,
            is_new = %is_new_session,
            has_models = %models.is_some(),
            has_modes = %modes.is_some(),
            "Processing session transfer"
        );

        // Find the client mapping for the source session
        let mapping = inner
            .session_manager
            .get_mapping_by_gateway_session(from_session);

        if let Some(mapping) = mapping {
            // Update the mapping to point to new connector/session
            // Note: connector_title is None here, will use connector_id as fallback
            if let Some(old_mapping) = inner.session_manager.update_mapping_connector(
                &mapping.client_session_id,
                to_connector.to_string(),
                None, // connector_title - could be looked up from ConnectorOperations
                to_session.to_string(),
            ) {
                tracing::info!(
                    client_session_id = %mapping.client_session_id,
                    old_connector = %old_mapping.connector_id,
                    new_connector = %to_connector,
                    "Session mapping updated for transfer"
                );

                // Notify client via SSE that the transfer completed
                use crate::sse::{SessionUpdateParams, SessionUpdateVariant};
                let update = SessionUpdateParams {
                    session_id: mapping.client_session_id.clone(),
                    update: SessionUpdateVariant::ConnectorChanged {
                        new_connector_id: to_connector.to_string(),
                        new_internal_session_id: to_session.to_string(),
                        is_new_session,
                    },
                    event_type_override: None,
                };

                match inner.sse_notifier.broadcast(&mapping.client_id, update) {
                    Ok(receivers) => {
                        tracing::info!(
                            client_id = %mapping.client_id,
                            session_id = %mapping.client_session_id,
                            receivers = receivers,
                            "Notified client about session transfer via SSE"
                        );
                    }
                    Err(_) => {
                        tracing::warn!(
                            client_id = %mapping.client_id,
                            "Failed to notify client about session transfer via SSE - client not subscribed"
                        );
                    }
                }

                // Send config_option_update with the new connector's modes/models
                // These are passed directly from the SessionTransferred event, which
                // carries the initialization data from the new connector's session/new response.
                // See: https://agentclientprotocol.com/rfds/session-config-options
                {
                    let mut config_options = vec![];

                    if let Some(ref modes_state) = modes {
                        config_options.push(crate::sse::modes_to_config_option(modes_state));
                    }

                    if let Some(ref models_state) = models {
                        config_options.push(crate::sse::models_to_config_option(models_state));
                    }

                    // Send config_option_update if we have any options
                    if !config_options.is_empty() {
                        let config_update = SessionUpdateParams {
                            session_id: mapping.client_session_id.clone(),
                            update: SessionUpdateVariant::ConfigOptionUpdate { config_options },
                            event_type_override: None,
                        };

                        if let Ok(receivers) = inner
                            .sse_notifier
                            .broadcast(&mapping.client_id, config_update)
                        {
                            tracing::info!(
                                client_id = %mapping.client_id,
                                session_id = %mapping.client_session_id,
                                receivers = receivers,
                                "Sent config_option_update after transfer"
                            );
                        }
                    } else {
                        tracing::debug!(
                            client_id = %mapping.client_id,
                            "No modes/models to send in config_option_update after transfer"
                        );
                    }
                }

                // Get available commands from new connector and send update
                if let Some(ref core_handle) = inner.core_handle {
                    match core_handle.get_connector_commands(to_connector).await {
                        Ok(commands) => {
                            let commands_update = SessionUpdateParams {
                                session_id: mapping.client_session_id.clone(),
                                update: SessionUpdateVariant::AvailableCommandsUpdate {
                                    available_commands: commands.clone(),
                                },
                                event_type_override: None,
                            };

                            match inner
                                .sse_notifier
                                .broadcast(&mapping.client_id, commands_update)
                            {
                                Ok(receivers) => {
                                    tracing::info!(
                                        client_id = %mapping.client_id,
                                        session_id = %mapping.client_session_id,
                                        commands = commands.len(),
                                        receivers = receivers,
                                        "Sent AvailableCommandsUpdate after transfer"
                                    );
                                }
                                Err(_) => {
                                    tracing::warn!(
                                        client_id = %mapping.client_id,
                                        "Failed to send AvailableCommandsUpdate after transfer"
                                    );
                                }
                            }
                        }
                        Err(e) => {
                            tracing::warn!(
                                connector_id = %to_connector,
                                error = %e,
                                "Failed to get connector commands after transfer"
                            );
                        }
                    }
                } else {
                    tracing::debug!("No core handle available - skipping metadata/commands query after transfer");
                }

                // Send confirmation message so the user sees the transfer completed
                let mode_text = if is_new_session {
                    "new session"
                } else {
                    "existing session"
                };
                let confirmation_text = format!("Connected to {} ({}).", to_connector, mode_text);
                let confirmation_update = SessionUpdateParams {
                    session_id: mapping.client_session_id.clone(),
                    update: SessionUpdateVariant::AgentMessageChunk {
                        content: dirigent_protocol::types::ContentBlock::Text {
                            text: confirmation_text,
                        },
                    },
                    event_type_override: None,
                };

                match inner
                    .sse_notifier
                    .broadcast(&mapping.client_id, confirmation_update)
                {
                    Ok(receivers) => {
                        tracing::info!(
                            client_id = %mapping.client_id,
                            session_id = %mapping.client_session_id,
                            receivers = receivers,
                            "Sent transfer confirmation message"
                        );
                    }
                    Err(_) => {
                        tracing::warn!(
                            client_id = %mapping.client_id,
                            "Failed to send transfer confirmation message"
                        );
                    }
                }

                // After transfer, send SessionIdle to give control back to client
                let idle_update = SessionUpdateParams {
                    session_id: mapping.client_session_id.clone(),
                    update: SessionUpdateVariant::SessionIdle {},
                    event_type_override: None,
                };

                match inner
                    .sse_notifier
                    .broadcast(&mapping.client_id, idle_update)
                {
                    Ok(receivers) => {
                        tracing::info!(
                            client_id = %mapping.client_id,
                            session_id = %mapping.client_session_id,
                            receivers = receivers,
                            "Sent SessionIdle after transfer to give control back"
                        );
                    }
                    Err(_) => {
                        tracing::warn!(
                            client_id = %mapping.client_id,
                            "Failed to send SessionIdle after transfer"
                        );
                    }
                }

            }
        } else {
            tracing::warn!(
                from_session = %from_session,
                "No client mapping found for transferred session - session may have been created outside ACP Server"
            );
        }
    }

    /// Internal handler for forwarding panic (static method for use in process_event)
    async fn handle_forwarding_panic_internal(
        inner: &EventBridgeInner,
        connector_id: &str,
        session_id: &str,
        reason: &str,
        fallback_gateway_session: Option<&str>,
    ) {
        tracing::warn!(
            connector_id = %connector_id,
            session_id = %session_id,
            reason = %reason,
            "Handling forwarding panic"
        );

        // Find mapping by the failed connector's session
        let mapping = inner.session_manager.get_mapping_by_internal_id(session_id);

        if let Some(mapping) = mapping {
            // Determine fallback session
            let fallback_session = fallback_gateway_session
                .map(String::from)
                .unwrap_or_else(|| format!("fallback-{}", uuid::Uuid::new_v4()));

            // Find a Gateway connector to fall back to
            let gateway_connector = Self::find_gateway_connector_internal(inner).await;

            if let Some(gateway_id) = gateway_connector {
                inner.session_manager.update_mapping_connector(
                    &mapping.client_session_id,
                    gateway_id.clone(),
                    Some("Gateway".to_string()), // Fallback to Gateway
                    fallback_session.clone(),
                );

                // Log the error - client notification would happen through the Gateway
                // when they send their next message and see they're back on Gateway
                tracing::warn!(
                    client_id = %mapping.client_id,
                    connector_id = %connector_id,
                    reason = %reason,
                    "Connector failed, client mapping reverted to Gateway"
                );

                tracing::info!(
                    client_id = %mapping.client_id,
                    fallback_connector = %gateway_id,
                    "Client reverted to Gateway after forwarding panic"
                );
            } else {
                tracing::error!("No Gateway connector available for fallback!");
            }
        }
    }

    /// Internal method to find an available Gateway connector
    async fn find_gateway_connector_internal(inner: &EventBridgeInner) -> Option<String> {
        // Query CoreHandle for connectors
        if let Some(ref core_handle) = inner.core_handle {
            match core_handle.list_connectors().await {
                Ok(connectors) => connectors
                    .into_iter()
                    .find(|c| c.connector_type.to_lowercase() == "gateway" && c.available)
                    .map(|c| c.id),
                Err(e) => {
                    tracing::error!("Failed to list connectors: {:?}", e);
                    None
                }
            }
        } else {
            None
        }
    }

    /// Internal handler for agent requests (T024-T030, Phase 2)
    ///
    /// This method implements the complete bidirectional request/response flow:
    /// 1. Translate internal session_id to client session_id
    /// 2. Check ownership to determine routing (UI permission modal vs external client)
    /// 3. Route to appropriate handler
    async fn handle_agent_request_internal(
        inner: &EventBridgeInner,
        connector_id: &str,
        session_id: &str,
        request_id: serde_json::Value,
        method: &str,
        params: serde_json::Value,
    ) {
        tracing::info!(
            connector_id = %connector_id,
            session_id = %session_id,
            method = %method,
            request_id = %request_id,
            "Handling agent request"
        );

        // Get mapping with ownership information
        let mapping = match inner.session_manager.get_mapping_by_internal_id(session_id) {
            Some(m) => m,
            None => {
                tracing::warn!(
                    connector_id = %connector_id,
                    session_id = %session_id,
                    "No client mapping found for agent request - cannot forward to client"
                );
                return;
            }
        };

        // Phase 2: Check if we should forward to external client
        if let Some(forward_client_id) = mapping.ownership.forward_to_client() {
            // Forward to external client with capabilities
            Self::forward_to_external_client(
                inner,
                forward_client_id,
                connector_id,
                &mapping.client_session_id,
                request_id,
                method,
                params,
            )
            .await;
        } else {
            // Show permission modal in UI (current behavior)
            Self::handle_permission_request_for_ui(
                inner,
                &mapping.client_id,
                &mapping.client_session_id,
                connector_id,
                request_id,
                method,
                params,
            )
            .await;
        }
    }

    /// Handle permission request for UI client (original flow)
    ///
    /// This method implements the UI permission modal flow:
    /// 1. Register request with tracker
    /// 2. Broadcast SSE notification to UI
    /// 3. Wait for UI response (30s timeout)
    /// 4. Send response back to connector
    async fn handle_permission_request_for_ui(
        inner: &EventBridgeInner,
        client_id: &str,
        client_session_id: &str,
        connector_id: &str,
        request_id: serde_json::Value,
        method: &str,
        params: serde_json::Value,
    ) {
        use crate::sse::{SessionUpdateParams, SessionUpdateVariant};

        tracing::debug!(
            client_id = %client_id,
            client_session_id = %client_session_id,
            method = %method,
            "Routing agent request to UI permission modal"
        );

        // Register request with tracker and get receiver
        let response_rx = inner
            .agent_request_tracker
            .register(client_id, request_id.clone());

        tracing::debug!(
            client_id = %client_id,
            request_id = %request_id,
            "Registered agent request with tracker"
        );

        // Create SessionUpdateParams with AgentRequest variant
        // Translate sessionId in params from internal to client-facing
        let mut translated_params = params.clone();
        if let Some(params_obj) = translated_params.as_object_mut() {
            params_obj.insert(
                "sessionId".to_string(),
                serde_json::json!(client_session_id),
            );
        }

        let update = SessionUpdateParams {
            session_id: client_session_id.to_string(),
            update: SessionUpdateVariant::AgentRequest {
                request_id: request_id.clone(),
                method: method.to_string(),
                params: translated_params,
            },
            event_type_override: None,
        };

        // Broadcast SSE event to UI client
        match inner.sse_notifier.broadcast(client_id, update) {
            Ok(receivers) => {
                tracing::info!(
                    client_id = %client_id,
                    request_id = %request_id,
                    receivers = receivers,
                    "Broadcasted agent request via SSE to UI"
                );
            }
            Err(()) => {
                tracing::warn!(
                    client_id = %client_id,
                    request_id = %request_id,
                    "Failed to broadcast agent request - UI not subscribed to SSE"
                );
                // Clean up the tracker entry since we can't deliver the request
                inner
                    .agent_request_tracker
                    .timeout(client_id, request_id.clone());
                return;
            }
        }

        // Wait for UI response with 30s timeout
        let response =
            match tokio::time::timeout(tokio::time::Duration::from_secs(30), response_rx).await {
                Ok(Ok(response_value)) => {
                    tracing::info!(
                        client_id = %client_id,
                        request_id = %request_id,
                        "Received UI response for agent request"
                    );
                    response_value
                }
                Ok(Err(_)) => {
                    tracing::warn!(
                        client_id = %client_id,
                        request_id = %request_id,
                        "Response channel dropped for agent request"
                    );
                    return;
                }
                Err(_) => {
                    tracing::warn!(
                        client_id = %client_id,
                        request_id = %request_id,
                        "Agent request timed out after 30s - UI did not respond"
                    );
                    // Clean up tracker
                    inner
                        .agent_request_tracker
                        .timeout(client_id, request_id.clone());
                    return;
                }
            };

        // Send response to connector via send_agent_response
        if let Some(ref core_handle) = inner.core_handle {
            match core_handle
                .send_agent_response(connector_id, request_id.clone(), response)
                .await
            {
                Ok(()) => {
                    tracing::info!(
                        connector_id = %connector_id,
                        request_id = %request_id,
                        "Sent agent response back to connector"
                    );
                }
                Err(e) => {
                    tracing::error!(
                        connector_id = %connector_id,
                        request_id = %request_id,
                        error = %e,
                        "Failed to send agent response to connector"
                    );
                }
            }
        } else {
            tracing::error!(
                connector_id = %connector_id,
                request_id = %request_id,
                "No core handle available - cannot send agent response to connector"
            );
        }
    }

    /// Forward an agent request to an external client via SSE (Phase 2)
    ///
    /// This method implements external client forwarding:
    /// 1. Register request with tracker
    /// 2. Translate sessionId in params
    /// 3. Broadcast SSE notification to external client
    /// 4. Wait for external client response (60s timeout - longer than UI)
    /// 5. Send response back to connector
    async fn forward_to_external_client(
        inner: &EventBridgeInner,
        client_id: &str,
        connector_id: &str,
        client_session_id: &str,
        request_id: serde_json::Value,
        method: &str,
        params: serde_json::Value,
    ) {
        use crate::sse::{SessionUpdateParams, SessionUpdateVariant};

        tracing::info!(
            client_id = %client_id,
            method = %method,
            "Forwarding agent request to external client"
        );

        // Register with tracker (reuse existing mechanism)
        let response_rx = inner
            .agent_request_tracker
            .register(client_id, request_id.clone());

        // Translate sessionId in params from internal to client-facing
        let mut translated_params = params.clone();
        if let Some(params_obj) = translated_params.as_object_mut() {
            params_obj.insert(
                "sessionId".to_string(),
                serde_json::json!(client_session_id),
            );
        }

        // Create SSE notification (same format as permission requests)
        let notification = SessionUpdateParams {
            session_id: client_session_id.to_string(),
            update: SessionUpdateVariant::AgentRequest {
                request_id: request_id.clone(),
                method: method.to_string(),
                params: translated_params,
            },
            event_type_override: None,
        };

        // Broadcast to external client
        if inner
            .sse_notifier
            .broadcast(client_id, notification)
            .is_err()
        {
            tracing::warn!(client_id = %client_id, "External client not subscribed to SSE");
            inner.agent_request_tracker.timeout(client_id, request_id);
            return;
        }

        // Wait for response with longer timeout for external clients (60s)
        let response = match tokio::time::timeout(tokio::time::Duration::from_secs(60), response_rx)
            .await
        {
            Ok(Ok(response)) => {
                tracing::info!(client_id = %client_id, "Received response from external client");
                response
            }
            Ok(Err(_)) => {
                tracing::warn!(client_id = %client_id, "External client response channel dropped");
                return;
            }
            Err(_) => {
                tracing::warn!(client_id = %client_id, "External client request timed out after 60s");
                inner
                    .agent_request_tracker
                    .timeout(client_id, request_id.clone());
                return;
            }
        };

        // Send back to connector
        if let Some(ref core_handle) = inner.core_handle {
            if let Err(e) = core_handle
                .send_agent_response(connector_id, request_id.clone(), response)
                .await
            {
                tracing::error!(connector_id = %connector_id, error = %e, "Failed to send response to connector");
            }
        }
    }

    /// Stop the event bridge
    ///
    /// Signals the background task to shut down gracefully. The task will
    /// complete processing any current event and then exit.
    pub fn stop(&self) {
        info!("Stopping event bridge");
        self.inner.shutdown.store(true, Ordering::SeqCst);

        // Also send via channel for faster shutdown
        if let Some(tx) = &self.shutdown_tx {
            let _ = tx.try_send(());
        }
    }

    /// Check if the event bridge is running
    ///
    /// Returns `true` if the bridge has been started and has not been stopped.
    /// Note that this doesn't guarantee the background task is still alive -
    /// use the `JoinHandle` returned by `start()` for that.
    pub fn is_running(&self) -> bool {
        !self.inner.shutdown.load(Ordering::SeqCst)
    }

    /// Get a reference to the session manager
    pub fn session_manager(&self) -> &SessionManager {
        &self.inner.session_manager
    }

    /// Get a reference to the SSE notifier
    pub fn sse_notifier(&self) -> &SseNotifier {
        &self.inner.sse_notifier
    }
}

/// Target for an event notification
enum EventTarget {
    /// Send to a specific client
    Client(String),

    /// Broadcast to all clients (future use for system-wide notifications)
    #[allow(dead_code)]
    AllClients,

    /// No target found, drop the event
    NoTarget,
}

impl std::fmt::Debug for EventBridge {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EventBridge")
            .field("running", &self.is_running())
            .field("config", &self.inner.config)
            .finish()
    }
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use dirigent_protocol::{ContentBlock, Message, MessageRole, MessageStatus, SessionUpdate};
    use tokio_stream::StreamExt;

    fn create_test_bridge() -> EventBridge {
        let session_manager = SessionManager::new();
        let sse_notifier = SseNotifier::new();
        let agent_request_tracker = Arc::new(crate::agent_requests::AgentRequestTracker::new());
        EventBridge::new(session_manager, sse_notifier, agent_request_tracker)
    }

    #[test]
    fn test_event_bridge_new() {
        let bridge = create_test_bridge();
        assert!(bridge.is_running()); // Not started, but not stopped either
    }

    #[test]
    fn test_event_bridge_with_config() {
        let session_manager = SessionManager::new();
        let sse_notifier = SseNotifier::new();
        let agent_request_tracker = Arc::new(crate::agent_requests::AgentRequestTracker::new());
        let config = EventBridgeConfig {
            broadcast_system_errors: false,
            log_unmapped_events: false,
        };

        let bridge =
            EventBridge::with_config(session_manager, sse_notifier, agent_request_tracker, config);
        assert!(!bridge.inner.config.broadcast_system_errors);
        assert!(!bridge.inner.config.log_unmapped_events);
    }

    #[test]
    fn test_event_bridge_stop() {
        let bridge = create_test_bridge();
        assert!(bridge.is_running());

        bridge.stop();
        assert!(!bridge.is_running());
    }

    #[tokio::test]
    async fn test_event_bridge_start_and_stop() {
        let session_manager = SessionManager::new();
        let sse_notifier = SseNotifier::new();
        let agent_request_tracker = Arc::new(crate::agent_requests::AgentRequestTracker::new());
        let mut bridge = EventBridge::new(session_manager, sse_notifier, agent_request_tracker);

        // Create an empty stream that will end immediately
        let stream = tokio_stream::empty::<Event>();

        let handle = bridge.start(stream);

        // Wait for the task to complete (stream ends immediately)
        tokio::time::timeout(tokio::time::Duration::from_secs(1), handle)
            .await
            .expect("Task should complete quickly")
            .expect("Task should not panic");
    }

    #[tokio::test]
    async fn test_event_bridge_with_events() {
        let session_manager = SessionManager::new();
        let sse_notifier = SseNotifier::new();

        // Register a client and create a session
        let client_id = session_manager.register_client(None);
        let mapping = session_manager.create_mapping(
            &client_id,
            Some("test-session".to_string()),
            "connector-1".to_string(),
        );

        // Subscribe to SSE
        let mut sse_stream = sse_notifier.subscribe(&client_id);

        let agent_request_tracker = Arc::new(crate::agent_requests::AgentRequestTracker::new());
        let mut bridge = EventBridge::new(session_manager, sse_notifier, agent_request_tracker);

        // Create a stream with one event
        let event = Event::SessionIdle {
            connector_id: "test-connector".to_string(),
            session_id: mapping.internal_session_id.clone(),
        };
        let stream = tokio_stream::iter(vec![event]);

        let handle = bridge.start(stream);

        // Wait for the event to be processed
        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

        // Check that the client received the notification
        let notification =
            tokio::time::timeout(tokio::time::Duration::from_millis(500), sse_stream.next()).await;

        // The notification should have been received
        assert!(notification.is_ok());
        let notification = notification.unwrap();
        assert!(notification.is_some());

        let notification = notification.unwrap();
        assert!(notification.is_ok());

        let update_params = notification.unwrap();
        // The session_id in the update params should be the CLIENT session ID
        // because get_client_for_event_and_translate() translates internal IDs to client IDs
        assert_eq!(update_params.session_id, mapping.client_session_id);

        // Verify it's a SessionIdle update
        match update_params.update {
            crate::sse::SessionUpdateVariant::SessionIdle {} => {
                // Expected
            }
            other => panic!("Expected SessionIdle variant, got {:?}", other),
        }

        // Stop and wait for completion
        bridge.stop();
        let _ = tokio::time::timeout(tokio::time::Duration::from_secs(1), handle).await;
    }

    #[tokio::test]
    async fn test_event_bridge_filtered_events() {
        let session_manager = SessionManager::new();
        let sse_notifier = SseNotifier::new();

        // Register a client
        let client_id = session_manager.register_client(None);

        // Subscribe to SSE
        let mut sse_stream = sse_notifier.subscribe(&client_id);

        let agent_request_tracker = Arc::new(crate::agent_requests::AgentRequestTracker::new());
        let mut bridge = EventBridge::new(session_manager, sse_notifier, agent_request_tracker);

        // Create events that should be filtered (not forwarded)
        let events = vec![
            Event::Error {
                message: "System-wide error".to_string(),
            },
            Event::MessageFailed {
                message_id: "msg-1".to_string(),
                error: "Failed".to_string(),
            },
            Event::Connected,
        ];
        let stream = tokio_stream::iter(events);

        let _handle = bridge.start(stream);

        // Wait for processing
        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

        // No notifications should be received (filtered events)
        let result =
            tokio::time::timeout(tokio::time::Duration::from_millis(200), sse_stream.next()).await;

        // Timeout expected since no events should be forwarded
        assert!(
            result.is_err(),
            "Expected timeout since events should be filtered"
        );

        bridge.stop();
    }

    #[test]
    fn test_event_bridge_debug() {
        let bridge = create_test_bridge();
        let debug_str = format!("{:?}", bridge);

        assert!(debug_str.contains("EventBridge"));
        assert!(debug_str.contains("running"));
        assert!(debug_str.contains("config"));
    }

    #[test]
    fn test_event_bridge_clone() {
        let bridge1 = create_test_bridge();
        let bridge2 = bridge1.clone();

        // Both should share the same inner state
        bridge1.stop();
        assert!(!bridge2.is_running());
    }

    #[test]
    fn test_extract_internal_session_id() {
        // SessionUpdate
        let event = Event::SessionUpdate {
            connector_id: "conn".to_string(),
            session_id: "sess-123".to_string(),
            update: SessionUpdate::AgentMessageChunk {
                message_id: "msg-1".to_string(),
                content: ContentBlock::Text {
                    text: "Hello".to_string(),
                },
                _meta: None,
            },
        };
        assert_eq!(
            EventBridge::extract_internal_session_id(&event),
            Some("sess-123".to_string())
        );

        // SessionIdle
        let event = Event::SessionIdle {
            connector_id: "test-connector".to_string(),
            session_id: "sess-456".to_string(),
        };
        assert_eq!(
            EventBridge::extract_internal_session_id(&event),
            Some("sess-456".to_string())
        );

        // MessageCompleted
        let event = Event::MessageCompleted {
            connector_id: "conn".to_string(),
            message: Message {
                id: "msg-1".to_string(),
                session_id: "sess-789".to_string(),
                role: MessageRole::Assistant,
                created_at: chrono::Utc::now(),
                content: vec![],
                status: MessageStatus::Completed,
                metadata: None,
            },
        };
        assert_eq!(
            EventBridge::extract_internal_session_id(&event),
            Some("sess-789".to_string())
        );

        // Error (no session)
        let event = Event::Error {
            message: "error".to_string(),
        };
        assert_eq!(EventBridge::extract_internal_session_id(&event), None);
    }

    #[test]
    fn test_event_bridge_config_default() {
        let config = EventBridgeConfig::default();
        assert!(config.broadcast_system_errors);
        assert!(config.log_unmapped_events);
    }

    #[tokio::test]
    async fn test_handle_session_transferred() {
        let session_manager = SessionManager::new();
        let sse_notifier = SseNotifier::new();
        let agent_request_tracker = Arc::new(crate::agent_requests::AgentRequestTracker::new());
        let bridge = EventBridge::new(session_manager.clone(), sse_notifier, agent_request_tracker);

        // Setup initial mapping
        let client_id = session_manager.register_client(None);
        session_manager.create_mapping(
            &client_id,
            Some("client-session-1".to_string()),
            "gateway-1".to_string(),
        );

        // Get the internal session ID (which represents the gateway session)
        let mapping = session_manager.get_mapping("client-session-1").unwrap();
        let gateway_session_id = mapping.internal_session_id.clone();

        // Simulate SessionTransferred event via internal handler
        EventBridge::handle_session_transferred_internal(
            &bridge.inner,
            "gateway-1",
            &gateway_session_id,
            "opencode-1",
            "opencode-session-new",
            true,
            None,
            None,
        )
        .await;

        // Verify mapping updated
        let updated_mapping = session_manager.get_mapping("client-session-1").unwrap();
        assert_eq!(updated_mapping.connector_id, "opencode-1");
        assert_eq!(updated_mapping.internal_session_id, "opencode-session-new");
        assert_eq!(updated_mapping.client_session_id, "client-session-1"); // Should remain unchanged
    }

    #[tokio::test]
    async fn test_handle_session_transferred_no_mapping() {
        let session_manager = SessionManager::new();
        let sse_notifier = SseNotifier::new();
        let agent_request_tracker = Arc::new(crate::agent_requests::AgentRequestTracker::new());
        let bridge = EventBridge::new(session_manager, sse_notifier, agent_request_tracker);

        // Try to transfer a session that doesn't have a client mapping
        // This should log a warning but not panic
        EventBridge::handle_session_transferred_internal(
            &bridge.inner,
            "gateway-1",
            "nonexistent-session",
            "opencode-1",
            "opencode-session-new",
            true,
            None,
            None,
        )
        .await;

        // Test passes if no panic occurred
    }

    #[tokio::test]
    async fn test_process_event_session_transferred() {
        let session_manager = SessionManager::new();
        let sse_notifier = SseNotifier::new();
        let agent_request_tracker = Arc::new(crate::agent_requests::AgentRequestTracker::new());
        let mut bridge =
            EventBridge::new(session_manager.clone(), sse_notifier, agent_request_tracker);

        // Setup initial mapping
        let client_id = session_manager.register_client(None);
        session_manager.create_mapping(
            &client_id,
            Some("client-session-1".to_string()),
            "gateway-1".to_string(),
        );

        let mapping = session_manager.get_mapping("client-session-1").unwrap();
        let gateway_session_id = mapping.internal_session_id.clone();

        // Create a SessionTransferred event
        let event = Event::SessionTransferred {
            from_connector: "gateway-1".to_string(),
            from_session: gateway_session_id,
            to_connector: "opencode-1".to_string(),
            to_session: "opencode-session-new".to_string(),
            is_new_session: true,
            models: None,
            modes: None,
        };

        let stream = tokio_stream::iter(vec![event]);
        let handle = bridge.start(stream);

        // Wait for processing
        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

        // Verify mapping was updated
        let updated_mapping = session_manager.get_mapping("client-session-1").unwrap();
        assert_eq!(updated_mapping.connector_id, "opencode-1");
        assert_eq!(updated_mapping.internal_session_id, "opencode-session-new");

        bridge.stop();
        let _ = tokio::time::timeout(tokio::time::Duration::from_secs(1), handle).await;
    }

    #[tokio::test]
    async fn test_handle_agent_request_forwards_to_external_client() {
        use dirigent_protocol::SessionOwnership;

        // Setup with external ownership
        let session_manager = SessionManager::new();
        let sse_notifier = SseNotifier::new();
        let agent_request_tracker = Arc::new(crate::agent_requests::AgentRequestTracker::new());

        // Register external client with capabilities
        let client_capabilities =
            serde_json::json!({"fs": {"readTextFile": true}, "terminal": true});
        let client_id = session_manager.register_client(Some(client_capabilities.clone()));

        // Create mapping
        let mapping = session_manager.create_mapping(
            &client_id,
            Some("client-session-1".to_string()),
            "connector-1".to_string(),
        );

        // Update ownership to external with forwarding
        let ownership =
            SessionOwnership::external_forwarded(client_id.clone(), Some(client_capabilities));
        session_manager.update_mapping_ownership("client-session-1", ownership);

        // Subscribe external client to SSE
        let mut sse_stream = sse_notifier.subscribe(&client_id);

        // Create event bridge
        let bridge = EventBridge::new(
            session_manager.clone(),
            sse_notifier.clone(),
            agent_request_tracker.clone(),
        );

        // Simulate an agent request
        tokio::spawn({
            let bridge_inner = bridge.inner.clone();
            let internal_session_id = mapping.internal_session_id.clone();
            async move {
                EventBridge::handle_agent_request_internal(
                    &bridge_inner,
                    "connector-1",
                    &internal_session_id,
                    serde_json::json!("req-123"),
                    "fs/readTextFile",
                    serde_json::json!({"path": "/test.txt"}),
                )
                .await;
            }
        });

        // External client should receive the request
        let notification =
            tokio::time::timeout(tokio::time::Duration::from_secs(1), sse_stream.next())
                .await
                .expect("Should receive notification")
                .expect("Should have notification")
                .expect("Should parse notification");

        // Verify the request was forwarded
        assert_eq!(notification.session_id, "client-session-1");
        match notification.update {
            crate::sse::SessionUpdateVariant::AgentRequest {
                request_id,
                method,
                params,
            } => {
                assert_eq!(request_id, serde_json::json!("req-123"));
                assert_eq!(method, "fs/readTextFile");
                assert_eq!(params["sessionId"], serde_json::json!("client-session-1"));
                assert_eq!(params["path"], serde_json::json!("/test.txt"));
            }
            _ => panic!("Expected AgentRequest variant"),
        }

        // Note: Full roundtrip testing (client response -> connector) would require
        // a mock CoreHandle implementation, which is out of scope for this unit test.
        // Integration tests should verify the complete flow.
    }

}
