//! Connection-owned capture-sink and per-app audio routing.
//!
//! Two cooperating layers:
//!
//! - **Native graph actor** (libpipewire on a dedicated OS thread): owns a
//!   non-lingering per-PID sink named `pixelpass_capture_<pid>`. When
//!   [`HostOpts::app`] is set, the same actor finds matching
//!   `Stream/Output/Audio` nodes and writes `target.object` so WirePlumber
//!   reroutes them to that sink. The sink disappears with the actor's PipeWire
//!   connection, including after SIGKILL.
//!
//! - **Pulse loopbacks** (bounded pactl shell-outs): by default one loopback
//!   mirrors the default sink's monitor into the native capture sink. Once at
//!   least one selected app stream is routed, that loopback is unloaded so the
//!   viewer does not hear the app twice.
//!
//! - **Local monitor** (pactl shell-out, app mode only): rerouting *moves*
//!   the chosen app off the sharer's speakers into the null-sink, so without
//!   this the sharer would go deaf to the very content they're sharing. We
//!   mirror the null-sink's monitor back to `@DEFAULT_SINK@` so the sharer
//!   hears it too. Only the chosen app is in the null-sink — never the
//!   desktop/call — so this can't echo back into the capture. It is loaded on
//!   the first routed stream (after the default-sink loopback is gone, so the
//!   two never coexist and feed back) and unloaded when the app stops.
//!
//! Shutdown quiesces route writes, unloads every dependent Pulse loopback, and
//! only then releases the actor connection and native sink.

use anyhow::{Context, Result, bail};
use std::collections::BTreeMap;
use std::io::{self, Read};
use std::process::{Child, Command, ExitStatus, Stdio};
use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::cli::HostOpts;
use crate::common::contained;
use crate::host::graph::{AudioGraphOwner, CaptureSinkSpec, GraphEvent, QuiesceOutcome};
use crate::host::health;
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 native graph actor plus its pactl-loaded dependent modules. Drop
/// unloads modules as a backstop; prefer [`Routing::shutdown`] explicitly,
/// which is the only path that can reconcile a load whose outcome was never
/// observed and acknowledge route restoration.
pub struct Routing {
    /// 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,
    graph_owner: Option<AudioGraphOwner>,
    event_task: Option<tokio::task::JoinHandle<()>>,
    health: health::Reporter,
}

