//! JSON-RPC Request Handler for ACP Server
//!
//! This module implements the RPC handler that processes incoming JSON-RPC requests
//! from ACP clients. It dispatches requests to appropriate method handlers and manages
//! the request/response lifecycle.
//!
//! ## Architecture
//!
//! The `RpcHandler` uses a `ConnectorOperations` trait to abstract connector access.
//! This design avoids circular dependencies - the web server implements this trait
//! to bridge to `CoreHandle`, allowing `dirigent_acp_api` to remain independent of
//! `dirigent_core`.
//!
//! ## Module Organization
//!
//! - `types` - Request/response parameter types and `ConnectorOperations` trait
//! - `noop` - No-op implementation for testing
//!
//! ## Supported Methods
//!
//! - `initialize` - Client handshake, returns server capabilities
//! - `session/new` - Create a new session
//! - `session/load` - Load an existing session
//! - `session/prompt` - Send a prompt to a session
//! - `session/cancel` - Cancel active generation
//! - `session/close` - Close a session
//! - `session/set_mode` - Set the session mode (legacy)
//! - `session/set_model` - Set the session model (legacy)
//! - `session/set_config_option` - Set a configuration option (unified mode/model setter)
//! - `session/resume` - Resume an existing session without history replay

// Submodules
mod noop;
pub mod types;

// Re-exports
pub use noop::NoOpConnectorOperations;
pub use types::{
    AgentCapabilities, AgentInfo, AuthMethod, ConnectorInfo, ConnectorOperations, ContentBlock,
    InitializeParams, InitializeResult, McpCapabilities, PromptCapabilities, PromptContent,
    SessionCancelParams, SessionCancelResult, SessionCapabilities, SessionCloseParams,
    SessionCloseResult, SessionInfo, SessionListEntry, SessionListParams, SessionListResult,
    SessionLoadParams, SessionLoadResult, SessionNewParams, SessionNewResult, SessionPromptParams,
    SessionPromptResult, SessionResumeParams, SessionResumeResult, SessionSetConfigOptionParams,
    SessionSetModeParams, SessionSetModelParams,
};

use tracing::{debug, error, info, warn};

use crate::error::JsonRpcErrorObject;
use crate::jsonrpc::{
    JsonRpcId, JsonRpcRequest, JsonRpcRequestBatch, JsonRpcResponse, JsonRpcResponseBatch,
};
use crate::session_manager::SessionManager;
use crate::sse::{models_to_config_option, modes_to_config_option};

/// Protocol version supported by this server
pub const ACP_PROTOCOL_VERSION: u32 = 1;

/// Server name for capability responses
pub const SERVER_NAME: &str = "dirigent-acp-server";

// ============================================================================
// RpcHandler Implementation
// ============================================================================

/// JSON-RPC request handler for the ACP Server
///
/// The RpcHandler processes incoming JSON-RPC requests, dispatches them to
/// the appropriate method handlers, and manages session state through the
/// SessionManager.
///
/// ## Thread Safety
///
/// The RpcHandler is designed to be cloned and shared across async tasks.
/// The SessionManager uses internal locking for thread-safe access.
pub struct RpcHandler<C: ConnectorOperations> {
    /// Session manager for tracking client sessions
    session_manager: SessionManager,

    /// Connector operations implementation
    connector_ops: C,

    /// SSE notifier for broadcasting events to clients
    sse_notifier: crate::sse::SseNotifier,

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

impl<C: ConnectorOperations> RpcHandler<C> {
    /// Create a new RPC handler
    ///
    /// # Parameters
    /// - `session_manager`: The session manager for tracking sessions
    /// - `connector_ops`: Implementation of connector operations
    /// - `sse_notifier`: The SSE notifier for broadcasting events to clients
    /// - `agent_request_tracker`: The agent request tracker for bidirectional requests
    pub fn new(
        session_manager: SessionManager,
        connector_ops: C,
        sse_notifier: crate::sse::SseNotifier,
        agent_request_tracker: std::sync::Arc<crate::agent_requests::AgentRequestTracker>,
    ) -> Self {
        Self {
            session_manager,
            connector_ops,
            sse_notifier,
            agent_request_tracker,
        }
    }

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

    /// Get a reference to the agent request tracker
    pub fn agent_request_tracker(
        &self,
    ) -> &std::sync::Arc<crate::agent_requests::AgentRequestTracker> {
        &self.agent_request_tracker
    }

    /// Resolve a connector ID from a preference string (ID or magic word)
    ///
    /// # Parameters
    /// - `preference`: Either a connector ID or a magic word (claude, codex, gemini)
    ///
    /// # Returns
    /// The resolved connector ID, or an error if not found
    async fn resolve_connector_by_preference(
        &self,
        preference: &str,
    ) -> Result<String, crate::error::AcpServerError> {
        // First, try exact ID match by listing all connectors
        let connectors = self.connector_ops.list_connectors().await?;

        // Try exact ID match
        if connectors.iter().any(|c| c.id == preference) {
            return Ok(preference.to_string());
        }

        // Try to match by agent_type magic word
        // Magic words: claude, codex (alias: openai), gemini (alias: google)
        let magic_word = preference.to_lowercase();
        let is_magic_word = matches!(
            magic_word.as_str(),
            "claude" | "codex" | "openai" | "gemini" | "google"
        );

        if is_magic_word {
            // Find first connector whose name/type contains the magic word
            for connector in &connectors {
                // Match if connector name/type contains the magic word (or its canonical form)
                let name_lower = connector.name.to_lowercase();
                let type_lower = connector.connector_type.to_lowercase();

                let matches = match magic_word.as_str() {
                    "claude" => name_lower.contains("claude") || type_lower.contains("claude"),
                    "codex" | "openai" => {
                        name_lower.contains("codex")
                            || name_lower.contains("openai")
                            || type_lower.contains("codex")
                            || type_lower.contains("openai")
                    }
                    "gemini" | "google" => {
                        name_lower.contains("gemini")
                            || name_lower.contains("google")
                            || type_lower.contains("gemini")
                            || type_lower.contains("google")
                    }
                    _ => false,
                };

                if matches {
                    debug!(
                        "Matched magic word '{}' to connector: {} ({})",
                        preference, connector.id, connector.name
                    );
                    return Ok(connector.id.clone());
                }
            }

            return Err(crate::error::AcpServerError::ConnectorNotFound(format!(
                "No connector found matching agent type '{}'",
                preference
            )));
        }

        // No match found
        Err(crate::error::AcpServerError::ConnectorNotFound(format!(
            "No connector found matching preference '{}'",
            preference
        )))
    }

