IdeA/crates/application/tests/session_limit_service.rs

//! LS4 — tests unitaires QA du `SessionLimitService` (ARCHITECTURE §21.5),
//! **100 % fakes** des ports : `Clock` fixe, `Scheduler` enregistreur/contrôlable,
//! `EventBus` espion, `AgentResumer` espion/contrôlable.
//!
//! Couvre les trois responsabilités du service :
//! - (a) détection → planification (`on_rate_limited`) : armement + ordre des events ;
//! - (b) exécution de la reprise (`execute_resume`) : prompt, event, propagation d'erreur ;
//! - (c) annulation (`cancel_resume`) : contrat anti-course (cancel `false` ⇒ pas d'event).

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

use async_trait::async_trait;

use application::{AgentResumer, AppError, SessionLimitService, RESUME_PROMPT};
use domain::ids::{AgentId, NodeId, ScheduleId};
use domain::ports::{Clock, EventBus, EventStream, ScheduledTask, Scheduler};
use domain::DomainEvent;
use uuid::Uuid;

fn aid(n: u128) -> AgentId {
    AgentId::from_uuid(Uuid::from_u128(n))
}
fn nid(n: u128) -> NodeId {
    NodeId::from_uuid(Uuid::from_u128(n))
}

// ---------------------------------------------------------------------------
// Fakes des ports
// ---------------------------------------------------------------------------

/// Horloge fixe (déterminisme) : `now` injecté.
struct FixedClock(i64);
impl Clock for FixedClock {
    fn now_millis(&self) -> i64 {
        self.0
    }
}

/// EventBus espion : journalise les events publiés, dans l'ordre.
#[derive(Default, Clone)]
struct SpyBus(Arc<Mutex<Vec<DomainEvent>>>);
impl SpyBus {
    fn events(&self) -> Vec<DomainEvent> {
        self.0.lock().unwrap().clone()
    }
}
impl EventBus for SpyBus {
    fn publish(&self, event: DomainEvent) {
        self.0.lock().unwrap().push(event);
    }
    fn subscribe(&self) -> EventStream {
        Box::new(std::iter::empty())
    }
}

/// Scheduler fake : enregistre chaque `arm` (deadline + task) et chaque `cancel`,
/// distribue des `ScheduleId` déterministes, et rend un résultat de `cancel`
/// contrôlable (pour simuler le cas « déjà tiré »).
#[derive(Clone)]
struct FakeScheduler {
    armed: Arc<Mutex<Vec<(i64, ScheduledTask)>>>,
    issued: Arc<Mutex<Vec<ScheduleId>>>,
    cancels: Arc<Mutex<Vec<ScheduleId>>>,
    next_id: Arc<Mutex<u128>>,
    cancel_result: Arc<AtomicBool>,
}
impl FakeScheduler {
    fn new() -> Self {
        Self {
            armed: Arc::new(Mutex::new(Vec::new())),
            issued: Arc::new(Mutex::new(Vec::new())),
            cancels: Arc::new(Mutex::new(Vec::new())),
            next_id: Arc::new(Mutex::new(1)),
            cancel_result: Arc::new(AtomicBool::new(true)),
        }
    }
    fn set_cancel_result(&self, v: bool) {
        self.cancel_result.store(v, Ordering::SeqCst);
    }
    fn armed(&self) -> Vec<(i64, ScheduledTask)> {
        self.armed.lock().unwrap().clone()
    }
    fn issued(&self) -> Vec<ScheduleId> {
        self.issued.lock().unwrap().clone()
    }
    fn cancels(&self) -> Vec<ScheduleId> {
        self.cancels.lock().unwrap().clone()
    }
}
impl Scheduler for FakeScheduler {
    fn arm(&self, deadline_ms: i64, task: ScheduledTask) -> ScheduleId {
        self.armed.lock().unwrap().push((deadline_ms, task));
        let mut n = self.next_id.lock().unwrap();
        let id = ScheduleId::from_uuid(Uuid::from_u128(*n));
        *n += 1;
        self.issued.lock().unwrap().push(id);
        id
    }
    fn cancel(&self, id: ScheduleId) -> bool {
        self.cancels.lock().unwrap().push(id);
        self.cancel_result.load(Ordering::SeqCst)
    }
}

