diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index a82c6cf6..c7a800c5 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -33,9 +33,9 @@ pub use args::{ pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv}; use render::{ - dial_agent_socket, format_matrix_summary, local_reminders, local_todos, loose_end_kind_label, - mark_local_todo_done, mark_local_todos_done, matrix_unread_summary, parse_loose_end_kind, - render_loose_ends, reply_err, + dial_agent_socket, format_matrix_summary, local_questions, local_reminders, local_todos, + loose_end_kind_label, mark_local_todo_done, mark_local_todos_done, matrix_unread_summary, + parse_loose_end_kind, render_loose_ends, reply_err, }; /// Write (or remove) the status file in the agent's own `state/` directory. @@ -233,14 +233,15 @@ impl AgentServer { } #[tool( - description = "List loose ends pending against this agent: pending reminders \ - you scheduled, plus — for the manager only — pending \ + description = "List loose ends pending against this agent: unanswered questions \ + where you are the asker (waiting on someone) or the target (someone's waiting on \ + you), pending reminders you scheduled, plus — for the manager only — pending \ approvals you submitted that the operator hasn't acted on yet (agents with the \ `approvals` tool group also see their own pending approvals). Also lists active \ local tasks published by external MCP daemons (e.g. running bash tasks). Cheap sweep, no args. Useful \ at turn start to remember what you owe / what's owed to you without scrolling \ inbox history. Output is a short bulleted list with ids, ages in seconds, and \ - the relevant context. A `reminder` row can be cancelled by \ + the relevant context. Each `question` or `reminder` row can be cancelled by \ passing its id + kind to `cancel_loose_end`. Empty result is reported clearly.\n\ Todos are a to-do list, not an inbox to empty on sight — a row that's still \ genuinely in flight (a running bash task, say) is expected to sit there and \ @@ -292,6 +293,16 @@ impl AgentServer { if is_self_query && let Some(reminders) = local_reminders().await { loose_ends.extend(reminders); } + // Merge local mirrored questions — same self-query-only + // restriction as todos/reminders above. c0re no longer sources + // `Question` rows for `for_agent`/`hive_wide` (see + // `hive-c0re::loose_ends` doc), so this is the only place a + // self-query sees its own questions now; a manager query for a + // child still sees that child's approvals (unaffected) but no + // longer its questions, matching the reminders precedent. + if is_self_query && let Some(questions) = local_questions().await { + loose_ends.extend(questions); + } annotate_retries(render_loose_ends(&loose_ends), retries) }) .await @@ -354,16 +365,20 @@ impl AgentServer { .await } - #[tool(description = "Cancel an open thread you own — a `reminder` \ - you scheduled (hard-deleted before it fires). `kind` is `\"reminder\"`; \ - `id` is the row id from the matching `get_loose_ends` entry. Auth: you can only \ - cancel rows where you're the owner. Returns `ok` or an error string.\n\ + #[tool( + description = "Cancel an open thread you own — a `question` you asked (the \ + asker gets `[cancelled by ]` as the answer and unblocks) or a `reminder` \ + you scheduled (hard-deleted before it fires). `kind` is `\"question\"` or \ + `\"reminder\"`; `id` is the row id from the matching `get_loose_ends` entry \ + or the `question_queued` reply you got when you submitted. Auth: you can only \ + cancel rows where you're the asker / owner. Returns `ok` or an error string.\n\ `kind` may also be `\"approval\"` to withdraw a pending approval you submitted \ (before the operator acts on it) — root agent (`ruth`) only; the server rejects \ `approval` kind for all other callers.\n\ `kind` may also be `\"todo\"` to clear one of your own loose-ends-v2 todos (the id \ from a `get_loose_ends` `todo #N` line) — dials the in-agent socket directly, safe \ - to call repeatedly.")] + to call repeatedly." + )] async fn cancel_loose_end(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); let id = args.id; @@ -413,6 +428,15 @@ impl AgentServer { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::CancelLooseEnd { kind, id }) .await; + if resp.is_ok() && kind == hive_sh4re::inbox::CancelLooseEndKind::Question { + // Best-effort — cancel is ownership-gated to the asker on + // the c0re side, so a successful cancel here always means + // *this* agent's own `asked` mirror row for `id`. Known gap + // (documented on `Questions`): the target isn't notified, + // so their `answering` row always lingers until it goes + // stale — `answer()` no longer exists as an option. + let _ = dial_agent_socket(&hive_agent_sock::Request::ClearQuestion { id }).await; + } annotate_retries( format_ack( resp, diff --git a/hive-agent-mcp/src/mcp/render.rs b/hive-agent-mcp/src/mcp/render.rs index bb6b3466..66a09d61 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -408,6 +408,17 @@ pub(super) async fn local_reminders() -> Option } } +/// Query the harness's in-agent socket for this agent's mirrored questions +/// (both roles — asked and answering). Same best-effort +/// contract as [`local_reminders`]; c0re stays the actual `Ask`/`Answer` +/// routing, this only mirrors the durable "still owed a reply" view. +pub(super) async fn local_questions() -> Option> { + match dial_agent_socket(&hive_agent_sock::Request::ListQuestions).await? { + hive_agent_sock::Response::LooseEnds { loose_ends } => Some(loose_ends), + _ => None, + } +} + /// Mark one of this agent's local todos (loose-ends v2) done by id, via /// the harness's in-agent socket — reachable through `cancel_loose_end` /// kind `"todo"` so clearing a todo never has to shell out through a diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs index aab4e2e7..8fb6641b 100644 --- a/hive-agent-sock/src/lib.rs +++ b/hive-agent-sock/src/lib.rs @@ -4,9 +4,9 @@ //! ops below is a plain string, not a closed set: any user-configured MCP //! server declared in an agent's `agent.nix` can dial this socket and push //! its own todos the same way. Carries the loose-ends-v2 *todo* op family -//! plus the harness-local *reminder* op family; more in-agent request -//! families may be added over time (the socket is deliberately named for -//! the agent, not the todos). +//! plus the harness-local *reminder* and *question* op families; more +//! in-agent request families may be added over time (the socket is +//! deliberately named for the agent, not the todos). //! //! Distinct from `hive-core-agent-sock`, the *host*-served core↔agent //! protocol on `/run/hive/mcp.sock`: this socket never leaves the @@ -123,6 +123,30 @@ pub enum Request { /// through this variant, since a human watching the dashboard isn't /// waiting on a wake. Compact { wake_prompt: Option }, + /// Mirror an outstanding question this agent asked (`ask()` succeeded). + /// `target` is who it's waiting on (`"operator"` when asked with + /// `to: None`). Part of the questions-mirror increment — see + /// `hive-agent::questions`. + RecordAskedQuestion { + id: i64, + target: String, + question: String, + }, + /// Mirror an outstanding question this agent was asked (a + /// `question_asked` system event arrived in the inbox). `asker` is who's + /// waiting on this agent for a reply. + RecordAnsweringQuestion { + id: i64, + asker: String, + question: String, + }, + /// Drop the mirror row for `id` (either role) — the question resolved + /// from this agent's side (answered, or the `question_answered` event + /// for a question this agent asked arrived). + ClearQuestion { id: i64 }, + /// List this agent's mirrored questions (both roles) — single-agent + /// scope, same shape as `ListReminders`. + ListQuestions, } /// A response on the in-agent socket. Serialised with a `kind` tag, @@ -134,8 +158,8 @@ pub enum Response { Ok, /// Op succeeded and touched `count` rows (clear / mark-done). Acked { count: u64 }, - /// `ListTodos` / `ListReminders` result (each wraps its rows as the - /// matching [`LooseEnd`] variant). + /// `ListTodos` / `ListReminders` / `ListQuestions` result (each wraps + /// its rows as the matching [`LooseEnd`] variant). LooseEnds { loose_ends: Vec }, /// `CountPendingReminders` result. PendingRemindersCount { count: u64 }, diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index d59d6205..b6a8028e 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -22,6 +22,7 @@ mod otel_turn_metrics; mod paths; mod plugins; mod prompt; +mod questions; mod reminder_timer; mod reminders; mod serve_common; @@ -108,7 +109,17 @@ async fn main() -> Result<()> { /// `ContainerCrash`, reparent notifications, and friends; the parse /// and log path is identical. Quiet no-op when `from` isn't /// `SYSTEM_SENDER`. -fn log_system_event(bus: &Bus, from: &str, body: &str) { +/// +/// Also keeps the harness-local questions mirror in sync on the two +/// question-shaped events: `QuestionAsked` (a peer/manager asked *this* +/// agent — mirror an `Answering` row) and +/// `QuestionAnswered` (a question *this* agent asked got a reply — +/// clear the mirrored `Asked` row). Best-effort loopback dial of the +/// in-agent socket (see `todo_server::dial`'s docs on why a dial beats +/// threading an `Arc` through this whole call chain) — a +/// dial failure is a missed mirror update, not a turn failure, so it's +/// logged and swallowed rather than propagated. +async fn log_system_event(bus: &Bus, from: &str, body: &str) { if from != SYSTEM_SENDER { return; } @@ -118,11 +129,62 @@ fn log_system_event(bus: &Bus, from: &str, body: &str) { } else { tracing::info!(%from, %body, "system message"); } + match parsed { + Some(HelperEvent::QuestionAsked { + id, + asker, + question, + .. + }) => { + mirror_question(hive_agent_sock::Request::RecordAnsweringQuestion { + id, + asker, + question, + }) + .await; + } + Some(HelperEvent::QuestionAnswered { id, .. }) => { + mirror_question(hive_agent_sock::Request::ClearQuestion { id }).await; + } + _ => {} + } bus.emit(LiveEvent::Note { text: format!("[system] {body}"), }); } +/// Open the harness-local questions mirror against the consolidated state +/// db — same open-alongside shape as the reminders store +/// (`reminders::Reminders::open`) opened just above this call site in +/// `serve_main`, sharing the same file (distinct table, see +/// `questions::SCHEMA`). `None` on open failure disables question ops the +/// same way a failed reminders open disables reminder ops +/// (`no_questions_store` in `todo_server`). Pulled out to its own fn to +/// keep `serve_main` under the pedantic line-count lint. +fn open_question_store() -> Option> { + match questions::Questions::open(&paths::state_db()) { + Ok(store) => Some(Arc::new(store)), + Err(e) => { + tracing::error!(error = ?e, "open questions db failed — question mirror disabled"); + None + } + } +} + +/// Fire-and-forget helper for `log_system_event`'s questions-mirror dial: +/// logs a warning on dial failure / an `Err` response, otherwise silent. +async fn mirror_question(req: hive_agent_sock::Request) { + match todo_server::dial(&req).await { + Some(hive_agent_sock::Response::Err { message }) => { + tracing::warn!(%message, ?req, "questions mirror dial returned an error"); + } + None => { + tracing::warn!(?req, "questions mirror dial failed (socket unavailable?)"); + } + Some(_) => {} + } +} + /// Body string for the turn-failure notification we route to /// `` on `TurnError::Failed`. Reads the hive-qualified /// identity so the receiver sees `agent@hive` rather than relying on @@ -420,22 +482,23 @@ impl Surface for AgentSurface { // ---------- generic turn loop ---------- /// Opens the todos store and spawns the in-agent todo socket (loose-ends -/// v2 + harness-local reminders): the harness owns the todo + reminder -/// stores locally and serves the in-container producers on -/// `HIVE_AGENT_SOCKET`. A new/changed todo upsert fires the returned -/// `Notify` so the serve loop drives a turn directly — no broker +/// v2 + harness-local reminders + questions): the harness owns the todo + +/// reminder + question stores locally and serves the in-container +/// producers on `HIVE_AGENT_SOCKET`. A new/changed todo upsert fires the +/// returned `Notify` so the serve loop drives a turn directly — no broker /// round-trip, no marker files. Best-effort: if the todos store can't -/// open, the whole socket isn't served (reminder ops ride along on the -/// same listener, so they're gated on the same store — acceptable since a -/// from-scratch harness boot either has a writable harness dir or -/// doesn't). Split out of `serve_main` to keep it under clippy's -/// `too_many_lines` limit; kept alongside the returned `Notify` so the -/// serve loop's `LocalTodo` arm can gate a wake on `has_any()` before -/// spawning a turn — see its doc comment (the phantom-todo-wake issue: a -/// burst of same-turn upserts can arm a second `Notify` permit that -/// outlives the turn that already drained its payload). +/// open, the whole socket isn't served (reminder + question ops ride +/// along on the same listener, so they're gated on the same store — +/// acceptable since a from-scratch harness boot either has a writable +/// harness dir or doesn't). Split out of `serve_main` to keep it under +/// clippy's `too_many_lines` limit; kept alongside the returned `Notify` +/// so the serve loop's `LocalTodo` arm can gate a wake on `has_any()` +/// before spawning a turn — see its doc comment (the phantom-todo-wake +/// issue: a burst of same-turn upserts can arm a second `Notify` permit +/// that outlives the turn that already drained its payload). fn spawn_todo_socket( reminder_store: Option>, + question_store: Option>, bus: &Bus, ) -> (Arc, Option>) { let todo_wake = Arc::new(tokio::sync::Notify::new()); @@ -446,8 +509,14 @@ fn spawn_todo_socket( let bus_for_socket = bus.clone(); let store_for_socket = store.clone(); tokio::spawn(async move { - if let Err(e) = - todo_server::run(store_for_socket, wake, reminder_store, bus_for_socket).await + if let Err(e) = todo_server::run( + store_for_socket, + wake, + reminder_store, + question_store, + bus_for_socket, + ) + .await { tracing::error!(error = %e, "in-agent todo socket exited with error"); } @@ -577,7 +646,9 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { } }; tokio::spawn(reminder_timer::run(reminder_store.clone(), reminder_tx)); - let (todo_wake, todos_store) = spawn_todo_socket(reminder_store.clone(), &bus); + let question_store = open_question_store(); + let (todo_wake, todos_store) = + spawn_todo_socket(reminder_store.clone(), question_store.clone(), &bus); if matches!(initial, LoginState::NeedsLogin) { login::wait_for_login( &claude_dir, @@ -902,7 +973,7 @@ async fn handle_turn( let body = first.body; let redelivered = first.redelivered; let msg_id = first.id; - log_system_event(bus, &from, &body); + log_system_event(bus, &from, &body).await; tracing::info!(%from, %body, %redelivered, "inbox"); let unread = S::inbox_unread(socket).await; bus.emit(LiveEvent::TurnStart { diff --git a/hive-agent/src/paths.rs b/hive-agent/src/paths.rs index dad8007f..45158138 100644 --- a/hive-agent/src/paths.rs +++ b/hive-agent/src/paths.rs @@ -40,25 +40,19 @@ pub fn harness_dir() -> PathBuf { hive_agent_sock::paths::harness_dir() } -/// Consolidated harness-local state db — todos + reminders, one table -/// each — mutable per-agent state the harness owns, kept out of the -/// append-only `hyperhive-events.sqlite` sink. Per mara's call ("not yet -/// another sqlite! todos, reminders, questions should be like three tiny -/// tables in one 500kb sqlite"), this file is the shared home for the -/// loose-ends-v2 stores; each store's `open()` only applies its own +/// Consolidated harness-local state db — todos + reminders + the questions +/// mirror, one table each — mutable per-agent state the harness owns, kept +/// out of the append-only `hyperhive-events.sqlite` sink. Per mara's call +/// ("not yet another sqlite! todos, reminders, questions should be like +/// three tiny tables in one 500kb sqlite"), this file is the shared home +/// for all loose-ends-v2 stores; each store's `open()` only applies its own /// `CREATE TABLE IF NOT EXISTS`, so opening multiple stores against the -/// same path is safe (distinct table names, no schema collision). Both -/// stores (todos, reminders) open this same path directly (see their -/// `open()` call sites) — distinct table names mean no schema collision, -/// so there's no need for per-store path wrapper fns here. +/// same path is safe (distinct table names, no schema collision). +/// All three stores (todos, reminders, questions) open this same path +/// directly (see their `open()` call sites) — distinct table names mean no +/// schema collision, so there's no need for per-store path wrapper fns here. /// -/// ⚠️ A pre-existing harness may still carry a third table here, `questions` -/// — the mirror this file's own doc comment used to describe as one of -/// three. It's gone now that the ask/answer mechanism itself has been -/// removed: nothing opens or writes it any more, so any rows already in -/// it are inert, harmless leftovers, not a store to migrate or clean up. -/// -/// Before the todos/reminders consolidation, they lived in their own +/// Before this consolidation, todos and reminders lived in their own /// `hyperhive-todos.sqlite` / `hyperhive-reminders.sqlite` files; a /// one-time boot migration (`db_migrate::run`) folds those into this path /// the first time a harness boots after the upgrade. diff --git a/hive-agent/src/questions.rs b/hive-agent/src/questions.rs new file mode 100644 index 00000000..62a3fd95 --- /dev/null +++ b/hive-agent/src/questions.rs @@ -0,0 +1,235 @@ +//! Harness-local questions mirror — the second increment of the +//! loose-ends-v2 migration's questions phase (see the design comment on +//! the tracking issue). c0re stays the `Ask`/`Answer` routing + delivery +//! rendezvous (mara's "2a" call); this store only holds the *durable "I +//! still owe/am owed a reply" view* `get_loose_ends` renders, so that view +//! survives a hive migration the same way todos/reminders already do. +//! +//! One row per outstanding question **from this agent's point of view**, +//! keyed by the c0re-assigned question id (globally unique — an agent is +//! never both asker and target of the same question, self-asks are +//! rejected at the c0re layer): +//! +//! - `role = "asked"`: this agent called `ask()`; `peer` is the target +//! (`"operator"` when asked with `to: None`). Cleared when the +//! `question_answered` system event for `id` arrives (see `main.rs`'s +//! inbound-event hook). +//! - `role = "answering"`: this agent received a `question_asked` system +//! event for `id`; `peer` is the asker. Cleared when this agent calls +//! `answer()` for `id` (see `hive-agent-mcp`'s tool impl). +//! +//! Known gap: if the asker cancels their own question, the target is not +//! proactively notified today (`hive-c0re::questions::handle_cancel_loose_end` +//! only notifies a *different* canceller than the asker, which never +//! happens via the ownership-gated agent-socket cancel path). A target's +//! `answering` row is cleared the next time they call `answer()` — c0re's +//! "already answered"/"not found" rejection is treated as resolved-without- +//! us and clears the mirror row (see `hive-agent-mcp`'s `answer()` tool) — +//! but if they never call `answer()` at all, the row lingers with no +//! proactive nudge. Flagged on the tracking issue rather than fully fixed. + +use std::path::Path; +use std::sync::Mutex; + +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; +use rusqlite::{Connection, params}; + +const SCHEMA: &str = r" +CREATE TABLE IF NOT EXISTS questions ( + id INTEGER PRIMARY KEY, + role TEXT NOT NULL, + peer TEXT NOT NULL, + question TEXT NOT NULL, + asked_at INTEGER NOT NULL +); +"; + +/// Which side of the question this agent is on. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Role { + /// This agent asked; `peer` is who it's waiting on. + Asked, + /// This agent was asked; `peer` is who's waiting on it. + Answering, +} + +impl Role { + const fn as_str(self) -> &'static str { + match self { + Role::Asked => "asked", + Role::Answering => "answering", + } + } + + fn parse(s: &str) -> Option { + match s { + "asked" => Some(Role::Asked), + "answering" => Some(Role::Answering), + _ => None, + } + } +} + +/// One mirrored question row. +#[derive(Debug, Clone)] +pub struct QuestionMirror { + pub id: i64, + pub role: Role, + pub peer: String, + pub question: String, + pub asked_at: DateTime, +} + +/// The harness-local questions mirror. Same sharing/locking shape as +/// [`crate::reminders::Reminders`] — cheap behind an `Arc`, short sqlite +/// writes guarded by a `Mutex`. +pub struct Questions { + conn: Mutex, +} + +impl Questions { + /// Open (creating if needed) the questions mirror at `path`. + /// + /// # Errors + /// + /// Propagates sqlite open / schema-apply failures. + pub fn open(path: &Path) -> Result { + let conn = Connection::open(path) + .with_context(|| format!("open questions db {}", path.display()))?; + conn.execute_batch(SCHEMA) + .context("apply questions schema")?; + Ok(Self { + conn: Mutex::new(conn), + }) + } + + /// Record a new outstanding row. `INSERT OR REPLACE` so a caller that + /// re-observes the same id (e.g. a redelivered `question_asked`) is a + /// harmless no-op rather than a unique-constraint error. + /// + /// # Errors + /// + /// Propagates the sqlite insert failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn record(&self, id: i64, role: Role, peer: &str, question: &str) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "INSERT OR REPLACE INTO questions (id, role, peer, question, asked_at) \ + VALUES (?1, ?2, ?3, ?4, ?5)", + params![id, role.as_str(), peer, question, Utc::now().timestamp()], + )?; + Ok(()) + } + + /// Drop the mirror row for `id` (either role) — the question resolved + /// (answered/cancelled) from this agent's side. Returns the number of + /// rows removed (0 = no local row for that id, a harmless no-op). + /// + /// # Errors + /// + /// Propagates the sqlite delete failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn clear(&self, id: i64) -> Result { + let conn = self.conn.lock().unwrap(); + let n = conn.execute("DELETE FROM questions WHERE id = ?1", params![id])?; + Ok(n) + } + + /// List every mirrored row, oldest-asked first — for `get_loose_ends` + /// rendering. A row whose `role` column doesn't parse (corruption — + /// should never happen via this module's own writes) is logged and + /// skipped rather than either propagating a hard error (which would + /// hide every other, valid row) or silently misattributing it as + /// `Asked` (which would lie about who owes whom a reply). + /// + /// # Errors + /// + /// Propagates the sqlite prepare / query / column-read failures. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn list(&self) -> Result> { + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare( + "SELECT id, role, peer, question, asked_at FROM questions ORDER BY asked_at ASC", + )?; + let mut out = Vec::new(); + let mut rows = stmt.query([])?; + while let Some(row) = rows.next()? { + let id: i64 = row.get(0)?; + let role_str: String = row.get(1)?; + let Some(role) = Role::parse(&role_str) else { + tracing::warn!(%id, %role_str, "questions mirror: unknown role, skipping row"); + continue; + }; + let asked_at_secs: i64 = row.get(4)?; + out.push(QuestionMirror { + id, + role, + peer: row.get(2)?, + question: row.get(3)?, + asked_at: chrono::DateTime::from_timestamp(asked_at_secs, 0) + .unwrap_or_else(Utc::now), + }); + } + Ok(out) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn store() -> (tempfile::TempDir, Questions) { + let dir = tempfile::tempdir().unwrap(); + let db = Questions::open(&dir.path().join("questions.sqlite")).unwrap(); + (dir, db) + } + + #[test] + fn record_and_list_both_roles() { + let (_dir, s) = store(); + s.record(1, Role::Asked, "atlas", "are we there yet") + .unwrap(); + s.record(2, Role::Answering, "mara", "is this fine") + .unwrap(); + let rows = s.list().unwrap(); + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].id, 1); + assert_eq!(rows[0].role, Role::Asked); + assert_eq!(rows[0].peer, "atlas"); + assert_eq!(rows[1].id, 2); + assert_eq!(rows[1].role, Role::Answering); + assert_eq!(rows[1].peer, "mara"); + } + + #[test] + fn clear_removes_the_row() { + let (_dir, s) = store(); + s.record(1, Role::Asked, "atlas", "q").unwrap(); + assert_eq!(s.clear(1).unwrap(), 1); + assert!(s.list().unwrap().is_empty()); + assert_eq!(s.clear(1).unwrap(), 0, "already-cleared id is a no-op"); + assert_eq!(s.clear(999).unwrap(), 0, "unknown id is a no-op"); + } + + #[test] + fn record_is_idempotent_on_redelivery() { + let (_dir, s) = store(); + s.record(1, Role::Answering, "mara", "q").unwrap(); + s.record(1, Role::Answering, "mara", "q").unwrap(); + assert_eq!( + s.list().unwrap().len(), + 1, + "re-observing the same id doesn't duplicate" + ); + } +} diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index c7fa7828..307d874d 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -1,20 +1,24 @@ //! In-agent socket server (loose-ends v2 + harness-local reminders + -//! self-service compact). Binds the harness-owned `HIVE_AGENT_SOCKET` and -//! serves the `hive-agent-sock` protocol to the in-container producers — -//! matrix / bash daemons and forge-notify are the built-in ones, but any -//! user-configured MCP server can dial the same socket and push its own -//! todos — and to `hive-agent-mcp`'s `remind`/`get_loose_ends`/ +//! questions mirror + self-service compact). Binds the harness-owned +//! `HIVE_AGENT_SOCKET` and serves the `hive-agent-sock` protocol to the +//! in-container producers — matrix / bash daemons and forge-notify are the +//! built-in ones, but any user-configured MCP server can dial the same +//! socket and push its own todos — and to +//! `hive-agent-mcp`'s `ask`/`answer`/`remind`/`get_loose_ends`/ //! `cancel_loose_end`/`compact` tool impls. Todo ops hit the harness-local //! [`Todos`] store; a new-or-changed upsert fires an in-process [`Notify`] //! so the serve loop drives a turn. Reminder ops hit the harness-local //! [`Reminders`] store (`None` when the store failed to open — every //! reminder op then returns `Response::Err`); a reminder *firing* is a //! separate path (`reminder_timer`), not driven through this socket. -//! `Request::Compact` is the odd one out — it doesn't touch any store, just -//! the harness's [`Bus`] (gate-checked context usage, then the same -//! deferred `compact_pending` flag the operator dashboard's `/compact` -//! button sets). No hive-c0re round-trip, no broker long-poll, no marker -//! files. +//! Question ops hit the harness-local [`Questions`] mirror the same way +//! (`None` when it failed to open) — c0re stays the actual `Ask`/`Answer` +//! routing + delivery rendezvous, this store only mirrors the durable +//! "still owed a reply" view for `get_loose_ends`. `Request::Compact` is +//! the odd one out — it doesn't touch any store, just the harness's +//! [`Bus`] (gate-checked context usage, then the same deferred +//! `compact_pending` flag the operator dashboard's `/compact` button +//! sets). No hive-c0re round-trip, no broker long-poll, no marker files. //! //! One request/response line per connection, matching the producers' //! existing best-effort JSON-line clients (they just change which socket @@ -32,6 +36,7 @@ use tokio::net::{UnixListener, UnixStream}; use tokio::sync::Notify; use crate::events::Bus; +use crate::questions::{QuestionMirror, Questions, Role}; use crate::reminders::{Reminder, Reminders}; use crate::todos::{Todo, Todos}; @@ -83,6 +88,7 @@ pub async fn run( store: Arc, wake: Arc, reminders: Option>, + questions: Option>, bus: Bus, ) -> Result<()> { let Some(path) = socket_path() else { @@ -97,10 +103,18 @@ pub async fn run( let store = store.clone(); let wake = wake.clone(); let reminders = reminders.clone(); + let questions = questions.clone(); let bus = bus.clone(); tokio::spawn(async move { - if let Err(e) = - handle_conn(stream, &store, &wake, reminders.as_deref(), &bus).await + if let Err(e) = handle_conn( + stream, + &store, + &wake, + reminders.as_deref(), + questions.as_deref(), + &bus, + ) + .await { tracing::warn!(error = ?e, "in-agent todo connection failed"); } @@ -145,6 +159,7 @@ async fn handle_conn( store: &Todos, wake: &Notify, reminders: Option<&Reminders>, + questions: Option<&Questions>, bus: &Bus, ) -> Result<()> { let (read, mut write) = stream.into_split(); @@ -154,7 +169,7 @@ async fn handle_conn( return Ok(()); } let resp = match serde_json::from_str::(line.trim()) { - Ok(req) => dispatch(req, store, wake, reminders, bus), + Ok(req) => dispatch(req, store, wake, reminders, questions, bus), Err(e) => Response::Err { message: format!("bad request: {e}"), }, @@ -170,15 +185,16 @@ async fn handle_conn( /// the serve loop runs a turn. Each arm calls a small named handler function /// directly — no sub-match/`unreachable!()` indirection per family (that /// pattern got reviewed out of the todo family in the diagnostic-logging -/// follow-up PR; kept the reminder family consistent with it here rather -/// than reintroducing it). `reminders` is `None` when that store failed to -/// open at boot, in which case every reminder op returns an `Err` — each +/// follow-up PR; kept the reminder and question families consistent with it +/// here rather than reintroducing it). `reminders`/`questions` are `None` when that store failed to open at +/// boot, in which case every op in that family returns an `Err` — each /// handler checks for its own `None` case. fn dispatch( req: Request, store: &Todos, wake: &Notify, reminders: Option<&Reminders>, + questions: Option<&Questions>, bus: &Bus, ) -> Response { match req { @@ -214,6 +230,18 @@ fn dispatch( Request::CancelReminder { id } => cancel_reminder(reminders, id), Request::CountPendingReminders => count_pending_reminders(reminders), Request::ReminderRollup { since_secs } => reminder_rollup(reminders, since_secs), + Request::RecordAskedQuestion { + id, + target, + question, + } => record_asked_question(questions, id, &target, &question), + Request::RecordAnsweringQuestion { + id, + asker, + question, + } => record_answering_question(questions, id, &asker, &question), + Request::ClearQuestion { id } => clear_question(questions, id), + Request::ListQuestions => list_questions(questions), Request::Compact { wake_prompt } => compact(bus, wake_prompt), } } @@ -284,6 +312,65 @@ fn reminder_rollup(reminders: Option<&Reminders>, since_secs: u64) -> Response { } } +/// `RecordAskedQuestion` handler: mirror a question this agent asked. +fn record_asked_question( + questions: Option<&Questions>, + id: i64, + target: &str, + question: &str, +) -> Response { + let Some(q) = questions else { + return no_questions_store(); + }; + match q.record(id, Role::Asked, target, question) { + Ok(()) => Response::Ok, + Err(e) => err(&e), + } +} + +/// `RecordAnsweringQuestion` handler: mirror a question this agent owes a +/// reply to. +fn record_answering_question( + questions: Option<&Questions>, + id: i64, + asker: &str, + question: &str, +) -> Response { + let Some(q) = questions else { + return no_questions_store(); + }; + match q.record(id, Role::Answering, asker, question) { + Ok(()) => Response::Ok, + Err(e) => err(&e), + } +} + +/// `ClearQuestion` handler: drop the mirror row for `id` (either role). +fn clear_question(questions: Option<&Questions>, id: i64) -> Response { + let Some(q) = questions else { + return no_questions_store(); + }; + match q.clear(id) { + Ok(count) => Response::Acked { + count: u64::try_from(count).unwrap_or(0), + }, + Err(e) => err(&e), + } +} + +/// `ListQuestions` handler: this agent's mirrored questions (both roles). +fn list_questions(questions: Option<&Questions>) -> Response { + let Some(q) = questions else { + return no_questions_store(); + }; + match q.list() { + Ok(rows) => Response::LooseEnds { + loose_ends: rows.into_iter().map(question_to_loose_end).collect(), + }, + Err(e) => err(&e), + } +} + /// `UpsertTodo` handler: writes/refreshes a todo row, logs the outcome, and /// fires `wake` on a new-or-changed upsert so the serve loop runs a turn. fn upsert_todo( @@ -424,6 +511,34 @@ fn no_reminders_store() -> Response { } } +/// Shared "questions mirror unavailable" response for every question op +/// when the store failed to open at boot (see `main.rs`'s best-effort open). +fn no_questions_store() -> Response { + Response::Err { + message: "questions mirror unavailable on this agent".to_owned(), + } +} + +/// Map a mirrored [`QuestionMirror`] to a [`LooseEnd::Question`]. `asker`/ +/// `target` are derived from `role` — this agent's own label fills whichever +/// side `role` says is us, `peer` fills the other. +fn question_to_loose_end(q: QuestionMirror) -> LooseEnd { + let now = chrono::Utc::now().timestamp(); + let age = u64::try_from(now.saturating_sub(q.asked_at.timestamp())).unwrap_or(0); + let me = crate::identity::label(); + let (asker, target) = match q.role { + Role::Asked => (me, Some(q.peer)), + Role::Answering => (q.peer, Some(me)), + }; + LooseEnd::Question { + id: q.id, + asker, + target, + question: q.question, + age_seconds: age, + } +} + /// Map a stored [`Reminder`] to a [`LooseEnd::Reminder`], deriving /// `age_seconds` from `created_at` (mirrors the old c0re rendering — /// "age" is how long the reminder has been *scheduled*, not how soon