    /// Handle an incoming request (single or batch)
    ///
    /// This is the main entry point for processing JSON-RPC requests.
    /// It handles both single requests and batch requests.
    ///
    /// # Parameters
    /// - `body`: The raw JSON body of the request
    /// - `client_id`: Optional client ID for authenticated requests
    ///
    /// # Returns
    /// A JSON-RPC response (single or batch)
    pub async fn handle_request(
        &self,
        body: &str,
        client_id: Option<&str>,
    ) -> JsonRpcResponseBatch {
        // Try to parse as batch or single request
        match serde_json::from_str::<JsonRpcRequestBatch>(body) {
            Ok(batch) => self.handle_batch(batch, client_id).await,
            Err(e) => {
                error!("Failed to parse JSON-RPC request: {}", e);
                JsonRpcResponseBatch::Single(JsonRpcResponse::error_with_null_id(
                    JsonRpcErrorObject::parse_error(format!("Invalid JSON: {}", e)),
                ))
            }
        }
    }

    /// Handle a batch of requests
    ///
    /// Processes each request in the batch and returns a batch of responses.
    /// Notifications (requests without id) do not generate responses.
    ///
    /// # Parameters
    /// - `batch`: The batch of requests
    /// - `client_id`: Optional client ID
    ///
    /// # Returns
    /// A batch of responses
    pub async fn handle_batch(
        &self,
        batch: JsonRpcRequestBatch,
        client_id: Option<&str>,
    ) -> JsonRpcResponseBatch {
        match batch {
            JsonRpcRequestBatch::Single(request) => {
                if request.is_notification() {
                    // Notifications don't get responses, but we still process them
                    self.dispatch_request(&request, client_id).await;
                    // Return an empty batch for notifications
                    JsonRpcResponseBatch::Batch(vec![])
                } else {
                    let response = self.dispatch_request(&request, client_id).await;
                    JsonRpcResponseBatch::Single(response)
                }
            }
            JsonRpcRequestBatch::Batch(requests) => {
                if requests.is_empty() {
                    // Empty batch is an invalid request
                    return JsonRpcResponseBatch::Single(JsonRpcResponse::error_with_null_id(
                        JsonRpcErrorObject::invalid_request("Empty batch request"),
                    ));
                }

                let mut responses = Vec::new();

                for request in requests {
                    if request.is_notification() {
                        // Process notification but don't add response
                        self.dispatch_request(&request, client_id).await;
                    } else {
                        let response = self.dispatch_request(&request, client_id).await;
                        responses.push(response);
                    }
                }

                // If all requests were notifications, return empty batch
                if responses.is_empty() {
                    JsonRpcResponseBatch::Batch(vec![])
                } else {
                    JsonRpcResponseBatch::from_vec(responses)
                }
            }
        }
    }

    /// Dispatch a single request to the appropriate handler
    async fn dispatch_request(
        &self,
        request: &JsonRpcRequest,
        client_id: Option<&str>,
    ) -> JsonRpcResponse {
        let id = request.id.clone().unwrap_or(JsonRpcId::Null);

        // Validate the request
        if let Err(e) = request.validate() {
            return JsonRpcResponse::error(JsonRpcErrorObject::invalid_request(e), id);
        }

        debug!("Dispatching RPC method: {}", request.method);

        // Route to appropriate handler
        let result = match request.method.as_str() {
            "initialize" => self.handle_initialize(request, client_id).await,
            "session/new" => self.handle_session_new(request, client_id).await,
            "session/load" => self.handle_session_load(request, client_id).await,
            "session/prompt" => self.handle_session_prompt(request, client_id).await,
            "session/cancel" => self.handle_session_cancel(request, client_id).await,
            "session/close" => self.handle_session_close(request, client_id).await,
            "session/set_mode" => self.handle_session_set_mode(request, client_id).await,
            "session/set_model" => self.handle_session_set_model(request, client_id).await,
            "session/set_config_option" => {
                self.handle_session_set_config_option(request, client_id)
                    .await
            }
            "session/list" => self.handle_session_list(request, client_id).await,
            "session/resume" => self.handle_session_resume(request, client_id).await,
            _ => {
                warn!("Unknown method: {}", request.method);
                Err(JsonRpcErrorObject::method_not_found(&request.method))
            }
        };

        match result {
            Ok(value) => JsonRpcResponse::success(value, id),
            Err(error) => JsonRpcResponse::error(error, id),
        }
    }

    // ========================================================================
    // Request Handlers
    // ========================================================================

    /// Handle the initialize request
    ///
    /// Registers the client and returns server capabilities.
    async fn handle_initialize(
        &self,
        request: &JsonRpcRequest,
        client_id: Option<&str>,
    ) -> Result<serde_json::Value, JsonRpcErrorObject> {
        // Parse parameters (optional)
        let params: InitializeParams = request
            .params
            .as_ref()
            .map(|p| serde_json::from_value(p.clone()))
            .transpose()
            .map_err(|e| {
                JsonRpcErrorObject::invalid_params(format!("Invalid initialize params: {}", e))
            })?
            .unwrap_or(InitializeParams {
                capabilities: None,
                client_name: None,
                client_version: None,
            });

        // Use provided client_id from X-Client-Id header if available,
        // otherwise register a new client
        let new_client_id = if let Some(id) = client_id {
            info!(
                client_id = %id,
                "Using provided client_id from X-Client-Id header"
            );
            // Register this client_id with the session manager
            self.session_manager
                .register_client_with_id(id.to_string(), params.capabilities);
            id.to_string()
        } else {
            // Fallback: generate a new client_id
            let id = self.session_manager.register_client(params.capabilities);
            info!(
                client_id = %id,
                "Generated new client_id (no X-Client-Id header provided)"
            );
            id
        };

        info!(
            "Client initialized: {} (name: {:?}, version: {:?})",
            new_client_id, params.client_name, params.client_version
        );

        let result = InitializeResult {
            protocol_version: ACP_PROTOCOL_VERSION,
            agent_capabilities: AgentCapabilities::default(),
            agent_info: AgentInfo {
                name: SERVER_NAME.to_string(),
                title: "Dirigent ACP Server".to_string(),
                version: env!("CARGO_PKG_VERSION").to_string(),
            },
            auth_methods: vec![],
            client_id: new_client_id,
        };

        serde_json::to_value(result)
            .map_err(|e| JsonRpcErrorObject::internal_error(format!("Serialization error: {}", e)))
    }