impl Routing {
    /// Create the per-PID native sink and graph actor, plus the default-monitor
    /// loopback when the selected routing mode permits it.
    pub(super) async fn start(opts: &HostOpts, health: health::Reporter) -> Result<Self> {
        let pid = std::process::id();
        let sink_name = repair_plan::sink_name_for(pid);
        let ledger = ModuleLedger::new();
        // Construct Routing before either ownership layer mutates the graph. Any
        // error or cancellation below drops a real owner whose backstop closes,
        // reconciles, and unloads the ledger before releasing the native sink.
        let mut routing = Self {
            ledger: Arc::clone(&ledger),
            sink_name: sink_name.clone(),
            graph_owner: None,
            event_task: None,
            health: health.clone(),
        };

        // S4: the capture sink is a native, non-lingering PipeWire object owned
        // by this actor connection, not a module owned by pipewire-pulse.  The
        // actor also absorbs the per-app router so every sink-owning mode has one
        // graph lifetime and one readiness handshake.
        let (graph_owner, mut event_rx, _identity_rx) = AudioGraphOwner::start(
            opts.app.clone(),
            CaptureSinkSpec::for_pid(pid),
            health.clone(),
        )
        .await
        .context("failed to start the connection-owned audio graph")?;
        debug_assert_eq!(graph_owner.identity().name, sink_name);
        routing.graph_owner = Some(graph_owner);

        // In strict per-app mode we never mirror the default sink: the viewer
        // must hear *only* the chosen app, never the whole desktop (which would
        // leak e.g. a voice call the sharer is in back to viewers — the echo
        // bug A23). Without strict mode (whole-desktop share, or best-effort
        // app filtering) we load the monitor loopback so the viewer hears
        // system audio immediately and during any gap before the app routes.
        // 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;
        if !strict_app
            && let Err(error) = load_module(&ledger, Shape::LoopbackIntoCapture, pid).await
        {
            // Once the actor exists, an ordinary constructor error gets a full
            // async teardown rather than falling through the narrower
            // synchronous Drop backstop and quarantining a responsive thread.
            routing.shutdown().await;
            return Err(error)
                .context("failed to load module-loopback (connection-owned sink cleaned up)");
        }

        tracing::info!(
            strict_app,
            %sink_name,
            "audio routing: connection-owned sink ready (loopback skipped in strict app mode)"
        );

        if opts.app.is_some() {
            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 {
                        GraphEvent::FirstRoutedStream => {
                            tracing::info!(
                                "audio routing: first stream routed → unloading default-sink loopback"
                            );
                            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
                            // the two never coexist (which would feed back), and
                            // sourced from the null-sink monitor — the chosen app
                            // only, never the desktop/call — so it can't echo into
                            // the capture.
                            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"
                                );
                            }
                            // Tell the front-end the chosen app's audio is live.
                            output::emit(output::Event::AppAudio {
                                state: AppAudioState::Routed,
                            });
                        }
                        GraphEvent::LastRoutedStreamGone => {
                            // Routed app exited/paused mid-session. Notify the
                            // front-end either way; the recovery differs by mode.
                            output::emit(output::Event::AppAudio {
                                state: AppAudioState::Lost,
                            });
                            // 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.
                            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
                                // produces audio again — never the rest of the
                                // desktop (call included).
                                tracing::info!(
                                    "audio routing: strict mode — last routed stream gone, leaving viewers silent"
                                );
                                continue;
                            }
                            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"
                            );
                            ensure_loaded(&ledger_for_task, Shape::LoopbackIntoCapture, pid).await;
                        }
                    }
                }
            });
            routing.event_task = Some(event_task);
        }

        // Strict per-app mode suppresses the default-sink loopback, so until the
        // chosen app's first stream routes the viewer hears *silence*. Emit an
        // initial `lost` at capture start (capture is lazy — this runs on the
        // first viewer) so the front-end can warn from the outset rather than
        // only after an app that *was* routed later stops (audit A23 P2/F1):
        // `LastRoutedStreamGone`→`lost` never fires for an app that never routed.
        if let Some(state) = initial_app_audio_state(opts) {
            crate::common::output::emit(crate::common::output::Event::AppAudio { state });
        }

        Ok(routing)
    }

    pub fn sink_name(&self) -> &str {
        &self.sink_name
    }

    /// Quiesce graph mutations, stop the event task, settle anything the ledger
    /// is unsure about, unload every dependent loopback, then release the native
    /// sink. PipeWire can leave zombie links if a sink is destroyed with active
    /// inputs, so that final ordering is load-bearing.
    ///
    /// 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
        // graph event while shutdown is in progress.
        self.ledger.close();
        let graph_quiesced = if let Some(graph_owner) = self.graph_owner.as_mut() {
            graph_owner.quiesce().await == QuiesceOutcome::Confirmed
        } else {
            true
        };
        if let Some(mut task) = self.event_task.take() {
            if !graph_quiesced {
                // An unresponsive graph actor may still own its event sender.
                // Cancel and await the task before ledger reconciliation.
                task.abort();
                let _ = task.await;
            } else {
                // Quiesce closes the event sender while retaining the native
                // sink. Abort is the fallback and is awaited through `&mut
                // JoinHandle`, so any in-flight affine permit is dropped before
                // reconciliation reads the ledger.
                match tokio::time::timeout(PACTL_BUDGET, &mut task).await {
                    Ok(Ok(())) => {}
                    Ok(Err(e)) => {
                        self.health
                            .poison(format!("audio routing event task failed: {e}"));
                    }
                    Err(_) => {
                        tracing::warn!(
                            "audio routing: the event task did not finish within {PACTL_BUDGET:?}; \
                             cancelling it"
                        );
                        self.health.poison(format!(
                            "audio routing event task did not stop within {PACTL_BUDGET:?}"
                        ));
                        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}");
            self.health
                .poison(format!("audio module-operation wait task failed: {e}"));
        }
        cleanup_modules(&self.ledger).await;

        // Loopbacks are gone before the actor connection is released.  This is
        // the S4 ordering invariant: dependent Pulse modules never outlive the
        // native sink they reference during an ordinary shutdown.
        if let Some(graph_owner) = self.graph_owner.take()
            && !graph_owner.shutdown().await
        {
            tracing::warn!("audio routing: AudioGraphOwner shutdown was not confirmed");
        }

        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"
            );
            self.health.poison(
                "audio routing teardown left module ownership unresolved; repair is required",
            );
        }
    }
}

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(task) = self.event_task.take() {
            task.abort();
        }
        self.ledger.close_and_wait();
        cleanup_modules_blocking(&self.ledger);
        // Keep the actor/sink alive until dependent modules have been handled,
        // even on this synchronous error/unwind backstop.
        if let Some(graph_owner) = self.graph_owner.take() {
            drop(graph_owner);
        }
        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"
            );
            self.health
                .poison("audio routing Drop left module ownership unresolved; repair is required");
        }
    }
}

