//! Core runtime for connector orchestration
//!
//! This module implements the `CoreRuntime`, which is the central orchestrator
//! for managing long-lived connectors and routing events in the Dirigent system.
//!
//! # Architecture
//!
//! The CoreRuntime:
//! - Maintains a registry of active connectors keyed by ConnectorId
//! - Provides lifecycle management (create, start, stop, remove)
//! - Routes commands to specific connectors
//! - Broadcasts global events to all subscribers
//! - Enforces basic ownership and authorization
//!
//! # Module Organization
//!
//! The runtime functionality is split across focused submodules:
//! - `config_manager` - Configuration persistence and updates
//! - `summary_cache` - Connector summary cache for sync callbacks
//!
//! # Usage
//!
//! ```no_run
//! use dirigent_core::{CoreRuntime, CoreConfig, CoreHandle};
//!
//! # async fn example() {
//! // Create a runtime with default config
//! let config = CoreConfig::default();
//! let runtime = CoreRuntime::new(config, None);
//! let handle = CoreHandle::new(runtime);
//!
//! // List all connectors (none initially)
//! let connectors = handle.list_connectors(None).await;
//! assert_eq!(connectors.len(), 0);
//!
//! // Subscribe to events via the SharingBus
//! let _bus_rx = handle.sharing_bus().subscribe_all().await;
//! # }
//! ```

// Submodules
pub(crate) mod config_manager;
mod session_transfer;
mod summary_cache;
#[cfg(feature = "server")]
pub mod zed_detection;

use crate::config::{AcpServerConfig, ConnectorConfig, CoreConfig};
use crate::connectors::acp::{AcpConfig, AcpConnector};
use crate::connectors::gateway::{
    ConnectorSummaryInfo, SessionTransferCallback, SessionTransferRequest, SessionTransferResult,
};
use crate::connectors::{Connector, ConnectorCommand, ConnectorHandle};
use crate::sharing::bus::SharingBus;
use crate::types::{ConnectorId, ConnectorKind, ConnectorState, ConnectorSummary, User, UserId};
use crate::{CoreError, OpenCodeConfig, OpenCodeConnector};

use std::collections::HashMap;
use std::ops::Deref;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};

/// Error type for [`CoreRuntime::replay_session_to_stream`].
///
/// Wraps the standalone [`crate::sharing::replay::ReplayError`] with the
/// two runtime-level preconditions that the replay fn itself has no way to
/// check (missing stream id, missing archivist).
#[cfg(feature = "server")]
#[derive(Debug, thiserror::Error)]
pub enum ReplaySessionError {
    /// No stream with the given id is currently registered.
    #[error("stream not found: {0:?}")]
    StreamNotFound(crate::sharing::StreamId),
    /// The runtime has no archivist configured, so there's nothing to
    /// replay from.
    #[error("no archivist configured")]
    NoArchivist,
    /// The replay itself failed (session missing, archive I/O, …).
    #[error(transparent)]
    Replay(#[from] crate::sharing::replay::ReplayError),
}

/// Tracks an active session transfer for fallback purposes
#[derive(Clone, Debug)]
struct TransferredSession {
    /// Client's session (in Gateway before transfer)
    gateway_session_id: String,
    /// Gateway connector ID (for fallback)
    gateway_connector_id: String,
    /// Target connector that received the session
    target_connector_id: String,
    /// Session ID in target connector
    target_session_id: String,
    /// When the transfer occurred
    transferred_at: std::time::Instant,
}

/// Core runtime for connector orchestration
///
/// The CoreRuntime is the central component that manages all connectors
/// in the Dirigent system. It maintains:
/// - A registry of active connectors
/// - Configuration state
/// - A global event broadcast channel
/// - User registry (for ownership and authorization)
///
/// # Thread Safety
///
/// CoreRuntime uses tokio's async-aware RwLock for interior mutability,
/// allowing safe concurrent access from multiple async tasks. Read operations
/// can proceed in parallel, while write operations require exclusive access.
///
/// # Lifecycle
///
/// 1. Create with `CoreRuntime::new(config, None)`
/// 2. Wrap in `CoreHandle` for cheap cloning
/// 3. Create and manage connectors via runtime methods
/// 4. Subscribe to global events for system-wide event streaming
pub struct CoreRuntime {
    /// Runtime configuration
    ///
    /// Wrapped in Arc<RwLock> to allow dynamic updates while maintaining
    /// shared ownership across clones.
    config: Arc<RwLock<CoreConfig>>,

    /// Registry of active connectors
    ///
    /// Maps ConnectorId to ConnectorHandle. Protected by RwLock to allow
    /// concurrent reads (list/get operations) while serializing writes
    /// (create/remove operations).
    connectors: RwLock<HashMap<ConnectorId, ConnectorHandle>>,

    /// Primary event fan-out: a filtered, routing-aware event bus.
    ///
    /// All connector and runtime events are published onto the `SharingBus`.
    /// Subscribers pick an `EventFilter` and receive only matching events.
    /// See `docs/plans/2026-04-21-archivist-phase4-design.md`.
    sharing_bus: Arc<SharingBus>,

    /// Registry of users
    ///
    /// Maps UserId to User. Currently a simple in-memory registry.
    /// Future versions will integrate with authentication providers.
    // TODO: Multi-user support - user registry for authorization and ownership tracking
    #[allow(dead_code)]
    users: RwLock<HashMap<UserId, User>>,

    /// Optional archivist for persistent storage
    ///
    /// If Some, the archivist coordinates connector UID generation and
    /// persists session/message data. If None, the runtime operates
    /// in ephemeral mode with no archival storage.
    #[cfg(feature = "server")]
    archivist: Arc<tokio::sync::RwLock<Option<Arc<dirigent_archivist::Archivist>>>>,

    /// Mapping from connector_id (user-facing) to archivist connector_uid (internal UUID)
    ///
    /// This mapping enables archive-first reads to work with non-UUID connector IDs.
    /// Only populated when archivist is available and connectors are successfully registered.
    #[cfg(feature = "server")]
    archivist_connector_uids: Arc<RwLock<HashMap<String, uuid::Uuid>>>,

    /// Sync-accessible cache of connector summaries for use by GatewayConnector callbacks
    ///
    /// This cache is updated whenever connectors are added or removed. It uses a sync RwLock
    /// (std::sync::RwLock) so that it can be accessed from synchronous callback functions.
    /// The cache is wrapped in Arc so it can be shared with callbacks.
    connector_summary_cache: Arc<std::sync::RwLock<Vec<ConnectorSummaryInfo>>>,

    /// Weak reference to self (set when wrapped in Arc by CoreHandle)
    ///
    /// This allows creating callbacks that need to call runtime methods asynchronously.
    /// The weak reference prevents circular references and allows the runtime to be dropped.
    self_weak: Arc<RwLock<Option<std::sync::Weak<CoreRuntime>>>>,

    /// Tracks sessions that were transferred, for fallback handling
    ///
    /// Maps target_session_id -> TransferredSession. When a connector fails,
    /// we check this map to see if any transferred sessions need to fall back
    /// to Gateway with user notification.
    transferred_sessions: Arc<RwLock<HashMap<String, TransferredSession>>>,

    /// Optional inspector registry for internal visualization and introspection.
    ///
    /// When present, connectors and services register themselves as nodes
    /// in the inspector tree, enabling monitoring of state, process health,
    /// and system resources.
    #[cfg(feature = "server")]
    inspector: Option<Arc<dirigent_inspector::InspectorRegistry>>,

    /// Optional process group manager for lifecycle management of child processes.
    ///
    /// When present, ACP connectors using StdioTransport will create per-process
    /// lifecycle handles so spawned agents are tracked in the platform job object /
    /// process group and are shut down gracefully on disconnect.
    #[cfg(feature = "server")]
    process_manager: Option<Arc<dyn dirigent_process::ProcessGroupManager>>,

    /// Optional task runner for background process management
    #[cfg(feature = "server")]
    task_runner: Arc<tokio::sync::RwLock<Option<Arc<dirigent_taskrunner::TaskRunner>>>>,

    /// Optional Matrix service for session sharing
    #[cfg(feature = "server")]
    matrix_service: Arc<tokio::sync::RwLock<Option<Arc<dirigent_matrix::MatrixService>>>>,

    /// Registry of active streams wired to the `SharingBus`.
    ///
    /// Streams are attached at boot from `[[streams]]` config, or at runtime
    /// via [`CoreRuntime::attach_stream`]. See
    /// `docs/plans/2026-04-21-archivist-phase4-design.md`.
    stream_registry: Arc<crate::sharing::StreamRegistry>,

    /// Factory lookup used to build concrete streams from `StreamConfig`
    /// blocks. Populated by callers at boot with their compiled-in factories;
    /// defaults to an empty registry for tests and pre-Phase-4 call sites.
    stream_factories: Arc<crate::sharing::StreamFactoryRegistry>,
}

impl CoreRuntime {
    /// Create a new CoreRuntime with the given configuration
    ///
    /// This initializes all runtime state with defaults:
    /// - Empty connector registry
    /// - Global event channel with capacity 1000
    /// - Empty user registry
    /// - Optional archivist for persistent storage
    ///
    /// # Arguments
    ///
    /// * `config` - Core configuration (port, project directory, etc.)
    /// * `archivist` - Optional archivist for persistence (server feature only)
    ///
    /// # Returns
    ///
    /// A new CoreRuntime ready to manage connectors
    ///
    /// # Example
    ///
    /// ```no_run
    /// use dirigent_core::{CoreRuntime, CoreConfig};
    ///
    /// let config = CoreConfig::default();
    /// let runtime = CoreRuntime::new(config, None);
    /// ```
    #[cfg(feature = "server")]
    pub fn new(config: CoreConfig, archivist: Option<Arc<dirigent_archivist::Archivist>>) -> Self {
        Self::new_with_inspector(config, archivist, None)
    }

    /// Create a new CoreRuntime with archivist and inspector.
    ///
    /// This is a thin wrapper around [`Self::new_with_factories`] that
    /// supplies an empty [`StreamFactoryRegistry`]. Callers that want to
    /// attach streams at runtime should use `new_with_factories` directly
    /// and pre-register their compiled-in factories.
    #[cfg(feature = "server")]
    pub fn new_with_inspector(
        config: CoreConfig,
        archivist: Option<Arc<dirigent_archivist::Archivist>>,
        inspector: Option<Arc<dirigent_inspector::InspectorRegistry>>,
    ) -> Self {
        Self::new_with_factories(
            config,
            archivist,
            inspector,
            Arc::new(crate::sharing::StreamFactoryRegistry::default()),
        )
    }

    /// Create a new CoreRuntime with archivist, inspector, and a
    /// pre-populated [`StreamFactoryRegistry`].
    ///
    /// Callers at boot build the factory registry with their compiled-in
    /// `StreamFactory` implementations (e.g. Matrix, Langfuse, …) and hand
    /// it in here so that [`CoreRuntime::attach_stream`] can resolve
    /// `StreamConfig::kind` strings without further plumbing.
    ///
    /// Existing call sites that don't yet know about streams should keep
    /// calling [`Self::new`] / [`Self::new_with_inspector`], which forward
    /// to this constructor with an empty factory registry.
    #[cfg(feature = "server")]
    pub fn new_with_factories(
        config: CoreConfig,
        archivist: Option<Arc<dirigent_archivist::Archivist>>,
        inspector: Option<Arc<dirigent_inspector::InspectorRegistry>>,
        stream_factories: Arc<crate::sharing::StreamFactoryRegistry>,
    ) -> Self {
        // Primary event fan-out. All connector/runtime events flow through
        // this bus; subscribers choose an `EventFilter` to receive only the
        // events they care about.
        let sharing_bus = SharingBus::new();
        let stream_registry = Arc::new(crate::sharing::StreamRegistry::new(Arc::clone(&sharing_bus)));

        Self {
            config: Arc::new(RwLock::new(config)),
            connectors: RwLock::new(HashMap::new()),
            sharing_bus,
            users: RwLock::new(HashMap::new()),
            archivist: Arc::new(tokio::sync::RwLock::new(archivist)),
            archivist_connector_uids: Arc::new(RwLock::new(HashMap::new())),
            connector_summary_cache: Arc::new(std::sync::RwLock::new(Vec::new())),
            self_weak: Arc::new(RwLock::new(None)),
            transferred_sessions: Arc::new(RwLock::new(HashMap::new())),
            inspector,
            process_manager: None,
            task_runner: Arc::new(tokio::sync::RwLock::new(None)),
            matrix_service: Arc::new(tokio::sync::RwLock::new(None)),
            stream_registry,
            stream_factories,
        }
    }

    /// Create a new CoreRuntime without archivist (non-server builds)
    #[cfg(not(feature = "server"))]
    pub fn new(config: CoreConfig) -> Self {
        let sharing_bus = SharingBus::new();
        let stream_registry = Arc::new(crate::sharing::StreamRegistry::new(Arc::clone(&sharing_bus)));
        let stream_factories = Arc::new(crate::sharing::StreamFactoryRegistry::default());

        Self {
            config: Arc::new(RwLock::new(config)),
            connectors: RwLock::new(HashMap::new()),
            sharing_bus,
            users: RwLock::new(HashMap::new()),
            connector_summary_cache: Arc::new(std::sync::RwLock::new(Vec::new())),
            self_weak: Arc::new(RwLock::new(None)),
            transferred_sessions: Arc::new(RwLock::new(HashMap::new())),
            stream_registry,
            stream_factories,
        }
    }

    /// Set the process group manager used for lifecycle management of stdio agent processes.
    ///
    /// Call this before creating ACP connectors to ensure newly spawned processes are
    /// tracked in the platform job object (Windows) or process group (Unix). After this
    /// is set, every new `StdioTransport` will receive a per-process lifecycle handle.
    #[cfg(feature = "server")]
    pub fn set_process_manager(&mut self, mgr: Arc<dyn dirigent_process::ProcessGroupManager>) {
        self.process_manager = Some(mgr);
    }

    /// Get a reference to the process group manager, if configured.
    #[cfg(feature = "server")]
    pub fn process_manager(&self) -> Option<&Arc<dyn dirigent_process::ProcessGroupManager>> {
        self.process_manager.as_ref()
    }

    /// List all connectors, optionally filtered by owner
    ///
    /// Returns a summary of all connectors in the registry. If an owner
    /// is specified, only connectors owned by that user are returned.
    ///
    /// # Arguments
    ///
    /// * `owner` - Optional user ID to filter by ownership
    ///
    /// # Returns
    ///
    /// A vector of `ConnectorSummary` structs containing id, kind, owner,
    /// title, and current state for each matching connector.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig};
    /// # async fn example(runtime: &CoreRuntime) {
    /// // List all connectors
    /// let all = runtime.list_connectors(None).await;
    ///
    /// // List connectors for a specific user
    /// let user_connectors = runtime.list_connectors(Some(uuid::Uuid::nil())).await;
    /// # }
    /// ```
    pub async fn list_connectors(&self, owner: Option<UserId>) -> Vec<ConnectorSummary> {
        let connectors = self.connectors.read().await;

        // Collect handles first, then build summaries with async config reading
        let filtered_handles: Vec<_> = connectors
            .values()
            .filter(|handle| {
                // Filter by owner if provided
                if let Some(ref owner_id) = owner {
                    handle.owner() == owner_id
                } else {
                    true
                }
            })
            .cloned()
            .collect();

        // Drop the connectors lock before async operations
        drop(connectors);

        let mut summaries = Vec::with_capacity(filtered_handles.len());

        for handle in filtered_handles {
            // Extract supported_features and agent_type based on connector type
            let (supported_features, agent_type) = match handle.kind() {
                ConnectorKind::OpenCode => {
                    // OpenCode connectors support cancellation via stop_session,
                    // session resume, and session listing from the connector API
                    (
                        vec![
                            "cancellation".to_string(),
                            "session_resume".to_string(),
                            "session_list".to_string(),
                        ],
                        None,
                    )
                }
                ConnectorKind::Acp => {
                    // Read supported_features and agent_type from the persisted ConnectorConfig
                    // (not from handle.get_config_cloned() which returns AcpConfig params
                    // that don't contain supported_features)
                    let config_guard = self.config.read().await;
                    let (features, agent_type) = config_guard
                        .connectors
                        .iter()
                        .find(|c| c.id.as_deref() == Some(handle.id().as_str()))
                        .map(|c| {
                            let features = c.supported_features.clone();
                            let agent_type = serde_json::from_value::<crate::connectors::acp::config::AcpConfig>(
                                c.params.clone(),
                            )
                            .ok()
                            .map(|cfg| cfg.agent_type);
                            (features, agent_type)
                        })
                        .unwrap_or_default();
                    drop(config_guard);
                    (features, agent_type)
                }
                ConnectorKind::Gateway => {
                    // Gateway supports session resume via Archivist and session listing
                    (
                        vec![
                            "session_resume".to_string(),
                            "session_list".to_string(),
                        ],
                        None,
                    )
                }
                ConnectorKind::Acceptor | ConnectorKind::Mock => {
                    // These connectors don't have special features
                    (vec![], None)
                }
            };

            // Determine if archiving can be toggled for this connector
            // - OpenCode, ACP, Gateway: archiving can always be toggled
            // - Mock, Acceptor: archiving toggle not supported (test/routing connectors)
            let archiving_toggleable = match handle.kind() {
                ConnectorKind::OpenCode | ConnectorKind::Acp | ConnectorKind::Gateway => true,
                ConnectorKind::Mock | ConnectorKind::Acceptor => false,
            };

            // Look up tool_configuration and other config fields from saved config
            let (tool_configuration, plugin_assignments, project_assignments, use_in_new_projects, source, zed_agent_name) = {
                let config_lock = self.config.read().await;
                config_lock
                    .connectors
                    .iter()
                    .find(|c| c.id.as_deref() == Some(handle.id().as_str()))
                    .map(|c| (
                        c.tool_configuration.clone(),
                        c.plugin_assignments.clone(),
                        vec![], // project_assignments live on ConnectorSummary, not ConnectorConfig
                        c.use_in_new_projects,
                        c.source.clone(),
                        c.zed_agent_name.clone(),
                    ))
                    .unwrap_or((None, vec![], vec![], true, None, None))
            };

            summaries.push(ConnectorSummary {
                id: handle.id().clone(),
                kind: handle.kind(),
                owner: handle.owner().clone(),
                title: handle.title().to_string(),
                state: handle.state(),
                working_directory: handle.working_directory().map(|p| p.display().to_string()),
                supported_features,
                icon_path: handle.icon_path().map(|s| s.to_string()),
                show_type_overlay: handle.show_type_overlay(),
                archiving_toggleable,
                agent_type,
                tool_configuration,
                plugin_assignments,
                project_assignments,
                use_in_new_projects,
                source,
                zed_agent_name,
                error_kind: handle.error_kind(),
            });
        }

        summaries
    }