    /// Handle session/new request
    ///
    /// Creates a new session via the target connector.
    async fn handle_session_new(
        &self,
        request: &JsonRpcRequest,
        client_id: Option<&str>,
    ) -> Result<serde_json::Value, JsonRpcErrorObject> {
        // Parse parameters
        let params: SessionNewParams = request
            .params
            .as_ref()
            .map(|p| serde_json::from_value(p.clone()))
            .transpose()
            .map_err(|e| {
                JsonRpcErrorObject::invalid_params(format!("Invalid session/new params: {}", e))
            })?
            .unwrap_or(SessionNewParams {
                connector_id: None,
                cwd: None,
                session_id: None,
            });

        // Get or generate a client ID
        let client_id = client_id
            .map(|s| s.to_string())
            .unwrap_or_else(|| format!("http-client-{}", uuid::Uuid::now_v7()));

        // Determine which connector to use
        let connector_id = match params.connector_id {
            Some(id) => id,
            None => {
                // Check if client has preferred_connector (from --select-connector)
                let client_info = self.session_manager.get_client(&client_id);
                let preferred_connector = client_info
                    .as_ref()
                    .and_then(|c| c.preferred_connector.clone());

                if let Some(ref preferred) = preferred_connector {
                    // Try to resolve the preferred connector
                    match self.resolve_connector_by_preference(preferred).await {
                        Ok(connector_id) => {
                            info!(
                                "Resolved preferred connector '{}' to connector_id: {}",
                                preferred, connector_id
                            );
                            connector_id
                        }
                        Err(e) => {
                            warn!(
                                "Failed to resolve preferred connector '{}': {}. Falling back to default.",
                                preferred, e
                            );
                            self.connector_ops
                                .default_connector_id()
                                .await
                                .ok_or_else(|| {
                                    JsonRpcErrorObject::invalid_params(
                                        "No connector_id specified and no default connector configured",
                                    )
                                })?
                        }
                    }
                } else {
                    // Use default connector if available
                    self.connector_ops
                        .default_connector_id()
                        .await
                        .ok_or_else(|| {
                            JsonRpcErrorObject::invalid_params(
                                "No connector_id specified and no default connector configured",
                            )
                        })?
                }
            }
        };

        // Get client capabilities from the session manager
        let client_capabilities: Option<serde_json::Value> = self
            .session_manager
            .get_client(&client_id)
            .and_then(|client| client.capabilities);

        // Create external ownership with capability forwarding
        let ownership = dirigent_protocol::SessionOwnership::external_forwarded(
            client_id.clone(),
            client_capabilities,
        );

        // Create session via connector operations
        let session_info = self
            .connector_ops
            .create_session(&connector_id, params.cwd, ownership.clone())
            .await
            .map_err(JsonRpcErrorObject::from)?;

        // Create the session mapping
        let mapping = self.session_manager.create_mapping(
            &client_id,
            params.session_id,
            connector_id.clone(),
        );

        // Update the mapping with the actual connector session ID
        self.session_manager.update_mapping_internal_session(
            &mapping.client_session_id,
            session_info.session_id.clone(),
        );

        // Apply ownership model to the mapping for permission request routing
        self.session_manager
            .update_mapping_ownership(&mapping.client_session_id, ownership);

        info!(
            "Created session: {} -> {} on connector {}",
            mapping.client_session_id, mapping.internal_session_id, connector_id
        );

        // Build config_options from modes and models
        let config_options = {
            let mut options = Vec::new();
            if let Some(ref modes) = session_info.modes {
                options.push(modes_to_config_option(modes));
            }
            if let Some(ref models) = session_info.models {
                options.push(models_to_config_option(models));
            }
            if options.is_empty() {
                None
            } else {
                Some(options)
            }
        };

        let result = SessionNewResult {
            session_id: mapping.client_session_id.clone(),
            title: session_info.title,
            connector_id: session_info.connector_id,
            created_at: session_info.created_at,
            models: session_info.models.clone(),
            modes: session_info.modes.clone(),
            config_options,
        };

        // Send available_commands_update notification
        let commands = match self
            .connector_ops
            .get_connector_commands(&connector_id)
            .await
        {
            Ok(cmds) => cmds,
            Err(e) => {
                warn!(
                    "Failed to get connector commands for {}: {}. Using empty list.",
                    connector_id, e
                );
                Vec::new()
            }
        };

        let update_params = crate::sse::SessionUpdateParams {
            session_id: result.session_id.clone(),
            update: crate::sse::SessionUpdateVariant::AvailableCommandsUpdate {
                available_commands: commands.clone(),
            },
            event_type_override: None,
        };

        // Broadcast to this client (ignore errors)
        if let Err(_) = self.sse_notifier.broadcast(&client_id, update_params) {
            debug!(
                "Client {} not subscribed to SSE yet, commands update not sent",
                client_id
            );
        }

        serde_json::to_value(result)
            .map_err(|e| JsonRpcErrorObject::internal_error(format!("Serialization error: {}", e)))
    }

