reovim-server 0.14.4

//! Session - a named editing context.
//!
//! # Per-Client State (Phase 11.2)
//!
//! Sessions track connected clients via the `clients` map. Each client has a role:
//! - **Owner**: Owns editing state (mode, cursor, etc.)
//! - **Follow**: Read-only spectator
//! - **Share**: Bidirectional co-edit with owner
//!
//! The `presence` map tracks display preferences (cursor position for rendering).
//! The `clients` map tracks editing roles and state ownership.

use std::collections::HashMap;

use reovim_kernel::api::v1::ServiceRegistry;

use parking_lot::RwLock;
#[cfg(feature = "grpc")]
use {reovim_protocol::v2::Notification, tokio::sync::broadcast};

#[cfg(feature = "grpc")]
use super::CaptureTracker;
#[cfg(feature = "grpc")]
use super::PresenceMap;
use {reovim_driver_session::ExtensionMap, reovim_kernel::api::v1::RegisterContent};

use super::{Client, ClientId, SessionId, SessionState};

/// Default channel capacity for notifications.
#[cfg(feature = "grpc")]
const NOTIFICATION_CHANNEL_CAPACITY: usize = 256;

/// A session is a named editing context.
///
/// Sessions hold the kernel state (buffers, options, etc.) and can have
/// multiple clients attached. Think of it like a tmux session.
///
/// # Client Management (Phase 11.2)
///
/// The session tracks connected clients via two maps:
/// - `clients`: Role and editing state ownership (Owner/Follow/Share)
/// - `presence`: Display preferences and cursor positions (for rendering)
pub struct Session {
    /// Unique session identifier.
    id: SessionId,

    /// Session state protected by `RwLock`.
    state: RwLock<SessionState>,

    /// Per-client roles and editing state (Phase 11.2).
    ///
    /// Maps `ClientId` to `Client` enum which tracks:
    /// - Owner: Has own `EditingState`
    /// - Follow: References another client (read-only)
    /// - Share: Co-edits with owner (bidirectional)
    clients: RwLock<HashMap<ClientId, Client>>,

    /// Notification broadcast channel (gRPC only).
    #[cfg(feature = "grpc")]
    notification_tx: broadcast::Sender<Notification>,

    /// Capture request tracker for CLI→Server→TUI→Server→CLI relay (gRPC only).
    #[cfg(feature = "grpc")]
    capture_tracker: CaptureTracker,

    /// Multi-client presence tracking (Phase 14, gRPC only).
    #[cfg(feature = "grpc")]
    presence: PresenceMap,
}

/// Log client disconnect with crash dump path.
///
/// Dumps the client's ring buffer to a file and logs the path via `pr_info!`.
/// The `pr_info!` macro requires a global logger (`OnceLock`) which is not
/// reliably initialized in unit tests, making this path untestable.
#[cfg_attr(coverage_nightly, coverage(off))]
fn log_client_disconnect(client_id: ClientId, ring_buffer: &super::ring_buffer::ClientRingBuffer) {
    use reovim_kernel::api::v1::pr_info;

    if let Some(path) = super::crash_dump::try_dump_client_to_file(client_id, ring_buffer) {
        pr_info!("CLIENT_DISCONNECT client_id={} dump={}", client_id.as_usize(), path.display());
    }
}

impl Session {
    /// Create a new session with the given ID.
    #[must_use]
    pub fn new(id: SessionId) -> Self {
        #[cfg(feature = "grpc")]
        let (notification_tx, _) = broadcast::channel(NOTIFICATION_CHANNEL_CAPACITY);

        Self {
            id,
            state: RwLock::new(SessionState::default()),
            clients: RwLock::new(HashMap::new()),
            #[cfg(feature = "grpc")]
            notification_tx,
            #[cfg(feature = "grpc")]
            capture_tracker: CaptureTracker::new(),
            #[cfg(feature = "grpc")]
            presence: PresenceMap::new(),
        }
    }

    /// Create a new session with a custom state.
    ///
    /// This allows the runner to inject module-initialized registries into sessions.
    /// The state should be created with populated registries from module initialization.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use reovim_server::{Session, SessionId, SessionState};
    ///
    /// // Create state with populated registries from modules
    /// let state = SessionState::with_registries(
    ///     kernel, initial_mode, vfs,
    ///     mode_registry, command_registry, keymap_registry, resolver_registry,
    ///     compositor,
    /// );
    ///
    /// let session = Session::from_state(SessionId::new("main"), state);
    /// ```
    #[must_use]
    #[allow(clippy::missing_const_for_fn)] // Contains RwLock::new which is not const
    pub fn from_state(id: SessionId, state: SessionState) -> Self {
        #[cfg(feature = "grpc")]
        let (notification_tx, _) = broadcast::channel(NOTIFICATION_CHANNEL_CAPACITY);

        Self {
            id,
            state: RwLock::new(state),
            clients: RwLock::new(HashMap::new()),
            #[cfg(feature = "grpc")]
            notification_tx,
            #[cfg(feature = "grpc")]
            capture_tracker: CaptureTracker::new(),
            #[cfg(feature = "grpc")]
            presence: PresenceMap::new(),
        }
    }

