chore: roll up security migration

crates/tinyflows-module/src/run_manager/actor_recovery.rs

@@ -83,9 +83,14 @@ impl Drop for ActorRecoveryGuard {
         // environment, so even cancellation before its first poll releases
         // this entry without removing a replacement actor.
         let active_cleanup = ActiveCleanupGuard(Arc::clone(&self.active));
+        #[cfg(test)]
+        let run_id = self.run_id.clone();
         runtime.spawn(async move {
             let _active_cleanup = active_cleanup;
             recovery.reconcile().await;
+            drop(_active_cleanup);
+            #[cfg(test)]
+            super::actor_recovery_tests::notify_recovery_finished_for_test(&run_id);
         });
     }
 }

crates/tinyflows-module/src/run_manager_actor_recovery_tests.rs

@@ -31,7 +31,12 @@ async fn dropping_an_unpolled_actor_future_reconciles_only_its_live_run() {
     // Tokio can drop a spawned future before its first poll during shutdown.
     // The actor factory must have captured recovery ownership before it
     // returned this future, so the admitted row cannot be left running.
+    let (reconciled_tx, reconciled_rx) = oneshot::channel();
+    install_recovery_completion(&fixture.run_id, reconciled_tx);
     drop(actor);
+    reconciled_rx
+        .await
+        .expect("actor recovery task must finish");
 
     assert_reconciled(
         &fixture.catalog,
@@ -60,9 +65,14 @@ async fn aborting_an_admitted_actor_reconciles_its_running_row() {
     ));
     admitted_tx.send(()).unwrap();
     arrived_rx.await.unwrap();
+    let (reconciled_tx, reconciled_rx) = oneshot::channel();
+    install_recovery_completion(&fixture.run_id, reconciled_tx);
     task.abort();
     assert!(task.await.unwrap_err().is_cancelled());
     drop(release_tx);
+    reconciled_rx
+        .await
+        .expect("actor recovery task must finish");
 
     assert_reconciled(
         &fixture.catalog,
@@ -468,6 +478,29 @@ fn install_after_admission_gate(
 static AFTER_ADMISSION_GATES: LazyLock<Mutex<HashMap<String, AdmissionGate>>> =
     LazyLock::new(|| Mutex::new(HashMap::new()));
 
+#[cfg(test)]
+static RECOVERY_COMPLETIONS: LazyLock<Mutex<HashMap<String, oneshot::Sender<()>>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
+
+#[cfg(test)]
+fn install_recovery_completion(run_id: &str, finished: oneshot::Sender<()>) {
+    RECOVERY_COMPLETIONS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .insert(run_id.to_owned(), finished);
+}
+
+#[cfg(test)]
+pub(super) fn notify_recovery_finished_for_test(run_id: &str) {
+    if let Some(finished) = RECOVERY_COMPLETIONS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .remove(run_id)
+    {
+        let _ = finished.send(());
+    }
+}
+
 #[cfg(test)]
 pub(super) async fn pause_after_admission_for_test(run_id: &str) {
     let gate = AFTER_ADMISSION_GATES

crates/tinyflows-schedule/src/runtime/mod.rs

@@ -198,8 +198,9 @@ impl Scheduler {
             let config = self.config.clone();
             let cancellation = self.cancellation.subscribe();
             let permits = Arc::clone(&self.permits);
+            let job_id = claim.job.id.clone();
             self.tasks.spawn(async move {
-                run_claim_with_renewal(
+                let outcome = run_claim_with_renewal(
                     &ports,
                     &config,
                     claim,
@@ -208,8 +209,17 @@ impl Scheduler {
                     Some(permits),
                     cancellation,
                 )
-                .await
-                .map(|_| ())
+                .await?;
+                // The run record holds the result; a failed delivery has no
+                // other trace in the poll loop.
+                if !outcome.success {
+                    tracing::warn!(
+                        %job_id,
+                        delivery_status = ?outcome.delivery_status,
+                        "scheduled run did not succeed"
+                    );
+                }
+                Ok(())
             });
             count += 1;
         }