//! Gateway connector implementation
//!
//! This module provides a connector that handles messages locally with
//! configurable behavior, including echo mode and command processing.
//! It serves as the default connector for incoming ACP sessions before
//! they are routed to an external agent.
//!
//! # Features
//!
//! - **Echo Mode**: Echoes user messages back as assistant responses
//! - **Command System**: Built-in commands for session management
//! - **Session Routing**: Transfer sessions to other connectors via commands
//!
//! # Commands
//!
//! The GatewayConnector supports these commands:
//! - `/echo on|off` - Enable/disable echo mode
//! - `/list-connectors` - Show available connectors
//! - `/select-connector <id>` - Transfer session to a connector
//! - `/claude` - Shortcut to transfer to a Claude connector
//! - `/help` - Show available commands
//!
//! # Usage
//!
//! ```no_run
//! use dirigent_core::connectors::gateway::{GatewayConnector, GatewayConfig};
//! use dirigent_core::connectors::{Connector, ConnectorCommand};
//!
//! # async fn example() {
//! let config = GatewayConfig::default();
//! let connector = GatewayConnector::new(
//!     "gateway-1".to_string(),
//!     uuid::Uuid::now_v7(),
//!     config,
//! );
//!
//! // Start the connector task
//! let task_handle = connector.start_task().await;
//!
//! // Subscribe to events
//! let mut events = connector.subscribe();
//!
//! // Send a message (will echo if echo mode is on)
//! let cmd_tx = connector.command_tx();
//! cmd_tx.send(ConnectorCommand::SendMessage {
//!     session_id: "session-1".to_string(),
//!     text: "Hello!".to_string(),
//! }).await.ok();
//! # }
//! ```

pub mod commands;
pub mod echo;
pub mod mappings;

use crate::connectors::{Connector, ConnectorCommand};
use crate::types::{ConnectorId, ConnectorKind, ConnectorState, UserId};
use chrono::Utc;
use dirigent_protocol::events::TurnCompleteTrigger;
use dirigent_protocol::session::{ModelInfo, SessionMode, SessionModeState, SessionModelState};
use dirigent_protocol::types::ContentBlock;
use dirigent_protocol::{
    Event, Message, MessageRole, MessageStatus, Session, SessionMetadata, SessionUpdate,
};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{broadcast, mpsc, RwLock};
use tokio::task::JoinHandle;
use tracing::{debug, info, warn};
use uuid::Uuid;

pub use commands::{Command, CommandResult};
pub use echo::EchoConfig;

use tokio::sync::oneshot;

/// Callback type for listing connectors
///
/// This is used to allow the GatewayConnector to query available
/// connectors without having a direct dependency on CoreHandle.
pub type ConnectorListCallback = Arc<dyn Fn() -> Vec<ConnectorSummaryInfo> + Send + Sync>;

/// Request to transfer a session to another connector
#[derive(Debug)]
pub struct SessionTransferRequest {
    /// ID of the Gateway connector
    pub gateway_connector_id: String,
    /// ID of the session being transferred (in Gateway)
    pub gateway_session_id: String,
    /// Target connector ID
    pub target_connector_id: String,
    /// Optional: specific session ID to load in target connector
    pub target_session_id: Option<String>,
    /// Current mode ID from the Gateway session (e.g., "yolo", "write", "ask")
    pub current_mode_id: String,
    /// Current model ID from the Gateway session (e.g., "high", "default", "simple")
    pub current_model_id: String,
    /// Channel to send the result back
    pub result_tx: oneshot::Sender<SessionTransferResult>,
}

/// Result of a session transfer attempt
#[derive(Debug, Clone)]
pub enum SessionTransferResult {
    /// Transfer succeeded
    Transferred {
        /// ID of the connector transferred to
        connector_id: String,
        /// Session ID in the target connector
        session_id: String,
        /// Whether a new session was created
        is_new: bool,
        /// Model state from the target agent (if available)
        models: Option<SessionModelState>,
        /// Mode state from the target agent (if available)
        modes: Option<SessionModeState>,
    },
    /// Transfer failed
    Failed(String),
}

/// Callback type for transferring a session to another connector
///
/// The callback receives a transfer request and should process it asynchronously.
/// Results are sent back via the oneshot channel in the request.
pub type SessionTransferCallback = Arc<dyn Fn(SessionTransferRequest) + Send + Sync>;

/// Summary info about a connector (lightweight, no CoreHandle dependency)
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ConnectorSummaryInfo {
    pub id: String,
    pub title: String,
    pub kind: String,
    pub state: String,
    /// Whether sessions can be transferred to this connector
    pub supports_session_transfer: bool,
    /// Agent type (for ACP connectors only, None otherwise)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub agent_type: Option<crate::connectors::acp::config::ConnectorAgentType>,
}

/// Session state tracked by the GatewayConnector
#[derive(Clone, Debug)]
pub struct GatewaySession {
    /// Session ID
    pub id: String,
    /// Session title
    pub title: String,
    /// Whether echo mode is enabled for this session
    pub echo_enabled: bool,
    /// Messages in this session
    pub messages: Vec<Message>,
    /// Created timestamp
    pub created_at: chrono::DateTime<Utc>,
    /// Updated timestamp
    pub updated_at: chrono::DateTime<Utc>,
    /// Current mode ID (Gateway naming) - default: "ask"
    pub current_mode_id: String,
    /// Current model ID (Gateway naming) - default: "default"
    pub current_model_id: String,
    /// Optional project ID associated with this session
    pub project_id: Option<String>,
}

