klieo-ops 3.3.0

//! KV-backed WorkLog impl. KV is the single source of truth for items,
//! adjacency, and status indexes — enabling correct multi-process operation
//! (spec § 1.10). No in-process DashMap cache; every read goes to KV.
//!
//! Bucket layout:
//! - `ops.worklog.items/<id>`          — JSON-serialised `WorkItem`
//! - `ops.worklog.parents/<id>`        — JSON-encoded `HashSet<WorkId>`
//! - `ops.worklog.children/<id>`       — JSON-encoded `HashSet<WorkId>`
//! - `ops.worklog.status/<status_key>` — JSON-encoded `HashSet<WorkId>`
//! - `ops.worklog.meta/count`          — u64 item count (CAS-maintained)
//!
//! Status-index buckets (one per `WorkStatus` variant) enable correct
//! `ready()` and `list_by_status()` without a KV listing primitive.
//! Each transition updates the old-status index (remove id) and the
//! new-status index (add id) via CAS-retry (up to 5 attempts with backoff).
//!
//! `dag()` performs BFS over parents/children via KV reads. Cycle
//! detection in `would_cycle()` does the same upward BFS from `on`.
//! Dependencies satisfied check reads each parent's item status from KV.
//!
//! Phase B M7 cap-enforcement reads the count key; the count is
//! incremented on `plan` and never decremented (items are not deleted in
//! this phase — orphan TTL is operator-cron territory per design spec § 2.5).

use super::trait_::{WorkDag, WorkId, WorkItem, WorkLog, WorkLogError, WorkStatus};
use crate::audit::OpsAuditSink;
use crate::ops_event::OpsEvent;
use async_trait::async_trait;
use bytes::Bytes;
use chrono::Utc;
use futures::future::try_join_all;
use futures::stream;
use futures_core::stream::BoxStream;
use klieo_core::error::BusError;
use klieo_core::ids::RunId;
use klieo_core::memory::EpisodicMemory;
use klieo_core::KvStore;
use std::collections::{HashSet, VecDeque};
use std::sync::Arc;
use tokio::sync::broadcast;
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::StreamExt;
use ulid::Ulid;

const BUCKET_ITEMS: &str = "ops.worklog.items";
const BUCKET_PARENTS: &str = "ops.worklog.parents";
const BUCKET_CHILDREN: &str = "ops.worklog.children";
const BUCKET_STATUS: &str = "ops.worklog.status";
const BUCKET_META: &str = "ops.worklog.meta";

const META_COUNT_KEY: &str = "count";
const CAS_MAX_RETRIES: u32 = 5;
const DEFAULT_MAX_ITEMS: usize = 10_000;
/// Number of shards for the status-index hot keys. Each status variant is
/// distributed across `STATUS_INDEX_SHARDS` KV keys instead of one, so
/// concurrent transitions to the same status do not contend on a single CAS
/// key. `list_by_status` reads all shards (16 reads) and unions the results.
const STATUS_INDEX_SHARDS: usize = 16;

/// KV-backed worklog. All state lives in KV; correct across processes
/// sharing the same `Arc<dyn KvStore>`.
pub struct KvWorkLog {
    kv: Arc<dyn KvStore>,
    cap: usize,
    ready_tx: broadcast::Sender<WorkId>,
    audit: Option<OpsAuditSink>,
}

// ── Construction ────────────────────────────────────────────────────────────

impl KvWorkLog {
    /// Build with the default 10 000-item cap.
    #[must_use]
    pub fn new(kv: Arc<dyn KvStore>) -> Self {
        Self::with_cap(kv, DEFAULT_MAX_ITEMS)
    }

    /// Build with a custom DAG size cap.
    #[must_use]
    pub fn with_cap(kv: Arc<dyn KvStore>, cap: usize) -> Self {
        let (tx, _rx) = broadcast::channel(1024);
        Self {
            kv,
            cap,
            ready_tx: tx,
            audit: None,
        }
    }

    /// Wire an audit sink.
    #[must_use]
    pub fn with_audit(mut self, episodic: Arc<dyn EpisodicMemory>, run_id: RunId) -> Self {
        self.audit = Some(OpsAuditSink::new(
            episodic,
            run_id,
            "klieo.ops.worklog.audit",
        ));
        self
    }
}

// ── Internal helpers — KV reads ──────────────────────────────────────────────

