fix(audio): make module teardown cancellation-safe

Teardown used to `event_task.abort()` and then read three `Option<u32>`s.
The task's work was a synchronous `pactl` call with no await point, so the
abort could not land until the load had already returned: teardown saw
`None`, unloaded the sink, and the task then stored the new module's id into
a mutex nobody would read again. An orphan loopback, pointing at a sink that
no longer existed.

Three changes close it:

- Loads and unloads are `tokio::process::Command` with `kill_on_drop` and a
  bound, so cancellation is expressible at all. They are deliberately not
  `select!`ed against a cancel signal — dropping a completed load's index on
  the floor is the defect, not the fix. Cancellation happens by dropping the
  future, and the permit's `Drop` turns that into a question.
- `Routing::shutdown` is async and *awaits* the event task through
  `&mut JoinHandle`, falling back to abort-then-await. Dropping the handle
  would detach the task, which is how a load could still land after teardown
  believed it had finished. It then runs two reconcile-then-unload rounds:
  one round can raise exactly one new question, and a second settles it.
- `Drop` stays as the narrower synchronous backstop for the paths that never
  reach `shutdown`. It cannot await or reconcile, so when the ledger is left
  unexplained it says so and names `--repair`.

A load whose outcome cannot be observed is now distinguished from one the
server refused: a clean non-zero `pactl` exit abandons the permit (nothing
was created), while a signal death, a timeout, an unreadable index or
`PA_INVALID_INDEX` all leave it unsettled for reconciliation.

Two live gates, both A/B against the real module table: teardown leaves it
byte-identical with both modules carrying owner tokens, and a load cancelled
mid-flight is reconciled rather than orphaned. The second asserts the slot is
pending *before* reconciling, so it cannot pass by aborting before the load
ever began. Both mutate global state, so they need `--test-threads=1` —
running them in parallel makes each see the other's modules, which is how the
first run failed.

273 tests, clippy clean under `-D warnings`, `--doctor` all checks pass.

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

src/host/audio.rs

@@ -33,28 +33,41 @@
 use anyhow::{Context, Result, bail};
 use std::cell::RefCell;
 use std::collections::BTreeMap;
-use std::process::Command;
+use std::io::{self, Read};
+use std::process::{Child, Command, ExitStatus, Stdio};
 use std::rc::Rc;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
 
 use crate::cli::HostOpts;
-use crate::repair::plan::{self as repair_plan, Shape};
+use crate::host::ledger::{self, LedgerError, ModuleLedger, UnloadOutcome};
+use crate::repair::plan::{self as repair_plan, Fingerprint, Shape};
+
+/// How long a `pactl load-module` worker may run before it is killed and reaped.
+///
+/// A bound, not a calibration: a local Pulse socket answers in milliseconds, and
+/// this exists only so a wedged server cannot hang teardown forever. It is
+/// deliberately generous because exceeding it is no longer destructive: the
+/// ledger records the attempt, and reconciliation finds whatever the server
+/// actually did. Unloads use PulseSession's independently bounded native
+/// connect/list/unload requests instead of a second pactl connection.
+const PACTL_BUDGET: Duration = Duration::from_secs(5);
+const PACTL_REAP_BUDGET: Duration = Duration::from_secs(1);
 
 /// Owns the pactl-loaded modules plus, when filtering is active, the
 /// libpipewire stream-router thread. Drop unloads modules as a backstop;
