mediad: a websocket for programs, next to the datachannel for people

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

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

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

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<Option<Media>>,
+}
+
+/// 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::<String>(QUEUE);
+    let (outbound, mut outbound_rx) = mpsc::channel::<String>(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<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>
+    {
+        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<tokio::net::TcpStream>,
+        >,
+        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}"
+        );
+    }
+}

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;

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::<Option<mediad::session::Media>>(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::<Option<mediad::session::Media>>(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

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()),
     )
 }
 

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(),