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) }