//! The pure half of `--repair`: turn one observed module snapshot into an
//! ordered list of unloads, with no I/O and no destruction.
//!
//! # Why this is a separate, pure module
//!
//! Repair destroys server-side state belonging to processes it does not own, so
//! every interesting property is a *decision* property — which pid is dead, which
//! module belongs to it, in what order to unload — and none of them need PipeWire
//! to be exercised. Splitting the decision out means the safety rules below are
//! unit-testable exactly, and the I/O shell in the parent module has nothing left
//! in it worth arguing about.
//!
//! # The discovery rule (phase 0c)
//!
//! Before 0c the capture sink was a `module-null-sink`, so repair could learn a
//! host's pid from that module and then match its loopbacks. After 0c the sink is
//! a **connection-owned native node**: it disappears on its own when the host
//! dies, and it is not a Pulse module at all. A dead host therefore leaves
//! loopbacks with **no null-sink module to learn the pid from**, and the old
//! discovery could not see them — repair did not get smaller, it went blind.
//!
//! So candidate pids are derived **independently from all three shapes**
//! ([`Shape`]), and a shape's absence is never taken as permission to skip
//! another shape's cleanup.
//!
//! # The safety rules, in order of how much damage they prevent
//!
//! 0. **A pid is not an owner.** The same number is a different process in a
//!    different pid namespace, so before liveness can even be *asked*, the module
//!    must prove which machine, boot and namespace its pid belongs to — see
//!    [`OwnerToken`]. Modules that cannot be attributed are never touched, and
//!    their pids are never even looked up: asking is the bug, because the answer
//!    would be meaningless. This rule comes first because it gates the others.
//! 1. **A live pid is never touched**, even if it is not pixelpass. Pid reuse is
//!    real, so "this pid is alive" always wins over "this module looks orphaned".
//!    Leaving a stale module behind is recoverable; unloading a live host's audio
//!    is not. **An *undecidable* pid counts as live** ([`Liveness::Unknown`]):
//!    "I cannot see whether that process exists" must never become "it is dead,
//!    go ahead".
//! 2. **Native nodes are never destroyed.** Repair only ever unloads Pulse
//!    modules it can fingerprint. It has no business touching a live graph object,
//!    and after 0c the sink cleans itself up anyway.
//! 3. **A plan is not a licence.** Pulse module indices are reused verbatim, so an
//!    id planned against one module can name a *different* live module by the time
//!    the unload runs. Every action therefore carries a full [`Fingerprint`] which
//!    the caller must re-verify against a fresh snapshot immediately before each
//!    unload ([`Fingerprint::still_matches`]). Anything that does not match
//!    exactly is skipped, never unloaded.
//! 4. **Ordering is not a licence either.** Planning loopbacks before the sink
//!    they reference is necessary but not sufficient: an unload can *fail* or be
//!    skipped, and a loopback can be created after the plan was made. So the sink
//!    unload is additionally gated at execution time on
//!    [`sink_still_referenced`] against the fresh snapshot — never on the plan's
//!    own ordering having been followed.
//! 5. **Only the canonical forms are ours.** A module is recognised only if its
//!    recorded argument string matches, exactly, what pixelpass itself would have
//!    written ([`Shape::template`]). Recognising "a loopback with one
//!    pixelpass-looking endpoint" would let repair unload a third party's module
//!    that merely names one of our sinks.
//!
//! # Why the templates are generated, not written out
//!
//! The matcher's prefixes and suffixes are derived at runtime from the *same*
//! renderer the loader uses — [`Shape::render_args`], which owns the module name
//! too. Hard-coding `latency_msec=20` in a matcher would mean that changing the
//! loader silently blinds repair to every module the new version loads: the
//! fail-closed-and-silent failure this project has now been bitten by three times.
//! With one source of truth, a loader change moves the matcher with it.
//!
//! The template is only a pre-filter, too. [`classify`] re-renders the pid it
//! extracted and demands byte equality, so the renderer — not a derived pair of
//! strings — is always the authority on what one of our modules looks like.
//!
//! Blindness is also reported rather than assumed impossible:
//! [`unrecognised_pixelpass_modules`] finds modules that name a
//! `pixelpass_capture_*` sink but do **not** match any canonical form, so the I/O
//! shell can say so loudly instead of quietly cleaning up nothing.

use std::collections::{BTreeMap, BTreeSet};

/// Every pixelpass capture sink is named `pixelpass_capture_<pid>`; the pid in
/// that name is the only owner identity these modules carry.
pub const SINK_NAME_PREFIX: &str = "pixelpass_capture_";

/// Loopback latency. Pulse's default of 200 ms is perceptible; 20 ms keeps the
/// mirrored audio tight. Shared with the matcher, so it cannot drift.
pub const LOOPBACK_LATENCY_MSEC: u32 = 20;

/// The capture sink name for a host pid.
pub fn sink_name_for(pid: u32) -> String {
    format!("{SINK_NAME_PREFIX}{pid}")
}

/// How the server records an argument vector we passed as separate argv entries.
///
/// Measured on pactl 17.0 against a live pipewire-pulse: the arguments come back
/// byte-for-byte as passed, joined with single spaces, in the order given, with
/// `@DEFAULT_SINK@` **not** resolved to the concrete device name. Both facts are
/// load-bearing for exact-form matching, so both have their own test.
pub fn recorded_argument(args: &[String]) -> String {
    args.join(" ")
}

/// One module exactly as the server reported it.
///
/// `args` is the **exact** argument string from `pa_module_info`, not a normalised
/// one, and not a reconstruction from a text listing. Normalising would only make
/// two genuinely different arguments compare equal — whitespace inside a quoted
/// property value is not layout — and every snapshot within one invocation comes
/// from the same connection, so there is no re-rendering to absorb.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ModuleObservation {
    pub id: u32,
    pub name: String,
    pub args: String,
}

impl ModuleObservation {
    pub fn new(id: u32, name: &str, args: &str) -> Self {
        Self {
            id,
            name: name.to_string(),
            args: args.to_string(),
        }
    }
}

/// Whether the owner of a module is still around.
///
/// Three states, not two, because "I cannot see that process" and "that process
/// does not exist" are different answers and only one of them permits destroying
/// anything. A pid can be alive and invisible: inside a pid namespace every
/// process in the parent namespace is, and `hidepid` hides others while leaving
/// `self` visible. Collapsing those into "absent" would point the wrong way.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Liveness {
    Alive,
    Dead,
    /// Undecidable. Treated exactly like `Alive` for the purposes of destroying
    /// anything, and reported separately so the user knows repair held back.
    Unknown,
}