    /// Get a handle to a specific connector by ID
    ///
    /// Returns a cloned ConnectorHandle if the connector exists, or None
    /// if no connector with the given ID is registered.
    ///
    /// # Arguments
    ///
    /// * `id` - The unique identifier of the connector to retrieve
    ///
    /// # Returns
    ///
    /// `Some(ConnectorHandle)` if found, `None` otherwise
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig};
    /// # use dirigent_core::connectors::Connector;
    /// # async fn example(runtime: &CoreRuntime) {
    /// if let Some(handle) = runtime.get_connector(&"conn-123".to_string()).await {
    ///     println!("Found connector: {}", handle.title());
    ///     println!("State: {:?}", handle.state());
    /// } else {
    ///     println!("Connector not found");
    /// }
    /// # }
    /// ```
    pub async fn get_connector(&self, id: &ConnectorId) -> Option<ConnectorHandle> {
        let connectors = self.connectors.read().await;
        connectors.get(id).cloned()
    }

    /// Get the inspector registry, if configured.
    #[cfg(feature = "server")]
    pub fn inspector(&self) -> Option<&Arc<dirigent_inspector::InspectorRegistry>> {
        self.inspector.as_ref()
    }

    /// Create a new connector from configuration
    ///
    /// This method instantiates a connector of the specified type with the given
    /// configuration. The connector is created but not started - call `start_connector`
    /// to begin its background task.
    ///
    /// # Arguments
    ///
    /// * `owner` - User ID of the connector owner (for authorization)
    /// * `cfg` - Connector configuration including type and parameters
    ///
    /// # Returns
    ///
    /// The unique ID of the created connector, or an error if creation failed.
    ///
    /// # Errors
    ///
    /// - `AlreadyExists` - A connector with the specified ID already exists
    /// - `InvalidConfig` - Configuration validation failed or deserialization error
    /// - `Internal` - Other errors during connector instantiation
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig, ConnectorConfig, ConnectorKind};
    /// # use serde_json::json;
    /// # async fn example(runtime: &CoreRuntime) -> Result<(), Box<dyn std::error::Error>> {
    /// let cfg = ConnectorConfig {
    ///     id: None,
    ///     kind: ConnectorKind::OpenCode,
    ///     owner: None,
    ///     title: Some("My Connector".to_string()),
    ///     params: json!({
    ///         "base_url": "http://localhost:12225",
    ///         "title": "My Connector",
    ///         "initial_session": null
    ///     }),
    /// };
    ///
    /// let connector_id = runtime.create_connector(uuid::Uuid::now_v7(), cfg).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn create_connector(
        &self,
        owner: UserId,
        mut cfg: ConnectorConfig,
    ) -> Result<ConnectorId, CoreError> {
        // Generate ID if not provided
        let connector_id = cfg
            .id
            .clone()
            .unwrap_or_else(|| uuid::Uuid::now_v7().to_string());

        // Log connector count before creation
        let count_before = {
            let connectors = self.connectors.read().await;
            connectors.len()
        };

        info!(
            connector_id = %connector_id,
            kind = ?cfg.kind,
            owner = %owner,
            connector_count = count_before,
            "Creating connector (current count: {})", count_before
        );

        // Check if connector already exists
        {
            let connectors = self.connectors.read().await;
            if connectors.contains_key(&connector_id) {
                error!(connector_id = %connector_id, "Connector already exists");
                return Err(CoreError::AlreadyExists);
            }
        }

        // Override owner from config with the provided owner
        cfg.owner = Some(owner.clone());

        // Match on connector kind and instantiate the appropriate type
        // For OpenCode connectors, we start the task immediately since they can't be restarted
        let handle = match cfg.kind {
            ConnectorKind::OpenCode => {
                // Deserialize params to OpenCodeConfig
                let oc_config: OpenCodeConfig = serde_json::from_value(cfg.params.clone())
                    .map_err(|e| {
                        error!(
                            connector_id = %connector_id,
                            error = %e,
                            "Failed to deserialize OpenCodeConfig"
                        );
                        CoreError::InvalidConfig
                    })?;

                debug!(
                    connector_id = %connector_id,
                    base_url = %oc_config.base_url,
                    "Creating and starting OpenCode connector"
                );

                // Get title from main config, with fallback
                let title = cfg.title.clone().unwrap_or_else(|| "OpenCode".to_string());

                // Create the connector
                let oc_connector = OpenCodeConnector::new(
                    connector_id.clone(),
                    owner.clone(),
                    title,
                    oc_config,
                    Arc::clone(&self.sharing_bus),
                );

                // Resolve working directory using global config
                let config_lock = self.config.read().await;
                let working_directory = crate::resolve_default_runtime_working_directory(&cfg, &config_lock);

                // Create a handle from the connector's channels
                // IMPORTANT: Use new_with_state to share the same state Arc
                let mut handle = ConnectorHandle::new_with_state(
                    connector_id.clone(),
                    oc_connector.kind(),
                    owner.clone(),
                    oc_connector.title().to_string(),
                    oc_connector.state_arc(), // Share the state Arc from the connector
                    oc_connector.command_tx(),
                    oc_connector.events_sender(),
                    cfg.params.clone(), // Store config for restart
                    Some(working_directory),
                    cfg.icon_path.clone(),
                    cfg.show_type_overlay,
                );
                // Share the error_kind Arc so the handle observes connector error classification
                handle.set_error_kind_arc(oc_connector.error_kind_arc());

                // Start the connector task immediately (before dropping the connector)
                // This is necessary because OpenCodeConnector consumes the cmd_rx on start
                let task_handle = oc_connector.start_task().await;

                // Store the task handle in the ConnectorHandle
                handle.set_task_handle(task_handle).await;

                // Events are published directly to the SharingBus by the
                // connector (Task 9); no forwarder task needed.

                info!(
                    connector_id = %connector_id,
                    "OpenCode connector task started"
                );

                handle
            }
            ConnectorKind::Acp => {
                // Deserialize params to AcpConfig
                let acp_config: AcpConfig =
                    serde_json::from_value(cfg.params.clone()).map_err(|e| {
                        error!(
                            connector_id = %connector_id,
                            error = %e,
                            "Failed to deserialize AcpConfig"
                        );
                        CoreError::InvalidConfig
                    })?;

                // Extract title from main config or use default
                let title = cfg
                    .title
                    .clone()
                    .unwrap_or_else(|| "ACP Connector".to_string());

                debug!(
                    connector_id = %connector_id,
                    title = %title,
                    transport = ?acp_config.transport,
                    "Creating and starting ACP connector"
                );

                // Create the connector
                let acp_connector = {
                    let connector = AcpConnector::new(
                        connector_id.clone(),
                        owner.clone(),
                        title,
                        acp_config,
                        Arc::clone(&self.sharing_bus),
                    )
                    .map_err(|e| {
                        error!(
                            connector_id = %connector_id,
                            error = %e,
                            "Failed to create ACP connector"
                        );
                        CoreError::InvalidConfig
                    })?;
                    let connector =
                        connector.with_tool_configuration(cfg.tool_configuration.clone());
                    #[cfg(feature = "server")]
                    let connector = connector.with_inspector(self.inspector.clone());
                    #[cfg(feature = "server")]
                    let connector = connector.with_process_manager(self.process_manager.clone());
                    connector
                };

                // Resolve working directory using global config
                let config_lock = self.config.read().await;
                let working_directory = crate::resolve_default_runtime_working_directory(&cfg, &config_lock);

                // Create a handle from the connector's channels
                // IMPORTANT: Use new_with_state to share the same state Arc
                let handle = ConnectorHandle::new_with_state(
                    connector_id.clone(),
                    acp_connector.kind(),
                    owner.clone(),
                    acp_connector.title().to_string(),
                    acp_connector.state_arc(), // Share the state Arc from the connector
                    acp_connector.command_tx(),
                    acp_connector.events_sender(),
                    cfg.params.clone(), // Store config for restart
                    Some(working_directory),
                    cfg.icon_path.clone(),
                    cfg.show_type_overlay,
                );

                // Start the connector task immediately
                let task_handle = acp_connector.start_task().await;

                // Store the task handle in the ConnectorHandle
                handle.set_task_handle(task_handle).await;

                // The connector publishes events directly to the SharingBus
                // (Task 9). We still subscribe to the per-connector broadcast
                // here to intercept `available_commands_update` notifications
                // and mirror the dynamic commands onto the handle so that
                // `get_available_commands()` can return them.
                let mut connector_events = handle.subscribe();
                let conn_id_for_intercept: ConnectorId = connector_id.clone();
                let handle_for_commands = handle.clone();
                tokio::spawn(async move {
                    while let Ok(event) = connector_events.recv().await {
                        if let dirigent_protocol::Event::SessionUpdate {
                            update:
                                dirigent_protocol::SessionUpdate::Unknown { ref data },
                            ..
                        } = event
                        {
                            if data
                                .get("sessionUpdate")
                                .and_then(|s| s.as_str())
                                == Some("available_commands_update")
                            {
                                if let Some(commands_json) = data.get("availableCommands") {
                                    if let Ok(commands) = serde_json::from_value::<
                                        Vec<crate::acp::protocol::streaming::Command>,
                                    >(
                                        commands_json.clone()
                                    ) {
                                        tracing::debug!(
                                            connector_id = %conn_id_for_intercept,
                                            count = commands.len(),
                                            "Storing dynamic available commands on handle"
                                        );
                                        handle_for_commands.set_available_commands(commands);
                                    }
                                }
                            }
                        }
                    }
                    debug!(connector_id = %conn_id_for_intercept, "available_commands interceptor ended");
                });

                info!(
                    connector_id = %connector_id,
                    "ACP connector task started"
                );

                handle
            }
            ConnectorKind::Mock => {
                // Mock connector is only for testing
                error!(connector_id = %connector_id, "Mock connector cannot be created via API");
                return Err(CoreError::InvalidConfig);
            }
            ConnectorKind::Acceptor => {
                // Create an Acceptor connector (represents incoming ACP connections)
                // This is a lightweight connector that shows in the UI but delegates
                // actual session handling to the ACP Server and SessionManager

                let title = cfg
                    .title
                    .clone()
                    .unwrap_or_else(|| "Acceptor (ACP)".to_string());

                debug!(
                    connector_id = %connector_id,
                    title = %title,
                    "Creating Acceptor connector for incoming connections"
                );

                // Create channels (mostly unused for Acceptor, but needed for ConnectorHandle)
                let (cmd_tx, _cmd_rx) = tokio::sync::mpsc::channel::<ConnectorCommand>(100);
                let (events_tx, _) =
                    tokio::sync::broadcast::channel::<dirigent_protocol::Event>(1000);

                // Acceptor is immediately "Ready" since it's passive (waits for incoming connections)
                let state = std::sync::Arc::new(tokio::sync::RwLock::new(ConnectorState::Ready));

                // Resolve working directory using global config
                let config_lock = self.config.read().await;
                let working_directory = crate::resolve_default_runtime_working_directory(&cfg, &config_lock);

                // Create handle
                let handle = ConnectorHandle::new_with_state(
                    connector_id.clone(),
                    ConnectorKind::Acceptor,
                    owner.clone(),
                    title,
                    state,
                    cmd_tx,
                    events_tx.clone(),
                    cfg.params.clone(),
                    Some(working_directory),
                    cfg.icon_path.clone(),
                    cfg.show_type_overlay,
                );

                // Emit Connected event for consistency
                let _ = events_tx.send(dirigent_protocol::Event::Connected);

                info!(
                    connector_id = %connector_id,
                    "Acceptor connector created and ready for incoming connections"
                );

                handle
            }
            ConnectorKind::Gateway => {
                // Deserialize params to GatewayConfig
                let gateway_config: crate::connectors::gateway::GatewayConfig =
                    serde_json::from_value(cfg.params.clone()).map_err(|e| {
                        error!(
                            connector_id = %connector_id,
                            error = %e,
                            "Failed to deserialize GatewayConfig"
                        );
                        CoreError::InvalidConfig
                    })?;

                debug!(
                    connector_id = %connector_id,
                    title = %gateway_config.title,
                    echo_enabled = gateway_config.default_echo_enabled,
                    "Creating and starting Gateway connector"
                );

                // Create the connector
                let mut gateway_connector = crate::connectors::gateway::GatewayConnector::new(
                    connector_id.clone(),
                    owner.clone(),
                    gateway_config,
                    Arc::clone(&self.sharing_bus),
                );

                // Wire up callbacks for /list-connectors command
                // The cache is a sync RwLock that gets updated whenever connectors change
                let cache_for_list = self.connector_summary_cache();
                let list_callback: crate::connectors::gateway::ConnectorListCallback =
                    Arc::new(move || {
                        // Read from the sync cache (non-blocking)
                        match cache_for_list.read() {
                            Ok(cache) => cache.clone(),
                            Err(_) => Vec::new(), // Return empty if lock is poisoned
                        }
                    });
                gateway_connector.set_connector_list_callback(list_callback);

                // Wire up session transfer callback for /select-connector command
                let self_weak_clone = Arc::clone(&self.self_weak);
                let transfer_callback: SessionTransferCallback = Arc::new(move |request| {
                    let self_weak = Arc::clone(&self_weak_clone);
                    tokio::spawn(async move {
                        // Try to upgrade the weak reference
                        let weak_guard = self_weak.read().await;
                        if let Some(ref weak) = *weak_guard {
                            if let Some(runtime) = weak.upgrade() {
                                runtime.transfer_session(request).await;
                            } else {
                                tracing::warn!(
                                    "Runtime has been dropped, cannot process transfer request"
                                );
                                let _ = request.result_tx.send(SessionTransferResult::Failed(
                                    "Runtime is shutting down".to_string(),
                                ));
                            }
                        } else {
                            tracing::error!("Runtime weak reference not initialized");
                            let _ = request.result_tx.send(SessionTransferResult::Failed(
                                "Runtime not properly initialized".to_string(),
                            ));
                        }
                    });
                });
                gateway_connector.set_session_transfer_callback(transfer_callback);

                // Wire up inspector for session tracking
                #[cfg(feature = "server")]
                gateway_connector.set_inspector(self.inspector.clone());

                // Resolve working directory using global config
                let config_lock = self.config.read().await;
                let working_directory = crate::resolve_default_runtime_working_directory(&cfg, &config_lock);

                // Create a handle from the connector's channels
                let handle = ConnectorHandle::new_with_state(
                    connector_id.clone(),
                    gateway_connector.kind(),
                    owner.clone(),
                    gateway_connector.title().to_string(),
                    gateway_connector.state_arc(),
                    gateway_connector.command_tx(),
                    gateway_connector.events_sender(),
                    cfg.params.clone(),
                    Some(working_directory),
                    cfg.icon_path.clone(),
                    cfg.show_type_overlay,
                );

                // Start the connector task immediately
                let task_handle = gateway_connector.start_task().await;

                // Store the task handle in the ConnectorHandle
                handle.set_task_handle(task_handle).await;

                // Events are published directly to the SharingBus by the
                // connector (Task 9); no forwarder task needed.

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

                handle
            }
        };

        // Insert the handle into the connectors map
        {
            let mut connectors = self.connectors.write().await;
            connectors.insert(connector_id.clone(), handle);
        }

        // Update the connector summary cache for GatewayConnector callbacks
        self.update_connector_summary_cache().await;

        // Save values we'll need after moving cfg
        let connector_kind = cfg.kind.clone();
        let connector_title = cfg
            .title
            .clone()
            .unwrap_or_else(|| format!("{:?} Connector", cfg.kind));

        // Register connector with archivist (if available and server feature enabled)
        #[cfg(feature = "server")]
        if let Some(archivist) = self.archivist.read().await.clone() {
            // Determine custom_uid: use connector_id if it's a UUID, otherwise let archivist generate one
            let custom_uid = uuid::Uuid::parse_str(&connector_id).ok();

            // Compute a deterministic fingerprint for archivist re-association across restarts
            let fingerprint =
                crate::connectors::fingerprint::compute_fingerprint(&cfg.kind, &cfg.params);

            // Create registration request
            let register_req = dirigent_archivist::types::RegisterConnectorRequest {
                custom_uid, // Some(uuid) if connector_id was UUID, None otherwise
                r#type: format!("{:?}", connector_kind),
                title: connector_title.clone(),
                client_native_id: connector_id.clone(),
                metadata: serde_json::json!({}),
                fingerprint: fingerprint.clone(),
            };

            // Attempt registration (non-blocking, best-effort)
            match archivist.register_connector(register_req, None).await {
                Ok(response) => {
                    // Store the mapping for archive-first reads (works for both UUID and custom IDs)
                    self.archivist_connector_uids
                        .write()
                        .await
                        .insert(connector_id.clone(), response.connector_uid);

                    // Backfill fingerprint on existing records that were registered
                    // before fingerprinting was introduced (Task 11)
                    if response.status == dirigent_archivist::types::RegisterStatus::Aliased {
                        if let Some(ref fp) = fingerprint {
                            let _ = archivist
                                .update_connector_fingerprint(
                                    response.connector_uid,
                                    fp.clone(),
                                    None,
                                )
                                .await;
                        }
                    }

                    info!(
                        connector_id = %connector_id,
                        connector_uid = %response.connector_uid,
                        status = ?response.status,
                        "Registered connector with archivist"
                    );
                }
                Err(e) => {
                    warn!(
                        connector_id = %connector_id,
                        error = %e,
                        "Failed to register connector with archivist (non-fatal)"
                    );
                }
            }
        }

        // Register connector with inspector (if available)
        #[cfg(feature = "server")]
        if let Some(ref inspector) = self.inspector {
            let node_id =
                dirigent_inspector::NodeId::new(format!("dirigent/connectors/{}", connector_id));
            let parent_id = dirigent_inspector::NodeId::new("dirigent/connectors");
            let mut metadata = dirigent_inspector::NodeMetadata::new(
                dirigent_inspector::NodeKind::Connector,
                &connector_title,
            )
            .with_state(dirigent_inspector::NodeState::Initializing)
            .with_property("kind", serde_json::json!(format!("{:?}", connector_kind)))
            .with_property("owner", serde_json::json!(&owner));

            // Extract command/executable from params for display in inspector
            if let Some(params_obj) = cfg.params.as_object() {
                if let Some(transport) = params_obj.get("transport") {
                    match transport.get("type").and_then(|t| t.as_str()) {
                        Some("stdio") => {
                            if let Some(cmd) = transport.get("command").and_then(|c| c.as_str()) {
                                let args = transport
                                    .get("args")
                                    .and_then(|a| a.as_array())
                                    .map(|arr| {
                                        arr.iter()
                                            .filter_map(|v| v.as_str())
                                            .collect::<Vec<_>>()
                                            .join(" ")
                                    })
                                    .unwrap_or_default();
                                let full_command = if args.is_empty() {
                                    cmd.to_string()
                                } else {
                                    format!("{} {}", cmd, args)
                                };
                                metadata = metadata.with_property(
                                    "command",
                                    serde_json::json!(full_command),
                                );
                            }
                        }
                        Some("http") => {
                            if let Some(url) =
                                transport.get("base_url").and_then(|u| u.as_str())
                            {
                                metadata = metadata
                                    .with_property("command", serde_json::json!(url));
                            }
                        }
                        _ => {}
                    }
                } else if let Some(base_url) = params_obj.get("base_url").and_then(|u| u.as_str())
                {
                    // OpenCode connectors have base_url at the top level
                    metadata =
                        metadata.with_property("command", serde_json::json!(base_url));
                }
            }
            match inspector
                .register(node_id, &parent_id, metadata, None)
                .await
            {
                Ok(mut handle) => {
                    handle.detach();
                    info!(
                        connector_id = %connector_id,
                        "Registered connector with inspector"
                    );
                }
                Err(e) => {
                    warn!(
                        connector_id = %connector_id,
                        error = %e,
                        "Failed to register connector with inspector (non-fatal)"
                    );
                }
            }
        }

        // Auto-save: persist the new connector to the configuration file
        // This ensures the connector will be restored on server restart
        {
            let mut config_lock = self.config.write().await;
            // Ensure the connector config has the generated ID set
            cfg.id = Some(connector_id.clone());

            // Check if connector already exists in config
            if let Some(existing_idx) = config_lock
                .connectors
                .iter()
                .position(|c| c.id == Some(connector_id.clone()))
            {
                // Update existing entry instead of appending
                debug!(connector_id = %connector_id, "Updating existing connector in config");
                config_lock.connectors[existing_idx] = cfg;
            } else {
                // Append new entry
                debug!(connector_id = %connector_id, "Adding new connector to config");
                config_lock.connectors.push(cfg);
            }
        }

        // Save configuration to disk (log warning on error, don't fail operation)
        if let Err(e) = self.save_config().await {
            warn!(
                connector_id = %connector_id,
                error = %e,
                "Failed to save configuration after connector creation"
            );
        }

        // Log connector count after creation
        let count_after = {
            let connectors = self.connectors.read().await;
            connectors.len()
        };

        info!(
            connector_id = %connector_id,
            connector_count = count_after,
            "Connector created successfully (count: {} -> {})",
            count_before,
            count_after
        );

        // Log special case: first connector created
        if count_before == 0 && count_after == 1 {
            info!("First connector created - transitioning from empty state");
        }

        // Emit ConnectorCreated onto the SharingBus.
        let bus_event = dirigent_protocol::streaming::BusEvent::from_connector_event(
            dirigent_protocol::Event::ConnectorCreated {
                connector_id: connector_id.clone(),
                kind: format!("{:?}", connector_kind),
                title: connector_title,
            },
            None, // Task 9 will wire connector_uid
            connector_id.clone(),
        );
        self.sharing_bus.publish(bus_event).await;

        Ok(connector_id)
    }

    /// Start a connector's background task
    ///
    /// This method starts the connector's async task loop, which manages the
    /// connection to the underlying agent system and processes commands and events.
    ///
    /// # Arguments
    ///
    /// * `id` - The unique identifier of the connector to start
    ///
    /// # Returns
    ///
    /// Ok(()) if the connector was started successfully, or an error.
    ///
    /// # Errors
    ///
    /// - `NotFound` - No connector with the specified ID exists
    /// - `InvalidState` - Connector is already running or in an invalid state
    /// - `StartFailed` - Failed to start the connector task
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig};
    /// # async fn example(runtime: &CoreRuntime, connector_id: &str) -> Result<(), Box<dyn std::error::Error>> {
    /// runtime.start_connector(&connector_id.to_string()).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn start_connector(&self, id: &ConnectorId) -> Result<(), CoreError> {
        info!(connector_id = %id, "Starting connector");

        // Get the connector handle
        let handle = {
            let connectors = self.connectors.read().await;
            connectors.get(id).cloned().ok_or_else(|| {
                error!(connector_id = %id, "Connector not found");
                CoreError::NotFound
            })?
        };

        // Check if already started (has task handle)
        if handle.take_task_handle().await.is_some() {
            warn!(connector_id = %id, "Connector already started");
            return Err(CoreError::InvalidState);
        }

        // Start the connector based on its kind
        match handle.kind() {
            ConnectorKind::OpenCode => {
                // For OpenCode, we need to recreate the connector and start its task
                // This is necessary because the start_task() method consumes the command receiver

                // We'll create a new OpenCodeConnector instance with the same config
                // Extract config from handle metadata
                // Since we can't easily get the config back, we'll use the command pattern

                // For now, we'll document that connectors should be started immediately after creation
                // A better approach would be to store the connector trait object alongside the handle
                error!(
                    connector_id = %id,
                    "Starting existing connector not yet supported - start immediately after creation"
                );
                Err(CoreError::Internal(
                    "Starting existing connector not yet supported - start immediately after creation"
                        .to_string(),
                ))
            }
            ConnectorKind::Acp => {
                error!(connector_id = %id, "ACP connector not yet implemented");
                Err(CoreError::Internal(
                    "ACP connector not yet implemented".to_string(),
                ))
            }
            ConnectorKind::Mock => {
                // Mock connectors don't have background tasks
                debug!(connector_id = %id, "Mock connector has no background task");
                Ok(())
            }
            ConnectorKind::Acceptor => {
                // Acceptors don't have background tasks - they are event-driven
                debug!(connector_id = %id, "Acceptor connector has no background task");
                Ok(())
            }
            ConnectorKind::Gateway => {
                // Gateway connectors are started immediately after creation
                // like OpenCode connectors
                error!(
                    connector_id = %id,
                    "Starting existing Gateway connector not yet supported - start immediately after creation"
                );
                Err(CoreError::Internal(
                    "Starting existing Gateway connector not yet supported - start immediately after creation"
                        .to_string(),
                ))
            }
        }
    }

    /// Stop a running connector gracefully
    ///
    /// This method sends a shutdown command to the connector and waits for its
    /// background task to complete. The connector will clean up resources and
    /// transition to the Stopped state.
    ///
    /// # Arguments
    ///
    /// * `id` - The unique identifier of the connector to stop
    ///
    /// # Returns
    ///
    /// Ok(()) if the connector was stopped successfully, or an error.
    ///
    /// # Errors
    ///
    /// - `NotFound` - No connector with the specified ID exists
    /// - `Internal` - Failed to send shutdown command or task join failed
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig};
    /// # async fn example(runtime: &CoreRuntime, connector_id: &str) -> Result<(), Box<dyn std::error::Error>> {
    /// runtime.stop_connector(&connector_id.to_string()).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn stop_connector(&self, id: &ConnectorId) -> Result<(), CoreError> {
        info!(connector_id = %id, "Stopping connector");

        // Get the connector handle
        let handle = {
            let connectors = self.connectors.read().await;
            connectors.get(id).cloned().ok_or_else(|| {
                error!(connector_id = %id, "Connector not found");
                CoreError::NotFound
            })?
        };

        // Check if connector has been started
        let current_state = handle.state();
        if matches!(current_state, ConnectorState::Initializing) {
            // Connector was never started, so there's no task to shutdown
            debug!(connector_id = %id, "Connector is in Initializing state, skipping shutdown command");
        } else {
            // Send shutdown command
            let cmd_tx = handle.command_tx();
            match cmd_tx.send(ConnectorCommand::Shutdown).await {
                Ok(_) => {
                    debug!(connector_id = %id, "Shutdown command sent");
                }
                Err(e) => {
                    // Command send failed - likely the receiver was dropped (connector not started)
                    debug!(connector_id = %id, error = %e, "Failed to send shutdown command (connector may not have been started)");
                }
            }
        }

        // Wait for task to complete (with timeout)
        if let Some(task_handle) = handle.take_task_handle().await {
            debug!(connector_id = %id, "Waiting for connector task to complete");

            // Use tokio::time::timeout to avoid waiting forever
            match tokio::time::timeout(Duration::from_secs(30), task_handle).await {
                Ok(Ok(())) => {
                    info!(connector_id = %id, "Connector task completed successfully");
                }
                Ok(Err(e)) if e.is_panic() => {
                    // Task panicked - log but don't treat as error
                    // The task is stopped (panicked tasks are stopped), which is what we wanted
                    error!(connector_id = %id, "Connector task panicked during shutdown: {:?}", e);
                    info!(connector_id = %id, "Treating panicked task as stopped (goal achieved)");
                }
                Ok(Err(e)) => {
                    // Other JoinHandle errors (e.g., task was cancelled)
                    warn!(connector_id = %id, error = %e, "Connector task terminated with error");
                    // Don't return error - task is stopped regardless
                }
                Err(_) => {
                    warn!(connector_id = %id, "Connector task did not complete within timeout");
                    // Don't return error - the connector may still be shutting down
                }
            }
        } else {
            debug!(connector_id = %id, "No task handle found (connector may not have been started)");
        }

        // Update state to Stopped
        {
            let state_lock = handle.state_lock();
            let mut state = state_lock.write().await;
            *state = crate::types::ConnectorState::Stopped;
        }

        info!(connector_id = %id, "Connector stopped successfully");

        Ok(())
    }

    /// Restart a stopped connector
    ///
    /// This method recreates a stopped connector's background task with fresh
    /// channels while preserving its identity (ID, owner, configuration).
    /// The connector must be in the `Stopped` or `Error` state to be restarted.
    ///
    /// # Arguments
    ///
    /// * `id` - The unique identifier of the connector to restart
    ///
    /// # Returns
    ///
    /// Ok(()) if the connector was restarted successfully, or an error.
    ///
    /// # Errors
    ///
    /// - `NotFound` - No connector with the specified ID exists
    /// - `InvalidState` - Connector is not in `Stopped` or `Error` state
    /// - `InvalidConfig` - Failed to deserialize connector configuration
    /// - `Internal` - Other internal errors during restart
    ///
    /// # State Transitions
    ///
    /// The connector will transition through the following states:
    /// - `Stopped` or `Error` → `Initializing` → `Connecting` → `Ready`
    ///
    /// # Preservation
    ///
    /// The following are preserved across restart:
    /// - Connector ID (remains the same)
    /// - Owner (authorization unchanged)
    /// - Configuration (same parameters)
    /// - Event broadcast channel (existing subscribers continue receiving)
    /// - State Arc (external observers see state transitions)
    ///
    /// The following are recreated:
    /// - Command channel (new sender/receiver pair)
    /// - Connector instance (fresh OpenCodeConnector, etc.)
    /// - Background task (new spawn)
    /// - Task handle (new JoinHandle)
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig};
    /// # async fn example(runtime: &CoreRuntime, connector_id: &str) -> Result<(), Box<dyn std::error::Error>> {
    /// // Stop a connector
    /// runtime.stop_connector(&connector_id.to_string()).await?;
    ///
    /// // Restart it
    /// runtime.restart_connector(&connector_id.to_string()).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn restart_connector(&self, id: &ConnectorId) -> Result<(), CoreError> {
        info!(connector_id = %id, "Restarting connector");

        // Get the connector handle from registry
        let handle = {
            let connectors = self.connectors.read().await;
            connectors.get(id).cloned()
        };

        let handle = match handle {
            Some(h) => h,
            None => {
                error!(connector_id = %id, "Connector not found");
                return Err(CoreError::NotFound);
            }
        };

        // Verify state is Stopped or Error (only valid states for restart)
        let current_state = handle.state();
        match current_state {
            crate::types::ConnectorState::Stopped | crate::types::ConnectorState::Error(_) => {
                debug!(connector_id = %id, state = ?current_state, "Connector is in restartable state");
            }
            _ => {
                error!(
                    connector_id = %id,
                    state = ?current_state,
                    "Cannot restart connector in current state. Stop it first."
                );
                return Err(CoreError::InvalidState);
            }
        }

        // Get config from handle
        let config_json = handle.get_config_cloned().await;
        debug!(connector_id = %id, "Retrieved connector config for restart");

        // Match on connector kind and recreate
        match handle.kind() {
            ConnectorKind::OpenCode => {
                // Deserialize config to OpenCodeConfig
                let oc_config: crate::connectors::opencode::OpenCodeConfig =
                    serde_json::from_value(config_json).map_err(|e| {
                        error!(
                            connector_id = %id,
                            error = %e,
                            "Failed to deserialize OpenCodeConfig during restart"
                        );
                        CoreError::InvalidConfig
                    })?;

                debug!(
                    connector_id = %id,
                    base_url = %oc_config.base_url,
                    "Creating new OpenCode connector instance for restart"
                );

                // Use existing title from handle
                let title = handle.title().to_string();

                // Create new connector instance
                let oc_connector = crate::connectors::opencode::OpenCodeConnector::new(
                    id.clone(),
                    handle.owner().clone(),
                    title,
                    oc_config,
                    Arc::clone(&self.sharing_bus),
                );

                // Create new handle with same identity but fresh channels
                // IMPORTANT: Use new connector's events_sender, not old handle's
                // The new connector creates a new broadcast channel in its constructor
                let mut new_handle = ConnectorHandle::new_with_state(
                    id.clone(),
                    oc_connector.kind(),
                    handle.owner().clone(),
                    oc_connector.title().to_string(),
                    handle.state_lock(), // Reuse existing state Arc
                    oc_connector.command_tx(),
                    oc_connector.events_sender(), // NEW: Use fresh events channel
                    handle.get_config_cloned().await, // Preserve config
                    handle.working_directory(),   // Preserve working directory
                    handle.icon_path().map(|s| s.to_string()),
                    handle.show_type_overlay(),
                );
                // Share the error_kind Arc so the handle observes connector error classification
                new_handle.set_error_kind_arc(oc_connector.error_kind_arc());

                // Start the new connector task
                let task_handle = oc_connector.start_task().await;

                // Store the task handle
                new_handle.set_task_handle(task_handle).await;

                // Events are published directly to the SharingBus by the
                // connector (Task 9); no forwarder task needed.

                info!(
                    connector_id = %id,
                    "OpenCode connector task restarted"
                );

                // Replace the handle in the registry
                {
                    let mut connectors = self.connectors.write().await;
                    connectors.insert(id.clone(), new_handle);
                }
            }
            ConnectorKind::Acp => {
                // Deserialize config to AcpConfig
                let acp_config: AcpConfig = serde_json::from_value(config_json).map_err(|e| {
                    error!(
                        connector_id = %id,
                        error = %e,
                        "Failed to deserialize AcpConfig during restart"
                    );
                    CoreError::InvalidConfig
                })?;

                // Use existing title from handle
                let title = handle.title().to_string();

                debug!(
                    connector_id = %id,
                    title = %title,
                    transport = ?acp_config.transport,
                    "Creating new ACP connector instance for restart"
                );

                // Create new connector instance
                let acp_connector = AcpConnector::new(
                    id.clone(),
                    handle.owner().clone(),
                    title,
                    acp_config,
                    Arc::clone(&self.sharing_bus),
                )
                .map_err(|e| {
                    error!(
                        connector_id = %id,
                        error = %e,
                        "Failed to create ACP connector during restart"
                    );
                    CoreError::InvalidConfig
                })?;

                // Look up tool configuration from saved config
                let tool_config = {
                    let config_lock = self.config.read().await;
                    config_lock
                        .connectors
                        .iter()
                        .find(|c| c.id.as_deref() == Some(id.as_str()))
                        .and_then(|c| c.tool_configuration.clone())
                };
                let acp_connector =
                    acp_connector.with_tool_configuration(tool_config);
                #[cfg(feature = "server")]
                let acp_connector = acp_connector.with_process_manager(self.process_manager.clone());

                // Create new handle with same identity but fresh channels
                // IMPORTANT: Use new connector's events_sender, not old handle's
                // The new connector creates a new broadcast channel in its constructor
                let new_handle = ConnectorHandle::new_with_state(
                    id.clone(),
                    acp_connector.kind(),
                    handle.owner().clone(),
                    acp_connector.title().to_string(),
                    handle.state_lock(), // Reuse existing state Arc
                    acp_connector.command_tx(),
                    acp_connector.events_sender(), // NEW: Use fresh events channel
                    handle.get_config_cloned().await, // Preserve config
                    handle.working_directory(),    // Preserve working directory
                    handle.icon_path().map(|s| s.to_string()),
                    handle.show_type_overlay(),
                );

                // Start the new connector task
                let task_handle = acp_connector.start_task().await;

                // Store the task handle
                new_handle.set_task_handle(task_handle).await;

                // The connector publishes events directly to the SharingBus
                // (Task 9). Subscribe to the per-connector broadcast only to
                // intercept `available_commands_update` notifications and
                // mirror dynamic commands onto the handle.
                let mut connector_events = new_handle.subscribe();
                let conn_id_for_intercept: ConnectorId = id.clone();
                let handle_for_commands = new_handle.clone();
                tokio::spawn(async move {
                    while let Ok(event) = connector_events.recv().await {
                        if let dirigent_protocol::Event::SessionUpdate {
                            update:
                                dirigent_protocol::SessionUpdate::Unknown { ref data },
                            ..
                        } = event
                        {
                            if data
                                .get("sessionUpdate")
                                .and_then(|s| s.as_str())
                                == Some("available_commands_update")
                            {
                                if let Some(commands_json) = data.get("availableCommands") {
                                    if let Ok(commands) = serde_json::from_value::<
                                        Vec<crate::acp::protocol::streaming::Command>,
                                    >(
                                        commands_json.clone()
                                    ) {
                                        handle_for_commands.set_available_commands(commands);
                                    }
                                }
                            }
                        }
                    }
                    debug!(connector_id = %conn_id_for_intercept, "available_commands interceptor ended for restarted ACP connector");
                });

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

                // Replace the handle in the registry
                {
                    let mut connectors = self.connectors.write().await;
                    connectors.insert(id.clone(), new_handle);
                }
            }
            ConnectorKind::Mock => {
                // Mock connectors don't have background tasks, so restart is a no-op
                debug!(connector_id = %id, "Mock connector restart is a no-op");
            }
            ConnectorKind::Acceptor => {
                // Acceptors don't have background tasks, so restart is a no-op
                debug!(connector_id = %id, "Acceptor connector restart is a no-op");
            }
            ConnectorKind::Gateway => {
                // Deserialize config to GatewayConfig
                let gateway_config: crate::connectors::gateway::GatewayConfig =
                    serde_json::from_value(config_json).map_err(|e| {
                        error!(
                            connector_id = %id,
                            error = %e,
                            "Failed to deserialize GatewayConfig during restart"
                        );
                        CoreError::InvalidConfig
                    })?;

                debug!(
                    connector_id = %id,
                    title = %gateway_config.title,
                    echo_enabled = gateway_config.default_echo_enabled,
                    "Creating new Gateway connector instance for restart"
                );

                // Create new connector instance
                let mut gateway_connector = crate::connectors::gateway::GatewayConnector::new(
                    id.clone(),
                    handle.owner().clone(),
                    gateway_config,
                    Arc::clone(&self.sharing_bus),
                );

                // Wire up inspector for session tracking
                #[cfg(feature = "server")]
                gateway_connector.set_inspector(self.inspector.clone());

                // Create new handle with same identity but fresh channels
                // IMPORTANT: Use new connector's events_sender, not old handle's
                // The new connector creates a new broadcast channel in its constructor
                let new_handle = ConnectorHandle::new_with_state(
                    id.clone(),
                    gateway_connector.kind(),
                    handle.owner().clone(),
                    gateway_connector.title().to_string(),
                    handle.state_lock(),
                    gateway_connector.command_tx(),
                    gateway_connector.events_sender(), // NEW: Use fresh events channel
                    handle.get_config_cloned().await,
                    handle.working_directory(),
                    handle.icon_path().map(|s| s.to_string()),
                    handle.show_type_overlay(),
                );

                // Start the new connector task
                let task_handle = gateway_connector.start_task().await;

                // Store the task handle
                new_handle.set_task_handle(task_handle).await;

                // Events are published directly to the SharingBus by the
                // connector (Task 9); no forwarder task needed.

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

                // Replace the handle in the registry
                {
                    let mut connectors = self.connectors.write().await;
                    connectors.insert(id.clone(), new_handle);
                }
            }
        }

        info!(connector_id = %id, "Connector restarted successfully");

        Ok(())
    }

    /// Update a connector's configuration
    ///
    /// This method updates the stored configuration for a connector.
    /// The changes are persisted immediately but only take effect when
    /// the connector is restarted.
    ///
    /// # Arguments
    ///
    /// * `id` - The connector ID
    /// * `patch` - JSON patch to merge into existing config
    ///
    /// # Returns
    ///
    /// Ok(()) on success, or CoreError on failure
    ///
    /// # Errors
    ///
    /// - `NotFound` - No connector with the specified ID exists
    /// - `Internal` - Failed to save configuration to disk
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig};
    /// # use serde_json::json;
    /// # async fn example(runtime: &CoreRuntime, connector_id: &str) -> Result<(), Box<dyn std::error::Error>> {
    /// // Update the connector's title
    /// runtime.update_connector_config(
    ///     &connector_id.to_string(),
    ///     json!({
    ///         "title": "New Title"
    ///     })
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn update_connector_config(
        &self,
        id: &ConnectorId,
        patch: serde_json::Value,
    ) -> Result<(), CoreError> {
        info!(connector_id = %id, "Updating connector configuration");

        // Get the connector handle
        let handle = {
            let connectors = self.connectors.read().await;
            connectors.get(id).cloned().ok_or_else(|| {
                error!(connector_id = %id, "Connector not found");
                CoreError::NotFound
            })?
        };

        // Get current config
        let mut config = handle.get_config_cloned().await;
        debug!(connector_id = %id, "Retrieved current config for update");

        // Merge the patch into the config (shallow merge)
        if let (Some(base_obj), Some(patch_obj)) = (config.as_object_mut(), patch.as_object()) {
            for (key, value) in patch_obj {
                base_obj.insert(key.clone(), value.clone());
            }
        }

        // Update the handle's config
        handle.set_config(config.clone()).await;
        debug!(connector_id = %id, "Updated connector handle config");

        // Update the config in CoreConfig's connectors list and live ConnectorHandle
        {
            let mut config_lock = self.config.write().await;
            let mut connectors_lock = self.connectors.write().await;

            // Update persisted config
            if let Some(connector_cfg) = config_lock
                .connectors
                .iter_mut()
                .find(|c| c.id.as_ref() == Some(id))
            {
                // Handle top-level orchestration fields
                if let Some(patch_obj) = patch.as_object() {
                    // Update title if present in patch
                    if let Some(title_value) = patch_obj.get("title") {
                        let new_title = title_value.as_str().map(|s| s.to_string());
                        connector_cfg.title = new_title.clone();

                        // Also update the live ConnectorHandle
                        if let Some(live_handle) = connectors_lock.get_mut(id) {
                            if let Some(title) = new_title {
                                live_handle.set_title(title);
                            }
                        }
                    }

                    // Update working_directory if present in patch
                    if let Some(wd_value) = patch_obj.get("working_directory") {
                        let new_wd = if wd_value.is_null() {
                            None
                        } else {
                            wd_value.as_str().map(|s| std::path::PathBuf::from(s))
                        };
                        connector_cfg.working_directory = new_wd.clone();

                        // Also update the live ConnectorHandle (async operation)
                        if let Some(live_handle) = connectors_lock.get(id) {
                            live_handle.set_working_directory(new_wd).await;
                        }
                    }

                    // Update supported_features if present in patch
                    if let Some(features_value) = patch_obj.get("supported_features") {
                        if let Some(features_array) = features_value.as_array() {
                            let new_features: Vec<String> = features_array
                                .iter()
                                .filter_map(|v| v.as_str().map(|s| s.to_string()))
                                .collect();
                            connector_cfg.supported_features = new_features;
                        }
                    }

                    // Update icon_path if present in patch
                    if let Some(icon_value) = patch_obj.get("icon_path") {
                        let new_icon_path = if icon_value.is_null() {
                            None
                        } else {
                            icon_value.as_str().map(|s| s.to_string())
                        };
                        connector_cfg.icon_path = new_icon_path.clone();

                        // Also update the live ConnectorHandle
                        if let Some(live_handle) = connectors_lock.get_mut(id) {
                            live_handle.set_icon_path(new_icon_path);
                        }
                    }

                    // Update show_type_overlay if present in patch
                    if let Some(overlay_value) = patch_obj.get("show_type_overlay") {
                        if let Some(overlay_bool) = overlay_value.as_bool() {
                            connector_cfg.show_type_overlay = overlay_bool;

                            // Also update the live ConnectorHandle
                            if let Some(live_handle) = connectors_lock.get_mut(id) {
                                live_handle.set_show_type_overlay(overlay_bool);
                            }
                        }
                    }

                    // Merge other fields into the stored connector params
                    // ONLY connector-specific fields should go in params
                    if let Some(params_obj) = connector_cfg.params.as_object_mut() {
                        for (key, value) in patch_obj {
                            // Skip ALL orchestration fields - they belong in main config, not params
                            if key != "title"
                                && key != "working_directory"
                                && key != "supported_features"
                                && key != "icon_path"
                                && key != "show_type_overlay"
                                && key != "kind"
                            {
                                params_obj.insert(key.clone(), value.clone());
                            }
                        }
                    }
                }
                debug!(connector_id = %id, "Updated connector config in CoreConfig and live handle");
            }
        }

        // Persist changes to disk
        if let Err(e) = self.save_config().await {
            warn!(
                connector_id = %id,
                error = %e,
                "Failed to save configuration after connector update"
            );
            return Err(CoreError::Internal(format!(
                "Failed to save configuration: {}",
                e
            )));
        }

        info!(connector_id = %id, "Connector configuration updated successfully");

        Ok(())
    }

    /// Remove a connector from the runtime
    ///
    /// This method stops the connector (if running) and removes it from the
    /// connector registry. After removal, the connector ID can be reused.
    ///
    /// # Arguments
    ///
    /// * `id` - The unique identifier of the connector to remove
    ///
    /// # Returns
    ///
    /// Ok(()) if the connector was removed successfully, or an error.
    ///
    /// # Errors
    ///
    /// - `NotFound` - No connector with the specified ID exists
    /// - Other errors from `stop_connector` if the connector is running
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig};
    /// # async fn example(runtime: &CoreRuntime, connector_id: &str) -> Result<(), Box<dyn std::error::Error>> {
    /// runtime.remove_connector(&connector_id.to_string()).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn remove_connector(&self, id: &ConnectorId) -> Result<(), CoreError> {
        // Log connector count before removal
        let count_before = {
            let connectors = self.connectors.read().await;
            connectors.len()
        };

        info!(
            connector_id = %id,
            connector_count = count_before,
            "Removing connector (current count: {})", count_before
        );

        // Check if connector exists
        {
            let connectors = self.connectors.read().await;
            if !connectors.contains_key(id) {
                error!(connector_id = %id, "Connector not found");
                return Err(CoreError::NotFound);
            }
        }

        // Stop the connector first (if running)
        // We ignore NotFound errors here since we already checked above
        match self.stop_connector(id).await {
            Ok(()) => {
                debug!(connector_id = %id, "Connector stopped before removal");
            }
            Err(CoreError::NotFound) => {
                // This shouldn't happen, but handle it gracefully
                debug!(connector_id = %id, "Connector not found during stop (concurrent removal?)");
            }
            Err(e) => {
                warn!(connector_id = %id, error = %e, "Error stopping connector, continuing with removal");
            }
        }

        // Remove from registry
        {
            let mut connectors = self.connectors.write().await;
            connectors.remove(id);
        }

        // Update the connector summary cache for GatewayConnector callbacks
        self.update_connector_summary_cache().await;

        // Auto-save: remove the connector from the configuration file
        // This ensures the connector won't be restored on server restart
        {
            let mut config_lock = self.config.write().await;
            // If this is a Zed-sourced connector, extract its title so we can
            // add it to the dismissed list (preventing re-addition on restart)
            let dismissed_title = config_lock
                .connectors
                .iter()
                .find(|cfg| cfg.id.as_ref().map(|cfg_id| cfg_id == id).unwrap_or(false))
                .filter(|cfg| cfg.source.as_deref() == Some("zed"))
                .and_then(|cfg| cfg.title.clone());

            if let Some(ref title) = dismissed_title {
                if !config_lock.dismissed_zed_agents.contains(title) {
                    info!(
                        connector_id = %id,
                        title = %title,
                        "Adding Zed connector to dismissed list"
                    );
                    config_lock.dismissed_zed_agents.push(title.clone());
                }
            }

            config_lock.connectors.retain(|cfg| {
                // Remove the connector by its ID
                if cfg.id.as_ref().map(|cfg_id| cfg_id == id).unwrap_or(false) {
                    return false;
                }
                // Also remove null-ID Zed entries that match the dismissed title
                // (these are transient entries from enrichment that haven't been
                // assigned IDs yet)
                if let Some(ref dismissed) = dismissed_title {
                    if cfg.id.is_none()
                        && cfg.source.as_deref() == Some("zed")
                        && cfg.title.as_deref() == Some(dismissed)
                    {
                        return false;
                    }
                }
                true
            });
        }

        // Save configuration to disk (log warning on error, don't fail operation)
        if let Err(e) = self.save_config().await {
            warn!(
                connector_id = %id,
                error = %e,
                "Failed to save configuration after connector removal"
            );
        }

        // Log connector count after removal
        let count_after = {
            let connectors = self.connectors.read().await;
            connectors.len()
        };

        info!(
            connector_id = %id,
            connector_count = count_after,
            "Connector removed successfully (count: {} -> {})",
            count_before,
            count_after
        );

        // Log special case: last connector removed
        if count_before == 1 && count_after == 0 {
            info!("Last connector removed - transitioning to empty state");
        }

        // Deregister connector from inspector (if available)
        #[cfg(feature = "server")]
        if let Some(ref inspector) = self.inspector {
            let node_id = dirigent_inspector::NodeId::new(format!("dirigent/connectors/{}", id));
            let _ = inspector.deregister_subtree(&node_id).await;
        }

        // Emit ConnectorRemoved event onto the SharingBus.
        let bus_event = dirigent_protocol::streaming::BusEvent::from_connector_event(
            dirigent_protocol::Event::ConnectorRemoved {
                connector_id: id.clone(),
            },
            None, // Task 9 will wire connector_uid
            id.clone(),
        );
        self.sharing_bus.publish(bus_event).await;

        Ok(())
    }

    /// Send a command to a specific connector
    ///
    /// This method validates ownership and then sends the command to the
    /// connector's command channel for async processing.
    ///
    /// # Arguments
    ///
    /// * `id` - The unique identifier of the connector
    /// * `cmd` - The command to send
    ///
    /// # Returns
    ///
    /// Ok(()) if the command was sent successfully, or an error.
    ///
    /// # Errors
    ///
    /// - `NotFound` - No connector with the specified ID exists
    /// - `Internal` - Failed to send command to the connector's channel
    ///
    /// # Note
    ///
    /// This method does not perform ownership validation yet. In a future
    /// version, it will check that the requesting user owns the connector.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig, ConnectorCommand};
    /// # async fn example(runtime: &CoreRuntime, connector_id: &str) -> Result<(), Box<dyn std::error::Error>> {
    /// runtime.send_command(
    ///     &connector_id.to_string(),
    ///     ConnectorCommand::ListSessions
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn send_command(
        &self,
        id: &ConnectorId,
        cmd: ConnectorCommand,
    ) -> Result<(), CoreError> {
        debug!(connector_id = %id, command = ?cmd, "Sending command to connector");

        // Get the connector handle
        let handle = {
            let connectors = self.connectors.read().await;
            connectors.get(id).cloned().ok_or_else(|| {
                error!(connector_id = %id, "Connector not found");
                CoreError::NotFound
            })?
        };

        // TODO: Add ownership validation here
        // For now, we allow any command to any connector
        // In the future, we'll check:
        // if handle.owner() != current_user {
        //     return Err(CoreError::Unauthorized);
        // }

        // Send the command
        let cmd_tx = handle.command_tx();
        cmd_tx.send(cmd).await.map_err(|e| {
            error!(connector_id = %id, error = %e, "Failed to send command to connector");
            CoreError::Internal(format!("Failed to send command: {}", e))
        })?;

        debug!(connector_id = %id, "Command sent successfully");

        Ok(())
    }

    /// Transfer a session to another connector
    ///
    /// This is called by the Gateway connector when a user invokes /select-connector.
    /// It validates the target, creates or loads a session, and emits events.
    ///
    /// # Arguments
    /// * `request` - The transfer request from Gateway
    ///
    /// # Returns
    /// The result is sent via the oneshot channel in the request
    pub async fn transfer_session(&self, request: SessionTransferRequest) {
        let result = self.execute_transfer(&request).await;

        // Emit event on success
        // Models/modes are included directly in the SessionTransferred event
        // so the event bridge can emit config_option_update without needing
        // a separate SessionMetadataReceived event.
        if let SessionTransferResult::Transferred {
            ref connector_id,
            ref session_id,
            is_new,
            ref models,
            ref modes,
        } = result
        {
            let event = dirigent_protocol::Event::SessionTransferred {
                from_connector: "gateway".to_string(), // Could track actual gateway ID
                from_session: request.gateway_session_id.clone(),
                to_connector: connector_id.clone(),
                to_session: session_id.clone(),
                is_new_session: is_new,
                models: models.clone(),
                modes: modes.clone(),
            };

            // Broadcast onto the SharingBus.
            let bus_event = dirigent_protocol::streaming::BusEvent::from_connector_event(
                event.clone(),
                None, // Task 9 will wire connector_uid
                connector_id.clone(),
            );
            self.sharing_bus.publish(bus_event).await;

            // Also emit to Gateway connector's local channel so send_message() can receive it
            if let Some(gateway_connector) = self.get_connector(&request.gateway_connector_id).await
            {
                let _ = gateway_connector.events_sender().send(event.clone());
            }

            // Track the transfer for potential fallback
            // Find a Gateway connector to use as fallback
            let gateway_connector_id = self
                .find_gateway_connector()
                .await
                .unwrap_or_else(|| "gateway-placeholder".to_string());

            let transfer_record = TransferredSession {
                gateway_session_id: request.gateway_session_id.clone(),
                gateway_connector_id,
                target_connector_id: connector_id.clone(),
                target_session_id: session_id.clone(),
                transferred_at: std::time::Instant::now(),
            };

            let mut transfers = self.transferred_sessions.write().await;
            transfers.insert(session_id.clone(), transfer_record);
        }

        // Send result back to Gateway
        let _ = request.result_tx.send(result);
    }

    /// Get the agent_type from a connector's configuration
    ///
    /// Returns `ConnectorAgentType::Custom` if the connector is not found or not an ACP connector.
    /// Used by the ACP server to apply mode/model mappings when setting session modes/models.
    pub async fn get_connector_agent_type(
        &self,
        connector_id: &str,
    ) -> crate::connectors::acp::config::ConnectorAgentType {
        use crate::connectors::acp::config::ConnectorAgentType;

        // Read config to get connector config
        let config_guard = self.config.read().await;

        // Find the connector config by ID
        if let Some(connector_cfg) = config_guard
            .connectors
            .iter()
            .find(|c| c.id.as_deref() == Some(connector_id))
        {
            // Only ACP connectors have agent_type
            if connector_cfg.kind == ConnectorKind::Acp {
                // Try to deserialize to AcpConfig to extract agent_type
                if let Ok(acp_config) = serde_json::from_value::<
                    crate::connectors::acp::config::AcpConfig,
                >(connector_cfg.params.clone())
                {
                    return acp_config.agent_type;
                }
            }
        }

        // Default to Custom if not found or not an ACP connector
        ConnectorAgentType::Custom
    }

    /// Execute the transfer logic
    ///
    /// Creates or loads a session in the target connector. The target connector's
    /// actual modes/models are forwarded to the editor via `config_option_update`
    /// (handled by the event bridge).
    async fn execute_transfer(&self, request: &SessionTransferRequest) -> SessionTransferResult {
        // 1. Validate target connector exists
        let connector = match self.get_connector(&request.target_connector_id).await {
            Some(c) => c,
            None => {
                return SessionTransferResult::Failed(format!(
                    "Connector '{}' not found",
                    request.target_connector_id
                ));
            }
        };

        // 2. Check connector kind supports transfer
        if !connector.kind().supports_session_transfer() {
            return SessionTransferResult::Failed(format!(
                "Connector '{}' does not support session transfer",
                request.target_connector_id
            ));
        }

        // 3. Check connector is Ready
        let state = connector.state();
        if state != ConnectorState::Ready {
            return SessionTransferResult::Failed(format!(
                "Connector '{}' is not ready (state: {:?})",
                request.target_connector_id, state
            ));
        }

        // 4. Create or load session in target connector
        let cmd_tx = connector.command_tx();
        let mut events = connector.subscribe();

        if let Some(ref target_session_id) = request.target_session_id {
            // Try to load existing session
            let transfer_cwd = connector
                .working_directory()
                .map(|p| p.to_string_lossy().to_string())
                .unwrap_or_else(|| ".".to_string());
            if let Err(e) = cmd_tx
                .send(ConnectorCommand::LoadSession {
                    session_id: target_session_id.clone(),
                    cwd: transfer_cwd,
                    mcp_servers: None,
                })
                .await
            {
                return SessionTransferResult::Failed(format!(
                    "Failed to send load command: {}",
                    e
                ));
            }

            // Wait for response
            match session_transfer::wait_for_session_event(
                &mut events,
                target_session_id,
                Duration::from_secs(10),
            )
            .await
            {
                Ok(session_id) => {
                    // For loaded sessions, we don't have models/modes from the event
                    // They will be populated via SessionMetadataReceived event if the connector sends it
                    return SessionTransferResult::Transferred {
                        connector_id: request.target_connector_id.clone(),
                        session_id,
                        is_new: false,
                        models: None,
                        modes: None,
                    };
                }
                Err(_) => {
                    // Load failed, create new session instead
                    tracing::info!(
                        "Session '{}' not found in connector '{}', creating new session",
                        target_session_id,
                        request.target_connector_id
                    );
                    match session_transfer::create_session_in_connector(&cmd_tx, &mut events).await
                    {
                        Ok((id, models, modes)) => {
                            return SessionTransferResult::Transferred {
                                connector_id: request.target_connector_id.clone(),
                                session_id: id,
                                is_new: true,
                                models,
                                modes,
                            };
                        }
                        Err(e) => return SessionTransferResult::Failed(e),
                    }
                }
            }
        } else {
            // Create new session
            match session_transfer::create_session_in_connector(&cmd_tx, &mut events).await {
                Ok((id, models, modes)) => {
                    return SessionTransferResult::Transferred {
                        connector_id: request.target_connector_id.clone(),
                        session_id: id,
                        is_new: true,
                        models,
                        modes,
                    };
                }
                Err(e) => return SessionTransferResult::Failed(e),
            }
        }
    }

    /// Find an available Gateway connector for fallback
    ///
    /// Returns the ID of a Ready Gateway connector, or None if none found.
    async fn find_gateway_connector(&self) -> Option<String> {
        let connectors = self.connectors.read().await;

        for (id, handle) in connectors.iter() {
            if handle.kind() == ConnectorKind::Gateway && handle.state() == ConnectorState::Ready {
                return Some(id.clone());
            }
        }

        None
    }

    /// Called when a connector's state changes
    ///
    /// Checks if any transferred sessions need to fall back to Gateway.
    pub async fn check_transferred_sessions_fallback(
        &self,
        connector_id: &str,
        new_state: &ConnectorState,
    ) {
        // Only care about failure states
        if !matches!(
            new_state,
            ConnectorState::Error(_) | ConnectorState::Stopped
        ) {
            return;
        }

        let affected_sessions: Vec<TransferredSession> = {
            let transfers = self.transferred_sessions.read().await;

            transfers
                .values()
                .filter(|t| t.target_connector_id == connector_id)
                .cloned()
                .collect()
        };

        for session in affected_sessions {
            self.trigger_fallback_to_gateway(&session, new_state).await;
        }
    }

    /// Trigger fallback for a transferred session
    async fn trigger_fallback_to_gateway(
        &self,
        session: &TransferredSession,
        failed_state: &ConnectorState,
    ) {
        let reason = match failed_state {
            ConnectorState::Error(msg) => format!("Connector error: {}", msg),
            ConnectorState::Stopped => "Connector stopped".to_string(),
            _ => "Unknown failure".to_string(),
        };

        warn!(
            target_connector = %session.target_connector_id,
            target_session = %session.target_session_id,
            gateway_session = %session.gateway_session_id,
            reason = %reason,
            "Triggering fallback to Gateway for transferred session"
        );

        // Emit ForwardingPanic onto the SharingBus.
        let event = dirigent_protocol::Event::ForwardingPanic {
            connector_id: session.target_connector_id.clone(),
            session_id: session.target_session_id.clone(),
            reason: reason.clone(),
            fallback_gateway_session: Some(session.gateway_session_id.clone()),
        };
        let bus_event = dirigent_protocol::streaming::BusEvent::from_connector_event(
            event,
            None, // Task 9 will wire connector_uid
            session.target_connector_id.clone(),
        );
        self.sharing_bus.publish(bus_event).await;

        // Remove from tracking
        let mut transfers = self.transferred_sessions.write().await;
        transfers.remove(&session.target_session_id);

        // Notify Gateway to create/restore session with error message
        self.notify_gateway_fallback(session, &reason).await;
    }

    /// Notify Gateway about a fallback
    async fn notify_gateway_fallback(&self, session: &TransferredSession, reason: &str) {
        // Get Gateway connector
        let gateway = self.get_connector(&session.gateway_connector_id).await;

        if let Some(gateway) = gateway {
            // Send a message to Gateway to display to the user
            let cmd_tx = gateway.command_tx();
            let error_message = format!(
                "**Connection Lost**\n\n\
                The session with {} was interrupted: {}\n\n\
                You have been returned to Gateway. Use `/list-connectors` to reconnect.",
                session.target_connector_id, reason
            );

            // Ensure the Gateway session exists and send error message
            let _ = cmd_tx
                .send(ConnectorCommand::SendMessage {
                    session_id: session.gateway_session_id.clone(),
                    text: format!("__SYSTEM_ERROR__:{}", error_message),
                })
                .await;
        }
    }

    /// Remove stale transfer records older than the given duration
    pub fn cleanup_stale_transfers(&self, max_age: Duration) {
        let now = std::time::Instant::now();

        if let Ok(mut transfers) = self.transferred_sessions.try_write() {
            transfers.retain(|_, t| now.duration_since(t.transferred_at) < max_age);
        }
    }

    /// Return a reference to the primary event bus.
    ///
    /// The bus is the primary fan-out for all runtime/connector events.
    /// Callers can subscribe with an
    /// [`dirigent_protocol::streaming::EventFilter`] to receive only the
    /// events they care about.
    pub fn sharing_bus(&self) -> &Arc<SharingBus> {
        &self.sharing_bus
    }

    /// Return a reference to the live [`StreamRegistry`].
    ///
    /// Useful when callers want direct access (e.g. for telemetry) without
    /// going through the [`attach_stream`](Self::attach_stream) /
    /// [`detach_stream`](Self::detach_stream) convenience methods.
    pub fn stream_registry(&self) -> &Arc<crate::sharing::StreamRegistry> {
        &self.stream_registry
    }

    /// Build a stream from `config` using the registered factory for
    /// `config.kind`, then attach it to the bus.
    ///
    /// Returns the freshly assigned [`StreamId`] on success. Errors if no
    /// factory is registered for the given kind, or if the factory's
    /// `build` call fails (see [`StreamBuildError`]).
    pub async fn attach_stream(
        &self,
        config: crate::sharing::StreamConfig,
    ) -> Result<crate::sharing::StreamId, crate::sharing::StreamBuildError> {
        let factory = self
            .stream_factories
            .get(&config.kind)
            .cloned()
            .ok_or_else(|| crate::sharing::StreamBuildError::UnknownKind(config.kind.clone()))?;
        let stream = factory.build(&config).await?;
        let id = self.stream_registry.attach(config.name, stream).await;
        Ok(id)
    }

    /// Detach a previously attached stream. Returns `true` if the id was
    /// live and the stream has now been shut down; `false` if the id was
    /// already absent from the registry.
    pub async fn detach_stream(&self, id: crate::sharing::StreamId) -> bool {
        self.stream_registry.detach(id).await.is_some()
    }

    /// Snapshot every registered stream. See [`StreamRegistry::list`].
    pub async fn list_streams(&self) -> Vec<crate::sharing::StreamInfo> {
        self.stream_registry.list().await
    }

    /// Replay the archived messages of `scroll_id` onto the stream
    /// identified by `stream_id`.
    ///
    /// Reads metadata + messages from the attached archivist and dispatches
    /// synthetic `BusEvent`s with `EventOrigin::Replay { .. }` directly to
    /// the target stream, bypassing the `SharingBus`. Live events remain
    /// unaffected.
    ///
    /// Errors if the stream isn't registered, no archivist is configured on
    /// this runtime, or the archive read fails fatally. Per-event stream
    /// failures are counted in the returned [`ReplayReport`] instead of
    /// propagated.
    #[cfg(feature = "server")]
    pub async fn replay_session_to_stream(
        &self,
        scroll_id: uuid::Uuid,
        stream_id: crate::sharing::StreamId,
        opts: crate::sharing::ReplayOptions,
    ) -> Result<crate::sharing::ReplayReport, ReplaySessionError> {
        let stream = self
            .stream_registry
            .get_stream(stream_id)
            .await
            .ok_or(ReplaySessionError::StreamNotFound(stream_id))?;
        let archivist_guard = self.archivist.read().await;
        let archivist = archivist_guard
            .as_ref()
            .ok_or(ReplaySessionError::NoArchivist)?;
        crate::sharing::replay::replay_session_to_stream(
            archivist.as_ref(),
            scroll_id,
            stream,
            opts,
        )
        .await
        .map_err(ReplaySessionError::Replay)
    }

    /// Emit a runtime-origin event onto the primary event bus.
    ///
    /// This allows non-connector components (e.g., system task registry,
    /// archivist-adjacent flows, import jobs) to publish events that are
    /// delivered to all bus subscribers.
    ///
    /// # Return value
    ///
    /// Returns `0` for historical signature compatibility with the
    /// pre-Phase-4 API (which reported the number of `global_events`
    /// receivers that saw the event). Routing is now fire-and-forget
    /// through the bus worker, so this number is no longer available
    /// synchronously and no remaining caller inspects it.
    pub fn emit_event(&self, event: dirigent_protocol::Event) -> usize {
        let bus = Arc::clone(&self.sharing_bus);
        let bus_event = dirigent_protocol::streaming::BusEvent {
            routing: dirigent_protocol::streaming::EventRouting::default(),
            origin: dirigent_protocol::streaming::EventOrigin::Runtime,
            event: Arc::new(event),
        };
        // `publish` is async; we have no connector context here and must
        // remain `fn` (non-async) for the existing call-site ergonomics.
        tokio::spawn(async move {
            bus.publish(bus_event).await;
        });
        0
    }

    /// Get a reference to the runtime configuration
    ///
    /// Returns an Arc to the RwLock-protected configuration. This allows
    /// reading and potentially updating the configuration at runtime.
    ///
    /// # Returns
    ///
    /// Arc<RwLock<CoreConfig>> for shared access to configuration
    pub fn config(&self) -> Arc<RwLock<CoreConfig>> {
        Arc::clone(&self.config)
    }

    /// Get a reference to the connectors registry
    ///
    /// This is primarily for internal use by methods that need direct
    /// access to the connector map for mutations (create, remove).
    ///
    /// # Returns
    ///
    /// Reference to the RwLock-protected connector HashMap
    // TODO: Low-level runtime access - internal methods for advanced connector/user management operations
    #[allow(dead_code)]
    pub(crate) fn connectors_lock(&self) -> &RwLock<HashMap<ConnectorId, ConnectorHandle>> {
        &self.connectors
    }

    /// Get a reference to the users registry
    ///
    /// This is primarily for internal use by methods that need to manage
    /// user data and perform authorization checks.
    ///
    /// # Returns
    ///
    /// Reference to the RwLock-protected user HashMap
    // TODO: Low-level runtime access - internal methods for advanced connector/user management operations
    #[allow(dead_code)]
    pub(crate) fn users_lock(&self) -> &RwLock<HashMap<UserId, User>> {
        &self.users
    }

    /// Get a reference to the archivist (server feature only)
    ///
    /// Returns None if no archivist is configured or if not running in server mode.
    ///
    /// # Returns
    ///
    /// Option<Arc<Archivist>> for archival storage operations
    #[cfg(feature = "server")]
    pub async fn archivist(&self) -> Option<Arc<dirigent_archivist::Archivist>> {
        self.archivist.read().await.clone()
    }

    /// Set the archivist at runtime (for hot-reload activation)
    #[cfg(feature = "server")]
    pub async fn set_archivist(&self, archivist: Option<Arc<dirigent_archivist::Archivist>>) {
        let mut guard = self.archivist.write().await;
        *guard = archivist;
    }

    /// Get a reference to the archivist slot (for sharing with AppState)
    #[cfg(feature = "server")]
    pub fn archivist_slot(&self) -> &Arc<tokio::sync::RwLock<Option<Arc<dirigent_archivist::Archivist>>>> {
        &self.archivist
    }

    /// Get the task runner slot (shared reference for AppState integration)
    #[cfg(feature = "server")]
    pub fn task_runner_slot(
        &self,
    ) -> &Arc<tokio::sync::RwLock<Option<Arc<dirigent_taskrunner::TaskRunner>>>> {
        &self.task_runner
    }

    // -----------------------------------------------------------------------
    // Matrix integration
    // -----------------------------------------------------------------------

    /// Initialize and start the Matrix service if configured.
    ///
    /// Reads the `matrix` behavior config and resolves the referenced account,
    /// then creates and starts the service.
    /// Call during server startup after CoreRuntime is created.
    #[cfg(feature = "server")]
    pub async fn start_matrix_service(&self) -> std::result::Result<(), String> {
        let config_guard = self.config.read().await;
        let behavior = match &config_guard.matrix {
            Some(b) => b.clone(),
            None => {
                debug!("No [matrix] configuration found, skipping Matrix service startup");
                return Ok(());
            }
        };

        let account = config_guard
            .accounts
            .get(&behavior.account)
            .ok_or_else(|| {
                format!(
                    "Matrix config references account '{}' but it is not defined in [accounts]",
                    behavior.account
                )
            })?
            .clone();
        drop(config_guard);

        if account.kind != dirigent_auth::AccountKind::Matrix {
            return Err(format!(
                "Account '{}' has type {:?}, expected 'matrix'",
                behavior.account, account.kind
            ));
        }

        let data_dir = dirigent_config::DirigentPaths::resolve()
            .map_err(|e| format!("Failed to resolve data directory: {}", e))?
            .data_dir()
            .to_path_buf();

        let service = dirigent_matrix::MatrixService::from_account(&account, behavior, data_dir)
            .map_err(|e| format!("Failed to create Matrix service: {}", e))?;

        service
            .login()
            .await
            .map_err(|e| format!("Matrix login failed: {}", e))?;
        service
            .start_sync()
            .await
            .map_err(|e| format!("Matrix sync start failed: {}", e))?;

        let service_arc = Arc::new(service);
        *self.matrix_service.write().await = Some(service_arc);

        info!("Matrix service started successfully");
        Ok(())
    }

    /// Create a Matrix session share.
    ///
    /// Creates a Matrix room and starts a bidirectional bridge between the
    /// specified connector session and the room.
    ///
    /// Returns the Matrix room ID string on success.
    #[cfg(feature = "server")]
    pub async fn create_matrix_share(
        &self,
        connector_id: &str,
        session_id: &str,
        session_title: Option<&str>,
    ) -> std::result::Result<String, String> {
        let service_guard = self.matrix_service.read().await;
        let service = service_guard
            .as_ref()
            .ok_or("Matrix service not started")?;

        let client = service
            .client_cloned()
            .await
            .ok_or("Matrix client not logged in")?;

        // Get the connector
        let connectors = self.connectors.read().await;
        let connector = connectors
            .get(connector_id)
            .ok_or_else(|| format!("Connector '{}' not found", connector_id))?;

        let event_rx = connector.subscribe();
        let connector_cmd_tx = connector.command_tx();
        drop(connectors);

        // Create room
        let room_name =
            dirigent_matrix::room::room_name_for_session(connector_id, session_title);
        let invite = service.behavior().default_invite.clone();

        let room_id = dirigent_matrix::room::create_share_room(
            &client,
            dirigent_matrix::CreateRoomOptions {
                name: room_name,
                topic: Some(format!(
                    "Dirigent session share: {} / {}",
                    connector_id, session_id
                )),
                invite,
            },
        )
        .await
        .map_err(|e| format!("Failed to create Matrix room: {}", e))?;

        let room_id_str = room_id.to_string();

        // Get the Room handle for the share
        let room = client
            .get_room(&room_id)
            .ok_or_else(|| format!("Room {} not found after creation", room_id_str))?;

        // Start the share — returns (share, command_rx)
        let (share, mut command_rx) = dirigent_matrix::MatrixSessionShare::start(
            connector_id.to_string(),
            session_id.to_string(),
            room_id_str.clone(),
            room,
            event_rx,
        );

        // Spawn proxy task: command_rx -> ConnectorCommand::SendMessage
        let cmd_tx = connector_cmd_tx.clone();
        tokio::spawn(async move {
            while let Some(proxy) = command_rx.recv().await {
                let cmd = crate::connectors::ConnectorCommand::SendMessage {
                    session_id: proxy.session_id,
                    text: proxy.text,
                };
                if cmd_tx.send(cmd).await.is_err() {
                    break;
                }
            }
        });

        // Register share
        service
            .register_share(share)
            .await
            .map_err(|e| format!("Failed to register share: {}", e))?;

        // Persist sharing metadata in archivist
        if let Some(archivist) = self.archivist().await {
            if let Some(connector_uid) = self.get_archivist_connector_uid(connector_id).await {
                if let Ok(scroll_id) = archivist.resolve_session(connector_uid, session_id, None).await {
                    if let Err(e) = archivist.update_session_sharing(
                        scroll_id,
                        Some(room_id_str.clone()),
                        true,
                        None,
                    ).await {
                        tracing::warn!("Failed to persist sharing metadata: {}", e);
                    }
                }
            }
        }

        info!(
            connector_id = %connector_id,
            session_id = %session_id,
            room_id = %room_id_str,
            "Matrix session share created"
        );

        Ok(room_id_str)
    }

    /// Stop the Matrix service (if running).
    ///
    /// Shuts down all active shares and releases the client, then clears the
    /// service slot. Returns `true` if a service was running and was stopped.
    #[cfg(feature = "server")]
    pub async fn stop_matrix_service(&self) -> bool {
        let mut guard = self.matrix_service.write().await;
        if let Some(service) = guard.take() {
            service.shutdown().await;
            info!("Matrix service stopped");
            true
        } else {
            false
        }
    }

    /// Restart the Matrix service.
    ///
    /// Stops any running instance, then starts a fresh one from the current
    /// configuration. This picks up any config changes made since the last
    /// start.
    #[cfg(feature = "server")]
    pub async fn restart_matrix_service(&self) -> std::result::Result<(), String> {
        self.stop_matrix_service().await;
        self.start_matrix_service().await
    }

    /// Get the Matrix service (if started).
    #[cfg(feature = "server")]
    pub async fn matrix_service(&self) -> Option<Arc<dirigent_matrix::MatrixService>> {
        self.matrix_service.read().await.clone()
    }

    /// Register an existing connector with the archivist (for hot-reload)
    ///
    /// Used during hot-reload to register connectors that were created
    /// before the archivist was activated.
    #[cfg(feature = "server")]
    pub async fn register_connector_with_archivist(
        &self,
        connector_id: &str,
    ) -> Result<(), CoreError> {
        let archivist = self
            .archivist()
            .await
            .ok_or_else(|| CoreError::Internal("Archivist not configured".to_string()))?;

        let connector = self
            .get_connector(&connector_id.to_string())
            .await
            .ok_or(CoreError::NotFound)?;

        let connector_kind = connector.kind();
        let connector_title = connector.title().to_string();
        let fingerprint = {
            let config_arc = self.config.read().await;
            config_arc
                .connectors
                .iter()
                .find(|c| c.id.as_deref() == Some(connector_id))
                .and_then(|cfg| {
                    crate::connectors::fingerprint::compute_fingerprint(&cfg.kind, &cfg.params)
                })
        };

        let register_req = dirigent_archivist::types::RegisterConnectorRequest {
            custom_uid: uuid::Uuid::try_parse(connector_id).ok(),
            r#type: format!("{:?}", connector_kind),
            title: connector_title,
            client_native_id: connector_id.to_string(),
            metadata: serde_json::json!({}),
            fingerprint: fingerprint.clone(),
        };

        match archivist.register_connector(register_req, None).await {
            Ok(response) => {
                self.archivist_connector_uids
                    .write()
                    .await
                    .insert(connector_id.to_string(), response.connector_uid);
                info!(
                    connector_id = %connector_id,
                    connector_uid = %response.connector_uid,
                    "Registered existing connector with archivist (hot-reload)"
                );
                Ok(())
            }
            Err(e) => {
                warn!(
                    connector_id = %connector_id,
                    error = %e,
                    "Failed to register connector with archivist"
                );
                Err(CoreError::Internal(format!(
                    "Failed to register connector: {}",
                    e
                )))
            }
        }
    }

    /// Get the archivist connector_uid for a given connector_id
    ///
    /// This mapping enables archive-first reads to work with non-UUID connector IDs.
    /// When a connector is created with a custom ID (e.g., "opencode-1" from config),
    /// the archivist still uses a UUID internally. This method retrieves that UUID.
    ///
    /// # Arguments
    ///
    /// * `connector_id` - The connector ID (may be UUID or custom string)
    ///
    /// # Returns
    ///
    /// Some(Uuid) if the connector is registered with the archivist, None otherwise
    #[cfg(feature = "server")]
    pub async fn get_archivist_connector_uid(&self, connector_id: &str) -> Option<uuid::Uuid> {
        self.archivist_connector_uids
            .read()
            .await
            .get(connector_id)
            .copied()
    }

    /// Get the connector_id for a given archivist connector_uid (reverse lookup)
    ///
    /// This is the inverse of `get_archivist_connector_uid`. Since the map is small
    /// (typically < 10 entries), iterating is efficient.
    ///
    /// # Arguments
    /// * `uid` - The archivist connector UUID
    ///
    /// # Returns
    /// Some(String) with the connector_id if found, None otherwise
    #[cfg(feature = "server")]
    pub async fn get_connector_id_by_uid(&self, uid: uuid::Uuid) -> Option<String> {
        self.archivist_connector_uids
            .read()
            .await
            .iter()
            .find(|(_, &v)| v == uid)
            .map(|(k, _)| k.clone())
    }

    /// Backfill sessions from a connector into the archivist
    ///
    /// This method imports existing sessions from a connector that supports
    /// `list_sessions()` and `list_messages()` operations into the archivist.
    ///
    /// # Arguments
    ///
    /// * `connector_id` - The unique identifier of the connector to backfill from
    ///
    /// # Returns
    ///
    /// Statistics about the backfill operation including number of sessions and messages imported
    ///
    /// # Errors
    ///
    /// - `NotFound` - Connector with the specified ID doesn't exist
    /// - `Internal` - Archivist is not configured or other internal errors
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig};
    /// # async fn example(runtime: &CoreRuntime) -> Result<(), Box<dyn std::error::Error>> {
    /// let stats = runtime.backfill_connector_sessions("opencode-1").await?;
    /// println!("Imported {} sessions with {} messages",
    ///     stats.sessions_imported, stats.messages_imported);
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "server")]
    pub async fn backfill_connector_sessions(
        &self,
        connector_id: &str,
    ) -> Result<dirigent_archivist::BackfillStats, CoreError> {
        use futures::future::BoxFuture;
        use std::time::Duration;

        info!(connector_id = %connector_id, "Starting connector session backfill");

        // Get archivist
        let archivist = self
            .archivist
            .read()
            .await
            .clone()
            .ok_or_else(|| CoreError::Internal("Archivist not configured".to_string()))?;

        // Get connector handle
        let connector = self
            .get_connector(&connector_id.to_string())
            .await
            .ok_or_else(|| {
                error!(connector_id = %connector_id, "Connector not found for backfill");
                CoreError::NotFound
            })?;

        // Get connector UID from archivist mapping
        // This is the same UID used for archive-first reads and session registration
        let connector_uid = self
            .archivist_connector_uids
            .read()
            .await
            .get(connector_id)
            .copied()
            .ok_or_else(|| {
                error!(
                    connector_id = %connector_id,
                    "Connector not registered with archivist (no connector_uid mapping)"
                );
                CoreError::Internal(format!(
                    "Connector {} not registered with archivist. \
                     This should not happen - connector registration happens on creation.",
                    connector_id
                ))
            })?;

        debug!(
            connector_id = %connector_id,
            connector_uid = %connector_uid,
            "Using archivist connector_uid for backfill"
        );

        // Send ListSessions command and wait for response
        let mut events = connector.subscribe();
        let cmd_tx = connector.command_tx();

        cmd_tx
            .send(ConnectorCommand::ListSessions)
            .await
            .map_err(|e| {
                error!(connector_id = %connector_id, error = %e, "Failed to send ListSessions command");
                CoreError::Internal(format!("Failed to send command: {}", e))
            })?;

        // Wait for SessionsListed event with timeout
        let sessions = tokio::time::timeout(Duration::from_secs(30), async {
            while let Ok(event) = events.recv().await {
                if let dirigent_protocol::Event::SessionsListed {
                    connector_id: _,
                    sessions,
                } = event
                {
                    return Ok(sessions);
                }
            }
            Err(CoreError::Internal(
                "No SessionsListed event received".to_string(),
            ))
        })
        .await
        .map_err(|_| {
            error!(connector_id = %connector_id, "Timeout waiting for sessions list");
            CoreError::Internal("Timeout waiting for sessions list".to_string())
        })??;

        debug!(
            connector_id = %connector_id,
            session_count = sessions.len(),
            "Received {} sessions from connector",
            sessions.len()
        );

        // Create closure to fetch messages for each session
        let connector_clone = connector.clone();
        let connector_id_clone = connector_id.to_string();
        let fetch_messages = move |session_id: &str| {
            let session_id = session_id.to_string();
            let connector = connector_clone.clone();
            let connector_id = connector_id_clone.clone();

            Box::pin(async move {
                debug!(
                    connector_id = %connector_id,
                    session_id = %session_id,
                    "Fetching messages for session"
                );

                // Subscribe to events
                let mut events = connector.subscribe();
                let cmd_tx = connector.command_tx();

                // Send ListMessages command
                cmd_tx
                    .send(ConnectorCommand::ListMessages {
                        session_id: session_id.clone(),
                    })
                    .await
                    .map_err(|e| {
                        dirigent_archivist::ArchivistError::InvalidRequest(format!(
                            "Failed to send command: {}",
                            e
                        ))
                    })?;

                // Wait for MessagesListed event
                let messages = tokio::time::timeout(Duration::from_secs(30), async {
                    while let Ok(event) = events.recv().await {
                        if let dirigent_protocol::Event::MessagesListed { messages } = event {
                            // Note: MessagesListed doesn't have a session_id field
                            // We assume the messages are for the session we requested
                            return Ok(messages);
                        }
                    }
                    Err(dirigent_archivist::ArchivistError::InvalidRequest(
                        "No MessagesListed event received".to_string(),
                    ))
                })
                .await
                .map_err(|_| {
                    dirigent_archivist::ArchivistError::InvalidRequest(
                        "Timeout waiting for messages list".to_string(),
                    )
                })??;

                debug!(
                    connector_id = %connector_id,
                    session_id = %session_id,
                    message_count = messages.len(),
                    "Fetched {} messages for session",
                    messages.len()
                );

                Ok(messages)
            })
                as BoxFuture<
                    'static,
                    Result<Vec<dirigent_protocol::Message>, dirigent_archivist::ArchivistError>,
                >
        };

        // Perform backfill
        let stats = dirigent_archivist::backfill_from_sessions(
            &*archivist,
            connector_uid,
            sessions,
            fetch_messages,
        )
        .await
        .map_err(|e| {
            error!(
                connector_id = %connector_id,
                error = %e,
                "Backfill operation failed"
            );
            CoreError::Internal(format!("Backfill failed: {}", e))
        })?;

        info!(
            connector_id = %connector_id,
            sessions_imported = stats.sessions_imported,
            messages_imported = stats.messages_imported,
            errors = stats.errors.len(),
            "Connector session backfill completed"
        );

        Ok(stats)
    }

    /// Update ACP Server configuration
    ///
    /// This method updates the ACP Server configuration in the runtime's CoreConfig.
    /// After updating, you should call `save_config()` to persist the changes to disk.
    ///
    /// # Arguments
    ///
    /// * `config` - The new ACP Server configuration
    ///
    /// # Returns
    ///
    /// Ok(()) if the configuration was updated successfully.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig, AcpServerConfig};
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let runtime = CoreRuntime::new(CoreConfig::default());
    /// let acp_config = AcpServerConfig {
    ///     enabled: true,
    ///     port: 3001,
    ///     allowed_origins: None,
    ///     max_connections: 100,
    ///     default_connector_id: None,
    /// };
    /// runtime.update_acp_server_config(acp_config).await?;
    /// runtime.save_config().await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn update_acp_server_config(&self, config: AcpServerConfig) -> Result<(), CoreError> {
        config_manager::update_acp_server_config(&self.config, config).await
    }

    /// Save the current configuration to disk
    ///
    /// This method serializes the current runtime configuration to JSON or TOML format
    /// and writes it to the filesystem atomically (write to temp file, then rename).
    ///
    /// The save path is determined as follows:
    /// 1. The file the config was originally loaded from, promoted to user config dir if project-local
    /// 2. User config directory default (`~/.config/dirigent/dirigent.toml`)
    ///
    /// The file format is determined by the extension of the save path.
    ///
    /// # Returns
    ///
    /// Ok(()) if the configuration was saved successfully, or an error.
    ///
    /// # Errors
    ///
    /// Returns `CoreError::Internal` if:
    /// - Failed to serialize the configuration
    /// - Failed to write to the temp file
    /// - Failed to rename the temp file to the final path
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::{CoreRuntime, CoreConfig};
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let runtime = CoreRuntime::new(CoreConfig::default(, None));
    /// runtime.save_config().await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn save_config(&self) -> Result<(), CoreError> {
        config_manager::save_config(&self.config).await
    }

    /// Update the connector summary cache from the current connectors map
    ///
    /// This method rebuilds the sync-accessible cache used by GatewayConnector
    /// callbacks. It should be called after any connector is added or removed.
    ///
    /// The cache uses a std::sync::RwLock so it can be accessed from synchronous
    /// callback functions that don't have access to an async runtime.
    pub async fn update_connector_summary_cache(&self) {
        summary_cache::update_connector_summary_cache(
            &self.connectors,
            &self.config,
            &self.connector_summary_cache,
        )
        .await
    }

    /// Get a clone of the connector summary cache Arc
    ///
    /// This returns the Arc<std::sync::RwLock<Vec<ConnectorSummaryInfo>>> that can
    /// be shared with synchronous callbacks (e.g., in GatewayConnector).
    pub fn connector_summary_cache(&self) -> Arc<std::sync::RwLock<Vec<ConnectorSummaryInfo>>> {
        Arc::clone(&self.connector_summary_cache)
    }
}

