harness_loop/orchestrator/
run.rs

1//! `Run` — one user goal, executed as a DAG of Jobs, plus its top-level
2//! state machine and budget.
3
4use crate::orchestrator::dag::Dag;
5use crate::orchestrator::job::JobState;
6use serde::{Deserialize, Serialize};
7
8pub type RunId = String;
9
10/// Lifecycle of a whole Run. Mirrors the article's Run state model.
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
12#[serde(rename_all = "snake_case")]
13pub enum RunState {
14    Created,
15    Planning,
16    Executing,
17    /// All runnable Jobs are blocked on something external (a retry backoff,
18    /// or — with a planner — awaiting the next replan).
19    Waiting,
20    Aggregating,
21    Completed,
22    Failed,
23    Cancelled,
24}
25
26impl RunState {
27    pub fn is_terminal(self) -> bool {
28        matches!(
29            self,
30            RunState::Completed | RunState::Failed | RunState::Cancelled
31        )
32    }
33    pub fn label(self) -> &'static str {
34        match self {
35            RunState::Created => "created",
36            RunState::Planning => "planning",
37            RunState::Executing => "executing",
38            RunState::Waiting => "waiting",
39            RunState::Aggregating => "aggregating",
40            RunState::Completed => "completed",
41            RunState::Failed => "failed",
42            RunState::Cancelled => "cancelled",
43        }
44    }
45}
46
47/// A run-level spend ceiling. The orchestrator tallies every Job's token
48/// usage against this and stops the Run if it is exceeded — the cost
49/// governance the async-orchestration literature usually leaves out.
50#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
51pub struct RunBudget {
52    pub max_total_tokens: Option<u64>,
53}
54
55impl RunBudget {
56    pub fn unlimited() -> Self {
57        Self::default()
58    }
59    pub fn max_total_tokens(n: u64) -> Self {
60        Self {
61            max_total_tokens: Some(n),
62        }
63    }
64    /// True if `spent` is over the ceiling.
65    pub fn exceeded(&self, spent: u64) -> bool {
66        matches!(self.max_total_tokens, Some(m) if spent > m)
67    }
68}
69
70/// The persistent record of a Run: its goal, state, budget, and DAG. This is
71/// what a [`crate::orchestrator::RunStore`] saves and reloads for crash recovery.
72#[derive(Debug, Clone, Serialize, Deserialize)]
73pub struct Run {
74    pub id: RunId,
75    pub goal: String,
76    pub state: RunState,
77    pub budget: RunBudget,
78    pub dag: Dag,
79    pub spent_tokens: u64,
80}
81
82impl Run {
83    pub fn new(id: impl Into<RunId>, goal: impl Into<String>, dag: Dag) -> Self {
84        Self {
85            id: id.into(),
86            goal: goal.into(),
87            state: RunState::Created,
88            budget: RunBudget::unlimited(),
89            dag,
90            spent_tokens: 0,
91        }
92    }
93
94    pub fn with_budget(mut self, b: RunBudget) -> Self {
95        self.budget = b;
96        self
97    }
98}
99
100/// A read-only summary returned when a Run finishes. Suitable for delivery
101/// and for aggregating Job results.
102#[derive(Debug, Clone)]
103pub struct RunReport {
104    pub run_id: RunId,
105    pub goal: String,
106    pub state: RunState,
107    pub spent_tokens: u64,
108    /// Every Job's outcome. DAG order not implied.
109    pub jobs: Vec<JobReport>,
110}
111
112/// One Job's outcome, as a reader of the run needs it.
113///
114/// `error` is here because without it a failed run is unreadable: a job shows
115/// as `DeadLettered` with an empty body and the reason — which the job had all
116/// along in `last_error` — never leaves the graph. Measured on a real run, a
117/// `grep` job died on an invalid regex and the report said only "DeadLettered".
118/// `visits` is here for the same reason, one level up: a loop that hit its cap
119/// looks identical to one that failed once.
120#[derive(Debug, Clone, Serialize, Deserialize)]
121pub struct JobReport {
122    pub id: String,
123    pub state: JobState,
124    /// What it produced, if it succeeded.
125    pub result: Option<String>,
126    /// Why it failed, if it did.
127    pub error: Option<String>,
128    /// Attempts of the latest entry, and how many times it was entered.
129    pub attempts: u32,
130    pub visits: u32,
131}
132
133impl RunReport {
134    pub fn succeeded(&self) -> usize {
135        self.jobs
136            .iter()
137            .filter(|j| j.state == JobState::Succeeded)
138            .count()
139    }
140    pub fn dead_lettered(&self) -> usize {
141        self.jobs
142            .iter()
143            .filter(|j| j.state == JobState::DeadLettered)
144            .count()
145    }
146
147    pub fn render(&self) -> String {
148        let mut s = format!(
149            "Run `{}` [{}] — {} goal: {}\n",
150            self.run_id,
151            self.state.label(),
152            format_args!("{} jobs, {} tokens;", self.jobs.len(), self.spent_tokens),
153            self.goal,
154        );
155        for j in &self.jobs {
156            s.push_str(&format!("  - {}: {}", j.id, j.state.label()));
157            if j.visits > 1 {
158                // A loop that hit its cap otherwise reads like a single failure.
159                s.push_str(&format!(" (entered {}×)", j.visits));
160            }
161            // Char-safe truncation (text may contain multi-byte chars / emoji).
162            let brief = |t: &str| t.trim().chars().take(80).collect::<String>();
163            if let Some(t) = &j.result {
164                s.push_str(&format!(" — {}", brief(t)));
165            }
166            // The reason comes last and always: a failed job with an empty line
167            // is the report saying nothing at exactly the moment it matters.
168            if let Some(e) = &j.error {
169                s.push_str(&format!(" — failed: {}", brief(e)));
170            }
171            s.push('\n');
172        }
173        s
174    }
175}
176
177#[cfg(test)]
178mod tests {
179    use super::*;
180    use crate::orchestrator::job::JobState;
181
182    #[test]
183    fn render_truncates_on_char_boundary_with_multibyte_text() {
184        // A job result whose 80th byte lands inside a multi-byte char must not
185        // panic when rendered (regression: byte-slicing `&t[..80]`).
186        let long = format!("🔄 {}", "更".repeat(100));
187        let report = RunReport {
188            run_id: "r".into(),
189            goal: "g".into(),
190            state: RunState::Completed,
191            spent_tokens: 0,
192            jobs: vec![JobReport {
193                id: "j".into(),
194                state: JobState::Succeeded,
195                result: Some(long),
196                error: None,
197                attempts: 1,
198                visits: 1,
199            }],
200        };
201        let out = report.render(); // must not panic
202        assert!(out.contains("j: succeeded"));
203    }
204}
harness_loop/orchestrator/run.rs

harness_loop/orchestrator/
run.rs
