From 8a810857702ed3e773a8674c7e55c7b92af3b732 Mon Sep 17 00:00:00 2001 From: damocles Date: Tue, 28 Jul 2026 09:39:10 +0200 Subject: [PATCH] harness todos: ack reconciled todos instead of deleting them --- hive-agent/src/disk_watch.rs | 5 +- hive-agent/src/todos.rs | 237 +++++++++++++++++++++++++++++++---- hive-agent/src/vacuum.rs | 18 +++ 3 files changed, 233 insertions(+), 27 deletions(-) diff --git a/hive-agent/src/disk_watch.rs b/hive-agent/src/disk_watch.rs index a38f1b63..a434566b 100644 --- a/hive-agent/src/disk_watch.rs +++ b/hive-agent/src/disk_watch.rs @@ -15,7 +15,10 @@ //! loose-ends list instead of waking the agent every tick. Only crossing //! into a new bucket (or a change in which directories are big enough to //! list) speaks up again. Dropping back under the threshold clears the -//! todo. +//! todo. Acknowledging the todo (`cancel_loose_end`) does not defeat this — +//! `Todos::mark_done` acks the row rather than deleting it, so the next +//! probe still has the prior summary to compare against instead of seeing +//! an empty table and re-announcing an unchanged condition as new. //! //! Scoped by design: the `statvfs` reads the *whole filesystem*, which on //! a shared host volume (several agents' state dirs on the same diff --git a/hive-agent/src/todos.rs b/hive-agent/src/todos.rs index a5d7da2f..63deb11d 100644 --- a/hive-agent/src/todos.rs +++ b/hive-agent/src/todos.rs @@ -15,9 +15,18 @@ //! 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 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. +//! 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; @@ -26,6 +35,10 @@ use anyhow::{Context, Result}; use hive_sh4re::wire_time::now_unix; 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, @@ -42,6 +55,17 @@ 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. Timestamps are unix seconds; the /// consumer derives `age_seconds` from `updated_at`. #[derive(Debug, Clone)] @@ -75,6 +99,18 @@ impl Todos { 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), }) @@ -84,10 +120,14 @@ impl Todos { /// 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). + /// 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 /// @@ -121,8 +161,12 @@ impl Todos { 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 WHERE id = ?4", + "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)); @@ -175,24 +219,35 @@ impl Todos { Ok(n) } - /// The agent marks one of its todos done, by id. Returns the number of - /// rows deleted (0 when the id was unknown / already gone). + /// 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 delete failure. + /// 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("DELETE FROM todos WHERE id = ?1", params![id])?; + let n = conn.execute( + "UPDATE todos SET acked = 1, acked_at = ?1 WHERE id = ?2 AND acked = 0", + params![now_unix(), id], + )?; Ok(n) } /// List todos, newest-updated first. `subsystem = Some(..)` filters to - /// one producer's set; `None` returns all. + /// 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 /// @@ -206,7 +261,7 @@ impl Todos { let mut stmt = conn.prepare( "SELECT id, subsystem, subsystem_key, summary, source, updated_at \ FROM todos \ - WHERE (?1 IS NULL OR subsystem = ?1) \ + WHERE acked = 0 AND (?1 IS NULL OR subsystem = ?1) \ ORDER BY updated_at DESC, id DESC", )?; let rows = stmt @@ -224,12 +279,12 @@ impl Todos { Ok(rows) } - /// Cheap existence check — `true` when at least one 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. + /// 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 /// @@ -240,10 +295,36 @@ impl Todos { /// 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)", [], |row| row.get(0))?; + 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)] @@ -291,13 +372,20 @@ mod tests { } #[test] - fn clear_and_mark_done_remove_rows() { + 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); - assert_eq!(s.mark_done(id).unwrap(), 1); - assert!(s.list(None).unwrap().is_empty()); + 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] @@ -324,4 +412,101 @@ mod tests { 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"); + } + + /// `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![now_unix() - 1000, old_id], + ) + .unwrap(); + } + + let removed = s.reap_acked(now_unix() - 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; + } } diff --git a/hive-agent/src/vacuum.rs b/hive-agent/src/vacuum.rs index c95e2ebc..1c77d86d 100644 --- a/hive-agent/src/vacuum.rs +++ b/hive-agent/src/vacuum.rs @@ -32,6 +32,12 @@ const STREAM_KEEP_SECS: i64 = 14 * 24 * 3600; /// before reaping them — same window as `STREAM_KEEP_SECS`, kept around /// only to serve the trailing-window `ReminderRollup` stats. const REMINDER_KEEP_SECS: i64 = 14 * 24 * 3600; +/// Keep acked (agent-dismissed) todo rows this long before reaping them — +/// see `todos.rs`'s module doc for why acking doesn't delete outright. Long +/// enough that a genuinely quiet month is the only way to trigger the "one +/// spurious re-announcement" fallback path, short enough the table doesn't +/// grow meaningfully from one-shot todos that will never be upserted again. +const TODO_ACKED_KEEP_SECS: i64 = 30 * 24 * 3600; /// Terminal bash-task statuses whose files are eligible for deletion. const TERMINAL_STATUSES: &[&str] = &["done", "timed_out", "interrupted"]; @@ -72,6 +78,11 @@ fn sweep_once() { Ok(n) => tracing::info!(removed = n, "reminders vacuum"), Err(e) => tracing::warn!(error = ?e, "reminders vacuum failed"), } + match vacuum_todos(&state_db) { + Ok(0) => {} + Ok(n) => tracing::info!(removed = n, "todos vacuum"), + Err(e) => tracing::warn!(error = ?e, "todos vacuum failed"), + } } } @@ -84,6 +95,13 @@ fn vacuum_reminders(path: &Path) -> anyhow::Result { store.prune_delivered_older_than(now_unix() - REMINDER_KEEP_SECS) } +/// Reap acked todo rows older than [`TODO_ACKED_KEEP_SECS`] via the typed +/// store API — same own-short-lived-connection shape as `vacuum_reminders`. +fn vacuum_todos(path: &Path) -> anyhow::Result { + let store = crate::todos::Todos::open(path)?; + store.reap_acked(now_unix() - TODO_ACKED_KEEP_SECS) +} + /// Delete eligible bash-task trios in `dir`. Returns the count of `.json` /// sentinels removed (each represents one task; `.out`/`.err` deletions are /// not counted separately).