/// Handle to a CoreRuntime
///
/// CoreHandle is a lightweight, cloneable wrapper around CoreRuntime.
/// It uses Arc internally, so cloning is cheap and all clones refer
/// to the same underlying runtime.
///
/// # Purpose
///
/// CoreHandle provides ergonomic access to the runtime:
/// - Cheap to clone and pass between tasks
/// - Implements Deref to CoreRuntime for transparent method access
/// - Can be stored in application state (e.g., Axum/Dioxus)
///
/// # Usage
///
/// ```no_run
/// use dirigent_core::{CoreRuntime, CoreConfig, CoreHandle};
///
/// # async fn example() {
/// let runtime = CoreRuntime::new(CoreConfig::default(, None));
/// let handle = CoreHandle::new(runtime);
///
/// // Clone is cheap - just increments Arc refcount
/// let handle2 = handle.clone();
///
/// // Both handles refer to the same runtime
/// let connectors1 = handle.list_connectors(None).await;
/// let connectors2 = handle2.list_connectors(None).await;
/// assert_eq!(connectors1.len(), connectors2.len());
/// # }
/// ```
#[derive(Clone)]
pub struct CoreHandle {
    /// The underlying runtime wrapped in Arc for cheap cloning
    runtime: Arc<CoreRuntime>,
}

