//! Versioned, authenticated, transport-neutral operator control plane.

#![allow(clippy::missing_errors_doc)]

use crate::backend::BackendFuture;
use crate::config::SecretString;
use crate::observability::{MetricsSnapshot, StructuredEvent};
use libremetaverse_types::compat::{CancellationToken, CancellationTokenSource};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::error::Error;
use std::fmt;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, Weak};
use std::time::{Duration, Instant};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{mpsc, oneshot};
use tokio::task::{JoinHandle, JoinSet};

pub const CONTROL_PROTOCOL_VERSION: u16 = 1;
const MAX_REQUEST_ID_BYTES: usize = 96;
const MAX_TOKEN_BYTES: usize = 16 * 1024;
const MAX_OPERATOR_MESSAGE_BYTES: usize = 16 * 1024;
const MAX_PAGE_SIZE: u16 = 100;
const MAX_EVENT_BYTES: usize = 4 * 1024;

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ControlRole {
    Observer,
    Operator,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ControlErrorCode {
    AuthenticationFailed,
    VersionMismatch,
    PermissionDenied,
    InvalidRequest,
    Replay,
    NotFound,
    Conflict,
    Cancelled,
    TimedOut,
    Busy,
    Backpressure,
    FrameTooLarge,
    IdleTimeout,
    TransportClosed,
    Internal,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct ControlErrorBody {
    pub code: ControlErrorCode,
    pub message: String,
    pub retryable: bool,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ControlError {
    pub code: ControlErrorCode,
    pub message: &'static str,
    pub retryable: bool,
}

impl ControlError {
    const fn new(code: ControlErrorCode, message: &'static str, retryable: bool) -> Self {
        Self {
            code,
            message,
            retryable,
        }
    }

    fn body(&self) -> ControlErrorBody {
        ControlErrorBody {
            code: self.code,
            message: self.message.to_owned(),
            retryable: self.retryable,
        }
    }
}

impl fmt::Display for ControlError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.message)
    }
}

impl Error for ControlError {}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ControlLimits {
    pub max_frame_bytes: usize,
    pub max_connections: usize,
    pub max_in_flight_per_connection: usize,
    pub command_queue: usize,
    pub event_queue: usize,
    pub event_history: usize,
    pub max_events_per_second: usize,
    pub max_subscriptions: usize,
    pub replay_history: usize,
    pub request_timeout: Duration,
    pub idle_timeout: Duration,
    pub write_timeout: Duration,
}

impl Default for ControlLimits {
    fn default() -> Self {
        Self {
            max_frame_bytes: 64 * 1024,
            max_connections: 16,
            max_in_flight_per_connection: 16,
            command_queue: 64,
            event_queue: 128,
            event_history: 512,
            max_events_per_second: 1_024,
            max_subscriptions: 32,
            replay_history: 1_024,
            request_timeout: Duration::from_secs(15),
            idle_timeout: Duration::from_mins(2),
            write_timeout: Duration::from_secs(5),
        }
    }
}

