//! The taint-engine fixture matrix — design v3.4 §12 plus the impl plan's
//! degenerate-snapshot addition.
//!
//! **Every test asserts an exact partition of the whole candidate universe**,
//! never a spot check on one named node. Checking only the node a test cares
//! about constrains nothing about the others, and an exclude-everything
//! implementation would pass. So each test names *every* `Stream/Output/Audio`
//! node in its graph as either eligible or excluded-with-a-reason-code.
//!
//! Fixture realism that is load-bearing: an ordinary app holds **one client
//! connection**, each pactl module holds its own, and every device node
//! shares the session manager's. Getting that wrong makes the engine look
//! broken (or, worse, makes a leak look fine).

use std::collections::BTreeSet;

use super::fixture::{Graph, NodeRef, PULSE_PID, app};
use super::owner::{OwnerCtx, OwnerKey, strongest_shared_key};
use super::snapshot::{MediaRole, NodeProps, PortDirection, Serial};
use super::{Decisions, Eligibility, ExclusionCtx, ObjectRef, Reason, StickyState, evaluate};

fn ctx() -> ExclusionCtx {
    ExclusionCtx {
        aec_module_id: None,
        pipewire_pulse_pid: Some(PULSE_PID),
        pixelpass_owned: BTreeSet::new(),
        graph_ready: true,
    }
}

fn run(graph: &Graph, ctx: &ExclusionCtx) -> Decisions {
    evaluate(&graph.build(), ctx, &StickyState::default()).0
}

/// Assert the complete candidate partition. `excluded` names reason codes.
#[track_caller]
fn assert_partition(
    decisions: &Decisions,
    eligible: &[(&str, NodeRef)],
    excluded: &[(&str, NodeRef, &str)],
) {
    let mut expected_eligible: Vec<Serial> = eligible.iter().map(|(_, n)| n.serial).collect();
    expected_eligible.sort_unstable();
    let mut expected_excluded: Vec<(Serial, &str)> = excluded
        .iter()
        .map(|(_, n, code)| (n.serial, *code))
        .collect();
    expected_excluded.sort_unstable();

    let label = |serial: Serial| -> String {
        eligible
            .iter()
            .map(|(name, n)| (*name, *n))
            .chain(excluded.iter().map(|(name, n, _)| (*name, *n)))
            .find(|(_, n)| n.serial == serial)
            .map_or_else(|| format!("<unnamed {serial:?}>"), |(name, _)| name.into())
    };
    let actual: Vec<String> = decisions
        .candidates
        .values()
        .map(|d| match d.reason() {
            None => format!("{} = eligible", label(d.serial)),
            Some(reason) => format!("{} = {}", label(d.serial), reason.code()),
        })
        .collect();

    assert_eq!(
        (decisions.eligible(), decisions.excluded()),
        (expected_eligible, expected_excluded),
        "actual decisions: {actual:#?}"
    );
}

#[track_caller]
fn assert_tainted(decisions: &Decisions, node: NodeRef, reason: &str) {
    let entry = decisions
        .taint
        .get(&node.serial)
        .unwrap_or_else(|| panic!("expected {:?} to be tainted", node.serial));
    assert_eq!(entry.reason.code(), reason);
}

#[track_caller]
fn assert_untainted(decisions: &Decisions, node: NodeRef) {
    assert_eq!(
        decisions.taint.get(&node.serial).map(|e| e.reason.code()),
        None,
        "expected {:?} to be untainted",
        node.serial
    );
}

// ──────────────────────────────────────────────────────────────────────
// Boundary / degenerate cases — the fail-closed default
// ──────────────────────────────────────────────────────────────────────

#[test]
fn empty_snapshot_yields_nothing_eligible() {
    let decisions = run(&Graph::new(), &ctx());
    assert!(decisions.candidates.is_empty());
    assert!(decisions.eligible().is_empty());
}

#[test]
fn a_graph_with_only_untainted_streams_still_partitions() {
    // The other half of the degenerate case: "nothing eligible" must not be
    // reached by excluding everything unconditionally.
    let mut graph = Graph::new();
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    assert_partition(&run(&graph, &ctx()), &[("firefox", firefox)], &[]);
}

#[test]
fn graph_not_ready_excludes_every_candidate() {
    let mut graph = Graph::new();
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    let decisions = run(
        &graph,
        &ExclusionCtx {
            graph_ready: false,
            ..ctx()
        },
    );
    assert_partition(&decisions, &[], &[("firefox", firefox, "graph-not-ready")]);
}

#[test]
fn decisions_do_not_depend_on_insertion_order() {
    // The reported reason must come from an explicit priority, not from
    // traversal order, or the audit output is unstable between recomputes.
    let build = |reverse: bool| {
        let mut graph = Graph::new();
        let sink = graph.device_node("sink", MediaRole::Sink);
        let call = graph.peerspeak_node("peerspeak", 7);
        let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
        if reverse {
            graph.link(firefox, sink);
            graph.link(call, sink);
        } else {
            graph.link(call, sink);
            graph.link(firefox, sink);
        }
        run(&graph, &ctx()).excluded()
    };
    assert_eq!(build(false), build(true));
}

// ──────────────────────────────────────────────────────────────────────
// Node-local roots (v3.4 §5.1, §5.2, §6.2, §6.6)
// ──────────────────────────────────────────────────────────────────────

#[test]
fn peerspeak_tagged_nodes_are_excluded_and_plain_apps_are_not() {
    let mut graph = Graph::new();
    let sink = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak-call", 7);
    let mpv = graph.peerspeak_node("peerspeak-mpv", 8);
    let notify = graph.peerspeak_node("peerspeak-notify", 9);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    for node in [call, mpv, notify, firefox] {
        graph.link(node, sink);
    }

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("mpv", mpv, "peerspeak-owned"),
            ("notify", notify, "peerspeak-owned"),
        ],
    );
    // The sink carried peerspeak audio, so it is tainted — that is what
    // makes every monitor reader downstream of it tainted too.
    assert_tainted(&decisions, sink, "tainted-upstream");
}

/// Each ownership carrier must work **alone** (v3.5 §5.1).
///
/// ⚠️ The phase-3r lesson, applied deliberately: a gate that asserts a value
/// two sources can satisfy gates neither. `peerspeak_tagged_nodes_…` above
/// uses nodes carrying both carriers, so it would keep passing if either
/// were deleted. These are the rows that actually pin them.
#[test]
fn either_ownership_carrier_alone_taints_the_node() {
    let mut graph = Graph::new();
    let sink = graph.device_node("hw-sink", MediaRole::Sink);
    // Carrier 1: the property, on a node whose name says nothing.
    let prop_only = graph.peerspeak_node_prop_only("some-playback-stream", 7);
    // Carrier 2: the name prefix, property absent — the F1 case.
    let name_only = graph.peerspeak_node_name_only("mpv", 31_284);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11_114);
    for node in [prop_only, name_only, firefox] {
        graph.link(node, sink);
    }

    assert_partition(
        &run(&graph, &ctx()),
        &[("firefox", firefox)],
        &[
            ("prop_only", prop_only, "peerspeak-owned"),
            ("name_only", name_only, "peerspeak-owned"),
        ],
    );
}

/// **R10-1, the F2 fix.** Neither carrier is a security boundary — both are
/// strings any unprivileged process can set on its own node — so the tag is
/// honoured only on `Stream/Output/Audio`, the one role peerspeak ever tags.
///
/// Without the restriction, a tagged `Stream/Input/Audio` **with no links at
/// all** is a tainted *reader* (`receivers` includes nodes by role, no link
/// required), and an unbounded one, so `propagate_unresolved_owner` fails
/// every candidate on the machine closed. That is a whole-feature denial from
/// an unprivileged process, reproduced live during the phase-1 review.
#[test]
fn an_ownership_tag_on_a_non_producer_is_not_a_taint_root() {
    for role in [
        MediaRole::StreamInput,
        MediaRole::Sink,
        MediaRole::Source,
        MediaRole::Duplex,
        MediaRole::Other,
    ] {
        let mut graph = Graph::new();
        let sink = graph.device_node("hw-sink", MediaRole::Sink);
        let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11_114);
        graph.link(firefox, sink);
        // Deliberately unlinked: the F2 shape needs no edges whatsoever.
        let impostor = graph.peerspeak_tagged_node("rogue", role, 4_242);

        let decisions = run(&graph, &ctx());
        assert_untainted(&decisions, impostor);
        assert!(
            decisions.taint.is_empty(),
            "{role:?} impostor tainted something: {:?}",
            decisions.taint.keys().collect::<Vec<_>>()
        );
        // The whole point: the eligible half stays non-empty.
        assert_partition(&decisions, &[("firefox", firefox)], &[]);
    }
}

/// **The live F2 reproduction, verbatim.** The measured impostor was an
/// *unbounded* reader — `client.id` present, `application.process.id` absent
/// — which is what turns "one bogus tainted node" into "nothing on this
/// machine is shareable": `propagate_unresolved_owner` cannot prove any
/// candidate independent of a reader it cannot attribute to an owner.
///
/// Measured before the fix: `BASELINE eligible=1 excluded=[]` →
/// `WITH IMPOSTOR eligible=0 excluded=[firefox → unresolved-owner]`.
///
/// Distinct from the row above, which uses a *bounded* impostor and so would
/// still pass if only the cheap half of the fix were present.
#[test]
fn an_unbounded_tagged_impostor_cannot_exclude_a_bystander_app() {
    let mut graph = Graph::new();
    let sink = graph.device_node("hw-sink", MediaRole::Sink);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11_114);
    let mpv = graph.app_node("mpv", MediaRole::StreamOutput, 31_284);
    for node in [firefox, mpv] {
        graph.link(node, sink);
    }

    let baseline = run(&graph, &ctx());
    assert_partition(&baseline, &[("firefox", firefox), ("mpv", mpv)], &[]);

    // Both carriers, no pid, no links — everything an unprivileged process
    // can arrange for itself in one `pw-cli` invocation.
    let rogue_client = graph.client(Some(PULSE_PID));
    let impostor = graph.node(
        &format!("{}rogue_4242", super::PEERSPEAK_OWNED_NODE_PREFIX),
        MediaRole::StreamInput,
        NodeProps {
            peerspeak_owned: true,
            client_id: Some(rogue_client),
            ..NodeProps::default()
        },
    );

    let decisions = run(&graph, &ctx());
    assert_untainted(&decisions, impostor);
    assert_partition(&decisions, &[("firefox", firefox), ("mpv", mpv)], &[]);
}

/// A tag that R10-1 ignores is still reported, so that neither a peerspeak
/// tagging bug nor an impersonation attempt is silent.
#[test]
fn ignored_ownership_tags_are_surfaced_for_diagnostics() {
    let mut graph = Graph::new();
    let sink = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("call", 7);
    graph.link(call, sink);
    let impostor = graph.peerspeak_tagged_node("rogue", MediaRole::StreamInput, 4_242);

    let snapshot = graph.build();
    let misplaced: Vec<Serial> = super::misplaced_ownership_tags(&snapshot)
        .iter()
        .map(|node| node.serial)
        .collect();

    // Exactly the ignored one: the honoured producer is not "misplaced".
    assert_eq!(misplaced, vec![impostor.serial]);
    assert_ne!(impostor.serial, call.serial);
}

/// The prefix is a **prefix**, not a substring: an unrelated app must not be
/// excluded because the literal appears somewhere in its name. Over-exclusion
/// is the safe direction, but it is still wrong, and the phase-5 gate now
/// asserts exact partitions in both halves.
#[test]
fn the_owned_prefix_matches_only_at_the_start_of_node_name() {
    let mut graph = Graph::new();
    let sink = graph.device_node("hw-sink", MediaRole::Sink);
    let impostor = graph.app_node(
        &format!("recorder-of-{}stuff", super::PEERSPEAK_OWNED_NODE_PREFIX),
        MediaRole::StreamOutput,
        11_114,
    );
    graph.link(impostor, sink);

    assert_partition(&run(&graph, &ctx()), &[("impostor", impostor)], &[]);
}