impl CoreHandle {
    /// Create a new CoreHandle wrapping the given runtime
    ///
    /// # Arguments
    ///
    /// * `runtime` - The CoreRuntime to wrap
    ///
    /// # Returns
    ///
    /// A new CoreHandle that can be cloned and shared
    ///
    /// # Example
    ///
    /// ```no_run
    /// use dirigent_core::{CoreRuntime, CoreConfig, CoreHandle};
    ///
    /// let runtime = CoreRuntime::new(CoreConfig::default(, None));
    /// let handle = CoreHandle::new(runtime);
    /// ```
    pub fn new(runtime: CoreRuntime) -> Self {
        let arc = Arc::new(runtime);

        // Set the weak reference to self
        let weak = Arc::downgrade(&arc);
        {
            // Use try_write() instead of blocking_write() to avoid blocking the async runtime
            let mut self_weak = arc
                .self_weak
                .try_write()
                .expect("self_weak lock should not be contended during construction");
            *self_weak = Some(weak);
        } // Drop the lock guard here

        Self { runtime: arc }
    }

    /// Get a reference to the underlying Arc<CoreRuntime>
    ///
    /// This is useful when you need direct access to the Arc, for example
    /// when storing the runtime in application state.
    ///
    /// # Returns
    ///
    /// A reference to the Arc<CoreRuntime>
    pub fn inner(&self) -> &Arc<CoreRuntime> {
        &self.runtime
    }

