rmux-server 0.10.0

use std::collections::HashMap;
use std::future::Future;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use rmux_core::LifecycleEvent;
use rmux_os::identity::UserIdentity;
use rmux_proto::{SessionId, TerminalSize};
use tokio::sync::{mpsc, watch};

use super::{
    attach_support::{switch_control_session_for_client, ControlResizeClient},
    client_support::SwitchTargetSelection,
    current_client_activity_timestamp, update_environment_from_client, QueuedLifecycleEvent,
    RequestHandler, SelectionTargetTransitionSnapshot,
};
use crate::client_names::control_client_name;
use crate::control::{
    ControlClientFlags, ControlCommandResponseEvent, ControlModeUpgrade, ControlServerEvent,
};
use crate::control_notifications::{collect_control_notifications, ControlClientSnapshot};
use crate::handler_support::{ambiguous_attached_client, attached_client_required};
use crate::outer_terminal::OuterTerminalContext;
use crate::pane_io::PaneOutputSender;
use crate::pane_terminals::HandlerState;
use crate::server_access::ServerAccessAdmission;
#[cfg(test)]
use crate::server_access::{
    current_owner_uid, pause_before_access_registration, AccessRegistrationKind,
};

#[path = "handler_control/session_attach.rs"]
mod session_attach;

const CONTROL_QUEUE_DRAIN_REGISTRATION_TIMEOUT: Duration = Duration::from_secs(5);
const DEFAULT_CONTROL_CLIENT_WIDTH: u16 = 80;
const DEFAULT_CONTROL_CLIENT_HEIGHT: u16 = 24;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ControlRegistrationError {
    QueueDrainTimedOut { requester_pid: u32 },
    AccessRevoked,
}

impl ControlRegistrationError {
    pub(crate) fn into_rmux_error(self) -> rmux_proto::RmuxError {
        match self {
            Self::QueueDrainTimedOut { requester_pid } => rmux_proto::RmuxError::Server(format!(
                "timed out waiting for the previous control queue for client {requester_pid} to drain"
            )),
            Self::AccessRevoked => {
                rmux_proto::RmuxError::Server("access not allowed".to_owned())
            }
        }
    }
}

#[derive(Debug, Default)]
pub(super) struct ActiveControlState {
    next_id: u64,
    pub(super) by_pid: HashMap<u32, ActiveControl>,
}

#[derive(Debug)]
pub(super) struct ActiveControl {
    pub(super) id: u64,
    pub(super) last_activity_sequence: u64,
    pub(super) activity_at: i64,
    pub(super) client_width: u16,
    pub(super) client_height: u16,
    /// Whether this client ever announced a size with `refresh-client -C`.
    ///
    /// A control client owns no window geometry until it has: tmux 3.7b's
    /// `ignore_client_size()` skips a `CLIENT_CONTROL` client that has no
    /// `CLIENT_SIZECHANGED`, so `client_width`/`client_height` stay at
    /// `DEFAULT_CONTROL_CLIENT_WIDTH`/`DEFAULT_CONTROL_CLIENT_HEIGHT` — a
    /// placeholder for reporting only — and never reach a policy. Measured
    /// 2026-07-25 across `window-size` latest/largest/smallest/manual: a
    /// control client attaching to a 200x50 session leaves it at 200x50 in
    /// tmux, whether or not another client is attached.
    pub(super) size_declared: bool,
    pub(super) size_sequence: u64,
    pub(super) session_name: Option<rmux_proto::SessionName>,
    pub(super) session_id: Option<SessionId>,
    pub(super) last_session: Option<rmux_proto::SessionName>,
    pub(super) last_session_id: Option<SessionId>,
    pub(super) flags: ControlClientFlags,
    pub(super) uid: u32,
    pub(super) user: UserIdentity,
    pub(super) can_write: bool,
    pub(super) terminal_context: OuterTerminalContext,
    event_tx: mpsc::Sender<ControlServerEvent>,
    pub(super) closing: Arc<AtomicBool>,
    client_detached_prepared: bool,
    queue_draining: bool,
    queue_drain_finished: watch::Sender<bool>,
}

