//! ACP Connector implementation
//!
//! This module provides the main connector implementation for the Agent-Client Protocol.
//! It integrates the transport and protocol layers with the Connector trait to provide
//! a complete, production-ready ACP client.
//!
//! # Architecture
//!
//! The AcpConnector follows the same pattern as OpenCodeConnector:
//! - Background task for event processing and reconnection
//! - Command channel for control operations
//! - Event broadcast for publishing events to subscribers
//! - State machine for lifecycle management
//! - Automatic reconnection with exponential backoff
//!
//! # Lifecycle
//!
//! 1. Create connector with `AcpConnector::new(id, owner, config)`
//! 2. Start the connector task with `start_task()` which returns a JoinHandle
//! 3. Send commands via the command channel
//! 4. Receive events via the event broadcast channel
//! 5. Stop with `Connector::stop()` which sends Shutdown command

use crate::connectors::acp::{
    config::{AcpConfig, TransportKind},
    error::{AcpError, AcpResult},
    idle_detector::{check_idle_sessions, PendingCompletion, SessionState},
    protocol::{
        self, build_initialize_request, build_session_cancel_request, build_session_new_request,
        build_session_prompt_request, ProtocolHandler,
    },
    state::{InternalState, SessionInfo, SessionStatus},
    title_utils::derive_title_from_text,
    transport::{AcpTransport, HttpTransport, StdioTransport},
};
use crate::connectors::{Connector, ConnectorCommand};
use crate::sharing::bus::SharingBus;
use crate::types::{ConnectorId, ConnectorKind, ConnectorState, UserId};
use chrono::Utc;
use dirigent_protocol::types::meta::Meta;
use dirigent_protocol::types::{ContentBlock, ToolOrigin};
use dirigent_protocol::{
    Event, Message, MessagePart, MessageRole, MessageStatus, Session, SessionMetadata,
    SessionOwnership, SessionUpdate, TurnCompleteTrigger,
};
use serde_json::{json, Value};
use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{broadcast, mpsc, Mutex, RwLock};
use tokio::task::JoinHandle;
use tracing::{debug, error, info, trace, warn};
use uuid::Uuid;

/// Debug log to file macro
/// Writes to acp_connector_debug.log in the logs directory (no console output).
/// For deep protocol debugging. Use tracing for normal logging.
macro_rules! debug_log {
    ($($arg:tt)*) => {{
        let msg = format!("[{}] {}\n", chrono::Utc::now().format("%H:%M:%S%.3f"), format!($($arg)*));
        let log_path = dirigent_config::DirigentPaths::resolve()
            .map(|p| {
                let dir = p.logs_dir();
                let _ = std::fs::create_dir_all(&dir);
                dir.join("acp_connector_debug.log")
            })
            .unwrap_or_else(|_| std::path::PathBuf::from("acp_connector_debug.log"));
        if let Ok(mut file) = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&log_path)
        {
            let _ = std::io::Write::write_all(&mut file, msg.as_bytes());
        }
        // No console output - file-only logging for deep protocol debugging
    }};
}

/// Register or update a session node in the inspector registry.
///
/// Creates the node if it doesn't exist, or updates its properties and state.
/// Called after `upsert_session` and on session metadata changes.
#[cfg(feature = "server")]
async fn inspector_upsert_session(
    inspector: &Arc<dirigent_inspector::InspectorRegistry>,
    connector_id: &str,
    session: &SessionInfo,
) {
    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 node_state = match session.status {
        SessionStatus::Active => dirigent_inspector::NodeState::Running,
        SessionStatus::Processing => dirigent_inspector::NodeState::Busy("Generating".to_string()),
        SessionStatus::Idle => dirigent_inspector::NodeState::Idle,
        SessionStatus::Ended => dirigent_inspector::NodeState::Stopped,
    };

    let label = session.title.as_deref().unwrap_or(&session.id);

    // Try to register; if already exists, update instead
    let meta =
        dirigent_inspector::NodeMetadata::new(dirigent_inspector::NodeKind::AsyncTask, label)
            .with_state(node_state.clone())
            .with_property("session_id", serde_json::json!(&session.id))
            .with_property("status", serde_json::json!(format!("{:?}", session.status)));

    let meta = if let Some(ref model) = session.model {
        meta.with_property("model", serde_json::json!(model))
    } else {
        meta
    };

    let meta = if let Some(ref title) = session.title {
        meta.with_property("title", serde_json::json!(title))
    } else {
        meta
    };

    let meta = if session.message_count > 0 {
        meta.with_property("message_count", serde_json::json!(session.message_count))
    } else {
        meta
    };

    match inspector
        .register(sess_node_id.clone(), &parent_id, meta, None)
        .await
    {
        Ok(mut handle) => {
            handle.detach();
            trace!(connector_id = %connector_id, session_id = %session.id, "Registered session with inspector");
        }
        Err(_) => {
            // Already registered — update state and properties
            let _ = inspector.update_state(&sess_node_id, node_state).await;
            let mut props = HashMap::new();
            props.insert(
                "status".to_string(),
                serde_json::json!(format!("{:?}", session.status)),
            );
            if let Some(ref title) = session.title {
                props.insert("title".to_string(), serde_json::json!(title));
            }
            if let Some(ref model) = session.model {
                props.insert("model".to_string(), serde_json::json!(model));
            }
            if session.message_count > 0 {
                props.insert(
                    "message_count".to_string(),
                    serde_json::json!(session.message_count),
                );
            }
            let _ = inspector.update_properties(&sess_node_id, props).await;
            trace!(connector_id = %connector_id, session_id = %session.id, "Updated session in inspector");
        }
    }
}

/// Update only the inspector state for a session (lightweight, no property changes).
#[cfg(feature = "server")]
async fn inspector_update_session_state(
    inspector: &Arc<dirigent_inspector::InspectorRegistry>,
    connector_id: &str,
    session_id: &str,
    state: dirigent_inspector::NodeState,
) {
    let sess_node_id = dirigent_inspector::NodeId::new(format!(
        "dirigent/connectors/{}/sessions/{}",
        connector_id, session_id
    ));
    let _ = inspector.update_state(&sess_node_id, state).await;
}

/// Deregister all session nodes for a connector from the inspector.
#[cfg(feature = "server")]
async fn inspector_deregister_all_sessions(
    inspector: &Arc<dirigent_inspector::InspectorRegistry>,
    connector_id: &str,
    internal_state: &InternalState,
) {
    let sessions = internal_state.list_sessions().await;
    for session in sessions {
        let sess_node_id = dirigent_inspector::NodeId::new(format!(
            "dirigent/connectors/{}/sessions/{}",
            connector_id, session.id
        ));
        let _ = inspector.deregister_subtree(&sess_node_id).await;
    }
}

/// Helper that publishes an event to both the per-connector broadcast
/// channel and the global `SharingBus`. The bus publish goes first so any
/// subscriber reading from both sees the bus event no later than the
/// broadcast event.
async fn emit_event(
    sharing_bus: &Arc<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);
}

/// ACP connector implementation
///
/// Provides integration with ACP-compliant agents via stdio or HTTP transport.
/// Implements the `Connector` trait for use with CoreRuntime.
pub struct AcpConnector {
    /// 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,

    /// Human-readable title for this connector
    ///
    /// Used in UI display and logging. This is stored separately from
    /// AcpConfig as it's an orchestration-level field.
    title: String,

    /// Connector configuration
    config: AcpConfig,

    /// Tool configuration inherited from the connector config.
    ///
    /// When present, tool calls are checked against these directives
    /// before being emitted as events. Denied tools are marked as errors;
    /// hidden tools are silently dropped.
    tool_configuration: Option<crate::tools::ToolConfiguration>,

    /// Shared connector state (Initializing, Connecting, Ready, Error, Stopped)
    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.
    ///
    /// Every event emitted by this connector is published here in addition
    /// to `events_tx`, eliminating the forwarder task tier in the runtime.
    sharing_bus: Arc<SharingBus>,

    /// Internal state (protocol version, capabilities, sessions)
    internal_state: Arc<InternalState>,

    /// Pending agent requests awaiting external responses
    ///
    /// Tracks request_ids that are waiting for user response (e.g., permission prompts).
    /// No timeout - the system waits indefinitely for the user to respond.
    /// Used to validate that incoming AgentResponse commands match a real pending request.
    pending_agent_requests: Arc<Mutex<HashSet<String>>>,

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

    /// Optional process group manager for lifecycle management of stdio processes.
    ///
    /// When set, each StdioTransport created by this connector receives a
    /// per-process `ProcessLifecycle` handle so that spawned agents are
    /// tracked in the platform job object / process group, and are shut down
    /// gracefully on close.
    #[cfg(feature = "server")]
    process_manager: Option<Arc<dyn dirigent_process::ProcessGroupManager>>,
}