/// The consumer half of the cross-repo contract test (impl plan §3
/// requirement 2). peerspeak runs the mirror of this against a byte-identical
/// copy of the same file, and asserts the environment a real child `Command`
/// would carry produces exactly these literals.
///
/// This proves the two repos agree on the *literals*. That pixelpass actually
/// *listens* is proven by the two carrier tests above, and against the live
/// graph by the phase 5 dry-run.
#[test]
fn ownership_carriers_match_the_cross_repo_fixture() {
    const FIXTURE: &str = include_str!("../../../tests/fixtures/ownership-tag-contract.txt");

    let pinned: Vec<(&str, &str)> = FIXTURE
        .lines()
        .map(str::trim)
        .filter(|line| !line.is_empty() && !line.starts_with('#'))
        .map(|line| line.split_once('=').expect("fixture line is key=value"))
        .collect();

    // ⚠️ Refuse a duplicated key rather than resolving it (Codex phase-1
    // review, finding 3). This side takes the first match and peerspeak's
    // took the last, so a duplicate in a byte-identical file could leave both
    // repos green having selected *different* contracts.
    for (index, (key, _)) in pinned.iter().enumerate() {
        assert!(
            !pinned[..index].iter().any(|(seen, _)| seen == key),
            "fixture defines {key:?} twice; the two repos would disagree on which wins"
        );
    }
    let get = |key: &str| -> &str {
        pinned
            .iter()
            .find(|(k, _)| *k == key)
            .unwrap_or_else(|| panic!("fixture has no key {key:?}"))
            .1
    };

    assert_eq!(super::PEERSPEAK_OWNED_PROP, get("prop_key"));
    assert_eq!(super::PEERSPEAK_OWNED_NODE_PREFIX, get("node_name_prefix"));
    // ⚠️ **Equality, and that is now the whole rule**: carrier 1 is matched
    // exactly, not as "anything but false/0" (round 10, R10-4). This assert
    // used to be followed by a weaker `value != "false" && value != "0"`
    // check, which described a leniency that no longer exists — the round-10
    // review's finding 6, and a real trap: a future producer reading the old
    // fixture prose could emit "true" and silently lose this carrier.
    //
    // That this consumer actually *listens* to the fixture's value, through
    // the production observer wiring rather than a helper, is asserted by
    // `observer::adapter::tests::the_fixture_value_is_the_only_owned_spelling`.
    assert_eq!(super::PEERSPEAK_OWNED_VALUE, get("prop_value"));

    // And the fixture's own worked example must be one this engine excludes,
    // through carrier 2, exactly as written in the shared file.
    let mut graph = Graph::new();
    let sink = graph.device_node("hw-sink", MediaRole::Sink);
    let example = graph.app_node(get("node_name_example"), MediaRole::StreamOutput, 31_284);
    graph.link(example, sink);
    assert_partition(
        &run(&graph, &ctx()),
        &[],
        &[("example", example, "peerspeak-owned")],
    );
}

#[test]
fn aec_identity_is_exact_equality_and_other_modules_stay_eligible() {
    let mut graph = Graph::new();
    let ours = graph.module_node("echo-cancel-playback", MediaRole::StreamOutput, 536_870_919);
    // A tunnel/RTP module: it has a `pulse.module.id`, just not ours.
    // "Has any pulse.module.id" is explicitly rejected as an exclusion rule
    // — such a module may be the only carrier of audio the user wants
    // shared (v3.4 §5.2 correction 2).
    let tunnel = graph.module_node("tunnel-out", MediaRole::StreamOutput, 536_870_921);
    let plain = graph.app_node("firefox", MediaRole::StreamOutput, 11114);

    let decisions = run(
        &graph,
        &ExclusionCtx {
            aec_module_id: Some(536_870_919),
            ..ctx()
        },
    );
    assert_partition(
        &decisions,
        &[("tunnel", tunnel), ("plain", plain)],
        &[("ours", ours, "aec-identity")],
    );
}

#[test]
fn aec_module_id_is_compared_beyond_u32() {
    // Guards the phase 0a widening at the engine boundary too: these two
    // differ only above bit 32.
    let big = u64::from(u32::MAX) + 7;
    let mut graph = Graph::new();
    let ours = graph.module_node("aec", MediaRole::StreamOutput, big);
    let other = graph.module_node("other", MediaRole::StreamOutput, big ^ 0x1_0000_0000);
    let decisions = run(
        &graph,
        &ExclusionCtx {
            aec_module_id: Some(big),
            ..ctx()
        },
    );
    assert_partition(
        &decisions,
        &[("other", other)],
        &[("ours", ours, "aec-identity")],
    );
}

#[test]
fn foreign_echo_cancel_group_is_excluded_not_shared() {
    // Decision D3: warn and exclude. The group prefix is hazard detection,
    // never ownership — it cannot tell peerspeak's AEC from anyone else's.
    let mut graph = Graph::new();
    let foreign = graph.group_node(
        "echo-cancel-playback",
        MediaRole::StreamOutput,
        "echo-cancel-9999-13",
        4321,
    );
    let plain = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    assert_partition(
        &run(&graph, &ctx()),
        &[("plain", plain)],
        &[("foreign", foreign, "foreign-echo-cancel")],
    );
}

#[test]
fn pixelpass_capture_sink_and_its_downstream_are_excluded() {
    // v3.4 §6.2 cycle prevention, and impl-plan §5.1 row 7: observing the
    // capture sink is not enough — a *downstream* candidate must be named,
    // or recognising `pixelpass_capture_*` as a mere sink name would pass
    // without any transitive propagation.
    let mut graph = Graph::new();
    let other_host_sink = graph.module_node("pixelpass_capture_31337", MediaRole::Sink, 42);
    let fwd_in = graph.module_node("fwd-in", MediaRole::StreamInput, 77);
    let fwd_out = graph.module_node("fwd-out", MediaRole::StreamOutput, 77);
    graph.link(other_host_sink, fwd_in);
    let plain = graph.app_node("firefox", MediaRole::StreamOutput, 11114);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[("plain", plain)],
        &[("fwd-out", fwd_out, "tainted-owner-bridge")],
    );
    assert_tainted(&decisions, other_host_sink, "pixelpass-owned");
    assert_tainted(&decisions, fwd_in, "tainted-upstream");
}

#[test]
fn pixelpass_owned_serials_are_roots_even_without_a_capture_sink_name() {
    let mut graph = Graph::new();
    let ours = graph.device_node("some-sink", MediaRole::Sink);
    let reader = graph.app_node("reader-in", MediaRole::StreamInput, 5150);
    let leg = graph.app_node("reader-out", MediaRole::StreamOutput, 5150);
    graph.link(ours, reader);
    let decisions = run(
        &graph,
        &ExclusionCtx {
            pixelpass_owned: BTreeSet::from([ours.serial]),
            ..ctx()
        },
    );
    assert_partition(&decisions, &[], &[("leg", leg, "tainted-owner-bridge")]);
    assert_tainted(&decisions, ours, "pixelpass-owned");
}

#[test]
fn port_exclusive_and_passthrough_are_local_exclusions() {
    let mut graph = Graph::new();
    let exclusive = graph.app_node("exclusive", MediaRole::StreamOutput, 100);
    graph.port(exclusive, PortDirection::Out, true);
    let client = graph.client_of_app(101);
    let passthrough = graph.node(
        "passthrough",
        MediaRole::StreamOutput,
        NodeProps {
            passthrough: true,
            ..app(client, 101)
        },
    );
    let ok = graph.app_node("ok", MediaRole::StreamOutput, 102);
    graph.port(ok, PortDirection::Out, false);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[("ok", ok)],
        &[
            ("exclusive", exclusive, "port-exclusive"),
            ("passthrough", passthrough, "passthrough"),
        ],
    );
    // Neither is hazardous — an unlinkable stream must not taint anything.
    assert_untainted(&decisions, exclusive);
    assert_untainted(&decisions, passthrough);
}

#[test]
fn a_node_with_no_props_at_all_is_eligible_and_non_streams_are_not_candidates() {
    let mut graph = Graph::new();
    let bare = graph.node("bare", MediaRole::StreamOutput, NodeProps::default());
    graph.node("video", MediaRole::Other, NodeProps::default());
    graph.node("mic", MediaRole::Source, NodeProps::default());
    assert_partition(&run(&graph, &ctx()), &[("bare", bare)], &[]);
}

// ──────────────────────────────────────────────────────────────────────
// Edge type 2 — sink → monitor, free at node granularity
// ──────────────────────────────────────────────────────────────────────

#[test]
fn taint_crosses_app_then_sink_then_monitor_reader() {
    let mut graph = Graph::new();
    let sink = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, sink);
    // A recorder reading the sink's monitor. The monitor connection IS a
    // real Link whose output node is the sink itself, so no synthetic edge
    // is needed at node granularity.
    let rec_in = graph.app_node("recorder-in", MediaRole::StreamInput, 555);
    graph.link(sink, rec_in);
    let rec_out = graph.app_node("recorder-out", MediaRole::StreamOutput, 555);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("rec-out", rec_out, "tainted-owner-bridge"),
        ],
    );
    assert_tainted(&decisions, rec_in, "tainted-upstream");
}

#[test]
fn playing_into_a_tainted_sink_does_not_taint_the_player() {
    // Taint flows downstream only. Every app on the box plays into the same
    // sink peerspeak does; if that tainted them, nothing would ever be
    // shareable and the feature would ship as silence.
    let mut graph = Graph::new();
    let sink = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    graph.link(call, sink);
    graph.link(firefox, sink);

    assert_partition(
        &run(&graph, &ctx()),
        &[("firefox", firefox)],
        &[("call", call, "peerspeak-owned")],
    );
}

// ──────────────────────────────────────────────────────────────────────
// Edge type 3 — the owner bridge (v3.4 §6.1.1, §6.1.2)
// ──────────────────────────────────────────────────────────────────────

/// The measured `module-null-sink` + `module-loopback` shape. Walking Links
/// alone from the leaking output leg finds *no inbound links at all* — a
/// dead end that reads as "clean".
#[test]
fn loopback_forwarder_output_leg_is_excluded_via_the_owner_bridge() {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let null_sink = graph.module_node("fabletest_sink", MediaRole::Sink, 536_870_917);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, null_sink);

    // The tainted forwarder: legs share `pulse.module.id`, no Link between.
    let lb_in = graph.module_node("input.loopback", MediaRole::StreamInput, 536_870_918);
    let lb_out = graph.module_node("output.loopback", MediaRole::StreamOutput, 536_870_918);
    graph.link(null_sink, lb_in);
    graph.link(lb_out, hw);

    // The same shape with a clean input — impl-plan §5.1 row 1's eligible
    // half. Without it the test would also pass on "exclude all forwarders".
    let clean_sink = graph.device_node("clean-sink", MediaRole::Sink);
    let clean_in = graph.module_node("clean-in", MediaRole::StreamInput, 536_870_920);
    let clean_out = graph.module_node("clean-out", MediaRole::StreamOutput, 536_870_920);
    graph.link(clean_sink, clean_in);
    graph.link(clean_out, hw);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[("clean-out", clean_out)],
        &[
            ("call", call, "peerspeak-owned"),
            ("lb-out", lb_out, "tainted-owner-bridge"),
        ],
    );
    // ...and it was the bridge, naming the key — not a Link walk.
    assert_eq!(
        decisions.taint[&lb_out.serial].reason,
        Reason::TaintedOwnerBridge {
            key: Some(OwnerKey::PulseModuleId)
        }
    );
}