impl KvWorkLog {
    async fn fetch_item(&self, id: &WorkId) -> Result<Option<WorkItem>, WorkLogError> {
        let entry = self
            .kv
            .get(BUCKET_ITEMS, &id.0)
            .await
            .map_err(|e| WorkLogError::Storage {
                message: e.to_string(),
                source: Some(Box::new(e)),
            })?;
        match entry {
            None => Ok(None),
            Some(e) => {
                let item = serde_json::from_slice::<WorkItem>(&e.value)
                    .map_err(|e| WorkLogError::Internal(format!("deserialise item: {e}")))?;
                Ok(Some(item))
            }
        }
    }

    async fn require_item(&self, id: &WorkId) -> Result<WorkItem, WorkLogError> {
        self.fetch_item(id)
            .await?
            .ok_or_else(|| WorkLogError::UnknownWorkItem(id.clone()))
    }

    /// So a caller can CAS-write the item back conditioned on the revision
    /// being unchanged since this read.
    async fn fetch_item_with_rev(
        &self,
        id: &WorkId,
    ) -> Result<Option<(WorkItem, u64)>, WorkLogError> {
        let entry = self
            .kv
            .get(BUCKET_ITEMS, &id.0)
            .await
            .map_err(|e| WorkLogError::Storage {
                message: e.to_string(),
                source: Some(Box::new(e)),
            })?;
        match entry {
            None => Ok(None),
            Some(e) => {
                let item = serde_json::from_slice::<WorkItem>(&e.value)
                    .map_err(|e| WorkLogError::Internal(format!("deserialise item: {e}")))?;
                Ok(Some((item, e.revision)))
            }
        }
    }

    async fn fetch_id_set(
        &self,
        bucket: &str,
        key: &str,
    ) -> Result<(HashSet<WorkId>, Option<u64>), WorkLogError> {
        let entry = self
            .kv
            .get(bucket, key)
            .await
            .map_err(|e| WorkLogError::Storage {
                message: e.to_string(),
                source: Some(Box::new(e)),
            })?;
        match entry {
            None => Ok((HashSet::new(), None)),
            Some(e) => {
                let set = serde_json::from_slice::<HashSet<WorkId>>(&e.value)
                    .map_err(|e| WorkLogError::Internal(format!("deserialise id set: {e}")))?;
                Ok((set, Some(e.revision)))
            }
        }
    }

    async fn fetch_parents(&self, id: &WorkId) -> Result<HashSet<WorkId>, WorkLogError> {
        let (set, _) = self.fetch_id_set(BUCKET_PARENTS, &id.0).await?;
        Ok(set)
    }

    async fn fetch_children(&self, id: &WorkId) -> Result<HashSet<WorkId>, WorkLogError> {
        let (set, _) = self.fetch_id_set(BUCKET_CHILDREN, &id.0).await?;
        Ok(set)
    }

    /// Read all shards for `status` and union the id sets. The returned
    /// revision is `None` because it spans multiple KV keys; callers that
    /// need per-shard CAS use `fetch_status_shard` directly.
    async fn fetch_status_index(
        &self,
        status: WorkStatus,
    ) -> Result<(HashSet<WorkId>, Option<u64>), WorkLogError> {
        let shard_reads: Vec<_> = (0..STATUS_INDEX_SHARDS)
            .map(|i| {
                let key = status_shard_key(status, i);
                async move { self.fetch_id_set(BUCKET_STATUS, &key).await }
            })
            .collect();
        let shards = try_join_all(shard_reads).await?;
        let union = shards
            .into_iter()
            .flat_map(|(set, _)| set)
            .collect::<HashSet<WorkId>>();
        Ok((union, None))
    }

    /// Short-circuit variant of [`fetch_status_index`]: fetches shard indexes
    /// sequentially in batches of 4 (via `buffer_unordered`) and stops once the
    /// collected unique-id count reaches `limit`. Avoids the full 16-read fan-out
    /// when the caller only needs a bounded sample (e.g. the `ready()` poll).
    async fn fetch_status_index_bounded(
        &self,
        status: WorkStatus,
        limit: usize,
    ) -> Result<HashSet<WorkId>, WorkLogError> {
        const CONCURRENCY: usize = 4;
        let mut collected: HashSet<WorkId> = HashSet::new();

        // Use UFCS to avoid ambiguity with tokio_stream::StreamExt which is
        // also in scope via tokio_stream imports elsewhere in the file.
        let raw = futures::StreamExt::map(stream::iter(0..STATUS_INDEX_SHARDS), |i| {
            let key = status_shard_key(status, i);
            async move { self.fetch_id_set(BUCKET_STATUS, &key).await }
        });
        let mut shard_stream = futures::StreamExt::buffer_unordered(raw, CONCURRENCY);

        while let Some(result) = futures::StreamExt::next(&mut shard_stream).await {
            let (shard_ids, _) = result?;
            collected.extend(shard_ids);
            if collected.len() >= limit {
                break;
            }
        }
        Ok(collected)
    }

