slice-5: EventSink seam + wall-clock started_at (review M3)

Lifecycle events now flow through a trait the ADR-0005 Valkey publisher
can implement; Channel carries SystemTime so a CDR can exist.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012QndwfhjyTiZcUYp87dwW8
Signed-off-by: Aaron D. Lee <himself@adlee.work>

crates/rutster-call-model/src/lib.rs

@@ -57,6 +57,14 @@ mod tests {
         // Zero-sized type — confirms the "no extra allocation" claim.
         assert_eq!(std::mem::size_of::<TapHandle>(), 0);
     }
+
+    #[test]
+    fn channel_carries_wall_clock_start() {
+        let before = std::time::SystemTime::now();
+        let ch = Channel::new_inbound();
+        let after = std::time::SystemTime::now();
+        assert!(ch.started_at >= before && ch.started_at <= after);
+    }
 }
 
 use std::time::Instant;
@@ -150,6 +158,13 @@ pub struct Channel {
     /// user could change mid-call. The monotonic clock is the right
     /// tool for "has this peer been silent for 60 seconds?"
     pub created_at: Instant,
+    /// Wall-clock start (slice-5, review M3). `created_at: Instant` above
+    /// measures elapsed time (idle timeout); THIS field anchors the CDR —
+    /// "when did the call start" as a timestamp a bill or an audit can
+    /// use. A monotonic Instant cannot be converted to wall-clock after
+    /// the fact, which is why both exist: Instant for arithmetic,
+    /// SystemTime for the record.
+    pub started_at: std::time::SystemTime,
     /// NEW (slice-2, spec §5.2, §6): None until Connected, set on Connected,
     /// cleared on Closing. State invariant — see the table below.
     pub tap: Option<TapHandle>,
@@ -163,6 +178,7 @@ impl Channel {
             state: ChannelState::New,
             direction: Direction::Inbound,
             created_at: Instant::now(),
+            started_at: std::time::SystemTime::now(),
             tap: None,
         }
     }

crates/rutster/src/event_sink.rs

@@ -0,0 +1,72 @@
+//! # event_sink — durable-lifecycle seam (slice-5, review M3)
+//!
+//! ADR-0005 promises call lifecycle → Valkey streams → durable CDR
+//! pipeline. Until that lands, lifecycle events die with the process —
+//! an OOM-killed node erases all evidence of its calls. This module is
+//! the SEAM: the media thread emits `CallEvent`s through `EventSink`;
+//! today's impl logs, the ADR-0005 impl will publish. The emission
+//! points and the event shape are the part that calcifies — the backend
+//! is plumbing.
+
+use std::sync::Arc;
+use std::time::SystemTime;
+
+use rutster_call_model::ChannelId;
+use rutster_tap::MetricsSnapshot;
+
+/// CDR-shaped lifecycle events. `SessionEnded` carries both timestamps so
+/// a consumer computes duration without correlating two events (a node
+/// can die between them — the whole point of emitting durably).
+#[derive(Debug, Clone)]
+pub enum CallEvent {
+    SessionRegistered {
+        id: ChannelId,
+        at: SystemTime,
+    },
+    SessionConnected {
+        id: ChannelId,
+        at: SystemTime,
+    },
+    SessionEnded {
+        id: ChannelId,
+        started_at: SystemTime,
+        ended_at: SystemTime,
+        reason: EndReason,
+        /// Tap counters at teardown — fulfills the tap_engine.rs promise
+        /// ("read a MetricsSnapshot for log/CDR emission on teardown").
+        /// Snapshot is taken just before the engine task is torn down, so
+        /// counts may trail the final frame by a tick — fine for a CDR.
+        tap_metrics: Option<MetricsSnapshot>,
+    },
+}
+
+/// Why the session ended — the CDR disposition embryo.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum EndReason {
+    /// DELETE /v1/sessions/:id (API-driven hangup).
+    Deleted,
+    /// The session closed itself (peer close / idle timeout).
+    Closed,
+    /// Node shutdown dropped it (the mass-hangup path drain exists to avoid).
+    Shutdown,
+}
+
+/// The seam. CONTRACT: `emit` is called from the media std::thread —
+/// implementations MUST NOT block or perform I/O inline. Buffer to your
+/// own task (the future Valkey impl does channel → tokio publisher).
+pub trait EventSink: Send + Sync {
+    fn emit(&self, event: CallEvent);
+}
+
+/// Log-backed sink: structured tracing on target "rutster::events". Not
+/// durable — the placeholder that makes the emission points real today.
+pub struct TracingEventSink;
+
+impl EventSink for TracingEventSink {
+    fn emit(&self, event: CallEvent) {
+        tracing::info!(target: "rutster::events", event = ?event, "call event");
+    }
+}
+
+/// Convenience alias for the opts field / spawn signature.
+pub type SharedEventSink = Arc<dyn EventSink>;

