harness_loop/orchestrator/
job.rs

1//! `Job` — one node in a Run's DAG, and its state machine.
2//!
3//! A Job is a single unit of work (typically one sub-agent invocation) with
4//! dependencies on other Jobs, a retry policy, and a place to hold its
5//! result. The two-level state model (Run + Job) is what lets an
6//! orchestrator recover from a crash: a Job stuck in `Running` when the
7//! process died is reset to `Pending` on resume.
8
9use serde::{Deserialize, Serialize};
10use std::time::Duration;
11
12pub type JobId = String;
13
14/// Lifecycle of a single Job. Mirrors the states a durable task system
15/// needs to recover and retry.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
17#[serde(rename_all = "snake_case")]
18pub enum JobState {
19    /// Created; waiting for its dependencies to succeed.
20    Pending,
21    /// Dependencies satisfied; eligible to run, awaiting a concurrency slot.
22    Queued,
23    /// Currently executing.
24    Running,
25    /// Finished successfully; `result` is populated.
26    Succeeded,
27    /// This attempt failed; may transition to `Retrying` or `DeadLettered`.
28    Failed,
29    /// Waiting out a backoff delay before the next attempt.
30    Retrying,
31    /// Exhausted its retries; needs human attention. Blocks dependents.
32    DeadLettered,
33    /// Cancelled (e.g. the Run was cancelled or hit its budget).
34    Cancelled,
35}
36
37impl JobState {
38    pub fn is_terminal(self) -> bool {
39        matches!(
40            self,
41            JobState::Succeeded | JobState::DeadLettered | JobState::Cancelled
42        )
43    }
44    pub fn label(self) -> &'static str {
45        match self {
46            JobState::Pending => "pending",
47            JobState::Queued => "queued",
48            JobState::Running => "running",
49            JobState::Succeeded => "succeeded",
50            JobState::Failed => "failed",
51            JobState::Retrying => "retrying",
52            JobState::DeadLettered => "dead-lettered",
53            JobState::Cancelled => "cancelled",
54        }
55    }
56}
57
58/// What a Job produced. Kept in memory — this is a single-machine
59/// orchestrator, so there's no Claim-Check indirection (that's a
60/// distributed concern, explicitly out of scope).
61#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
62pub struct JobResult {
63    pub text: String,
64    pub input_tokens: u64,
65    pub output_tokens: u64,
66}
67
68impl JobResult {
69    pub fn new(text: impl Into<String>) -> Self {
70        Self {
71            text: text.into(),
72            input_tokens: 0,
73            output_tokens: 0,
74        }
75    }
76    pub fn with_tokens(mut self, input: u64, output: u64) -> Self {
77        self.input_tokens = input;
78        self.output_tokens = output;
79        self
80    }
81    pub fn total_tokens(&self) -> u64 {
82        self.input_tokens + self.output_tokens
83    }
84}
85
86/// Backoff strategy between retry attempts.
87#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
88#[serde(rename_all = "snake_case")]
89pub enum Backoff {
90    /// Retry immediately.
91    None,
92    /// Wait a fixed duration before each retry.
93    Fixed(Duration),
94    /// `base * factor^(attempt-1)`, capped at `max`.
95    Exponential {
96        base: Duration,
97        factor: u32,
98        max: Duration,
99    },
100}
101
102impl Backoff {
103    /// Delay before the given attempt number (1-based: attempt 1 is the
104    /// first retry, i.e. the 2nd run overall).
105    pub fn delay(self, attempt: u32) -> Duration {
106        match self {
107            Backoff::None => Duration::ZERO,
108            Backoff::Fixed(d) => d,
109            Backoff::Exponential { base, factor, max } => {
110                let mult = factor.saturating_pow(attempt.saturating_sub(1));
111                base.saturating_mul(mult).min(max)
112            }
113        }
114    }
115}
116
117/// How many times a Job may run, and how long to wait between attempts.
118#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
119pub struct RetryPolicy {
120    /// Total attempts allowed, including the first. `1` means no retry.
121    pub max_attempts: u32,
122    pub backoff: Backoff,
123}
124
125impl Default for RetryPolicy {
126    fn default() -> Self {
127        Self {
128            max_attempts: 1,
129            backoff: Backoff::None,
130        }
131    }
132}
133
134impl RetryPolicy {
135    /// No retries.
136    pub fn once() -> Self {
137        Self::default()
138    }
139    /// `attempts` total tries with the given backoff.
140    pub fn new(max_attempts: u32, backoff: Backoff) -> Self {
141        Self {
142            max_attempts: max_attempts.max(1),
143            backoff,
144        }
145    }
146}
147
148/// One node in the Run DAG.
149#[derive(Debug, Clone, Serialize, Deserialize)]
150pub struct Job {
151    pub id: JobId,
152    /// Ids of Jobs that must `Succeed` before this one can run.
153    pub deps: Vec<JobId>,
154    /// The task description handed to the runner (e.g. a sub-agent prompt).
155    pub prompt: String,
156    pub retry: RetryPolicy,
157    pub state: JobState,
158    pub result: Option<JobResult>,
159    /// How many times this Job has been run so far.
160    pub attempts: u32,
161    /// How many times this Job has been *entered* — one per run, plus one for
162    /// every time a [`Next::Goto`](crate::orchestrator::Next) sent the graph back through it.
163    /// Distinct from `attempts`, which counts retries of a single entry: a job
164    /// that fails twice and is revisited once has 2 attempts and 2 visits, and
165    /// only the second bounds the loop.
166    #[serde(default)]
167    pub visits: u32,
168    /// Error from the most recent failed attempt, if any.
169    pub last_error: Option<String>,
170    /// What this job produced on the previous lap, and why it was sent back.
171    /// Set only by a [`Next::Goto`](crate::orchestrator::Next); the runner puts both in front
172    /// of the job so it can improve on its own answer instead of reproducing it.
173    #[serde(default)]
174    pub prior_attempt: Option<String>,
175    #[serde(default)]
176    pub feedback: Option<String>,
177}
178
179impl Job {
180    pub fn new(id: impl Into<JobId>, prompt: impl Into<String>) -> Self {
181        Self {
182            id: id.into(),
183            deps: Vec::new(),
184            prompt: prompt.into(),
185            retry: RetryPolicy::default(),
186            state: JobState::Pending,
187            result: None,
188            attempts: 0,
189            visits: 0,
190            last_error: None,
191            prior_attempt: None,
192            feedback: None,
193        }
194    }
195
196    pub fn with_deps<I, S>(mut self, deps: I) -> Self
197    where
198        I: IntoIterator<Item = S>,
199        S: Into<JobId>,
200    {
201        self.deps = deps.into_iter().map(Into::into).collect();
202        self
203    }
204
205    pub fn with_retry(mut self, retry: RetryPolicy) -> Self {
206        self.retry = retry;
207        self
208    }
209}
210
211#[cfg(test)]
212mod tests {
213    use super::*;
214
215    #[test]
216    fn exponential_backoff_grows_and_caps() {
217        let b = Backoff::Exponential {
218            base: Duration::from_secs(1),
219            factor: 2,
220            max: Duration::from_secs(10),
221        };
222        assert_eq!(b.delay(1), Duration::from_secs(1)); // base * 2^0
223        assert_eq!(b.delay(2), Duration::from_secs(2)); // base * 2^1
224        assert_eq!(b.delay(3), Duration::from_secs(4)); // base * 2^2
225        assert_eq!(b.delay(5), Duration::from_secs(10)); // 16 capped to 10
226    }
227
228    #[test]
229    fn retry_policy_min_one_attempt() {
230        assert_eq!(RetryPolicy::new(0, Backoff::None).max_attempts, 1);
231    }
232
233    #[test]
234    fn terminal_states() {
235        assert!(JobState::Succeeded.is_terminal());
236        assert!(JobState::DeadLettered.is_terminal());
237        assert!(!JobState::Pending.is_terminal());
238        assert!(!JobState::Running.is_terminal());
239    }
240}
harness_loop/orchestrator/job.rs

harness_loop/orchestrator/
job.rs
