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

ito_core/orchestrate/
state.rs