#[test]
fn easyeffects_shape_bridges_from_an_audio_sink_to_its_re_emitting_leg() {
    // The canonical installed case. The tainted member here is an
    // `Audio/Sink`, not a `Stream/Input/Audio` — if the bridge only fired
    // from stream inputs, EasyEffects would forward the whole call into the
    // share at full level.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let ee_sink = graph.group_node("ee_sink", MediaRole::Sink, "ee", 4321);
    let ee_out = graph.group_node("easyeffects-out", MediaRole::StreamOutput, "ee", 4321);
    graph.link(ee_out, hw);
    let call = graph.peerspeak_node("peerspeak", 7);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    graph.link(call, ee_sink);
    graph.link(firefox, ee_sink);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("ee-out", ee_out, "tainted-owner-bridge"),
        ],
    );
    assert_eq!(
        decisions.taint[&ee_out.serial].reason,
        Reason::TaintedOwnerBridge {
            key: Some(OwnerKey::LinkGroup)
        }
    );
}

#[test]
fn gstreamer_split_clients_bridge_on_process_id() {
    // The C2 refutation, measured: one `gst-launch pulsesrc ! pulsesink`
    // process produced TWO client objects (209 and 210) for its two legs.
    let mut graph = Graph::new();
    let tainted_sink = graph.device_node("null-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, tainted_sink);

    let client_in = graph.client(Some(PULSE_PID));
    let client_out = graph.client(Some(PULSE_PID));
    let gst_in = graph.node("gst-in", MediaRole::StreamInput, app(client_in, 20172));
    let gst_out = graph.node("gst-out", MediaRole::StreamOutput, app(client_out, 20172));
    graph.link(tainted_sink, gst_in);

    // Control: the same split-client shape reading an *untainted* source.
    let mic = graph.device_node("mic", MediaRole::Source);
    let ok_client_in = graph.client(Some(PULSE_PID));
    let ok_client_out = graph.client(Some(PULSE_PID));
    let ok_in = graph.node("ok-in", MediaRole::StreamInput, app(ok_client_in, 30000));
    let ok_out = graph.node("ok-out", MediaRole::StreamOutput, app(ok_client_out, 30000));
    graph.link(mic, ok_in);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[("ok-out", ok_out)],
        &[
            ("call", call, "peerspeak-owned"),
            ("gst-out", gst_out, "tainted-owner-bridge"),
        ],
    );
    assert_eq!(
        decisions.taint[&gst_out.serial].reason,
        Reason::TaintedOwnerBridge {
            key: Some(OwnerKey::ProcessId)
        }
    );
}

#[test]
fn owner_key_union_falls_through_a_present_but_unequal_key() {
    // The §6.1.2 wording trap, asserted on the key resolver itself: a
    // first-present implementation stops at `client.id`, sees 209 != 210,
    // concludes "different owners", and the leak survives.
    let mut graph = Graph::new();
    let a_client = graph.client(Some(PULSE_PID));
    let b_client = graph.client(Some(PULSE_PID));
    let a = graph.node("a", MediaRole::StreamInput, app(a_client, 20172));
    let b = graph.node("b", MediaRole::StreamOutput, app(b_client, 20172));
    let snapshot = graph.build();
    let (a, b) = (
        snapshot.node(a.serial).unwrap(),
        snapshot.node(b.serial).unwrap(),
    );
    assert_ne!(a.props.client_id, b.props.client_id);
    let owner_ctx = OwnerCtx::new(&snapshot, Some(PULSE_PID));
    assert_eq!(
        strongest_shared_key(a, b, &owner_ctx),
        Some(OwnerKey::ProcessId)
    );
}

#[test]
fn the_strongest_shared_key_wins_when_several_match() {
    let mut graph = Graph::new();
    let a = graph.group_node("a", MediaRole::StreamInput, "g", 500);
    let b = graph.group_node("b", MediaRole::StreamOutput, "g", 500);
    let snapshot = graph.build();
    let owner_ctx = OwnerCtx::new(&snapshot, Some(PULSE_PID));
    assert_eq!(
        strongest_shared_key(
            snapshot.node(a.serial).unwrap(),
            snapshot.node(b.serial).unwrap(),
            &owner_ctx
        ),
        Some(OwnerKey::LinkGroup)
    );
}

#[test]
fn the_pipewire_pulse_pid_does_not_fuse_unrelated_modules() {
    // impl-plan §5.1 row 3: two Pulse modules, one tainted input. The other
    // module's output MUST stay eligible — that is the only thing that
    // makes wrong pipewire-pulse-PID fusion observable at all.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);

    let a_in = graph.module_node("a-in", MediaRole::StreamInput, 1);
    let a_out = graph.module_node("a-out", MediaRole::StreamOutput, 1);
    graph.link(hw, a_in);

    let mic = graph.device_node("mic", MediaRole::Source);
    let b_in = graph.module_node("b-in", MediaRole::StreamInput, 2);
    let b_out = graph.module_node("b-out", MediaRole::StreamOutput, 2);
    graph.link(mic, b_in);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[("b-out", b_out)],
        &[
            ("call", call, "peerspeak-owned"),
            ("a-out", a_out, "tainted-owner-bridge"),
        ],
    );
    assert_untainted(&decisions, b_in);
}

/// **R10-3, the fix.** A native PipeWire client puts no
/// `application.process.id` on its node — only `client.id` — so before the
/// Client fallback it had no key 4, was therefore *unbounded*, and
/// `propagate_unresolved_owner` excluded it the moment any tainted reader
/// existed anywhere on the machine.
///
/// Measured live: an untagged mpv was eligible alone, and became
/// `unresolved-owner` the instant peerspeak played audio. Since peerspeak
/// playing audio is the only situation in which this feature runs at all, that
/// amounted to "native-PipeWire apps are never shareable".
#[test]
fn a_native_client_is_bounded_by_its_clients_sec_pid() {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);

    // The tainted reader that arms the unresolved-owner arm. Bounded itself
    // (a real pid), exactly as the live `sunshine` was — so this is the
    // bounded-reader arm, not the keyless-reader one.
    let sunshine = graph.app_node("sunshine", MediaRole::StreamInput, 3_838);
    graph.link(hw, sunshine);

    // mpv on its default ao: client.id only, pid on the Client.
    let mpv = graph.native_client_node("mpv", MediaRole::StreamOutput, 31_284);
    graph.link(mpv, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("mpv", mpv)],
        &[("call", call, "peerspeak-owned")],
    );
}

/// The fallback must bridge a native app's *own* legs, or it has bought
/// boundedness without buying correctness: an app that reads the call and
/// re-emits it on a second native node would be declared clean.
#[test]
fn the_sec_pid_fallback_still_bridges_a_native_apps_own_legs() {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);

    // One native process, two nodes, no link between them — the forwarder
    // shape, in the native flavour.
    let leg_in = graph.native_client_node("forwarder-in", MediaRole::StreamInput, 50_000);
    let leg_out = graph.native_client_node("forwarder-out", MediaRole::StreamOutput, 50_000);
    graph.link(hw, leg_in);

    let decisions = run(&graph, &ctx());
    assert_tainted(&decisions, leg_out, "tainted-owner-bridge");
    assert_partition(
        &decisions,
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("forwarder-out", leg_out, "tainted-owner-bridge"),
        ],
    );
}

/// **The risk the fallback creates, and the guard on it.** Every
/// Pulse-emulated Client carries pipewire-pulse's own PID as `sec_pid` —
/// measured, 15 unrelated Clients sharing 2528 on this host. An unguarded
/// fallback would give all of them key 4 with the *same* value and fuse them
/// into one owner, so a single tainted Pulse app would exclude every other
/// Pulse app on the machine.
///
/// Exception 1 therefore applies to the fallback exactly as it does to the
/// node's own property. Without that, this row goes red.
#[test]
fn the_sec_pid_fallback_does_not_fuse_every_pulse_client() {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);

    // Three unrelated Pulse-emulated apps, each on its own Client, none
    // exposing a node-level pid — so each can only reach key 4 through its
    // Client, whose sec_pid is the daemon's.
    let pulse_app = |graph: &mut Graph, name: &str, role| {
        let client = graph.client(Some(PULSE_PID));
        graph.node(
            name,
            role,
            NodeProps {
                client_id: Some(client),
                ..NodeProps::default()
            },
        )
    };
    // One of them reads the tainted sink; the other two must not care.
    let reader = pulse_app(&mut graph, "recorder", MediaRole::StreamInput);
    graph.link(hw, reader);
    let other_a = pulse_app(&mut graph, "player-a", MediaRole::StreamOutput);
    let other_b = pulse_app(&mut graph, "player-b", MediaRole::StreamOutput);

    let decisions = run(&graph, &ctx());
    // They are unbounded (`client.id` alone never bounds an owner), so the
    // fail-closed arm still excludes them — but as `unresolved-owner`, NOT as
    // `tainted-owner-bridge`. That distinction is the whole assertion: a
    // bridge reason here would mean the daemon pid had fused three unrelated
    // applications into one owner, and unlike fail-closed exclusion, fusion
    // does not go away when the apps are given real pids
    // (`distinct_sec_pids_bound_each_native_app_separately` is that half).
    assert_tainted(&decisions, other_a, "unresolved-owner");
    assert_tainted(&decisions, other_b, "unresolved-owner");
    for node in [other_a, other_b] {
        assert_ne!(
            decisions.taint.get(&node.serial).map(|e| e.reason.code()),
            Some("tainted-owner-bridge"),
            "the daemon pid must not bridge unrelated Pulse clients"
        );
    }
}

/// The same three apps, given **real per-app** `sec_pid`s: now the fallback
/// fires, all three are bounded, and only the one actually reading the call is
/// affected. This is the row that proves the guard above suppresses the daemon
/// pid *specifically* rather than disabling the fallback outright.
#[test]
fn distinct_sec_pids_bound_each_native_app_separately() {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);

    let reader = graph.native_client_node("recorder", MediaRole::StreamInput, 6_001);
    graph.link(hw, reader);
    let other_a = graph.native_client_node("player-a", MediaRole::StreamOutput, 6_002);
    let other_b = graph.native_client_node("player-b", MediaRole::StreamOutput, 6_003);

    assert_partition(
        &run(&graph, &ctx()),
        &[("player-a", other_a), ("player-b", other_b)],
        &[("call", call, "peerspeak-owned")],
    );
}

/// An **ambiguous** `client.id` — two live Clients claiming it, meaning the
/// observer missed a removal — must not yield a fallback pid. Inventing an
/// owner key is the one direction that can *reduce* taint, so resolving the
/// ambiguity by coin toss is the wrong kind of guess.
#[test]
fn an_ambiguous_client_id_yields_no_fallback_pid() {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let sunshine = graph.app_node("sunshine", MediaRole::StreamInput, 3_838);
    graph.link(hw, sunshine);

    // Two Clients, one id, distinct real pids.
    let shared_id = graph.client(Some(6_010));
    graph.client_with_id(shared_id, Some(6_011));
    let app = graph.node(
        "native-app",
        MediaRole::StreamOutput,
        NodeProps {
            client_id: Some(shared_id),
            ..NodeProps::default()
        },
    );
    graph.link(app, hw);

    // Unbounded ⇒ fails closed, exactly as before R10-3.
    assert_partition(
        &run(&graph, &ctx()),
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("native-app", app, "unresolved-owner"),
        ],
    );
}

