IdeA/crates/application/tests/session_limit_t4.rs

//! LS4 — réconciliation §21.2-T4 au niveau applicatif : `drain_with_readiness_outcome`
//! traduit un tour clos par une **limite de session** (un `RateLimited` vu, pas de
//! `Final`) en `Ok(TurnOutcome::RateLimited{..})` — une **fin gracieuse**, pas une
//! erreur. **100 % fakes** (fake `AgentSession` scriptable + fake `InputMediator`).
//!
//! On vérifie AUSSI la NON-RÉGRESSION des signatures historiques `Result<String>` :
//! - `drain_with_readiness` (et `send_blocking`) ⇒ un tour limité reste une `Io`
//!   (zéro régression pour l'appelant orchestrateur) ;
//! - un flux clos SANS `Final` ET SANS `RateLimited` reste une `Io` (tour tronqué).

use std::sync::Mutex;

use async_trait::async_trait;

use application::{drain_with_readiness, drain_with_readiness_outcome, send_blocking, TurnOutcome};
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 aid(n: u128) -> AgentId {
    AgentId::from_uuid(Uuid::from_u128(n))
}

// ---------------------------------------------------------------------------
// Fake AgentSession : `send` rejoue une liste fixe d'événements.
// ---------------------------------------------------------------------------

