dpm_engine/query/
next.rs

1//! `next`: claimable work ranked by an explainable score, with the schedule risk of each candidate.
2
3use super::{downstream_counts, explain, is_unestimated, simulation_config};
4use crate::{EngineError, Transition, gates};
5use chrono::{DateTime, Utc};
6use dpm_model::{Plan, Priority, Timeline, WorkItem, WorkItemId};
7use dpm_schedule::{deterministic_remaining_at, drag_remaining_at, simulate_remaining_at};
8use serde::{Deserialize, Serialize};
9use std::collections::{BTreeMap, BTreeSet};
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
12#[serde(rename_all = "snake_case")]
13/// Which projection a candidate's criticality component comes from.
14pub enum CriticalityBasis {
15    /// The fraction of seeded simulations in which the task is critical.
16    Simulated,
17    /// Whether the task is on the deterministic critical path, as 1 or 0.
18    Deterministic,
19}
20
21#[derive(Debug, Clone, Serialize, Deserialize)]
22/// Ready task with a reproducible priority score and its explanation.
23pub struct NextWorkCandidate {
24    /// Work-item projection represented by this result.
25    pub work: WorkItem,
26    /// Ranking score composed from criticality, priority, downstream work, and float.
27    pub score: f64,
28    /// Whether total float is within numerical tolerance of zero.
29    pub critical: bool,
30    /// Fraction of sampled schedules in which each activity is critical.
31    pub criticality: Option<f64>,
32    /// Projection the score's criticality component used.
33    #[serde(default = "deterministic_basis")]
34    pub criticality_basis: CriticalityBasis,
35    /// Hours the deterministic finish would move if this task took no time: zero off the
36    /// critical path, and negative when shortening it would delay the finish. Reported for
37    /// context; the score does not use it.
38    #[serde(default)]
39    pub drag_hours: f64,
40    /// Delay available without postponing project completion.
41    pub total_float_hours: f64,
42    /// Count of distinct transitively dependent work items.
43    pub downstream_count: usize,
44    /// Whether this work fits the requested capability set.
45    pub capability_match: bool,
46    /// Human-readable factors contributing to this recommendation.
47    pub reasons: Vec<String>,
48}
49
50fn deterministic_basis() -> CriticalityBasis {
51    CriticalityBasis::Deterministic
52}
53
54#[derive(Debug, Clone, Serialize, Deserialize)]
55/// Eligibility and scheduling inputs for a work recommendation.
56pub struct NextWorkQuery {
57    /// Required capabilities the caller provides; an empty set disables capability filtering.
58    pub capabilities: BTreeSet<String>,
59    /// Whether ranking should use sampled criticality instead of deterministic criticality.
60    pub use_probabilistic_criticality: bool,
61}
62
63impl Default for NextWorkQuery {
64    fn default() -> Self {
65        Self {
66            capabilities: BTreeSet::new(),
67            use_probabilistic_criticality: true,
68        }
69    }
70}
71
72/// Recommend only tasks claimable at an adapter-supplied time and compatible with capabilities.
73pub fn next_work(
74    plan: &Plan,
75    query: &NextWorkQuery,
76    now: DateTime<Utc>,
77) -> Result<Vec<NextWorkCandidate>, EngineError> {
78    plan.validate()?;
79    let timeline = Timeline::at(plan, now);
80    let schedule = deterministic_remaining_at(plan, &timeline)?;
81    let simulation = if query.use_probabilistic_criticality
82        && plan
83            .work_items
84            .values()
85            .any(|w| w.schedule.estimate.is_some())
86    {
87        Some(simulate_remaining_at(plan, simulation_config(), &timeline)?)
88    } else {
89        None
90    };
91    let reports = super::claim_reports(plan, &timeline);
92    let eligible: BTreeSet<WorkItemId> = reports
93        .iter()
94        .filter(|(_, r)| r.ready)
95        .filter_map(|(id, _)| plan.work_items.get(id))
96        .filter(|w| {
97            query.capabilities.is_empty() || w.contract.capabilities.is_subset(&query.capabilities)
98        })
99        .map(|w| w.id)
100        .collect();
101    let downstream = downstream_counts(plan, &timeline, eligible.iter().copied());
102    // Only the candidates are recomputed, so drag never costs a pass per critical activity.
103    let drags = drag_remaining_at(plan, &timeline, &eligible)?;
104
105    let mut candidates = plan
106        .work_items
107        .values()
108        .filter(|work| eligible.contains(&work.id))
109        .filter_map(|work| reports.get(&work.id).map(|r| (work, r)))
110        // Claimable work is applicable, so the active-graph schedule always covers it.
111        .filter_map(|(work, claim)| {
112            let finish = gates::evaluate(plan, work, Transition::Submit, &timeline);
113            let activity = schedule.activities.get(&work.id)?;
114            let mut candidate = candidate(
115                work,
116                query,
117                (claim, &finish),
118                activity,
119                &downstream,
120                simulation.as_ref(),
121            );
122            candidate.drag_hours = drags.get(&work.id).copied().unwrap_or(0.0);
123            candidate.reasons.push(drag_reason(candidate.drag_hours));
124            if is_unestimated(&timeline, work) {
125                candidate.reasons.push(explain::UNESTIMATED_REASON.into());
126            }
127            Some(candidate)
128        })
129        .collect::<Vec<_>>();
130
131    candidates.sort_by(|a, b| {
132        b.score
133            .total_cmp(&a.score)
134            .then_with(|| a.work.key.natural_cmp(&b.work.key))
135    });
136    Ok(candidates)
137}
138
139/// The hours carry no space before their unit, so no drag statement reads as the unestimated
140/// "0 h" statement.
141fn drag_reason(hours: f64) -> String {
142    format!(
143        "drag {hours:.1}h is reported but not ranked: the score uses criticality, priority, downstream work and float"
144    )
145}
146
147fn candidate(
148    work: &WorkItem,
149    query: &NextWorkQuery,
150    gates: (&crate::GateReport, &crate::GateReport),
151    activity: &dpm_schedule::ActivitySchedule,
152    downstream: &BTreeMap<WorkItemId, usize>,
153    simulation: Option<&dpm_schedule::SimulationSummary>,
154) -> NextWorkCandidate {
155    let downstream_count = downstream.get(&work.id).copied().unwrap_or(0);
156    let criticality = simulation.and_then(|summary| summary.criticality.get(&work.id).copied());
157    let capability_match = work.contract.capabilities.is_empty()
158        || work.contract.capabilities.is_subset(&query.capabilities)
159        || query.capabilities.is_empty();
160
161    // Explicit components keep recommendation scores independently explainable.
162    let critical_component = criticality.unwrap_or(if activity.critical { 1.0 } else { 0.0 });
163    let capability_component = if capability_match { 15.0 } else { -100.0 };
164    let priority_component = f64::from(work.schedule.priority.rank()) * 8.0;
165    let downstream_component = libm::log(downstream_count as f64 + 1.0) * 6.0;
166    let float_penalty = activity.total_float_hours.min(100.0) * 0.05;
167    let score = critical_component * 100.0
168        + capability_component
169        + priority_component
170        + downstream_component
171        - float_penalty;
172
173    NextWorkCandidate {
174        work: work.clone(),
175        score,
176        critical: activity.critical,
177        criticality,
178        criticality_basis: if criticality.is_some() {
179            CriticalityBasis::Simulated
180        } else {
181            CriticalityBasis::Deterministic
182        },
183        drag_hours: 0.0,
184        total_float_hours: activity.total_float_hours,
185        downstream_count,
186        capability_match,
187        reasons: reasons(
188            work,
189            gates,
190            activity,
191            criticality,
192            downstream_count,
193            capability_match,
194        ),
195    }
196}
197
198fn reasons(
199    work: &WorkItem,
200    (claim, finish): (&crate::GateReport, &crate::GateReport),
201    activity: &dpm_schedule::ActivitySchedule,
202    criticality: Option<f64>,
203    downstream_count: usize,
204    capability_match: bool,
205) -> Vec<String> {
206    let mut reasons = Vec::new();
207    reasons.push(
208        "start gates are satisfied: every FS/SS predecessor event and positive lag has elapsed"
209            .into(),
210    );
211    reasons.extend(
212        claim
213            .provisional
214            .iter()
215            .map(crate::ProvisionalRelease::describe),
216    );
217    let pending_finish = finish
218        .unmet
219        .iter()
220        .filter(|g| matches!(g, crate::UnmetGate::Dependency { .. }))
221        .count();
222    if pending_finish > 0 {
223        reasons.push(format!(
224            "submission additionally waits for {pending_finish} FF/SF prerequisite(s); see explain transitions.submit"
225        ));
226    }
227    reasons.push(match criticality {
228        Some(value) => format!(
229            "criticality basis is simulated: critical in {:.0}% of schedule simulations",
230            value * 100.0
231        ),
232        None if activity.critical => {
233            "criticality basis is deterministic: on the deterministic critical path".into()
234        }
235        None => "criticality basis is deterministic: off the deterministic critical path".into(),
236    });
237    if downstream_count > 0 {
238        reasons.push(format!("unblocks {downstream_count} downstream work items"));
239    }
240    if work.schedule.priority <= Priority::P1 {
241        reasons.push(format!("human priority is {:?}", work.schedule.priority));
242    }
243    if !work.contract.capabilities.is_empty() {
244        reasons.push(if capability_match {
245            "requested capabilities match".into()
246        } else {
247            "requested capabilities do not fully match".into()
248        });
249    }
250    reasons
251}
252
253#[cfg(test)]
254mod tests;
dpm_engine/query/next.rs

dpm_engine/query/
next.rs
