ipfrs_tensorlogic/
session_manager.rs

1//! Distributed Inference Session Manager
2//!
3//! Production-hardening component for managing concurrent distributed
4//! inference sessions with fault tolerance, lifecycle tracking, GC,
5//! and atomic metrics collection.
6//!
7//! # Overview
8//!
9//! [`DistributedSessionManager`] acts as the central registry for all
10//! in-flight and recently-completed distributed reasoning sessions.
11//! Each session is identified by a [`SessionId`] (a 128-bit random value
12//! rendered as 32 lowercase hex digits).
13//!
14//! ## Limits
15//!
16//! At most [`MAX_CONCURRENT_SESSIONS`] (256) sessions may be active at once.
17//! Attempts to exceed this limit return [`SessionError::CapacityExceeded`].
18//!
19//! ## Garbage Collection
20//!
21//! Call [`DistributedSessionManager::gc_expired_sessions`] periodically to
22//! remove sessions whose age exceeds the supplied `max_age` duration.
23//!
24//! # Examples
25//!
26//! ```
27//! use ipfrs_tensorlogic::session_manager::{
28//!     DistributedSessionManager, PeerId, SessionStatus,
29//! };
30//! use std::time::Duration;
31//!
32//! let mgr = DistributedSessionManager::new();
33//! let id = mgr.create_session("parent(X, Y)", vec![PeerId::new("peer-1")]).expect("example: should succeed in docs");
34//!
35//! // Transition the session to Running
36//! mgr.set_running(&id).expect("example: should succeed in docs");
37//!
38//! // Inspect status
39//! assert!(matches!(mgr.session_status(&id), Some(SessionStatus::Running { .. })));
40//!
41//! // Cancel it
42//! mgr.cancel_session(&id).expect("example: should succeed in docs");
43//! assert!(matches!(mgr.session_status(&id), Some(SessionStatus::Cancelled)));
44//!
45//! // GC sessions older than 1 second
46//! mgr.gc_expired_sessions(Duration::from_secs(1));
47//! ```
48
49use parking_lot::RwLock;
50use std::collections::HashMap;
51use std::fmt;
52use std::sync::atomic::{AtomicU64, Ordering};
53use std::sync::Arc;
54use std::time::{Duration, Instant};
55use thiserror::Error;
56
57// ─── Constants ──────────────────────────────────────────────────────────────
58
59/// Maximum number of concurrent sessions the manager will accept.
60pub const MAX_CONCURRENT_SESSIONS: usize = 256;
61
62// ─── PeerId ──────────────────────────────────────────────────────────────────
63
64/// Opaque identifier for a remote peer participating in a session.
65///
66/// Represented as a plain `String` for compatibility with libp2p `PeerId`
67/// string encoding, but kept as a newtype for type safety.
68#[derive(Debug, Clone, PartialEq, Eq, Hash)]
69pub struct PeerId(pub String);
70
71impl PeerId {
72    /// Create a [`PeerId`] from any string-like value.
73    pub fn new(s: impl Into<String>) -> Self {
74        Self(s.into())
75    }
76}
77
78impl fmt::Display for PeerId {
79    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
80        f.write_str(&self.0)
81    }
82}
83
84impl From<String> for PeerId {
85    fn from(s: String) -> Self {
86        Self(s)
87    }
88}
89
90impl From<&str> for PeerId {
91    fn from(s: &str) -> Self {
92        Self(s.to_string())
93    }
94}
95
96// ─── SessionId ──────────────────────────────────────────────────────────────
97
98/// 128-bit random session identifier.
99///
100/// Displayed as 32 lowercase hex characters (no dashes).
101#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
102pub struct SessionId([u8; 16]);
103
104impl SessionId {
105    /// Generate a new random [`SessionId`] using the OS entropy source.
106    ///
107    /// # Errors
108    ///
109    /// Returns [`SessionError::IdGeneration`] if the OS fails to supply
110    /// random bytes.
111    pub fn new() -> Result<Self, SessionError> {
112        let mut buf = [0u8; 16];
113        getrandom::fill(&mut buf).map_err(|e| SessionError::IdGeneration(e.to_string()))?;
114        Ok(Self(buf))
115    }
116
117    /// Construct a [`SessionId`] directly from raw bytes.
118    pub fn from_bytes(bytes: [u8; 16]) -> Self {
119        Self(bytes)
120    }
121
122    /// Return a reference to the underlying raw bytes.
123    pub fn as_bytes(&self) -> &[u8; 16] {
124        &self.0
125    }
126}
127
128impl fmt::Display for SessionId {
129    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
130        for byte in &self.0 {
131            write!(f, "{byte:02x}")?;
132        }
133        Ok(())
134    }
135}
136
137// ─── SessionStatus ──────────────────────────────────────────────────────────
138
139/// Lifecycle state of a distributed inference session.
140#[derive(Debug, Clone)]
141pub enum SessionStatus {
142    /// Session has been created but not yet started.
143    Pending,
144
145    /// Session is actively running.
146    Running {
147        /// Wall-clock instant at which `Running` state was entered.
148        started_at: Instant,
149    },
150
151    /// Session finished successfully.
152    Completed {
153        /// Optional content-identifier for the assembled proof / result.
154        result_cid: Option<String>,
155    },
156
157    /// Session terminated due to an error.
158    Failed {
159        /// Human-readable error description.
160        reason: String,
161    },
162
163    /// Session was explicitly cancelled by the caller.
164    Cancelled,
165}
166
167// ─── SessionRecord ───────────────────────────────────────────────────────────
168
169/// Internal record stored per session in the registry.
170#[derive(Debug)]
171struct SessionRecord {
172    /// Canonical goal string supplied at creation time.
173    goal: String,
174
175    /// Participating peers at the time of creation.
176    peers: Vec<PeerId>,
177
178    /// Current lifecycle status.
179    status: SessionStatus,
180
181    /// Wall-clock time at which the session was first created.
182    created_at: Instant,
183}
184
185// ─── SessionError ────────────────────────────────────────────────────────────
186
187/// Error type for all session manager operations.
188#[derive(Debug, Error)]
189pub enum SessionError {
190    /// Returned when the random-byte generation fails.
191    #[error("session id generation failed: {0}")]
192    IdGeneration(String),
193
194    /// Returned when the requested session does not exist.
195    #[error("session not found")]
196    NotFound,
197
198    /// Returned when attempting to transition from an incompatible state.
199    #[error("invalid state transition: {0}")]
200    InvalidTransition(String),
201
202    /// Returned when the concurrent-session limit is reached.
203    #[error("capacity exceeded: maximum {MAX_CONCURRENT_SESSIONS} concurrent sessions")]
204    CapacityExceeded,
205}
206
207// ─── SessionMetrics ──────────────────────────────────────────────────────────
208
209/// Atomic counters tracking aggregate session outcomes.
210///
211/// Individual counter values are accessed via [`SessionMetrics::snapshot`]
212/// which returns a plain [`SessionMetricsSnapshot`] with no atomics.
213#[derive(Debug)]
214pub struct SessionMetrics {
215    /// Total sessions ever created by this manager instance.
216    pub total_created: AtomicU64,
217
218    /// Total sessions that transitioned to [`SessionStatus::Completed`].
219    pub total_completed: AtomicU64,
220
221    /// Total sessions that transitioned to [`SessionStatus::Failed`].
222    pub total_failed: AtomicU64,
223
224    /// Total sessions that were [`SessionStatus::Cancelled`].
225    pub total_cancelled: AtomicU64,
226}
227
228impl SessionMetrics {
229    /// Create zeroed metrics.
230    fn new() -> Self {
231        Self {
232            total_created: AtomicU64::new(0),
233            total_completed: AtomicU64::new(0),
234            total_failed: AtomicU64::new(0),
235            total_cancelled: AtomicU64::new(0),
236        }
237    }
238
239    /// Return a plain snapshot of current counter values.
240    ///
241    /// Counters are read with `Relaxed` ordering; individual values are
242    /// consistent but the snapshot is not a linearisable point-in-time view.
243    pub fn snapshot(&self) -> SessionMetricsSnapshot {
244        SessionMetricsSnapshot {
245            total_created: self.total_created.load(Ordering::Relaxed),
246            total_completed: self.total_completed.load(Ordering::Relaxed),
247            total_failed: self.total_failed.load(Ordering::Relaxed),
248            total_cancelled: self.total_cancelled.load(Ordering::Relaxed),
249        }
250    }
251}
252
253// ─── SessionMetricsSnapshot ──────────────────────────────────────────────────
254
255/// A point-in-time copy of [`SessionMetrics`] with plain `u64` fields.
256#[derive(Debug, Clone, Copy, PartialEq, Eq)]
257pub struct SessionMetricsSnapshot {
258    /// Total sessions ever created.
259    pub total_created: u64,
260
261    /// Total sessions completed successfully.
262    pub total_completed: u64,
263
264    /// Total sessions that failed.
265    pub total_failed: u64,
266
267    /// Total sessions that were cancelled.
268    pub total_cancelled: u64,
269}
270
271// ─── DistributedSessionManager ───────────────────────────────────────────────
272
273/// Manages concurrent distributed inference sessions.
274///
275/// The manager is cheap to clone because all state is held behind an
276/// `Arc`-guarded lock.  Clone the manager to share it across threads or
277/// async tasks.
278///
279/// # Thread safety
280///
281/// All public methods acquire a short-lived read or write lock on the
282/// internal registry.  Metrics are updated via lock-free atomics.
283#[derive(Debug, Clone)]
284pub struct DistributedSessionManager {
285    inner: Arc<ManagerInner>,
286}
287
288#[derive(Debug)]
289struct ManagerInner {
290    registry: RwLock<HashMap<SessionId, SessionRecord>>,
291    metrics: SessionMetrics,
292}
293
294impl DistributedSessionManager {
295    /// Create a new, empty session manager.
296    pub fn new() -> Self {
297        Self {
298            inner: Arc::new(ManagerInner {
299                registry: RwLock::new(HashMap::new()),
300                metrics: SessionMetrics::new(),
301            }),
302        }
303    }
304
305    // ── Lifecycle operations ─────────────────────────────────────────────────
306
307    /// Create a new distributed inference session.
308    ///
309    /// # Errors
310    ///
311    /// - [`SessionError::CapacityExceeded`] when there are already
312    ///   [`MAX_CONCURRENT_SESSIONS`] active sessions.
313    /// - [`SessionError::IdGeneration`] when OS entropy is unavailable.
314    pub fn create_session(
315        &self,
316        goal: &str,
317        peers: Vec<PeerId>,
318    ) -> Result<SessionId, SessionError> {
319        // Check capacity under a read lock first (fast path).
320        {
321            let guard = self.inner.registry.read();
322            if guard.len() >= MAX_CONCURRENT_SESSIONS {
323                return Err(SessionError::CapacityExceeded);
324            }
325        }
326
327        let id = SessionId::new()?;
328
329        let record = SessionRecord {
330            goal: goal.to_string(),
331            peers,
332            status: SessionStatus::Pending,
333            created_at: Instant::now(),
334        };
335
336        // Re-check capacity under the write lock to close the TOCTOU window.
337        let mut guard = self.inner.registry.write();
338        if guard.len() >= MAX_CONCURRENT_SESSIONS {
339            return Err(SessionError::CapacityExceeded);
340        }
341        guard.insert(id, record);
342        self.inner
343            .metrics
344            .total_created
345            .fetch_add(1, Ordering::Relaxed);
346
347        Ok(id)
348    }
349
350    /// Transition a session to [`SessionStatus::Running`].
351    ///
352    /// Only valid from the [`SessionStatus::Pending`] state.
353    ///
354    /// # Errors
355    ///
356    /// - [`SessionError::NotFound`] when `id` is not registered.
357    /// - [`SessionError::InvalidTransition`] when the current state is not
358    ///   `Pending`.
359    pub fn set_running(&self, id: &SessionId) -> Result<(), SessionError> {
360        let mut guard = self.inner.registry.write();
361        let record = guard.get_mut(id).ok_or(SessionError::NotFound)?;
362        match &record.status {
363            SessionStatus::Pending => {
364                record.status = SessionStatus::Running {
365                    started_at: Instant::now(),
366                };
367                Ok(())
368            }
369            other => Err(SessionError::InvalidTransition(format!(
370                "expected Pending, got {other:?}"
371            ))),
372        }
373    }
374
375    /// Transition a session to [`SessionStatus::Completed`].
376    ///
377    /// Valid from either `Pending` or `Running`.
378    ///
379    /// # Errors
380    ///
381    /// - [`SessionError::NotFound`] when `id` is not registered.
382    /// - [`SessionError::InvalidTransition`] when the state is not
383    ///   `Pending` or `Running`.
384    pub fn set_completed(
385        &self,
386        id: &SessionId,
387        result_cid: Option<String>,
388    ) -> Result<(), SessionError> {
389        let mut guard = self.inner.registry.write();
390        let record = guard.get_mut(id).ok_or(SessionError::NotFound)?;
391        match &record.status {
392            SessionStatus::Pending | SessionStatus::Running { .. } => {
393                record.status = SessionStatus::Completed { result_cid };
394                self.inner
395                    .metrics
396                    .total_completed
397                    .fetch_add(1, Ordering::Relaxed);
398                Ok(())
399            }
400            other => Err(SessionError::InvalidTransition(format!(
401                "cannot complete from {other:?}"
402            ))),
403        }
404    }
405
406    /// Transition a session to [`SessionStatus::Failed`].
407    ///
408    /// Valid from either `Pending` or `Running`.
409    ///
410    /// # Errors
411    ///
412    /// - [`SessionError::NotFound`] when `id` is not registered.
413    /// - [`SessionError::InvalidTransition`] when the state is terminal.
414    pub fn set_failed(
415        &self,
416        id: &SessionId,
417        reason: impl Into<String>,
418    ) -> Result<(), SessionError> {
419        let mut guard = self.inner.registry.write();
420        let record = guard.get_mut(id).ok_or(SessionError::NotFound)?;
421        match &record.status {
422            SessionStatus::Pending | SessionStatus::Running { .. } => {
423                record.status = SessionStatus::Failed {
424                    reason: reason.into(),
425                };
426                self.inner
427                    .metrics
428                    .total_failed
429                    .fetch_add(1, Ordering::Relaxed);
430                Ok(())
431            }
432            other => Err(SessionError::InvalidTransition(format!(
433                "cannot fail from {other:?}"
434            ))),
435        }
436    }
437
438    /// Cancel a session.
439    ///
440    /// The session is transitioned to [`SessionStatus::Cancelled`].  Only
441    /// non-terminal sessions (`Pending`, `Running`) may be cancelled.
442    ///
443    /// # Errors
444    ///
445    /// - [`SessionError::NotFound`] when `id` is not registered.
446    /// - [`SessionError::InvalidTransition`] when the session is already in
447    ///   a terminal state.
448    pub fn cancel_session(&self, id: &SessionId) -> Result<(), SessionError> {
449        let mut guard = self.inner.registry.write();
450        let record = guard.get_mut(id).ok_or(SessionError::NotFound)?;
451        match &record.status {
452            SessionStatus::Pending | SessionStatus::Running { .. } => {
453                record.status = SessionStatus::Cancelled;
454                self.inner
455                    .metrics
456                    .total_cancelled
457                    .fetch_add(1, Ordering::Relaxed);
458                Ok(())
459            }
460            other => Err(SessionError::InvalidTransition(format!(
461                "cannot cancel a session in state {other:?}"
462            ))),
463        }
464    }
465
466    // ── Query operations ─────────────────────────────────────────────────────
467
468    /// Return the current status of a session, or `None` if it does not exist.
469    pub fn session_status(&self, id: &SessionId) -> Option<SessionStatus> {
470        let guard = self.inner.registry.read();
471        guard.get(id).map(|r| r.status.clone())
472    }
473
474    /// Return the goal string associated with a session.
475    pub fn session_goal(&self, id: &SessionId) -> Option<String> {
476        let guard = self.inner.registry.read();
477        guard.get(id).map(|r| r.goal.clone())
478    }
479
480    /// Return the peers associated with a session.
481    pub fn session_peers(&self, id: &SessionId) -> Option<Vec<PeerId>> {
482        let guard = self.inner.registry.read();
483        guard.get(id).map(|r| r.peers.clone())
484    }
485
486    /// Return the number of sessions currently in the registry.
487    ///
488    /// This includes both active and terminal (Completed / Failed / Cancelled)
489    /// sessions that have not yet been garbage-collected.
490    pub fn active_count(&self) -> usize {
491        self.inner.registry.read().len()
492    }
493
494    // ── Garbage collection ───────────────────────────────────────────────────
495
496    /// Remove sessions whose `created_at` age exceeds `max_age`.
497    ///
498    /// This includes sessions in *any* state (Pending, Running, Completed,
499    /// Failed, Cancelled) that are older than the threshold.  The method
500    /// returns the number of sessions that were removed.
501    pub fn gc_expired_sessions(&self, max_age: Duration) -> usize {
502        let now = Instant::now();
503        let mut guard = self.inner.registry.write();
504        let before = guard.len();
505        guard.retain(|_id, record| now.duration_since(record.created_at) <= max_age);
506        before - guard.len()
507    }
508
509    // ── Metrics ──────────────────────────────────────────────────────────────
510
511    /// Return a reference to the live metrics object.
512    ///
513    /// Use [`SessionMetrics::snapshot`] to get a plain-value copy.
514    pub fn metrics(&self) -> &SessionMetrics {
515        &self.inner.metrics
516    }
517}
518
519impl Default for DistributedSessionManager {
520    fn default() -> Self {
521        Self::new()
522    }
523}
524
525// ─── Tests ───────────────────────────────────────────────────────────────────
526
527#[cfg(test)]
528mod tests {
529    use super::*;
530    use std::time::Duration;
531
532    fn peers(names: &[&str]) -> Vec<PeerId> {
533        names.iter().map(|s| PeerId::new(*s)).collect()
534    }
535
536    // ── 1. SessionId display format ──────────────────────────────────────────
537
538    #[test]
539    fn test_session_id_display_is_32_hex_chars() {
540        let id = SessionId::new().expect("getrandom must succeed in tests");
541        let s = id.to_string();
542        assert_eq!(s.len(), 32, "SessionId display must be 32 hex chars");
543        assert!(
544            s.chars().all(|c| c.is_ascii_hexdigit()),
545            "All chars must be hex digits"
546        );
547        // Must be lowercase
548        assert_eq!(s, s.to_lowercase(), "Hex must be lowercase");
549    }
550
551    #[test]
552    fn test_session_id_from_bytes_roundtrip() {
553        let bytes = [
554            0xde, 0xad, 0xbe, 0xef, 0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99,
555            0xaa, 0xbb,
556        ];
557        let id = SessionId::from_bytes(bytes);
558        assert_eq!(id.as_bytes(), &bytes);
559        assert_eq!(id.to_string(), "deadbeef00112233445566778899aabb");
560    }
561
562    // ── 2. Session creation and initial status ───────────────────────────────
563
564    #[test]
565    fn test_create_session_returns_pending() {
566        let mgr = DistributedSessionManager::new();
567        let id = mgr
568            .create_session("parent(X, Y)", peers(&["p1"]))
569            .expect("create must succeed");
570
571        assert!(
572            matches!(mgr.session_status(&id), Some(SessionStatus::Pending)),
573            "Newly created session must be Pending"
574        );
575    }
576
577    #[test]
578    fn test_create_session_increments_active_count() {
579        let mgr = DistributedSessionManager::new();
580        assert_eq!(mgr.active_count(), 0);
581        let _ = mgr
582            .create_session("goal_a", peers(&[]))
583            .expect("create must succeed");
584        assert_eq!(mgr.active_count(), 1);
585        let _ = mgr
586            .create_session("goal_b", peers(&[]))
587            .expect("create must succeed");
588        assert_eq!(mgr.active_count(), 2);
589    }
590
591    // ── 3. State transitions ─────────────────────────────────────────────────
592
593    #[test]
594    fn test_set_running_transitions_from_pending() {
595        let mgr = DistributedSessionManager::new();
596        let id = mgr
597            .create_session("foo", peers(&[]))
598            .expect("test: should succeed");
599        mgr.set_running(&id).expect("set_running must succeed");
600        assert!(matches!(
601            mgr.session_status(&id),
602            Some(SessionStatus::Running { .. })
603        ));
604    }
605
606    #[test]
607    fn test_set_running_fails_when_not_pending() {
608        let mgr = DistributedSessionManager::new();
609        let id = mgr
610            .create_session("foo", peers(&[]))
611            .expect("test: should succeed");
612        mgr.set_running(&id).expect("test: should succeed");
613        // Second call must fail
614        let result = mgr.set_running(&id);
615        assert!(
616            matches!(result, Err(SessionError::InvalidTransition(_))),
617            "set_running on Running state must fail"
618        );
619    }
620
621    #[test]
622    fn test_set_completed_from_running() {
623        let mgr = DistributedSessionManager::new();
624        let id = mgr
625            .create_session("foo", peers(&[]))
626            .expect("test: should succeed");
627        mgr.set_running(&id).expect("test: should succeed");
628        mgr.set_completed(&id, Some("bafybei...".to_string()))
629            .expect("complete must succeed");
630        assert!(matches!(
631            mgr.session_status(&id),
632            Some(SessionStatus::Completed { .. })
633        ));
634    }
635
636    #[test]
637    fn test_set_failed_from_running() {
638        let mgr = DistributedSessionManager::new();
639        let id = mgr
640            .create_session("foo", peers(&[]))
641            .expect("test: should succeed");
642        mgr.set_running(&id).expect("test: should succeed");
643        mgr.set_failed(&id, "peer timed out")
644            .expect("fail must succeed");
645        assert!(matches!(
646            mgr.session_status(&id),
647            Some(SessionStatus::Failed { .. })
648        ));
649    }
650
651    // ── 4. Cancellation ──────────────────────────────────────────────────────
652
653    #[test]
654    fn test_cancel_pending_session() {
655        let mgr = DistributedSessionManager::new();
656        let id = mgr
657            .create_session("foo", peers(&[]))
658            .expect("test: should succeed");
659        mgr.cancel_session(&id).expect("cancel must succeed");
660        assert!(matches!(
661            mgr.session_status(&id),
662            Some(SessionStatus::Cancelled)
663        ));
664    }
665
666    #[test]
667    fn test_cancel_running_session() {
668        let mgr = DistributedSessionManager::new();
669        let id = mgr
670            .create_session("foo", peers(&[]))
671            .expect("test: should succeed");
672        mgr.set_running(&id).expect("test: should succeed");
673        mgr.cancel_session(&id)
674            .expect("cancel of running must succeed");
675        assert!(matches!(
676            mgr.session_status(&id),
677            Some(SessionStatus::Cancelled)
678        ));
679    }
680
681    #[test]
682    fn test_cancel_completed_session_fails() {
683        let mgr = DistributedSessionManager::new();
684        let id = mgr
685            .create_session("foo", peers(&[]))
686            .expect("test: should succeed");
687        mgr.set_completed(&id, None).expect("test: should succeed");
688        let result = mgr.cancel_session(&id);
689        assert!(
690            matches!(result, Err(SessionError::InvalidTransition(_))),
691            "Cancelling a Completed session must fail"
692        );
693    }
694
695    #[test]
696    fn test_cancel_nonexistent_session_returns_not_found() {
697        let mgr = DistributedSessionManager::new();
698        let id = SessionId::from_bytes([0u8; 16]);
699        let result = mgr.cancel_session(&id);
700        assert!(matches!(result, Err(SessionError::NotFound)));
701    }
702
703    // ── 5. GC of expired sessions ────────────────────────────────────────────
704
705    #[test]
706    fn test_gc_removes_old_sessions() {
707        let mgr = DistributedSessionManager::new();
708        let _id = mgr
709            .create_session("old", peers(&[]))
710            .expect("test: should succeed");
711        assert_eq!(mgr.active_count(), 1);
712
713        // GC with zero max_age should remove everything (age > 0 immediately).
714        let removed = mgr.gc_expired_sessions(Duration::ZERO);
715        assert_eq!(removed, 1, "GC must remove the old session");
716        assert_eq!(mgr.active_count(), 0);
717    }
718
719    #[test]
720    fn test_gc_preserves_fresh_sessions() {
721        let mgr = DistributedSessionManager::new();
722        let _ = mgr
723            .create_session("fresh", peers(&[]))
724            .expect("test: should succeed");
725        assert_eq!(mgr.active_count(), 1);
726
727        // A very long max_age should not remove anything.
728        let removed = mgr.gc_expired_sessions(Duration::from_secs(3600));
729        assert_eq!(
730            removed, 0,
731            "No sessions should be GCed with a 1-hour window"
732        );
733        assert_eq!(mgr.active_count(), 1);
734    }
735
736    // ── 6. Max session limit enforcement ────────────────────────────────────
737
738    #[test]
739    fn test_capacity_exceeded_returns_error() {
740        let mgr = DistributedSessionManager::new();
741
742        for i in 0..MAX_CONCURRENT_SESSIONS {
743            mgr.create_session(&format!("goal_{i}"), peers(&[]))
744                .unwrap_or_else(|e| panic!("create #{i} must succeed: {e}"));
745        }
746
747        // The 257th session must fail.
748        let result = mgr.create_session("one_too_many", peers(&[]));
749        assert!(
750            matches!(result, Err(SessionError::CapacityExceeded)),
751            "Must return CapacityExceeded when limit reached"
752        );
753    }
754
755    // ── 7. Metrics accumulation ──────────────────────────────────────────────
756
757    #[test]
758    fn test_metrics_count_created() {
759        let mgr = DistributedSessionManager::new();
760        let snap_before = mgr.metrics().snapshot();
761
762        let _ = mgr
763            .create_session("a", peers(&[]))
764            .expect("test: should succeed");
765        let _ = mgr
766            .create_session("b", peers(&[]))
767            .expect("test: should succeed");
768
769        let snap_after = mgr.metrics().snapshot();
770        assert_eq!(
771            snap_after.total_created - snap_before.total_created,
772            2,
773            "Two sessions must be counted as created"
774        );
775    }
776
777    #[test]
778    fn test_metrics_count_completed_failed_cancelled() {
779        let mgr = DistributedSessionManager::new();
780
781        let id1 = mgr
782            .create_session("complete_me", peers(&[]))
783            .expect("test: should succeed");
784        let id2 = mgr
785            .create_session("fail_me", peers(&[]))
786            .expect("test: should succeed");
787        let id3 = mgr
788            .create_session("cancel_me", peers(&[]))
789            .expect("test: should succeed");
790
791        mgr.set_completed(&id1, None).expect("test: should succeed");
792        mgr.set_failed(&id2, "oops").expect("test: should succeed");
793        mgr.cancel_session(&id3).expect("test: should succeed");
794
795        let snap = mgr.metrics().snapshot();
796        assert_eq!(snap.total_completed, 1);
797        assert_eq!(snap.total_failed, 1);
798        assert_eq!(snap.total_cancelled, 1);
799    }
800
801    #[test]
802    fn test_metrics_snapshot_is_plain_values() {
803        // Ensure the snapshot type carries no atomics – just copy it.
804        let mgr = DistributedSessionManager::new();
805        let _ = mgr
806            .create_session("x", peers(&[]))
807            .expect("test: should succeed");
808        let snap: SessionMetricsSnapshot = mgr.metrics().snapshot();
809        let _snap2 = snap; // Must be Copy
810        assert_eq!(snap.total_created, 1);
811    }
812
813    // ── 8. Goal and peer retrieval ────────────────────────────────────────────
814
815    #[test]
816    fn test_session_goal_and_peers_accessible() {
817        let mgr = DistributedSessionManager::new();
818        let goal = "ancestor(alice, bob)";
819        let p = peers(&["peer-alpha", "peer-beta"]);
820        let id = mgr
821            .create_session(goal, p.clone())
822            .expect("test: should succeed");
823
824        assert_eq!(mgr.session_goal(&id).as_deref(), Some(goal));
825        assert_eq!(mgr.session_peers(&id), Some(p));
826    }
827
828    // ── 9. Status of unknown session ─────────────────────────────────────────
829
830    #[test]
831    fn test_status_of_unknown_session_returns_none() {
832        let mgr = DistributedSessionManager::new();
833        let id = SessionId::from_bytes([0xff; 16]);
834        assert!(mgr.session_status(&id).is_none());
835    }
836
837    // ── 10. Clone shares state ───────────────────────────────────────────────
838
839    #[test]
840    fn test_cloned_manager_shares_registry() {
841        let mgr1 = DistributedSessionManager::new();
842        let mgr2 = mgr1.clone();
843
844        let id = mgr1
845            .create_session("shared", peers(&[]))
846            .expect("create must succeed");
847
848        // mgr2 should see the session created via mgr1
849        assert!(
850            mgr2.session_status(&id).is_some(),
851            "Cloned manager must share the same session registry"
852        );
853    }
854}
ipfrs_tensorlogic/session_manager.rs

ipfrs_tensorlogic/
session_manager.rs
