1 files changed, 160 insertions, 0 deletions
diff --git a/crates/asist-app/src/session.rs b/crates/asist-app/src/session.rs
new file mode 100644
index 0000000..b74a4f3
--- /dev/null
+++ b/crates/asist-app/src/session.rs
@@ -0,0 +1,160 @@
+//! Estado compartido entre los hilos del pipeline.
+//!
+//! Sólo hay dos cosas que de verdad tienen que compartirse: qué turno es el
+//! vigente, y las banderas que permiten cortar lo que está en marcha. Todo lo
+//! demás viaja por canales. Mantener esta superficie pequeña es lo que hace
+//! que la interrupción sea razonable de seguir: cortar es subir el turno y
+//! levantar dos banderas.
+
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::Arc;
+
+use asist_core::event::TurnId;
+use asist_core::http::Cancel;
+
+#[derive(Clone)]
+pub struct Session {
+    inner: Arc<Inner>,
+}
+
+struct Inner {
+    /// Turno que se está atendiendo. Cualquier trabajo de un turno anterior
+    /// que aparezca por un canal es basura y se tira.
+    current: AtomicU64,
+    /// El asistente está hablando (o a punto de hacerlo).
+    speaking: AtomicBool,
+    /// Cierre solicitado.
+    stopping: AtomicBool,
+    /// Corta la respuesta del modelo en curso.
+    llm: Cancel,
+    /// Corta la síntesis en curso.
+    tts: Cancel,
+}
+
+impl Default for Session {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Session {
+    pub fn new() -> Self {
+        Self {
+            inner: Arc::new(Inner {
+                current: AtomicU64::new(0),
+                speaking: AtomicBool::new(false),
+                stopping: AtomicBool::new(false),
+                llm: Cancel::new(),
+                tts: Cancel::new(),
+            }),
+        }
+    }
+
+    pub fn current(&self) -> TurnId {
+        TurnId(self.inner.current.load(Ordering::SeqCst))
+    }
+
+    /// Abre un turno nuevo y devuelve su identificador.
+    pub fn begin_turn(&self) -> TurnId {
+        let id = TurnId(self.inner.current.fetch_add(1, Ordering::SeqCst) + 1);
+        self.inner.llm.reset();
+        self.inner.tts.reset();
+        id
+    }
+
+    /// ¿Sigue siendo `turn` el turno vigente?
+    ///
+    /// Lo consultan los hilos antes de gastar trabajo: una frase que
+    /// pertenece a un turno ya superado no debe sintetizarse ni oírse.
+    pub fn is_current(&self, turn: TurnId) -> bool {
+        turn == self.current()
+    }
+
+    /// Corta todo lo que esté en marcha para el turno actual.
+    pub fn interrupt(&self) {
+        self.inner.llm.cancel();
+        self.inner.tts.cancel();
+        self.inner.speaking.store(false, Ordering::SeqCst);
+    }
+
+    pub fn llm_cancel(&self) -> &Cancel {
+        &self.inner.llm
+    }
+
+    pub fn tts_cancel(&self) -> &Cancel {
+        &self.inner.tts
+    }
+
+    pub fn set_speaking(&self, speaking: bool) {
+        self.inner.speaking.store(speaking, Ordering::SeqCst);
+    }
+
+    pub fn is_speaking(&self) -> bool {
+        self.inner.speaking.load(Ordering::SeqCst)
+    }
+
+    pub fn request_stop(&self) {
+        self.inner.stopping.store(true, Ordering::SeqCst);
+        self.interrupt();
+    }
+
+    pub fn is_stopping(&self) -> bool {
+        self.inner.stopping.load(Ordering::SeqCst)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn cada_turno_recibe_un_identificador_creciente() {
+        let session = Session::new();
+        assert_eq!(session.begin_turn(), TurnId(1));
+        assert_eq!(session.begin_turn(), TurnId(2));
+        assert_eq!(session.current(), TurnId(2));
+    }
+
+    #[test]
+    fn el_trabajo_de_un_turno_viejo_deja_de_ser_vigente() {
+        let session = Session::new();
+        let primero = session.begin_turn();
+        session.begin_turn();
+        assert!(
+            !session.is_current(primero),
+            "el turno viejo debe descartarse"
+        );
+    }
+
+    #[test]
+    fn abrir_turno_rearma_las_cancelaciones() {
+        let session = Session::new();
+        session.begin_turn();
+        session.interrupt();
+        assert!(session.llm_cancel().is_cancelled());
+
+        session.begin_turn();
+        assert!(
+            !session.llm_cancel().is_cancelled(),
+            "un turno nuevo no puede heredar la cancelación del anterior"
+        );
+    }
+
+    #[test]
+    fn interrumpir_calla_y_cancela_las_dos_etapas() {
+        let session = Session::new();
+        session.begin_turn();
+        session.set_speaking(true);
+        session.interrupt();
+        assert!(session.tts_cancel().is_cancelled());
+        assert!(!session.is_speaking());
+    }
+
+    #[test]
+    fn pedir_el_cierre_tambien_interrumpe() {
+        let session = Session::new();
+        session.request_stop();
+        assert!(session.is_stopping());
+        assert!(session.llm_cancel().is_cancelled());
+    }
+}