// Request handlers intentionally spell out their deserialized protocol fields.
// Keeping those authority and payload values explicit at this boundary is safer
// than passing an unvalidated wire request deeper into the service.
#![allow(clippy::too_many_arguments)]

use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::time::{SystemTime, UNIX_EPOCH};

use clusterflux_core::{
    Actor, AgentId, ApiError, ApiErrorCategory, ApiErrorCode, ArtifactRegistry,
    CapabilityReportError, CredentialKind, Digest, DownloadError, LimitError, NativeQuicTransport,
    NodeDescriptor, NodeId, PanelError, PanelState, Placement, ProcessId, ProjectId, RateLimit,
    TenantId, TransportError, UserId,
};
use thiserror::Error;

use crate::{Coordinator, CoordinatorError, NodeScopeKey};

mod admin;
mod artifacts;
mod authenticated;
mod authorization;
mod debug;
mod debug_requests;
mod durable_runtime;
mod keys;
mod logs;
mod main_runtime;
mod nodes;
mod panels;
mod process_launch;
mod processes;
mod protocol;
mod quota;
mod relay;
mod routing;
mod signed_nodes;
mod summaries;
mod tcp;
mod wire_protocol;
use authorization::authorize_authenticated_user_operation;
use durable_runtime::RuntimeDurableStore;
use keys::{
    artifact_id_from_path, enrollment_grant_key, EnrollmentGrantKey, PanelStopKey,
    ProcessControlKey, TaskAssignmentKey, TaskControlKey, TaskRestartKey,
};
pub use protocol::{
    ArtifactAvailability, ArtifactRetentionState, ArtifactSummary, ArtifactTransferAssignment,
    AuthenticatedCoordinatorRequest, CoordinatorRequest, CoordinatorResponse,
    DebugAcknowledgementState, DebugAuditEvent, DebugEpochSummary, DebugParticipantAcknowledgement,
    NodeSummary, ProcessActivityState, ProcessFinalResult, ProcessLifecycleState, ProcessSummary,
    RecentLogEntry, SourcePreparationDisposition, SourcePreparationStatus, TaskAssignment,
    TaskAttemptSnapshot, TaskAttemptState, TaskCancellationTarget, TaskCompletionEvent,
    TaskExecutor, TaskFailureResolution, TaskLogStream, TaskReplacementBundle, TaskTerminalState,
    VirtualProcessStatus, WorkflowActor,
};
pub use quota::CoordinatorQuotaConfiguration;
pub use relay::{
    ArtifactRelayConfiguration, ArtifactRelayDurableState, ArtifactRelayError, ArtifactRelayUsage,
};
pub use tcp::{bind_listener, ClientAuthorityMode};
pub use wire_protocol::CoordinatorWireRequest;