/// The app-audio state to announce at capture start, if any. Only strict per-app
/// mode warrants one: there the loopback is suppressed, so the viewer hears
/// silence until the chosen app's first stream routes — surface that as an
/// initial `lost`. In every other mode (whole-desktop, or best-effort app
/// filtering) the loopback keeps audio flowing from the outset, so there is no
/// initial gap to report. Pure: no I/O, so the emit decision is unit-testable.
pub(super) fn initial_app_audio_state(
    opts: &HostOpts,
) -> Option<crate::common::output::AppAudioState> {
    (opts.app.is_some() && opts.strict_audio).then_some(crate::common::output::AppAudioState::Lost)
}

// ──────────────────────────────────────────────────────────────────────
// App enumeration (interactive picker source)
// ──────────────────────────────────────────────────────────────────────

/// One deduplicated app currently producing audio. The picker in
/// interactive mode shows these as the per-app capture choices.
#[derive(Debug, Clone)]
pub struct App {
    pub name: String,
    pub stream_count: u32,
}

/// Enumerate apps currently sending audio to any sink, deduplicated by
/// `application.name`. Returns an empty Vec if nothing is playing.
pub fn list_playing_apps() -> Result<Vec<App>> {
    let output = Command::new("pactl")
        .args(["-f", "json", "list", "sink-inputs"])
        .output()
        .context("failed to run `pactl -f json list sink-inputs`")?;
    if !output.status.success() {
        bail!(
            "pactl list sink-inputs failed: {}",
            String::from_utf8_lossy(&output.stderr).trim()
        );
    }
    parse_sink_inputs(&output.stdout)
}

fn parse_sink_inputs(stdout: &[u8]) -> Result<Vec<App>> {
    let entries: Vec<SinkInput> =
        serde_json::from_slice(stdout).context("pactl returned unparseable JSON")?;
    let mut counts: BTreeMap<String, u32> = BTreeMap::new();
    for entry in entries {
        let Some(name) = entry.properties.application_name else {
            continue;
        };
        let trimmed = name.trim();
        if trimmed.is_empty() {
            continue;
        }
        *counts.entry(trimmed.to_string()).or_insert(0) += 1;
    }
    Ok(counts
        .into_iter()
        .map(|(name, stream_count)| App { name, stream_count })
        .collect())
}

#[derive(serde::Deserialize)]
struct SinkInput {
    properties: SinkInputProperties,
}

#[derive(serde::Deserialize)]
struct SinkInputProperties {
    #[serde(rename = "application.name")]
    application_name: Option<String>,
}

