IdeA/crates/application/tests/send_blocking_d1.rs

//! LOT D1 (ARCHITECTURE §17.1 / §17.4 / §17.9 ligne D1) — tests unitaires du
//! helper applicatif `send_blocking`, **100 % fakes**.
//!
//! `send_blocking` draine le flux d'un tour jusqu'au [`ReplyEvent::Final`] :
//! - cas nominal : `TextDelta*` puis `Final{content}` ⇒ renvoie `content` (les
//!   deltas/activités traversés sont ignorés par le rendez-vous synchrone) ;
//! - flux **sans** `Final` (épuisé) ⇒ [`AgentSessionError::Io`] ;
//! - `send` renvoyant `Err(Decode/Io)` ⇒ propagé tel quel ;
//! - **timeout** ⇒ [`AgentSessionError::Timeout`] **sans tuer la session**
//!   (aucun `shutdown` appelé) ; le fake retarde son `send` de façon
//!   déterministe (au-delà du `timeout`) pour forcer l'expiration.
//!
//! Fake `AgentSession` local et **scriptable** (inspiré du fake D0 du domaine) :
//! il produit le `ReplyStream` qu'on lui a scripté, ou une erreur, ou un délai ;
//! il compte ses `shutdown` pour prouver la non-mise-à-mort sur timeout.

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

use async_trait::async_trait;

use application::send_blocking;
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))
}

/// Ce que le fake doit faire au prochain `send`.
enum Script {
    /// Renvoyer ce flux d'événements (consommé tel quel).
    Stream(Vec<ReplyEvent>),
    /// `send` échoue avec cette erreur.
    Err(AgentSessionError),
    /// `send` dort `delay` avant de renvoyer le flux (force le timeout).
    Delayed {
        delay: Duration,
        events: Vec<ReplyEvent>,
    },
}

/// Fake scriptable d'`AgentSession`. Mono-usage (un `send` scripté).
struct ScriptedSession {
    id: SessionId,
    script: std::sync::Mutex<Option<Script>>,
    shutdowns: AtomicUsize,
}

impl ScriptedSession {
    fn new(script: Script) -> Self {
        Self {
            id: sid(1),
            script: std::sync::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) => {
                let stream: ReplyStream = Box::new(events.into_iter());
                Ok(stream)
            }
            Script::Err(e) => Err(e),
            Script::Delayed { delay, events } => {
                tokio::time::sleep(delay).await;
                let stream: ReplyStream = Box::new(events.into_iter());
                Ok(stream)
            }
        }
    }
    async fn shutdown(&self) -> Result<(), AgentSessionError> {
        self.shutdowns.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
}

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

// ---------------------------------------------------------------------------
// Cas nominal
// ---------------------------------------------------------------------------

#[tokio::test]
async fn returns_final_content_ignoring_deltas_and_tools() {
    let session = ScriptedSession::new(Script::Stream(vec![
        delta("hel"),
        tool("reads a file"),
        delta("lo"),
        final_("hello world"),
    ]));

    let out = send_blocking(&session, "ping", None).await;
    assert_eq!(out, Ok("hello world".to_owned()));
    // Rendez-vous synchrone : on ne tue pas la session sur succès.
    assert_eq!(session.shutdown_count(), 0);
}

#[tokio::test]
async fn returns_final_content_with_no_intermediate_events() {
    // Flux = juste le Final déterministe.
    let session = ScriptedSession::new(Script::Stream(vec![final_("done")]));
    let out = send_blocking(&session, "x", Some(Duration::from_secs(5))).await;
    assert_eq!(out, Ok("done".to_owned()));
}

#[tokio::test]
async fn stops_at_first_final_even_if_more_events_follow() {
    // Robustesse : le drain s'arrête au PREMIER Final et renvoie son contenu.
    let session = ScriptedSession::new(Script::Stream(vec![
        delta("a"),
        final_("first"),
        final_("second-should-be-ignored"),
    ]));
    let out = send_blocking(&session, "x", None).await;
    assert_eq!(out, Ok("first".to_owned()));
}

// ---------------------------------------------------------------------------
// Bord : flux sans Final
// ---------------------------------------------------------------------------

#[tokio::test]
async fn stream_without_final_is_io_error() {
    let session = ScriptedSession::new(Script::Stream(vec![delta("a"), tool("b")]));
    let out = send_blocking(&session, "x", None).await;
    assert!(
        matches!(out, Err(AgentSessionError::Io(_))),
        "flux épuisé sans Final ⇒ Io, obtenu {out:?}"
    );
}

#[tokio::test]
async fn empty_stream_is_io_error() {
    let session = ScriptedSession::new(Script::Stream(vec![]));
    let out = send_blocking(&session, "x", None).await;
    assert!(
        matches!(out, Err(AgentSessionError::Io(_))),
        "flux vide ⇒ Io, obtenu {out:?}"
    );
}

// ---------------------------------------------------------------------------
// Bord : erreur de `send` propagée
// ---------------------------------------------------------------------------

#[tokio::test]
async fn send_decode_error_is_propagated() {
    let session = ScriptedSession::new(Script::Err(AgentSessionError::Decode(
        "bad json".to_owned(),
    )));
    let out = send_blocking(&session, "x", None).await;
    assert_eq!(out, Err(AgentSessionError::Decode("bad json".to_owned())));
}

#[tokio::test]
async fn send_io_error_is_propagated() {
    let session =
        ScriptedSession::new(Script::Err(AgentSessionError::Io("broken pipe".to_owned())));
    let out = send_blocking(&session, "x", Some(Duration::from_secs(5))).await;
    assert_eq!(out, Err(AgentSessionError::Io("broken pipe".to_owned())));
}

// ---------------------------------------------------------------------------
// Bord : timeout — Timeout renvoyé ET session NON tuée
// ---------------------------------------------------------------------------

#[tokio::test]
async fn timeout_returns_timeout_error_and_does_not_kill_session() {
    // `send` dort 1 s ; timeout fixé à 20 ms ⇒ l'attente expire avant le Final.
    let session = ScriptedSession::new(Script::Delayed {
        delay: Duration::from_secs(1),
        events: vec![final_("too late")],
    });

    let out = send_blocking(&session, "x", Some(Duration::from_millis(20))).await;
    assert_eq!(out, Err(AgentSessionError::Timeout));
    // §17.1 : on ne `shutdown` rien sur timeout — la session reste vivante.
    assert_eq!(
        session.shutdown_count(),
        0,
        "timeout ne doit PAS tuer la session"
    );
}

#[tokio::test]
async fn no_timeout_bound_waits_for_final() {
    // `timeout = None` ⇒ pas de borne : on attend le Final même après un délai.
    let session = ScriptedSession::new(Script::Delayed {
        delay: Duration::from_millis(10),
        events: vec![final_("eventually")],
    });
    let out = send_blocking(&session, "x", None).await;
    assert_eq!(out, Ok("eventually".to_owned()));
}