    async fn fetch_item_count(&self) -> Result<(usize, Option<u64>), WorkLogError> {
        let entry = self
            .kv
            .get(BUCKET_META, META_COUNT_KEY)
            .await
            .map_err(|e| WorkLogError::Storage {
                message: e.to_string(),
                source: Some(Box::new(e)),
            })?;
        match entry {
            None => Ok((0, None)),
            Some(e) => {
                let n = serde_json::from_slice::<u64>(&e.value)
                    .map_err(|e| WorkLogError::Internal(format!("deserialise count: {e}")))?;
                Ok((n as usize, Some(e.revision)))
            }
        }
    }
}

// ── Internal helpers — KV writes ─────────────────────────────────────────────

impl KvWorkLog {
    async fn persist_item(&self, item: &WorkItem) -> Result<(), WorkLogError> {
        let body = serde_json::to_vec(item)
            .map_err(|e| WorkLogError::Internal(format!("serialise item: {e}")))?;
        self.kv
            .put(BUCKET_ITEMS, &item.id.0, Bytes::from(body))
            .await
            .map(|_| ())
            .map_err(|e| WorkLogError::Storage {
                message: e.to_string(),
                source: Some(Box::new(e)),
            })
    }

    /// CAS-retry: add `id` to the set stored at `(bucket, key)`.
    async fn cas_add_to_set(
        &self,
        bucket: &str,
        key: &str,
        id: WorkId,
    ) -> Result<(), WorkLogError> {
        for attempt in 0..CAS_MAX_RETRIES {
            let (mut set, rev) = self.fetch_id_set(bucket, key).await?;
            set.insert(id.clone());
            let encoded = encode_id_set(&set)?;
            match self.kv.cas(bucket, key, encoded, rev).await {
                Ok(_) => return Ok(()),
                Err(BusError::CasConflict { .. }) => {
                    if attempt + 1 < CAS_MAX_RETRIES {
                        backoff_delay(attempt).await;
                        continue;
                    }
                    return Err(WorkLogError::Storage {
                        message: "cas_add_to_set: too many CAS conflicts".into(),
                        source: None,
                    });
                }
                Err(e) => {
                    return Err(WorkLogError::Storage {
                        message: e.to_string(),
                        source: Some(Box::new(e)),
                    })
                }
            }
        }
        unreachable!()
    }

    /// CAS-retry: remove `id` from the set stored at `(bucket, key)`.
    async fn cas_remove_from_set(
        &self,
        bucket: &str,
        key: &str,
        id: &WorkId,
    ) -> Result<(), WorkLogError> {
        for attempt in 0..CAS_MAX_RETRIES {
            let (mut set, rev) = self.fetch_id_set(bucket, key).await?;
            if !set.remove(id) {
                // Already absent — idempotent success.
                return Ok(());
            }
            let encoded = encode_id_set(&set)?;
            match self.kv.cas(bucket, key, encoded, rev).await {
                Ok(_) => return Ok(()),
                Err(BusError::CasConflict { .. }) => {
                    if attempt + 1 < CAS_MAX_RETRIES {
                        backoff_delay(attempt).await;
                        continue;
                    }
                    return Err(WorkLogError::Storage {
                        message: "cas_remove_from_set: too many CAS conflicts".into(),
                        source: None,
                    });
                }
                Err(e) => {
                    return Err(WorkLogError::Storage {
                        message: e.to_string(),
                        source: Some(Box::new(e)),
                    })
                }
            }
        }
        unreachable!()
    }