    /// Subscribe to notifications (gRPC only).
    ///
    /// Returns a receiver for the notification broadcast channel.
    /// Used by `NotificationService` to stream updates to clients.
    #[cfg(feature = "grpc")]
    #[must_use]
    pub fn subscribe_notifications(&self) -> broadcast::Receiver<Notification> {
        self.notification_tx.subscribe()
    }

    /// Emit a notification to all subscribers (gRPC only).
    ///
    /// Sends a notification to all connected clients via the broadcast channel.
    /// If no clients are subscribed, the notification is silently dropped.
    #[cfg(feature = "grpc")]
    pub fn emit_notification(&self, notification: Notification) {
        // Ignore send errors (no subscribers)
        let _ = self.notification_tx.send(notification);
    }

    /// Get the capture tracker for CLI→Server→TUI→Server→CLI relay (gRPC only).
    #[cfg(feature = "grpc")]
    #[must_use]
    pub const fn capture_tracker(&self) -> &CaptureTracker {
        &self.capture_tracker
    }

    /// Get the presence map for multi-client tracking (Phase 14, gRPC only).
    #[cfg(feature = "grpc")]
    #[must_use]
    pub const fn presence(&self) -> &PresenceMap {
        &self.presence
    }

    // =========================================================================
    // Client Management (Phase 11.2)
    // =========================================================================

    /// Add a client to the session as independent.
    ///
    /// New clients default to independent (no relation) with their own editing state.
    /// The client's mode stack is initialized with the session's home mode.
    /// The client's windows are initialized with the session's active buffer.
    /// Call `set_client_relation()` to change to Following/Sharing.
    ///
    /// # Per-Client Windows (#471)
    ///
    /// Each client gets their own `WindowLayout` with independent cursors.
    /// If the session has an active buffer, a window is created for it.
    pub fn add_client(&self, client_id: ClientId) {
        self.add_client_with_metadata(client_id, super::ClientMetadata::default());
    }

    /// Add a client with metadata.
    ///
    /// Creates an independent client with the given metadata.
    /// This is the preferred method for gRPC handlers that have client info.
    pub fn add_client_with_metadata(&self, client_id: ClientId, metadata: super::ClientMetadata) {
        use {reovim_driver_session::Window, reovim_kernel::api::v1::ModeStack};

        // Per-client state (#471, #491): Initialize new clients with session's home mode
        // stored in SessionShared. After this, the client's per-client mode stack is used.
        // active_buffer: new clients get the first kernel buffer (scratch buffer).
        let state = self.state.read();
        let home_mode = state.home_mode().clone();
        let active_buffer = state.app.kernel.buffers.list().first().copied();
        // #474: Clone shared compositor for per-client ownership
        let compositor = state
            .driver_session
            .shared
            .compositor
            .as_ref()
            .map(|c| c.boxed_clone());
        drop(state); // Release lock before acquiring clients lock

        tracing::debug!(
            %client_id,
            mode_module = %home_mode.module(),
            mode_name = %home_mode.name(),
            ?active_buffer,
            has_compositor = compositor.is_some(),
            "Initializing client with home mode and per-client windows"
        );

        let mode_stack = ModeStack::new(home_mode);
        let mut client = Client::with_mode_stack(client_id, metadata, mode_stack);

        // Per-client active_buffer: initialize with first kernel buffer
        client.state.active_buffer = active_buffer;

        // #474: Set per-client compositor and create windows with matching IDs.
        // The compositor's window IDs must match the per-client WindowLayout IDs
        // so that cursor notifications (which use WindowLayout IDs) align with
        // layout notifications (which use compositor IDs).
        if let Some(compositor) = compositor {
            if let Some(buffer_id) = active_buffer {
                let (tw, th) = client.state.terminal_size;
                let screen = reovim_driver_layout::Rect::new(0, 0, tw, th);
                let result = compositor.composite(screen);
                for p in &result.placements {
                    let window = Window::with_id_and_buffer(p.window_id, buffer_id);
                    client.state.windows.add(window);
                }
                if let Some(focused) = result.focused {
                    client.state.windows.set_active(focused);
                }
            }
            client.state.compositor = Some(compositor);
        } else if let Some(buffer_id) = active_buffer {
            // No compositor — create window with new ID (fallback)
            let window = Window::with_buffer(buffer_id);
            client.state.windows.add(window);
        }

        let mut clients = self.clients.write();
        clients.insert(client_id, client);
    }

