klieo_ops_redispatcher/

lib.rs

#![deny(missing_docs)]
#![deny(rustdoc::broken_intra_doc_links)]
//! Library API for the klieo-ops redispatcher.
//!
//! The `klieo-ops-redispatcher` binary is a thin CLI wrapper. Downstream
//! integration tests and operator harnesses import this crate directly and
//! call [`redispatch`].

use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use klieo_ops::worklog::{WorkLog, WorkStatus};
use std::time::Duration;

const LIST_LIMIT: usize = 10_000;

/// Whether to apply the transition or just report what would happen.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RedispatchMode {
    /// Transition every stale item back to Ready.
    Apply,
    /// Log and count what would transition but do not write.
    DryRun,
}

/// Scan the worklog for stale items and transition them back to `Ready`.
///
/// An item is considered stale when its current status is one of `statuses`
/// AND its `last_transition_at` timestamp is older than `threshold` ago.
///
/// Returns the count of items transitioned (or that would have been
/// transitioned when `mode` is [`RedispatchMode::DryRun`]).
///
/// Re-dispatching a [`WorkStatus::AwaitingApproval`] item discards a pending
/// human-approval (four-eyes) decision; each such transition is logged at WARN.
/// The downstream consumer of `Ready` items MUST re-wrap effectful tools in
/// `GatedTool` so a redispatched item re-enters the approval gate rather than
/// running unguarded.
pub async fn redispatch(
    worklog: &dyn WorkLog,
    threshold: Duration,
    statuses: &[WorkStatus],
    mode: RedispatchMode,
) -> Result<usize> {
    let cutoff: DateTime<Utc> = Utc::now()
        - chrono::Duration::from_std(threshold).context("threshold to chrono::Duration")?;
    let mut transitioned = 0usize;

    for &status in statuses {
        for item in worklog.list_by_status(status, LIST_LIMIT).await {
            // A malformed or contended item must not abort the sweep.
            let ts: DateTime<Utc> = match item.last_transition_at.parse() {
                Ok(ts) => ts,
                Err(err) => {
                    tracing::warn!(
                        target: "klieo.ops.redispatch",
                        work_id = %item.id,
                        error = %err,
                        "unparseable last_transition_at; skipping item",
                    );
                    continue;
                }
            };

            if ts >= cutoff {
                continue;
            }

            if mode == RedispatchMode::DryRun {
                tracing::info!(work_id = %item.id, status = ?status, "would re-dispatch (dry_run)");
                transitioned += 1;
                continue;
            }

            if status == WorkStatus::AwaitingApproval {
                tracing::warn!(
                    target: "audit.redispatch",
                    work_id = %item.id,
                    "re-dispatching an AwaitingApproval item to Ready discards a pending \
                     human-approval decision; the Ready consumer must re-gate effectful tools",
                );
            }

            // CAS to Ready only if the item is still in the stale status we
            // observed. If another redispatcher or a worker already moved it,
            // `Ok(false)` — we neither re-transition nor count it.
            match worklog
                .transition_if_status(item.id.clone(), status, WorkStatus::Ready)
                .await
            {
                Ok(true) => {
                    tracing::info!(work_id = %item.id, status = ?status, "re-dispatched to Ready");
                    transitioned += 1;
                }
                Ok(false) => tracing::debug!(
                    work_id = %item.id,
                    status = ?status,
                    "item left stale status before transition; another actor handled it",
                ),
                Err(err) => tracing::warn!(
                    target: "klieo.ops.redispatch",
                    work_id = %item.id,
                    error = %err,
                    "transition failed; skipping item",
                ),
            }
        }
    }

    Ok(transitioned)
}