    /// CAS-retry increment of the item count. Returns the new count.
    async fn cas_increment_count(&self) -> Result<usize, WorkLogError> {
        for attempt in 0..CAS_MAX_RETRIES {
            let (current, rev) = self.fetch_item_count().await?;
            let next = current + 1;
            let encoded = serde_json::to_vec(&(next as u64))
                .map_err(|e| WorkLogError::Internal(format!("serialise count: {e}")))?;
            match self
                .kv
                .cas(BUCKET_META, META_COUNT_KEY, Bytes::from(encoded), rev)
                .await
            {
                Ok(_) => return Ok(next),
                Err(BusError::CasConflict { .. }) => {
                    if attempt + 1 < CAS_MAX_RETRIES {
                        backoff_delay(attempt).await;
                        continue;
                    }
                    return Err(WorkLogError::Storage {
                        message: "cas_increment_count: too many CAS conflicts".into(),
                        source: None,
                    });
                }
                Err(e) => {
                    return Err(WorkLogError::Storage {
                        message: e.to_string(),
                        source: Some(Box::new(e)),
                    })
                }
            }
        }
        unreachable!()
    }

    /// Move `id` from the old-status shard to the new-status shard in KV.
    /// Each id is consistently mapped to one shard by `shard_for_id` so
    /// the item lands in and is removed from the same shard on every call.
    async fn update_status_index(
        &self,
        id: &WorkId,
        from: WorkStatus,
        to: WorkStatus,
    ) -> Result<(), WorkLogError> {
        if from == to {
            return Ok(());
        }
        let shard = shard_for_id(id);
        let from_key = status_shard_key(from, shard);
        let to_key = status_shard_key(to, shard);
        let remove = self.cas_remove_from_set(BUCKET_STATUS, &from_key, id);
        let add = self.cas_add_to_set(BUCKET_STATUS, &to_key, id.clone());
        // Independent shards — run concurrently.
        let (r, a) = tokio::join!(remove, add);
        r?;
        a
    }

    async fn emit(&self, event: OpsEvent) {
        if let Some(sink) = &self.audit {
            sink.emit(event).await;
        }
    }

    fn announce_ready(&self, id: WorkId) {
        if let Err(err) = self.ready_tx.send(id) {
            tracing::debug!(
                target: "klieo.ops.worklog",
                work_id = %err.0,
                "ready-stream has no live receivers; dropping notification"
            );
        }
    }
}

// ── Cross-process graph algorithms ───────────────────────────────────────────

impl KvWorkLog {
    /// BFS upward from `on` along KV-stored parents; returns true if `child`
    /// is reachable (which would form a cycle).
    async fn would_cycle(&self, child: &WorkId, on: &WorkId) -> Result<bool, WorkLogError> {
        let mut queue: VecDeque<WorkId> = VecDeque::new();
        let mut visited: HashSet<WorkId> = HashSet::new();
        queue.push_back(on.clone());
        while let Some(node) = queue.pop_front() {
            if &node == child {
                return Ok(true);
            }
            if !visited.insert(node.clone()) {
                continue;
            }
            for parent in self.fetch_parents(&node).await? {
                queue.push_back(parent);
            }
        }
        Ok(false)
    }

    /// Returns true when all parents of `id` are in status `Done` per KV.
    async fn dependencies_satisfied(&self, id: &WorkId) -> Result<bool, WorkLogError> {
        let parents = self.fetch_parents(id).await?;
        for parent_id in &parents {
            match self.fetch_item(parent_id).await? {
                Some(item) if item.status == WorkStatus::Done => continue,
                _ => return Ok(false),
            }
        }
        Ok(true)
    }
}

// ── WorkLog impl ─────────────────────────────────────────────────────────────

#[async_trait]
impl WorkLog for KvWorkLog {
    async fn plan(&self, mut item: WorkItem) -> Result<WorkId, WorkLogError> {
        // Cap check via KV count (cross-process correct).
        let count = self.fetch_item_count().await?.0;
        if count >= self.cap {
            return Err(WorkLogError::CapExceeded { cap: self.cap });
        }

        let id = WorkId(format!("wrk_{}", Ulid::new()));
        item.id = id.clone();
        item.last_transition_at = Utc::now().to_rfc3339();

        let initial_deps = std::mem::take(&mut item.depends_on);

        for parent in &initial_deps {
            if self.fetch_item(parent).await?.is_none() {
                return Err(WorkLogError::UnknownWorkItem(parent.clone()));
            }
        }

        // A freshly-minted id cannot appear in any existing parent chain,
        // so we skip cycle-checking for initial deps.
        item.status = if initial_deps.is_empty() {
            WorkStatus::Ready
        } else {
            WorkStatus::Planned
        };

        // Persist item first so other processes can see it immediately.
        self.persist_item(&item).await?;
        self.cas_increment_count().await?;

        let edge_futures: Vec<_> = initial_deps
            .iter()
            .map(|parent| {
                let parent_id = parent.clone();
                let child_id = id.clone();
                async move {
                    let add_parent =
                        self.cas_add_to_set(BUCKET_PARENTS, &child_id.0, parent_id.clone());
                    let add_child =
                        self.cas_add_to_set(BUCKET_CHILDREN, &parent_id.0, child_id.clone());
                    let (r1, r2) = tokio::join!(add_parent, add_child);
                    r1?;
                    r2
                }
            })
            .collect();
        for fut in edge_futures {
            fut.await?;
        }

        let shard = shard_for_id(&id);
        self.cas_add_to_set(
            BUCKET_STATUS,
            &status_shard_key(item.status, shard),
            id.clone(),
        )
        .await?;

        self.emit(OpsEvent::WorkPlanned {
            tenant: item.tenant.clone(),
            work_id: id.0.clone(),
            title: item.title.clone(),
            depends_on: initial_deps.iter().map(|p| p.0.clone()).collect(),
        })
        .await;

        if item.status == WorkStatus::Ready {
            self.announce_ready(id.clone());
        }
        Ok(id)
    }