/// The ambiguity guard must not depend on the *first* Client claiming an id
/// having a `sec_pid`.
///
/// Found by auditing R10-3 rather than by a failing case: the first cut
/// detected a duplicate id by looking it up in the pid map, which is only
/// populated for Clients that carry a pid at all. A pid-less Client therefore
/// left no trace, and the next Client claiming the same id was treated as
/// unique — resolving an ambiguous id, which is exactly the guess the guard
/// exists to refuse. Pid-less Clients are ordinary here (`device_node`'s
/// session client is one), so this is reachable, not theoretical.
#[test]
fn a_pidless_first_client_still_makes_its_id_ambiguous() {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let sunshine = graph.app_node("sunshine", MediaRole::StreamInput, 3_838);
    graph.link(hw, sunshine);

    // First claimant has NO sec_pid; second has one.
    let shared_id = graph.client(None);
    graph.client_with_id(shared_id, Some(6_011));
    let app = graph.node(
        "native-app",
        MediaRole::StreamOutput,
        NodeProps {
            client_id: Some(shared_id),
            ..NodeProps::default()
        },
    );
    graph.link(app, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("native-app", app, "unresolved-owner"),
        ],
    );
}

/// A process using **two** Clients cannot escape the bridge by presenting a
/// bogus pid on one leg and none on the other.
///
/// ⚠️ **This is the round-10 review's finding 1, and it was a real leak while
/// key 4 was `node.or_else(client)`.** The node's `application.process.id` is
/// client-controlled; the Client's `pipewire.sec.pid` is protected. Letting
/// the node's value *replace* the Client's meant the reader was bounded by
/// `12_345` and the output leg by `50_000`, so they shared no key, did not
/// bridge, and — both being bounded — neither tripped the unbounded sweep.
/// The output stayed eligible while re-emitting the call.
///
/// Carrying both values fixes it: the two legs share the Client pid.
///
/// Reachability, stated honestly: `evaluate()` today is reached only by the
/// dry-run audit, which creates no links, so this could not echo on this
/// branch. It becomes live the moment phase 6 consumes these decisions.
#[test]
fn one_process_with_two_clients_cannot_split_its_pid_to_escape_the_bridge() {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);

    // One native process, two Clients, one protected pid.
    let reader_client = graph.client(Some(50_000));
    let output_client = graph.client(Some(50_000));

    // Its reading leg claims a pid that is not its own.
    let reader = graph.node(
        "two-client-reader",
        MediaRole::StreamInput,
        NodeProps {
            client_id: Some(reader_client),
            process_id: Some(12_345),
            ..NodeProps::default()
        },
    );
    graph.link(hw, reader);

    // Its re-emitting leg claims no pid at all.
    let output = graph.node(
        "two-client-output",
        MediaRole::StreamOutput,
        NodeProps {
            client_id: Some(output_client),
            process_id: None,
            ..NodeProps::default()
        },
    );
    graph.link(output, hw);

    // A genuinely unrelated app must survive, or "exclude everything" would
    // pass this test — the §5.1 eligible-half rule.
    let bystander = graph.app_node("mpv", MediaRole::StreamOutput, 9_001);
    graph.link(bystander, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("mpv", bystander)],
        &[
            ("call", call, "peerspeak-owned"),
            ("two-client-output", output, "tainted-owner-bridge"),
        ],
    );
}

/// The node's own `application.process.id` is used even when its Client's
/// `sec_pid` is the daemon's — the single most common shape here, since a
/// Pulse-emulated node's pid is the app's while its Client's is
/// pipewire-pulse's.
///
/// ⚠️ Both values are now carried (round-10 review, finding 1), so this is no
/// longer "the node's wins" but "exception 1 is applied per value": the
/// daemon's `sec_pid` is dropped and the node's real pid is kept, leaving the
/// same single key as before.
#[test]
fn the_nodes_own_process_id_wins_over_its_clients() {
    let mut graph = Graph::new();
    // `app_node` is exactly that shape: node pid 11_114, Client sec_pid
    // PULSE_PID. If the Client's won, exception 1 would suppress key 4 and
    // this node would be unbounded.
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let sunshine = graph.app_node("sunshine", MediaRole::StreamInput, 3_838);
    graph.link(hw, sunshine);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11_114);
    graph.link(firefox, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("firefox", firefox)],
        &[("call", call, "peerspeak-owned")],
    );
}

// ──────────────────────────────────────────────────────────────────────
// F11-1 — a self-claimed pid is not provenance: the five Client cases
// ──────────────────────────────────────────────────────────────────────

/// The scaffold every F11-1 row needs: peerspeak's call reaching the hardware
/// sink, a **bounded** tainted reader, and an ordinary bystander.
///
/// ⚠️ The reader must be *bounded* (`sunshine` carries a real pid). An
/// unbounded tainted reader trips `propagate_unresolved_owner`'s other tier,
/// which sweeps **every** output candidate on the box regardless of its own
/// keys — the three "unbounded" rows below would then pass without testing
/// anything. The bystander is the other half of that guard: it is bounded via
/// the ordinary Pulse shape, so an implementation that unbounded everything
/// fails every row instead of passing three of them.
///
/// Returns the graph, the hardware sink to hang nodes off, and the two nodes
/// every row must name in its partition.
fn armed_with_a_bounded_reader() -> (Graph, NodeRef, NodeRef, NodeRef) {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let sunshine = graph.app_node("sunshine", MediaRole::StreamInput, 3_838);
    graph.link(hw, sunshine);
    let bystander = graph.app_node("mpv", MediaRole::StreamOutput, 9_001);
    graph.link(bystander, hw);
    (graph, hw, call, bystander)
}

/// Case 1 of 5 — **Client absent.** A node that names no Client at all has
/// nothing but its own word for who owns it, so it cannot be bounded.
#[test]
fn an_absent_client_leaves_a_self_claimed_pid_unbounded() {
    let (mut graph, hw, call, bystander) = armed_with_a_bounded_reader();
    let orphan = graph.node(
        "no-client",
        MediaRole::StreamOutput,
        NodeProps {
            process_id: Some(70_001),
            ..NodeProps::default()
        },
    );
    graph.link(orphan, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("mpv", bystander)],
        &[
            ("call", call, "peerspeak-owned"),
            ("no-client", orphan, "unresolved-owner"),
        ],
    );
}

/// Case 2 of 5 — **Client ambiguous.** Two live Clients claim the id, so the
/// observer missed a removal and we do not know who owns this node. A
/// self-claimed pid must not paper over that: this is step 2 of the recorded
/// leak path, and before F11-1 the claim bounded the node and spared it.
#[test]
fn an_ambiguous_client_leaves_a_self_claimed_pid_unbounded() {
    let (mut graph, hw, call, bystander) = armed_with_a_bounded_reader();
    let shared_id = graph.client(Some(70_010));
    graph.client_with_id(shared_id, Some(70_011));
    let app = graph.node(
        "ambiguous-client",
        MediaRole::StreamOutput,
        NodeProps {
            client_id: Some(shared_id),
            process_id: Some(70_012),
            ..NodeProps::default()
        },
    );
    graph.link(app, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("mpv", bystander)],
        &[
            ("call", call, "peerspeak-owned"),
            ("ambiguous-client", app, "unresolved-owner"),
        ],
    );
}

/// Case 3 of 5 — **Client unique but pid-less**, and *the row that decides
/// which rule is implemented*.
///
/// A unique Client object exists, so "resolved = a unique Client exists" would
/// call this node bounded — leaving the self-claimed-pid hole wide open under a
/// rule that looks like it closed it. `sec_pid` is what carries protected
/// identity, so `None` means unresolved, and pid-less Clients are ordinary
/// (the session manager's is one).
///
/// A two-case absent/resolved matrix skips this silently. That is why it is
/// written out.
#[test]
fn a_unique_but_pidless_client_leaves_a_self_claimed_pid_unbounded() {
    let (mut graph, hw, call, bystander) = armed_with_a_bounded_reader();
    let pidless = graph.client(None);
    let app = graph.node(
        "pidless-client",
        MediaRole::StreamOutput,
        NodeProps {
            client_id: Some(pidless),
            process_id: Some(70_020),
            ..NodeProps::default()
        },
    );
    graph.link(app, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("mpv", bystander)],
        &[
            ("call", call, "peerspeak-owned"),
            ("pidless-client", app, "unresolved-owner"),
        ],
    );
}

/// Case 4 of 5 — **Client resolved, native.** `pipewire.sec.pid` is the app's
/// own, so provenance and key 4 are the same value and the node is bounded
/// without claiming anything itself.
#[test]
fn a_resolved_native_client_bounds_its_node() {
    let (mut graph, hw, call, bystander) = armed_with_a_bounded_reader();
    let mpv2 = graph.native_client_node("mpv-native", MediaRole::StreamOutput, 70_030);
    graph.link(mpv2, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("mpv", bystander), ("mpv-native", mpv2)],
        &[("call", call, "peerspeak-owned")],
    );
}

/// Case 5 of 5 — **Client resolved to pipewire-pulse.** The row that stops
/// this rule from being the blunt fix.
///
/// Every Pulse-emulated app looks like this: the Client's `sec_pid` is the
/// daemon's — suppressed as a *grouping* key, because it would fuse fifteen
/// unrelated apps — while the node's own `application.process.id` is the app's.
/// Provenance is read **before** that suppression, so the app keeps its bound
/// and stays eligible. Reading it after would unbound every Pulse app on the
/// box and empty the eligible half of the §5.1 matrix, which is the §6.1.1
/// catastrophe arriving through the boundedness door.
#[test]
fn a_client_resolving_to_pipewire_pulse_still_bounds_its_node() {
    let (mut graph, hw, call, bystander) = armed_with_a_bounded_reader();
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 70_040);
    graph.link(firefox, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("mpv", bystander), ("firefox", firefox)],
        &[("call", call, "peerspeak-owned")],
    );
}

/// **The recorded leak path, end to end** (round 11 review, finding 1).
///
/// One process, two Clients. Its reading leg claims the daemon's pid — which
/// exception 1 suppresses — while its Client holds a real protected pid `A`, so
/// the union bounds the reader by `A` and the *unbounded-reader* tier never
/// arms. Its re-emitting leg sits on a second Client whose id is **ambiguous**
/// (one of the two claimants even holds `A`, so this is not "the guess would
/// have been wrong" — it is "a guess is not evidence"), and claims a pid of its
/// own. The two legs share no key, so the bridge does not fire either.
///
/// Before F11-1 the self-claim bounded the output leg, both tiers stayed quiet,
/// and it re-emitted the call while eligible. Now the leg is unbounded, the
/// bounded-reader tier sweeps it, and `mpv` shows the sweep is still targeted.
#[test]
fn a_self_claimed_pid_cannot_spare_an_output_leg_the_bridge_cannot_reach() {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);

    let reader_client = graph.client(Some(80_000));
    let reader = graph.node(
        "forwarder-in",
        MediaRole::StreamInput,
        NodeProps {
            client_id: Some(reader_client),
            process_id: Some(PULSE_PID),
            ..NodeProps::default()
        },
    );
    graph.link(hw, reader);

    let ambiguous = graph.client(Some(80_000));
    graph.client_with_id(ambiguous, Some(80_001));
    let output = graph.node(
        "forwarder-out",
        MediaRole::StreamOutput,
        NodeProps {
            client_id: Some(ambiguous),
            process_id: Some(80_002),
            ..NodeProps::default()
        },
    );
    graph.link(output, hw);

    let bystander = graph.app_node("mpv", MediaRole::StreamOutput, 9_001);
    graph.link(bystander, hw);

    // `mpv` staying eligible is what proves the reader is bounded: an
    // unbounded tainted reader sweeps **every** output candidate, `mpv`
    // included, and this row would then be testing the wrong tier.
    assert_partition(
        &run(&graph, &ctx()),
        &[("mpv", bystander)],
        &[
            ("call", call, "peerspeak-owned"),
            ("forwarder-out", output, "unresolved-owner"),
        ],
    );
}

