klieo-ops 3.5.0

//! `WorkLog` trait + supporting types.

use crate::types::TenantId;
use async_trait::async_trait;
use futures_core::stream::BoxStream;
use serde::{Deserialize, Serialize};
use thiserror::Error;

/// Server-assigned work identifier.
#[derive(Clone, Debug, Hash, Eq, PartialEq, Serialize, Deserialize)]
pub struct WorkId(pub String);

impl std::fmt::Display for WorkId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

/// Lifecycle status of a work item.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum WorkStatus {
    /// Planned but dependencies not yet satisfied.
    Planned,
    /// All dependencies satisfied; ready for dispatch.
    Ready,
    /// Dispatched to the JobQueue; awaiting / executing.
    InProgress,
    /// Completed successfully.
    Done,
    /// Failed (terminal).
    Failed,
    /// Awaiting human approval (Phase B FourEyesGate suspend path).
    AwaitingApproval,
    /// Cancelled (terminal — operator action or kill-switch).
    Cancelled,
}

impl WorkStatus {
    /// Whether this status is terminal (Done / Failed / Cancelled).
    #[must_use]
    pub fn is_terminal(self) -> bool {
        matches!(self, Self::Done | Self::Failed | Self::Cancelled)
    }
}

/// A single work item in the DAG.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[non_exhaustive]
pub struct WorkItem {
    /// Server-assigned id (set by `plan` — callers leave empty).
    pub id: WorkId,
    /// Tenant tag.
    pub tenant: Option<TenantId>,
    /// Short title / label for audit + operator UI.
    pub title: String,
    /// Free-form payload for the agent runtime to consume on dispatch.
    pub payload: serde_json::Value,
    /// Current status.
    pub status: WorkStatus,
    /// Direct parent dependencies (must reach Done before this item enters Ready).
    pub depends_on: Vec<WorkId>,
    /// RFC 3339 timestamp of last status transition.
    pub last_transition_at: String,
}

impl WorkItem {
    /// Construct a `WorkItem` for planning. `id` and `last_transition_at` are
    /// left empty — the `WorkLog` impl fills them in on `plan`. Use this
    /// instead of struct literal syntax — `#[non_exhaustive]` blocks
    /// cross-crate struct literals.
    pub fn new(
        title: impl Into<String>,
        payload: serde_json::Value,
        tenant: Option<TenantId>,
        depends_on: Vec<WorkId>,
    ) -> Self {
        Self {
            id: WorkId(String::new()),
            tenant,
            title: title.into(),
            payload,
            status: WorkStatus::Planned,
            depends_on,
            last_transition_at: String::new(),
        }
    }
}

/// DAG snapshot returned by `WorkLog::dag`.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct WorkDag {
    /// All items reachable from the root (inclusive).
    pub items: Vec<WorkItem>,
}

/// Errors raised by WorkLog calls.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum WorkLogError {
    /// Underlying storage unavailable.
    #[error("worklog storage unavailable: {message}")]
    Storage {
        /// Context describing the storage operation that failed.
        message: String,
        /// Underlying cause preserved for `std::error::Error::source()`;
        /// `None` for storage faults with no lower-level error (e.g. CAS
        /// retry exhaustion).
        #[source]
        source: Option<Box<dyn std::error::Error + Send + Sync + 'static>>,
    },
    /// Caller-supplied id is unknown.
    #[error("unknown work item: {0}")]
    UnknownWorkItem(WorkId),
    /// Adding the dependency would create a cycle.
    #[error("dependency would create a cycle: {child} -> {on}")]
    CycleDetected {
        /// Child id.
        child: WorkId,
        /// Dependency parent id.
        on: WorkId,
    },
    /// Max-DAG-size cap exceeded.
    #[error("DAG size cap {cap} exceeded")]
    CapExceeded {
        /// Configured cap.
        cap: usize,
    },
    /// JobQueue write failed during dispatch.
    #[error("dispatch failed: {0}")]
    DispatchFailed(String),
    /// Other internal.
    #[error("internal: {0}")]
    Internal(String),
}

