shepherd-core 6.6.0

//! Native live-process budget facts for one bounded dispatch parent.
//!
//! Logical graph width is never process authority. The adapter supplies facts
//! reopened from the host, project configuration, verified plan, parent role,
//! and remaining run budget; this module computes the only effective live
//! ceiling and rejects an overlapping launch before provider creation.

#[cfg(feature = "alloc")]
use alloc::{format, string::String};

use crate::Harness;

use super::{DispatchError, DispatchId, DispatchResult, LaneId, ProjectId, Role, RunId, SessionId};

pub const DISPATCH_BUDGET_SCHEMA: &str = "shepherd.dispatch-budget/1";

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct DispatchBudgetInput {
    pub project_id: ProjectId,
    pub run: RunId,
    pub root_session_id: SessionId,
    pub parent_dispatch_id: Option<DispatchId>,
    pub lane: Option<LaneId>,
    pub parent_role: Role,
    pub harness: Harness,
    pub host_concurrent_limit: Option<u32>,
    pub project_parallel_limit: u32,
    pub plan_parallel_limit: u32,
    pub parent_parallel_limit: u32,
    pub remaining_run_dispatches: u32,
    pub observed_live: u32,
    pub observed_total: u32,
    pub measured_at: i64,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct DispatchBudget {
    pub schema: String,
    pub project_id: ProjectId,
    pub run: RunId,
    pub root_session_id: SessionId,
    pub parent_dispatch_id: Option<DispatchId>,
    pub lane: Option<LaneId>,
    pub parent_role: Role,
    pub harness: Harness,
    pub host_concurrent_limit: Option<u32>,
    pub project_parallel_limit: u32,
    pub plan_parallel_limit: u32,
    pub parent_parallel_limit: u32,
    pub remaining_run_dispatches: u32,
    pub effective_live_limit: u32,
    pub observed_live: u32,
    pub observed_total: u32,
    pub measured_at: i64,
}

impl DispatchBudget {
    pub fn measure(input: DispatchBudgetInput) -> DispatchResult<Self> {
        if input.measured_at < 0
            || input.project_parallel_limit == 0
            || input.plan_parallel_limit == 0
            || input.parent_parallel_limit == 0
            || input.remaining_run_dispatches == 0
            || input.host_concurrent_limit == Some(0)
        {
            return Err(DispatchError::InvalidBudget(
                "dispatch budget inputs must be positive native facts".into(),
            ));
        }
        if input.parent_dispatch_id.is_none() != input.parent_role.is_root() {
            return Err(DispatchError::InvalidBudget(
                "root and child parent ancestry disagree".into(),
            ));
        }
        if input.parent_role == Role::Conductor && input.lane.is_none() {
            return Err(DispatchError::InvalidBudget(
                "Conductor budget requires one verified lane".into(),
            ));
        }
        let mut effective = input
            .project_parallel_limit
            .min(input.plan_parallel_limit)
            .min(input.parent_parallel_limit)
            .min(input.remaining_run_dispatches);
        if let Some(host) = input.host_concurrent_limit {
            effective = effective.min(host);
        }
        let value = Self {
            schema: DISPATCH_BUDGET_SCHEMA.into(),
            project_id: input.project_id,
            run: input.run,
            root_session_id: input.root_session_id,
            parent_dispatch_id: input.parent_dispatch_id,
            lane: input.lane,
            parent_role: input.parent_role,
            harness: input.harness,
            host_concurrent_limit: input.host_concurrent_limit,
            project_parallel_limit: input.project_parallel_limit,
            plan_parallel_limit: input.plan_parallel_limit,
            parent_parallel_limit: input.parent_parallel_limit,
            remaining_run_dispatches: input.remaining_run_dispatches,
            effective_live_limit: effective,
            observed_live: input.observed_live,
            observed_total: input.observed_total,
            measured_at: input.measured_at,
        };
        value.validate()?;
        Ok(value)
    }

    pub fn validate(&self) -> DispatchResult<()> {
        let expected = Self::measure_unchecked_min(
            self.host_concurrent_limit,
            self.project_parallel_limit,
            self.plan_parallel_limit,
            self.parent_parallel_limit,
            self.remaining_run_dispatches,
        )?;
        if self.schema != DISPATCH_BUDGET_SCHEMA
            || self.measured_at < 0
            || self.effective_live_limit != expected
            || self.parent_dispatch_id.is_none() != self.parent_role.is_root()
            || (self.parent_role == Role::Conductor && self.lane.is_none())
        {
            return Err(DispatchError::InvalidBudget(
                "persisted dispatch budget is inconsistent".into(),
            ));
        }
        Ok(())
    }

    /// Reject the next overlapping provider before process creation.
    pub fn authorize_next(&self) -> DispatchResult<()> {
        self.validate()?;
        let next_live = self
            .observed_live
            .checked_add(1)
            .ok_or_else(|| DispatchError::InvalidBudget("live count overflow".into()))?;
        if next_live > self.effective_live_limit {
            return Err(DispatchError::HarnessLimit {
                harness: self.harness,
                kind: "effective concurrent child",
                limit: self.effective_live_limit,
                observed: next_live,
            });
        }
        if self.remaining_run_dispatches == 0 {
            return Err(DispatchError::InvalidBudget(
                "run dispatch budget is exhausted".into(),
            ));
        }
        Ok(())
    }

    fn measure_unchecked_min(
        host: Option<u32>,
        project: u32,
        plan: u32,
        parent: u32,
        remaining: u32,
    ) -> DispatchResult<u32> {
        if project == 0 || plan == 0 || parent == 0 || remaining == 0 || host == Some(0) {
            return Err(DispatchError::InvalidBudget(format!(
                "zero dispatch budget input: host={host:?} project={project} plan={plan} parent={parent} remaining={remaining}"
            )));
        }
        let effective = project.min(plan).min(parent).min(remaining);
        Ok(host.map_or(effective, |limit| effective.min(limit)))
    }
}