    async fn depend(&self, child: WorkId, on: WorkId) -> Result<(), WorkLogError> {
        if self.fetch_item(&child).await?.is_none() {
            return Err(WorkLogError::UnknownWorkItem(child));
        }
        if self.fetch_item(&on).await?.is_none() {
            return Err(WorkLogError::UnknownWorkItem(on));
        }
        if self.would_cycle(&child, &on).await? {
            return Err(WorkLogError::CycleDetected { child, on });
        }
        let add_parent = self.cas_add_to_set(BUCKET_PARENTS, &child.0, on.clone());
        let add_child = self.cas_add_to_set(BUCKET_CHILDREN, &on.0, child.clone());
        let (r1, r2) = tokio::join!(add_parent, add_child);
        r1?;
        r2?;

        let tenant = self
            .fetch_item(&child)
            .await?
            .and_then(|i| i.tenant.clone());
        self.emit(OpsEvent::WorkDependencyAdded {
            tenant,
            child: child.0,
            on: on.0,
        })
        .await;
        Ok(())
    }

    async fn ready(&self, limit: usize) -> Vec<WorkId> {
        // Use the bounded variant to short-circuit once enough ids are
        // collected; avoids the full 16-shard fan-out when limit is small.
        match self
            .fetch_status_index_bounded(WorkStatus::Ready, limit)
            .await
        {
            Ok(set) => set.into_iter().take(limit).collect(),
            Err(e) => {
                tracing::warn!(target: "klieo.ops.worklog", error = %e, "ready() index read failed");
                vec![]
            }
        }
    }