    /// Handle session/load request
    ///
    /// Loads an existing session from a connector.
    async fn handle_session_load(
        &self,
        request: &JsonRpcRequest,
        client_id: Option<&str>,
    ) -> Result<serde_json::Value, JsonRpcErrorObject> {
        let params_value = request
            .params
            .as_ref()
            .ok_or_else(|| JsonRpcErrorObject::invalid_params("Missing params for session/load"))?
            .clone();

        let params: SessionLoadParams = serde_json::from_value(params_value).map_err(|e| {
            JsonRpcErrorObject::invalid_params(format!("Invalid session/load params: {}", e))
        })?;

        let client_id = client_id
            .map(|s| s.to_string())
            .unwrap_or_else(|| format!("http-client-{}", uuid::Uuid::now_v7()));

        // Resolve connector ID: explicit param -> session mapping -> archivist lookup -> client sessions -> default
        let connector_id = if let Some(ref cid) = params.connector_id {
            cid.clone()
        } else if let Some(mapping) = self.session_manager.get_mapping(&params.session_id) {
            // Session was already mapped (e.g. from a prior session/list call)
            mapping.connector_id
        } else if let Some(connector) = self.connector_ops.resolve_session_connector(&params.session_id).await {
            // Archivist knows which connector owns this session
            connector
        } else if let Some(connector) = self.resolve_connector_from_client_sessions(&client_id) {
            connector
        } else if let Some(default) = self.connector_ops.default_connector_id().await {
            default
        } else {
            return Err(JsonRpcErrorObject::invalid_params(
                "No connector_id specified and no default connector available",
            ));
        };

        // Pre-route: if resolved to gateway but archivist knows the real owner, override
        let connector_id = if self.is_gateway_connector(&connector_id).await {
            if let Some(real_connector) = self.connector_ops.resolve_session_connector(&params.session_id).await {
                info!(
                    "Pre-routing session {} from gateway {} to owning connector {}",
                    params.session_id, connector_id, real_connector
                );
                real_connector
            } else {
                connector_id
            }
        } else {
            connector_id
        };

        // Load session via connector operations
        let mcp_servers_value = params.mcp_servers.as_ref().map(|servers| {
            serde_json::Value::Array(servers.clone())
        });
        let session_info = self
            .connector_ops
            .load_session(&connector_id, &params.session_id, params.cwd.clone(), mcp_servers_value)
            .await
            .map_err(JsonRpcErrorObject::from)?;

        // Create the session mapping
        let mapping = self.session_manager.create_mapping(
            &client_id,
            Some(params.session_id.clone()),
            connector_id.clone(),
        );

        // Update the mapping with the actual connector session ID
        self.session_manager.update_mapping_internal_session(
            &mapping.client_session_id,
            session_info.session_id.clone(),
        );

        info!(
            "Loaded session: {} -> {} from connector {}",
            mapping.client_session_id, mapping.internal_session_id, connector_id
        );

        // Build config_options from modes and models
        let config_options = {
            let mut options = Vec::new();
            if let Some(ref modes) = session_info.modes {
                options.push(modes_to_config_option(modes));
            }
            if let Some(ref models) = session_info.models {
                options.push(models_to_config_option(models));
            }
            if options.is_empty() {
                None
            } else {
                Some(options)
            }
        };

        let result = SessionLoadResult {
            session_id: mapping.client_session_id,
            title: session_info.title,
            connector_id: session_info.connector_id,
            created_at: session_info.created_at,
            models: session_info.models.clone(),
            modes: session_info.modes.clone(),
            config_options,
        };

        serde_json::to_value(result)
            .map_err(|e| JsonRpcErrorObject::internal_error(format!("Serialization error: {}", e)))
    }

    /// Handle session/list request
    ///
    /// Lists available sessions from the target connector.
    async fn handle_session_list(
        &self,
        request: &JsonRpcRequest,
        client_id: Option<&str>,
    ) -> Result<serde_json::Value, JsonRpcErrorObject> {
        // params are optional for session/list
        let params: SessionListParams = request
            .params
            .as_ref()
            .map(|v| serde_json::from_value(v.clone()))
            .transpose()
            .map_err(|e| {
                JsonRpcErrorObject::invalid_params(format!("Invalid session/list params: {}", e))
            })?
            .unwrap_or(SessionListParams {
                connector_id: None,
                cwd: None,
                cursor: None,
            });

        let client_id = client_id
            .map(|s| s.to_string())
            .unwrap_or_else(|| format!("http-client-{}", uuid::Uuid::now_v7()));

        // Resolve connector ID: explicit param → client's active session → default
        let connector_id = if let Some(ref cid) = params.connector_id {
            cid.clone()
        } else if let Some(connector) = self.resolve_connector_from_client_sessions(&client_id) {
            connector
        } else if let Some(default) = self.connector_ops.default_connector_id().await {
            default
        } else {
            return Err(JsonRpcErrorObject::invalid_params(
                "No connector_id specified and no default connector available",
            ));
        };

        // Gateways: list sessions from all connectors via archivist
        if self.is_gateway_connector(&connector_id).await {
            info!(
                "Connector {} is a gateway, listing sessions across all connectors (client: {})",
                connector_id, client_id
            );

            let all_sessions = self
                .connector_ops
                .list_all_sessions()
                .await
                .map_err(JsonRpcErrorObject::from)?;

            let entries: Vec<SessionListEntry> = all_sessions
                .into_iter()
                .map(|info| SessionListEntry {
                    session_id: info.session_id,
                    cwd: info.cwd.unwrap_or_else(|| ".".to_string()),
                    title: info.title,
                    updated_at: Some(info.created_at),
                    meta: Some(serde_json::json!({ "connectorId": info.connector_id })),
                })
                .collect();

            info!(
                "Listed {} sessions across all connectors for client {}",
                entries.len(),
                client_id
            );

            let result = SessionListResult {
                sessions: entries,
                next_cursor: None,
            };
            return serde_json::to_value(result).map_err(|e| {
                JsonRpcErrorObject::internal_error(format!("Serialization error: {}", e))
            });
        }

        info!(
            "Listing sessions for connector: {} (client: {})",
            connector_id, client_id
        );

        // List sessions via connector operations
        let sessions = self
            .connector_ops
            .list_sessions(&connector_id)
            .await
            .map_err(JsonRpcErrorObject::from)?;

        // Convert to ACP-spec SessionListEntry format
        // Note: We pass through the connector's raw session IDs here.
        // No session mappings are created — session/list is a read-only query.
        // Mappings are created later when session/load or session/resume is called.
        let entries: Vec<SessionListEntry> = sessions
            .into_iter()
            .map(|info| SessionListEntry {
                session_id: info.session_id,
                cwd: info.cwd.unwrap_or_else(|| ".".to_string()),
                title: info.title,
                updated_at: Some(info.created_at),
                meta: None,
            })
            .collect();

        info!(
            "Listed {} sessions from connector {} for client {}",
            entries.len(),
            connector_id,
            client_id
        );

        // No server-side pagination for now (connector already resolved all pages)
        let result = SessionListResult {
            sessions: entries,
            next_cursor: None,
        };

        serde_json::to_value(result)
            .map_err(|e| JsonRpcErrorObject::internal_error(format!("Serialization error: {}", e)))
    }

