topology: drop manager-root special case + notify three agents on reparent (#743)

hive-c0re/src/coordinator.rs

@@ -509,6 +509,89 @@ impl Coordinator {
         }
     }
 
+    /// Apply a topology reparent + fan the resulting notifications
+    /// out to the three affected agents (#743). On success, drops a
+    /// one-line system message into the inbox of:
+    ///
+    /// 1. The **old parent** (if any) — `"{child} moved out of your
+    ///    subtree to {new_parent_or_root}"`.
+    /// 2. The **new parent** (if any) — `"{child} just moved into
+    ///    your subtree (was previously under
+    ///    {old_parent_or_root})"`.
+    /// 3. The **moved agent** — `"your parent changed from
+    ///    {old_parent_or_root} to {new_parent_or_root}"`.
+    ///
+    /// `_or_root` resolves to the literal string `"<root>"` when the
+    /// slot is `None`, keeping the wording consistent with the
+    /// `<parent>` sentinel's "root → operator" routing (#692). The
+    /// notifications fire as ordinary broker messages with
+    /// `from = hive_sh4re::SYSTEM_SENDER` so the dashboard renders
+    /// them under the existing system-source styling.
+    ///
+    /// Idempotent: if `topology::set_parent` skipped the disk write
+    /// (same-parent no-op), no messages fire. The `topology` module's
+    /// validation (`apply_set_parent`: unknown agent, cycle, etc.)
+    /// runs *before* any message is sent, so a refused move never
+    /// notifies anyone.
+    ///
+    /// Also drives a `rescan_containers_and_emit` after the write so
+    /// the dashboard tree re-renders without polling.
+    pub async fn reparent_with_notify(
+        self: &Arc<Self>,
+        child: &str,
+        new_parent: Option<&str>,
+    ) -> std::result::Result<(), String> {
+        // Snapshot the old parent BEFORE the write so the
+        // notifications can describe both sides of the change.
+        let topo_before = crate::topology::read();
+        let old_parent = topo_before.get(child).cloned().flatten();
+
+        // The disk write happens here; topology validation +
+        // idempotent fast-path inside `set_parent` may short-circuit
+        // (same parent → no-op). We mirror the same idempotent shape
+        // for the notifications: if nothing changed, send nothing.
+        crate::topology::set_parent(child, new_parent)?;
+
+        let changed = old_parent.as_deref() != new_parent;
+        if changed {
+            let old_label = old_parent.as_deref().unwrap_or("<root>");
+            let new_label = new_parent.unwrap_or("<root>");
+            // System-source notifications. Best-effort: a failed broker
+            // send is logged + ignored so a transient sqlite error
+            // doesn't bubble out and unwind the topology write.
+            if let Some(op) = old_parent.as_deref() {
+                let _ = self.broker.send(&hive_sh4re::Message {
+                    from: hive_sh4re::SYSTEM_SENDER.to_owned(),
+                    to: op.to_owned(),
+                    body: format!("{child} moved out of your subtree to {new_label}"),
+                    in_reply_to: None,
+                });
+            }
+            if let Some(np) = new_parent {
+                let _ = self.broker.send(&hive_sh4re::Message {
+                    from: hive_sh4re::SYSTEM_SENDER.to_owned(),
+                    to: np.to_owned(),
+                    body: format!(
+                        "{child} just moved into your subtree (was previously under {old_label})"
+                    ),
+                    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,
+            });
+        }
+
+        // Rescan + diff-emit regardless of whether messages fired —
+        // even an idempotent no-op might have happened concurrently
+        // with another lifecycle event the dashboard cares about.
+        self.rescan_containers_and_emit().await;
+        Ok(())
+    }
+
     /// Read-only snapshot of the last cached container view. Used by
     /// `/api/state` to cold-load page-open clients without re-running
     /// `nixos-container list` themselves; the