    async fn ready_stream(&self) -> BoxStream<'static, WorkId> {
        let rx = self.ready_tx.subscribe();
        let stream = BroadcastStream::new(rx).filter_map(|r| r.ok());
        Box::pin(stream)
    }

    async fn dispatch(&self, id: WorkId) -> Result<(), WorkLogError> {
        let item = self.require_item(&id).await?;
        if item.status != WorkStatus::Ready {
            return Err(WorkLogError::Internal(format!(
                "cannot dispatch item in status {:?}",
                item.status
            )));
        }
        let mut updated = item.clone();
        updated.status = WorkStatus::InProgress;
        updated.last_transition_at = Utc::now().to_rfc3339();
        self.persist_item(&updated).await?;
        self.update_status_index(&id, WorkStatus::Ready, WorkStatus::InProgress)
            .await?;
        self.emit(OpsEvent::WorkDispatched {
            tenant: updated.tenant.clone(),
            work_id: id.0,
        })
        .await;
        Ok(())
    }

    async fn transition(&self, id: WorkId, status: WorkStatus) -> Result<(), WorkLogError> {
        let item = self.require_item(&id).await?;
        if item.status.is_terminal() && item.status != status {
            return Err(WorkLogError::Internal(format!(
                "cannot transition from terminal status {:?} to {:?}",
                item.status, status
            )));
        }
        let prev_status = item.status;
        let mut updated = item;
        updated.status = status;
        updated.last_transition_at = Utc::now().to_rfc3339();
        self.persist_item(&updated).await?;
        self.update_status_index(&id, prev_status, status).await?;

        self.emit(OpsEvent::WorkTransition {
            tenant: updated.tenant.clone(),
            work_id: id.0.clone(),
            from: format!("{prev_status:?}").to_lowercase(),
            to: format!("{status:?}").to_lowercase(),
            reason: None,
        })
        .await;

        if status == WorkStatus::Done {
            self.cascade_ready_children(&id).await?;
        }
        Ok(())
    }

    async fn transition_if_status(
        &self,
        id: WorkId,
        from: WorkStatus,
        to: WorkStatus,
    ) -> Result<bool, WorkLogError> {
        let (item, rev) = match self.fetch_item_with_rev(&id).await? {
            Some(pair) => pair,
            None => return Err(WorkLogError::UnknownWorkItem(id)),
        };
        // A concurrent actor already moved it off `from` and won the race.
        if item.status != from {
            return Ok(false);
        }
        if item.status.is_terminal() && item.status != to {
            return Err(WorkLogError::Internal(format!(
                "cannot transition from terminal status {:?} to {:?}",
                item.status, to
            )));
        }

        let mut updated = item;
        updated.status = to;
        updated.last_transition_at = Utc::now().to_rfc3339();
        let body = serde_json::to_vec(&updated)
            .map_err(|e| WorkLogError::Internal(format!("serialise item: {e}")))?;

        // CAS the item write on the revision read above: if anyone else wrote
        // the item between the read and here, the revision moved and we lose the
        // race rather than clobbering their transition.
        match self
            .kv
            .cas(BUCKET_ITEMS, &id.0, Bytes::from(body), Some(rev))
            .await
        {
            Ok(_) => {}
            Err(BusError::CasConflict { .. }) => return Ok(false),
            Err(e) => {
                return Err(WorkLogError::Storage {
                    message: e.to_string(),
                    source: Some(Box::new(e)),
                })
            }
        }
        self.update_status_index(&id, from, to).await?;

        self.emit(OpsEvent::WorkTransition {
            tenant: updated.tenant.clone(),
            work_id: id.0.clone(),
            from: format!("{from:?}").to_lowercase(),
            to: format!("{to:?}").to_lowercase(),
            reason: None,
        })
        .await;

        if to == WorkStatus::Done {
            self.cascade_ready_children(&id).await?;
        }
        Ok(true)
    }

    async fn get(&self, id: WorkId) -> Option<WorkItem> {
        match self.fetch_item(&id).await {
            Ok(opt) => opt,
            Err(err) => {
                tracing::warn!(
                    target: "klieo.ops.worklog",
                    work_id = %id.0,
                    error = %err,
                    "WorkLog::get fetch failed; returning None which is indistinguishable from missing"
                );
                None
            }
        }
    }

    async fn list_by_status(&self, filter: WorkStatus, limit: usize) -> Vec<WorkItem> {
        let ids = match self.fetch_status_index(filter).await {
            Ok((set, _)) => set,
            Err(e) => {
                tracing::warn!(target: "klieo.ops.worklog", error = %e, "list_by_status index read failed");
                return vec![];
            }
        };
        let fetches: Vec<_> = ids
            .into_iter()
            .take(limit)
            .map(|id| async move { self.fetch_item(&id).await })
            .collect();
        let results = match try_join_all(fetches).await {
            Ok(items) => items,
            Err(err) => {
                tracing::warn!(
                    target: "klieo.ops.worklog",
                    error = %err,
                    "list_by_status fetch_item failed; returning partial results"
                );
                return Vec::new();
            }
        };
        results.into_iter().flatten().collect()
    }

    async fn dag(&self, root: WorkId) -> WorkDag {
        let mut visited: HashSet<WorkId> = HashSet::new();
        let mut queue: VecDeque<WorkId> = VecDeque::new();
        queue.push_back(root);
        let mut out: Vec<WorkItem> = Vec::new();
        while let Some(node) = queue.pop_front() {
            if !visited.insert(node.clone()) {
                continue;
            }
            match self.fetch_item(&node).await {
                Ok(Some(item)) => out.push(item),
                Ok(None) => {}
                Err(e) => {
                    tracing::warn!(target: "klieo.ops.worklog", error = %e, "dag fetch failed");
                }
            }
            match self.fetch_children(&node).await {
                Ok(children) => {
                    for c in children {
                        queue.push_back(c);
                    }
                }
                Err(e) => {
                    tracing::warn!(target: "klieo.ops.worklog", error = %e, "dag children fetch failed")
                }
            }
            match self.fetch_parents(&node).await {
                Ok(parents) => {
                    for p in parents {
                        queue.push_back(p);
                    }
                }
                Err(e) => {
                    tracing::warn!(target: "klieo.ops.worklog", error = %e, "dag parents fetch failed")
                }
            }
        }
        WorkDag { items: out }
    }
}