    /// Start a background task to monitor connector state changes and handle fallbacks
    ///
    /// This task subscribes to the SharingBus and watches for connector
    /// errors and disconnections. When a connector that has transferred sessions
    /// fails, it triggers automatic fallback to Gateway.
    ///
    /// # Returns
    ///
    /// A JoinHandle for the monitoring task
    pub fn start_fallback_monitor(&self) -> tokio::task::JoinHandle<()> {
        let runtime = self.runtime.clone();

        tokio::spawn(async move {
            let mut bus_rx = runtime.sharing_bus.subscribe_all().await;

            info!("Started fallback monitor for transferred sessions");

            while let Some(bus_event) = bus_rx.rx.recv().await {
                // Check for connector failure events
                match bus_event.event.as_ref() {
                    dirigent_protocol::Event::Error { .. }
                    | dirigent_protocol::Event::Disconnected => {
                        // When we see an error or disconnect, check all connectors
                        // to see if any are in failed states with transferred sessions
                        let connectors = runtime.connectors.read().await;
                        for (id, handle) in connectors.iter() {
                            let state = handle.state();
                            runtime
                                .check_transferred_sessions_fallback(id, &state)
                                .await;
                        }
                    }
                    _ => {}
                }
            }
            info!("Bus event stream closed, stopping fallback monitor");
        })
    }
}