impl ControlLimits {
    #[must_use]
    pub fn is_valid(self) -> bool {
        (1_024..=1024 * 1024).contains(&self.max_frame_bytes)
            && (1..=128).contains(&self.max_connections)
            && (1..=128).contains(&self.max_in_flight_per_connection)
            && (1..=8_192).contains(&self.command_queue)
            && (1..=8_192).contains(&self.event_queue)
            && (1..=8_192).contains(&self.event_history)
            && (1..=65_536).contains(&self.max_events_per_second)
            && (1..=128).contains(&self.max_subscriptions)
            && (1..=16_384).contains(&self.replay_history)
            && !self.request_timeout.is_zero()
            && self.request_timeout <= Duration::from_mins(2)
            && self.idle_timeout >= Duration::from_secs(5)
            && self.idle_timeout <= Duration::from_hours(1)
            && !self.write_timeout.is_zero()
            && self.write_timeout <= Duration::from_secs(30)
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PageRequest {
    pub cursor: Option<u64>,
    pub limit: u16,
}

impl Default for PageRequest {
    fn default() -> Self {
        Self {
            cursor: None,
            limit: 50,
        }
    }
}

impl PageRequest {
    fn valid(&self) -> bool {
        (1..=MAX_PAGE_SIZE).contains(&self.limit)
    }
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Page<T> {
    pub items: Vec<T>,
    pub next_cursor: Option<u64>,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct HealthView {
    pub service_state: String,
    pub ready: bool,
    pub uptime_seconds: u64,
    pub protocol_version: u16,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct RuntimeView {
    pub grid_state: String,
    pub generation: u64,
    pub transport_connected: bool,
    pub agent_ready: bool,
    pub region_id: Option<String>,
    pub region_name: Option<String>,
    pub position: Option<[f64; 3]>,
    pub behavior_mode: String,
    pub control_queue_used: usize,
    pub control_queue_capacity: usize,
    pub budget_tool_calls_used: u64,
    pub budget_movement_millimeters_used: u64,
    pub active_build_transaction: Option<String>,
    pub build_progress: Option<String>,
    pub build_orphan_ids: Vec<String>,
    pub active_visual_capture: Option<String>,
    pub visual_progress: Option<String>,
    pub visual_image_sha256: Option<String>,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct SessionMetadataView {
    pub session_id: String,
    pub avatar_id: String,
    pub channel: String,
    pub created_unix_millis: u64,
    pub last_active_unix_millis: u64,
    pub turns: usize,
    pub bytes: usize,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct ScheduledJobView {
    pub job_id: String,
    pub kind: String,
    pub enabled: bool,
    pub next_run_unix_millis: Option<u64>,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct PendingApprovalView {
    pub approval_id: u64,
    pub tool: String,
    pub principal: String,
    pub expires_unix_seconds: u64,
    pub movement_millimeters: u64,
    pub inventory_operations: u64,
    pub build_prims: u64,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct AuditEventView {
    pub sequence: u64,
    pub unix_millis: u64,
    pub principal: String,
    pub operation: String,
    pub outcome: String,
    pub authorization_id: Option<u64>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
pub enum ControlPayload {
    Health(HealthView),
    Metrics(MetricsSnapshot),
    Runtime(RuntimeView),
    Sessions(Page<SessionMetadataView>),
    ScheduledJobs(Page<ScheduledJobView>),
    PendingApprovals(Page<PendingApprovalView>),
    AuditEvents(Page<AuditEventView>),
    ObservabilityEvents(Page<StructuredEvent>),
    Accepted { operation_id: String },
    Completed,
    Subscribed { current_sequence: u64 },
    Cancelled { request_id: String },
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ConversationChannelView {
    PublicChat,
    DirectIm,
}

#[derive(Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "method", content = "parameters", rename_all = "snake_case")]
pub enum ControlRequest {
    Health,
    Metrics,
    Runtime,
    ListSessions {
        page: PageRequest,
    },
    ListScheduledJobs {
        page: PageRequest,
    },
    ListPendingApprovals {
        page: PageRequest,
    },
    ListAuditEvents {
        page: PageRequest,
    },
    ListObservabilityEvents {
        page: PageRequest,
    },
    SubscribeEvents {
        after_sequence: Option<u64>,
    },
    CancelRequest {
        target_request_id: String,
    },
    PauseAutonomy,
    ResumeAutonomy,
    CancelAction {
        action_id: String,
    },
    DecideApproval {
        approval_id: u64,
        approve: bool,
    },
    ForceReconnect,
    ExpireConversation {
        avatar_id: String,
        channel: ConversationChannelView,
    },
    SetRoamingJob {
        job_id: String,
        enabled: bool,
    },
    InjectOperatorMessage {
        message: String,
    },
    GracefulShutdown,
}

impl fmt::Debug for ControlRequest {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        if let Self::InjectOperatorMessage { message } = self {
            return formatter
                .debug_struct("InjectOperatorMessage")
                .field("message_bytes", &message.len())
                .finish();
        }
        formatter.write_str(request_name(self))
    }
}

impl ControlRequest {
    fn mutating(&self) -> bool {
        !matches!(
            self,
            Self::Health
                | Self::Metrics
                | Self::Runtime
                | Self::ListSessions { .. }
                | Self::ListScheduledJobs { .. }
                | Self::ListPendingApprovals { .. }
                | Self::ListAuditEvents { .. }
                | Self::ListObservabilityEvents { .. }
                | Self::SubscribeEvents { .. }
        )
    }

    fn valid(&self) -> bool {
        match self {
            Self::ListSessions { page }
            | Self::ListScheduledJobs { page }
            | Self::ListPendingApprovals { page }
            | Self::ListAuditEvents { page }
            | Self::ListObservabilityEvents { page } => page.valid(),
            Self::CancelRequest { target_request_id } => valid_identifier(target_request_id),
            Self::CancelAction { action_id } => valid_identifier(action_id),
            Self::DecideApproval { approval_id, .. } => *approval_id != 0,
            Self::ExpireConversation { avatar_id, .. } => valid_uuid_text(avatar_id),
            Self::SetRoamingJob { job_id, .. } => valid_identifier(job_id),
            Self::InjectOperatorMessage { message } => {
                !message.trim().is_empty()
                    && message.len() <= MAX_OPERATOR_MESSAGE_BYTES
                    && !message.contains('\0')
            }
            _ => true,
        }
    }
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ControlRequestEnvelope {
    pub version: u16,
    pub request_id: String,
    pub request: ControlRequest,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ControlResponseEnvelope {
    pub version: u16,
    pub request_id: String,
    pub result: Result<ControlPayload, ControlErrorBody>,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
pub enum ControlEventKind {
    Mutation {
        principal: String,
        operation: String,
        outcome: String,
    },
    StateChanged {
        component: String,
        state: String,
    },
    Gap {
        first_available: u64,
        last_missed: u64,
    },
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct ControlEvent {
    pub version: u16,
    pub sequence: u64,
    pub unix_millis: u64,
    pub event: ControlEventKind,
}

pub type ControlFuture<'a> = BackendFuture<'a, Result<ControlPayload, ControlError>>;

/// Authenticated request identity supplied to the management implementation.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ControlContext {
    pub role: ControlRole,
    pub principal: String,
}

/// Management implementation injected beneath both transports. Protocol types
/// deliberately do not expose internal runtime structs or secret-bearing data.
pub trait ControlTarget: Send + Sync + 'static {
    fn execute(
        &self,
        context: ControlContext,
        request: ControlRequest,
        cancellation: CancellationToken,
    ) -> ControlFuture<'_>;
}

struct TokenSet {
    observer: Option<SecretString>,
    operator: SecretString,
}

struct EventHubState {
    next_sequence: u64,
    history: VecDeque<ControlEvent>,
    subscribers: BTreeMap<u64, mpsc::Sender<ControlEvent>>,
    next_subscriber: u64,
    rate_window: Instant,
    events_in_window: usize,
}

struct EventHub {
    state: Mutex<EventHubState>,
    history_capacity: usize,
    subscriber_capacity: usize,
    max_events_per_second: usize,
    max_subscriptions: usize,
}

impl EventHub {
    fn new(
        history_capacity: usize,
        subscriber_capacity: usize,
        max_events_per_second: usize,
        max_subscriptions: usize,
    ) -> Self {
        Self {
            state: Mutex::new(EventHubState {
                next_sequence: 1,
                history: VecDeque::with_capacity(history_capacity),
                subscribers: BTreeMap::new(),
                next_subscriber: 1,
                rate_window: Instant::now(),
                events_in_window: 0,
            }),
            history_capacity,
            subscriber_capacity,
            max_events_per_second,
            max_subscriptions,
        }
    }

    fn publish(&self, event: ControlEventKind) -> bool {
        if serde_json::to_vec(&event).map_or(true, |body| body.len() > MAX_EVENT_BYTES) {
            return false;
        }
        let mut state = lock(&self.state);
        if state.rate_window.elapsed() >= Duration::from_secs(1) {
            state.rate_window = Instant::now();
            state.events_in_window = 0;
        }
        if state.events_in_window >= self.max_events_per_second {
            return false;
        }
        state.events_in_window = state.events_in_window.saturating_add(1);
        let sequence = state.next_sequence;
        state.next_sequence = state.next_sequence.saturating_add(1);
        let event = ControlEvent {
            version: CONTROL_PROTOCOL_VERSION,
            sequence,
            unix_millis: unix_millis(),
            event,
        };
        if state.history.len() == self.history_capacity {
            state.history.pop_front();
        }
        state.history.push_back(event.clone());
        state
            .subscribers
            .retain(|_, subscriber| subscriber.try_send(event.clone()).is_ok());
        true
    }

    fn subscribe(
        self: &Arc<Self>,
        after: Option<u64>,
    ) -> Result<(u64, ControlSubscription), ControlError> {
        let mut state = lock(&self.state);
        if state.subscribers.len() >= self.max_subscriptions {
            return Err(ControlError::new(
                ControlErrorCode::Busy,
                "event subscription limit reached",
                true,
            ));
        }
        let current = state.next_sequence.saturating_sub(1);
        let after = after.unwrap_or(current);
        let retained_first = state
            .history
            .front()
            .map_or(state.next_sequence, |item| item.sequence);
        let replay_first = state
            .next_sequence
            .saturating_sub(u64::try_from(self.subscriber_capacity).unwrap_or(u64::MAX));
        let first = retained_first.max(replay_first);
        let mut initial = VecDeque::new();
        if after.saturating_add(1) < first {
            initial.push_back(ControlEvent {
                version: CONTROL_PROTOCOL_VERSION,
                sequence: first.saturating_sub(1),
                unix_millis: unix_millis(),
                event: ControlEventKind::Gap {
                    first_available: first,
                    last_missed: first.saturating_sub(1),
                },
            });
        }
        initial.extend(
            state
                .history
                .iter()
                .filter(|item| item.sequence >= first && item.sequence > after)
                .cloned(),
        );
        let (sender, receiver) = mpsc::channel(self.subscriber_capacity);
        let id = state.next_subscriber;
        state.next_subscriber = state.next_subscriber.saturating_add(1);
        state.subscribers.insert(id, sender);
        Ok((
            current,
            ControlSubscription {
                id,
                hub: Arc::downgrade(self),
                initial,
                receiver,
            },
        ))
    }
}

pub struct ControlSubscription {
    id: u64,
    hub: Weak<EventHub>,
    initial: VecDeque<ControlEvent>,
    receiver: mpsc::Receiver<ControlEvent>,
}

impl Drop for ControlSubscription {
    fn drop(&mut self) {
        if let Some(hub) = self.hub.upgrade() {
            lock(&hub.state).subscribers.remove(&self.id);
        }
    }
}

impl ControlSubscription {
    pub async fn recv(&mut self) -> Option<ControlEvent> {
        if let Some(event) = self.initial.pop_front() {
            Some(event)
        } else {
            self.receiver.recv().await
        }
    }
}

pub struct ControlPlane {
    target: Arc<dyn ControlTarget>,
    tokens: TokenSet,
    limits: ControlLimits,
    events: Arc<EventHub>,
    next_connection: AtomicU64,
    active_connections: AtomicUsize,
}

impl fmt::Debug for ControlPlane {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ControlPlane")
            .field("limits", &self.limits)
            .field("tokens", &"[REDACTED]")
            .field(
                "active_connections",
                &self.active_connections.load(Ordering::Acquire),
            )
            .finish_non_exhaustive()
    }
}

impl ControlPlane {
    /// Creates an authenticated in-process operator connection with a fresh
    /// capability that is never exposed through configuration or diagnostics.
    pub fn integrated(
        target: Arc<dyn ControlTarget>,
        limits: ControlLimits,
    ) -> Result<(Arc<Self>, InProcessControlClient), ControlError> {
        let random = libremetaverse_types::UUID::secure_random().map_err(|_| {
            ControlError::new(
                ControlErrorCode::Internal,
                "cannot generate integrated control capability",
                false,
            )
        })?;
        let value = format!("integrated-{random}");
        let token = SecretString::new("control.integrated_token", value.clone()).map_err(|_| {
            ControlError::new(
                ControlErrorCode::Internal,
                "cannot construct integrated control capability",
                false,
            )
        })?;
        let plane = Self::new(target, token, None, limits)?;
        let client = plane.connect(&value)?;
        Ok((plane, client))
    }

    pub fn new(
        target: Arc<dyn ControlTarget>,
        operator_token: SecretString,
        observer_token: Option<SecretString>,
        limits: ControlLimits,
    ) -> Result<Arc<Self>, ControlError> {
        if !limits.is_valid()
            || observer_token.as_ref().is_some_and(|token| {
                constant_time_eq(token.expose_secret(), operator_token.expose_secret())
            })
        {
            return Err(ControlError::new(
                ControlErrorCode::InvalidRequest,
                "unsafe control-plane configuration",
                false,
            ));
        }
        Ok(Arc::new(Self {
            target,
            tokens: TokenSet {
                observer: observer_token,
                operator: operator_token,
            },
            limits,
            events: Arc::new(EventHub::new(
                limits.event_history,
                limits.event_queue,
                limits.max_events_per_second,
                limits.max_subscriptions,
            )),
            next_connection: AtomicU64::new(1),
            active_connections: AtomicUsize::new(0),
        }))
    }

    pub fn connect(self: &Arc<Self>, token: &str) -> Result<InProcessControlClient, ControlError> {
        let core = self.open(token)?;
        Ok(InProcessControlClient { core })
    }

    fn open(self: &Arc<Self>, token: &str) -> Result<Arc<ConnectionCore>, ControlError> {
        let role = self.authenticate(token)?;
        let prior = self.active_connections.fetch_add(1, Ordering::AcqRel);
        if prior >= self.limits.max_connections {
            self.active_connections.fetch_sub(1, Ordering::AcqRel);
            return Err(ControlError::new(
                ControlErrorCode::Busy,
                "connection limit reached",
                true,
            ));
        }
        let id = self.next_connection.fetch_add(1, Ordering::Relaxed);
        Ok(Arc::new(ConnectionCore {
            id,
            role,
            plane: Arc::clone(self),
            replay: Mutex::new((VecDeque::new(), BTreeSet::new())),
            active: Mutex::new(BTreeMap::new()),
            in_flight: AtomicUsize::new(0),
        }))
    }

    fn authenticate(&self, token: &str) -> Result<ControlRole, ControlError> {
        if token.is_empty() || token.len() > MAX_TOKEN_BYTES || token.contains(['\0', '\r', '\n']) {
            return Err(auth_error());
        }
        if constant_time_eq(token, self.tokens.operator.expose_secret()) {
            return Ok(ControlRole::Operator);
        }
        if self
            .tokens
            .observer
            .as_ref()
            .is_some_and(|expected| constant_time_eq(token, expected.expose_secret()))
        {
            return Ok(ControlRole::Observer);
        }
        Err(auth_error())
    }

    #[must_use]
    pub fn active_connections(&self) -> usize {
        self.active_connections.load(Ordering::Acquire)
    }

    #[must_use]
    pub fn retained_event_count(&self) -> usize {
        lock(&self.events.state).history.len()
    }

    pub fn publish(&self, event: ControlEventKind) -> bool {
        self.events.publish(event)
    }
}

struct ConnectionCore {
    id: u64,
    role: ControlRole,
    plane: Arc<ControlPlane>,
    replay: Mutex<(VecDeque<String>, BTreeSet<String>)>,
    active: Mutex<BTreeMap<String, CancellationTokenSource>>,
    in_flight: AtomicUsize,
}

impl Drop for ConnectionCore {
    fn drop(&mut self) {
        for source in lock(&self.active).values() {
            source.cancel();
        }
        self.plane.active_connections.fetch_sub(1, Ordering::AcqRel);
    }
}

impl ConnectionCore {
    async fn request(
        self: &Arc<Self>,
        envelope: ControlRequestEnvelope,
    ) -> (ControlResponseEnvelope, Option<ControlSubscription>) {
        let request_id = envelope.request_id.clone();
        let mutation = envelope
            .request
            .mutating()
            .then(|| request_name(&envelope.request).to_owned());
        let result = self.execute_envelope(envelope).await;
        if let Some(operation) = mutation {
            self.plane.events.publish(ControlEventKind::Mutation {
                principal: format!("control:{}:{}", role_name(self.role), self.id),
                operation,
                outcome: match &result {
                    Ok((ControlPayload::Accepted { .. }, _)) => "accepted",
                    Ok(_) => "completed",
                    Err(_) => "rejected",
                }
                .to_owned(),
            });
        }
        match result {
            Ok((payload, subscription)) => (response_ok(request_id, payload), subscription),
            Err(error) => (response_error(request_id, &error), None),
        }
    }

    async fn execute_envelope(
        self: &Arc<Self>,
        envelope: ControlRequestEnvelope,
    ) -> Result<(ControlPayload, Option<ControlSubscription>), ControlError> {
        if envelope.version != CONTROL_PROTOCOL_VERSION {
            return Err(ControlError::new(
                ControlErrorCode::VersionMismatch,
                "unsupported control protocol version",
                false,
            ));
        }
        if !valid_identifier(&envelope.request_id) || !envelope.request.valid() {
            return Err(ControlError::new(
                ControlErrorCode::InvalidRequest,
                "invalid bounded control request",
                false,
            ));
        }
        self.remember(&envelope.request_id)?;
        if envelope.request.mutating() && self.role != ControlRole::Operator {
            return Err(ControlError::new(
                ControlErrorCode::PermissionDenied,
                "operator permission required",
                false,
            ));
        }
        if let ControlRequest::CancelRequest { target_request_id } = &envelope.request {
            let cancelled = lock(&self.active)
                .get(target_request_id)
                .is_some_and(|source| {
                    source.cancel();
                    true
                });
            if !cancelled {
                return Err(ControlError::new(
                    ControlErrorCode::NotFound,
                    "active request not found",
                    false,
                ));
            }
            return Ok((
                ControlPayload::Cancelled {
                    request_id: target_request_id.clone(),
                },
                None,
            ));
        }
        if let ControlRequest::SubscribeEvents { after_sequence } = envelope.request {
            let (current, subscription) = self.plane.events.subscribe(after_sequence)?;
            return Ok((
                ControlPayload::Subscribed {
                    current_sequence: current,
                },
                Some(subscription),
            ));
        }
        let prior = self.in_flight.fetch_add(1, Ordering::AcqRel);
        if prior >= self.plane.limits.max_in_flight_per_connection {
            self.in_flight.fetch_sub(1, Ordering::AcqRel);
            return Err(ControlError::new(
                ControlErrorCode::Busy,
                "request concurrency limit reached",
                true,
            ));
        }
        let source = CancellationTokenSource::new();
        lock(&self.active).insert(envelope.request_id.clone(), source.clone());
        let context = ControlContext {
            role: self.role,
            principal: format!("control:{}:{}", role_name(self.role), self.id),
        };
        let timed = tokio::time::timeout(
            self.plane.limits.request_timeout,
            self.plane
                .target
                .execute(context, envelope.request, source.token()),
        )
        .await;
        if timed.is_err() {
            source.cancel();
        }
        lock(&self.active).remove(&envelope.request_id);
        self.in_flight.fetch_sub(1, Ordering::AcqRel);
        let result = timed.map_err(|_| {
            ControlError::new(
                ControlErrorCode::TimedOut,
                "control request timed out",
                true,
            )
        })?;
        let payload = result?;
        if !payload_valid(&payload, self.plane.limits.max_frame_bytes) {
            return Err(ControlError::new(
                ControlErrorCode::FrameTooLarge,
                "control response exceeds configured bound",
                false,
            ));
        }
        Ok((payload, None))
    }

    fn remember(&self, request_id: &str) -> Result<(), ControlError> {
        let mut replay = lock(&self.replay);
        if replay.1.contains(request_id) {
            return Err(ControlError::new(
                ControlErrorCode::Replay,
                "request ID replayed",
                false,
            ));
        }
        if replay.0.len() == self.plane.limits.replay_history
            && let Some(expired) = replay.0.pop_front()
        {
            replay.1.remove(&expired);
        }
        replay.0.push_back(request_id.to_owned());
        replay.1.insert(request_id.to_owned());
        Ok(())
    }
}

#[derive(Clone)]
pub struct InProcessControlClient {
    core: Arc<ConnectionCore>,
}

impl InProcessControlClient {
    pub async fn request(&self, envelope: ControlRequestEnvelope) -> ControlResponseEnvelope {
        self.core.request(envelope).await.0
    }

    pub async fn subscribe(
        &self,
        envelope: ControlRequestEnvelope,
    ) -> Result<(ControlResponseEnvelope, ControlSubscription), ControlError> {
        let (response, subscription) = self.core.request(envelope).await;
        if let Some(subscription) = subscription {
            return Ok((response, subscription));
        }
        if let Err(error) = &response.result {
            return Err(ControlError {
                code: error.code,
                message: "event subscription rejected",
                retryable: error.retryable,
            });
        }
        Err(ControlError::new(
            ControlErrorCode::InvalidRequest,
            "request did not create a subscription",
            false,
        ))
    }

    #[must_use]
    pub fn role(&self) -> ControlRole {
        self.core.role
    }
}

#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct ClientHello {
    version: u16,
    token: String,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
struct ServerHello {
    version: u16,
    role: Option<ControlRole>,
    error: Option<ControlErrorBody>,
}

#[derive(Serialize, Deserialize)]
#[serde(tag = "frame", content = "body", rename_all = "snake_case")]
enum ClientFrame {
    Hello(ClientHello),
    Request(ControlRequestEnvelope),
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "frame", content = "body", rename_all = "snake_case")]
enum ServerFrame {
    Hello(ServerHello),
    Response(Box<ControlResponseEnvelope>),
    Event(ControlEvent),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct TcpControlConfig {
    pub listen: SocketAddr,
    pub limits: ControlLimits,
}

impl Default for TcpControlConfig {
    fn default() -> Self {
        Self {
            listen: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 7943),
            limits: ControlLimits::default(),
        }
    }
}

pub struct TcpControlServer {
    address: SocketAddr,
    cancellation: CancellationTokenSource,
    task: Option<JoinHandle<()>>,
    shutdown_timeout: Duration,
}

impl Drop for TcpControlServer {
    fn drop(&mut self) {
        self.cancellation.cancel();
        if let Some(task) = &self.task {
            task.abort();
        }
    }
}

impl TcpControlServer {
    pub async fn bind(
        plane: Arc<ControlPlane>,
        config: TcpControlConfig,
    ) -> Result<Self, ControlError> {
        if !config.listen.ip().is_loopback()
            || !config.limits.is_valid()
            || config.limits != plane.limits
        {
            return Err(ControlError::new(
                ControlErrorCode::InvalidRequest,
                "plain TCP control transport is loopback-only",
                false,
            ));
        }
        Self::bind_listener(plane, config, None).await
    }

    /// Binds an explicitly configured TLS listener. Unlike [`Self::bind`],
    /// this may use a non-loopback address because plaintext never leaves the
    /// TLS session and token authentication still applies inside it.
    pub async fn bind_tls(
        plane: Arc<ControlPlane>,
        config: TcpControlConfig,
        server_config: Arc<rustls::ServerConfig>,
    ) -> Result<Self, ControlError> {
        if !config.limits.is_valid() || config.limits != plane.limits {
            return Err(ControlError::new(
                ControlErrorCode::InvalidRequest,
                "invalid TLS control transport configuration",
                false,
            ));
        }
        Self::bind_listener(
            plane,
            config,
            Some(tokio_rustls::TlsAcceptor::from(server_config)),
        )
        .await
    }

    async fn bind_listener(
        plane: Arc<ControlPlane>,
        config: TcpControlConfig,
        tls: Option<tokio_rustls::TlsAcceptor>,
    ) -> Result<Self, ControlError> {
        let listener = TcpListener::bind(config.listen).await.map_err(|_| {
            ControlError::new(
                ControlErrorCode::TransportClosed,
                "cannot bind control transport",
                true,
            )
        })?;
        let address = listener.local_addr().map_err(|_| {
            ControlError::new(
                ControlErrorCode::TransportClosed,
                "cannot read control transport address",
                true,
            )
        })?;
        let cancellation = CancellationTokenSource::new();
        let task_cancellation = cancellation.token();
        let shutdown_timeout = config.limits.write_timeout;
        let task = tokio::spawn(async move {
            let mut connections = JoinSet::new();
            loop {
                let accepted = tokio::select! {
                    () = task_cancellation.cancelled() => break,
                    Some(_) = connections.join_next(), if !connections.is_empty() => continue,
                    accepted = listener.accept() => accepted,
                };
                let Ok((stream, peer)) = accepted else {
                    continue;
                };
                if (tls.is_none() && !peer.ip().is_loopback())
                    || connections.len() >= plane.limits.max_connections
                {
                    continue;
                }
                let connection_plane = Arc::clone(&plane);
                let connection_cancel = task_cancellation.clone();
                let connection_tls = tls.clone();
                connections.spawn(async move {
                    if let Some(acceptor) = connection_tls {
                        if let Ok(Ok(stream)) = tokio::time::timeout(
                            connection_plane.limits.idle_timeout,
                            acceptor.accept(stream),
                        )
                        .await
                        {
                            let _ = serve_control_connection(
                                connection_plane,
                                stream,
                                connection_cancel,
                            )
                            .await;
                        }
                    } else {
                        let _ =
                            serve_control_connection(connection_plane, stream, connection_cancel)
                                .await;
                    }
                });
            }
            connections.abort_all();
            while connections.join_next().await.is_some() {}
        });
        Ok(Self {
            address,
            cancellation,
            task: Some(task),
            shutdown_timeout,
        })
    }

    #[must_use]
    pub const fn local_addr(&self) -> SocketAddr {
        self.address
    }

    pub async fn shutdown(mut self) -> Result<(), ControlError> {
        self.cancellation.cancel();
        if let Some(task) = self.task.take() {
            tokio::time::timeout(self.shutdown_timeout, task)
                .await
                .map_err(|_| {
                    ControlError::new(
                        ControlErrorCode::TimedOut,
                        "control server shutdown timed out",
                        true,
                    )
                })?
                .map_err(|_| {
                    ControlError::new(
                        ControlErrorCode::Internal,
                        "control server task failed",
                        false,
                    )
                })?;
        }
        Ok(())
    }
}

#[allow(clippy::too_many_lines)] // Ordered handshake, reader, writer, and task ownership.
async fn serve_control_connection<S>(
    plane: Arc<ControlPlane>,
    mut stream: S,
    cancellation: CancellationToken,
) -> Result<(), ControlError>
where
    S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
    let hello: ClientFrame = read_frame(
        &mut stream,
        plane.limits.max_frame_bytes,
        plane.limits.idle_timeout,
    )
    .await?;
    let ClientFrame::Hello(hello) = hello else {
        return Err(ControlError::new(
            ControlErrorCode::AuthenticationFailed,
            "authentication required",
            false,
        ));
    };
    if hello.version != CONTROL_PROTOCOL_VERSION {
        let error = ControlError::new(
            ControlErrorCode::VersionMismatch,
            "unsupported control protocol version",
            false,
        );
        let _ = write_frame(
            &mut stream,
            &ServerFrame::Hello(ServerHello {
                version: CONTROL_PROTOCOL_VERSION,
                role: None,
                error: Some(error.body()),
            }),
            plane.limits.max_frame_bytes,
            plane.limits.write_timeout,
        )
        .await;
        return Err(error);
    }
    let core = match plane.open(&hello.token) {
        Ok(core) => core,
        Err(error) => {
            let _ = write_frame(
                &mut stream,
                &ServerFrame::Hello(ServerHello {
                    version: CONTROL_PROTOCOL_VERSION,
                    role: None,
                    error: Some(error.body()),
                }),
                plane.limits.max_frame_bytes,
                plane.limits.write_timeout,
            )
            .await;
            return Err(error);
        }
    };
    write_frame(
        &mut stream,
        &ServerFrame::Hello(ServerHello {
            version: CONTROL_PROTOCOL_VERSION,
            role: Some(core.role),
            error: None,
        }),
        plane.limits.max_frame_bytes,
        plane.limits.write_timeout,
    )
    .await?;
    let (mut reader, mut writer) = tokio::io::split(stream);
    let (outbound, mut outbound_rx) = mpsc::channel::<ServerFrame>(plane.limits.command_queue);
    let connection_cancel = CancellationTokenSource::new();
    let writer_cancel = connection_cancel.token();
    let writer_source = connection_cancel.clone();
    let write_timeout = plane.limits.write_timeout;
    let max_frame = plane.limits.max_frame_bytes;
    let writer_task = tokio::spawn(async move {
        loop {
            let frame = tokio::select! {
                () = writer_cancel.cancelled() => break,
                frame = outbound_rx.recv() => frame,
            };
            let Some(frame) = frame else {
                break;
            };
            if write_frame(&mut writer, &frame, max_frame, write_timeout)
                .await
                .is_err()
            {
                writer_source.cancel();
                break;
            }
        }
    });
    let mut tasks = JoinSet::new();
    loop {
        while tasks.try_join_next().is_some() {}
        let frame = tokio::select! {
            () = cancellation.cancelled() => break,
            () = connection_cancel.token().cancelled() => break,
            frame = read_frame::<_, ClientFrame>(&mut reader, plane.limits.max_frame_bytes, plane.limits.idle_timeout) => frame,
        };
        let Ok(ClientFrame::Request(envelope)) = frame else {
            break;
        };
        if tasks.len() >= plane.limits.max_in_flight_per_connection {
            let response = response_error(
                envelope.request_id,
                &ControlError::new(
                    ControlErrorCode::Busy,
                    "connection task limit reached",
                    true,
                ),
            );
            if outbound
                .try_send(ServerFrame::Response(Box::new(response)))
                .is_err()
            {
                break;
            }
            continue;
        }
        let request_core = Arc::clone(&core);
        let request_outbound = outbound.clone();
        let request_cancel = connection_cancel.clone();
        tasks.spawn(async move {
            let (response, subscription) = request_core.request(envelope).await;
            if request_outbound
                .try_send(ServerFrame::Response(Box::new(response)))
                .is_err()
            {
                request_cancel.cancel();
                return;
            }
            if let Some(mut subscription) = subscription {
                while let Some(event) = subscription.recv().await {
                    if request_outbound
                        .try_send(ServerFrame::Event(event))
                        .is_err()
                    {
                        request_cancel.cancel();
                        break;
                    }
                }
            }
        });
    }
    connection_cancel.cancel();
    tasks.abort_all();
    while tasks.join_next().await.is_some() {}
    drop(outbound);
    let _ = tokio::time::timeout(plane.limits.write_timeout, writer_task).await;
    Ok(())
}

struct TcpClientShared {
    writer: tokio::sync::Mutex<Box<dyn ControlWrite>>,
    pending: Mutex<BTreeMap<String, oneshot::Sender<ControlResponseEnvelope>>>,
    events: tokio::sync::Mutex<mpsc::Receiver<ControlEvent>>,
    limits: ControlLimits,
    closed: CancellationTokenSource,
}

#[derive(Clone)]
pub struct TcpControlClient {
    shared: Arc<TcpClientShared>,
    role: ControlRole,
}

impl TcpControlClient {
    pub async fn connect(
        address: SocketAddr,
        token: &str,
        limits: ControlLimits,
    ) -> Result<Self, ControlError> {
        if !address.ip().is_loopback() || !limits.is_valid() {
            return Err(ControlError::new(
                ControlErrorCode::InvalidRequest,
                "TCP control client requires bounded loopback configuration",
                false,
            ));
        }
        let stream = tokio::time::timeout(limits.request_timeout, TcpStream::connect(address))
            .await
            .map_err(|_| {
                ControlError::new(
                    ControlErrorCode::TimedOut,
                    "control connection timed out",
                    true,
                )
            })?
            .map_err(|_| {
                ControlError::new(
                    ControlErrorCode::TransportClosed,
                    "cannot connect to control server",
                    true,
                )
            })?;
        Self::connect_stream(stream, token, limits).await
    }

    /// Connects to an explicitly configured TLS control endpoint. Certificate
    /// roots and server-name policy are supplied by the embedding client.
    pub async fn connect_tls(
        address: SocketAddr,
        server_name: &str,
        token: &str,
        limits: ControlLimits,
        client_config: Arc<rustls::ClientConfig>,
    ) -> Result<Self, ControlError> {
        if !limits.is_valid() {
            return Err(ControlError::new(
                ControlErrorCode::InvalidRequest,
                "TLS control client requires bounded configuration",
                false,
            ));
        }
        let server_name =
            rustls::pki_types::ServerName::try_from(server_name.to_owned()).map_err(|_| {
                ControlError::new(
                    ControlErrorCode::InvalidRequest,
                    "invalid TLS control server name",
                    false,
                )
            })?;
        let stream = tokio::time::timeout(limits.request_timeout, TcpStream::connect(address))
            .await
            .map_err(|_| {
                ControlError::new(
                    ControlErrorCode::TimedOut,
                    "TLS control connection timed out",
                    true,
                )
            })?
            .map_err(|_| {
                ControlError::new(
                    ControlErrorCode::TransportClosed,
                    "cannot connect to TLS control server",
                    true,
                )
            })?;
        let stream = tokio::time::timeout(
            limits.idle_timeout,
            tokio_rustls::TlsConnector::from(client_config).connect(server_name, stream),
        )
        .await
        .map_err(|_| {
            ControlError::new(
                ControlErrorCode::TimedOut,
                "TLS control handshake timed out",
                true,
            )
        })?
        .map_err(|_| {
            ControlError::new(
                ControlErrorCode::AuthenticationFailed,
                "TLS control server authentication failed",
                false,
            )
        })?;
        Self::connect_stream(stream, token, limits).await
    }

    async fn connect_stream<S>(
        mut stream: S,
        token: &str,
        limits: ControlLimits,
    ) -> Result<Self, ControlError>
    where
        S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
    {
        write_frame(
            &mut stream,
            &ClientFrame::Hello(ClientHello {
                version: CONTROL_PROTOCOL_VERSION,
                token: token.to_owned(),
            }),
            limits.max_frame_bytes,
            limits.write_timeout,
        )
        .await?;
        let hello: ServerFrame =
            read_frame(&mut stream, limits.max_frame_bytes, limits.idle_timeout).await?;
        let ServerFrame::Hello(hello) = hello else {
            return Err(ControlError::new(
                ControlErrorCode::AuthenticationFailed,
                "invalid server handshake",
                false,
            ));
        };
        if let Some(error) = hello.error {
            return Err(ControlError {
                code: error.code,
                message: "control handshake rejected",
                retryable: error.retryable,
            });
        }
        let role = hello.role.ok_or_else(auth_error)?;
        let (reader, writer) = tokio::io::split(stream);
        let (event_tx, event_rx) = mpsc::channel(limits.event_queue);
        let shared = Arc::new(TcpClientShared {
            writer: tokio::sync::Mutex::new(Box::new(writer)),
            pending: Mutex::new(BTreeMap::new()),
            events: tokio::sync::Mutex::new(event_rx),
            limits,
            closed: CancellationTokenSource::new(),
        });
        let reader_shared = Arc::clone(&shared);
        tokio::spawn(async move {
            tcp_client_reader(reader_shared, Box::new(reader), event_tx).await;
        });
        Ok(Self { shared, role })
    }

    pub async fn request(
        &self,
        envelope: ControlRequestEnvelope,
    ) -> Result<ControlResponseEnvelope, ControlError> {
        if !valid_identifier(&envelope.request_id) {
            return Err(ControlError::new(
                ControlErrorCode::InvalidRequest,
                "invalid request ID",
                false,
            ));
        }
        let (sender, receiver) = oneshot::channel();
        if lock(&self.shared.pending)
            .insert(envelope.request_id.clone(), sender)
            .is_some()
        {
            return Err(ControlError::new(
                ControlErrorCode::Replay,
                "request ID already pending",
                false,
            ));
        }
        let write = {
            let mut writer = self.shared.writer.lock().await;
            write_frame(
                &mut **writer,
                &ClientFrame::Request(envelope.clone()),
                self.shared.limits.max_frame_bytes,
                self.shared.limits.write_timeout,
            )
            .await
        };
        if let Err(error) = write {
            lock(&self.shared.pending).remove(&envelope.request_id);
            return Err(error);
        }
        let result = tokio::time::timeout(
            self.shared.limits.request_timeout + self.shared.limits.write_timeout,
            receiver,
        )
        .await
        .map_err(|_| {
            ControlError::new(
                ControlErrorCode::TimedOut,
                "control response timed out",
                true,
            )
        })?
        .map_err(|_| {
            ControlError::new(
                ControlErrorCode::TransportClosed,
                "control response channel closed",
                true,
            )
        });
        if result.is_err() {
            lock(&self.shared.pending).remove(&envelope.request_id);
        }
        result
    }

    pub async fn next_event(&self) -> Option<ControlEvent> {
        self.shared.events.lock().await.recv().await
    }

    #[must_use]
    pub fn role(&self) -> ControlRole {
        self.role
    }
}

impl Drop for TcpControlClient {
    fn drop(&mut self) {
        if Arc::strong_count(&self.shared) == 2 {
            self.shared.closed.cancel();
        }
    }
}

async fn tcp_client_reader(
    shared: Arc<TcpClientShared>,
    mut reader: Box<dyn ControlRead>,
    events: mpsc::Sender<ControlEvent>,
) {
    loop {
        let frame = tokio::select! {
            () = shared.closed.token().cancelled() => break,
            frame = read_frame::<_, ServerFrame>(&mut reader, shared.limits.max_frame_bytes, shared.limits.idle_timeout) => frame,
        };
        match frame {
            Ok(ServerFrame::Response(response)) => {
                if let Some(sender) = lock(&shared.pending).remove(&response.request_id) {
                    let _ = sender.send(*response);
                }
            }
            Ok(ServerFrame::Event(event)) => {
                if events.try_send(event).is_err() {
                    break;
                }
            }
            _ => break,
        }
    }
    shared.closed.cancel();
    lock(&shared.pending).clear();
}

trait ControlRead: AsyncRead + Unpin + Send {}
impl<T: AsyncRead + Unpin + Send> ControlRead for T {}

trait ControlWrite: AsyncWrite + Unpin + Send {}
impl<T: AsyncWrite + Unpin + Send> ControlWrite for T {}

async fn read_frame<R, T>(reader: &mut R, maximum: usize, idle: Duration) -> Result<T, ControlError>
where
    R: AsyncRead + Unpin,
    T: for<'de> Deserialize<'de>,
{
    let mut header = [0_u8; 4];
    tokio::time::timeout(idle, reader.read_exact(&mut header))
        .await
        .map_err(|_| {
            ControlError::new(
                ControlErrorCode::IdleTimeout,
                "control connection idle timeout",
                true,
            )
        })?
        .map_err(|_| {
            ControlError::new(
                ControlErrorCode::TransportClosed,
                "control frame header closed",
                true,
            )
        })?;
    let length = usize::try_from(u32::from_be_bytes(header)).map_err(|_| {
        ControlError::new(
            ControlErrorCode::FrameTooLarge,
            "control frame length is invalid",
            false,
        )
    })?;
    if length == 0 || length > maximum {
        return Err(ControlError::new(
            ControlErrorCode::FrameTooLarge,
            "control frame exceeds configured bound",
            false,
        ));
    }
    let mut body = vec![0_u8; length];
    tokio::time::timeout(idle, reader.read_exact(&mut body))
        .await
        .map_err(|_| {
            ControlError::new(
                ControlErrorCode::IdleTimeout,
                "partial control frame timed out",
                true,
            )
        })?
        .map_err(|_| {
            ControlError::new(
                ControlErrorCode::TransportClosed,
                "partial control frame closed",
                true,
            )
        })?;
    serde_json::from_slice(&body).map_err(|_| {
        ControlError::new(
            ControlErrorCode::InvalidRequest,
            "control frame JSON is invalid",
            false,
        )
    })
}

async fn write_frame<W, T>(
    writer: &mut W,
    value: &T,
    maximum: usize,
    timeout: Duration,
) -> Result<(), ControlError>
where
    W: AsyncWrite + Unpin + ?Sized,
    T: Serialize,
{
    let body = serde_json::to_vec(value).map_err(|_| {
        ControlError::new(
            ControlErrorCode::Internal,
            "control frame serialization failed",
            false,
        )
    })?;
    if body.is_empty() || body.len() > maximum || body.len() > u32::MAX as usize {
        return Err(ControlError::new(
            ControlErrorCode::FrameTooLarge,
            "control frame exceeds configured bound",
            false,
        ));
    }
    let length = u32::try_from(body.len()).map_err(|_| {
        ControlError::new(
            ControlErrorCode::FrameTooLarge,
            "control frame length overflow",
            false,
        )
    })?;
    tokio::time::timeout(timeout, async {
        writer.write_all(&length.to_be_bytes()).await?;
        writer.write_all(&body).await?;
        writer.flush().await
    })
    .await
    .map_err(|_| {
        ControlError::new(
            ControlErrorCode::TimedOut,
            "control frame write timed out",
            true,
        )
    })?
    .map_err(|_| {
        ControlError::new(
            ControlErrorCode::TransportClosed,
            "control frame write failed",
            true,
        )
    })
}

fn response_ok(request_id: String, payload: ControlPayload) -> ControlResponseEnvelope {
    ControlResponseEnvelope {
        version: CONTROL_PROTOCOL_VERSION,
        request_id,
        result: Ok(payload),
    }
}

fn response_error(request_id: String, error: &ControlError) -> ControlResponseEnvelope {
    ControlResponseEnvelope {
        version: CONTROL_PROTOCOL_VERSION,
        request_id,
        result: Err(error.body()),
    }
}

fn auth_error() -> ControlError {
    ControlError::new(
        ControlErrorCode::AuthenticationFailed,
        "control authentication failed",
        false,
    )
}

fn valid_identifier(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= MAX_REQUEST_ID_BYTES
        && value.chars().all(|character| {
            character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.' | ':')
        })
}

fn valid_uuid_text(value: &str) -> bool {
    libremetaverse_types::UUID::parse(value.to_owned()).is_ok()
}

fn payload_valid(payload: &ControlPayload, maximum_frame_bytes: usize) -> bool {
    let page_is_bounded = match payload {
        ControlPayload::Sessions(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE),
        ControlPayload::ScheduledJobs(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE),
        ControlPayload::PendingApprovals(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE),
        ControlPayload::AuditEvents(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE),
        _ => true,
    };
    page_is_bounded
        && serde_json::to_vec(payload).is_ok_and(|body| {
            // Reserve space for the versioned response envelope and frame tag.
            body.len().saturating_add(512) <= maximum_frame_bytes
        })
}

fn constant_time_eq(left: &str, right: &str) -> bool {
    let left = left.as_bytes();
    let right = right.as_bytes();
    let mut difference = left.len() ^ right.len();
    let maximum = left.len().max(right.len());
    for index in 0..maximum {
        difference |= usize::from(
            left.get(index).copied().unwrap_or(0) ^ right.get(index).copied().unwrap_or(0),
        );
    }
    difference == 0
}

fn role_name(role: ControlRole) -> &'static str {
    match role {
        ControlRole::Observer => "observer",
        ControlRole::Operator => "operator",
    }
}

fn request_name(request: &ControlRequest) -> &'static str {
    match request {
        ControlRequest::Health => "health",
        ControlRequest::Metrics => "metrics",
        ControlRequest::Runtime => "runtime",
        ControlRequest::ListSessions { .. } => "list_sessions",
        ControlRequest::ListScheduledJobs { .. } => "list_scheduled_jobs",
        ControlRequest::ListPendingApprovals { .. } => "list_pending_approvals",
        ControlRequest::ListAuditEvents { .. } => "list_audit_events",
        ControlRequest::ListObservabilityEvents { .. } => "list_observability_events",
        ControlRequest::SubscribeEvents { .. } => "subscribe_events",
        ControlRequest::CancelRequest { .. } => "cancel_request",
        ControlRequest::PauseAutonomy => "pause_autonomy",
        ControlRequest::ResumeAutonomy => "resume_autonomy",
        ControlRequest::CancelAction { .. } => "cancel_action",
        ControlRequest::DecideApproval { approve: true, .. } => "approve_proposal",
        ControlRequest::DecideApproval { approve: false, .. } => "deny_proposal",
        ControlRequest::ForceReconnect => "force_reconnect",
        ControlRequest::ExpireConversation { .. } => "expire_conversation",
        ControlRequest::SetRoamingJob { enabled: true, .. } => "enable_roaming_job",
        ControlRequest::SetRoamingJob { enabled: false, .. } => "disable_roaming_job",
        ControlRequest::InjectOperatorMessage { .. } => "inject_operator_message",
        ControlRequest::GracefulShutdown => "graceful_shutdown",
    }
}

fn unix_millis() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |duration| {
            u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
        })
}

fn lock<T>(value: &Mutex<T>) -> MutexGuard<'_, T> {
    value
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}