fix(audio): close the teardown orphan race by wiring in the ledger

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>

.gitignore

@@ -1 +1,7 @@
 /target
+
+# Nix: the symlink `nix build` drops, and direnv's local cache. flake.nix and
+# flake.lock ARE tracked — the lock is what pins the toolchain.
+/result
+/result-*
+/.direnv/

flake.lock

@@ -0,0 +1,48 @@
+{
+  "nodes": {
+    "nixpkgs": {
+      "locked": {
+        "lastModified": 1785989512,
+        "narHash": "sha256-HFQhkQcl5D1hUNoen3SGHCSFCt2Bg6uP+HgbrnA3InQ=",
+        "owner": "nixos",
+        "repo": "nixpkgs",
+        "rev": "445d861c6d31b4af0c79d8d4be2331f762a361d7",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nixos",
+        "ref": "nixos-26.05",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
+    "root": {
+      "inputs": {
+        "nixpkgs": "nixpkgs",
+        "rust-overlay": "rust-overlay"
+      }
+    },
+    "rust-overlay": {
+      "inputs": {
+        "nixpkgs": [
+          "nixpkgs"
+        ]
+      },
+      "locked": {
+        "lastModified": 1786076960,
+        "narHash": "sha256-jfR6OhwurCKn1tREyfOcK/Omxf1Q/DzDDFbnEr1mBLs=",
+        "owner": "oxalica",
+        "repo": "rust-overlay",
+        "rev": "57a23bfaf4f7017267294b161175db1e32eb1c85",
+        "type": "github"
+      },
+      "original": {
+        "owner": "oxalica",
+        "repo": "rust-overlay",
+        "type": "github"
+      }
+    }
+  },
+  "root": "root",
+  "version": 7
+}

flake.nix

@@ -0,0 +1,147 @@
+{
+  description = "PixelPass — P2P screen sharing CLI over iroh";
+
+  inputs = {
+    # Same channel the hosts run (nixos-config tracks nixos-26.05). The capture
+    # path talks to the live PipeWire daemon and the system PulseAudio server,
+    # so the client libraries here should come from the same release the server
+    # did.
+    nixpkgs.url = "github:nixos/nixpkgs/nixos-26.05";
+
+    # The Rust toolchain is pinned SEPARATELY from the system libraries, so a
+    # nixpkgs bump cannot move the compiler under the lint gate. nixpkgs 26.05
+    # ships 1.95.0; this crate was developed and verified on 1.97.1, and
+    # peerspeak — the sibling project this one is built against — has a clippy
+    # lint that differs between exactly those two versions. Keeping both repos
+    # on one pinned compiler means a check that passes here passes there.
+    #
+    # This is the reproducible alternative to rustup: the same exact-version
+    # control, but recorded in flake.lock, so a fresh clone resolves the
+    # identical toolchain rather than whatever rustup fetches that day.
+    rust-overlay = {
+      url = "github:oxalica/rust-overlay";
+      inputs.nixpkgs.follows = "nixpkgs";
+    };
+  };
+
+  outputs =
+    { nixpkgs, rust-overlay, ... }:
+    let
+      system = "x86_64-linux";
+      pkgs = import nixpkgs {
+        inherit system;
+        overlays = [ rust-overlay.overlays.default ];
+      };
+
+      # Matches what CachyOS shipped (rust 1:1.97.1-1) and what peerspeak pins.
+      # `default` is the rustup "default" profile — rustc, cargo, rust-std,
+      # rustfmt and clippy — so those are NOT listed separately below. No
+      # windows-gnu target here: pixelpass is Linux-only (portal/PipeWire/X11
+      # capture), unlike peerspeak which has a Windows port.
+      rustToolchain = pkgs.rust-bin.stable."1.97.1".default;
+
+      # GStreamer is driven as a SUBPROCESS (gst-launch-1.0 / gst-inspect-1.0),
+      # not linked — there is no gstreamer-sys in Cargo.lock. So these are PATH
+      # dependencies at runtime rather than build inputs, and `deps.rs` refuses
+      # to start a share if any are missing.
+      gstPlugins = with pkgs; [
+        gst_all_1.gstreamer # gst-launch-1.0 / gst-inspect-1.0
+        gst_all_1.gst-plugins-base # videoscale (quality-preset downscale)
+        gst_all_1.gst-plugins-good # pulsesrc, ximagesrc
+        gst_all_1.gst-plugins-bad # h264parse, mpegtsmux, aacparse, vah264enc
+        gst_all_1.gst-plugins-ugly # x264enc (software-encode fallback)
+        gst_all_1.gst-libav # avenc_aac
+        pipewire # pipewiresrc (Wayland capture; ships in this pkg)
+      ];
+
+      # Opened with dlopen by the optional `--gui` front end (eframe/egui_glow/
+      # winit/glutin), never linked. Harmless for the default headless build.
+      guiRuntimeLibs = with pkgs; [
+        libGL
+        libxkbcommon
+        wayland
+        libx11
+        libxcursor
+        libxrandr
+        libxi
+      ];
+    in
+    {
+      devShells.${system}.default = pkgs.mkShell {
+        nativeBuildInputs =
+          [ rustToolchain ]
+          ++ (with pkgs; [
+            # Debian packaging (`cargo deb --no-build`). Build the binary
+            # inside a Debian/Ubuntu distrobox first so it links that distro's
+            # glibc — see the packaging notes in Cargo.toml.
+            cargo-deb
+
+            pkg-config
+
+            # pipewire-sys, libspa-sys and libpulse-sys all generate bindings
+            # with bindgen, which needs a real libclang at build time.
+            clang
+          ])
+          ++ gstPlugins
+          ++ [
+            # The rest of what `deps::check_host_binaries` looks for.
+            pkgs.pulseaudio # `pactl` (PipeWire stays the actual audio server)
+            pkgs.mpv # the viewer-side player
+            pkgs.xwininfo # the `--window` click-picker on X11
+
+            # `--doctor` shells out to vainfo to confirm the VA-API H.264
+            # ENCODE entrypoint really exists. Without it the report can only
+            # say "vah264enc and a render node are present" and has to leave
+            # hardware encode unconfirmed — which matters, because a GPU
+            # missing that entrypoint produces no video at all under the
+            # default encoder rather than failing loudly.
+            pkgs.libva-utils
+          ];
+
+        buildInputs =
+          with pkgs;
+          [
+            pipewire # pipewire-sys + libspa-sys
+            libpulseaudio # libpulse-sys: --repair reads/unloads Pulse modules
+
+            # x11rb is declared `default-features = false` here, which by itself
+            # is pure Rust — but Cargo unifies features across the graph, and
+            # arboard pulls x11rb with its `libxcb` feature on. That drags in
+            # as-raw-xcb-connection and makes the final link need -lxcb. It is a
+            # real link-time dependency of the binary, not an optional extra.
+            libxcb
+          ]
+          ++ guiRuntimeLibs;
+
+        # bindgen finds libclang through this variable specifically — having
+        # clang on PATH is not sufficient.
+        LIBCLANG_PATH = "${pkgs.llvmPackages.libclang.lib}/lib";
+
+        # NixOS keeps every GStreamer plugin in its own store path, so
+        # gst-launch-1.0 discovers them ONLY through this search path. Without
+        # it, pixelpass's `gst-inspect-1.0 --exists pipewiresrc` preflight fails
+        # even though the plugins are installed. Same reasoning as the
+        # GST_PLUGIN_SYSTEM_PATH_1_0 block in nixos-config hosts/darp5.
+        GST_PLUGIN_SYSTEM_PATH_1_0 = pkgs.lib.makeSearchPathOutput "lib" "lib/gstreamer-1.0" gstPlugins;
+
+        LD_LIBRARY_PATH = pkgs.lib.makeLibraryPath guiRuntimeLibs;
+
+        # Only greet an interactive shell. shellHook also runs under
+        # `nix develop --command …`, where printing this would interleave the
+        # banner with the command's own output (and corrupt it outright for
+        # anything whose stdout is parsed, such as pixelpass's `--output json`).
+        shellHook = ''
+          if [ -t 1 ]; then
+            echo "pixelpass — rustc $(rustc --version | cut -d' ' -f2) / cargo $(cargo --version | cut -d' ' -f2)"
+            echo "  cargo build --release                 headless build (what peerspeak spawns)"
+            echo "  cargo build --release --features gui  with the egui front end"
+            echo "  cargo test                            unit + integration tests"
+            echo "  ./target/debug/pixelpass --doctor     verify this machine can host"
+            echo
+            echo "GStreamer, pactl, mpv and xwininfo are on PATH in this shell, so"
+            echo "capture works here without a system rebuild."
+          fi
+        '';
+      };
+    };
+}

src/host/audio.rs

@@ -35,26 +35,45 @@ use std::cell::RefCell;
 use std::collections::BTreeMap;
 use std::process::Command;
 use std::rc::Rc;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 use std::thread::JoinHandle;
+use std::time::Duration;
 
 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 single `pactl load-module` / `unload-module` may take.
+///
+/// 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.
+const PACTL_BUDGET: Duration = Duration::from_secs(5);
+
+/// How many reconcile-then-unload rounds teardown runs.
+///
+/// Two, because one round can create exactly one new question: an ambiguous load
+/// resolves to a module that then needs unloading, and an uncertain unload
+/// resolves to a module that is either gone or still there. A second round
+/// settles either. Anything still unresolved after that is left to `--repair`
+/// rather than looped over.
+const TEARDOWN_ROUNDS: usize = 2;
 
 /// 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 +85,17 @@ 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();
 
         // 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,29 +107,20 @@ 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),
+            ledger: Arc::clone(&ledger),
             sink_name: sink_name.clone(),
             stream_router: None,
             event_task: None,
@@ -114,21 +128,17 @@ impl Routing {
 
         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);
-                            }
+                            tracing::info!(
+                                "audio routing: first stream routed → unloading default-sink loopback"
+                            );
+                            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,20 +146,7 @@ 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)"
-                                        );
-                                        *local_monitor_for_task.lock().unwrap() = Some(id);
-                                    }
-                                    Err(e) => tracing::warn!(
-                                        "audio routing: failed to load local monitor loopback: {e:#}"
-                                    ),
-                                }
-                            }
+                            ensure_loaded(&ledger_for_task, Shape::LoopbackOutOfCapture, pid).await;
                             // Tell the front-end the chosen app's audio is live.
                             output::emit(output::Event::AppAudio {
                                 state: AppAudioState::Routed,
@@ -164,13 +161,7 @@ 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);
-                            }
+                            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
@@ -183,23 +174,12 @@ impl Routing {
                             }
                             // 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() {