impl AcpConnector {
    /// Create a new ACP connector
    ///
    /// Initializes the connector with the given configuration but does not
    /// start it. Call `start_task()` to begin the connector's background task.
    ///
    /// # Arguments
    ///
    /// * `id` - Unique identifier for this connector
    /// * `owner` - User ID of the connector owner
    /// * `title` - Human-readable title for this connector
    /// * `config` - ACP configuration (transport, protocol version, etc.)
    ///
    /// # Returns
    ///
    /// A new AcpConnector in Initializing state, or an error if configuration is invalid.
    pub fn new(
        id: ConnectorId,
        owner: UserId,
        title: String,
        config: AcpConfig,
        sharing_bus: Arc<SharingBus>,
    ) -> AcpResult<Self> {
        // Validate configuration
        config.validate().map_err(AcpError::config)?;

        // Create command channel (capacity 100 for buffering commands)
        let (cmd_tx, cmd_rx) = mpsc::channel(100);

        // Create event broadcast channel (capacity 1000 for event buffering)
        let (events_tx, _) = broadcast::channel(1000);

        // Create internal state
        let internal_state = Arc::new(InternalState::new());

        info!(
            connector_id = %id,
            owner = %owner,
            title = %title,
            transport = ?config.transport,
            "Creating ACP connector"
        );

        Ok(Self {
            id,
            connector_uid: None,
            owner,
            title,
            config,
            tool_configuration: None,
            state: Arc::new(RwLock::new(ConnectorState::Initializing)),
            cmd_tx,
            cmd_rx: Arc::new(RwLock::new(Some(cmd_rx))),
            events_tx,
            sharing_bus,
            internal_state,
            pending_agent_requests: Arc::new(Mutex::new(HashSet::new())),
            #[cfg(feature = "server")]
            inspector: None,
            #[cfg(feature = "server")]
            process_manager: 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 tool configuration for this connector.
    ///
    /// When set, tool calls are checked against these directives before
    /// being emitted as events. This configuration is also cloned into
    /// each session's metadata at creation time.
    pub fn with_tool_configuration(
        mut self,
        tool_config: Option<crate::tools::ToolConfiguration>,
    ) -> Self {
        self.tool_configuration = tool_config;
        self
    }

    /// Set the inspector registry for PID tracking of stdio processes.
    #[cfg(feature = "server")]
    pub fn with_inspector(
        mut self,
        inspector: Option<Arc<dirigent_inspector::InspectorRegistry>>,
    ) -> Self {
        self.inspector = inspector;
        self
    }

    /// Set the process group manager for lifecycle management of stdio processes.
    ///
    /// When set, each `StdioTransport` spawned by this connector will receive a
    /// per-process `ProcessLifecycle` handle so agents are tracked in the
    /// platform job object / process group and are shut down gracefully.
    #[cfg(feature = "server")]
    pub fn with_process_manager(
        mut self,
        process_manager: Option<Arc<dyn dirigent_process::ProcessGroupManager>>,
    ) -> Self {
        self.process_manager = process_manager;
        self
    }

    /// 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
    ///
    /// This spawns an async task that:
    /// - Establishes transport connection
    /// - Performs ACP initialization handshake
    /// - Processes incoming notifications
    /// - Handles commands from the command channel
    /// - Manages automatic reconnection on failures
    ///
    /// # Returns
    ///
    /// A JoinHandle for the background task
    ///
    /// # Panics
    ///
    /// Panics if called more than once (the command receiver can only be taken once).
    pub async fn start_task(&self) -> JoinHandle<()> {
        let id = self.id.clone();
        let connector_uid = self.connector_uid;
        let config = self.config.clone();
        let tool_configuration = self.tool_configuration.clone();
        let state = Arc::clone(&self.state);
        let events_tx = self.events_tx.clone();
        let sharing_bus = Arc::clone(&self.sharing_bus);
        let internal_state = Arc::clone(&self.internal_state);
        let pending_agent_requests = Arc::clone(&self.pending_agent_requests);
        #[cfg(feature = "server")]
        let inspector = self.inspector.clone();
        #[cfg(feature = "server")]
        let process_manager = self.process_manager.clone();

        // Create session states map for idle detection
        let session_states = Arc::new(Mutex::new(HashMap::<String, SessionState>::new()));

        // Take the command receiver (this can only be done once)
        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 ACP connector task");

        tokio::spawn(async move {
            Self::run_task(
                id,
                connector_uid,
                config,
                tool_configuration,
                state,
                events_tx,
                sharing_bus,
                internal_state,
                pending_agent_requests,
                session_states,
                cmd_rx,
                #[cfg(feature = "server")]
                inspector,
                #[cfg(feature = "server")]
                process_manager,
            )
            .await;
        })
    }

    /// Main connector task loop
    ///
    /// This is the core of the connector's async behavior. It manages:
    /// - Transport connection and reconnection
    /// - ACP initialization handshake
    /// - Notification processing
    /// - Command handling
    /// - Error handling and recovery
    async fn run_task(
        id: ConnectorId,
        connector_uid: Option<Uuid>,
        config: AcpConfig,
        tool_configuration: Option<crate::tools::ToolConfiguration>,
        state: Arc<RwLock<ConnectorState>>,
        events_tx: broadcast::Sender<Event>,
        sharing_bus: Arc<SharingBus>,
        internal_state: Arc<InternalState>,
        pending_agent_requests: Arc<Mutex<HashSet<String>>>,
        session_states: Arc<Mutex<HashMap<String, SessionState>>>,
        mut cmd_rx: mpsc::Receiver<ConnectorCommand>,
        #[cfg(feature = "server")] inspector: Option<Arc<dirigent_inspector::InspectorRegistry>>,
        #[cfg(feature = "server")] process_manager: Option<Arc<dyn dirigent_process::ProcessGroupManager>>,
    ) {
        debug_log!("🚀 ACP connector {} task started", id);
        info!(connector_id = %id, "ACP connector task started");

        // Reconnection state
        let mut retry_count = 0;
        let mut last_init_success: Option<tokio::time::Instant> = None;
        let max_retries = config.retry.max_retries;
        let retry_delays: Vec<Duration> = config
            .retry
            .retry_delays_ms
            .iter()
            .map(|ms| Duration::from_millis(*ms))
            .collect();

        // Main loop: connect, process events, handle reconnects
        'reconnect_loop: loop {
            // Update state to Connecting
            {
                let mut state_guard = state.write().await;
                *state_guard = ConnectorState::Connecting;
            }
            emit_event(
                &sharing_bus,
                &events_tx,
                &id,
                connector_uid,
                Event::ConnectorStateChanged {
                connector_id: id.clone(),
                state: "Connecting".to_string(),
                error_kind: None,
            },
            )
            .await;
            debug!(connector_id = %id, "Attempting to connect to ACP agent");

            // Create transport based on config
            let transport_result = Self::create_transport(
                &config,
                #[cfg(feature = "server")]
                process_manager.as_ref(),
            )
            .await;

            let mut transport = match transport_result {
                Ok(t) => {
                    info!(connector_id = %id, "Successfully connected to ACP agent");
                    if let Some(init_time) = last_init_success {
                        if init_time.elapsed() > Duration::from_secs(30) {
                            debug!(connector_id = %id, "Previous connection was stable, resetting retry count");
                            retry_count = 0;
                        }
                    }
                    t
                }
                Err(e) => {
                    error!(
                        connector_id = %id,
                        error = %e,
                        "Failed to connect to ACP agent"
                    );

                    // Update state to Error
                    {
                        let mut state_guard = state.write().await;
                        *state_guard = ConnectorState::Error(format!("Connection failed: {}", e));
                    }
                    emit_event(
                        &sharing_bus,
                        &events_tx,
                        &id,
                        connector_uid,
                        Event::ConnectorStateChanged {
                        connector_id: id.clone(),
                        state: format!("Error(Connection failed: {})", e),
                        error_kind: Some("connection_failed".to_string()),
                    },
                    )
                    .await;
                    // Emit error event
                    emit_event(
                        &sharing_bus,
                        &events_tx,
                        &id,
                        connector_uid,
                        Event::Error {
                        message: format!("Failed to connect: {}", e),
                    },
                    )
                    .await;
                    // Handle retry logic
                    if retry_count < max_retries {
                        let delay = retry_delays[retry_count.min(retry_delays.len() - 1)];
                        warn!(
                            connector_id = %id,
                            retry_count = retry_count + 1,
                            max_retries = max_retries,
                            delay_secs = delay.as_secs(),
                            "Retrying connection"
                        );

                        retry_count += 1;
                        tokio::time::sleep(delay).await;
                        continue 'reconnect_loop;
                    } else {
                        error!(
                            connector_id = %id,
                            "Max retries exceeded, staying in Error state until manual Reconnect"
                        );

                        // Stay in Error state and wait for commands
                        loop {
                            match cmd_rx.recv().await {
                                Some(ConnectorCommand::Reconnect) => {
                                    info!(connector_id = %id, "Received Reconnect command, resetting retry count");
                                    retry_count = 0;
                                    internal_state.clear().await;
                                    pending_agent_requests.lock().await.clear();
                                    continue 'reconnect_loop;
                                }
                                Some(ConnectorCommand::Shutdown) => {
                                    info!(connector_id = %id, "Received Shutdown command");
                                    break 'reconnect_loop;
                                }
                                Some(cmd) => {
                                    warn!(
                                        connector_id = %id,
                                        command = ?cmd,
                                        "Received command while in Error state, ignoring"
                                    );
                                }
                                None => {
                                    error!(connector_id = %id, "Command channel closed");
                                    break 'reconnect_loop;
                                }
                            }
                        }
                    }
                }
            };

            // Perform ACP initialization
            let protocol_handler = ProtocolHandler::new();

            // Create ACP adapter for notification translation
            let adapter = dirigent_protocol::adapters::AcpAdapter::new();

            match Self::initialize_acp(
                &id,
                &mut transport,
                &protocol_handler,
                &config,
                &internal_state,
            )
            .await
            {
                Ok(()) => {
                    info!(connector_id = %id, "ACP initialization successful");
                    retry_count = 0;
                    last_init_success = Some(tokio::time::Instant::now());

                    // Update state to Ready
                    {
                        let mut state_guard = state.write().await;
                        *state_guard = ConnectorState::Ready;
                    }
                    emit_event(
                        &sharing_bus,
                        &events_tx,
                        &id,
                        connector_uid,
                        Event::ConnectorStateChanged {
                        connector_id: id.clone(),
                        state: "Ready".to_string(),
                        error_kind: None,
                    },
                    )
                    .await;
                    // Emit Connected event
                    emit_event(&sharing_bus, &events_tx, &id, connector_uid, Event::Connected).await;
                    // Register stdio process PID with inspector (if available)
                    #[cfg(feature = "server")]
                    if let Some(ref inspector) = inspector {
                        if let Some(pid) = transport.pid().await {
                            let process_node_id = dirigent_inspector::NodeId::new(format!(
                                "dirigent/connectors/{}/process",
                                id
                            ));
                            let parent_node_id = dirigent_inspector::NodeId::new(format!(
                                "dirigent/connectors/{}",
                                id
                            ));
                            let meta = dirigent_inspector::NodeMetadata::new(
                                dirigent_inspector::NodeKind::Process,
                                "stdio-process",
                            )
                            .with_state(dirigent_inspector::NodeState::Running)
                            .with_property("pid", serde_json::json!(pid))
                            .with_property("transport", serde_json::json!("stdio"));
                            if let Ok(mut handle) = inspector
                                .register(process_node_id, &parent_node_id, meta, None)
                                .await
                            {
                                handle.detach();
                                info!(connector_id = %id, pid = pid, "Registered stdio process with inspector");
                            }
                        }
                    }

                    // Auto-discover sessions if agent supports session/list
                    if internal_state.agent_supports_list_sessions().await {
                        info!(connector_id = %id, "Agent supports session/list — auto-discovering sessions");
                        Self::execute_list_sessions(
                            &mut transport,
                            &protocol_handler,
                            &events_tx, &sharing_bus, connector_uid,
                            &id,
                            &config,
                            &internal_state,
                            &pending_agent_requests,
                            &mut cmd_rx,
                        ).await;
                    } else {
                        debug!(connector_id = %id, "Agent does not support session/list — skipping auto-discovery");
                    }
                }
                Err(e) => {
                    error!(
                        connector_id = %id,
                        error = %e,
                        "ACP initialization failed"
                    );

                    // Close transport
                    let _ = transport.close().await;

                    // Update state to Error
                    {
                        let mut state_guard = state.write().await;
                        *state_guard =
                            ConnectorState::Error(format!("Initialization failed: {}", e));
                    }
                    emit_event(
                        &sharing_bus,
                        &events_tx,
                        &id,
                        connector_uid,
                        Event::ConnectorStateChanged {
                        connector_id: id.clone(),
                        state: format!("Error(Initialization failed: {})", e),
                        error_kind: Some("initialization_failed".to_string()),
                    },
                    )
                    .await;
                    // Emit error event
                    emit_event(
                        &sharing_bus,
                        &events_tx,
                        &id,
                        connector_uid,
                        Event::Error {
                        message: format!("Initialization failed: {}", e),
                    },
                    )
                    .await;
                    // Retry if retriable
                    if e.is_retriable() && retry_count < max_retries {
                        let delay = retry_delays[retry_count.min(retry_delays.len() - 1)];
                        warn!(
                            connector_id = %id,
                            retry_count = retry_count + 1,
                            "Retrying after initialization failure"
                        );

                        retry_count += 1;
                        tokio::time::sleep(delay).await;
                        continue 'reconnect_loop;
                    } else {
                        // Enter Error state
                        loop {
                            match cmd_rx.recv().await {
                                Some(ConnectorCommand::Reconnect) => {
                                    info!(connector_id = %id, "Received Reconnect command");
                                    retry_count = 0;
                                    internal_state.clear().await;
                                    pending_agent_requests.lock().await.clear();
                                    continue 'reconnect_loop;
                                }
                                Some(ConnectorCommand::Shutdown) => {
                                    info!(connector_id = %id, "Received Shutdown command");
                                    break 'reconnect_loop;
                                }
                                Some(cmd) => {
                                    warn!(connector_id = %id, command = ?cmd, "Ignoring command in Error state");
                                }
                                None => {
                                    error!(connector_id = %id, "Command channel closed");
                                    break 'reconnect_loop;
                                }
                            }
                        }
                    }
                }
            }

            // Event processing loop
            // Get notification receiver from protocol handler
            let mut notifications = protocol_handler
                .take_notification_receiver()
                .await
                .expect("Notification receiver should be available");

            // Idle check interval for housekeeping (50ms interval)
            let mut idle_check_interval = tokio::time::interval(Duration::from_millis(50));
            // Don't let missed ticks accumulate
            idle_check_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

            'event_loop: loop {
                tokio::select! {
                    biased;  // Enforce priority: transport > notifications > commands > housekeeping

                    // PRIORITY 1: Handle incoming notifications from transport
                    message = transport.recv() => {
                        match message {
                            Ok(Some(msg)) => {
                                let masked_msg = dirigent_protocol::log_utils::format_for_log(&msg);
                                info!(connector_id = %id, message = %masked_msg, "📬 Received message from transport in event loop");

                                // Route message through protocol handler
                                use crate::connectors::acp::protocol::MessageHandlerResult;
                                match protocol_handler.handle_message(msg).await {
                                    MessageHandlerResult::None => {
                                        // No response needed (notification processed)
                                    }
                                    MessageHandlerResult::Response(response) => {
                                        // Send immediate response back to agent
                                        info!(connector_id = %id, "📤 Sending response to agent request");
                                        if let Err(e) = transport.send(response).await {
                                            error!(connector_id = %id, error = %e, "Failed to send response to agent");
                                        }
                                    }
                                    MessageHandlerResult::AgentRequest { request_id, method, params } => {
                                        // Agent is requesting something from the client (e.g., permission)
                                        info!(
                                            connector_id = %id,
                                            method = %method,
                                            request_id = %request_id,
                                            "🔔 Agent request received - emitting Event::AgentRequest"
                                        );

                                        // Extract session_id from params (standard ACP location)
                                        let session_id = params
                                            .get("sessionId")
                                            .and_then(|v| v.as_str())
                                            .unwrap_or("unknown")
                                            .to_string();

                                        // Determine if this is a forwarded (external) session
                                        // Check session ownership to decide routing
                                        let is_forwarded = if let Some(session_info) = internal_state.get_session(&session_id).await {
                                            // Session found - check if it's external (owned by ACP client)
                                            session_info.ownership.is_external()
                                        } else {
                                            // Session not found - default to false (internal)
                                            // This shouldn't happen in normal operation, but we default to showing UI modal
                                            warn!(
                                                connector_id = %id,
                                                session_id = %session_id,
                                                "Session not found when processing AgentRequest - defaulting to internal (UI modal)"
                                            );
                                            false
                                        };

                                        // Store the pending request (no timeout - wait indefinitely for user)
                                        let request_id_str = request_id.to_string();
                                        {
                                            let mut pending = pending_agent_requests.lock().await;
                                            pending.insert(request_id_str.clone());
                                        }
                                        info!(
                                            connector_id = %id,
                                            request_id = %request_id_str,
                                            is_forwarded = is_forwarded,
                                            "Stored pending agent request (waiting for user response)"
                                        );

                                        // Emit Event::AgentRequest for routing
                                        // - If is_forwarded=true: EventBridge forwards to external client
                                        // - If is_forwarded=false: Web UI shows permission modal
                                        let event = Event::AgentRequest {
                                            connector_id: id.clone(),
                                            session_id,
                                            request_id: request_id.clone(),
                                            method,
                                            params,
                                            is_forwarded,
                                        };

                                        let bus_event_for_agent_request = dirigent_protocol::streaming::BusEvent::from_connector_event(
                                            event.clone(),
                                            connector_uid,
                                            id.clone(),
                                        );
                                        sharing_bus.publish(bus_event_for_agent_request).await;
                                        if let Err(e) = events_tx.send(event) {
                                            error!(
                                                connector_id = %id,
                                                error = %e,
                                                "Failed to emit AgentRequest event"
                                            );
                                        }

                                        // Response will be handled when AgentResponse command is received
                                        // No timeout - the system waits indefinitely for user input
                                    }
                                }
                            }
                            Ok(None) => {
                                warn!(connector_id = %id, "Transport stream closed");

                                // Update state to Connecting (will reconnect)
                                {
                                    let mut state_guard = state.write().await;
                                    *state_guard = ConnectorState::Connecting;
                                }
                                emit_event(
                                    &sharing_bus,
                                    &events_tx,
                                    &id,
                                    connector_uid,
                                    Event::ConnectorStateChanged {
                                    connector_id: id.clone(),
                                    state: "Connecting".to_string(),
                                    error_kind: None,
                                },
                                )
                                .await;
                                // Emit disconnected event
                                emit_event(&sharing_bus, &events_tx, &id, connector_uid, Event::Disconnected).await;
                                // Deregister session nodes (will be re-created on reconnect)
                                #[cfg(feature = "server")]
                                if let Some(ref inspector) = inspector {
                                    inspector_deregister_all_sessions(inspector, &id, &internal_state).await;
                                }

                                // Cancel all pending requests with crash context before closing
                                let crash_reason = Self::build_crash_reason(&transport, "Connection lost: transport closed").await;
                                protocol_handler.cancel_all_pending(&crash_reason).await;

                                // Close transport and clear stale requests before reconnect
                                let _ = transport.close().await;
                                pending_agent_requests.lock().await.clear();
                                break 'event_loop;
                            }
                            Err(e) => {
                                error!(connector_id = %id, error = %e, "Transport error");

                                // Update state to Error
                                {
                                    let mut state_guard = state.write().await;
                                    *state_guard = ConnectorState::Error(format!("Transport error: {}", e));
                                }
                                emit_event(
                                    &sharing_bus,
                                    &events_tx,
                                    &id,
                                    connector_uid,
                                    Event::ConnectorStateChanged {
                                    connector_id: id.clone(),
                                    state: format!("Error(Transport error: {})", e),
                                    error_kind: Some("transport_error".to_string()),
                                },
                                )
                                .await;
                                // Emit error event
                                emit_event(
                                    &sharing_bus,
                                    &events_tx,
                                    &id,
                                    connector_uid,
                                    Event::Error {
                                    message: format!("Transport error: {}", e),
                                },
                                )
                                .await;
                                // Deregister session nodes (will be re-created on reconnect)
                                #[cfg(feature = "server")]
                                if let Some(ref inspector) = inspector {
                                    inspector_deregister_all_sessions(inspector, &id, &internal_state).await;
                                }

                                // Cancel all pending requests with crash context before closing
                                let crash_reason = Self::build_crash_reason(&transport, &format!("Connection lost: {}", e)).await;
                                protocol_handler.cancel_all_pending(&crash_reason).await;

                                // Close transport and clear stale requests before reconnect
                                let _ = transport.close().await;
                                pending_agent_requests.lock().await.clear();
                                break 'event_loop;
                            }
                        }
                    }

                    // Handle notifications from protocol handler
                    notification = notifications.recv() => {
                        match notification {
                            Some(notif) => {
                                let masked_notif = dirigent_protocol::log_utils::format_for_log(&notif);
                                debug_log!("🔔 Notification received: {}", masked_notif);
                                info!(connector_id = %id, notification = %masked_notif, "🔔 Received ACP notification from protocol handler");

                                // Translate notification to Dirigent event using adapter
                                if let Err(e) = Self::handle_notification(&id, notif, &events_tx, &sharing_bus, connector_uid, &adapter, &internal_state, &session_states, &tool_configuration).await {
                                    debug_log!("⚠️ Failed to handle notification: {}", e);
                                    warn!(connector_id = %id, error = %e, "Failed to handle notification");
                                }
                            }
                            None => {
                                warn!(connector_id = %id, "Notification channel closed");
                                break 'event_loop;
                            }
                        }
                    }

                    // Handle commands
                    cmd = cmd_rx.recv() => {
                        match cmd {
                            Some(ConnectorCommand::ListSessions) => {
                                debug!(connector_id = %id, "Processing ListSessions command");
                                Self::execute_list_sessions(
                                    &mut transport,
                                    &protocol_handler,
                                    &events_tx, &sharing_bus, connector_uid,
                                    &id,
                                    &config,
                                    &internal_state,
                                    &pending_agent_requests,
                                    &mut cmd_rx,
                                ).await;
                            }

                            Some(ConnectorCommand::CreateSession { cwd, project_id, ownership }) => {
                                debug!(connector_id = %id, "Processing CreateSession command");
                                debug_log!("🆕 CreateSession command received, cwd={:?}, ownership={:?}", cwd, ownership);

                                // Build session/new request (use provided cwd or default from config)
                                let session_cwd = cwd.as_ref().unwrap_or(&config.cwd);
                                let request = build_session_new_request(session_cwd, None);
                                debug_log!("📤 session/new request: {}", serde_json::to_string_pretty(&request).unwrap_or_default());

                                // Send request and wait for response
                                match Self::send_request(&mut transport, &protocol_handler, request, Some(&events_tx), Some(&sharing_bus), connector_uid, &id, Some(&pending_agent_requests), Some(&mut cmd_rx), Some(&internal_state)).await {
                                    Ok(response) => {
                                        debug_log!("📥 session/new response: {}", serde_json::to_string_pretty(&response).unwrap_or_default());

                                        // Extract session ID and metadata from result
                                        if let Some(result) = response.get("result") {
                                            if let Some(session_id) = result.get("sessionId").and_then(|s| s.as_str()) {
                                                debug_log!("✅ session/new returned sessionId: {}", session_id);
                                                info!(connector_id = %id, session_id = session_id, "Session created successfully");

                                                // Parse optional models field (UNSTABLE in ACP spec but Claude uses it)
                                                let models: Option<dirigent_protocol::SessionModelState> = result
                                                    .get("models")
                                                    .and_then(|m| {
                                                        serde_json::from_value(m.clone())
                                                            .map_err(|e| {
                                                                warn!(connector_id = %id, "Failed to parse models from session/new: {}", e);
                                                                e
                                                            })
                                                            .ok()
                                                    });

                                                // Parse optional modes field (Stable in ACP spec)
                                                let modes: Option<dirigent_protocol::SessionModeState> = result
                                                    .get("modes")
                                                    .and_then(|m| {
                                                        serde_json::from_value(m.clone())
                                                            .map_err(|e| {
                                                                warn!(connector_id = %id, "Failed to parse modes from session/new: {}", e);
                                                                e
                                                            })
                                                            .ok()
                                                    });

                                                // Parse optional configOptions field (ACP spec replacement for modes/models)
                                                let config_options: Option<Vec<dirigent_protocol::ConfigOption>> = result
                                                    .get("configOptions")
                                                    .and_then(|co| {
                                                        serde_json::from_value(co.clone())
                                                            .map_err(|e| {
                                                                warn!(connector_id = %id, "Failed to parse configOptions from session/new: {}", e);
                                                                e
                                                            })
                                                            .ok()
                                                    });

                                                if models.is_some() || modes.is_some() || config_options.is_some() {
                                                    debug_log!("📊 session/new metadata: models={}, modes={}, configOptions={}",
                                                        models.as_ref().map(|m| format!("{} models, current={}", m.available_models.len(), m.current_model_id)).unwrap_or_else(|| "none".to_string()),
                                                        modes.as_ref().map(|m| format!("{} modes, current={}", m.available_modes.len(), m.current_mode_id)).unwrap_or_else(|| "none".to_string()),
                                                        config_options.as_ref().map(|co| format!("{} options", co.len())).unwrap_or_else(|| "none".to_string())
                                                    );
                                                }

                                                // Extract current_mode_id for SessionMetadata
                                                let current_mode_id = modes.as_ref().map(|m| m.current_mode_id.clone());

                                                // Extract current model name for legacy model field
                                                let current_model_name = models.as_ref().and_then(|m| {
                                                    m.available_models.iter()
                                                        .find(|model| model.model_id == m.current_model_id)
                                                        .map(|model| model.name.clone())
                                                });

                                                // Update internal state with metadata
                                                let now = Utc::now();
                                                let session_info = SessionInfo {
                                                    id: session_id.to_string(),
                                                    title: None,
                                                    cwd: session_cwd.clone(),
                                                    message_count: 0,
                                                    model: current_model_name.clone(),
                                                    created_at: now,
                                                    last_activity: now,
                                                    status: SessionStatus::Active,
                                                    models: models.clone(),
                                                    modes: modes.clone(),
                                                    config_options: config_options.clone(),
                                                    ownership: ownership.clone(),
                                                };
                                                internal_state.upsert_session(session_info.clone()).await;

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

                                                // Build _meta with tool configuration if present
                                                let session_meta = Self::build_session_meta(&tool_configuration);

                                                // Emit SessionCreated event
                                                emit_event(
                                                    &sharing_bus,
                                                    &events_tx,
                                                    &id,
                                                    connector_uid,
                                                    Event::SessionCreated {
                                                        connector_id: id.clone(),
                                                        session: dirigent_protocol::Session {
                                                            id: session_id.to_string(),
                                                            title: String::new(),
                                                            created_at: Utc::now(),
                                                            updated_at: Utc::now(),
                                                            metadata: dirigent_protocol::SessionMetadata {
                                                                project_path: session_cwd.clone(),
                                                                model: current_model_name,
                                                                total_messages: 0,
                                                                system_message: None,
                                                                current_mode_id: current_mode_id.clone(),
                                                                _meta: session_meta,
                                                                project_id: project_id.as_ref().and_then(|p| uuid::Uuid::parse_str(p).ok()),
                                                            },
                                                            cwd: Some(session_cwd.clone()),
                                                            // Include ACP models/modes state in Session
                                                            models: models.clone(),
                                                            modes: modes.clone(),
                                                            config_options: config_options.clone(),
                                                            // ACP connector sessions don't have a client ID (that's for incoming connections)
                                                            acp_client_id: None,
                                                        },
                                                    },
                                                )
                                                .await;

                                                // Emit SessionMetadataReceived event if we have any metadata
                                                if models.is_some() || modes.is_some() || config_options.is_some() {
                                                    emit_event(
                                                        &sharing_bus,
                                                        &events_tx,
                                                        &id,
                                                        connector_uid,
                                                        Event::SessionMetadataReceived {
                                                        connector_id: id.clone(),
                                                        session_id: session_id.to_string(),
                                                        models,
                                                        modes,
                                                        config_options,
                                                    },
                                                    )
                                                    .await;
                                                }

                                                // Auto-trigger ListSessions after session creation.
                                                // This populates the session list for any deferred
                                                // obligations (e.g., a client called session/list on
                                                // a gateway, got transferred here, and is waiting for
                                                // the real session list from the upstream agent).
                                                Self::execute_list_sessions(
                                                    &mut transport,
                                                    &protocol_handler,
                                                    &events_tx, &sharing_bus, connector_uid,
                                                    &id,
                                                    &config,
                                                    &internal_state,
                                                    &pending_agent_requests,
                                                    &mut cmd_rx,
                                                ).await;
                                            } else {
                                                error!(connector_id = %id, "Session creation response missing sessionId");
                                                emit_event(
                                                    &sharing_bus,
                                                    &events_tx,
                                                    &id,
                                                    connector_uid,
                                                    Event::Error {
                                                    message: "Session creation failed: invalid response".to_string(),
                                                },
                                                )
                                                .await;
                                            }
                                        }
                                    }
                                    Err(e) => {
                                        error!(connector_id = %id, error = %e, "Failed to create session");
                                        emit_event(
                                            &sharing_bus,
                                            &events_tx,
                                            &id,
                                            connector_uid,
                                            Event::Error {
                                            message: format!("Failed to create session: {}", e),
                                        },
                                        )
                                        .await;
                                    }
                                }
                            }

                            Some(ConnectorCommand::ListMessages { session_id }) => {
                                debug!(connector_id = %id, session_id = %session_id, "Processing ListMessages command");

                                // For ACP, message listing is not part of the core protocol
                                // This would need to be handled via custom agent implementation
                                warn!(connector_id = %id, "ListMessages not yet implemented for ACP");

                                emit_event(
                                    &sharing_bus,
                                    &events_tx,
                                    &id,
                                    connector_uid,
                                    Event::MessagesListed { messages: vec![] },
                                )
                                .await;
                            }

                            Some(ConnectorCommand::LoadSession { session_id, cwd, mcp_servers }) => {
                                debug!(connector_id = %id, session_id = %session_id, "Processing LoadSession command");

                                // Check agent capabilities and session state
                                let supports_load = internal_state.agent_supports_load_session().await;
                                let supports_resume = internal_state.agent_supports_session_resume().await;
                                let already_loaded = internal_state.is_session_loaded(&session_id).await;

                                let request = if already_loaded && supports_resume {
                                    // Session already loaded in current process — use resume (fast re-attach, no history replay)
                                    debug!(connector_id = %id, session_id = %session_id, "Using session/resume (already loaded in process)");
                                    Some(protocol::build_session_resume_request(&session_id, Some(&cwd), mcp_servers.clone()))
                                } else if supports_load {
                                    // Full restore from disk with history replay
                                    debug!(connector_id = %id, session_id = %session_id, "Using session/load (full restore)");
                                    Some(protocol::build_session_load_request(&session_id, &cwd, mcp_servers))
                                } else if supports_resume {
                                    // Fallback: agent doesn't support load but supports resume
                                    debug!(connector_id = %id, session_id = %session_id, "Using session/resume (load not supported, fallback)");
                                    Some(protocol::build_session_resume_request(&session_id, Some(&cwd), mcp_servers))
                                } else {
                                    warn!(connector_id = %id, "Agent supports neither session/load nor session/resume");
                                    emit_event(
                                        &sharing_bus,
                                        &events_tx,
                                        &id,
                                        connector_uid,
                                        Event::Error {
                                        message: "Session loading is not supported by this agent".to_string(),
                                    },
                                    )
                                    .await;
                                    None
                                };

                                if let Some(request) = request {

                                    // Send request and wait for response
                                    match Self::send_request(&mut transport, &protocol_handler, request, Some(&events_tx), Some(&sharing_bus), connector_uid, &id, Some(&pending_agent_requests), Some(&mut cmd_rx), Some(&internal_state)).await {
                                        Ok(response) => {
                                            info!(connector_id = %id, session_id = %session_id, "Session loaded successfully");

                                            // Extract result for metadata parsing
                                            let result = response.get("result");

                                            // Parse optional models field (UNSTABLE in ACP spec but Claude uses it)
                                            let models: Option<dirigent_protocol::SessionModelState> = result
                                                .and_then(|r| r.get("models"))
                                                .and_then(|m| {
                                                    serde_json::from_value(m.clone())
                                                        .map_err(|e| {
                                                            warn!(connector_id = %id, "Failed to parse models from session/load: {}", e);
                                                            e
                                                        })
                                                        .ok()
                                                });

                                            // Parse optional modes field (Stable in ACP spec)
                                            let modes: Option<dirigent_protocol::SessionModeState> = result
                                                .and_then(|r| r.get("modes"))
                                                .and_then(|m| {
                                                    serde_json::from_value(m.clone())
                                                        .map_err(|e| {
                                                            warn!(connector_id = %id, "Failed to parse modes from session/load: {}", e);
                                                            e
                                                        })
                                                        .ok()
                                                });

                                            // Parse optional configOptions field (ACP spec replacement for modes/models)
                                            let config_options: Option<Vec<dirigent_protocol::ConfigOption>> = result
                                                .and_then(|r| r.get("configOptions"))
                                                .and_then(|co| {
                                                    serde_json::from_value(co.clone())
                                                        .map_err(|e| {
                                                            warn!(connector_id = %id, "Failed to parse configOptions from session/load: {}", e);
                                                            e
                                                        })
                                                        .ok()
                                                });

                                            if models.is_some() || modes.is_some() || config_options.is_some() {
                                                debug_log!("📊 session/load metadata: models={}, modes={}, configOptions={}",
                                                    models.as_ref().map(|m| format!("{} models, current={}", m.available_models.len(), m.current_model_id)).unwrap_or_else(|| "none".to_string()),
                                                    modes.as_ref().map(|m| format!("{} modes, current={}", m.available_modes.len(), m.current_mode_id)).unwrap_or_else(|| "none".to_string()),
                                                    config_options.as_ref().map(|co| format!("{} options", co.len())).unwrap_or_else(|| "none".to_string())
                                                );
                                            }

                                            // Extract current_mode_id for SessionMetadata
                                            let current_mode_id = modes.as_ref().map(|m| m.current_mode_id.clone());

                                            // Extract current model name for legacy model field
                                            let current_model_name = models.as_ref().and_then(|m| {
                                                m.available_models.iter()
                                                    .find(|model| model.model_id == m.current_model_id)
                                                    .map(|model| model.name.clone())
                                            });

                                            // Update internal state with metadata
                                            let now = Utc::now();
                                            let session_info = SessionInfo {
                                                id: session_id.clone(),
                                                title: None,
                                                cwd: config.cwd.clone(),
                                                message_count: 0,
                                                model: current_model_name.clone(),
                                                created_at: now,
                                                last_activity: now,
                                                status: SessionStatus::Active,
                                                models: models.clone(),
                                                modes: modes.clone(),
                                                config_options: config_options.clone(),
                                                ownership: SessionOwnership::default(),
                                            };
                                            internal_state.upsert_session(session_info.clone()).await;

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

                                            // Build _meta with tool configuration if present
                                            let session_meta = Self::build_session_meta(&tool_configuration);

                                            // Emit SessionCreated event so archivist can register the session
                                            // This prevents resolve_session() failures when messages arrive
                                            emit_event(
                                                &sharing_bus,
                                                &events_tx,
                                                &id,
                                                connector_uid,
                                                Event::SessionCreated {
                                                    connector_id: id.clone(),
                                                    session: dirigent_protocol::Session {
                                                        id: session_id.clone(),
                                                        title: String::new(),
                                                        created_at: now,
                                                        updated_at: now,
                                                        metadata: dirigent_protocol::SessionMetadata {
                                                            project_path: config.cwd.clone(),
                                                            model: current_model_name,
                                                            total_messages: 0,
                                                            system_message: None,
                                                            current_mode_id: current_mode_id.clone(),
                                                            _meta: session_meta,
                                                            project_id: None,
                                                        },
                                                        cwd: Some(config.cwd.clone()),
                                                        // Include ACP models/modes state in Session
                                                        models: models.clone(),
                                                        modes: modes.clone(),
                                                        config_options: config_options.clone(),
                                                        // ACP connector sessions don't have a client ID (that's for incoming connections)
                                                        acp_client_id: None,
                                                    },
                                                },
                                            )
                                            .await;

                                            // Emit SessionMetadataReceived event if we have any metadata
                                            if models.is_some() || modes.is_some() || config_options.is_some() {
                                                emit_event(
                                                    &sharing_bus,
                                                    &events_tx,
                                                    &id,
                                                    connector_uid,
                                                    Event::SessionMetadataReceived {
                                                    connector_id: id.clone(),
                                                    session_id: session_id.clone(),
                                                    models,
                                                    modes,
                                                    config_options,
                                                },
                                                )
                                                .await;
                                            }

                                            // Track that this session is now loaded in this process
                                            // so subsequent loads use session/resume instead of session/load
                                            internal_state.mark_session_loaded(&session_id).await;

                                            // Register session with idle detector so that SessionIdle
                                            // fires after history replay notifications stop arriving.
                                            // This triggers archivist finalization (MessageAccumulator.finalize())
                                            // which writes the replayed messages to disk.
                                            {
                                                let mut states = session_states.lock().await;
                                                if let Some(state) = states.get_mut(&session_id) {
                                                    state.mark_awaiting_idle();
                                                } else {
                                                    let mut new_state = SessionState::new(session_id.clone());
                                                    new_state.mark_awaiting_idle();
                                                    states.insert(session_id.clone(), new_state);
                                                }
                                            }

                                            // History replay happens via session/update notifications
                                            // which are handled by the notification handler
                                        }
                                        Err(e) => {
                                            error!(connector_id = %id, session_id = %session_id, error = %e, "Failed to load session");

                                            // Check if it's a "session not found" error (not recoverable)
                                            // Other errors (transport, timeout) may be recoverable
                                            let (error_message, is_recoverable) = if matches!(e, AcpError::AgentError { code, .. } if code == -32602 || code == -32001) {
                                                (format!("Session not found: {}", session_id), false)
                                            } else {
                                                (format!("Failed to load session: {}", e), e.is_retriable())
                                            };

                                            let error_code = if matches!(e, AcpError::AgentError { code, .. } if code == -32602 || code == -32001) {
                                                "SESSION_NOT_FOUND"
                                            } else {
                                                "SESSION_LOAD_FAILED"
                                            };

                                            emit_event(
                                                &sharing_bus,
                                                &events_tx,
                                                &id,
                                                connector_uid,
                                                Event::SessionError {
                                                connector_id: id.clone(),
                                                session_id: session_id.clone(),
                                                error_message,
                                                is_recoverable,
                                                error_code: Some(error_code.to_string()),
                                                technical_details: Some(format!("{:?}", e)),
                                                context: Some(serde_json::json!({
                                                    "operation": "session/load",
                                                    "connector_id": id.clone(),
                                                })),
                                            },
                                            )
                                            .await;
                                            debug_log!("📛 Emitted SessionError for LoadSession failure: session={}, error={}", session_id, e);
                                        }
                                    }
                                }
                            }

                            Some(ConnectorCommand::CancelGeneration { session_id }) => {
                                debug!(connector_id = %id, session_id = %session_id, "Processing CancelGeneration command");

                                // Build session/cancel request
                                let request = build_session_cancel_request(&session_id);

                                // Send request (fire and forget, don't wait for response)
                                match Self::send_request(&mut transport, &protocol_handler, request, Some(&events_tx), Some(&sharing_bus), connector_uid, &id, Some(&pending_agent_requests), Some(&mut cmd_rx), Some(&internal_state)).await {
                                    Ok(_) => {
                                        info!(connector_id = %id, session_id = %session_id, "Generation cancelled");
                                    }
                                    Err(e) => {
                                        warn!(connector_id = %id, session_id = %session_id, error = %e, "Failed to cancel generation");
                                    }
                                }
                            }

                            Some(ConnectorCommand::SendMessage { session_id, text }) => {
                                debug_log!("📨 SendMessage command: session={}, text_len={}", session_id, text.len());
                                info!(connector_id = %id, session_id = %session_id, text_len = text.len(), "Processing SendMessage command");

                                // Generate UUIDv7 for the user message (for temporal ordering)
                                let user_message_id = uuid::Uuid::now_v7().to_string();
                                debug_log!("📝 Generated user message ID: {}", user_message_id);
                                info!(connector_id = %id, user_message_id = %user_message_id, "Generated user message ID");

                                // Build prompt request
                                let prompt = json!([{
                                    "type": "text",
                                    "text": text
                                }]);

                                let request = build_session_prompt_request(&session_id, prompt);
                                let masked_request = dirigent_protocol::log_utils::format_for_log(&request);
                                debug!(connector_id = %id, request = %masked_request, "Built prompt request");

                                // NON-BLOCKING: Prepare request and send it, then spawn task to handle response
                                // Prepare request (adds ID and creates response channel)
                                let (message_with_id, response_rx) = protocol_handler.prepare_request(request).await;

                                let request_id = message_with_id.get("id").cloned().unwrap_or(json!(null));
                                debug!(connector_id = %id, request_id = %request_id, "Prepared non-blocking prompt request");

                                // Send via transport (non-blocking - just writes to transport)
                                let send_result = transport.send(message_with_id).await;
                                debug!(connector_id = %id, request_id = %request_id, "Non-blocking send completed");

                                // Mark session as Processing in inspector
                                #[cfg(feature = "server")]
                                if let Some(ref inspector) = inspector {
                                    inspector_update_session_state(
                                        inspector,
                                        &id,
                                        &session_id,
                                        dirigent_inspector::NodeState::Busy("Generating".to_string()),
                                    ).await;
                                }

                                // Emit user message IMMEDIATELY via SessionUpdate for instant UI feedback
                                // This ensures the user message appears BEFORE the agent starts responding
                                emit_event(
                                    &sharing_bus,
                                    &events_tx,
                                    &id,
                                    connector_uid,
                                    Event::SessionUpdate {
                                        connector_id: id.clone(),
                                        session_id: session_id.clone(),
                                        update: SessionUpdate::UserMessageChunk {
                                            message_id: user_message_id.clone(),
                                            content: ContentBlock::Text {
                                                text: text.clone(),
                                            },
                                            _meta: None,
                                        },
                                    },
                                )
                                .await;
                                debug!(connector_id = %id, session_id = %session_id, user_message_id = %user_message_id, "Emitted UserMessageChunk immediately after send");

                                // Clone values for spawned task
                                let session_id_clone = session_id.clone();
                                let text_clone = text.clone();
                                let user_message_id_clone = user_message_id.clone();
                                let id_clone = id.clone();
                                let events_tx_clone = events_tx.clone();
                                let sharing_bus_clone = Arc::clone(&sharing_bus);
                                let internal_state_clone = Arc::clone(&internal_state);
                                let session_states_clone = Arc::clone(&session_states);
                                #[cfg(feature = "server")]
                                let inspector_clone = inspector.clone();

                                // Spawn task to wait for response and perform post-processing
                                // This allows the main event loop to continue polling notifications
                                tokio::spawn(async move {
                                    match send_result {
                                        Err(e) => {
                                            error!(connector_id = %id_clone, session_id = %session_id_clone, error = %e, "Failed to send prompt (transport error)");

                                            let is_recoverable = true; // Transport errors are generally recoverable
                                            let error_string = format!("{}", e);
                                            let debug_string = format!("{:?}", e);

                                            // Extract received content preview if present (for JSON parse errors)
                                            let received_preview = if error_string.contains("Line:") {
                                                error_string
                                                    .split("Line:")
                                                    .nth(1)
                                                    .map(|s| {
                                                        let trimmed = s.trim();
                                                        if trimmed.len() > 500 {
                                                            format!("{}... (truncated, {} bytes total)", &trimmed[..500], trimmed.len())
                                                        } else {
                                                            trimmed.to_string()
                                                        }
                                                    })
                                            } else {
                                                None
                                            };

                                            emit_event(
                                                &sharing_bus_clone,
                                                &events_tx_clone,
                                                &id_clone,
                                                connector_uid,
                                                Event::SessionError {
                                                    connector_id: id_clone.clone(),
                                                    session_id: session_id_clone.clone(),
                                                    error_message: "Failed to send message: Transport error".to_string(),
                                                    is_recoverable,
                                                    error_code: Some("TRANSPORT_ERROR".to_string()),
                                                    technical_details: Some(debug_string),
                                                    context: Some(serde_json::json!({
                                                        "operation": "session/prompt",
                                                        "connector_id": id_clone.clone(),
                                                        "error_display": error_string,
                                                        "received_preview": received_preview,
                                                        "user_message_preview": if text_clone.len() > 100 {
                                                            format!("{}...", &text_clone[..100])
                                                        } else {
                                                            text_clone.clone()
                                                        },
                                                    })),
                                                },
                                            )
                                            .await;

                                            // Mark session as awaiting idle even on error
                                            // The idle check will emit SessionIdle after the threshold
                                            {
                                                let mut states = session_states_clone.lock().await;
                                                if let Some(state) = states.get_mut(&session_id_clone) {
                                                    state.mark_awaiting_idle();
                                                } else {
                                                    // Create new state if not exists
                                                    states.insert(session_id_clone.clone(), SessionState::new(session_id_clone.clone()));
                                                    states.get_mut(&session_id_clone).unwrap().mark_awaiting_idle();
                                                }
                                            }
                                            return;
                                        }
                                        Ok(_) => {
                                            debug!(connector_id = %id_clone, "Waiting for prompt response...");
                                        }
                                    }

                                    // Wait for response via protocol handler correlation
                                    match response_rx.await {
                                        Ok(response) => {
                                            let masked_response = dirigent_protocol::log_utils::format_for_log(&response);
                                            debug_log!("✅ Prompt response received (non-blocking): {}", masked_response);
                                            info!(connector_id = %id_clone, response = %masked_response, "Prompt sent successfully, got response");

                                            // Check for JSON-RPC error in response
                                            if let Some(error) = response.get("error") {
                                                let message = error
                                                    .get("message")
                                                    .and_then(|m| m.as_str())
                                                    .unwrap_or("Unknown error")
                                                    .to_string();

                                                let error_code_num = error.get("code").and_then(|c| c.as_i64());
                                                let error_data = error.get("data").cloned();

                                                // Detect cancel_all_pending responses (code -32000 = connection lost)
                                                let (error_code_str, error_msg) = if error_code_num == Some(-32000) {
                                                    error!(connector_id = %id_clone, session_id = %session_id_clone, error = %message, "Request cancelled due to connection loss");
                                                    ("CONNECTION_LOST".to_string(), message.clone())
                                                } else {
                                                    error!(connector_id = %id_clone, session_id = %session_id_clone, error = %message, "Prompt request failed with JSON-RPC error");
                                                    ("JSONRPC_ERROR".to_string(), format!("Agent error: {}", message))
                                                };

                                                emit_event(
                                                    &sharing_bus_clone,
                                                    &events_tx_clone,
                                                    &id_clone,
                                                    connector_uid,
                                                    Event::SessionError {
                                                        connector_id: id_clone.clone(),
                                                        session_id: session_id_clone.clone(),
                                                        error_message: error_msg,
                                                        is_recoverable: true,
                                                        error_code: Some(error_code_str),
                                                        technical_details: Some(format!("JSON-RPC error code: {:?}, message: {}", error_code_num, message)),
                                                        context: Some(serde_json::json!({
                                                            "operation": "session/prompt",
                                                            "jsonrpc_error_code": error_code_num,
                                                            "jsonrpc_error_message": message,
                                                            "jsonrpc_error_data": error_data,
                                                        })),
                                                    },
                                                )
                                                .await;

                                                // Mark session as awaiting idle after error
                                                {
                                                    let mut states = session_states_clone.lock().await;
                                                    if let Some(state) = states.get_mut(&session_id_clone) {
                                                        state.mark_awaiting_idle();
                                                    } else {
                                                        states.insert(session_id_clone.clone(), SessionState::new(session_id_clone.clone()));
                                                        states.get_mut(&session_id_clone).unwrap().mark_awaiting_idle();
                                                    }
                                                }
                                                return;
                                            }

                                            // Emit user message event AFTER successful prompt send
                                            // User messages are atomic (not streamed), so emit MessageCompleted directly
                                            let user_message = Message {
                                                id: user_message_id_clone.clone(),
                                                session_id: session_id_clone.clone(),
                                                role: MessageRole::User,
                                                created_at: Utc::now(),
                                                content: vec![
                                                    MessagePart::Text {
                                                        text: text_clone.clone(),
                                                    }
                                                ],
                                                status: MessageStatus::Completed,
                                                metadata: None,
                                            };

                                            emit_event(
                                                &sharing_bus_clone,
                                                &events_tx_clone,
                                                &id_clone,
                                                connector_uid,
                                                Event::MessageCompleted {
                                                    connector_id: id_clone.clone(),
                                                    message: user_message,
                                                },
                                            )
                                            .await;
                                            debug_log!("✅ Emitted user MessageCompleted for {}", user_message_id_clone);
                                            info!(connector_id = %id_clone, session_id = %session_id_clone, user_message_id = %user_message_id_clone, "✅ Emitted user MessageCompleted");

                                            // Emit TurnComplete for user message so archivist can finalize and write it
                                            // User messages are atomic (not streamed), so turn is complete immediately
                                            emit_event(
                                                &sharing_bus_clone,
                                                &events_tx_clone,
                                                &id_clone,
                                                connector_uid,
                                                Event::TurnComplete {
                                                    connector_id: id_clone.clone(),
                                                    session_id: session_id_clone.clone(),
                                                    message_id: user_message_id_clone.clone(),
                                                    trigger: TurnCompleteTrigger::ResponseReceived,
                                                },
                                            )
                                            .await;
                                            debug_log!("✅ Emitted user TurnComplete for {}", user_message_id_clone);

                                            // Derive title from first user message (Task 1.2: Title Propagation)
                                            // Check if this is the first user message for this session
                                            let should_set_title = {
                                                let session_info = internal_state_clone.get_session(&session_id_clone).await;
                                                match session_info {
                                                    Some(info) => info.title.is_none() && info.message_count == 0,
                                                    None => false, // Session not tracked, skip title derivation
                                                }
                                            };

                                            if should_set_title {
                                                // Derive title from text content (first 50 chars, trimmed, word boundary)
                                                let title = derive_title_from_text(&text_clone);

                                                debug_log!("📝 Derived title for session {}: {}", session_id_clone, title);
                                                info!(connector_id = %id_clone, session_id = %session_id_clone, title = %title, "📝 Derived session title from first user message");

                                                // Update internal state with title
                                                if let Some(mut info) = internal_state_clone.get_session(&session_id_clone).await {
                                                    info.title = Some(title.clone());
                                                    internal_state_clone.upsert_session(info.clone()).await;

                                                    // Update inspector with new title
                                                    #[cfg(feature = "server")]
                                                    if let Some(ref inspector) = inspector_clone {
                                                        inspector_upsert_session(inspector, &id_clone, &info).await;
                                                    }
                                                }

                                                // Emit SessionMetadataUpdated event
                                                emit_event(
                                                    &sharing_bus_clone,
                                                    &events_tx_clone,
                                                    &id_clone,
                                                    connector_uid,
                                                    Event::SessionMetadataUpdated {
                                                        connector_id: id_clone.clone(),
                                                        session_id: session_id_clone.clone(),
                                                        title: Some(title.clone()),
                                                        total_messages: None,
                                                        model: None,
                                                    },
                                                )
                                                .await;
                                                debug_log!("✅ Emitted SessionMetadataUpdated with title: {}", title);
                                                info!(connector_id = %id_clone, session_id = %session_id_clone, title = %title, "✅ Emitted SessionMetadataUpdated");
                                            }

                                            // Response streaming is handled via notifications
                                            // Note: Notifications may arrive AFTER this response, so there might not be
                                            // an active message yet. We still need to emit SessionIdle to clear is_generating.

                                            // Always try to emit MessageCompleted, even if no chunks arrived yet.
                                            // This handles the race where the prompt response arrives before the first chunk.
                                            let message_id = internal_state_clone.clear_active_message(&session_id_clone).await;

                                            // Build pending completion data if we have an active message
                                            // MessageCompleted will be emitted when SessionIdle fires (after 300ms of inactivity)
                                            // This ensures all notifications are processed before finalization
                                            let pending = match message_id {
                                                Some(message_id) => {
                                                    debug_log!("🎯 Preparing deferred MessageCompleted: session={}, message_id={}", session_id_clone, message_id);
                                                    info!(connector_id = %id_clone, session_id = %session_id_clone, message_id = %message_id, "🎯 Storing pending MessageCompleted (will emit on SessionIdle)");

                                                    // Get message start time (when first chunk arrived)
                                                    let created_at = internal_state_clone
                                                        .get_message_start_time(&message_id)
                                                        .await
                                                        .unwrap_or_else(Utc::now); // Fallback to now if not tracked

                                                    // Create a minimal Message struct to signal completion
                                                    // The cache already has all the chunks accumulated
                                                    let message = Message {
                                                        id: message_id.clone(),
                                                        session_id: session_id_clone.clone(),
                                                        role: MessageRole::Assistant,
                                                        created_at, // Use message start time, not finalization time
                                                        content: vec![], // Chunks already accumulated in cache
                                                        status: MessageStatus::Completed,
                                                        metadata: None,
                                                    };

                                                    // Clean up message start time to prevent memory leak
                                                    internal_state_clone.clear_message_start_time(&message_id).await;

                                                    Some(PendingCompletion {
                                                        connector_id: id_clone.clone(),
                                                        message,
                                                    })
                                                }
                                                None => {
                                                    debug_log!("⚠️ No active message to finalize (normal for sessions with no streaming)");
                                                    None
                                                }
                                            };

                                            // Mark session as awaiting idle and store pending completion
                                            // MessageCompleted + SessionIdle will be emitted by check_idle_sessions()
                                            // after IDLE_THRESHOLD (300ms) of inactivity
                                            {
                                                let mut states = session_states_clone.lock().await;
                                                if let Some(state) = states.get_mut(&session_id_clone) {
                                                    state.mark_awaiting_idle();
                                                    state.pending_completion = pending;
                                                } else {
                                                    // Create new state if not exists
                                                    let mut new_state = SessionState::new(session_id_clone.clone());
                                                    new_state.mark_awaiting_idle();
                                                    new_state.pending_completion = pending;
                                                    states.insert(session_id_clone.clone(), new_state);
                                                }
                                            }
                                            debug_log!("✅ Marked session {} as awaiting idle with pending completion", session_id_clone);
                                            info!(connector_id = %id_clone, session_id = %session_id_clone, "🎯 Marked session as awaiting idle after prompt completion (MessageCompleted deferred)");
                                        }
                                        Err(_) => {
                                            error!(connector_id = %id_clone, session_id = %session_id_clone, "Response channel dropped (connection lost)");

                                            emit_event(
                                                &sharing_bus_clone,
                                                &events_tx_clone,
                                                &id_clone,
                                                connector_uid,
                                                Event::SessionError {
                                                    connector_id: id_clone.clone(),
                                                    session_id: session_id_clone.clone(),
                                                    error_message: "Connection to agent was lost".to_string(),
                                                    is_recoverable: true,
                                                    error_code: Some("CONNECTION_LOST".to_string()),
                                                    technical_details: Some("Response channel was dropped — agent process likely exited".to_string()),
                                                    context: Some(serde_json::json!({
                                                        "operation": "session/prompt",
                                                        "reason": "channel_dropped",
                                                    })),
                                                },
                                            )
                                            .await;

                                            // Mark session as awaiting idle after timeout/cancellation
                                            {
                                                let mut states = session_states_clone.lock().await;
                                                if let Some(state) = states.get_mut(&session_id_clone) {
                                                    state.mark_awaiting_idle();
                                                } else {
                                                    states.insert(session_id_clone.clone(), SessionState::new(session_id_clone.clone()));
                                                    states.get_mut(&session_id_clone).unwrap().mark_awaiting_idle();
                                                }
                                            }
                                        }
                                    }
                                });

                                // CRITICAL: Return immediately to event loop!
                                // The spawned task will handle all response processing
                                // This allows notifications to be processed in real-time
                            }

                            Some(ConnectorCommand::Reconnect) => {
                                info!(connector_id = %id, "Received Reconnect command, restarting connection");

                                // Close transport
                                let _ = transport.close().await;

                                // Reset retry count and clear stale state
                                retry_count = 0;
                                internal_state.clear().await;
                                pending_agent_requests.lock().await.clear();
                                break 'event_loop;
                            }

                            Some(ConnectorCommand::AgentResponse { request_id, response }) => {
                                info!(
                                    connector_id = %id,
                                    request_id = %request_id,
                                    "Received AgentResponse command"
                                );

                                // Convert request_id to string for HashSet lookup
                                let request_id_str = request_id.to_string();

                                // Check if this request is still pending and remove it
                                let was_pending = {
                                    let mut pending = pending_agent_requests.lock().await;
                                    pending.remove(&request_id_str)
                                };

                                if was_pending {
                                    // Send response directly to agent via transport
                                    info!(
                                        connector_id = %id,
                                        request_id = %request_id_str,
                                        "Sending agent response to transport"
                                    );

                                    if let Err(e) = transport.send(response).await {
                                        error!(
                                            connector_id = %id,
                                            request_id = %request_id_str,
                                            error = %e,
                                            "Failed to send agent response to transport"
                                        );
                                    } else {
                                        info!(
                                            connector_id = %id,
                                            request_id = %request_id_str,
                                            "Successfully sent agent response to transport"
                                        );
                                    }
                                } else {
                                    warn!(
                                        connector_id = %id,
                                        request_id = %request_id_str,
                                        "Received AgentResponse for unknown request_id (not found in pending requests)"
                                    );
                                }
                            }

                            Some(ConnectorCommand::SetSessionMode { session_id, mode_id }) => {
                                info!(
                                    connector_id = %id,
                                    session_id = %session_id,
                                    mode_id = %mode_id,
                                    "Processing SetSessionMode command"
                                );

                                // Check if this session has config_options — if so, use the new API
                                let has_config_options = internal_state.get_session(&session_id).await
                                    .and_then(|s| s.config_options.as_ref().map(|opts| !opts.is_empty()))
                                    .unwrap_or(false);

                                let request = if has_config_options {
                                    debug!(connector_id = %id, "Using session/set_config_option for mode change");
                                    protocol::build_set_config_option_request(&session_id, "mode", &mode_id)
                                } else {
                                    // Legacy path for agents that don't support configOptions
                                    json!({
                                        "jsonrpc": "2.0",
                                        "method": "session/set_mode",
                                        "params": {
                                            "sessionId": session_id,
                                            "modeId": mode_id
                                        }
                                    })
                                };

                                let using_config_option = has_config_options;

                                // Send request and wait for response
                                match Self::send_request(
                                    &mut transport,
                                    &protocol_handler,
                                    request,
                                    Some(&events_tx), Some(&sharing_bus), connector_uid,
                                    &id,
                                    Some(&pending_agent_requests),
                                    Some(&mut cmd_rx),
                                    Some(&internal_state),
                                ).await {
                                    Ok(response) => {
                                        info!(
                                            connector_id = %id,
                                            session_id = %session_id,
                                            mode_id = %mode_id,
                                            "Mode changed successfully"
                                        );

                                        // When using set_config_option, parse configOptions response
                                        if using_config_option {
                                            if let Some(result) = response.get("result") {
                                                let config_options: Option<Vec<dirigent_protocol::ConfigOption>> = result
                                                    .get("configOptions")
                                                    .and_then(|co| serde_json::from_value(co.clone()).ok());

                                                if let Some(ref opts) = config_options {
                                                    if let Some(mut session_info) = internal_state.get_session(&session_id).await {
                                                        session_info.config_options = Some(opts.clone());
                                                        internal_state.upsert_session(session_info).await;
                                                    }

                                                    emit_event(
                                                        &sharing_bus,
                                                        &events_tx,
                                                        &id,
                                                        connector_uid,
                                                        Event::SessionMetadataReceived {
                                                        connector_id: id.clone(),
                                                        session_id: session_id.clone(),
                                                        models: None,
                                                        modes: None,
                                                        config_options,
                                                    },
                                                    )
                                                    .await;
                                                }
                                            }
                                        }
                                    }
                                    Err(e) => {
                                        error!(
                                            connector_id = %id,
                                            session_id = %session_id,
                                            mode_id = %mode_id,
                                            error = %e,
                                            "Failed to change mode"
                                        );

                                        emit_event(
                                            &sharing_bus,
                                            &events_tx,
                                            &id,
                                            connector_uid,
                                            Event::SessionError {
                                            connector_id: id.clone(),
                                            session_id: session_id.clone(),
                                            error_message: format!("Failed to set mode: {}", e),
                                            is_recoverable: true,
                                            error_code: Some("SET_MODE_FAILED".to_string()),
                                            technical_details: Some(format!("{:?}", e)),
                                            context: Some(serde_json::json!({
                                                "operation": "session/set_mode",
                                                "mode_id": mode_id,
                                            })),
                                        },
                                        )
                                        .await;
                                    }
                                }
                            }

                            Some(ConnectorCommand::SetSessionModel { session_id, model_id }) => {
                                info!(
                                    connector_id = %id,
                                    session_id = %session_id,
                                    model_id = %model_id,
                                    "Processing SetSessionModel command"
                                );

                                // Check if this session has config_options — if so, use the new API
                                let has_config_options = internal_state.get_session(&session_id).await
                                    .and_then(|s| s.config_options.as_ref().map(|opts| !opts.is_empty()))
                                    .unwrap_or(false);

                                let request = if has_config_options {
                                    debug!(connector_id = %id, "Using session/set_config_option for model change");
                                    protocol::build_set_config_option_request(&session_id, "model", &model_id)
                                } else {
                                    // Legacy path for agents that don't support configOptions
                                    json!({
                                        "jsonrpc": "2.0",
                                        "method": "session/set_model",
                                        "params": {
                                            "sessionId": session_id,
                                            "modelId": model_id
                                        }
                                    })
                                };

                                let using_config_option = has_config_options;

                                // Send request and wait for response
                                match Self::send_request(
                                    &mut transport,
                                    &protocol_handler,
                                    request,
                                    Some(&events_tx), Some(&sharing_bus), connector_uid,
                                    &id,
                                    Some(&pending_agent_requests),
                                    Some(&mut cmd_rx),
                                    Some(&internal_state),
                                ).await {
                                    Ok(response) => {
                                        info!(
                                            connector_id = %id,
                                            session_id = %session_id,
                                            model_id = %model_id,
                                            "Model changed successfully"
                                        );

                                        // When using set_config_option, parse configOptions response
                                        if using_config_option {
                                            if let Some(result) = response.get("result") {
                                                let config_options: Option<Vec<dirigent_protocol::ConfigOption>> = result
                                                    .get("configOptions")
                                                    .and_then(|co| serde_json::from_value(co.clone()).ok());

                                                if let Some(ref opts) = config_options {
                                                    if let Some(mut session_info) = internal_state.get_session(&session_id).await {
                                                        session_info.config_options = Some(opts.clone());
                                                        internal_state.upsert_session(session_info).await;
                                                    }

                                                    emit_event(
                                                        &sharing_bus,
                                                        &events_tx,
                                                        &id,
                                                        connector_uid,
                                                        Event::SessionMetadataReceived {
                                                        connector_id: id.clone(),
                                                        session_id: session_id.clone(),
                                                        models: None,
                                                        modes: None,
                                                        config_options,
                                                    },
                                                    )
                                                    .await;
                                                }
                                            }
                                        }
                                    }
                                    Err(e) => {
                                        error!(
                                            connector_id = %id,
                                            session_id = %session_id,
                                            model_id = %model_id,
                                            error = %e,
                                            "Failed to change model"
                                        );

                                        emit_event(
                                            &sharing_bus,
                                            &events_tx,
                                            &id,
                                            connector_uid,
                                            Event::SessionError {
                                            connector_id: id.clone(),
                                            session_id: session_id.clone(),
                                            error_message: format!("Failed to set model: {}", e),
                                            is_recoverable: true,
                                            error_code: Some("SET_MODEL_FAILED".to_string()),
                                            technical_details: Some(format!("{:?}", e)),
                                            context: Some(serde_json::json!({
                                                "operation": "session/set_model",
                                                "model_id": model_id,
                                            })),
                                        },
                                        )
                                        .await;
                                    }
                                }
                            }

                            Some(ConnectorCommand::SetConfigOption { session_id, config_id, value }) => {
                                debug!(connector_id = %id, session_id = %session_id, config_id = %config_id, value = %value, "Processing SetConfigOption command");

                                let request = protocol::build_set_config_option_request(&session_id, &config_id, &value);
                                match Self::send_request(
                                    &mut transport,
                                    &protocol_handler,
                                    request,
                                    Some(&events_tx), Some(&sharing_bus), connector_uid,
                                    &id,
                                    Some(&pending_agent_requests),
                                    Some(&mut cmd_rx),
                                    Some(&internal_state),
                                ).await {
                                    Ok(response) => {
                                        info!(connector_id = %id, session_id = %session_id, config_id = %config_id, "Config option set successfully");

                                        // The response contains the complete updated configOptions array
                                        if let Some(result) = response.get("result") {
                                            let config_options: Option<Vec<dirigent_protocol::ConfigOption>> = result
                                                .get("configOptions")
                                                .and_then(|co| serde_json::from_value(co.clone()).ok());

                                            if let Some(ref opts) = config_options {
                                                // Update internal state
                                                if let Some(mut session_info) = internal_state.get_session(&session_id).await {
                                                    session_info.config_options = Some(opts.clone());
                                                    internal_state.upsert_session(session_info).await;
                                                }

                                                // Emit SessionMetadataReceived with updated config options
                                                emit_event(
                                                    &sharing_bus,
                                                    &events_tx,
                                                    &id,
                                                    connector_uid,
                                                    Event::SessionMetadataReceived {
                                                    connector_id: id.clone(),
                                                    session_id: session_id.clone(),
                                                    models: None,
                                                    modes: None,
                                                    config_options,
                                                },
                                                )
                                                .await;
                                            }
                                        }
                                    }
                                    Err(e) => {
                                        warn!(connector_id = %id, session_id = %session_id, error = %e, "Failed to set config option");
                                        emit_event(
                                            &sharing_bus,
                                            &events_tx,
                                            &id,
                                            connector_uid,
                                            Event::Error {
                                            message: format!("Failed to set config option: {}", e),
                                        },
                                        )
                                        .await;
                                    }
                                }
                            }

                            Some(ConnectorCommand::CloseSession { session_id }) => {
                                debug!(connector_id = %id, session_id = %session_id, "Processing CloseSession command");

                                let supports_close = internal_state.agent_supports_session_close().await;
                                if !supports_close {
                                    warn!(connector_id = %id, "Agent does not support session/close");
                                    emit_event(
                                        &sharing_bus,
                                        &events_tx,
                                        &id,
                                        connector_uid,
                                        Event::Error {
                                        message: "Agent does not support session close".to_string(),
                                    },
                                    )
                                    .await;
                                } else {
                                    let request = protocol::build_session_close_request(&session_id);
                                    match Self::send_request(
                                        &mut transport, &protocol_handler, request,
                                        Some(&events_tx), Some(&sharing_bus), connector_uid, &id, Some(&pending_agent_requests),
                                        Some(&mut cmd_rx), Some(&internal_state),
                                    ).await {
                                        Ok(_response) => {
                                            info!(connector_id = %id, session_id = %session_id, "Session closed successfully");

                                            // Remove from loaded sessions tracking
                                            internal_state.unmark_session_loaded(&session_id).await;

                                            // Update internal state — mark session as idle
                                            if let Some(mut session_info) = internal_state.get_session(&session_id).await {
                                                session_info.status = SessionStatus::Idle;
                                                internal_state.upsert_session(session_info).await;
                                            }

                                            // Emit SessionClosed event
                                            emit_event(
                                                &sharing_bus,
                                                &events_tx,
                                                &id,
                                                connector_uid,
                                                Event::SessionClosed {
                                                connector_id: id.clone(),
                                                session_id: session_id.clone(),
                                            },
                                            )
                                            .await;
                                        }
                                        Err(e) => {
                                            warn!(connector_id = %id, session_id = %session_id, error = %e, "Failed to close session");
                                            emit_event(
                                                &sharing_bus,
                                                &events_tx,
                                                &id,
                                                connector_uid,
                                                Event::Error {
                                                message: format!("Failed to close session: {}", e),
                                            },
                                            )
                                            .await;
                                        }
                                    }
                                }
                            }

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

                                // Close transport
                                let _ = transport.close().await;

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

                                // Update state to Stopped
                                {
                                    let mut state_guard = state.write().await;
                                    *state_guard = ConnectorState::Stopped;
                                }
                                emit_event(
                                    &sharing_bus,
                                    &events_tx,
                                    &id,
                                    connector_uid,
                                    Event::ConnectorStateChanged {
                                    connector_id: id.clone(),
                                    state: "Offline".to_string(),
                                    error_kind: None,
                                },
                                )
                                .await;
                                // Exit both loops
                                break 'reconnect_loop;
                            }

                            None => {
                                error!(connector_id = %id, "Command channel closed, stopping connector");

                                // Close transport
                                let _ = transport.close().await;

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

                                // Update state to Stopped
                                {
                                    let mut state_guard = state.write().await;
                                    *state_guard = ConnectorState::Stopped;
                                }
                                emit_event(
                                    &sharing_bus,
                                    &events_tx,
                                    &id,
                                    connector_uid,
                                    Event::ConnectorStateChanged {
                                    connector_id: id.clone(),
                                    state: "Offline".to_string(),
                                    error_kind: None,
                                },
                                )
                                .await;
                                // Exit both loops
                                break 'reconnect_loop;
                            }
                        }
                    }

                    // PRIORITY 4: Housekeeping (idle detection and timeout cleanup)
                    _ = idle_check_interval.tick() => {
                        // Check for sessions that should emit SessionIdle
                        let idled = check_idle_sessions(&session_states, &events_tx, &sharing_bus, connector_uid, &id).await;

                        // Update inspector state for sessions that just went idle
                        #[cfg(feature = "server")]
                        if !idled.is_empty() {
                            if let Some(ref inspector) = inspector {
                                for session_id in &idled {
                                    inspector_update_session_state(
                                        inspector,
                                        &id,
                                        session_id,
                                        dirigent_inspector::NodeState::Idle,
                                    ).await;
                                }
                            }
                        }
                        // TODO: Task 02 cleanup_timed_out_requests() can be added here if needed
                    }
                }
            }

            // If we exited the event loop (not due to Shutdown), attempt reconnect
            info!(connector_id = %id, "Event loop exited, attempting reconnect");
        }

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

    /// Create transport based on configuration
    async fn create_transport(
        config: &AcpConfig,
        #[cfg(feature = "server")] process_manager: Option<&Arc<dyn dirigent_process::ProcessGroupManager>>,
    ) -> AcpResult<Box<dyn AcpTransport>> {
        match &config.transport {
            TransportKind::Stdio {
                command,
                args,
                cwd,
                env,
            } => {
                let mut transport = StdioTransport::new(command, args);

                // Set working directory if specified
                if let Some(cwd_path) = cwd {
                    transport.set_cwd(cwd_path.clone());
                }

                // Set environment variables if specified
                for (key, value) in env {
                    transport.set_env(key.clone(), value.clone());
                }

                // Set protocol logging directory: explicit config > data_dir/logs/acp/
                let log_dir = config.acp_log_dir.clone().or_else(|| {
                    #[cfg(feature = "server")]
                    {
                        Some(dirigent_config::DirigentPaths::resolve()
                            .map(|p| p.logs_dir().join("acp"))
                            .unwrap_or_else(|_| PathBuf::from("logs/acp")))
                    }
                    #[cfg(not(feature = "server"))]
                    { None }
                });
                if let Some(log_dir) = log_dir {
                    transport.set_log_dir(log_dir);
                }

                // Wire process lifecycle for graceful shutdown management
                #[cfg(feature = "server")]
                if let Some(mgr) = process_manager {
                    transport.set_process_lifecycle(mgr.create_lifecycle());
                }

                transport.connect().await?;
                Ok(Box::new(transport))
            }
            TransportKind::Http {
                base_url,
                timeout_ms,
            } => {
                // Generate client_id for SSE subscription
                // This ensures SSE connection can be established before RPC calls
                let client_id = uuid::Uuid::new_v4().to_string();

                info!(
                    base_url = %base_url,
                    client_id = %client_id,
                    "Generated client_id for HTTP transport"
                );

                let mut transport = HttpTransport::new(base_url, client_id);

                // Set timeout if specified
                if let Some(timeout) = timeout_ms {
                    transport.set_timeout(Duration::from_millis(*timeout));
                }

                transport.connect().await?;
                Ok(Box::new(transport))
            }
        }
    }

    /// Perform ACP initialization handshake
    async fn initialize_acp(
        id: &str,
        transport: &mut Box<dyn AcpTransport>,
        protocol_handler: &ProtocolHandler,
        config: &AcpConfig,
        internal_state: &Arc<InternalState>,
    ) -> AcpResult<()> {
        // Build initialize request with capabilities from default ownership
        let capabilities = config.default_ownership.capabilities_for_agent();
        let request = build_initialize_request(config.protocol_version, capabilities);

        // Send request and wait for response
        // Note: During initialization, we don't have access to event channels, pending requests, cmd_rx, or internal_state,
        // so we pass None for those parameters. Agent requests during init are unexpected.
        let response = Self::send_request(
            transport,
            protocol_handler,
            request,
            None,
            None,
            None,
            "init",
            None,
            None,
            None,
        )
        .await?;

        // Extract result
        let result = response
            .get("result")
            .ok_or_else(|| AcpError::protocol("Initialize response missing result field"))?;

        // Extract protocol version
        let protocol_version = result
            .get("protocolVersion")
            .and_then(|v| v.as_u64())
            .ok_or_else(|| AcpError::protocol("Initialize result missing protocolVersion"))?
            as u32;

        // Extract agent capabilities
        let capabilities = result
            .get("agentCapabilities")
            .cloned()
            .unwrap_or(json!({}));

        info!(
            connector_id = %id,
            protocol_version = protocol_version,
            "ACP initialization complete"
        );

        // Update internal state
        internal_state.set_protocol_version(protocol_version).await;
        internal_state.set_agent_capabilities(capabilities).await;

        Ok(())
    }

    /// Send a request via transport with protocol handler correlation
    ///
    /// This method properly integrates the protocol handler for request/response correlation.
    /// It prepares the request with a unique ID, sends it via transport, and waits for the
    /// correlated response with timeout.
    ///
    /// **IMPORTANT**: This method concurrently receives messages from the transport while waiting
    /// for the response, avoiding deadlock when the transport requires active receiving.
    async fn send_request(
        transport: &mut Box<dyn AcpTransport>,
        protocol_handler: &ProtocolHandler,
        request: Value,
        events_sender: Option<&broadcast::Sender<Event>>,
        sharing_bus: Option<&Arc<SharingBus>>,
        connector_uid: Option<Uuid>,
        connector_id: &str,
        pending_agent_requests: Option<&Arc<Mutex<HashSet<String>>>>,
        mut cmd_rx: Option<&mut mpsc::Receiver<ConnectorCommand>>,
        internal_state: Option<&Arc<InternalState>>,
    ) -> AcpResult<Value> {
        // Prepare request (adds ID and creates response channel)
        let (message_with_id, mut response_rx) = protocol_handler.prepare_request(request).await;

        let request_id = message_with_id.get("id").cloned().unwrap_or(json!(null));
        debug!(request_id = %request_id, "Prepared request with ID");

        // Send via transport
        let masked_message = dirigent_protocol::log_utils::format_for_log(&message_with_id);
        debug!(request_id = %request_id, message = %masked_message, "Calling transport.send()");
        transport.send(message_with_id).await?;
        debug!(request_id = %request_id, "transport.send() completed successfully");

        // Wait for correlated response while also receiving messages from transport
        // This prevents deadlock when the agent sends the response immediately
        // Timeout set to 5 minutes to allow for tool use, thinking, and complex operations
        let timeout_duration = Duration::from_millis(300_000);
        let deadline = tokio::time::Instant::now() + timeout_duration;

        loop {
            tokio::select! {
                // Check if we got the response via protocol handler
                response = &mut response_rx => {
                    match response {
                        Ok(response) => {
                            // Check for JSON-RPC error in response
                            if let Some(error) = response.get("error") {
                                let code = error.get("code").and_then(|c| c.as_i64()).unwrap_or(-1);
                                let message = error
                                    .get("message")
                                    .and_then(|m| m.as_str())
                                    .unwrap_or("Unknown error")
                                    .to_string();
                                let data = error.get("data").cloned();

                                return Err(AcpError::AgentError {
                                    code,
                                    message,
                                    data,
                                });
                            }

                            return Ok(response);
                        }
                        Err(_) => {
                            return Err(AcpError::internal("Response channel dropped"));
                        }
                    }
                }

                // Continue receiving messages from transport and route them
                message = transport.recv() => {
                    match message {
                        Ok(Some(msg)) => {
                            // Route through protocol handler (may complete response_rx)
                            use crate::connectors::acp::protocol::MessageHandlerResult;
                            match protocol_handler.handle_message(msg).await {
                                MessageHandlerResult::None => {
                                    // Notification processed, continue
                                }
                                MessageHandlerResult::Response(response) => {
                                    debug!("📤 Sending response to agent request during send_request");
                                    if let Err(e) = transport.send(response).await {
                                        error!(error = %e, "Failed to send response to agent during send_request");
                                    }
                                }
                                MessageHandlerResult::AgentRequest { request_id, method, params } => {
                                    // Agent request during send_request - handle it if we have the channels
                                    if let (Some(events_tx), Some(pending_reqs)) = (events_sender, pending_agent_requests) {
                                        info!(
                                            connector_id = connector_id,
                                            request_id = %request_id,
                                            method = %method,
                                            "Received agent request during send_request, emitting event"
                                        );

                                        // Extract session_id from params (standard ACP location)
                                        let session_id = params
                                            .get("sessionId")
                                            .and_then(|v| v.as_str())
                                            .unwrap_or("unknown")
                                            .to_string();

                                        // Determine if this is a forwarded (external) session
                                        let is_forwarded = if let Some(state) = internal_state {
                                            if let Some(session_info) = state.get_session(&session_id).await {
                                                session_info.ownership.is_external()
                                            } else {
                                                warn!(
                                                    connector_id = connector_id,
                                                    session_id = %session_id,
                                                    "Session not found when processing AgentRequest in send_request - defaulting to internal"
                                                );
                                                false
                                            }
                                        } else {
                                            // No internal_state provided (e.g., during initialization) - default to internal
                                            false
                                        };

                                        // Store the pending request (no timeout - wait indefinitely for user)
                                        let request_id_str = request_id.to_string();
                                        {
                                            let mut pending = pending_reqs.lock().await;
                                            pending.insert(request_id_str.clone());
                                        }
                                        info!(
                                            connector_id = connector_id,
                                            request_id = %request_id_str,
                                            is_forwarded = is_forwarded,
                                            "Stored pending agent request (waiting for user response)"
                                        );

                                        // Emit Event::AgentRequest for routing
                                        // - If is_forwarded=true: EventBridge forwards to external client
                                        // - If is_forwarded=false: Web UI shows permission modal
                                        let event = Event::AgentRequest {
                                            connector_id: connector_id.to_string(),
                                            session_id,
                                            request_id: request_id.clone(),
                                            method,
                                            params,
                                            is_forwarded,
                                        };

                                        if let Some(bus) = sharing_bus {
                                            let bus_event = dirigent_protocol::streaming::BusEvent::from_connector_event(
                                                event.clone(),
                                                connector_uid,
                                                connector_id.to_string(),
                                            );
                                            bus.publish(bus_event).await;
                                        }
                                        if let Err(e) = events_tx.send(event) {
                                            error!(
                                                connector_id = connector_id,
                                                error = %e,
                                                "Failed to emit AgentRequest event"
                                            );
                                        }

                                        // Response will be handled when AgentResponse command is received
                                        // No timeout - the system waits indefinitely for user input
                                    } else {
                                        warn!(
                                            connector_id = connector_id,
                                            "Received agent request during send_request, but no event channel available (ignoring)"
                                        );
                                    }
                                }
                            }
                            // Continue loop to check response_rx again
                        }
                        Ok(None) => {
                            return Err(AcpError::transport("Transport closed while waiting for response"));
                        }
                        Err(e) => {
                            return Err(AcpError::transport(format!("Transport error: {}", e)));
                        }
                    }
                }

                // Handle incoming commands (specifically AgentResponse to avoid deadlock)
                cmd = async {
                    if let Some(ref mut rx) = cmd_rx {
                        rx.recv().await
                    } else {
                        std::future::pending().await // Never resolves if no cmd_rx
                    }
                } => {
                    match cmd {
                        Some(ConnectorCommand::AgentResponse { request_id, response }) => {
                            info!(
                                connector_id = connector_id,
                                request_id = %request_id,
                                "Received AgentResponse command during send_request"
                            );

                            // Convert request_id to string for HashSet lookup
                            let request_id_str = request_id.to_string();

                            // Check if this request is still pending and remove it
                            if let Some(pending_reqs) = pending_agent_requests {
                                let was_pending = {
                                    let mut pending = pending_reqs.lock().await;
                                    pending.remove(&request_id_str)
                                };

                                if was_pending {
                                    // Send response directly to agent via transport
                                    info!(
                                        connector_id = connector_id,
                                        request_id = %request_id_str,
                                        "Sending agent response to transport during send_request"
                                    );

                                    if let Err(e) = transport.send(response).await {
                                        error!(
                                            connector_id = connector_id,
                                            request_id = %request_id_str,
                                            error = %e,
                                            "Failed to send agent response to transport during send_request"
                                        );
                                    } else {
                                        info!(
                                            connector_id = connector_id,
                                            request_id = %request_id_str,
                                            "Successfully sent agent response to transport during send_request"
                                        );
                                    }
                                } else {
                                    warn!(
                                        connector_id = connector_id,
                                        request_id = %request_id_str,
                                        "Received AgentResponse for unknown request_id during send_request"
                                    );
                                }
                            }
                            // Continue loop to check response_rx again
                        }
                        Some(other_cmd) => {
                            warn!(
                                connector_id = connector_id,
                                "Received non-AgentResponse command during send_request (ignoring): {:?}",
                                std::mem::discriminant(&other_cmd)
                            );
                            // Put the command back or drop it - for now we'll drop it
                            // In the future, we might want to buffer these for later processing
                        }
                        None => {
                            warn!(
                                connector_id = connector_id,
                                "Command channel closed during send_request"
                            );
                            // Continue waiting for response
                        }
                    }
                }

                // Check for timeout
                _ = tokio::time::sleep_until(deadline) => {
                    return Err(AcpError::timeout(timeout_duration.as_millis() as u64));
                }
            }
        }
    }

    /// Execute session listing: fetch from upstream agent if supported, fall back to internal state.
    ///
    /// Emits `SessionsListed` (and optionally `SessionMetadataReceived` per session).
    /// Called by both the `ListSessions` command handler and automatically after
    /// `CreateSession` to populate the session list for deferred obligations.
    async fn execute_list_sessions(
        transport: &mut Box<dyn AcpTransport>,
        protocol_handler: &ProtocolHandler,
        events_tx: &broadcast::Sender<Event>,
        sharing_bus: &Arc<SharingBus>,
        connector_uid: Option<Uuid>,
        connector_id: &str,
        config: &AcpConfig,
        internal_state: &Arc<InternalState>,
        pending_agent_requests: &Arc<Mutex<HashSet<String>>>,
        cmd_rx: &mut mpsc::Receiver<ConnectorCommand>,
    ) {
        debug!(connector_id = %connector_id, "Executing ListSessions");

        // Check if the upstream agent supports session/list
        let supports_list = internal_state.agent_supports_list_sessions().await;

        // Try to fetch from upstream if supported
        let upstream_sessions: Option<Vec<SessionInfo>> = if supports_list {
            let mut all_upstream: Vec<serde_json::Value> = Vec::new();
            let mut cursor: Option<String> = None;
            let mut fetch_ok = true;

            loop {
                let request = protocol::build_session_list_request(
                    Some(&config.cwd),
                    cursor.as_deref(),
                );

                match Self::send_request(
                    transport,
                    protocol_handler,
                    request,
                    Some(events_tx),
                    Some(sharing_bus),
                    connector_uid,
                    connector_id,
                    Some(pending_agent_requests),
                    Some(cmd_rx),
                    Some(internal_state),
                )
                .await
                {
                    Ok(response) => {
                        if let Some(result) = response.get("result") {
                            if let Some(sessions) =
                                result.get("sessions").and_then(|s| s.as_array())
                            {
                                all_upstream.extend(sessions.iter().cloned());
                            }
                            cursor = result
                                .get("nextCursor")
                                .and_then(|c| c.as_str())
                                .map(|s| s.to_string());
                            if cursor.is_none() {
                                break;
                            }
                        } else {
                            warn!(connector_id = %connector_id, "session/list response missing result");
                            fetch_ok = false;
                            break;
                        }
                    }
                    Err(e) => {
                        warn!(connector_id = %connector_id, error = %e, "session/list failed, falling back to internal state");
                        fetch_ok = false;
                        break;
                    }
                }
            }

            if fetch_ok {
                // Parse upstream sessions into SessionInfo
                let mut session_map: HashMap<String, SessionInfo> = HashMap::new();

                for session_json in &all_upstream {
                    let session_id = session_json
                        .get("sessionId")
                        .and_then(|v| v.as_str())
                        .unwrap_or_default()
                        .to_string();

                    if session_id.is_empty() {
                        continue;
                    }

                    let title = session_json
                        .get("title")
                        .and_then(|v| v.as_str())
                        .map(|s| s.to_string());

                    let cwd = session_json
                        .get("cwd")
                        .and_then(|v| v.as_str())
                        .unwrap_or(&config.cwd)
                        .to_string();

                    let updated_at = session_json
                        .get("updatedAt")
                        .and_then(|v| v.as_str())
                        .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
                        .map(|dt| dt.with_timezone(&Utc))
                        .unwrap_or_else(Utc::now);

                    let meta = session_json.get("_meta");
                    let message_count = meta
                        .and_then(|m| m.get("messageCount"))
                        .and_then(|v| v.as_u64())
                        .unwrap_or(0) as u32;

                    let info = SessionInfo {
                        id: session_id.clone(),
                        title,
                        cwd,
                        message_count,
                        model: None,
                        created_at: updated_at,
                        last_activity: updated_at,
                        status: SessionStatus::Idle,
                        models: None,
                        modes: None,
                        config_options: None,
                        ownership: SessionOwnership::default(),
                    };

                    session_map.insert(session_id, info);
                }

                // Union merge: add internal state sessions that aren't already upstream
                let internal_sessions = internal_state.list_sessions().await;
                for internal in internal_sessions {
                    session_map.entry(internal.id.clone()).or_insert(internal);
                }

                // Upsert newly discovered sessions into internal state
                for info in session_map.values() {
                    internal_state.upsert_session(info.clone()).await;
                }

                info!(connector_id = %connector_id, count = session_map.len(), "Listed sessions from upstream agent (merged with internal state)");
                Some(session_map.into_values().collect())
            } else {
                None
            }
        } else {
            None
        };

        // Use upstream sessions or fall back to internal state
        let sessions = match upstream_sessions {
            Some(s) => s,
            None => internal_state.list_sessions().await,
        };

        // Convert SessionInfo to protocol::Session and emit metadata events
        let mut protocol_sessions: Vec<Session> = Vec::with_capacity(sessions.len());

        for info in sessions {
            let current_mode_id = info.modes.as_ref().map(|m| m.current_mode_id.clone());

            protocol_sessions.push(Session {
                id: info.id.clone(),
                title: info
                    .title
                    .clone()
                    .unwrap_or_else(|| "Untitled Session".to_string()),
                created_at: info.created_at,
                updated_at: info.last_activity,
                cwd: Some(info.cwd.clone()),
                metadata: SessionMetadata {
                    project_path: info.cwd,
                    model: info.model,
                    total_messages: info.message_count,
                    system_message: None,
                    current_mode_id,
                    _meta: None,
                    project_id: None,
                },
                models: info.models.clone(),
                modes: info.modes.clone(),
                config_options: None,
                acp_client_id: None,
            });

            if info.models.is_some() || info.modes.is_some() || info.config_options.is_some() {
                emit_event(
                    sharing_bus,
                    events_tx,
                    &connector_id.to_string(),
                    connector_uid,
                    Event::SessionMetadataReceived {
                        connector_id: connector_id.to_string(),
                        session_id: info.id,
                        models: info.models,
                        modes: info.modes,
                        config_options: info.config_options,
                    },
                )
                .await;
            }
        }

        emit_event(
            sharing_bus,
            events_tx,
            &connector_id.to_string(),
            connector_uid,
            Event::SessionsListed {
                connector_id: connector_id.to_string(),
                sessions: protocol_sessions,
            },
        )
        .await;
    }

    /// Prepare request without sending (for use in protocol handler)
    // TODO: Request preparation - utility for building ACP requests with validation and defaults
    #[allow(dead_code)]
    async fn prepare_request(
        protocol_handler: &ProtocolHandler,
        request: Value,
    ) -> (Value, tokio::sync::oneshot::Receiver<Value>) {
        protocol_handler.prepare_request(request).await
    }

    /// Handle a notification from the agent using ACP adapter
    ///
    /// Translates ACP JSON-RPC notifications to Dirigent protocol events.
    async fn handle_notification(
        id: &str,
        notification: Value,
        events_tx: &broadcast::Sender<Event>,
        sharing_bus: &Arc<SharingBus>,
        connector_uid: Option<Uuid>,
        adapter: &dirigent_protocol::adapters::AcpAdapter,
        internal_state: &Arc<InternalState>,
        session_states: &Arc<Mutex<HashMap<String, SessionState>>>,
        tool_configuration: &Option<crate::tools::ToolConfiguration>,
    ) -> AcpResult<()> {
        // Extract session ID and update activity timestamp
        if let Some(params) = notification.get("params") {
            if let Some(notif_session_id) = params.get("sessionId").and_then(|s| s.as_str()) {
                debug_log!(
                    "🔔 Notification sessionId: {} (method: {})",
                    notif_session_id,
                    notification
                        .get("method")
                        .and_then(|m| m.as_str())
                        .unwrap_or("unknown")
                );

                // Update session activity timestamp
                {
                    let mut states = session_states.lock().await;
                    if let Some(state) = states.get_mut(notif_session_id) {
                        state.touch();
                        trace!(session_id = %notif_session_id, "Updated session activity timestamp");
                    }
                }
            }
        }
        let masked_notification = dirigent_protocol::log_utils::format_for_log(&notification);
        info!(connector_id = %id, notification = %masked_notification, "🔄 Processing ACP notification in handle_notification");

        // Save notification data for fallback before it's moved
        let notification_clone = notification.clone();

        // Translate notification to Dirigent event using adapter
        match adapter.translate_notification(notification).await {
            Ok(event) => {
                info!(connector_id = %id, "✅ Successfully translated notification to event");

                // Replace ACP message_id with Dirigent message_id for streaming chunks
                // and inject connector_id
                let event = match event {
                    Event::SessionUpdate {
                        connector_id: _,
                        session_id,
                        update,
                    } => {
                        match update {
                            SessionUpdate::AgentMessageChunk {
                                message_id: acp_message_id,
                                content,
                                _meta,
                            } => {
                                // Get or create the Dirigent message_id for this session
                                let dirigent_message_id = internal_state
                                    .get_or_create_active_message(&session_id)
                                    .await;

                                debug_log!(
                                    "🔄 Message ID translation: ACP={} → Dirigent={}",
                                    acp_message_id,
                                    dirigent_message_id
                                );
                                info!(
                                    connector_id = %id,
                                    session_id = %session_id,
                                    acp_message_id = %acp_message_id,
                                    dirigent_message_id = %dirigent_message_id,
                                    "🔄 Translating ACP message_id to Dirigent message_id"
                                );

                                // Store the original ACP message_id in metadata
                                let mut extra = HashMap::new();
                                extra.insert("acp_message_id".to_string(), json!(acp_message_id));
                                let meta = Meta {
                                    provider: None,
                                    extra,
                                };

                                Event::SessionUpdate {
                                    connector_id: id.to_string(),
                                    session_id,
                                    update: SessionUpdate::AgentMessageChunk {
                                        message_id: dirigent_message_id,
                                        content,
                                        _meta: Some(meta),
                                    },
                                }
                            }
                            SessionUpdate::ToolCall {
                                message_id: acp_message_id,
                                mut tool_call,
                                _meta,
                            } => {
                                // Use same Dirigent message_id as surrounding chunks
                                let dirigent_message_id = internal_state
                                    .get_or_create_active_message(&session_id)
                                    .await;

                                // Mark as external - Claude Code executes these tools
                                tool_call.origin = Some(ToolOrigin::External);

                                debug_log!(
                                    "🔄 ToolCall message ID translation: ACP={} → Dirigent={}",
                                    acp_message_id,
                                    dirigent_message_id
                                );
                                info!(
                                    connector_id = %id,
                                    session_id = %session_id,
                                    acp_message_id = %acp_message_id,
                                    dirigent_message_id = %dirigent_message_id,
                                    tool_name = %tool_call.tool_name,
                                    "🔄 Translating ToolCall message_id to Dirigent message_id"
                                );

                                Event::SessionUpdate {
                                    connector_id: id.to_string(),
                                    session_id,
                                    update: SessionUpdate::ToolCall {
                                        message_id: dirigent_message_id,
                                        tool_call,
                                        _meta,
                                    },
                                }
                            }
                            SessionUpdate::ToolCallUpdate {
                                message_id: acp_message_id,
                                tool_call_id,
                                mut tool_call,
                                _meta,
                            } => {
                                // Use same Dirigent message_id as surrounding chunks
                                let dirigent_message_id = internal_state
                                    .get_or_create_active_message(&session_id)
                                    .await;

                                // Mark as external - Claude Code executes these tools
                                tool_call.origin = Some(ToolOrigin::External);

                                debug_log!("🔄 ToolCallUpdate message ID translation: ACP={} → Dirigent={}", acp_message_id, dirigent_message_id);
                                info!(
                                    connector_id = %id,
                                    session_id = %session_id,
                                    acp_message_id = %acp_message_id,
                                    dirigent_message_id = %dirigent_message_id,
                                    tool_call_id = %tool_call_id,
                                    "🔄 Translating ToolCallUpdate message_id to Dirigent message_id"
                                );

                                Event::SessionUpdate {
                                    connector_id: id.to_string(),
                                    session_id,
                                    update: SessionUpdate::ToolCallUpdate {
                                        message_id: dirigent_message_id,
                                        tool_call_id,
                                        tool_call,
                                        _meta,
                                    },
                                }
                            }
                            // Check for config_option_update, available_commands_update, etc.
                            SessionUpdate::Unknown { ref data } => {
                                // Check the sessionUpdate type for special handling
                                let merged_event = if let Some(session_update_type) =
                                    data.get("sessionUpdate").and_then(|s| s.as_str())
                                {
                                    if session_update_type == "config_option_update" {
                                        // Parse configOptions from the notification
                                        if let Some(config_options_json) = data.get("configOptions") {
                                            if let Ok(config_options) = serde_json::from_value::<
                                                Vec<dirigent_protocol::ConfigOption>,
                                            >(config_options_json.clone()) {
                                                info!(
                                                    connector_id = %id,
                                                    session_id = %session_id,
                                                    option_count = config_options.len(),
                                                    "📊 Received config_option_update notification"
                                                );

                                                // Update internal state with new config options
                                                if let Some(mut session_info) = internal_state.get_session(&session_id).await {
                                                    session_info.config_options = Some(config_options.clone());
                                                    session_info.last_activity = Utc::now();
                                                    internal_state.upsert_session(session_info).await;
                                                }

                                                // Emit SessionMetadataReceived event with config options
                                                // This allows downstream consumers (event bridge, archivist, UI) to react
                                                emit_event(
                                                    sharing_bus,
                                                    events_tx,
                                                    &id.to_string(),
                                                    connector_uid,
                                                    Event::SessionMetadataReceived {
                                                        connector_id: id.to_string(),
                                                        session_id: session_id.clone(),
                                                        models: None,
                                                        modes: None,
                                                        config_options: Some(config_options),
                                                    },
                                                )
                                                .await;

                                                // Also pass through as Unknown for the event bridge to forward
                                                Some(Event::SessionUpdate {
                                                    connector_id: id.to_string(),
                                                    session_id: session_id.clone(),
                                                    update: SessionUpdate::Unknown { data: data.clone() },
                                                })
                                            } else {
                                                warn!(connector_id = %id, "Failed to parse configOptions from config_option_update notification");
                                                None
                                            }
                                        } else {
                                            warn!(connector_id = %id, "config_option_update notification missing configOptions field");
                                            None
                                        }
                                    } else if session_update_type == "available_commands_update" {
                                        // Extract commands from the data
                                        if let Some(commands_json) = data.get("availableCommands") {
                                            // Parse commands and merge with transient commands
                                            if let Ok(mut commands) = serde_json::from_value::<
                                                Vec<crate::acp::protocol::streaming::Command>,
                                            >(
                                                commands_json.clone()
                                            ) {
                                                info!(
                                                    connector_id = %id,
                                                    session_id = %session_id,
                                                    original_count = commands.len(),
                                                    "🔄 Intercepted available_commands_update, merging transient commands"
                                                );

                                                // Merge transient commands
                                                commands = crate::connectors::gateway::merge_with_transient_commands(commands);

                                                info!(
                                                    connector_id = %id,
                                                    session_id = %session_id,
                                                    merged_count = commands.len(),
                                                    "✅ Merged transient commands into available_commands_update"
                                                );

                                                // Reconstruct the update with merged commands
                                                let mut merged_data = data.clone();
                                                if let Some(obj) = merged_data.as_object_mut() {
                                                    obj.insert(
                                                        "availableCommands".to_string(),
                                                        serde_json::to_value(&commands)
                                                            .unwrap_or_default(),
                                                    );
                                                }

                                                Some(Event::SessionUpdate {
                                                    connector_id: id.to_string(),
                                                    session_id: session_id.clone(),
                                                    update: SessionUpdate::Unknown {
                                                        data: merged_data,
                                                    },
                                                })
                                            } else {
                                                None
                                            }
                                        } else {
                                            None
                                        }
                                    } else {
                                        None
                                    }
                                } else {
                                    None
                                };

                                // Return merged event if we processed it, otherwise pass through unchanged
                                merged_event.unwrap_or_else(|| Event::SessionUpdate {
                                    connector_id: id.to_string(),
                                    session_id,
                                    update: SessionUpdate::Unknown { data: data.clone() },
                                })
                            }
                            // Pass through all other update types unchanged, but add connector_id
                            other_update => Event::SessionUpdate {
                                connector_id: id.to_string(),
                                session_id,
                                update: other_update,
                            },
                        }
                    }
                    // Pass through all other event types unchanged
                    other_event => other_event,
                };

                // --- Tool directive intercept ---
                // Check ToolCall events against the connector's tool configuration.
                // Denied tools are marked as errors; hidden tools are silently dropped.
                let event = if let Some(ref tool_config) = tool_configuration {
                    Self::apply_tool_directive(event, tool_config, id)
                } else {
                    Some(event)
                };

                // Broadcast the event (if not hidden by tool directive)
                if let Some(event) = event {
                    debug_log!("📢 Broadcasting event: {:?}", event);
                    let bus_event = dirigent_protocol::streaming::BusEvent::from_connector_event(
                        event.clone(),
                        connector_uid,
                        id.to_string(),
                    );
                    sharing_bus.publish(bus_event).await;
                    match events_tx.send(event) {
                        Ok(receiver_count) => {
                            debug_log!("✅ Event sent to {} receivers", receiver_count);
                            info!(connector_id = %id, receiver_count = receiver_count, "📢 Event broadcast to {} receivers", receiver_count);
                        }
                        Err(e) => {
                            debug_log!("⚠️ No receivers for event");
                            warn!(connector_id = %id, error = ?e, "⚠️ No receivers for event (this is normal if no clients are listening)");
                        }
                    }
                }

                Ok(())
            }
            Err(e) => {
                // Translation failed
                match e {
                    dirigent_protocol::adapters::AcpTranslationError::Duplicate => {
                        // Duplicates are expected and should be silently skipped
                        debug!(connector_id = %id, "Skipped duplicate notification");
                        Ok(())
                    }
                    dirigent_protocol::adapters::AcpTranslationError::UnknownMethod(method) => {
                        // Unknown methods are logged at debug, not errors
                        debug!(connector_id = %id, method = method, "Unknown notification method");
                        Ok(())
                    }
                    dirigent_protocol::adapters::AcpTranslationError::InvalidValue {
                        ref field,
                        ref reason,
                    } => {
                        // Invalid values (like missing type field) are logged at debug
                        debug!(connector_id = %id, field = field, reason = reason, "Skipped notification with invalid value");
                        Ok(())
                    }
                    dirigent_protocol::adapters::AcpTranslationError::MissingField(ref field) => {
                        // Log warning when falling back to Unknown event
                        warn!(
                            connector_id = %id,
                            field = field,
                            "Missing field in notification, forwarding as Unknown"
                        );

                        // Extract session ID from notification
                        let session_id = notification_clone
                            .get("params")
                            .and_then(|p| p.get("sessionId"))
                            .and_then(|s| s.as_str())
                            .unwrap_or("unknown")
                            .to_string();

                        // Create Unknown event with raw JSON for pass-through
                        let event = Event::SessionUpdate {
                            connector_id: id.to_string(),
                            session_id,
                            update: SessionUpdate::Unknown {
                                data: notification_clone,
                            },
                        };

                        // Broadcast the event
                        debug_log!("📢 Broadcasting Unknown event: {:?}", event);
                        let bus_event = dirigent_protocol::streaming::BusEvent::from_connector_event(
                            event.clone(),
                            connector_uid,
                            id.to_string(),
                        );
                        sharing_bus.publish(bus_event).await;
                        match events_tx.send(event) {
                            Ok(receiver_count) => {
                                debug_log!("✅ Unknown event sent to {} receivers", receiver_count);
                                info!(connector_id = %id, receiver_count = receiver_count, "📢 Unknown event broadcast to {} receivers", receiver_count);
                            }
                            Err(e) => {
                                debug_log!("⚠️ No receivers for Unknown event");
                                warn!(connector_id = %id, error = ?e, "⚠️ No receivers for Unknown event (this is normal if no clients are listening)");
                            }
                        }

                        Ok(())
                    }
                }
            }
        }
    }

    /// Build a `Meta` value containing the tool configuration (if present).
    ///
    /// Used when emitting `SessionCreated` events so the archivist can persist
    /// the tool configuration in the session metadata.
    fn build_session_meta(
        tool_configuration: &Option<crate::tools::ToolConfiguration>,
    ) -> Option<Meta> {
        let tool_config = tool_configuration.as_ref()?;
        let tool_config_value = serde_json::to_value(tool_config).ok()?;
        let mut extra = HashMap::new();
        extra.insert("tool_configuration".to_string(), tool_config_value);
        Some(Meta {
            provider: None,
            extra,
        })
    }

    /// Apply tool directive intercept to an event.
    ///
    /// Checks `SessionUpdate::ToolCall` events against the connector's tool
    /// configuration:
    /// - `Deny`: marks the tool call as failed with an error message
    /// - `Hide`: returns `None` to suppress the event entirely
    /// - Other handlers or unconfigured tools: pass through unchanged
    fn apply_tool_directive(
        event: Event,
        tool_config: &crate::tools::ToolConfiguration,
        connector_id: &str,
    ) -> Option<Event> {
        use crate::tools::ToolHandler;
        use dirigent_protocol::types::ToolCallStatus;

        match event {
            Event::SessionUpdate {
                connector_id: cid,
                session_id,
                update:
                    SessionUpdate::ToolCall {
                        message_id,
                        mut tool_call,
                        _meta,
                    },
            } => {
                if tool_config.should_intercept(&tool_call.tool_name) {
                    match tool_config.active_handler(&tool_call.tool_name) {
                        Some(ToolHandler::Hide) => {
                            info!(
                                connector_id = %connector_id,
                                tool_name = %tool_call.tool_name,
                                "Tool call hidden by directive"
                            );
                            return None;
                        }
                        Some(ToolHandler::Deny) => {
                            info!(
                                connector_id = %connector_id,
                                tool_name = %tool_call.tool_name,
                                "Tool call denied by directive"
                            );
                            tool_call.status = ToolCallStatus::Error;
                            tool_call.error =
                                Some("Tool denied by directive".to_string());
                        }
                        _ => {
                            // Other interceptable handlers (Dirigent, Editor,
                            // Plugin) pass through for now -- future tasks will
                            // route them to the appropriate handler.
                        }
                    }
                }
                Some(Event::SessionUpdate {
                    connector_id: cid,
                    session_id,
                    update: SessionUpdate::ToolCall {
                        message_id,
                        tool_call,
                        _meta,
                    },
                })
            }
            // All other events pass through unchanged.
            other => Some(other),
        }
    }

    /// Build a crash reason string from transport crash context.
    async fn build_crash_reason(transport: &Box<dyn AcpTransport>, default_reason: &str) -> String {
        if let Some(ctx) = transport.get_crash_context().await {
            let mut reason = "Connection lost: agent process exited".to_string();
            if let Some(status) = ctx.exit_status {
                reason.push_str(&format!(" ({})", status));
            }
            if !ctx.recent_stderr.is_empty() {
                let last_lines: Vec<&str> = ctx.recent_stderr
                    .iter()
                    .rev()
                    .take(5)
                    .map(|s| s.as_str())
                    .collect::<Vec<_>>()
                    .into_iter()
                    .rev()
                    .collect();
                reason.push_str(&format!(". Last stderr: {}", last_lines.join(" | ")));
            }
            reason
        } else {
            default_reason.to_string()
        }
    }
}

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

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

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

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

