IdeA/crates/application/tests/drain_with_readiness_lot1.rs

//! Lot 1 (readiness/heartbeat model-agnostique) — tests unitaires QA **indépendants**
//! du helper applicatif `drain_with_readiness`, **100 % fakes**.
//!
//! Couvre les points 5 et 6 du périmètre QA :
//!
//! - **Point 5** : un flux se terminant par `Final` appelle `mark_idle(agent)`
//!   **exactement une fois** ; les `Heartbeat`/deltas/activités intermédiaires
//!   n'appellent **jamais** `mark_idle`.
//! - **Point 6 (le bug d'origine)** : un agent cible qui **ne fait que renvoyer un
//!   `Final`** (sans jamais appeler `idea_reply`) débloque bien la file via
//!   `mark_idle` — c'est exactement la cause racine du blocage `Busy`.
//!
//! On teste au niveau `drain_with_readiness` (le rendez-vous synchrone qu'`ask_agent`
//! utilise sur la session structurée d'une cible) avec :
//! - un fake `AgentSession` scriptable (calqué sur `send_blocking_d1.rs`) ;
//! - un fake `InputMediator` qui **compte** les `mark_idle` par agent et journalise
//!   l'ordre des appels, sans seconde file ni I/O.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Mutex;
use std::time::Duration;

use async_trait::async_trait;

use application::drain_with_readiness;
use domain::ids::AgentId;
use domain::input::{AgentBusyState, InputMediator};
use domain::mailbox::{PendingReply, Ticket};
use domain::ports::{AgentSession, AgentSessionError, ReplyEvent, ReplyStream};
use domain::SessionId;
use uuid::Uuid;

fn sid(n: u128) -> SessionId {
    SessionId::from_uuid(Uuid::from_u128(n))
}

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

// ---------------------------------------------------------------------------
// Fake AgentSession scriptable (mono-usage), repris de send_blocking_d1.rs
// ---------------------------------------------------------------------------

enum Script {
    Stream(Vec<ReplyEvent>),
    Err(AgentSessionError),
}

struct ScriptedSession {
    id: SessionId,
    script: Mutex<Option<Script>>,
    shutdowns: AtomicUsize,
}

impl ScriptedSession {
    fn new(script: Script) -> Self {
        Self {
            id: sid(1),
            script: Mutex::new(Some(script)),
            shutdowns: AtomicUsize::new(0),
        }
    }
    fn shutdown_count(&self) -> usize {
        self.shutdowns.load(Ordering::SeqCst)
    }
}

