diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index 60f3dc1a..088556b9 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -279,6 +279,78 @@ impl Broker { Ok(n > 0) } + /// Send a "your parent changed from X to Y" notification to `child`, + /// coalescing with any existing undelivered one so that multiple moves + /// while the agent is offline collapse into a single message spanning + /// the full arc (e.g. A→B then B→C becomes "your parent changed from A to C"). + /// + /// If an undelivered system reparent notification for `child` already + /// exists, its body is updated in-place preserving the original "from" + /// label. If none exists, a fresh message is inserted with `old_label` + /// as the source. + pub fn send_coalescing_reparent( + &self, + child: &str, + old_label: &str, + new_label: &str, + ) -> Result<()> { + const PREFIX: &str = "your parent changed from "; + const SEPARATOR: &str = " to "; + let conn = self.conn.lock().unwrap(); + let now = now_unix(); + let existing: Option<(i64, String)> = conn + .query_row( + "SELECT id, body FROM messages + WHERE recipient = ?1 + AND sender = ?2 + AND body LIKE ?3 + AND delivered_at IS NULL + LIMIT 1", + params![child, hive_sh4re::SYSTEM_SENDER, format!("{PREFIX}%")], + |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)), + ) + .optional()?; + if let Some((row_id, old_body)) = existing { + // Preserve the original "from" label from the earlier notification. + let original_from = old_body + .strip_prefix(PREFIX) + .and_then(|rest| rest.split(SEPARATOR).next()) + .unwrap_or(old_label); + let new_body = format!("{PREFIX}{original_from}{SEPARATOR}{new_label}"); + conn.execute( + "UPDATE messages SET body = ?1, sent_at = ?2 WHERE id = ?3", + params![new_body, now, row_id], + )?; + drop(conn); + let _ = self.events.send(MessageEvent::Sent { + id: row_id, + from: hive_sh4re::SYSTEM_SENDER.to_owned(), + to: child.to_owned(), + body: new_body, + at: now, + in_reply_to: None, + }); + } else { + let body = format!("{PREFIX}{old_label}{SEPARATOR}{new_label}"); + conn.execute( + "INSERT INTO messages (sender, recipient, body, sent_at, in_reply_to) + VALUES (?1, ?2, ?3, ?4, NULL)", + params![hive_sh4re::SYSTEM_SENDER, child, body, now], + )?; + let row_id = conn.last_insert_rowid(); + drop(conn); + let _ = self.events.send(MessageEvent::Sent { + id: row_id, + from: hive_sh4re::SYSTEM_SENDER.to_owned(), + to: child.to_owned(), + body, + at: now, + in_reply_to: None, + }); + } + Ok(()) + } + /// Long-poll variant of `recv_batch`: returns immediately if any /// row is pending (popping up to `max`); otherwise waits up to /// `timeout` for the broker to emit a `Sent { to: recipient }` diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index f9d3f7d7..212d7ec3 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -616,12 +616,12 @@ impl Coordinator { in_reply_to: None, }); } - let _ = self.broker.send(&hive_sh4re::Message { - from: hive_sh4re::SYSTEM_SENDER.to_owned(), - to: child.to_owned(), - body: format!("your parent changed from {old_label} to {new_label}"), - in_reply_to: None, - }); + // Coalesce: if a prior move notification is already pending in + // the broker (agent was offline for multiple moves), update it + // in-place so the agent sees "A to C" not "A to B" then "B to C". + let _ = self + .broker + .send_coalescing_reparent(child, old_label, new_label); } // Rescan + diff-emit regardless of whether messages fired —