fix(daemon): probes survive a cancelled reconcile tick

The cache was written only after probe() returned, so a single probe
longer than the lane's 25 s budget was cancelled before depositing and
restarted forever. Each probe now runs as a detached task that writes
the cache itself; a cancelled tick loses nothing and the next one reads
the result.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

crates/arr-daemon/src/import.rs

@@ -76,10 +76,12 @@ pub struct ImportAction {
     /// Probe results by path, kept across ticks. The reconcile lane cancels
     /// the whole action after its 25 s budget while one probe alone may take
     /// up to 60 s, so without this a large multi-file torrent would restart
-    /// from the first file every tick and never finish. Transient state,
-    /// rebuilt by re-probing after a restart (§8); entries are dropped once
-    /// their grab settles.
-    probed: tokio::sync::Mutex<HashMap<PathBuf, ProbeOutcome>>,
+    /// from the first file every tick and never finish. Shared with the
+    /// detached probe tasks, which is what lets a probe outlive a cancelled
+    /// tick and still deposit its result. Transient state, rebuilt by
+    /// re-probing after a restart (§8); entries are dropped once their grab
+    /// settles.
+    probed: std::sync::Arc<tokio::sync::Mutex<HashMap<PathBuf, ProbeOutcome>>>,
 }
 
 /// A movie grab Transmission finished downloading, not yet imported.
@@ -101,13 +103,21 @@ impl ImportAction {
         Self {
             transmission,
             prober,
-            probed: tokio::sync::Mutex::new(HashMap::new()),
+            probed: std::sync::Arc::new(tokio::sync::Mutex::new(HashMap::new())),
         }
     }
 
     /// Probe every path, reusing results settled on earlier ticks, and
     /// return the readable video files.
     ///
+    /// Each probe runs as a detached task that writes the cache itself, so a
+    /// tick cancelled by the lane's 25 s budget mid-probe loses nothing: the
+    /// probe finishes in the background (its own 60 s limit still applies)
+    /// and the next tick reads the deposited result. A tick that re-requests
+    /// a path an orphaned probe is still on starts a second probe of the same
+    /// file — bounded overlap, identical result, chosen over tracking
+    /// in-flight probes.
+    ///
     /// Only facts about a file are cached; a prober failure — missing
     /// binary, timeout, unparseable output — is returned so the tick retries.
     async fn probe_all(
@@ -120,19 +130,22 @@ impl ImportAction {
             let outcome = if let Some(outcome) = cached {
                 outcome
             } else {
-                let outcome = match self.prober.probe(path.clone()).await {
-                    Ok(file) => ProbeOutcome::Media(Box::new(file)),
-                    Err(error) if error.is_about_the_file() => {
-                        tracing::debug!(path = %path.display(), %error, "not a video file, skipping");
-                        ProbeOutcome::NotMedia
-                    }
-                    Err(error) => return Err(error.into()),
-                };
-                self.probed
-                    .lock()
-                    .await
-                    .insert(path.clone(), outcome.clone());
-                outcome
+                let prober = self.prober.clone();
+                let cache = std::sync::Arc::clone(&self.probed);
+                let target = path.clone();
+                tokio::spawn(async move {
+                    let outcome = match prober.probe(target.clone()).await {
+                        Ok(file) => ProbeOutcome::Media(Box::new(file)),
+                        Err(error) if error.is_about_the_file() => {
+                            tracing::debug!(path = %target.display(), %error, "not a video file, skipping");
+                            ProbeOutcome::NotMedia
+                        }
+                        Err(error) => return Err(error),
+                    };
+                    cache.lock().await.insert(target, outcome.clone());
+                    Ok(outcome)
+                })
+                .await??
             };
             if let ProbeOutcome::Media(file) = outcome {
                 files.push(*file);
@@ -905,6 +918,52 @@ mod tests {
         );
     }
 
+    /// The probe outlives a cancelled tick: the detached task deposits its
+    /// result after the caller's future is dropped, and the next tick reads
+    /// it instead of restarting the same probe forever.
+    #[tokio::test]
+    async fn a_cancelled_probe_still_deposits_its_result() {
+        let dir = tempfile::tempdir().unwrap();
+        let media = dir.path().join("film.mkv");
+        std::fs::write(&media, b"bytes").unwrap();
+        let counter = dir.path().join("count");
+        let script = dir.path().join("ffprobe");
+        std::fs::write(
+            &script,
+            format!(
+                "#!/bin/sh\nsleep 0.3\necho x >> {}\ncat <<'PROBE_EOF'\n{HDR10_PROBE}\nPROBE_EOF\n",
+                counter.display()
+            ),
+        )
+        .unwrap();
+        let mut permissions = std::fs::metadata(&script).unwrap().permissions();
+        permissions.set_mode(0o755);
+        std::fs::set_permissions(&script, permissions).unwrap();
+        let action = ImportAction::new(
+            TransmissionClient::new("http://127.0.0.1:1").unwrap(),
+            Prober::new().with_binary(&script),
+        );
+        let paths = vec![media];
+
+        // The lane's budget, in miniature: the tick is cancelled mid-probe.
+        let cancelled = tokio::time::timeout(
+            std::time::Duration::from_millis(50),
+            action.probe_all(&paths),
+        )
+        .await;
+        assert!(cancelled.is_err());
+
+        // The detached probe finishes on its own and deposits the result.
+        tokio::time::sleep(std::time::Duration::from_millis(600)).await;
+        let files = action.probe_all(&paths).await.unwrap();
+        assert_eq!(files.len(), 1);
+        assert_eq!(
+            std::fs::read_to_string(&counter).unwrap().lines().count(),
+            1,
+            "the next tick reused the deposited result instead of re-probing"
+        );
+    }
+
     /// The `EXDEV` fallback path lands whole files via rename (§7.2).
     #[test]
     fn the_copy_fallback_lands_a_whole_file_and_cleans_up() {