enact-core 0.0.2

//! Inbox Store - Storage for inbox messages
//!
//! ## Invariants
//!
//! - **INV-INBOX-002**: Control messages (pause/cancel) MUST be processed first
//!   - Implemented via `priority_order()` sorting in `drain_messages()`
//! - **INV-INBOX-004**: Messages are scoped to a specific ExecutionId
//!   - All operations require ExecutionId parameter
//!
//! @see docs/TECHNICAL/31-MID-EXECUTION-GUIDANCE.md

use super::message::InboxMessage;
use crate::kernel::ExecutionId;
use std::collections::HashMap;
use std::sync::{Arc, RwLock};

/// InboxStore trait - async storage for inbox messages
///
/// Implementations must ensure thread-safety for concurrent access.
pub trait InboxStore: Send + Sync {
    /// Push a message to the inbox for a specific execution
    ///
    /// ## Arguments
    /// * `execution_id` - Target execution (INV-INBOX-004)
    /// * `message` - The message to push
    fn push(&self, execution_id: &ExecutionId, message: InboxMessage);

    /// Get the number of pending messages for an execution
    fn len(&self, execution_id: &ExecutionId) -> usize;

    /// Check if there are any pending messages
    fn is_empty(&self, execution_id: &ExecutionId) -> bool {
        self.len(execution_id) == 0
    }

    /// Check if there are any control messages (highest priority)
    ///
    /// Used for fast-path cancellation/pause checks without draining.
    fn has_control_messages(&self, execution_id: &ExecutionId) -> bool;

    /// Drain all messages for an execution, sorted by priority
    ///
    /// ## Invariant INV-INBOX-002
    /// Messages are returned sorted by priority_order():
    /// 1. Control (pause/resume/cancel) - highest
    /// 2. Evidence (contradicts_plan)
    /// 3. Evidence (other)
    /// 4. Guidance (high priority)
    /// 5. Guidance (other)
    /// 6. A2A - lowest
    fn drain_messages(&self, execution_id: &ExecutionId) -> Vec<InboxMessage>;

    /// Peek at the next message without removing it
    fn peek(&self, execution_id: &ExecutionId) -> Option<InboxMessage>;

    /// Pop the highest-priority message
    fn pop(&self, execution_id: &ExecutionId) -> Option<InboxMessage>;

    /// Clear all messages for an execution
    fn clear(&self, execution_id: &ExecutionId);
}

/// In-memory inbox store implementation
///
/// Thread-safe storage using RwLock. Suitable for single-node deployments.
/// For distributed deployments, use Redis-backed implementation.
#[derive(Default)]
pub struct InMemoryInboxStore {
    /// Messages keyed by ExecutionId
    messages: RwLock<HashMap<String, Vec<InboxMessage>>>,
}

impl InMemoryInboxStore {
    /// Create a new empty inbox store
    pub fn new() -> Self {
        Self {
            messages: RwLock::new(HashMap::new()),
        }
    }

    /// Create an Arc-wrapped instance for sharing
    pub fn shared() -> Arc<Self> {
        Arc::new(Self::new())
    }
}

impl InboxStore for InMemoryInboxStore {
    fn push(&self, execution_id: &ExecutionId, message: InboxMessage) {
        let mut guard = self.messages.write().expect("lock poisoned");
        guard
            .entry(execution_id.to_string())
            .or_default()
            .push(message);
    }

    fn len(&self, execution_id: &ExecutionId) -> usize {
        let guard = self.messages.read().expect("lock poisoned");
        guard
            .get(&execution_id.to_string())
            .map(|v| v.len())
            .unwrap_or(0)
    }

    fn has_control_messages(&self, execution_id: &ExecutionId) -> bool {
        let guard = self.messages.read().expect("lock poisoned");
        guard
            .get(&execution_id.to_string())
            .map(|v| v.iter().any(|m| m.is_control()))
            .unwrap_or(false)
    }