    /// Add a client with a specific initial state.
    ///
    /// Used for restoring clients or creating clients with pre-configured state.
    pub fn add_client_with_state(&self, client: Client) {
        let mut clients = self.clients.write();
        clients.insert(client.id, client);
    }

    /// Remove a client from the session.
    ///
    /// Returns the removed client if found.
    ///
    /// # Debug Infrastructure (#481)
    ///
    /// Before removing a client, this method dumps the client's ring buffer
    /// to a file at `~/.local/share/reovim/crash/client-{id}-{timestamp}.log` for
    /// post-mortem analysis. A `CLIENT_DISCONNECT` entry is logged to the server
    /// ring buffer with the dump file path.
    pub fn remove_client(&self, client_id: ClientId) -> Option<Client> {
        let mut clients = self.clients.write();

        // Phase #481: Dump ring buffer before removal
        if let Some(client) = clients.get(&client_id) {
            log_client_disconnect(client_id, &client.ring_buffer);
        }

        clients.remove(&client_id)
    }

    /// Get a client's role (immutable).
    #[must_use]
    pub fn get_client(&self, client_id: ClientId) -> Option<Client> {
        let clients = self.clients.read();
        clients.get(&client_id).cloned()
    }

    /// Set a client's relation with validation.
    ///
    /// Use this to change between Independent/Following/Sharing modes.
    /// Pass `None` for independent, `Some(ClientRelation::Following { target })`
    /// for following, or `Some(ClientRelation::Sharing { with })` for sharing.
    ///
    /// # Validation
    ///
    /// Validates the transition:
    /// - Cannot target self
    /// - Target must exist
    /// - Cannot create cycles (A → B → A)
    /// - Following → Sharing upgrade may require cursor sync (returns `RequiresCursorSync`)
    ///
    /// # Errors
    ///
    /// Returns `Err(TransitionResult)` if:
    /// - Client not found (`TargetNotFound`)
    /// - Attempting to target self (`CannotTargetSelf`)
    /// - Change would create a cycle (`WouldCreateCycle`)
    /// - Following → Sharing requires cursor sync first (`RequiresCursorSync`)
    pub fn set_client_relation(
        &self,
        client_id: ClientId,
        relation: Option<super::ClientRelation>,
    ) -> Result<(), super::TransitionResult> {
        let mut clients = self.clients.write();

        // First validate without mutation
        let validation_result = {
            let Some(client) = clients.get(&client_id) else {
                return Err(super::TransitionResult::TargetNotFound(client_id));
            };
            Client::validate_relation_change(client, relation, &clients)
        };

        // If validation passed, apply the change
        let result = match validation_result {
            super::TransitionResult::Ok => {
                if let Some(client) = clients.get_mut(&client_id) {
                    client.set_relation_unchecked(relation);
                }
                Ok(())
            }
            other => Err(other),
        };

        drop(clients);
        result
    }

    /// Set a client's relation without validation.
    ///
    /// **Use sparingly** - prefer `set_client_relation()` for safety.
    /// This is useful for initialization where validation isn't needed.
    ///
    /// # Returns
    ///
    /// `true` if the client was found and relation set, `false` otherwise.
    pub fn set_client_relation_unchecked(
        &self,
        client_id: ClientId,
        relation: Option<super::ClientRelation>,
    ) -> bool {
        let mut clients = self.clients.write();
        clients.get_mut(&client_id).is_some_and(|client| {
            client.set_relation_unchecked(relation);
            true
        })
    }