impl GatewaySession {
    /// Create a new session
    pub fn new(id: String, title: String, echo_enabled: bool, project_id: Option<String>) -> Self {
        let now = Utc::now();
        Self {
            id,
            title,
            echo_enabled,
            messages: Vec::new(),
            created_at: now,
            updated_at: now,
            current_mode_id: mappings::gateway_modes::ASK.to_string(),
            current_model_id: mappings::gateway_models::DAILYDRIVER.to_string(),
            project_id,
        }
    }

    /// Convert to a dirigent_protocol::Session
    pub fn to_protocol_session(&self) -> Session {
        Session {
            id: self.id.clone(),
            title: self.title.clone(),
            created_at: self.created_at,
            updated_at: self.updated_at,
            metadata: SessionMetadata {
                project_path: ".".to_string(),
                model: Some("gateway".to_string()),
                total_messages: self.messages.len() as u32,
                system_message: None,
                current_mode_id: None,
                _meta: None,
                project_id: self.project_id.as_ref().and_then(|p| uuid::Uuid::parse_str(p).ok()),
            },
            cwd: None,
            models: Some(build_gateway_model_state(&self.current_model_id)),
            modes: Some(build_gateway_mode_state(&self.current_mode_id)),
            // Gateway connector doesn't have ACP client ID
            config_options: None,
            acp_client_id: None,
        }
    }
}

/// Build the default mode state advertised by Gateway.
pub fn build_gateway_mode_state(current_mode_id: &str) -> SessionModeState {
    SessionModeState {
        current_mode_id: current_mode_id.to_string(),
        available_modes: vec![
            SessionMode {
                id: mappings::gateway_modes::PLAN.to_string(),
                name: "Plan Mode".to_string(),
                description: Some("Analyze and plan without modifying files".to_string()),
            },
            SessionMode {
                id: mappings::gateway_modes::READONLY.to_string(),
                name: "Read Only".to_string(),
                description: Some("Read files but don't modify anything".to_string()),
            },
            SessionMode {
                id: mappings::gateway_modes::ASK.to_string(),
                name: "Ask Permission".to_string(),
                description: Some("Ask before making changes (default)".to_string()),
            },
            SessionMode {
                id: mappings::gateway_modes::WRITE.to_string(),
                name: "Write".to_string(),
                description: Some("Automatically accept file edits".to_string()),
            },
            SessionMode {
                id: mappings::gateway_modes::YOLO.to_string(),
                name: "YOLO".to_string(),
                description: Some("Bypass all permission prompts".to_string()),
            },
        ],
    }
}

/// Build the default model state advertised by Gateway.
pub fn build_gateway_model_state(current_model_id: &str) -> SessionModelState {
    SessionModelState {
        current_model_id: current_model_id.to_string(),
        available_models: vec![
            ModelInfo {
                model_id: mappings::gateway_models::SIMPLE.to_string(),
                name: "Simple".to_string(),
                description: Some("Fastest, best for quick answers".to_string()),
            },
            ModelInfo {
                model_id: mappings::gateway_models::DAILYDRIVER.to_string(),
                name: "Daily Driver".to_string(),
                description: Some("Balanced for everyday tasks (recommended)".to_string()),
            },
            ModelInfo {
                model_id: mappings::gateway_models::HIGH.to_string(),
                name: "High".to_string(),
                description: Some("Most capable for complex work".to_string()),
            },
        ],
    }
}

/// Configuration for the Gateway connector
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct GatewayConfig {
    /// Human-readable title for this connector
    pub title: String,

    /// Whether echo mode is enabled by default for new sessions
    #[serde(default)]
    pub default_echo_enabled: bool,

    /// Configuration for echo behavior
    #[serde(default)]
    pub echo: EchoConfig,
}

impl Default for GatewayConfig {
    fn default() -> Self {
        Self {
            title: "Gateway".to_string(),
            default_echo_enabled: false,
            echo: EchoConfig::default(),
        }
    }
}

/// Register a Gateway session node in the inspector registry.
#[cfg(feature = "server")]
async fn inspector_register_gateway_session(
    inspector: &Arc<dirigent_inspector::InspectorRegistry>,
    connector_id: &str,
    session: &GatewaySession,
) {
    let sess_node_id = dirigent_inspector::NodeId::new(format!(
        "dirigent/connectors/{}/sessions/{}",
        connector_id, session.id
    ));
    let parent_id =
        dirigent_inspector::NodeId::new(format!("dirigent/connectors/{}", connector_id));

    let meta = dirigent_inspector::NodeMetadata::new(
        dirigent_inspector::NodeKind::AsyncTask,
        &session.title,
    )
    .with_state(dirigent_inspector::NodeState::Running)
    .with_property("session_id", serde_json::json!(&session.id))
    .with_property("status", serde_json::json!("Active"))
    .with_property("message_count", serde_json::json!(session.messages.len()));

    match inspector
        .register(sess_node_id, &parent_id, meta, None)
        .await
    {
        Ok(mut handle) => {
            handle.detach();
        }
        Err(_) => {
            // Already registered — that's fine
        }
    }
}

