1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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());
}
}
|
