klieo-workflow-api 3.8.1

//! Startup orphan sweep.
//!
//! After a restart, a run left `Running` OR `Pending` in the durable store is
//! orphaned — its executor (or its about-to-spawn submit) died with the
//! process and will never write a terminal status. The host calls
//! [`sweep_orphaned_runs`] once at startup (before serving traffic) to mark
//! each such run `Aborted`, appending the terminal audit entry to the chain
//! FIRST. Reclaiming `Pending` is safe precisely because the sweep runs
//! before any live submit — a `Pending` at startup is genuinely stranded.
//! With an ephemeral (non-enumerable / empty) store this is a no-op.
//!
//! Per-orphan failures are isolated: one orphan's transient failure (sqlite
//! lock, append error) is logged with its run id and the sweep continues, so
//! one bad orphan never blocks the rest. The returned [`SweepSummary`]
//! reports the swept count and any run ids that could not be swept.

use crate::provenance::{RunProvenance, STATUS_ABORTED};
use crate::run_store::{RunStatus, RunStore, StoreError};
use klieo_provenance::{ProvenanceError, ProvenanceRepository};
use std::sync::Arc;

/// Reason recorded for a run orphaned by a process restart.
const RESTART_REASON: &str = "aborted: executor lost to a process restart";

/// Outcome of a startup sweep: how many orphans were reclaimed, and the run
/// ids of any that could not be (already logged, left for the next sweep).
#[derive(Debug, Default, PartialEq, Eq)]
#[non_exhaustive]
pub struct SweepSummary {
    /// Orphaned runs successfully marked `Aborted`.
    pub swept: usize,
    /// Run ids whose reclaim failed this pass.
    pub failed: Vec<String>,
}

/// Failure that aborts the sweep before it can process any orphan (e.g. the
/// run store could not be enumerated). Per-orphan failures do NOT surface
/// here — they are isolated into [`SweepSummary::failed`].
#[non_exhaustive]
#[derive(Debug, thiserror::Error)]
pub enum SweepError {
    /// Reading or updating the run store failed.
    #[error("run store error during sweep: {0}")]
    Store(#[from] StoreError),
    /// Appending a terminal audit entry failed.
    #[error("provenance append error during sweep: {0}")]
    Provenance(#[from] ProvenanceError),
}

/// Reclaim every orphaned (`Running` or `Pending`) run, chain-first. Isolates
/// per-orphan failures; only an inability to enumerate the store returns
/// `Err`. Call once at startup, before serving traffic.
pub async fn sweep_orphaned_runs(
    run_store: &RunStore,
    provenance: &Arc<dyn ProvenanceRepository>,
) -> Result<SweepSummary, SweepError> {
    let mut summary = SweepSummary::default();
    for run in run_store.list_active().await? {
        if !is_orphaned(run.status) {
            continue;
        }
        match abort_orphan(
            run_store,
            provenance,
            &run.run_id,
            &run.workflow_id,
            &run.author,
        )
        .await
        {
            Ok(()) => summary.swept += 1,
            Err(e) => {
                tracing::error!(run_id = run.run_id, error = %e, "failed to reclaim orphaned run; continuing");
                summary.failed.push(run.run_id);
            }
        }
    }
    Ok(summary)
}

/// A run left mid-flight by a dead process (before serving traffic resumes).
fn is_orphaned(status: RunStatus) -> bool {
    matches!(status, RunStatus::Running | RunStatus::Pending)
}

async fn abort_orphan(
    run_store: &RunStore,
    provenance: &Arc<dyn ProvenanceRepository>,
    run_id: &str,
    workflow_id: &str,
    author: &str,
) -> Result<(), SweepError> {
    // The orphan-abort marker carries no payload hash — the run's own
    // `AgentRunStarted` entry already bound it to the canonical definition;
    // this entry only records that the run ended by restart.
    let binding = RunProvenance {
        repo: Arc::clone(provenance),
        run_id: run_id.to_string(),
        author: author.to_string(),
        workflow_id: workflow_id.to_string(),
        content_hash: String::new(),
    };
    binding
        .record_terminal(STATUS_ABORTED, Some(RESTART_REASON.to_string()))
        .await?;
    run_store
        .set_aborted(run_id, RESTART_REASON.to_string())
        .await?;
    Ok(())
}