-                                continue;
-                            }
+                            // 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 +205,90 @@ 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) {
+        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;
+            }
+        }
+
+        for _ in 0..TEARDOWN_ROUNDS {
+            if let Err(e) = ledger::reconcile_pending(&self.ledger).await {
+                tracing::warn!("audio routing: could not reconcile the module ledger: {e:#}");
+            }
+            let loaded = self.ledger.loaded();
+            if loaded.is_empty() {
+                break;
+            }
+            for fp in loaded {
+                unload_module(&self.ledger, fp.shape).await;
+            }
+        }
+
+        if !self.ledger.is_settled() {
+            tracing::warn!(
+                "audio routing: some audio modules could not be accounted for; \
+                 `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) {
         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);
+        for fp in self.ledger.loaded() {
+            if self.ledger.begin_unload(fp.shape).is_none() {
+                continue;
+            }
+            let outcome = blocking_unload(fp.id);
+            self.ledger.finish_unload(fp.shape, outcome);
         }
-        // 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 !self.ledger.is_settled() {
+            tracing::warn!(
+                "audio routing: torn down without settling the module ledger; \
+                 run `pixelpass --repair` to clean up anything left behind"
+            );
         }
-        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,54 +402,143 @@ 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")
-        .arg("load-module")
+    let permit = ledger
+        .begin_load(shape, pid, owner.clone())
+        .map_err(anyhow::Error::new)
+        .with_context(|| format!("cannot load the {} module", shape.label()))?;
+
+    let mut cmd = tokio::process::Command::new("pactl");
+    cmd.arg("load-module")
         .arg(shape.module_name())
         .args(shape.render_args(pid, Some(&owner)))