/// Worklog primitive.
#[async_trait]
pub trait WorkLog: Send + Sync {
    /// Plan a new work item. Returns the server-assigned `WorkId`.
    /// Items start in `WorkStatus::Planned` and transition to `Ready`
    /// when their dependencies are all `Done`. An item with no
    /// dependencies starts at `Ready`.
    async fn plan(&self, item: WorkItem) -> Result<WorkId, WorkLogError>;

    /// Register a dependency relationship between two existing items.
    /// Rejects cycles.
    async fn depend(&self, child: WorkId, on: WorkId) -> Result<(), WorkLogError>;

    /// List up to `limit` items currently in `WorkStatus::Ready`.
    async fn ready(&self, limit: usize) -> Vec<WorkId>;

    /// Stream items as they transition into `Ready`. Implementation may
    /// poll; bounds + cadence are impl-defined.
    async fn ready_stream(&self) -> BoxStream<'static, WorkId>;

    /// Dispatch a `Ready` item to the JobQueue + transition to
    /// `InProgress`. Returns error if the item is not currently `Ready`.
    async fn dispatch(&self, id: WorkId) -> Result<(), WorkLogError>;

    /// Transition an item to the given status. Validates that the
    /// transition is legal (e.g. cannot go from `Done` back to `Planned`).
    async fn transition(&self, id: WorkId, status: WorkStatus) -> Result<(), WorkLogError>;

    /// Atomically transition an item to `to` **only if** its current status is
    /// still `from`. Returns `Ok(true)` when the transition was applied and
    /// `Ok(false)` when the item had already moved off `from` (a concurrent
    /// actor won the race) — no transition happens in that case.
    ///
    /// This is the compare-and-swap the redispatcher needs: two redispatchers
    /// (or a redispatcher racing a worker) that both observe an item as stale
    /// must not both transition it, or the item is dispatched twice.
    ///
    /// The default implementation is **best-effort, not atomic** — it reads the
    /// status then transitions, so a third-party `WorkLog` must override this
    /// with a store-native CAS to get a true guarantee.
    async fn transition_if_status(
        &self,
        id: WorkId,
        from: WorkStatus,
        to: WorkStatus,
    ) -> Result<bool, WorkLogError> {
        match self.get(id.clone()).await {
            Some(item) if item.status == from => {
                self.transition(id, to).await?;
                Ok(true)
            }
            Some(_) => Ok(false),
            None => Err(WorkLogError::UnknownWorkItem(id)),
        }
    }

    /// Read a single item by id.
    async fn get(&self, id: WorkId) -> Option<WorkItem>;

    /// Snapshot the DAG reachable from `root` (inclusive). For full-DAG
    /// queries pass any node id.
    async fn dag(&self, root: WorkId) -> WorkDag;

    /// List up to `limit` items whose current status matches `filter`.
    ///
    /// Used by the redispatcher cron to locate stale `InProgress` /
    /// `AwaitingApproval` items without a known root. Implementations may
    /// bound the scan; callers should set `limit` conservatively.
    ///
    /// The default implementation returns an empty vec so existing
    /// third-party `WorkLog` impls compile without changes.
    async fn list_by_status(&self, filter: WorkStatus, limit: usize) -> Vec<WorkItem> {
        let _ = (filter, limit);
        Vec::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::error::Error as StdError;

    #[test]
    fn storage_preserves_source_chain() {
        let inner = std::io::Error::other("kv connection refused");
        let e = WorkLogError::Storage {
            message: "fetch item".into(),
            source: Some(Box::new(inner)),
        };
        assert_eq!(e.to_string(), "worklog storage unavailable: fetch item");
        let src = e.source().expect("source must be preserved");
        assert_eq!(src.to_string(), "kv connection refused");
    }

    #[test]
    fn storage_without_source_returns_none() {
        let e = WorkLogError::Storage {
            message: "cas exhaustion".into(),
            source: None,
        };
        assert!(e.source().is_none());
    }
}