dpm_engine/query/
flow.rs

1//! Flow of work through execution: cycle and lead time, throughput, aging work in progress and
2//! how reliably a claim ends in verified work.
3
4use super::calibration::{HistoryIndex, ascending, elapsed_hours, exclusions};
5use crate::{Command, Exclusion, ExclusionReason, Operation};
6use chrono::{DateTime, Datelike, TimeDelta, Utc};
7use dpm_model::{ActorId, ActorKind, Key, Plan, WorkItem, WorkItemId, WorkStatus};
8use dpm_schedule::percentile;
9use serde::{Deserialize, Serialize};
10use std::collections::{BTreeMap, BTreeSet};
11
12/// ISO weeks the throughput history covers, the current one included.
13pub const THROUGHPUT_WEEKS: usize = 8;
14
15/// Flow metrics at one clock reading.
16#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
17pub struct FlowReport {
18    /// Elapsed hours from start to verification of verified tasks not recorded in bulk.
19    pub cycle_time: Durations,
20    /// Elapsed hours from the first claim to verification of verified tasks not recorded in bulk.
21    pub lead_time: Durations,
22    /// Tasks verified in recent periods.
23    pub throughput: Throughput,
24    /// Every claimed, started, blocked or submitted task, oldest claim first.
25    pub aging: Vec<AgingWork>,
26    /// How claims ended, overall and per holder kind.
27    pub reliability: Reliability,
28}
29
30/// Distribution of elapsed durations.
31#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
32pub struct Durations {
33    /// Number of tasks measured.
34    pub count: usize,
35    /// Median elapsed hours; absent without tasks.
36    pub median_hours: Option<f64>,
37    /// 85th percentile elapsed hours; absent without tasks.
38    pub p85_hours: Option<f64>,
39    /// Verified tasks not measured, grouped by reason.
40    pub excluded: Vec<Exclusion>,
41}
42
43/// Tasks verified in recent periods, counted by their verification time.
44#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45pub struct Throughput {
46    /// Verified in the seven days up to the clock reading.
47    pub last_7_days: usize,
48    /// Verified in the 28 days up to the clock reading.
49    pub last_28_days: usize,
50    /// Verified per ISO week (UTC), the current week last.
51    pub weeks: Vec<WeekCount>,
52}
53
54/// Tasks verified in one ISO week.
55#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
56pub struct WeekCount {
57    /// ISO week such as `2026-W40`.
58    pub week: String,
59    /// Tasks verified in it.
60    pub verified: usize,
61}
62
63/// One task in progress and how long it has been held.
64#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
65pub struct AgingWork {
66    /// Task key.
67    pub key: Key,
68    /// Lifecycle state.
69    pub status: WorkStatus,
70    /// Current holder.
71    pub owner: Option<ActorId>,
72    /// Elapsed hours since the current holder claimed it or received it by handoff, when known.
73    pub hours_since_claim: Option<f64>,
74    /// Elapsed hours since its recorded start.
75    pub hours_since_start: Option<f64>,
76}
77
78/// How claims ended: each claim, or handoff to a new holder, opens one episode.
79#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
80pub struct Reliability {
81    /// Every episode in the log.
82    pub total: ClaimOutcomes,
83    /// Episodes per holder kind.
84    pub by_holder: Vec<HolderOutcomes>,
85}
86
87/// Episodes of one holder kind.
88#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
89pub struct HolderOutcomes {
90    /// Kind of the holder.
91    pub holder: ActorKind,
92    /// How its episodes ended.
93    #[serde(flatten)]
94    pub outcomes: ClaimOutcomes,
95}
96
97/// Counts of claim episodes by how they ended.
98#[derive(Debug, Clone, Copy, Default, PartialEq, Serialize, Deserialize)]
99pub struct ClaimOutcomes {
100    /// Episodes in total.
101    pub episodes: usize,
102    /// Ended in verification.
103    pub verified: usize,
104    /// Ended in verification after at least one rejection.
105    pub verified_after_rejection: usize,
106    /// The holder gave the claim back.
107    pub released: usize,
108    /// A human or service moved the work to another holder; a transfer, not a failure.
109    pub handed_off: usize,
110    /// Still held.
111    pub open: usize,
112    /// `verified / (verified + released)`: handed-off and open episodes count on neither side;
113    /// absent while no episode was verified or released.
114    pub verified_fraction: Option<f64>,
115}
116
117/// Flow metrics of a plan and the log it results from.
118///
119/// Cycle and lead time skip the work calibration found recorded in bulk (`bulk`): its start and
120/// submission are when someone typed them in, not when the work happened.
121pub(crate) fn flow(
122    plan: &Plan,
123    index: &HistoryIndex<'_>,
124    bulk: &BTreeSet<WorkItemId>,
125    now: DateTime<Utc>,
126) -> FlowReport {
127    let done: Vec<&WorkItem> = plan
128        .work_items
129        .values()
130        .filter(|w| w.is_executable() && w.execution.status.satisfies_dependency())
131        .collect();
132    let mut cycle = Measured::default();
133    let mut lead = Measured::default();
134    let mut verified = Vec::new();
135    for work in done {
136        let Some(at) = work.execution.events.verified_at else {
137            cycle.skip(ExclusionReason::VerificationUnrecorded, work);
138            lead.skip(ExclusionReason::VerificationUnrecorded, work);
139            continue;
140        };
141        verified.push(at);
142        if bulk.contains(&work.id) {
143            cycle.skip(ExclusionReason::BulkRecorded, work);
144            lead.skip(ExclusionReason::BulkRecorded, work);
145            continue;
146        }
147        match work.execution.events.started_at {
148            Some(started) => cycle.hours.push(elapsed_hours(started, at)),
149            None => cycle.skip(ExclusionReason::StartUnrecorded, work),
150        }
151        let claim = index
152            .of(work.id)
153            .iter()
154            .find(|op| matches!(op.command, Command::Claim { .. }));
155        match claim {
156            Some(claim) => lead.hours.push(elapsed_hours(claim.timestamp, at)),
157            None => lead.skip(ExclusionReason::ClaimedBeforeHistory, work),
158        }
159    }
160    FlowReport {
161        cycle_time: cycle.durations(),
162        lead_time: lead.durations(),
163        throughput: throughput(verified.iter().copied(), now),
164        aging: aging(plan, index, now),
165        reliability: reliability(index.work_operations()),
166    }
167}
168
169/// Durations measured so far and the tasks skipped, with why.
170#[derive(Default)]
171struct Measured {
172    hours: Vec<f64>,
173    skipped: Vec<(ExclusionReason, Key)>,
174}
175
176impl Measured {
177    fn skip(&mut self, reason: ExclusionReason, work: &WorkItem) {
178        self.skipped.push((reason, work.key.clone()));
179    }
180
181    fn durations(self) -> Durations {
182        let sorted = ascending(self.hours);
183        let known = !sorted.is_empty();
184        Durations {
185            count: sorted.len(),
186            median_hours: known.then(|| percentile(&sorted, 0.5)),
187            p85_hours: known.then(|| percentile(&sorted, 0.85)),
188            excluded: exclusions(self.skipped),
189        }
190    }
191}
192
193fn throughput(
194    verified: impl Iterator<Item = DateTime<Utc>> + Clone,
195    now: DateTime<Utc>,
196) -> Throughput {
197    let within = |days: i64| {
198        verified
199            .clone()
200            .filter(|at| *at <= now && now - *at < TimeDelta::days(days))
201            .count()
202    };
203    let week_of = |at: DateTime<Utc>| {
204        let week = at.iso_week();
205        (week.year(), week.week())
206    };
207    let weeks = (0..THROUGHPUT_WEEKS)
208        .rev()
209        .filter_map(|back| now.checked_sub_signed(TimeDelta::weeks(back as i64)))
210        .map(|at| {
211            let (year, week) = week_of(at);
212            WeekCount {
213                week: format!("{year}-W{week:02}"),
214                verified: verified
215                    .clone()
216                    .filter(|v| week_of(*v) == (year, week))
217                    .count(),
218            }
219        })
220        .collect();
221    Throughput {
222        last_7_days: within(7),
223        last_28_days: within(28),
224        weeks,
225    }
226}
227
228fn aging(plan: &Plan, index: &HistoryIndex<'_>, now: DateTime<Utc>) -> Vec<AgingWork> {
229    const HELD: [WorkStatus; 4] = [
230        WorkStatus::Claimed,
231        WorkStatus::InProgress,
232        WorkStatus::Blocked,
233        WorkStatus::Submitted,
234    ];
235    let mut aging: Vec<AgingWork> = plan
236        .work_items
237        .values()
238        .filter(|w| w.is_executable() && HELD.contains(&w.execution.status))
239        .map(|w| {
240            let held_since = w.execution.events.claimed_at.or_else(|| {
241                index
242                    .last(w.id, |c| {
243                        matches!(c, Command::Claim { .. } | Command::Handoff { .. })
244                    })
245                    .map(|op| op.timestamp)
246            });
247            AgingWork {
248                key: w.key.clone(),
249                status: w.execution.status,
250                owner: w.execution.owner.clone(),
251                hours_since_claim: held_since.map(|at| elapsed_hours(at, now)),
252                hours_since_start: w
253                    .execution
254                    .events
255                    .started_at
256                    .map(|at| elapsed_hours(at, now)),
257            }
258        })
259        .collect();
260    aging.sort_by(|a, b| {
261        let age = |w: &AgingWork| w.hours_since_claim.unwrap_or(f64::NEG_INFINITY);
262        age(b)
263            .total_cmp(&age(a))
264            .then_with(|| a.key.natural_cmp(&b.key))
265    });
266    aging
267}
268
269/// How an episode ended.
270#[derive(Clone, Copy)]
271enum Ending {
272    Verified { rejected: bool },
273    Released,
274    HandedOff,
275    Open,
276}
277
278fn reliability(operations: &[&Operation]) -> Reliability {
279    let mut open: BTreeMap<WorkItemId, (ActorKind, bool)> = BTreeMap::new();
280    let mut ended: Vec<(ActorKind, Ending)> = Vec::new();
281    for operation in operations {
282        match &operation.command {
283            Command::Claim { work } => {
284                open.insert(*work, (operation.actor.kind, false));
285            }
286            Command::Handoff { work, to, .. } => {
287                if let Some((holder, _)) = open.insert(*work, (to.kind, false)) {
288                    ended.push((holder, Ending::HandedOff));
289                }
290            }
291            Command::Release { work, .. } => {
292                if let Some((holder, _)) = open.remove(work) {
293                    ended.push((holder, Ending::Released));
294                }
295            }
296            Command::Reject { work, .. } => {
297                if let Some(episode) = open.get_mut(work) {
298                    episode.1 = true;
299                }
300            }
301            Command::Verify { work, .. } => {
302                if let Some((holder, rejected)) = open.remove(work) {
303                    ended.push((holder, Ending::Verified { rejected }));
304                }
305            }
306            _ => {}
307        }
308    }
309    ended.extend(open.into_values().map(|(holder, _)| (holder, Ending::Open)));
310    let mut by_holder: BTreeMap<ActorKind, ClaimOutcomes> = BTreeMap::new();
311    let mut total = ClaimOutcomes::default();
312    for (holder, ending) in ended {
313        count(&mut total, ending);
314        count(by_holder.entry(holder).or_default(), ending);
315    }
316    Reliability {
317        total: finish(total),
318        by_holder: by_holder
319            .into_iter()
320            .map(|(holder, outcomes)| HolderOutcomes {
321                holder,
322                outcomes: finish(outcomes),
323            })
324            .collect(),
325    }
326}
327
328fn count(outcomes: &mut ClaimOutcomes, ending: Ending) {
329    outcomes.episodes = outcomes.episodes.wrapping_add(1);
330    let slot = match ending {
331        Ending::Verified { rejected } => {
332            if rejected {
333                outcomes.verified_after_rejection =
334                    outcomes.verified_after_rejection.wrapping_add(1);
335            }
336            &mut outcomes.verified
337        }
338        Ending::Released => &mut outcomes.released,
339        Ending::HandedOff => &mut outcomes.handed_off,
340        Ending::Open => &mut outcomes.open,
341    };
342    *slot = slot.wrapping_add(1);
343}
344
345/// A handoff transfers the work rather than failing it, so its episode counts on neither side of
346/// `verified_fraction`.
347fn finish(mut outcomes: ClaimOutcomes) -> ClaimOutcomes {
348    let closed = outcomes.verified + outcomes.released;
349    outcomes.verified_fraction = (closed > 0).then(|| outcomes.verified as f64 / closed as f64);
350    outcomes
351}
352
353#[cfg(test)]
354mod tests;
dpm_engine/query/flow.rs

dpm_engine/query/
flow.rs