crates/rutster/src/lib.rs

@@ -23,6 +23,7 @@
 //! - [ARCHITECTURE.md](../../docs/ARCHITECTURE.md) — fused vertical.
 
 pub mod config;
+pub mod event_sink;
 pub mod media_thread;
 pub mod routes;
 pub mod session_map;

crates/rutster/src/main.rs

@@ -48,6 +48,8 @@ async fn main() {
         media_cfg,
         max_sessions: rutster::config::max_sessions(std::env::var("RUTSTER_MAX_SESSIONS").ok())
             .expect("RUTSTER_MAX_SESSIONS must be a number"),
+        // sink: TracingEventSink via Default — ADR-0005 Valkey impl replaces it at the same seam.
+        ..Default::default()
     };
     let (app_state, media_thread) = app_state
         .spawn_media_thread_with(opts, tokio::runtime::Handle::current())

crates/rutster/src/media_thread.rs

@@ -39,8 +39,9 @@ const META_TICK: Duration = Duration::from_millis(10);
 const CMD_CHANNEL_CAPACITY: usize = 64;
 
 /// Knobs the binary resolves from env and hands to the media thread.
-/// Manual Default (not derive) because later slices add non-derivable
-/// fields (Task 7's `Arc<dyn EventSink>`). Defaults reproduce the
+/// Manual Default (not derive) because `sink: Arc<dyn EventSink>` has no
+/// meaningful derived default — `Arc::new(TracingEventSink)` is a
+/// deliberate choice, not a zero value. Defaults reproduce the
 /// pre-slice-5 single-node dev behavior exactly.
 pub struct MediaThreadOpts {
     pub media_cfg: rutster_media::MediaAddressConfig,
@@ -48,18 +49,17 @@ pub struct MediaThreadOpts {
     /// benchmark produces the real per-node number — the gauge below is
     /// that benchmark's primary readout.
     pub max_sessions: usize,
+    /// Lifecycle event seam (Task 7, review M3). Default is the log-backed
+    /// `TracingEventSink`; ADR-0005's Valkey publisher swaps in here.
+    pub sink: crate::event_sink::SharedEventSink,
 }
 
-// clippy sees a two-field, all-derivable struct today and suggests
-// `#[derive(Default)]`; we keep the manual impl on purpose (see doc comment
-// above) so Task 7's `Arc<dyn EventSink>` field doesn't force a churn-y
-// derive → manual flip.
-#[allow(clippy::derivable_impls)]
 impl Default for MediaThreadOpts {
     fn default() -> Self {
         Self {
             media_cfg: rutster_media::MediaAddressConfig::default(),
             max_sessions: 64,
+            sink: std::sync::Arc::new(crate::event_sink::TracingEventSink),
         }
     }
 }
@@ -220,6 +220,7 @@ fn run_media_thread(
 ) {
     let mut sessions: HashMap<ChannelId, ThreadSession> = HashMap::new();
     let max_sessions = opts.max_sessions;
+    let sink = opts.sink.clone();
     let mut tick_overruns: u64 = 0;
     let mut last_tick_micros: u64 = 0;
     let mut draining = false;
@@ -255,6 +256,7 @@ fn run_media_thread(
                     match RtcSession::new_with_config(&opts.media_cfg) {
                         Ok(session) => {
                             let id = session.channel_id();
+                            let started_at = session.channel.started_at;
                             let app_state = crate::session_map::AppState::new(
                                 media_cmd_tx.clone(),
                                 default_tap_url.clone(),
@@ -268,6 +270,10 @@ fn run_media_thread(
                                     tap_conn: None,
                                 },
                             );
+                            sink.emit(crate::event_sink::CallEvent::SessionRegistered {
+                                id,
+                                at: started_at,
+                            });
                             let _ = reply.send(Ok(id));
                             debug!(channel_id = %id, "session registered");
                         }
@@ -285,6 +291,16 @@ fn run_media_thread(
                 }
                 MediaCmd::Delete { id, reply } => {
                     if let Some(mut s) = sessions.remove(&id) {
+                        // Snapshot BEFORE take(): after the take the conn
+                        // (and its metrics Arc) belongs to the teardown task.
+                        let tap_metrics = s.tap_conn.as_ref().map(|c| c.metrics.snapshot());
+                        sink.emit(crate::event_sink::CallEvent::SessionEnded {
+                            id,
+                            started_at: s.rtc.channel.started_at,
+                            ended_at: std::time::SystemTime::now(),
+                            reason: crate::event_sink::EndReason::Deleted,
+                            tap_metrics,
+                        });
                         if let Some(conn) = s.tap_conn.take() {
                             // Hand the 750ms bounded teardown to tokio —
                             // the tick loop never waits on brain I/O
@@ -301,6 +317,15 @@ fn run_media_thread(
                         "media thread shutdown; dropping {} sessions",
                         sessions.len()
                     );
+                    for (id, s) in sessions.iter() {
+                        sink.emit(crate::event_sink::CallEvent::SessionEnded {
+                            id: *id,
+                            started_at: s.rtc.channel.started_at,
+                            ended_at: std::time::SystemTime::now(),
+                            reason: crate::event_sink::EndReason::Shutdown,
+                            tap_metrics: s.tap_conn.as_ref().map(|c| c.metrics.snapshot()),
+                        });
+                    }
                     sessions.clear();
                     let _ = reply.send(());
                     return;
@@ -378,6 +403,10 @@ fn run_media_thread(
                     session.rtc.channel.tap = Some(rutster_call_model::TapHandle::new());
                     session.tap_conn = Some(conn);
                     info!(channel_id = %id, "tap engine + reflex + local VAD wired on Connected");
+                    sink.emit(crate::event_sink::CallEvent::SessionConnected {
+                        id: *id,
+                        at: std::time::SystemTime::now(),
+                    });
                     continue;
                 }
             }
@@ -387,7 +416,15 @@ fn run_media_thread(
             }
         }
         for id in closed_ids {
-            sessions.remove(&id);
+            if let Some(s) = sessions.remove(&id) {
+                sink.emit(crate::event_sink::CallEvent::SessionEnded {
+                    id,
+                    started_at: s.rtc.channel.started_at,
+                    ended_at: std::time::SystemTime::now(),
+                    reason: crate::event_sink::EndReason::Closed,
+                    tap_metrics: s.tap_conn.as_ref().map(|c| c.metrics.snapshot()),
+                });
+            }
             debug!(channel_id = %id, "session evicted after close");
         }
 
@@ -515,6 +552,51 @@ mod tests {
         thread.shutdown();
     }
 
+    #[test]
+    fn lifecycle_events_emit_registered_then_ended_deleted() {
+        use crate::event_sink::{CallEvent, EndReason, EventSink};
+        struct TestSink(std::sync::Mutex<Vec<CallEvent>>);
+        impl EventSink for TestSink {
+            fn emit(&self, event: CallEvent) {
+                self.0.lock().unwrap().push(event);
+            }
+        }
+        let sink = std::sync::Arc::new(TestSink(std::sync::Mutex::new(Vec::new())));
+
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let handle = rt.handle().clone();
+        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
+        let opts = MediaThreadOpts {
+            sink: sink.clone(),
+            ..Default::default()
+        };
+        let thread = MediaThread::spawn(url, opts, handle).expect("spawn");
+
+        let (reply, rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Register {
+                tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
+                reply,
+            })
+            .unwrap();
+        let id = rx.blocking_recv().unwrap().unwrap();
+        let (reply, rx) = oneshot::channel();
+        thread
+            .cmd_tx()
+            .blocking_send(MediaCmd::Delete { id, reply })
+            .unwrap();
+        rx.blocking_recv().unwrap();
+        thread.shutdown();
+
+        let events = sink.0.lock().unwrap();
+        assert!(matches!(&events[0], CallEvent::SessionRegistered { id: eid, .. } if *eid == id));
+        assert!(events.iter().any(|e| matches!(
+            e,
+            CallEvent::SessionEnded { id: eid, reason: EndReason::Deleted, .. } if *eid == id
+        )));
+    }
+
     #[test]
     fn media_thread_register_and_shutdown_round_trips() {
         let rt = tokio::runtime::Runtime::new().unwrap();