// ──────────────────────────────────────────────────────────────────────
// pactl module helpers
// ──────────────────────────────────────────────────────────────────────

/// Mint an ownership token for one module load.
///
/// **Per load, not per session.** The nonce is what makes two loads by the same pid
/// render different arguments, which is what lets a fingerprint tell a module from
/// its replacement at the same index. A token minted once and reused for every
/// reload would be a host-session nonce and would not do that, so the counter is
/// bumped on every call and mixed with the clock.
fn owner_token(pid: u32) -> Result<repair_plan::OwnerToken> {
    use std::sync::atomic::{AtomicU64, Ordering};
    static LOADS: AtomicU64 = AtomicU64::new(0);

    let local = crate::repair::local_identity()?;
    // A nonce only has to be unlikely to repeat, not unguessable. The counter makes
    // two loads within the same clock tick distinct; the clock keeps two runs of the
    // same process distinct.
    let counter = LOADS.fetch_add(1, Ordering::Relaxed);
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64)
        .unwrap_or(0);
    Ok(repair_plan::OwnerToken {
        machine: local.machine,
        boot: local.boot,
        pid_ns: local.pid_ns,
        nonce: nanos ^ (counter << 48) ^ (u64::from(pid) << 32),
    })
}

/// Load the Pulse module for one [`Shape`] and return its index.
///
/// Both the module name and its arguments come from the shape itself
/// ([`crate::repair::plan::Shape`]) rather than being written out here, so that
/// `--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.
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 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(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");
            }
        };

        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")?
}

/// 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!(
                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;
        }
        self.kill_group();
        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 kill_group(&mut self) {
        let _ = contained::signal_group(self.child.id(), nix::sys::signal::Signal::SIGKILL);
        // Backstop in case the group disappeared between lookup and signal.
        let _ = self.child.kill();
    }

    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: contained::spawn(command)?,
        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 {
            child.kill_group();
            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;
        }
    }
}

