diff --git a/Cargo.lock b/Cargo.lock index 18bc515..9621b90 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1250,8 +1250,6 @@ checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" [[package]] name = "device-info" version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2ca8e71544c1b67dcdbc2699ab258828aff985e5bc8d5f6b486d90d7df2f848" dependencies = [ "core-foundation 0.10.1", "jni 0.21.1", @@ -2839,8 +2837,6 @@ dependencies = [ [[package]] name = "libwebrtc" version = "0.3.55" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0399e372f69d3bb3ad8d51f6354f3090c57317de4f61521aa2f5a542c06fe1f" dependencies = [ "cxx", "jni 0.21.1", @@ -2902,8 +2898,6 @@ checksum = "11d3d7f243d5c5a8b9bb5d6dd2b1602c0cb0b9db1621bafc7ed66e35ff9fe092" [[package]] name = "livekit" version = "0.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "51f6a9dd86d3255487f778582ec0d631092fd379d781ae11995479498f12daf4" dependencies = [ "bmrng", "bytes", @@ -2931,8 +2925,6 @@ dependencies = [ [[package]] name = "livekit-api" version = "0.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb58ad37e249137bded70b959ef86474f464df6064b54c007093a97a38e81123" dependencies = [ "http", "livekit-protocol", @@ -2950,8 +2942,6 @@ dependencies = [ [[package]] name = "livekit-capture" version = "0.1.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b3a14d36a02c28182d73245d0dc57876a6ab3cf6d2d92cf1cb979fc5b2fc4587" dependencies = [ "bytes", "chrono", @@ -2967,8 +2957,6 @@ dependencies = [ [[package]] name = "livekit-common" version = "0.1.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "728a0e2cc2991a3392e2a1c9321b92f3f49d35f02ffc3e071ec75ea66fa9a5bc" dependencies = [ "livekit-protocol", ] @@ -2976,8 +2964,6 @@ dependencies = [ [[package]] name = "livekit-data-stream" version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97cf9d2dc503dc26c365fb90f994e0a5fed0eaf8800cd32abfc53e4645f09c39" dependencies = [ "async-compression", "bmrng", @@ -3000,8 +2986,6 @@ dependencies = [ [[package]] name = "livekit-datatrack" version = "0.2.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a610cb4b2a6927fb21dee3d089bfb5d39da1b71de383201a89b7ac9bf80e1c74" dependencies = [ "anyhow", "bytes", @@ -3019,8 +3003,6 @@ dependencies = [ [[package]] name = "livekit-net" version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92ee1131d54701875cf3d0a1ef5ea09f8e328c93afc8845ac65b36cdadf80422" dependencies = [ "async-trait", "base64 0.21.7", @@ -3039,8 +3021,6 @@ dependencies = [ [[package]] name = "livekit-protocol" version = "0.8.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1329f59d49f876eb15aae91b1a6bb72b8d75ea4f9eceba606191af468c2c231d" dependencies = [ "pbjson", "pbjson-types", @@ -3051,8 +3031,6 @@ dependencies = [ [[package]] name = "livekit-region" version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e94621b7d7cbb4544cd568c8e632852cc174d383055b0e77f4638eb91e43d3b" dependencies = [ "serde", ] @@ -3060,8 +3038,6 @@ dependencies = [ [[package]] name = "livekit-rpc" version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64484e0a669d6b6f1ec1572caef775fcab5a67325f3cc41ebc255a86d9e309db" dependencies = [ "livekit-common", "livekit-data-stream", @@ -3074,11 +3050,22 @@ dependencies = [ "uuid", ] +[[package]] +name = "livekit-session" +version = "0.1.0" +dependencies = [ + "chrono", + "futures-util", + "livekit", + "livekit-token-source", + "log", + "thiserror 2.0.18", + "tokio", +] + [[package]] name = "livekit-signaling" version = "0.1.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e4c039320b98bc615890653143735c8f7f45789bca074dc97a6392ea04390a5" dependencies = [ "base64 0.21.7", "device-info", @@ -3095,14 +3082,13 @@ dependencies = [ "serde_json", "thiserror 2.0.18", "tokio", + "tokio-util", "url", ] [[package]] name = "livekit-token" version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "179d8858e2fa6c72126f04c72351490bac15d1d79443f0aff65102fefe74b945" dependencies = [ "hmac", "jsonwebtoken", @@ -3116,8 +3102,6 @@ dependencies = [ [[package]] name = "livekit-token-source" version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "15fae1735a88f5dda8554f57a6186a3136f73297d9a9045c27d5910293875bfc" dependencies = [ "async-trait", "base64 0.21.7", @@ -4779,6 +4763,7 @@ checksum = "4ade083ccbb4bf536df69d1f6432cc23deb7acccff86b183f3923a6fd56a1153" name = "rust-dev-client" version = "0.1.0" dependencies = [ + "async-trait", "eframe", "egui", "egui-wgpu", @@ -4788,6 +4773,8 @@ dependencies = [ "livekit", "livekit-api", "livekit-capture", + "livekit-protocol", + "livekit-session", "livekit-token-source", "log", "parking_lot", @@ -6207,8 +6194,6 @@ dependencies = [ [[package]] name = "webrtc-sys" version = "0.3.52" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bae66695bde7b250a9e5f7976db6015e30fa06033afc1745e5499d55aa51612a" dependencies = [ "cc", "cxx", @@ -6222,8 +6207,6 @@ dependencies = [ [[package]] name = "webrtc-sys-build" version = "0.3.21" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8d67326bcdd69af2a902c37d2f19101eaf4298fc81b699b89a2017a9a0c46267" dependencies = [ "anyhow", "fs2", @@ -7174,8 +7157,6 @@ dependencies = [ [[package]] name = "yuv-sys" version = "0.3.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "769d9a5cace19cbafd5cde04054c0bbaaf948dea1fc317121919bac791b1b53d" dependencies = [ "bindgen", "cc", diff --git a/Cargo.toml b/Cargo.toml index f108566..7011502 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,7 @@ version = "0.1.0" edition = "2024" [dependencies] +async-trait = "0.1" eframe = { version = "0.37.0", features = ["default_fonts", "wgpu", "persistence"] } egui = "0.37.0" egui-wgpu = "0.37.0" @@ -18,18 +19,22 @@ tokio = { version = "1", features = ["full", "parking_lot"] } # egui-wgpu 0.37 requires wgpu ^30.0; "30.0" resolves to a compatible 30.0.x. wgpu = "30.0" winit = { version = "0.30.13", features = [ "android-native-activity" ] } -livekit = { version = "0.10.0", features = ["rustls-tls-native-roots"] } -livekit-api = { version = "0.10.0", default-features = false, features = ["access-token"] } -livekit-token-source = { version = "0.1.4", features = ["rustls-tls-native-roots"] } -livekit-capture = { version = "0.1.8", default-features = false, features = ["source-pattern", "source-clock"] } +# livekit = { version = "0.10.0", features = ["rustls-tls-native-roots"] } +# livekit-api = { version = "0.10.0", default-features = false, features = ["access-token"] } +# livekit-token-source = { version = "0.1.4", features = ["rustls-tls-native-roots"] } +# livekit-capture = { version = "0.1.8", default-features = false, features = ["source-pattern", "source-clock"] } +# livekit-protocol = "0.8.2" +# livekit-session = { version = "0.1.0", features = ["rustls-tls-native-roots"] } # Building against a local rust-sdks checkout (see "Building against a local # rust-sdks" in README.md). To build against the released crates instead, # uncomment out the three lines below and comment the ones above: -# livekit = { path = "../rust-sdks/livekit", features = ["rustls-tls-native-roots"] } -# livekit-api = { path = "../rust-sdks/livekit-api", default-features = false, features = ["access-token"] } -# livekit-token-source = { path = "../rust-sdks/livekit-token-source", features = ["rustls-tls-native-roots"] } -# livekit-capture = { path = "../rust-sdks/livekit-capture", default-features = false, features = ["source-pattern", "source-clock"] } +livekit = { path = "../rust-sdks/livekit", features = ["rustls-tls-native-roots"] } +livekit-api = { path = "../rust-sdks/livekit-api", default-features = false, features = ["access-token"] } +livekit-token-source = { path = "../rust-sdks/livekit-token-source", features = ["rustls-tls-native-roots"] } +livekit-capture = { path = "../rust-sdks/livekit-capture", default-features = false, features = ["source-pattern", "source-clock"] } +livekit-protocol = { path = "../rust-sdks/livekit-protocol" } +livekit-session = { path = "../rust-sdks/livekit-session", features = ["rustls-tls-native-roots"] } [package.metadata.bundle] name = "LiveKit Client" diff --git a/README.md b/README.md index f1f3ecd..d62781f 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,8 @@ An example of building a cross-platform, GUI application using the [LiveKit Rust - [x] Simulate fault scenarios (e.g., reconnect, migration, etc.) - [x] Send/receive remote procedure calls (RPC) - [x] Send/receive data streams +- [x] Connect to an agent through a session ([livekit-session](https://github.com/livekit/rust-sdks/tree/main/livekit-session)): text chat, voice, live transcripts, and avatar video +- [x] Dispatch a named agent, optionally targeting one of its deployments, into the room you join (API key and TokenSource methods) ## Building @@ -45,6 +47,7 @@ To build against a local checkout of the [LiveKit Rust SDK](https://github.com/l livekit-api = { path = "../rust-sdks/livekit-api", default-features = false, features = ["access-token"] } ``` These are plain `path` dependencies, so cargo re-resolves automatically — no `cargo update` needed. + `livekit-session` is not published yet, so for now it is always a `path` dependency and the app needs a local `rust-sdks` checkout that includes it. 3. `cargo run` — and rust-analyzer — now build against the local sources. To return to the published crates, reverse step 2: uncomment the two version lines and comment the `path` lines back out. diff --git a/cspell.yml b/cspell.yml index 39397f5..08ec8e8 100644 --- a/cspell.yml +++ b/cspell.yml @@ -36,6 +36,7 @@ words: - tsid - unmultiplied - unorm + - upserted - videotexture - viewports - webrtc diff --git a/src/agent/mod.rs b/src/agent/mod.rs new file mode 100644 index 0000000..598deac --- /dev/null +++ b/src/agent/mod.rs @@ -0,0 +1,9 @@ +//! The agent feature: a window driving a `livekit_session::Session`, with the +//! agent's state, the conversation feed, and its avatar. Depends on the +//! reusable toolkit in [`crate::ui`] and reuses the participants panel and +//! status bar from [`crate::room`]. + +pub mod service; +pub mod window; + +pub use window::AgentWindow; diff --git a/src/agent/service.rs b/src/agent/service.rs new file mode 100644 index 0000000..3c7adc5 --- /dev/null +++ b/src/agent/service.rs @@ -0,0 +1,214 @@ +use crate::connect::AgentConnectSettings; +use crate::room::participants::{TrackCmd, TrackCommands}; +use livekit::prelude::*; +use livekit_session::{Message, Session, SessionConnectionState, SessionEvent, SessionState}; +use std::sync::Arc; +use tokio::sync::{mpsc, watch}; + +#[derive(Debug)] +pub enum AgentCmd { + Start, + End, + SendText(String), + Track(TrackCmd), +} + +#[derive(Debug)] +pub enum AgentEvent { + /// `Err` is a human-readable message. + StartResult(Result<(), String>), + /// A message was added, or a transcript grew. + Message(Message), + SendFailed(String), + /// The agent's avatar video, or `None` once it is gone. + Avatar(Option), +} + +/// The UI-side handle of an agent session: commands go to [`AgentActor`], +/// events come back. State that is cheap to read synchronously (the session +/// state and the room) is read from the session directly. +pub struct AgentService { + session: Session, + state: watch::Receiver, + cmd_tx: mpsc::UnboundedSender, + ui_rx: mpsc::UnboundedReceiver, +} + +impl AgentService { + pub fn new(runtime: &tokio::runtime::Handle, settings: &AgentConnectSettings) -> Self { + let session = Session::from(settings); + + let (cmd_tx, cmd_rx) = mpsc::unbounded_channel(); + let (ui_tx, ui_rx) = mpsc::unbounded_channel(); + let actor = AgentActor { + session: session.clone(), + ui_tx, + room_events: None, + avatar_sid: None, + }; + runtime.spawn(actor.run(cmd_rx)); + + Self { + state: session.state(), + session, + cmd_tx, + ui_rx, + } + } + + pub fn send(&self, cmd: AgentCmd) { + let _ = self.cmd_tx.send(cmd); + } + + pub fn try_recv(&mut self) -> Option { + self.ui_rx.try_recv().ok() + } + + pub fn state(&self) -> SessionState { + *self.state.borrow() + } + + pub fn room(&self) -> Option> { + self.session.room() + } +} + +impl TrackCommands for AgentService { + fn track_cmd(&self, cmd: TrackCmd) { + self.send(AgentCmd::Track(cmd)); + } +} + +enum Input { + Cmd(Option), + Feed(SessionEvent), + Room(RoomEvent), + StateChanged, +} + +/// Owns the session's async side: starting and ending it, and forwarding the +/// message feed and avatar changes to the UI. The microphone and the agent's +/// audio are the session's own (`SessionOptions::platform_audio`). +struct AgentActor { + session: Session, + ui_tx: mpsc::UnboundedSender, + /// The connected room's events, `None` while there is no room. + room_events: Option>, + avatar_sid: Option, +} + +impl AgentActor { + async fn run(mut self, mut cmd_rx: mpsc::UnboundedReceiver) { + let mut feed = self.session.subscribe(); + let mut state = self.session.state(); + + loop { + let input = tokio::select! { + cmd = cmd_rx.recv() => Input::Cmd(cmd), + Some(event) = feed.recv() => Input::Feed(event), + Some(event) = recv_room_event(&mut self.room_events) => Input::Room(event), + Ok(()) = state.changed() => Input::StateChanged, + }; + match input { + Input::Cmd(None) => break, + Input::Cmd(Some(cmd)) => self.handle(cmd).await, + Input::Feed(SessionEvent::MessageUpserted(message)) => { + self.emit(AgentEvent::Message(message)); + } + // Only `MessagesRestored`, which follows `Session::restore_messages`; + // this client never restores a feed. + Input::Feed(_) => {} + Input::Room(event) => self.on_room_event(event), + Input::StateChanged => { + let connection = state.borrow_and_update().connection; + if connection == SessionConnectionState::Disconnected + && self.room_events.is_some() + { + self.release(); + } + } + } + } + + self.session.end().await; + } + + async fn handle(&mut self, cmd: AgentCmd) { + match cmd { + AgentCmd::Start => self.start().await, + AgentCmd::End => { + self.session.end().await; + self.release(); + } + AgentCmd::SendText(text) => { + if let Err(err) = self.session.send_text(&text).await { + self.emit(AgentEvent::SendFailed(err.to_string())); + } + } + AgentCmd::Track(cmd) => cmd.apply(), + } + } + + async fn start(&mut self) { + if let Err(err) = self.session.start().await { + log::error!("failed to start agent session: {err}"); + self.emit(AgentEvent::StartResult(Err(err.to_string()))); + return; + } + let Some(room) = self.session.room() else { + self.emit(AgentEvent::StartResult(Err( + "session disconnected while starting".to_string(), + ))); + return; + }; + log::info!("agent session connected to room: {}", room.name()); + + self.room_events = Some(room.subscribe()); + self.emit(AgentEvent::StartResult(Ok(()))); + self.refresh_avatar(); + } + + /// Drops what belonged to the room once it is gone. + fn release(&mut self) { + self.room_events = None; + self.refresh_avatar(); + } + + fn on_room_event(&mut self, event: RoomEvent) { + if matches!( + event, + RoomEvent::TrackSubscribed { .. } + | RoomEvent::TrackUnsubscribed { .. } + | RoomEvent::ParticipantDisconnected(_) + ) { + self.refresh_avatar(); + } + } + + /// Reports the avatar's video track when it changes. + fn refresh_avatar(&mut self) { + let track = self + .room_events + .as_ref() + .and_then(|_| self.session.agent_avatar_video_track()); + let sid = track.as_ref().map(|track| track.sid()); + if sid != self.avatar_sid { + self.avatar_sid = sid; + self.emit(AgentEvent::Avatar(track)); + } + } + + fn emit(&self, event: AgentEvent) { + let _ = self.ui_tx.send(event); + } +} + +/// The next event of the connected room; pending while there is none. +async fn recv_room_event( + room_events: &mut Option>, +) -> Option { + match room_events { + Some(events) => events.recv().await, + None => std::future::pending().await, + } +} diff --git a/src/agent/window.rs b/src/agent/window.rs new file mode 100644 index 0000000..306b791 --- /dev/null +++ b/src/agent/window.rs @@ -0,0 +1,282 @@ +use crate::{ + agent::service::{AgentCmd, AgentEvent, AgentService}, + connect::AgentConnectSettings, + media::VideoRenderer, + room::participants::ParticipantsPanel, + room::status_bar::{StatusBar, StatusBarActions}, + ui::{ + chat_message::ChatMessage, + placeholder_tile::{PlaceholderTile, placeholder_texture}, + status_badge::StatusBadge, + video_tile::{VideoTile, fit_size}, + }, +}; +use livekit_session::{AgentActivity, AgentFailure, AgentState, Message, MessageOrigin}; +use std::collections::BTreeMap; + +/// Share of the panel height the avatar may take; the feed gets the rest. +const AVATAR_HEIGHT_FRACTION: f32 = 0.4; +const AVATAR_OVERLAY_MARGIN: f32 = 5.0; +const SEND_BUTTON_WIDTH: f32 = 60.0; + +/// State and UI of a single agent session window. The session starts +/// immediately on creation with the settings handed over by the connect screen. +pub struct AgentWindow { + runtime: tokio::runtime::Handle, + render_state: egui_wgpu::RenderState, + service: AgentService, + /// Reconnecting rejoins the same room, where the agent may not be + /// dispatched again. + reuses_room: bool, + /// The feed keyed by [`Message::seq`], which a growing transcript keeps. + messages: BTreeMap, + draft: String, + avatar: Option, + connecting: bool, + connection_failure: Option, + send_failure: Option, +} + +impl AgentWindow { + pub fn new( + runtime: tokio::runtime::Handle, + render_state: egui_wgpu::RenderState, + settings: &AgentConnectSettings, + ) -> Self { + let mut window = Self { + service: AgentService::new(&runtime, settings), + runtime, + render_state, + reuses_room: settings.auth.reuses_room(), + messages: BTreeMap::new(), + draft: String::new(), + avatar: None, + connecting: false, + connection_failure: None, + send_failure: None, + }; + window.connect(); + window + } + + fn connect(&mut self) { + self.connecting = true; + self.connection_failure = None; + self.service.send(AgentCmd::Start); + } + + pub fn disconnect(&self) { + self.service.send(AgentCmd::End); + } + + fn event(&mut self, event: AgentEvent) { + match event { + AgentEvent::StartResult(result) => { + self.connecting = false; + self.connection_failure = result.err(); + } + AgentEvent::Message(message) => { + self.messages.insert(message.seq, message); + } + AgentEvent::SendFailed(err) => self.send_failure = Some(err), + AgentEvent::Avatar(track) => { + self.avatar = track.map(|track| { + VideoRenderer::new(&self.runtime, self.render_state.clone(), track.rtc_track()) + }); + } + } + } + + pub fn ui(&mut self, ui: &mut egui::Ui) { + while let Some(event) = self.service.try_recv() { + self.event(event); + } + + let state = self.service.state(); + let room = self.service.room(); + let mut actions = StatusBarActions::default(); + let bar_frame = + egui::Frame::central_panel(ui.style()).inner_margin(egui::Margin::symmetric(10, 6)); + + egui::Panel::bottom("status_bar") + .frame(bar_frame) + .show(ui, |ui| { + ui.add(StatusBar { + room: room.as_deref(), + connecting: self.connecting, + connection_failure: self.connection_failure.as_deref(), + detail: Some(StatusBadge::from(state.agent)), + reconnect_hint: self.reuses_room.then_some( + "A pre-generated token rejoins the same room, where the agent may not \ + be dispatched again", + ), + actions: &mut actions, + }); + }); + + egui::Panel::right("right_panel") + .frame(egui::Frame::central_panel(ui.style()).outer_margin(0.)) + .resizable(true) + .size_range(20.0..=360.0) + .show(ui, |ui| { + ui.heading("Participants"); + ui.separator(); + ui.add(ParticipantsPanel { + room: room.as_deref(), + commands: &self.service, + }); + }); + + egui::Panel::bottom("composer") + .frame(bar_frame) + .show(ui, |ui| self.composer(ui, state.connection.is_connected())); + + egui::CentralPanel::default().show(ui, |ui| { + self.agent_panel(ui, state.agent); + ui.separator(); + + egui::ScrollArea::vertical() + .id_salt("agent_feed") + .stick_to_bottom(true) + .auto_shrink(false) + .show(ui, |ui| { + if self.messages.is_empty() { + ui.weak("No messages yet"); + } + for message in self.messages.values() { + ui.add( + ChatMessage::new(speaker(message), &message.text) + .local(is_local(message)) + .pending(!message.is_final), + ); + ui.add_space(4.0); + } + }); + }); + + if actions.disconnect { + self.disconnect(); + } else if actions.reconnect { + self.connect(); + } + + ui.ctx().request_repaint(); + } + + /// The agent's avatar (or placeholder once it joined) with its state on top; + /// before that, only the state. The avatar keeps its aspect ratio: it is + /// scaled to fit the panel width and a share of its height, and centered. + fn agent_panel(&self, ui: &mut egui::Ui, state: AgentState) { + let bounds = egui::vec2( + ui.available_width(), + ui.available_height() * AVATAR_HEIGHT_FRACTION, + ); + if !has_joined(state) { + let (rect, _) = ui.allocate_exact_size(bounds, egui::Sense::hover()); + ui.painter().rect_filled( + rect.shrink(1.0), + egui::CornerRadius::default(), + ui.style().visuals.code_bg_color, + ); + ui.place(rect, StatusBadge::from(state)); + return; + } + + let speaking = state == AgentState::Connected(AgentActivity::Speaking); + let texture = self.avatar.as_ref().and_then(VideoRenderer::texture); + let size = texture.map_or(bounds, |texture| fit_size(bounds, texture.size)); + let (row, _) = ui.allocate_exact_size(egui::vec2(bounds.x, size.y), egui::Sense::hover()); + let rect = egui::Rect::from_center_size(row.center(), size); + if self.avatar.is_some() { + ui.place(rect, VideoTile::new(texture, "Avatar", speaking)); + } else { + let placeholder = placeholder_texture(ui.ctx()); + ui.place( + rect, + PlaceholderTile::new(placeholder.id(), "Agent", speaking), + ); + } + ui.new_child( + egui::UiBuilder::new() + .max_rect(rect.shrink(AVATAR_OVERLAY_MARGIN)) + .layout(egui::Layout::top_down(egui::Align::Min)), + ) + .add(StatusBadge::from(state)); + } + + /// The text input with Enter-to-send, and the last send failure. + fn composer(&mut self, ui: &mut egui::Ui, connected: bool) { + ui.horizontal(|ui| { + let width = ui.available_width() - SEND_BUTTON_WIDTH - ui.spacing().item_spacing.x; + let edit = ui.add_enabled( + connected, + egui::TextEdit::singleline(&mut self.draft) + .hint_text("Message the agent") + .desired_width(width), + ); + let submitted = edit.lost_focus() && ui.input(|i| i.key_pressed(egui::Key::Enter)); + let has_text = !self.draft.trim().is_empty(); + let send = egui::Button::new("Send").min_size(egui::vec2(SEND_BUTTON_WIDTH, 0.0)); + let clicked = ui.add_enabled(connected && has_text, send).clicked(); + if (submitted || clicked) && has_text { + let text = self.draft.trim().to_owned(); + self.draft.clear(); + self.send_failure = None; + self.service.send(AgentCmd::SendText(text)); + edit.request_focus(); + } + }); + if let Some(err) = &self.send_failure { + ui.add(StatusBadge::error(format!("Failed to send: {err}"))); + } + } +} + +/// Whether an agent participant is in the room. +fn has_joined(state: AgentState) -> bool { + matches!(state, AgentState::Connected(_)) +} + +impl From for StatusBadge { + fn from(state: AgentState) -> Self { + match state { + AgentState::Disconnected => StatusBadge::muted("Agent disconnected"), + AgentState::Connecting => StatusBadge::neutral("Waiting for agent..."), + AgentState::Connected(activity) => StatusBadge::from(activity), + AgentState::Failed(AgentFailure::Timeout) => { + StatusBadge::error("Agent did not connect in time") + } + AgentState::Failed(AgentFailure::Left) => StatusBadge::error("Agent left"), + state => StatusBadge::neutral(format!("Agent {state:?}")), + } + } +} + +impl From for StatusBadge { + fn from(activity: AgentActivity) -> Self { + match activity { + AgentActivity::Initializing => StatusBadge::neutral("Agent initializing"), + AgentActivity::Idle => StatusBadge::neutral("Agent idle"), + AgentActivity::Listening => StatusBadge::ok("Agent listening"), + AgentActivity::Thinking => StatusBadge::ok("Agent thinking"), + AgentActivity::Speaking => StatusBadge::ok("Agent speaking"), + activity => StatusBadge::neutral(format!("Agent {activity:?}")), + } + } +} + +fn speaker(message: &Message) -> &str { + match message.origin { + MessageOrigin::AgentTranscript => "Agent", + MessageOrigin::UserTranscript => "You (transcribed)", + MessageOrigin::UserInput => "You", + _ => message.participant_identity.as_str(), + } +} + +fn is_local(message: &Message) -> bool { + matches!( + message.origin, + MessageOrigin::UserTranscript | MessageOrigin::UserInput + ) +} diff --git a/src/connect.rs b/src/connect.rs deleted file mode 100644 index d3b043a..0000000 --- a/src/connect.rs +++ /dev/null @@ -1,452 +0,0 @@ -use crate::ui::{labeled_field::LabeledTextEdit, prominent_button::ProminentButton}; -use livekit_token_source::{ - TokenSourceConfigurable, TokenSourceFetchOptions, development_token_server, -}; - -/// How a room connection is authenticated. The methods that target a known -/// server carry its URL; the token-source method learns it from the development token server. -#[derive(Clone, Debug)] -pub enum Auth { - /// A pre-generated access token (the room is encoded in it). - Token { url: String, token: String }, - /// API credentials from which a join token is generated on demand. - ApiKey { - url: String, - api_key: String, - api_secret: String, - identity: String, - room: String, - }, - /// A LiveKit Cloud development token server, which provides both the server - /// URL and the join token. `options` customizes the request; unset - /// fields are left to server defaults. - DevelopmentTokenServer { - token_server_id: String, - options: TokenSourceFetchOptions, - }, -} - -impl Auth { - /// Resolve to `(server_url, token)`: the JWT is generated locally for the - /// API-key method, fetched over HTTP for the token-source method. - /// - /// Async because of that fetch — call it from the service task, not the UI - /// thread. - pub async fn connection_details(&self) -> Result<(String, String), String> { - match self { - Auth::Token { url, token } => Ok((url.clone(), token.clone())), - Auth::ApiKey { - url, - api_key, - api_secret, - identity, - room, - } => livekit_api::access_token::AccessToken::with_api_key(api_key, api_secret) - .with_identity(identity) - .with_name(identity) - .with_grants(livekit_api::access_token::VideoGrants { - room_join: true, - room: room.clone(), - ..Default::default() - }) - .to_jwt() - .map(|token| (url.clone(), token)) - .map_err(|e| e.to_string()), - Auth::DevelopmentTokenServer { - token_server_id, - options, - } => { - let token_source = development_token_server(token_server_id); - let response = token_source - .fetch(options) - .await - .map_err(|e| e.to_string())?; - Ok((response.server_url, response.participant_token)) - } - } - } - - /// Short label of the connection target for window titles: the server URL - /// when known up front, otherwise the token server id. - pub fn target_label(&self) -> &str { - match self { - Auth::Token { url, .. } | Auth::ApiKey { url, .. } => url, - Auth::DevelopmentTokenServer { - token_server_id, .. - } => token_server_id, - } - } -} - -/// Settings for connecting to a room, entered on the connect screen; one copy -/// per application, passed to each room window as it is opened. -#[derive(Clone)] -pub struct ConnectSettings { - pub auth: Auth, - pub key: String, - pub auto_subscribe: bool, - pub dynacast: bool, - pub enable_e2ee: bool, -} - -/// Which authentication method the connect form is editing. -#[derive( - Clone, Copy, Debug, PartialEq, Eq, Hash, Default, serde::Serialize, serde::Deserialize, -)] -enum AuthMethod { - #[default] - ApiKey, - Token, - DevelopmentTokenServer, -} - -#[derive(Default, serde::Serialize, serde::Deserialize)] -#[serde(default)] -struct TokenSourceOptions { - room_name: String, - participant_name: String, - participant_identity: String, - participant_metadata: String, - agent_name: String, - agent_metadata: String, - agent_deployment: String, -} - -impl From<&TokenSourceOptions> for TokenSourceFetchOptions { - fn from(value: &TokenSourceOptions) -> Self { - // Empty (or whitespace-only) fields are left unset so the - // token server applies its defaults. - let opt = |s: &str| { - let s = s.trim(); - (!s.is_empty()).then(|| s.to_string()) - }; - let mut options = TokenSourceFetchOptions::new(); - if let Some(v) = opt(&value.room_name) { - options = options.with_room_name(v); - } - if let Some(v) = opt(&value.participant_name) { - options = options.with_participant_name(v); - } - if let Some(v) = opt(&value.participant_identity) { - options = options.with_participant_identity(v); - } - if let Some(v) = opt(&value.participant_metadata) { - options = options.with_participant_metadata(v); - } - if let Some(v) = opt(&value.agent_name) { - options = options.with_agent_name(v); - } - if let Some(v) = opt(&value.agent_metadata) { - options = options.with_agent_metadata(v); - } - if let Some(v) = opt(&value.agent_deployment) { - options = options.with_deployment(v); - } - options - } -} - -/// The root window: a welcome screen holding the only connect form in the app. -/// Each successful Connect click spawns a dedicated room window. Inputs for both -/// auth methods are kept so switching tabs doesn't lose what was typed. -/// -/// Persisted between runs via eframe storage (see `AppRoot`); `#[serde(default)]` -/// lets a stored blob from an older version fall back to `Default` (env-seeded) -/// for any missing field. -#[derive(serde::Serialize, serde::Deserialize)] -#[serde(default)] -pub struct ConnectView { - method: AuthMethod, - url: String, - token: String, - token_server_id: String, - token_source_options: TokenSourceOptions, - api_key: String, - api_secret: String, - identity: String, - room: String, - key: String, - auto_subscribe: bool, - dynacast: bool, - enable_e2ee: bool, - // Not persisted: secrets always start hidden on launch. - #[serde(skip)] - show_secrets: bool, -} - -impl Default for ConnectView { - fn default() -> Self { - // Seed the fields from the standard LiveKit env vars when present, so a - // configured shell or `.env` pre-fills the form instead of the defaults. - let env_or = |key: &str, default: &str| { - std::env::var(key) - .ok() - .filter(|v| !v.is_empty()) - .unwrap_or_else(|| default.to_string()) - }; - Self { - method: AuthMethod::default(), - url: env_or("LIVEKIT_URL", "ws://localhost:7880"), - token: env_or("LIVEKIT_TOKEN", ""), - token_server_id: String::new(), - token_source_options: TokenSourceOptions::default(), - api_key: env_or("LIVEKIT_API_KEY", "devkey"), - api_secret: env_or("LIVEKIT_API_SECRET", "secret"), - identity: "participant-0".to_string(), - room: "dev-room".to_string(), - key: String::new(), - auto_subscribe: true, - dynacast: true, - enable_e2ee: false, - show_secrets: false, - } - } -} - -impl ConnectView { - fn is_connect_enabled(&self) -> bool { - // The URL only matters for the methods that use it; the token-source - // method gets its server URL from the development token server response. - match self.method { - AuthMethod::ApiKey => { - !self.url.trim().is_empty() - && !self.api_key.trim().is_empty() - && !self.api_secret.trim().is_empty() - && !self.identity.trim().is_empty() - && !self.room.trim().is_empty() - } - AuthMethod::Token => !self.url.trim().is_empty() && !self.token.trim().is_empty(), - AuthMethod::DevelopmentTokenServer => !self.token_server_id.trim().is_empty(), - } - } - - fn current_settings(&self) -> ConnectSettings { - let auth = match self.method { - AuthMethod::ApiKey => Auth::ApiKey { - url: self.url.clone(), - api_key: self.api_key.clone(), - api_secret: self.api_secret.clone(), - identity: self.identity.clone(), - room: self.room.clone(), - }, - AuthMethod::Token => Auth::Token { - url: self.url.clone(), - token: self.token.clone(), - }, - AuthMethod::DevelopmentTokenServer => { - let options = TokenSourceFetchOptions::from(&self.token_source_options); - Auth::DevelopmentTokenServer { - token_server_id: self.token_server_id.clone(), - options, - } - } - }; - ConnectSettings { - auth, - key: self.key.clone(), - auto_subscribe: self.auto_subscribe, - dynacast: self.dynacast, - enable_e2ee: self.enable_e2ee, - } - } - - /// If the identity ends in `-` (e.g. `alice-0`), increment that - /// - /// Identities must be unique within a room, so bumping it after each - /// Connect lets you spin up several participants in the same room with one click. - /// - fn bump_identity(&mut self) { - let Some((base, suffix)) = self.identity.rsplit_once('-') else { - return; - }; - let Ok(n) = suffix.parse::() else { - return; - }; - self.identity = format!("{base}-{}", n.saturating_add(1)); - } - - /// Returns the settings to open a room with when Connect is clicked. - pub fn ui(&mut self, ui: &mut egui::Ui) -> Option { - let mut request = None; - - let frame = egui::Frame::central_panel(ui.style()).inner_margin(16.0); - - egui::Panel::bottom("connect_info").show(ui, |ui| { - ui.horizontal(|ui| { - let version_string = format!("SDK Version: {}", livekit::SDK_VERSION); - let text = egui::RichText::new(version_string).text_style(egui::TextStyle::Small); - ui.label(text); - egui::Aligned::new(egui::Align2::RIGHT_CENTER) - .id_salt("debug_warning") - .show(ui, |ui| { - egui::warn_if_debug_build(ui); - }); - }); - }); - - egui::Panel::bottom("connect_action") - .frame(frame) - .show_separator_line(false) - .show(ui, |ui| { - let connect = - ui.add(ProminentButton::new("Connect").enabled(self.is_connect_enabled())); - if connect.clicked() { - request = Some(self.current_settings()); - if self.method == AuthMethod::ApiKey { - self.bump_identity(); - } - } - }); - - egui::CentralPanel::default().frame(frame).show(ui, |ui| { - ui.add(ConnectForm { view: self }); - }); - - request - } -} - -/// The connect form: URL, an auth-method tab selector, the active method's -/// fields, and the shared E2EE / options. The Connect button lives in the -/// connect screen's bottom panel, so this widget only edits the borrowed view. -struct ConnectForm<'a> { - view: &'a mut ConnectView, -} - -impl egui::Widget for ConnectForm<'_> { - fn ui(self, ui: &mut egui::Ui) -> egui::Response { - let view = self.view; - ui.vertical(|ui| { - ui.label(egui::RichText::new("Connect to a Room").text_style(egui::TextStyle::Heading)); - ui.add_space(8.0); - - ui.horizontal(|ui| { - ui.selectable_value(&mut view.method, AuthMethod::ApiKey, "API Key"); - ui.selectable_value(&mut view.method, AuthMethod::Token, "Token"); - ui.selectable_value( - &mut view.method, - AuthMethod::DevelopmentTokenServer, - "TokenSource", - ); - egui::Aligned::new(egui::Align2::RIGHT_CENTER) - .id_salt("secrets_toggle") - .show(ui, |ui| { - let toggle = ui - .add(egui::Button::selectable(view.show_secrets, "👁")) - .on_hover_text("Show/hide secrets"); - if toggle.clicked() { - view.show_secrets = !view.show_secrets; - } - }); - }); - ui.add_space(8.0); - - // Mask the secret fields (token / API key / secret / E2EE key) unless - // the eye toggle above is on. - let mask = !view.show_secrets; - - // Scope the method's fields under a *constant* id. The single-field - // tabs (Token / TokenSource) render the same full-width widget at the - // same rect, so a per-method salt would give that rect a different id - // each switch — which is exactly what egui's "rect changed id between - // passes" warning flags. A stable salt keeps the id constant. - // The URL lives inside the Token / API Key tabs (shared between - // them): the token-source method gets its URL from the development token server. - ui.push_id("auth_method_fields", |ui| match view.method { - AuthMethod::Token => { - ui.add(LabeledTextEdit::singleline("URL", &mut view.url)); - ui.add_space(8.0); - ui.add(LabeledTextEdit::singleline("Token", &mut view.token).password(mask)); - ui.add_space(8.0); - } - AuthMethod::ApiKey => { - ui.add(LabeledTextEdit::singleline("URL", &mut view.url)); - ui.add_space(8.0); - ui.columns(2, |columns| { - columns[0].add( - LabeledTextEdit::singleline("API Key", &mut view.api_key) - .password(mask), - ); - columns[1].add( - LabeledTextEdit::singleline("API Secret", &mut view.api_secret) - .password(mask), - ); - }); - ui.add_space(8.0); - ui.columns(2, |columns| { - columns[0].add(LabeledTextEdit::singleline("Identity", &mut view.identity)); - columns[1].add(LabeledTextEdit::singleline("Room", &mut view.room)); - }); - ui.add_space(8.0); - } - AuthMethod::DevelopmentTokenServer => { - ui.add(LabeledTextEdit::singleline( - "Token Server Id", - &mut view.token_server_id, - )); - ui.add_space(8.0); - - ui.label( - egui::RichText::new( - "Optional overrides — empty fields use server defaults", - ) - .text_style(egui::TextStyle::Small), - ); - ui.add_space(8.0); - ui.columns(2, |columns| { - columns[0].add(LabeledTextEdit::singleline( - "Room Name", - &mut view.token_source_options.room_name, - )); - columns[1].add(LabeledTextEdit::singleline( - "Participant Name", - &mut view.token_source_options.participant_name, - )); - }); - ui.add_space(8.0); - ui.columns(2, |columns| { - columns[0].add(LabeledTextEdit::singleline( - "Participant Identity", - &mut view.token_source_options.participant_identity, - )); - columns[1].add(LabeledTextEdit::singleline( - "Participant Metadata", - &mut view.token_source_options.participant_metadata, - )); - }); - ui.add_space(8.0); - ui.columns(2, |columns| { - columns[0].add(LabeledTextEdit::singleline( - "Agent Name", - &mut view.token_source_options.agent_name, - )); - columns[1].add(LabeledTextEdit::singleline( - "Agent Deployment", - &mut view.token_source_options.agent_deployment, - )); - }); - ui.add_space(8.0); - ui.add(LabeledTextEdit::singleline( - "Agent Metadata", - &mut view.token_source_options.agent_metadata, - )); - ui.add_space(8.0); - } - }); - - ui.add( - LabeledTextEdit::singleline("E2EE Key", &mut view.key) - .enabled(view.enable_e2ee) - .password(mask), - ); - ui.add_space(8.0); - - ui.checkbox(&mut view.enable_e2ee, "Enable E2EE"); - ui.checkbox(&mut view.auto_subscribe, "Auto Subscribe"); - ui.checkbox(&mut view.dynacast, "Dynacast"); - - ui.add_space(8.0); - }) - .response - } -} diff --git a/src/connect/agent.rs b/src/connect/agent.rs new file mode 100644 index 0000000..d150e73 --- /dev/null +++ b/src/connect/agent.rs @@ -0,0 +1,188 @@ +use super::{Auth, e2ee_options}; +use async_trait::async_trait; +use livekit::prelude::RoomOptions; +use livekit_api::access_token::{AccessToken, VideoGrants}; +use livekit_protocol::{RoomAgentDispatch, RoomConfiguration}; +use livekit_session::{Session, SessionOptions}; +use livekit_token_source::{ + TokenSourceError, TokenSourceFixed, TokenSourceResponse, TokenSourceResult, + development_token_server, literal, +}; +use std::hash::{BuildHasher, Hasher, RandomState}; +use std::time::Duration; + +/// Settings for starting an agent session, entered on the connect screen. +#[derive(Clone)] +pub struct AgentConnectSettings { + pub auth: Auth, + /// The shared E2EE key, if encryption is enabled. + pub e2ee_key: Option, + /// Capture the microphone and play the agent's audio through the + /// platform's audio devices; off for a conversation by text alone. + pub voice: bool, + pub agent_connect_timeout: Duration, +} + +/// A session that connects as the settings say; nothing happens until it is +/// started. +impl From<&AgentConnectSettings> for Session { + fn from(settings: &AgentConnectSettings) -> Self { + let mut room_options = RoomOptions::default(); + room_options.encryption = e2ee_options(settings.e2ee_key.as_deref()); + let session_options = SessionOptions::default() + .with_agent_connect_timeout(settings.agent_connect_timeout) + .with_room_options(room_options) + .with_platform_audio(settings.voice); + match settings.auth.clone() { + Auth::Token { url, token } => Session::new(literal(url, token), session_options), + Auth::ApiKey(token_source) => Session::new(token_source, session_options), + Auth::DevelopmentTokenServer { + token_server_id, + options: fetch_options, + } => Session::with_fetch_options( + development_token_server(token_server_id), + fetch_options, + session_options, + ), + } + } +} + +/// Mints a join token locally from API credentials. +#[derive(Clone, Debug)] +pub struct LocalTokenSource { + pub url: String, + pub api_key: String, + pub api_secret: String, + pub identity: String, + /// The room to join, or `None` for a new `agent-` room per fetch, + /// so that every session start dispatches the agent again. + pub room: Option, + /// Empty to dispatch no agent explicitly, leaving it to automatic dispatch. + pub agent_name: String, + pub agent_metadata: String, + /// Empty to target the agent's production deployment. + pub agent_deployment: String, +} + +impl LocalTokenSource { + /// A join token for the room, dispatching the agent if the room is new. + fn mint(&self) -> Result { + let room = self.room.clone().unwrap_or_else(random_room_name); + let mut token = AccessToken::with_api_key(&self.api_key, &self.api_secret) + .with_identity(&self.identity) + .with_name(&self.identity) + .with_grants(VideoGrants { + room_join: true, + room, + ..Default::default() + }); + if !self.agent_name.is_empty() { + token = token.with_room_config(RoomConfiguration { + agents: vec![RoomAgentDispatch { + agent_name: self.agent_name.clone(), + metadata: self.agent_metadata.clone(), + deployment: self.agent_deployment.clone(), + ..Default::default() + }], + ..Default::default() + }); + } + token.to_jwt().map_err(|e| e.to_string()) + } +} + +#[async_trait] +impl TokenSourceFixed for LocalTokenSource { + async fn fetch(&self) -> TokenSourceResult { + let participant_token = self.mint().map_err(|body| TokenSourceError::Server { + // `TokenSourceError` has no variant for local failures. + status: 0, + body: format!("failed to mint token: {body}"), + })?; + Ok(TokenSourceResponse { + server_url: self.url.clone(), + participant_token, + }) + } +} + +/// A room name unlikely to exist yet, so the agent is dispatched into it. +fn random_room_name() -> String { + // Every `RandomState` is seeded differently, which is all the randomness needed here. + let suffix = RandomState::new().build_hasher().finish() as u32; + format!("agent-{suffix:08x}") +} + +#[cfg(test)] +mod tests { + use super::*; + use livekit_api::access_token::TokenVerifier; + + fn source(agent_name: &str) -> LocalTokenSource { + LocalTokenSource { + url: "ws://localhost:7880".to_string(), + api_key: "devkey".to_string(), + api_secret: "secret".to_string(), + identity: "participant-0".to_string(), + room: None, + agent_name: agent_name.to_string(), + agent_metadata: "{}".to_string(), + agent_deployment: String::new(), + } + } + + fn claims(response: &TokenSourceResponse) -> livekit_api::access_token::Claims { + TokenVerifier::with_api_key("devkey", "secret") + .verify(&response.participant_token) + .expect("minted token verifies") + } + + #[tokio::test] + async fn every_fetch_uses_a_new_room() { + let source = source("my-agent"); + let first = claims(&source.fetch().await.unwrap()); + let second = claims(&source.fetch().await.unwrap()); + assert!(first.video.room.starts_with("agent-")); + assert_ne!(first.video.room, second.video.room); + } + + #[tokio::test] + async fn a_named_room_is_joined_as_is() { + let source = LocalTokenSource { + room: Some("dev-room".to_string()), + ..source("my-agent") + }; + let first = claims(&source.fetch().await.unwrap()); + let second = claims(&source.fetch().await.unwrap()); + assert_eq!(first.video.room, "dev-room"); + assert_eq!(second.video.room, "dev-room"); + } + + #[tokio::test] + async fn dispatches_the_named_agent() { + let response = source("my-agent").fetch().await.unwrap(); + let agents = claims(&response).room_config.unwrap().agents; + assert_eq!(agents.len(), 1); + assert_eq!(agents[0].agent_name, "my-agent"); + assert_eq!(agents[0].metadata, "{}"); + assert_eq!(agents[0].deployment, ""); + } + + #[tokio::test] + async fn targets_the_named_deployment() { + let source = LocalTokenSource { + agent_deployment: "preview".to_string(), + ..source("my-agent") + }; + let response = source.fetch().await.unwrap(); + let agents = claims(&response).room_config.unwrap().agents; + assert_eq!(agents[0].deployment, "preview"); + } + + #[tokio::test] + async fn no_agent_name_dispatches_nothing() { + let response = source("").fetch().await.unwrap(); + assert!(claims(&response).room_config.is_none()); + } +} diff --git a/src/connect/mod.rs b/src/connect/mod.rs new file mode 100644 index 0000000..53f8700 --- /dev/null +++ b/src/connect/mod.rs @@ -0,0 +1,645 @@ +use crate::ui::{labeled_field::LabeledTextEdit, prominent_button::ProminentButton}; +use livekit::e2ee::{ + E2eeOptions, EncryptionType, + key_provider::{KeyProvider, KeyProviderOptions}, +}; +use livekit_token_source::{ + TokenSourceConfigurable, TokenSourceFetchOptions, TokenSourceFixed, development_token_server, +}; +use std::time::Duration; + +mod agent; + +pub use agent::{AgentConnectSettings, LocalTokenSource}; + +/// How a room or agent connection is authenticated. The methods that target a +/// known server carry its URL; the token-source method learns it from the +/// development token server. +#[derive(Clone, Debug)] +pub enum Auth { + /// A pre-generated access token; the room, and any agent dispatch, are + /// encoded in it. + Token { url: String, token: String }, + /// API credentials from which a join token is minted on demand, + /// dispatching the source's agent if the room is new. + ApiKey(LocalTokenSource), + /// A LiveKit Cloud development token server, which provides both the server + /// URL and the join token. `options` customizes the request; unset + /// fields are left to server defaults. + DevelopmentTokenServer { + token_server_id: String, + options: TokenSourceFetchOptions, + }, +} + +impl Auth { + /// Resolve to `(server_url, token)`: the JWT is minted locally for the + /// API-key method, fetched over HTTP for the token-source method. + /// + /// Async because of that fetch — call it from the service task, not the UI + /// thread. + pub async fn connection_details(&self) -> Result<(String, String), String> { + let response = match self { + Auth::Token { url, token } => return Ok((url.clone(), token.clone())), + Auth::ApiKey(source) => source.fetch().await, + Auth::DevelopmentTokenServer { + token_server_id, + options, + } => { + development_token_server(token_server_id) + .fetch(options) + .await + } + }; + response + .map(|response| (response.server_url, response.participant_token)) + .map_err(|e| e.to_string()) + } + + /// Short label of the connection target for window titles: the agent being + /// dispatched, else the server URL when known up front, else the token + /// server id. + pub fn target_label(&self) -> &str { + match self { + Auth::Token { url, .. } => url, + Auth::ApiKey(source) if !source.agent_name.is_empty() => &source.agent_name, + Auth::ApiKey(source) => &source.url, + Auth::DevelopmentTokenServer { + token_server_id, .. + } => token_server_id, + } + } + + /// Whether reconnecting rejoins the same room, where an agent may not be + /// dispatched again. An agent session never asks a token server for a + /// room name (see `TokenSourceOptions::for_agent`). + pub fn reuses_room(&self) -> bool { + match self { + Auth::Token { .. } => true, + Auth::ApiKey(source) => source.room.is_some(), + Auth::DevelopmentTokenServer { .. } => false, + } + } +} + +/// Settings for connecting to a room, entered on the connect screen; one copy +/// per application, passed to each room window as it is opened. +#[derive(Clone)] +pub struct ConnectSettings { + pub auth: Auth, + /// The shared E2EE key, if encryption is enabled. + pub e2ee_key: Option, + pub auto_subscribe: bool, + pub dynacast: bool, +} + +/// Room encryption with the shared `key`, or `None` without one. +pub fn e2ee_options(key: Option<&str>) -> Option { + let key = key?; + Some(E2eeOptions { + encryption_type: EncryptionType::Gcm, + key_provider: KeyProvider::with_shared_key( + KeyProviderOptions::default(), + key.as_bytes().to_vec(), + ), + }) +} + +/// What a Connect click asks the app to open. +pub enum ConnectRequest { + Room(ConnectSettings), + Agent(AgentConnectSettings), +} + +/// Whether the connect form joins a room directly or starts an agent session. +#[derive( + Clone, Copy, Debug, PartialEq, Eq, Hash, Default, serde::Serialize, serde::Deserialize, +)] +enum ConnectMode { + #[default] + Room, + Agent, +} + +/// Which authentication method the connect form is editing. +#[derive( + Clone, Copy, Debug, PartialEq, Eq, Hash, Default, serde::Serialize, serde::Deserialize, +)] +enum AuthMethod { + #[default] + ApiKey, + Token, + DevelopmentTokenServer, +} + +#[derive(Clone, Default, serde::Serialize, serde::Deserialize)] +#[serde(default)] +struct TokenSourceOptions { + room_name: String, + participant_name: String, + participant_identity: String, + participant_metadata: String, + agent_name: String, + agent_metadata: String, + agent_deployment: String, +} + +impl TokenSourceOptions { + /// The options for an agent session: as for a room, but never naming the + /// room, so the agent is dispatched into a new one. + fn for_agent(&self) -> TokenSourceFetchOptions { + TokenSourceFetchOptions::from(&TokenSourceOptions { + room_name: String::new(), + ..self.clone() + }) + } +} + +impl From<&TokenSourceOptions> for TokenSourceFetchOptions { + fn from(value: &TokenSourceOptions) -> Self { + // Empty (or whitespace-only) fields are left unset so the + // token server applies its defaults. + let opt = |s: &str| { + let s = s.trim(); + (!s.is_empty()).then(|| s.to_string()) + }; + let mut options = TokenSourceFetchOptions::new(); + if let Some(v) = opt(&value.room_name) { + options = options.with_room_name(v); + } + if let Some(v) = opt(&value.participant_name) { + options = options.with_participant_name(v); + } + if let Some(v) = opt(&value.participant_identity) { + options = options.with_participant_identity(v); + } + if let Some(v) = opt(&value.participant_metadata) { + options = options.with_participant_metadata(v); + } + if let Some(v) = opt(&value.agent_name) { + options = options.with_agent_name(v); + } + if let Some(v) = opt(&value.agent_metadata) { + options = options.with_agent_metadata(v); + } + if let Some(v) = opt(&value.agent_deployment) { + options = options.with_deployment(v); + } + options + } +} + +/// The root window: a welcome screen holding the only connect form in the app. +/// Each successful Connect click spawns a dedicated room or agent window. Inputs +/// for every auth method are kept, and shared between the room and agent modes, +/// so switching tabs doesn't lose what was typed. +/// +/// Persisted between runs via eframe storage (see `AppRoot`); `#[serde(default)]` +/// lets a stored blob from an older version fall back to `Default` (env-seeded) +/// for any missing field. +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(default)] +pub struct ConnectView { + mode: ConnectMode, + method: AuthMethod, + url: String, + token: String, + token_server_id: String, + token_source_options: TokenSourceOptions, + api_key: String, + api_secret: String, + identity: String, + room: String, + key: String, + auto_subscribe: bool, + dynacast: bool, + enable_e2ee: bool, + voice: bool, + agent_connect_timeout_secs: u64, + // Not persisted: secrets always start hidden on launch. + #[serde(skip)] + show_secrets: bool, +} + +impl Default for ConnectView { + fn default() -> Self { + // Seed the fields from the standard LiveKit env vars when present, so a + // configured shell or `.env` pre-fills the form instead of the defaults. + let env_or = |key: &str, default: &str| { + std::env::var(key) + .ok() + .filter(|v| !v.is_empty()) + .unwrap_or_else(|| default.to_string()) + }; + Self { + mode: ConnectMode::default(), + method: AuthMethod::default(), + url: env_or("LIVEKIT_URL", "ws://localhost:7880"), + token: env_or("LIVEKIT_TOKEN", ""), + token_server_id: String::new(), + token_source_options: TokenSourceOptions::default(), + api_key: env_or("LIVEKIT_API_KEY", "devkey"), + api_secret: env_or("LIVEKIT_API_SECRET", "secret"), + identity: "participant-0".to_string(), + room: "dev-room".to_string(), + key: String::new(), + auto_subscribe: true, + dynacast: true, + enable_e2ee: false, + voice: true, + agent_connect_timeout_secs: 20, + show_secrets: false, + } + } +} + +impl ConnectView { + fn is_connect_enabled(&self) -> bool { + let filled = |s: &str| !s.trim().is_empty(); + // The URL only matters for the methods that use it; the token-source + // method gets its server URL from the development token server response. + // Agent sessions never name a room. + match self.method { + AuthMethod::ApiKey => { + filled(&self.url) + && filled(&self.api_key) + && filled(&self.api_secret) + && filled(&self.identity) + && (self.mode == ConnectMode::Agent || filled(&self.room)) + } + AuthMethod::Token => filled(&self.url) && filled(&self.token), + AuthMethod::DevelopmentTokenServer => filled(&self.token_server_id), + } + } + + fn current_request(&self) -> ConnectRequest { + match self.mode { + ConnectMode::Room => ConnectRequest::Room(self.room_settings()), + ConnectMode::Agent => ConnectRequest::Agent(self.agent_settings()), + } + } + + fn local_token_source(&self) -> LocalTokenSource { + let options = &self.token_source_options; + LocalTokenSource { + url: self.url.clone(), + api_key: self.api_key.clone(), + api_secret: self.api_secret.clone(), + identity: self.identity.clone(), + // An agent session gets a new room each time; see `LocalTokenSource::room`. + room: (self.mode == ConnectMode::Room).then(|| self.room.clone()), + agent_name: options.agent_name.trim().to_string(), + agent_metadata: options.agent_metadata.trim().to_string(), + agent_deployment: options.agent_deployment.trim().to_string(), + } + } + + /// The selected method's credentials, for the selected mode. + fn auth(&self) -> Auth { + match self.method { + AuthMethod::ApiKey => Auth::ApiKey(self.local_token_source()), + AuthMethod::Token => Auth::Token { + url: self.url.clone(), + token: self.token.clone(), + }, + AuthMethod::DevelopmentTokenServer => Auth::DevelopmentTokenServer { + token_server_id: self.token_server_id.clone(), + options: match self.mode { + ConnectMode::Room => TokenSourceFetchOptions::from(&self.token_source_options), + ConnectMode::Agent => self.token_source_options.for_agent(), + }, + }, + } + } + + fn e2ee_key(&self) -> Option { + self.enable_e2ee.then(|| self.key.clone()) + } + + fn room_settings(&self) -> ConnectSettings { + ConnectSettings { + auth: self.auth(), + e2ee_key: self.e2ee_key(), + auto_subscribe: self.auto_subscribe, + dynacast: self.dynacast, + } + } + + fn agent_settings(&self) -> AgentConnectSettings { + AgentConnectSettings { + auth: self.auth(), + e2ee_key: self.e2ee_key(), + voice: self.voice, + agent_connect_timeout: Duration::from_secs(self.agent_connect_timeout_secs), + } + } + + /// If the identity ends in `-` (e.g. `alice-0`), increment that + /// + /// Identities must be unique within a room, so bumping it after each + /// Connect lets you spin up several participants in the same room with one click. + /// + fn bump_identity(&mut self) { + let Some((base, suffix)) = self.identity.rsplit_once('-') else { + return; + }; + let Ok(n) = suffix.parse::() else { + return; + }; + self.identity = format!("{base}-{}", n.saturating_add(1)); + } + + /// Returns what to open when Connect is clicked. + pub fn ui(&mut self, ui: &mut egui::Ui) -> Option { + let mut request = None; + + let frame = egui::Frame::central_panel(ui.style()).inner_margin(16.0); + + egui::Panel::bottom("connect_info").show(ui, |ui| { + ui.horizontal(|ui| { + let version_string = format!("SDK Version: {}", livekit::SDK_VERSION); + let text = egui::RichText::new(version_string).text_style(egui::TextStyle::Small); + ui.label(text); + egui::Aligned::new(egui::Align2::RIGHT_CENTER) + .id_salt("debug_warning") + .show(ui, |ui| { + egui::warn_if_debug_build(ui); + }); + }); + }); + + egui::Panel::bottom("connect_action") + .frame(frame) + .show_separator_line(false) + .show(ui, |ui| { + let connect = + ui.add(ProminentButton::new("Connect").enabled(self.is_connect_enabled())); + if connect.clicked() { + request = Some(self.current_request()); + // Agent sessions get a new room each time, so only a room + // join needs a fresh identity. + if self.mode == ConnectMode::Room && self.method == AuthMethod::ApiKey { + self.bump_identity(); + } + } + }); + + egui::CentralPanel::default().frame(frame).show(ui, |ui| { + ui.add(ConnectForm { view: self }); + }); + + request + } +} + +/// The connect form: a room / agent mode selector, an auth-method tab selector, +/// the active method's fields, and the mode's options. The Connect button lives +/// in the connect screen's bottom panel, so this widget only edits the borrowed view. +struct ConnectForm<'a> { + view: &'a mut ConnectView, +} + +impl egui::Widget for ConnectForm<'_> { + fn ui(self, ui: &mut egui::Ui) -> egui::Response { + let view = self.view; + ui.vertical(|ui| { + ui.horizontal(|ui| { + for (mode, label) in [ + (ConnectMode::Room, "Connect to a Room"), + (ConnectMode::Agent, "Connect to an Agent"), + ] { + let text = egui::RichText::new(label).text_style(egui::TextStyle::Heading); + ui.selectable_value(&mut view.mode, mode, text); + } + }); + ui.add_space(8.0); + + // The method is shared between the modes, so switching keeps the tab. + let mode = view.mode; + let method = &mut view.method; + ui.horizontal(|ui| { + ui.selectable_value(method, AuthMethod::ApiKey, "API Key") + .on_hover_text(match mode { + ConnectMode::Room => "Generate a join token from API credentials", + ConnectMode::Agent => { + "Generate a token dispatching the agent into a new room" + } + }); + ui.selectable_value(method, AuthMethod::Token, "Token") + .on_hover_text(match mode { + ConnectMode::Room => "Join with a pre-generated token", + ConnectMode::Agent => { + "Join with a pre-generated token that dispatches the agent" + } + }); + ui.selectable_value(method, AuthMethod::DevelopmentTokenServer, "TokenSource") + .on_hover_text("Fetch credentials from a development token server"); + egui::Aligned::new(egui::Align2::RIGHT_CENTER) + .id_salt("secrets_toggle") + .show(ui, |ui| { + let toggle = ui + .add(egui::Button::selectable(view.show_secrets, "👁")) + .on_hover_text("Show/hide secrets"); + if toggle.clicked() { + view.show_secrets = !view.show_secrets; + } + }); + }); + ui.add_space(8.0); + + // Mask the secret fields (token / API key / secret / E2EE key) unless + // the eye toggle above is on. + let mask = !view.show_secrets; + + // Scope the method's fields under a *constant* id. The single-field + // tabs (Token / TokenSource) render the same full-width widget at the + // same rect, so a per-method salt would give that rect a different id + // each switch — which is exactly what egui's "rect changed id between + // passes" warning flags. A stable salt keeps the id constant. + // The URL lives inside the Token / API Key tabs (shared between + // them): the token-source method gets its URL from the development token server. + ui.push_id("auth_method_fields", |ui| match view.method { + AuthMethod::Token => { + token_fields(ui, view, mask); + } + AuthMethod::ApiKey => { + api_key_fields(ui, view, mask, mode); + } + AuthMethod::DevelopmentTokenServer => { + token_server_fields(ui, view, mode); + } + }); + + ui.add( + LabeledTextEdit::singleline("E2EE Key", &mut view.key) + .enabled(view.enable_e2ee) + .password(mask), + ); + ui.add_space(8.0); + + ui.checkbox(&mut view.enable_e2ee, "Enable E2EE"); + match mode { + ConnectMode::Room => { + ui.checkbox(&mut view.auto_subscribe, "Auto Subscribe"); + ui.checkbox(&mut view.dynacast, "Dynacast"); + } + ConnectMode::Agent => { + ui.checkbox(&mut view.voice, "Voice").on_hover_text( + "Capture the microphone and play the agent's audio through the \ + platform's audio devices; uncheck to talk by text only", + ); + ui.horizontal(|ui| { + ui.label("Agent connect timeout"); + ui.add( + egui::DragValue::new(&mut view.agent_connect_timeout_secs) + .range(1..=300) + .suffix(" s"), + ); + }) + .response + .on_hover_text( + "How long a dispatched agent gets to become conversational before \ + the session reports it failed", + ); + } + } + + ui.add_space(8.0); + }) + .response + } +} + +fn token_fields(ui: &mut egui::Ui, view: &mut ConnectView, mask: bool) { + ui.add(LabeledTextEdit::singleline("URL", &mut view.url)); + ui.add_space(8.0); + ui.add(LabeledTextEdit::singleline("Token", &mut view.token).password(mask)); + ui.add_space(8.0); +} + +/// The API credentials, identity, room, and the agent to dispatch. +fn api_key_fields(ui: &mut egui::Ui, view: &mut ConnectView, mask: bool, mode: ConnectMode) { + ui.add(LabeledTextEdit::singleline("URL", &mut view.url)); + ui.add_space(8.0); + ui.columns(2, |columns| { + columns[0].add(LabeledTextEdit::singleline("API Key", &mut view.api_key).password(mask)); + columns[1] + .add(LabeledTextEdit::singleline("API Secret", &mut view.api_secret).password(mask)); + }); + ui.add_space(8.0); + ui.columns(2, |columns| { + columns[0].add(LabeledTextEdit::singleline("Identity", &mut view.identity)); + columns[1] + .add( + LabeledTextEdit::singleline("Room", &mut view.room) + .enabled(mode == ConnectMode::Room), + ) + .on_disabled_hover_text(NEW_ROOM_HINT); + }); + ui.add_space(8.0); + let options = &mut view.token_source_options; + ui.columns(2, |columns| { + columns[0] + .add(LabeledTextEdit::singleline( + "Agent Name", + &mut options.agent_name, + )) + .on_hover_text( + "Dispatched when the room is created, for room joins as well as agent \ + sessions; leave empty to rely on automatic agent dispatch", + ); + columns[1] + .add(LabeledTextEdit::singleline( + "Agent Deployment", + &mut options.agent_deployment, + )) + .on_hover_text( + "Targets a deployment of the named agent; leave empty for its production \ + deployment", + ); + }); + ui.add_space(8.0); + ui.add(LabeledTextEdit::singleline( + "Agent Metadata", + &mut options.agent_metadata, + )); + ui.add_space(8.0); +} + +/// Why the room name fields are disabled in agent mode. They stay in place to +/// keep the layout from shifting between the modes. +const NEW_ROOM_HINT: &str = + "Agent sessions always use a new room, so the agent is dispatched into it"; + +/// The development token server id and its optional overrides. Agent sessions +/// leave out the room name, so each one is dispatched into a new room. +fn token_server_fields(ui: &mut egui::Ui, view: &mut ConnectView, mode: ConnectMode) { + ui.add(LabeledTextEdit::singleline( + "Token Server Id", + &mut view.token_server_id, + )); + ui.add_space(8.0); + + ui.label( + egui::RichText::new("Optional overrides — empty fields use server defaults") + .text_style(egui::TextStyle::Small), + ); + ui.add_space(8.0); + let options = &mut view.token_source_options; + ui.columns(2, |columns| { + columns[0] + .add( + LabeledTextEdit::singleline("Room Name", &mut options.room_name) + .enabled(mode == ConnectMode::Room), + ) + .on_disabled_hover_text(NEW_ROOM_HINT); + columns[1].add(LabeledTextEdit::singleline( + "Participant Name", + &mut options.participant_name, + )); + }); + ui.add_space(8.0); + ui.columns(2, |columns| { + columns[0].add(LabeledTextEdit::singleline( + "Participant Identity", + &mut options.participant_identity, + )); + columns[1].add(LabeledTextEdit::singleline( + "Participant Metadata", + &mut options.participant_metadata, + )); + }); + ui.add_space(8.0); + ui.columns(2, |columns| { + columns[0].add(LabeledTextEdit::singleline( + "Agent Name", + &mut options.agent_name, + )); + columns[1].add(LabeledTextEdit::singleline( + "Agent Deployment", + &mut options.agent_deployment, + )); + }); + ui.add_space(8.0); + ui.add(LabeledTextEdit::singleline( + "Agent Metadata", + &mut options.agent_metadata, + )); + ui.add_space(8.0); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn agent_options_never_name_the_room() { + let options = TokenSourceOptions { + room_name: "dev-room".to_string(), + agent_name: "my-agent".to_string(), + ..Default::default() + }; + let expected = TokenSourceFetchOptions::new().with_agent_name("my-agent"); + assert_eq!(options.for_agent(), expected); + } +} diff --git a/src/lib.rs b/src/lib.rs index de8ba2f..46d0531 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,9 +1,11 @@ -use crate::connect::{ConnectSettings, ConnectView}; +use crate::agent::AgentWindow; +use crate::connect::{ConnectRequest, ConnectView}; use crate::room::RoomWindow; use parking_lot::Mutex; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; +mod agent; mod connect; mod media; mod room; @@ -17,15 +19,38 @@ pub static APP_NAME: &str = "LiveKit Client"; /// eframe storage key under which the connect form is persisted between runs. const CONNECT_STORAGE_KEY: &str = "connect"; -/// A room window, shown as a deferred viewport so it repaints independently of -/// the connect screen and of other rooms. +/// The kinds of window a Connect click can open. +enum AppWindow { + // Both are large; boxed so the enum stays small. + Room(Box), + Agent(Box), +} + +impl AppWindow { + fn ui(&mut self, ui: &mut egui::Ui) { + match self { + AppWindow::Room(window) => window.ui(ui), + AppWindow::Agent(window) => window.ui(ui), + } + } + + fn disconnect(&self) { + match self { + AppWindow::Room(window) => window.disconnect(), + AppWindow::Agent(window) => window.disconnect(), + } + } +} + +/// A room or agent window, shown as a deferred viewport so it repaints +/// independently of the connect screen and of other windows. struct WindowEntry { id: u64, title: String, open: Arc, /// Shared with the deferred viewport callback, which runs outside of /// [`AppRoot::ui`] and therefore needs `Send + Sync` access. - window: Arc>, + window: Arc>, } pub struct AppRoot { @@ -63,16 +88,22 @@ impl AppRoot { } } - fn open_room(&mut self, request: ConnectSettings) { + fn open_window(&mut self, request: ConnectRequest) { let id = self.next_window_id; self.next_window_id += 1; - let title = format!("{} - {}", APP_NAME, request.auth.target_label()); - let window = RoomWindow::new( - self.async_runtime.handle().clone(), - self.render_state.clone(), - request, - ); + let runtime = self.async_runtime.handle().clone(); + let render_state = self.render_state.clone(); + let (title, window) = match request { + ConnectRequest::Room(settings) => ( + format!("{} - {}", APP_NAME, settings.auth.target_label()), + AppWindow::Room(Box::new(RoomWindow::new(runtime, render_state, settings))), + ), + ConnectRequest::Agent(settings) => ( + format!("{} - Agent - {}", APP_NAME, settings.auth.target_label()), + AppWindow::Agent(Box::new(AgentWindow::new(runtime, render_state, &settings))), + ), + }; self.windows.push(WindowEntry { id, @@ -95,13 +126,13 @@ impl eframe::App for AppRoot { fn ui(&mut self, ui: &mut egui::Ui, _frame: &mut eframe::Frame) { if let Some(request) = self.connect.ui(ui) { - self.open_room(request); + self.open_window(request); } self.windows.retain(|window| { let open = window.open.load(Ordering::Relaxed); if !open { - // Dropping LkService alone doesn't close the room; disconnect first. + // Dropping a room window's service alone doesn't close the room; disconnect first. window.window.lock().disconnect(); } open diff --git a/src/main.rs b/src/main.rs index af6e353..c171206 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,7 +7,7 @@ fn main() { .init(); let viewport = egui::ViewportBuilder::default() - .with_inner_size([380.0, 480.0]) + .with_inner_size([420.0, 600.0]) .with_resizable(false) .with_minimize_button(false); @@ -15,7 +15,7 @@ fn main() { viewport, centered: true, // Persist only the connect form (see AppRoot::save), not window geometry — - // the root window is fixed-size and centered. + // the root window always opens centered at its default size. persist_window: false, ..Default::default() }; diff --git a/src/media/video_renderer.rs b/src/media/video_renderer.rs index 07e0b72..092eb94 100644 --- a/src/media/video_renderer.rs +++ b/src/media/video_renderer.rs @@ -1,3 +1,4 @@ +use egui::load::SizedTexture; use futures::StreamExt; use livekit::webrtc::native::yuv_helper; use livekit::webrtc::prelude::*; @@ -102,9 +103,12 @@ impl VideoRenderer { } } - // Returns the texture id, can be used to draw the texture on the UI - pub fn texture_id(&self) -> Option { - self.internal.lock().egui_texture + /// The latest frame as a sized texture, or `None` before the first frame. + pub fn texture(&self) -> Option { + let internal = self.internal.lock(); + let id = internal.egui_texture?; + let size = egui::vec2(internal.width as f32, internal.height as f32); + Some(SizedTexture::new(id, size)) } } diff --git a/src/room/context.rs b/src/room/context.rs index 77a3c56..a56c32f 100644 --- a/src/room/context.rs +++ b/src/room/context.rs @@ -3,6 +3,11 @@ use livekit::prelude::Room; /// Shared, per-frame context passed to a room window's widgets: the service and /// the optional connected room. +/// +/// The rule for widgets: one that only reads the room takes `Option<&Room>` +/// directly (like `StatusBar` and `ParticipantsPanel`), which also lets the +/// agent window reuse it; one that sends [`AsyncCmd`](crate::service::AsyncCmd)s +/// takes `&RoomContext`. pub struct RoomContext<'a> { pub service: &'a LkService, pub room: Option<&'a Room>, diff --git a/src/room/mod.rs b/src/room/mod.rs index aa91fd2..c6a11cc 100644 --- a/src/room/mod.rs +++ b/src/room/mod.rs @@ -1,5 +1,6 @@ //! The room feature: the per-room window plus its room-context widgets and -//! state. Depends on the reusable toolkit in [`crate::ui`]. +//! state. Depends on the reusable toolkit in [`crate::ui`]. The [`participants`] +//! and [`status_bar`] widgets are also used by [`crate::agent`]. pub mod context; pub mod data_streams; diff --git a/src/room/participants.rs b/src/room/participants.rs index 94a5f2c..344bc8a 100644 --- a/src/room/participants.rs +++ b/src/room/participants.rs @@ -1,19 +1,47 @@ -use crate::room::RoomContext; -use crate::service::AsyncCmd; use crate::ui::status_badge::StatusBadge; use livekit::{e2ee::EncryptionType, prelude::*, track::VideoQuality}; +/// A change to a remote track subscription, requested from the participants list. +#[derive(Debug)] +pub enum TrackCmd { + Subscribe(RemoteTrackPublication), + Unsubscribe(RemoteTrackPublication), + SetVideoQuality(RemoteTrackPublication, VideoQuality), +} + +impl TrackCmd { + /// Must run inside the Tokio runtime, hence the round trip through a + /// service actor: `set_subscribed` calls `tokio::spawn` through the + /// subscription callback, which panics on the egui thread. + pub fn apply(self) { + match self { + TrackCmd::Subscribe(publication) => publication.set_subscribed(true), + TrackCmd::Unsubscribe(publication) => publication.set_subscribed(false), + TrackCmd::SetVideoQuality(publication, quality) => { + publication.set_video_quality(quality) + } + } + } +} + +/// Where the participants list sends its [`TrackCmd`]s, so it works with any +/// service that owns a room. +pub trait TrackCommands { + fn track_cmd(&self, cmd: TrackCmd); +} + /// Scrollable list of the local and remote participants and their track /// publications. pub struct ParticipantsPanel<'a> { - pub ctx: &'a RoomContext<'a>, + pub room: Option<&'a Room>, + pub commands: &'a dyn TrackCommands, } impl egui::Widget for ParticipantsPanel<'_> { fn ui(self, ui: &mut egui::Ui) -> egui::Response { - let ctx = self.ctx; + let ParticipantsPanel { room, commands } = self; ui.scope(|ui| { - let Some(room) = ctx.room else { + let Some(room) = room else { ui.label("Not connected"); return; }; @@ -24,7 +52,7 @@ impl egui::Widget for ParticipantsPanel<'_> { // Local participant first, so "you" are always visible and // stable at the top. ui.add(ParticipantCard { - ctx, + commands, participant: Participant::Local(room.local_participant()), }); @@ -39,7 +67,7 @@ impl egui::Widget for ParticipantsPanel<'_> { for psid in sorted_participants { let participant = participants.get(&psid).unwrap().clone(); ui.add(ParticipantCard { - ctx, + commands, participant: Participant::Remote(participant), }); } @@ -52,13 +80,16 @@ impl egui::Widget for ParticipantsPanel<'_> { /// One participant: a collapsible section titled by its identity, with a row per /// track publication in the body. struct ParticipantCard<'a> { - ctx: &'a RoomContext<'a>, + commands: &'a dyn TrackCommands, participant: Participant, } impl egui::Widget for ParticipantCard<'_> { fn ui(self, ui: &mut egui::Ui) -> egui::Response { - let ParticipantCard { ctx, participant } = self; + let ParticipantCard { + commands, + participant, + } = self; let is_local = matches!(participant, Participant::Local(_)); let identity = participant.identity().0; let title = if is_local { @@ -88,7 +119,10 @@ impl egui::Widget for ParticipantCard<'_> { for tsid in sorted_tracks { let publication = tracks.get(&tsid).unwrap().clone(); - ui.add(TrackPublicationRow { ctx, publication }); + ui.add(TrackPublicationRow { + commands, + publication, + }); } }) .header_response @@ -99,13 +133,16 @@ impl egui::Widget for ParticipantCard<'_> { /// The quality menu and subscribe/unsubscribe controls apply only to remote /// tracks, so they are hidden for the local participant's own publications. struct TrackPublicationRow<'a> { - ctx: &'a RoomContext<'a>, + commands: &'a dyn TrackCommands, publication: TrackPublication, } impl egui::Widget for TrackPublicationRow<'_> { fn ui(self, ui: &mut egui::Ui) -> egui::Response { - let TrackPublicationRow { ctx, publication } = self; + let TrackPublicationRow { + commands, + publication, + } = self; ui.vertical(|ui| { ui.horizontal(|ui| { ui.label("Encrypted - "); @@ -135,20 +172,20 @@ impl egui::Widget for TrackPublicationRow<'_> { { ui.menu_button("Set Quality", |ui| { if ui.button("Low").clicked() { - let _ = ctx.service.send(AsyncCmd::SetVideoQuality { - publication: remote.clone(), - quality: VideoQuality::Low, - }); + commands.track_cmd(TrackCmd::SetVideoQuality( + remote.clone(), + VideoQuality::Low, + )); } else if ui.button("Medium").clicked() { - let _ = ctx.service.send(AsyncCmd::SetVideoQuality { - publication: remote.clone(), - quality: VideoQuality::Medium, - }); + commands.track_cmd(TrackCmd::SetVideoQuality( + remote.clone(), + VideoQuality::Medium, + )); } else if ui.button("High").clicked() { - let _ = ctx.service.send(AsyncCmd::SetVideoQuality { - publication: remote.clone(), - quality: VideoQuality::High, - }); + commands.track_cmd(TrackCmd::SetVideoQuality( + remote.clone(), + VideoQuality::High, + )); } }); } @@ -170,14 +207,10 @@ impl egui::Widget for TrackPublicationRow<'_> { if remote.is_subscribed() { if ui.button("Unsubscribe").clicked() { - let _ = ctx.service.send(AsyncCmd::UnsubscribeTrack { - publication: remote.clone(), - }); + commands.track_cmd(TrackCmd::Unsubscribe(remote.clone())); } } else if ui.button("Subscribe").clicked() { - let _ = ctx.service.send(AsyncCmd::SubscribeTrack { - publication: remote.clone(), - }); + commands.track_cmd(TrackCmd::Subscribe(remote.clone())); } } }); diff --git a/src/room/right_panel.rs b/src/room/right_panel.rs index 0cb1efd..4bee8f5 100644 --- a/src/room/right_panel.rs +++ b/src/room/right_panel.rs @@ -43,7 +43,10 @@ impl egui::Widget for RightPanel<'_> { match state.tab { RightTab::Participants => { - ui.add(ParticipantsPanel { ctx }); + ui.add(ParticipantsPanel { + room: ctx.room, + commands: ctx.service, + }); } RightTab::Rpc => { ui.add(RpcPanel { diff --git a/src/room/status_bar.rs b/src/room/status_bar.rs index 1c747af..52ab83b 100644 --- a/src/room/status_bar.rs +++ b/src/room/status_bar.rs @@ -1,4 +1,3 @@ -use crate::room::RoomContext; use crate::ui::status_badge::StatusBadge; use livekit::prelude::*; @@ -6,8 +5,8 @@ use livekit::prelude::*; /// between Disconnect / Reconnect / Connecting. const BUTTON_WIDTH: f32 = 100.0; -/// Connect/disconnect intents raised by the status bar. It only holds -/// `&RoomContext`, so it records the intent here and the window applies it +/// Connect/disconnect intents raised by the status bar. The bar borrows the +/// window's state, so it records the intent here and the window applies it /// (with `&mut self`) after the borrow ends. #[derive(Default)] pub struct StatusBarActions { @@ -15,64 +14,114 @@ pub struct StatusBarActions { pub disconnect: bool, } -/// Bottom status bar: the current connection status on the left and a -/// connect/disconnect button floated to the right. +/// Bottom status bar: the connection status on the left, an optional detail +/// next to it, and a connect/disconnect button floated to the right. Shared by +/// the room and agent windows. pub struct StatusBar<'a> { - pub ctx: &'a RoomContext<'a>, + pub room: Option<&'a Room>, pub connecting: bool, pub connection_failure: Option<&'a str>, + /// Shown after the connection status, e.g. the agent's state. + pub detail: Option, + /// Hover text for the Reconnect button. + pub reconnect_hint: Option<&'a str>, pub actions: &'a mut StatusBarActions, } impl egui::Widget for StatusBar<'_> { fn ui(self, ui: &mut egui::Ui) -> egui::Response { let StatusBar { - ctx, + room, connecting, connection_failure, + detail, + reconnect_hint, actions, } = self; - let room = ctx - .room - .filter(|r| r.connection_state() == ConnectionState::Connected); + let state = room.map(Room::connection_state); ui.horizontal(|ui| { if let Some(err) = connection_failure { ui.add(StatusBadge::error(format!("Connection failed: {err}"))); } else if connecting { ui.label("Connecting..."); - } else if let Some(room) = room { - ui.label(format!( - "Connected to '{}' as '{}'", - room.name(), - room.local_participant().identity() - )); } else { - ui.label("Disconnected"); + match room.zip(state) { + Some((room, ConnectionState::Connected)) => { + ui.label(format!( + "Connected to '{}' as '{}'", + room.name(), + room.local_participant().identity() + )); + } + Some((_, ConnectionState::Reconnecting)) => { + ui.label("Reconnecting..."); + } + _ => { + ui.label("Disconnected"); + } + } } + if let Some(detail) = detail { + ui.separator(); + ui.add(detail); + } + + ui.add(ConnectionButton { + connecting, + connected: matches!( + state, + Some(ConnectionState::Connected | ConnectionState::Reconnecting) + ), + reconnect_hint, + actions, + }); + }) + .response + } +} - egui::Aligned::new(egui::Align2::RIGHT_CENTER) - .id_salt("status_actions") - .show(ui, |ui| { - let size = egui::vec2(BUTTON_WIDTH, 0.0); - if connecting { - ui.add_enabled(false, egui::Button::new("Connecting...").min_size(size)); - } else if room.is_some() { - if ui - .add(egui::Button::new("Disconnect").min_size(size)) - .clicked() - { - actions.disconnect = true; - } - } else if ui - .add(egui::Button::new("Reconnect").min_size(size)) +/// The Connecting / Disconnect / Reconnect button, floated to the right of the +/// status bar. +struct ConnectionButton<'a> { + connecting: bool, + connected: bool, + reconnect_hint: Option<&'a str>, + actions: &'a mut StatusBarActions, +} + +impl egui::Widget for ConnectionButton<'_> { + fn ui(self, ui: &mut egui::Ui) -> egui::Response { + let ConnectionButton { + connecting, + connected, + reconnect_hint, + actions, + } = self; + egui::Aligned::new(egui::Align2::RIGHT_CENTER) + .id_salt("status_actions") + .show(ui, |ui| { + let size = egui::vec2(BUTTON_WIDTH, 0.0); + if connecting { + ui.add_enabled(false, egui::Button::new("Connecting...").min_size(size)); + } else if connected { + if ui + .add(egui::Button::new("Disconnect").min_size(size)) .clicked() { + actions.disconnect = true; + } + } else { + let mut reconnect = ui.add(egui::Button::new("Reconnect").min_size(size)); + if let Some(hint) = reconnect_hint { + reconnect = reconnect.on_hover_text(hint); + } + if reconnect.clicked() { actions.reconnect = true; } - }); - }) - .response + } + }) + .response } } diff --git a/src/room/track_grid_view.rs b/src/room/track_grid_view.rs index 0b8cc92..35f5e90 100644 --- a/src/room/track_grid_view.rs +++ b/src/room/track_grid_view.rs @@ -76,12 +76,12 @@ impl egui::Widget for TrackGridView<'_> { else { continue; }; - if renderer.texture_id().is_none() { + let Some(texture) = renderer.texture() else { continue; // no frame decoded yet - } + }; ui.track_frame(|ui| { ui.add(VideoTile::new( - renderer.texture_id(), + Some(texture), identity.as_str(), speaking, )); diff --git a/src/room/window.rs b/src/room/window.rs index 57b1682..58e2807 100644 --- a/src/room/window.rs +++ b/src/room/window.rs @@ -59,8 +59,7 @@ impl RoomWindow { auth: Box::new(self.request.auth.clone()), auto_subscribe: self.request.auto_subscribe, dynacast: self.request.dynacast, - enable_e2ee: self.request.enable_e2ee, - key: self.request.key.clone(), + e2ee_key: self.request.e2ee_key.clone(), }); } @@ -220,9 +219,11 @@ impl RoomWindow { ) .show(ui, |ui| { ui.add(StatusBar { - ctx: &ctx, + room: ctx.room, connecting: self.connecting, connection_failure: self.connection_failure.as_deref(), + detail: None, + reconnect_hint: None, actions: &mut actions, }); }); diff --git a/src/service.rs b/src/service.rs index 345b0ab..781ec47 100644 --- a/src/service.rs +++ b/src/service.rs @@ -1,13 +1,9 @@ -use crate::connect::Auth; +use crate::connect::{Auth, e2ee_options}; use crate::media::{ CaptureSource, CaptureTrack, MicTrack, OnUnpublished, SineParameters, SineTrack, }; -use livekit::{ - SendByteOptions, SendTextOptions, SimulateScenario, - e2ee::{E2eeOptions, EncryptionType, key_provider::*}, - prelude::*, - track::VideoQuality, -}; +use crate::room::participants::{TrackCmd, TrackCommands}; +use livekit::{SendByteOptions, SendTextOptions, SimulateScenario, prelude::*}; use parking_lot::Mutex; use std::collections::HashMap; use std::sync::Arc; @@ -21,8 +17,8 @@ pub enum AsyncCmd { auth: Box, auto_subscribe: bool, dynacast: bool, - enable_e2ee: bool, - key: String, + /// The shared E2EE key, if encryption is enabled. + e2ee_key: Option, }, RoomDisconnect, SimulateScenario { @@ -31,16 +27,7 @@ pub enum AsyncCmd { TogglePublish { source: LocalSource, }, - SubscribeTrack { - publication: RemoteTrackPublication, - }, - UnsubscribeTrack { - publication: RemoteTrackPublication, - }, - SetVideoQuality { - publication: RemoteTrackPublication, - quality: VideoQuality, - }, + Track(TrackCmd), E2eeKeyRatchet, LogStats, RpcSendRequest { @@ -200,6 +187,12 @@ impl LkService { } } +impl TrackCommands for LkService { + fn track_cmd(&self, cmd: TrackCmd) { + let _ = self.send(AsyncCmd::Track(cmd)); + } +} + async fn service_task(inner: Arc, mut cmd_rx: mpsc::UnboundedReceiver) { struct RunningState { room: Arc, @@ -337,8 +330,7 @@ async fn service_task(inner: Arc, mut cmd_rx: mpsc::UnboundedRecei auth, auto_subscribe, dynacast, - enable_e2ee, - key, + e2ee_key, } => { // Resolved here rather than UI-side: the token-source method // fetches the connection details over HTTP, which must not @@ -354,17 +346,10 @@ async fn service_task(inner: Arc, mut cmd_rx: mpsc::UnboundedRecei log::info!("connecting to room: {}", url); - let key_provider = - KeyProvider::with_shared_key(KeyProviderOptions::default(), key.into_bytes()); - let e2ee = enable_e2ee.then_some(E2eeOptions { - encryption_type: EncryptionType::Gcm, - key_provider, - }); - let mut options = RoomOptions::default(); options.auto_subscribe = auto_subscribe; options.dynacast = dynacast; - options.encryption = e2ee; + options.encryption = e2ee_options(e2ee_key.as_deref()); let res = Room::connect(&url, &token, options).await; @@ -437,18 +422,7 @@ async fn service_task(inner: Arc, mut cmd_rx: mpsc::UnboundedRecei PublishState::Unpublished }); } - AsyncCmd::SubscribeTrack { publication } => { - publication.set_subscribed(true); - } - AsyncCmd::UnsubscribeTrack { publication } => { - publication.set_subscribed(false); - } - AsyncCmd::SetVideoQuality { - publication, - quality, - } => { - publication.set_video_quality(quality); - } + AsyncCmd::Track(cmd) => cmd.apply(), AsyncCmd::E2eeKeyRatchet => { if let Some(state) = running_state.as_ref() { let e2ee_manager = state.room.e2ee_manager(); diff --git a/src/ui/chat_message.rs b/src/ui/chat_message.rs new file mode 100644 index 0000000..fec5093 --- /dev/null +++ b/src/ui/chat_message.rs @@ -0,0 +1,60 @@ +use crate::style::Palette; +use egui::{Align, Frame, Layout, Response, RichText, Ui, Widget}; + +/// Widget: one chat message, the speaker above the text. The local user's +/// messages are right-aligned on the accent fill; pending text, still being +/// received (e.g. a growing transcript), is shown weak. +pub struct ChatMessage<'a> { + speaker: &'a str, + text: &'a str, + is_local: bool, + pending: bool, +} + +impl<'a> ChatMessage<'a> { + pub fn new(speaker: &'a str, text: &'a str) -> Self { + Self { + speaker, + text, + is_local: false, + pending: false, + } + } + + pub fn local(mut self, is_local: bool) -> Self { + self.is_local = is_local; + self + } + + /// Whether more text is still to come. Defaults to `false`. + pub fn pending(mut self, pending: bool) -> Self { + self.pending = pending; + self + } +} + +impl Widget for ChatMessage<'_> { + fn ui(self, ui: &mut Ui) -> Response { + let palette = Palette::for_theme(ui.theme()); + let (align, fill) = if self.is_local { + (Align::Max, palette.bg_accent) + } else { + (Align::Min, palette.bg_3) + }; + let max_width = ui.available_width() * 0.8; + + ui.with_layout(Layout::top_down(align), |ui| { + Frame::new() + .fill(fill) + .corner_radius(6.0) + .inner_margin(egui::Margin::symmetric(8, 6)) + .show(ui, |ui| { + ui.set_max_width(max_width); + ui.label(RichText::new(self.speaker).small().weak()); + let text = RichText::new(self.text); + ui.label(if self.pending { text.weak() } else { text }); + }); + }) + .response + } +} diff --git a/src/ui/mod.rs b/src/ui/mod.rs index 1d38a6d..27c45fe 100644 --- a/src/ui/mod.rs +++ b/src/ui/mod.rs @@ -2,6 +2,7 @@ //! service / `RoomCtx` dependency. The room feature in `crate::room` depends on //! this layer, never the reverse. +pub mod chat_message; pub mod data_chart; pub mod labeled_field; pub mod placeholder_tile; diff --git a/src/ui/video_tile.rs b/src/ui/video_tile.rs index 8accc6b..c535065 100644 --- a/src/ui/video_tile.rs +++ b/src/ui/video_tile.rs @@ -1,15 +1,18 @@ -/// Widget: paints a video frame to fill the available space, with a speaking -/// border and a participant-identity overlay. Decoupled from any video source — -/// the caller passes the already-resolved texture id, so this stays a pure UI -/// primitive. +use egui::load::SizedTexture; + +/// Widget: paints a video frame into the available space, with a speaking +/// border and a participant-identity overlay. The frame keeps its aspect ratio: +/// it is scaled to fit and centered, and the rest of the cell shows the tile +/// background. Decoupled from any video source — the caller passes the +/// already-resolved texture, so this stays a pure UI primitive. pub struct VideoTile<'a> { - texture: Option, + texture: Option, identity: &'a str, speaking: bool, } impl<'a> VideoTile<'a> { - pub fn new(texture: Option, identity: &'a str, speaking: bool) -> Self { + pub fn new(texture: Option, identity: &'a str, speaking: bool) -> Self { Self { texture, identity, @@ -33,10 +36,16 @@ impl egui::Widget for VideoTile<'_> { ); } - if let Some(tex) = self.texture { + ui.painter().rect_filled( + inner_rect, + egui::CornerRadius::default(), + ui.style().visuals.code_bg_color, + ); + + if let Some(texture) = self.texture { ui.painter().image( - tex, - inner_rect, + texture.id, + fit_rect(inner_rect, texture.size), egui::Rect::from_min_max(egui::pos2(0.0, 0.0), egui::pos2(1.0, 1.0)), egui::Color32::WHITE, ); @@ -53,3 +62,41 @@ impl egui::Widget for VideoTile<'_> { response } } + +/// `frame` scaled to the largest size with its aspect ratio that fits in `bounds`. +pub fn fit_size(bounds: egui::Vec2, frame: egui::Vec2) -> egui::Vec2 { + let scale = (bounds.x / frame.x).min(bounds.y / frame.y); + frame * scale +} + +/// The largest rect with `frame`'s aspect ratio that fits inside `cell`, centered. +fn fit_rect(cell: egui::Rect, frame: egui::Vec2) -> egui::Rect { + egui::Rect::from_center_size(cell.center(), fit_size(cell.size(), frame)) +} + +#[cfg(test)] +mod tests { + use super::*; + + const CELL: egui::Rect = + egui::Rect::from_min_max(egui::pos2(0.0, 0.0), egui::pos2(320.0, 180.0)); + + #[test] + fn portrait_frame_fits_height() { + let rect = fit_rect(CELL, egui::vec2(720.0, 1280.0)); + assert_eq!(rect.size(), egui::vec2(101.25, 180.0)); + assert_eq!(rect.center(), CELL.center()); + } + + #[test] + fn wide_frame_fits_width() { + let rect = fit_rect(CELL, egui::vec2(640.0, 200.0)); + assert_eq!(rect.size(), egui::vec2(320.0, 100.0)); + assert_eq!(rect.center(), CELL.center()); + } + + #[test] + fn matching_frame_fills_cell() { + assert_eq!(fit_rect(CELL, egui::vec2(1280.0, 720.0)), CELL); + } +}