klieo-ops 0.41.2

//! `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: {0}")]
    Storage(String),
    /// 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>;

    /// 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()
    }
}