    /// Handle session/resume request
    ///
    /// Resumes an existing session without history replay.
    /// Structurally similar to session/load but semantically different
    /// (no session/update history notifications are sent).
    async fn handle_session_resume(
        &self,
        request: &JsonRpcRequest,
        client_id: Option<&str>,
    ) -> Result<serde_json::Value, JsonRpcErrorObject> {
        let params_value = request
            .params
            .as_ref()
            .ok_or_else(|| {
                JsonRpcErrorObject::invalid_params("Missing params for session/resume")
            })?
            .clone();

        let params: SessionResumeParams = serde_json::from_value(params_value).map_err(|e| {
            JsonRpcErrorObject::invalid_params(format!("Invalid session/resume params: {}", e))
        })?;

        let client_id = client_id
            .map(|s| s.to_string())
            .unwrap_or_else(|| format!("http-client-{}", uuid::Uuid::now_v7()));

        // Resolve connector ID: explicit param -> session mapping -> client sessions -> default
        let connector_id = if let Some(ref cid) = params.connector_id {
            cid.clone()
        } else if let Some(mapping) = self.session_manager.get_mapping(&params.session_id) {
            // Session was already mapped (e.g. from a prior session/list call)
            mapping.connector_id
        } else if let Some(connector) = self.resolve_connector_from_client_sessions(&client_id) {
            connector
        } else if let Some(default) = self.connector_ops.default_connector_id().await {
            default
        } else {
            return Err(JsonRpcErrorObject::invalid_params(
                "No connector_id specified and no default connector available",
            ));
        };

        // Resume uses load_session under the hood — the connector decides
        // whether to send session/resume or session/load to the upstream agent
        let mcp_servers_value = params.mcp_servers.as_ref().map(|servers| {
            serde_json::Value::Array(servers.clone())
        });
        let session_info = self
            .connector_ops
            .load_session(&connector_id, &params.session_id, params.cwd.clone(), mcp_servers_value)
            .await
            .map_err(JsonRpcErrorObject::from)?;

        // Create the session mapping
        let mapping = self.session_manager.create_mapping(
            &client_id,
            Some(params.session_id.clone()),
            connector_id.clone(),
        );

        self.session_manager.update_mapping_internal_session(
            &mapping.client_session_id,
            session_info.session_id.clone(),
        );

        info!(
            "Resumed session: {} -> {} from connector {}",
            mapping.client_session_id, mapping.internal_session_id, connector_id
        );

        // Build config_options from modes and models
        let config_options = {
            let mut options = Vec::new();
            if let Some(ref modes) = session_info.modes {
                options.push(modes_to_config_option(modes));
            }
            if let Some(ref models) = session_info.models {
                options.push(models_to_config_option(models));
            }
            if options.is_empty() {
                None
            } else {
                Some(options)
            }
        };

        let result = SessionResumeResult {
            session_id: mapping.client_session_id,
            title: session_info.title,
            connector_id: session_info.connector_id,
            created_at: session_info.created_at,
            models: session_info.models.clone(),
            modes: session_info.modes.clone(),
            config_options,
        };

        serde_json::to_value(result)
            .map_err(|e| JsonRpcErrorObject::internal_error(format!("Serialization error: {}", e)))
    }

    /// Handle session/prompt request
    ///
    /// Sends a prompt to a session and starts streaming the response.
    async fn handle_session_prompt(
        &self,
        request: &JsonRpcRequest,
        _client_id: Option<&str>,
    ) -> Result<serde_json::Value, JsonRpcErrorObject> {
        let params_value = request
            .params
            .as_ref()
            .ok_or_else(|| JsonRpcErrorObject::invalid_params("Missing params for session/prompt"))?
            .clone();

        let params: SessionPromptParams = serde_json::from_value(params_value).map_err(|e| {
            JsonRpcErrorObject::invalid_params(format!("Invalid session/prompt params: {}", e))
        })?;

        // Look up the session mapping
        let mapping = self
            .session_manager
            .get_mapping(&params.session_id)
            .ok_or_else(|| {
                JsonRpcErrorObject::new(
                    crate::error::error_codes::SESSION_NOT_FOUND,
                    format!("Session not found: {}", params.session_id),
                )
            })?;

        // Convert prompt content to text
        let text = params.prompt.to_text();

        info!(
            "Received prompt for session {}: {} chars",
            params.session_id,
            text.len()
        );

        // Send message via connector operations and wait for completion
        let stop_reason = self
            .connector_ops
            .send_message(&mapping.connector_id, &mapping.internal_session_id, text)
            .await
            .map_err(JsonRpcErrorObject::from)?;

        info!(
            "Prompt completed for session {} with stop_reason: {}",
            params.session_id, stop_reason
        );

        let result = SessionPromptResult { stop_reason };

        serde_json::to_value(&result)
            .map_err(|e| JsonRpcErrorObject::internal_error(format!("Serialization error: {}", e)))
    }