pub(crate) struct ControlRegistration {
    pub(crate) event_tx: mpsc::Sender<ControlServerEvent>,
    pub(crate) closing: Arc<AtomicBool>,
    pub(crate) uid: u32,
    pub(crate) user: UserIdentity,
    pub(crate) can_write: bool,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ControlQueueDrainLease {
    Acquired,
    Unavailable,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum ManagedClient {
    Attach { pid: u32, attach_id: u64 },
    Control(ControlClientIdentity),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ControlClientIdentity {
    requester_pid: u32,
    control_id: u64,
}

#[derive(Debug, Clone)]
pub(crate) struct ControlCommandResponseSink {
    identity: ControlClientIdentity,
    sender: mpsc::Sender<ControlCommandResponseEvent>,
    closing: Arc<AtomicBool>,
}

pub(super) struct ControlClientDetachOutcome {
    pub(in crate::handler) lifecycle_event: Option<QueuedLifecycleEvent>,
}

pub(super) struct ControlClientsDetachOutcome {
    pub(in crate::handler) lifecycle_events: Vec<QueuedLifecycleEvent>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ControlOutputStart {
    Current,
    Oldest,
}

/// Whether the caller wants `client-session-changed` published at the commit
/// point, before any window geometry the session change reconciles.
///
/// tmux 3.7b, measured 2026-07-25 with two control clients on one session
/// (101x41 and 60x20, `window-size largest`): after the 101x41 client runs
/// `switch-client -t target`, the surviving client receives
///     %client-session-changed client-71338 $1 target
///     %layout-change @0 a1dd,60x20,0,0,0 a1dd,60x20,0,0,0 *
/// and the switching client receives `%session-changed $1 target` before its
/// own `%layout-change @1 ...`. Both orders come from the same notification, so
/// it has to be published before the reconciles.
#[derive(Clone, Copy, PartialEq, Eq)]
enum ControlSessionChangedNotice {
    Skip,
    Publish,
    PublishIfChanged,
}

struct ControlSessionUpdate<'a> {
    target_selection: Option<SwitchTargetSelection>,
    client_environment: Option<&'a HashMap<String, String>>,
    output_start: ControlOutputStart,
    session_changed_notice: ControlSessionChangedNotice,
}

impl<'a> ControlSessionUpdate<'a> {
    fn existing(
        target_selection: Option<SwitchTargetSelection>,
        client_environment: Option<&'a HashMap<String, String>>,
    ) -> Self {
        Self {
            target_selection,
            client_environment,
            output_start: ControlOutputStart::Current,
            session_changed_notice: ControlSessionChangedNotice::Skip,
        }
    }

    fn switched(
        target_selection: Option<SwitchTargetSelection>,
        client_environment: Option<&'a HashMap<String, String>>,
    ) -> Self {
        Self {
            session_changed_notice: ControlSessionChangedNotice::Publish,
            ..Self::existing(target_selection, client_environment)
        }
    }

    fn attached_existing() -> Self {
        Self {
            session_changed_notice: ControlSessionChangedNotice::PublishIfChanged,
            ..Self::existing(None, None)
        }
    }

    fn created() -> Self {
        Self {
            target_selection: None,
            client_environment: None,
            output_start: ControlOutputStart::Oldest,
            session_changed_notice: ControlSessionChangedNotice::Skip,
        }
    }
}

struct DestroyControlSwitchCommitOutcome {
    target_session_name: rmux_proto::SessionName,
    client_session_changed: QueuedLifecycleEvent,
}

#[cfg(test)]
#[derive(Debug, Default)]
pub(in crate::handler) struct ControlSwitchPostCommitPause {
    pub(in crate::handler) reached: tokio::sync::Notify,
    pub(in crate::handler) release: tokio::sync::Notify,
}

#[cfg(test)]
static CONTROL_SWITCH_POST_COMMIT_PAUSE: std::sync::Mutex<
    Vec<(u32, Arc<ControlSwitchPostCommitPause>)>,
> = std::sync::Mutex::new(Vec::new());

impl ControlClientIdentity {
    pub(crate) const fn new(requester_pid: u32, control_id: u64) -> Self {
        Self {
            requester_pid,
            control_id,
        }
    }

    pub(crate) const fn requester_pid(self) -> u32 {
        self.requester_pid
    }

    pub(crate) const fn control_id(self) -> u64 {
        self.control_id
    }
}

impl ControlCommandResponseSink {
    pub(crate) fn new(
        identity: ControlClientIdentity,
        sender: mpsc::Sender<ControlCommandResponseEvent>,
        closing: Arc<AtomicBool>,
    ) -> Self {
        Self {
            identity,
            sender,
            closing,
        }
    }

    fn try_send(&self, event: ControlCommandResponseEvent) -> bool {
        if self.closing.load(Ordering::SeqCst) {
            return false;
        }
        if self.sender.try_send(event).is_ok() {
            return true;
        }
        self.closing.store(true, Ordering::SeqCst);
        false
    }
}

tokio::task_local! {
    static CONTROL_QUEUE_IDENTITY: ControlClientIdentity;
    static CONTROL_QUEUE_EOF_CANCELLATION: ControlQueueEofCancellation;
    static CONTROL_COMMAND_RESPONSE_SINK: ControlCommandResponseSink;
    static CONTROL_COMMAND_RESPONSE_CAPTURE: ();
}

#[derive(Debug, Clone)]
pub(crate) struct ControlQueueEofCancellation {
    identity: ControlClientIdentity,
    cancelled: watch::Sender<bool>,
    wait_cancelled: Arc<AtomicBool>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(in crate::handler) enum ControlQueueEofAction {
    Continue,
    StopFrame,
}

impl ControlQueueEofCancellation {
    pub(crate) fn new(identity: ControlClientIdentity) -> Self {
        let (cancelled, _receiver) = watch::channel(false);
        Self {
            identity,
            cancelled,
            wait_cancelled: Arc::new(AtomicBool::new(false)),
        }
    }

    pub(crate) fn cancel_for_eof(&self) {
        self.cancelled.send_replace(true);
    }

    pub(in crate::handler) fn is_cancelled(&self) -> bool {
        *self.cancelled.borrow()
    }

    pub(in crate::handler) fn mark_wait_cancelled(&self) {
        self.wait_cancelled.store(true, Ordering::SeqCst);
    }

    fn action(&self) -> ControlQueueEofAction {
        if self.wait_cancelled.load(Ordering::SeqCst) {
            ControlQueueEofAction::StopFrame
        } else {
            ControlQueueEofAction::Continue
        }
    }

    pub(in crate::handler) async fn cancelled(&self) {
        let mut receiver = self.cancelled.subscribe();
        if *receiver.borrow_and_update() {
            return;
        }
        let _ = receiver.wait_for(|cancelled| *cancelled).await;
    }
}

pub(crate) async fn with_control_queue_identity<T, F>(
    identity: ControlClientIdentity,
    future: F,
) -> T
where
    F: Future<Output = T>,
{
    CONTROL_QUEUE_IDENTITY.scope(identity, future).await
}

pub(crate) async fn with_control_queue_eof_cancellation<T, F>(
    cancellation: ControlQueueEofCancellation,
    future: F,
) -> T
where
    F: Future<Output = T>,
{
    CONTROL_QUEUE_EOF_CANCELLATION
        .scope(cancellation, future)
        .await
}

pub(crate) async fn with_control_command_response_sink<T, F>(
    sink: ControlCommandResponseSink,
    future: F,
) -> T
where
    F: Future<Output = T>,
{
    CONTROL_COMMAND_RESPONSE_SINK.scope(sink, future).await
}

pub(in crate::handler) async fn with_control_command_response_capture<T, F>(future: F) -> T
where
    F: Future<Output = T>,
{
    CONTROL_COMMAND_RESPONSE_CAPTURE.scope((), future).await
}

fn current_control_command_response_sink() -> Option<ControlCommandResponseSink> {
    CONTROL_COMMAND_RESPONSE_CAPTURE.try_with(|()| ()).ok()?;
    CONTROL_COMMAND_RESPONSE_SINK.try_with(Clone::clone).ok()
}

pub(crate) fn control_command_response_stream_is_active(identity: ControlClientIdentity) -> bool {
    CONTROL_COMMAND_RESPONSE_SINK
        .try_with(|sink| sink.identity == identity && !sink.closing.load(Ordering::SeqCst))
        .unwrap_or(false)
}

pub(crate) fn send_control_command_response_event(
    identity: ControlClientIdentity,
    event: ControlCommandResponseEvent,
) -> bool {
    CONTROL_COMMAND_RESPONSE_SINK
        .try_with(|sink| (sink.identity == identity).then(|| sink.try_send(event)))
        .ok()
        .flatten()
        .unwrap_or(false)
}

pub(in crate::handler) fn current_control_queue_identity(
    requester_pid: u32,
) -> Option<ControlClientIdentity> {
    CONTROL_QUEUE_IDENTITY
        .try_with(|identity| (identity.requester_pid() == requester_pid).then_some(*identity))
        .ok()
        .flatten()
}

pub(in crate::handler) fn current_control_queue_eof_cancellation(
) -> Option<ControlQueueEofCancellation> {
    let identity = CONTROL_QUEUE_IDENTITY.try_with(|identity| *identity).ok()?;
    CONTROL_QUEUE_EOF_CANCELLATION
        .try_with(|cancellation| (cancellation.identity == identity).then(|| cancellation.clone()))
        .ok()
        .flatten()
}

pub(in crate::handler) fn control_queue_eof_action(
    identity: Option<ControlClientIdentity>,
) -> ControlQueueEofAction {
    let identity = identity.or_else(|| CONTROL_QUEUE_IDENTITY.try_with(|identity| *identity).ok());
    let Some(identity) = identity else {
        return ControlQueueEofAction::Continue;
    };
    CONTROL_QUEUE_EOF_CANCELLATION
        .try_with(|cancellation| {
            if cancellation.identity == identity {
                cancellation.action()
            } else {
                ControlQueueEofAction::Continue
            }
        })
        .unwrap_or(ControlQueueEofAction::Continue)
}

impl RequestHandler {
    #[cfg(test)]
    pub(crate) async fn register_control_with_closing(
        &self,
        requester_pid: u32,
        upgrade: ControlModeUpgrade,
        event_tx: mpsc::Sender<ControlServerEvent>,
        closing: Arc<AtomicBool>,
    ) -> u64 {
        self.register_control_with_access(
            requester_pid,
            upgrade,
            ControlRegistration {
                event_tx,
                closing,
                uid: current_owner_uid(),
                user: self.server_owner_identity(),
                can_write: true,
            },
        )
        .await
        .expect("test control registration must not outlive the bounded queue drain")
    }

    #[cfg(test)]
    pub(crate) async fn register_control_with_access(
        &self,
        requester_pid: u32,
        upgrade: ControlModeUpgrade,
        registration: ControlRegistration,
    ) -> Result<u64, ControlRegistrationError> {
        self.register_control_with_access_timeout(
            requester_pid,
            upgrade,
            registration,
            None,
            CONTROL_QUEUE_DRAIN_REGISTRATION_TIMEOUT,
        )
        .await
    }

    pub(crate) async fn register_control_with_server_access(
        &self,
        requester_pid: u32,
        upgrade: ControlModeUpgrade,
        registration: ControlRegistration,
        admission: ServerAccessAdmission,
    ) -> Result<u64, ControlRegistrationError> {
        self.register_control_with_access_timeout(
            requester_pid,
            upgrade,
            registration,
            Some(admission),
            CONTROL_QUEUE_DRAIN_REGISTRATION_TIMEOUT,
        )
        .await
    }

    #[cfg(test)]
    pub(crate) async fn register_control_with_access_timeout_for_test(
        &self,
        requester_pid: u32,
        upgrade: ControlModeUpgrade,
        registration: ControlRegistration,
        drain_timeout: Duration,
    ) -> Result<u64, ControlRegistrationError> {
        self.register_control_with_access_timeout(
            requester_pid,
            upgrade,
            registration,
            None,
            drain_timeout,
        )
        .await
    }

    async fn register_control_with_access_timeout(
        &self,
        requester_pid: u32,
        upgrade: ControlModeUpgrade,
        registration: ControlRegistration,
        admission: Option<ServerAccessAdmission>,
        drain_timeout: Duration,
    ) -> Result<u64, ControlRegistrationError> {
        #[cfg(test)]
        if admission.is_some() {
            pause_before_access_registration(AccessRegistrationKind::Control, requester_pid).await;
        }
        let drain_deadline = tokio::time::Instant::now() + drain_timeout;
        let (control_id, replaced_session, detached_event) = loop {
            let activity_sequence = self.next_client_activity_sequence();
            // Session lifecycle events and the logical client replacement
            // share the established state -> active-control lock order. This
            // keeps the detach ticket at the exact replacement boundary.
            let mut state = self.state.lock().await;
            let mut active_control = self.active_control.lock().await;
            let drain_finished = active_control
                .by_pid
                .get(&requester_pid)
                .filter(|active| active.queue_draining)
                .map(|active| active.queue_drain_finished.subscribe());
            if let Some(mut drain_finished) = drain_finished {
                drop(active_control);
                drop(state);
                if tokio::time::timeout_at(
                    drain_deadline,
                    drain_finished.wait_for(|finished| *finished),
                )
                .await
                .is_err()
                {
                    return Err(ControlRegistrationError::QueueDrainTimedOut { requester_pid });
                }
                continue;
            }

            let (control_id, replaced_session, detached_event) = {
                // Keep the client-state lock across policy revalidation and
                // publication. ACL mutations commit to the store first, then take
                // this lock to update or disconnect every published control client.
                let server_access = admission.as_ref().map(|_| {
                    self.server_access
                        .lock()
                        .expect("server access mutex must not be poisoned")
                });
                let can_write = match (server_access.as_ref(), admission.as_ref()) {
                    (Some(server_access), Some(admission)) => server_access
                        .revalidate_admission(admission, &registration.user)
                        .ok_or(ControlRegistrationError::AccessRevoked)?
                        .can_write(),
                    (None, None) => registration.can_write,
                    _ => unreachable!("an admission and its policy guard are created together"),
                };

                let control_id = active_control.next_id;
                active_control.next_id += 1;
                let (queue_drain_finished, _queue_drain_pending) = watch::channel(false);
                let previous = active_control.by_pid.insert(
                    requester_pid,
                    ActiveControl {
                        id: control_id,
                        last_activity_sequence: activity_sequence,
                        activity_at: current_client_activity_timestamp(),
                        client_width: DEFAULT_CONTROL_CLIENT_WIDTH,
                        client_height: DEFAULT_CONTROL_CLIENT_HEIGHT,
                        size_declared: false,
                        size_sequence: 0,
                        session_name: None,
                        session_id: None,
                        last_session: None,
                        last_session_id: None,
                        flags: ControlClientFlags::default(),
                        uid: registration.uid,
                        user: registration.user,
                        can_write,
                        terminal_context: upgrade.terminal_context,
                        event_tx: registration.event_tx,
                        closing: registration.closing,
                        client_detached_prepared: false,
                        queue_draining: false,
                        queue_drain_finished,
                    },
                );
                let (replaced_session, detached_event) = if let Some(previous) = previous {
                    let replaced_session = previous.session_name.clone().zip(previous.session_id);
                    let detached_event =
                        replaced_session
                            .as_ref()
                            .and_then(|(session_name, session_id)| {
                                (!previous.client_detached_prepared).then(|| {
                                    let mut event = super::prepare_lifecycle_event(
                                        &mut state,
                                        &LifecycleEvent::ClientDetached {
                                            session_name: session_name.clone(),
                                            client_name: Some(control_client_name(requester_pid)),
                                        },
                                    );
                                    event.control_session_identity = Some(*session_id);
                                    event
                                })
                            });
                    close_control_with_exit(&previous, None);
                    (replaced_session, detached_event)
                } else {
                    (None, None)
                };
                drop(server_access);
                (control_id, replaced_session, detached_event)
            };
            drop(active_control);
            drop(state);
            break (control_id, replaced_session, detached_event);
        };

        if let Some(event) = detached_event {
            self.emit_prepared(event).await;
        }
        if let Some(session_identity) = replaced_session {
            self.destroy_unattached_sessions(vec![session_identity])
                .await;
        }

        for line in self.take_startup_config_error_notifications().await {
            self.send_control_notification_to(requester_pid, line).await;
        }

        Ok(control_id)
    }

    pub(crate) async fn begin_control_queue_drain(
        &self,
        identity: ControlClientIdentity,
    ) -> ControlQueueDrainLease {
        let mut active_control = self.active_control.lock().await;
        let Some(active) = active_control.by_pid.get_mut(&identity.requester_pid()) else {
            return ControlQueueDrainLease::Unavailable;
        };
        if active.id != identity.control_id() {
            return ControlQueueDrainLease::Unavailable;
        }
        // The active frame may itself have marked the control client closing
        // immediately before EOF (for example by emitting Exit). Its exact
        // registration still owns all already-accepted frames until the
        // drain observes that Exit and stops.
        active.queue_draining = true;
        ControlQueueDrainLease::Acquired
    }

    pub(crate) async fn control_queue_identity_is_open(
        &self,
        identity: ControlClientIdentity,
    ) -> bool {
        let active_control = self.active_control.lock().await;
        active_control
            .by_pid
            .get(&identity.requester_pid())
            .is_some_and(|active| {
                active.id == identity.control_id() && !active.closing.load(Ordering::SeqCst)
            })
    }

    pub(crate) async fn finish_control(&self, requester_pid: u32, control_id: u64) {
        let (removed, removed_session, detached_event) = {
            let mut state = self.state.lock().await;
            let mut active_control = self.active_control.lock().await;
            let removed_control = if active_control
                .by_pid
                .get(&requester_pid)
                .is_some_and(|active| active.id == control_id)
            {
                active_control.by_pid.remove(&requester_pid)
            } else {
                None
            };
            match removed_control {
                Some(removed) => {
                    removed.queue_drain_finished.send_replace(true);
                    let prepare_detached = !removed.client_detached_prepared;
                    let removed_session = removed.session_name.zip(removed.session_id);
                    let detached_event =
                        removed_session
                            .as_ref()
                            .and_then(|(session_name, session_id)| {
                                prepare_detached.then(|| {
                                    let mut event = super::prepare_lifecycle_event(
                                        &mut state,
                                        &LifecycleEvent::ClientDetached {
                                            session_name: session_name.clone(),
                                            client_name: Some(control_client_name(requester_pid)),
                                        },
                                    );
                                    event.control_session_identity = Some(*session_id);
                                    event
                                })
                            });
                    (true, removed_session, detached_event)
                }
                None => (false, None, None),
            }
        };
        if let Some(event) = detached_event {
            self.emit_prepared(event).await;
        }
        if let Some(session_identity) = removed_session {
            let _ = self
                .reconcile_attached_session_identity_size_and_emit(session_identity.1)
                .await;
            self.destroy_unattached_sessions(vec![session_identity])
                .await;
        }
        if removed {
            let _ = self.request_shutdown_if_pending();
        }
    }

    pub(super) async fn attached_count(&self, session_name: &rmux_proto::SessionName) -> usize {
        let attach_count = {
            let active_attach = self.active_attach.lock().await;
            active_attach.attached_count(session_name)
        };
        let control_count = {
            let active_control = self.active_control.lock().await;
            active_control.attached_count(session_name)
        };

        attach_count.saturating_add(control_count)
    }

    pub(super) async fn attached_count_for_session_identity(
        &self,
        session_name: &rmux_proto::SessionName,
        session_id: SessionId,
    ) -> usize {
        let attach_count = {
            let active_attach = self.active_attach.lock().await;
            active_attach
                .by_pid
                .values()
                .filter(|active| {
                    &active.session_name == session_name
                        && active.session_id == session_id
                        && !active.suspended
                })
                .count()
        };
        let control_count = {
            let active_control = self.active_control.lock().await;
            active_control
                .by_pid
                .values()
                .filter(|active| {
                    active.session_name.as_ref() == Some(session_name)
                        && active.session_id == Some(session_id)
                })
                .count()
        };

        attach_count.saturating_add(control_count)
    }

    pub(super) async fn attached_count_after_switch(
        &self,
        session_name: &rmux_proto::SessionName,
        client: ManagedClient,
    ) -> usize {
        let attached_count = self.attached_count(session_name).await;

        match client {
            ManagedClient::Attach {
                pid: attach_pid,
                attach_id,
            } => {
                let active_attach = self.active_attach.lock().await;
                if active_attach.by_pid.get(&attach_pid).is_some_and(|active| {
                    active.id == attach_id && &active.session_name == session_name
                }) {
                    attached_count
                } else {
                    attached_count.saturating_add(1)
                }
            }
            ManagedClient::Control(identity) => {
                let active_control = self.active_control.lock().await;
                if active_control
                    .by_pid
                    .get(&identity.requester_pid())
                    .filter(|active| active.id == identity.control_id())
                    .and_then(|active| active.session_name.as_ref())
                    .is_some_and(|active| active == session_name)
                {
                    attached_count
                } else {
                    attached_count.saturating_add(1)
                }
            }
        }
    }

    pub(super) async fn rename_control_session(
        &self,
        session_name: &rmux_proto::SessionName,
        session_id: SessionId,
        new_name: &rmux_proto::SessionName,
    ) {
        let mut active_control = self.active_control.lock().await;
        for active in active_control.by_pid.values_mut() {
            if active.session_name.as_ref() == Some(session_name)
                && active.session_id == Some(session_id)
            {
                active.session_name = Some(new_name.clone());
                let _ = try_send_control_event(
                    active,
                    ControlServerEvent::SessionChanged(Some(new_name.clone())),
                );
            }
            if active.last_session.as_ref() == Some(session_name)
                && active.last_session_id == Some(session_id)
            {
                active.last_session = Some(new_name.clone());
            }
        }
    }

    pub(super) async fn current_session_candidate(
        &self,
        requester_pid: u32,
    ) -> Option<rmux_proto::SessionName> {
        if let Some(identity) = current_control_queue_identity(requester_pid) {
            let state = self.state.lock().await;
            let active_control = self.active_control.lock().await;
            Self::validate_control_queue_identity_locked(
                &state,
                &active_control,
                requester_pid,
                identity.control_id(),
            )
            .ok()?;
            return active_control
                .by_pid
                .get(&requester_pid)
                .and_then(|active| active.session_name.clone());
        }

        {
            let active_attach = self.active_attach.lock().await;
            if let Some(candidate) = active_attach.current_session_candidate(requester_pid) {
                return Some(candidate);
            }
        }

        let candidate = {
            let active_control = self.active_control.lock().await;
            active_control.current_session_candidate(requester_pid)
        }?;
        let state = self.state.lock().await;
        state
            .sessions
            .session(&candidate.0)
            .is_some_and(|session| session.id() == candidate.1)
            .then_some(candidate.0)
    }

    pub(super) async fn validate_control_queue_session_identity(
        &self,
        requester_pid: u32,
        expected_control_id: u64,
    ) -> Result<(), rmux_proto::RmuxError> {
        let state = self.state.lock().await;
        let active_control = self.active_control.lock().await;
        Self::validate_control_queue_identity_locked(
            &state,
            &active_control,
            requester_pid,
            expected_control_id,
        )
    }

    pub(super) fn validate_control_queue_identity_locked(
        state: &HandlerState,
        active_control: &ActiveControlState,
        requester_pid: u32,
        expected_control_id: u64,
    ) -> Result<(), rmux_proto::RmuxError> {
        let Some(active) = active_control.by_pid.get(&requester_pid) else {
            return Err(attached_client_required("control command"));
        };
        if active.id != expected_control_id {
            return Err(attached_client_required("control command"));
        }
        if active.closing.load(Ordering::SeqCst) {
            return Err(rmux_proto::RmuxError::Server(
                "control client is closing".to_owned(),
            ));
        }
        match (active.session_name.as_ref(), active.session_id) {
            (None, None) => Ok(()),
            (Some(session_name), Some(session_id))
                if state
                    .sessions
                    .session(session_name)
                    .is_some_and(|session| session.id() == session_id) =>
            {
                Ok(())
            }
            (Some(session_name), _) => Err(rmux_proto::RmuxError::SessionNotFound(
                session_name.to_string(),
            )),
            (None, Some(_)) => Err(rmux_proto::RmuxError::Server(
                "control client has an invalid session identity".to_owned(),
            )),
        }
    }

    #[cfg(test)]
    pub(super) async fn control_queue_client_id(
        &self,
        requester_pid: u32,
    ) -> Result<u64, rmux_proto::RmuxError> {
        let active_control = self.active_control.lock().await;
        let active = active_control
            .by_pid
            .get(&requester_pid)
            .ok_or_else(|| attached_client_required("control command"))?;
        if active.closing.load(Ordering::SeqCst) {
            return Err(rmux_proto::RmuxError::Server(
                "control client is closing".to_owned(),
            ));
        }
        Ok(active.id)
    }

    pub(super) async fn resolve_managed_client(
        &self,
        requester_pid: u32,
        command_name: &str,
    ) -> Result<ManagedClient, rmux_proto::RmuxError> {
        if let Some(identity) = current_control_queue_identity(requester_pid) {
            let state = self.state.lock().await;
            let active_control = self.active_control.lock().await;
            Self::validate_control_queue_identity_locked(
                &state,
                &active_control,
                requester_pid,
                identity.control_id(),
            )?;
            return Ok(ManagedClient::Control(identity));
        }

        {
            let active_attach = self.active_attach.lock().await;
            if let Some(active) = active_attach
                .by_pid
                .get(&requester_pid)
                .filter(|active| !active.closing.load(Ordering::SeqCst))
            {
                return Ok(ManagedClient::Attach {
                    pid: requester_pid,
                    attach_id: active.id,
                });
            }
        }
        {
            let active_control = self.active_control.lock().await;
            if let Some(active) = active_control
                .by_pid
                .get(&requester_pid)
                .filter(|active| !active.closing.load(Ordering::SeqCst))
            {
                return Ok(ManagedClient::Control(ControlClientIdentity::new(
                    requester_pid,
                    active.id,
                )));
            }
        }

        let attach_candidates = {
            let active_attach = self.active_attach.lock().await;
            active_attach
                .by_pid
                .iter()
                .filter(|(_, active)| !active.closing.load(Ordering::SeqCst))
                .map(|(&pid, active)| (pid, active.id))
                .collect::<Vec<_>>()
        };
        let control_candidates = {
            let active_control = self.active_control.lock().await;
            active_control
                .by_pid
                .iter()
                .filter(|(_, active)| !active.closing.load(Ordering::SeqCst))
                .map(|(&pid, active)| ControlClientIdentity::new(pid, active.id))
                .collect::<Vec<_>>()
        };

        match attach_candidates.len() + control_candidates.len() {
            0 if command_name == "show-messages" => Err(rmux_proto::RmuxError::Message(
                "no current client".to_owned(),
            )),
            0 => Err(attached_client_required(command_name)),
            1 => {
                if let Some((pid, attach_id)) = attach_candidates.first().copied() {
                    Ok(ManagedClient::Attach { pid, attach_id })
                } else {
                    Ok(ManagedClient::Control(
                        control_candidates
                            .first()
                            .copied()
                            .expect("single control candidate"),
                    ))
                }
            }
            _ => Err(ambiguous_attached_client(command_name)),
        }
    }

    pub(crate) async fn control_session_name(
        &self,
        requester_pid: u32,
    ) -> Option<rmux_proto::SessionName> {
        let expected_control_id =
            current_control_queue_identity(requester_pid).map(ControlClientIdentity::control_id);
        let active_control = self.active_control.lock().await;
        active_control
            .by_pid
            .get(&requester_pid)
            .filter(|active| {
                expected_control_id.is_none_or(|expected| {
                    active.id == expected && !active.closing.load(Ordering::SeqCst)
                })
            })
            .and_then(|active| active.session_name.clone())
    }

    pub(in crate::handler) async fn control_queue_can_write(
        &self,
        identity: ControlClientIdentity,
    ) -> bool {
        let state = self.state.lock().await;
        let active_control = self.active_control.lock().await;
        if Self::validate_control_queue_identity_locked(
            &state,
            &active_control,
            identity.requester_pid(),
            identity.control_id(),
        )
        .is_err()
        {
            return false;
        }
        active_control
            .by_pid
            .get(&identity.requester_pid())
            .is_some_and(|active| active.can_write)
    }

    pub(crate) async fn is_control_client(&self, requester_pid: u32) -> bool {
        let expected_control_id =
            current_control_queue_identity(requester_pid).map(ControlClientIdentity::control_id);
        let active_control = self.active_control.lock().await;
        active_control
            .by_pid
            .get(&requester_pid)
            .is_some_and(|active| {
                expected_control_id.is_none_or(|expected| active.id == expected)
                    && !active.closing.load(Ordering::SeqCst)
            })
    }

    #[cfg(test)]
    pub(in crate::handler) fn install_control_switch_post_commit_pause(
        &self,
        requester_pid: u32,
    ) -> Arc<ControlSwitchPostCommitPause> {
        let pause = Arc::new(ControlSwitchPostCommitPause::default());
        let mut installed = CONTROL_SWITCH_POST_COMMIT_PAUSE
            .lock()
            .expect("control switch post-commit pause lock");
        installed.retain(|(paused_pid, _)| *paused_pid != requester_pid);
        installed.push((requester_pid, pause.clone()));
        pause
    }

    #[cfg(test)]
    async fn pause_after_control_switch_commit(&self, requester_pid: u32) {
        let pause = {
            let mut installed = CONTROL_SWITCH_POST_COMMIT_PAUSE
                .lock()
                .expect("control switch post-commit pause lock");
            installed
                .iter()
                .position(|(paused_pid, _)| *paused_pid == requester_pid)
                .map(|position| installed.swap_remove(position).1)
        };
        let Some(pause) = pause else {
            return;
        };
        pause.reached.notify_one();
        pause.release.notified().await;
    }

    #[cfg(test)]
    pub(super) async fn set_control_session(
        &self,
        requester_pid: u32,
        next_session_name: Option<rmux_proto::SessionName>,
    ) -> Result<Option<rmux_proto::SessionName>, rmux_proto::RmuxError> {
        self.set_control_session_with_expected_identity(
            requester_pid,
            next_session_name,
            None,
            None,
            ControlSessionUpdate::existing(None, None),
        )
        .await
    }

    #[cfg(test)]
    pub(super) async fn set_control_session_identity(
        &self,
        requester_pid: u32,
        next_session_name: rmux_proto::SessionName,
        expected_session_id: SessionId,
    ) -> Result<Option<rmux_proto::SessionName>, rmux_proto::RmuxError> {
        self.set_control_session_with_expected_identity(
            requester_pid,
            Some(next_session_name),
            Some(expected_session_id),
            None,
            ControlSessionUpdate::existing(None, None),
        )
        .await
    }

    /// `switch-client` for a control client: same commit as
    /// the regular identity-checked control-session update, with the
    /// `client-session-changed` notification published at the commit point so
    /// it precedes the `%layout-change` lines the reconciles produce.
    pub(super) async fn switch_control_session_for_client_identity(
        &self,
        requester_pid: u32,
        expected_control_id: u64,
        next_session_name: rmux_proto::SessionName,
        expected_session_id: SessionId,
        target_selection: Option<SwitchTargetSelection>,
        client_environment: Option<&HashMap<String, String>>,
    ) -> Result<Option<rmux_proto::SessionName>, rmux_proto::RmuxError> {
        self.set_control_session_with_expected_identity(
            requester_pid,
            Some(next_session_name),
            Some(expected_session_id),
            Some(expected_control_id),
            ControlSessionUpdate::switched(target_selection, client_environment),
        )
        .await
    }

    pub(in crate::handler) async fn attach_existing_control_session_for_client_identity(
        &self,
        requester_pid: u32,
        expected_control_id: u64,
        next_session_name: rmux_proto::SessionName,
        expected_session_id: SessionId,
    ) -> Result<Option<rmux_proto::SessionName>, rmux_proto::RmuxError> {
        self.set_control_session_with_expected_identity(
            requester_pid,
            Some(next_session_name),
            Some(expected_session_id),
            Some(expected_control_id),
            ControlSessionUpdate::attached_existing(),
        )
        .await
    }

    pub(in crate::handler) async fn set_created_control_session_for_client_identity(
        &self,
        requester_pid: u32,
        expected_control_id: u64,
        next_session_name: rmux_proto::SessionName,
        expected_session_id: SessionId,
    ) -> Result<Option<rmux_proto::SessionName>, rmux_proto::RmuxError> {
        self.set_control_session_with_expected_identity(
            requester_pid,
            Some(next_session_name),
            Some(expected_session_id),
            Some(expected_control_id),
            ControlSessionUpdate::created(),
        )
        .await
    }

    #[cfg_attr(not(test), allow(dead_code))]
    pub(in crate::handler) async fn switch_control_session_after_destroy(
        &self,
        requester_pid: u32,
        expected_control_id: u64,
        expected_current_session_id: SessionId,
        target_session_id: SessionId,
    ) -> Option<rmux_proto::SessionName> {
        let (target_session_name, client_session_changed) = self
            .prepare_control_session_switch_after_destroy(
                requester_pid,
                expected_control_id,
                expected_current_session_id,
                target_session_id,
            )
            .await?;
        self.emit_prepared(client_session_changed).await;
        Some(target_session_name)
    }

    pub(in crate::handler) async fn prepare_control_session_switch_after_destroy(
        &self,
        requester_pid: u32,
        expected_control_id: u64,
        expected_current_session_id: SessionId,
        target_session_id: SessionId,
    ) -> Option<(rmux_proto::SessionName, QueuedLifecycleEvent)> {
        let outcome = self
            .switch_control_session_after_destroy_inner(
                requester_pid,
                expected_control_id,
                expected_current_session_id,
                target_session_id,
            )
            .await?;
        #[cfg(test)]
        self.pause_after_control_switch_commit(requester_pid).await;
        Some((outcome.target_session_name, outcome.client_session_changed))
    }

    async fn switch_control_session_after_destroy_inner(
        &self,
        requester_pid: u32,
        expected_control_id: u64,
        expected_current_session_id: SessionId,
        target_session_id: SessionId,
    ) -> Option<DestroyControlSwitchCommitOutcome> {
        let mut state = self.state.lock().await;
        let target_session_name = state
            .sessions
            .session_by_id(target_session_id)?
            .name()
            .clone();
        let pane_sequences = current_pane_output_sequences(&state, &target_session_name).ok()?;
        let size_sequence = self.next_client_size_sequence();
        let mut active_control = self.active_control.lock().await;
        let active = active_control
            .by_pid
            .get_mut(&requester_pid)
            .filter(|active| {
                active.id == expected_control_id
                    && active.session_id == Some(expected_current_session_id)
                    && !active.closing.load(Ordering::SeqCst)
            })?;
        let (_, delivered) = update_control_session(
            active,
            Some(target_session_name.clone()),
            Some(target_session_id),
            Some(pane_sequences),
        );
        if !delivered {
            return None;
        }
        active.size_sequence = size_sequence;
        state
            .sessions
            .session_mut(&target_session_name)
            .expect("stable destroy-switch target remains locked")
            .touch_attached();
        let mut client_session_changed = super::prepare_lifecycle_event(
            &mut state,
            &LifecycleEvent::ClientSessionChanged {
                session_name: target_session_name.clone(),
                client_name: Some(control_client_name(requester_pid)),
            },
        );
        client_session_changed.control_session_identity = Some(target_session_id);
        Some(DestroyControlSwitchCommitOutcome {
            target_session_name,
            client_session_changed,
        })
    }

    async fn set_control_session_with_expected_identity(
        &self,
        requester_pid: u32,
        next_session_name: Option<rmux_proto::SessionName>,
        expected_session_id: Option<SessionId>,
        expected_control_id: Option<u64>,
        update: ControlSessionUpdate<'_>,
    ) -> Result<Option<rmux_proto::SessionName>, rmux_proto::RmuxError> {
        let ControlSessionUpdate {
            target_selection,
            client_environment,
            output_start,
            session_changed_notice,
        } = update;
        let exact_client_identity = expected_control_id.is_some();
        let command_name = if exact_client_identity {
            "switch-client"
        } else {
            "control session"
        };
        let touch_attached = exact_client_identity;
        let queued_control_id =
            current_control_queue_identity(requester_pid).map(ControlClientIdentity::control_id);
        if expected_control_id
            .zip(queued_control_id)
            .is_some_and(|(explicit, queued)| explicit != queued)
        {
            return Err(attached_client_required(command_name));
        }
        let expected_control_id = expected_control_id.or(queued_control_id);
        let mut state = self.state.lock().await;
        let next_session_id = match next_session_name.as_ref() {
            Some(session_name) => {
                let session_id = state
                    .sessions
                    .session(session_name)
                    .ok_or_else(|| {
                        rmux_proto::RmuxError::SessionNotFound(session_name.to_string())
                    })?
                    .id();
                if expected_session_id.is_some_and(|expected| expected != session_id) {
                    return Err(rmux_proto::RmuxError::SessionNotFound(
                        session_name.to_string(),
                    ));
                }
                Some(session_id)
            }
            None => None,
        };
        let pane_sequences = match output_start {
            ControlOutputStart::Current => next_session_name
                .as_ref()
                .map(|session_name| current_pane_output_sequences(&state, session_name))
                .transpose()?,
            ControlOutputStart::Oldest => None,
        };
        let active_attach = self.active_attach.lock().await;
        let mut active_control = self.active_control.lock().await;
        let Some(active) = active_control.by_pid.get(&requester_pid) else {
            return Err(attached_client_required(command_name));
        };
        if expected_control_id.is_some_and(|expected| active.id != expected)
            || active.closing.load(Ordering::SeqCst)
        {
            return Err(attached_client_required(command_name));
        }
        if let Some(selection) = target_selection.as_ref() {
            let session_name = next_session_name
                .as_ref()
                .expect("a switch target selection carries a session");
            let session_id = next_session_id
                .expect("a switch target selection carries a stable session identity");
            selection.validate_for_session_identity(&state, session_name, session_id)?;
        }
        let selection_transition = target_selection
            .as_ref()
            .map(|selection| {
                SelectionTargetTransitionSnapshot::capture(&state, selection.window_target())
            })
            .transpose()?;
        let control_size = active.size_declared.then_some(TerminalSize {
            cols: active.client_width,
            rows: active.client_height,
        });
        let session_changed =
            active.session_name != next_session_name || active.session_id != next_session_id;
        let previous_session_id = active.session_id;
        let size_sequence = next_session_name
            .as_ref()
            .map(|_| self.next_client_size_sequence());
        let prepared_update =
            prepare_control_session_update(active, next_session_name.clone(), pane_sequences)
                .ok_or_else(|| attached_client_required(command_name))?;
        let refresh_sessions = if exact_client_identity {
            let (session_name, session_id) = next_session_name
                .as_ref()
                .zip(next_session_id)
                .expect("switch-client always carries a target session identity");
            switch_control_session_for_client(
                &mut state,
                ControlResizeClient::new(
                    &active_attach,
                    &active_control,
                    requester_pid,
                    control_size,
                    size_sequence.expect("switch-client carries a target session"),
                ),
                session_name,
                session_id,
                target_selection.as_ref(),
            )?
        } else {
            Vec::new()
        };
        if let (Some(session_name), Some(client_environment)) =
            (next_session_name.as_ref(), client_environment)
        {
            update_environment_from_client(&mut state, session_name, client_environment);
        }
        let active = active_control
            .by_pid
            .get_mut(&requester_pid)
            .expect("validated control client remains locked");
        let previous = commit_control_session_update(
            active,
            next_session_name.clone(),
            next_session_id,
            prepared_update,
        );
        if let Some(size_sequence) = size_sequence {
            active.size_sequence = size_sequence;
        }
        if touch_attached {
            let session_name = next_session_name
                .as_ref()
                .expect("switch-client always carries a target session");
            state
                .sessions
                .session_mut(session_name)
                .expect("target session stayed locked across the control update")
                .touch_attached();
        }
        let selection_events = selection_transition
            .map(|transition| transition.prepare(&mut state))
            .unwrap_or_default();
        drop(active_control);
        drop(active_attach);
        drop(state);
        for event in selection_events {
            self.emit_prepared(event).await;
        }
        if session_changed_notice == ControlSessionChangedNotice::Publish
            || (session_changed_notice == ControlSessionChangedNotice::PublishIfChanged
                && session_changed)
        {
            if let (Some(session_name), Some(session_id)) =
                (next_session_name.as_ref(), next_session_id)
            {
                self.emit_client_session_changed(
                    control_client_name(requester_pid),
                    session_name.clone(),
                    session_id,
                )
                .await;
            }
        }
        // The destination resize was recorded by the geometry chokepoint and is
        // published by `emit_client_session_changed` above (or by the dispatch
        // backstop when this update carries no session-changed notice).
        self.publish_applied_window_resizes().await;
        self.refresh_linked_window_sessions(refresh_sessions).await;
        if let Some(previous_session_id) =
            previous_session_id.filter(|previous| Some(*previous) != next_session_id)
        {
            let _ = self
                .reconcile_attached_session_identity_size_and_emit(previous_session_id)
                .await;
        }
        Ok(previous)
    }

    pub(in crate::handler) async fn attach_control_session_for_queue(
        &self,
        identity: ControlClientIdentity,
        session_name: &rmux_proto::SessionName,
        expected_session_id: Option<SessionId>,
    ) -> Result<bool, rmux_proto::RmuxError> {
        self.attach_control_session_for_queue_with_output_start(
            identity,
            session_name,
            expected_session_id,
            false,
        )
        .await
    }

    pub(in crate::handler) async fn attach_created_control_session_for_queue(
        &self,
        identity: ControlClientIdentity,
        session_name: &rmux_proto::SessionName,
        expected_session_id: Option<SessionId>,
    ) -> Result<bool, rmux_proto::RmuxError> {
        self.attach_control_session_for_queue_with_output_start(
            identity,
            session_name,
            expected_session_id,
            true,
        )
        .await
    }

    async fn attach_control_session_for_queue_with_output_start(
        &self,
        identity: ControlClientIdentity,
        session_name: &rmux_proto::SessionName,
        expected_session_id: Option<SessionId>,
        replay_from_oldest: bool,
    ) -> Result<bool, rmux_proto::RmuxError> {
        let mut state = self.state.lock().await;
        let mut active_control = self.active_control.lock().await;
        Self::validate_control_queue_identity_locked(
            &state,
            &active_control,
            identity.requester_pid(),
            identity.control_id(),
        )?;

        let session_id = state
            .sessions
            .session(session_name)
            .ok_or_else(|| rmux_proto::RmuxError::SessionNotFound(session_name.to_string()))?
            .id();
        if expected_session_id.is_some_and(|expected| expected != session_id) {
            return Err(rmux_proto::RmuxError::SessionNotFound(
                session_name.to_string(),
            ));
        }
        if active_control
            .by_pid
            .get(&identity.requester_pid())
            .is_some_and(|active| {
                active.session_name.as_ref() == Some(session_name)
                    && active.session_id == Some(session_id)
            })
        {
            return Ok(false);
        }

        let pane_sequences = if replay_from_oldest {
            None
        } else {
            Some(current_pane_output_sequences(&state, session_name)?)
        };
        let size_sequence = self.next_client_size_sequence();

        let delivered = {
            let active = active_control
                .by_pid
                .get_mut(&identity.requester_pid())
                .expect("validated control client remains registered while locked");
            update_control_session(
                active,
                Some(session_name.clone()),
                Some(session_id),
                pane_sequences,
            )
            .1
        };
        if !delivered {
            return Err(attached_client_required("control session"));
        }
        active_control
            .by_pid
            .get_mut(&identity.requester_pid())
            .expect("delivered control client remains registered while locked")
            .size_sequence = size_sequence;
        state
            .sessions
            .session_mut(session_name)
            .expect("validated session remains present while locked")
            .touch_attached();
        Ok(true)
    }

    pub(super) async fn refresh_control_session(&self, session_name: &rmux_proto::SessionName) {
        let mut active_control = self.active_control.lock().await;
        for active in active_control.by_pid.values_mut() {
            if active.session_name.as_ref() != Some(session_name) {
                continue;
            }
            let _ = try_send_control_event(active, ControlServerEvent::Refresh);
        }
    }

    pub(in crate::handler) async fn refresh_control_session_for_session_identity(
        &self,
        session_name: &rmux_proto::SessionName,
        session_id: SessionId,
    ) {
        let current = {
            let state = self.state.lock().await;
            state
                .sessions
                .session(session_name)
                .is_some_and(|session| session.id() == session_id)
        };
        if !current {
            return;
        }
        let mut active_control = self.active_control.lock().await;
        for active in active_control.by_pid.values_mut() {
            if active.session_name.as_ref() != Some(session_name)
                || active.session_id != Some(session_id)
            {
                continue;
            }
            let _ = try_send_control_event(active, ControlServerEvent::Refresh);
        }
    }

    pub(super) async fn refresh_control_client_for_identity(
        &self,
        identity: ControlClientIdentity,
    ) -> Result<(), rmux_proto::RmuxError> {
        let mut active_control = self.active_control.lock().await;
        let Some(active) = active_control
            .by_pid
            .get_mut(&identity.requester_pid())
            .filter(|active| {
                active.id == identity.control_id() && !active.closing.load(Ordering::SeqCst)
            })
        else {
            return Err(attached_client_required("refresh-client"));
        };
        if !try_send_control_event(active, ControlServerEvent::Refresh) {
            return Err(attached_client_required("refresh-client"));
        }
        Ok(())
    }

    pub(super) async fn exit_control_session_identity(
        &self,
        session_name: &rmux_proto::SessionName,
        session_id: SessionId,
        reason: Option<String>,
    ) {
        let mut active_control = self.active_control.lock().await;
        for active in active_control.by_pid.values_mut() {
            if active.last_session.as_ref() == Some(session_name)
                && active.last_session_id == Some(session_id)
            {
                active.last_session = None;
                active.last_session_id = None;
            }
            if active.session_name.as_ref() != Some(session_name)
                || active.session_id != Some(session_id)
            {
                continue;
            }
            close_control_with_exit(active, reason.clone());
        }
    }

    pub(super) async fn detach_control_clients_for_session_identity(
        &self,
        session_id: SessionId,
        control_clients: Vec<ControlClientIdentity>,
        reason: Option<String>,
    ) -> ControlClientsDetachOutcome {
        let mut state = self.state.lock().await;
        let Some(session_name) = state
            .sessions
            .session_by_id(session_id)
            .map(|session| session.name().clone())
        else {
            return ControlClientsDetachOutcome {
                lifecycle_events: Vec::new(),
            };
        };
        let mut active_control = self.active_control.lock().await;
        let lifecycle_events = control_clients
            .into_iter()
            .filter_map(|identity| {
                let requester_pid = identity.requester_pid();
                let matches_snapshot =
                    active_control
                        .by_pid
                        .get(&requester_pid)
                        .is_some_and(|active| {
                            active.id == identity.control_id()
                                && active.session_id == Some(session_id)
                        });
                if !matches_snapshot {
                    return None;
                }
                let active = active_control.by_pid.remove(&requester_pid)?;
                let prepare_detached = !active.client_detached_prepared;
                close_control_with_exit(&active, reason.clone());
                prepare_detached.then(|| {
                    let mut event = super::prepare_lifecycle_event(
                        &mut state,
                        &LifecycleEvent::ClientDetached {
                            session_name: session_name.clone(),
                            client_name: Some(control_client_name(requester_pid)),
                        },
                    );
                    event.control_session_identity = Some(session_id);
                    event
                })
            })
            .collect();

        ControlClientsDetachOutcome { lifecycle_events }
    }

    pub(super) async fn refresh_all_control_sessions(&self) {
        let session_names = {
            let active_control = self.active_control.lock().await;
            active_control
                .by_pid
                .values()
                .filter(|active| !active.closing.load(Ordering::SeqCst))
                .filter_map(|active| active.session_name.clone())
                .collect::<Vec<_>>()
        };

        for session_name in session_names {
            self.refresh_control_session(&session_name).await;
        }
    }

    pub(super) async fn send_control_notification_to(&self, requester_pid: u32, line: String) {
        let mut active_control = self.active_control.lock().await;
        deliver_control_notification(&mut active_control, requester_pid, line);
    }

    pub(in crate::handler) async fn send_control_response_notification_to(
        &self,
        requester_pid: u32,
        line: String,
    ) {
        let mut active_control = self.active_control.lock().await;
        deliver_control_response_notification(&mut active_control, requester_pid, line);
    }

    pub(in crate::handler) async fn send_control_response_notification_to_queue(
        &self,
        identity: ControlClientIdentity,
        line: String,
    ) -> bool {
        let mut active_control = self.active_control.lock().await;
        if active_control
            .by_pid
            .get(&identity.requester_pid())
            .is_some_and(|active| {
                active.id == identity.control_id() && !active.closing.load(Ordering::SeqCst)
            })
        {
            return deliver_control_response_notification(
                &mut active_control,
                identity.requester_pid(),
                line,
            );
        }
        false
    }

    pub(super) async fn dispatch_control_notifications(&self, event: &QueuedLifecycleEvent) {
        let state = self.state.lock().await;
        let mut active_control = self.active_control.lock().await;
        let control_clients = control_clients_snapshot_locked(&state, &active_control);
        if control_clients.is_empty() {
            return;
        }

        let notifications = collect_control_notifications(
            &state,
            &event.event,
            event.control_session_identity,
            &control_clients,
        );
        #[cfg(test)]
        self.pause_before_control_notification_delivery().await;
        for notification in notifications {
            deliver_control_notification(&mut active_control, notification.pid, notification.line);
        }
    }

    #[cfg(test)]
    pub(in crate::handler) fn install_control_notification_delivery_pause(
        &self,
    ) -> Arc<super::ControlNotificationDeliveryPause> {
        let pause = Arc::new(super::ControlNotificationDeliveryPause::default());
        *self
            .control_notification_delivery_pause
            .lock()
            .expect("control notification delivery pause") = Some(pause.clone());
        pause
    }

    #[cfg(test)]
    async fn pause_before_control_notification_delivery(&self) {
        let pause = self
            .control_notification_delivery_pause
            .lock()
            .expect("control notification delivery pause")
            .take();
        if let Some(pause) = pause {
            pause.reached.notify_one();
            pause.release.notified().await;
        }
    }

    pub(super) async fn refresh_control_sessions_for_event(&self, event: &LifecycleEvent) {
        match event {
            LifecycleEvent::PaneModeChanged { .. }
            | LifecycleEvent::WindowLayoutChanged { .. }
            | LifecycleEvent::WindowPaneChanged { .. }
            | LifecycleEvent::WindowUnlinked { .. }
            | LifecycleEvent::WindowLinked { .. }
            | LifecycleEvent::WindowRenamed { .. }
            | LifecycleEvent::ClientSessionChanged { .. }
            | LifecycleEvent::ClientResized { .. }
            | LifecycleEvent::ClientDetached { .. }
            | LifecycleEvent::SessionRenamed { .. }
            | LifecycleEvent::SessionCreated { .. }
            | LifecycleEvent::SessionWindowChanged { .. }
            | LifecycleEvent::PasteBufferChanged { .. }
            | LifecycleEvent::PasteBufferDeleted { .. } => {
                self.refresh_all_control_sessions().await;
            }
            LifecycleEvent::SessionClosed {
                session_name,
                session_id,
            } => {
                if let Some(session_id) = session_id {
                    self.exit_control_session_identity(
                        session_name,
                        SessionId::new(*session_id),
                        None,
                    )
                    .await;
                }
                self.refresh_all_control_sessions().await;
            }
            LifecycleEvent::ClientActive { .. }
            | LifecycleEvent::ClientAttached { .. }
            | LifecycleEvent::ClientFocusIn { .. }
            | LifecycleEvent::ClientFocusOut { .. }
            | LifecycleEvent::ClientLightTheme { .. }
            | LifecycleEvent::ClientDarkTheme { .. }
            | LifecycleEvent::AlertBell { .. }
            | LifecycleEvent::AlertActivity { .. }
            | LifecycleEvent::AlertSilence { .. }
            | LifecycleEvent::PaneExited { .. }
            | LifecycleEvent::PaneDied { .. }
            | LifecycleEvent::PaneFocusIn { .. }
            | LifecycleEvent::PaneFocusOut { .. }
            | LifecycleEvent::PaneSetClipboard { .. }
            | LifecycleEvent::PaneTitleChanged { .. }
            | LifecycleEvent::WindowResized { .. }
            | LifecycleEvent::AfterSelectWindow { .. }
            | LifecycleEvent::AfterSelectPane { .. }
            | LifecycleEvent::AfterSendKeys { .. }
            | LifecycleEvent::AfterSetOption { .. } => {}
        }
    }

    pub(super) async fn exit_control_client_for_identity(
        &self,
        requester_pid: u32,
        expected_control_id: u64,
        reason: Option<String>,
    ) -> Result<ControlClientDetachOutcome, rmux_proto::RmuxError> {
        self.exit_control_client_with_expected_id(requester_pid, expected_control_id, reason, None)
            .await
    }

    pub(in crate::handler) async fn exit_control_client_for_identity_from_mode_tree(
        &self,
        requester: super::mode_tree_support::ModeTreeActionIdentity,
        requester_pid: u32,
        expected_control_id: u64,
        reason: Option<String>,
    ) -> Result<ControlClientDetachOutcome, rmux_proto::RmuxError> {
        self.exit_control_client_with_expected_id(
            requester_pid,
            expected_control_id,
            reason,
            Some(requester),
        )
        .await
    }

    async fn exit_control_client_with_expected_id(
        &self,
        requester_pid: u32,
        expected_control_id: u64,
        reason: Option<String>,
        mode_tree_requester: Option<super::mode_tree_support::ModeTreeActionIdentity>,
    ) -> Result<ControlClientDetachOutcome, rmux_proto::RmuxError> {
        let mut state = self.state.lock().await;
        let _mode_tree_guard = if let Some(identity) = mode_tree_requester {
            let active_attach = self.active_attach.lock().await;
            if !identity.matches_active(&state, &active_attach) {
                return Err(attached_client_required("detach-client"));
            }
            Some(active_attach)
        } else {
            None
        };
        let mut active_control = self.active_control.lock().await;
        let Some(active) = active_control.by_pid.get_mut(&requester_pid) else {
            return Err(attached_client_required("detach-client"));
        };
        if active.id != expected_control_id {
            return Err(attached_client_required("detach-client"));
        }
        let session_name = active.session_name.clone();
        let session_id = active.session_id;
        let prepare_detached = session_name.is_some() && !active.client_detached_prepared;
        active.client_detached_prepared |= prepare_detached;
        close_control_with_exit(active, reason);
        let lifecycle_event = session_name
            .filter(|_| prepare_detached)
            .map(|session_name| {
                let mut event = super::prepare_lifecycle_event(
                    &mut state,
                    &LifecycleEvent::ClientDetached {
                        session_name,
                        client_name: Some(control_client_name(requester_pid)),
                    },
                );
                event.control_session_identity = session_id;
                event
            });
        Ok(ControlClientDetachOutcome { lifecycle_event })
    }

    pub(crate) async fn control_client_flags(
        &self,
        requester_pid: u32,
    ) -> Option<ControlClientFlags> {
        let expected_control_id =
            current_control_queue_identity(requester_pid).map(ControlClientIdentity::control_id);
        let active_control = self.active_control.lock().await;
        active_control
            .by_pid
            .get(&requester_pid)
            .filter(|active| {
                expected_control_id.is_none_or(|expected| {
                    active.id == expected && !active.closing.load(Ordering::SeqCst)
                })
            })
            .map(|active| active.flags)
    }

    #[cfg(test)]
    pub(crate) async fn control_session_panes(
        &self,
        session_name: &rmux_proto::SessionName,
    ) -> Result<Vec<(u32, PaneOutputSender)>, rmux_proto::RmuxError> {
        let state = self.state.lock().await;
        state.session_pane_outputs(session_name)
    }

    pub(crate) async fn control_session_panes_for_identity(
        &self,
        control_identity: ControlClientIdentity,
        session_name: &rmux_proto::SessionName,
    ) -> Result<Vec<(u32, PaneOutputSender)>, rmux_proto::RmuxError> {
        let session_id = {
            let active_control = self.active_control.lock().await;
            active_control
                .by_pid
                .get(&control_identity.requester_pid())
                .filter(|active| {
                    active.id == control_identity.control_id()
                        && !active.closing.load(Ordering::SeqCst)
                        && active.session_name.as_ref() == Some(session_name)
                })
                .and_then(|active| active.session_id)
                .ok_or_else(|| rmux_proto::RmuxError::SessionNotFound(session_name.to_string()))?
        };
        let state = self.state.lock().await;
        let current_name = state
            .sessions
            .session_by_id(session_id)
            .map(|session| session.name())
            .filter(|current_name| *current_name == session_name)
            .ok_or_else(|| rmux_proto::RmuxError::SessionNotFound(session_name.to_string()))?;
        state.session_pane_outputs(current_name)
    }

    #[cfg(test)]
    pub(crate) async fn set_control_subscription_identity_for_test(
        &self,
        control_identity: ControlClientIdentity,
        session_name: rmux_proto::SessionName,
        session_id: SessionId,
    ) {
        let mut active_control = self.active_control.lock().await;
        let active = active_control
            .by_pid
            .get_mut(&control_identity.requester_pid())
            .filter(|active| active.id == control_identity.control_id())
            .expect("test control identity remains registered");
        active.session_name = Some(session_name);
        active.session_id = Some(session_id);
        active.size_sequence = self.next_client_size_sequence();
    }

    async fn take_startup_config_error_notifications(&self) -> Vec<String> {
        let mut errors = self.startup_config_errors.lock().await;
        errors
            .drain(..)
            .flat_map(|error| match error {
                rmux_proto::RmuxError::Server(message) => message
                    .lines()
                    .filter(|line| !line.is_empty())
                    .map(|line| format!("%config-error {line}"))
                    .collect::<Vec<_>>(),
                other => vec![format!("%config-error {other}")],
            })
            .collect()
    }
}

struct PreparedControlSessionUpdate {
    event: ControlServerEvent,
    permit: mpsc::OwnedPermit<ControlServerEvent>,
}

fn prepare_control_session_update(
    active: &ActiveControl,
    next_session_name: Option<rmux_proto::SessionName>,
    pane_sequences: Option<Vec<(u32, u64)>>,
) -> Option<PreparedControlSessionUpdate> {
    if active.closing.load(Ordering::SeqCst) {
        return None;
    }
    let event = match (next_session_name, pane_sequences) {
        (Some(session_name), Some(pane_sequences)) => ControlServerEvent::SessionChangedAt {
            session_name,
            pane_sequences,
        },
        (next_session_name, None) => ControlServerEvent::SessionChanged(next_session_name),
        (None, Some(_)) => unreachable!("pane cursors require a control session"),
    };
    let permit = match active.event_tx.clone().try_reserve_owned() {
        Ok(permit) => permit,
        Err(_) => {
            active.closing.store(true, Ordering::SeqCst);
            return None;
        }
    };
    Some(PreparedControlSessionUpdate { event, permit })
}

fn commit_control_session_update(
    active: &mut ActiveControl,
    next_session_name: Option<rmux_proto::SessionName>,
    next_session_id: Option<SessionId>,
    prepared: PreparedControlSessionUpdate,
) -> Option<rmux_proto::SessionName> {
    let previous = active.session_name.clone();
    if let (Some(previous_session), Some(previous_session_id), Some(next_session), Some(next_id)) = (
        previous.as_ref(),
        active.session_id,
        next_session_name.as_ref(),
        next_session_id,
    ) {
        if previous_session != next_session || previous_session_id != next_id {
            active.last_session = Some(previous_session.clone());
            active.last_session_id = Some(previous_session_id);
        }
    }
    active.session_name = next_session_name.clone();
    active.session_id = next_session_id;
    let _ = prepared.permit.send(prepared.event);
    previous
}

fn update_control_session(
    active: &mut ActiveControl,
    next_session_name: Option<rmux_proto::SessionName>,
    next_session_id: Option<SessionId>,
    pane_sequences: Option<Vec<(u32, u64)>>,
) -> (Option<rmux_proto::SessionName>, bool) {
    let previous = active.session_name.clone();
    let Some(prepared) =
        prepare_control_session_update(active, next_session_name.clone(), pane_sequences)
    else {
        return (previous, false);
    };
    (
        commit_control_session_update(active, next_session_name, next_session_id, prepared),
        true,
    )
}

fn current_pane_output_sequences(
    state: &HandlerState,
    session_name: &rmux_proto::SessionName,
) -> Result<Vec<(u32, u64)>, rmux_proto::RmuxError> {
    state.session_pane_outputs(session_name).map(|outputs| {
        outputs
            .into_iter()
            .map(|(pane_id, sender)| {
                let (sequence, ()) = sender.capture_with_next_sequence(|| ());
                (pane_id, sequence)
            })
            .collect()
    })
}

fn deliver_control_notification(
    active_control: &mut ActiveControlState,
    requester_pid: u32,
    line: String,
) -> bool {
    let Some(active) = active_control.by_pid.get_mut(&requester_pid) else {
        return false;
    };
    try_send_control_event(active, ControlServerEvent::Notification(line))
}

fn deliver_control_response_notification(
    active_control: &mut ActiveControlState,
    requester_pid: u32,
    line: String,
) -> bool {
    let Some(active) = active_control.by_pid.get_mut(&requester_pid) else {
        return false;
    };
    let target_identity = ControlClientIdentity::new(requester_pid, active.id);
    if let Some(sink) =
        current_control_command_response_sink().filter(|sink| sink.identity == target_identity)
    {
        if active.closing.load(Ordering::SeqCst) {
            return false;
        }
        if sink.try_send(ControlCommandResponseEvent::Notification(line)) {
            return true;
        }
        active.closing.store(true, Ordering::SeqCst);
        return false;
    }
    try_send_control_event(active, ControlServerEvent::Notification(line))
}

fn try_send_control_event(active: &ActiveControl, event: ControlServerEvent) -> bool {
    // Callers hold `active_control`: never await capacity for a client that is not draining.
    if active.closing.load(Ordering::SeqCst) {
        return false;
    }
    if active.event_tx.try_send(event).is_ok() {
        return true;
    }

    active.closing.store(true, Ordering::SeqCst);
    false
}

fn close_control_with_exit(active: &ActiveControl, reason: Option<String>) {
    if active.closing.swap(true, Ordering::SeqCst) {
        return;
    }
    let _ = active.event_tx.try_send(ControlServerEvent::Exit(reason));
}

fn control_clients_snapshot_locked(
    state: &HandlerState,
    active_control: &ActiveControlState,
) -> Vec<ControlClientSnapshot> {
    active_control
        .by_pid
        .iter()
        .filter_map(|(&pid, active)| {
            let session_name = match (&active.session_name, active.session_id) {
                (None, None) => None,
                (Some(_), Some(session_id)) => {
                    Some(state.sessions.session_by_id(session_id)?.name().clone())
                }
                (None, Some(_)) | (Some(_), None) => return None,
            };
            Some(ControlClientSnapshot { pid, session_name })
        })
        .collect()
}

impl ActiveControlState {
    pub(super) fn attached_count(&self, session_name: &rmux_proto::SessionName) -> usize {
        self.by_pid
            .values()
            .filter(|active| active.session_name.as_ref() == Some(session_name))
            .count()
    }

    fn current_session_candidate(
        &self,
        requester_pid: u32,
    ) -> Option<(rmux_proto::SessionName, SessionId)> {
        if let Some(active) = self.by_pid.get(&requester_pid) {
            if active.closing.load(Ordering::SeqCst) {
                return None;
            }
            return active.session_name.clone().zip(active.session_id);
        }

        if self.by_pid.len() == 1 {
            return self
                .by_pid
                .values()
                .next()
                .filter(|active| !active.closing.load(Ordering::SeqCst))
                .and_then(|active| active.session_name.clone().zip(active.session_id));
        }

        None
    }
}