From 6685b33c9d4080c135d66400d1834a84fa226bc1 Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 19 Jul 2026 01:15:17 +0200 Subject: [PATCH 1/4] feat(#2569): DB-backed per-agent todo store + mcp.sock upsert/clear/list/mark-done ops --- hive-agent-mcp/src/mcp/render.rs | 21 +++ hive-c0re/src/coordinator.rs | 8 + hive-c0re/src/loose_ends.rs | 27 +++ hive-c0re/src/main.rs | 1 + hive-c0re/src/socket_server/mod.rs | 106 +++++++++++ hive-c0re/src/stores/mod.rs | 1 + hive-c0re/src/stores/todos.rs | 292 +++++++++++++++++++++++++++++ hive-sh4re/src/lib.rs | 49 +++++ 8 files changed, 505 insertions(+) create mode 100644 hive-c0re/src/stores/todos.rs diff --git a/hive-agent-mcp/src/mcp/render.rs b/hive-agent-mcp/src/mcp/render.rs index db1a6eb2..28f55bca 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -228,6 +228,27 @@ pub(super) fn render_loose_ends(loose_ends: &[hive_sh4re::LooseEnd]) -> String { ); } } + hive_sh4re::LooseEnd::Todo { + id, + subsystem, + subsystem_key, + summary, + source, + age_seconds, + } => { + let key = subsystem_key + .as_deref() + .map(|k| format!(" {k}")) + .unwrap_or_default(); + let src = source + .as_deref() + .map(|s| format!(" — {s}")) + .unwrap_or_default(); + let _ = writeln!( + out, + "- todo #{id} [{subsystem}{key}, {age_seconds}s old]: {summary}{src} (mark_todo_done to clear)" + ); + } } } out diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 2e488e16..6a2b6a8d 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -16,6 +16,7 @@ use crate::container_view::{self, ContainerView}; use crate::dashboard_events::DashboardEvent; use crate::operator_questions::OperatorQuestions; use crate::socket_server::{self, AgentSocket}; +use crate::todos::Todos; /// Capacity of the dashboard event channel. Slow browser subscribers /// (idle tab, throttled connection) drop frames past this — that's @@ -32,6 +33,11 @@ pub struct Coordinator { pub broker: Arc, pub approvals: Arc, pub questions: Arc, + /// Dynamic, subsystem-pushed todos (loose-ends v2, #2569). In-agent + /// subsystems (matrix, forge, bash) upsert/clear todos over mcp.sock + /// instead of firing wakes directly; `get_todos` merges these with + /// the computed static loose ends. + pub todos: Arc, /// Scheduled-prompts queue. One sqlite connection, /// internal mutex; the worker drains due rows and the manager /// handlers insert / cancel through the same handle. @@ -460,6 +466,7 @@ impl Coordinator { let broker = Broker::open(db_path).context("open broker")?; let approvals = Approvals::open(db_path).context("open approvals")?; let questions = OperatorQuestions::open(db_path).context("open operator_questions")?; + let todos = Todos::open(db_path).context("open todos")?; let scheduled_prompts = crate::scheduled_prompts::ScheduledPrompts::open(db_path) .context("open scheduled_prompts")?; // BuildLogs wants a directory (it picks its own `build_logs.sqlite` @@ -489,6 +496,7 @@ impl Coordinator { broker: Arc::new(broker), approvals: Arc::new(approvals), questions: Arc::new(questions), + todos: Arc::new(todos), scheduled_prompts: Arc::new(scheduled_prompts), build_logs, audit_log, diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index 741d0789..bd0174c4 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -97,9 +97,36 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { age_seconds: saturating_age(now, r.created_at.timestamp()), }); } + // Dynamic, subsystem-pushed todos (loose-ends v2, #2569). Scoped to + // this agent; the producing subsystem or the agent itself clears them. + out.extend(todos_for(coord, agent, None)?); Ok(out) } +/// This agent's dynamic todos as `LooseEnd::Todo` rows, optionally +/// filtered to one `subsystem`. Shared by [`for_agent`] and the +/// `ListTodos` handler so the row-mapping lives in one place. +pub fn todos_for( + coord: &Coordinator, + agent: &str, + subsystem: Option<&str>, +) -> Result> { + let now = now_unix(); + Ok(coord + .todos + .list(agent, subsystem)? + .into_iter() + .map(|t| LooseEnd::Todo { + id: t.id, + subsystem: t.subsystem, + subsystem_key: t.subsystem_key, + summary: t.summary, + source: t.source, + age_seconds: saturating_age(now, t.updated_at.timestamp()), + }) + .collect()) +} + /// Hive-wide loose-ends view: EVERY pending approval + EVERY /// unanswered question + EVERY pending reminder. Manager surface /// only; sub-agents can't see each other's threads via the agent diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index a97aff7b..dd704ad4 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -43,6 +43,7 @@ pub(crate) use stats::{ }; pub(crate) use stores::{ approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts, + todos, }; pub(crate) use workers::{ agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, reminder_scheduler, diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index e23b411f..41a7ca95 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -575,6 +575,31 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> since_secs, agent: target, } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs), + // Todos (loose-ends v2, #2569): in-container subsystems push/clear + // their own; the agent lists / marks its own done. Scoped to the + // calling agent (the socket identity) — no cross-agent access. + AgentRequest::UpsertTodo { + subsystem, + key, + summary, + source, + } => handle_upsert_todo( + coord, + agent, + subsystem, + key.as_deref(), + summary, + source.as_deref(), + ), + AgentRequest::ClearTodo { + subsystem, + key, + all, + } => handle_clear_todo(coord, agent, subsystem, key.as_deref(), *all), + AgentRequest::ListTodos { subsystem } => { + handle_list_todos(coord, agent, subsystem.as_deref()) + } + AgentRequest::MarkTodoDone { id } => handle_mark_todo_done(coord, agent, *id), // Orchestration / diagnostics verbs — gated per-verb on tool-group // membership or topology (see `dispatch_orchestration`). _ => dispatch_orchestration(req, agent, coord).await, @@ -777,6 +802,87 @@ fn handle_get_loose_ends( } } +/// `UpsertTodo` — a subsystem pushes/updates one of this agent's todos. +/// Coalesces a wake ONLY when the row is new or actually changed, so +/// re-pushing an identical keyed todo is a silent no-op. +fn handle_upsert_todo( + coord: &Arc, + agent: &str, + subsystem: &str, + key: Option<&str>, + summary: &str, + source: Option<&str>, +) -> AgentResponse { + match coord.todos.upsert(agent, subsystem, key, summary, source) { + Ok((_, changed)) => { + if changed { + let _ = coord.broker.send(&Message { + from: "todo".to_owned(), + to: agent.to_owned(), + body: "you have todos — call get_todos".to_owned(), + in_reply_to: None, + }); + } + AgentResponse::Ok + } + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, + } +} + +/// `ClearTodo` — a producer clears a resolved todo by `(subsystem, key)`, +/// or wipes its whole set when `all` (cancel-and-recreate on restart). +fn handle_clear_todo( + coord: &Arc, + agent: &str, + subsystem: &str, + key: Option<&str>, + all: bool, +) -> AgentResponse { + let result = if all { + coord.todos.clear_subsystem(agent, subsystem) + } else { + coord.todos.clear(agent, subsystem, key) + }; + match result { + Ok(count) => AgentResponse::Acked { + count: u64::try_from(count).unwrap_or(0), + }, + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, + } +} + +/// `ListTodos` — enumerate this agent's todos (optionally one subsystem's) +/// as `LooseEnd::Todo` rows, so a producer can reconcile its own set. +fn handle_list_todos( + coord: &Arc, + agent: &str, + subsystem: Option<&str>, +) -> AgentResponse { + match crate::loose_ends::todos_for(coord, agent, subsystem) { + Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends }, + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, + } +} + +/// `MarkTodoDone` — the agent clears one of its own todos by id (scoped to +/// the agent, so it can't touch another agent's). +fn handle_mark_todo_done(coord: &Arc, agent: &str, id: i64) -> AgentResponse { + match coord.todos.mark_done(agent, id) { + Ok(count) => AgentResponse::Acked { + count: u64::try_from(count).unwrap_or(0), + }, + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, + } +} + /// `CountPendingReminders` — resolve the target (own / subtree free, else /// `QueryAgentState`) then count its pending reminders. fn handle_count_pending_reminders( diff --git a/hive-c0re/src/stores/mod.rs b/hive-c0re/src/stores/mod.rs index 1c02820c..45e676c6 100644 --- a/hive-c0re/src/stores/mod.rs +++ b/hive-c0re/src/stores/mod.rs @@ -12,3 +12,4 @@ pub mod db; pub mod operator_questions; pub mod power; pub mod scheduled_prompts; +pub mod todos; diff --git a/hive-c0re/src/stores/todos.rs b/hive-c0re/src/stores/todos.rs new file mode 100644 index 00000000..718c2870 --- /dev/null +++ b/hive-c0re/src/stores/todos.rs @@ -0,0 +1,292 @@ +//! Todo store — the persistent, DB-backed half of the "todos" +//! (loose-ends v2) system (issue #2569). +//! +//! Subsystems inside an agent's container (matrix, forge-notify, bash, +//! …) push *todos* to the agent over the mcp.sock protocol instead of +//! firing wakes directly. Todos are scoped to the owning `agent` (the +//! socket identity of the pushing container) and tagged with a +//! `subsystem` marker plus an optional `subsystem_key` (a matrix room +//! id, a bash task id, …); together `(agent, subsystem, subsystem_key)` +//! is the upsert/dedup key, so re-pushing the same item is idempotent +//! (no duplicate) and a producer can list / clear / rebuild only its own +//! set (e.g. matrix wipes + recreates its todos on daemon restart). +//! +//! Removal has two paths (mara's call on #2569): the producing subsystem +//! `clear`s a todo it has resolved (keyed by subsystem + key), or the +//! agent itself `mark_done`s one by id. Both delete the row. +//! +//! This table holds only the *dynamic* subsystem-pushed todos. The +//! static ones (pending approvals / questions / reminders / undelivered +//! messages) are still computed on demand in `loose_ends.rs`; `get_todos` +//! merges the two. Folding the static kinds into this table is a later +//! increment. + +use std::path::Path; +use std::sync::Mutex; + +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; +use hive_sh4re::wire_time::now_unix; +use rusqlite::{Connection, params}; +use serde::Serialize; + +use crate::db::Migration; + +const SCHEMA: &str = r" +CREATE TABLE IF NOT EXISTS todos ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + agent TEXT NOT NULL, + subsystem TEXT NOT NULL, + subsystem_key TEXT, + summary TEXT NOT NULL, + source TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); +-- (agent, subsystem, subsystem_key) is the upsert/dedup key. A NULL key +-- never conflicts (SQLite treats NULLs as distinct), so keyless todos +-- always insert as one-offs; keyed todos update in place. +CREATE UNIQUE INDEX IF NOT EXISTS idx_todos_dedup + ON todos (agent, subsystem, subsystem_key); +"; + +// New table — no legacy rows to migrate past the initial schema. +const MIGRATIONS: &[Migration] = &[]; + +/// One dynamic, subsystem-pushed todo. +#[derive(Debug, Clone, Serialize)] +pub struct Todo { + pub id: i64, + /// Owning agent (the container whose subsystem pushed it). + pub agent: String, + /// Producing subsystem marker (`"matrix"`, `"forge"`, `"bash"`, …). + pub subsystem: String, + /// Optional subsystem-specific dedup key (matrix room id, bash task + /// id, …). `None` = a keyless one-off todo. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub subsystem_key: Option, + /// Human-readable one-line summary shown to the agent. + pub summary: String, + /// Optional free-text provenance (e.g. the room name / task label). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source: Option, + pub created_at: DateTime, + pub updated_at: DateTime, +} + +pub struct Todos { + conn: Mutex, +} + +impl Todos { + pub fn open(path: &Path) -> Result { + let conn = crate::db::open(path, "todos")?; + conn.execute_batch(SCHEMA).context("apply todos schema")?; + crate::db::apply_versioned_migrations(&conn, "todos", MIGRATIONS)?; + Ok(Self { + conn: Mutex::new(conn), + }) + } + + /// Insert a todo for `agent`, or update the existing one for + /// `(agent, subsystem, key)` when `key` is `Some` and already + /// present. A `None` key never conflicts, so it always inserts a + /// fresh row. + /// + /// Returns `(id, changed)` where `changed` is `true` when the row is + /// new OR its `summary`/`source` actually differed — the caller uses + /// this to decide whether to coalesce a wake (re-pushing an identical + /// keyed todo is a no-op and must not re-wake). + pub fn upsert( + &self, + agent: &str, + subsystem: &str, + key: Option<&str>, + summary: &str, + source: Option<&str>, + ) -> Result<(i64, bool)> { + let conn = self.conn.lock().unwrap(); + let now = now_unix(); + let existing: Option<(i64, String, Option)> = if key.is_some() { + conn.query_row( + "SELECT id, summary, source FROM todos \ + WHERE agent = ?1 AND subsystem = ?2 AND subsystem_key IS ?3", + params![agent, subsystem, key], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .ok() + } else { + None + }; + if let Some((id, cur_summary, cur_source)) = existing { + let unchanged = cur_summary == summary && cur_source.as_deref() == source; + if unchanged { + return Ok((id, false)); + } + conn.execute( + "UPDATE todos SET summary = ?1, source = ?2, updated_at = ?3 WHERE id = ?4", + params![summary, source, now, id], + )?; + return Ok((id, true)); + } + conn.execute( + "INSERT INTO todos \ + (agent, subsystem, subsystem_key, summary, source, created_at, updated_at) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?6)", + params![agent, subsystem, key, summary, source, now], + )?; + Ok((conn.last_insert_rowid(), true)) + } + + /// Clear a producer-resolved todo, keyed by `(agent, subsystem, key)`. + /// Returns the number of rows deleted (0 when nothing matched). + pub fn clear(&self, agent: &str, subsystem: &str, key: Option<&str>) -> Result { + let conn = self.conn.lock().unwrap(); + let n = conn.execute( + "DELETE FROM todos WHERE agent = ?1 AND subsystem = ?2 AND subsystem_key IS ?3", + params![agent, subsystem, key], + )?; + Ok(n) + } + + /// Clear every todo `agent`'s `subsystem` owns — used by a producer + /// that rebuilds its whole set on restart (cancel-and-recreate). + /// Returns the number of rows deleted. + pub fn clear_subsystem(&self, agent: &str, subsystem: &str) -> Result { + let conn = self.conn.lock().unwrap(); + let n = conn.execute( + "DELETE FROM todos WHERE agent = ?1 AND subsystem = ?2", + params![agent, subsystem], + )?; + Ok(n) + } + + /// The agent marks one of *its own* todos done, by id. Scoped to + /// `agent` so one agent can't clear another's. Returns the number of + /// rows deleted (0 when the id was unknown / not owned / already gone). + pub fn mark_done(&self, agent: &str, id: i64) -> Result { + let conn = self.conn.lock().unwrap(); + let n = conn.execute( + "DELETE FROM todos WHERE id = ?1 AND agent = ?2", + params![id, agent], + )?; + Ok(n) + } + + /// List `agent`'s todos, newest-updated first. `subsystem = Some(..)` + /// filters to one producer's set (so a producer can enumerate + + /// reconcile only its own); `None` returns all of the agent's. + pub fn list(&self, agent: &str, subsystem: Option<&str>) -> Result> { + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare( + "SELECT id, agent, subsystem, subsystem_key, summary, source, created_at, updated_at \ + FROM todos \ + WHERE agent = ?1 AND (?2 IS NULL OR subsystem = ?2) \ + ORDER BY updated_at DESC, id DESC", + )?; + let rows = stmt + .query_map(params![agent, subsystem], |row| { + let created: i64 = row.get(6)?; + let updated: i64 = row.get(7)?; + Ok(Todo { + id: row.get(0)?, + agent: row.get(1)?, + subsystem: row.get(2)?, + subsystem_key: row.get(3)?, + summary: row.get(4)?, + source: row.get(5)?, + created_at: DateTime::from_timestamp(created, 0).unwrap_or_default(), + updated_at: DateTime::from_timestamp(updated, 0).unwrap_or_default(), + }) + })? + .collect::>>()?; + Ok(rows) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + // Return the `TempDir` alongside the store so it outlives the test — + // dropping it early deletes the dir and SQLite fails with + // `SQLITE_READONLY_DBMOVED`. + fn store() -> (tempfile::TempDir, Todos) { + let dir = tempfile::tempdir().unwrap(); + let db = Todos::open(&dir.path().join("todos.sqlite")).unwrap(); + (dir, db) + } + + #[test] + fn keyed_upsert_dedups_and_reports_changed() { + let (_dir, s) = store(); + let (id1, changed1) = s + .upsert("alice", "matrix", Some("!room:x"), "1 unread", None) + .unwrap(); + assert!(changed1, "first push is new → changed"); + // Same key + same summary → no-op, not changed (must not re-wake). + let (id2, changed2) = s + .upsert("alice", "matrix", Some("!room:x"), "1 unread", None) + .unwrap(); + assert_eq!(id1, id2, "keyed upsert updates in place, same row"); + assert!(!changed2, "identical re-push is a no-op"); + // Same key, new summary → updates, changed. + let (id3, changed3) = s + .upsert("alice", "matrix", Some("!room:x"), "3 unread", None) + .unwrap(); + assert_eq!(id1, id3); + assert!(changed3); + assert_eq!(s.list("alice", Some("matrix")).unwrap().len(), 1); + } + + #[test] + fn todos_are_scoped_per_agent() { + let (_dir, s) = store(); + // Same subsystem+key for two agents → distinct rows. + s.upsert("alice", "matrix", Some("!r:x"), "u", None) + .unwrap(); + s.upsert("bob", "matrix", Some("!r:x"), "u", None).unwrap(); + assert_eq!(s.list("alice", None).unwrap().len(), 1); + assert_eq!(s.list("bob", None).unwrap().len(), 1); + // bob can't mark alice's todo done. + let alice_id = s.list("alice", None).unwrap()[0].id; + assert_eq!(s.mark_done("bob", alice_id).unwrap(), 0); + assert_eq!(s.mark_done("alice", alice_id).unwrap(), 1); + } + + #[test] + fn keyless_todos_always_insert() { + let (_dir, s) = store(); + let (a, _) = s.upsert("alice", "bash", None, "task done", None).unwrap(); + let (b, _) = s.upsert("alice", "bash", None, "task done", None).unwrap(); + assert_ne!(a, b, "keyless pushes are distinct one-offs"); + assert_eq!(s.list("alice", Some("bash")).unwrap().len(), 2); + } + + #[test] + fn clear_and_mark_done_remove_rows() { + let (_dir, s) = store(); + s.upsert("alice", "matrix", Some("!a:x"), "unread", None) + .unwrap(); + let (id, _) = s + .upsert("alice", "forge", Some("pr-1"), "review", None) + .unwrap(); + assert_eq!(s.clear("alice", "matrix", Some("!a:x")).unwrap(), 1); + assert_eq!(s.mark_done("alice", id).unwrap(), 1); + assert!(s.list("alice", None).unwrap().is_empty()); + } + + #[test] + fn clear_subsystem_wipes_only_its_own() { + let (_dir, s) = store(); + s.upsert("alice", "matrix", Some("!a:x"), "u", None) + .unwrap(); + s.upsert("alice", "matrix", Some("!b:x"), "u", None) + .unwrap(); + s.upsert("alice", "forge", Some("pr-1"), "r", None).unwrap(); + assert_eq!(s.clear_subsystem("alice", "matrix").unwrap(), 2); + let left = s.list("alice", None).unwrap(); + assert_eq!(left.len(), 1); + assert_eq!(left[0].subsystem, "forge"); + } +} diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index f373bfdb..837ee064 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -269,6 +269,23 @@ pub enum LooseEnd { #[serde(default)] summary: String, }, + /// A dynamic, subsystem-pushed todo (loose-ends v2, #2569). Produced + /// by an in-container subsystem (matrix / forge / bash) via + /// `UpsertTodo`. Cleared by that subsystem (`ClearTodo`) or by the + /// agent itself (`MarkTodoDone`, by `id`). + Todo { + id: i64, + /// Producing subsystem marker (`"matrix"`, `"forge"`, `"bash"`, …). + subsystem: String, + /// Optional subsystem-specific key (matrix room id, bash task id). + #[serde(default, skip_serializing_if = "Option::is_none")] + subsystem_key: Option, + summary: String, + /// Optional free-text provenance (room name / task label). + #[serde(default, skip_serializing_if = "Option::is_none")] + source: Option, + age_seconds: u64, + }, } /// Kind discriminator for `CancelLooseEnd`. Per-kind store + @@ -422,6 +439,38 @@ pub enum Request { #[serde(default, skip_serializing_if = "Option::is_none")] agent: Option, }, + /// Upsert a *todo* (loose-ends v2, #2569) from an in-container + /// subsystem (matrix / forge / bash). `subsystem` is the producer + /// marker; `key` is the optional subsystem-specific dedup key (a + /// matrix room id, a bash task id). Re-pushing an identical keyed + /// todo is a no-op; a new-or-changed one coalesces a wake to the + /// agent. Keyless todos always insert as one-offs. + UpsertTodo { + subsystem: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + key: Option, + summary: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + source: Option, + }, + /// Clear a producer-resolved todo by `(subsystem, key)`. `key = None` + /// targets the keyless one-off; `all = true` wipes the producer's + /// whole set (cancel-and-recreate on daemon restart). + ClearTodo { + subsystem: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + key: Option, + #[serde(default)] + all: bool, + }, + /// List todos, optionally filtered to one `subsystem` (a producer + /// enumerating its own set). `None` = all. + ListTodos { + #[serde(default, skip_serializing_if = "Option::is_none")] + subsystem: Option, + }, + /// The agent marks one of its own todos done, by id. + MarkTodoDone { id: i64 }, /// Count of pending (un-delivered) reminders. On the agent socket: /// same target rules as `GetLooseEnds` (self/children free; /// non-children require `query_agent_state`; `"*"` rejected). From 711e0ece2a137e05359953a2e0ec49def07b03ee Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 19 Jul 2026 01:40:32 +0200 Subject: [PATCH 2/4] =?UTF-8?q?feat(#2569):=20matrix=20producer=20?= =?UTF-8?q?=E2=80=94=20sweep=5Funread=20pushes=20per-room=20todos=20instea?= =?UTF-8?q?d=20of=20direct=20wakes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hive-matrix-mcp/src/handlers.rs | 30 +++++++-- hive-matrix-mcp/src/main.rs | 5 ++ hive-matrix-mcp/src/timeline.rs | 105 +++++++++++++------------------- hive-matrix-mcp/src/wake.rs | 60 ++++++++++++++++-- 4 files changed, 127 insertions(+), 73 deletions(-) diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index c69bf547..9dd83b26 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -901,6 +901,21 @@ pub fn unread_count(client: &Client) -> DaemonResponse { /// mind if latency becomes a concern. #[must_use] pub async fn collect_unread(client: &Client) -> Vec { + collect_unread_with_ids(client) + .await + .into_iter() + .map(|(_, ru)| ru) + .collect() +} + +/// Like [`collect_unread`] but pairs each entry with its `OwnedRoomId`. +/// The todo producer (loose-ends v2, #2569) needs the room id as the +/// per-room upsert/dedup key, which the claude-facing `RoomUnread` +/// payload intentionally doesn't carry. +#[must_use] +pub async fn collect_unread_with_ids( + client: &Client, +) -> Vec<(matrix_sdk::ruma::OwnedRoomId, crate::protocol::RoomUnread)> { use crate::protocol::RoomUnread; let mut result = Vec::new(); for room in client.joined_rooms() { @@ -915,12 +930,15 @@ pub async fn collect_unread(client: &Client) -> Vec } else { (None, None) }; - result.push(RoomUnread { - label, - count, - last_body, - last_sender, - }); + result.push(( + room.room_id().to_owned(), + RoomUnread { + label, + count, + last_body, + last_sender, + }, + )); } result } diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index 49773875..bfba0573 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -299,6 +299,11 @@ async fn bring_up_account( // each sweep (see the sweep fns) so re-invites / new messages re-wake. let invite_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); let unread_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); + // Startup cancel-and-recreate (loose-ends v2, #2569): wipe this agent's + // matrix todos so stale ones (rooms read while the daemon was down) don't + // linger, then let the first sweep rebuild the set to match current + // unread reality. Best-effort; the sweep converges regardless. + let _ = crate::wake::send_todo_clear(hyperhive_socket, None, true).await; let sync_loop: SyncLoop = Box::pin(async move { sync_client .sync_with_callback(SyncSettings::default(), move |_response| { diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index 9947866c..79ac2ad9 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -20,23 +20,19 @@ use tokio::sync::Mutex; use crate::{handlers, wake}; -/// Wake the agent for any joined room carrying unread notifications it has -/// not yet been successfully woken about. Driven from the post-sync -/// callback — the same reliable path `sweep_invites` uses — rather than a -/// one-shot `m.room.message` event handler: a message-wake whose -/// `send_wake` raced a hive-c0re / socket-down window (a container rebuild) -/// was dropped with no retry, so the agent went deaf until manually -/// prompted. The sync callback fires on every sync response (so promptly on -/// new activity), so this re-checks unread each tick and retries a dropped -/// wake on the next one — self-healing once the socket is back. +/// Push a *todo* (loose-ends v2, #2569) for each joined room carrying +/// unread notifications, and clear the todo for rooms that have been read. +/// Replaces the old direct-wake path: instead of firing an all-rooms wake, +/// each unread room becomes a per-room `upsert_todo` keyed by its room id, +/// and hive-c0re coalesces exactly one wake when the todo set changes. /// -/// `notified` dedups so a standing unread wakes the agent once, not on -/// every sync tick; it is pruned to the current unread set each pass so a -/// room that has been read and then receives a new message wakes again. On -/// a wake-send failure the freshly-seen rooms are rolled back out of -/// `notified` so the next pass retries them. `account_tag` is `Some(name)` -/// only in multi-account mode, prepended so the agent knows which account -/// to `read_room` on. +/// Driven from the post-sync callback — the same reliable path +/// `sweep_invites` uses — so it re-checks unread each tick; a dropped +/// upsert is retried on the next one (self-healing once the socket is +/// back). `notified` tracks which rooms currently have an active todo so +/// a room that becomes read gets its todo cleared. `account_tag` is +/// `Some(name)` only in multi-account mode, prepended so the agent knows +/// which account to `read_room` on. /// /// Self-sent messages don't raise an unread notification server-side, so /// no explicit self-filter is needed here. @@ -46,55 +42,42 @@ pub async fn sweep_unread( notified: &Mutex>, account_tag: Option<&str>, ) { - // Current set of joined rooms carrying unread notifications. - let unread_ids: HashSet = client - .joined_rooms() - .into_iter() - .filter(|r| r.unread_notification_counts().notification_count > 0) - .map(|r| r.room_id().to_owned()) - .collect(); + let unread = handlers::collect_unread_with_ids(client).await; + let unread_ids: HashSet = unread.iter().map(|(id, _)| id.clone()).collect(); - // Which unread rooms are newly-seen (not already successfully woken)? - let mut fresh = Vec::new(); - { - let mut seen = notified.lock().await; - // Drop rooms no longer unread (read / left) so a future new message - // in them wakes the agent again. - seen.retain(|id| unread_ids.contains(id)); - for id in &unread_ids { - if seen.insert(id.clone()) { - fresh.push(id.clone()); + // Rooms we previously pushed a todo for that are no longer unread → the + // agent read them; clear their todo. Snapshot under the lock, send + // outside it, drop from `notified` on success (retry next tick on fail). + let stale: Vec = { + let active = notified.lock().await; + active.difference(&unread_ids).cloned().collect() + }; + for id in stale { + if wake::send_todo_clear(socket, Some(id.as_str()), false) + .await + .is_ok() + { + notified.lock().await.remove(&id); + } + } + + // Upsert a todo per currently-unread room. hive-c0re coalesces the wake + // iff the summary is new or changed, so re-upserting an unchanged room + // every sync tick is a cheap server-side no-op (no re-wake). + for (id, ru) in &unread { + let summary = wake::tag_account( + account_tag, + wake::format_unread_summary(std::slice::from_ref(ru)), + ); + match wake::send_todo_upsert(socket, id.as_str(), &summary).await { + Ok(()) => { + notified.lock().await.insert(id.clone()); + } + Err(e) => { + tracing::warn!(error = %e, room = %id, "matrix: todo upsert failed; will retry next sync"); } } } - if fresh.is_empty() { - return; - } - - // Roll the freshly-seen rooms back out of `notified` so the next sweep - // retries them. Used on both the sync-race (empty collect) and the - // wake-send-failure paths. - let rollback = || async { - let mut seen = notified.lock().await; - for id in &fresh { - seen.remove(id); - } - }; - - // Same all-rooms unread summary the wake body has always used. Empty - // only under a sync race (counts changed between the scan and collect). - let unread = handlers::collect_unread(client).await; - if unread.is_empty() { - rollback().await; - return; - } - let body = wake::tag_account(account_tag, wake::format_unread_summary(&unread)); - if let Err(e) = wake::send_wake(socket, &body).await { - rollback().await; - tracing::warn!(error = %e, "matrix: unread-sweep wake failed; will retry next sync"); - } else { - tracing::info!(rooms = fresh.len(), "matrix: unread-sweep wake delivered"); - } } /// Wake the agent for any pending room invite it has not yet been diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index 9fe371c7..9e041a98 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -34,15 +34,65 @@ use tokio::net::UnixStream; /// Returns an error on socket connect failure, serialisation failure, /// or I/O error writing to or reading from the socket. pub async fn send_wake(socket: &Path, body: impl AsRef) -> Result<()> { - use tokio::io::AsyncBufReadExt; - let payload = serde_json::json!({ "cmd": "wake", "from": "matrix", "body": body.as_ref(), "transient": true, }); - let line = format!("{}\n", serde_json::to_string(&payload)?); + send_line(socket, &payload).await +} + +/// Upsert a matrix-subsystem *todo* (loose-ends v2, #2569) on the +/// hyperhive control socket — the replacement for a direct wake. `key` is +/// the room id (the dedup key); hive-c0re coalesces a wake iff the todo is +/// new or its `summary` changed. Best-effort like [`send_wake`]. +/// +/// # Errors +/// +/// Returns an error on socket connect failure, serialisation failure, +/// or I/O error writing to or reading from the socket. +pub async fn send_todo_upsert(socket: &Path, key: &str, summary: impl AsRef) -> Result<()> { + let payload = serde_json::json!({ + "cmd": "upsert_todo", + "subsystem": "matrix", + "key": key, + "summary": summary.as_ref(), + }); + send_line(socket, &payload).await +} + +/// Clear matrix-subsystem todos. `key = Some(room)` clears one room's +/// todo (it was read); `all = true` wipes the whole matrix set +/// (cancel-and-recreate on daemon restart). Best-effort. +/// +/// # Errors +/// +/// Returns an error on socket connect failure, serialisation failure, +/// or I/O error writing to or reading from the socket. +pub async fn send_todo_clear(socket: &Path, key: Option<&str>, all: bool) -> Result<()> { + let payload = serde_json::json!({ + "cmd": "clear_todo", + "subsystem": "matrix", + "key": key, + "all": all, + }); + send_line(socket, &payload).await +} + +/// Write one JSON request line to the hyperhive control socket and drain +/// the response line (best-effort — the reply is not acted on, we just +/// read it so the server doesn't get ECONNRESET on its write-back). +/// Shared by [`send_wake`] and the todo senders. +/// +/// # Errors +/// +/// Returns an error on socket connect failure, serialisation failure, +/// or I/O error writing to or reading from the socket. +async fn send_line(socket: &Path, payload: &serde_json::Value) -> Result<()> { + use tokio::io::AsyncBufReadExt; + + let line = format!("{}\n", serde_json::to_string(payload)?); let stream = UnixStream::connect(socket) .await .with_context(|| format!("connect hyperhive socket {}", socket.display()))?; @@ -50,13 +100,11 @@ pub async fn send_wake(socket: &Path, body: impl AsRef) -> Result<()> { write .write_all(line.as_bytes()) .await - .with_context(|| format!("write wake to {}", socket.display()))?; + .with_context(|| format!("write to {}", socket.display()))?; write .shutdown() .await .with_context(|| format!("shutdown write to {}", socket.display()))?; - // Drain the response line so the server doesn't get ECONNRESET on - // its write-back. We don't act on the response — best-effort wake. let mut reader = tokio::io::BufReader::new(read); let mut resp = String::new(); let _ = reader.read_line(&mut resp).await; From 8149dc7633fe2a9b2e5d2a174370006c081c7987 Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 19 Jul 2026 01:41:23 +0200 Subject: [PATCH 3/4] fix(#2569): wake body points at get_loose_ends (get_todos rename is a later increment); strip tracker tags from source comments per hive-rules --- hive-c0re/src/coordinator.rs | 2 +- hive-c0re/src/loose_ends.rs | 2 +- hive-c0re/src/socket_server/mod.rs | 4 ++-- hive-c0re/src/stores/todos.rs | 4 ++-- hive-matrix-mcp/src/handlers.rs | 2 +- hive-matrix-mcp/src/main.rs | 2 +- hive-matrix-mcp/src/timeline.rs | 2 +- hive-matrix-mcp/src/wake.rs | 2 +- hive-sh4re/src/lib.rs | 4 ++-- 9 files changed, 12 insertions(+), 12 deletions(-) diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 6a2b6a8d..88390fed 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -33,7 +33,7 @@ pub struct Coordinator { pub broker: Arc, pub approvals: Arc, pub questions: Arc, - /// Dynamic, subsystem-pushed todos (loose-ends v2, #2569). In-agent + /// Dynamic, subsystem-pushed todos (loose-ends v2). In-agent /// subsystems (matrix, forge, bash) upsert/clear todos over mcp.sock /// instead of firing wakes directly; `get_todos` merges these with /// the computed static loose ends. diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index bd0174c4..78e8689d 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -97,7 +97,7 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { age_seconds: saturating_age(now, r.created_at.timestamp()), }); } - // Dynamic, subsystem-pushed todos (loose-ends v2, #2569). Scoped to + // Dynamic, subsystem-pushed todos (loose-ends v2). Scoped to // this agent; the producing subsystem or the agent itself clears them. out.extend(todos_for(coord, agent, None)?); Ok(out) diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index 41a7ca95..3d3f2cba 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -575,7 +575,7 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> since_secs, agent: target, } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs), - // Todos (loose-ends v2, #2569): in-container subsystems push/clear + // Todos (loose-ends v2): in-container subsystems push/clear // their own; the agent lists / marks its own done. Scoped to the // calling agent (the socket identity) — no cross-agent access. AgentRequest::UpsertTodo { @@ -819,7 +819,7 @@ fn handle_upsert_todo( let _ = coord.broker.send(&Message { from: "todo".to_owned(), to: agent.to_owned(), - body: "you have todos — call get_todos".to_owned(), + body: "you have todos — call get_loose_ends to see them".to_owned(), in_reply_to: None, }); } diff --git a/hive-c0re/src/stores/todos.rs b/hive-c0re/src/stores/todos.rs index 718c2870..e8038451 100644 --- a/hive-c0re/src/stores/todos.rs +++ b/hive-c0re/src/stores/todos.rs @@ -1,5 +1,5 @@ //! Todo store — the persistent, DB-backed half of the "todos" -//! (loose-ends v2) system (issue #2569). +//! (loose-ends v2) system. //! //! Subsystems inside an agent's container (matrix, forge-notify, bash, //! …) push *todos* to the agent over the mcp.sock protocol instead of @@ -11,7 +11,7 @@ //! (no duplicate) and a producer can list / clear / rebuild only its own //! set (e.g. matrix wipes + recreates its todos on daemon restart). //! -//! Removal has two paths (mara's call on #2569): the producing subsystem +//! Removal has two paths (mara's call): the producing subsystem //! `clear`s a todo it has resolved (keyed by subsystem + key), or the //! agent itself `mark_done`s one by id. Both delete the row. //! diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 9dd83b26..7c4d906e 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -909,7 +909,7 @@ pub async fn collect_unread(client: &Client) -> Vec } /// Like [`collect_unread`] but pairs each entry with its `OwnedRoomId`. -/// The todo producer (loose-ends v2, #2569) needs the room id as the +/// The todo producer (loose-ends v2) needs the room id as the /// per-room upsert/dedup key, which the claude-facing `RoomUnread` /// payload intentionally doesn't carry. #[must_use] diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index bfba0573..42d768dd 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -299,7 +299,7 @@ async fn bring_up_account( // each sweep (see the sweep fns) so re-invites / new messages re-wake. let invite_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); let unread_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); - // Startup cancel-and-recreate (loose-ends v2, #2569): wipe this agent's + // Startup cancel-and-recreate (loose-ends v2): wipe this agent's // matrix todos so stale ones (rooms read while the daemon was down) don't // linger, then let the first sweep rebuild the set to match current // unread reality. Best-effort; the sweep converges regardless. diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index 79ac2ad9..50ebc082 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -20,7 +20,7 @@ use tokio::sync::Mutex; use crate::{handlers, wake}; -/// Push a *todo* (loose-ends v2, #2569) for each joined room carrying +/// Push a *todo* (loose-ends v2) for each joined room carrying /// unread notifications, and clear the todo for rooms that have been read. /// Replaces the old direct-wake path: instead of firing an all-rooms wake, /// each unread room becomes a per-room `upsert_todo` keyed by its room id, diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index 9e041a98..93baf9d6 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -43,7 +43,7 @@ pub async fn send_wake(socket: &Path, body: impl AsRef) -> Result<()> { send_line(socket, &payload).await } -/// Upsert a matrix-subsystem *todo* (loose-ends v2, #2569) on the +/// Upsert a matrix-subsystem *todo* (loose-ends v2) on the /// hyperhive control socket — the replacement for a direct wake. `key` is /// the room id (the dedup key); hive-c0re coalesces a wake iff the todo is /// new or its `summary` changed. Best-effort like [`send_wake`]. diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 837ee064..439d2096 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -269,7 +269,7 @@ pub enum LooseEnd { #[serde(default)] summary: String, }, - /// A dynamic, subsystem-pushed todo (loose-ends v2, #2569). Produced + /// A dynamic, subsystem-pushed todo (loose-ends v2). Produced /// by an in-container subsystem (matrix / forge / bash) via /// `UpsertTodo`. Cleared by that subsystem (`ClearTodo`) or by the /// agent itself (`MarkTodoDone`, by `id`). @@ -439,7 +439,7 @@ pub enum Request { #[serde(default, skip_serializing_if = "Option::is_none")] agent: Option, }, - /// Upsert a *todo* (loose-ends v2, #2569) from an in-container + /// Upsert a *todo* (loose-ends v2) from an in-container /// subsystem (matrix / forge / bash). `subsystem` is the producer /// marker; `key` is the optional subsystem-specific dedup key (a /// matrix room id, a bash task id). Re-pushing an identical keyed From 565b1b90fc00e284ae92a5012b1d2c78b920833c Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 19 Jul 2026 02:03:44 +0200 Subject: [PATCH 4/4] =?UTF-8?q?docs(#2569):=20address=20argus=20review=20?= =?UTF-8?q?=E2=80=94=20add=20Errors/Panics=20doc=20sections=20+=20fix=20Cl?= =?UTF-8?q?earTodo=20keyless-clear=20semantics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hive-c0re/src/loose_ends.rs | 4 +++ hive-c0re/src/stores/todos.rs | 53 +++++++++++++++++++++++++++++++++-- hive-sh4re/src/lib.rs | 9 ++++-- 3 files changed, 61 insertions(+), 5 deletions(-) diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index 78e8689d..b4803b37 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -106,6 +106,10 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { /// This agent's dynamic todos as `LooseEnd::Todo` rows, optionally /// filtered to one `subsystem`. Shared by [`for_agent`] and the /// `ListTodos` handler so the row-mapping lives in one place. +/// +/// # Errors +/// +/// Propagates the todo-store query failure. pub fn todos_for( coord: &Coordinator, agent: &str, diff --git a/hive-c0re/src/stores/todos.rs b/hive-c0re/src/stores/todos.rs index e8038451..8effdd65 100644 --- a/hive-c0re/src/stores/todos.rs +++ b/hive-c0re/src/stores/todos.rs @@ -79,6 +79,11 @@ pub struct Todos { } impl Todos { + /// Open (creating if needed) the todo store at `path`. + /// + /// # Errors + /// + /// Propagates sqlite open / schema-apply / migration failures. pub fn open(path: &Path) -> Result { let conn = crate::db::open(path, "todos")?; conn.execute_batch(SCHEMA).context("apply todos schema")?; @@ -97,6 +102,14 @@ impl Todos { /// new OR its `summary`/`source` actually differed — the caller uses /// this to decide whether to coalesce a wake (re-pushing an identical /// keyed todo is a no-op and must not re-wake). + /// + /// # Errors + /// + /// Propagates sqlite query / execute failures. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. pub fn upsert( &self, agent: &str, @@ -138,8 +151,20 @@ impl Todos { Ok((conn.last_insert_rowid(), true)) } - /// Clear a producer-resolved todo, keyed by `(agent, subsystem, key)`. - /// Returns the number of rows deleted (0 when nothing matched). + /// Clear producer-resolved todo(s) by `(agent, subsystem, key)`. + /// `key = Some(k)` targets the one keyed row; `key = None` matches + /// `subsystem_key IS NULL`, i.e. **all** keyless todos for that + /// subsystem (keyless rows have no distinguishing key — clear a + /// specific one via [`Todos::mark_done`] by id instead). Returns the + /// number of rows deleted (0 when nothing matched). + /// + /// # Errors + /// + /// Propagates the sqlite delete failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. pub fn clear(&self, agent: &str, subsystem: &str, key: Option<&str>) -> Result { let conn = self.conn.lock().unwrap(); let n = conn.execute( @@ -152,6 +177,14 @@ impl Todos { /// Clear every todo `agent`'s `subsystem` owns — used by a producer /// that rebuilds its whole set on restart (cancel-and-recreate). /// Returns the number of rows deleted. + /// + /// # Errors + /// + /// Propagates the sqlite delete failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. pub fn clear_subsystem(&self, agent: &str, subsystem: &str) -> Result { let conn = self.conn.lock().unwrap(); let n = conn.execute( @@ -164,6 +197,14 @@ impl Todos { /// The agent marks one of *its own* todos done, by id. Scoped to /// `agent` so one agent can't clear another's. Returns the number of /// rows deleted (0 when the id was unknown / not owned / already gone). + /// + /// # Errors + /// + /// Propagates the sqlite delete failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. pub fn mark_done(&self, agent: &str, id: i64) -> Result { let conn = self.conn.lock().unwrap(); let n = conn.execute( @@ -176,6 +217,14 @@ impl Todos { /// List `agent`'s todos, newest-updated first. `subsystem = Some(..)` /// filters to one producer's set (so a producer can enumerate + /// reconcile only its own); `None` returns all of the agent's. + /// + /// # Errors + /// + /// Propagates the sqlite prepare / query failures. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. pub fn list(&self, agent: &str, subsystem: Option<&str>) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 439d2096..73180185 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -453,9 +453,12 @@ pub enum Request { #[serde(default, skip_serializing_if = "Option::is_none")] source: Option, }, - /// Clear a producer-resolved todo by `(subsystem, key)`. `key = None` - /// targets the keyless one-off; `all = true` wipes the producer's - /// whole set (cancel-and-recreate on daemon restart). + /// Clear producer-resolved todo(s) by `(subsystem, key)`. `key = + /// Some(k)` clears the one keyed row; `key = None` clears **all** of + /// the subsystem's keyless todos (rows with no key can't be told + /// apart — clear a specific one via `MarkTodoDone` by id). `all = + /// true` wipes the producer's whole set (cancel-and-recreate on + /// daemon restart). ClearTodo { subsystem: String, #[serde(default, skip_serializing_if = "Option::is_none")]