    fn drain_messages(&self, execution_id: &ExecutionId) -> Vec<InboxMessage> {
        let mut guard = self.messages.write().expect("lock poisoned");
        let mut messages = guard.remove(&execution_id.to_string()).unwrap_or_default();

        // INV-INBOX-002: Sort by priority (control messages first)
        messages.sort_by_key(|m| m.priority_order());

        messages
    }

    fn peek(&self, execution_id: &ExecutionId) -> Option<InboxMessage> {
        let guard = self.messages.read().expect("lock poisoned");
        guard.get(&execution_id.to_string()).and_then(|v| {
            // Return highest priority message
            v.iter().min_by_key(|m| m.priority_order()).cloned()
        })
    }

    fn pop(&self, execution_id: &ExecutionId) -> Option<InboxMessage> {
        let mut guard = self.messages.write().expect("lock poisoned");
        let messages = guard.get_mut(&execution_id.to_string())?;

        if messages.is_empty() {
            return None;
        }

        // Find index of highest priority message (INV-INBOX-002)
        let min_idx = messages
            .iter()
            .enumerate()
            .min_by_key(|(_, m)| m.priority_order())
            .map(|(i, _)| i)?;

        Some(messages.remove(min_idx))
    }

    fn clear(&self, execution_id: &ExecutionId) {
        let mut guard = self.messages.write().expect("lock poisoned");
        guard.remove(&execution_id.to_string());
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::inbox::message::{
        ControlAction, ControlMessage, EvidenceImpact, EvidenceSource, EvidenceUpdate,
        GuidanceMessage,
    };

    fn test_execution_id() -> ExecutionId {
        ExecutionId::new()
    }

    #[test]
    fn test_push_and_drain() {
        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        // Push messages
        let guidance =
            InboxMessage::Guidance(GuidanceMessage::from_user(exec_id.clone(), "Focus on EU"));
        store.push(&exec_id, guidance);

        assert_eq!(store.len(&exec_id), 1);
        assert!(!store.is_empty(&exec_id));

        // Drain
        let messages = store.drain_messages(&exec_id);
        assert_eq!(messages.len(), 1);
        assert!(store.is_empty(&exec_id));
    }

    #[test]
    fn test_priority_sorting_inv_inbox_002() {
        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        // Push in reverse priority order
        let guidance =
            InboxMessage::Guidance(GuidanceMessage::from_user(exec_id.clone(), "low priority"));
        let evidence = InboxMessage::Evidence(EvidenceUpdate::new(
            exec_id.clone(),
            EvidenceSource::Discovery,
            "Found something",
            serde_json::json!({}),
            EvidenceImpact::Informational,
        ));
        let control = InboxMessage::Control(ControlMessage::new(
            exec_id.clone(),
            ControlAction::Pause,
            "admin",
        ));

        store.push(&exec_id, guidance);
        store.push(&exec_id, evidence);
        store.push(&exec_id, control);

        // Drain should return sorted by priority
        let messages = store.drain_messages(&exec_id);
        assert_eq!(messages.len(), 3);

        // INV-INBOX-002: Control first
        assert!(messages[0].is_control());
        assert!(matches!(messages[1], InboxMessage::Evidence(_)));
        assert!(matches!(messages[2], InboxMessage::Guidance(_)));
    }

    #[test]
    fn test_has_control_messages() {
        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        // No messages
        assert!(!store.has_control_messages(&exec_id));

        // Add guidance only
        let guidance = InboxMessage::Guidance(GuidanceMessage::from_user(exec_id.clone(), "test"));
        store.push(&exec_id, guidance);
        assert!(!store.has_control_messages(&exec_id));

        // Add control message
        let control = InboxMessage::Control(ControlMessage::new(
            exec_id.clone(),
            ControlAction::Cancel,
            "admin",
        ));
        store.push(&exec_id, control);
        assert!(store.has_control_messages(&exec_id));
    }

    #[test]
    fn test_pop_returns_highest_priority() {
        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        // Push low priority first, then high priority
        let guidance = InboxMessage::Guidance(GuidanceMessage::from_user(exec_id.clone(), "low"));
        let control = InboxMessage::Control(ControlMessage::new(
            exec_id.clone(),
            ControlAction::Pause,
            "admin",
        ));

        store.push(&exec_id, guidance);
        store.push(&exec_id, control);

        // Pop should return control first (highest priority)
        let msg = store.pop(&exec_id).unwrap();
        assert!(msg.is_control());

        // Next pop returns guidance
        let msg = store.pop(&exec_id).unwrap();
        assert!(matches!(msg, InboxMessage::Guidance(_)));

        // Empty now
        assert!(store.pop(&exec_id).is_none());
    }

    #[test]
    fn test_peek_does_not_remove() {
        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        let control = InboxMessage::Control(ControlMessage::new(
            exec_id.clone(),
            ControlAction::Pause,
            "admin",
        ));
        store.push(&exec_id, control);

        // Peek multiple times - message remains
        let msg1 = store.peek(&exec_id);
        let msg2 = store.peek(&exec_id);

        assert!(msg1.is_some());
        assert!(msg2.is_some());
        assert_eq!(store.len(&exec_id), 1);
    }

    #[test]
    fn test_execution_isolation_inv_inbox_004() {
        let store = InMemoryInboxStore::new();
        let exec_id_1 = test_execution_id();
        let exec_id_2 = test_execution_id();

        // Push to different executions
        let control_1 = InboxMessage::Control(ControlMessage::new(
            exec_id_1.clone(),
            ControlAction::Pause,
            "admin",
        ));
        let control_2 = InboxMessage::Control(ControlMessage::new(
            exec_id_2.clone(),
            ControlAction::Cancel,
            "admin",
        ));

        store.push(&exec_id_1, control_1);
        store.push(&exec_id_2, control_2);

        // Each execution has its own messages
        assert_eq!(store.len(&exec_id_1), 1);
        assert_eq!(store.len(&exec_id_2), 1);

        // Drain exec_1 doesn't affect exec_2
        let msgs = store.drain_messages(&exec_id_1);
        assert_eq!(msgs.len(), 1);
        assert_eq!(store.len(&exec_id_1), 0);
        assert_eq!(store.len(&exec_id_2), 1);
    }

    #[test]
    fn test_clear() {
        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        for _ in 0..5 {
            let control = InboxMessage::Control(ControlMessage::new(
                exec_id.clone(),
                ControlAction::Pause,
                "admin",
            ));
            store.push(&exec_id, control);
        }

        assert_eq!(store.len(&exec_id), 5);

        store.clear(&exec_id);

        assert_eq!(store.len(&exec_id), 0);
        assert!(store.is_empty(&exec_id));
    }

    #[test]
    fn test_inbox_drain_ordering() {
        // Verify messages are drained in FIFO order within the same priority
        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        // Push 3 guidance messages in order: msg1, msg2, msg3
        // All have the same priority so FIFO should be preserved
        let msg1 = InboxMessage::Guidance(GuidanceMessage::from_user(exec_id.clone(), "first"));
        let msg2 = InboxMessage::Guidance(GuidanceMessage::from_user(exec_id.clone(), "second"));
        let msg3 = InboxMessage::Guidance(GuidanceMessage::from_user(exec_id.clone(), "third"));

        // Store IDs to verify order
        let id1 = msg1.id().to_string();
        let id2 = msg2.id().to_string();
        let id3 = msg3.id().to_string();

        store.push(&exec_id, msg1);
        store.push(&exec_id, msg2);
        store.push(&exec_id, msg3);

        // Drain and verify they come out in same order (FIFO within priority)
        let messages = store.drain_messages(&exec_id);
        assert_eq!(messages.len(), 3);

        // Since stable sort is used and all have same priority, FIFO is preserved
        assert_eq!(messages[0].id(), id1);
        assert_eq!(messages[1].id(), id2);
        assert_eq!(messages[2].id(), id3);
    }

    #[test]
    fn test_control_priority() {
        // Verify control messages (Cancel, Pause) are processed before other messages
        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        // Push a guidance message first
        let guidance =
            InboxMessage::Guidance(GuidanceMessage::from_user(exec_id.clone(), "low priority"));
        store.push(&exec_id, guidance);

        // Push a control message second
        let control = InboxMessage::Control(ControlMessage::new(
            exec_id.clone(),
            ControlAction::Cancel,
            "admin",
        ));
        let control_id = control.id().to_string();
        store.push(&exec_id, control);

        // Drain - control should come first despite being pushed second
        let messages = store.drain_messages(&exec_id);
        assert_eq!(messages.len(), 2);
        assert!(messages[0].is_control());
        assert_eq!(messages[0].id(), control_id);
        assert!(matches!(messages[1], InboxMessage::Guidance(_)));
    }

    #[test]
    fn test_inbox_scoped_to_execution() {
        // Verify messages for one execution don't leak to another (INV-INBOX-004)
        let store = InMemoryInboxStore::new();
        let exec_id_1 = test_execution_id();
        let exec_id_2 = test_execution_id();

        // Push message to exec_id_1 only
        let guidance = InboxMessage::Guidance(GuidanceMessage::from_user(
            exec_id_1.clone(),
            "message for exec1",
        ));
        let msg_id = guidance.id().to_string();
        store.push(&exec_id_1, guidance);

        // Drain exec_id_2 - should be empty
        let messages_2 = store.drain_messages(&exec_id_2);
        assert!(messages_2.is_empty(), "exec_id_2 should have no messages");

        // Drain exec_id_1 - should have the message
        let messages_1 = store.drain_messages(&exec_id_1);
        assert_eq!(messages_1.len(), 1);
        assert_eq!(messages_1[0].id(), msg_id);
    }

    #[test]
    fn test_pause_resume_execution() {
        // Test pause/resume control flow
        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        // Push pause control message
        let pause = InboxMessage::Control(ControlMessage::new(
            exec_id.clone(),
            ControlAction::Pause,
            "admin",
        ));
        store.push(&exec_id, pause);

        // Verify has_control_messages returns true
        assert!(
            store.has_control_messages(&exec_id),
            "should have control messages after push"
        );

        // Drain and verify it's a pause
        let messages = store.drain_messages(&exec_id);
        assert_eq!(messages.len(), 1);
        if let InboxMessage::Control(ctrl) = &messages[0] {
            assert_eq!(ctrl.action, ControlAction::Pause);
        } else {
            panic!("expected Control message");
        }

        // Push resume message
        let resume = InboxMessage::Control(ControlMessage::new(
            exec_id.clone(),
            ControlAction::Resume,
            "admin",
        ));
        store.push(&exec_id, resume);

        // Drain and verify it's a resume
        let messages = store.drain_messages(&exec_id);
        assert_eq!(messages.len(), 1);
        if let InboxMessage::Control(ctrl) = &messages[0] {
            assert_eq!(ctrl.action, ControlAction::Resume);
        } else {
            panic!("expected Control message");
        }
    }

    #[test]
    fn test_cancel_long_running() {
        // Test cancel control message with reason
        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        // Push cancel control message with reason
        let cancel_reason = "Execution timed out after 30 minutes";
        let cancel = InboxMessage::Control(
            ControlMessage::new(exec_id.clone(), ControlAction::Cancel, "system")
                .with_reason(cancel_reason),
        );
        store.push(&exec_id, cancel);

        // Drain and verify
        let messages = store.drain_messages(&exec_id);
        assert_eq!(messages.len(), 1);

        // Verify it's a Control message
        assert!(messages[0].is_control(), "should be a Control message");

        // Verify action is Cancel and reason is preserved
        if let InboxMessage::Control(ctrl) = &messages[0] {
            assert_eq!(ctrl.action, ControlAction::Cancel);
            assert_eq!(
                ctrl.reason.as_deref(),
                Some(cancel_reason),
                "reason should be preserved"
            );
            assert_eq!(ctrl.actor, "system");
        } else {
            panic!("expected Control message");
        }
    }

    #[test]
    fn test_approval_flow_hitl() {
        // Test Human-in-the-Loop plan approval/rejection flow
        // This verifies the HITL integration where ApprovePlan/RejectPlan
        // sends guidance messages to the inbox with high priority
        use crate::inbox::message::GuidancePriority;

        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        // Simulate ApprovePlan: push PLAN_APPROVED guidance with high priority
        let approval_msg = GuidanceMessage::from_user(
            exec_id.clone(),
            "PLAN_APPROVED: User approved the proposed plan. Proceed with execution.",
        )
        .with_priority(GuidancePriority::High);
        let approval_id = approval_msg.id.clone();
        store.push(&exec_id, InboxMessage::Guidance(approval_msg));

        // Verify message is in inbox
        assert_eq!(store.len(&exec_id), 1);
        assert!(
            !store.has_control_messages(&exec_id),
            "guidance is not a control message"
        );

        // Drain and verify approval message content
        let messages = store.drain_messages(&exec_id);
        assert_eq!(messages.len(), 1);

        if let InboxMessage::Guidance(g) = &messages[0] {
            assert_eq!(g.id, approval_id);
            assert!(g.content.contains("PLAN_APPROVED"));
            assert_eq!(g.priority, GuidancePriority::High);
            assert_eq!(g.from, crate::inbox::message::GuidanceSource::User);
        } else {
            panic!("expected Guidance message for approval");
        }

        // Test rejection flow
        let rejection_msg = GuidanceMessage::from_user(
            exec_id.clone(),
            "PLAN_REJECTED: User rejected the plan. Reason: Need more details on approach.",
        )
        .with_priority(GuidancePriority::High);
        let rejection_id = rejection_msg.id.clone();
        store.push(&exec_id, InboxMessage::Guidance(rejection_msg));

        let messages = store.drain_messages(&exec_id);
        assert_eq!(messages.len(), 1);

        if let InboxMessage::Guidance(g) = &messages[0] {
            assert_eq!(g.id, rejection_id);
            assert!(g.content.contains("PLAN_REJECTED"));
            assert_eq!(g.priority, GuidancePriority::High);
        } else {
            panic!("expected Guidance message for rejection");
        }
    }

    #[test]
    fn test_hitl_priority_ordering() {
        // Verify high-priority HITL guidance is processed before normal guidance
        // but after control messages (INV-INBOX-002)
        use crate::inbox::message::GuidancePriority;

        let store = InMemoryInboxStore::new();
        let exec_id = test_execution_id();

        // Push messages in reverse priority order
        // 1. Normal guidance (lowest)
        let normal_guidance = InboxMessage::Guidance(GuidanceMessage::from_user(
            exec_id.clone(),
            "normal guidance",
        ));

        // 2. High priority HITL approval (higher)
        let hitl_approval = InboxMessage::Guidance(
            GuidanceMessage::from_user(exec_id.clone(), "PLAN_APPROVED")
                .with_priority(GuidancePriority::High),
        );

        // 3. Control message (highest)
        let control = InboxMessage::Control(ControlMessage::new(
            exec_id.clone(),
            ControlAction::Pause,
            "admin",
        ));

        // Push in reverse order
        store.push(&exec_id, normal_guidance);
        store.push(&exec_id, hitl_approval);
        store.push(&exec_id, control);

        // Drain - should be sorted by priority
        let messages = store.drain_messages(&exec_id);
        assert_eq!(messages.len(), 3);

        // INV-INBOX-002: Control first
        assert!(messages[0].is_control(), "control message should be first");

        // High priority guidance second
        if let InboxMessage::Guidance(g) = &messages[1] {
            assert!(
                g.content.contains("PLAN_APPROVED"),
                "high priority HITL should be second"
            );
            assert_eq!(g.priority, GuidancePriority::High);
        } else {
            panic!("expected high priority guidance second");
        }

        // Normal guidance last
        if let InboxMessage::Guidance(g) = &messages[2] {
            assert!(
                g.content.contains("normal guidance"),
                "normal guidance should be last"
            );
            assert_eq!(g.priority, GuidancePriority::Medium);
        } else {
            panic!("expected normal guidance last");
        }
    }
}