/// Deregister all Gateway session nodes from the inspector.
#[cfg(feature = "server")]
async fn inspector_deregister_gateway_sessions(
    inspector: &Arc<dirigent_inspector::InspectorRegistry>,
    connector_id: &str,
    sessions: &HashMap<String, GatewaySession>,
) {
    for session_id in sessions.keys() {
        let sess_node_id = dirigent_inspector::NodeId::new(format!(
            "dirigent/connectors/{}/sessions/{}",
            connector_id, session_id
        ));
        let _ = inspector.deregister_subtree(&sess_node_id).await;
    }
}

/// Gateway connector implementation
///
/// A connector that handles messages locally with configurable behavior.
/// It supports echo mode and built-in commands for session management.
pub struct GatewayConnector {
    /// Unique identifier for this connector instance
    id: ConnectorId,

    /// Optional connector UID used for BusEvent routing.
    ///
    /// Populated by the runtime when the connector is registered; legacy
    /// constructor paths default this to `None` and later wiring can fill it in.
    pub connector_uid: Option<Uuid>,

    /// User who owns this connector
    owner: UserId,

    /// Connector configuration
    config: GatewayConfig,

    /// Active sessions tracked by this connector
    sessions: Arc<RwLock<HashMap<String, GatewaySession>>>,

    /// Shared connector state
    state: Arc<RwLock<ConnectorState>>,

    /// Sender for commands to the connector
    cmd_tx: mpsc::Sender<ConnectorCommand>,

    /// Receiver for commands (internal to the connector task)
    cmd_rx: Arc<RwLock<Option<mpsc::Receiver<ConnectorCommand>>>>,

    /// Broadcast sender for events
    events_tx: broadcast::Sender<Event>,

    /// Shared event bus for direct-to-bus publishes.
    sharing_bus: Arc<crate::sharing::bus::SharingBus>,

    /// Optional callback for listing available connectors
    connector_list_callback: Option<ConnectorListCallback>,

    /// Optional callback for transferring sessions
    session_transfer_callback: Option<SessionTransferCallback>,

    /// Optional inspector registry for session tracking
    #[cfg(feature = "server")]
    inspector: Option<Arc<dirigent_inspector::InspectorRegistry>>,
}

/// Helper that publishes an event to both the per-connector broadcast
/// channel and the global `SharingBus`.
async fn emit_event(
    sharing_bus: &Arc<crate::sharing::bus::SharingBus>,
    events_tx: &broadcast::Sender<Event>,
    connector_id: &ConnectorId,
    connector_uid: Option<Uuid>,
    event: Event,
) {
    let bus_event = dirigent_protocol::streaming::BusEvent::from_connector_event(
        event.clone(),
        connector_uid,
        connector_id.clone(),
    );
    sharing_bus.publish(bus_event).await;
    let _ = events_tx.send(event);
}

impl GatewayConnector {
    /// Create a new Gateway connector
    ///
    /// # Arguments
    ///
    /// * `id` - Unique identifier for this connector
    /// * `owner` - User ID of the connector owner
    /// * `config` - Gateway configuration
    pub fn new(
        id: ConnectorId,
        owner: UserId,
        config: GatewayConfig,
        sharing_bus: Arc<crate::sharing::bus::SharingBus>,
    ) -> Self {
        let (cmd_tx, cmd_rx) = mpsc::channel(100);
        let (events_tx, _) = broadcast::channel(1000);

        info!(
            connector_id = %id,
            owner = %owner,
            title = %config.title,
            default_echo = config.default_echo_enabled,
            "Creating Gateway connector"
        );

        Self {
            id,
            connector_uid: None,
            owner,
            config,
            sessions: Arc::new(RwLock::new(HashMap::new())),
            state: Arc::new(RwLock::new(ConnectorState::Initializing)),
            cmd_tx,
            cmd_rx: Arc::new(RwLock::new(Some(cmd_rx))),
            events_tx,
            sharing_bus,
            connector_list_callback: None,
            session_transfer_callback: None,
            #[cfg(feature = "server")]
            inspector: None,
        }
    }

    /// Wrap a raw connector `Event` in a `BusEvent` populated with this
    /// connector's routing identity (`connector_uid` + `connector_id`).
    ///
    /// Call sites that currently broadcast a raw `Event` via `events_tx` can
    /// use this helper when migrating to the `BusEvent` pipeline. This is an
    /// additive helper — existing `events_tx.send(event)` emissions remain
    /// unchanged.
    pub fn to_bus_event(
        &self,
        event: dirigent_protocol::Event,
    ) -> dirigent_protocol::streaming::BusEvent {
        dirigent_protocol::streaming::BusEvent::from_connector_event(
            event,
            self.connector_uid,
            self.id.clone(),
        )
    }

    /// Set the inspector registry for session tracking.
    #[cfg(feature = "server")]
    pub fn set_inspector(&mut self, inspector: Option<Arc<dirigent_inspector::InspectorRegistry>>) {
        self.inspector = inspector;
    }