    /// Handle session/cancel request
    ///
    /// Cancels any active generation on the session.
    async fn handle_session_cancel(
        &self,
        request: &JsonRpcRequest,
        _client_id: Option<&str>,
    ) -> Result<serde_json::Value, JsonRpcErrorObject> {
        let params_value = request
            .params
            .as_ref()
            .ok_or_else(|| JsonRpcErrorObject::invalid_params("Missing params for session/cancel"))?
            .clone();

        let params: SessionCancelParams = serde_json::from_value(params_value).map_err(|e| {
            JsonRpcErrorObject::invalid_params(format!("Invalid session/cancel params: {}", e))
        })?;

        // Look up the session mapping
        let mapping = self
            .session_manager
            .get_mapping(&params.session_id)
            .ok_or_else(|| {
                JsonRpcErrorObject::new(
                    crate::error::error_codes::SESSION_NOT_FOUND,
                    format!("Session not found: {}", params.session_id),
                )
            })?;

        info!("Cancelling generation for session {}", params.session_id);

        // Cancel via connector operations
        self.connector_ops
            .cancel_generation(&mapping.connector_id, &mapping.internal_session_id)
            .await
            .map_err(JsonRpcErrorObject::from)?;

        let result = SessionCancelResult { cancelled: true };

        serde_json::to_value(result)
            .map_err(|e| JsonRpcErrorObject::internal_error(format!("Serialization error: {}", e)))
    }

    /// Handle session/close request
    ///
    /// Closes a session and removes the session mapping.
    async fn handle_session_close(
        &self,
        request: &JsonRpcRequest,
        _client_id: Option<&str>,
    ) -> Result<serde_json::Value, JsonRpcErrorObject> {
        let params_value = request
            .params
            .as_ref()
            .ok_or_else(|| JsonRpcErrorObject::invalid_params("Missing params for session/close"))?
            .clone();

        let params: SessionCloseParams = serde_json::from_value(params_value).map_err(|e| {
            JsonRpcErrorObject::invalid_params(format!("Invalid session/close params: {}", e))
        })?;

        // Remove the session mapping
        let removed = self.session_manager.remove_mapping(&params.session_id);

        if let Some(mapping) = removed {
            info!(
                "Closed session: {} (internal: {}, connector: {})",
                params.session_id, mapping.internal_session_id, mapping.connector_id
            );
        } else {
            warn!(
                "Session not found during close: {} (treating as success)",
                params.session_id
            );
        }

        let result = SessionCloseResult { closed: true };
        serde_json::to_value(result)
            .map_err(|e| JsonRpcErrorObject::internal_error(format!("Serialization error: {}", e)))
    }

    /// Handle session/set_mode request
    ///
    /// Sets the mode for a session. This is the legacy mode-setting method.
    async fn handle_session_set_mode(
        &self,
        request: &JsonRpcRequest,
        _client_id: Option<&str>,
    ) -> Result<serde_json::Value, JsonRpcErrorObject> {
        let params_value = request
            .params
            .as_ref()
            .ok_or_else(|| {
                JsonRpcErrorObject::invalid_params("Missing params for session/set_mode")
            })?
            .clone();

        let params: SessionSetModeParams = serde_json::from_value(params_value).map_err(|e| {
            JsonRpcErrorObject::invalid_params(format!("Invalid session/set_mode params: {}", e))
        })?;

        // Look up the session mapping
        let mapping = self
            .session_manager
            .get_mapping(&params.session_id)
            .ok_or_else(|| {
                JsonRpcErrorObject::new(
                    crate::error::error_codes::SESSION_NOT_FOUND,
                    format!("Session not found: {}", params.session_id),
                )
            })?;

        // Apply forward mapping for legacy mode (Gateway → Agent)
        let mapped_mode_id = self
            .apply_forward_mode_mapping(&mapping.connector_id, &params.mode_id)
            .await;

        info!(
            "Setting mode for session {} to mode_id: {} (mapped from: {})",
            params.session_id, mapped_mode_id, params.mode_id
        );

        // Forward to connector via connector operations
        self.connector_ops
            .set_session_mode(
                &mapping.connector_id,
                &mapping.internal_session_id,
                &mapped_mode_id,
            )
            .await
            .map_err(JsonRpcErrorObject::from)?;

        // Per ACP spec, return empty object on success
        Ok(serde_json::json!({}))
    }

    /// Handle session/set_model request
    ///
    /// Sets the model for a session. This is the legacy model-setting method.
    async fn handle_session_set_model(
        &self,
        request: &JsonRpcRequest,
        _client_id: Option<&str>,
    ) -> Result<serde_json::Value, JsonRpcErrorObject> {
        let params_value = request
            .params
            .as_ref()
            .ok_or_else(|| {
                JsonRpcErrorObject::invalid_params("Missing params for session/set_model")
            })?
            .clone();

        let params: SessionSetModelParams = serde_json::from_value(params_value).map_err(|e| {
            JsonRpcErrorObject::invalid_params(format!("Invalid session/set_model params: {}", e))
        })?;

        // Look up the session mapping
        let mapping = self
            .session_manager
            .get_mapping(&params.session_id)
            .ok_or_else(|| {
                JsonRpcErrorObject::new(
                    crate::error::error_codes::SESSION_NOT_FOUND,
                    format!("Session not found: {}", params.session_id),
                )
            })?;

        // Apply forward mapping for legacy model (Gateway → Agent)
        let mapped_model_id = self
            .apply_forward_model_mapping(&mapping.connector_id, &params.model_id)
            .await;

        info!(
            "Setting model for session {} to model_id: {} (mapped from: {})",
            params.session_id, mapped_model_id, params.model_id
        );

        // Forward to connector via connector operations
        self.connector_ops
            .set_session_model(
                &mapping.connector_id,
                &mapping.internal_session_id,
                &mapped_model_id,
            )
            .await
            .map_err(JsonRpcErrorObject::from)?;

        // Per ACP spec, return empty object on success
        Ok(serde_json::json!({}))
    }