/// AgentResumer fake : enregistre l'appel `resume` (et le prompt reçu), et peut
/// être configuré pour échouer (propagation d'erreur).
#[derive(Clone)]
struct FakeResumer {
    calls: Arc<Mutex<Vec<(AgentId, NodeId, Option<String>, String)>>>,
    fail: Arc<AtomicBool>,
}
impl FakeResumer {
    fn new() -> Self {
        Self {
            calls: Arc::new(Mutex::new(Vec::new())),
            fail: Arc::new(AtomicBool::new(false)),
        }
    }
    fn set_fail(&self, v: bool) {
        self.fail.store(v, Ordering::SeqCst);
    }
    fn calls(&self) -> Vec<(AgentId, NodeId, Option<String>, String)> {
        self.calls.lock().unwrap().clone()
    }
}
#[async_trait]
impl AgentResumer for FakeResumer {
    async fn resume(
        &self,
        agent_id: AgentId,
        node_id: NodeId,
        conversation_id: Option<String>,
        resume_prompt: &str,
    ) -> Result<(), AppError> {
        self.calls
            .lock()
            .unwrap()
            .push((agent_id, node_id, conversation_id, resume_prompt.to_owned()));
        if self.fail.load(Ordering::SeqCst) {
            Err(AppError::Internal("reprise échouée (fake)".to_owned()))
        } else {
            Ok(())
        }
    }
}

/// Banc d'essai : assemble le service avec ses fakes (et garde les poignées).
struct Env {
    service: SessionLimitService,
    scheduler: FakeScheduler,
    bus: SpyBus,
    resumer: FakeResumer,
}
fn env_at(now_ms: i64) -> Env {
    let scheduler = FakeScheduler::new();
    let bus = SpyBus::default();
    let resumer = FakeResumer::new();
    let service = SessionLimitService::new(
        Arc::new(FixedClock(now_ms)),
        Arc::new(scheduler.clone()),
        Arc::new(bus.clone()),
        Arc::new(resumer.clone()),
    );
    Env {
        service,
        scheduler,
        bus,
        resumer,
    }
}

// ===========================================================================
// (a) Détection → planification
// ===========================================================================

const NOW: i64 = 1_700_000_000_000;

/// `on_rate_limited(Some(reset futur))` ⇒ EXACTEMENT un `arm(fire_at_ms, ResumeAgent{..})`
/// avec `fire_at_ms == reset` (futur) ; events `AgentRateLimited` PUIS `AgentResumeScheduled`
/// dans CET ordre.
#[test]
fn on_rate_limited_future_arms_and_emits_in_order() {
    let env = env_at(NOW);
    let reset = NOW + 60_000;
    env.service
        .on_rate_limited(aid(1), nid(2), Some("conv-1".to_owned()), Some(reset));

    // Exactement un arm, avec la bonne échéance et la bonne tâche.
    let armed = env.scheduler.armed();
    assert_eq!(armed.len(), 1, "exactement un arm");
    assert_eq!(armed[0].0, reset, "fire_at_ms == reset (futur)");
    assert_eq!(
        armed[0].1,
        ScheduledTask::ResumeAgent {
            agent_id: aid(1),
            node_id: nid(2),
            conversation_id: Some("conv-1".to_owned()),
        }
    );

    // Ordre des events : RateLimited puis ResumeScheduled.
    assert_eq!(
        env.bus.events(),
        vec![
            DomainEvent::AgentRateLimited {
                agent_id: aid(1),
                resets_at_ms: Some(reset),
            },
            DomainEvent::AgentResumeScheduled {
                agent_id: aid(1),
                fire_at_ms: reset,
            },
        ]
    );
}

