fix(maestro): poll awaiting-input until leaderless state holds

The pipeline leader exits as soon as both stages are forked, which can
beat the surviving stage opening the tty and parking on read — the one-
shot assertion then ran against a group with no parked member yet.
Poll for the answer, accepting it only while the foreground group is
still leaderless so a shell that regained the tty cannot satisfy it.

Fixes #113

crates/maestro-daemon/tests/activity.rs

@@ -189,8 +189,23 @@ fn a_foreground_pipeline_whose_leader_exited_is_still_awaiting_input() {
         "never observed a foreground group whose leader had exited"
     );
 
-    assert!(
-        awaiting_input(pane.pane_pid, &pane.tty, &ProcGroups::snapshot()),
-        "a surviving pipeline member still holds the tty"
-    );
+    // The leader can be gone while the surviving stage is still setting up:
+    // `true` exits as soon as both stages are forked, before the reader has
+    // opened the tty and parked. Poll for the answer, but only accept it
+    // while the state this test exists for still holds — otherwise a shell
+    // that regained the terminal would satisfy it vacuously.
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    let mut holds = false;
+    while std::time::Instant::now() < deadline {
+        let groups = ProcGroups::snapshot();
+        let still_leaderless = pgrp_and_tpgid(pane.pane_pid).is_some_and(|(pgrp, tpgid)| {
+            tpgid > 0 && tpgid != pgrp && pgrp_and_tpgid(tpgid).is_none()
+        });
+        if still_leaderless && awaiting_input(pane.pane_pid, &pane.tty, &groups) {
+            holds = true;
+            break;
+        }
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    assert!(holds, "a surviving pipeline member still holds the tty");
 }

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
@@ -848,6 +878,17 @@ async fn natural_exit_of_a_pristine_worktree_removes_it() {
     })
     .await;
 
+    // (`Dead` is observable a moment before the settle step removes the tree.)
+    wait_for_async(Duration::from_secs(3), || async {
+        !wt_path.exists()
+            && fx
+                .retained_store()
+                .load_all()
+                .map(|all| all.is_empty())
+                .unwrap_or(false)
+    })
+    .await;
+
     assert!(
         !wt_path.exists(),
         "a pristine worktree leaves nothing behind"