From 94712016981d184043569db89b119116fb48b63b Mon Sep 17 00:00:00 2001 From: damocles Date: Thu, 23 Jul 2026 20:55:54 +0200 Subject: [PATCH 1/3] feat(#2635): add harness-local questions mirror store (inc2 pt2, unwired) --- hive-agent/src/main.rs | 1 + hive-agent/src/questions.rs | 228 ++++++++++++++++++++++++++++++++++++ 2 files changed, 229 insertions(+) create mode 100644 hive-agent/src/questions.rs diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 78efa744..16a2e820 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -21,6 +21,7 @@ mod mcp_config; mod paths; mod plugins; mod prompt; +mod questions; mod reminder_timer; mod reminders; mod serve_common; diff --git a/hive-agent/src/questions.rs b/hive-agent/src/questions.rs new file mode 100644 index 00000000..b576df7e --- /dev/null +++ b/hive-agent/src/questions.rs @@ -0,0 +1,228 @@ +//! Harness-local questions mirror — increment 2 part 2 of #2635 (see the +//! design comment on the 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, so `id` alone +//! is the primary key — 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 in this agent's +//! inbox (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 +//! (`cancel_loose_end` kind `"question"`), the target is not notified +//! today (`hive-c0re::questions::handle_cancel_loose_end` only notifies +//! a *different* canceller than the asker, which never happens on the +//! agent-socket path since cancel is ownership-gated to the asker). A +//! target's `answering` row then lingers until they call `answer()` +//! (which now returns a "no such question" error from c0re rather than +//! silently succeeding) or the row goes stale. Flagged on #2635 rather +//! than fixed here — fixing it means teaching `handle_cancel_loose_end` +//! to also notify the target, a small but separate change. + +use std::path::Path; +use std::sync::Mutex; + +use anyhow::{Context, Result}; +use hive_sh4re::wire_time::now_unix; +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: i64, +} + +/// 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, now_unix()], + )?; + 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. + /// + /// # Errors + /// + /// Propagates the sqlite prepare / query 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 rows = stmt + .query_map([], row_to_question)? + .collect::>>()?; + Ok(rows) + } +} + +fn row_to_question(row: &rusqlite::Row) -> rusqlite::Result { + let role_str: String = row.get(1)?; + let role = Role::parse(&role_str).unwrap_or(Role::Asked); + Ok(QuestionMirror { + id: row.get(0)?, + role, + peer: row.get(2)?, + question: row.get(3)?, + asked_at: row.get(4)?, + }) +} + +#[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" + ); + } +} From e5ef5a72be75406e69a7e0623b9ddada9866e269 Mon Sep 17 00:00:00 2001 From: damocles Date: Thu, 23 Jul 2026 21:10:51 +0200 Subject: [PATCH 2/3] feat(#2635): wire harness-local questions mirror (inc2 pt2) --- hive-agent-mcp/src/mcp/mod.rs | 56 ++++++- hive-agent-mcp/src/mcp/render.rs | 11 ++ hive-agent-sock/src/lib.rs | 34 +++- hive-agent/src/main.rs | 156 +++++++++++++----- hive-agent/src/paths.rs | 22 ++- hive-agent/src/questions.rs | 39 ++--- hive-agent/src/todo_server.rs | 273 ++++++++++++++++++++++++------- hive-c0re/src/loose_ends.rs | 71 +++----- 8 files changed, 468 insertions(+), 194 deletions(-) diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index d3a14b91..6e16b4ef 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -35,9 +35,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, 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, 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. @@ -186,9 +186,13 @@ impl AgentServer { Ok(to) => to, Err(reason) => return format!("invalid `to` agent name: {reason}"), }; + let question = args.question; + let target = to + .as_ref() + .map_or_else(|| "operator".to_owned(), std::string::ToString::to_string); let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::Ask { - question: args.question, + question: question.clone(), options: args.options, multi: args.multi, ttl_seconds: args.ttl_seconds, @@ -196,10 +200,22 @@ impl AgentServer { }) .await; let s = match resp { - Ok(hive_core_agent_sock::Response::QuestionQueued { id }) => format!( - "question queued (id={id}); answer will arrive as a system \ - `question_answered` event in your inbox" - ), + Ok(hive_core_agent_sock::Response::QuestionQueued { id }) => { + // Best-effort local questions-mirror record — a dial + // failure just means `get_loose_ends` won't show this + // row locally; the actual question is already queued + // in c0re regardless. + let _ = dial_agent_socket(&hive_agent_sock::Request::RecordAskedQuestion { + id, + target, + question, + }) + .await; + format!( + "question queued (id={id}); answer will arrive as a system \ + `question_answered` event in your inbox" + ) + } other => reply_err(other, "ask"), }; annotate_retries(s, retries) @@ -226,6 +242,11 @@ impl AgentServer { answer: args.answer, }) .await; + if resp.is_ok() { + // Best-effort — this agent is done owing a reply for `id`; + // drop its `answering` mirror row. + let _ = dial_agent_socket(&hive_agent_sock::Request::ClearQuestion { id }).await; + } annotate_retries( format_ack(resp, "answer", format!("answered question {id}")), retries, @@ -350,6 +371,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 @@ -476,6 +507,15 @@ impl AgentServer { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::CancelLooseEnd { kind, id }) .await; + if resp.is_ok() && kind == hive_sh4re::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 lingers until they call + // `answer()` or it goes stale. + 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 6ac53a37..ddac1248 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -351,6 +351,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 a3ff9a45..e3cd83ed 100644 --- a/hive-agent-sock/src/lib.rs +++ b/hive-agent-sock/src/lib.rs @@ -1,9 +1,9 @@ //! Wire types for the *in-agent* socket, served by the hive-agent harness //! to the in-container producers (matrix / bash MCP daemons) and //! `forge_notify`. 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). +//! 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 @@ -91,6 +91,30 @@ pub enum Request { /// deferred `compact_pending` flag the operator's button sets (consumed /// at the next turn boundary), so it never races a live claude process. Compact, + /// 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, @@ -102,8 +126,8 @@ pub enum Response { Ok, /// Op succeeded and touched `count` rows (clear / mark-done). Acked { count: u64 }, - /// `ListTodos` / `ListReminders` result (the latter wraps each row as - /// [`LooseEnd::Reminder`]). + /// `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 16a2e820..4a4db3b3 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -84,21 +84,82 @@ 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; } let parsed = serde_json::from_str::(body).ok(); - if let Some(event) = parsed { + if let Some(event) = &parsed { tracing::info!(?event, "helper event"); } 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 @@ -393,6 +454,56 @@ impl Surface for AgentSurface { // ---------- generic turn loop ---------- +/// Opens the todos store and spawns the in-agent todo socket (loose-ends +/// 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 + 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()); + let todos_store: Option> = match todos::Todos::open(&paths::state_db()) { + Ok(store) => { + let store = Arc::new(store); + let wake = todo_wake.clone(); + 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, + question_store, + bus_for_socket, + ) + .await + { + tracing::error!(error = %e, "in-agent todo socket exited with error"); + } + }); + Some(store) + } + Err(e) => { + tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled"); + None + } + }; + (todo_wake, todos_store) +} + /// Boot — wires up the web UI, login state, stats, plugins, forge /// notifier, and either drops into `serve_loop` directly (`Online`) or /// parks on the login flow first (`NeedsLogin`). See @@ -491,42 +602,9 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { } }; tokio::spawn(reminder_timer::run(reminder_store.clone(), reminder_tx)); - // 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 `todo_wake` 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). - let todo_wake = Arc::new(tokio::sync::Notify::new()); - // Kept alongside `todo_wake` 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). - let todos_store: Option> = match todos::Todos::open(&paths::state_db()) { - Ok(store) => { - let store = Arc::new(store); - let wake = todo_wake.clone(); - let reminders = reminder_store.clone(); - 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, reminders, bus_for_socket).await - { - tracing::error!(error = %e, "in-agent todo socket exited with error"); - } - }); - Some(store) - } - Err(e) => { - tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled"); - None - } - }; + 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, login_state.clone(), &bus, poll_ms).await; } else { @@ -690,7 +768,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 ebd4e200..bfcf378f 100644 --- a/hive-agent/src/paths.rs +++ b/hive-agent/src/paths.rs @@ -40,19 +40,17 @@ pub fn harness_dir() -> PathBuf { hive_sh4re::paths::harness_dir() } -/// Consolidated harness-local state db — currently 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 all -/// 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). A -/// questions mirror table is the planned third tenant (a following -/// increment), not part of this schema yet. -/// Both the todos and reminders stores 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. /// /// Before this consolidation, todos and reminders lived in their own /// `hyperhive-todos.sqlite` / `hyperhive-reminders.sqlite` files; a diff --git a/hive-agent/src/questions.rs b/hive-agent/src/questions.rs index b576df7e..9e333f0c 100644 --- a/hive-agent/src/questions.rs +++ b/hive-agent/src/questions.rs @@ -1,33 +1,30 @@ -//! Harness-local questions mirror — increment 2 part 2 of #2635 (see the -//! design comment on the 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. +//! 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, so `id` alone -//! is the primary key — an agent is never both asker and target of the -//! same question, self-asks are rejected at the c0re layer): +//! 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 in this agent's -//! inbox (see `main.rs`'s inbound-event hook). +//! `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 -//! (`cancel_loose_end` kind `"question"`), the target is not notified -//! today (`hive-c0re::questions::handle_cancel_loose_end` only notifies -//! a *different* canceller than the asker, which never happens on the -//! agent-socket path since cancel is ownership-gated to the asker). A -//! target's `answering` row then lingers until they call `answer()` -//! (which now returns a "no such question" error from c0re rather than -//! silently succeeding) or the row goes stale. Flagged on #2635 rather -//! than fixed here — fixing it means teaching `handle_cancel_loose_end` -//! to also notify the target, a small but separate change. +//! Known gap: if the asker cancels their own question, the target is not +//! 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 then lingers until they call `answer()` (now a "no +//! such question" error from c0re) or the row goes stale. Flagged on the +//! tracking issue rather than fixed here. use std::path::Path; use std::sync::Mutex; diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index 1336f1fb..3c2e4a99 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -1,18 +1,22 @@ //! 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, forge-notify) and to `hive-agent-mcp`'s -//! `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 either -//! 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. +//! 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, forge-notify) 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. +//! 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 @@ -29,6 +33,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}; @@ -84,6 +89,7 @@ pub async fn run( store: Arc, wake: Arc, reminders: Option>, + questions: Option>, bus: Bus, ) -> Result<()> { let Some(path) = socket_path() else { @@ -98,10 +104,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"); } @@ -133,6 +147,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(); @@ -142,7 +157,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}"), }, @@ -155,13 +170,19 @@ async fn handle_conn( } /// Apply one request to the store, firing `wake` on a new/changed upsert so -/// the serve loop runs a turn. `reminders` is `None` when that store -/// failed to open at boot — every reminder op then returns an `Err`. +/// 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 #2679; 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 { @@ -189,54 +210,152 @@ fn dispatch( message, timing, file_path, - } => match reminders { - Some(r) => { - match crate::reminder_timer::store(r, &message, &timing, file_path.as_deref()) { - Ok(_id) => Response::Ok, - Err(message) => Response::Err { message }, - } - } - None => no_reminders_store(), - }, - Request::ListReminders => match reminders { - Some(r) => match r.list_pending() { - Ok(rows) => Response::LooseEnds { - loose_ends: rows.into_iter().map(reminder_to_loose_end).collect(), - }, - Err(e) => err(&e), - }, - None => no_reminders_store(), - }, - Request::CancelReminder { id } => match reminders { - Some(r) => match r.cancel(id) { - Ok(count) => Response::Acked { - count: u64::try_from(count).unwrap_or(0), - }, - Err(e) => err(&e), - }, - None => no_reminders_store(), - }, - Request::CountPendingReminders => match reminders { - Some(r) => match r.count_pending() { - Ok(count) => Response::PendingRemindersCount { count }, - Err(e) => err(&e), - }, - None => no_reminders_store(), - }, - Request::ReminderRollup { since_secs } => match reminders { - Some(r) => { - let since_secs = i64::try_from(since_secs).unwrap_or(i64::MAX); - match r.rollup(since_secs) { - Ok(stats) => Response::ReminderRollup { stats }, - Err(e) => err(&e), - } - } - None => no_reminders_store(), - }, + } => store_reminder(reminders, &message, &timing, file_path.as_deref()), + Request::ListReminders => list_reminders(reminders), + 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 => compact(bus), } } +/// `StoreReminder` handler: persists a reminder due at `timing`. +fn store_reminder( + reminders: Option<&Reminders>, + message: &str, + timing: &hive_sh4re::ReminderTiming, + file_path: Option<&str>, +) -> Response { + let Some(r) = reminders else { + return no_reminders_store(); + }; + match crate::reminder_timer::store(r, message, timing, file_path) { + Ok(_id) => Response::Ok, + Err(message) => Response::Err { message }, + } +} + +/// `ListReminders` handler: this agent's own pending reminders. +fn list_reminders(reminders: Option<&Reminders>) -> Response { + let Some(r) = reminders else { + return no_reminders_store(); + }; + match r.list_pending() { + Ok(rows) => Response::LooseEnds { + loose_ends: rows.into_iter().map(reminder_to_loose_end).collect(), + }, + Err(e) => err(&e), + } +} + +/// `CancelReminder` handler: drops one pending reminder by id. +fn cancel_reminder(reminders: Option<&Reminders>, id: i64) -> Response { + let Some(r) = reminders else { + return no_reminders_store(); + }; + match r.cancel(id) { + Ok(count) => Response::Acked { + count: u64::try_from(count).unwrap_or(0), + }, + Err(e) => err(&e), + } +} + +/// `CountPendingReminders` handler. +fn count_pending_reminders(reminders: Option<&Reminders>) -> Response { + let Some(r) = reminders else { + return no_reminders_store(); + }; + match r.count_pending() { + Ok(count) => Response::PendingRemindersCount { count }, + Err(e) => err(&e), + } +} + +/// `ReminderRollup` handler: scheduled/delivered/pending counts over a +/// trailing `since_secs` window (`0` = all time). +fn reminder_rollup(reminders: Option<&Reminders>, since_secs: u64) -> Response { + let Some(r) = reminders else { + return no_reminders_store(); + }; + let since_secs = i64::try_from(since_secs).unwrap_or(i64::MAX); + match r.rollup(since_secs) { + Ok(stats) => Response::ReminderRollup { stats }, + Err(e) => err(&e), + } +} + +/// `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( @@ -352,6 +471,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 = hive_sh4re::wire_time::now_unix(); + let age = u64::try_from(now.saturating_sub(q.asked_at)).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 diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index c206b19f..621419fb 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -1,17 +1,22 @@ -//! Loose-ends aggregator. Walks the `approvals` + `operator_questions` -//! tables once per call and assembles a `Vec` for either -//! a single agent (`for_agent`) or the whole hive (`hive_wide`). -//! `Request::GetLooseEnds` from either the agent or manager socket -//! lands here so the routing logic + age-seconds derivation stay in -//! one place. Reminders are agent-local (in-container store) and no -//! longer sourced from here. +//! Loose-ends aggregator. Walks the `approvals` table once per call and +//! assembles a `Vec` for either a single agent (`for_agent`) or +//! the whole hive (`hive_wide`). `Request::GetLooseEnds` from either the +//! agent or manager socket lands here so the routing logic + age-seconds +//! derivation stay in one place. Reminders AND questions are agent-local +//! (in-container stores, `hive-agent::reminders` / `hive-agent::questions`) +//! and no longer sourced from here (loose-ends-v2's questions phase) — +//! c0re remains the `Ask`/`Answer` routing + delivery rendezvous +//! (`coord.questions`), it just isn't asked for the *pending-view* +//! rendering anymore. The +//! operator dashboard's questions pane is unaffected: it reads +//! `coord.questions.pending_all()` directly (`dashboard/state_snapshot.rs`), +//! independent of this module. //! //! Call frequency is low (an agent doing self-introspection between //! turns), so the sweep happens fresh every time — no caching, no -//! mutation events. If the sweep ever shows up in a profile, the -//! sqlite queries already filter on the same indexes -//! (`idx_approvals_pending` + `idx_operator_questions_pending`) that -//! the dashboard uses, so the bottleneck would be json +//! mutation events. If the sweep ever shows up in a profile, the sqlite +//! queries already filter on the same index (`idx_approvals_pending`) +//! that the dashboard uses, so the bottleneck would be json //! (de)serialisation, not the read. use anyhow::Result; @@ -26,19 +31,16 @@ use hive_sh4re::wire_time::now_unix; /// - pending approvals where this agent is the submitter (a parent /// agent with the `approvals` group submits for its children; the /// root submits for top-level agents). Legacy rows with no recorded -/// submitter count as the root's; -/// - unanswered questions where `agent` is the asker (waiting on -/// someone) OR the target (owes a reply). +/// submitter count as the root's. /// -/// Ordered `pending_messages` (when non-zero) → approvals → questions -/// within the returned vector. Within each kind, source-of-truth -/// ordering (sqlite's `pending()` queries return newest-first within -/// their indexes). +/// Ordered `pending_messages` (when non-zero) → approvals within the +/// returned vector. Within each kind, source-of-truth ordering (sqlite's +/// `pending()` query returns newest-first within its index). /// /// # Errors /// -/// Propagates errors from `count_pending` and the pending-approval / -/// question sqlite queries. +/// Propagates errors from `count_pending` and the pending-approval +/// sqlite query. pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { let now = now_unix(); let mut out = Vec::new(); @@ -72,26 +74,12 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { age_seconds: saturating_age(now, a.requested_at.timestamp()), }); } - for q in coord.questions.pending_all()? { - let role_match = q.asker == agent || q.target.as_deref() == Some(agent); - if !role_match { - continue; - } - out.push(LooseEnd::Question { - id: q.id, - asker: q.asker, - target: q.target, - question: q.question, - age_seconds: saturating_age(now, q.asked_at.timestamp()), - }); - } Ok(out) } -/// Hive-wide loose-ends view: EVERY pending approval + EVERY -/// unanswered question. Manager surface only; sub-agents can't see -/// each other's threads via the agent surface (`for_agent` filters by -/// name). +/// Hive-wide loose-ends view: EVERY pending approval. Manager surface +/// only; sub-agents can't see each other's threads via the agent surface +/// (`for_agent` filters by name). pub fn hive_wide(coord: &Coordinator) -> Result> { let now = now_unix(); let mut out = Vec::new(); @@ -104,15 +92,6 @@ pub fn hive_wide(coord: &Coordinator) -> Result> { age_seconds: saturating_age(now, a.requested_at.timestamp()), }); } - for q in coord.questions.pending_all()? { - out.push(LooseEnd::Question { - id: q.id, - asker: q.asker, - target: q.target, - question: q.question, - age_seconds: saturating_age(now, q.asked_at.timestamp()), - }); - } Ok(out) } From 7b2645078aec7f01fb100be7ced104d521425b9f Mon Sep 17 00:00:00 2001 From: damocles Date: Thu, 23 Jul 2026 22:01:27 +0200 Subject: [PATCH 3/3] fix(#2635): close review nits on the questions mirror (inc2 pt2) --- hive-agent-mcp/src/mcp/mod.rs | 19 ++++++++++-- hive-agent/src/questions.rs | 56 ++++++++++++++++++++--------------- hive-agent/src/todo_server.rs | 6 ++-- 3 files changed, 51 insertions(+), 30 deletions(-) diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index 6e16b4ef..f2b1fd8c 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -242,9 +242,22 @@ impl AgentServer { answer: args.answer, }) .await; - if resp.is_ok() { - // Best-effort — this agent is done owing a reply for `id`; - // drop its `answering` mirror row. + // Clear the local `answering` mirror row whenever this agent is + // done owing a reply for `id`: either a genuine success, or + // c0re telling us the question already resolved without us + // (the asker cancelled/answered it first, surfacing as an + // "already answered"/"not found" rejection) — in both cases + // nothing is still owed, so the row would otherwise linger + // stale. Any other rejection (e.g. wrong answerer) means the + // question is still genuinely outstanding, so leave it be. + let should_clear = match &resp { + Ok(hive_core_agent_sock::Response::Ok) => true, + Ok(hive_core_agent_sock::Response::Err { message }) => { + message.contains("not found") || message.contains("already answered") + } + _ => false, + }; + if should_clear { let _ = dial_agent_socket(&hive_agent_sock::Request::ClearQuestion { id }).await; } annotate_retries( diff --git a/hive-agent/src/questions.rs b/hive-agent/src/questions.rs index 9e333f0c..c2c5f8ca 100644 --- a/hive-agent/src/questions.rs +++ b/hive-agent/src/questions.rs @@ -19,12 +19,14 @@ //! `answer()` for `id` (see `hive-agent-mcp`'s tool impl). //! //! Known gap: if the asker cancels their own question, the target is not -//! 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 then lingers until they call `answer()` (now a "no -//! such question" error from c0re) or the row goes stale. Flagged on the -//! tracking issue rather than fixed here. +//! 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; @@ -141,11 +143,15 @@ impl Questions { } /// List every mirrored row, oldest-asked first — for `get_loose_ends` - /// rendering. + /// 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 failures. + /// Propagates the sqlite prepare / query / column-read failures. /// /// # Panics /// @@ -155,25 +161,27 @@ impl Questions { let mut stmt = conn.prepare( "SELECT id, role, peer, question, asked_at FROM questions ORDER BY asked_at ASC", )?; - let rows = stmt - .query_map([], row_to_question)? - .collect::>>()?; - Ok(rows) + 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; + }; + out.push(QuestionMirror { + id, + role, + peer: row.get(2)?, + question: row.get(3)?, + asked_at: row.get(4)?, + }); + } + Ok(out) } } -fn row_to_question(row: &rusqlite::Row) -> rusqlite::Result { - let role_str: String = row.get(1)?; - let role = Role::parse(&role_str).unwrap_or(Role::Asked); - Ok(QuestionMirror { - id: row.get(0)?, - role, - peer: row.get(2)?, - question: row.get(3)?, - asked_at: row.get(4)?, - }) -} - #[cfg(test)] mod tests { use super::*; diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index 3c2e4a99..835b8850 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -172,9 +172,9 @@ async fn handle_conn( /// Apply one request to the store, firing `wake` on a new/changed upsert so /// 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 #2679; 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 +/// pattern got reviewed out of the todo family in the diagnostic-logging +/// 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(