struct FakeSession {
    events: Vec<ReplyEvent>,
}
#[async_trait]
impl AgentSession for FakeSession {
    fn id(&self) -> SessionId {
        SessionId::from_uuid(Uuid::from_u128(1))
    }
    fn conversation_id(&self) -> Option<String> {
        None
    }
    async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
        Ok(Box::new(self.events.clone().into_iter()))
    }
    async fn shutdown(&self) -> Result<(), AgentSessionError> {
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Fake InputMediator : journalise mark_idle / mark_alive (pour prouver que la
// readiness reste pilotée par le Final, pas par un RateLimited).
// ---------------------------------------------------------------------------

#[derive(Default)]
struct RecordingMediator {
    calls: Mutex<Vec<&'static str>>,
}
impl InputMediator for RecordingMediator {
    fn enqueue(&self, _agent: AgentId, _ticket: Ticket) -> PendingReply {
        PendingReply::new(Box::pin(std::future::pending()))
    }
    fn preempt(&self, _agent: AgentId) {}
    fn mark_idle(&self, _agent: AgentId) {
        self.calls.lock().unwrap().push("idle");
    }
    fn mark_alive(&self, _agent: AgentId) {
        self.calls.lock().unwrap().push("alive");
    }
    fn busy_state(&self, _agent: AgentId) -> AgentBusyState {
        AgentBusyState::Idle
    }
}

// ===========================================================================
// drain_with_readiness_outcome — issue riche T4
// ===========================================================================

/// `[RateLimited{Some(t)}]` sans Final ⇒ `Ok(TurnOutcome::RateLimited{Some(t)})`
/// (fin gracieuse, PAS d'Err).
#[tokio::test]
async fn outcome_rate_limited_some_without_final_is_graceful() {
    let session = FakeSession {
        events: vec![ReplyEvent::RateLimited {
            resets_at_ms: Some(1_700_000_000_000),
        }],
    };
    let mediator = RecordingMediator::default();
    let out = drain_with_readiness_outcome(&session, "go", None, &mediator, aid(1))
        .await
        .expect("un tour limité est une fin gracieuse, pas une erreur");
    assert_eq!(
        out,
        TurnOutcome::RateLimited {
            resets_at_ms: Some(1_700_000_000_000)
        }
    );
    // Readiness : un RateLimited ne marque PAS Idle (piloté par Final uniquement).
    assert!(
        !mediator.calls.lock().unwrap().contains(&"idle"),
        "un RateLimited ne doit pas marquer l'agent Idle"
    );
}

/// `[RateLimited{None}]` sans Final ⇒ `Ok(TurnOutcome::RateLimited{None})`.
#[tokio::test]
async fn outcome_rate_limited_none_without_final_is_graceful() {
    let session = FakeSession {
        events: vec![ReplyEvent::RateLimited { resets_at_ms: None }],
    };
    let mediator = RecordingMediator::default();
    let out = drain_with_readiness_outcome(&session, "go", None, &mediator, aid(1))
        .await
        .expect("fin gracieuse");
    assert_eq!(out, TurnOutcome::RateLimited { resets_at_ms: None });
}

/// `[.., RateLimited, Final]` ⇒ `Ok(TurnOutcome::Completed(contenu))` : le Final
/// l'emporte toujours (le tour a réellement abouti).
#[tokio::test]
async fn outcome_rate_limited_then_final_is_completed() {
    let session = FakeSession {
        events: vec![
            ReplyEvent::TextDelta { text: "a".into() },
            ReplyEvent::RateLimited {
                resets_at_ms: Some(42),
            },
            ReplyEvent::Final {
                content: "fini".into(),
            },
        ],
    };
    let mediator = RecordingMediator::default();
    let out = drain_with_readiness_outcome(&session, "go", None, &mediator, aid(1))
        .await
        .expect("ok");
    assert_eq!(out, TurnOutcome::Completed("fini".to_owned()));
    // Le Final marque bien Idle.
    assert!(mediator.calls.lock().unwrap().contains(&"idle"));
}

/// `[TextDelta]` seul (ni Final ni RateLimited) ⇒ `Err(Io)` INCHANGÉ : un vrai flux
/// tronqué reste une erreur (non-régression critique).
#[tokio::test]
async fn outcome_truncated_stream_without_final_or_ratelimit_is_io_error() {
    let session = FakeSession {
        events: vec![ReplyEvent::TextDelta { text: "a".into() }],
    };
    let mediator = RecordingMediator::default();
    let err = drain_with_readiness_outcome(&session, "go", None, &mediator, aid(1))
        .await
        .expect_err("flux tronqué sans limite ⇒ erreur");
    assert!(matches!(err, AgentSessionError::Io(_)), "vu: {err:?}");
}

// ===========================================================================
// Non-régression des signatures historiques Result<String>
// ===========================================================================

/// `drain_with_readiness` (signature historique) : un tour limité reste `Err(Io)`
/// (zéro régression pour l'appelant orchestrateur).
#[tokio::test]
async fn drain_with_readiness_rate_limited_is_io_error() {
    let session = FakeSession {
        events: vec![ReplyEvent::RateLimited {
            resets_at_ms: Some(1),
        }],
    };
    let mediator = RecordingMediator::default();
    let err = drain_with_readiness(&session, "go", None, &mediator, aid(1))
        .await
        .expect_err("limite ⇒ Io sur la signature historique");
    assert!(matches!(err, AgentSessionError::Io(_)), "vu: {err:?}");
}

/// `send_blocking` (signature historique) : un tour limité reste `Err(Io)`.
#[tokio::test]
async fn send_blocking_rate_limited_is_io_error() {
    let session = FakeSession {
        events: vec![ReplyEvent::RateLimited { resets_at_ms: None }],
    };
    let err = send_blocking(&session, "go", None)
        .await
        .expect_err("limite ⇒ Io");
    assert!(matches!(err, AgentSessionError::Io(_)), "vu: {err:?}");
}

/// `drain_with_readiness` nominal : `[.., Final]` ⇒ le contenu, et `mark_idle` au Final.
#[tokio::test]
async fn drain_with_readiness_nominal_still_completes() {
    let session = FakeSession {
        events: vec![
            ReplyEvent::TextDelta { text: "a".into() },
            ReplyEvent::Final {
                content: "fini".into(),
            },
        ],
    };
    let mediator = RecordingMediator::default();
    let content = drain_with_readiness(&session, "go", None, &mediator, aid(1))
        .await
        .expect("ok");
    assert_eq!(content, "fini");
    assert!(mediator.calls.lock().unwrap().contains(&"idle"));
}