#[async_trait]
impl AgentSession for ScriptedSession {
    fn id(&self) -> SessionId {
        self.id
    }
    fn conversation_id(&self) -> Option<String> {
        None
    }
    async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
        let script = self
            .script
            .lock()
            .unwrap()
            .take()
            .expect("send scripté une seule fois");
        match script {
            Script::Stream(events) => Ok(Box::new(events.into_iter())),
            Script::Err(e) => Err(e),
        }
    }
    async fn shutdown(&self) -> Result<(), AgentSessionError> {
        self.shutdowns.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Fake InputMediator : compte les mark_idle par agent + journalise les appels.
// ---------------------------------------------------------------------------

struct RecordingMediator {
    /// (agent, "mark_idle") dans l'ordre des appels.
    calls: Mutex<Vec<(AgentId, &'static str)>>,
}

impl RecordingMediator {
    fn new() -> Self {
        Self {
            calls: Mutex::new(Vec::new()),
        }
    }
    fn mark_idle_count(&self, agent: AgentId) -> usize {
        self.calls
            .lock()
            .unwrap()
            .iter()
            .filter(|(a, kind)| *a == agent && *kind == "mark_idle")
            .count()
    }
    fn total_calls(&self) -> usize {
        self.calls.lock().unwrap().len()
    }
}

impl InputMediator for RecordingMediator {
    fn enqueue(&self, _agent: AgentId, _ticket: Ticket) -> PendingReply {
        // Jamais utilisé par drain_with_readiness ; un future qui ne résout pas.
        PendingReply::new(Box::pin(std::future::pending()))
    }
    fn preempt(&self, agent: AgentId) {
        self.calls.lock().unwrap().push((agent, "preempt"));
    }
    fn mark_idle(&self, agent: AgentId) {
        self.calls.lock().unwrap().push((agent, "mark_idle"));
    }
    fn busy_state(&self, _agent: AgentId) -> AgentBusyState {
        AgentBusyState::Idle
    }
}

// ---------------------------------------------------------------------------
// Helpers d'événements
// ---------------------------------------------------------------------------

fn delta(t: &str) -> ReplyEvent {
    ReplyEvent::TextDelta { text: t.to_owned() }
}
fn tool(l: &str) -> ReplyEvent {
    ReplyEvent::ToolActivity {
        label: l.to_owned(),
    }
}
fn heartbeat() -> ReplyEvent {
    ReplyEvent::Heartbeat
}
fn final_(c: &str) -> ReplyEvent {
    ReplyEvent::Final {
        content: c.to_owned(),
    }
}

// ===========================================================================
// Point 6 (LE point qui compte) : un Final SEUL débloque la file via mark_idle,
// sans aucun idea_reply.
// ===========================================================================

#[tokio::test]
async fn final_only_stream_unblocks_queue_via_mark_idle() {
    // L'agent cible ne fait QUE renvoyer un Final (jamais d'idea_reply).
    let agent = aid(42);
    let session = ScriptedSession::new(Script::Stream(vec![final_("done")]));
    let mediator = RecordingMediator::new();

    let out = drain_with_readiness(&session, "tâche", None, &mediator, agent).await;

    assert_eq!(out, Ok("done".to_owned()), "le Final rend bien son contenu");
    assert_eq!(
        mediator.mark_idle_count(agent),
        1,
        "le Final déterministe DOIT marquer l'agent Idle (fix de la cause racine du blocage Busy)"
    );
    // Aucun autre appel parasite, et la session reste vivante (pas de shutdown).
    assert_eq!(mediator.total_calls(), 1);
    assert_eq!(session.shutdown_count(), 0);
}

// ===========================================================================
// Point 5 : exactement UN mark_idle ; les events intermédiaires n'en émettent pas.
// ===========================================================================

#[tokio::test]
async fn intermediate_events_do_not_mark_idle_only_final_does() {
    let agent = aid(7);
    // Heartbeats + deltas + activité PUIS le Final.
    let session = ScriptedSession::new(Script::Stream(vec![
        heartbeat(),
        delta("hel"),
        tool("lit un fichier"),
        heartbeat(),
        delta("lo"),
        final_("hello"),
    ]));
    let mediator = RecordingMediator::new();

    let out = drain_with_readiness(&session, "x", None, &mediator, agent).await;

    assert_eq!(out, Ok("hello".to_owned()));
    assert_eq!(
        mediator.mark_idle_count(agent),
        1,
        "mark_idle exactement UNE fois (au Final), jamais sur heartbeat/delta/activité"
    );
    assert_eq!(
        mediator.total_calls(),
        1,
        "aucun appel mediator hormis l'unique mark_idle"
    );
}

#[tokio::test]
async fn heartbeats_alone_never_mark_idle_before_final() {
    // Que des heartbeats avant le Final : aucun ne doit marquer Idle.
    let agent = aid(9);
    let session = ScriptedSession::new(Script::Stream(vec![
        heartbeat(),
        heartbeat(),
        heartbeat(),
        final_("fini"),
    ]));
    let mediator = RecordingMediator::new();

    let out = drain_with_readiness(&session, "x", None, &mediator, agent).await;
    assert_eq!(out, Ok("fini".to_owned()));
    assert_eq!(mediator.mark_idle_count(agent), 1);
}

// ===========================================================================
// Bords : pas de Final ⇒ pas de mark_idle ; erreur de send propagée ; mauvais agent.
// ===========================================================================

#[tokio::test]
async fn stream_without_final_does_not_mark_idle_and_is_io_error() {
    let agent = aid(3);
    let session = ScriptedSession::new(Script::Stream(vec![heartbeat(), delta("a"), tool("b")]));
    let mediator = RecordingMediator::new();

    let out = drain_with_readiness(&session, "x", None, &mediator, agent).await;
    assert!(
        matches!(out, Err(AgentSessionError::Io(_))),
        "flux épuisé sans Final ⇒ Io, obtenu {out:?}"
    );
    assert_eq!(
        mediator.mark_idle_count(agent),
        0,
        "sans Final, on ne marque JAMAIS Idle (jamais de faux Idle)"
    );
    assert_eq!(mediator.total_calls(), 0);
}

#[tokio::test]
async fn send_error_is_propagated_and_no_mark_idle() {
    let agent = aid(5);
    let session =
        ScriptedSession::new(Script::Err(AgentSessionError::Decode("bad json".to_owned())));
    let mediator = RecordingMediator::new();

    let out = drain_with_readiness(&session, "x", None, &mediator, agent).await;
    assert_eq!(out, Err(AgentSessionError::Decode("bad json".to_owned())));
    assert_eq!(mediator.mark_idle_count(agent), 0);
}

#[tokio::test]
async fn mark_idle_targets_the_drained_agent_only() {
    // Le mark_idle doit porter sur l'agent passé à drain_with_readiness, pas un autre.
    let drained = aid(100);
    let other = aid(200);
    let session = ScriptedSession::new(Script::Stream(vec![final_("ok")]));
    let mediator = RecordingMediator::new();

    let _ = drain_with_readiness(&session, "x", None, &mediator, drained).await;
    assert_eq!(mediator.mark_idle_count(drained), 1);
    assert_eq!(
        mediator.mark_idle_count(other),
        0,
        "mark_idle ne doit cibler que l'agent drainé"
    );
}

// ===========================================================================
// Timeout (garde-fou du tour) : Timeout renvoyé, pas de mark_idle, session vivante.
// (Le helper borne via tokio::time::timeout ; on force l'expiration avec un Final
//  reporté — réutilise un fake retardé minimal local.)
// ===========================================================================

struct DelayedSession {
    delay: Duration,
    shutdowns: AtomicUsize,
}

#[async_trait]
impl AgentSession for DelayedSession {
    fn id(&self) -> SessionId {
        sid(2)
    }
    fn conversation_id(&self) -> Option<String> {
        None
    }
    async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
        tokio::time::sleep(self.delay).await;
        Ok(Box::new(vec![final_("too late")].into_iter()))
    }
    async fn shutdown(&self) -> Result<(), AgentSessionError> {
        self.shutdowns.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
}

#[tokio::test]
async fn timeout_returns_timeout_no_mark_idle_session_alive() {
    let agent = aid(11);
    let session = DelayedSession {
        delay: Duration::from_secs(1),
        shutdowns: AtomicUsize::new(0),
    };
    let mediator = RecordingMediator::new();

    let out =
        drain_with_readiness(&session, "x", Some(Duration::from_millis(20)), &mediator, agent).await;
    assert_eq!(out, Err(AgentSessionError::Timeout));
    assert_eq!(
        mediator.mark_idle_count(agent),
        0,
        "un timeout ne marque pas Idle (le tour n'a pas rendu son Final)"
    );
    assert_eq!(
        session.shutdowns.load(Ordering::SeqCst),
        0,
        "timeout ne tue PAS la session (§17.1)"
    );
}