    /// Handle session/set_config_option request
    ///
    /// Sets a configuration option for a session.
    async fn handle_session_set_config_option(
        &self,
        request: &JsonRpcRequest,
        _client_id: Option<&str>,
    ) -> Result<serde_json::Value, JsonRpcErrorObject> {
        let params_value = request
            .params
            .as_ref()
            .ok_or_else(|| {
                JsonRpcErrorObject::invalid_params("Missing params for session/set_config_option")
            })?
            .clone();

        let params: SessionSetConfigOptionParams =
            serde_json::from_value(params_value).map_err(|e| {
                JsonRpcErrorObject::invalid_params(format!(
                    "Invalid session/set_config_option params: {}",
                    e
                ))
            })?;

        // Look up the session mapping
        let mapping = self
            .session_manager
            .get_mapping(&params.session_id)
            .ok_or_else(|| {
                JsonRpcErrorObject::new(
                    crate::error::error_codes::SESSION_NOT_FOUND,
                    format!("Session not found: {}", params.session_id),
                )
            })?;

        info!(
            "Setting config option for session {}: {}={}",
            params.session_id, params.config_id, params.value
        );

        // Map config_id to appropriate handler
        match params.config_id.as_str() {
            "mode" => {
                self.connector_ops
                    .set_session_mode(
                        &mapping.connector_id,
                        &mapping.internal_session_id,
                        &params.value,
                    )
                    .await
                    .map_err(JsonRpcErrorObject::from)?;
            }
            "model" => {
                self.connector_ops
                    .set_session_model(
                        &mapping.connector_id,
                        &mapping.internal_session_id,
                        &params.value,
                    )
                    .await
                    .map_err(JsonRpcErrorObject::from)?;
            }
            other => {
                warn!(
                    "Unknown config option ID: {} - ignoring (session: {})",
                    other, params.session_id
                );
            }
        }

        // Per ACP spec, return empty object on success
        Ok(serde_json::json!({}))
    }

    // ========================================================================
    // Forward Mapping Helpers (Gateway → Agent)
    // ========================================================================

    /// Apply forward mapping for mode (Gateway → Agent)
    async fn apply_forward_mode_mapping(&self, connector_id: &str, gateway_mode: &str) -> String {
        // Get the agent type for this connector
        let agent_type = self
            .connector_ops
            .get_connector_agent_type(connector_id)
            .await
            .ok()
            .flatten();

        // Apply forward mapping based on agent type
        let mapped = match agent_type.as_deref() {
            Some("claude") => match gateway_mode {
                "ask" => "default",
                "plan" => "plan",
                "readonly" => "plan",
                "write" => "acceptEdits",
                "yolo" => "bypassPermissions",
                _ => gateway_mode,
            },
            Some("codex") | Some("gemini") => gateway_mode,
            _ => gateway_mode,
        };

        if mapped != gateway_mode {
            tracing::debug!(
                connector_id = %connector_id,
                gateway_mode = %gateway_mode,
                mapped_mode = %mapped,
                "Forward mode mapping applied"
            );
        }

        mapped.to_string()
    }

    /// Apply forward mapping for model (Gateway → Agent)
    async fn apply_forward_model_mapping(&self, connector_id: &str, gateway_model: &str) -> String {
        // Get the agent type for this connector
        let agent_type = self
            .connector_ops
            .get_connector_agent_type(connector_id)
            .await
            .ok()
            .flatten();

        // Apply forward mapping based on agent type
        let mapped = match agent_type.as_deref() {
            Some("claude") => match gateway_model {
                "simple" => "haiku",
                "dailydriver" => "sonnet",
                "high" => "opus",
                _ => gateway_model,
            },
            Some("codex") | Some("gemini") => gateway_model,
            _ => gateway_model,
        };

        if mapped != gateway_model {
            tracing::debug!(
                connector_id = %connector_id,
                gateway_model = %gateway_model,
                mapped_model = %mapped,
                "Forward model mapping applied"
            );
        }

        mapped.to_string()
    }

    /// Check if a connector is a gateway type
    ///
    /// Gateway connectors don't have real sessions — they act as routers
    /// that transfer clients to actual agent connectors. When a session/list
    /// Resolve the connector ID from the client's existing session mappings.
    /// Returns the connector from the most recent session, if any.
    fn resolve_connector_from_client_sessions(&self, client_id: &str) -> Option<String> {
        let session_ids = self.session_manager.list_client_sessions(client_id);
        // Take the last session (most recently added) and look up its connector
        session_ids.last().and_then(|sid| {
            self.session_manager
                .get_mapping(sid)
                .map(|m| m.connector_id)
        })
    }

    /// request targets a gateway, the RPC handler should delay the response
    /// until a transfer occurs.
    async fn is_gateway_connector(&self, connector_id: &str) -> bool {
        match self.connector_ops.list_connectors().await {
            Ok(connectors) => connectors
                .iter()
                .any(|c| c.id == connector_id && c.connector_type.to_lowercase() == "gateway"),
            Err(_) => false,
        }
    }
}

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

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

    fn create_test_handler() -> RpcHandler<NoOpConnectorOperations> {
        RpcHandler::new(
            SessionManager::new(),
            NoOpConnectorOperations,
            crate::sse::SseNotifier::new(),
            std::sync::Arc::new(crate::agent_requests::AgentRequestTracker::new()),
        )
    }

    #[tokio::test]
    async fn test_handle_initialize() {
        let handler = create_test_handler();

        let request = json!({
            "jsonrpc": "2.0",
            "method": "initialize",
            "params": {
                "client_name": "test-client",
                "client_version": "1.0.0"
            },
            "id": 1
        });

        let response = handler.handle_request(&request.to_string(), None).await;

        match response {
            JsonRpcResponseBatch::Single(resp) => {
                assert!(resp.is_success());
                let result = resp.result.unwrap();
                assert_eq!(result["agentInfo"]["name"], SERVER_NAME);
                assert_eq!(result["protocolVersion"], ACP_PROTOCOL_VERSION);
                assert!(!result["clientId"].as_str().unwrap().is_empty());
            }
            _ => panic!("Expected single response"),
        }
    }

    #[tokio::test]
    async fn test_handle_session_new() {
        let handler = create_test_handler();

        // First initialize to get a client ID
        let init_request = json!({
            "jsonrpc": "2.0",
            "method": "initialize",
            "id": 1
        });

        let init_response = handler
            .handle_request(&init_request.to_string(), None)
            .await;
        let client_id = match init_response {
            JsonRpcResponseBatch::Single(resp) => resp.result.unwrap()["clientId"]
                .as_str()
                .unwrap()
                .to_string(),
            _ => panic!("Expected single response"),
        };

        // Now create a session
        let request = json!({
            "jsonrpc": "2.0",
            "method": "session/new",
            "params": {
                "connector_id": "test-connector"
            },
            "id": 2
        });

        let response = handler
            .handle_request(&request.to_string(), Some(&client_id))
            .await;

        match response {
            JsonRpcResponseBatch::Single(resp) => {
                assert!(resp.is_success());
                let result = resp.result.unwrap();
                assert!(!result["sessionId"].as_str().unwrap().is_empty());
            }
            _ => panic!("Expected single response"),
        }
    }

