fix(sidecar): harden telemetry lifecycle state

datadog-sidecar/src/entry.rs

@@ -113,7 +113,7 @@ where
     // Initialize telemetry sender synchronously before spawning the receiver task
     // This ensures the sender is available immediately, avoiding race conditions
     // where FFI calls might try to send telemetry before the receiver task starts
-    if let Some(rx) = init_telemetry_sender() {
+    if let Some(rx) = init_telemetry_sender(&server) {
         tokio::spawn(telemetry_action_receiver_task(server.clone(), rx));
     }
 

datadog-sidecar/src/service/session_info.rs

@@ -98,12 +98,9 @@ impl SessionInfo {
     }
 
     /// Shuts down all running instances in the session.
+    #[cfg(test)]
     pub(crate) async fn shutdown_running_instances(&self) {
-        let runtimes: Vec<RuntimeInfo> = self
-            .lock_runtimes()
-            .drain()
-            .map(|(_, instance)| instance)
-            .collect();
+        let runtimes = self.take_running_instances();
 
         let instances_shutting_down: Vec<_> = runtimes
             .into_iter()
@@ -113,11 +110,28 @@ impl SessionInfo {
         future::join_all(instances_shutting_down).await;
     }
 
+    #[cfg(test)]
+    pub(crate) fn take_running_instances(&self) -> Vec<RuntimeInfo> {
+        self.lock_runtimes()
+            .drain()
+            .map(|(_, instance)| instance)
+            .collect()
+    }
+
+    pub(crate) fn take_runtime(&self, runtime_id: &str) -> Option<RuntimeInfo> {
+        self.lock_runtimes().remove(runtime_id)
+    }
+
+    pub(crate) fn find_runtime(&self, runtime_id: &str) -> Option<RuntimeInfo> {
+        self.lock_runtimes().get(runtime_id).cloned()
+    }
+
     /// Shuts down a specific runtime in the session.
     ///
     /// # Arguments
     ///
     /// * `runtime_id` - The ID of the runtime.
+    #[cfg(test)]
     pub(crate) async fn shutdown_runtime(&self, runtime_id: &str) {
         let maybe_runtime = {
             let mut runtimes = self.lock_runtimes();