// ── Cascade helper ────────────────────────────────────────────────────────────

impl KvWorkLog {
    /// When an item transitions to Done, check each child: if all its parents
    /// are Done, promote it from Planned → Ready and persist + announce.
    async fn cascade_ready_children(&self, done_id: &WorkId) -> Result<(), WorkLogError> {
        let children = self.fetch_children(done_id).await?;
        let mut newly_ready: Vec<WorkItem> = Vec::new();

        for child_id in children {
            let child = match self.fetch_item(&child_id).await? {
                Some(c) if c.status == WorkStatus::Planned => c,
                _ => continue,
            };
            if self.dependencies_satisfied(&child_id).await? {
                let mut updated = child;
                updated.status = WorkStatus::Ready;
                updated.last_transition_at = Utc::now().to_rfc3339();
                newly_ready.push(updated);
            }
        }

        // Persist all newly-ready items concurrently.
        let persists = newly_ready.iter().map(|snap| self.persist_item(snap));
        try_join_all(persists).await?;

        // Update status indexes concurrently.
        let index_updates = newly_ready
            .iter()
            .map(|snap| self.update_status_index(&snap.id, WorkStatus::Planned, WorkStatus::Ready));
        try_join_all(index_updates).await?;

        for snap in &newly_ready {
            self.announce_ready(snap.id.clone());
        }
        Ok(())
    }
}

// ── Utility fns ───────────────────────────────────────────────────────────────

/// Return the status prefix string for a given `WorkStatus` variant.
fn status_prefix(status: WorkStatus) -> &'static str {
    match status {
        WorkStatus::Planned => "planned",
        WorkStatus::Ready => "ready",
        WorkStatus::InProgress => "in_progress",
        WorkStatus::Done => "done",
        WorkStatus::Failed => "failed",
        WorkStatus::AwaitingApproval => "awaiting_approval",
        WorkStatus::Cancelled => "cancelled",
    }
}

/// Compute the shard index for a `WorkId` by hashing its string representation
/// modulo [`STATUS_INDEX_SHARDS`].
///
/// ## Stability contract
///
/// `SipHasher13` with fixed keys guarantees the same shard assignment across
/// Rust releases. `std::collections::hash_map::DefaultHasher` explicitly
/// provides no such guarantee (stdlib docs: "should not be relied upon over
/// releases"). The keys below are chosen to be human-readable in hex dumps:
///
/// - `k0 = 0x4b6c_6965_6f31_3300` — ASCII "klieo13\0"
/// - `k1 = 0x7368_6172_645f_6964` — ASCII "shard_id"
///
/// **Do not change these constants.** Doing so silently re-shards every
/// persisted `WorkId`, causing items to be visible in the wrong shard and
/// breaking `ready()` / `list_by_status()` on existing KV state.
fn shard_for_id(id: &WorkId) -> usize {
    use std::hash::{Hash, Hasher};
    let mut hasher = siphasher::sip::SipHasher13::new_with_keys(
        0x4b6c_6965_6f31_3300, // "klieo13\0" — readable in hex dumps
        0x7368_6172_645f_6964, // "shard_id"
    );
    id.0.hash(&mut hasher);
    (hasher.finish() % STATUS_INDEX_SHARDS as u64) as usize
}

/// Build the sharded KV key for a status index entry:
/// `<status_prefix>/shard_<N>`.
fn status_shard_key(status: WorkStatus, shard: usize) -> String {
    format!("{}/shard_{shard}", status_prefix(status))
}

fn encode_id_set(set: &HashSet<WorkId>) -> Result<Bytes, WorkLogError> {
    let body = serde_json::to_vec(set)
        .map_err(|e| WorkLogError::Internal(format!("serialise id set: {e}")))?;
    Ok(Bytes::from(body))
}

async fn backoff_delay(attempt: u32) {
    let ms = 5u64 * (1 << attempt.min(6));
    tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
}

#[cfg(test)]
mod tests {
    use super::*;

    use klieo_bus_memory::MemoryBus;
    use klieo_core::{KvEntry, Lease, Revision};
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::Duration;

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

    /// `transition_if_status` from a terminal status to a different status is a
    /// rejected transition.
    #[tokio::test]
    async fn transition_if_status_rejects_terminal_from() {
        let bus = MemoryBus::new();
        let wl = KvWorkLog::new(bus.kv.clone());
        let id = wl.plan(stale_item("done")).await.expect("plan");
        wl.transition(id.clone(), WorkStatus::Done)
            .await
            .expect("to done");

        let err = wl
            .transition_if_status(id, WorkStatus::Done, WorkStatus::Ready)
            .await
            .expect_err("terminal -> Ready must be rejected");
        assert!(matches!(err, WorkLogError::Internal(_)), "got {err:?}");
    }

