ito_core/orchestrate/
state.rs

1//! Orchestrator run state persistence.
2//!
3//! Persists run state under `.ito/.state/orchestrate/runs/<run-id>/`.
4
5use crate::errors::{CoreError, CoreResult};
6use crate::orchestrate::plan::{PlannedGate, RunPlan};
7use crate::orchestrate::types::{FailurePolicy, GateOutcome, OrchestrateRunStatus};
8use chrono::{DateTime, Utc};
9use serde::{Deserialize, Serialize};
10use std::fs::OpenOptions;
11use std::io::Write;
12use std::path::{Path, PathBuf};
13
14#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
15/// Persisted `run.json` record for an orchestration run.
16pub struct OrchestrateRun {
17    /// Run id (`YYYYMMDD-HHMMSS-<short-uuid>`).
18    pub run_id: String,
19    /// RFC3339 timestamp for run start.
20    pub started_at: String,
21    /// RFC3339 timestamp for run completion.
22    pub finished_at: Option<String>,
23    /// Run status.
24    pub status: OrchestrateRunStatus,
25    /// Preset name used for this run.
26    pub preset: String,
27    /// Maximum number of concurrent change pipelines.
28    pub max_parallel: usize,
29    /// Failure policy applied to this run.
30    pub failure_policy: FailurePolicy,
31}
32
33#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
34/// One terminal gate result recorded for a change.
35pub struct OrchestrateGateRecord {
36    /// Gate identifier.
37    pub gate: String,
38    /// Gate outcome.
39    pub outcome: GateOutcome,
40    /// RFC3339 timestamp for gate completion.
41    pub finished_at: String,
42    #[serde(default)]
43    /// Optional error payload (for failed gates).
44    pub error: Option<String>,
45}
46
47#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
48/// Persisted per-change state (`changes/<change-id>.json`).
49pub struct OrchestrateChangeState {
50    /// Canonical change id.
51    pub change_id: String,
52    #[serde(default)]
53    /// Terminal gate outcomes observed so far.
54    pub gates: Vec<OrchestrateGateRecord>,
55    /// RFC3339 timestamp for last update.
56    pub updated_at: String,
57}
58
59#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
60/// One record appended to `events.jsonl`.
61pub struct OrchestrateEvent {
62    /// RFC3339 timestamp.
63    pub ts: String,
64    /// Event payload.
65    pub kind: OrchestrateEventKind,
66}
67
68#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
69#[serde(tag = "event", rename_all = "kebab-case")]
70/// Event kinds written to `events.jsonl`.
71pub enum OrchestrateEventKind {
72    /// Run started.
73    RunStart {
74        /// Run id.
75        run_id: String,
76        /// Preset name.
77        preset: String,
78        /// Maximum parallel pipelines.
79        max_parallel: usize,
80        /// Failure policy.
81        failure_policy: FailurePolicy,
82    },
83    /// Run completed.
84    RunComplete {
85        /// Run id.
86        run_id: String,
87        /// Terminal status.
88        status: OrchestrateRunStatus,
89    },
90    /// Gate started for a change.
91    GateStart {
92        /// Run id.
93        run_id: String,
94        /// Change id.
95        change_id: String,
96        /// Gate name.
97        gate: String,
98    },
99    /// Gate passed for a change.
100    GatePass {
101        /// Run id.
102        run_id: String,
103        /// Change id.
104        change_id: String,
105        /// Gate name.
106        gate: String,
107    },
108    /// Gate failed for a change.
109    GateFail {
110        /// Run id.
111        run_id: String,
112        /// Change id.
113        change_id: String,
114        /// Gate name.
115        gate: String,
116        /// Error payload.
117        error: String,
118    },
119    /// Gate skipped for a change.
120    GateSkip {
121        /// Run id.
122        run_id: String,
123        /// Change id.
124        change_id: String,
125        /// Gate name.
126        gate: String,
127    },
128    /// Worker dispatched.
129    WorkerDispatch {
130        /// Run id.
131        run_id: String,
132        /// Change id.
133        change_id: String,
134        /// Gate name.
135        gate: String,
136        /// Suggested role label (preset-driven).
137        role: String,
138    },
139    /// Worker completed.
140    WorkerComplete {
141        /// Run id.
142        run_id: String,
143        /// Change id.
144        change_id: String,
145        /// Gate name.
146        gate: String,
147        /// Worker-reported outcome.
148        outcome: GateOutcome,
149    },
150    /// Remediation dispatched after a gate failure.
151    RemediationDispatch {
152        /// Run id.
153        run_id: String,
154        /// Change id.
155        change_id: String,
156        /// Gate name that failed.
157        failed_gate: String,
158    },
159}
160
161/// Initialize the on-disk run state directory and write `run.json` + `plan.json`.
162pub fn init_orchestrate_run_state(
163    ito_path: &Path,
164    run: &OrchestrateRun,
165    plan: &RunPlan,
166) -> CoreResult<()> {
167    let root = run_root(ito_path, &run.run_id);
168    std::fs::create_dir_all(root.join("changes"))
169        .map_err(|e| CoreError::io("create orchestrate run state directory", e))?;
170
171    write_json_pretty(&root.join("run.json"), run)
172        .map_err(|e| CoreError::io("write orchestrate run.json", e))?;
173    write_json_pretty(&root.join("plan.json"), plan)
174        .map_err(|e| CoreError::io("write orchestrate plan.json", e))?;
175
176    // Ensure the event log exists.
177    let events = root.join("events.jsonl");
178    if !events.exists() {
179        std::fs::write(&events, "")
180            .map_err(|e| CoreError::io("create orchestrate events.jsonl", e))?;
181    }
182
183    Ok(())
184}
185
186/// Load the persisted `run.json` record for a previously started orchestrate run.
187///
188/// Use this when resuming a run or when an adapter needs to report the run's
189/// current top-level status without re-reading the full event log.
190pub fn load_orchestrate_run(ito_path: &Path, run_id: &str) -> CoreResult<OrchestrateRun> {
191    let path = run_root(ito_path, run_id).join("run.json");
192    read_json_file(&path, "orchestrate run.json")
193}
194
195/// Load the resolved `plan.json` for a previously started orchestrate run.
196///
197/// This is the canonical persisted plan used for resuming or inspecting a run,
198/// so callers do not need to rebuild planning inputs from change metadata.
199pub fn load_orchestrate_plan(ito_path: &Path, run_id: &str) -> CoreResult<RunPlan> {
200    let path = run_root(ito_path, run_id).join("plan.json");
201    read_json_file(&path, "orchestrate plan.json")
202}
203
204/// Append an event to `events.jsonl`.
205pub fn append_orchestrate_event(
206    ito_path: &Path,
207    run_id: &str,
208    event: &OrchestrateEvent,
209) -> CoreResult<()> {
210    let path = run_root(ito_path, run_id).join("events.jsonl");
211    if let Some(parent) = path.parent() {
212        std::fs::create_dir_all(parent)
213            .map_err(|e| CoreError::io("create orchestrate events directory", e))?;
214    }
215
216    let json = serde_json::to_string(event).map_err(|e| CoreError::Serde {
217        context: "serialize orchestrate event".to_string(),
218        message: e.to_string(),
219    })?;
220    let mut f = OpenOptions::new()
221        .create(true)
222        .append(true)
223        .open(&path)
224        .map_err(|e| {
225            CoreError::io(
226                format!("open orchestrate events.jsonl: {}", path.display()),
227                e,
228            )
229        })?;
230    writeln!(f, "{json}").map_err(|e| CoreError::io("append orchestrate event", e))?;
231    f.flush()
232        .map_err(|e| CoreError::io("flush orchestrate event log", e))?;
233    Ok(())
234}
235
236/// Load the per-change state file if present.
237pub fn load_orchestrate_change_state(
238    ito_path: &Path,
239    run_id: &str,
240    change_id: &str,
241) -> CoreResult<Option<OrchestrateChangeState>> {
242    let path = change_state_path(ito_path, run_id, change_id);
243    if !path.exists() {
244        return Ok(None);
245    }
246    let state = read_json_file(&path, "orchestrate change state")?;
247    Ok(Some(state))
248}
249
250/// Write the per-change state file.
251pub fn write_orchestrate_change_state(
252    ito_path: &Path,
253    run_id: &str,
254    state: &OrchestrateChangeState,
255) -> CoreResult<()> {
256    let path = change_state_path(ito_path, run_id, &state.change_id);
257    if let Some(parent) = path.parent() {
258        std::fs::create_dir_all(parent)
259            .map_err(|e| CoreError::io("create orchestrate change state directory", e))?;
260    }
261    write_json_pretty(&path, state).map_err(|e| {
262        CoreError::io(
263            format!("write orchestrate change state: {}", path.display()),
264            e,
265        )
266    })?;
267    Ok(())
268}
269
270/// Generate a sortable run id in the form `YYYYMMDD-HHMMSS-<short-uuid>`.
271pub fn generate_orchestrate_run_id(now: DateTime<Utc>) -> String {
272    let ts = now.format("%Y%m%d-%H%M%S").to_string();
273    let uuid = uuid::Uuid::new_v4().simple().to_string();
274    let short = uuid.chars().take(8).collect::<String>();
275    format!("{ts}-{short}")
276}
277
278/// Compute remaining gates for a change when resuming an interrupted run.
279///
280/// Gates that have terminal `pass` or `skip` outcomes are skipped.
281/// Execution resumes from the first gate that is missing, failed, or otherwise incomplete.
282pub fn remaining_gates_for_change(
283    planned: &[PlannedGate],
284    state: Option<&OrchestrateChangeState>,
285) -> Vec<PlannedGate> {
286    let Some(state) = state else {
287        return planned.to_vec();
288    };
289
290    let mut outcomes: std::collections::BTreeMap<&str, GateOutcome> =
291        std::collections::BTreeMap::new();
292    for g in &state.gates {
293        outcomes.insert(&g.gate, g.outcome);
294    }
295
296    let mut start = 0;
297    for (i, g) in planned.iter().enumerate() {
298        match outcomes.get(g.name.as_str()) {
299            Some(GateOutcome::Pass) | Some(GateOutcome::Skip) => {
300                start = i + 1;
301                continue;
302            }
303            Some(GateOutcome::Fail) | None => {
304                start = i;
305                break;
306            }
307        }
308    }
309
310    planned[start..].to_vec()
311}
312
313fn run_root(ito_path: &Path, run_id: &str) -> PathBuf {
314    ito_path
315        .join(".state")
316        .join("orchestrate")
317        .join("runs")
318        .join(run_id)
319}
320
321fn change_state_path(ito_path: &Path, run_id: &str, change_id: &str) -> PathBuf {
322    run_root(ito_path, run_id)
323        .join("changes")
324        .join(format!("{change_id}.json"))
325}
326
327fn write_json_pretty<T: Serialize>(path: &Path, value: &T) -> std::io::Result<()> {
328    let json = serde_json::to_string_pretty(value)
329        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
330    std::fs::write(path, json)
331}
332
333fn read_json_file<T: for<'de> Deserialize<'de>>(path: &Path, label: &str) -> CoreResult<T> {
334    let contents = std::fs::read_to_string(path)
335        .map_err(|e| CoreError::io(format!("read {label}: {}", path.display()), e))?;
336    serde_json::from_str(&contents).map_err(|e| CoreError::Serde {
337        context: format!("parse {label}"),
338        message: e.to_string(),
339    })
340}
ito_core/orchestrate/state.rs

ito_core/orchestrate/
state.rs