    /// Set the callback for listing connectors
    ///
    /// This callback is invoked when the `/list-connectors` command is used.
    pub fn set_connector_list_callback(&mut self, callback: ConnectorListCallback) {
        self.connector_list_callback = Some(callback);
    }

    /// Set the callback for transferring sessions
    ///
    /// This callback is invoked when the `/select-connector` or `/claude` commands are used.
    pub fn set_session_transfer_callback(&mut self, callback: SessionTransferCallback) {
        self.session_transfer_callback = Some(callback);
    }

    /// Get the events broadcast sender
    pub fn events_sender(&self) -> broadcast::Sender<Event> {
        self.events_tx.clone()
    }

    /// Get the state Arc
    pub fn state_arc(&self) -> Arc<RwLock<ConnectorState>> {
        Arc::clone(&self.state)
    }

    /// Start the connector's background task
    pub async fn start_task(&self) -> JoinHandle<()> {
        let id = self.id.clone();
        let connector_uid = self.connector_uid;
        let config = self.config.clone();
        let sessions = Arc::clone(&self.sessions);
        let state = Arc::clone(&self.state);
        let events_tx = self.events_tx.clone();
        let sharing_bus = Arc::clone(&self.sharing_bus);
        let connector_list_callback = self.connector_list_callback.clone();
        let session_transfer_callback = self.session_transfer_callback.clone();
        #[cfg(feature = "server")]
        let inspector = self.inspector.clone();

        let cmd_rx = self
            .cmd_rx
            .write()
            .await
            .take()
            .expect("start_task() called more than once - command receiver already taken");

        info!(connector_id = %id, "Starting Gateway connector task");

        tokio::spawn(async move {
            Self::run_task(
                id,
                connector_uid,
                config,
                sessions,
                state,
                events_tx,
                sharing_bus,
                cmd_rx,
                connector_list_callback,
                session_transfer_callback,
                #[cfg(feature = "server")]
                inspector,
            )
            .await;
        })
    }