#[test]
fn an_unknown_pipewire_pulse_pid_over_excludes_rather_than_leaks() {
    // v3.4 §6.1.2's failure-mode paragraph: if pixelpass cannot identify
    // the daemon PID, key 4 stops being suppressed, every module-created
    // stream fuses into one owner, and the result is broad over-exclusion —
    // annoying and safe. This test pins that direction; flipping it to
    // "leave key 4 out when the PID is unknown" would be a leak.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let a_in = graph.module_node("a-in", MediaRole::StreamInput, 1);
    let a_out = graph.module_node("a-out", MediaRole::StreamOutput, 1);
    graph.link(hw, a_in);
    let mic = graph.device_node("mic", MediaRole::Source);
    let b_in = graph.module_node("b-in", MediaRole::StreamInput, 2);
    let b_out = graph.module_node("b-out", MediaRole::StreamOutput, 2);
    graph.link(mic, b_in);

    let decisions = run(
        &graph,
        &ExclusionCtx {
            pipewire_pulse_pid: None,
            ..ctx()
        },
    );
    assert_partition(
        &decisions,
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("a-out", a_out, "tainted-owner-bridge"),
            ("b-out", b_out, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn coarse_keys_do_not_bridge_device_nodes() {
    // ⚠️ A rule added during implementation, NOT present in design v3.4.
    // Every ALSA device node is created by one WirePlumber process, so all
    // of them share one `client.id` and one `application.process.id`.
    // peerspeak's playback taints the default sink on every recompute; if
    // coarse keys bridged devices, that taint would jump to the microphone
    // source, and then every app holding a mic would lose its playback —
    // the exact §6.1.1 catastrophe by another route.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let mic = graph.device_node("mic", MediaRole::Source);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let ff_in = graph.app_node("firefox-mic", MediaRole::StreamInput, 11114);
    let ff_out = graph.app_node("firefox-out", MediaRole::StreamOutput, 11114);
    graph.link(mic, ff_in);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[("ff-out", ff_out)],
        &[("call", call, "peerspeak-owned")],
    );
    assert_tainted(&decisions, hw, "tainted-upstream");
    assert_untainted(&decisions, mic);
    assert_untainted(&decisions, ff_in);
}

#[test]
fn a_module_owned_device_still_bridges_on_its_strong_key() {
    // The other side of that rule: exception 2 must not disarm the bridge
    // for virtual sinks that genuinely belong to a module.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    let virt = graph.module_node("virtual-sink", MediaRole::Sink, 900_001);
    let leg = graph.module_node("virtual-out", MediaRole::StreamOutput, 900_001);
    graph.link(call, virt);
    graph.link(leg, hw);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("leg", leg, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn firefox_three_cases() {
    // v3.4 §6.1.1's table, as one graph with an exact partition.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let mic = graph.device_node("mic", MediaRole::Source);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);

    // 1. music only
    let music = graph.app_node("ff-music", MediaRole::StreamOutput, 1001);
    graph.link(music, hw);
    // 2. a Meet call: a mic input leg on an untainted source
    let meet_in = graph.app_node("ff-meet-in", MediaRole::StreamInput, 1002);
    let meet_out = graph.app_node("ff-meet-out", MediaRole::StreamOutput, 1002);
    graph.link(mic, meet_in);
    graph.link(meet_out, hw);
    // 3. screen-sharing with desktop audio: an input leg on a tainted monitor
    let share_in = graph.app_node("ff-share-in", MediaRole::StreamInput, 1003);
    let share_out = graph.app_node("ff-share-out", MediaRole::StreamOutput, 1003);
    graph.link(hw, share_in);
    graph.link(share_out, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("music", music), ("meet-out", meet_out)],
        &[
            ("call", call, "peerspeak-owned"),
            ("share-out", share_out, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn an_unbounded_module_forwarder_fails_closed_and_sweeps_the_desktop() {
    // v3.4 §12: "a module forwarder with neither link-group nor
    // pulse.module.id ⇒ unresolved ⇒ excluded". Its process id is the
    // daemon's (so key 4 is suppressed) and its legs carry different
    // `client.id`s, so nothing can enumerate its siblings.
    //
    // ⚠️ Because the reading leg is itself *unbounded*, the whole desktop's
    // output is swept — an ordinary app is NOT spared (Codex round 4: a
    // real app can present no PID on its reader leg, so "unbounded ⇒ not an
    // app" is unsound; over-exclude instead). The trigger is anomalous: a
    // keyless reader actively consuming the call.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);

    let in_client = graph.client(Some(PULSE_PID));
    let out_client = graph.client(Some(PULSE_PID));
    let fwd_in = graph.node("fwd-in", MediaRole::StreamInput, app(in_client, PULSE_PID));
    let fwd_out = graph.node(
        "fwd-out",
        MediaRole::StreamOutput,
        app(out_client, PULSE_PID),
    );
    graph.link(hw, fwd_in);

    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    graph.link(firefox, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("fwd-out", fwd_out, "unresolved-owner"),
            ("firefox", firefox, "unresolved-owner"),
        ],
    );
}

#[test]
fn a_bounded_tainted_reader_leaves_ordinary_apps_alone() {
    // The blast-radius guarantee survives for the *bounded* tier (round-1
    // finding 4): a tainted reader with a real strong key excludes only the
    // unbounded output legs that could share its identity, not real apps.
    // Here an EasyEffects-shaped reader (link-group) reads the call, and its
    // own re-emitting leg is excluded via the bridge — but firefox is not.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let ee_in = graph.group_node("ee-in", MediaRole::StreamInput, "ee", 5000);
    let ee_out = graph.group_node("ee-out", MediaRole::StreamOutput, "ee", 5000);
    graph.link(hw, ee_in);
    graph.link(ee_out, hw);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    graph.link(firefox, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("ee-out", ee_out, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn audio_duplex_over_taints_known_accepted() {
    // v3.4 §6.1 edge-type-2 caveat, asserted so that fixing it later is a
    // deliberate change and not a silent regression. A duplex device whose
    // playback side is tainted has its capture side treated as tainted too,
    // so an app recording from it loses its playback leg.
    let mut graph = Graph::new();
    let duplex = graph.device_node("duplex-card", MediaRole::Duplex);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, duplex);
    let rec_in = graph.app_node("rec-in", MediaRole::StreamInput, 4444);
    let rec_out = graph.app_node("rec-out", MediaRole::StreamOutput, 4444);
    graph.link(duplex, rec_in);

    assert_partition(
        &run(&graph, &ctx()),
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("rec-out", rec_out, "tainted-owner-bridge"),
        ],
    );
}

// ──────────────────────────────────────────────────────────────────────
// Unresolved ancestry — fail closed (v3.4 §6.1.4)
// ──────────────────────────────────────────────────────────────────────

#[test]
fn a_link_from_an_unknown_node_fails_closed() {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let mystery_in = graph.app_node("mystery-in", MediaRole::StreamInput, 6000);
    let mystery_out = graph.app_node("mystery-out", MediaRole::StreamOutput, 6000);
    let ghost = graph.dangling_id();
    graph.link_ids(ghost, mystery_in.id);
    graph.link(mystery_out, hw);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[("firefox", firefox)],
        &[("mystery-out", mystery_out, "tainted-owner-bridge")],
    );
    assert_tainted(&decisions, mystery_in, "unresolved-ancestry");
}

#[test]
fn an_ambiguous_recycled_global_id_fails_closed() {
    // Two live nodes claiming one id means the observer missed a removal,
    // so every edge touching that id is untrustworthy.
    let mut graph = Graph::new();
    let shared = graph.dangling_id();
    let client = graph.client_of_app(7001);
    let first = graph.node_with_id("first", MediaRole::StreamOutput, shared, app(client, 7001));
    let second = graph.node_with_id("second", MediaRole::StreamOutput, shared, app(client, 7002));
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);

    assert_partition(
        &run(&graph, &ctx()),
        &[("firefox", firefox)],
        &[
            ("first", first, "unresolved-ancestry"),
            ("second", second, "unresolved-ancestry"),
        ],
    );
}

// ──────────────────────────────────────────────────────────────────────
// Uncertainty is not history — it never enters sticky state
// (round 9, from a live phase-5 audit run; see `Uncertainty` in mod.rs)
// ──────────────────────────────────────────────────────────────────────

#[test]
fn unresolved_ancestry_does_not_survive_being_resolved() {
    // Measured live on a desktop: a link is observed while its output node is
    // still unbound, the input side fails closed — correctly — and then that
    // fail-closed mark became *sticky*, so a hardware sink stayed excluded for
    // the process lifetime even after the node resolved and turned out to be
    // an ordinary game. Phase 3r's bind-everything observer widens that window
    // to every node, so this must clear.
    let mut graph = Graph::new();
    let ghost = graph.dangling_id();
    let client = graph.client_of_app(6000);
    let victim = graph.node("victim-in", MediaRole::StreamInput, app(client, 6000));
    let sibling = graph.node("victim-out", MediaRole::StreamOutput, app(client, 6000));
    graph.link_ids(ghost, victim.id);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    let c = ctx();

    // While the ancestry is genuinely unresolved, the decision is unchanged:
    // fail closed, both the victim and its sibling excluded.
    let (first, sticky) = evaluate(&graph.build(), &c, &StickyState::default());
    assert_partition(
        &first,
        &[("firefox", firefox)],
        &[("victim-out", sibling, "tainted-owner-bridge")],
    );
    assert_tainted(&first, victim, "unresolved-ancestry");

    // The node behind that id turns up — nothing tainted, it was simply not
    // observed yet. The uncertainty is gone, so nothing may remain of it.
    let late_client = graph.client_of_app(7100);
    let resolved = graph.node_with_id(
        "was-unbound",
        MediaRole::StreamOutput,
        ghost,
        app(late_client, 7100),
    );
    let (second, _) = evaluate(&graph.build(), &c, &sticky);
    assert_partition(
        &second,
        &[
            ("firefox", firefox),
            ("victim-out", sibling),
            ("was-unbound", resolved),
        ],
        &[],
    );
}

#[test]
fn uncertainty_laundered_into_downstream_taint_is_not_sticky_either() {
    // Retiring by reason *code* would not be enough: an unresolved node
    // propagates `tainted-upstream`, which is indistinguishable from real
    // contamination once recorded. The split has to be by provenance, so a
    // node two hops from the uncertainty must clear too.
    let mut graph = Graph::new();
    let ghost = graph.dangling_id();
    let forwarder_client = graph.client_of_app(6100);
    let forwarder_in = graph.node(
        "fwd-in",
        MediaRole::StreamInput,
        app(forwarder_client, 6100),
    );
    let forwarder_out = graph.node(
        "fwd-out",
        MediaRole::StreamOutput,
        app(forwarder_client, 6100),
    );
    let downstream_client = graph.client_of_app(6200);
    let downstream = graph.node("downstream", MediaRole::Sink, app(downstream_client, 6200));
    let downstream_leg = graph.node(
        "downstream-out",
        MediaRole::StreamOutput,
        app(downstream_client, 6200),
    );
    graph.link_ids(ghost, forwarder_in.id);
    graph.link(forwarder_out, downstream);
    let c = ctx();

    let (first, sticky) = evaluate(&graph.build(), &c, &StickyState::default());
    assert_tainted(&first, forwarder_in, "unresolved-ancestry");
    assert_tainted(&first, downstream, "tainted-upstream");
    assert!(
        first.candidates[&downstream_leg.serial].reason().is_some(),
        "while the ancestry is unresolved the downstream owner is excluded too"
    );

    let late_client = graph.client_of_app(7200);
    graph.node_with_id(
        "was-unbound",
        MediaRole::StreamOutput,
        ghost,
        app(late_client, 7200),
    );
    let (second, _) = evaluate(&graph.build(), &c, &sticky);
    assert_eq!(
        second.candidates[&downstream_leg.serial].reason(),
        None,
        "nothing derived from the uncertainty may outlive it"
    );
    assert_eq!(
        second.candidates[&forwarder_out.serial].reason(),
        None,
        "including the unresolved node's own owner siblings"
    );
}

#[test]
fn real_taint_is_still_sticky_when_its_topology_goes_away() {
    // The other half of the same rule, stated positively: *evidence* is
    // history and must survive. This is the guard on the change above — if
    // provenance splitting ever leaks into the evidence path, peerspeak's own
    // audio starts escaping.
    let (graph, call, rec_in, rec_out, firefox) = sticky_scene();
    let c = ctx();
    let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default());
    let (second, _) = evaluate(&graph.build_without(&[rec_in]), &c, &sticky);
    assert_partition(
        &second,
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("rec-out", rec_out, "tainted-owner-bridge"),
        ],
    );
}

// ──────────────────────────────────────────────────────────────────────
// Stickiness and lifetime-awareness (v3.4 §6.1.3)
// ──────────────────────────────────────────────────────────────────────

/// The scene the sticky tests share: one owner with a tainted input leg and
/// an output leg, plus an untouched bystander.
fn sticky_scene() -> (Graph, NodeRef, NodeRef, NodeRef, NodeRef) {
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let rec_in = graph.app_node("rec-in", MediaRole::StreamInput, 8080);
    let rec_out = graph.app_node("rec-out", MediaRole::StreamOutput, 8080);
    graph.link(hw, rec_in);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    (graph, call, rec_in, rec_out, firefox)
}

#[test]
fn sticky_taint_survives_the_tainted_input_leg_disappearing() {
    // The C3 buffered-audio defect: the recorder stops capturing, but its
    // 5-second ring buffer is still full of peerspeak's audio and no graph
    // event marks the moment it drains.
    let (graph, call, rec_in, rec_out, firefox) = sticky_scene();
    let c = ctx();
    let (first, sticky) = evaluate(&graph.build(), &c, &StickyState::default());
    assert_partition(
        &first,
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("rec-out", rec_out, "tainted-owner-bridge"),
        ],
    );

    let (second, _) = evaluate(&graph.build_without(&[rec_in]), &c, &sticky);
    assert_partition(
        &second,
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("rec-out", rec_out, "tainted-owner-bridge"),
        ],
    );
    assert!(
        matches!(
            second.candidates[&rec_out.serial].eligibility,
            Eligibility::NotEligible { sticky: true, .. }
        ),
        "the second exclusion must be recorded as sticky, not re-derived"
    );
}

#[test]
fn a_new_leg_of_a_still_tainted_owner_inherits_the_taint() {
    // Stickiness is per *owner*, not per node: the recorder opening a
    // second output stream after its input leg closed must not escape.
    let (mut graph, _, rec_in, _, _) = sticky_scene();
    let c = ctx();
    let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default());
    let late = graph.app_node("rec-out-2", MediaRole::StreamOutput, 8080);
    let (second, _) = evaluate(&graph.build_without(&[rec_in]), &c, &sticky);
    assert_eq!(
        second.candidates[&late.serial].reason().map(Reason::code),
        Some("tainted-owner-bridge")
    );
}

