agents can transcribe an audio file: the transcribe_audio tool, through the harness and the speech-to-text service

hive-agent/src/main.rs

@@ -48,6 +48,7 @@ mod term_msg;
 mod todo_server;
 mod todos;
 mod todos_gauge;
+mod transcribe;
 mod turn;
 mod turn_stats;
 mod vacuum;

hive-agent/src/todo_server.rs

@@ -183,8 +183,11 @@ async fn handle_conn(
     reminders: Option<&Reminders>,
     bus: &Bus,
 ) -> Result<()> {
-    serve_one(stream, REQUEST_DEADLINES, |req| {
-        dispatch(req, store, wake, reminders, bus)
+    serve_one(stream, REQUEST_DEADLINES, async |req| match req {
+        // Waits on the swarm's speech-to-text service, so it is the one
+        // request that does not answer from the local store at once.
+        Request::Transcribe { path } => crate::transcribe::run(&path).await,
+        req => dispatch(req, store, wake, reminders, bus),
     })
     .await
 }
@@ -201,7 +204,7 @@ const REQUEST_DEADLINES: LineDeadlines = LineDeadlines {
 async fn serve_one(
     stream: UnixStream,
     deadlines: LineDeadlines,
-    apply: impl FnOnce(Request) -> Response,
+    apply: impl AsyncFnOnce(Request) -> Response,
 ) -> Result<()> {
     let (read, mut write) = stream.into_split();
     let mut reader = BufReader::new(read);
@@ -214,7 +217,7 @@ async fn serve_one(
         Response::Err { message }
     } else {
         match serde_json::from_slice::<Request>(line.trim_ascii()) {
-            Ok(req) => apply(req),
+            Ok(req) => apply(req).await,
             Err(e) => Response::Err {
                 message: format!("bad request: {e}"),
             },
@@ -276,6 +279,11 @@ fn dispatch(
         Request::CountPendingReminders => count_pending_reminders(reminders),
         Request::ReminderRollup { since_secs } => reminder_rollup(reminders, since_secs),
         Request::Compact { wake_prompt } => compact(bus, wake_prompt),
+        // Answered by `handle_conn` before this runs: it waits on the queue,
+        // which this store-only function never does.
+        Request::Transcribe { .. } => Response::Err {
+            message: "transcribe is answered by the connection handler, not the store".to_owned(),
+        },
     }
 }
 
@@ -695,7 +703,7 @@ mod tests {
     /// Send `input` (then keep the write half open), return the one response line.
     async fn exchange(input: Vec<u8>, deadlines: LineDeadlines) -> (Response, Result<()>) {
         let (client, server) = UnixStream::pair().expect("socketpair");
-        let handler = tokio::spawn(serve_one(server, deadlines, |_| Response::Ok));
+        let handler = tokio::spawn(serve_one(server, deadlines, async |_| Response::Ok));
         let (read, mut write) = client.into_split();
         let _writer = tokio::spawn(async move {
             let _ = write.write_all(&input).await;

hive-agent/src/transcribe.rs

@@ -0,0 +1,119 @@
+//! `Request::Transcribe`: the text of an audio file, from the swarm's
+//! speech-to-text service.
+//!
+//! The harness holds the swarm queue connection, so an agent's tool call
+//! reaches the service through it: read the file, publish the bytes on this
+//! agent's request subject (`swarm_wire::stt`), and wait for the service's
+//! JSON answer.
+
+use std::path::Path;
+
+use hive_agent_sock::Response;
+use swarm_wire::stt;
+
+/// The largest file sent. The queue carries 8 MiB a message; the service
+/// refuses what it was configured not to take, so this only keeps the harness
+/// from reading something huge into memory.
+const MAX_BYTES: u64 = 6 * 1024 * 1024;
+
+/// Ask the service for the text of the audio file at `path`.
+pub async fn run(path: &str) -> Response {
+    let agent = crate::identity::label();
+    if agent.is_empty() {
+        return err("this agent has no label, so the service has no subject to answer on");
+    }
+    let bytes = match read_audio(Path::new(path)).await {
+        Ok(bytes) => bytes,
+        Err(message) => return err(message),
+    };
+    let client = crate::swarm_queue::client().await;
+    let request = client.request(stt::request_subject(&agent), bytes.into());
+    match tokio::time::timeout(stt::REPLY_DEADLINE, request).await {
+        Err(_) => err("the speech-to-text service did not answer in time"),
+        Ok(Err(e)) => err(request_failure(&e)),
+        Ok(Ok(reply)) => from_wire(&reply.payload),
+    }
+}
+
+/// The file's bytes, if it is a regular file small enough to send.
+async fn read_audio(path: &Path) -> Result<Vec<u8>, String> {
+    let meta = tokio::fs::metadata(path)
+        .await
+        .map_err(|e| format!("cannot read {}: {e}", path.display()))?;
+    if !meta.is_file() {
+        return Err(format!("{} is not a regular file", path.display()));
+    }
+    if meta.len() > MAX_BYTES {
+        return Err(format!(
+            "{} is {} bytes, over the {MAX_BYTES} byte limit",
+            path.display(),
+            meta.len()
+        ));
+    }
+    tokio::fs::read(path)
+        .await
+        .map_err(|e| format!("cannot read {}: {e}", path.display()))
+}
+
+/// What a failed publish-and-wait means to the agent.
+fn request_failure(e: &async_nats::RequestError) -> &'static str {
+    if e.kind() == async_nats::client::RequestErrorKind::NoResponders {
+        "no speech-to-text service is running in this swarm"
+    } else {
+        "the speech-to-text service could not be reached"
+    }
+}
+
+/// The service's JSON answer as the socket's response.
+fn from_wire(payload: &[u8]) -> Response {
+    match serde_json::from_slice::<stt::Response>(payload) {
+        Ok(stt::Response::Ok { text, cached }) => Response::Transcript { text, cached },
+        Ok(stt::Response::Failed { message }) => Response::Err { message },
+        Err(_) => err("the speech-to-text service sent an answer this harness does not understand"),
+    }
+}
+
+fn err(message: impl Into<String>) -> Response {
+    Response::Err {
+        message: message.into(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn the_services_answers_map_to_socket_responses() {
+        let ok = br#"{"status":"ok","text":"hello","cached":true}"#;
+        assert!(matches!(
+            from_wire(ok),
+            Response::Transcript { ref text, cached: true } if text == "hello"
+        ));
+        let failed = br#"{"status":"failed","message":"the audio could not be decoded"}"#;
+        assert!(matches!(
+            from_wire(failed),
+            Response::Err { ref message } if message == "the audio could not be decoded"
+        ));
+        assert!(matches!(from_wire(b"nonsense"), Response::Err { .. }));
+    }
+
+    #[tokio::test]
+    async fn only_a_small_regular_file_is_read() {
+        let dir = tempfile::tempdir().unwrap();
+        let small = dir.path().join("memo.ogg");
+        std::fs::write(&small, b"OggS").unwrap();
+        assert_eq!(read_audio(&small).await.unwrap(), b"OggS");
+
+        let missing = read_audio(&dir.path().join("none.ogg")).await.unwrap_err();
+        assert!(missing.contains("cannot read"), "{missing}");
+        let directory = read_audio(dir.path()).await.unwrap_err();
+        assert!(directory.contains("not a regular file"), "{directory}");
+
+        let big = dir.path().join("big.ogg");
+        let file = std::fs::File::create(&big).unwrap();
+        file.set_len(MAX_BYTES + 1).unwrap();
+        let too_big = read_audio(&big).await.unwrap_err();
+        assert!(too_big.contains("over the"), "{too_big}");
+    }
+}