/// Parse a comma-separated list of snake_case `WorkStatus` names.
///
/// Accepted values: `in_progress`, `awaiting_approval`, `planned`,
/// `ready`, `done`, `failed`, `cancelled`.
pub fn parse_statuses(raw: &str) -> Result<Vec<WorkStatus>> {
    raw.split(',')
        .map(|s| match s.trim() {
            "planned" => Ok(WorkStatus::Planned),
            "ready" => Ok(WorkStatus::Ready),
            "in_progress" => Ok(WorkStatus::InProgress),
            "done" => Ok(WorkStatus::Done),
            "failed" => Ok(WorkStatus::Failed),
            "awaiting_approval" => Ok(WorkStatus::AwaitingApproval),
            "cancelled" => Ok(WorkStatus::Cancelled),
            other => anyhow::bail!("unknown WorkStatus: {other:?}"),
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use futures_core::stream::BoxStream;
    use klieo_bus_memory::MemoryBus;
    use klieo_ops::worklog::{KvWorkLog, WorkDag, WorkId, WorkItem, WorkLogError};
    use std::sync::Mutex;

    fn fresh_item(title: &str) -> WorkItem {
        WorkItem::new(title, serde_json::json!({}), None, vec![])
    }

    /// Returns a controlled candidate list and records every successful
    /// `transition_if_status`. The transition outcome is keyed off the id
    /// prefix: `skip*` → `Ok(false)` (lost the race), `err*` → `Err`, anything
    /// else → `Ok(true)`. Only the two methods `redispatch` calls are
    /// meaningful; the rest are never reached.
    struct MockWorkLog {
        items: Vec<WorkItem>,
        transitioned: Mutex<Vec<WorkId>>,
    }

    fn item_at(id: &str, last_transition_at: &str) -> WorkItem {
        let mut item = WorkItem::new(id, serde_json::json!({}), None, vec![]);
        item.id = WorkId(id.to_string());
        item.last_transition_at = last_transition_at.to_string();
        item
    }

    const STALE: &str = "2020-01-01T00:00:00Z";

    #[async_trait::async_trait]
    impl WorkLog for MockWorkLog {
        async fn list_by_status(&self, _filter: WorkStatus, _limit: usize) -> Vec<WorkItem> {
            self.items.clone()
        }

        async fn transition_if_status(
            &self,
            id: WorkId,
            _from: WorkStatus,
            _to: WorkStatus,
        ) -> Result<bool, WorkLogError> {
            if id.0.starts_with("skip") {
                return Ok(false);
            }
            if id.0.starts_with("err") {
                return Err(WorkLogError::Storage {
                    message: "transition failed".into(),
                    source: None,
                });
            }
            self.transitioned.lock().unwrap().push(id);
            Ok(true)
        }

        async fn plan(&self, _: WorkItem) -> Result<WorkId, WorkLogError> {
            unreachable!()
        }
        async fn depend(&self, _: WorkId, _: WorkId) -> Result<(), WorkLogError> {
            unreachable!()
        }
        async fn ready(&self, _: usize) -> Vec<WorkId> {
            unreachable!()
        }
        async fn ready_stream(&self) -> BoxStream<'static, WorkId> {
            unreachable!()
        }
        async fn dispatch(&self, _: WorkId) -> Result<(), WorkLogError> {
            unreachable!()
        }
        async fn transition(&self, _: WorkId, _: WorkStatus) -> Result<(), WorkLogError> {
            unreachable!()
        }
        async fn get(&self, _: WorkId) -> Option<WorkItem> {
            unreachable!()
        }
        async fn dag(&self, _: WorkId) -> WorkDag {
            unreachable!()
        }
    }

    async fn redispatch_apply(items: Vec<WorkItem>) -> (usize, Vec<WorkId>) {
        let wl = MockWorkLog {
            items,
            transitioned: Mutex::new(vec![]),
        };
        let n = redispatch(
            &wl,
            Duration::ZERO,
            &[WorkStatus::InProgress],
            RedispatchMode::Apply,
        )
        .await
        .expect("a per-item failure must not abort the sweep");
        let transitioned = wl.transitioned.lock().unwrap().clone();
        (n, transitioned)
    }

    /// An unparseable `last_transition_at` is logged and skipped — the sweep
    /// continues and still transitions the valid stale item.
    #[tokio::test]
    async fn redispatch_skips_unparseable_item_and_continues() {
        let (n, transitioned) = redispatch_apply(vec![
            item_at("bad-1", "not-a-timestamp"),
            item_at("good-1", STALE),
        ])
        .await;
        assert_eq!(n, 1, "bad item skipped, good item still transitioned");
        assert_eq!(transitioned.as_slice(), &[WorkId("good-1".into())]);
    }

    /// An item another actor already moved (`transition_if_status` → `Ok(false)`)
    /// is not counted, and the sweep still transitions the rest.
    #[tokio::test]
    async fn redispatch_skips_item_lost_to_concurrent_actor() {
        let (n, transitioned) =
            redispatch_apply(vec![item_at("skip-1", STALE), item_at("good-1", STALE)]).await;
        assert_eq!(n, 1, "the Ok(false) item is not counted");
        assert_eq!(transitioned.as_slice(), &[WorkId("good-1".into())]);
    }

    /// A `transition_if_status` error on one item is logged and skipped — it
    /// must not abort the sweep.
    #[tokio::test]
    async fn redispatch_skips_item_whose_transition_errors() {
        let (n, transitioned) =
            redispatch_apply(vec![item_at("err-1", STALE), item_at("good-1", STALE)]).await;
        assert_eq!(
            n, 1,
            "the erroring item is skipped, the good one transitions"
        );
        assert_eq!(transitioned.as_slice(), &[WorkId("good-1".into())]);
    }

    #[tokio::test]
    async fn redispatch_with_empty_log_returns_zero() {
        let bus = MemoryBus::new();
        let wl = KvWorkLog::new(bus.kv.clone());
        let n = redispatch(
            &wl,
            Duration::from_secs(60),
            &[WorkStatus::InProgress],
            RedispatchMode::DryRun,
        )
        .await
        .expect("redispatch ok");
        assert_eq!(n, 0);
    }

    #[tracing_test::traced_test]
    #[tokio::test]
    async fn redispatch_warns_when_discarding_awaiting_approval() {
        let bus = MemoryBus::new();
        let wl = KvWorkLog::new(bus.kv.clone());
        let id = wl.plan(fresh_item("needs approval")).await.expect("plan");
        wl.transition(id.clone(), WorkStatus::AwaitingApproval)
            .await
            .expect("to awaiting_approval");

        let n = redispatch(
            &wl,
            Duration::ZERO,
            &[WorkStatus::AwaitingApproval],
            RedispatchMode::Apply,
        )
        .await
        .expect("redispatch ok");

        assert_eq!(n, 1, "the stale awaiting-approval item is redispatched");
        assert!(
            logs_contain("discards a pending"),
            "redispatching an AwaitingApproval item must warn (four-eyes decision discarded)"
        );
        assert_eq!(
            wl.get(id).await.expect("item exists").status,
            WorkStatus::Ready
        );
    }

    #[tokio::test]
    async fn parse_statuses_accepts_valid_input() {
        let statuses = parse_statuses("in_progress,awaiting_approval").expect("parse ok");
        assert_eq!(
            statuses,
            vec![WorkStatus::InProgress, WorkStatus::AwaitingApproval]
        );
    }

    #[tokio::test]
    async fn parse_statuses_rejects_unknown_name() {
        assert!(parse_statuses("bogus").is_err());
    }

    #[test]
    fn cli_default_statuses_excludes_awaiting_approval() {
        // The binary's `--statuses` default (main.rs) is this string; redispatching
        // AwaitingApproval discards a four-eyes decision, so it must NOT be default.
        let statuses = parse_statuses("in_progress").expect("parse ok");
        assert_eq!(statuses, vec![WorkStatus::InProgress]);
        assert!(!statuses.contains(&WorkStatus::AwaitingApproval));
    }

    #[tokio::test]
    async fn fresh_item_not_re_dispatched_when_threshold_is_one_hour() {
        // Items dispatched NOW are not stale against a 3600s threshold.
        let bus = MemoryBus::new();
        let wl = KvWorkLog::new(bus.kv.clone());
        let id = wl.plan(fresh_item("recent")).await.expect("plan");
        wl.dispatch(id).await.expect("dispatch");

        let n = redispatch(
            &wl,
            Duration::from_secs(3600),
            &[WorkStatus::InProgress],
            RedispatchMode::Apply,
        )
        .await
        .expect("redispatch");
        assert_eq!(n, 0, "recently dispatched item must not be re-dispatched");
    }
}