diff --git a/docs/design/architecture.md b/docs/design/architecture.md index 1eae59f2..c1969fa6 100644 --- a/docs/design/architecture.md +++ b/docs/design/architecture.md @@ -422,6 +422,11 @@ Same API behind all of them. "Run an LLM on a server that controls the robot" becomes: open a WebSocket, poll a frame, send intents — a few dozen lines, no media stack. That is what makes it genuinely easy. +`mediad::agent` is that socket, at `/agent` on the port the console is already +served from. Same API is not a promise here, it is the implementation: a line +arriving there runs through `session::run` against `route`, which is the same +dispatcher and the same table the datachannel uses. + Note also that LLM latency (hundreds of ms to seconds) means the agent is a **high-level** controller: "go to the kitchen", "look at the person". Reactive control stays local in `robotd`. This is the correct split regardless of diff --git a/docs/design/remote-webrtc.md b/docs/design/remote-webrtc.md index d1b1b1f9..c8f4a9bd 100644 --- a/docs/design/remote-webrtc.md +++ b/docs/design/remote-webrtc.md @@ -571,10 +571,12 @@ along with a transport. `remote-access-design.md` §9 carries it as open. ## 12. Deferred, with reasons -- **A WebSocket surface for server-side programs** (`architecture.md` §5.3). Same JSON-RPC, no - media stack, `get_frame` returning a JPEG. It is a few dozen lines once §5's routing exists, and - it is what makes "an LLM drives the robot" easy — but it is a second transport and the first one - should work. +- ~~**A WebSocket surface for server-side programs**~~ — built. `mediad::agent` serves it at + `/agent` on the console's own listener, and it was the few dozen lines this predicted: §5's + routing already existed, and `session::run` never knew what carried its lines, so the whole of + the transport is pumping text frames into one channel and out of the other. What an agent may + call is what a console peer may call, decided in `route` for both — a method is routed once, for + every transport, or refused everywhere. `refusal` stopped naming WebRTC for the same reason. - **The `teleop` datachannel.** Not the near-term priority; §6 covers what deferring it removes, what it costs in the meantime, and the sequence numbers it will need. - **Multi-peer video.** One media session at a time, plus control-only clients. Simulcast and diff --git a/mediad/Cargo.toml b/mediad/Cargo.toml index 061a2bf6..36a4d03a 100644 --- a/mediad/Cargo.toml +++ b/mediad/Cargo.toml @@ -44,7 +44,7 @@ image = { version = "0.25", default-features = false, features = ["jpeg", "png"] # # Default features off: this serves one file over GET, and JSON extractors, multipart, forms and # tracing middleware are all weight for routes that do not exist. -axum = { version = "0.8.9", default-features = false, features = ["http1", "tokio"] } +axum = { version = "0.8.9", default-features = false, features = ["http1", "tokio", "ws"] } clap = { workspace = true, features = ["derive"] } # The relay's half of the bridge: HTTPS to the rendezvous service — SSE inbound, `POST /send` # outbound. Same line `updater` uses, minus `form` (nothing here posts a form), so the TLS stack diff --git a/mediad/src/agent.rs b/mediad/src/agent.rs new file mode 100644 index 00000000..f4153cce --- /dev/null +++ b/mediad/src/agent.rs @@ -0,0 +1,229 @@ +//! A WebSocket for programs, next to the WebRTC console for people. +//! +//! `architecture.md` §5.3 argues that an LLM-driven controller should not be pushed through +//! WebRTC: an agent does not want a 30 fps H.264 track to decode, it wants a frame every second or +//! two and a state blob, and making it do ICE and DTLS and a decode pipeline first is a poor +//! trade. What it wants is "open a socket, poll a frame, send intents". That is this. +//! +//! **It is the same API, not a second one.** A line arriving here goes through +//! [`crate::session::run`] — the same dispatcher the datachannel uses, against the same +//! [`crate::route`] table — so what an agent may call is what a console peer may call, decided in +//! one place. Adding a method here is not a thing anybody can do; a method is routed once, for +//! every transport, or it is refused everywhere. +//! +//! That is also why this is small. `session::run` takes a pair of `String` channels and knows +//! nothing about what carries them, so the whole of the transport is pumping text frames into one +//! and out of the other. + +use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; +use axum::response::Response; +use futures_util::{SinkExt, StreamExt}; +use tokio::sync::{mpsc, watch}; + +use crate::session::{self, Media}; +use crate::upstream::{Pool, Sockets}; + +/// Lines buffered in each direction before a slow peer starts costing something. +/// +/// An agent's own calls are request-response and never queue, so this is really the size of a +/// subscription's backlog — `robot.state` at the control rate is the fast one. Deep enough to ride +/// out a garbage collection at the other end, shallow enough that a peer which has stopped reading +/// is noticed rather than buffered for ever. +const QUEUE: usize = 256; + +/// What an agent connection needs to serve: where the daemons are, and the media that may not +/// exist yet. +/// +/// The media arrives late by construction — the console is served before the pipeline is built, +/// because a page that cannot bind must not cost the video — so it comes as a watch rather than a +/// value. A connection reads it once, at accept, which is also when a `None` is honest: no +/// pipeline, no frames to hand out. +#[derive(Clone)] +pub struct State { + pub sockets: Sockets, + pub video: watch::Receiver>, +} + +/// The upgrade handler, for the router to hang on a path. +pub async fn upgrade(ws: WebSocketUpgrade, state: State) -> Response { + ws.on_upgrade(move |socket| serve(socket, state)) +} + +/// One agent, until it goes away. +async fn serve(socket: WebSocket, state: State) { + let (mut sink, mut stream) = socket.split(); + let (inbound, inbound_rx) = mpsc::channel::(QUEUE); + let (outbound, mut outbound_rx) = mpsc::channel::(QUEUE); + + // Its own connections to the daemons, per peer rather than shared, for the reason the + // datachannel's are: one peer's minutes-long update must not silence another's telemetry. + let pool = Pool::new(state.sockets.clone(), outbound.clone()); + let media = state.video.borrow().clone(); + if media.is_none() { + // Worth saying once. Every control call still works; `media.*` is what will refuse, and + // "the camera answers nothing" is otherwise a silent property of having connected early. + tracing::debug!("an agent connected before the pipeline; frames are not available yet"); + } + + let writer = tokio::spawn(async move { + while let Some(line) = outbound_rx.recv().await { + if sink.send(Message::Text(line.into())).await.is_err() { + break; + } + } + }); + + let session = tokio::spawn(session::run(inbound_rx, outbound, pool, media)); + + while let Some(message) = stream.next().await { + match message { + // One JSON-RPC object per message, which is what a WebSocket already frames for us — + // so unlike every other transport here there is no newline to reassemble. + Ok(Message::Text(text)) => { + if inbound.send(text.to_string()).await.is_err() { + break; + } + } + Ok(Message::Close(_)) => break, + // Ping and pong are the runtime's; binary is not something this speaks. + Ok(_) => {} + Err(e) => { + tracing::debug!(error = %e, "an agent's socket ended"); + break; + } + } + } + + // Dropping `inbound` ends `session::run`, which drops the last `outbound` and ends the writer. + drop(inbound); + let _ = session.await; + writer.abort(); + tracing::debug!("agent disconnected"); +} + +#[cfg(test)] +mod tests { + use super::*; + use duck_ipc_proto as proto; + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + + /// A daemon that reads one line and answers with a canned one, so a call that is *routed* can + /// be told from one that is merely accepted. + fn fake_daemon(path: std::path::PathBuf, reply: String) { + let listener = std::os::unix::net::UnixListener::bind(&path).unwrap(); + listener.set_nonblocking(true).unwrap(); + let listener = tokio::net::UnixListener::from_std(listener).unwrap(); + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let reply = reply.clone(); + tokio::spawn(async move { + let (read, mut write) = stream.into_split(); + let mut lines = BufReader::new(read).lines(); + while let Ok(Some(_)) = lines.next_line().await { + let _ = write.write_all(format!("{reply}\n").as_bytes()).await; + let _ = write.flush().await; + } + }); + } + }); + } + + /// Serve the real router on an ephemeral port and open a real WebSocket to it. + async fn connect( + dir: &std::path::Path, + ) -> tokio_tungstenite::WebSocketStream> + { + let sockets = Sockets { + robot: dir.join("robotd.sock"), + updater: dir.join("updaterd.sock"), + config: dir.join("configd.sock"), + ..Sockets::default() + }; + + let (_video_tx, video) = watch::channel(None); + let state = State { sockets, video }; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let router = axum::Router::new().route( + "/agent", + axum::routing::get(move |ws| upgrade(ws, state.clone())), + ); + tokio::spawn(async move { + let _ = axum::serve(listener, router).await; + }); + + let (socket, _) = tokio_tungstenite::connect_async(format!("ws://127.0.0.1:{port}/agent")) + .await + .expect("the agent socket accepts a websocket"); + socket + } + + async fn ask( + socket: &mut tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, + >, + call: &proto::Call, + ) -> serde_json::Value { + let request = proto::Request::call(proto::Id::Number(1), call); + socket + .send(tokio_tungstenite::tungstenite::Message::Text( + serde_json::to_string(&request).unwrap().into(), + )) + .await + .unwrap(); + let answer = socket.next().await.unwrap().unwrap(); + serde_json::from_str(answer.to_text().unwrap()).unwrap() + } + + /// **The point of the whole module: a program gets the surface a person gets.** + /// + /// `hello` is answered, a call `route` permits reaches the daemon that owns it, and one it + /// refuses is refused here too — all through `session::run`, which is what makes "same API + /// behind all of them" true rather than a thing to keep in step by hand. + #[tokio::test] + async fn an_agent_gets_the_same_surface_a_console_peer_gets() { + let dir = tempfile::tempdir().unwrap(); + fake_daemon( + dir.path().join("robotd.sock"), + r#"{"jsonrpc":"2.0","id":1,"result":{"healthy":true}}"#.to_owned(), + ); + // `hello` belongs to updaterd, like it does on every other transport. + fake_daemon( + dir.path().join("updaterd.sock"), + format!( + r#"{{"jsonrpc":"2.0","id":1,"result":{{"api_version":{}}}}}"#, + proto::API_VERSION + ), + ); + let mut socket = connect(dir.path()).await; + + let hello = ask( + &mut socket, + &proto::Call::Hello(proto::HelloParams { + api_version: proto::API_VERSION, + }), + ) + .await; + assert_eq!( + hello["result"]["api_version"], + proto::API_VERSION, + "a version handshake must work before anything else: {hello}" + ); + + let health = ask(&mut socket, &proto::Call::RobotHealth).await; + assert_eq!( + health["result"]["healthy"], true, + "a permitted call must reach robotd: {health}" + ); + + // Refused over WebRTC, so refused here: the PIN authorises a phone, and a transport that + // hands it out is a transport that hands out the robot. + let pin = ask(&mut socket, &proto::Call::SystemPairingPin).await; + assert_eq!( + pin["error"]["code"], + proto::code::METHOD_NOT_FOUND, + "the agent socket must not widen what a transport may call: {pin}" + ); + } +} diff --git a/mediad/src/lib.rs b/mediad/src/lib.rs index fbb8bd02..f1e3d0db 100644 --- a/mediad/src/lib.rs +++ b/mediad/src/lib.rs @@ -17,6 +17,9 @@ //! signalling server in this process, `mpph264enc` in front of it, and a `control` datachannel per //! peer wired to [`session::run`]. +/// The WebSocket a program drives the robot over, as opposed to the datachannel a person does. +/// `architecture.md` §5.3 has the argument; the surface is the same one either way. +pub mod agent; /// What the camera's geometry is — the intrinsics a consumer needs to turn pixels into /// directions, and which sensor mode they belong to. pub mod camera; diff --git a/mediad/src/main.rs b/mediad/src/main.rs index 52b5d1d9..ef99c4a1 100644 --- a/mediad/src/main.rs +++ b/mediad/src/main.rs @@ -299,11 +299,22 @@ fn main() -> ExitCode { // a port already in use, which `Restart=always` cannot fix by trying again; a robot that // streams and answers control calls with no console is much better than one that does // neither. So this is logged at error and the daemon carries on. + // Made here rather than beside the relay, because the console's listener now carries the + // agent socket too and both want the same late-arriving media. + let (video_tx, video_rx) = + tokio::sync::watch::channel::>(None); + let page = mediad::web::page(args.port); let (web_host, web_port) = (args.host.clone(), args.web_port); let web_frame_socket = args.frame_socket.clone(); + let agent = mediad::agent::State { + sockets: args.sockets(), + video: video_rx.clone(), + }; tokio::spawn(async move { - if let Err(e) = mediad::web::serve(&web_host, web_port, page, web_frame_socket).await { + if let Err(e) = + mediad::web::serve(&web_host, web_port, page, web_frame_socket, agent).await + { tracing::error!( error = %format!("{e:#}"), "the console is not being served; video and control are unaffected" @@ -341,9 +352,6 @@ fn main() -> ExitCode { // once something has tried to set it, and there are no frames to encode before then — and // the relay below is spawned before that on purpose, so the answer has to be able to // arrive late rather than be a value passed in now. - let (video_tx, video_rx) = - tokio::sync::watch::channel::>(None); - // The outward half of remote access, and it is deliberately *after* the producer is // learned: the name a client sees in the service's listing comes from the same place the // local `meta` gets it, and a relay that registered first would publish an unnamed robot diff --git a/mediad/src/route.rs b/mediad/src/route.rs index 977cc11a..c5688443 100644 --- a/mediad/src/route.rs +++ b/mediad/src/route.rs @@ -309,10 +309,14 @@ pub fn route_for(call: &proto::Call) -> Route { } /// The refusal a peer gets, naming the method so a client can report which call was declined. +/// +/// "this transport" rather than "WebRTC": the table is shared with the agent socket +/// (`crate::agent`), and a websocket client told its call was refused over WebRTC would go looking +/// for a second table that does not exist. pub fn refusal(call: &proto::Call) -> proto::Error { proto::Error::new( proto::code::METHOD_NOT_FOUND, - format!("{} is not available over WebRTC", call.method()), + format!("{} is not available over this transport", call.method()), ) } diff --git a/mediad/src/session.rs b/mediad/src/session.rs index 6c65132e..974d3205 100644 --- a/mediad/src/session.rs +++ b/mediad/src/session.rs @@ -466,7 +466,10 @@ mod tests { .unwrap(); let reply = h.to_peer.recv().await.unwrap(); - assert!(reply.contains("not available over WebRTC"), "{reply}"); + assert!( + reply.contains("not available over this transport"), + "{reply}" + ); assert!(reply.contains(r#""id":7"#), "{reply}"); assert!( config_seen.try_recv().is_err(), diff --git a/mediad/src/web.rs b/mediad/src/web.rs index 2bd649ae..ee3da96b 100644 --- a/mediad/src/web.rs +++ b/mediad/src/web.rs @@ -74,7 +74,13 @@ pub fn page(signalling_port: u32) -> String { /// Returns only on failure — a bind that was refused, or a listener that died. The caller decides /// what that costs; in `mediad` it costs the page and not the video, because a robot that streams /// and answers control calls with no console is a great deal better than one that does neither. -pub async fn serve(host: &str, port: u16, page: String, frame_socket: PathBuf) -> Result<()> { +pub async fn serve( + host: &str, + port: u16, + page: String, + frame_socket: PathBuf, + agent: crate::agent::State, +) -> Result<()> { let address: SocketAddr = format!("{host}:{port}") .parse() .with_context(|| format!("{host}:{port} is not an address to listen on"))?; @@ -82,14 +88,19 @@ pub async fn serve(host: &str, port: u16, page: String, frame_socket: PathBuf) - .await .with_context(|| format!("could not listen on {address}"))?; - tracing::info!(%address, "serving the console"); - axum::serve(listener, router(page, frame_socket)) + tracing::info!(%address, "serving the console and the agent socket"); + axum::serve(listener, router(page, frame_socket, agent)) .await .context("the console's listener stopped") } -/// The console and its bounded, uncached snapshot endpoint. -fn router(page: String, frame_socket: PathBuf) -> Router { +/// The page, a still of what the camera sees, and the WebSocket a program drives the robot over. +/// +/// One listener for all three because they are one interface seen from different sides, and +/// because a second port is a second thing to configure, open and explain. `/` is for a person, +/// `/frame` is the picture on its own for anything that only wants a still, `/agent` is for a +/// program, and `crate::agent` says why that one exists. +fn router(page: String, frame_socket: PathBuf, agent: crate::agent::State) -> Router { let slots = Arc::new(tokio::sync::Semaphore::new(4)); Router::new() .route("/", get(move || std::future::ready(Html(page)))) @@ -97,6 +108,10 @@ fn router(page: String, frame_socket: PathBuf) -> Router { "/frame", get(move || snapshot(frame_socket.clone(), slots.clone())), ) + .route( + "/agent", + get(move |ws| crate::agent::upgrade(ws, agent.clone())), + ) } async fn snapshot(socket: PathBuf, slots: Arc) -> axum::response::Response { @@ -192,7 +207,12 @@ mod tests { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let server = tokio::spawn(async move { - axum::serve(listener, router(page(8443), socket)) + let (_video_tx, video) = tokio::sync::watch::channel(None); + let agent = crate::agent::State { + sockets: crate::upstream::Sockets::default(), + video, + }; + axum::serve(listener, router(page(8443), socket, agent)) .await .unwrap(); }); @@ -294,9 +314,14 @@ mod tests { .expect("a loopback port"); let address = listener.local_addr().expect("the port it took"); tokio::spawn(async move { + let (_video_tx, video) = tokio::sync::watch::channel(None); + let agent = crate::agent::State { + sockets: crate::upstream::Sockets::default(), + video, + }; let _ = axum::serve( listener, - router(page(8443), PathBuf::from(proto::socket::MEDIA)), + router(page(8443), PathBuf::from(proto::socket::MEDIA), agent), ) .await; });