-        .output()
-        .context("failed to run pactl load-module")?;
+        .kill_on_drop(true);
+
+    // Deliberately **not** `select!`ed against a cancellation signal: a completed
+    // load whose index was then dropped on the floor is precisely the defect the
+    // ledger exists to prevent. Cancellation here happens by dropping this whole
+    // future, and the permit's `Drop` turns that into a question reconciliation
+    // can answer, rather than into silence.
+    let output = match tokio::time::timeout(PACTL_BUDGET, cmd.output()).await {
+        Ok(Ok(output)) => output,
+        Ok(Err(e)) => {
+            // Spawning or reading failed. Whether the server was ever reached is
+            // not knowable from here, so leave the permit unsettled: an
+            // unnecessary reconcile costs one listing, a missed one costs an
+            // orphan.
+            drop(permit);
+            return Err(e).context("failed to run pactl load-module");
+        }
+        Err(_) => {
+            drop(permit);
+            bail!("pactl load-module did not finish within {PACTL_BUDGET:?}");
+        }
+    };
+
     if !output.status.success() {
-        bail!(
-            "pactl load-module failed: {}",
-            String::from_utf8_lossy(&output.stderr).trim()
-        );
+        let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
+        if output.status.code().is_some() {
+            // pactl exited of its own accord, having reported the server's
+            // refusal: nothing was created, so there is nothing to reconcile.
+            permit.abandon();
+        } else {
+            // Killed by a signal, which may have arrived *after* the server
+            // created the module.
+            drop(permit);
+        }
+        bail!("pactl load-module failed: {stderr}");
     }
-    let id_str = String::from_utf8(output.stdout)
-        .context("pactl returned non-UTF-8")?
-        .trim()
-        .to_string();
+
+    let id_str = String::from_utf8_lossy(&output.stdout).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:?}"))
+    let Ok(index) = id_str.parse::<u32>() else {
+        // The load may well have succeeded — we simply cannot say which module it
+        // produced, which is exactly what reconciliation is for.
+        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)
 }
 
