meerkat-core 0.8.12

//! Canonical completion delivery seam for async operation terminals.
//!
//! The [`CompletionFeed`] trait provides a monotonic, append-only read handle
//! over operation terminal events. The runtime registry is the sole writer;
//! consumers (agent boundary, idle wake) are read-only.
//!
//! This replaces the prior parallel-truth paths (shell job projection and
//! detached wake booleans).

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;

use crate::ops::OperationId;
use crate::ops_lifecycle::{OperationKind, OperationTerminalOutcome};

/// Monotonic sequence number assigned to each completion event.
pub type CompletionSeq = u64;

/// A single completion event in the feed.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct CompletionEntry {
    pub seq: CompletionSeq,
    pub operation_id: OperationId,
    pub kind: OperationKind,
    pub display_name: String,
    pub terminal_outcome: OperationTerminalOutcome,
    pub completed_at_ms: Option<u64>,
}

/// A batch of completion entries with an atomically-captured watermark.
///
/// The `watermark` is captured at the same instant as the entry snapshot,
/// ensuring no completions land between the read and the watermark check.
#[derive(Debug, Clone)]
pub struct CompletionBatch {
    pub entries: Vec<CompletionEntry>,
    /// Feed watermark at the time entries were read.
    pub watermark: CompletionSeq,
    /// Highest sequence that has been evicted from the bounded read window.
    ///
    /// A consumer whose cursor is lower than this value has an explicit gap:
    /// the feed can no longer prove whether one of the unavailable entries
    /// carried an obligation for that consumer.
    pub dropped_through: CompletionSeq,
}

/// Typed proof that a completion-feed cursor fell behind bounded retention.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CompletionFeedGap {
    pub requested_after_seq: CompletionSeq,
    pub dropped_through: CompletionSeq,
    pub watermark: CompletionSeq,
}

impl CompletionBatch {
    /// Return a typed retention gap when `after_seq` predates evicted entries.
    #[must_use]
    pub fn gap_after(&self, after_seq: CompletionSeq) -> Option<CompletionFeedGap> {
        (after_seq < self.dropped_through).then_some(CompletionFeedGap {
            requested_after_seq: after_seq,
            dropped_through: self.dropped_through,
            watermark: self.watermark,
        })
    }
}

/// Read-only handle to the canonical completion feed.
///
/// Consumers call [`list_since`](CompletionFeed::list_since) with a
/// machine-authorized cursor, and
/// [`wait_for_advance`](CompletionFeed::wait_for_advance) to block until the
/// watermark advances past a given sequence. The feed watermark is not cursor
/// authority for delivery consumers.
pub trait CompletionFeed: Send + Sync + std::fmt::Debug {
    /// Current feed watermark (highest seq written).
    fn watermark(&self) -> CompletionSeq;

    /// Return all entries with `seq > after_seq`, plus the current watermark.
    ///
    /// The watermark in the returned batch is captured atomically with the
    /// entry snapshot — no entry can land between the snapshot and the
    /// watermark read.
    fn list_since(&self, after_seq: CompletionSeq) -> CompletionBatch;

    /// Block until the feed watermark advances past `after_seq`.
    ///
    /// Returns the new watermark. If the watermark is already past `after_seq`,
    /// returns immediately.
    fn wait_for_advance(
        &self,
        after_seq: CompletionSeq,
    ) -> Pin<Box<dyn Future<Output = CompletionSeq> + Send + '_>>;
}

// ---------------------------------------------------------------------------
// Enrichment provider for agent-boundary completion projection
// ---------------------------------------------------------------------------

/// Shell-level enrichment for a completed operation.
///
/// Carries display details (job ID, detail string) that the agent boundary
/// needs for `BackgroundJobCompleted` events. These come from the shell
/// `JobManager`, not from the feed itself.
#[derive(Debug, Clone)]
pub struct CompletionEnrichmentData {
    pub job_id: String,
    pub detail: String,
}

