//! Harness-local todo store — the persistent, DB-backed half of the //! "todos" (loose-ends v2) system, owned by the in-container harness. //! //! In-container subsystems (matrix, forge-notify, bash) push *todos* to //! the harness over the in-agent socket instead of firing wakes directly. //! The harness owns this store locally (one sqlite db under the harness //! dir) and signals its own turn loop on a new/changed row — hive-c0re is //! not involved (no broker round-trip, no marker files). //! //! Because the store lives inside a single agent's container, todos are //! **not** agent-scoped here (unlike the old c0re store): every row //! belongs to this agent. A todo is tagged with a `subsystem` marker plus //! an optional `subsystem_key` (a matrix room id, a bash task id, …), and //! `(subsystem, subsystem_key)` is the upsert/dedup key — re-pushing the //! same item is idempotent, and a producer can list / clear / rebuild only //! its own set (e.g. matrix wipes + recreates its todos on daemon restart). //! //! Removal/dismissal has two different paths. The **producing subsystem** //! `clear`s a todo it has itself resolved (keyed by subsystem + key) — a //! one-shot event has nothing left to represent once resolved, so this //! hard-deletes the row. The **agent** `mark_done`s one by id (via //! `cancel_loose_end`) — that only means "stop showing it to me", so it sets //! `acked` instead of deleting: a *reconciled* producer like `disk_watch` //! (re-derives its summary on a timer, not a one-shot event) needs the row //! to survive ack so [`Todos::upsert`] can still tell "unchanged" from //! "genuinely different". Acked rows stay out of [`Todos::list`] but keep //! anchoring that comparison; a materially different summary un-acks + //! wakes again. Acked rows are reaped after a retention window //! (`vacuum.rs`) so the table doesn't grow unbounded. use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use hive_sh4re::wire_time; use rusqlite::{Connection, params}; /// SQL bootstrap. `CREATE TABLE IF NOT EXISTS` so first-boot agents and /// existing ones converge on the same base shape; additive columns land via /// [`MIGRATIONS`] as try-and-ignore `ALTER TABLE`s so existing dbs catch up /// without a hard schema-version bump. const SCHEMA: &str = r" CREATE TABLE IF NOT EXISTS todos ( id INTEGER PRIMARY KEY AUTOINCREMENT, subsystem TEXT NOT NULL, subsystem_key TEXT, summary TEXT NOT NULL, source TEXT, updated_at INTEGER NOT NULL ); -- (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 (subsystem, subsystem_key); "; /// Additive column migrations — see `turn_stats.rs`'s identical pattern. /// Each runs unconditionally and ignores `duplicate column name` errors, so /// re-running on an already-migrated db is a silent no-op. const MIGRATIONS: &[&str] = &[ "ALTER TABLE todos ADD COLUMN acked INTEGER NOT NULL DEFAULT 0", // No default: only meaningful once `acked = 1`; NULL on every // pre-migration / never-acked row, which `reap_acked`'s `acked_at < ?` // comparison naturally excludes (NULL compares false, not true). "ALTER TABLE todos ADD COLUMN acked_at INTEGER", ]; /// One dynamic, subsystem-pushed todo. The consumer derives `age_seconds` /// from `updated_at`. #[derive(Debug, Clone)] pub struct Todo { pub id: i64, /// 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. 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). pub source: Option, pub updated_at: DateTime, } /// The harness-local todo store. Cheap to share behind an `Arc`; the inner /// connection is guarded by a `Mutex` (todo ops are short sqlite writes). pub struct Todos { conn: Mutex, } impl Todos { /// Open (creating if needed) the todo store at `path`. /// /// # Errors /// /// Propagates sqlite open / schema-apply failures. pub fn open(path: &Path) -> Result { let conn = Connection::open(path).with_context(|| format!("open todos db {}", path.display()))?; conn.execute_batch(SCHEMA).context("apply todos schema")?; for stmt in MIGRATIONS { // Ignore "duplicate column name" — the migration already ran. // Any other error is logged but doesn't fail open(): the base // schema still works and we'd rather keep the harness alive // than crash on an upgrade hiccup. if let Err(e) = conn.execute(stmt, []) { let msg = e.to_string(); if !msg.contains("duplicate column name") { tracing::warn!(error = %msg, stmt, "todos migration failed"); } } } Ok(Self { conn: Mutex::new(conn), }) } /// Insert a todo, or update the existing one for `(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 signal the turn loop (re-pushing an identical keyed /// todo is a no-op and must not re-wake). This holds **regardless of /// whether the existing row is acked**: an identical re-push of an /// acked row stays quiet (and stays acked/hidden), while a genuinely /// different summary un-acks it and reports `changed = true` — the /// agent's earlier dismissal doesn't suppress a real re-deterioration. /// /// # Errors /// /// Propagates sqlite query / execute failures. /// /// # Panics /// /// Panics if the connection mutex is poisoned. pub fn upsert( &self, subsystem: &str, key: Option<&str>, summary: &str, source: Option<&str>, ) -> Result<(i64, bool)> { let conn = self.conn.lock().unwrap(); let now = Utc::now().timestamp(); let existing: Option<(i64, String, Option)> = if key.is_some() { conn.query_row( "SELECT id, summary, source FROM todos \ WHERE subsystem = ?1 AND subsystem_key IS ?2", params![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)); } // A materially different summary clears any prior ack — this is // new information, not a re-announcement of what was dismissed. conn.execute( "UPDATE todos SET summary = ?1, source = ?2, updated_at = ?3, \ acked = 0, acked_at = NULL \ WHERE id = ?4", params![summary, source, now, id], )?; return Ok((id, true)); } conn.execute( "INSERT INTO todos \ (subsystem, subsystem_key, summary, source, updated_at) \ VALUES (?1, ?2, ?3, ?4, ?5)", params![subsystem, key, summary, source, now], )?; Ok((conn.last_insert_rowid(), true)) } /// Clear producer-resolved todo(s) by `(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 (clear a specific keyless one via [`Todos::mark_done`] by /// id instead). Returns the number of rows deleted. /// /// # Errors /// /// Propagates the sqlite delete failure. /// /// # Panics /// /// Panics if the connection mutex is poisoned. pub fn clear(&self, subsystem: &str, key: Option<&str>) -> Result { let conn = self.conn.lock().unwrap(); let n = conn.execute( "DELETE FROM todos WHERE subsystem = ?1 AND subsystem_key IS ?2", params![subsystem, key], )?; Ok(n) } /// Clear every todo `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, subsystem: &str) -> Result { let conn = self.conn.lock().unwrap(); let n = conn.execute("DELETE FROM todos WHERE subsystem = ?1", params![subsystem])?; Ok(n) } /// The agent marks one of its todos done, by id — i.e. dismisses it from /// its own list. Sets `acked` rather than deleting the row (see the /// module doc comment for why): the row stays intact so a reconciled /// producer's next `upsert` can still tell "unchanged" from "genuinely /// different" instead of seeing an empty table and treating a re-poll of /// the same condition as brand new. Returns the number of rows newly /// acked (0 when the id was unknown, already gone, or already acked — /// acking twice is not a new action). /// /// # Errors /// /// Propagates the sqlite update failure. /// /// # Panics /// /// Panics if the connection mutex is poisoned. pub fn mark_done(&self, id: i64) -> Result { let conn = self.conn.lock().unwrap(); let n = conn.execute( "UPDATE todos SET acked = 1, acked_at = ?1 WHERE id = ?2 AND acked = 0", params![Utc::now().timestamp(), id], )?; Ok(n) } /// Bulk-ack a specific, explicit list of todo ids in one shot. Exists /// for the case a per-id `mark_done` loop isn't worth the round-trips: /// an agent that hasn't called `get_loose_ends` in a long stretch (or /// one whose producers pile up faster than it triages) can end up with /// a backlog large enough that clearing it one call at a time is /// impractical. Deliberately **explicit ids, not a `<= threshold` /// range** — a reviewer's call on the design: a range-based /// bulk-ack risks silently acking something the agent never actually /// looked at, since todos are heterogeneous unrelated items (bash / /// matrix / forge) rather than a sequentially-read stream the way inbox /// messages are. The caller is expected to have looked at each id /// (typically the ids `get_loose_ends` just rendered) before passing /// them here. Same `acked`-not-deleted semantics as [`Self::mark_done`] /// (a reconciled producer's next `upsert` still sees the row to compare /// against). Unknown/already-acked ids are silently skipped — same "not /// a new action" idempotence as the single-id path. Returns the number /// of rows newly acked. /// /// # Errors /// /// Propagates the sqlite update failure. /// /// # Panics /// /// Panics if the connection mutex is poisoned. pub fn mark_done_many(&self, ids: &[i64]) -> Result { if ids.is_empty() { return Ok(0); } let conn = self.conn.lock().unwrap(); let now = Utc::now().timestamp(); let mut total = 0usize; for id in ids { total += conn.execute( "UPDATE todos SET acked = 1, acked_at = ?1 WHERE acked = 0 AND id = ?2", params![now, id], )?; } Ok(total) } /// List todos, newest-updated first. `subsystem = Some(..)` filters to /// one producer's set; `None` returns all. Excludes acked rows — once /// the agent has dismissed a todo it stays out of its own list, even /// though the row itself survives (see `mark_done`). /// /// # Errors /// /// Propagates the sqlite prepare / query failures. /// /// # Panics /// /// Panics if the connection mutex is poisoned. pub fn list(&self, subsystem: Option<&str>) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "SELECT id, subsystem, subsystem_key, summary, source, updated_at \ FROM todos \ WHERE acked = 0 AND (?1 IS NULL OR subsystem = ?1) \ ORDER BY updated_at DESC, id DESC", )?; let rows = stmt .query_map(params![subsystem], |row| { let updated_at_secs: i64 = row.get(5)?; Ok(Todo { id: row.get(0)?, subsystem: row.get(1)?, subsystem_key: row.get(2)?, summary: row.get(3)?, source: row.get(4)?, updated_at: wire_time::from_secs(updated_at_secs), }) })? .collect::>>()?; Ok(rows) } /// Cheap existence check — `true` when at least one un-acked todo row /// exists (across every subsystem). Used to gate a `todo_wake` /// notification against being turned into a turn when its payload has /// already been drained by an earlier turn (see the serve loop's /// `LocalTodo` arm): an `EXISTS` probe, not a full `list` + row /// materialization, since this runs on every wake. /// /// # Errors /// /// Propagates the sqlite query failure. /// /// # Panics /// /// Panics if the connection mutex is poisoned. pub fn has_any(&self) -> Result { let conn = self.conn.lock().unwrap(); let any: bool = conn.query_row( "SELECT EXISTS(SELECT 1 FROM todos WHERE acked = 0)", [], |row| row.get(0), )?; Ok(any) } /// Reap acked rows whose `acked_at` is older than `cutoff` (a unix /// timestamp). Called by the harness's hourly vacuum sweep /// (`vacuum.rs`), not on any hot path — bounds the table against acked /// one-shot todos (bash/matrix/forge) that will never naturally be /// upserted again and so would otherwise sit forever. Never touches /// un-acked rows. Returns the number of rows deleted. /// /// # Errors /// /// Propagates the sqlite delete failure. /// /// # Panics /// /// Panics if the connection mutex is poisoned. pub fn reap_acked(&self, cutoff: i64) -> Result { let conn = self.conn.lock().unwrap(); let n = conn.execute( "DELETE FROM todos WHERE acked = 1 AND acked_at < ?1", params![cutoff], )?; Ok(n) } } #[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("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("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("matrix", Some("!room:x"), "3 unread", None) .unwrap(); assert_eq!(id1, id3); assert!(changed3); assert_eq!(s.list(Some("matrix")).unwrap().len(), 1); } #[test] fn keyless_todos_always_insert() { let (_dir, s) = store(); let (a, _) = s.upsert("bash", None, "task done", None).unwrap(); let (b, _) = s.upsert("bash", None, "task done", None).unwrap(); assert_ne!(a, b, "keyless pushes are distinct one-offs"); assert_eq!(s.list(Some("bash")).unwrap().len(), 2); } #[test] fn clear_deletes_and_mark_done_hides_from_list() { let (_dir, s) = store(); s.upsert("matrix", Some("!a:x"), "unread", None).unwrap(); let (id, _) = s.upsert("forge", Some("pr-1"), "review", None).unwrap(); assert_eq!(s.clear("matrix", Some("!a:x")).unwrap(), 1, "clear deletes"); assert_eq!( s.mark_done(id).unwrap(), 1, "mark_done acks, doesn't delete" ); assert!( s.list(None).unwrap().is_empty(), "both are gone from the agent-visible list either way" ); } #[test] fn has_any_reflects_emptiness() { let (_dir, s) = store(); assert!(!s.has_any().unwrap(), "fresh store has no todos"); let (id, _) = s.upsert("bash", None, "task done", None).unwrap(); assert!(s.has_any().unwrap()); s.mark_done(id).unwrap(); assert!( !s.has_any().unwrap(), "empty again after draining the only row" ); } #[test] fn clear_subsystem_wipes_only_its_own() { let (_dir, s) = store(); s.upsert("matrix", Some("!a:x"), "u", None).unwrap(); s.upsert("matrix", Some("!b:x"), "u", None).unwrap(); s.upsert("forge", Some("pr-1"), "r", None).unwrap(); assert_eq!(s.clear_subsystem("matrix").unwrap(), 2); let left = s.list(None).unwrap(); assert_eq!(left.len(), 1); assert_eq!(left[0].subsystem, "forge"); } /// The whole point of the fix: acking a reconciled todo must not defeat /// the producer's own anti-nag comparison. A re-push with an identical /// summary after ack stays quiet AND stays out of the list — the `disk_watch` /// bug this closes. #[test] fn acked_row_survives_unchanged_reupsert_and_stays_hidden() { let (_dir, s) = store(); let (id, changed1) = s.upsert("disk", Some("usage"), "over 85%", None).unwrap(); assert!(changed1); assert_eq!(s.mark_done(id).unwrap(), 1, "first ack succeeds"); assert!(s.list(None).unwrap().is_empty(), "acked row is hidden"); // Same producer, same reconcile tick, identical summary — must NOT // look like a new todo just because the agent dismissed it. let (id2, changed2) = s.upsert("disk", Some("usage"), "over 85%", None).unwrap(); assert_eq!(id, id2, "same row, not a fresh insert"); assert!(!changed2, "identical summary stays quiet even though acked"); assert!( s.list(None).unwrap().is_empty(), "still hidden — an unchanged re-push does not un-ack" ); } /// A materially different summary on an acked row is real information — /// it must un-ack, re-surface in `list`, and report `changed = true`. #[test] fn acked_row_resurfaces_on_genuine_change() { let (_dir, s) = store(); let (id, _) = s.upsert("disk", Some("usage"), "over 85%", None).unwrap(); s.mark_done(id).unwrap(); assert!(s.list(None).unwrap().is_empty()); let (id2, changed) = s.upsert("disk", Some("usage"), "over 90%", None).unwrap(); assert_eq!(id, id2); assert!(changed, "a genuine re-deterioration must not be silenced"); let visible = s.list(None).unwrap(); assert_eq!(visible.len(), 1, "un-acked row is visible again"); assert_eq!(visible[0].summary, "over 90%"); } /// Acking an already-acked (or unknown) id is a no-op, not a new action. #[test] fn mark_done_twice_is_idempotent() { let (_dir, s) = store(); let (id, _) = s.upsert("bash", None, "task done", None).unwrap(); assert_eq!(s.mark_done(id).unwrap(), 1); assert_eq!(s.mark_done(id).unwrap(), 0, "already acked"); assert_eq!(s.mark_done(999_999).unwrap(), 0, "unknown id"); } /// `has_any` must also treat an acked-only table as empty — it gates a /// wake against a payload the agent has already drained. #[test] fn has_any_excludes_acked_rows() { let (_dir, s) = store(); let (id, _) = s.upsert("bash", None, "task done", None).unwrap(); assert!(s.has_any().unwrap()); s.mark_done(id).unwrap(); assert!(!s.has_any().unwrap(), "acked-only table reads as empty"); } /// `mark_done_many` acks exactly the ids passed, leaves the rest /// untouched — no threshold/range semantics. #[test] fn mark_done_many_acks_only_the_listed_ids() { let (_dir, s) = store(); let (id1, _) = s.upsert("bash", None, "task 1", None).unwrap(); let (id2, _) = s.upsert("bash", None, "task 2", None).unwrap(); let (id3, _) = s.upsert("bash", None, "task 3", None).unwrap(); assert_eq!( s.mark_done_many(&[id1, id3]).unwrap(), 2, "acks id1 and id3, not id2 — not a range" ); let left = s.list(None).unwrap(); assert_eq!(left.len(), 1); assert_eq!(left[0].id, id2); // Idempotent: re-running over the same ids acks nothing new. assert_eq!(s.mark_done_many(&[id1, id3]).unwrap(), 0); // Empty input is a no-op, not an error. assert_eq!(s.mark_done_many(&[]).unwrap(), 0); } /// `reap_acked` only removes acked rows past the cutoff — a recent ack /// and any un-acked row both survive. #[test] fn reap_acked_only_removes_old_acked_rows() { let (dir, s) = store(); let (old_id, _) = s.upsert("bash", None, "old task", None).unwrap(); let (recent_id, _) = s.upsert("bash", None, "recent task", None).unwrap(); let (never_acked_id, _) = s.upsert("forge", Some("pr-1"), "review", None).unwrap(); s.mark_done(old_id).unwrap(); s.mark_done(recent_id).unwrap(); // Backdate the old row's ack past the cutoff directly — mark_done // always acks "now", so simulate age via the raw connection. { let conn = s.conn.lock().unwrap(); conn.execute( "UPDATE todos SET acked_at = ?1 WHERE id = ?2", params![Utc::now().timestamp() - 1000, old_id], ) .unwrap(); } let removed = s.reap_acked(Utc::now().timestamp() - 500).unwrap(); assert_eq!(removed, 1, "only the backdated row is past the cutoff"); let conn = Connection::open(dir.path().join("todos.sqlite")).unwrap(); let remaining: i64 = conn .query_row("SELECT COUNT(*) FROM todos", [], |r| r.get(0)) .unwrap(); assert_eq!( remaining, 2, "recent ack + never-acked row both survive the reap" ); let _ = never_acked_id; } }