/// Reset DÉJÀ PASSÉ ⇒ `plan_resume` clampe à `now` ⇒ `fire_at_ms == now` (jamais dans
/// le passé) ; le `AgentRateLimited` garde l'heure brute (passée), le `ResumeScheduled`
/// porte le `now` clampé.
#[test]
fn on_rate_limited_past_reset_clamps_fire_at_to_now() {
    let env = env_at(NOW);
    let past = NOW - 60_000;
    env.service
        .on_rate_limited(aid(1), nid(2), None, Some(past));

    let armed = env.scheduler.armed();
    assert_eq!(armed.len(), 1);
    assert_eq!(armed[0].0, NOW, "fire_at_ms clampé à now (jamais le passé)");

    assert_eq!(
        env.bus.events(),
        vec![
            DomainEvent::AgentRateLimited {
                agent_id: aid(1),
                resets_at_ms: Some(past), // l'heure brute (passée) est conservée dans l'event
            },
            DomainEvent::AgentResumeScheduled {
                agent_id: aid(1),
                fire_at_ms: NOW, // clampé
            },
        ]
    );
}

/// `on_rate_limited(None)` ⇒ AUCUN arm ; events `AgentRateLimited{None}` puis
/// `AgentRateLimitSuspected{None}` (filet humain).
#[test]
fn on_rate_limited_without_reset_is_human_fallback_no_arm() {
    let env = env_at(NOW);
    env.service
        .on_rate_limited(aid(1), nid(2), Some("conv-1".to_owned()), None);

    assert!(env.scheduler.armed().is_empty(), "aucun arm sans heure de reset");
    assert!(env.scheduler.cancels().is_empty(), "aucun cancel non plus");
    assert_eq!(
        env.bus.events(),
        vec![
            DomainEvent::AgentRateLimited {
                agent_id: aid(1),
                resets_at_ms: None,
            },
            DomainEvent::AgentRateLimitSuspected {
                agent_id: aid(1),
                resets_at_ms: None,
            },
        ]
    );
}

/// Dédoublonnage (§21.10-4) : deux `on_rate_limited` successifs pour le MÊME agent ⇒
/// l'ancien `ScheduleId` est annulé sur le Scheduler (cancel interne), un nouveau réveil
/// est armé, et AUCUN `AgentResumeCancelled` n'est émis (le dédoublonnage est interne).
#[test]
fn on_rate_limited_twice_same_agent_dedups_cancelling_previous() {
    let env = env_at(NOW);
    let reset1 = NOW + 60_000;
    let reset2 = NOW + 120_000;
    env.service
        .on_rate_limited(aid(1), nid(2), Some("conv-1".to_owned()), Some(reset1));
    env.service
        .on_rate_limited(aid(1), nid(2), Some("conv-1".to_owned()), Some(reset2));

    // Deux arms (un par signal), ids distincts.
    let issued = env.scheduler.issued();
    assert_eq!(issued.len(), 2, "deux arms (rafraîchissement, pas empilement)");
    // Le premier id émis a été annulé par le dédoublonnage du 2ᵉ signal.
    assert_eq!(
        env.scheduler.cancels(),
        vec![issued[0]],
        "l'ancien ScheduleId est cancel-é avant de réarmer"
    );

    // Aucun AgentResumeCancelled : le dédoublonnage est silencieux.
    let cancelled: Vec<_> = env
        .bus
        .events()
        .into_iter()
        .filter(|e| matches!(e, DomainEvent::AgentResumeCancelled { .. }))
        .collect();
    assert!(
        cancelled.is_empty(),
        "le dédoublonnage interne n'émet PAS AgentResumeCancelled"
    );
}

// ===========================================================================
// (b) Exécution de la reprise
// ===========================================================================

/// `execute_resume(ResumeAgent{..})` ⇒ `AgentResumer::resume` appelé avec
/// (agent, node, conv, RESUME_PROMPT) ; `AgentResumed` publié ; l'entrée armée est
/// retirée (un `cancel_resume` ultérieur ⇒ false).
#[tokio::test]
async fn execute_resume_calls_resumer_with_prompt_and_emits_resumed() {
    let env = env_at(NOW);
    // Arme d'abord (pour prouver que l'entrée est ensuite retirée).
    env.service
        .on_rate_limited(aid(1), nid(2), Some("conv-1".to_owned()), Some(NOW + 60_000));

    let task = ScheduledTask::ResumeAgent {
        agent_id: aid(1),
        node_id: nid(2),
        conversation_id: Some("conv-1".to_owned()),
    };
    env.service.execute_resume(task).await.expect("resume ok");

    // Le resumer a vu exactement l'appel attendu, avec le prompt constant.
    assert_eq!(
        env.resumer.calls(),
        vec![(aid(1), nid(2), Some("conv-1".to_owned()), RESUME_PROMPT.to_owned())]
    );

    // AgentResumed publié.
    assert!(
        env.bus
            .events()
            .iter()
            .any(|e| *e == DomainEvent::AgentResumed { agent_id: aid(1) }),
        "AgentResumed doit être publié"
    );

    // L'entrée armée a été retirée ⇒ cancel_resume ultérieur = false.
    assert!(
        !env.service.cancel_resume(aid(1)),
        "après execute_resume, plus rien à annuler"
    );
}