/// The three module shapes pixelpass is capable of loading. Each one carries the
/// owner pid in a different place, which is exactly why discovery must consider
/// all three independently.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum Shape {
    /// `module-loopback source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_<pid>
    /// latency_msec=20` — the default-sink mirror, so the viewer hears system
    /// audio.
    LoopbackIntoCapture,
    /// `module-loopback source=pixelpass_capture_<pid>.monitor sink=@DEFAULT_SINK@
    /// latency_msec=20` — the local monitor, so the sharer hears the app they are
    /// sharing.
    LoopbackOutOfCapture,
    /// `module-null-sink sink_name=pixelpass_capture_<pid>` — the legacy capture
    /// sink. Post-0c hosts do not load this at all; it exists for hosts that
    /// predate the connection-owned sink, and for mixed sessions.
    ///
    /// Declared **last** on purpose: `Ord` gives the unload ordering, and the
    /// sink must go after the loopbacks that reference it.
    LegacyCaptureSink,
}

/// Every shape, in unload order.
pub const ALL_SHAPES: [Shape; 3] = [
    Shape::LoopbackIntoCapture,
    Shape::LoopbackOutOfCapture,
    Shape::LegacyCaptureSink,
];

// ──────────────────────────────────────────────────────────────────────
// Ownership
// ──────────────────────────────────────────────────────────────────────

/// The Pulse property every pixelpass module carries its owner token in.
///
/// Measured on the live server before being relied on: all three shapes accept a
/// property-list argument (`sink_properties` / `sink_input_properties` /
/// `source_output_properties`), the recorded argument comes back byte-identical, and
/// the property really does land on the resulting sink, sink-input and
/// source-output.
pub const OWNER_PROPERTY: &str = "pixelpass.owner";

/// Bumped if the token's shape ever changes. A token this build cannot parse is
/// **not** treated as ours, so an older `--repair` meeting a newer token refuses it
/// and reports it rather than guessing.
pub const OWNER_TOKEN_VERSION: u32 = 1;

/// Proof of *which* machine, boot and pid namespace a module's pid refers to.
///
/// # Why a pid alone is not ownership
///
/// The pid in `pixelpass_capture_<pid>` is only a number, and a number means
/// different processes in different pid namespaces. Repair running inside a
/// container that can reach the host's Pulse socket sees a live host's modules,
/// asks about that pid in *its own* namespace, is told nothing exists, and unloads
/// a running host's audio. No negative signal closes that: `NSpid == 1` does not
/// prove the initial namespace, because its leftmost value is relative to whichever
/// procfs was mounted.
///
/// So the module carries the answer with it. If the token's machine, boot and pid
/// namespace all match ours, then its pid is a number we can meaningfully ask
/// about. Otherwise the only safe verdict is [`Liveness::Unknown`].
///
/// The `nonce` is per-load, and it narrows the residual ABA window as a side
/// effect: two loads by the same pid no longer render byte-identical arguments, so
/// a module that vanished and a replacement that took its index are distinguishable.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OwnerToken {
    /// `/etc/machine-id`, dash-free.
    pub machine: String,
    /// `/proc/sys/kernel/random/boot_id`, dash-free — a fresh value per boot.
    pub boot: String,
    /// Inode of `/proc/self/ns/pid`: which pid namespace the pid belongs to.
    pub pid_ns: u64,
    /// Distinguishes two loads that are otherwise identical.
    pub nonce: u64,
}

impl OwnerToken {
    /// `<version>-<machine>-<boot>-<pid_ns>-<nonce>`.
    ///
    /// Every component is dash-free and free of spaces and `=`, so the whole token
    /// is a single unquoted Pulse property value and survives the round trip
    /// through the module's recorded argument untouched.
    pub fn render(&self) -> String {
        format!(
            "{OWNER_TOKEN_VERSION}-{}-{}-{}-{}",
            self.machine, self.boot, self.pid_ns, self.nonce
        )
    }

    /// Parse a token, or `None` for anything this build does not fully understand.
    pub fn parse(raw: &str) -> Option<Self> {
        let mut parts = raw.split('-');
        let version = parts.next()?.parse::<u32>().ok()?;
        if version != OWNER_TOKEN_VERSION {
            return None;
        }
        let machine = parts.next()?;
        let boot = parts.next()?;
        let pid_ns = parts.next()?.parse::<u64>().ok()?;
        let nonce = parts.next()?.parse::<u64>().ok()?;
        if parts.next().is_some() {
            return None;
        }
        // Identities are hex strings; anything else is not a token we wrote.
        let identity_ok = |s: &str| !s.is_empty() && s.bytes().all(|b| b.is_ascii_hexdigit());
        if !identity_ok(machine) || !identity_ok(boot) {
            return None;
        }
        Some(Self {
            machine: machine.to_string(),
            boot: boot.to_string(),
            pid_ns,
            nonce,
        })
    }
}

/// Who *this* process is, for comparison against a module's [`OwnerToken`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LocalIdentity {
    pub machine: String,
    pub boot: String,
    pub pid_ns: u64,
}

impl LocalIdentity {
    /// Does this token describe a pid we can meaningfully ask about?
    ///
    /// All three must agree. A different boot means the pid space has been recycled
    /// wholesale; a different machine means the token came from somewhere else
    /// entirely; a different namespace means the number is not ours to interpret.
    pub fn can_judge(&self, token: &OwnerToken) -> bool {
        self.machine == token.machine && self.boot == token.boot && self.pid_ns == token.pid_ns
    }
}

/// Whether a module proved which pid space its pid belongs to.
///
/// Passed to the liveness callback because the answer changes *how* the question
/// may be asked: an attributed pid needs no guessing about namespaces, while an
/// untagged one is a bare number whose meaning has to be guarded some other way.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Attribution {
    /// Carries a token matching this machine, boot and pid namespace.
    Tokened,
    /// Carries no token; judged by pid alone, and only under
    /// [`UntaggedPolicy::CleanByPidAlone`].
    Untagged,
}

/// What repair may do about modules that carry no token at all.
///
/// Every module loaded before tokens existed is untagged, and there is no way to
/// establish whose pid it names. The default therefore refuses them: leaving an old
/// orphan behind is recoverable, while unloading a live host's routing is not.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UntaggedPolicy {
    /// Untagged modules are [`Liveness::Unknown`]: reported, never unloaded.
    Refuse,
    /// Judge untagged modules by pid liveness alone — the pre-token heuristic,
    /// available only behind an explicit flag.
    CleanByPidAlone,
}

/// Everything the planner needs in order to decide ownership.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Policy {
    pub local: LocalIdentity,
    pub untagged: UntaggedPolicy,
}

/// The pid used to render a shape's argument string when deriving its template.
/// Any value works as long as its decimal form appears exactly once in the
/// rendered arguments, which [`Template::derive`] asserts — unconditionally, so a
/// future shape that repeats the pid cannot slip through a release build.
const TEMPLATE_SENTINEL_PID: u32 = u32::MAX;