    /// Sync cursor and set relation.
    ///
    /// Use this when `set_client_relation()` returns `RequiresCursorSync`.
    /// This syncs the cursor first, then sets the relation.
    ///
    /// # Errors
    ///
    /// Returns `Err(TransitionResult)` if:
    /// - Client or target not found (`TargetNotFound`)
    /// - Attempting to target self (`CannotTargetSelf`)
    /// - Change would create a cycle (`WouldCreateCycle`)
    pub fn sync_and_set_relation(
        &self,
        client_id: ClientId,
        target_id: ClientId,
        relation: Option<super::ClientRelation>,
    ) -> Result<(), super::TransitionResult> {
        let mut clients = self.clients.write();

        // Sync cursor first
        let target_cursor = clients
            .get(&target_id)
            .and_then(|c| c.state.windows.active())
            .map(|w| w.cursor);

        if let (Some(cursor), Some(client)) = (target_cursor, clients.get_mut(&client_id))
            && let Some(window) = client.state.windows.active_mut()
        {
            window.cursor = cursor;
        }

        // Validate without mutation
        let validation_result = {
            let Some(client) = clients.get(&client_id) else {
                return Err(super::TransitionResult::TargetNotFound(client_id));
            };
            Client::validate_relation_change(client, relation, &clients)
        };

        // If validation passed, apply the change
        let result = match validation_result {
            super::TransitionResult::Ok => {
                if let Some(client) = clients.get_mut(&client_id) {
                    client.set_relation_unchecked(relation);
                }
                Ok(())
            }
            other => Err(other),
        };

        drop(clients);
        result
    }

    /// Get the effective editing state for a client.
    ///
    /// - Owner: Returns own state
    /// - Follow: Returns target's state (read-only access)
    /// - Share: Returns owner's state (for display)
    ///
    /// Returns `None` if client not found or target chain is broken.
    #[must_use]
    pub fn client_state(&self, client_id: ClientId) -> Option<super::EditingState> {
        let clients = self.clients.read();
        clients
            .get(&client_id)
            .and_then(|c| c.effective_state(&clients))
            .cloned()
    }

    /// Update a client's editing state via closure.
    ///
    /// - Independent: Updates own state
    /// - Following: No-op (input ignored)
    /// - Sharing: Updates target's state
    ///
    /// Returns `true` if state was updated.
    pub fn update_client_state<F>(&self, client_id: ClientId, f: F) -> bool
    where
        F: FnOnce(&mut super::EditingState),
    {
        let mut clients = self.clients.write();

        // Find the target client ID based on relation
        let Some(client) = clients.get(&client_id) else {
            return false;
        };

        let target_id = match client.relation {
            None => client_id, // Independent - update own state
            Some(super::ClientRelation::Sharing { with }) => with, // Sharing - update target's state
            Some(super::ClientRelation::Following { .. }) => return false, // Following - input ignored
        };

        // Update the target's state
        if let Some(target_client) = clients.get_mut(&target_id) {
            f(&mut target_client.state);
            true
        } else {
            false
        }
    }