impl Deref for CoreHandle {
    type Target = CoreRuntime;

    fn deref(&self) -> &Self::Target {
        &self.runtime
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::{ConnectorKind, ConnectorState};
    use tokio::sync::{broadcast, mpsc};

    #[tokio::test]
    async fn test_core_runtime_new() {
        // `SharingBus::new` spawns a worker task, so we need a Tokio
        // runtime in scope. `CoreRuntime::new` itself remains sync.
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Runtime should be initialized with empty state. We can't
        // easily test internal state without async, but we can verify
        // construction succeeded.
        drop(runtime);
    }

    #[tokio::test]
    async fn test_list_connectors_empty() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        let connectors = runtime.list_connectors(None).await;
        assert_eq!(connectors.len(), 0);
    }

    #[tokio::test]
    async fn test_list_connectors_with_data() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Add a test connector
        let (cmd_tx, _cmd_rx) = mpsc::channel(100);
        let (events_tx, _events_rx) = broadcast::channel(1000);
        let handle = ConnectorHandle::new(
            "test-conn".to_string(),
            ConnectorKind::Mock,
            uuid::Uuid::nil(),
            "Test Connector".to_string(),
            cmd_tx,
            events_tx,
            serde_json::json!({}),
            None,
            None,  // icon_path
            false, // show_type_overlay
        );