    /// `transition_if_status` on an unknown id is `UnknownWorkItem`, not a
    /// silent `Ok(false)`.
    #[tokio::test]
    async fn transition_if_status_unknown_item_errors() {
        let bus = MemoryBus::new();
        let wl = KvWorkLog::new(bus.kv.clone());
        let err = wl
            .transition_if_status(
                WorkId("nope".into()),
                WorkStatus::InProgress,
                WorkStatus::Ready,
            )
            .await
            .expect_err("unknown id must error");
        assert!(
            matches!(err, WorkLogError::UnknownWorkItem(_)),
            "got {err:?}"
        );
    }

    /// Wraps a `KvStore`, injecting exactly one `CasConflict` on the next `cas`
    /// to the items bucket — simulating a concurrent writer landing between a
    /// `transition_if_status` read and its CAS write.
    struct CasConflictOnceKv {
        inner: Arc<dyn KvStore>,
        armed: AtomicBool,
    }

    #[async_trait]
    impl KvStore for CasConflictOnceKv {
        async fn get(&self, bucket: &str, key: &str) -> Result<Option<KvEntry>, BusError> {
            self.inner.get(bucket, key).await
        }
        async fn put(&self, bucket: &str, key: &str, value: Bytes) -> Result<Revision, BusError> {
            self.inner.put(bucket, key, value).await
        }
        async fn cas(
            &self,
            bucket: &str,
            key: &str,
            value: Bytes,
            expected: Option<Revision>,
        ) -> Result<Revision, BusError> {
            if bucket == BUCKET_ITEMS && self.armed.swap(false, Ordering::SeqCst) {
                return Err(BusError::CasConflict {
                    expected: expected.unwrap_or(0),
                    actual: expected.unwrap_or(0) + 1,
                });
            }
            self.inner.cas(bucket, key, value, expected).await
        }
        async fn delete(&self, bucket: &str, key: &str) -> Result<(), BusError> {
            self.inner.delete(bucket, key).await
        }
        async fn lease(&self, bucket: &str, key: &str, ttl: Duration) -> Result<Lease, BusError> {
            self.inner.lease(bucket, key, ttl).await
        }
        async fn keys(&self, bucket: &str) -> Result<Vec<String>, BusError> {
            self.inner.keys(bucket).await
        }
    }

    /// When the item write loses the CAS race (revision moved between read and
    /// write), `transition_if_status` returns `Ok(false)` — no double-dispatch.
    #[tokio::test]
    async fn transition_if_status_returns_false_on_cas_conflict() {
        let bus = MemoryBus::new();
        // Seed the item via the plain bus so plan's writes succeed un-trapped.
        let wl_seed = KvWorkLog::new(bus.kv.clone());
        let id = wl_seed.plan(stale_item("contended")).await.expect("plan");
        wl_seed.dispatch(id.clone()).await.expect("dispatch"); // -> InProgress

        let trap = Arc::new(CasConflictOnceKv {
            inner: bus.kv.clone(),
            armed: AtomicBool::new(true),
        });
        let wl = KvWorkLog::new(trap);

        let applied = wl
            .transition_if_status(id.clone(), WorkStatus::InProgress, WorkStatus::Ready)
            .await
            .expect("cas conflict is Ok(false), not Err");
        assert!(!applied, "a lost CAS race must not transition the item");
        // Status is unchanged — still InProgress.
        assert_eq!(
            wl_seed.get(id).await.unwrap().status,
            WorkStatus::InProgress
        );
    }

    /// Shard distribution sanity: 1000 synthetic WorkIds must hit at least
    /// 12 of 16 shards, confirming the SipHasher13 keys produce a reasonably
    /// uniform distribution across the shard space.
    #[test]
    fn shard_distribution_covers_at_least_12_of_16() {
        let mut hit_shards = std::collections::HashSet::new();
        for i in 0..1000usize {
            let id = WorkId(format!("wrk_{i}"));
            hit_shards.insert(shard_for_id(&id));
        }
        assert!(
            hit_shards.len() >= 12,
            "expected >= 12 shards hit out of 16; got {} — hash distribution may be degenerate",
            hit_shards.len()
        );
    }
}