    /// Main connector task loop
    async fn run_task(
        id: ConnectorId,
        connector_uid: Option<Uuid>,
        config: GatewayConfig,
        sessions: Arc<RwLock<HashMap<String, GatewaySession>>>,
        state: Arc<RwLock<ConnectorState>>,
        events_tx: broadcast::Sender<Event>,
        sharing_bus: Arc<crate::sharing::bus::SharingBus>,
        mut cmd_rx: mpsc::Receiver<ConnectorCommand>,
        connector_list_callback: Option<ConnectorListCallback>,
        session_transfer_callback: Option<SessionTransferCallback>,
        #[cfg(feature = "server")] inspector: Option<Arc<dirigent_inspector::InspectorRegistry>>,
    ) {
        info!(connector_id = %id, "Gateway connector task started");

        // Immediately transition to Ready state (no external connection needed)
        {
            let mut state_guard = state.write().await;
            *state_guard = ConnectorState::Ready;
        }
        emit_event(&sharing_bus, &events_tx, &id, connector_uid, Event::Connected).await;
        // Process commands
        while let Some(cmd) = cmd_rx.recv().await {
            match cmd {
                ConnectorCommand::ListSessions => {
                    debug!(connector_id = %id, "Processing ListSessions command");
                    let sessions_guard = sessions.read().await;
                    let session_list: Vec<Session> = sessions_guard
                        .values()
                        .map(|s| s.to_protocol_session())
                        .collect();
                    emit_event(
                        &sharing_bus,
                        &events_tx,
                        &id,
                        connector_uid,
                        Event::SessionsListed {
                        connector_id: id.clone(),
                        sessions: session_list,
                    },
                    )
                    .await;
                }

                ConnectorCommand::CreateSession { cwd, project_id, ownership } => {
                    debug!(connector_id = %id, cwd = ?cwd, ownership = ?ownership, "Processing CreateSession command");
                    let session_id = uuid::Uuid::new_v4().to_string();
                    let title = format!("Session {}", &session_id[..8]);
                    let session = GatewaySession::new(
                        session_id.clone(),
                        title.clone(),
                        config.default_echo_enabled,
                        project_id,
                    );
                    let protocol_session = session.to_protocol_session();

                    // Register session with inspector
                    #[cfg(feature = "server")]
                    if let Some(ref inspector) = inspector {
                        inspector_register_gateway_session(inspector, &id, &session).await;
                    }

                    sessions.write().await.insert(session_id.clone(), session);

                    emit_event(
                        &sharing_bus,
                        &events_tx,
                        &id,
                        connector_uid,
                        Event::SessionCreated {
                        connector_id: id.clone(),
                        session: protocol_session,
                    },
                    )
                    .await;
                }

                ConnectorCommand::LoadSession { session_id, .. } => {
                    debug!(connector_id = %id, session_id = %session_id, "Processing LoadSession command");
                    let sessions_guard = sessions.read().await;
                    if let Some(session) = sessions_guard.get(&session_id) {
                        // Session exists, emit updated event
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::SessionUpdated {
                            connector_id: id.clone(),
                            session: session.to_protocol_session(),
                        },
                        )
                        .await;
                    } else {
                        // Session doesn't exist - create it
                        drop(sessions_guard);
                        let title = format!("Session {}", &session_id[..8.min(session_id.len())]);
                        let session = GatewaySession::new(
                            session_id.clone(),
                            title,
                            config.default_echo_enabled,
                            None,
                        );
                        let protocol_session = session.to_protocol_session();

                        // Register session with inspector
                        #[cfg(feature = "server")]
                        if let Some(ref inspector) = inspector {
                            inspector_register_gateway_session(inspector, &id, &session).await;
                        }

                        sessions.write().await.insert(session_id.clone(), session);

                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::SessionCreated {
                            connector_id: id.clone(),
                            session: protocol_session,
                        },
                        )
                        .await;
                    }
                }

                ConnectorCommand::ListMessages { session_id } => {
                    debug!(connector_id = %id, session_id = %session_id, "Processing ListMessages command");
                    let sessions_guard = sessions.read().await;
                    if let Some(session) = sessions_guard.get(&session_id) {
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::MessagesListed {
                            messages: session.messages.clone(),
                        },
                        )
                        .await;
                    } else {
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::MessagesListed { messages: vec![] },
                        )
                        .await;
                    }
                }

                ConnectorCommand::SendMessage { session_id, text } => {
                    debug!(connector_id = %id, session_id = %session_id, text_len = text.len(), "Processing SendMessage command");

                    // Get or create session
                    let mut sessions_guard = sessions.write().await;
                    let session = sessions_guard.entry(session_id.clone()).or_insert_with(|| {
                        GatewaySession::new(
                            session_id.clone(),
                            format!("Session {}", &session_id[..8.min(session_id.len())]),
                            config.default_echo_enabled,
                            None,
                        )
                    });

                    // Check for system error message from runtime
                    if text.starts_with("__SYSTEM_ERROR__:") {
                        let error_content = text.strip_prefix("__SYSTEM_ERROR__:").unwrap_or(&text);

                        // Create error message in session
                        let assistant_message_id = uuid::Uuid::new_v4().to_string();
                        let assistant_message = Message {
                            id: assistant_message_id.clone(),
                            session_id: session_id.clone(),
                            role: MessageRole::Assistant,
                            content: vec![dirigent_protocol::MessagePart::Text {
                                text: error_content.to_string(),
                            }],
                            created_at: Utc::now(),
                            status: MessageStatus::Completed,
                            metadata: None,
                        };
                        session.messages.push(assistant_message.clone());
                        session.updated_at = Utc::now();

                        // Emit as assistant message
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::MessageStarted {
                            connector_id: id.clone(),
                            message: assistant_message.clone(),
                        },
                        )
                        .await;
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::MessageCompleted {
                            connector_id: id.clone(),
                            message: assistant_message.clone(),
                        },
                        )
                        .await;
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::TurnComplete {
                            connector_id: id.clone(),
                            session_id: session_id.clone(),
                            message_id: assistant_message_id.clone(),
                            trigger: TurnCompleteTrigger::ExplicitSignal,
                        },
                        )
                        .await;
                        // Don't process as command or echo
                        continue;
                    }

                    // Create user message
                    let user_message_id = uuid::Uuid::new_v4().to_string();
                    let user_message = Message {
                        id: user_message_id.clone(),
                        session_id: session_id.clone(),
                        role: MessageRole::User,
                        content: vec![dirigent_protocol::MessagePart::Text { text: text.clone() }],
                        created_at: Utc::now(),
                        status: MessageStatus::Completed,
                        metadata: None,
                    };
                    session.messages.push(user_message.clone());
                    session.updated_at = Utc::now();

                    // Emit user message completed
                    emit_event(
                        &sharing_bus,
                        &events_tx,
                        &id,
                        connector_uid,
                        Event::MessageCompleted {
                        connector_id: id.clone(),
                        message: user_message,
                    },
                    )
                    .await;
                    // Track if session was transferred to skip further events
                    let mut session_transferred = false;

                    // Check for commands first
                    if let Some(command) = commands::parse_command(&text) {
                        let result = commands::execute_command(
                            command,
                            &id,
                            &session_id,
                            session,
                            connector_list_callback.as_ref(),
                            session_transfer_callback.as_ref(),
                        )
                        .await;

                        // Check if session was transferred
                        session_transferred =
                            matches!(result, CommandResult::SessionTransferred { .. });

                        // For transfers, the EventBridge handles the confirmation message
                        // and SessionIdle. We only emit messages for non-transfer commands.
                        if !session_transferred {
                            let assistant_message_id = uuid::Uuid::new_v4().to_string();
                            let response_text = match result {
                                CommandResult::Message(msg) => msg,
                                CommandResult::Error(err) => format!("Error: {}", err),
                                CommandResult::SessionTransferred { .. } => {
                                    // This branch won't be reached due to the if guard above
                                    unreachable!()
                                }
                            };

                            let assistant_message = Message {
                                id: assistant_message_id.clone(),
                                session_id: session_id.clone(),
                                role: MessageRole::Assistant,
                                content: vec![dirigent_protocol::MessagePart::Text {
                                    text: response_text.clone(),
                                }],
                                created_at: Utc::now(),
                                status: MessageStatus::Completed,
                                metadata: None,
                            };
                            session.messages.push(assistant_message.clone());
                            session.updated_at = Utc::now();

                            // Stream the response
                            emit_event(
                                &sharing_bus,
                                &events_tx,
                                &id,
                                connector_uid,
                                Event::MessageStarted {
                                connector_id: id.clone(),
                                message: assistant_message.clone(),
                            },
                            )
                            .await;
                            // Emit chunk update
                            emit_event(
                                &sharing_bus,
                                &events_tx,
                                &id,
                                connector_uid,
                                Event::SessionUpdate {
                                    connector_id: id.clone(),
                                    session_id: session_id.clone(),
                                    update: SessionUpdate::AgentMessageChunk {
                                        message_id: assistant_message_id.clone(),
                                        content: ContentBlock::Text {
                                            text: response_text,
                                        },
                                        _meta: None,
                                    },
                                },
                            )
                            .await;

                            emit_event(
                                &sharing_bus,
                                &events_tx,
                                &id,
                                connector_uid,
                                Event::MessageCompleted {
                                connector_id: id.clone(),
                                message: assistant_message.clone(),
                            },
                            )
                            .await;
                            emit_event(
                                &sharing_bus,
                                &events_tx,
                                &id,
                                connector_uid,
                                Event::TurnComplete {
                                connector_id: id.clone(),
                                session_id: session_id.clone(),
                                message_id: assistant_message_id.clone(),
                                trigger: TurnCompleteTrigger::ExplicitSignal,
                            },
                            )
                            .await;
                        }
                        // For transfers, the EventBridge handles confirmation + SessionIdle
                    } else if session.echo_enabled {
                        // Echo mode is on - echo the message back
                        let echo_response = echo::generate_echo_response(&text, &config.echo);

                        let assistant_message_id = uuid::Uuid::new_v4().to_string();
                        let assistant_message = Message {
                            id: assistant_message_id.clone(),
                            session_id: session_id.clone(),
                            role: MessageRole::Assistant,
                            content: vec![dirigent_protocol::MessagePart::Text {
                                text: echo_response.clone(),
                            }],
                            created_at: Utc::now(),
                            status: MessageStatus::Completed,
                            metadata: None,
                        };
                        session.messages.push(assistant_message.clone());
                        session.updated_at = Utc::now();

                        // Check if streaming simulation is enabled
                        if config.echo.simulate_streaming {
                            // Stream the echo response with delays between chunks
                            echo::stream_echo_response(
                                &id,
                                connector_uid,
                                &session_id,
                                &assistant_message_id,
                                &echo_response,
                                &config.echo,
                                &events_tx,
                                &sharing_bus,
                            )
                            .await;
                        } else {
                            // Send message events without streaming delays
                            // Pattern: MessageStarted (empty) -> AgentMessageChunk (content) -> MessageCompleted
                            // This matches OpenCode behavior and tests chunk handling
                            let placeholder_message = Message {
                                id: assistant_message_id.clone(),
                                session_id: session_id.clone(),
                                role: MessageRole::Assistant,
                                content: vec![], // Empty - content comes via AgentMessageChunk
                                created_at: Utc::now(),
                                status: MessageStatus::Streaming,
                                metadata: None,
                            };
                            emit_event(
                                &sharing_bus,
                                &events_tx,
                                &id,
                                connector_uid,
                                Event::MessageStarted {
                                connector_id: id.clone(),
                                message: placeholder_message,
                            },
                            )
                            .await;
                            emit_event(
                                &sharing_bus,
                                &events_tx,
                                &id,
                                connector_uid,
                                Event::SessionUpdate {
                                    connector_id: id.clone(),
                                    session_id: session_id.clone(),
                                    update: SessionUpdate::AgentMessageChunk {
                                        message_id: assistant_message_id,
                                        content: ContentBlock::Text {
                                            text: echo_response,
                                        },
                                        _meta: None,
                                    },
                                },
                            )
                            .await;
                        }

                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::MessageCompleted {
                            connector_id: id.clone(),
                            message: assistant_message.clone(),
                        },
                        )
                        .await;
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::TurnComplete {
                            connector_id: id.clone(),
                            session_id: session_id.clone(),
                            message_id: assistant_message.id.clone(),
                            trigger: TurnCompleteTrigger::ExplicitSignal,
                        },
                        )
                        .await;
                    } else {
                        // Echo mode is off - provide helpful response
                        let assistant_message_id = uuid::Uuid::new_v4().to_string();
                        let response_text = "Echo mode is disabled. Use `/echo on` to enable, or `/help` for available commands.".to_string();
                        let assistant_message = Message {
                            id: assistant_message_id.clone(),
                            session_id: session_id.clone(),
                            role: MessageRole::Assistant,
                            content: vec![dirigent_protocol::MessagePart::Text {
                                text: response_text.clone(),
                            }],
                            created_at: Utc::now(),
                            status: MessageStatus::Completed,
                            metadata: None,
                        };
                        session.messages.push(assistant_message.clone());
                        session.updated_at = Utc::now();

                        // Send message events using OpenCode-compatible pattern:
                        // MessageStarted (empty) -> AgentMessageChunk (content) -> MessageCompleted
                        let placeholder_message = Message {
                            id: assistant_message_id.clone(),
                            session_id: session_id.clone(),
                            role: MessageRole::Assistant,
                            content: vec![], // Empty - content comes via AgentMessageChunk
                            created_at: Utc::now(),
                            status: MessageStatus::Streaming,
                            metadata: None,
                        };
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::MessageStarted {
                            connector_id: id.clone(),
                            message: placeholder_message,
                        },
                        )
                        .await;
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::SessionUpdate {
                                connector_id: id.clone(),
                                session_id: session_id.clone(),
                                update: SessionUpdate::AgentMessageChunk {
                                    message_id: assistant_message_id.clone(),
                                    content: ContentBlock::Text {
                                        text: response_text,
                                    },
                                    _meta: None,
                                },
                            },
                        )
                        .await;
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::MessageCompleted {
                            connector_id: id.clone(),
                            message: assistant_message.clone(),
                        },
                        )
                        .await;
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::TurnComplete {
                            connector_id: id.clone(),
                            session_id: session_id.clone(),
                            message_id: assistant_message_id,
                            trigger: TurnCompleteTrigger::ExplicitSignal,
                        },
                        )
                        .await;
                    }

                    // Emit session idle (unless session was transferred)
                    if !session_transferred {
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::SessionIdle {
                            connector_id: id.to_string(),
                            session_id: session_id.clone(),
                        },
                        )
                        .await;
                    }
                }

                ConnectorCommand::CancelGeneration { session_id } => {
                    debug!(connector_id = %id, session_id = %session_id, "Processing CancelGeneration (no-op for Gateway)");
                    // No-op for Gateway connector since responses are instant
                }

                ConnectorCommand::Reconnect => {
                    debug!(connector_id = %id, "Processing Reconnect (no-op for Gateway)");
                    // No-op since we don't have external connections
                }

                ConnectorCommand::AgentResponse { .. } => {
                    // Gateway connector does not support agent-initiated requests
                    warn!(
                        connector_id = %id,
                        "Received AgentResponse command but Gateway connector does not support agent-initiated requests"
                    );
                }

                ConnectorCommand::SetSessionMode {
                    session_id,
                    mode_id,
                } => {
                    if let Some(session) = sessions.write().await.get_mut(&session_id) {
                        session.current_mode_id = mode_id.clone();
                        session.updated_at = Utc::now();

                        // Emit SessionMetadataReceived event to notify clients
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::SessionMetadataReceived {
                            connector_id: id.clone(),
                            session_id: session_id.clone(),
                            models: Some(build_gateway_model_state(&session.current_model_id)),
                            modes: Some(build_gateway_mode_state(&session.current_mode_id)),
                            config_options: None,
                        },
                        )
                        .await;
                        debug!(session_id = %session_id, mode_id = %mode_id, "Gateway session mode updated");
                    } else {
                        warn!(session_id = %session_id, "SetSessionMode: session not found");
                    }
                }

                ConnectorCommand::CloseSession { session_id } => {
                    warn!(connector_id = %id, session_id = %session_id, "CloseSession not supported by Gateway connector");
                }

                ConnectorCommand::SetSessionModel {
                    session_id,
                    model_id,
                } => {
                    if let Some(session) = sessions.write().await.get_mut(&session_id) {
                        session.current_model_id = model_id.clone();
                        session.updated_at = Utc::now();

                        // Emit SessionMetadataReceived event to notify clients
                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::SessionMetadataReceived {
                            connector_id: id.clone(),
                            session_id: session_id.clone(),
                            models: Some(build_gateway_model_state(&session.current_model_id)),
                            modes: Some(build_gateway_mode_state(&session.current_mode_id)),
                            config_options: None,
                        },
                        )
                        .await;
                        debug!(session_id = %session_id, model_id = %model_id, "Gateway session model updated");
                    } else {
                        warn!(session_id = %session_id, "SetSessionModel: session not found");
                    }
                }

                ConnectorCommand::SetConfigOption { .. } => {
                    warn!(connector_id = %id, "SetConfigOption not supported by Gateway connector");
                }

                ConnectorCommand::Shutdown => {
                    info!(connector_id = %id, "Received Shutdown command, stopping connector");

                    // Deregister session nodes from inspector
                    #[cfg(feature = "server")]
                    if let Some(ref inspector) = inspector {
                        let sessions_guard = sessions.read().await;
                        inspector_deregister_gateway_sessions(inspector, &id, &sessions_guard)
                            .await;
                    }

                    {
                        let mut state_guard = state.write().await;
                        *state_guard = ConnectorState::Stopped;
                    }
                    break;
                }
            }
        }

        info!(connector_id = %id, "Gateway connector task stopped");
    }
}

