From 9de95b303c00579230e978f2a96f92bac08559e4 Mon Sep 17 00:00:00 2001
From: Nils Andreas Svee <me@lochnair.net>
Date: Sun, 9 Aug 2026 20:21:12 +0700
Subject: [PATCH] feat(managed): support subscribed blocking start

---
 crates/irtt-client/src/managed/blocking.rs    | 31 ++++++++++++++++--
 .../irtt-client/src/managed/blocking_tests.rs | 32 ++++++++++++++++++-
 crates/irtt-client/src/managed/model.rs       |  3 ++
 3 files changed, 63 insertions(+), 3 deletions(-)

diff --git a/crates/irtt-client/src/managed/blocking.rs b/crates/irtt-client/src/managed/blocking.rs
index 15fbeed..4febf65 100644
--- a/crates/irtt-client/src/managed/blocking.rs
+++ b/crates/irtt-client/src/managed/blocking.rs
@@ -7,8 +7,8 @@ use thiserror::Error;
 use tokio::runtime::{Builder, Runtime};
 
 use super::{
-    ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedConfigError, ManagedOutcome,
-    ManagedTargetConfig,
+    ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask, ManagedConfigError,
+    ManagedEventSubscription, ManagedOutcome, ManagedSubscribeError, ManagedTargetConfig,
 };
 
 const BLOCKING_RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(100);
@@ -56,6 +56,10 @@ pub enum BlockingManagedStartError {
     #[error(transparent)]
     Config(#[from] ManagedConfigError),
 
+    /// The managed event subscription could not be created before startup.
+    #[error(transparent)]
+    Subscribe(#[from] ManagedSubscribeError),
+
     /// The dedicated Tokio runtime could not be created.
     #[error("failed to create blocking managed Tokio runtime: {source}")]
     Runtime {
@@ -98,6 +102,29 @@ impl BlockingManagedClient {
         targets: Vec<ManagedTargetConfig>,
     ) -> Result<Self, BlockingManagedStartError> {
         let (task, handle) = ManagedClient::task(config, targets)?;
+        Self::start_task(task, handle)
+    }
+
+    /// Validate configuration, subscribe to presentation events, and start a
+    /// dedicated managed worker.
+    ///
+    /// The returned subscription exists before the managed task receives its
+    /// first poll, so it cannot miss startup events because of subscription
+    /// ordering.
+    pub fn start_with_subscription(
+        config: ManagedClientConfig,
+        targets: Vec<ManagedTargetConfig>,
+    ) -> Result<(Self, ManagedEventSubscription), BlockingManagedStartError> {
+        let (task, handle) = ManagedClient::task(config, targets)?;
+        let subscription = handle.subscribe()?;
+        let owner = Self::start_task(task, handle)?;
+        Ok((owner, subscription))
+    }
+
+    fn start_task(
+        task: ManagedClientTask,
+        handle: ManagedClientHandle,
+    ) -> Result<Self, BlockingManagedStartError> {
         let runtime = Builder::new_current_thread()
             .enable_io()
             .enable_time()
diff --git a/crates/irtt-client/src/managed/blocking_tests.rs b/crates/irtt-client/src/managed/blocking_tests.rs
index 96cf767..e3fe66d 100644
--- a/crates/irtt-client/src/managed/blocking_tests.rs
+++ b/crates/irtt-client/src/managed/blocking_tests.rs
@@ -14,7 +14,8 @@ use tokio::{runtime::Builder, time::timeout};
 use super::{
     blocking::{join_worker, WorkerRuntime},
     BlockingManagedClient, BlockingManagedJoinError, ManagedClientConfig, ManagedCompletionPolicy,
-    ManagedEndReason, ManagedLifecycle, ManagedPacing, ManagedTargetConfig, ManagedTargetEndReason,
+    ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedPacing, ManagedTargetConfig,
+    ManagedTargetEndReason,
 };
 use crate::ClientConfig;
 
@@ -183,6 +184,35 @@ fn starts_without_caller_runtime_and_finishes_naturally() {
     server.finish();
 }
 
+#[test]
+fn subscribed_start_captures_events_from_a_short_run() {
+    let server = EchoServer::start();
+    let (owner, mut events) = BlockingManagedClient::start_with_subscription(
+        config(ManagedCompletionPolicy::FinishWhenQuiescent),
+        vec![target(&server)],
+    )
+    .unwrap();
+
+    server.wait_probe();
+    let outcome = owner.join().unwrap();
+    assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete);
+
+    let mut saw_started = false;
+    let mut saw_finished = false;
+    let mut saw_completed = false;
+    while let Ok(event) = events.try_recv() {
+        match event {
+            ManagedEvent::Started => saw_started = true,
+            ManagedEvent::TargetFinished { .. } => saw_finished = true,
+            ManagedEvent::Completed { .. } => saw_completed = true,
+            _ => {}
+        }
+    }
+    assert!(saw_started && saw_finished && saw_completed);
+    server.wait_close();
+    server.finish();
+}
+
 #[test]
 fn external_handle_stop_ends_with_graceful_close() {
     let server = EchoServer::start();
diff --git a/crates/irtt-client/src/managed/model.rs b/crates/irtt-client/src/managed/model.rs
index da785de..f972c2d 100644
--- a/crates/irtt-client/src/managed/model.rs
+++ b/crates/irtt-client/src/managed/model.rs
@@ -173,6 +173,9 @@ pub enum ManagedEvent {
 /// Receiving half of the lossy managed event stream.
 pub type ManagedEventSubscription = tokio::sync::broadcast::Receiver<ManagedEvent>;
 
+/// Immediate result of attempting to receive a managed presentation event.
+pub type ManagedEventTryRecvError = tokio::sync::broadcast::error::TryRecvError;
+
 /// Final authoritative outcome returned by the task.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct ManagedOutcome {