-/// prefer [`Routing::shutdown`] explicitly so failures get logged.
+/// prefer [`Routing::shutdown`] explicitly, which is the only path that can
+/// reconcile a load whose outcome was never observed.
 pub struct Routing {
-    sink_module: Option<u32>,
-    /// Shared with the event task so it can `take()` and unload on the
-    /// first successful route. `Routing::shutdown` unloads whatever
-    /// remains.
-    loopback_module: Arc<Mutex<Option<u32>>>,
-    /// The `null-sink.monitor → @DEFAULT_SINK@` loopback that lets the sharer
-    /// hear the routed app. Shared with the event task, which loads it on the
-    /// first routed stream and unloads it when the app stops. `None` outside
-    /// app mode and whenever no app is currently routed.
-    local_monitor_module: Arc<Mutex<Option<u32>>>,
+    /// Every module this host has loaded, is loading, or must ask the server
+    /// about. Shared with the event task, which loads and unloads the two
+    /// loopbacks as the routed app comes and goes.
+    ///
+    /// This replaced three `Option<u32>`s. The reason is in [`crate::host::ledger`]:
+    /// an `Option` cannot say "a load is in flight", so a cancelled load looked
+    /// exactly like no load at all and its module was left behind.
+    ledger: Arc<ModuleLedger>,
     sink_name: String,
     stream_router: Option<StreamRouter>,
     event_task: Option<tokio::task::JoinHandle<()>>,
@@ -66,13 +79,28 @@ impl Routing {
     pub async fn start(opts: &HostOpts) -> Result<Self> {
         let pid = std::process::id();
         let sink_name = repair_plan::sink_name_for(pid);
+        let ledger = ModuleLedger::new();
+        // Construct the owner before the first mutation. Any error or cancellation
+        // below now drops a real `Routing`, whose backstop closes, quiesces,
+        // reconciles, and unloads this ledger. Previously the owner did not exist
+        // until both initial modules had loaded, so constructor failure leaked
+        // everything loaded up to that point.
+        let mut routing = Self {
+            ledger: Arc::clone(&ledger),
+            sink_name: sink_name.clone(),
+            stream_router: None,
+            event_task: None,
+        };
 
         // Every module this host loads carries an ownership token, minted per
         // load, so `--repair` can tell whose pid the name refers to instead of
         // assuming the number means the same thing everywhere. Without it a repair
         // run in another pid namespace can unload a live host's audio; see
-        // `repair::plan::OwnerToken`.
-        let sink_module = load_module(Shape::LegacyCaptureSink, pid)
+        // `repair::plan::OwnerToken`. That same per-load nonce is what lets
+        // reconciliation identify a module whose load was interrupted before its
+        // index was ever read.
+        load_module(&ledger, Shape::LegacyCaptureSink, pid)
+            .await
             .context("failed to load module-null-sink")?;
 
         // In strict per-app mode we never mirror the default sink: the viewer
@@ -84,51 +112,32 @@ impl Routing {
         // 20ms loopback latency keeps the mirrored audio tight; pactl's
         // default of 200ms is enough to be perceptible.
         let strict_app = opts.app.is_some() && opts.strict_audio;
-        let loopback_module = if strict_app {
-            None
-        } else {
-            Some(
-                load_module(Shape::LoopbackIntoCapture, pid)
-                    .context("failed to load module-loopback (null-sink cleaned up on Drop)")?,
-            )
-        };
+        if !strict_app {
+            load_module(&ledger, Shape::LoopbackIntoCapture, pid)
+                .await
+                .context("failed to load module-loopback (null-sink cleaned up on Drop)")?;
+        }
 
         tracing::info!(
-            sink_module,
-            ?loopback_module,
             strict_app,
             %sink_name,
             "audio routing: null-sink ready (loopback skipped in strict app mode)"
         );
 
-        let loopback_arc = Arc::new(Mutex::new(loopback_module));
-        let local_monitor_arc = Arc::new(Mutex::new(None));
-        let mut routing = Self {
-            sink_module: Some(sink_module),
-            loopback_module: Arc::clone(&loopback_arc),
-            local_monitor_module: Arc::clone(&local_monitor_arc),
-            sink_name: sink_name.clone(),
-            stream_router: None,
-            event_task: None,
-        };
-
         if let Some(app) = &opts.app {
             let (router, mut event_rx) = StreamRouter::spawn(app.clone(), sink_name.clone())?;
-            let loopback_for_task = Arc::clone(&loopback_arc);
-            let local_monitor_for_task = Arc::clone(&local_monitor_arc);
+            let ledger_for_task = Arc::clone(&ledger);
             let strict = opts.strict_audio;
             let event_task = tokio::spawn(async move {
                 use crate::common::output::{self, AppAudioState};
                 while let Some(ev) = event_rx.recv().await {
                     match ev {
                         Event::FirstRoutedStream => {
-                            let mid = loopback_for_task.lock().unwrap().take();
-                            if let Some(id) = mid {
                             tracing::info!(
                                 "audio routing: first stream routed → unloading default-sink loopback"
                             );
-                                unload_module(id);
-                            }
+                            let mirror_absent =
+                                unload_module(&ledger_for_task, Shape::LoopbackIntoCapture).await;
                             // Mirror the routed app back to the sharer's own
                             // speakers so they hear the content they're sharing.
                             // Loaded *after* the default-sink loopback is gone so
@@ -136,19 +145,14 @@ impl Routing {
                             // sourced from the null-sink monitor — the chosen app
                             // only, never the desktop/call — so it can't echo into
                             // the capture.
-                            if local_monitor_for_task.lock().unwrap().is_none() {
-                                match load_module(Shape::LoopbackOutOfCapture, pid) {
-                                    Ok(id) => {
-                                        tracing::info!(
-                                            module = id,
-                                            "audio routing: local monitor loaded (sharer hears the shared app)"
+                            if mirror_absent {
+                                ensure_loaded(&ledger_for_task, Shape::LoopbackOutOfCapture, pid)
+                                    .await;
+                            } else {
+                                tracing::warn!(
+                                    "audio routing: default-sink mirror absence was not confirmed; \
+                                     refusing to load the inverse local monitor"
                                 );
-                                        *local_monitor_for_task.lock().unwrap() = Some(id);
-                                    }
-                                    Err(e) => tracing::warn!(
-                                        "audio routing: failed to load local monitor loopback: {e:#}"
-                                    ),
-                                }
                             }
                             // Tell the front-end the chosen app's audio is live.
                             output::emit(output::Event::AppAudio {
@@ -164,13 +168,8 @@ impl Routing {
                             // The shared app is gone, so its null-sink is silent:
                             // stop mirroring it to the sharer's speakers. Re-loads
                             // on the next FirstRoutedStream if the app resumes.
-                            if let Some(id) = local_monitor_for_task.lock().unwrap().take() {
-                                tracing::info!(
-                                    module = id,
-                                    "audio routing: last routed stream gone → unloading local monitor"
-                                );
-                                unload_module(id);
-                            }
+                            let local_monitor_absent =
+                                unload_module(&ledger_for_task, Shape::LoopbackOutOfCapture).await;
                             if strict {
                                 // Strict mode: do NOT restore the whole-desktop
                                 // loopback. Viewers hear silence until the app
@@ -181,25 +180,21 @@ impl Routing {
                                 );
                                 continue;
                             }
-                            // Best-effort mode: restore the default-sink loopback
-                            // so the viewer hears system audio again instead of
-                            // silence.
-                            if loopback_for_task.lock().unwrap().is_some() {
+                            if !local_monitor_absent {
+                                tracing::warn!(
+                                    "audio routing: local-monitor absence was not confirmed; \
+                                     refusing to restore the inverse default-sink mirror"
+                                );
                                 continue;
                             }
+                            // Best-effort mode: restore the default-sink loopback
+                            // so the viewer hears system audio again instead of
+                            // silence. Already loaded is not an error — the ledger
+                            // refuses the load and `ensure_loaded` says so quietly.
                             tracing::info!(
                                 "audio routing: last routed stream gone → restoring default-sink loopback"
                             );
-                            match load_module(Shape::LoopbackIntoCapture, pid) {
-                                Ok(id) => {
-                                    *loopback_for_task.lock().unwrap() = Some(id);
-                                }
-                                Err(e) => {
-                                    tracing::warn!(
-                                        "audio routing: failed to re-load loopback: {e:#}"
-                                    );
-                                }
-                            }
+                            ensure_loaded(&ledger_for_task, Shape::LoopbackIntoCapture, pid).await;
                         }
                     }
                 }
@@ -225,42 +220,92 @@ impl Routing {
         &self.sink_name
     }
 
-    /// Stop the stream router (if any), then unload loopback (if still
-    /// loaded), then unload the null-sink. Order matters: PipeWire can
-    /// leave zombie links if you destroy a sink with active inputs.
+    /// Stop the stream router and the event task, settle anything the ledger is
+    /// unsure about, then unload every module in shape order — the loopbacks
+    /// before the sink they reference, because PipeWire can leave zombie links if
+    /// a sink is destroyed with active inputs.
     ///
-    /// Every step is a `take()`, so this is idempotent — `Drop` calls it again
-    /// as a backstop and the second run is a no-op.
-    fn cleanup(&mut self) {
+    /// The event task is **awaited, not merely aborted**. Aborting and walking
+    /// away is what left orphans behind: the task's load is an await point now,
+    /// so dropping its future marks the slot ambiguous rather than losing the
+    /// module — but only a path that then reconciles can actually clean it up.
+    /// `Drop` cannot await, which is why it is the narrower backstop.
+    pub async fn shutdown(mut self) {
+        // Closing is synchronous and happens first: after this point the event
+        // task cannot register another mutation even if it receives one last
+        // router event while shutdown is in progress.
+        self.ledger.close();
+        if let Some(router) = self.stream_router.take() {
+            // ⚠️ Still an unbounded join: a wedged PipeWire thread parks this
+            // task indefinitely. That is the pre-existing defect S3b exists for.
+            // Nothing here makes it worse, and the ledger is what will make
+            // bounding it safe when it lands.
+            router.shutdown();
+        }
+        if let Some(mut task) = self.event_task.take() {
+            // The router's exit drops the event senders, so the task normally
+            // ends by itself. Abort is the fallback, and it is awaited through
+            // `&mut JoinHandle` so the future is genuinely dropped — and with it
+            // any in-flight permit — before reconciliation reads the ledger.
+            // Dropping the handle instead would *detach* the task, which is how a
+            // load could still land after teardown believed it was finished.
+            if tokio::time::timeout(PACTL_BUDGET, &mut task).await.is_err() {
+                tracing::warn!(
+                    "audio routing: the event task did not finish within {PACTL_BUDGET:?}; \
+                     cancelling it"
+                );
+                task.abort();
+                let _ = task.await;
+            }
+        }
+
+        // A cancelled `spawn_blocking` await detaches its worker. Every worker is
+        // registered before it can be spawned, so this is a real ordering
+        // boundary: reconciliation cannot overtake late module creation/removal.
+        let ledger_for_wait = Arc::clone(&self.ledger);
+        if let Err(e) = tokio::task::spawn_blocking(move || {
+            ledger_for_wait.wait_for_operations();
+        })
+        .await
+        {
+            tracing::warn!("audio routing: module-operation wait task failed: {e}");
+        }
+        cleanup_modules(&self.ledger).await;
+
+        if !self.ledger.is_clean() {
+            tracing::warn!(
+                settled = self.ledger.is_settled(),
+                "audio routing: some audio modules could not be removed safely; \
+                 `pixelpass --repair` will clean up anything left behind"
+            );
+        }
+    }
+}
+
+impl Drop for Routing {
+    /// Synchronous backstop for the paths that never reach [`Routing::shutdown`]
+    /// — an error on the way up, or a panic. It cannot await, so it can neither
+    /// wait for the event task nor reconcile; it unloads what the ledger can name
+    /// and says so plainly when something is left unexplained.
+    ///
+    /// After a completed `shutdown` the ledger holds nothing and this does nothing.
+    fn drop(&mut self) {
+        self.ledger.close();
         if let Some(router) = self.stream_router.take() {
             router.shutdown();
         }
         if let Some(task) = self.event_task.take() {
             task.abort();
         }
-        if let Some(id) = self.loopback_module.lock().unwrap().take() {
-            unload_module(id);
+        self.ledger.close_and_wait();
+        cleanup_modules_blocking(&self.ledger);
+        if !self.ledger.is_clean() {
+            tracing::warn!(
+                settled = self.ledger.is_settled(),
+                "audio routing: torn down with modules that could not be removed safely; \
+                 run `pixelpass --repair` to clean up anything left behind"
+            );
         }
-        // Unload the local monitor before the null-sink it reads from, so the
-        // sink has no active loopback reader when it's destroyed.
-        if let Some(id) = self.local_monitor_module.lock().unwrap().take() {
-            unload_module(id);
-        }
-        if let Some(id) = self.sink_module.take() {
-            unload_module(id);
-        }
-    }
-
-    /// Consume the routing and tear it all down now. `Drop` is the backstop;
-    /// the real work lives in [`cleanup`](Self::cleanup).
-    pub fn shutdown(mut self) {
-        self.cleanup();
-    }
-}
-
-impl Drop for Routing {
-    fn drop(&mut self) {
-        self.cleanup();
     }
 }
 
@@ -374,53 +419,362 @@ fn owner_token(pid: u32) -> Result<repair_plan::OwnerToken> {
 /// `--repair`'s exact-form matcher and this loader are one source of truth. A
 /// latency or argument change that moved only one of them would leave repair
 /// silently unable to recognise the modules this build loads.
-fn load_module(shape: Shape, pid: u32) -> Result<u32> {
+async fn load_module(ledger: &Arc<ModuleLedger>, shape: Shape, pid: u32) -> Result<Fingerprint> {
     let owner = owner_token(pid).context("could not build an audio ownership token")?;
-    let output = Command::new("pactl")
+    let permit = ledger
+        .begin_load(shape, pid, owner.clone())
+        .map_err(anyhow::Error::new)
+        .with_context(|| format!("cannot load the {} module", shape.label()))?;
+    let args = shape.render_args(pid, Some(&owner));
+
+    // The affine permit moves into the blocking worker. Dropping this await does
+    // not cancel `spawn_blocking`; the worker remains registered, owns and reaps
+    // its child, and settles the slot before teardown's quiescence barrier opens.
+    tokio::task::spawn_blocking(move || -> Result<Fingerprint> {
+        let output = permit.with_server_operation(|| {
+            let mut command = Command::new("pactl");
+            command
                 .arg("load-module")
                 .arg(shape.module_name())
-        .args(shape.render_args(pid, Some(&owner)))
-        .output()
-        .context("failed to run pactl load-module")?;
-    if !output.status.success() {
-        bail!(
-            "pactl load-module failed: {}",
-            String::from_utf8_lossy(&output.stderr).trim()
-        );
+                .args(args);
+            bounded_output(&mut command, PACTL_BUDGET)
+        });
+        let output = match output {
+            Ok(output) => output,
+            Err(e) => {
+                // Whether the server was reached is unknown. Leaving the permit
+                // unsettled makes its Drop create a reconciliation question.
+                drop(permit);
+                return Err(e).context("failed to run pactl load-module");
             }
-    let id_str = String::from_utf8(output.stdout)
-        .context("pactl returned non-UTF-8")?
-        .trim()
-        .to_string();
-    // Genuinely 32-bit, unlike `object.serial`: this is a PulseAudio module
-    // index (`pa_module.index`, `uint32_t`), which `pactl unload-module` takes
-    // back verbatim. Do not widen it.
-    id_str
-        .parse::<u32>()
-        .with_context(|| format!("pactl returned unexpected module ID: {id_str:?}"))
+        };
+
+        if output.timed_out {
+            drop(permit);
+            bail!("pactl load-module did not finish within {PACTL_BUDGET:?}");
+        }
+        if !output.status.success() {
+            let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
+            if output.status.code().is_some() {
+                // pactl exited normally and reported the server's refusal.
+                permit.abandon();
+            } else {
+                drop(permit);
+            }
+            bail!("pactl load-module failed: {stderr}");
+        }
+
+        let id_str = String::from_utf8_lossy(&output.stdout).trim().to_string();
+        // Genuinely 32-bit: this is a Pulse module index, not object.serial.
+        let Ok(index) = id_str.parse::<u32>() else {
+            drop(permit);
+            bail!("pactl returned unexpected module ID: {id_str:?}");
+        };
+        let fp = permit.commit(index)?;
+        tracing::info!(
+            module = fp.id,
+            shape = shape.label(),
+            "audio routing: loaded pactl module"
+        );
+        Ok(fp)
+    })
+    .await
+    .context("the pactl load worker failed")?
 }
 
-fn unload_module(id: u32) {
-    let result = Command::new("pactl")
-        .arg("unload-module")
-        .arg(id.to_string())
-        .output();
-    match result {
-        Ok(output) if output.status.success() => {
-            tracing::info!(module = id, "audio routing: unloaded pactl module");
-        }
-        Ok(output) => {
-            tracing::warn!(
-                module = id,
-                stderr = %String::from_utf8_lossy(&output.stderr).trim(),
-                "audio routing: pactl unload-module exited non-zero"
-            );
+/// Load `shape` unless the slot already holds it.
+///
+/// A busy slot is not a failure on the oscillation path: `FirstRoutedStream` and
+/// `LastRoutedStreamGone` can both ask for a module that is already in the state
+/// they want, and the ledger is what decides that rather than a separate flag.
+async fn ensure_loaded(ledger: &Arc<ModuleLedger>, shape: Shape, pid: u32) {
+    let Err(e) = load_module(ledger, shape, pid).await else {
+        return;
+    };
+    match e.downcast_ref::<LedgerError>() {
+        Some(LedgerError::Busy { state, .. }) => tracing::debug!(
+            shape = shape.label(),
+            state,
+            "audio routing: nothing to load, the slot is already occupied"
+        ),
+        _ => tracing::warn!(
+            "audio routing: failed to load the {} module: {e:#}",
+            shape.label()
+        ),
     }
+}
+
+/// Unload whatever the ledger holds for `shape`, and record how it went.
+///
+/// A no-op for a slot holding nothing. An outcome that cannot be confirmed is
+/// recorded as uncertain rather than assumed done, so the module keeps being
+/// named until the server is asked about it.
+async fn unload_module(ledger: &Arc<ModuleLedger>, shape: Shape) -> bool {
+    unload_module_inner(ledger, shape, false).await
+}
+
+async fn unload_module_inner(ledger: &Arc<ModuleLedger>, shape: Shape, cleanup: bool) -> bool {
+    let begun = if cleanup {
+        ledger.begin_cleanup_unload(shape)
+    } else {
+        ledger.begin_unload(shape)
+    };
+    let permit = match begun {
+        Ok(Some(permit)) => permit,
+        Ok(None) => return true,
         Err(e) => {
             tracing::warn!(
-                module = id,
-                "audio routing: failed to run pactl unload-module: {e}"
+                shape = shape.label(),
+                "audio routing: refusing module unload: {e}"
             );
+            return false;
+        }
+    };
+    match tokio::task::spawn_blocking(move || finish_verified_unload(permit)).await {
+        Ok(confirmed_absent) => confirmed_absent,
+        Err(e) => {
+            // A panicking worker drops its affine permit and therefore leaves an
+            // unload reconciliation question behind.
+            tracing::warn!(
+                shape = shape.label(),
+                "audio routing: unload worker failed: {e}"
+            );
+            false
+        }
+    }
+}
+
+fn finish_verified_unload(permit: ledger::UnloadPermit) -> bool {
+    let fp = permit.fingerprint().clone();
+    let result = permit.with_server_operation(|| verified_unload(&fp));
+    match result {
+        Ok(()) => {
+            permit.finish(UnloadOutcome::Confirmed);
+            true
+        }
+        Err(e) => {
+            permit.finish(UnloadOutcome::Uncertain(format!("{e:#}")));
+            false
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum UnloadPresence {
+    Exact,
+    Absent,
+    Replaced,
+}
+
+fn unload_presence(fp: &Fingerprint, current: &[repair_plan::ModuleObservation]) -> UnloadPresence {
+    match current.iter().find(|module| module.id == fp.id) {
+        None => UnloadPresence::Absent,
+        Some(observed) if fp.still_matches(observed) => UnloadPresence::Exact,
+        Some(_) => UnloadPresence::Replaced,
+    }
+}
+
+/// Re-list and unload through one verified-local Pulse connection. The exact
+/// fingerprint is checked immediately before destruction; if the index vanished
+/// or was reused, our module is already absent and the replacement is left alone.
+fn verified_unload(fp: &Fingerprint) -> Result<()> {
+    let mut session = crate::repair::introspect::PulseSession::connect()
+        .context("could not connect to verify a module unload")?;
+    let current = session
+        .list_modules()
+        .context("could not list modules immediately before unload")?;
+    match unload_presence(fp, &current) {
+        UnloadPresence::Exact => {}
+        UnloadPresence::Absent => {
+            tracing::info!(
+                module = fp.id,
+                shape = fp.shape.label(),
+                "audio routing: tracked module was already absent"
+            );
+            return Ok(());
+        }
+        UnloadPresence::Replaced => {
+            tracing::warn!(
+                module = fp.id,
+                shape = fp.shape.label(),
+                "audio routing: module index was reused; leaving the replacement alone"
+            );
+            return Ok(());
+        }
+    }
+    session
+        .unload_module(fp.id)
+        .with_context(|| format!("the server did not confirm unloading module #{}", fp.id))?;
+    tracing::info!(
+        module = fp.id,
+        shape = fp.shape.label(),
+        "audio routing: unloaded verified Pulse module"
+    );
+    Ok(())
+}
+
+/// One bounded child result. On deadline the child is killed and synchronously
+/// reaped before this returns, so server reconciliation cannot overtake a late
+/// `pactl` request merely because its async waiter was cancelled.
+struct BoundedOutput {
+    status: ExitStatus,
+    stdout: Vec<u8>,
+    stderr: Vec<u8>,
+    timed_out: bool,
+}
+
+/// Ensures every early-return and panic after spawn kills and reaps the child.
+/// The normal path marks it reaped after `try_wait`/`wait` obtained the status.
+struct ReapedChild {
+    child: Child,
+    reaped: bool,
+}
+
+impl Drop for ReapedChild {
+    fn drop(&mut self) {
+        if self.reaped {
+            return;
+        }
+        let _ = self.child.kill();
+        match self.reap_within(PACTL_REAP_BUDGET) {
+            Ok(Some(_)) => {}
+            Ok(None) => tracing::warn!(
+                "audio routing: killed pactl child was not reaped within {PACTL_REAP_BUDGET:?}"
+            ),
+            Err(e) => tracing::warn!("audio routing: could not reap killed pactl child: {e}"),
+        }
+    }
+}
+
+impl ReapedChild {
+    fn reap_within(&mut self, budget: Duration) -> io::Result<Option<ExitStatus>> {
+        let deadline = Instant::now() + budget;
+        loop {
+            if let Some(status) = self.child.try_wait()? {
+                self.reaped = true;
+                return Ok(Some(status));
+            }
+            if Instant::now() >= deadline {
+                return Ok(None);
+            }
+            std::thread::sleep(Duration::from_millis(5));
+        }
+    }
+}
+
+fn bounded_output(command: &mut Command, budget: Duration) -> io::Result<BoundedOutput> {
+    command.stdout(Stdio::piped()).stderr(Stdio::piped());
+    let mut child = ReapedChild {
+        child: command.spawn()?,
+        reaped: false,
+    };
+    let stdout = child
+        .child
+        .stdout
+        .take()
+        .ok_or_else(|| io::Error::other("pactl stdout was not piped"))?;
+    let stderr = child
+        .child
+        .stderr
+        .take()
+        .ok_or_else(|| io::Error::other("pactl stderr was not piped"))?;
+    let stdout_reader = std::thread::Builder::new()
+        .name("pixelpass-pactl-stdout".to_string())
+        .spawn(move || {
+            let mut bytes = Vec::new();
+            let mut stdout = stdout;
+            stdout.read_to_end(&mut bytes).map(|_| bytes)
+        })?;
+    let stderr_reader = std::thread::Builder::new()
+        .name("pixelpass-pactl-stderr".to_string())
+        .spawn(move || {
+            let mut bytes = Vec::new();
+            let mut stderr = stderr;
+            stderr.read_to_end(&mut bytes).map(|_| bytes)
+        })?;
+
+    let deadline = Instant::now() + budget;
+    let (status, timed_out) = loop {
+        if let Some(status) = child.child.try_wait()? {
+            child.reaped = true;
+            break (status, false);
+        }
+        if Instant::now() >= deadline {
+            let _ = child.child.kill();
+            let Some(status) = child.reap_within(PACTL_REAP_BUDGET)? else {
+                return Err(io::Error::new(
+                    io::ErrorKind::TimedOut,
+                    format!("pactl did not exit within {PACTL_REAP_BUDGET:?} after SIGKILL"),
+                ));
+            };
+            break (status, true);
+        }
+        std::thread::sleep(Duration::from_millis(5));
+    };
+    let stdout = stdout_reader
+        .join()
+        .map_err(|_| io::Error::other("pactl stdout reader panicked"))??;
+    let stderr = stderr_reader
+        .join()
+        .map_err(|_| io::Error::other("pactl stderr reader panicked"))??;
+    Ok(BoundedOutput {
+        status,
+        stdout,
+        stderr,
+        timed_out,
+    })
+}
+
+/// Async teardown: keep reconciling and unloading while a pass changes state.
+/// This replaces the arbitrary two-round count. With loads closed, the state
+/// graph is monotonic except for `Ambiguous(Unload) -> Loaded -> Ambiguous` when
+/// the same unload remains uncertain; that produces an identical snapshot and
+/// stops here for `--repair` rather than spinning.
+async fn cleanup_modules(ledger: &Arc<ModuleLedger>) {
+    loop {
+        let before = ledger.snapshot();
+        if ledger.is_clean() {
+            break;
+        }
+        if let Err(e) = ledger::reconcile_pending(ledger).await {
+            tracing::warn!("audio routing: could not reconcile the module ledger: {e:#}");
+        }
+        for fp in ledger.loaded() {
+            unload_module_inner(ledger, fp.shape, true).await;
+        }
+        if ledger.is_clean() || ledger.snapshot() == before {
+            break;
+        }
+    }
+}
+
+/// Synchronous teardown backstop. PulseSession bounds every connect/list/unload
+/// request, and `close_and_wait` has already drained any registered pactl worker.
+fn cleanup_modules_blocking(ledger: &Arc<ModuleLedger>) {
+    loop {
+        let before = ledger.snapshot();
+        if ledger.is_clean() {
+            break;
+        }
+        if let Err(e) = ledger::reconcile_pending_blocking(ledger) {
+            tracing::warn!("audio routing: could not reconcile the module ledger: {e:#}");
+        }
+        for fp in ledger.loaded() {
+            let permit = match ledger.begin_cleanup_unload(fp.shape) {
+                Ok(Some(permit)) => permit,
+                Ok(None) => continue,
+                Err(e) => {
+                    tracing::warn!(
+                        shape = fp.shape.label(),
+                        "audio routing: refusing blocking module unload: {e}"
+                    );
+                    continue;
+                }
+            };
+            finish_verified_unload(permit);
+        }
+        if ledger.is_clean() || ledger.snapshot() == before {
+            break;
         }
     }
 }
@@ -718,6 +1072,175 @@ fn try_flush(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::host::ledger::SlotState;
+    use crate::repair::plan::{ModuleObservation, classify};
+
+    /// Whole-desktop routing: no app filter, so no PipeWire thread and no event
+    /// task — just the null-sink and its default-sink loopback.
+    fn whole_desktop_opts() -> HostOpts {
+        HostOpts {
+            window: false,
+            app: None,
+            strict_audio: false,
+            display_server: None,
+            quality: crate::cli::Quality::Auto,
+            bitrate: None,
+            framerate: None,
+            max_height: None,
+            no_hwencode: false,
+            max_viewers: None,
+            interactive: false,
+            relay: None,
+        }
+    }
+
+    /// The module table exactly as `--repair` observes it.
+    fn module_snapshot() -> Vec<(u32, String, String)> {
+        let mut session =
+            crate::repair::introspect::PulseSession::connect().expect("a local Pulse server");
+        session
+            .list_modules()
+            .expect("the server lists its modules")
+            .into_iter()
+            .map(|m| (m.id, m.name, m.args))
+            .collect()
+    }
+
+    #[test]
+    fn normal_unload_requires_the_full_fingerprint_not_only_the_index() {
+        fn observation(id: u32, pid: u32, nonce: u64) -> ModuleObservation {
+            let token = repair_plan::OwnerToken {
+                machine: "abc123".to_string(),
+                boot: "def456".to_string(),
+                pid_ns: 4_026_531_836,
+                nonce,
+            };
+            ModuleObservation::new(
+                id,
+                Shape::LoopbackIntoCapture.module_name(),
+                &repair_plan::recorded_argument(
+                    &Shape::LoopbackIntoCapture.render_args(pid, Some(&token)),
+                ),
+            )
+        }
+
+        let ours = observation(5, 42, 7);
+        let fp = classify(&ours).expect("the fixture is canonical");
+        assert_eq!(
+            unload_presence(&fp, std::slice::from_ref(&ours)),
+            UnloadPresence::Exact
+        );
+        assert_eq!(unload_presence(&fp, &[]), UnloadPresence::Absent);
+
+        // Same live index, but another perfectly canonical host module. This is
+        // the non-vacuous reuse case: an id-only normal unload would destroy it.
+        let replacement = observation(5, 99, 8);
+        assert_eq!(
+            unload_presence(&fp, std::slice::from_ref(&replacement)),
+            UnloadPresence::Replaced
+        );
+    }
+
+    /// A/B against the live graph: routing must leave the module table exactly
+    /// as it found it. The same shape as `--repair`'s field gate, because the
+    /// property is the same one — nothing of ours outlives the session.
+    #[tokio::test]
+    #[ignore = "loads real Pulse modules; run with --ignored --test-threads=1"]
+    async fn live_teardown_leaves_the_module_table_as_it_found_it() {
+        let before = module_snapshot();
+        let routing = Routing::start(&whole_desktop_opts())
+            .await
+            .expect("routing starts");
+
+        let during = module_snapshot();
+        let ours: Vec<_> = during
+            .iter()
+            .filter(|m| !before.iter().any(|b| b.0 == m.0))
+            .collect();
+        assert_eq!(
+            ours.len(),
+            2,
+            "the null-sink and its default-sink loopback must both be loaded"
+        );
+        for (id, name, args) in ours {
+            let fp = classify(&ModuleObservation::new(*id, name, args))
+                .expect("a module we loaded must match one of our canonical forms");
+            assert!(
+                fp.owner.is_some(),
+                "every module we load carries an owner token, or --repair cannot \
+                 attribute it to this host's pid namespace"
+            );
+        }
+
+        routing.shutdown().await;
+        assert_eq!(
+            module_snapshot(),
+            before,
+            "teardown must leave the module table byte-identical"
+        );
+    }
+
+    /// The orphan race, staged against a real server: a load cancelled while
+    /// `pactl` is in flight must still be findable and removable.
+    ///
+    /// Whether the server got as far as creating the module is genuinely racy,
+    /// and that is the point — the gate does not care which way it went, only
+    /// that the ledger can account for both. With the permit's `Drop` disarmed
+    /// and the module created, the final comparison fails.
+    #[tokio::test]
+    #[ignore = "loads real Pulse modules; run with --ignored --test-threads=1"]
+    async fn live_a_cancelled_load_is_reconciled_not_orphaned() {
+        let before = module_snapshot();
+        let ledger = ModuleLedger::new();
+        let pid = std::process::id();
+
+        let ledger_for_task = Arc::clone(&ledger);
+        let task = tokio::spawn(async move {
+            let _ = load_module(&ledger_for_task, Shape::LegacyCaptureSink, pid).await;
+        });
+        // Wait until the affine permit is registered. A single yield is not a
+        // scheduling guarantee and made the old version of this gate capable of
+        // aborting before the task had started.
+        tokio::time::timeout(PACTL_BUDGET, async {
+            while matches!(ledger.state(Shape::LegacyCaptureSink), SlotState::Vacant) {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("the load registers its operation");
+        task.abort();
+        let _ = task.await;
+
+        // Cancellation detaches `spawn_blocking`; closing plus this registered-
+        // operation wait is the ordering boundary that prevents reconciliation
+        // from overtaking the worker's late server mutation.
+        ledger.close();
+        let ledger_for_wait = Arc::clone(&ledger);
+        tokio::task::spawn_blocking(move || ledger_for_wait.wait_for_operations())
+            .await
+            .expect("the operation wait runs");
+
+        // The worker either committed a known module or left a reconciliation
+        // question. Both are correct; vacancy here would mean the live operation
+        // disappeared from the ledger.
+        assert_eq!(
+            ledger.pending().len() + ledger.loaded().len(),
+            1,
+            "the cancelled load must retain exactly one tracked outcome"
+        );
+
+        cleanup_modules(&ledger).await;
+
+        assert!(
+            ledger.is_clean(),
+            "every tracked module must be removed, not merely explained"
+        );
+        assert_eq!(
+            module_snapshot(),
+            before,
+            "a cancelled load must leave nothing behind"
+        );
+    }
 
     #[test]
     fn object_serial_parses_past_u32() {

src/host/mod.rs

@@ -2,6 +2,7 @@ pub mod aec;
 pub mod audio;
 pub mod audit;
 mod capture;
+pub mod ledger;
 mod observer;
 mod pipeline;
 mod quality;

src/host/pipeline.rs

@@ -47,7 +47,7 @@ impl CaptureHandle {
             let _ = child.start_kill();
         }
         if let Some(audio) = self.audio.take() {
-            audio.shutdown();
+            audio.shutdown().await;
         }
         if let Some(serve) = self.serve.take() {
             serve.shutdown().await;