From 94712016981d184043569db89b119116fb48b63b Mon Sep 17 00:00:00 2001 From: damocles Date: Thu, 23 Jul 2026 20:55:54 +0200 Subject: [PATCH] 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" + ); + } +}