impl Connector for GatewayConnector {
    fn id(&self) -> &ConnectorId {
        &self.id
    }

    fn kind(&self) -> ConnectorKind {
        ConnectorKind::Gateway
    }

    fn owner(&self) -> &UserId {
        &self.owner
    }

    fn title(&self) -> &str {
        &self.config.title
    }

    fn state(&self) -> ConnectorState {
        match self.state.try_read() {
            Ok(state_guard) => state_guard.clone(),
            Err(_) => ConnectorState::Initializing,
        }
    }

    fn command_tx(&self) -> mpsc::Sender<ConnectorCommand> {
        self.cmd_tx.clone()
    }

    fn subscribe(&self) -> broadcast::Receiver<Event> {
        self.events_tx.subscribe()
    }

    fn stop(&self) {
        let cmd_tx = self.cmd_tx.clone();
        tokio::spawn(async move {
            let _ = cmd_tx.send(ConnectorCommand::Shutdown).await;
        });
    }

    fn get_available_commands(&self) -> Vec<crate::acp::protocol::streaming::Command> {
        // Return the built-in Gateway commands
        vec![
            crate::acp::protocol::streaming::Command {
                name: "echo".to_string(),
                description: Some("Enable or disable echo mode (/echo on|off)".to_string()),
            },
            crate::acp::protocol::streaming::Command {
                name: "help".to_string(),
                description: Some("Show available commands".to_string()),
            },
            crate::acp::protocol::streaming::Command {
                name: "list-connectors".to_string(),
                description: Some("Show available connectors".to_string()),
            },
            crate::acp::protocol::streaming::Command {
                name: "select-connector".to_string(),
                description: Some(
                    "Transfer session to a connector (/select-connector <id>)".to_string(),
                ),
            },
            crate::acp::protocol::streaming::Command {
                name: "claude".to_string(),
                description: Some("Transfer session to a Claude connector".to_string()),
            },
        ]
    }
}