/// The token used when deriving a template. Its rendering must not contain the pid
/// sentinel's digits, which [`Template::derive`] also asserts.
fn template_sentinel_token() -> OwnerToken {
    OwnerToken {
        machine: "ffffffffffffffff".to_string(),
        boot: "eeeeeeeeeeeeeeee".to_string(),
        pid_ns: u64::MAX,
        nonce: u64::MAX - 1,
    }
}

/// An exact-match matcher for one shape, derived from that shape's own renderer.
///
/// The template is only ever a *pre-filter*: it locates the candidate pid and token
/// cheaply, and [`classify`] then re-renders both through the real renderer and
/// demands byte equality. So the authority is always the renderer the loader uses,
/// never these derived strings.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Template {
    pub module_name: &'static str,
    prefix: String,
    /// Between the pid and the token. `None` for the legacy, token-less form, which
    /// has only one hole.
    mid: Option<String>,
    suffix: String,
}

impl Template {
    /// Split a shape's rendered arguments around its variable parts.
    fn derive(shape: Shape, tokened: bool) -> Self {
        let sentinel_token = template_sentinel_token();
        let token = tokened.then(|| sentinel_token.clone());
        let rendered = recorded_argument(&shape.render_args(TEMPLATE_SENTINEL_PID, token.as_ref()));
        let pid_sentinel = TEMPLATE_SENTINEL_PID.to_string();

        let (prefix, rest) = rendered
            .split_once(&pid_sentinel)
            .expect("a rendered shape must contain its pid");
        assert!(
            !rest.contains(&pid_sentinel),
            "a shape must name its pid exactly once, but {shape:?} rendered {rendered:?}"
        );

        let (mid, suffix) = match &token {
            None => (None, rest.to_string()),
            Some(token) => {
                let token_sentinel = token.render();
                assert!(
                    !token_sentinel.contains(&pid_sentinel),
                    "the sentinel token must not contain the sentinel pid's digits"
                );
                let (mid, suffix) = rest
                    .split_once(&token_sentinel)
                    .expect("a tokened shape must contain its token");
                assert!(
                    !suffix.contains(&token_sentinel),
                    "a shape must carry its token exactly once, but {shape:?} rendered \
                     {rendered:?}"
                );
                (Some(mid.to_string()), suffix.to_string())
            }
        };

        Self {
            module_name: shape.module_name(),
            prefix: prefix.to_string(),
            mid,
            suffix,
        }
    }

    /// The pid and token this argument string names, if it is *exactly* this shape.
    ///
    /// Total: the argument must equal `prefix ++ pid ++ mid ++ token ++ suffix` with
    /// nothing left over. A canonical decimal pid is required — no sign, no leading
    /// zeroes, no whitespace — because `u32::from_str` accepts a leading `+`, and
    /// because we only ever render a pid one way, so `pixelpass_capture_007` is not
    /// a name we wrote.
    pub fn parse(&self, args: &str) -> Option<(u32, Option<OwnerToken>)> {
        let rest = args.strip_prefix(self.prefix.as_str())?;
        let (digits, token) = match &self.mid {
            None => (rest.strip_suffix(self.suffix.as_str())?, None),
            Some(mid) => {
                let (digits, after) = rest.split_once(mid.as_str())?;
                let raw = after.strip_suffix(self.suffix.as_str())?;
                (digits, Some(OwnerToken::parse(raw)?))
            }
        };
        if digits.is_empty() || !digits.bytes().all(|b| b.is_ascii_digit()) {
            return None;
        }
        if digits.len() > 1 && digits.starts_with('0') {
            return None;
        }
        Some((digits.parse::<u32>().ok()?, token))
    }
}

impl Shape {
    /// The Pulse module this shape loads. Owned here rather than written out at
    /// the call site, so the module *name* is as drift-proof as the arguments —
    /// the loader asks for it too.
    pub fn module_name(self) -> &'static str {
        match self {
            Shape::LoopbackIntoCapture | Shape::LoopbackOutOfCapture => "module-loopback",
            Shape::LegacyCaptureSink => "module-null-sink",
        }
    }

    /// Which property-list argument this shape carries its owner token in.
    fn owner_property_argument(self) -> &'static str {
        match self {
            // The mirror's stream is a sink-input on our capture sink.
            Shape::LoopbackIntoCapture => "sink_input_properties",
            // The local monitor's stream is a source-output on our capture monitor.
            Shape::LoopbackOutOfCapture => "source_output_properties",
            // The sink itself carries the property.
            Shape::LegacyCaptureSink => "sink_properties",
        }
    }

    /// The exact `pactl load-module` arguments for this shape, for `pid`, optionally
    /// carrying an owner token.
    ///
    /// **This is the single source of truth.** `host/audio.rs` loads through it and
    /// `--repair` matches through it, so a change here moves both at once.
    ///
    /// `token: None` renders the **legacy** form — what every pixelpass before
    /// ownership tokens loaded. It is still rendered, because repair must be able to
    /// recognise those modules in order to report them.
    pub fn render_args(self, pid: u32, token: Option<&OwnerToken>) -> Vec<String> {
        let sink = sink_name_for(pid);
        let mut args = match self {
            // The default-sink mirror: the viewer hears system audio.
            Shape::LoopbackIntoCapture => vec![
                "source=@DEFAULT_SINK@.monitor".to_string(),
                format!("sink={sink}"),
                format!("latency_msec={LOOPBACK_LATENCY_MSEC}"),
            ],
            // The local monitor: the sharer hears the app they are sharing.
            Shape::LoopbackOutOfCapture => vec![
                format!("source={sink}.monitor"),
                "sink=@DEFAULT_SINK@".to_string(),
                format!("latency_msec={LOOPBACK_LATENCY_MSEC}"),
            ],
            // The legacy capture sink (pre-0c hosts only).
            Shape::LegacyCaptureSink => vec![format!("sink_name={sink}")],
        };
        if let Some(token) = token {
            args.push(format!(
                "{}={OWNER_PROPERTY}={}",
                self.owner_property_argument(),
                token.render()
            ));
        }
        args
    }

    /// The exact-match pre-filters for this shape, tokened form first.
    ///
    /// Both are generated from `render_args`, so neither can drift from the loader.
    pub fn templates(self) -> [Template; 2] {
        [Template::derive(self, true), Template::derive(self, false)]
    }

    /// Human label for reporting.
    pub fn label(self) -> &'static str {
        match self {
            Shape::LoopbackIntoCapture => "default-sink mirror",
            Shape::LoopbackOutOfCapture => "local monitor",
            Shape::LegacyCaptureSink => "legacy capture sink",
        }
    }
}