        {
            let mut connectors = runtime.connectors.write().await;
            connectors.insert("test-conn".to_string(), handle);
        }

        // List all connectors
        let all = runtime.list_connectors(None).await;
        assert_eq!(all.len(), 1);
        assert_eq!(all[0].id, "test-conn");
        assert_eq!(all[0].title, "Test Connector");
        assert_eq!(all[0].owner, uuid::Uuid::nil());

        // List connectors for specific owner
        let user1 = runtime.list_connectors(Some(uuid::Uuid::nil())).await;
        assert_eq!(user1.len(), 1);

        let user2 = runtime
            .list_connectors(Some(uuid::Uuid::from_u128(2)))
            .await;
        assert_eq!(user2.len(), 0);
    }

    #[tokio::test]
    async fn test_list_connectors_filters_by_owner() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Add connectors for different users
        let (cmd_tx1, _) = mpsc::channel(100);
        let (events_tx1, _) = broadcast::channel(1000);
        let handle1 = ConnectorHandle::new(
            "conn-1".to_string(),
            ConnectorKind::Mock,
            uuid::Uuid::nil(),
            "Connector 1".to_string(),
            cmd_tx1,
            events_tx1,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        let (cmd_tx2, _) = mpsc::channel(100);
        let (events_tx2, _) = broadcast::channel(1000);
        let handle2 = ConnectorHandle::new(
            "conn-2".to_string(),
            ConnectorKind::OpenCode,
            uuid::Uuid::from_u128(2),
            "Connector 2".to_string(),
            cmd_tx2,
            events_tx2,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        {
            let mut connectors = runtime.connectors.write().await;
            connectors.insert("conn-1".to_string(), handle1);
            connectors.insert("conn-2".to_string(), handle2);
        }

        // List all
        let all = runtime.list_connectors(None).await;
        assert_eq!(all.len(), 2);

        // List for user-1
        let user1 = runtime.list_connectors(Some(uuid::Uuid::nil())).await;
        assert_eq!(user1.len(), 1);
        assert_eq!(user1[0].id, "conn-1");

        // List for user-2
        let user2 = runtime
            .list_connectors(Some(uuid::Uuid::from_u128(2)))
            .await;
        assert_eq!(user2.len(), 1);
        assert_eq!(user2[0].id, "conn-2");
    }

    #[tokio::test]
    async fn test_get_connector_exists() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Add a test connector
        let (cmd_tx, _cmd_rx) = mpsc::channel(100);
        let (events_tx, _events_rx) = broadcast::channel(1000);
        let handle = ConnectorHandle::new(
            "test-conn".to_string(),
            ConnectorKind::Mock,
            uuid::Uuid::nil(),
            "Test Connector".to_string(),
            cmd_tx,
            events_tx,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        {
            let mut connectors = runtime.connectors.write().await;
            connectors.insert("test-conn".to_string(), handle);
        }

        // Get the connector
        let result = runtime.get_connector(&"test-conn".to_string()).await;
        assert!(result.is_some());

        let retrieved = result.unwrap();
        assert_eq!(retrieved.id(), "test-conn");
        assert_eq!(retrieved.title(), "Test Connector");
    }

    #[tokio::test]
    async fn test_get_connector_not_found() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        let result = runtime.get_connector(&"nonexistent".to_string()).await;
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn test_core_handle_new() {
        // `SharingBus::new` spawns a worker task, so we need a Tokio
        // runtime in scope.
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);
        let handle = CoreHandle::new(runtime);

        // Handle should be created successfully
        drop(handle);
    }

    #[tokio::test]
    async fn test_core_handle_deref() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);
        let handle = CoreHandle::new(runtime);

        // Should be able to call runtime methods directly via Deref
        let connectors = handle.list_connectors(None).await;
        assert_eq!(connectors.len(), 0);
    }

    #[tokio::test]
    async fn test_core_handle_clone() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);
        let handle1 = CoreHandle::new(runtime);

        // Clone should be cheap and share the same runtime
        let handle2 = handle1.clone();

        // Add a connector via handle1
        let (cmd_tx, _) = mpsc::channel(100);
        let (events_tx, _) = broadcast::channel(1000);
        let connector = ConnectorHandle::new(
            "test-conn".to_string(),
            ConnectorKind::Mock,
            uuid::Uuid::nil(),
            "Test".to_string(),
            cmd_tx,
            events_tx,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        {
            let mut connectors = handle1.connectors_lock().write().await;
            connectors.insert("test-conn".to_string(), connector);
        }

        // handle2 should see the same connector
        let connectors = handle2.list_connectors(None).await;
        assert_eq!(connectors.len(), 1);
    }

    #[tokio::test]
    async fn test_core_handle_inner() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);
        let handle = CoreHandle::new(runtime);

        // Should be able to get the inner Arc
        let inner = handle.inner();
        assert!(Arc::strong_count(inner) >= 1);
    }

    // Tests for create_connector (T024)

    #[tokio::test]
    async fn test_create_connector_generates_id() {
        use crate::ConnectorConfig;
        use serde_json::json;

        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        let cfg = ConnectorConfig {
            id: None, // No ID provided
            kind: ConnectorKind::OpenCode,
            owner: None,
            title: Some("Test".to_string()),
            working_directory: None,
            params: json!({
                "base_url": "http://localhost:12225",
                "title": "Test",
                "initial_session": null
            }),
            ..Default::default()
        };

        let result = runtime.create_connector(uuid::Uuid::nil(), cfg).await;

        assert!(result.is_ok());
        let connector_id = result.unwrap();
        assert!(!connector_id.is_empty());

        // Verify connector was created
        let connector = runtime.get_connector(&connector_id).await;
        assert!(connector.is_some());
    }

    #[tokio::test]
    async fn test_create_connector_with_id() {
        use crate::ConnectorConfig;
        use serde_json::json;

        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        let cfg = ConnectorConfig {
            id: Some("my-connector".to_string()),
            kind: ConnectorKind::OpenCode,
            owner: None,
            title: Some("Test".to_string()),
            working_directory: None,
            params: json!({
                "base_url": "http://localhost:12225",
                "title": "Test",
                "initial_session": null
            }),
            ..Default::default()
        };

        let result = runtime.create_connector(uuid::Uuid::nil(), cfg).await;

        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "my-connector");
    }

    #[tokio::test]
    async fn test_create_connector_already_exists() {
        use crate::ConnectorConfig;
        use serde_json::json;

        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        let cfg = ConnectorConfig {
            id: Some("duplicate".to_string()),
            kind: ConnectorKind::OpenCode,
            owner: None,
            title: Some("Test".to_string()),
            working_directory: None,
            params: json!({
                "base_url": "http://localhost:12225",
                "title": "Test",
                "initial_session": null
            }),
            ..Default::default()
        };

        // Create first connector
        let result1 = runtime
            .create_connector(uuid::Uuid::nil(), cfg.clone())
            .await;
        assert!(result1.is_ok());

        // Try to create duplicate
        let result2 = runtime.create_connector(uuid::Uuid::nil(), cfg).await;
        assert!(result2.is_err());
        assert_eq!(result2.unwrap_err(), CoreError::AlreadyExists);
    }

    #[tokio::test]
    async fn test_create_connector_invalid_config() {
        use crate::ConnectorConfig;
        use serde_json::json;

        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        let cfg = ConnectorConfig {
            id: None,
            kind: ConnectorKind::OpenCode,
            owner: None,
            title: Some("Test".to_string()),
            working_directory: None,
            params: json!({
                // Missing required fields
                "invalid": "config"
            }),
            ..Default::default()
        };

        let result = runtime.create_connector(uuid::Uuid::nil(), cfg).await;
        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), CoreError::InvalidConfig);
    }

    #[tokio::test]
    async fn test_create_connector_mock_not_allowed() {
        use crate::ConnectorConfig;
        use serde_json::json;

        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        let cfg = ConnectorConfig {
            id: None,
            kind: ConnectorKind::Mock,
            owner: None,
            title: Some("Test".to_string()),
            working_directory: None,
            params: json!({}),
            ..Default::default()
        };

        let result = runtime.create_connector(uuid::Uuid::nil(), cfg).await;
        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), CoreError::InvalidConfig);
    }

    // Tests for stop_connector (T026)

    #[tokio::test]
    async fn test_stop_connector_not_found() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        let result = runtime.stop_connector(&"nonexistent".to_string()).await;
        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), CoreError::NotFound);
    }

    #[tokio::test]
    async fn test_stop_connector_success() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Add a mock connector
        let (cmd_tx, _cmd_rx) = mpsc::channel(100);
        let (events_tx, _events_rx) = broadcast::channel(1000);
        let handle = ConnectorHandle::new(
            "test-conn".to_string(),
            ConnectorKind::Mock,
            uuid::Uuid::nil(),
            "Test".to_string(),
            cmd_tx,
            events_tx,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        {
            let mut connectors = runtime.connectors.write().await;
            connectors.insert("test-conn".to_string(), handle);
        }

        // Stop should succeed even without a running task
        let result = runtime.stop_connector(&"test-conn".to_string()).await;
        assert!(result.is_ok());

        // Verify state was updated to Stopped
        let connector = runtime.get_connector(&"test-conn".to_string()).await;
        assert!(connector.is_some());
        assert_eq!(connector.unwrap().state(), ConnectorState::Stopped);
    }

    // Tests for remove_connector (T027)

    #[tokio::test]
    async fn test_remove_connector_not_found() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        let result = runtime.remove_connector(&"nonexistent".to_string()).await;
        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), CoreError::NotFound);
    }

    #[tokio::test]
    async fn test_remove_connector_success() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Add a mock connector
        let (cmd_tx, _cmd_rx) = mpsc::channel(100);
        let (events_tx, _events_rx) = broadcast::channel(1000);
        let handle = ConnectorHandle::new(
            "test-conn".to_string(),
            ConnectorKind::Mock,
            uuid::Uuid::nil(),
            "Test".to_string(),
            cmd_tx,
            events_tx,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        {
            let mut connectors = runtime.connectors.write().await;
            connectors.insert("test-conn".to_string(), handle);
        }

        // Verify it exists
        assert!(runtime
            .get_connector(&"test-conn".to_string())
            .await
            .is_some());

        // Remove it
        let result = runtime.remove_connector(&"test-conn".to_string()).await;
        assert!(result.is_ok());

        // Verify it's gone
        assert!(runtime
            .get_connector(&"test-conn".to_string())
            .await
            .is_none());
    }

    // Tests for send_command (T028)

    #[tokio::test]
    async fn test_send_command_not_found() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        let result = runtime
            .send_command(&"nonexistent".to_string(), ConnectorCommand::ListSessions)
            .await;
        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), CoreError::NotFound);
    }

    #[tokio::test]
    async fn test_send_command_success() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Add a mock connector
        let (cmd_tx, mut cmd_rx) = mpsc::channel(100);
        let (events_tx, _events_rx) = broadcast::channel(1000);
        let handle = ConnectorHandle::new(
            "test-conn".to_string(),
            ConnectorKind::Mock,
            uuid::Uuid::nil(),
            "Test".to_string(),
            cmd_tx,
            events_tx,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        {
            let mut connectors = runtime.connectors.write().await;
            connectors.insert("test-conn".to_string(), handle);
        }

        // Send a command
        let result = runtime
            .send_command(&"test-conn".to_string(), ConnectorCommand::ListSessions)
            .await;
        assert!(result.is_ok());

        // Verify the command was received
        let received = cmd_rx.try_recv();
        assert!(received.is_ok());
        matches!(received.unwrap(), ConnectorCommand::ListSessions);
    }

    #[tokio::test]
    async fn test_send_command_all_types() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Add a mock connector
        let (cmd_tx, mut cmd_rx) = mpsc::channel(100);
        let (events_tx, _events_rx) = broadcast::channel(1000);
        let handle = ConnectorHandle::new(
            "test-conn".to_string(),
            ConnectorKind::Mock,
            uuid::Uuid::nil(),
            "Test".to_string(),
            cmd_tx,
            events_tx,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        {
            let mut connectors = runtime.connectors.write().await;
            connectors.insert("test-conn".to_string(), handle);
        }

        // Test ListSessions
        runtime
            .send_command(&"test-conn".to_string(), ConnectorCommand::ListSessions)
            .await
            .unwrap();
        matches!(cmd_rx.recv().await.unwrap(), ConnectorCommand::ListSessions);

        // Test ListMessages
        runtime
            .send_command(
                &"test-conn".to_string(),
                ConnectorCommand::ListMessages {
                    session_id: "sess-1".to_string(),
                },
            )
            .await
            .unwrap();
        let cmd = cmd_rx.recv().await.unwrap();
        if let ConnectorCommand::ListMessages { session_id } = cmd {
            assert_eq!(session_id, "sess-1");
        } else {
            panic!("Expected ListMessages command");
        }

        // Test SendMessage
        runtime
            .send_command(
                &"test-conn".to_string(),
                ConnectorCommand::SendMessage {
                    session_id: "sess-1".to_string(),
                    text: "Hello".to_string(),
                },
            )
            .await
            .unwrap();
        let cmd = cmd_rx.recv().await.unwrap();
        if let ConnectorCommand::SendMessage { session_id, text } = cmd {
            assert_eq!(session_id, "sess-1");
            assert_eq!(text, "Hello");
        } else {
            panic!("Expected SendMessage command");
        }

        // Test Reconnect
        runtime
            .send_command(&"test-conn".to_string(), ConnectorCommand::Reconnect)
            .await
            .unwrap();
        matches!(cmd_rx.recv().await.unwrap(), ConnectorCommand::Reconnect);

        // Test Shutdown
        runtime
            .send_command(&"test-conn".to_string(), ConnectorCommand::Shutdown)
            .await
            .unwrap();
        matches!(cmd_rx.recv().await.unwrap(), ConnectorCommand::Shutdown);
    }

    // Tests for the SharingBus-based event fan-out (T029 replacement).
    //
    // Originally T029 exercised `subscribe_global()`; that API was removed
    // in Phase 4 Task 10. The replacement test verifies the SharingBus
    // fan-out directly: publishing a `BusEvent` must reach every
    // `subscribe_all()` subscriber.
    #[tokio::test]
    async fn test_sharing_bus_subscribe_all() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Two independent subscribers to the primary event fan-out.
        let mut rx1 = runtime.sharing_bus.subscribe_all().await;
        let mut rx2 = runtime.sharing_bus.subscribe_all().await;

        let event = dirigent_protocol::Event::Connected;
        let bus_event = dirigent_protocol::streaming::BusEvent {
            routing: dirigent_protocol::streaming::EventRouting::default(),
            origin: dirigent_protocol::streaming::EventOrigin::Runtime,
            event: Arc::new(event),
        };
        runtime.sharing_bus.publish(bus_event).await;

        // Both receivers should get it.
        let received1 = rx1.rx.recv().await.expect("rx1 should receive");
        let received2 = rx2.rx.recv().await.expect("rx2 should receive");

        matches!(received1.event.as_ref(), dirigent_protocol::Event::Connected);
        matches!(received2.event.as_ref(), dirigent_protocol::Event::Connected);
    }

    // Integration test: create, list, remove

    #[tokio::test]
    async fn test_integration_create_list_remove() {
        use crate::ConnectorConfig;
        use serde_json::json;

        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Initially empty
        let list = runtime.list_connectors(None).await;
        assert_eq!(list.len(), 0);

        // Create a connector
        let cfg = ConnectorConfig {
            id: Some("conn-1".to_string()),
            kind: ConnectorKind::OpenCode,
            owner: None,
            title: Some("Test 1".to_string()),
            working_directory: None,
            params: json!({
                "base_url": "http://localhost:12225",
                "title": "Test 1",
                "initial_session": null
            }),
            ..Default::default()
        };

        let connector_id = runtime
            .create_connector(uuid::Uuid::nil(), cfg)
            .await
            .unwrap();
        assert_eq!(connector_id, "conn-1");

        // List should show 1 connector
        let list = runtime.list_connectors(None).await;
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].id, "conn-1");
        assert_eq!(list[0].owner, uuid::Uuid::nil());

        // Remove the connector
        runtime.remove_connector(&connector_id).await.unwrap();

        // List should be empty again
        let list = runtime.list_connectors(None).await;
        assert_eq!(list.len(), 0);
    }

    #[tokio::test]
    async fn test_integration_multiple_connectors() {
        use crate::ConnectorConfig;
        use serde_json::json;

        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Create multiple connectors
        for i in 1..=3 {
            let cfg = ConnectorConfig {
                id: Some(format!("conn-{}", i)),
                kind: ConnectorKind::OpenCode,
                owner: None,
                title: Some(format!("Test {}", i)),
                working_directory: None,
                params: json!({
                    "base_url": "http://localhost:12225",
                    "title": format!("Test {}", i),
                    "initial_session": null
                }),
                ..Default::default()
            };

            runtime
                .create_connector(uuid::Uuid::nil(), cfg)
                .await
                .unwrap();
        }

        // List should show all 3
        let list = runtime.list_connectors(None).await;
        assert_eq!(list.len(), 3);

        // Filter by owner
        let user1_list = runtime.list_connectors(Some(uuid::Uuid::nil())).await;
        assert_eq!(user1_list.len(), 3);

        let user2_list = runtime
            .list_connectors(Some(uuid::Uuid::from_u128(2)))
            .await;
        assert_eq!(user2_list.len(), 0);

        // Remove one
        runtime
            .remove_connector(&"conn-2".to_string())
            .await
            .unwrap();

        // List should show 2
        let list = runtime.list_connectors(None).await;
        assert_eq!(list.len(), 2);
        assert!(list.iter().any(|c| c.id == "conn-1"));
        assert!(list.iter().any(|c| c.id == "conn-3"));
        assert!(!list.iter().any(|c| c.id == "conn-2"));
    }

    #[tokio::test]
    async fn test_create_connector_after_removing_all() {
        use serde_json::json;

        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Verify empty state initially
        assert_eq!(runtime.list_connectors(None).await.len(), 0);

        // Create first connector
        let cfg1 = ConnectorConfig {
            id: Some("conn-1".to_string()),
            kind: ConnectorKind::OpenCode,
            owner: None,
            title: Some("First Connector".to_string()),
            working_directory: None,
            params: json!({
                "base_url": "http://localhost:12225",
                "title": "First Connector",
                "initial_session": null
            }),
            ..Default::default()
        };

        let id1 = runtime
            .create_connector(uuid::Uuid::nil(), cfg1)
            .await
            .unwrap();
        assert_eq!(id1, "conn-1");
        assert_eq!(runtime.list_connectors(None).await.len(), 1);

        // Remove connector (transition to empty state)
        runtime.remove_connector(&id1).await.unwrap();
        assert_eq!(runtime.list_connectors(None).await.len(), 0);

        // Create new connector after empty state (this is the key test)
        let cfg2 = ConnectorConfig {
            id: Some("conn-2".to_string()),
            kind: ConnectorKind::OpenCode,
            owner: None,
            title: Some("Second Connector".to_string()),
            working_directory: None,
            params: json!({
                "base_url": "http://localhost:12225",
                "title": "Second Connector",
                "initial_session": null
            }),
            ..Default::default()
        };

        let id2 = runtime
            .create_connector(uuid::Uuid::nil(), cfg2)
            .await
            .unwrap();
        assert_eq!(id2, "conn-2");
        assert_eq!(runtime.list_connectors(None).await.len(), 1);

        // Verify new connector is functional
        let connector = runtime.get_connector(&id2).await.unwrap();
        assert_eq!(connector.state(), ConnectorState::Initializing); // Will transition to Ready
        assert_ne!(id1, id2); // Different IDs

        // Verify we can list connectors
        let list = runtime.list_connectors(None).await;
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].id, "conn-2");
        assert_eq!(list[0].title, "Second Connector");

        // Clean up
        runtime.remove_connector(&id2).await.unwrap();
        assert_eq!(runtime.list_connectors(None).await.len(), 0);
    }

    #[tokio::test]
    async fn test_rapid_remove_create_cycles() {
        use serde_json::json;

        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Perform 5 rapid remove-create cycles
        for i in 1..=5 {
            let connector_id = format!("conn-{}", i);

            // Create
            let cfg = ConnectorConfig {
                id: Some(connector_id.clone()),
                kind: ConnectorKind::OpenCode,
                owner: None,
                title: Some(format!("Connector {}", i)),
                working_directory: None,
                params: json!({
                    "base_url": "http://localhost:12225",
                    "title": format!("Connector {}", i),
                    "initial_session": null
                }),
                ..Default::default()
            };

            runtime
                .create_connector(uuid::Uuid::nil(), cfg)
                .await
                .unwrap();
            assert_eq!(runtime.list_connectors(None).await.len(), 1);

            // Verify it exists
            let connector = runtime.get_connector(&connector_id).await;
            assert!(connector.is_some());

            // Immediately remove (no delay)
            runtime.remove_connector(&connector_id).await.unwrap();
            assert_eq!(runtime.list_connectors(None).await.len(), 0);

            // Verify it's gone
            let connector = runtime.get_connector(&connector_id).await;
            assert!(connector.is_none());
        }

        // Verify clean state after all cycles
        assert_eq!(runtime.list_connectors(None).await.len(), 0);
    }

    #[tokio::test]
    async fn test_fallback_on_connector_error() {
        let config = CoreConfig::default();
        let runtime = Arc::new(CoreRuntime::new(config, None));

        // Create a Gateway connector
        let (gateway_cmd_tx, _) = mpsc::channel(100);
        let (gateway_events_tx, _) = broadcast::channel(1000);
        let gateway_handle = ConnectorHandle::new(
            "gateway-1".to_string(),
            ConnectorKind::Gateway,
            uuid::Uuid::nil(),
            "Gateway".to_string(),
            gateway_cmd_tx,
            gateway_events_tx,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        // Create an OpenCode connector
        let (opencode_cmd_tx, _) = mpsc::channel(100);
        let (opencode_events_tx, _) = broadcast::channel(1000);
        let opencode_handle = ConnectorHandle::new(
            "opencode-1".to_string(),
            ConnectorKind::OpenCode,
            uuid::Uuid::nil(),
            "OpenCode".to_string(),
            opencode_cmd_tx,
            opencode_events_tx,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        // Set OpenCode to Ready state
        {
            let state_arc = opencode_handle.state_lock();
            let mut state = state_arc.write().await;
            *state = ConnectorState::Ready;
        }

        {
            let mut connectors = runtime.connectors.write().await;
            connectors.insert("gateway-1".to_string(), gateway_handle);
            connectors.insert("opencode-1".to_string(), opencode_handle);
        }

        // Simulate a transfer
        let transfer = TransferredSession {
            gateway_session_id: "gateway-session-1".to_string(),
            gateway_connector_id: "gateway-1".to_string(),
            target_connector_id: "opencode-1".to_string(),
            target_session_id: "opencode-session-1".to_string(),
            transferred_at: std::time::Instant::now(),
        };

        {
            let mut transfers = runtime.transferred_sessions.write().await;
            transfers.insert("opencode-session-1".to_string(), transfer);
        }

        // Subscribe to the SharingBus and wait for ForwardingPanic.
        let mut bus_rx = runtime.sharing_bus.subscribe_all().await;

        // Trigger connector failure
        runtime
            .check_transferred_sessions_fallback(
                "opencode-1",
                &ConnectorState::Error("Connection lost".to_string()),
            )
            .await;

        // Verify ForwardingPanic event emitted
        let bus_event = tokio::time::timeout(Duration::from_secs(1), bus_rx.rx.recv())
            .await
            .expect("Should receive event")
            .expect("Event channel should not be closed");

        match bus_event.event.as_ref() {
            dirigent_protocol::Event::ForwardingPanic {
                connector_id,
                reason,
                fallback_gateway_session,
                ..
            } => {
                assert_eq!(connector_id, "opencode-1");
                assert!(reason.contains("Connection lost"));
                assert_eq!(
                    fallback_gateway_session.as_deref(),
                    Some("gateway-session-1")
                );
            }
            other => panic!("Expected ForwardingPanic event, got {:?}", other),
        }

        // Verify transfer record removed
        let transfers = runtime.transferred_sessions.read().await;
        assert!(transfers.is_empty());
    }

    #[tokio::test]
    async fn test_cleanup_stale_transfers() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Add old transfer
        let old_transfer = TransferredSession {
            gateway_session_id: "old-session".to_string(),
            gateway_connector_id: "gateway-1".to_string(),
            target_connector_id: "opencode-1".to_string(),
            target_session_id: "old-session-1".to_string(),
            transferred_at: std::time::Instant::now() - Duration::from_secs(3600), // 1 hour ago
        };

        {
            let mut transfers = runtime.transferred_sessions.write().await;
            transfers.insert("old-session-1".to_string(), old_transfer);
        }

        // Add recent transfer
        let recent_transfer = TransferredSession {
            gateway_session_id: "recent-session".to_string(),
            gateway_connector_id: "gateway-1".to_string(),
            target_connector_id: "opencode-2".to_string(),
            target_session_id: "recent-session-1".to_string(),
            transferred_at: std::time::Instant::now(),
        };

        {
            let mut transfers = runtime.transferred_sessions.write().await;
            transfers.insert("recent-session-1".to_string(), recent_transfer);
        }

        // Cleanup with 30 min max age
        runtime.cleanup_stale_transfers(Duration::from_secs(1800));

        // Old removed, recent kept
        let transfers = runtime.transferred_sessions.read().await;
        assert!(!transfers.contains_key("old-session-1"));
        assert!(transfers.contains_key("recent-session-1"));
    }

    #[tokio::test]
    async fn test_find_gateway_connector() {
        let config = CoreConfig::default();
        let runtime = CoreRuntime::new(config, None);

        // Initially no Gateway
        assert!(runtime.find_gateway_connector().await.is_none());

        // Add a non-Gateway connector
        let (cmd_tx, _) = mpsc::channel(100);
        let (events_tx, _) = broadcast::channel(1000);
        let opencode_handle = ConnectorHandle::new(
            "opencode-1".to_string(),
            ConnectorKind::OpenCode,
            uuid::Uuid::nil(),
            "OpenCode".to_string(),
            cmd_tx,
            events_tx,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        {
            let mut connectors = runtime.connectors.write().await;
            connectors.insert("opencode-1".to_string(), opencode_handle);
        }

        // Still no Gateway
        assert!(runtime.find_gateway_connector().await.is_none());

        // Add a Gateway connector
        let (gateway_cmd_tx, _) = mpsc::channel(100);
        let (gateway_events_tx, _) = broadcast::channel(1000);
        let gateway_handle = ConnectorHandle::new(
            "gateway-1".to_string(),
            ConnectorKind::Gateway,
            uuid::Uuid::nil(),
            "Gateway".to_string(),
            gateway_cmd_tx,
            gateway_events_tx,
            serde_json::json!({}),
            None,
            None,
            false,
        );

        // Set Gateway to Ready state
        {
            let state_arc = gateway_handle.state_lock();
            let mut state = state_arc.write().await;
            *state = ConnectorState::Ready;
        }

        {
            let mut connectors = runtime.connectors.write().await;
            connectors.insert("gateway-1".to_string(), gateway_handle);
        }

        // Now should find Gateway
        let gateway_id = runtime.find_gateway_connector().await;
        assert_eq!(gateway_id, Some("gateway-1".to_string()));
    }
}