const MAX_TASK_LOG_TAIL_BYTES: usize = 256 * 1024;
const DEBUG_CONTROL_READ_BYTES: u64 = 1024;
const MAX_REPLAY_NONCES_PER_AUTHORITY: usize = 1_024;
const NODE_SIGNATURE_WINDOW_SECONDS: u64 = 30;
const MAX_NODE_REPLAY_NONCES_PER_AUTHORITY: usize = 4_096;
const MAX_ENROLLMENT_GRANTS_PER_PROJECT: usize = 64;
const MAX_TASK_EVENTS_PER_PROCESS: usize = 128;
const MAX_DEBUG_AUDIT_EVENTS_PER_PROCESS: usize = 256;
const MAX_RESTART_CHECKPOINTS_PER_PROCESS: usize = 128;
const MAX_TASK_EVENTS_TOTAL: usize = 8_192;
const MAX_DEBUG_AUDIT_EVENTS_TOTAL: usize = 8_192;
const MAX_RESTART_CHECKPOINTS_TOTAL: usize = 4_096;
const MAX_TASK_ATTEMPT_HISTORIES: usize = 1_000_000;
const MAX_IN_FLIGHT_TASKS_PER_PROCESS: usize = 256;
const MAX_NODE_REPORTED_OBJECTS_PER_KIND: usize = 1_024;
const MAX_RECENT_LOG_ENTRIES_PER_PROCESS: usize = 256;
const MAX_RECENT_LOG_ENTRIES_PER_PROJECT: usize = 1_024;
const MAX_RECENT_LOG_BYTES_PER_PROJECT: usize = 512 * 1024;
const MAX_RECENT_LOG_CHUNK_BYTES: usize = 16 * 1024;
const MAX_RECENT_PROCESS_SUMMARIES_PER_PROJECT: usize = 32;
const MAX_RECENT_PROCESS_SUMMARIES_TOTAL: usize = 8_192;
const DEFAULT_NODE_STALE_AFTER_SECONDS: u64 = 30;
fn bounded_ttl(requested: u64, maximum: u64) -> u64 {
    requested.clamp(1, maximum)
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CoordinatorMainRuntimeConfiguration {
    pub fuel_units_per_second: u64,
    pub fuel_burst_seconds: u64,
    pub memory_bytes: usize,
    pub nested_join_timeout_ms: u64,
    pub max_active_mains: usize,
    pub max_wakeups_per_minute: u64,
    pub max_output_bytes: usize,
    pub max_state_bytes: usize,
}

impl Default for CoordinatorMainRuntimeConfiguration {
    fn default() -> Self {
        Self {
            fuel_units_per_second: 10_000_000,
            fuel_burst_seconds: 60,
            memory_bytes: 256 * 1024 * 1024,
            nested_join_timeout_ms: 24 * 60 * 60 * 1_000,
            max_active_mains: usize::MAX,
            max_wakeups_per_minute: 6_000,
            max_output_bytes: MAX_TASK_LOG_TAIL_BYTES,
            max_state_bytes: clusterflux_core::MAX_WASM_TASK_ENVELOPE_BYTES,
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CoordinatorAdmission {
    pub workflow_placement_allowed: bool,
    pub max_node_enrollment_ttl_seconds: u64,
    pub max_artifact_download_ttl_seconds: u64,
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct CoordinatorOperationalMetrics {
    pub tenants: usize,
    pub users: usize,
    pub projects: usize,
    pub enrolled_nodes: usize,
    pub reported_nodes: usize,
    pub live_nodes: usize,
    pub active_processes: usize,
    pub active_coordinator_mains: usize,
    pub max_active_coordinator_mains: usize,
    pub active_tasks: usize,
    pub queued_tasks: usize,
    pub artifacts: usize,
    pub retained_download_links: usize,
    pub relay: ArtifactRelayUsage,
}

impl Default for CoordinatorAdmission {
    fn default() -> Self {
        Self {
            workflow_placement_allowed: true,
            max_node_enrollment_ttl_seconds: 15 * 60,
            max_artifact_download_ttl_seconds: 15 * 60,
        }
    }
}

#[derive(Debug, Error)]
pub enum CoordinatorServiceError {
    #[error("coordinator protocol I/O error: {0}")]
    Io(#[from] std::io::Error),
    #[error("coordinator protocol JSON error: {0}")]
    Json(#[from] serde_json::Error),
    #[error("coordinator protocol error: {0}")]
    Protocol(String),
    #[error("coordinator request failed: {0}")]
    Coordinator(#[from] CoordinatorError),
    #[error("artifact download request failed: {0}")]
    Download(#[from] clusterflux_core::DownloadError),
    #[error("scheduler placement failed: {0}")]
    Scheduler(#[from] clusterflux_core::PlacementError),
    #[error("transport request failed: {0}")]
    Transport(#[from] TransportError),
    #[error("resource limit failed: {0}")]
    Resource(#[from] LimitError),
    #[error("operator panel request failed: {0}")]
    Panel(#[from] clusterflux_core::PanelError),
    #[error("invalid node capability report: {0}")]
    CapabilityReport(#[from] CapabilityReportError),
    #[error("invalid VFS artifact path reported by node: {0}")]
    InvalidArtifactPath(String),
    #[error("invalid task log tail reported by node: {0}")]
    InvalidTaskLogTail(String),
    #[error("durable coordinator state failed: {0}")]
    Durable(String),
}

impl CoordinatorServiceError {
    pub fn api_error(&self, request_id: impl Into<String>) -> ApiError {
        let request_id = request_id.into();
        let message = self.to_string();
        let (code, category, retryable) = match self {
            Self::Io(_) => (
                ApiErrorCode::TemporaryCapacity,
                ApiErrorCategory::Availability,
                true,
            ),
            Self::Json(_)
            | Self::CapabilityReport(_)
            | Self::InvalidArtifactPath(_)
            | Self::InvalidTaskLogTail(_) => (
                ApiErrorCode::ValidationError,
                ApiErrorCategory::Validation,
                false,
            ),
            Self::Protocol(_) | Self::Coordinator(CoordinatorError::Unauthorized(_)) => {
                return ApiError::from_message(request_id, message);
            }
            Self::Coordinator(CoordinatorError::UnknownNode) => {
                (ApiErrorCode::NotFound, ApiErrorCategory::State, false)
            }
            Self::Coordinator(CoordinatorError::Enrollment(_)) => (
                ApiErrorCode::Unauthenticated,
                ApiErrorCategory::Authentication,
                false,
            ),
            Self::Coordinator(CoordinatorError::StaleProcessEpoch { .. }) => {
                (ApiErrorCode::Conflict, ApiErrorCategory::State, true)
            }
            Self::Download(DownloadError::NotFound) => {
                (ApiErrorCode::NotFound, ApiErrorCategory::State, false)
            }
            Self::Download(DownloadError::Unavailable)
            | Self::Download(DownloadError::DirectConnectivityUnavailable(_)) => (
                ApiErrorCode::ArtifactUnavailable,
                ApiErrorCategory::Availability,
                true,
            ),
            Self::Download(DownloadError::LimitExceeded { .. }) => (
                ApiErrorCode::ArtifactLimitExceeded,
                ApiErrorCategory::Resource,
                false,
            ),
            Self::Download(DownloadError::Unauthorized(_))
            | Self::Download(DownloadError::InvalidToken)
            | Self::Download(DownloadError::Expired)
            | Self::Download(DownloadError::Revoked) => (
                ApiErrorCode::Forbidden,
                ApiErrorCategory::Authorization,
                false,
            ),
            Self::Download(DownloadError::Usage(_)) | Self::Resource(_) => (
                ApiErrorCode::QuotaExceeded,
                ApiErrorCategory::Resource,
                true,
            ),
            Self::Scheduler(_) => (
                ApiErrorCode::NoCapableNode,
                ApiErrorCategory::Availability,
                true,
            ),
            Self::Transport(_) => (
                ApiErrorCode::NodeOffline,
                ApiErrorCategory::Availability,
                true,
            ),
            Self::Panel(PanelError::RateLimited) => (
                ApiErrorCode::QuotaExceeded,
                ApiErrorCategory::Resource,
                true,
            ),
            Self::Panel(PanelError::UnknownWidget(_)) => {
                (ApiErrorCode::NotFound, ApiErrorCategory::State, false)
            }
            Self::Panel(_) => (
                ApiErrorCode::Forbidden,
                ApiErrorCategory::Authorization,
                false,
            ),
            Self::Durable(_) => (
                ApiErrorCode::InternalError,
                ApiErrorCategory::Internal,
                true,
            ),
        };
        ApiError::new(code, category, message, retryable, request_id)
    }
}

pub struct CoordinatorService {
    coordinator: Coordinator,
    store: RuntimeDurableStore,
    node_descriptors: BTreeMap<NodeScopeKey, NodeDescriptor>,
    node_last_seen_epoch_seconds: BTreeMap<NodeScopeKey, u64>,
    node_stale_after_seconds: u64,
    debug_freeze_timeout: std::time::Duration,
    enrollment_grants: BTreeMap<EnrollmentGrantKey, clusterflux_core::EnrollmentGrant>,
    task_events: VecDeque<TaskCompletionEvent>,
    process_scope_history: VecDeque<ProcessControlKey>,
    process_summaries: BTreeMap<ProcessControlKey, summaries::StoredProcessSummary>,
    process_summary_order: VecDeque<ProcessControlKey>,
    next_process_summary_order: u64,
    task_terminal_states: BTreeMap<TaskRestartKey, TaskTerminalState>,
    recent_logs: BTreeMap<(TenantId, ProjectId), VecDeque<RecentLogEntry>>,
    recent_log_dropped_through: BTreeMap<ProcessControlKey, u64>,
    recent_log_accounted_bytes: BTreeMap<
        (
            TenantId,
            ProjectId,
            ProcessId,
            clusterflux_core::TaskInstanceId,
            String,
        ),
        u64,
    >,
    recent_log_truncated_streams: BTreeSet<(
        TenantId,
        ProjectId,
        ProcessId,
        clusterflux_core::TaskInstanceId,
        String,
    )>,
    next_recent_log_sequence: u64,
    debug_audit_events: VecDeque<DebugAuditEvent>,
    debug_epochs: BTreeMap<ProcessControlKey, u64>,
    debug_epoch_runtime: BTreeMap<ProcessControlKey, debug::DebugEpochRuntime>,
    debug_breakpoints: BTreeMap<ProcessControlKey, debug::DebugBreakpointPlan>,
    debug_commands: BTreeMap<TaskControlKey, debug::DebugPendingCommand>,
    task_assignments: BTreeMap<TaskAssignmentKey, VecDeque<TaskAssignment>>,
    task_restart_checkpoints: BTreeMap<TaskRestartKey, processes::TaskRestartCheckpoint>,
    task_restart_checkpoint_order: VecDeque<TaskRestartKey>,
    task_attempts: BTreeMap<TaskRestartKey, Vec<protocol::TaskAttemptSnapshot>>,
    restart_launches: BTreeSet<TaskRestartKey>,
    main_runtime: main_runtime::CoordinatorMainRuntime,
    pending_task_launches: VecDeque<processes::PendingTaskLaunch>,
    task_placements: BTreeMap<TaskControlKey, Placement>,
    active_tasks: BTreeSet<TaskControlKey>,
    task_cancellations: BTreeSet<TaskControlKey>,
    task_aborts: BTreeSet<TaskControlKey>,
    process_cancellations: BTreeSet<ProcessControlKey>,
    process_aborts: BTreeSet<ProcessControlKey>,
    agent_replay_nonces: BTreeMap<(TenantId, ProjectId, AgentId, String), u64>,
    node_replay_nonces: BTreeMap<(NodeScopeKey, String), u64>,
    panel_snapshots: BTreeMap<PanelStopKey, PanelState>,
    stopped_panels: BTreeSet<PanelStopKey>,
    panel_event_limits: BTreeMap<(TenantId, ProjectId, ProcessId, String), RateLimit>,
    artifact_registry: ArtifactRegistry,
    artifact_reverse_transfers: BTreeMap<String, artifacts::ArtifactReverseTransfer>,
    artifact_transfer_by_token: BTreeMap<Digest, String>,
    artifact_relay: relay::ArtifactRelayLedger,
    transport: NativeQuicTransport,
    quota: quota::CoordinatorQuota,
    admission: CoordinatorAdmission,
    #[cfg(test)]
    server_time_override: Option<u64>,
    admin_token_digest: Option<Digest>,
    admin_replay_nonces: BTreeMap<String, u64>,
}

impl CoordinatorService {
    pub fn configure_artifact_relay(
        &mut self,
        configuration: ArtifactRelayConfiguration,
    ) -> Result<(), CoordinatorServiceError> {
        let now_epoch_seconds = self.current_epoch_seconds()?;
        let mut candidate = self.artifact_relay.clone();
        candidate
            .configure(configuration)
            .map_err(|error| CoordinatorServiceError::Protocol(error.to_string()))?;
        candidate.reconcile_after_restart(now_epoch_seconds);
        self.commit_artifact_relay(candidate)
    }

    pub fn artifact_relay_usage(&self) -> ArtifactRelayUsage {
        self.artifact_relay.usage()
    }

    pub fn operational_metrics(&self) -> CoordinatorOperationalMetrics {
        CoordinatorOperationalMetrics {
            tenants: self.coordinator.tenant_count(),
            users: self.coordinator.user_count(),
            projects: self.coordinator.project_count(),
            enrolled_nodes: self.coordinator.node_identity_count(),
            reported_nodes: self.node_descriptors.len(),
            live_nodes: self
                .node_descriptors
                .keys()
                .filter(|scope| self.node_is_live(scope))
                .count(),
            active_processes: self.coordinator.active_process_count(),
            active_coordinator_mains: self.main_runtime.active_main_count(),
            max_active_coordinator_mains: self.main_runtime.max_active_mains(),
            active_tasks: self.active_tasks.len(),
            queued_tasks: self.pending_task_launches.len(),
            artifacts: self.artifact_registry.artifact_count(),
            retained_download_links: self.artifact_registry.retained_download_link_count(),
            relay: self.artifact_relay.usage(),
        }
    }

    fn commit_artifact_relay(
        &mut self,
        candidate: relay::ArtifactRelayLedger,
    ) -> Result<(), CoordinatorServiceError> {
        let previous = self.artifact_relay.clone();
        let previous_state = previous.durable_state();
        self.artifact_relay = candidate;
        self.coordinator
            .set_artifact_relay_state(self.artifact_relay.durable_state());
        if let Err(error) = self.persist_durable_state() {
            self.artifact_relay = previous;
            self.coordinator.set_artifact_relay_state(previous_state);
            return Err(error);
        }
        Ok(())
    }

    fn mutate_artifact_relay<T>(
        &mut self,
        mutation: impl FnOnce(&mut relay::ArtifactRelayLedger) -> Result<T, ArtifactRelayError>,
    ) -> Result<T, CoordinatorServiceError> {
        let mut candidate = self.artifact_relay.clone();
        let result = mutation(&mut candidate)
            .map_err(|error| CoordinatorServiceError::Protocol(error.to_string()))?;
        self.commit_artifact_relay(candidate)?;
        Ok(result)
    }

    pub fn set_debug_freeze_timeout(&mut self, timeout: std::time::Duration) {
        self.debug_freeze_timeout = timeout.max(std::time::Duration::from_millis(1));
    }

    pub fn configure_coordinator_main_runtime(
        &mut self,
        configuration: CoordinatorMainRuntimeConfiguration,
    ) -> Result<(), CoordinatorServiceError> {
        self.main_runtime.configure(configuration)
    }

    pub fn record_service_policy(
        &mut self,
        tenant: TenantId,
        name: impl Into<String>,
        digest: Digest,
    ) -> Result<crate::ServicePolicyRecord, CoordinatorServiceError> {
        let name = name.into();
        self.coordinator
            .upsert_service_policy_record(tenant.clone(), name.clone(), digest);
        self.persist_durable_state()?;
        Ok(self
            .coordinator
            .service_policy_record(&tenant, &name)
            .expect("service policy record was persisted immediately after insertion")
            .clone())
    }

    pub(super) fn authorize_node_for_process_or_termination(
        &self,
        node: &NodeId,
        tenant: &TenantId,
        project: &ProjectId,
        process: &ProcessId,
    ) -> Result<(), CoordinatorServiceError> {
        let process_key = keys::process_control_key(tenant, project, process);
        if self.process_cancellations.contains(&process_key)
            || self.process_aborts.contains(&process_key)
        {
            let identity = self
                .coordinator
                .node_identity(tenant, project, node)
                .ok_or(CoordinatorError::UnknownNode)?;
            debug_assert_eq!(&identity.tenant, tenant);
            debug_assert_eq!(&identity.project, project);
            return Ok(());
        }
        self.coordinator
            .authorize_node_for_process(node, tenant, project, process)?;
        Ok(())
    }

    pub fn new(coordinator_epoch: u64) -> Self {
        Self::new_with_optional_admin_token_and_admission(
            coordinator_epoch,
            std::env::var("CLUSTERFLUX_ADMIN_TOKEN").ok(),
            CoordinatorAdmission::default(),
        )
    }

    pub fn new_with_admin_token(coordinator_epoch: u64, admin_token: impl Into<String>) -> Self {
        Self::new_with_optional_admin_token_and_admission(
            coordinator_epoch,
            Some(admin_token.into()),
            CoordinatorAdmission::default(),
        )
    }

    pub fn new_with_admission(coordinator_epoch: u64, admission: CoordinatorAdmission) -> Self {
        Self::new_with_optional_admin_token_and_admission(
            coordinator_epoch,
            std::env::var("CLUSTERFLUX_ADMIN_TOKEN").ok(),
            admission,
        )
    }

    fn new_with_optional_admin_token_and_admission(
        coordinator_epoch: u64,
        admin_token: Option<String>,
        admission: CoordinatorAdmission,
    ) -> Self {
        Self::try_new_with_optional_admin_token_admission_and_database_url(
            coordinator_epoch,
            admin_token,
            admission,
            None,
            CoordinatorQuotaConfiguration::default(),
        )
        .expect("in-memory durable coordinator store initialization cannot fail")
    }

    pub fn new_with_database_url(
        coordinator_epoch: u64,
        database_url: Option<&str>,
    ) -> Result<Self, CoordinatorServiceError> {
        Self::try_new_with_optional_admin_token_admission_and_database_url(
            coordinator_epoch,
            std::env::var("CLUSTERFLUX_ADMIN_TOKEN").ok(),
            CoordinatorAdmission::default(),
            database_url,
            CoordinatorQuotaConfiguration::default(),
        )
    }

    pub fn new_with_admin_token_and_database_url(
        coordinator_epoch: u64,
        admin_token: impl Into<String>,
        database_url: Option<&str>,
    ) -> Result<Self, CoordinatorServiceError> {
        Self::new_with_admin_token_database_url_and_quota(
            coordinator_epoch,
            admin_token,
            database_url,
            CoordinatorQuotaConfiguration::default(),
        )
    }

    pub fn new_with_admin_token_database_url_and_quota(
        coordinator_epoch: u64,
        admin_token: impl Into<String>,
        database_url: Option<&str>,
        quota_configuration: CoordinatorQuotaConfiguration,
    ) -> Result<Self, CoordinatorServiceError> {
        Self::try_new_with_optional_admin_token_admission_and_database_url(
            coordinator_epoch,
            Some(admin_token.into()),
            CoordinatorAdmission::default(),
            database_url,
            quota_configuration,
        )
    }

    fn try_new_with_optional_admin_token_admission_and_database_url(
        coordinator_epoch: u64,
        admin_token: Option<String>,
        admission: CoordinatorAdmission,
        database_url: Option<&str>,
        quota_configuration: CoordinatorQuotaConfiguration,
    ) -> Result<Self, CoordinatorServiceError> {
        let mut store = RuntimeDurableStore::from_database_url(database_url)
            .map_err(CoordinatorServiceError::Durable)?;
        let coordinator = Coordinator::try_boot(&mut store, coordinator_epoch)
            .map_err(CoordinatorServiceError::Durable)?;
        let artifact_relay_state = coordinator.artifact_relay_state().clone();
        let admin_token_digest = admin_token
            .filter(|token| !token.trim().is_empty())
            .map(Digest::sha256);
        Ok(Self {
            coordinator,
            store,
            node_descriptors: BTreeMap::new(),
            node_last_seen_epoch_seconds: BTreeMap::new(),
            node_stale_after_seconds: std::env::var("CLUSTERFLUX_NODE_STALE_AFTER_SECONDS")
                .ok()
                .and_then(|value| value.parse::<u64>().ok())
                .filter(|seconds| *seconds > 0)
                .unwrap_or(DEFAULT_NODE_STALE_AFTER_SECONDS),
            debug_freeze_timeout: std::time::Duration::from_secs(5),
            enrollment_grants: BTreeMap::new(),
            task_events: VecDeque::new(),
            process_scope_history: VecDeque::new(),
            process_summaries: BTreeMap::new(),
            process_summary_order: VecDeque::new(),
            next_process_summary_order: 1,
            task_terminal_states: BTreeMap::new(),
            recent_logs: BTreeMap::new(),
            recent_log_dropped_through: BTreeMap::new(),
            recent_log_accounted_bytes: BTreeMap::new(),
            recent_log_truncated_streams: BTreeSet::new(),
            next_recent_log_sequence: 1,
            debug_audit_events: VecDeque::new(),
            debug_epochs: BTreeMap::new(),
            debug_epoch_runtime: BTreeMap::new(),
            debug_breakpoints: BTreeMap::new(),
            debug_commands: BTreeMap::new(),
            task_assignments: BTreeMap::new(),
            task_restart_checkpoints: BTreeMap::new(),
            task_restart_checkpoint_order: VecDeque::new(),
            task_attempts: BTreeMap::new(),
            restart_launches: BTreeSet::new(),
            main_runtime: main_runtime::CoordinatorMainRuntime::default(),
            pending_task_launches: VecDeque::new(),
            task_placements: BTreeMap::new(),
            active_tasks: BTreeSet::new(),
            task_cancellations: BTreeSet::new(),
            task_aborts: BTreeSet::new(),
            process_cancellations: BTreeSet::new(),
            process_aborts: BTreeSet::new(),
            agent_replay_nonces: BTreeMap::new(),
            node_replay_nonces: BTreeMap::new(),
            panel_snapshots: BTreeMap::new(),
            stopped_panels: BTreeSet::new(),
            panel_event_limits: BTreeMap::new(),
            artifact_registry: ArtifactRegistry::default(),
            artifact_reverse_transfers: BTreeMap::new(),
            artifact_transfer_by_token: BTreeMap::new(),
            artifact_relay: relay::ArtifactRelayLedger::from_durable(
                ArtifactRelayConfiguration::default(),
                artifact_relay_state,
            ),
            transport: NativeQuicTransport,
            quota: quota::CoordinatorQuota::new(quota_configuration),
            admission,
            #[cfg(test)]
            server_time_override: None,
            admin_token_digest,
            admin_replay_nonces: BTreeMap::new(),
        })
    }

    pub fn durable_store_kind(&self) -> &'static str {
        self.store.kind()
    }

    fn persist_durable_state(&mut self) -> Result<(), CoordinatorServiceError> {
        self.coordinator
            .try_persist(&mut self.store)
            .map_err(CoordinatorServiceError::Durable)
    }

    fn current_epoch_seconds(&self) -> Result<u64, CoordinatorServiceError> {
        #[cfg(test)]
        if let Some(now) = self.server_time_override {
            return Ok(now);
        }
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|duration| duration.as_secs())
            .map_err(|err| CoordinatorServiceError::Protocol(format!("system clock error: {err}")))
    }

    fn handle_quota_status(
        &self,
        tenant: String,
        project: String,
        actor_user: String,
    ) -> Result<CoordinatorResponse, CoordinatorServiceError> {
        let tenant = TenantId::new(tenant);
        let project = ProjectId::new(project);
        let actor = UserId::new(actor_user);
        let durable_project = self.coordinator.project(&project).ok_or_else(|| {
            CoordinatorError::Unauthorized("quota status requires an existing project".to_owned())
        })?;
        if durable_project.tenant != tenant {
            return Err(CoordinatorError::Unauthorized(
                "quota status project is outside the tenant scope".to_owned(),
            )
            .into());
        }
        let now_epoch_seconds = self.current_epoch_seconds()?;
        let status = self
            .quota
            .project_status(&tenant, &project, now_epoch_seconds);
        Ok(CoordinatorResponse::QuotaStatus {
            tenant,
            project,
            actor,
            policy_label: status.policy_label,
            limits: status.limits,
            window_seconds: status.window_seconds,
            usage: status.usage,
            window_started_epoch_seconds: status.window_started_epoch_seconds,
        })
    }

    #[cfg(test)]
    fn set_server_time(&mut self, now_epoch_seconds: u64) {
        self.server_time_override = Some(now_epoch_seconds);
    }

    pub fn issue_cli_session(
        &mut self,
        tenant: TenantId,
        project: ProjectId,
        user: UserId,
        session_secret: &str,
        expires_at_epoch_seconds: Option<u64>,
    ) -> Result<crate::CliSessionRecord, CoordinatorServiceError> {
        if let Some(existing) = self.coordinator.project(&project) {
            if existing.tenant != tenant {
                return Err(CoordinatorError::Unauthorized(
                    "CLI session project belongs to a different tenant".to_owned(),
                )
                .into());
            }
        } else {
            self.coordinator
                .upsert_project(tenant.clone(), project.clone(), "Session project");
        }
        self.coordinator
            .grant_project_debug(tenant.clone(), project.clone(), user.clone());
        let record = self.coordinator.issue_cli_session(
            tenant,
            project,
            user,
            session_secret,
            expires_at_epoch_seconds,
        );
        self.persist_durable_state()?;
        Ok(record)
    }

    pub fn authenticate_cli_session_context(
        &self,
        session_secret: &str,
    ) -> Result<clusterflux_core::AuthContext, CoordinatorServiceError> {
        Ok(self.coordinator.authenticate_cli_session(session_secret)?)
    }

    pub fn authenticate_cli_session_status_context(
        &self,
        session_secret: &str,
    ) -> Result<clusterflux_core::AuthContext, CoordinatorServiceError> {
        Ok(self
            .coordinator
            .authenticate_cli_session_for_status(session_secret)?)
    }

    pub fn revoke_cli_session(
        &mut self,
        session_secret: &str,
    ) -> Result<crate::CliSessionRecord, CoordinatorServiceError> {
        let record = self.coordinator.revoke_cli_session(session_secret)?;
        self.persist_durable_state()?;
        Ok(record)
    }
}

#[cfg(test)]
mod tests;