    fn state(&self) -> ConnectorState {
        // Try to read the state without blocking
        match self.state.try_read() {
            Ok(state_guard) => state_guard.clone(),
            Err(_) => {
                // Lock is held, return Initializing as a safe default
                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) {
        // Send shutdown command
        let cmd_tx = self.cmd_tx.clone();
        tokio::spawn(async move {
            let _ = cmd_tx.send(ConnectorCommand::Shutdown).await;
        });
    }
}

// Note: ProtocolHandler's generate_request_id is private,
// so we'll use the pattern from the protocol layer properly

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

    #[tokio::test]
    async fn test_acp_connector_creation() {
        let config = AcpConfig::stdio("test-agent", vec!["--stdio".to_string()]);

        let connector = AcpConnector::new(
            "test-conn".to_string(),
            uuid::Uuid::nil(),
            "Test ACP Connector".to_string(),
            config,
            SharingBus::new(),
        )
        .unwrap();

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

    #[tokio::test]
    async fn test_acp_connector_invalid_config() {
        let config = AcpConfig::stdio("", vec![]); // Empty command

        let result = AcpConnector::new(
            "test-conn".to_string(),
            uuid::Uuid::nil(),
            "Test ACP Connector".to_string(),
            config,
            SharingBus::new(),
        );

        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_acp_connector_command_tx() {
        let config = AcpConfig::stdio("test-agent", vec![]);

        let connector = AcpConnector::new(
            "test-conn".to_string(),
            uuid::Uuid::nil(),
            "Test ACP Connector".to_string(),
            config,
            SharingBus::new(),
        )
        .unwrap();

        // Should be able to clone the command sender
        let cmd_tx1 = connector.command_tx();
        let cmd_tx2 = connector.command_tx();

        drop(cmd_tx1);
        drop(cmd_tx2);
    }

    #[tokio::test]
    async fn test_acp_connector_subscribe() {
        let config = AcpConfig::stdio("test-agent", vec![]);

        let connector = AcpConnector::new(
            "test-conn".to_string(),
            uuid::Uuid::nil(),
            "Test ACP Connector".to_string(),
            config,
            SharingBus::new(),
        )
        .unwrap();

        // Should be able to create multiple subscriptions
        let _rx1 = connector.subscribe();
        let _rx2 = connector.subscribe();
    }

    // =====================================================================
    // Tool directive intercept tests
    // =====================================================================

    mod tool_directive_tests {
        use crate::tools::{ToolConfiguration, ToolDirective, ToolHandler};
        use dirigent_protocol::types::{ToolCall, ToolCallStatus, ToolOrigin};
        use dirigent_protocol::{Event, SessionUpdate};

        use super::AcpConnector;

        /// Helper to create a ToolCall event for testing.
        fn make_tool_call_event(tool_name: &str) -> Event {
            Event::SessionUpdate {
                connector_id: "test-conn".to_string(),
                session_id: "sess-1".to_string(),
                update: SessionUpdate::ToolCall {
                    message_id: "msg-1".to_string(),
                    tool_call: ToolCall {
                        id: "tc-1".to_string(),
                        tool_name: tool_name.to_string(),
                        status: ToolCallStatus::Running,
                        content: vec![],
                        raw_input: None,
                        raw_output: None,
                        title: None,
                        error: None,
                        metadata: None,
                        origin: Some(ToolOrigin::External),
                    },
                    _meta: None,
                },
            }
        }

        #[test]
        fn tool_directive_deny_marks_error() {
            let mut config = ToolConfiguration::new();
            config.set(ToolDirective::checked("read", ToolHandler::Deny));

            let event = make_tool_call_event("read");
            let result = AcpConnector::apply_tool_directive(event, &config, "test-conn");

            let result = result.expect("Deny should still emit an event");
            if let Event::SessionUpdate {
                update: SessionUpdate::ToolCall { tool_call, .. },
                ..
            } = result
            {
                assert_eq!(tool_call.status, ToolCallStatus::Error);
                assert_eq!(
                    tool_call.error.as_deref(),
                    Some("Tool denied by directive")
                );
            } else {
                panic!("Expected SessionUpdate::ToolCall event");
            }
        }

        #[test]
        fn tool_directive_hide_suppresses_event() {
            let mut config = ToolConfiguration::new();
            config.set(ToolDirective::checked("write", ToolHandler::Hide));

            let event = make_tool_call_event("write");
            let result = AcpConnector::apply_tool_directive(event, &config, "test-conn");

            assert!(result.is_none(), "Hide should suppress the event entirely");
        }

        #[test]
        fn tool_directive_unchecked_passes_through() {
            let mut config = ToolConfiguration::new();
            config.set(ToolDirective::passthrough("bash"));

            let event = make_tool_call_event("bash");
            let result = AcpConnector::apply_tool_directive(event, &config, "test-conn");

            assert!(result.is_some(), "Passthrough should not suppress the event");
            if let Some(Event::SessionUpdate {
                update: SessionUpdate::ToolCall { tool_call, .. },
                ..
            }) = result
            {
                assert_eq!(tool_call.status, ToolCallStatus::Running);
                assert!(tool_call.error.is_none());
            } else {
                panic!("Expected SessionUpdate::ToolCall event");
            }
        }

        #[test]
        fn tool_directive_unknown_tool_passes_through() {
            let config = ToolConfiguration::new(); // Empty config

            let event = make_tool_call_event("unknown_tool");
            let result = AcpConnector::apply_tool_directive(event, &config, "test-conn");

            assert!(
                result.is_some(),
                "Unknown tools should pass through unmodified"
            );
        }

        #[test]
        fn tool_directive_non_tool_event_passes_through() {
            let config = ToolConfiguration::new();

            // A non-ToolCall event should pass through unchanged
            let event = Event::SessionUpdate {
                connector_id: "test-conn".to_string(),
                session_id: "sess-1".to_string(),
                update: SessionUpdate::AgentMessageChunk {
                    message_id: "msg-1".to_string(),
                    content: dirigent_protocol::ContentBlock::Text {
                        text: "hello".to_string(),
                    },
                    _meta: None,
                },
            };
            let result = AcpConnector::apply_tool_directive(event, &config, "test-conn");

            assert!(
                result.is_some(),
                "Non-ToolCall events should pass through"
            );
        }

        #[test]
        fn tool_directive_agent_handler_not_intercepted() {
            let mut config = ToolConfiguration::new();
            // Agent handler with checked=true should NOT be intercepted
            config.set(ToolDirective::checked("file_write", ToolHandler::Agent));

            let event = make_tool_call_event("file_write");
            let result = AcpConnector::apply_tool_directive(event, &config, "test-conn");

            assert!(
                result.is_some(),
                "Agent handler should not intercept"
            );
            if let Some(Event::SessionUpdate {
                update: SessionUpdate::ToolCall { tool_call, .. },
                ..
            }) = result
            {
                assert_eq!(tool_call.status, ToolCallStatus::Running);
                assert!(tool_call.error.is_none());
            }
        }

        #[test]
        fn build_session_meta_with_tool_config() {
            let mut config = ToolConfiguration::new();
            config.set(ToolDirective::checked("shell_exec", ToolHandler::Deny));

            let meta = AcpConnector::build_session_meta(&Some(config.clone()));

            assert!(meta.is_some());
            let meta = meta.unwrap();
            assert!(meta.extra.contains_key("tool_configuration"));

            // Verify the serialized value can be deserialized back
            let restored: ToolConfiguration =
                serde_json::from_value(meta.extra["tool_configuration"].clone()).unwrap();
            assert_eq!(restored, config);
        }

        #[test]
        fn build_session_meta_without_tool_config() {
            let meta = AcpConnector::build_session_meta(&None);
            assert!(meta.is_none());
        }
    }
}