/// Everything that must *still* be true of a module at the moment it is
/// unloaded — not merely when the plan was made.
///
/// It is an identity of the *observable* module, not of a generation: if the
/// planned module vanishes and a byte-identical one takes its index, this compares
/// equal. That residual ABA window cannot be closed through an unload API whose only
/// argument is an index; what narrows it is the per-load nonce in [`OwnerToken`]
/// (two loads no longer render identical arguments), and what closes the rest in
/// practice is the liveness recheck nearer the unload.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Fingerprint {
    pub id: u32,
    pub module_name: String,
    /// The exact recorded argument string.
    pub args: String,
    pub pid: u32,
    pub shape: Shape,
    /// The owner token, or `None` for a legacy module loaded before tokens existed.
    /// A pid without one cannot be attributed to a namespace — see [`OwnerToken`].
    pub owner: Option<OwnerToken>,
}

impl Fingerprint {
    /// Does `obs` still present the same observable module this fingerprint was
    /// taken from?
    ///
    /// Deliberately total: id, module name, exact args, derived pid and shape must
    /// all agree. A module index that has been reused fails on the name or the
    /// args; a module whose arguments were rewritten fails on the args. Either way
    /// the caller must skip it rather than guess.
    pub fn still_matches(&self, obs: &ModuleObservation) -> bool {
        classify(obs).as_ref() == Some(self)
    }
}

/// The result of planning: what to unload, and what was deliberately left alone.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Plan {
    /// In execution order: every loopback before any legacy sink.
    pub unload: Vec<Fingerprint>,
    /// Pids that are still alive and were therefore skipped entirely. Includes
    /// this process and any other running pixelpass.
    pub live_pids: BTreeSet<u32>,
    /// Pids we concluded are gone.
    pub dead_pids: BTreeSet<u32>,
    /// Pids whose liveness could not be determined. Skipped like live ones, but
    /// reported separately: this is the case where repair is not safe rather than
    /// not needed.
    pub unknown_pids: BTreeSet<u32>,
    /// Modules recognised as ours but carrying no owner token, while the policy
    /// refuses to judge those. Reported so the user can see what an explicit
    /// legacy-cleanup run would act on.
    pub untagged: Vec<Fingerprint>,
    /// Modules whose token belongs to another machine, boot or pid namespace. Their
    /// pids are numbers this process cannot interpret, so they are never touched.
    pub foreign: Vec<Fingerprint>,
}

impl Plan {
    pub fn is_empty(&self) -> bool {
        self.unload.is_empty()
    }
}

/// Recognise one of pixelpass's three module shapes, or `None` for a module that
/// is not ours.
///
/// Exact-form only. A loopback that merely mentions one of our sink names — say a
/// third-party controller's `module-loopback source=some_mic
/// sink=pixelpass_capture_4242` — is **not** ours and must never be unloaded, and
/// a `sink=` token nested inside a quoted `sink_input_properties` value cannot be
/// mistaken for a top-level argument because the whole string must match.
///
/// The template only proposes a pid; the shape's own renderer decides. Re-rendering
/// and demanding byte equality means the loader is the authority, so a shape that
/// grows an argument, changes a latency, or repeats the pid cannot leave a matcher
/// quietly accepting the old form.
pub fn classify(obs: &ModuleObservation) -> Option<Fingerprint> {
    for shape in ALL_SHAPES {
        for template in shape.templates() {
            if obs.name != template.module_name {
                continue;
            }
            let Some((pid, owner)) = template.parse(&obs.args) else {
                continue;
            };
            if recorded_argument(&shape.render_args(pid, owner.as_ref())) != obs.args {
                continue;
            }
            return Some(Fingerprint {
                id: obs.id,
                module_name: obs.name.clone(),
                args: obs.args.clone(),
                pid,
                shape,
                owner,
            });
        }
    }
    None
}

/// Modules that name a `pixelpass_capture_*` sink but match no canonical form.
///
/// These are never touched. They exist to be *reported*: either a third party is
/// naming our sinks, or a newer pixelpass loads a shape this build does not
/// recognise. The second case is how repair would go silently blind, so it gets
/// said out loud instead of inferred from a clean exit.
pub fn unrecognised_pixelpass_modules(
    observations: &[ModuleObservation],
) -> Vec<&ModuleObservation> {
    observations
        .iter()
        .filter(|obs| obs.args.contains(SINK_NAME_PREFIX) && classify(obs).is_none())
        .collect()
}

/// Is anything else in this snapshot still attached to `pid`'s capture sink?
///
/// Returns the id of the first module that references it. Deliberately textual and
/// broad — any mention of the sink name by any *other* module counts, canonical or
/// not — because the question here is "would destroying this sink break something
/// that is attached to it", not "is that attachment ours". Answering it wrongly in
/// the permissive direction is the one thing rule 4 exists to prevent.
pub fn sink_still_referenced(
    observations: &[ModuleObservation],
    pid: u32,
    sink_module_id: u32,
) -> Option<u32> {
    let sink_name = sink_name_for(pid);
    observations
        .iter()
        .find(|obs| obs.id != sink_module_id && obs.args.contains(&sink_name))
        .map(|obs| obs.id)
}

