diff --git a/docs/4-implementation/turn-loop/mcp.md b/docs/4-implementation/turn-loop/mcp.md index cf554f7a9..97f578a3f 100644 --- a/docs/4-implementation/turn-loop/mcp.md +++ b/docs/4-implementation/turn-loop/mcp.md @@ -95,7 +95,7 @@ hive_name?, swarm_name?, matrix_accounts? }`. `matrix_accounts` is a `name` to query self. **Always-on, no permission needed**: `set_status(text)`, `compact()`, -`mark_todos_done(ids)`. +`mark_todos_done(ids)`, `transcribe_audio(path)`. - `set_status` — set a free-text status string visible on the dashboard. Single line, ≤ 200 chars. Persisted to @@ -112,6 +112,15 @@ hive_name?, swarm_name?, matrix_accounts? }`. `matrix_accounts` is a notes-checkpoint turn still fires first. Dispatched through the in-agent socket (`hive-agent-sock::Request::Compact`), not the broker — see `hive-agent/src/todo_server.rs`. +- `transcribe_audio(path)` — the text spoken in an audio file (a matrix + voice memo that `download_file` saved, or any Ogg, WAV, FLAC, MP3, AMR, + MP4/M4A or WebM file up to 6 MiB). Dials the in-agent socket + (`hive-agent-sock::Request::Transcribe`) with a 16-minute deadline; the + harness reads the file and sends it over the swarm queue on this agent's + request subject to the speech-to-text service + ([`stt.md`](../swarm/stt.md)), which keeps each transcript for a day by the + audio's hash. An error names why: no service running, a file that isn't + audio, too large, or the service busy. - `mark_todos_done(ids)` — bulk-clear specific loose-ends-v2 todo rows by id in one call, instead of `cancel_loose_end`ing each one. List-based: pass the ids `get_loose_ends` actually showed you reviewed; diff --git a/docs/5-project/process/conventions.md b/docs/5-project/process/conventions.md index 50678737f..a24a3be56 100644 --- a/docs/5-project/process/conventions.md +++ b/docs/5-project/process/conventions.md @@ -274,7 +274,7 @@ host access it has. | `metrics` | the metrics store: `swarm-metrics` queries. Checked by `swarm-metrics-agent-auth`, the same binary, for the metrics gateway. Off by default. | **Always-on tools** — `Permission::ALWAYS_ON_TOOLS` exposes `set_status`, -`compact`, and `mark_todos_done` to every agent regardless of its +`compact`, `mark_todos_done` and `transcribe_audio` to every agent regardless of its permissions. The operator dashboard depends on every agent being able to report its status chip, and the server-side `SetStatus` handler has no permission check (only length validation), so gating it would only desync @@ -282,6 +282,9 @@ the `--allowedTools` list from what the host actually accepts. Revoking `meta` therefore drops `get_agent_meta` but never `set_status`. `mark_todos_done` is here because producers push todos to an agent whether or not it holds `inbox`, and an agent without it still needs to clear them. +`transcribe_audio` is here because it only turns an audio file the agent +already holds into text through the swarm's service; it reaches nothing the +agent couldn't read. **Where they live** — in the swarm, not on the hive: the `permissions` field of the agent's key in the `agent-permissions` KV bucket diff --git a/hive-agent-mcp/src/mcp/args.rs b/hive-agent-mcp/src/mcp/args.rs index b1ef9db95..bd13aa383 100644 --- a/hive-agent-mcp/src/mcp/args.rs +++ b/hive-agent-mcp/src/mcp/args.rs @@ -13,6 +13,15 @@ pub struct MarkTodosDoneArgs { pub ids: Vec, } +/// MCP tool args for `transcribe_audio`. +#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] +pub struct TranscribeAudioArgs { + /// Absolute path of the audio file to transcribe (a regular file, at most + /// 6 MiB; Ogg, WAV, FLAC, MP3, AMR, `MP4`/`M4A` or `WebM`). For a matrix voice + /// memo, the `path` that `download_file` returned. + pub path: String, +} + /// MCP tool args for `remind`. Exactly one of `delay_seconds` or /// `at_unix_timestamp` must be set; both / neither is a tool-side error. /// Hides the tagged `ReminderTiming` enum behind a flatter schema so the diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index 3ba7a9b5f..545b44d37 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -24,13 +24,14 @@ mod args; mod render; pub use args::{ - CancelLooseEndArgs, CompactArgs, GetAgentMetaArgs, MarkTodosDoneArgs, RemindArgs, SetStatusArgs, + CancelLooseEndArgs, CompactArgs, GetAgentMetaArgs, MarkTodosDoneArgs, RemindArgs, + SetStatusArgs, TranscribeAudioArgs, }; pub use render::{annotate_retries, format_ack, format_agent_meta}; use render::{ dial_agent_socket, format_matrix_summary, local_reminders, local_todos, mark_local_todo_done, - mark_local_todos_done, matrix_unread_summary, render_loose_ends, reply_err, + mark_local_todos_done, matrix_unread_summary, render_loose_ends, reply_err, transcribe_local, }; /// Write (or remove) the status file in the agent's own `state/` directory. @@ -329,6 +330,37 @@ impl AgentServer { .await } + #[tool( + description = "Get the text spoken in an audio file, from the swarm's speech-to-text \ + service: a matrix voice memo (download it with the matrix `download_file` tool and pass \ + the returned `path`), or any Ogg, WAV, FLAC, MP3, AMR, MP4/M4A or WebM file up to 6 MiB. \ + The service keeps each transcript for a day, so the same memo heard by several agents is \ + transcribed once. Can take minutes when other memos are queued ahead. Returns the text \ + (empty for silence), or why it could not be transcribed." + )] + async fn transcribe_audio(&self, Parameters(args): Parameters) -> String { + let log = format!("{args:?}"); + run_tool_envelope("transcribe_audio", log, async move { + match transcribe_local(args.path).await { + Some(hive_agent_sock::Response::Transcript { text, .. }) => { + if text.trim().is_empty() { + "(no speech found)".to_owned() + } else { + text + } + } + Some(hive_agent_sock::Response::Err { message }) => { + format!("transcribe_audio failed: {message}") + } + Some(other) => format!("transcribe_audio unexpected response: {other:?}"), + None => "transcribe_audio: local agent socket unavailable \ + (HIVE_AGENT_SOCKET unset or harness unreachable)" + .to_owned(), + } + }) + .await + } + #[tool( description = "Schedule a reminder that lands in this agent's own inbox at a future \ time (sender will appear as `reminder`). Use for self-paced follow-ups: 'check task \ diff --git a/hive-agent-mcp/src/mcp/render.rs b/hive-agent-mcp/src/mcp/render.rs index a5cdc21d4..f5fc66a84 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -302,6 +302,26 @@ pub(super) async fn mark_local_todos_done(ids: Vec) -> Option Option { + let socket = std::env::var_os("HIVE_AGENT_SOCKET").map(std::path::PathBuf::from)?; + if !socket.exists() { + return None; + } + dial_socket( + &socket, + &hive_agent_sock::Request::Transcribe { path }, + TRANSCRIBE_DEADLINE, + ) + .await +} + +/// Longest a `transcribe_audio` call waits: the service's reply deadline (15 +/// minutes, `swarm_wire::stt::REPLY_DEADLINE`) plus a margin for the harness. +const TRANSCRIBE_DEADLINE: std::time::Duration = std::time::Duration::from_mins(16); + /// Format a `Vec` into a per-room summary string. /// Single room / single message collapses to one line; multi-room /// expands to a bulleted list. Returns an empty string for empty input. diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs index df42b873e..4fe5ad897 100644 --- a/hive-agent-sock/src/lib.rs +++ b/hive-agent-sock/src/lib.rs @@ -135,6 +135,13 @@ pub enum Request { /// through this variant, since a human watching the dashboard isn't /// waiting on a wake. Compact { wake_prompt: Option }, + /// Ask the swarm's speech-to-text service for the text of the audio file + /// at `path`. The harness reads the file (a regular file under the size + /// cap), sends the bytes on this agent's request subject + /// (`swarm_wire::stt`), and answers [`Response::Transcript`], or + /// [`Response::Err`] with the service's short reason. Can take minutes: + /// the service queues requests behind its one engine slot. + Transcribe { path: String }, } /// A response on the in-agent socket. Serialised with a `kind` tag, @@ -155,6 +162,9 @@ pub enum Response { ReminderRollup { stats: hive_sh4re::inbox::ReminderStats, }, + /// `Transcribe` result: the text spoken in the audio (empty for silence), + /// and whether the service had it already. + Transcript { text: String, cached: bool }, /// Op failed; `message` is operator-facing. Err { message: String }, } diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 91ef70315..149bc27a3 100644 --- a/hive-agent/src/main.rs +++ b/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; diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index a1e6cfdf8..c70a139c0 100644 --- a/hive-agent/src/todo_server.rs +++ b/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::(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, 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; diff --git a/hive-agent/src/transcribe.rs b/hive-agent/src/transcribe.rs new file mode 100644 index 000000000..7fcb19dfc --- /dev/null +++ b/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, 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::(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) -> 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}"); + } +} diff --git a/hive-matrix-mcp/src/mcp.rs b/hive-matrix-mcp/src/mcp.rs index a96a88e34..39fb84c96 100644 --- a/hive-matrix-mcp/src/mcp.rs +++ b/hive-matrix-mcp/src/mcp.rs @@ -525,7 +525,8 @@ impl MatrixMcp { first. `download_file` downloads an attachment from a message event \ to a local file and returns its path; pair with `read_room` which \ surfaces attachments as `[file: name]`, `[image: name]`, \ - `[audio: name]`, `[video: name]` markers. \ + `[audio: name]`, `[video: name]` markers. To hear a voice memo, \ + `download_file` it and pass the returned path to the `transcribe_audio` tool. \ IMPORTANT: `send_message`, `send_dm`, `send_file`, `send_reply` are \ all rejected if the room has unread messages — always call `read_room` \ then `mark_read` on the latest event before sending to a room you \ diff --git a/hive-types/src/permission.rs b/hive-types/src/permission.rs index 7acb781e3..683720350 100644 --- a/hive-types/src/permission.rs +++ b/hive-types/src/permission.rs @@ -110,8 +110,16 @@ impl Permission { /// holds `Inbox` (which only gates `get_loose_ends`/`cancel_loose_end`/ /// `remind`), so an agent without `Inbox` must still be able to clear /// them. - pub const ALWAYS_ON_TOOLS: &'static [&'static str] = - &["set_status", "compact", "mark_todos_done"]; + /// + /// `transcribe_audio` too: it asks the swarm's speech-to-text service for + /// the text of an audio file the agent already holds, and reaches nothing + /// the agent could not already read. + pub const ALWAYS_ON_TOOLS: &'static [&'static str] = &[ + "set_status", + "compact", + "mark_todos_done", + "transcribe_audio", + ]; /// The Claude built-in tool names this permission enables. Only /// `WebTools` returns a non-empty slice.