polyc-query 2026.9.0

//! Coordinates shared deployment admission for exact projected executions.

use std::sync::Arc;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio_util::sync::CancellationToken;

use super::{CoreExecutionError, CoreOperationContext, operation_wait};

/// Enforces a deployment-owned execution bound across request catalogs.
///
/// This slot is distinct from the durable audit `CoreExecutionPermit`.
/// The audit permit proves that one intent can execute.
/// This admission bound caps concurrent artifact and `DataFusion` work.
/// The future production composition must install one shared instance.
/// This uncomposed seam cannot prove process-wide scope by itself.
#[derive(Debug)]
pub(crate) struct CoreExecutionAdmission {
    slots: Arc<Semaphore>,
}

/// Names the deployment input for the aggregate execution bound.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct CoreExecutionAdmissionInput {
    pub(crate) max_concurrent_executions: usize,
}

impl TryFrom<CoreExecutionAdmissionInput> for CoreExecutionAdmission {
    type Error = CoreExecutionError;

    fn try_from(value: CoreExecutionAdmissionInput) -> Result<Self, Self::Error> {
        let CoreExecutionAdmissionInput {
            max_concurrent_executions,
        } = value;
        if max_concurrent_executions == 0 {
            return Err(CoreExecutionError::InvalidComposition(
                "the aggregate execution limit is zero",
            ));
        }
        // The semaphore asserts its own ceiling. A deployment value above it
        // would panic inside the constructor rather than refuse here, and this
        // module promises that no deployment value panics.
        if max_concurrent_executions > Semaphore::MAX_PERMITS {
            return Err(CoreExecutionError::InvalidComposition(
                "the aggregate execution limit exceeds the permit ceiling",
            ));
        }
        Ok(Self {
            slots: Arc::new(Semaphore::new(max_concurrent_executions)),
        })
    }
}

impl CoreExecutionAdmission {
    pub(super) async fn acquire(
        &self,
        operation: &CoreOperationContext,
        cancellation: &CancellationToken,
    ) -> Result<OwnedSemaphorePermit, CoreExecutionError> {
        operation_wait(
            operation,
            cancellation,
            Arc::clone(&self.slots).acquire_owned(),
        )
        .await?
        .map_err(|_| CoreExecutionError::ExecutionAdmissionClosed)
    }
}