/// Parse a PipeWire `object.serial` property value.
///
/// `object.serial` is a **64-bit** monotonically-increasing counter
/// (`pw_global`'s serial is `uint64_t`); it is *not* a `pw` object id
/// (those are `u32` and get recycled — the serial exists precisely so
/// that recycled ids can be disambiguated). Parsing it as `u32` silently
/// yields `None` past `u32::MAX`, which on a long-lived daemon means the
/// sink is never registered and no stream is ever routed.
///
/// Strict on purpose: PipeWire emits a bare decimal, so anything else
/// (empty, signed, whitespace-padded, non-numeric, overflowing) is a
/// property we do not understand and must not guess at. Leading zeroes
/// are accepted — they are unambiguous and parse to the same value.
pub(crate) fn parse_object_serial(raw: &str) -> Option<u64> {
    if raw.is_empty() || !raw.bytes().all(|b| b.is_ascii_digit()) {
        return None;
    }
    raw.parse::<u64>().ok()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::host::ledger::SlotState;
    use crate::repair::plan::{ModuleObservation, classify};

    /// Whole-desktop routing: the graph actor owns the native sink, while the
    /// ledger owns only its default-monitor 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,
            capture_mode: crate::cli::CaptureMode::Legacy,
            aec: crate::host::aec::AecConfig::Off,
            legacy_null_sink: false,
            relay: None,
        }
    }

    #[test]
    fn bounded_module_worker_uses_the_contained_spawn_path() {
        let mut command = Command::new("sh");
        command.args([
            "-c",
            "read pid comm state ppid pgrp rest < /proc/self/stat; printf '%s %s' \"$pid\" \"$pgrp\"",
        ]);
        let output = bounded_output(&mut command, Duration::from_secs(1))
            .expect("run contained module-worker fixture");
        assert!(output.status.success());
        let ids = String::from_utf8(output.stdout).expect("ascii pid/pgid");
        let mut ids = ids.split_whitespace();
        let pid = ids.next().expect("child pid");
        let pgid = ids.next().expect("child process group");
        assert_eq!(pid, pgid, "module worker must lead its own process group");
    }

    /// 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 (health, _) = health::channel();
        let routing = Routing::start(&whole_desktop_opts(), health)
            .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(),
            1,
            "only the default-monitor loopback is a Pulse module; the sink is native"
        );
        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"
        );
    }

    /// Exercise the real graph-actor command path with app routing enabled. The
    /// deliberately unmatched filter avoids moving an unrelated live stream.
    #[tokio::test]
    #[ignore = "uses the real Pulse/PipeWire graph; run with --ignored --test-threads=1"]
    async fn live_audio_graph_owner_stops_within_its_policy_budget() {
        let before = module_snapshot();
        let mut opts = whole_desktop_opts();
        opts.app = Some("__pixelpass_s3b_no_matching_application__".to_string());
        opts.strict_audio = true;
        let (health, _) = health::channel();
        let routing = Routing::start(&opts, health.clone())
            .await
            .expect("per-app routing starts");

        // Let the OS thread reach its normal running phase so this covers the
        // tighter steady-state budget rather than only startup containment.
        tokio::time::sleep(Duration::from_millis(100)).await;
        tokio::time::timeout(Duration::from_secs(10), routing.shutdown())
            .await
            .expect("actor and graph teardown stay globally bounded");

        assert!(
            health.fault().is_none(),
            "an observed shutdown and successful join must remain healthy"
        );
        assert_eq!(
            module_snapshot(),
            before,
            "per-app 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() {
        // The regression this fix exists for: a serial one past `u32::MAX`
        // used to parse as `None` and silently disable rerouting.
        let beyond = u64::from(u32::MAX) + 1;
        assert_eq!(parse_object_serial(&beyond.to_string()), Some(beyond));
        assert_eq!(
            parse_object_serial(&u64::MAX.to_string()),
            Some(u64::MAX),
            "the full 64-bit range must round-trip"
        );
    }

    #[test]
    fn object_serial_accepts_ordinary_serials() {
        // Without this the valid cases are only 1, 10 and 20 digits long, and
        // a length-gated mutant (`if (2..10).contains(&raw.len()) { None }`)
        // survives the whole suite while rejecting every serial a freshly
        // started daemon actually hands out. (Codex, round 1.)
        for serial in 0_u64..=1024 {
            assert_eq!(parse_object_serial(&serial.to_string()), Some(serial));
        }
        assert_eq!(parse_object_serial("123456789"), Some(123_456_789));
        assert_eq!(
            parse_object_serial("007"),
            Some(7),
            "leading zeroes are fine"
        );
    }

    #[test]
    fn object_serial_boundary_values() {
        assert_eq!(parse_object_serial("0"), Some(0));
        assert_eq!(parse_object_serial("1"), Some(1));
        let max32 = u64::from(u32::MAX);
        assert_eq!(parse_object_serial(&max32.to_string()), Some(max32));
        assert_eq!(
            parse_object_serial(&(max32 - 1).to_string()),
            Some(max32 - 1)
        );
    }

    #[test]
    fn object_serial_round_trips_through_the_metadata_string() {
        // `try_flush` writes the serial back out as a decimal string for
        // `target.object`; widening must not introduce a formatting change.
        for raw in ["0", "4294967296", "18446744073709551615"] {
            let parsed = parse_object_serial(raw).expect("valid serial");
            assert_eq!(parsed.to_string(), raw);
        }
    }

    #[test]
    fn object_serial_rejects_malformed() {
        for raw in [
            "",
            " 12",
            "12 ",
            "+12",
            "-1",
            "1.0",
            "0x10",
            "12a",
            "abc",
            // u64::MAX + 1 — overflow must be rejected, not wrapped.
            "18446744073709551616",
        ] {
            assert_eq!(parse_object_serial(raw), None, "should reject {raw:?}");
        }
    }
}