#[test]
fn sticky_taint_clears_once_every_owner_member_is_gone() {
    // Otherwise an app is unshareable forever, which is the over-exclusion
    // half of the trade and just as much a bug.
    let (mut graph, call, rec_in, rec_out, firefox) = sticky_scene();
    let c = ctx();
    let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default());

    // Full teardown: both legs and the client object.
    let recorder_client = graph.client_of_app(8080);
    graph.drop_clients(&[recorder_client]);
    let torn_down = graph.build_without(&[rec_in, rec_out]);
    let (_, sticky) = evaluate(&torn_down, &c, &sticky);
    assert!(
        sticky
            .owners
            .iter()
            .all(|owner| !owner.members.contains(&ObjectRef::Node(rec_out.serial))),
        "the recorder's sticky entry should have been dropped: {sticky:#?}"
    );

    // A brand-new recorder with the same PID starts clean.
    let fresh = graph.app_node("rec-out-fresh", MediaRole::StreamOutput, 8080);
    let (third, _) = evaluate(&graph.build_without(&[rec_in, rec_out]), &c, &sticky);
    assert_partition(
        &third,
        &[("firefox", firefox), ("fresh", fresh)],
        &[("call", call, "peerspeak-owned")],
    );
}

#[test]
fn recycled_ids_module_indices_and_link_groups_do_not_inherit_taint() {
    // Every one of these recycles on this stack — measured for module
    // indices, node ids and link-group strings. Only `object.serial` does
    // not, which is exactly why stickiness is keyed on it.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let old_in = graph.group_node("old-in", MediaRole::StreamInput, "loopback-2541-13", 6100);
    let old_out = graph.group_node("old-out", MediaRole::StreamOutput, "loopback-2541-13", 6100);
    graph.link(hw, old_in);

    let c = ctx();
    let (first, sticky) = evaluate(&graph.build(), &c, &StickyState::default());
    assert_eq!(
        first.candidates[&old_out.serial].reason().map(Reason::code),
        Some("tainted-owner-bridge")
    );

    // Teardown, then a brand-new module reusing the global id, the PID and
    // the link-group string verbatim — everything except the serial.
    let old_client = graph.client_of_app(6100);
    graph.drop_clients(&[old_client]);
    let reborn_client = graph.client(Some(PULSE_PID));
    let reborn = graph.node_with_id(
        "reborn-out",
        MediaRole::StreamOutput,
        old_out.id,
        super::fixture::link_group("loopback-2541-13", reborn_client, 6100),
    );
    let (second, _) = evaluate(&graph.build_without(&[old_in, old_out]), &c, &sticky);
    assert_partition(
        &second,
        &[("reborn", reborn)],
        &[("call", call, "peerspeak-owned")],
    );
}

// ──────────────────────────────────────────────────────────────────────
// Regressions from Codex round 1 — each of these was a leak or a lost
// taint history, and each mutation-tests the fix that closed it.
// ──────────────────────────────────────────────────────────────────────

#[test]
fn sticky_taint_survives_when_only_the_client_object_remains() {
    // Finding 1. An app can close every stream it holds while keeping its
    // PipeWire connection open, then open a fresh one — Firefox does this
    // constantly. Seeding sticky taint only from live *nodes* let the new
    // leg come back Eligible while the owner's buffers were still full of
    // the call.
    let (mut graph, call, rec_in, rec_out, firefox) = sticky_scene();
    let c = ctx();
    let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default());

    // Both legs vanish; the client object does not.
    let late = graph.app_node("rec-out-late", MediaRole::StreamOutput, 8080);
    let (next, _) = evaluate(&graph.build_without(&[rec_in, rec_out]), &c, &sticky);
    assert_partition(
        &next,
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("late", late, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn an_inbound_linked_node_of_unknown_role_still_bridges() {
    // Finding 2. "Receives audio" cannot be inferred from `media.class`
    // alone: a node with an absent or unexpected class sits on a real
    // inbound link carrying the call, and if it cannot start an owner
    // bridge its sibling output leg re-emits the call as Eligible.
    let mut graph = Graph::new();
    let sink = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, sink);

    let client = graph.client_of_app(8080);
    let unknown_in = graph.node("unknown-in", MediaRole::Other, app(client, 8080));
    let leaked_out = graph.node("leaked-out", MediaRole::StreamOutput, app(client, 8080));
    graph.link(sink, unknown_in);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("leaked-out", leaked_out, "tainted-owner-bridge"),
        ],
    );
    assert_tainted(&decisions, unknown_in, "tainted-upstream");
}

#[test]
fn a_candidate_that_is_itself_a_receiver_still_bridges_to_its_siblings() {
    // The other half of finding 2: a `Stream/Output/Audio` on an inbound
    // link is excluded by the link walk, but it must also be able to carry
    // taint across the owner bridge to its siblings.
    let mut graph = Graph::new();
    let sink = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, sink);

    let odd_in = graph.app_node("odd-in", MediaRole::StreamOutput, 9100);
    let sibling = graph.app_node("sibling-out", MediaRole::StreamOutput, 9100);
    graph.link(sink, odd_in);

    assert_partition(
        &run(&graph, &ctx()),
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("odd-in", odd_in, "tainted-upstream"),
            ("sibling", sibling, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn a_native_virtual_sink_bridges_on_its_client() {
    // Finding 3. The device exception is keyed on `device.id`, not on the
    // node's role: an app that creates its own `Audio/Sink` plus a
    // re-emitting stream on one client has no `device.id`, no link-group
    // and no `pulse.module.id`, so `client.id` is the *only* thing relating
    // the two. Suppressing coarse keys by role stripped it and leaked the
    // whole call.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    let virtual_sink = graph.native_virtual_node("native-virtual-sink", MediaRole::Sink, 4321);
    let reemitter = graph.native_virtual_node("native-output", MediaRole::StreamOutput, 4321);
    graph.link(call, virtual_sink);
    graph.link(reemitter, hw);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("reemitter", reemitter, "tainted-owner-bridge"),
        ],
    );
    assert_eq!(
        decisions.taint[&reemitter.serial].reason,
        Reason::TaintedOwnerBridge {
            key: Some(OwnerKey::ClientId)
        }
    );
}

#[test]
fn an_asymmetric_forwarder_fails_closed() {
    // Finding 4. The reader carries a `node.link-group` and is therefore
    // "bounded", but its re-emitting leg carries none, so the strong key
    // finds no sibling. Requiring the *source* to be unbounded let this
    // through as Eligible.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);

    let in_client = graph.client(Some(PULSE_PID));
    let out_client = graph.client(Some(PULSE_PID));
    let fwd_in = graph.node(
        "fwd-in",
        MediaRole::StreamInput,
        NodeProps {
            link_group: Some("asymmetric-forwarder".into()),
            client_id: Some(in_client),
            process_id: Some(PULSE_PID),
            ..NodeProps::default()
        },
    );
    let fwd_out = graph.node(
        "fwd-out",
        MediaRole::StreamOutput,
        app(out_client, PULSE_PID),
    );
    graph.link(hw, fwd_in);

    // Blast radius control: a real app is bounded by its own PID.
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    graph.link(firefox, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("fwd-out", fwd_out, "unresolved-owner"),
        ],
    );
}