/// Turn one snapshot into an ordered unload plan.
///
/// `liveness` is injected rather than read from `/proc` so the decision is
/// testable, and so the caller can re-ask at execution time — this plan is
/// evidence, not permission.
pub fn plan(
    observations: &[ModuleObservation],
    policy: &Policy,
    liveness: impl Fn(u32, Attribution) -> Liveness,
) -> Plan {
    // Deduplicate by module id: a snapshot should not repeat one, but a repeated
    // entry must not become a repeated unload of an index that has since been
    // reused by someone else.
    let mut seen: BTreeMap<u32, Fingerprint> = BTreeMap::new();
    for obs in observations {
        if let Some(fp) = classify(obs) {
            seen.entry(fp.id).or_insert(fp);
        }
    }

    // Ownership first, because it decides whether the pid is even a question we can
    // ask. A token from another machine, boot or pid namespace names a process this
    // one cannot see, and an absent token names nothing at all.
    let mut untagged: Vec<Fingerprint> = Vec::new();
    let mut foreign: Vec<Fingerprint> = Vec::new();
    let mut judgeable: Vec<(Fingerprint, Attribution)> = Vec::new();
    for fp in seen.into_values() {
        match &fp.owner {
            Some(token) if policy.local.can_judge(token) => {
                judgeable.push((fp, Attribution::Tokened))
            }
            Some(_) => foreign.push(fp),
            None => match policy.untagged {
                UntaggedPolicy::CleanByPidAlone => judgeable.push((fp, Attribution::Untagged)),
                UntaggedPolicy::Refuse => untagged.push(fp),
            },
        }
    }

    // Liveness is asked once per distinct pid, not once per module: a host with
    // three modules must not be able to change its own verdict mid-plan.
    let mut live_pids = BTreeSet::new();
    let mut dead_pids = BTreeSet::new();
    let mut unknown_pids = BTreeSet::new();
    for (pid, attribution) in judgeable
        .iter()
        .map(|(fp, attribution)| (fp.pid, *attribution))
        .collect::<BTreeMap<_, _>>()
    {
        match liveness(pid, attribution) {
            Liveness::Alive => live_pids.insert(pid),
            Liveness::Dead => dead_pids.insert(pid),
            Liveness::Unknown => unknown_pids.insert(pid),
        };
    }

    let mut unload: Vec<Fingerprint> = judgeable
        .into_iter()
        .map(|(fp, _)| fp)
        .filter(|fp| dead_pids.contains(&fp.pid))
        .collect();
    // `Shape`'s declaration order is the unload order: loopbacks before the sink
    // they reference. Id breaks ties so the plan is deterministic.
    unload.sort_by_key(|fp| (fp.shape, fp.id));
    untagged.sort_by_key(|fp| fp.id);
    foreign.sort_by_key(|fp| fp.id);

    Plan {
        unload,
        live_pids,
        dead_pids,
        unknown_pids,
        untagged,
        foreign,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The identity this test process pretends to be.
    fn our_identity() -> LocalIdentity {
        LocalIdentity {
            machine: "aa11bb22".to_string(),
            boot: "cc33dd44".to_string(),
            pid_ns: 4_026_531_836,
        }
    }

    /// A token minted by "us".
    fn our_token(nonce: u64) -> OwnerToken {
        let local = our_identity();
        OwnerToken {
            machine: local.machine,
            boot: local.boot,
            pid_ns: local.pid_ns,
            nonce,
        }
    }

    /// The default policy: our identity, untagged modules refused.
    fn our_policy() -> Policy {
        Policy {
            local: our_identity(),
            untagged: UntaggedPolicy::Refuse,
        }
    }

    fn obs(id: u32, shape: Shape, pid: u32, token: Option<&OwnerToken>) -> ModuleObservation {
        ModuleObservation::new(
            id,
            shape.module_name(),
            &recorded_argument(&shape.render_args(pid, token)),
        )
    }

    fn null_sink(id: u32, pid: u32) -> ModuleObservation {
        obs(id, Shape::LegacyCaptureSink, pid, Some(&our_token(1)))
    }

    fn mirror(id: u32, pid: u32) -> ModuleObservation {
        obs(id, Shape::LoopbackIntoCapture, pid, Some(&our_token(2)))
    }

    fn local_monitor(id: u32, pid: u32) -> ModuleObservation {
        obs(id, Shape::LoopbackOutOfCapture, pid, Some(&our_token(3)))
    }

    /// The pre-token form: recognisable as ours, but attributable to nobody.
    fn legacy_mirror(id: u32, pid: u32) -> ModuleObservation {
        obs(id, Shape::LoopbackIntoCapture, pid, None)
    }

    fn nothing_is_alive(_: u32, _: Attribution) -> Liveness {
        Liveness::Dead
    }

    fn ids(plan: &Plan) -> Vec<u32> {
        plan.unload.iter().map(|fp| fp.id).collect()
    }

    /// The renderers are the contract with the live server. These strings were
    /// measured on pactl 17.0 / pipewire-pulse: arguments come back joined with
    /// single spaces, in order, with `@DEFAULT_SINK@` unresolved. If this test is
    /// ever changed, the matcher's exactness claim has to be re-measured.
    #[test]
    fn the_canonical_argument_strings_are_what_the_server_records() {
        assert_eq!(
            recorded_argument(&Shape::LoopbackIntoCapture.render_args(4242, None)),
            "source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=20"
        );
        assert_eq!(
            recorded_argument(&Shape::LoopbackOutOfCapture.render_args(4242, None)),
            "source=pixelpass_capture_4242.monitor sink=@DEFAULT_SINK@ latency_msec=20"
        );
        assert_eq!(
            recorded_argument(&Shape::LegacyCaptureSink.render_args(4242, None)),
            "sink_name=pixelpass_capture_4242"
        );
        // And the tokened forms, which is what a host actually loads. These exact
        // strings were verified against the live server: all three shapes accept the
        // property argument and record it byte-identically.
        let token = our_token(7);
        assert_eq!(
            recorded_argument(&Shape::LoopbackIntoCapture.render_args(4242, Some(&token))),
            format!(
                "source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=20 \
                 sink_input_properties=pixelpass.owner={}",
                token.render()
            )
        );
        assert_eq!(
            recorded_argument(&Shape::LoopbackOutOfCapture.render_args(4242, Some(&token))),
            format!(
                "source=pixelpass_capture_4242.monitor sink=@DEFAULT_SINK@ latency_msec=20 \
                 source_output_properties=pixelpass.owner={}",
                token.render()
            )
        );
        assert_eq!(
            recorded_argument(&Shape::LegacyCaptureSink.render_args(4242, Some(&token))),
            format!(
                "sink_name=pixelpass_capture_4242 sink_properties=pixelpass.owner={}",
                token.render()
            )
        );
    }

    /// Every shape must round-trip through its own generated template. This is
    /// what makes the loader and the matcher one source of truth: change a
    /// renderer and this fails unless the template follows.
    #[test]
    fn every_shape_round_trips_through_its_generated_template() {
        for shape in ALL_SHAPES {
            let [tokened, legacy] = shape.templates();
            for pid in [1_u32, 7, 4242, 999_999, u32::MAX - 1] {
                let plain = recorded_argument(&shape.render_args(pid, None));
                assert_eq!(
                    legacy.parse(&plain),
                    Some((pid, None)),
                    "{shape:?} failed to round-trip legacy pid {pid}"
                );

                let token = our_token(u64::from(pid));
                let tagged = recorded_argument(&shape.render_args(pid, Some(&token)));
                assert_eq!(
                    tokened.parse(&tagged),
                    Some((pid, Some(token))),
                    "{shape:?} failed to round-trip tokened pid {pid}"
                );
            }
        }
    }

    /// The defect 0c introduces and this rewrite exists for: a dead host whose
    /// capture sink was connection-owned leaves loopbacks behind with **no**
    /// null-sink module to learn its pid from. The old discovery derived dead
    /// pids only from `module-null-sink`, so it found nothing here.
    #[test]
    fn orphan_loopbacks_are_found_without_any_null_sink() {
        let modules = [mirror(10, 4242), local_monitor(11, 4242)];
        let plan = plan(&modules, &our_policy(), nothing_is_alive);
        assert_eq!(ids(&plan), vec![10, 11]);
        assert_eq!(plan.dead_pids, BTreeSet::from([4242]));
    }

    /// Either loopback shape alone must be enough to identify the owner — the
    /// local monitor names the capture sink only as its *source*.
    #[test]
    fn each_loopback_shape_identifies_the_owner_on_its_own() {
        assert_eq!(
            ids(&plan(
                &[local_monitor(11, 7)],
                &our_policy(),
                nothing_is_alive
            )),
            vec![11]
        );
        assert_eq!(
            ids(&plan(&[mirror(10, 7)], &our_policy(), nothing_is_alive)),
            vec![10]
        );
    }

    /// The legacy shape still works, and the sink unloads *after* both
    /// loopbacks that reference it.
    #[test]
    fn legacy_sink_unloads_after_the_loopbacks_that_reference_it() {
        // Deliberately snapshot-ordered sink-first, so passing requires the
        // plan to reorder rather than to preserve input order.
        let modules = [
            null_sink(5, 4242),
            local_monitor(11, 4242),
            mirror(10, 4242),
        ];
        let plan = plan(&modules, &our_policy(), nothing_is_alive);
        assert_eq!(ids(&plan), vec![10, 11, 5]);
        assert_eq!(
            plan.unload.last().unwrap().shape,
            Shape::LegacyCaptureSink,
            "the sink must be last or PipeWire is asked to destroy a sink with a live loopback"
        );
    }

    /// A live pid is never touched, even when a dead one is being cleaned up in
    /// the same run.
    #[test]
    fn a_live_host_is_left_alone_while_a_dead_one_is_cleaned() {
        let modules = [
            null_sink(5, 100),
            mirror(10, 100),
            null_sink(6, 200),
            mirror(12, 200),
        ];
        let plan = plan(&modules, &our_policy(), |pid, _| {
            if pid == 200 {
                Liveness::Alive
            } else {
                Liveness::Dead
            }
        });
        assert_eq!(ids(&plan), vec![10, 5]);
        assert_eq!(plan.live_pids, BTreeSet::from([200]));
        assert_eq!(plan.dead_pids, BTreeSet::from([100]));
    }

    /// Undecidable liveness must behave exactly like alive. `/proc` answering
    /// "no" because of a pid namespace, a permission error or `hidepid` is the
    /// one way a wrong verdict destroys a *running* host's audio.
    #[test]
    fn an_undecidable_pid_is_never_touched() {
        let modules = [null_sink(5, 100), mirror(10, 100), local_monitor(11, 200)];
        let plan = plan(&modules, &our_policy(), |pid, _| {
            if pid == 100 {
                Liveness::Unknown
            } else {
                Liveness::Dead
            }
        });
        assert_eq!(
            ids(&plan),
            vec![11],
            "only the decidably-dead pid is planned"
        );
        assert_eq!(plan.unknown_pids, BTreeSet::from([100]));
        assert!(plan.dead_pids.contains(&200));
        assert!(
            !plan.dead_pids.contains(&100),
            "unknown must not be recorded as dead"
        );
    }

    /// Two live hosts: repair must be a complete no-op. This is the unit-level
    /// half of the two-host gate — the live half still has to be run for real.
    #[test]
    fn two_live_hosts_produce_no_actions_at_all() {
        let modules = [
            null_sink(5, 100),
            mirror(10, 100),
            local_monitor(11, 100),
            null_sink(6, 200),
            mirror(12, 200),
        ];
        let plan = plan(&modules, &our_policy(), |_, _| Liveness::Alive);
        assert!(plan.is_empty(), "no live host may be touched: {plan:?}");
        assert_eq!(plan.live_pids, BTreeSet::from([100, 200]));
        assert!(plan.dead_pids.is_empty());
    }

    /// Modules belonging to anything else are invisible to repair, including
    /// near-misses that mention a sink we do not own.
    #[test]
    fn unrelated_and_malformed_modules_are_never_planned() {
        let modules = [
            ModuleObservation::new(1, "module-null-sink", "sink_name=some_other_sink"),
            ModuleObservation::new(
                2,
                "module-loopback",
                "source=alsa_output.pci.monitor sink=x",
            ),
            ModuleObservation::new(3, "module-echo-cancel", "sink_name=pixelpass_capture_9"),
            // Our prefix, but no parseable pid — we do not guess an owner.
            ModuleObservation::new(4, "module-null-sink", "sink_name=pixelpass_capture_"),
            ModuleObservation::new(5, "module-null-sink", "sink_name=pixelpass_capture_abc"),
            ModuleObservation::new(6, "module-null-sink", "sink_name=pixelpass_capture_-1"),
            // `u32::from_str` accepts a leading `+`; a name we wrote never has one.
            ModuleObservation::new(7, "module-null-sink", "sink_name=pixelpass_capture_+1"),
            // Leading zeroes are not a name we render.
            ModuleObservation::new(8, "module-null-sink", "sink_name=pixelpass_capture_007"),
            ModuleObservation::new(9, "module-loopback", "latency_msec=20"),
        ];
        let plan = plan(&modules, &our_policy(), nothing_is_alive);
        assert!(plan.is_empty(), "{plan:?}");
        assert!(plan.dead_pids.is_empty(), "no owner may be invented");
    }

    /// A loopback that merely *names* one of our sinks is not ours. This is the
    /// case where over-eager recognition would unload a live third party's
    /// module: only one endpoint is a pixelpass name, so no canonical form
    /// matches.
    #[test]
    fn a_loopback_with_only_one_pixelpass_endpoint_is_not_ours() {
        let modules = [
            // A third-party controller routing some microphone into our sink.
            ModuleObservation::new(
                20,
                "module-loopback",
                "source=some_mic sink=pixelpass_capture_4242 latency_msec=20",
            ),
            // Our sink's monitor into somewhere that is not the default sink.
            ModuleObservation::new(
                21,
                "module-loopback",
                "source=pixelpass_capture_4242.monitor sink=other_sink latency_msec=20",
            ),
            // The canonical shape with a different latency — a version of
            // pixelpass this build does not know how to recognise.
            ModuleObservation::new(
                22,
                "module-loopback",
                "source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=50",
            ),
            // Extra arguments appended: not the string we write.
            ModuleObservation::new(
                23,
                "module-loopback",
                "source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=20 \
                 remix=false",
            ),
        ];
        let plan = plan(&modules, &our_policy(), nothing_is_alive);
        assert!(plan.is_empty(), "none of these are ours: {plan:?}");
        // …but every one of them is *reported*, so blindness is never silent.
        assert_eq!(unrecognised_pixelpass_modules(&modules).len(), 4);
    }

    /// A pixelpass-looking token nested inside a quoted property value must not
    /// promote a foreign module to ours. `module-loopback` really does accept
    /// `sink_input_properties`, so this argument is legal.
    #[test]
    fn a_nested_quoted_property_is_not_a_top_level_argument() {
        let obs = ModuleObservation::new(
            30,
            "module-loopback",
            "source=some_mic sink=real_sink \
             sink_input_properties=\"media.name=sink=pixelpass_capture_4242\" latency_msec=20",
        );
        assert_eq!(classify(&obs), None);
        assert!(plan(&[obs], &our_policy(), nothing_is_alive).is_empty());
    }

    /// A repeated observation of one module must not become two unloads of an
    /// index that may have been reused between them.
    #[test]
    fn a_duplicated_observation_yields_one_action() {
        let modules = [mirror(10, 4242), mirror(10, 4242)];
        assert_eq!(
            ids(&plan(&modules, &our_policy(), nothing_is_alive)),
            vec![10]
        );
    }

    /// Liveness is asked once per pid. Without this, a `liveness` that flips
    /// mid-plan could unload some of a host's modules and keep others — the
    /// worst possible outcome, since a half-repaired host is neither working
    /// nor cleanable. Two pids, counted separately: one pid cannot prove
    /// "once *per* pid".
    #[test]
    fn liveness_is_decided_once_per_pid_not_once_per_module() {
        use std::cell::RefCell;
        let calls: RefCell<BTreeMap<u32, u32>> = RefCell::new(BTreeMap::new());
        let modules = [
            null_sink(5, 42),
            mirror(10, 42),
            local_monitor(11, 42),
            mirror(12, 99),
            local_monitor(13, 99),
        ];
        let plan = plan(&modules, &our_policy(), |pid, _| {
            *calls.borrow_mut().entry(pid).or_insert(0) += 1;
            Liveness::Dead
        });
        let calls = calls.into_inner();
        assert_eq!(calls.get(&42), Some(&1), "one question for pid 42");
        assert_eq!(calls.get(&99), Some(&1), "one question for pid 99");
        assert_eq!(calls.len(), 2, "no pid asked that we have no module for");
        assert_eq!(ids(&plan), vec![10, 12, 11, 13, 5]);
    }

    /// Re-verification: the same module still matches, and a reused index
    /// carrying a different module does not.
    #[test]
    fn a_fingerprint_only_matches_the_module_it_was_taken_from() {
        let obs = mirror(10, 4242);
        let fp = classify(&obs).expect("ours");

        assert!(fp.still_matches(&obs));
        // Same index, someone else's module — the reuse case that makes a plan
        // unsafe to execute blind.
        assert!(!fp.still_matches(&mirror(10, 9999)));
        assert!(!fp.still_matches(&null_sink(10, 4242)));
        // Same module, different index.
        assert!(!fp.still_matches(&mirror(11, 4242)));
        // Same identity, arguments rewritten.
        assert!(!fp.still_matches(&ModuleObservation::new(
            10,
            "module-loopback",
            "source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=200"
        )));
        // Same shape and pid, but observed under the *other* loopback shape.
        assert!(!fp.still_matches(&local_monitor(10, 4242)));
    }

    /// The module *name* is part of the identity, and this is provable because the
    /// argument grammars overlap: `module-echo-cancel sink_name=pixelpass_capture_9`
    /// is byte-identical to the canonical null-sink argument. So a comparator using
    /// only `id + args` accepts a foreign module — and unloads it.
    ///
    /// (I previously argued this case could not be constructed non-vacuously, on the
    /// grounds that the name determines which grammar can match. That was wrong: the
    /// grammars are not disjoint across names.)
    #[test]
    fn an_identical_argument_under_another_module_name_is_not_a_match() {
        let ours = null_sink(5, 9);
        let fp = classify(&ours).expect("ours");
        let impostor = ModuleObservation::new(5, "module-echo-cancel", &ours.args);

        assert_eq!(fp.args, impostor.args, "the arguments really are identical");
        assert_eq!(fp.id, impostor.id, "and so is the index");
        assert!(
            !fp.still_matches(&impostor),
            "only the module name distinguishes these, so it must be compared"
        );
        assert_eq!(
            classify(&impostor),
            None,
            "and it is not ours to begin with"
        );
    }

    /// Whitespace is identity, not layout. The exact recorded argument is
    /// compared, so a re-spaced string is a *different* argument — inside a
    /// quoted property value that difference can be semantic.
    #[test]
    fn respaced_arguments_do_not_match() {
        let fp = classify(&mirror(10, 4242)).expect("ours");
        assert!(!fp.still_matches(&ModuleObservation::new(
            10,
            "module-loopback",
            "source=@DEFAULT_SINK@.monitor  sink=pixelpass_capture_4242 latency_msec=20"
        )));
    }

    /// Rule 4: the sink's unload is gated on nothing else being attached, not on
    /// the plan's ordering having been carried out. A loopback that failed to
    /// unload — or one created after planning — must block it.
    #[test]
    fn a_remaining_loopback_blocks_the_sink_unload() {
        // The mirror is still there (its unload failed, or it arrived late).
        let snapshot = [null_sink(5, 4242), mirror(10, 4242)];
        assert_eq!(sink_still_referenced(&snapshot, 4242, 5), Some(10));

        // A foreign module attached to our sink blocks it too: the question is
        // what would break, not who owns it.
        let foreign = [
            null_sink(5, 4242),
            ModuleObservation::new(
                77,
                "module-loopback",
                "source=some_mic sink=pixelpass_capture_4242 latency_msec=20",
            ),
        ];
        assert_eq!(sink_still_referenced(&foreign, 4242, 5), Some(77));

        // Nothing else attached: clear to unload. The sink's own argument names
        // itself, which must not count.
        let alone = [null_sink(5, 4242)];
        assert_eq!(sink_still_referenced(&alone, 4242, 5), None);

        // Another host's loopback is not a reference to *this* sink.
        let other_host = [null_sink(5, 4242), mirror(10, 9999)];
        assert_eq!(sink_still_referenced(&other_host, 4242, 5), None);
    }

    /// A token must survive the round trip through a module argument exactly, and
    /// anything this build does not fully understand must not parse at all.
    #[test]
    fn tokens_round_trip_and_reject_what_they_do_not_understand() {
        let token = our_token(42);
        assert_eq!(OwnerToken::parse(&token.render()), Some(token.clone()));

        for raw in [
            "",
            "1",
            "1-aa11bb22",
            "1-aa11bb22-cc33dd44",
            "1-aa11bb22-cc33dd44-4026531836",
            // A version this build does not know: not ours to touch.
            "2-aa11bb22-cc33dd44-4026531836-42",
            "0-aa11bb22-cc33dd44-4026531836-42",
            // Trailing junk.
            "1-aa11bb22-cc33dd44-4026531836-42-extra",
            // Non-hex identities.
            "1-zzzz-cc33dd44-4026531836-42",
            "1-aa11bb22-zzzz-4026531836-42",
            // Non-numeric namespace or nonce.
            "1-aa11bb22-cc33dd44-abc-42",
            "1-aa11bb22-cc33dd44-4026531836-abc",
            // Empty identity components.
            "1--cc33dd44-4026531836-42",
        ] {
            assert_eq!(OwnerToken::parse(raw), None, "should reject {raw:?}");
        }
    }

    /// The defect the token exists for: a pid means different processes in different
    /// pid namespaces, so a module whose token names another namespace, boot or
    /// machine must never be judged by asking about that number here.
    #[test]
    fn a_module_from_another_namespace_boot_or_machine_is_never_touched() {
        let local = our_identity();
        let elsewhere = [
            // Same machine and boot, different pid namespace: the number is not ours.
            OwnerToken {
                pid_ns: local.pid_ns + 1,
                ..our_token(1)
            },
            // Same machine, earlier boot: the whole pid space has been recycled.
            OwnerToken {
                boot: "ffffffff".to_string(),
                ..our_token(1)
            },
            // Another machine entirely.
            OwnerToken {
                machine: "99998888".to_string(),
                ..our_token(1)
            },
        ];

        for token in elsewhere {
            let modules = [obs(10, Shape::LoopbackIntoCapture, 4242, Some(&token))];
            // `nothing_is_alive` would happily call the pid dead, so a plan that
            // consults liveness at all here is already wrong.
            let plan = plan(&modules, &our_policy(), nothing_is_alive);
            assert!(
                plan.is_empty(),
                "a module from {token:?} must not be planned: {plan:?}"
            );
            assert_eq!(plan.foreign.len(), 1, "and it must be reported: {plan:?}");
            assert!(
                plan.dead_pids.is_empty(),
                "its pid must not even be considered"
            );
        }
    }

    /// Liveness must not be consulted at all for a module we cannot attribute —
    /// asking is the bug, because the answer is meaningless.
    #[test]
    fn an_unattributable_module_is_never_asked_about() {
        let foreign = OwnerToken {
            pid_ns: our_identity().pid_ns + 1,
            ..our_token(1)
        };
        let modules = [
            obs(10, Shape::LoopbackIntoCapture, 4242, Some(&foreign)),
            legacy_mirror(11, 5555),
        ];
        let asked = std::cell::RefCell::new(Vec::new());
        let plan = plan(&modules, &our_policy(), |pid, _| {
            asked.borrow_mut().push(pid);
            Liveness::Dead
        });
        assert!(
            asked.borrow().is_empty(),
            "no pid should have been asked about, but these were: {:?}",
            asked.borrow()
        );
        assert!(plan.is_empty());
        assert_eq!(plan.foreign.len(), 1);
        assert_eq!(plan.untagged.len(), 1);
    }

    /// Untagged (pre-token) modules are refused by default and reported, and only an
    /// explicit policy judges them by pid alone.
    #[test]
    fn untagged_modules_are_refused_by_default_and_only_cleaned_on_request() {
        let modules = [
            legacy_mirror(10, 4242),
            obs(11, Shape::LegacyCaptureSink, 4242, None),
        ];

        let refused = plan(&modules, &our_policy(), nothing_is_alive);
        assert!(
            refused.is_empty(),
            "default must not touch them: {refused:?}"
        );
        assert_eq!(refused.untagged.len(), 2);
        assert!(refused.dead_pids.is_empty());

        let opted_in = plan(
            &modules,
            &Policy {
                local: our_identity(),
                untagged: UntaggedPolicy::CleanByPidAlone,
            },
            nothing_is_alive,
        );
        assert_eq!(ids(&opted_in), vec![10, 11], "{opted_in:?}");
        assert!(opted_in.untagged.is_empty());
        // Even opted in, a live pid still wins.
        let live = plan(
            &modules,
            &Policy {
                local: our_identity(),
                untagged: UntaggedPolicy::CleanByPidAlone,
            },
            |_, _| Liveness::Alive,
        );
        assert!(live.is_empty(), "{live:?}");
    }

    /// The nonce narrows the ABA window: two loads by the same pid no longer render
    /// identical arguments, so a fingerprint taken from one does not match the other.
    #[test]
    fn the_nonce_distinguishes_two_loads_by_the_same_pid() {
        let first = obs(10, Shape::LoopbackIntoCapture, 4242, Some(&our_token(1)));
        let second = obs(10, Shape::LoopbackIntoCapture, 4242, Some(&our_token(2)));
        assert_ne!(first.args, second.args, "the nonce must reach the argument");

        let fp = classify(&first).expect("ours");
        assert!(fp.still_matches(&first));
        assert!(
            !fp.still_matches(&second),
            "a different load must not satisfy the first load's fingerprint"
        );
    }

    /// The attribution handed to the liveness callback decides *how* the question
    /// may be asked, so it must be right per module. Getting this wrong made the
    /// token useless in a container — the one place it exists for — because the
    /// probe's namespace guesswork answered `Unknown` for a pid the token had
    /// already proven local.
    #[test]
    fn attribution_is_reported_per_module() {
        let modules = [mirror(10, 100), legacy_mirror(11, 200)];
        let asked = std::cell::RefCell::new(Vec::new());
        let plan = plan(
            &modules,
            &Policy {
                local: our_identity(),
                untagged: UntaggedPolicy::CleanByPidAlone,
            },
            |pid, attribution| {
                asked.borrow_mut().push((pid, attribution));
                Liveness::Dead
            },
        );
        let asked = asked.into_inner();
        assert_eq!(
            asked,
            vec![(100, Attribution::Tokened), (200, Attribution::Untagged)],
            "each pid must be asked about with its own module's attribution"
        );
        assert_eq!(ids(&plan), vec![10, 11]);
    }

    /// Tokened and legacy forms must both classify, and carry the difference.
    #[test]
    fn both_forms_classify_and_record_whether_they_are_attributable() {
        let tokened = classify(&mirror(10, 4242)).expect("tokened is ours");
        assert_eq!(tokened.owner, Some(our_token(2)));

        let legacy = classify(&legacy_mirror(11, 4242)).expect("legacy is still ours");
        assert_eq!(legacy.owner, None);
        assert_eq!(legacy.shape, Shape::LoopbackIntoCapture);
    }

    /// The plan must be a pure function of the snapshot: same input, same
    /// order, every time.
    #[test]
    fn planning_is_deterministic_regardless_of_snapshot_order() {
        let a = [null_sink(5, 42), mirror(10, 42), local_monitor(11, 42)];
        let b = [local_monitor(11, 42), null_sink(5, 42), mirror(10, 42)];
        assert_eq!(
            plan(&a, &our_policy(), nothing_is_alive),
            plan(&b, &our_policy(), nothing_is_alive)
        );
    }
}