diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs index 480c668d..af436056 100644 --- a/hive-agent-sock/src/lib.rs +++ b/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, + #[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 diff --git a/hive-agent/src/db_migrate.rs b/hive-agent/src/db_migrate.rs index 9feca95b..368edf20 100644 --- a/hive-agent/src/db_migrate.rs +++ b/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() diff --git a/hive-agent/src/disk_watch.rs b/hive-agent/src/disk_watch.rs index a434566b..2a8f6d1f 100644 --- a/hive-agent/src/disk_watch.rs +++ b/hive-agent/src/disk_watch.rs @@ -94,7 +94,7 @@ pub async fn run(todos: Arc, wake: Arc) { /// 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(), diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index 00409d69..0352787c 100644 --- a/hive-agent/src/todo_server.rs +++ b/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 { diff --git a/hive-agent/src/todos.rs b/hive-agent/src/todos.rs index 53e5702e..aeaa253f 100644 --- a/hive-agent/src/todos.rs +++ b/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 - /// 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. + /// todo is a no-op and must not re-wake). A **materially different** + /// summary always un-acks + reports `changed = true`, unconditionally — + /// the agent's earlier dismissal never suppresses 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)> = if key.is_some() { + let existing: Option<(i64, String, Option, 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 { - let unchanged = cur_summary == summary && cur_source.as_deref() == source; - if unchanged { + if let Some((id, cur_summary, cur_source, cur_acked)) = existing { + let content_unchanged = cur_summary == summary && cur_source.as_deref() == source; + if content_unchanged && !(cur_acked && reopen_if_acked) { return Ok((id, false)); } - // A materially different summary clears any prior ack — this is - // new information, not a re-announcement of what was dismissed. + // Either the summary genuinely differs, or it's identical but + // 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 (b, _) = 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, 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(); - let (id, _) = s.upsert("forge", Some("pr-1"), "review", None).unwrap(); + s.upsert("matrix", Some("!a:x"), "unread", None, false) + .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("!b:x"), "u", None).unwrap(); - s.upsert("forge", Some("pr-1"), "r", None).unwrap(); + s.upsert("matrix", Some("!a:x"), "u", None, false).unwrap(); + s.upsert("matrix", Some("!b:x"), "u", None, false).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 (id2, _) = s.upsert("bash", None, "task 2", None).unwrap(); - let (id3, _) = s.upsert("bash", None, "task 3", None).unwrap(); + let (id1, _) = s.upsert("bash", None, "task 1", None, false).unwrap(); + let (id2, _) = s.upsert("bash", None, "task 2", None, false).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 (recent_id, _) = s.upsert("bash", None, "recent task", None).unwrap(); - let (never_acked_id, _) = s.upsert("forge", Some("pr-1"), "review", None).unwrap(); + let (old_id, _) = s.upsert("bash", None, "old task", None, false).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, false) + .unwrap(); s.mark_done(old_id).unwrap(); s.mark_done(recent_id).unwrap(); diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 231fe442..085d2bf7 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -967,6 +967,7 @@ pub async fn destroy(coord: &Arc, name: &str, purge: bool) -> Resul Some(format!("destroyed:{name}")), format!("agent '{name}' destroyed"), None, + false, ) .await; // Container row disappeared — rescan so the dashboard fires diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index bf08febc..1e8eb46e 100644 --- a/hive-c0re/src/coordinator.rs +++ b/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, summary: String, source: Option, + 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, ) -> 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 } diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 46319af1..d392bd78 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -178,6 +178,7 @@ async fn run_emit_rebuilt(coord: &Arc, agent: &str, dag_id: Option< Some(format!("rebuilt:{agent}")), summary, None, + false, ) .await; } @@ -441,6 +442,7 @@ async fn run_stop(coord: &Arc, name: &str) -> Result<()> { Some(format!("killed:{name}")), format!("agent '{name}' killed"), None, + false, ) .await; coord.rescan_containers_and_emit().await; diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 277483eb..eedb54d7 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -303,6 +303,7 @@ async fn handle_spawn(coord: &Arc, name: &str) -> Result, name: &str) -> Result, agent: &str, name: &st Some(format!("killed:{name}")), format!("agent '{name}' killed"), None, + false, ) .await; Response::Ok diff --git a/hive-c0re/src/workers/crash_watch.rs b/hive-c0re/src/workers/crash_watch.rs index 41c20735..6ee46982 100644 --- a/hive-c0re/src/workers/crash_watch.rs +++ b/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; } diff --git a/hive-c0re/src/workers/scheduled_prompts_worker.rs b/hive-c0re/src/workers/scheduled_prompts_worker.rs index 51d7bc5f..2f99f5bc 100644 --- a/hive-c0re/src/workers/scheduled_prompts_worker.rs +++ b/hive-c0re/src/workers/scheduled_prompts_worker.rs @@ -87,8 +87,14 @@ async fn tick(coord: &Arc) { /// 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 -/// schedule identity, not on the body happening to be byte-identical). +/// reviewed the last one yet, keyed on schedule identity rather than on +/// 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, schedule: &Schedule, now: i64) { let known: std::collections::HashSet = known_agents().await; for target_row in &schedule.targets { @@ -158,7 +164,9 @@ async fn fire_schedule(coord: &Arc, 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 }