/// Result of a best-effort completion-enrichment lookup.
///
/// `Busy` is intentionally distinct from `Missing`: a consumer may fall back
/// for a genuinely missing process-local record, but must retry a contended
/// lookup before advancing its durable delivery cursor.
#[derive(Debug, Clone)]
pub enum CompletionEnrichment {
    Found(CompletionEnrichmentData),
    Missing,
    Busy,
}

/// Provider of shell-level enrichment for completed operations.
///
/// The shell `JobManager` implements this trait. The agent boundary
/// calls [`enrich`](CompletionEnrichmentProvider::enrich) to look up
/// display details by operation ID.
pub trait CompletionEnrichmentProvider: Send + Sync {
    fn enrich(&self, operation_id: &OperationId) -> CompletionEnrichment;
}

// ---------------------------------------------------------------------------
// Arc delegation
// ---------------------------------------------------------------------------

impl<T: CompletionFeed + ?Sized> CompletionFeed for Arc<T> {
    fn watermark(&self) -> CompletionSeq {
        (**self).watermark()
    }

    fn list_since(&self, after_seq: CompletionSeq) -> CompletionBatch {
        (**self).list_since(after_seq)
    }

    fn wait_for_advance(
        &self,
        after_seq: CompletionSeq,
    ) -> Pin<Box<dyn Future<Output = CompletionSeq> + Send + '_>> {
        (**self).wait_for_advance(after_seq)
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, dead_code)]
pub(crate) mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU64, Ordering};

    /// Simple in-memory mock for testing feed-dependent code paths.
    #[derive(Debug)]
    pub struct MockCompletionFeed {
        watermark: AtomicU64,
        entries: std::sync::Mutex<Vec<CompletionEntry>>,
    }

    impl MockCompletionFeed {
        pub fn new() -> Self {
            Self {
                watermark: AtomicU64::new(0),
                entries: std::sync::Mutex::new(Vec::new()),
            }
        }

        /// Create a feed pre-seeded at a given watermark (simulates prior activity).
        pub fn with_watermark(watermark: CompletionSeq) -> Self {
            Self {
                watermark: AtomicU64::new(watermark),
                entries: std::sync::Mutex::new(Vec::new()),
            }
        }

        pub fn push(&self, entry: CompletionEntry) {
            let mut entries = self.entries.lock().unwrap();
            self.watermark.store(entry.seq, Ordering::Release);
            entries.push(entry);
        }
    }

    impl CompletionFeed for MockCompletionFeed {
        fn watermark(&self) -> CompletionSeq {
            self.watermark.load(Ordering::Acquire)
        }

        fn list_since(&self, after_seq: CompletionSeq) -> CompletionBatch {
            let entries = self.entries.lock().unwrap();
            CompletionBatch {
                entries: entries
                    .iter()
                    .filter(|e| e.seq > after_seq)
                    .cloned()
                    .collect(),
                watermark: self.watermark.load(Ordering::Acquire),
                dropped_through: 0,
            }
        }

        fn wait_for_advance(
            &self,
            _after_seq: CompletionSeq,
        ) -> Pin<Box<dyn Future<Output = CompletionSeq> + Send + '_>> {
            Box::pin(async move { self.watermark.load(Ordering::Acquire) })
        }
    }

    #[test]
    fn mock_feed_with_watermark_seeds_at_expected_value() {
        let feed = MockCompletionFeed::with_watermark(42);
        assert_eq!(feed.watermark(), 42);
        assert!(feed.list_since(0).entries.is_empty());
    }

    #[test]
    fn completion_batch_reports_only_real_retention_gaps() {
        let batch = CompletionBatch {
            entries: Vec::new(),
            watermark: 9,
            dropped_through: 4,
        };

        assert_eq!(
            batch.gap_after(3),
            Some(CompletionFeedGap {
                requested_after_seq: 3,
                dropped_through: 4,
                watermark: 9,
            })
        );
        assert_eq!(batch.gap_after(4), None);
        assert_eq!(batch.gap_after(8), None);
    }
}