1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
use crate::{
ActorId, Dependency, DependencyId, EventTime, Plan, Release, StartBasis, Timeline, WorkItem,
WorkItemId,
};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
/// One submission of a task, identified by its 1-based ordinal within that task.
///
/// The ordinal is deterministic, so every reload and replica of the same history agrees on it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SubmissionAttempt {
/// Position of this attempt in the task's submission history, starting at 1.
pub number: u32,
/// Time recorded by the submit command.
pub submitted_at: DateTime<Utc>,
/// Review result; only the latest attempt may still be pending.
pub outcome: AttemptOutcome,
}
/// Independent review result of one submission attempt.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum AttemptOutcome {
/// Awaiting independent review.
Pending,
/// Returned for rework; the attempt stays in history and never becomes valid again.
Rejected {
/// Independent reviewer.
actor: ActorId,
/// Time recorded by the reject command.
at: DateTime<Utc>,
/// Concrete unmet acceptance or evidence requirement.
reason: String,
},
/// Accepted by an independent verifier.
Verified {
/// Independent verifier.
actor: ActorId,
/// Time recorded by the verify command.
at: DateTime<Utc>,
},
}
impl AttemptOutcome {
/// Whether the attempt still awaited review at `now`: a start recorded before a later review
/// relied on an unreviewed result, whatever the review has since decided.
#[must_use]
pub fn pending_at(&self, now: DateTime<Utc>) -> bool {
match self {
Self::Pending => true,
Self::Rejected { at, .. } | Self::Verified { at, .. } => *at > now,
}
}
}
/// Immutable reference to the predecessor attempt a successor's execution relies on.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DependencyBasis {
/// Provisional edge whose start gate the attempt released.
pub dependency: DependencyId,
/// Predecessor that made the attempt.
pub predecessor: WorkItemId,
/// Ordinal of the relied-on predecessor attempt.
pub attempt: u32,
/// Time recorded by the operation that captured this basis.
pub recorded_at: DateTime<Utc>,
/// Operation that captured this basis.
pub source: BasisSource,
}
/// Operation that recorded a dependency basis.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum BasisSource {
/// The successor's owner started on the attempt through the provisional start gate.
Start,
/// An independent human or service reviewed the successor against a newer attempt.
Revalidation {
/// Accountable reviewer.
actor: ActorId,
/// Why the successor's work remains valid on the new attempt.
reason: String,
},
}
impl WorkItem {
/// Attempt with the given ordinal.
#[must_use]
pub fn attempt(&self, number: u32) -> Option<&SubmissionAttempt> {
let index = usize::try_from(number).ok()?.checked_sub(1)?;
self.execution.attempts.get(index)
}
/// Latest attempt unless it was rejected: the pending or verified result a basis may rely on.
#[must_use]
pub fn current_attempt(&self) -> Option<&SubmissionAttempt> {
self.execution
.attempts
.last()
.filter(|a| !matches!(a.outcome, AttemptOutcome::Rejected { .. }))
}
/// Latest basis recorded for each edge, in first-recorded edge order.
#[must_use]
pub fn effective_basis(&self) -> Vec<&DependencyBasis> {
let mut latest: Vec<&DependencyBasis> = Vec::new();
for entry in &self.execution.basis {
match latest.iter_mut().find(|b| b.dependency == entry.dependency) {
Some(slot) => *slot = entry,
None => latest.push(entry),
}
}
latest
}
}
/// Derived validity of a recorded basis; never stored.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum BasisState {
/// The relied-on attempt still awaits review.
Pending,
/// The relied-on attempt was verified.
Verified,
/// The relied-on attempt was rejected; only an explicit revalidation records a new basis.
Invalidated {
/// Reviewer who rejected the attempt.
rejected_by: ActorId,
/// Time of the rejection.
rejected_at: DateTime<Utc>,
/// Rejection reason.
reason: String,
},
}
/// A successor's effective basis on one provisional edge, with its derived state.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct BasisStatus {
/// Work whose execution relies on the attempt.
pub successor: WorkItemId,
/// Human-readable successor key.
pub successor_key: String,
/// Provisional edge carrying the basis.
pub dependency: DependencyId,
/// Work that made the attempt.
pub predecessor: WorkItemId,
/// Human-readable predecessor key.
pub predecessor_key: String,
/// The immutable recorded reference.
pub basis: DependencyBasis,
/// Derived from the predecessor's attempt record.
pub state: BasisState,
/// Predecessor's pending or verified attempt a revalidation may name, if any.
pub current_attempt: Option<u32>,
/// Whether the edge gates execution; a waived edge's basis is context only.
pub enforced: bool,
}
impl BasisStatus {
/// Whether an enforced basis rests on a rejected attempt and so gates finishing the successor.
#[must_use]
pub fn gates(&self) -> bool {
self.enforced && matches!(self.state, BasisState::Invalidated { .. })
}
}
/// Effective basis of a work item on each provisional edge, derived from attempt records.
#[must_use]
pub fn basis_status(plan: &Plan, work: &WorkItem) -> Vec<BasisStatus> {
work.effective_basis()
.into_iter()
.filter_map(|basis| status_of(plan, work, basis))
.collect()
}
/// Effective bases of other work that rely on attempts of this predecessor.
#[must_use]
pub fn basis_dependents(plan: &Plan, predecessor: WorkItemId) -> Vec<BasisStatus> {
plan.work_items
.values()
.flat_map(|work| basis_status(plan, work))
.filter(|status| status.predecessor == predecessor)
.collect()
}
fn status_of(plan: &Plan, work: &WorkItem, basis: &DependencyBasis) -> Option<BasisStatus> {
let predecessor = plan.work_items.get(&basis.predecessor)?;
let state = match &predecessor.attempt(basis.attempt)?.outcome {
AttemptOutcome::Pending => BasisState::Pending,
AttemptOutcome::Verified { .. } => BasisState::Verified,
AttemptOutcome::Rejected { actor, at, reason } => BasisState::Invalidated {
rejected_by: actor.clone(),
rejected_at: *at,
reason: reason.clone(),
},
};
Some(BasisStatus {
successor: work.id,
successor_key: work.key.to_string(),
dependency: basis.dependency,
predecessor: predecessor.id,
predecessor_key: predecessor.key.to_string(),
basis: basis.clone(),
state,
current_attempt: predecessor.current_attempt().map(|a| a.number),
enforced: plan
.find_dependency(basis.dependency)
.is_some_and(|edge| !edge.is_waived()),
})
}
/// Release of an edge for the successor's start, and the predecessor attempt it was measured from.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct StartRelease {
/// Release state at the timeline's clock reading.
pub release: Release,
/// Predecessor attempt whose submission time a provisional release is measured from.
pub attempt: Option<u32>,
/// Whether that attempt awaited review at the clock reading, so a start then relied on an
/// unverified result.
pub provisional: bool,
}
impl Timeline {
/// Evaluate an edge for the successor's claim or start.
///
/// A provisional edge releases on the predecessor's current attempt's submission plus positive
/// lag until its verified finish releases it, so verifying that attempt never pushes an
/// elapsing start gate later. Every other evaluation, including the successor's verification
/// and milestone reach, uses [`Timeline::edge`], so a submission never counts as a finish
/// there. A legacy submission without an attempt record has nothing a basis could reference,
/// so its edge waits for verification.
#[must_use]
pub fn start_edge(&self, plan: &Plan, edge: &Dependency) -> StartRelease {
let release = self.edge(plan, edge);
// Only an edge still waiting on the predecessor's own event may use its attempt; a
// not-selected branch stays governed by the choice, never by a submission.
let current = (edge.start_basis == StartBasis::Provisional
&& matches!(release, Release::AwaitingEvent | Release::Elapsing { .. })
&& !self.applicability(edge.predecessor).is_not_selected())
.then(|| plan.work_items.get(&edge.predecessor))
.flatten()
.and_then(WorkItem::current_attempt);
match current {
Some(attempt) => StartRelease {
release: Release::evaluate_with(
Some(EventTime::Recorded(attempt.submitted_at)),
edge.lag_hours,
self.now(),
|at, hours| plan.lag_end(edge, at, hours),
),
attempt: Some(attempt.number),
provisional: attempt.outcome.pending_at(self.now()),
},
None => StartRelease {
release,
attempt: None,
provisional: false,
},
}
}
/// The earliest instant after this timeline's clock reading at which a pending gate opens with
/// no change to the plan: an elapsing dependency lag (including a provisional start lag) or a
/// decision gate released with elapsing lag.
///
/// Time passing is the only change, so an adapter that evaluated at [`Timeline::now`] knows
/// when to evaluate again. The instant is the exact opening boundary: a reading strictly
/// before it still sees the gate elapsing, a reading at it sees it released. A fact recorded
/// with a future time is not reported as elapsing, so it is not included.
#[must_use]
pub fn next_release(&self, plan: &Plan) -> Option<DateTime<Utc>> {
let edges = plan
.dependencies
.iter()
.flat_map(|edge| [self.edge(plan, edge), self.start_edge(plan, edge).release]);
let decisions = plan.work_items.keys().flat_map(|id| {
self.decisions(plan, *id)
.into_iter()
.map(|(_, release)| release)
});
edges
.chain(decisions)
.filter_map(|release| match release {
Release::Elapsing { opens_at, .. } => Some(opens_at),
_ => None,
})
.min()
}
}
#[cfg(test)]
mod tests;