/// `AgentResumer` qui échoue ⇒ l'erreur est propagée ET `AgentResumed` n'est PAS publié.
#[tokio::test]
async fn execute_resume_propagates_error_without_emitting_resumed() {
    let env = env_at(NOW);
    env.resumer.set_fail(true);

    let task = ScheduledTask::ResumeAgent {
        agent_id: aid(1),
        node_id: nid(2),
        conversation_id: None,
    };
    let err = env
        .service
        .execute_resume(task)
        .await
        .expect_err("la reprise doit échouer");
    assert!(matches!(err, AppError::Internal(_)), "erreur propagée: {err:?}");

    assert!(
        !env
            .bus
            .events()
            .iter()
            .any(|e| matches!(e, DomainEvent::AgentResumed { .. })),
        "AgentResumed ne doit PAS être publié si la reprise a échoué"
    );
}

// ===========================================================================
// (c) Annulation
// ===========================================================================

/// `cancel_resume` après un armement, Scheduler renvoyant `true` ⇒ renvoie `true` et
/// publie `AgentResumeCancelled`.
#[test]
fn cancel_resume_after_arm_returns_true_and_emits_cancelled() {
    let env = env_at(NOW);
    env.service
        .on_rate_limited(aid(1), nid(2), Some("conv-1".to_owned()), Some(NOW + 60_000));
    let issued = env.scheduler.issued();

    assert!(env.service.cancel_resume(aid(1)), "cancel d'un réveil armé ⇒ true");
    // Le bon ScheduleId a été passé au Scheduler.
    assert_eq!(env.scheduler.cancels(), vec![issued[0]]);
    // AgentResumeCancelled publié.
    assert!(
        env.bus
            .events()
            .iter()
            .any(|e| *e == DomainEvent::AgentResumeCancelled { agent_id: aid(1) }),
        "AgentResumeCancelled doit être publié"
    );
}

/// `cancel_resume` sans armement préalable ⇒ `false`, aucun event, et le Scheduler
/// n'est même pas sollicité.
#[test]
fn cancel_resume_without_arm_is_false_no_event() {
    let env = env_at(NOW);
    assert!(!env.service.cancel_resume(aid(1)));
    assert!(env.scheduler.cancels().is_empty(), "Scheduler non sollicité");
    assert!(env.bus.events().is_empty(), "aucun event");
}

/// Contrat ANTI-COURSE : Scheduler renvoyant `false` (« déjà tiré ») ⇒ `cancel_resume`
/// renvoie `false` et n'émet PAS `AgentResumeCancelled` (la reprise suit son cours).
#[test]
fn cancel_resume_when_scheduler_already_fired_is_false_no_event() {
    let env = env_at(NOW);
    env.service
        .on_rate_limited(aid(1), nid(2), Some("conv-1".to_owned()), Some(NOW + 60_000));
    // Simule un réveil déjà tiré : cancel renvoie false.
    env.scheduler.set_cancel_result(false);

    assert!(
        !env.service.cancel_resume(aid(1)),
        "cancel pile au tir ⇒ false (la reprise suit son cours)"
    );
    // Le Scheduler a bien été sollicité (mais a répondu false).
    assert_eq!(env.scheduler.cancels().len(), 1);
    // AUCUN AgentResumeCancelled (pas d'event trompeur).
    assert!(
        !env
            .bus
            .events()
            .iter()
            .any(|e| matches!(e, DomainEvent::AgentResumeCancelled { .. })),
        "pas d'AgentResumeCancelled quand le réveil a déjà tiré"
    );
}