feat(#2635): add harness-local questions mirror store (inc2 pt2, unwired)

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;

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<Self> {
+        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<Connection>,
+}
+
+impl Questions {
+    /// Open (creating if needed) the questions mirror at `path`.
+    ///
+    /// # Errors
+    ///
+    /// Propagates sqlite open / schema-apply failures.
+    pub fn open(path: &Path) -> Result<Self> {
+        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<usize> {
+        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<Vec<QuestionMirror>> {
+        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::<rusqlite::Result<Vec<_>>>()?;
+        Ok(rows)
+    }
+}
+
+fn row_to_question(row: &rusqlite::Row) -> rusqlite::Result<QuestionMirror> {
+    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"
+        );
+    }
+}