    /// Execute a closure with read access to the clients map.
    pub fn with_clients<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&HashMap<ClientId, Client>) -> R,
    {
        let clients = self.clients.read();
        f(&clients)
    }

    /// Execute a closure with write access to the clients map.
    pub fn with_clients_mut<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&mut HashMap<ClientId, Client>) -> R,
    {
        let mut clients = self.clients.write();
        f(&mut clients)
    }

    /// Run a closure on a client's `ExtensionMap` without cloning.
    ///
    /// `EditingState::clone()` creates an empty `ExtensionMap` because
    /// `Box<dyn SessionExtensionDyn>` is not `Clone`. This method provides
    /// direct read access to extensions through the clients lock.
    ///
    /// Respects Follow/Share relations via `effective_state()`.
    pub fn with_client_extensions<F, R>(&self, client_id: ClientId, f: F) -> Option<R>
    where
        F: FnOnce(&ExtensionMap) -> R,
    {
        let clients = self.clients.read();
        let client = clients.get(&client_id)?;
        let state = client.effective_state(&clients)?;
        let result = f(&state.extensions);
        drop(clients);
        Some(result)
    }

    /// Run a closure with mutable access to a client's `ExtensionMap`.
    ///
    /// Used by bridge lifecycle hooks that need to mutate per-client state
    /// (e.g., auto-dismiss on mode change). Respects Follow/Share relations
    /// via [`find_input_target`](Self::find_input_target).
    ///
    /// Returns `None` if the client doesn't exist or input is ignored (Following).
    pub fn with_client_extensions_mut<F, R>(&self, client_id: ClientId, f: F) -> Option<R>
    where
        F: FnOnce(&mut ExtensionMap) -> R,
    {
        let mut clients = self.clients.write();
        let target_id = Self::find_input_target(&clients, client_id)?;
        let target_client = clients.get_mut(&target_id)?;
        let result = f(&mut target_client.state.extensions);
        drop(clients);
        Some(result)
    }

    /// Execute a tick closure with mutable access to client + shared extensions (#546).
    ///
    /// Lock order: clients (write) → state (write). Same order as
    /// [`resolve_key_for_client`](Self::resolve_key_for_client).
    /// Returns `None` if client not connected or input is ignored (Following).
    ///
    /// Used by `TokioTickScheduler` for periodic state advancement.
    #[cfg(feature = "grpc")]
    pub fn with_tick_mut<F, R>(&self, client_id: ClientId, f: F) -> Option<R>
    where
        F: FnOnce(&mut ExtensionMap, &mut ExtensionMap, &ServiceRegistry) -> R,
    {
        let mut clients = self.clients.write();
        let target_id = Self::find_input_target(&clients, client_id)?;
        let target_client = clients.get_mut(&target_id)?;

        let mut state = self.state.write();
        // Clone the Arc before taking mutable borrows on extensions (#555).
        let services = std::sync::Arc::clone(&state.app.services);
        let result = f(&mut target_client.state.extensions, &mut state.app.extensions, &services);
        drop(state);
        drop(clients);
        Some(result)
    }

    /// Get count of connected clients.
    #[must_use]
    pub fn client_count(&self) -> usize {
        self.clients.read().len()
    }

    /// Check if a client is connected.
    #[must_use]
    pub fn has_client(&self, client_id: ClientId) -> bool {
        self.clients.read().contains_key(&client_id)
    }

    /// Get the session ID.
    #[must_use]
    pub const fn id(&self) -> &SessionId {
        &self.id
    }

    /// Execute a closure with read access to the session state.
    ///
    /// This is the primary way to query session data.
    ///
    /// Note: Currently synchronous but kept async for future I/O operations.
    #[allow(clippy::unused_async)]
    pub async fn with_state<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&SessionState) -> R,
    {
        let state = self.state.read();
        f(&state)
    }

    /// Execute a closure with write access to the session state.
    ///
    /// Use this for mutations like inserting text, moving cursor, etc.
    ///
    /// Note: Currently synchronous but kept async for future I/O operations.
    #[allow(clippy::unused_async)]
    pub async fn with_state_mut<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&mut SessionState) -> R,
    {
        let mut state = self.state.write();
        f(&mut state)
    }

    /// Synchronous read access (for contexts where async isn't needed).
    pub fn with_state_sync<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&SessionState) -> R,
    {
        let state = self.state.read();
        f(&state)
    }

    /// Synchronous write access.
    pub fn with_state_mut_sync<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&mut SessionState) -> R,
    {
        let mut state = self.state.write();
        f(&mut state)
    }

    /// Execute a closure with combined read access to a client's extensions,
    /// shared extensions, and pre-collected opponent extension maps (#543).
    ///
    /// Acquires locks in established order: `clients` (read) first, then `state`
    /// (read). Resolves `effective_state()` for all clients to collect opponent
    /// data as driver-layer `ClientId` + `&ExtensionMap` pairs.
    ///
    /// Returns `None` if `client_id` is not connected or has no effective state.
    pub fn with_bridge_context<F, R>(&self, client_id: ClientId, f: F) -> Option<R>
    where
        F: FnOnce(
            &ExtensionMap,
            &ExtensionMap,
            &[(reovim_driver_session::ClientId, &ExtensionMap)],
        ) -> R,
    {
        let clients = self.clients.read();
        let client = clients.get(&client_id)?;
        let own_ext = &client.effective_state(&clients)?.extensions;

        // Pre-collect opponent extension maps with driver-layer ClientId.
        // The driver crate cannot see `Client`, so we resolve here.
        let opponents: Vec<(reovim_driver_session::ClientId, &ExtensionMap)> = clients
            .iter()
            .filter(|&(&id, _)| id != client_id)
            .filter_map(|(&id, c)| {
                c.effective_state(&clients).map(|state| {
                    (reovim_driver_session::ClientId::new(id.as_usize()), &state.extensions)
                })
            })
            .collect();

        let state = self.state.read();
        let shared_ext = &state.app.extensions;
        let result = f(own_ext, shared_ext, &opponents);
        drop(state);
        drop(clients);
        Some(result)
    }

    // =========================================================================
    // Per-Client Key Resolution (#471)
    // =========================================================================

    /// Ensure a client's per-client windows are populated.
    ///
    /// When a client joins before any buffers exist, their windows are empty.
    /// Later, when a buffer is created (e.g., via `:e`), only the shared session
    /// windows are updated. This helper syncs per-client windows with the session's
    /// active buffer when needed.
    ///
    /// # When this matters
    ///
    /// 1. Client connects (no buffers yet) → empty windows
    /// 2. `:e filename` creates buffer → shared windows updated
    /// 3. Client tries to move cursor → per-client windows still empty!
    ///
    /// This helper fixes step 3 by creating a window for the active buffer.
    fn ensure_client_has_window(editing_state: &mut super::EditingState) {
        use reovim_driver_session::Window;

        // Only sync if per-client windows are empty AND client has an active buffer
        if editing_state.windows.is_empty()
            && let Some(buffer_id) = editing_state.active_buffer
        {
            // #474: If per-client compositor exists, create windows with matching IDs
            if let Some(ref compositor) = editing_state.compositor {
                let (tw, th) = editing_state.terminal_size;
                let screen = reovim_driver_layout::Rect::new(0, 0, tw, th);
                let result = compositor.composite(screen);
                for p in &result.placements {
                    let window = Window::with_id_and_buffer(p.window_id, buffer_id);
                    editing_state.windows.add(window);
                }
                if let Some(focused) = result.focused {
                    editing_state.windows.set_active(focused);
                }
            } else {
                let window = Window::with_buffer(buffer_id);
                editing_state.windows.add(window);
            }
            tracing::debug!(?buffer_id, "Synced per-client windows with active buffer");
        }
    }

    /// Resolve a key with per-client mode stack (#471).
    ///
    /// This method provides access to both session state AND per-client mode stack,
    /// enabling multi-client mode isolation. The key is resolved using the client's
    /// mode stack instead of the shared session mode stack.
    ///
    /// # Arguments
    ///
    /// * `client_id` - Client ID to resolve for
    /// * `key` - Key event to resolve
    ///
    /// # Returns
    ///
    /// - `Some((ResolveResult, StateChanges))` - if key was resolved
    /// - `None` - if client not found, client is Following, or no resolver
    ///
    /// # Relation Behavior
    ///
    /// - **Independent**: Uses own mode stack and windows
    /// - **Following**: Returns `None` (input ignored)
    /// - **Sharing**: Uses target's mode stack and windows
    #[allow(clippy::unused_async, clippy::significant_drop_tightening)]
    pub async fn resolve_key_for_client(
        &self,
        client_id: ClientId,
        key: &reovim_driver_input::KeyEvent,
    ) -> Option<(reovim_driver_input::ResolveResult, reovim_driver_session::api::StateChanges)>
    {
        // Acquire both locks in consistent order to avoid deadlocks
        let mut clients = self.clients.write();
        let mut state = self.state.write();

        // Find the target client ID based on relation
        let target_id = Self::find_input_target(&clients, client_id)?;

        // Phase #471/#477/#480: Get mutable references to per-client state
        let target_client = clients.get_mut(&target_id)?;
        let editing_state = &mut target_client.state;

        // Ensure per-client windows are populated (fixes buffer-after-client-join issue)
        Self::ensure_client_has_window(editing_state);

        // Resolve key with per-client state (#471 Phase 5: pass client_id for undo origin)
        state.resolve_key_for_client(target_id.as_usize(), editing_state.client_context(), key)
    }

    /// Try `on_command_complete` with per-client state (#471, #477).
    ///
    /// Like `resolve_key_for_client`, but for post-command mode transitions.
    #[allow(clippy::unused_async, clippy::significant_drop_tightening)]
    pub async fn try_on_command_complete_for_client(
        &self,
        client_id: ClientId,
    ) -> Option<reovim_driver_input::ModeTransition> {
        // Acquire both locks in consistent order
        let mut clients = self.clients.write();
        let mut state = self.state.write();

        // Find the target client ID based on relation
        let target_id = Self::find_input_target(&clients, client_id)?;

        // Phase #471/#477/#480: Get mutable references to per-client state
        let target_client = clients.get_mut(&target_id)?;
        let editing_state = &mut target_client.state;

        // Ensure per-client windows are populated (fixes buffer-after-client-join issue)
        Self::ensure_client_has_window(editing_state);

        state.try_on_command_complete_for_client(
            target_id.as_usize(),
            editing_state.client_context(),
        )
    }

    /// Execute a command with per-client state (Phase #471).
    ///
    /// This enables multi-client mode isolation by operating on per-client
    /// mode and cursor state instead of shared session state.
    ///
    /// # Arguments
    ///
    /// * `client_id` - Client ID to execute for
    /// * `cmd_id` - Command ID to execute
    /// * `args` - Command arguments (count, register, etc.)
    ///
    /// # Returns
    ///
    /// - `Some((CommandResult, StateChanges))` - if command executed
    /// - `None` - if client not found, client is Following, or command not registered
    ///
    /// # Relation Behavior
    ///
    /// - **Independent**: Uses own mode stack, windows, and extensions
    /// - **Following**: Returns `None` (input ignored)
    /// - **Sharing**: Uses target's state
    #[allow(clippy::significant_drop_tightening)]
    pub fn execute_command_for_client(
        &self,
        client_id: ClientId,
        cmd_id: &reovim_kernel::api::v1::CommandId,
        args: &reovim_driver_command_types::CommandContext,
    ) -> Option<(
        reovim_driver_command::CommandResult,
        reovim_driver_session::api::StateChanges,
        Vec<reovim_driver_command_types::RuntimeSignal>,
    )> {
        // Acquire both locks in consistent order to avoid deadlocks
        let mut clients = self.clients.write();
        let mut state = self.state.write();

        // Find the target client ID based on relation
        let target_id = Self::find_input_target(&clients, client_id)?;

        // Phase #471/#477/#480: Get mutable references to per-client state
        let target_client = clients.get_mut(&target_id)?;
        let editing_state = &mut target_client.state;

        // Ensure per-client windows are populated (fixes buffer-after-client-join issue)
        Self::ensure_client_has_window(editing_state);

        // Execute command with per-client state, passing client_id for per-client undo (#471, #515)
        state.execute_command_for_client(
            target_id.as_usize(),
            editing_state.client_context(),
            cmd_id,
            args,
        )
    }

    /// Insert a character for a client, checking per-client extensions first (#477).
    ///
    /// For `InputTarget::Buffer`, inserts into the active buffer.
    /// For `InputTarget::Extension(type_id)`, looks in per-client extensions first,
    /// then falls back to shared session extensions.
    ///
    /// # Returns
    ///
    /// - `Some((BufferId, Some(Modification)))` if character was inserted into a buffer
    /// - `None` if inserted into extension or failed
    #[allow(clippy::significant_drop_tightening)]
    pub fn insert_char_for_client(
        &self,
        client_id: ClientId,
        ch: char,
        target: reovim_driver_input::InputTarget,
    ) -> Option<(
        reovim_kernel::api::v1::BufferId,
        Option<reovim_kernel::api::v1::events::kernel::Modification>,
    )> {
        use {
            reovim_driver_input::InputTarget,
            reovim_driver_undo::{UndoKey, UndoProviderRegistry},
            reovim_kernel::api::v1::{Edit, Position, events::kernel::Modification},
        };

        match target {
            InputTarget::Buffer => {
                // Insert into active buffer at client's cursor position
                // active_buffer is per-client (#471)
                let clients = self.clients.read();
                let buffer_id = clients.get(&client_id)?.state.active_buffer?;
                drop(clients);

                let state = self.state.read();
                let buffer_arc = state.buffer(buffer_id)?;

                // Get undo registry for recording edit (#471)
                let undo_registry = state.app.kernel.services.get::<UndoProviderRegistry>();

                drop(state); // Release lock before getting client

                // Get cursor position from client's window
                let mut clients = self.clients.write();
                let client = clients.get_mut(&client_id)?;
                let active_window = client.state.windows.active_mut()?;
                let cursor_before =
                    Position::new(active_window.cursor.line, active_window.cursor.column);

                // Compute start_byte BEFORE the mutation for incremental syntax parsing
                let start_byte = buffer_arc.read().position_to_byte(cursor_before);

                tracing::debug!(?buffer_id, ?ch, ?cursor_before, "Inserting into buffer");
                let ch_str = ch.to_string();
                buffer_arc.write().insert_at(cursor_before, &ch_str);

                // Update cursor position after insertion
                // For regular characters, move cursor one position right
                // For newlines, move to start of next line
                if ch == '\n' {
                    active_window.cursor.line += 1;
                    active_window.cursor.column = 0;
                } else {
                    active_window.cursor.column += 1;
                }

                let cursor_after =
                    Position::new(active_window.cursor.line, active_window.cursor.column);

                drop(clients);

                // Build Modification for incremental syntax parsing
                #[allow(clippy::cast_possible_truncation)] // cursor positions fit in u32
                let modification = Modification::Insert {
                    start: (cursor_before.line as u32, cursor_before.column as u32),
                    text: ch_str.clone(),
                    start_byte,
                };

                // Record edit for undo with client origin (#471)
                if let Some(undo_reg) = undo_registry
                    && let Some(undo_provider) = undo_reg.get(&UndoKey::Buffer)
                {
                    let edit = Edit::Insert {
                        position: cursor_before,
                        text: ch_str,
                    };
                    undo_provider.record_for_client(
                        buffer_id,
                        client_id.as_usize(),
                        vec![edit],
                        cursor_before,
                        cursor_after,
                    );
                }

                Some((buffer_id, Some(modification)))
            }
            InputTarget::Extension(type_id) => {
                // Phase #477: Check per-client extensions FIRST, then shared
                tracing::debug!(?type_id, ?ch, %client_id, "Routing to extension via TextInputSink");

                // Try per-client extensions first
                let mut clients = self.clients.write();
                if let Some(client) = clients.get_mut(&client_id)
                    && let Some(sink) = client.state.extensions.get_text_input_sink_by_id(type_id)
                {
                    sink.insert_char(ch);
                    tracing::debug!(?type_id, "Inserted char via per-client extension");
                    return None;
                }
                drop(clients);

                // Fallback to shared session extensions (#491)
                let mut state = self.state.write();
                if let Some(sink) = state.app.extensions.get_text_input_sink_by_id(type_id) {
                    sink.insert_char(ch);
                    tracing::debug!(?type_id, "Inserted char via shared extension");
                } else {
                    tracing::warn!(
                        ?type_id,
                        "Extension not found or doesn't implement TextInputSink"
                    );
                }
                drop(state); // Release lock early
                None
            }
        }
    }

    /// Get the current mode for a specific client (#471).
    ///
    /// Returns the mode from the client's per-client mode stack (if Independent/Sharing)
    /// or `None` for Following clients.
    #[must_use]
    pub fn client_current_mode(
        &self,
        client_id: ClientId,
    ) -> Option<reovim_kernel::api::v1::ModeId> {
        let clients = self.clients.read();

        // Find the target client ID based on relation
        let target_id = Self::find_input_target(&clients, client_id)?;

        // Get mode from target's mode stack
        clients
            .get(&target_id)
            .map(|c| c.state.mode_stack.current().clone())
    }

    /// Find the target client ID for input routing.
    ///
    /// - Independent: returns self
    /// - Following: returns None (input ignored)
    /// - Sharing: returns target
    fn find_input_target(
        clients: &HashMap<ClientId, Client>,
        client_id: ClientId,
    ) -> Option<ClientId> {
        let client = clients.get(&client_id)?;
        if client.is_independent() {
            Some(client_id)
        } else if client.is_sharing() {
            client.target_id()
        } else {
            // Following - input ignored
            None
        }
    }

    /// Get access to a client's ring buffer.
    ///
    /// Returns `None` if the client doesn't exist.
    pub fn with_client_ring_buffer<F, R>(&self, client_id: ClientId, f: F) -> Option<R>
    where
        F: FnOnce(&super::ring_buffer::ClientRingBuffer) -> R,
    {
        let clients = self.clients.read();
        clients.get(&client_id).map(|c| f(&c.ring_buffer))
    }

    /// Dump a client's ring buffer for debugging.
    ///
    /// Returns `None` if the client doesn't exist.
    #[must_use]
    pub fn dump_client_ring_buffer(&self, client_id: ClientId) -> Option<String> {
        self.with_client_ring_buffer(client_id, super::ring_buffer::ClientRingBuffer::dump)
    }

    // ========================================================================
    // Session-scoped registers (#515 Phase 5)
    // ========================================================================

    /// Get a session-shared register.
    ///
    /// Returns `None` if the register has not been set. Session registers
    /// are shared across all clients in this session.
    #[must_use]
    pub fn get_session_register(&self, key: char) -> Option<RegisterContent> {
        let state = self.state.read();
        state.session_registers.get(&key).cloned()
    }

    /// Set a session-shared register.
    ///
    /// The content is immediately visible to all clients in this session.
    pub fn set_session_register(&self, key: char, content: RegisterContent) {
        let mut state = self.state.write();
        state.session_registers.insert(key, content);
    }

    /// Read another client's history ring entry (`PeerHistory`).
    ///
    /// Returns `None` if the client doesn't exist or the index is out of range.
    /// This is a read-only operation - you cannot modify another client's history.
    #[must_use]
    pub fn get_peer_history(&self, client_id: ClientId, index: u8) -> Option<RegisterContent> {
        let clients = self.clients.read();
        clients
            .get(&client_id)
            .and_then(|client| client.state.clipboard_history.get_by_index(index))
            .cloned()
    }

    /// List connected client IDs (for peer history navigation).
    ///
    /// Returns a sorted list of client IDs currently in this session.
    #[must_use]
    pub fn connected_client_ids(&self) -> Vec<ClientId> {
        let mut ids: Vec<_> = self.clients.read().keys().copied().collect();
        ids.sort_unstable_by_key(ClientId::as_usize);
        ids
    }
}

#[cfg(test)]
#[path = "session_tests.rs"]
mod tests;