klieo_workflow_api/
sweep.rs

1//! Startup orphan sweep.
2//!
3//! After a restart, a run left `Running` OR `Pending` in the durable store is
4//! orphaned — its executor (or its about-to-spawn submit) died with the
5//! process and will never write a terminal status. The host calls
6//! [`sweep_orphaned_runs`] once at startup (before serving traffic) to mark
7//! each such run `Aborted`, appending the terminal audit entry to the chain
8//! FIRST. Reclaiming `Pending` is safe precisely because the sweep runs
9//! before any live submit — a `Pending` at startup is genuinely stranded.
10//! With an ephemeral (non-enumerable / empty) store this is a no-op.
11//!
12//! Per-orphan failures are isolated: one orphan's transient failure (sqlite
13//! lock, append error) is logged with its run id and the sweep continues, so
14//! one bad orphan never blocks the rest. The returned [`SweepSummary`]
15//! reports the swept count and any run ids that could not be swept.
16
17use crate::provenance::{RunProvenance, STATUS_ABORTED};
18use crate::run_store::{RunStatus, RunStore, StoreError};
19use klieo_provenance::{ProvenanceError, ProvenanceRepository};
20use std::sync::Arc;
21
22/// Reason recorded for a run orphaned by a process restart.
23const RESTART_REASON: &str = "aborted: executor lost to a process restart";
24
25/// Outcome of a startup sweep: how many orphans were reclaimed, and the run
26/// ids of any that could not be (already logged, left for the next sweep).
27#[derive(Debug, Default, PartialEq, Eq)]
28#[non_exhaustive]
29pub struct SweepSummary {
30    /// Orphaned runs successfully marked `Aborted`.
31    pub swept: usize,
32    /// Run ids whose reclaim failed this pass.
33    pub failed: Vec<String>,
34}
35
36/// Failure that aborts the sweep before it can process any orphan (e.g. the
37/// run store could not be enumerated). Per-orphan failures do NOT surface
38/// here — they are isolated into [`SweepSummary::failed`].
39#[non_exhaustive]
40#[derive(Debug, thiserror::Error)]
41pub enum SweepError {
42    /// Reading or updating the run store failed.
43    #[error("run store error during sweep: {0}")]
44    Store(#[from] StoreError),
45    /// Appending a terminal audit entry failed.
46    #[error("provenance append error during sweep: {0}")]
47    Provenance(#[from] ProvenanceError),
48}
49
50/// Reclaim every orphaned (`Running` or `Pending`) run, chain-first. Isolates
51/// per-orphan failures; only an inability to enumerate the store returns
52/// `Err`. Call once at startup, before serving traffic.
53pub async fn sweep_orphaned_runs(
54    run_store: &RunStore,
55    provenance: &Arc<dyn ProvenanceRepository>,
56) -> Result<SweepSummary, SweepError> {
57    let mut summary = SweepSummary::default();
58    for run in run_store.list_active().await? {
59        if !is_orphaned(run.status) {
60            continue;
61        }
62        match abort_orphan(
63            run_store,
64            provenance,
65            &run.run_id,
66            &run.workflow_id,
67            &run.author,
68        )
69        .await
70        {
71            Ok(()) => summary.swept += 1,
72            Err(e) => {
73                tracing::error!(run_id = run.run_id, error = %e, "failed to reclaim orphaned run; continuing");
74                summary.failed.push(run.run_id);
75            }
76        }
77    }
78    Ok(summary)
79}
80
81/// A run left mid-flight by a dead process (before serving traffic resumes).
82fn is_orphaned(status: RunStatus) -> bool {
83    matches!(status, RunStatus::Running | RunStatus::Pending)
84}
85
86async fn abort_orphan(
87    run_store: &RunStore,
88    provenance: &Arc<dyn ProvenanceRepository>,
89    run_id: &str,
90    workflow_id: &str,
91    author: &str,
92) -> Result<(), SweepError> {
93    // The orphan-abort marker carries no payload hash — the run's own
94    // `AgentRunStarted` entry already bound it to the canonical definition;
95    // this entry only records that the run ended by restart.
96    let binding = RunProvenance {
97        repo: Arc::clone(provenance),
98        run_id: run_id.to_string(),
99        author: author.to_string(),
100        workflow_id: workflow_id.to_string(),
101        content_hash: String::new(),
102    };
103    binding
104        .record_terminal(STATUS_ABORTED, Some(RESTART_REASON.to_string()))
105        .await?;
106    run_store
107        .set_aborted(run_id, RESTART_REASON.to_string())
108        .await?;
109    Ok(())
110}
klieo_workflow_api/sweep.rs

klieo_workflow_api/
sweep.rs
