Struct SessionBus 

pub struct SessionBus { /* private fields */ }

Unified event bus for a single session.

See the module docs for the ephemeral-vs-durable model. SessionBus is cheap to clone; every clone shares the same broadcast channel, durable sink, and optional legacy mpsc forwarder.

Examples

Fan out a single event to a broadcast subscriber and a durable sink:

use std::sync::{Arc, Mutex};
use codetether_agent::session::{
    DurableSink, SessionBus, SessionEvent, TokenDelta, TokenSource,
};

struct Mem(Arc<Mutex<usize>>);
impl DurableSink for Mem {
    fn write(&self, _e: &SessionEvent) -> std::io::Result<()> {
        *self.0.lock().unwrap() += 1;
        Ok(())
    }
}

let counter = Arc::new(Mutex::new(0));
let sink = Arc::new(Mem(counter.clone()));
let bus = SessionBus::new(64).with_durable_sink(sink);
let mut rx = bus.subscribe();

bus.emit(SessionEvent::TokenUsage(TokenDelta {
    source: TokenSource::Root,
    model: "m".into(),
    prompt_tokens: 10, completion_tokens: 5, duration_ms: 0,
}));

let ev = rx.recv().await.unwrap();
assert!(ev.is_durable());
assert_eq!(*counter.lock().unwrap(), 1);

Implementations

impl SessionBus

pub fn new(capacity: usize) -> Self

Construct a new bus with the given broadcast channel capacity.

capacity controls how many pending events the broadcast channel retains per subscriber before emitting RecvError::Lagged. 64–256 is typical for interactive TUIs.

Starts with NoopSink and no legacy forwarder. Use Self::with_durable_sink and Self::with_legacy_mpsc to attach either.

Panics

Panics if capacity == 0 (same contract as broadcast::channel).

Examples

use codetether_agent::session::SessionBus;

let bus = SessionBus::new(16);
assert_eq!(bus.subscriber_count(), 0);

pub fn with_durable_sink(self, sink: Arc<dyn DurableSink>) -> Self

Attach a durable write-ahead sink for is_durable() events.

Examples

use std::sync::Arc;
use codetether_agent::session::{NoopSink, SessionBus};

let bus = SessionBus::new(8).with_durable_sink(Arc::new(NoopSink));
assert_eq!(bus.subscriber_count(), 0);

pub fn with_legacy_mpsc(self, tx: Sender<SessionEvent>) -> Self

Forward every emission to an existing legacy mpsc::Sender for backward compatibility with code paths that have not migrated.

Emission is non-blocking: uses try_send and logs (does not propagate) failures. This preserves the lossy semantics the legacy channel already had.

Examples

use codetether_agent::session::{SessionBus, SessionEvent};
let (tx, mut rx) = tokio::sync::mpsc::channel(8);
let bus = SessionBus::new(8).with_legacy_mpsc(tx);
bus.emit(SessionEvent::Thinking);
let ev = rx.recv().await.unwrap();
assert!(matches!(ev, SessionEvent::Thinking));

pub fn subscribe(&self) -> Receiver<SessionEvent>

Subscribe to the ephemeral broadcast stream.

All events — ephemeral and durable — are republished here for live UIs. Slow consumers will see RecvError::Lagged(n); the durable sink is unaffected.

pub fn subscriber_count(&self) -> usize

Current number of active broadcast subscribers.

Examples

use codetether_agent::session::SessionBus;

let bus = SessionBus::new(4);
let _rx = bus.subscribe();
assert_eq!(bus.subscriber_count(), 1);

pub fn emit(&self, event: SessionEvent)

Publish one event.

For durable events the sink is written first (synchronously); on success the event is then broadcast. A sink failure is logged at warn and does not prevent the broadcast — observers still see the event even if durable persistence fell over.

Sending on an empty broadcast channel is not an error; it simply means no live subscribers.

Examples

use codetether_agent::session::{SessionBus, SessionEvent};
let bus = SessionBus::new(4);
bus.emit(SessionEvent::Done); // no subscribers → still ok

Trait Implementations

impl Clone for SessionBus

fn clone(&self) -> SessionBus

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.