#[test]
fn a_not_ready_snapshot_never_retires_sticky_owners() {
    // Finding 5. `graph_ready == false` says "this view of the graph is not
    // trustworthy". Retiring sticky owners because their members are absent
    // from an untrustworthy snapshot erases the taint history, and the next
    // ready recompute hands back a clean bill of health for an owner that
    // never went away.
    let (graph, call, rec_in, rec_out, firefox) = sticky_scene();
    let c = ctx();
    let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default());

    let not_ready = ExclusionCtx {
        graph_ready: false,
        ..ctx()
    };
    let (_, carried) = evaluate(
        &super::snapshot::GraphSnapshot::default(),
        &not_ready,
        &sticky,
    );
    assert_eq!(
        carried, sticky,
        "a not-ready recompute must not edit history"
    );

    let (after, _) = evaluate(&graph.build_without(&[rec_in]), &c, &carried);
    assert_partition(
        &after,
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("rec-out", rec_out, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn taint_crosses_two_chained_forwarders() {
    // Finding 6. Every earlier fixture needed at most one owner hop, so
    // replacing the fixpoint with a single pass of each rule survived the
    // whole suite. Two chained forwarders need two.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    let sink1 = graph.module_node("sink1", MediaRole::Sink, 100);
    graph.link(call, sink1);

    let fwd1_in = graph.module_node("fwd1-in", MediaRole::StreamInput, 101);
    let fwd1_out = graph.module_node("fwd1-out", MediaRole::StreamOutput, 101);
    graph.link(sink1, fwd1_in);

    let sink2 = graph.module_node("sink2", MediaRole::Sink, 102);
    graph.link(fwd1_out, sink2);
    let fwd2_in = graph.module_node("fwd2-in", MediaRole::StreamInput, 103);
    let fwd2_out = graph.module_node("fwd2-out", MediaRole::StreamOutput, 103);
    graph.link(sink2, fwd2_in);
    graph.link(fwd2_out, hw);

    // Negative control: an unrelated clean forwarder chain of the same shape.
    let clean_sink = graph.device_node("clean-sink", MediaRole::Sink);
    let clean_in = graph.module_node("clean-in", MediaRole::StreamInput, 200);
    let clean_out = graph.module_node("clean-out", MediaRole::StreamOutput, 200);
    graph.link(clean_sink, clean_in);
    graph.link(clean_out, hw);

    assert_partition(
        &run(&graph, &ctx()),
        &[("clean-out", clean_out)],
        &[
            ("call", call, "peerspeak-owned"),
            ("fwd1-out", fwd1_out, "tainted-owner-bridge"),
            ("fwd2-out", fwd2_out, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn a_stronger_reason_replaces_a_weaker_one_regardless_of_arrival_order() {
    // Finding 7. The old determinism test reversed two links that never
    // competed, so a first-write-wins `raise()` survived it. Here a node is
    // reached by the owner bridge on the first fixpoint pass and by a
    // direct tainted link on the second; the direct link is the stronger
    // reason and must win.
    let build = |reverse: bool| {
        let mut graph = Graph::new();
        let hw = graph.device_node("hw-sink", MediaRole::Sink);
        let call = graph.peerspeak_node("peerspeak", 7);
        graph.link(call, hw);
        let rec_in = graph.app_node("rec-in", MediaRole::StreamInput, 5000);
        let mid = graph.app_node("mid-out", MediaRole::StreamOutput, 5000);
        let owned_sink = graph.native_virtual_node("owned-sink", MediaRole::Sink, 5000);
        if reverse {
            graph.link(mid, owned_sink);
            graph.link(hw, rec_in);
        } else {
            graph.link(hw, rec_in);
            graph.link(mid, owned_sink);
        }
        let decisions = run(&graph, &ctx());
        (
            decisions.taint[&owned_sink.serial].reason.code(),
            decisions.excluded(),
        )
    };
    let forward = build(false);
    assert_eq!(
        forward.0, "tainted-upstream",
        "bridged first, then reached directly — the direct link is stronger"
    );
    assert_eq!(forward, build(true), "and it must not depend on link order");
}

#[test]
fn a_recycled_client_id_and_module_index_do_not_inherit_taint() {
    // Finding 8. The original recycling test never reused a
    // `pulse.module.id`, and its "fresh" client silently reused the dead
    // client's global id because the fixture cache was stale — so the node
    // pointed at a client object that no longer existed.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let old_in = graph.module_node("old-in", MediaRole::StreamInput, 536_870_919);
    let old_out = graph.module_node("old-out", MediaRole::StreamOutput, 536_870_919);
    graph.link(hw, old_in);

    let c = ctx();
    let (first, sticky) = evaluate(&graph.build(), &c, &StickyState::default());
    assert_eq!(
        first.candidates[&old_out.serial].reason().map(Reason::code),
        Some("tainted-owner-bridge")
    );

    // Full teardown, then a module that reuses the node id, the client id
    // and the module index verbatim. Only the serials are new — which is
    // the entire reason stickiness is keyed on them.
    let old_client = graph.client_of_module(536_870_919);
    graph.drop_clients(&[old_client]);
    let reborn_client = graph.client_with_id(old_client, Some(PULSE_PID));
    let reborn = graph.node_with_id(
        "reborn-out",
        MediaRole::StreamOutput,
        old_out.id,
        super::fixture::pulse_module(reborn_client, 536_870_919, PULSE_PID),
    );
    let (second, _) = evaluate(&graph.build_without(&[old_in, old_out]), &c, &sticky);
    assert_partition(
        &second,
        &[("reborn", reborn)],
        &[("call", call, "peerspeak-owned")],
    );
}

#[test]
fn a_deep_forwarder_chain_converges() {
    // Finding 9. The fixpoint is monotone so it terminates, but each pass
    // re-runs a full link BFS and a full owner scan, so a chain of owner
    // hops costs one pass per layer. This is the shape that would expose an
    // accidental blow-up; phase 5 measures the real distribution.
    const LAYERS: u64 = 60;
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    let mut upstream = graph.module_node("sink-0", MediaRole::Sink, 1);
    graph.link(call, upstream);
    let mut last_out = None;
    for layer in 1..=LAYERS {
        let module = 1000 + layer;
        let leg_in = graph.module_node(&format!("fwd-{layer}-in"), MediaRole::StreamInput, module);
        let leg_out =
            graph.module_node(&format!("fwd-{layer}-out"), MediaRole::StreamOutput, module);
        graph.link(upstream, leg_in);
        let next = graph.module_node(&format!("sink-{layer}"), MediaRole::Sink, 2000 + layer);
        graph.link(leg_out, next);
        upstream = next;
        last_out = Some(leg_out);
    }
    graph.link(last_out.expect("at least one layer"), hw);

    let decisions = run(&graph, &ctx());
    assert_eq!(
        decisions.eligible(),
        Vec::new(),
        "every leg of the chain carries the call"
    );
    assert_eq!(
        decisions.candidates.len(),
        LAYERS as usize + 1,
        "the whole chain plus peerspeak's own playback"
    );
}

// ──────────────────────────────────────────────────────────────────────
// Regressions from Codex round 2 — the verification round, where five of
// the round-1 fixes turned out to be partial. Two of these are worse than
// the bugs they were meant to close.
// ──────────────────────────────────────────────────────────────────────

#[test]
fn repeated_evaluation_does_not_contaminate_devices_through_sticky_clients() {
    // Round 2 finding 1, and the sharpest one so far: the round-1 sticky
    // fix smuggled the suppressed key back in. Recording the shared
    // `WirePlumber [export]` Client as a member of the tainted hardware
    // sink's owner meant the *second* recompute expanded that Client to
    // every sound card on the box, tainted the microphone, and excluded
    // every app holding one. The single-evaluate test could not see it.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let mic = graph.device_node("mic", MediaRole::Source);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let ff_in = graph.app_node("firefox-mic", MediaRole::StreamInput, 11114);
    let ff_out = graph.app_node("firefox-out", MediaRole::StreamOutput, 11114);
    graph.link(mic, ff_in);

    let snapshot = graph.build();
    let c = ctx();
    let (first, sticky) = evaluate(&snapshot, &c, &StickyState::default());
    assert_partition(
        &first,
        &[("ff-out", ff_out)],
        &[("call", call, "peerspeak-owned")],
    );

    // The identical graph, evaluated again. Nothing changed, so nothing
    // about the answer may change either.
    let (second, sticky) = evaluate(&snapshot, &c, &sticky);
    assert_partition(
        &second,
        &[("ff-out", ff_out)],
        &[("call", call, "peerspeak-owned")],
    );
    assert_untainted(&second, mic);

    // And it must not drift on the third, either.
    let (third, _) = evaluate(&snapshot, &c, &sticky);
    assert_partition(
        &third,
        &[("ff-out", ff_out)],
        &[("call", call, "peerspeak-owned")],
    );
}

#[test]
fn a_second_connection_of_a_still_tainted_process_inherits_the_taint() {
    // Round 2 finding 2. Following a surviving *connection* is not the same
    // as following a surviving *owner*: the process leaves its first client
    // idle and opens a second one, which the client expansion cannot see.
    // GStreamer opens one connection per stream as a matter of course.
    let (mut graph, call, rec_in, rec_out, firefox) = sticky_scene();
    let c = ctx();
    let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default());

    let client_b = graph.client(Some(PULSE_PID));
    let late = graph.node("rec-out-late", MediaRole::StreamOutput, app(client_b, 8080));
    let (next, _) = evaluate(&graph.build_without(&[rec_in, rec_out]), &c, &sticky);
    assert_partition(
        &next,
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("late", late, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn a_fingerprint_does_not_outlive_its_owner() {
    // The other side of that fix. A fingerprint is a recyclable PID, so it
    // may only be applied while some *serial* member of the owner is still
    // live; once the owner is fully gone, a new process handed the same PID
    // must start clean.
    //
    // ⚠️ The recycled node must exist in the **same** snapshot that first
    // sees the owner fully gone. A later snapshot proves nothing: the entry
    // has been retired by then, so the liveness guard is never consulted
    // and the test passes no matter what it does. (The first version of
    // this test made exactly that mistake and survived the mutation that
    // deletes the guard.)
    let (mut graph, call, rec_in, rec_out, firefox) = sticky_scene();
    let c = ctx();
    let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default());

    let recorder_client = graph.client_of_app(8080);
    graph.drop_clients(&[recorder_client]);
    // A different process that happens to be handed the same PID, present
    // in the very snapshot where the old owner disappears.
    let reborn_client = graph.client(Some(PULSE_PID));
    let reborn = graph.node("reborn", MediaRole::StreamOutput, app(reborn_client, 8080));

    let (after, _) = evaluate(&graph.build_without(&[rec_in, rec_out]), &c, &sticky);
    assert_partition(
        &after,
        &[("firefox", firefox), ("reborn", reborn)],
        &[("call", call, "peerspeak-owned")],
    );
}

#[test]
fn an_ambiguous_input_endpoint_makes_every_claimant_a_receiver() {
    // Round 2 finding 3. Both claimants were already tainted as unresolved
    // through their own ambiguous id — but taint without receiver status
    // cannot start an owner bridge, so both sibling output legs stayed
    // Eligible while one of them was re-emitting the call.
    let mut graph = Graph::new();
    let call = graph.peerspeak_node("call", 7);
    let shared = graph.dangling_id();
    let ca = graph.client_of_app(8000);
    let cb = graph.client_of_app(9000);
    let in_a = graph.node_with_id("in-a", MediaRole::Other, shared, app(ca, 8000));
    let in_b = graph.node_with_id("in-b", MediaRole::Other, shared, app(cb, 9000));
    let out_a = graph.node("out-a", MediaRole::StreamOutput, app(ca, 8000));
    let out_b = graph.node("out-b", MediaRole::StreamOutput, app(cb, 9000));
    graph.link_ids(call.id, shared);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("out-a", out_a, "tainted-owner-bridge"),
            ("out-b", out_b, "tainted-owner-bridge"),
        ],
    );
    assert_tainted(&decisions, in_a, "unresolved-ancestry");
    assert_tainted(&decisions, in_b, "unresolved-ancestry");
}

#[test]
fn an_unresolved_output_endpoint_to_an_unknown_role_node_still_bridges() {
    // Round 2 finding 6: a mutation the 42-test suite survived. The
    // unknown-role test used a *resolved* output endpoint, and the
    // unresolved-output test used a `StreamInput`, which the role union put
    // back into `receivers` anyway — so deleting the receiver insert from
    // the unresolved-output arm changed nothing. This fixture needs both.
    let mut graph = Graph::new();
    let ghost = graph.dangling_id();
    let client = graph.client_of_app(8080);
    let odd = graph.node("odd", MediaRole::Other, app(client, 8080));
    let out = graph.node("out", MediaRole::StreamOutput, app(client, 8080));
    graph.link_ids(ghost, odd.id);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[("firefox", firefox)],
        &[("out", out, "tainted-owner-bridge")],
    );
    assert_tainted(&decisions, odd, "unresolved-ancestry");
}

#[test]
fn a_device_associated_filter_still_bridges_on_its_client() {
    // Round 2 finding 4. `device.id` alone does not mean "passive device
    // node" — PipeWire defines it only as the Device a node belongs to. A
    // filter associated with a card would have lost both its coarse owner
    // keys *and* its ability to trip the fail-closed backstop, so the flag
    // is now a classification phase 3 owes (device.id AND device.api), and
    // anything unclassified is treated as not-a-device.
    let mut graph = Graph::new();
    let call = graph.peerspeak_node("call", 7);
    let filter_in = graph.device_associated_filter("card-filter-in", MediaRole::Sink, 4321);
    let leaked_out =
        graph.device_associated_filter("card-filter-out", MediaRole::StreamOutput, 4321);
    graph.link(call, filter_in);

    assert_partition(
        &run(&graph, &ctx()),
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("filter-out", leaked_out, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn a_not_ready_snapshot_still_records_new_taint() {
    // Round 2 finding 5: the round-1 fix stopped a not-ready epoch erasing
    // history, but also stopped it *recording* any. A reader can consume
    // and buffer the call during that epoch and vanish before readiness,
    // and its output leg came back Eligible. Readiness gates retirement
    // only.
    let (graph, call, rec_in, rec_out, firefox) = sticky_scene();
    let not_ready = ExclusionCtx {
        graph_ready: false,
        ..ctx()
    };
    let (_, sticky) = evaluate(&graph.build(), &not_ready, &StickyState::default());
    assert!(
        !sticky.is_empty(),
        "taint observed during a not-ready epoch is still taint"
    );

    let (ready, _) = evaluate(&graph.build_without(&[rec_in]), &ctx(), &sticky);
    assert_partition(
        &ready,
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("rec-out", rec_out, "tainted-owner-bridge"),
        ],
    );
}

// ──────────────────────────────────────────────────────────────────────
// Regressions from Codex round 3 (second verification round)
// ──────────────────────────────────────────────────────────────────────

#[test]
fn an_inverse_asymmetric_forwarder_fails_closed() {
    // Round 3 finding 1 — the mirror of round-1 finding 4, and a real leak.
    // The reader is unbounded (client.id only, daemon PID suppressed) while
    // its re-emitting leg carries an *unmatched* strong key, so the leg was
    // "bounded" and stayed Eligible. When the reader itself cannot be
    // bounded, its sibling could be any output, so a strong key that does
    // not match it back proves nothing.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw", MediaRole::Sink);
    let call = graph.peerspeak_node("call", 7);
    graph.link(call, hw);

    let in_client = graph.client(Some(PULSE_PID));
    let out_client = graph.client(Some(PULSE_PID));
    let fwd_in = graph.node("fwd-in", MediaRole::StreamInput, app(in_client, PULSE_PID));
    let fwd_out = graph.node(
        "fwd-out",
        MediaRole::StreamOutput,
        NodeProps {
            pulse_module_id: Some(77),
            client_id: Some(out_client),
            process_id: Some(PULSE_PID),
            ..NodeProps::default()
        },
    );
    graph.link(hw, fwd_in);

    let decisions = run(&graph, &ctx());
    assert_eq!(
        decisions.candidates[&fwd_out.serial]
            .reason()
            .map(Reason::code),
        Some("unresolved-owner")
    );
    assert!(decisions.eligible().is_empty());
}

#[test]
fn an_unbounded_tainted_reader_excludes_every_output() {
    // Round 4: conceded to Codex. An unbounded tainted reader could be a
    // real app that simply exposed no PID on its reader leg, so no output
    // property proves independence — exclude the whole candidate universe.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw", MediaRole::Sink);
    let call = graph.peerspeak_node("call", 7);
    graph.link(call, hw);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
    graph.link(firefox, hw);
    // A bounded module leg — even a strong key does not spare it.
    let daemon_leg = graph.module_node("daemon-leg", MediaRole::StreamOutput, 88);
    graph.link(daemon_leg, hw);

    // Without an unbounded reader, both are eligible.
    assert_partition(
        &run(&graph, &ctx()),
        &[("firefox", firefox), ("daemon-leg", daemon_leg)],
        &[("call", call, "peerspeak-owned")],
    );

    // Add a keyless reader with no PID at all — the exact shape the round-3
    // narrowing would have mishandled.
    let keyless_client = graph.client(Some(PULSE_PID));
    let leak_in = graph.node(
        "leak-in",
        MediaRole::StreamInput,
        NodeProps {
            client_id: Some(keyless_client),
            ..NodeProps::default()
        },
    );
    graph.link(hw, leak_in);
    assert_partition(
        &run(&graph, &ctx()),
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("firefox", firefox, "unresolved-owner"),
            ("daemon-leg", daemon_leg, "unresolved-owner"),
        ],
    );
}

#[test]
fn a_real_app_with_no_pid_on_its_reader_leg_does_not_leak() {
    // Codex round 4's exact counterexample to the narrowing I tried: one
    // process, reader leg with no PID (unbounded), output leg with a real
    // PID. The narrowing spared the output; the broad rule excludes it.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw", MediaRole::Sink);
    let call = graph.peerspeak_node("call", 7);
    graph.link(call, hw);

    let in_client = graph.client(Some(PULSE_PID));
    let out_client = graph.client(Some(PULSE_PID));
    let reader = graph.node(
        "reader",
        MediaRole::StreamInput,
        NodeProps {
            client_id: Some(in_client),
            ..NodeProps::default()
        },
    );
    let leaky_out = graph.node("leaky-out", MediaRole::StreamOutput, app(out_client, 4321));
    graph.link(hw, reader);

    assert_partition(
        &run(&graph, &ctx()),
        &[],
        &[
            ("call", call, "peerspeak-owned"),
            ("leaky-out", leaky_out, "unresolved-owner"),
        ],
    );
}

#[test]
fn a_strong_key_new_connection_of_a_still_tainted_owner_inherits_the_taint() {
    // Round 3 finding 5: a mutation that kept only PID fingerprints survived
    // the 49-test suite, because no fixture exercised a *strong-key*
    // fingerprint reaching a new connection. Here the owner is tainted via
    // its `pulse.module.id`, all its nodes vanish, its client stays live,
    // and a second client opens a new leg carrying the same module id.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let mod_in = graph.module_node("mod-in", MediaRole::StreamInput, 77);
    let mod_out = graph.module_node("mod-out", MediaRole::StreamOutput, 77);
    graph.link(hw, mod_in);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);

    let c = ctx();
    let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default());

    // The module's original client stays live; a new connection carries the
    // same module id. The daemon PID is suppressed, so only the module-id
    // fingerprint can catch this.
    let new_client = graph.client(Some(PULSE_PID));
    let late = graph.node(
        "mod-out-late",
        MediaRole::StreamOutput,
        super::fixture::pulse_module(new_client, 77, PULSE_PID),
    );
    let (next, _) = evaluate(&graph.build_without(&[mod_in, mod_out]), &c, &sticky);
    assert_partition(
        &next,
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("late", late, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn a_link_group_new_connection_of_a_still_tainted_owner_inherits_the_taint() {
    // Round 4 finding 3: a mutation dropping LinkGroup fingerprints (keeping
    // only module/PID) survived the 53-test suite, because the strong-key
    // fingerprint test used `pulse.module.id`, not `node.link-group`. Here a
    // link-group owner reads the call, its nodes vanish, its client stays
    // live, and a new client opens a leg with the same link-group and a
    // suppressed PID — only the link-group fingerprint can catch it.
    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);
    let g_in = graph.group_node("g-in", MediaRole::StreamInput, "filter-1", 6200);
    let g_out = graph.group_node("g-out", MediaRole::StreamOutput, "filter-1", 6200);
    graph.link(hw, g_in);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);

    let c = ctx();
    let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default());

    // New connection, same link-group, daemon PID (suppressed) so no usable
    // PID remains — the module/PID fingerprints cannot reach it.
    let new_client = graph.client(Some(PULSE_PID));
    let late = graph.node(
        "g-out-late",
        MediaRole::StreamOutput,
        super::fixture::link_group("filter-1", new_client, PULSE_PID),
    );
    let (next, _) = evaluate(&graph.build_without(&[g_in, g_out]), &c, &sticky);
    assert_partition(
        &next,
        &[("firefox", firefox)],
        &[
            ("call", call, "peerspeak-owned"),
            ("late", late, "tainted-owner-bridge"),
        ],
    );
}

#[test]
fn a_local_root_receiver_bridges_without_an_inbound_link() {
    // Round 5 finding 2: a mutation deleting the *role-based* receiver
    // insertion survived all 55 tests, because every tested bridge source
    // also had an inbound Link that put it in `receivers` anyway. A
    // pixelpass capture sink is a taint root the moment it exists — before
    // anything links into it — and its owner's re-emitting leg must bridge
    // from it on the strength of its role alone.
    let mut graph = Graph::new();
    // A capture sink (PixelpassOwned by name), sharing module id 55 with a
    // re-emitting output leg, and NO inbound link yet.
    let sink = graph.module_node("pixelpass_capture_4242", MediaRole::Sink, 55);
    let leg = graph.module_node("capture-reemit", MediaRole::StreamOutput, 55);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);

    let decisions = run(&graph, &ctx());
    assert_partition(
        &decisions,
        &[("firefox", firefox)],
        &[("leg", leg, "tainted-owner-bridge")],
    );
    assert_tainted(&decisions, sink, "pixelpass-owned");
}

#[test]
fn an_ambiguous_client_id_remembers_every_claimant_for_stickiness() {
    // Round 6 finding 2: nothing pinned the ambiguous-Client branch, so a
    // mutation remembering only the first claimant survived. Two live
    // clients claim one global id; the tainted owner's node references it.
    // If we remember only one and it is the one that later disappears, the
    // still-live claimant that reopens an output escapes.
    use super::snapshot::{ClientSnapshot, GlobalId};

    let mut graph = Graph::new();
    let hw = graph.device_node("hw-sink", MediaRole::Sink);
    let call = graph.peerspeak_node("peerspeak", 7);
    graph.link(call, hw);

    // Two clients share one global id (the observer saw an id collision).
    let shared_id = graph.dangling_id();
    let client_a = graph.client_with_id(shared_id, Some(PULSE_PID));
    let _client_b = graph.client_with_id(shared_id, Some(PULSE_PID));
    assert_eq!(client_a, shared_id);

    // The tainted reader references that (ambiguous) client id, no PID/keys.
    let reader = graph.node(
        "reader",
        MediaRole::StreamInput,
        NodeProps {
            client_id: Some(shared_id),
            ..NodeProps::default()
        },
    );
    let out = graph.node(
        "out",
        MediaRole::StreamOutput,
        NodeProps {
            client_id: Some(shared_id),
            ..NodeProps::default()
        },
    );
    graph.link(hw, reader);
    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);

    let c = ctx();
    let (first, sticky) = evaluate(&graph.build(), &c, &StickyState::default());
    assert_eq!(
        first.candidates[&out.serial].reason().map(Reason::code),
        Some("tainted-owner-bridge")
    );
    // Both claimants must be remembered, or a mutation keeping only one
    // could drop the surviving owner. At least both client serials appear.
    let client_members: usize = sticky
        .owners
        .iter()
        .flat_map(|o| o.members.iter())
        .filter(|m| matches!(m, super::ObjectRef::Client(_)))
        .count();
    assert!(
        client_members >= 2,
        "both ambiguous-id clients should be remembered: {sticky:#?}"
    );
    let _ = (
        ClientSnapshot {
            serial: Serial(0),
            id: GlobalId(0),
            sec_pid: None,
        },
        firefox,
    );
}