    #[tokio::test]
    async fn test_handle_session_close() {
        let handler = create_test_handler();

        // Initialize
        let init_request = json!({
            "jsonrpc": "2.0",
            "method": "initialize",
            "id": 1
        });
        let init_response = handler
            .handle_request(&init_request.to_string(), None)
            .await;
        let client_id = match init_response {
            JsonRpcResponseBatch::Single(resp) => resp.result.unwrap()["clientId"]
                .as_str()
                .unwrap()
                .to_string(),
            _ => panic!("Expected single response"),
        };

        // Create session
        let new_request = json!({
            "jsonrpc": "2.0",
            "method": "session/new",
            "params": {},
            "id": 2
        });
        let new_response = handler
            .handle_request(&new_request.to_string(), Some(&client_id))
            .await;
        let session_id = match new_response {
            JsonRpcResponseBatch::Single(resp) => resp.result.unwrap()["sessionId"]
                .as_str()
                .unwrap()
                .to_string(),
            _ => panic!("Expected single response"),
        };

        // Close session
        let close_request = json!({
            "jsonrpc": "2.0",
            "method": "session/close",
            "params": {
                "session_id": session_id
            },
            "id": 3
        });

        let response = handler
            .handle_request(&close_request.to_string(), Some(&client_id))
            .await;

        match response {
            JsonRpcResponseBatch::Single(resp) => {
                assert!(resp.is_success());
                let result = resp.result.unwrap();
                assert_eq!(result["closed"], true);
            }
            _ => panic!("Expected single response"),
        }
    }

    #[tokio::test]
    async fn test_handle_batch_request() {
        let handler = create_test_handler();

        let batch = json!([
            {
                "jsonrpc": "2.0",
                "method": "initialize",
                "id": 1
            },
            {
                "jsonrpc": "2.0",
                "method": "initialize",
                "id": 2
            }
        ]);

        let response = handler.handle_request(&batch.to_string(), None).await;

        match response {
            JsonRpcResponseBatch::Batch(responses) => {
                assert_eq!(responses.len(), 2);
                for resp in responses {
                    assert!(resp.is_success());
                }
            }
            _ => panic!("Expected batch response"),
        }
    }

    #[tokio::test]
    async fn test_handle_unknown_method() {
        let handler = create_test_handler();

        let request = json!({
            "jsonrpc": "2.0",
            "method": "unknown/method",
            "id": 1
        });

        let response = handler.handle_request(&request.to_string(), None).await;

        match response {
            JsonRpcResponseBatch::Single(resp) => {
                assert!(!resp.is_success());
                let error = resp.error.unwrap();
                assert_eq!(error.code, -32601); // Method not found
            }
            _ => panic!("Expected single response"),
        }
    }

    #[tokio::test]
    async fn test_handle_invalid_json() {
        let handler = create_test_handler();

        let response = handler.handle_request("invalid json", None).await;

        match response {
            JsonRpcResponseBatch::Single(resp) => {
                assert!(!resp.is_success());
                let error = resp.error.unwrap();
                assert_eq!(error.code, -32700); // Parse error
            }
            _ => panic!("Expected single response"),
        }
    }

    #[tokio::test]
    async fn test_handle_empty_batch() {
        let handler = create_test_handler();

        let response = handler.handle_request("[]", None).await;

        match response {
            JsonRpcResponseBatch::Single(resp) => {
                assert!(!resp.is_success());
                let error = resp.error.unwrap();
                assert_eq!(error.code, -32600); // Invalid request
            }
            _ => panic!("Expected single response"),
        }
    }

    #[test]
    fn test_prompt_content_to_text() {
        // Test simple text
        let text_content = PromptContent::Text("Hello world".to_string());
        assert_eq!(text_content.to_text(), "Hello world");

        // Test blocks with text
        let blocks_content = PromptContent::Blocks(vec![
            ContentBlock::Text {
                text: "First line".to_string(),
            },
            ContentBlock::Text {
                text: "Second line".to_string(),
            },
        ]);
        assert_eq!(blocks_content.to_text(), "First line\nSecond line");

        // Test blocks with mixed content (images filtered out)
        let mixed_content = PromptContent::Blocks(vec![
            ContentBlock::Text {
                text: "Text only".to_string(),
            },
            ContentBlock::Image {
                data: "base64data".to_string(),
                media_type: "image/png".to_string(),
            },
        ]);
        assert_eq!(mixed_content.to_text(), "Text only");
    }

    #[test]
    fn test_session_info_camel_case() {
        let info = SessionInfo {
            session_id: "test-123".to_string(),
            title: Some("Test Session".to_string()),
            connector_id: "conn-1".to_string(),
            cwd: None,
            created_at: "2024-01-01T00:00:00Z".to_string(),
            models: None,
            modes: None,
        };

        let json = serde_json::to_value(&info).unwrap();
        assert!(json.get("sessionId").is_some());
        assert!(json.get("connectorId").is_some());
        assert!(json.get("createdAt").is_some());
    }

    #[test]
    fn test_connector_info_camel_case() {
        let info = ConnectorInfo {
            id: "conn-1".to_string(),
            name: "Test Connector".to_string(),
            connector_type: "test".to_string(),
            available: true,
        };

        let json = serde_json::to_value(&info).unwrap();
        assert!(json.get("connectorType").is_some());
    }

    #[test]
    fn test_initialize_result_camel_case() {
        let result = InitializeResult {
            protocol_version: 1,
            agent_capabilities: AgentCapabilities::default(),
            agent_info: AgentInfo {
                name: "test".to_string(),
                title: "Test".to_string(),
                version: "1.0.0".to_string(),
            },
            auth_methods: vec![],
            client_id: "client-1".to_string(),
        };

        let json = serde_json::to_value(&result).unwrap();
        assert!(json.get("protocolVersion").is_some());
        assert!(json.get("agentCapabilities").is_some());
        assert!(json.get("agentInfo").is_some());
        assert!(json.get("authMethods").is_some());
        assert!(json.get("clientId").is_some());
    }
}