-fn unload_module(id: u32) {
-    let result = Command::new("pactl")
+/// 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) {
+    let Some(fp) = ledger.begin_unload(shape) else {
+        return;
+    };
+    let mut cmd = tokio::process::Command::new("pactl");
+    cmd.arg("unload-module")
+        .arg(fp.id.to_string())
+        .kill_on_drop(true);
+    let outcome = match tokio::time::timeout(PACTL_BUDGET, cmd.output()).await {
+        Ok(Ok(output)) if output.status.success() => {
+            tracing::info!(module = fp.id, "audio routing: unloaded pactl module");
+            UnloadOutcome::Confirmed
+        }
+        Ok(Ok(output)) => UnloadOutcome::Uncertain(format!(
+            "pactl unload-module exited {}: {}",
+            output.status,
+            String::from_utf8_lossy(&output.stderr).trim()
+        )),
+        Ok(Err(e)) => UnloadOutcome::Uncertain(format!("could not run pactl unload-module: {e}")),
+        Err(_) => {
+            UnloadOutcome::Uncertain(format!("pactl unload-module exceeded {PACTL_BUDGET:?}"))
+        }
+    };
+    ledger.finish_unload(shape, outcome);
+}
+
+/// The blocking unload [`Drop`] uses, since it has no runtime to await on.
+fn blocking_unload(id: u32) -> UnloadOutcome {
+    match Command::new("pactl")
         .arg("unload-module")
         .arg(id.to_string())
-        .output();
-    match result {
+        .output()
+    {
         Ok(output) if output.status.success() => {
             tracing::info!(module = id, "audio routing: unloaded pactl module");
+            UnloadOutcome::Confirmed
         }
-        Ok(output) => {
-            tracing::warn!(
-                module = id,
-                stderr = %String::from_utf8_lossy(&output.stderr).trim(),
-                "audio routing: pactl unload-module exited non-zero"
-            );
-        }
-        Err(e) => {
-            tracing::warn!(
-                module = id,
-                "audio routing: failed to run pactl unload-module: {e}"
-            );
-        }
+        Ok(output) => UnloadOutcome::Uncertain(format!(
+            "pactl unload-module exited {}: {}",
+            output.status,
+            String::from_utf8_lossy(&output.stderr).trim()
+        )),
+        Err(e) => UnloadOutcome::Uncertain(format!("could not run pactl unload-module: {e}")),
     }
 }
 
@@ -718,6 +835,130 @@ fn try_flush(
 #[cfg(test)]
 mod tests {
     use super::*;
+    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()
+    }
+
+    /// 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;
+        });
+        // Let the task run up to its first await — the spawned `pactl` — so the
+        // abort lands mid-flight rather than before the load ever started, which
+        // would make this gate vacuous.
+        tokio::task::yield_now().await;
+        task.abort();
+        let _ = task.await;
+
+        // Non-vacuity: the permit is taken *before* `pactl` is spawned, so a
+        // cancelled load must leave a question behind whichever side of the spawn
+        // the abort landed on. Without this the gate could pass while the abort
+        // fired before the load ever began, proving nothing.
+        assert_eq!(
+            ledger.pending().len(),
+            1,
+            "the cancelled load must have left exactly one question behind"
+        );
+
+        ledger::reconcile_pending(&ledger)
+            .await
+            .expect("the ledger reconciles against the server");
+        for fp in ledger.loaded() {
+            unload_module(&ledger, fp.shape).await;
+        }
+
+        assert!(
+            ledger.is_settled(),
+            "every slot must end in a state we can explain"
+        );
+        assert_eq!(
+            module_snapshot(),
+            before,
+            "a cancelled load must leave nothing behind"
+        );
+    }
 
     #[test]
     fn object_serial_parses_past_u32() {

src/host/ledger.rs

@@ -0,0 +1,890 @@
+//! What this host has put into the Pulse module table — and, where it cannot be
+//! sure, what it must go and find out before doing anything else.
+//!
+//! # The defect this exists for
+//!
+//! Module ids used to live in three `Option<u32>`s, two of them shared with the
+//! event task behind an `Arc<Mutex<…>>`. That representation cannot express *a
+//! load is in flight*, and the gap is reachable today: teardown calls
+//! `event_task.abort()` and then reads the ids, but the task's work is a
+//! synchronous `pactl` call with no await point inside it, so the abort cannot
+//! land until the load has already returned. Teardown therefore sees `None`,
+//! unloads the sink, and the still-running task stores the new module's id into a
+//! mutex nobody will ever read again — an orphan loopback pointing at a sink that
+//! no longer exists.
+//!
+//! A slot is consequently not an `Option<u32>` but a small state machine, and the
+//! transitions that matter are the ones that admit *we do not know*:
+//!
+//! ```text
+//!                begin_load                commit
+//!     Vacant ───────────────► Loading ───────────────► Loaded
+//!        ▲                    │    │                     │
+//!        │          abandon   │    │ permit dropped      │ begin_unload
+//!        └────────────────────┘    │ unsettled           ▼
+//!        ▲                         ▼                  Unloading
+//!        │  Resolution::Nothing  Ambiguous ◄──────────────┘
+//!        └───────────────────────► │  ▲    uncertain outcome
+//!                                  │  │
+//!              Resolution::Conflict▼  └── Resolution::Adopt ──► Loaded
+//!                              Poisoned
+//! ```
+//!
+//! # Why the permit is affine
+//!
+//! [`LoadPermit`] is not `Clone`, is consumed by value to settle, and its [`Drop`]
+//! marks the slot [`Reconcile::Load`] when it was never settled. That is what makes
+//! the guarantee structural rather than a discipline: a cancelled task drops its
+//! locals, so an aborted load *cannot* silently forget a module the server may
+//! already have created. Two permitted loads for one slot cannot both commit,
+//! because [`ModuleLedger::begin_load`] issues a permit only for a `Vacant` slot
+//! and every other state — including the ambiguous one — refuses.
+//!
+//! # Why an ambiguous slot blocks the next load
+//!
+//! An unresolved load may or may not have created a module carrying our sink name.
+//! Starting another one on top of it risks two sinks sharing a `node.name`, which
+//! is measurably not an error the server reports: it accepts both, and
+//! `pulsesrc device=<name>.monitor` attaches to the *older* one. A second capture
+//! session would then be silently stolen by the debris of the first. Refusing to
+//! proceed until the question is answered is the whole point.
+
+use std::collections::BTreeMap;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
+
+use anyhow::{Context as _, Result};
+
+use crate::repair::introspect::PulseSession;
+use crate::repair::plan::{
+    Fingerprint, ModuleObservation, OwnerToken, Shape, classify, recorded_argument,
+};
+
+/// PulseAudio's "no such index" — `PA_INVALID_INDEX`, `(uint32_t) -1`.
+///
+/// Every Pulse load callback reports failure by handing back this value rather
+/// than an index. We load through `pactl`, which reports failure by exiting
+/// non-zero instead, so seeing it on stdout would mean the tool printed a
+/// sentinel we must not mistake for a module: unloading it would be a request
+/// about something that cannot exist. Treated as a load whose outcome is unknown,
+/// never as a successful one.
+pub const PA_INVALID_INDEX: u32 = u32::MAX;
+
+/// What an unresolved slot needs looked up before it can be trusted again.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Reconcile {
+    /// A load whose outcome is unknown: `pactl` was killed, timed out, or printed
+    /// something we could not read as an index. The server may or may not have
+    /// created the module, so it is looked for **by owner token** — the nonce is
+    /// minted per load, so it names this attempt and no other.
+    Load {
+        shape: Shape,
+        pid: u32,
+        token: OwnerToken,
+    },
+    /// An unload whose outcome is unknown. The module is looked for by its full
+    /// fingerprint: still present means the unload did not happen, absent means it
+    /// did. An id alone would not do, because Pulse reuses module indices verbatim.
+    Unload { fp: Fingerprint },
+}
+
+/// One shape's slot in the module table.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SlotState {
+    /// Nothing loaded, and nothing outstanding.
+    Vacant,
+    /// A load is in flight under `permit`. The id is meaningless until it commits.
+    Loading { token: OwnerToken, permit: u64 },
+    /// A module we loaded and can name exactly.
+    Loaded { fp: Fingerprint },
+    /// An unload is in flight. The fingerprint is retained deliberately: an unload
+    /// that times out must not leave the module unrecorded.
+    Unloading { fp: Fingerprint },
+    /// The slot's real state is unknown and must be resolved against the server.
+    Ambiguous(Reconcile),
+    /// Resolution found something we refuse to act on. Terminal.
+    Poisoned { reason: String },
+}
+
+impl SlotState {
+    /// A short, stable label for logs and refusal messages.
+    fn label(&self) -> &'static str {
+        match self {
+            SlotState::Vacant => "vacant",
+            SlotState::Loading { .. } => "loading",
+            SlotState::Loaded { .. } => "loaded",
+            SlotState::Unloading { .. } => "unloading",
+            SlotState::Ambiguous(_) => "ambiguous",
+            SlotState::Poisoned { .. } => "poisoned",
+        }
+    }
+}
+
+/// Why the ledger refused an operation.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum LedgerError {
+    /// The slot was not free. Carries the state that refused, for the log line.
+    Busy { shape: Shape, state: &'static str },
+    /// The slot is poisoned and will not be used again this session.
+    Poisoned { shape: Shape, reason: String },
+    /// `pactl` reported `PA_INVALID_INDEX` where an index was expected.
+    InvalidIndex { shape: Shape },
+}
+
+impl std::fmt::Display for LedgerError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            LedgerError::Busy { shape, state } => {
+                write!(f, "the {} slot is {state}, not free", shape.label())
+            }
+            LedgerError::Poisoned { shape, reason } => {
+                write!(f, "the {} slot is poisoned: {reason}", shape.label())
+            }
+            LedgerError::InvalidIndex { shape } => write!(
+                f,
+                "pactl reported PA_INVALID_INDEX for the {} module",
+                shape.label()
+            ),
+        }
+    }
+}
+
+impl std::error::Error for LedgerError {}
+
+/// The outcome of an attempted unload, as the caller observed it.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum UnloadOutcome {
+    /// The server acknowledged the unload.
+    Confirmed,
+    /// The unload may or may not have happened: the command failed, was killed,
+    /// or its result could not be read.
+    Uncertain(String),
+}
+
+/// What a reconciliation found.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Resolution {
+    /// Exactly one module answers the question. Adopt it.
+    Adopt(Fingerprint),
+    /// No module answers it: the load never happened, or the unload did.
+    Nothing,
+    /// More than one module answers it. Refuse to guess — see [`Resolution`]'s use
+    /// in [`ModuleLedger::apply`], which poisons the slot rather than picking.
+    Conflict(usize),
+}
+
+/// The ledger proper: one slot per [`Shape`], plus the permit counter.
+///
+/// Held behind an `Arc` because the permit needs a way back to it from [`Drop`],
+/// which is what makes a cancelled load impossible to lose.
+pub struct ModuleLedger {
+    slots: Mutex<BTreeMap<Shape, SlotState>>,
+    next_permit: AtomicU64,
+}
+
+impl ModuleLedger {
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self {
+            slots: Mutex::new(BTreeMap::new()),
+            next_permit: AtomicU64::new(1),
+        })
+    }
+
+    /// The current state of one slot. Absent keys read as [`SlotState::Vacant`].
+    ///
+    /// Test-only: production code never needs to look a slot up, because every
+    /// decision that depends on one is made *by* the ledger — `begin_load`
+    /// refuses a busy slot and says which state refused, `begin_unload` returns
+    /// nothing for a slot holding nothing. An accessor callers could branch on
+    /// would invite exactly the check-then-act races the permit removes.
+    #[cfg(test)]
+    pub fn state(&self, shape: Shape) -> SlotState {
+        self.slots
+            .lock()
+            .unwrap()
+            .get(&shape)
+            .cloned()
+            .unwrap_or(SlotState::Vacant)
+    }
+
+    /// Take permission to load `shape`, moving the slot to
+    /// [`SlotState::Loading`].
+    ///
+    /// Refuses anything but a `Vacant` slot. In particular an *ambiguous* slot
+    /// refuses, so an unresolved load blocks the next one until it is reconciled.
+    pub fn begin_load(
+        self: &Arc<Self>,
+        shape: Shape,
+        pid: u32,
+        token: OwnerToken,
+    ) -> Result<LoadPermit, LedgerError> {
+        let mut slots = self.slots.lock().unwrap();
+        let current = slots.get(&shape).cloned().unwrap_or(SlotState::Vacant);
+        match current {
+            SlotState::Vacant => {}
+            SlotState::Poisoned { reason } => {
+                return Err(LedgerError::Poisoned { shape, reason });
+            }
+            other => {
+                return Err(LedgerError::Busy {
+                    shape,
+                    state: other.label(),
+                });
+            }
+        }
+        let permit = self.next_permit.fetch_add(1, Ordering::Relaxed);
+        slots.insert(
+            shape,
+            SlotState::Loading {
+                token: token.clone(),
+                permit,
+            },
+        );
+        drop(slots);
+        Ok(LoadPermit {
+            ledger: Arc::clone(self),
+            shape,
+            pid,
+            token,
+            permit,
+            settled: false,
+        })
+    }
+
+    /// Move a `Loaded` slot to `Unloading` and hand back what to unload.
+    ///
+    /// `None` for any other state: there is nothing to unload, or the slot is not
+    /// in a condition to be acted on.
+    pub fn begin_unload(&self, shape: Shape) -> Option<Fingerprint> {
+        let mut slots = self.slots.lock().unwrap();
+        let SlotState::Loaded { fp } = slots.get(&shape).cloned()? else {
+            return None;
+        };
+        slots.insert(shape, SlotState::Unloading { fp: fp.clone() });
+        Some(fp)
+    }
+
+    /// Record how an unload turned out. An uncertain one becomes ambiguous rather
+    /// than being assumed done — the module is not forgotten either way.
+    pub fn finish_unload(&self, shape: Shape, outcome: UnloadOutcome) {
+        let mut slots = self.slots.lock().unwrap();
+        let Some(SlotState::Unloading { fp }) = slots.get(&shape).cloned() else {
+            return;
+        };
+        let next = match outcome {
+            UnloadOutcome::Confirmed => SlotState::Vacant,
+            UnloadOutcome::Uncertain(why) => {
+                tracing::warn!(
+                    shape = fp.shape.label(),
+                    module = fp.id,
+                    "audio ledger: unload outcome uncertain ({why}); slot needs reconciling"
+                );
+                SlotState::Ambiguous(Reconcile::Unload { fp })
+            }
+        };
+        slots.insert(shape, next);
+    }
+
+    /// Every slot currently holding a module, in [`Shape`] declaration order —
+    /// which is unload order: the loopbacks that reference the capture sink come
+    /// before the sink itself.
+    pub fn loaded(&self) -> Vec<Fingerprint> {
+        self.slots
+            .lock()
+            .unwrap()
+            .values()
+            .filter_map(|state| match state {
+                SlotState::Loaded { fp } => Some(fp.clone()),
+                _ => None,
+            })
+            .collect()
+    }
+
+    /// Every question outstanding against the server, in shape order.
+    pub fn pending(&self) -> Vec<(Shape, Reconcile)> {
+        self.slots
+            .lock()
+            .unwrap()
+            .iter()
+            .filter_map(|(shape, state)| match state {
+                SlotState::Ambiguous(r) => Some((*shape, r.clone())),
+                _ => None,
+            })
+            .collect()
+    }
+
+    /// Is every slot in a state we can explain? False while anything is ambiguous
+    /// or poisoned.
+    pub fn is_settled(&self) -> bool {
+        self.slots
+            .lock()
+            .unwrap()
+            .values()
+            .all(|state| !matches!(state, SlotState::Ambiguous(_) | SlotState::Poisoned { .. }))
+    }
+
+    /// Apply a reconciliation result to an ambiguous slot.
+    ///
+    /// A slot that is no longer ambiguous is left alone: the answer is stale, and
+    /// overwriting a live state with it would be worse than ignoring it.
+    pub fn apply(&self, shape: Shape, resolution: Resolution) {
+        let mut slots = self.slots.lock().unwrap();
+        if !matches!(slots.get(&shape), Some(SlotState::Ambiguous(_))) {
+            return;
+        }
+        let next = match resolution {
+            Resolution::Adopt(fp) => {
+                tracing::info!(
+                    shape = shape.label(),
+                    module = fp.id,
+                    "audio ledger: reconciled — adopting the module the server actually has"
+                );
+                SlotState::Loaded { fp }
+            }
+            Resolution::Nothing => {
+                tracing::info!(
+                    shape = shape.label(),
+                    "audio ledger: reconciled — the server has no such module"
+                );
+                SlotState::Vacant
+            }
+            Resolution::Conflict(n) => {
+                let reason =
+                    format!("{n} modules answer to this slot's token; refusing to choose one");
+                tracing::error!(shape = shape.label(), "audio ledger: {reason}");
+                SlotState::Poisoned { reason }
+            }
+        };
+        slots.insert(shape, next);
+    }
+}
+
+/// Permission to perform exactly one load, which must be settled by value.
+///
+/// Dropping it unsettled is not an error — it is the *reporting* path for a
+/// cancelled or panicking load, and it marks the slot ambiguous so the module the
+/// server may have created is looked for rather than forgotten.
+#[must_use = "an unsettled permit marks the slot ambiguous when dropped"]
+pub struct LoadPermit {
+    ledger: Arc<ModuleLedger>,
+    shape: Shape,
+    pid: u32,
+    token: OwnerToken,
+    permit: u64,
+    settled: bool,
+}
+
+impl LoadPermit {
+    /// Record that the server created the module at `index`.
+    ///
+    /// Fails on [`PA_INVALID_INDEX`], leaving the permit unsettled so that
+    /// dropping it marks the slot ambiguous — a sentinel where an index belongs
+    /// means the load's outcome is precisely what we do not know.
+    pub fn commit(mut self, index: u32) -> Result<Fingerprint, LedgerError> {
+        if index == PA_INVALID_INDEX {
+            return Err(LedgerError::InvalidIndex { shape: self.shape });
+        }
+        let fp = expected_fingerprint(self.shape, self.pid, &self.token, index);
+        // A permit outlives its slot's `Loading` state only if something else has
+        // already moved the slot on — in which case this answer is stale and the
+        // live state wins.
+        let mut slots = self.ledger.slots.lock().unwrap();
+        match slots.get(&self.shape) {
+            Some(SlotState::Loading { permit, .. }) if *permit == self.permit => {
+                slots.insert(self.shape, SlotState::Loaded { fp: fp.clone() });
+            }
+            other => {
+                let state = other.map(SlotState::label).unwrap_or("vacant");
+                tracing::warn!(
+                    shape = self.shape.label(),
+                    module = index,
+                    "audio ledger: a load committed against a slot that is now {state}; \
+                     leaving the live state alone"
+                );
+            }
+        }
+        drop(slots);
+        self.settled = true;
+        Ok(fp)
+    }
+
+    /// Record that the server definitively created nothing.
+    ///
+    /// Only for a load that failed *cleanly* — `pactl` exiting non-zero of its own
+    /// accord, having reported the server's refusal. A killed or timed-out load is
+    /// not this: drop the permit instead and let the slot go ambiguous.
+    pub fn abandon(mut self) {
+        let mut slots = self.ledger.slots.lock().unwrap();
+        if let Some(SlotState::Loading { permit, .. }) = slots.get(&self.shape)
+            && *permit == self.permit
+        {
+            slots.insert(self.shape, SlotState::Vacant);
+        }
+        drop(slots);
+        self.settled = true;
+    }
+}
+
+impl Drop for LoadPermit {
+    fn drop(&mut self) {
+        if self.settled {
+            return;
+        }
+        let mut slots = self.ledger.slots.lock().unwrap();
+        // Only claim the slot if it is still *our* load. Anything else already
+        // moved past this permit.
+        if let Some(SlotState::Loading { permit, .. }) = slots.get(&self.shape)
+            && *permit == self.permit
+        {
+            tracing::warn!(
+                shape = self.shape.label(),
+                "audio ledger: a load was cancelled before its outcome was known; \
+                 the slot needs reconciling"
+            );
+            slots.insert(
+                self.shape,
+                SlotState::Ambiguous(Reconcile::Load {
+                    shape: self.shape,
+                    pid: self.pid,
+                    token: self.token.clone(),
+                }),
+            );
+        }
+    }
+}
+
+/// The fingerprint a successful load of `shape` for `pid` under `token` must have.
+///
+/// Built from the shape's own renderer, exactly as [`classify`] rebuilds it from
+/// an observation, so an adopted module and a committed one are the same value.
+fn expected_fingerprint(shape: Shape, pid: u32, token: &OwnerToken, id: u32) -> Fingerprint {
+    Fingerprint {
+        id,
+        module_name: shape.module_name().to_string(),
+        args: recorded_argument(&shape.render_args(pid, Some(token))),
+        pid,
+        shape,
+        owner: Some(token.clone()),
+    }
+}
+
+/// Answer one outstanding question against a snapshot of the module table.
+///
+/// Pure: the snapshot is the only input, so every branch is testable without a
+/// Pulse server.
+pub fn resolve(reconcile: &Reconcile, observations: &[ModuleObservation]) -> Resolution {
+    let matches: Vec<Fingerprint> = match reconcile {
+        // The nonce is minted per load, so token equality names this attempt and
+        // no other — including a previous load of the same shape by the same pid.
+        Reconcile::Load { shape, token, .. } => observations
+            .iter()
+            .filter_map(classify)
+            .filter(|fp| fp.shape == *shape && fp.owner.as_ref() == Some(token))
+            .collect(),
+        // A full fingerprint match, not an id: Pulse reuses module indices
+        // verbatim, so "something is at that index" is not "our module is".
+        Reconcile::Unload { fp } => observations
+            .iter()
+            .filter(|obs| fp.still_matches(obs))
+            .filter_map(classify)
+            .collect(),
+    };
+    match matches.len() {
+        0 => Resolution::Nothing,
+        1 => Resolution::Adopt(matches.into_iter().next().expect("length checked")),
+        n => Resolution::Conflict(n),
+    }
+}
+
+/// Resolve every outstanding question against one snapshot of the module table.
+///
+/// One listing answers all of them, so the slots are reconciled against a single
+/// server response rather than several that could disagree.
+///
+/// ⚠️ The session is deliberately short-lived — connect, list, disconnect — which
+/// is `--repair`'s pattern and not the long-lived host session round 19 sketched.
+/// `repair::introspect` documents why: on a request timeout the binding leaks the
+/// boxed callback until the context disconnects, which is bounded and harmless for
+/// a session that ends immediately, and is not acceptable for one held open for
+/// the life of a share. Reconciliation is rare and off the hot path, so paying a
+/// connection for it costs nothing that matters.
+pub async fn reconcile_pending(ledger: &Arc<ModuleLedger>) -> Result<()> {
+    let pending = ledger.pending();
+    if pending.is_empty() {
+        return Ok(());
+    }
+    tracing::info!(
+        n = pending.len(),
+        "audio ledger: reconciling unresolved module slots against the server"
+    );
+    let observations = tokio::task::spawn_blocking(|| -> Result<Vec<ModuleObservation>> {
+        let mut session = PulseSession::connect()?;
+        session.list_modules()
+    })
+    .await
+    .context("the Pulse listing task failed to run")?
+    .context("could not list Pulse modules to reconcile the audio ledger")?;
+
+    for (shape, reconcile) in pending {
+        ledger.apply(shape, resolve(&reconcile, &observations));
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn token(nonce: u64) -> OwnerToken {
+        OwnerToken {
+            machine: "abc123".to_string(),
+            boot: "def456".to_string(),
+            pid_ns: 4_026_531_836,
+            nonce,
+        }
+    }
+
+    /// A module observation as the server would report it for one of our loads.
+    fn observed(id: u32, shape: Shape, pid: u32, token: &OwnerToken) -> ModuleObservation {
+        ModuleObservation::new(
+            id,
+            shape.module_name(),
+            &recorded_argument(&shape.render_args(pid, Some(token))),
+        )
+    }
+
+    #[test]
+    fn a_committed_load_becomes_loaded() {
+        let ledger = ModuleLedger::new();
+        let permit = ledger
+            .begin_load(Shape::LegacyCaptureSink, 42, token(7))
+            .expect("a vacant slot issues a permit");
+        assert!(matches!(
+            ledger.state(Shape::LegacyCaptureSink),
+            SlotState::Loading { .. }
+        ));
+        let fp = permit.commit(9).expect("9 is a real index");
+        assert_eq!(fp.id, 9);
+        assert_eq!(
+            ledger.state(Shape::LegacyCaptureSink),
+            SlotState::Loaded { fp }
+        );
+    }
+
+    #[test]
+    fn dropping_a_permit_unsettled_marks_the_slot_ambiguous() {
+        // The orphan race in one test: a load that is cancelled between spawning
+        // pactl and reading its id must leave the slot asking a question, never
+        // looking empty.
+        let ledger = ModuleLedger::new();
+        let permit = ledger
+            .begin_load(Shape::LoopbackOutOfCapture, 42, token(7))
+            .expect("a vacant slot issues a permit");
+        drop(permit);
+        assert_eq!(
+            ledger.state(Shape::LoopbackOutOfCapture),
+            SlotState::Ambiguous(Reconcile::Load {
+                shape: Shape::LoopbackOutOfCapture,
+                pid: 42,
+                token: token(7),
+            }),
+            "a cancelled load must be remembered as a question, not as a vacancy"
+        );
+        assert!(!ledger.is_settled());
+    }
+
+    #[test]
+    fn abandoning_a_cleanly_failed_load_returns_the_slot_to_vacant() {
+        let ledger = ModuleLedger::new();
+        ledger
+            .begin_load(Shape::LoopbackIntoCapture, 42, token(7))
+            .expect("a vacant slot issues a permit")
+            .abandon();
+        assert_eq!(
+            ledger.state(Shape::LoopbackIntoCapture),
+            SlotState::Vacant,
+            "a load the server refused created nothing to reconcile"
+        );
+        assert!(ledger.is_settled());
+    }
+
+    #[test]
+    fn a_second_permit_is_refused_while_a_load_is_in_flight() {
+        let ledger = ModuleLedger::new();
+        let _first = ledger
+            .begin_load(Shape::LegacyCaptureSink, 42, token(7))
+            .expect("a vacant slot issues a permit");
+        let second = ledger.begin_load(Shape::LegacyCaptureSink, 42, token(8));
+        assert_eq!(
+            second.err(),
+            Some(LedgerError::Busy {
+                shape: Shape::LegacyCaptureSink,
+                state: "loading"
+            }),
+            "two loads for one slot must not both be permitted"
+        );
+    }
+
+    #[test]
+    fn an_ambiguous_slot_refuses_the_next_load_until_it_is_reconciled() {
+        // Why this matters: two sinks may share a node.name, and pulsesrc attaches
+        // to the older one — so loading over unresolved debris silently steals the
+        // next session's capture.
+        let ledger = ModuleLedger::new();
+        drop(
+            ledger
+                .begin_load(Shape::LegacyCaptureSink, 42, token(7))
+                .expect("a vacant slot issues a permit"),
+        );
+        assert_eq!(
+            ledger
+                .begin_load(Shape::LegacyCaptureSink, 42, token(8))
+                .err(),
+            Some(LedgerError::Busy {
+                shape: Shape::LegacyCaptureSink,
+                state: "ambiguous"
+            })
+        );
+
+        // Reconciling to "nothing was created" frees it again.
+        ledger.apply(Shape::LegacyCaptureSink, Resolution::Nothing);
+        assert!(
+            ledger
+                .begin_load(Shape::LegacyCaptureSink, 42, token(9))
+                .is_ok()
+        );
+    }
+
+    #[test]
+    fn an_invalid_index_is_not_adopted_and_leaves_a_question_behind() {
+        let ledger = ModuleLedger::new();
+        let permit = ledger
+            .begin_load(Shape::LegacyCaptureSink, 42, token(7))
+            .expect("a vacant slot issues a permit");
+        assert_eq!(
+            permit.commit(PA_INVALID_INDEX).err(),
+            Some(LedgerError::InvalidIndex {
+                shape: Shape::LegacyCaptureSink
+            })
+        );
+        assert!(
+            matches!(
+                ledger.state(Shape::LegacyCaptureSink),
+                SlotState::Ambiguous(_)
+            ),
+            "a sentinel where an index belongs is exactly the unknown outcome"
+        );
+    }
+
+    #[test]
+    fn an_uncertain_unload_is_not_forgotten() {
+        let ledger = ModuleLedger::new();
+        let fp = ledger
+            .begin_load(Shape::LoopbackIntoCapture, 42, token(7))
+            .expect("a vacant slot issues a permit")
+            .commit(3)
+            .expect("3 is a real index");
+        assert_eq!(
+            ledger.begin_unload(Shape::LoopbackIntoCapture),
+            Some(fp.clone())
+        );
+        ledger.finish_unload(
+            Shape::LoopbackIntoCapture,
+            UnloadOutcome::Uncertain("pactl was killed".to_string()),
+        );
+        assert_eq!(
+            ledger.state(Shape::LoopbackIntoCapture),
+            SlotState::Ambiguous(Reconcile::Unload { fp }),
+            "an unload whose outcome is unknown must keep naming the module"
+        );
+    }
+
+    #[test]
+    fn a_confirmed_unload_empties_the_slot() {
+        let ledger = ModuleLedger::new();
+        ledger
+            .begin_load(Shape::LoopbackIntoCapture, 42, token(7))
+            .expect("a vacant slot issues a permit")
+            .commit(3)
+            .expect("3 is a real index");
+        ledger.begin_unload(Shape::LoopbackIntoCapture);
+        ledger.finish_unload(Shape::LoopbackIntoCapture, UnloadOutcome::Confirmed);
+        assert_eq!(ledger.state(Shape::LoopbackIntoCapture), SlotState::Vacant);
+        assert!(ledger.is_settled());
+    }
+
+    #[test]
+    fn begin_unload_only_acts_on_a_loaded_slot() {
+        let ledger = ModuleLedger::new();
+        assert_eq!(ledger.begin_unload(Shape::LegacyCaptureSink), None);
+        let _permit = ledger
+            .begin_load(Shape::LegacyCaptureSink, 42, token(7))
+            .expect("a vacant slot issues a permit");
+        assert_eq!(
+            ledger.begin_unload(Shape::LegacyCaptureSink),
+            None,
+            "a load in flight has no id to unload yet"
+        );
+    }
+
+    #[test]
+    fn resolve_adopts_the_one_module_carrying_our_token() {
+        let mine = token(7);
+        let reconcile = Reconcile::Load {
+            shape: Shape::LegacyCaptureSink,
+            pid: 42,
+            token: mine.clone(),
+        };
+        let observations = vec![
+            // A previous load by the same pid and shape: same everything but the
+            // per-load nonce, which is the whole reason the nonce exists.
+            observed(4, Shape::LegacyCaptureSink, 42, &token(6)),
+            observed(5, Shape::LegacyCaptureSink, 42, &mine),
+            // Somebody else's module entirely.
+            ModuleObservation::new(6, "module-null-sink", "sink_name=other"),
+        ];
+        assert_eq!(
+            resolve(&reconcile, &observations),
+            Resolution::Adopt(expected_fingerprint(Shape::LegacyCaptureSink, 42, &mine, 5))
+        );
+    }
+
+    #[test]
+    fn resolve_reports_nothing_when_the_server_never_created_it() {
+        let reconcile = Reconcile::Load {
+            shape: Shape::LegacyCaptureSink,
+            pid: 42,
+            token: token(7),
+        };
+        let observations = vec![observed(4, Shape::LegacyCaptureSink, 42, &token(6))];
+        assert_eq!(resolve(&reconcile, &observations), Resolution::Nothing);
+    }
+
+    #[test]
+    fn resolve_fails_closed_when_more_than_one_module_answers() {
+        let mine = token(7);
+        let reconcile = Reconcile::Load {
+            shape: Shape::LegacyCaptureSink,
+            pid: 42,
+            token: mine.clone(),
+        };
+        // Two modules carrying the same token should be impossible. If it ever
+        // happens, guessing which to keep is how a live sink gets unloaded.
+        let observations = vec![
+            observed(4, Shape::LegacyCaptureSink, 42, &mine),
+            observed(5, Shape::LegacyCaptureSink, 42, &mine),
+        ];
+        assert_eq!(resolve(&reconcile, &observations), Resolution::Conflict(2));
+    }
+
+    #[test]
+    fn a_conflict_poisons_the_slot_and_it_stays_poisoned() {
+        let ledger = ModuleLedger::new();
+        drop(
+            ledger
+                .begin_load(Shape::LegacyCaptureSink, 42, token(7))
+                .expect("a vacant slot issues a permit"),
+        );
+        ledger.apply(Shape::LegacyCaptureSink, Resolution::Conflict(2));
+        let SlotState::Poisoned { reason } = ledger.state(Shape::LegacyCaptureSink) else {
+            panic!("a conflict must poison the slot");
+        };
+        assert!(!ledger.is_settled());
+        assert_eq!(
+            ledger
+                .begin_load(Shape::LegacyCaptureSink, 42, token(8))
+                .err(),
+            Some(LedgerError::Poisoned {
+                shape: Shape::LegacyCaptureSink,
+                reason
+            }),
+            "poisoning is terminal for the session"
+        );
+    }
+
+    #[test]
+    fn resolve_finds_a_module_an_uncertain_unload_left_behind() {
+        let mine = token(7);
+        let fp = expected_fingerprint(Shape::LoopbackIntoCapture, 42, &mine, 5);
+        let observations = vec![observed(5, Shape::LoopbackIntoCapture, 42, &mine)];
+        assert_eq!(
+            resolve(&Reconcile::Unload { fp: fp.clone() }, &observations),
+            Resolution::Adopt(fp.clone()),
+            "still present means the unload did not happen"
+        );
+        assert_eq!(
+            resolve(&Reconcile::Unload { fp }, &[]),
+            Resolution::Nothing,
+            "absent means it did"
+        );
+    }
+
+    #[test]
+    fn an_unload_reconcile_ignores_a_stranger_at_the_same_index() {
+        // Pulse reuses module indices verbatim, so "something is at index 5" must
+        // not be read as "our module is still at index 5".
+        //
+        // The stranger has to be a module that *classifies* — another pixelpass
+        // host's canonical loopback — or this gate is vacuous: a comparator using
+        // the id alone would still return `Nothing` for junk, because junk is
+        // discarded by `classify` regardless of how the match was made.
+        let fp = expected_fingerprint(Shape::LoopbackIntoCapture, 42, &token(7), 5);
+        let another_hosts = observed(5, Shape::LoopbackIntoCapture, 99, &token(3));
+        assert_eq!(
+            resolve(
+                &Reconcile::Unload { fp: fp.clone() },
+                std::slice::from_ref(&another_hosts)
+            ),
+            Resolution::Nothing,
+            "another host's module at our old index is not our module"
+        );
+        // And plain junk at that index is ignored too.
+        let junk = ModuleObservation::new(5, "module-loopback", "source=some_mic sink=theirs");
+        assert_eq!(
+            resolve(&Reconcile::Unload { fp }, std::slice::from_ref(&junk)),
+            Resolution::Nothing
+        );
+    }
+
+    #[test]
+    fn a_stale_answer_does_not_overwrite_a_live_slot() {
+        let ledger = ModuleLedger::new();
+        // Nothing is ambiguous, so an answer arriving late is meaningless.
+        let fp = ledger
+            .begin_load(Shape::LegacyCaptureSink, 42, token(7))
+            .expect("a vacant slot issues a permit")
+            .commit(3)
+            .expect("3 is a real index");
+        ledger.apply(Shape::LegacyCaptureSink, Resolution::Nothing);
+        assert_eq!(
+            ledger.state(Shape::LegacyCaptureSink),
+            SlotState::Loaded { fp },
+            "a live state must win over a stale reconciliation"
+        );
+    }
+
+    #[test]
+    fn loaded_lists_modules_in_shape_declaration_order() {
+        // Declaration order is unload order: the loopbacks that reference the
+        // capture sink must come before the sink itself.
+        let ledger = ModuleLedger::new();
+        for (shape, id) in [
+            (Shape::LegacyCaptureSink, 1),
+            (Shape::LoopbackIntoCapture, 2),
+            (Shape::LoopbackOutOfCapture, 3),
+        ] {
+            ledger
+                .begin_load(shape, 42, token(u64::from(id)))
+                .expect("a vacant slot issues a permit")
+                .commit(id)
+                .expect("a real index");
+        }
+        let order: Vec<Shape> = ledger.loaded().into_iter().map(|fp| fp.shape).collect();
+        assert_eq!(order, crate::repair::plan::ALL_SHAPES.to_vec());
+        assert_eq!(
+            order.last(),
+            Some(&Shape::LegacyCaptureSink),
+            "the sink must be unloaded after everything that references it"
+        );
+    }
+}

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;