fix(maestro): sync watcher re-emit test to real event flow

Two races. First, the watcher updates the snapshot and emits
SessionUpdated as two steps, so the drain could run between them: the
legitimate breakdown emission then arrived inside the quiet window.
Subscribe before the watcher starts and consume through the breakdown
event so the window begins after the last real change.

Second, the assertion treated any SessionUpdated as a failed
suppression, but the activity pass may legitimately announce an
Active→Idle transition in that window (ACTIVE_WINDOW is 3s), carrying
today's blocked_work unchanged. Only a changed indicator counts as a
re-emit now.

Fixes #113

crates/maestro-daemon/tests/session_manager.rs

@@ -397,6 +397,12 @@ async fn watcher_refreshes_the_blocked_work_indicator() {
     init_bare_repo(origin.path());
     push_all_to_origin(repo.path(), origin.path());
 
+    // Subscribe before the watcher starts so every pass's emission is
+    // captured — otherwise the breakdown event can land between the
+    // snapshot observation below and the drain, and the quiet window
+    // would see it as an illegitimate re-emit.
+    let mut events = fx.manager.events();
+
     fx.manager.start_liveness_watcher_with_blocked_work(
         Duration::from_millis(50),
         Duration::from_millis(100),
@@ -418,14 +424,38 @@ async fn watcher_refreshes_the_blocked_work_indicator() {
     })
     .await;
 
-    // A steady state must not re-emit every tick (#83 change-detection).
-    let mut events = fx.manager.events();
+    // Consume through the emission that carried the breakdown (#83
+    // change-detection): the quiet window below must start after the last
+    // legitimate change, not race it into the channel.
+    loop {
+        match tokio::time::timeout(Duration::from_secs(5), events.recv()).await {
+            Ok(Ok(CatalogEvent::SessionUpdated { session }))
+                if matches!(
+                    &session.blocked_work,
+                    Some(b) if b.uncommitted_files >= 1 && b.untracked_files >= 1
+                ) =>
+            {
+                break;
+            }
+            Ok(Ok(_)) => continue,
+            _ => panic!("never saw the SessionUpdated carrying the breakdown"),
+        }
+    }
     while events.try_recv().is_ok() {}
     tokio::time::sleep(Duration::from_millis(400)).await;
-    assert!(
-        !matches!(events.try_recv(), Ok(CatalogEvent::SessionUpdated { .. })),
-        "an unchanged worktree must not re-emit SessionUpdated"
-    );
+    // Only a *changed* indicator counts as a re-emit (#83 change-detection).
+    // The activity pass may still legitimately announce e.g. Active→Idle
+    // (its debounce is time-based, not worktree-based), and its payload
+    // carries today's blocked_work unchanged.
+    while let Ok(CatalogEvent::SessionUpdated { session }) = events.try_recv() {
+        assert!(
+            matches!(
+                &session.blocked_work,
+                Some(b) if b.uncommitted_files >= 1 && b.untracked_files >= 1
+            ),
+            "an unchanged worktree must not re-emit a changed indicator"
+        );
+    }
 }
 
 /// Cleanup on kill is unconditional once the guard passes (#79): even a spec