todos: reopen an acked keyed row when the caller says so

hive-agent-sock/src/lib.rs

@@ -41,6 +41,16 @@ pub enum Request {
     /// subsystem-specific dedup key (a matrix room id, a bash task id).
     /// A new-or-changed row signals the turn loop; an identical keyed
     /// re-push is a silent no-op. Keyless todos always insert as one-offs.
+    ///
+    /// `reopen_if_acked` (default `false`) governs what "changed" means
+    /// for an *acked* row with byte-identical `summary`/`source`: by
+    /// default it stays quiet — the right behavior for a reconciler
+    /// producer re-announcing a still-true condition (`disk_watch` polling
+    /// and finding the same problem). A producer whose keyed re-push
+    /// represents a genuinely new occurrence rather than a restatement
+    /// (a recurring scheduled prompt firing again with the same body) sets
+    /// this `true` so an acked row reopens and re-wakes instead of being
+    /// silently swallowed by the previous occurrence's ack.
     UpsertTodo {
         subsystem: String,
         #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -48,6 +58,8 @@ pub enum Request {
         summary: String,
         #[serde(default, skip_serializing_if = "Option::is_none")]
         source: Option<String>,
+        #[serde(default)]
+        reopen_if_acked: bool,
     },
     /// Clear producer-resolved todo(s). `key = Some(k)` clears the one
     /// keyed row; `key = None` clears the subsystem's keyless rows; `all

hive-agent/src/db_migrate.rs

@@ -136,7 +136,7 @@ mod tests {
         {
             let store = Todos::open(&legacy_todos).unwrap();
             store
-                .upsert("matrix", Some("!a:x"), "1 unread", None)
+                .upsert("matrix", Some("!a:x"), "1 unread", None, false)
                 .unwrap();
         }
         run(
@@ -184,7 +184,7 @@ mod tests {
         {
             Todos::open(&legacy_todos)
                 .unwrap()
-                .upsert("bash", None, "task done", None)
+                .upsert("bash", None, "task done", None, false)
                 .unwrap();
             Reminders::open(&legacy_reminders)
                 .unwrap()

hive-agent/src/disk_watch.rs

@@ -94,7 +94,7 @@ pub async fn run(todos: Arc<Todos>, wake: Arc<Notify>) {
 /// interaction is testable without a timer.
 fn reconcile(todos: &Todos, wake: &Notify, summary: Option<&str>) {
     match summary {
-        Some(summary) => match todos.upsert(SUBSYSTEM, Some(TODO_KEY), summary, None) {
+        Some(summary) => match todos.upsert(SUBSYSTEM, Some(TODO_KEY), summary, None, false) {
             // Only a genuine change wakes the agent — an identical summary
             // means the situation is unchanged and already in its list.
             Ok((_, true)) => wake.notify_one(),

hive-agent/src/todo_server.rs

@@ -203,6 +203,7 @@ fn dispatch(
             key,
             summary,
             source,
+            reopen_if_acked,
         } => upsert_todo(
             store,
             wake,
@@ -210,6 +211,7 @@ fn dispatch(
             key.as_deref(),
             &summary,
             source.as_deref(),
+            reopen_if_acked,
         ),
         Request::ClearTodo {
             subsystem,
@@ -378,8 +380,9 @@ fn upsert_todo(
     key: Option<&str>,
     summary: &str,
     source: Option<&str>,
+    reopen_if_acked: bool,
 ) -> Response {
-    match store.upsert(subsystem, key, summary, source) {
+    match store.upsert(subsystem, key, summary, source, reopen_if_acked) {
         Ok((id, changed)) => {
             tracing::debug!(%subsystem, ?key, id, changed, "todo upsert");
             if changed {

hive-agent/src/todos.rs

@@ -124,11 +124,18 @@ impl Todos {
     /// 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
+    /// todo is a no-op and must not re-wake). A **materially different**
-    /// whether the existing row is acked**: an identical re-push of an
+    /// summary always un-acks + reports `changed = true`, unconditionally —
-    /// acked row stays quiet (and stays acked/hidden), while a genuinely
+    /// the agent's earlier dismissal never suppresses a real
-    /// different summary un-acks it and reports `changed = true` — the
+    /// re-deterioration.
-    /// agent's earlier dismissal doesn't suppress a real re-deterioration.
+    ///
+    /// `reopen_if_acked` decides what an identical re-push against an
+    /// **acked** row means: `false` (a reconciler like `disk_watch`) stays
+    /// quiet — re-deriving the same still-true summary isn't new
+    /// information. `true` (an event-style producer like a scheduled
+    /// prompt) reopens anyway — each push is a distinct occurrence, not a
+    /// restatement, so acking the *previous* firing doesn't cover *this*
+    /// one.
     ///
     /// # Errors
     ///
@@ -143,27 +150,29 @@ impl Todos {
         key: Option<&str>,
         summary: &str,
         source: Option<&str>,
+        reopen_if_acked: bool,
     ) -> Result<(i64, bool)> {
         let conn = self.conn.lock().unwrap();
         let now = Utc::now().timestamp();
-        let existing: Option<(i64, String, Option<String>)> = if key.is_some() {
+        let existing: Option<(i64, String, Option<String>, bool)> = if key.is_some() {
             conn.query_row(
-                "SELECT id, summary, source FROM todos \
+                "SELECT id, summary, source, acked FROM todos \
                  WHERE subsystem = ?1 AND subsystem_key IS ?2",
                 params![subsystem, key],
-                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
+                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
             )
             .ok()
         } else {
             None
         };
-        if let Some((id, cur_summary, cur_source)) = existing {
+        if let Some((id, cur_summary, cur_source, cur_acked)) = existing {
-            let unchanged = cur_summary == summary && cur_source.as_deref() == source;
+            let content_unchanged = cur_summary == summary && cur_source.as_deref() == source;
-            if unchanged {
+            if content_unchanged && !(cur_acked && reopen_if_acked) {
                 return Ok((id, false));
             }
-            // A materially different summary clears any prior ack — this is
+            // Either the summary genuinely differs, or it's identical but
-            // new information, not a re-announcement of what was dismissed.
+            // the caller asked to treat an acked row's re-push as a new
+            // occurrence — either way this un-acks + reports changed.
             conn.execute(
                 "UPDATE todos SET summary = ?1, source = ?2, updated_at = ?3, \
                         acked = 0, acked_at = NULL \
@@ -387,18 +396,18 @@ mod tests {
     fn keyed_upsert_dedups_and_reports_changed() {
         let (_dir, s) = store();
         let (id1, changed1) = s
-            .upsert("matrix", Some("!room:x"), "1 unread", None)
+            .upsert("matrix", Some("!room:x"), "1 unread", None, false)
             .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)
+            .upsert("matrix", Some("!room:x"), "1 unread", None, false)
             .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)
+            .upsert("matrix", Some("!room:x"), "3 unread", None, false)
             .unwrap();
         assert_eq!(id1, id3);
         assert!(changed3);
@@ -408,8 +417,8 @@ mod tests {
     #[test]
     fn keyless_todos_always_insert() {
         let (_dir, s) = store();
-        let (a, _) = s.upsert("bash", None, "task done", None).unwrap();
+        let (a, _) = s.upsert("bash", None, "task done", None, false).unwrap();
-        let (b, _) = s.upsert("bash", None, "task done", None).unwrap();
+        let (b, _) = s.upsert("bash", None, "task done", None, false).unwrap();
         assert_ne!(a, b, "keyless pushes are distinct one-offs");
         assert_eq!(s.list(Some("bash")).unwrap().len(), 2);
     }
@@ -417,8 +426,11 @@ mod tests {
     #[test]
     fn clear_deletes_and_mark_done_hides_from_list() {
         let (_dir, s) = store();
-        s.upsert("matrix", Some("!a:x"), "unread", None).unwrap();
+        s.upsert("matrix", Some("!a:x"), "unread", None, false)
-        let (id, _) = s.upsert("forge", Some("pr-1"), "review", None).unwrap();
+            .unwrap();
+        let (id, _) = s
+            .upsert("forge", Some("pr-1"), "review", None, false)
+            .unwrap();
         assert_eq!(s.clear("matrix", Some("!a:x")).unwrap(), 1, "clear deletes");
         assert_eq!(
             s.mark_done(id).unwrap(),
@@ -435,7 +447,7 @@ mod tests {
     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();
+        let (id, _) = s.upsert("bash", None, "task done", None, false).unwrap();
         assert!(s.has_any().unwrap());
         s.mark_done(id).unwrap();
         assert!(
@@ -447,9 +459,9 @@ mod tests {
     #[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("!a:x"), "u", None, false).unwrap();
-        s.upsert("matrix", Some("!b:x"), "u", None).unwrap();
+        s.upsert("matrix", Some("!b:x"), "u", None, false).unwrap();
-        s.upsert("forge", Some("pr-1"), "r", None).unwrap();
+        s.upsert("forge", Some("pr-1"), "r", None, false).unwrap();
         assert_eq!(s.clear_subsystem("matrix").unwrap(), 2);
         let left = s.list(None).unwrap();
         assert_eq!(left.len(), 1);
@@ -463,14 +475,18 @@ mod tests {
     #[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();
+        let (id, changed1) = s
+            .upsert("disk", Some("usage"), "over 85%", None, false)
+            .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();
+        let (id2, changed2) = s
+            .upsert("disk", Some("usage"), "over 85%", None, false)
+            .unwrap();
         assert_eq!(id, id2, "same row, not a fresh insert");
         assert!(!changed2, "identical summary stays quiet even though acked");
         assert!(
@@ -479,16 +495,65 @@ mod tests {
         );
     }
 
+    /// The other half of the same mechanism: an event-style producer (a
+    /// recurring scheduled prompt is the real caller) sets
+    /// `reopen_if_acked = true` because each identical-content push is a
+    /// distinct occurrence, not a restatement — an acked row must reopen
+    /// and report `changed = true` even though the summary didn't move.
+    #[test]
+    fn reopen_if_acked_resurfaces_an_identical_acked_row() {
+        let (_dir, s) = store();
+        let (id, changed1) = s
+            .upsert(
+                "schedule",
+                Some("schedule:1"),
+                "check the thing",
+                None,
+                true,
+            )
+            .unwrap();
+        assert!(
+            changed1,
+            "first push is new → changed regardless of the flag"
+        );
+        assert_eq!(s.mark_done(id).unwrap(), 1, "agent reviewed fire #1");
+        assert!(s.list(None).unwrap().is_empty());
+
+        // Second fire, byte-identical body, but reopen_if_acked = true —
+        // this must NOT collapse into silence the way the reconciler test
+        // above expects it to.
+        let (id2, changed2) = s
+            .upsert(
+                "schedule",
+                Some("schedule:1"),
+                "check the thing",
+                None,
+                true,
+            )
+            .unwrap();
+        assert_eq!(id, id2, "same keyed row, not a fresh insert");
+        assert!(changed2, "acked + identical + reopen_if_acked still wakes");
+        assert_eq!(
+            s.list(None).unwrap().len(),
+            1,
+            "reopened row is visible again"
+        );
+    }
+
     /// 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();
+        let (id, _) = s
+            .upsert("disk", Some("usage"), "over 85%", None, false)
+            .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();
+        let (id2, changed) = s
+            .upsert("disk", Some("usage"), "over 90%", None, false)
+            .unwrap();
         assert_eq!(id, id2);
         assert!(changed, "a genuine re-deterioration must not be silenced");
         let visible = s.list(None).unwrap();
@@ -500,7 +565,7 @@ mod tests {
     #[test]
     fn mark_done_twice_is_idempotent() {
         let (_dir, s) = store();
-        let (id, _) = s.upsert("bash", None, "task done", None).unwrap();
+        let (id, _) = s.upsert("bash", None, "task done", None, false).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");
@@ -511,7 +576,7 @@ mod tests {
     #[test]
     fn has_any_excludes_acked_rows() {
         let (_dir, s) = store();
-        let (id, _) = s.upsert("bash", None, "task done", None).unwrap();
+        let (id, _) = s.upsert("bash", None, "task done", None, false).unwrap();
         assert!(s.has_any().unwrap());
         s.mark_done(id).unwrap();
         assert!(!s.has_any().unwrap(), "acked-only table reads as empty");
@@ -522,9 +587,9 @@ mod tests {
     #[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 (id1, _) = s.upsert("bash", None, "task 1", None, false).unwrap();
-        let (id2, _) = s.upsert("bash", None, "task 2", None).unwrap();
+        let (id2, _) = s.upsert("bash", None, "task 2", None, false).unwrap();
-        let (id3, _) = s.upsert("bash", None, "task 3", None).unwrap();
+        let (id3, _) = s.upsert("bash", None, "task 3", None, false).unwrap();
         assert_eq!(
             s.mark_done_many(&[id1, id3]).unwrap(),
             2,
@@ -544,9 +609,11 @@ mod tests {
     #[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 (old_id, _) = s.upsert("bash", None, "old task", None, false).unwrap();
-        let (recent_id, _) = s.upsert("bash", None, "recent task", None).unwrap();
+        let (recent_id, _) = s.upsert("bash", None, "recent task", None, false).unwrap();
-        let (never_acked_id, _) = s.upsert("forge", Some("pr-1"), "review", None).unwrap();
+        let (never_acked_id, _) = s
+            .upsert("forge", Some("pr-1"), "review", None, false)
+            .unwrap();
         s.mark_done(old_id).unwrap();
         s.mark_done(recent_id).unwrap();
 

hive-c0re/src/actions.rs

@@ -967,6 +967,7 @@ pub async fn destroy(coord: &Arc<Coordinator>, name: &str, purge: bool) -> Resul
             Some(format!("destroyed:{name}")),
             format!("agent '{name}' destroyed"),
             None,
+            false,
         )
         .await;
     // Container row disappeared — rescan so the dashboard fires

hive-c0re/src/coordinator.rs

@@ -1439,6 +1439,8 @@ impl Coordinator {
     /// (`let _ = coord.push_todo(...).await;`); callers that track a
     /// per-target delivery outcome (e.g. the scheduled-prompts worker's
     /// `last_result` column) use it instead of assuming success.
+    /// `reopen_if_acked` forwards to `UpsertTodo` (see its doc comment) —
+    /// `false` for every caller here except the scheduled-prompts worker.
     pub async fn push_todo(
         &self,
         agent: &str,
@@ -1446,6 +1448,7 @@ impl Coordinator {
         key: Option<String>,
         summary: String,
         source: Option<String>,
+        reopen_if_acked: bool,
     ) -> Result<(), String> {
         let Ok(ident) = hive_types::Ident::parse(agent) else {
             tracing::warn!(%agent, "push_todo: not a valid agent ident, skipping");
@@ -1461,6 +1464,7 @@ impl Coordinator {
             key,
             summary,
             source,
+            reopen_if_acked,
         };
         match hive_sock_client::request::<_, hive_agent_sock::Response>(
             &path,
@@ -1482,7 +1486,10 @@ impl Coordinator {
     }
 
     /// `push_todo` to whichever agent submitted approval `approval_id` —
-    /// same resolution `notify_submitter` uses.
+    /// same resolution `notify_submitter` uses. Every caller here is a
+    /// one-shot approval-resolution notice, so `reopen_if_acked` is
+    /// unconditionally `false` — there's no reconciler/event distinction
+    /// to make for an event that only ever fires once.
     pub async fn push_todo_submitter(
         &self,
         approval_id: i64,
@@ -1492,7 +1499,7 @@ impl Coordinator {
         source: Option<String>,
     ) -> Result<(), String> {
         let target = self.submitter_or_manager(approval_id);
-        self.push_todo(&target, subsystem, key, summary, source)
+        self.push_todo(&target, subsystem, key, summary, source, false)
             .await
     }
 

hive-c0re/src/job_queue/exec.rs

@@ -178,6 +178,7 @@ async fn run_emit_rebuilt(coord: &Arc<Coordinator>, agent: &str, dag_id: Option<
             Some(format!("rebuilt:{agent}")),
             summary,
             None,
+            false,
         )
         .await;
 }
@@ -441,6 +442,7 @@ async fn run_stop(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
             Some(format!("killed:{name}")),
             format!("agent '{name}' killed"),
             None,
+            false,
         )
         .await;
     coord.rescan_containers_and_emit().await;

hive-c0re/src/server.rs

@@ -303,6 +303,7 @@ async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostRespon
                     Some(format!("spawned:{name}")),
                     format!("agent '{name}' spawned"),
                     None,
+                    false,
                 )
                 .await;
             // Update tmpfiles.d so the new agent's dirs survive a reboot.
@@ -318,6 +319,7 @@ async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostRespon
                     Some(format!("spawned:{name}")),
                     format!("agent '{name}' spawn FAILED: {e:#}"),
                     None,
+                    false,
                 )
                 .await;
             return Err(e);

hive-c0re/src/socket_server/lifecycle_handlers.rs

@@ -145,6 +145,7 @@ pub(super) async fn handle_kill(coord: &Arc<Coordinator>, agent: &str, name: &st
                     Some(format!("killed:{name}")),
                     format!("agent '{name}' killed"),
                     None,
+                    false,
                 )
                 .await;
             Response::Ok

hive-c0re/src/workers/crash_watch.rs

@@ -143,6 +143,7 @@ async fn emit_login_transitions(
                 Some(format!("logged_in:{agent}")),
                 format!("agent '{agent}' logged in"),
                 None,
+                false,
             )
             .await;
     }
@@ -174,6 +175,7 @@ async fn emit_login_transitions(
                 Some(format!("needs_login:{agent}")),
                 format!("agent '{agent}' needs login"),
                 None,
+                false,
             )
             .await;
     }

hive-c0re/src/workers/scheduled_prompts_worker.rs

@@ -87,8 +87,14 @@ async fn tick(coord: &Arc<Coordinator>) {
 /// upsert-by-key dedup the same job a now-removed
 /// `has_pending_with_body` broker check used to do — collapsing a
 /// re-fire of the *same schedule* against a target that hasn't
-/// reviewed the last one yet — and does it more precisely (keyed on
+/// reviewed the last one yet, keyed on schedule identity rather than on
-/// schedule identity, not on the body happening to be byte-identical).
+/// the body happening to be byte-identical. `deliver_to_target` also
+/// passes `reopen_if_acked = true`, so this collapse only applies while
+/// the previous fire is genuinely still sitting there un-reviewed — once
+/// the target acks it, the *next* fire reopens the row and wakes again
+/// even with an identical body, instead of silently staying quiet
+/// forever — a recurring schedule with a static body used to wake its
+/// target once, ever, and then look dead.
 async fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
     let known: std::collections::HashSet<String> = known_agents().await;
     for target_row in &schedule.targets {
@@ -158,7 +164,9 @@ async fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64)
 /// to push into there), a `push_todo` otherwise, keyed on the
 /// schedule's own identity (`schedule:{schedule_id}`) so a re-fire
 /// against a target that hasn't reviewed the last one collapses via
-/// `push_todo`'s own upsert-by-key dedup. Shared by the periodic
+/// `push_todo`'s own upsert-by-key dedup — but `reopen_if_acked = true`
+/// means that collapse only holds while unreviewed; once acked, the next
+/// fire reopens regardless of whether the body changed. Shared by the periodic
 /// `fire_schedule` tick and the manual `fire_now` dashboard action —
 /// the only difference between them is what each caller does with the
 /// `Result` (log/prefix and per-target `last_result`/`FireNowReport`
@@ -188,6 +196,14 @@ async fn deliver_to_target(
                 Some(format!("schedule:{schedule_id}")),
                 body.to_owned(),
                 Some("scheduled".to_owned()),
+                // Each fire is a distinct occurrence, not a restatement of
+                // a persisting condition — an agent that already acked the
+                // *previous* firing hasn't acked *this* one, so an acked
+                // row must reopen even when the body is byte-identical
+                // (the common case: most schedules don't vary their text
+                // per fire). See `Todos::upsert`'s doc comment for the
+                // full reconciler-vs-event rationale.
+                true,
             )
             .await
     }