/// Returns the transient commands that should be available on all connectors
///
/// These are session management commands provided by the Gateway that should be
/// merged with any target connector's commands (unless the connector already
/// provides a command with the same name).
pub fn get_transient_commands() -> Vec<crate::acp::protocol::streaming::Command> {
    vec![
        crate::acp::protocol::streaming::Command {
            name: "list-connectors".to_string(),
            description: Some("Show available connectors".to_string()),
        },
        crate::acp::protocol::streaming::Command {
            name: "select-connector".to_string(),
            description: Some(
                "Transfer session to a connector (/select-connector <id>)".to_string(),
            ),
        },
        crate::acp::protocol::streaming::Command {
            name: "claude".to_string(),
            description: Some("Transfer session to a Claude connector".to_string()),
        },
    ]
}

/// Merges agent commands with transient gateway commands
///
/// Transient commands are added unless the agent already provides a command with the same name.
/// This ensures session management commands are available on all connectors.
pub fn merge_with_transient_commands(
    mut agent_commands: Vec<crate::acp::protocol::streaming::Command>,
) -> Vec<crate::acp::protocol::streaming::Command> {
    use std::collections::HashSet;

    let agent_names: HashSet<String> = agent_commands.iter().map(|c| c.name.clone()).collect();

    for cmd in get_transient_commands() {
        if !agent_names.contains(&cmd.name) {
            agent_commands.push(cmd);
        }
    }

    agent_commands
}

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

    #[test]
    fn test_gateway_config_default() {
        let config = GatewayConfig::default();
        assert_eq!(config.title, "Gateway");
        assert!(!config.default_echo_enabled);
    }

    #[test]
    fn test_gateway_config_serialization() {
        let config = GatewayConfig {
            title: "Test".to_string(),
            default_echo_enabled: true,
            echo: EchoConfig::default(),
        };

        let json = serde_json::to_string(&config).unwrap();
        let deserialized: GatewayConfig = serde_json::from_str(&json).unwrap();

        assert_eq!(deserialized.title, config.title);
        assert_eq!(
            deserialized.default_echo_enabled,
            config.default_echo_enabled
        );
    }

    #[tokio::test]
    async fn test_gateway_connector_creation() {
        let config = GatewayConfig::default();
        let connector = GatewayConnector::new(
            "test-conn".to_string(),
            uuid::Uuid::nil(),
            config,
            crate::sharing::bus::SharingBus::new(),
        );

        assert_eq!(connector.id(), "test-conn");
        assert_eq!(*connector.owner(), uuid::Uuid::nil());
        assert_eq!(connector.title(), "Gateway");
        assert_eq!(connector.kind(), ConnectorKind::Gateway);
        assert_eq!(connector.state(), ConnectorState::Initializing);
    }

    #[test]
    fn test_gateway_session_creation() {
        let session =
            GatewaySession::new("session-1".to_string(), "Test Session".to_string(), true, None);

        assert_eq!(session.id, "session-1");
        assert_eq!(session.title, "Test Session");
        assert!(session.echo_enabled);
        assert!(session.messages.is_empty());
    }

    #[test]
    fn test_gateway_session_to_protocol() {
        let session =
            GatewaySession::new("session-1".to_string(), "Test Session".to_string(), true, None);
        let protocol_session = session.to_protocol_session();

        assert_eq!(protocol_session.id, "session-1");
        assert_eq!(protocol_session.title, "Test Session");
        assert_eq!(protocol_session.metadata.model, Some("gateway".to_string()));
    }

    #[test]
    fn test_session_transfer_result_types() {
        let success = SessionTransferResult::Transferred {
            connector_id: "opencode-1".to_string(),
            session_id: "session-new".to_string(),
            is_new: true,
            models: None,
            modes: None,
        };

        let failure = SessionTransferResult::Failed("Connector not ready".to_string());

        match success {
            SessionTransferResult::Transferred { is_new, .. } => assert!(is_new),
            _ => panic!("Expected Transferred"),
        }

        match failure {
            SessionTransferResult::Failed(reason) => assert!(reason.contains("not ready")),
            _ => panic!("Expected Failed"),
        }
    }
}
