Merge branch 'feature/managed-task-phase-status'

crates/irtt-client/src/managed/model.rs

@@ -155,6 +155,10 @@ pub struct ManagedTargetStatus {
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct ManagedStatus {
     pub lifecycle: ManagedLifecycle,
+    /// Whether the task durably observed an explicit stop request.
+    ///
+    /// This remains false until the task observes the request latch. Natural
+    /// stopping and terminality alone do not imply a stop request was observed.
     pub stop_requested: bool,
     pub applied_command_sequence: u64,
     pub desired_target_count: usize,
@@ -171,9 +175,21 @@ pub struct ManagedStatus {
     pub discarded_target_outcomes: u64,
     pub targets: Arc<[ManagedTargetStatus]>,
     pub recent_target_outcomes: Arc<[ManagedTargetOutcome]>,
+    /// Authoritative final accounting, present if and only if `lifecycle` is
+    /// [`ManagedLifecycle::Completed`] or [`ManagedLifecycle::Failed`].
+    ///
+    /// Shares the terminal task phase's outcome with [`ManagedEvent::Completed`]
+    /// or [`ManagedEvent::Failed`]. Abandoned deliberately has no outcome.
     pub final_outcome: Option<Arc<ManagedOutcome>>,
 }
 
+/// Receiving half of authoritative durable latest-state observation.
+///
+/// The current snapshot is immediately readable. Updates may coalesce, and the
+/// final snapshot remains available after the sender closes at task termination.
+/// Use [`ManagedEventSubscription`] for lossy presentation events.
+pub type ManagedStatusSubscription = tokio::sync::watch::Receiver<Arc<ManagedStatus>>;
+
 /// Lossy presentation event emitted by the managed task.
 #[allow(clippy::large_enum_variant)]
 #[derive(Clone, Debug, Eq, PartialEq)]

crates/irtt-client/src/managed/task.rs

@@ -29,10 +29,10 @@ use super::{
     classify_client_error, ManagedClientConfig, ManagedCommandAcknowledgement,
     ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError,
     ManagedDriverFailure, ManagedEndReason, ManagedEvent, ManagedEventSubscription,
-    ManagedLifecycle, ManagedOutcome, ManagedPacing, ManagedStatus, ManagedSubscribeError,
+    ManagedLifecycle, ManagedOutcome, ManagedPacing, ManagedStatus, ManagedStatusSubscription,
-    ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure, ManagedTargetFailureKind,
+    ManagedSubscribeError, ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetFailure,
-    ManagedTargetFailurePhase, ManagedTargetLifecycle, ManagedTargetOutcome, ManagedTargetStatus,
+    ManagedTargetFailureKind, ManagedTargetFailurePhase, ManagedTargetLifecycle,
-    TargetInstance,
+    ManagedTargetOutcome, ManagedTargetStatus, TargetInstance,
 };
 
 const TARGET_WORK_BUDGET: usize = 128;
@@ -101,8 +101,6 @@ struct StopSignal {
     requested: AtomicBool,
     update_admission: AtomicU8,
     wake: watch::Sender<()>,
-    acknowledged: AtomicBool,
-    acknowledgement: watch::Sender<()>,
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -162,12 +160,6 @@ impl StopSignal {
     fn is_requested(&self) -> bool {
         self.requested.load(Ordering::Acquire)
     }
-
-    fn acknowledge(&self) {
-        if !self.acknowledged.swap(true, Ordering::AcqRel) {
-            self.acknowledgement.send_replace(());
-        }
-    }
 }
 
 /// Cloneable control and observation capability for a managed task.
@@ -194,6 +186,19 @@ impl ManagedClientHandle {
         Arc::clone(&self.status.borrow())
     }
 
+    /// Subscribe to authoritative durable latest-state observation.
+    ///
+    /// The current snapshot is marked as seen when subscribing.
+    /// The current snapshot is immediately available through `borrow()`. Updates
+    /// may coalesce; this is not a lossless event log. After the task terminates,
+    /// the sender closes but the final snapshot remains readable. Use
+    /// [`Self::subscribe`] for the lossy presentation-event stream.
+    pub fn subscribe_status(&self) -> ManagedStatusSubscription {
+        let mut status = self.status.clone();
+        drop(status.borrow_and_update());
+        status
+    }
+
     /// Subscribe to future lossy presentation events.
     pub fn subscribe(&self) -> Result<ManagedEventSubscription, ManagedSubscribeError> {
         self.events
@@ -205,7 +210,7 @@ impl ManagedClientHandle {
     /// Request idempotent graceful stop and return a durable-status receipt.
     pub fn stop(&self) -> ManagedStopReceipt {
         self.stop.request();
-        ManagedStopReceipt::new(Arc::clone(&self.stop))
+        ManagedStopReceipt::new(self.subscribe_status())
     }
 
     /// Submit the complete desired target set, replacing any previous one.
@@ -303,17 +308,33 @@ enum ManagedCommand {
 }
 
 /// Receipt resolving once stop is durably observed or the task is terminal.
+///
+/// Resolves from authoritative status as soon as `stop_requested` is true or
+/// the lifecycle is Completed, Failed, or Abandoned. It need not wait for
+/// graceful cleanup after stop observation.
 #[must_use = "await the receipt to observe durable stop acknowledgement"]
 pub struct ManagedStopReceipt {
     future: Pin<Box<dyn Future<Output = ()> + Send>>,
 }
 
 impl ManagedStopReceipt {
-    fn new(stop: Arc<StopSignal>) -> Self {
+    fn new(mut status: ManagedStatusSubscription) -> Self {
-        let mut acknowledgement = stop.acknowledgement.subscribe();
         let future = Box::pin(async move {
-            while !stop.acknowledged.load(Ordering::Acquire) {
+            while !{
-                let _ = acknowledgement.changed().await;
+                let snapshot = status.borrow();
+                snapshot.stop_requested
+                    || matches!(
+                        snapshot.lifecycle,
+                        ManagedLifecycle::Completed
+                            | ManagedLifecycle::Failed
+                            | ManagedLifecycle::Abandoned
+                    )
+            } {
+                // A closed sender retains the terminal snapshot. Never wait
+                // indefinitely if the channel closes.
+                if status.changed().await.is_err() {
+                    break;
+                }
             }
         });
         Self { future }
@@ -688,12 +709,48 @@ struct TaskResources {
     stop: Arc<StopSignal>,
 }
 
-#[derive(Clone, Copy, Eq, PartialEq)]
+enum TaskPhase {
-enum DriverState {
     NotStarted,
     Running,
     Stopping,
-    Finished,
+    Completed { outcome: Arc<ManagedOutcome> },
+    Failed { outcome: Arc<ManagedOutcome> },
+    Abandoned,
+}
+
+impl TaskPhase {
+    fn lifecycle(&self) -> ManagedLifecycle {
+        match self {
+            Self::NotStarted => ManagedLifecycle::NotStarted,
+            Self::Running => ManagedLifecycle::Running,
+            Self::Stopping => ManagedLifecycle::Stopping,
+            Self::Completed { .. } => ManagedLifecycle::Completed,
+            Self::Failed { .. } => ManagedLifecycle::Failed,
+            Self::Abandoned => ManagedLifecycle::Abandoned,
+        }
+    }
+
+    fn final_outcome(&self) -> Option<Arc<ManagedOutcome>> {
+        match self {
+            Self::Completed { outcome } | Self::Failed { outcome } => Some(Arc::clone(outcome)),
+            _ => None,
+        }
+    }
+
+    fn is_terminal(&self) -> bool {
+        matches!(
+            self,
+            Self::Completed { .. } | Self::Failed { .. } | Self::Abandoned
+        )
+    }
+
+    fn is_running(&self) -> bool {
+        matches!(self, Self::Running)
+    }
+
+    fn is_stopping(&self) -> bool {
+        matches!(self, Self::Stopping)
+    }
 }
 
 #[derive(Clone, Copy)]
@@ -707,8 +764,7 @@ enum OpenSessionFailureCleanup {
 pub struct ManagedClientTask {
     #[cfg(test)]
     timeout_inspections: std::cell::RefCell<Vec<usize>>,
-    state: DriverState,
+    phase: TaskPhase,
-    lifecycle: ManagedLifecycle,
     config: ManagedClientConfig,
     targets: Vec<TargetRuntime>,
     commands: mpsc::Receiver<ManagedCommand>,
@@ -725,8 +781,8 @@ pub struct ManagedClientTask {
     stagger_remaining: usize,
     send_gate: Option<Instant>,
     last_stagger_send: Option<Instant>,
+    // Explicit stop observation is independent of natural stopping or terminality.
     stop_observed: bool,
-    final_outcome: Option<Arc<ManagedOutcome>>,
     next_generation: u64,
     applied_command_sequence: u64,
 }
@@ -779,7 +835,7 @@ impl ManagedClientTask {
             })
             .collect::<Vec<_>>();
         Arc::new(ManagedStatus {
-            lifecycle: self.lifecycle,
+            lifecycle: self.phase.lifecycle(),
             stop_requested: self.stop_observed,
             applied_command_sequence: self.applied_command_sequence,
             desired_target_count: self
@@ -800,7 +856,7 @@ impl ManagedClientTask {
             discarded_target_outcomes: self.history.discarded,
             targets: Arc::from(targets.into_boxed_slice()),
             recent_target_outcomes: self.history.recent(),
-            final_outcome: self.final_outcome.clone(),
+            final_outcome: self.phase.final_outcome(),
         })
     }
 
@@ -890,8 +946,7 @@ impl ManagedClientTask {
 
     fn begin_running(&mut self) -> Result<(), ManagedDriverFailure> {
         tokio::runtime::Handle::try_current().map_err(|_| ManagedDriverFailure::NoTokioRuntime)?;
-        self.state = DriverState::Running;
+        self.phase = TaskPhase::Running;
-        self.lifecycle = ManagedLifecycle::Running;
         self.replace_status();
         self.publish_event(ManagedEvent::Started);
         Ok(())
@@ -903,15 +958,12 @@ impl ManagedClientTask {
         }
         self.stop_observed = true;
         self.resources().stop.begin_stopping();
-        let lifecycle_transition =
+        let lifecycle_transition = !self.phase.is_stopping();
-            self.state != DriverState::Stopping || self.lifecycle != ManagedLifecycle::Stopping;
+        self.phase = TaskPhase::Stopping;
-        self.state = DriverState::Stopping;
-        self.lifecycle = ManagedLifecycle::Stopping;
         self.replace_status();
         if lifecycle_transition {
             self.publish_event(ManagedEvent::Stopping);
         }
-        self.resources().stop.acknowledge();
         self.scan_remaining = self.targets.len();
         self.burst_remaining = 0;
     }
@@ -934,7 +986,7 @@ impl ManagedClientTask {
                     // This check is the driver-side application linearization point.
                     // `apply_targets` cannot suspend, so a stop observed here wins over
                     // this command; once it passes, this command may complete atomically.
-                    let result = if self.state != DriverState::Running
+                    let result = if !self.phase.is_running()
                         || self.resources().stop.update_admission() != UpdateAdmission::Open
                     {
                         Err(ManagedCommandApplyError::Stopping)
@@ -1015,8 +1067,7 @@ impl ManagedClientTask {
             && self.config.completion == ManagedCompletionPolicy::FinishWhenQuiescent;
         if stopping {
             self.resources().stop.begin_stopping();
-            self.state = DriverState::Stopping;
+            self.phase = TaskPhase::Stopping;
-            self.lifecycle = ManagedLifecycle::Stopping;
         }
 
         // A transaction has one externally visible durable point: all runtime and
@@ -1180,7 +1231,7 @@ impl ManagedClientTask {
 
     fn poll_target(&mut self, index: usize, cx: &mut Context<'_>, now: Instant) -> bool {
         let retirement = self.effective_retirement(index);
-        let stopping = self.state == DriverState::Stopping || retirement.is_some();
+        let stopping = self.phase.is_stopping() || retirement.is_some();
         match self.targets[index].phase_mut() {
             TargetPhase::Pending { .. } => {
                 if stopping {
@@ -1956,7 +2007,7 @@ impl ManagedClientTask {
         self.timeout_inspections.borrow_mut().push(index);
         match self.targets[index].phase() {
             TargetPhase::Active { client, .. } => {
-                self.state != DriverState::Stopping
+                !self.phase.is_stopping()
                     && self.effective_retirement(index).is_none()
                     && client
                         .next_probe_timeout_deadline()
@@ -2090,7 +2141,7 @@ impl ManagedClientTask {
                 _ => {}
             }
         }
-        if self.state == DriverState::Running
+        if self.phase.is_running()
             && self.config.pacing == ManagedPacing::Staggered
             && self.active_count() > 0
         {
@@ -2160,11 +2211,14 @@ impl ManagedClientTask {
             self.history
                 .outcome(end_reason, self.applied_command_sequence),
         );
-        self.final_outcome = Some(Arc::clone(&outcome));
+        self.phase = if failed {
-        self.lifecycle = if failed {
+            TaskPhase::Failed {
-            ManagedLifecycle::Failed
+                outcome: Arc::clone(&outcome),
+            }
         } else {
-            ManagedLifecycle::Completed
+            TaskPhase::Completed {
+                outcome: Arc::clone(&outcome),
+            }
         };
         self.replace_status();
         self.publish_event(if failed {
@@ -2176,11 +2230,9 @@ impl ManagedClientTask {
                 outcome: Arc::clone(&outcome),
             }
         });
-        self.resources().stop.acknowledge();
         if let Some(resources) = self.resources.as_mut() {
             resources.events.take();
         }
-        self.state = DriverState::Finished;
         self.wake = None;
         self.timer = None;
         self.resources.take();
@@ -2228,11 +2280,11 @@ impl Future for ManagedClientTask {
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.as_mut().get_mut();
-        if this.state == DriverState::Finished {
+        if this.phase.is_terminal() {
             panic!("ManagedClientTask polled after completion");
         }
 
-        if this.state == DriverState::NotStarted {
+        if matches!(this.phase, TaskPhase::NotStarted) {
             if this.resources().stop.is_requested() {
                 this.observe_stop();
             } else if let Err(failure) = this.begin_running() {
@@ -2251,7 +2303,7 @@ impl Future for ManagedClientTask {
         immediate |= this.poll_timeout_pass(now);
         immediate |= this.poll_target_pass(cx, now);
         this.prune_undesired_terminal();
-        if this.state == DriverState::Running {
+        if this.phase.is_running() {
             immediate |= match this.config.pacing {
                 ManagedPacing::Staggered => this.poll_staggered_send(cx, now),
                 ManagedPacing::Burst => this.poll_burst_sends(cx, now),
@@ -2263,8 +2315,8 @@ impl Future for ManagedClientTask {
         }
 
         if this.all_targets_terminal() {
-            match this.state {
+            match this.phase {
-                DriverState::Stopping => {
+                TaskPhase::Stopping => {
                     return this.seal(
                         if this.stop_observed {
                             ManagedEndReason::StopRequested
@@ -2274,11 +2326,10 @@ impl Future for ManagedClientTask {
                         false,
                     );
                 }
-                DriverState::Running
+                TaskPhase::Running
                     if this.config.completion == ManagedCompletionPolicy::FinishWhenQuiescent =>
                 {
-                    this.state = DriverState::Stopping;
+                    this.phase = TaskPhase::Stopping;
-                    this.lifecycle = ManagedLifecycle::Stopping;
                     this.resources().stop.begin_stopping();
                     this.replace_status();
                     this.publish_event(ManagedEvent::Stopping);
@@ -2301,16 +2352,14 @@ impl Future for ManagedClientTask {
 
 impl Drop for ManagedClientTask {
     fn drop(&mut self) {
-        if self.state == DriverState::Finished || self.resources.is_none() {
+        if self.phase.is_terminal() || self.resources.is_none() {
             return;
         }
-        self.lifecycle = ManagedLifecycle::Abandoned;
+        self.phase = TaskPhase::Abandoned;
-        self.final_outcome = None;
         self.resources().stop.close_updates();
         self.commands.close();
         self.replace_status();
         self.publish_event(ManagedEvent::Abandoned);
-        self.resources().stop.acknowledge();
         if let Some(resources) = self.resources.as_mut() {
             resources.events.take();
         }
@@ -2394,13 +2443,10 @@ fn build_task(
     let (event_sender, _) = broadcast::channel(config.event_capacity);
     let weak_events = event_sender.downgrade();
     let (wake_sender, wake_receiver) = watch::channel(());
-    let (acknowledgement_sender, _) = watch::channel(());
     let stop = Arc::new(StopSignal {
         requested: AtomicBool::new(false),
         update_admission: AtomicU8::new(UpdateAdmission::Open as u8),
         wake: wake_sender,
-        acknowledged: AtomicBool::new(false),
-        acknowledgement: acknowledgement_sender,
     });
     let history = OutcomeHistory::new(config.outcome_history_limit);
     let initial_targets = runtimes
@@ -2414,8 +2460,9 @@ fn build_task(
             remote: None,
         })
         .collect::<Vec<_>>();
+    let phase = TaskPhase::NotStarted;
     let initial = Arc::new(ManagedStatus {
-        lifecycle: ManagedLifecycle::NotStarted,
+        lifecycle: phase.lifecycle(),
         stop_requested: false,
         applied_command_sequence: 0,
         desired_target_count: runtimes.len(),
@@ -2432,7 +2479,7 @@ fn build_task(
         discarded_target_outcomes: 0,
         targets: Arc::from(initial_targets.into_boxed_slice()),
         recent_target_outcomes: history.recent(),
-        final_outcome: None,
+        final_outcome: phase.final_outcome(),
     });
     let (status_sender, status_receiver) = watch::channel(initial);
     let (command_sender, command_receiver) = mpsc::channel(config.command_capacity);
@@ -2445,8 +2492,7 @@ fn build_task(
     let task = ManagedClientTask {
         #[cfg(test)]
         timeout_inspections: std::cell::RefCell::new(Vec::new()),
-        state: DriverState::NotStarted,
+        phase,
-        lifecycle: ManagedLifecycle::NotStarted,
         config,
         targets: runtimes,
         commands: command_receiver,
@@ -2464,7 +2510,6 @@ fn build_task(
         send_gate: None,
         last_stagger_send: None,
         stop_observed: false,
-        final_outcome: None,
         next_generation,
         applied_command_sequence: 0,
     };
@@ -2547,7 +2592,7 @@ mod tests {
                 let client = AsyncClient::connect("127.0.0.1:9", ClientConfig::default())
                     .await
                     .unwrap();
-                task.state = DriverState::Running;
+                task.phase = TaskPhase::Running;
                 task.install_target_phase(
                     0,
                     TargetPhase::Active {

crates/irtt-client/tests/managed_control.rs

@@ -20,6 +20,32 @@ fn config() -> ManagedClientConfig {
 #[test]
 fn stop_resolves_every_accepted_update_and_closes_admission() {
     let (task, handle) = ManagedClient::task(config(), vec![]).unwrap();
+    let mut subscription: ManagedStatusSubscription = handle.subscribe_status();
+    assert_eq!(
+        subscription.borrow().lifecycle,
+        ManagedLifecycle::NotStarted
+    );
+    assert!(!subscription.borrow().stop_requested);
+    assert!(subscription.borrow().final_outcome.is_none());
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let mut task = pin!(task);
+    {
+        let _runtime = runtime.enter();
+        assert!(task
+            .as_mut()
+            .poll(&mut Context::from_waker(Waker::noop()))
+            .is_pending());
+    }
+    assert_eq!(handle.status().lifecycle, ManagedLifecycle::Running);
+    subscription = handle.subscribe_status();
+    assert_eq!(subscription.borrow().lifecycle, ManagedLifecycle::Running);
+    assert!(pin!(subscription.changed())
+        .as_mut()
+        .poll(&mut Context::from_waker(Waker::noop()))
+        .is_pending());
     let receipts: Vec<_> = (0..3)
         .map(|index| {
             handle
@@ -35,15 +61,16 @@ fn stop_resolves_every_accepted_update_and_closes_admission() {
         Err(ManagedCommandError::QueueFull)
     ));
 
-    let stop = handle.stop();
+    let mut stop = pin!(handle.stop());
+    assert!(stop
+        .as_mut()
+        .poll(&mut Context::from_waker(Waker::noop()))
+        .is_pending());
+    assert!(!handle.status().stop_requested);
     assert!(matches!(
         handle.update_targets(vec![]),
         Err(ManagedCommandError::Stopping)
     ));
-    let runtime = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()
-        .unwrap();
     runtime.block_on(async {
         tokio::time::timeout(Duration::from_secs(2), async {
             let outcome = task.await;
@@ -61,6 +88,16 @@ fn stop_resolves_every_accepted_update_and_closes_admission() {
             assert_eq!(status.lifecycle, ManagedLifecycle::Completed);
             assert!(status.stop_requested);
             assert_eq!(status.final_outcome.as_deref(), Some(&outcome));
+            subscription.changed().await.unwrap();
+            assert!(std::sync::Arc::ptr_eq(
+                &subscription.borrow_and_update(),
+                &status
+            ));
+            assert!(subscription.changed().await.is_err());
+            assert_eq!(
+                subscription.borrow().final_outcome.as_deref(),
+                Some(&outcome)
+            );
             assert!(matches!(
                 handle.update_targets(vec![]),
                 Err(ManagedCommandError::DriverClosed)
@@ -103,12 +140,14 @@ fn failed_driver_resolves_every_accepted_update_with_its_terminal_failure() {
     }
     let status = handle.status();
     assert_eq!(status.lifecycle, ManagedLifecycle::Failed);
+    assert!(!status.stop_requested);
     assert_eq!(status.final_outcome.as_deref(), Some(&outcome));
     assert!(matches!(
         handle.update_targets(vec![]),
         Err(ManagedCommandError::DriverClosed)
     ));
     assert!(pin!(handle.stop()).as_mut().poll(&mut context).is_ready());
+    assert!(!handle.status().stop_requested);
 }
 
 #[test]

crates/irtt-integration-tests/tests/managed_behavior.rs

@@ -238,6 +238,10 @@ async fn target_updates_preserve_identical_sessions_and_retire_removed_generatio
                     let status = handle.status();
                     assert_eq!(status.lifecycle, ManagedLifecycle::Completed);
                     assert_eq!(status.final_outcome.as_ref(), Some(&outcome));
+                    assert!(Arc::ptr_eq(
+                        status.final_outcome.as_ref().unwrap(),
+                        &outcome
+                    ));
                     break;
                 }
                 _ => {}

crates/irtt-integration-tests/tests/managed_lifecycle.rs

@@ -238,23 +238,50 @@ fn quiescent_rejects_empty_initial_targets() {
 
 #[test]
 fn explicit_empty_waits_for_stop() {
+    let server = start_server(ServerBehavior::Echo, None);
     let mut config = config(ManagedPacing::Staggered);
     config.completion = ManagedCompletionPolicy::ExplicitStop;
+    config.client.request.duration = None;
     let (task, handle) = ManagedClient::task(config, vec![]).unwrap();
-    let outcome = runtime().block_on(async move {
+    let mut events = handle.subscribe().unwrap();
-        let mut task = Box::pin(task);
+    let outcome = runtime().block_on(async {
-        assert!(
+        tokio::time::timeout(Duration::from_secs(2), async {
-            tokio::time::timeout(Duration::from_millis(20), task.as_mut())
+            let mut task = Box::pin(task);
-                .await
+            assert!(poll_once(task.as_mut()).is_pending());
-                .is_err()
+            assert_eq!(handle.status().lifecycle, ManagedLifecycle::Running);
-        );
+            let update = handle
-        assert_eq!(handle.status().lifecycle, ManagedLifecycle::Running);
+                .update_targets(vec![target("one", server.addr)])
-        let receipt = handle.stop();
+                .unwrap();
-        let outcome = task.await;
+            loop {
-        receipt.await;
+                tokio::select! {
-        outcome
+                    _ = task.as_mut() => panic!("explicit-stop task completed before stop"),
+                    event = events.recv() => {
+                        if matches!(event.unwrap(), ManagedEvent::Client {
+                            event: irtt_client::ClientEvent::EchoReply { .. }, ..
+                        }) {
+                            break;
+                        }
+                    }
+                }
+            }
+            update.await.unwrap();
+            let mut receipt = Box::pin(handle.stop());
+            assert!(poll_once(receipt.as_mut()).is_pending());
+            assert!(poll_once(task.as_mut()).is_pending());
+            let status = handle.status();
+            assert_eq!(status.lifecycle, ManagedLifecycle::Stopping);
+            assert!(status.stop_requested);
+            assert!(status.final_outcome.is_none());
+            // The driver is paused during graceful cleanup: receipt readiness must
+            // come from durable stop observation, not eventual task completion.
+            assert!(poll_once(receipt.as_mut()).is_ready());
+            task.await
+        })
+        .await
+        .expect("stop observation and cleanup must complete")
     });
     assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested);
+    assert!(has_close(&server.finish()));
 }
 
 #[test]
@@ -311,6 +338,8 @@ fn pre_poll_stop() {
     )
     .unwrap();
     let mut receipt = Box::pin(handle.stop());
+    assert!(poll_once(receipt.as_mut()).is_pending());
+    assert!(!handle.status().stop_requested);
     let mut task = Box::pin(task);
     let Poll::Ready(outcome) = poll_once(task.as_mut()) else {
         panic!("pre-stopped task did not complete")
@@ -469,6 +498,10 @@ fn event_loss_independence() {
     assert_eq!(outcome.successful_target_outcomes, 1);
     let status = handle.status();
     assert_eq!(status.lifecycle, ManagedLifecycle::Completed);
+    assert!(!status.stop_requested);
+    let mut receipt = Box::pin(handle.stop());
+    assert!(poll_once(receipt.as_mut()).is_ready());
+    assert!(Arc::ptr_eq(&status, &handle.status()));
     assert_eq!(status.targets.len(), 1);
     let target = &status.targets[0];
     assert!(target.desired);
@@ -511,16 +544,32 @@ fn dropping_all_handles_does_not_stop() {
 
 #[test]
 fn task_drop_abandonment() {
-    let (task, handle) = ManagedClient::task(
+    // A latched but unobserved stop must not turn abandonment into completion.
-        config(ManagedPacing::Staggered),
+    for request_stop in [false, true] {
-        vec![target("one", "127.0.0.1:9".parse().unwrap())],
+        let (task, handle) = ManagedClient::task(
-    )
+            config(ManagedPacing::Staggered),
-    .unwrap();
+            vec![target("one", "127.0.0.1:9".parse().unwrap())],
-    let mut events = handle.subscribe().unwrap();
+        )
-    drop(task);
+        .unwrap();
-    assert_eq!(handle.status().lifecycle, ManagedLifecycle::Abandoned);
+        let mut events = handle.subscribe().unwrap();
-    assert_eq!(handle.status().total_target_outcomes, 0);
+        let mut receipt = request_stop.then(|| Box::pin(handle.stop()));
-    assert!(matches!(events.try_recv(), Ok(ManagedEvent::Abandoned)));
+        if let Some(receipt) = &mut receipt {
+            assert!(poll_once(receipt.as_mut()).is_pending());
+        }
+        drop(task);
+        assert_eq!(handle.status().lifecycle, ManagedLifecycle::Abandoned);
+        assert_eq!(handle.status().total_target_outcomes, 0);
+        assert!(!handle.status().stop_requested);
+        assert!(handle.status().final_outcome.is_none());
+        assert!(matches!(events.try_recv(), Ok(ManagedEvent::Abandoned)));
+        if let Some(receipt) = &mut receipt {
+            assert!(poll_once(receipt.as_mut()).is_ready());
+        }
+        let mut late_stop = Box::pin(handle.stop());
+        assert!(poll_once(late_stop.as_mut()).is_ready());
+        assert_eq!(handle.status().lifecycle, ManagedLifecycle::Abandoned);
+        assert!(!handle.status().stop_requested);
+    }
 }
 
 #[test]