From 0977006ec60675441841963f04b4cd183049f9a4 Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 20 Jul 2026 23:15:14 +0200 Subject: [PATCH] refactor(#2569): remove the c0re todo store + handlers (todos now owned in-container) --- hive-c0re/src/coordinator.rs | 8 - hive-c0re/src/loose_ends.rs | 31 --- hive-c0re/src/main.rs | 1 - hive-c0re/src/socket_server/mod.rs | 100 --------- hive-c0re/src/stores/mod.rs | 1 - hive-c0re/src/stores/todos.rs | 341 ----------------------------- hive-core-agent-sock/src/lib.rs | 35 --- 7 files changed, 517 deletions(-) delete mode 100644 hive-c0re/src/stores/todos.rs diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 707ac73c..9d40377d 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -16,7 +16,6 @@ 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 @@ -33,11 +32,6 @@ pub struct Coordinator { pub broker: Arc, pub approvals: Arc, pub questions: Arc, - /// 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. - 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. @@ -466,7 +460,6 @@ 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` @@ -496,7 +489,6 @@ 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 72a9b0ee..b09228f8 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -97,40 +97,9 @@ 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). 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. -/// -/// # Errors -/// -/// Propagates the todo-store query failure. -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 dd704ad4..a97aff7b 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -43,7 +43,6 @@ 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 99be0eb2..8f14dfd5 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -596,29 +596,6 @@ async fn dispatch(req: &Request, agent: &str, coord: &Arc) -> Respo since_secs, agent: target, } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs), - // 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. - Request::UpsertTodo { - subsystem, - key, - summary, - source, - } => handle_upsert_todo( - coord, - agent, - subsystem, - key.as_deref(), - summary, - source.as_deref(), - ), - Request::ClearTodo { - subsystem, - key, - all, - } => handle_clear_todo(coord, agent, subsystem, key.as_deref(), *all), - Request::ListTodos { subsystem } => handle_list_todos(coord, agent, subsystem.as_deref()), - Request::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, @@ -813,83 +790,6 @@ fn handle_get_loose_ends(coord: &Arc, agent: &str, target: Option<& } } -/// `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>, -) -> Response { - 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_loose_ends to see them".to_owned(), - in_reply_to: None, - }); - } - Response::Ok - } - Err(e) => Response::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, -) -> Response { - let result = if all { - coord.todos.clear_subsystem(agent, subsystem) - } else { - coord.todos.clear(agent, subsystem, key) - }; - match result { - Ok(count) => Response::Acked { - count: u64::try_from(count).unwrap_or(0), - }, - Err(e) => Response::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>) -> Response { - match crate::loose_ends::todos_for(coord, agent, subsystem) { - Ok(loose_ends) => Response::LooseEnds { loose_ends }, - Err(e) => Response::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) -> Response { - match coord.todos.mark_done(agent, id) { - Ok(count) => Response::Acked { - count: u64::try_from(count).unwrap_or(0), - }, - Err(e) => Response::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 45e676c6..1c02820c 100644 --- a/hive-c0re/src/stores/mod.rs +++ b/hive-c0re/src/stores/mod.rs @@ -12,4 +12,3 @@ 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 deleted file mode 100644 index 8effdd65..00000000 --- a/hive-c0re/src/stores/todos.rs +++ /dev/null @@ -1,341 +0,0 @@ -//! Todo store — the persistent, DB-backed half of the "todos" -//! (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 -//! 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): 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 { - /// 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")?; - 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). - /// - /// # Errors - /// - /// Propagates sqlite query / execute failures. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - 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 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( - "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. - /// - /// # 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( - "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). - /// - /// # 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( - "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. - /// - /// # 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( - "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-core-agent-sock/src/lib.rs b/hive-core-agent-sock/src/lib.rs index e3ef2093..8c4b58ba 100644 --- a/hive-core-agent-sock/src/lib.rs +++ b/hive-core-agent-sock/src/lib.rs @@ -102,41 +102,6 @@ pub enum Request { #[serde(default, skip_serializing_if = "Option::is_none")] agent: Option, }, - /// 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 - /// 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 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")] - 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).