diff --git a/src/host/audio.rs b/src/host/audio.rs
index b782857..615538e 100644
--- a/src/host/audio.rs
+++ b/src/host/audio.rs
@@ -615,7 +615,7 @@ fn run_router(
 /// (empty, signed, whitespace-padded, non-numeric, overflowing) is a
 /// property we do not understand and must not guess at. Leading zeroes
 /// are accepted — they are unambiguous and parse to the same value.
-fn parse_object_serial(raw: &str) -> Option<u64> {
+pub(crate) fn parse_object_serial(raw: &str) -> Option<u64> {
     if raw.is_empty() || !raw.bytes().all(|b| b.is_ascii_digit()) {
         return None;
     }
diff --git a/src/host/observer/adapter.rs b/src/host/observer/adapter.rs
new file mode 100644
index 0000000..2fe7c89
--- /dev/null
+++ b/src/host/observer/adapter.rs
@@ -0,0 +1,600 @@
+//! PipeWire I/O adapter for the pure registry observer.
+//!
+//! This module owns a read-only PipeWire main-loop thread, translates registry
+//! callbacks into [`RegEvent`]s, and publishes the latest [`Projection`] for
+//! consumers running outside the PipeWire thread.
+
+use super::classify::DeviceClaim;
+use super::{LinkEndpoints, NodeObservation, Projection, RegEvent, RegistryModel};
+use crate::host::audio::parse_object_serial;
+use crate::host::taint::snapshot::{
+    ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial,
+};
+use anyhow::{Context, Result};
+use pipewire::{self as pw, types::ObjectType};
+use std::cell::{Cell, RefCell};
+use std::collections::{BTreeMap, VecDeque};
+use std::rc::Rc;
+use std::sync::{Arc, Mutex};
+use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
+
+const READINESS_TIMEOUT_MILLIS: u64 = 2_000;
+const TICK_INTERVAL: Duration = Duration::from_millis(250);
+
+/// Tokio-side access to the observer's most recent coherent projection.
+pub struct RegistryObserverHandle {
+    latest: Arc<Mutex<Option<Projection>>>,
+    shutdown_tx: pw::channel::Sender<()>,
+    thread: Option<JoinHandle<()>>,
+}
+
+impl RegistryObserverHandle {
+    /// Spawn the read-only PipeWire registry observer.
+    pub fn spawn() -> Result<Self> {
+        let latest = Arc::new(Mutex::new(None));
+        let latest_for_thread = Arc::clone(&latest);
+        let (shutdown_tx, shutdown_rx) = pw::channel::channel::<()>();
+        let thread = std::thread::Builder::new()
+            .name("pixelpass-pw-observer".to_string())
+            .spawn(move || {
+                if let Err(e) = run_observer(latest_for_thread, shutdown_rx) {
+                    tracing::warn!("registry observer: libpipewire thread exited with error: {e:#}");
+                }
+            })
+            .context("failed to spawn libpipewire registry observer thread")?;
+
+        Ok(Self {
+            latest,
+            shutdown_tx,
+            thread: Some(thread),
+        })
+    }
+
+    /// Return a clone of the latest projection, or `None` before the first
+    /// registry event has been applied.
+    pub fn latest(&self) -> Option<Projection> {
+        self.latest
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .clone()
+    }
+}
+
+impl Drop for RegistryObserverHandle {
+    fn drop(&mut self) {
+        let _ = self.shutdown_tx.send(());
+        if let Some(thread) = self.thread.take()
+            && let Err(e) = thread.join()
+        {
+            tracing::warn!("registry observer: pw thread join failed: {e:?}");
+        }
+    }
+}
+
+struct BoundLink {
+    _proxy: pw::link::Link,
+    _listener: pw::link::LinkListener,
+}
+
+#[derive(Default)]
+struct LiveGlobal {
+    bound_link: Option<BoundLink>,
+}
+
+struct ObserverState {
+    model: RegistryModel,
+    latest: Arc<Mutex<Option<Projection>>>,
+    last_candidate: Option<u32>,
+    live_globals: BTreeMap<GlobalId, VecDeque<LiveGlobal>>,
+}
+
+impl ObserverState {
+    fn new(latest: Arc<Mutex<Option<Projection>>>) -> Self {
+        Self {
+            model: RegistryModel::new(0, READINESS_TIMEOUT_MILLIS),
+            latest,
+            last_candidate: None,
+            live_globals: BTreeMap::new(),
+        }
+    }
+
+    fn apply(&mut self, event: RegEvent) {
+        self.model.apply(event);
+
+        let candidate = self.model.pulse_pid_candidate();
+        if candidate != self.last_candidate {
+            self.last_candidate = candidate;
+            if let Some(pid) = candidate {
+                let comm = std::fs::read_to_string(format!("/proc/{pid}/comm"))
+                    .ok()
+                    .map(|comm| comm.trim_end_matches(['\r', '\n']).to_string());
+                self.model.apply(RegEvent::ProcCommProbed { pid, comm });
+            }
+        }
+
+        self.publish();
+    }
+
+    fn publish(&self) {
+        *self
+            .latest
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(self.model.project());
+    }
+
+    /// Record the global's id and apply its add event as one step, so the
+    /// bound-link FIFO stays provably lockstep with the model's own `live_ids`
+    /// index. Recording only on *applied* adds (never on unknown object types
+    /// or globals dropped for a missing serial) is what keeps the two id
+    /// queues the same length per id — otherwise a phantom slot ahead of a
+    /// bound Link would be popped on removal, leaking that Link's proxy.
+    fn add(&mut self, id: GlobalId, event: RegEvent) {
+        self.live_globals.entry(id).or_default().push_back(LiveGlobal::default());
+        self.apply(event);
+    }
+
+    fn attach_bound_link(&mut self, id: GlobalId, bound_link: BoundLink) {
+        let Some(global) = self.live_globals.get_mut(&id).and_then(VecDeque::back_mut) else {
+            tracing::warn!(
+                global_id = id.0,
+                "registry observer: link bind completed without a live global slot"
+            );
+            return;
+        };
+        global.bound_link = Some(bound_link);
+    }
+
+    fn remove_global(&mut self, id: GlobalId) -> Option<BoundLink> {
+        let (bound_link, empty) = {
+            let globals = self.live_globals.get_mut(&id)?;
+            let bound_link = globals.pop_front().and_then(|global| global.bound_link);
+            (bound_link, globals.is_empty())
+        };
+        if empty {
+            self.live_globals.remove(&id);
+        }
+        bound_link
+    }
+}
+
+fn run_observer(
+    latest: Arc<Mutex<Option<Projection>>>,
+    shutdown_rx: pw::channel::Receiver<()>,
+) -> Result<()> {
+    let started_at = Instant::now();
+    let main_loop =
+        pw::main_loop::MainLoopRc::new(None).context("pw main loop construction failed")?;
+    let context =
+        pw::context::ContextRc::new(&main_loop, None).context("pw context construction failed")?;
+    let core = context
+        .connect_rc(None)
+        .context("pw core connect failed (is the daemon running?)")?;
+    let registry = core.get_registry_rc().context("pw get_registry failed")?;
+    let state = Rc::new(RefCell::new(ObserverState::new(latest)));
+
+    let main_loop_for_shutdown = main_loop.clone();
+    let _shutdown_receiver = shutdown_rx.attach(main_loop.loop_(), move |()| {
+        main_loop_for_shutdown.quit();
+    });
+
+    let pending_sync = Rc::new(Cell::new(None));
+    let pending_sync_for_done = Rc::clone(&pending_sync);
+    let state_for_done = Rc::clone(&state);
+    let _core_listener = core
+        .add_listener_local()
+        .done(move |id, seq| {
+            if id == pw::core::PW_ID_CORE && pending_sync_for_done.get() == Some(seq) {
+                pending_sync_for_done.set(None);
+                state_for_done.borrow_mut().apply(RegEvent::ServerSynced);
+            }
+        })
+        .error(|id, seq, res, message| {
+            tracing::warn!(
+                id,
+                seq,
+                result = res,
+                %message,
+                "registry observer: PipeWire core error"
+            );
+        })
+        .register();
+
+    let registry_weak = registry.downgrade();
+    let state_for_global = Rc::clone(&state);
+    let state_for_remove = Rc::clone(&state);
+    let _registry_listener = registry
+        .add_listener_local()
+        .global(move |obj| {
+            let id = GlobalId(obj.id);
+
+            match obj.type_ {
+                ObjectType::Node => {
+                    let Some(props) = obj.props.as_ref() else {
+                        tracing::warn!(
+                            node_id = obj.id,
+                            "registry observer: Node has no properties; dropping"
+                        );
+                        return;
+                    };
+                    let Some(serial) =
+                        parse_serial(obj.id, "Node", props.get("object.serial"))
+                    else {
+                        return;
+                    };
+                    let node_props = NodeProps {
+                        peerspeak_owned: truthy(props.get("peerspeak.owned")),
+                        pulse_module_id: props
+                            .get("pulse.module.id")
+                            .and_then(|value| value.parse::<u64>().ok()),
+                        link_group: props.get("node.link-group").map(str::to_owned),
+                        client_id: props
+                            .get("client.id")
+                            .and_then(|value| value.parse::<u32>().ok())
+                            .map(GlobalId),
+                        process_id: props
+                            .get("application.process.id")
+                            .and_then(|value| value.parse::<u32>().ok()),
+                        passthrough: truthy(props.get("node.passthrough")),
+                        session_device: false,
+                    };
+                    let observation = NodeObservation {
+                        serial,
+                        id,
+                        name: props.get("node.name").map(str::to_owned),
+                        role: MediaRole::parse(props.get("media.class")),
+                        props: node_props,
+                        device_claim: DeviceClaim {
+                            device_id: props
+                                .get("device.id")
+                                .and_then(|value| value.parse::<u32>().ok())
+                                .map(GlobalId),
+                            device_api: props.get("device.api").map(str::to_owned),
+                            factory_name: props.get("factory.name").map(str::to_owned),
+                        },
+                    };
+                    state_for_global
+                        .borrow_mut()
+                        .add(id, RegEvent::NodeAdded(observation));
+                }
+                ObjectType::Port => {
+                    let Some(props) = obj.props.as_ref() else {
+                        tracing::warn!(
+                            port_id = obj.id,
+                            "registry observer: Port has no properties; dropping"
+                        );
+                        return;
+                    };
+                    let Some(serial) =
+                        parse_serial(obj.id, "Port", props.get("object.serial"))
+                    else {
+                        return;
+                    };
+                    let Some(node) = props
+                        .get("node.id")
+                        .and_then(|value| value.parse::<u32>().ok())
+                        .map(GlobalId)
+                    else {
+                        tracing::warn!(
+                            port_id = obj.id,
+                            node_id = props.get("node.id").unwrap_or("<absent>"),
+                            "registry observer: Port has no usable node.id; dropping"
+                        );
+                        return;
+                    };
+                    let direction = match props.get("port.direction") {
+                        Some("in") => PortDirection::In,
+                        Some("out") => PortDirection::Out,
+                        direction => {
+                            tracing::warn!(
+                                port_id = obj.id,
+                                direction = direction.unwrap_or("<absent>"),
+                                "registry observer: Port has no usable direction; dropping"
+                            );
+                            return;
+                        }
+                    };
+                    state_for_global
+                        .borrow_mut()
+                        .add(id, RegEvent::PortAdded(PortSnapshot {
+                            serial,
+                            id,
+                            node,
+                            direction,
+                            exclusive: truthy(props.get("port.exclusive")),
+                            monitor: truthy(props.get("port.monitor")),
+                        }));
+                }
+                ObjectType::Client => {
+                    let Some(props) = obj.props.as_ref() else {
+                        tracing::warn!(
+                            client_id = obj.id,
+                            "registry observer: Client has no properties; dropping"
+                        );
+                        return;
+                    };
+                    let Some(serial) =
+                        parse_serial(obj.id, "Client", props.get("object.serial"))
+                    else {
+                        return;
+                    };
+                    state_for_global
+                        .borrow_mut()
+                        .add(id, RegEvent::ClientAdded(ClientSnapshot {
+                            serial,
+                            id,
+                            sec_pid: props
+                                .get("pipewire.sec.pid")
+                                .and_then(|value| value.parse::<u32>().ok()),
+                        }));
+                }
+                ObjectType::Device => {
+                    state_for_global
+                        .borrow_mut()
+                        .add(id, RegEvent::DeviceAdded { id });
+                }
+                ObjectType::Link => {
+                    let Some(props) = obj.props.as_ref() else {
+                        tracing::warn!(
+                            link_id = obj.id,
+                            "registry observer: Link has no properties; dropping"
+                        );
+                        return;
+                    };
+                    let Some(serial) =
+                        parse_serial(obj.id, "Link", props.get("object.serial"))
+                    else {
+                        return;
+                    };
+                    let endpoints = link_endpoints_from_props(props);
+                    state_for_global.borrow_mut().add(
+                        id,
+                        RegEvent::LinkAdded {
+                            serial,
+                            id,
+                            endpoints,
+                        },
+                    );
+                    if endpoints.is_some() {
+                        return;
+                    }
+
+                    let Some(registry) = registry_weak.upgrade() else {
+                        return;
+                    };
+                    let link: pw::link::Link = match registry.bind(obj) {
+                        Ok(link) => link,
+                        Err(e) => {
+                            tracing::warn!(
+                                link_id = obj.id,
+                                "registry observer: failed to bind Link for endpoints: {e}"
+                            );
+                            return;
+                        }
+                    };
+                    let resolved = Rc::new(Cell::new(false));
+                    let resolved_for_info = Rc::clone(&resolved);
+                    let state_for_info = Rc::downgrade(&state_for_global);
+                    let listener = link
+                        .add_listener_local()
+                        .info(move |info| {
+                            if resolved_for_info.replace(true) {
+                                return;
+                            }
+                            let endpoints = LinkEndpoints {
+                                output_node: GlobalId(info.output_node_id()),
+                                input_node: GlobalId(info.input_node_id()),
+                                output_port: optional_global_id(info.output_port_id()),
+                                input_port: optional_global_id(info.input_port_id()),
+                            };
+                            if let Some(state) = state_for_info.upgrade() {
+                                state.borrow_mut().apply(RegEvent::LinkEndpointsResolved {
+                                    serial,
+                                    endpoints,
+                                });
+                            }
+                        })
+                        .register();
+                    state_for_global
+                        .borrow_mut()
+                        .attach_bound_link(id, BoundLink {
+                            _proxy: link,
+                            _listener: listener,
+                        });
+                }
+                _ => {}
+            }
+        })
+        .global_remove(move |id| {
+            let id = GlobalId(id);
+            let bound_link = state_for_remove.borrow_mut().remove_global(id);
+            state_for_remove
+                .borrow_mut()
+                .apply(RegEvent::Removed { id });
+            drop(bound_link);
+        })
+        .register();
+
+    pending_sync.set(Some(
+        core.sync(0)
+            .context("registry observer: initial core.sync failed")?,
+    ));
+
+    let state_for_tick = Rc::clone(&state);
+    let timer = main_loop.loop_().add_timer(move |_| {
+        let now = u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
+        state_for_tick.borrow_mut().apply(RegEvent::Tick { now });
+    });
+    timer
+        .update_timer(Some(TICK_INTERVAL), Some(TICK_INTERVAL))
+        .into_result()
+        .context("registry observer: failed to arm readiness timer")?;
+
+    tracing::info!("registry observer: pw thread running");
+    main_loop.run();
+    tracing::info!("registry observer: pw thread exiting");
+    Ok(())
+}
+
+fn parse_serial(id: u32, kind: &str, raw: Option<&str>) -> Option<Serial> {
+    match raw.and_then(parse_object_serial) {
+        Some(serial) => Some(Serial(serial)),
+        None => {
+            tracing::warn!(
+                global_id = id,
+                object_type = kind,
+                serial = raw.unwrap_or("<absent>"),
+                "registry observer: global has no usable object.serial; dropping"
+            );
+            None
+        }
+    }
+}
+
+fn truthy(value: Option<&str>) -> bool {
+    value.is_some_and(|value| value != "false" && value != "0")
+}
+
+fn link_endpoints_from_props(props: &pw::spa::utils::dict::DictRef) -> Option<LinkEndpoints> {
+    let output_node = props.get("link.output.node")?.parse::<u32>().ok()?;
+    let input_node = props.get("link.input.node")?.parse::<u32>().ok()?;
+    Some(LinkEndpoints {
+        output_node: GlobalId(output_node),
+        input_node: GlobalId(input_node),
+        output_port: props
+            .get("link.output.port")
+            .and_then(|value| value.parse::<u32>().ok())
+            .map(GlobalId),
+        input_port: props
+            .get("link.input.port")
+            .and_then(|value| value.parse::<u32>().ok())
+            .map(GlobalId),
+    })
+}
+
+fn optional_global_id(id: u32) -> Option<GlobalId> {
+    (id != pw::constants::ID_ANY).then_some(GlobalId(id))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::process::Command;
+
+    struct PactlModule {
+        id: Option<u32>,
+    }
+
+    impl PactlModule {
+        fn load(name: &str, args: &[String]) -> Self {
+            let output = Command::new("pactl")
+                .arg("load-module")
+                .arg(name)
+                .args(args)
+                .output()
+                .expect("pactl must be installed for the live observer test");
+            assert!(
+                output.status.success(),
+                "pactl load-module {name} failed: {}",
+                String::from_utf8_lossy(&output.stderr).trim()
+            );
+            let id = String::from_utf8(output.stdout)
+                .expect("pactl module id must be UTF-8")
+                .trim()
+                .parse::<u32>()
+                .expect("pactl module id must be a u32");
+            Self { id: Some(id) }
+        }
+
+        fn unload(mut self) {
+            let id = self.id.take().expect("module must still be loaded");
+            let output = Command::new("pactl")
+                .arg("unload-module")
+                .arg(id.to_string())
+                .output()
+                .expect("pactl must be installed for the live observer test");
+            assert!(
+                output.status.success(),
+                "pactl unload-module {id} failed: {}",
+                String::from_utf8_lossy(&output.stderr).trim()
+            );
+        }
+    }
+
+    impl Drop for PactlModule {
+        fn drop(&mut self) {
+            if let Some(id) = self.id.take() {
+                let _ = Command::new("pactl")
+                    .arg("unload-module")
+                    .arg(id.to_string())
+                    .output();
+            }
+        }
+    }
+
+    fn wait_for(
+        observer: &RegistryObserverHandle,
+        predicate: impl Fn(&Projection) -> bool,
+    ) -> Projection {
+        let deadline = Instant::now() + Duration::from_secs(10);
+        while Instant::now() < deadline {
+            if let Some(projection) = observer.latest()
+                && predicate(&projection)
+            {
+                return projection;
+            }
+            std::thread::sleep(Duration::from_millis(50));
+        }
+        panic!("timed out waiting for the registry projection");
+    }
+
+    fn has_node(projection: &Projection, name: &str) -> bool {
+        projection
+            .snapshot
+            .nodes()
+            .any(|node| node.name.as_deref() == Some(name))
+    }
+
+    #[test]
+    #[ignore = "needs live pipewire"]
+    fn live_topology_diff_tracks_null_sink_and_loopback() {
+        pw::init();
+        let observer = RegistryObserverHandle::spawn().expect("observer thread must spawn");
+        let baseline = wait_for(&observer, |projection| projection.graph_ready);
+        let baseline_links = baseline.snapshot.links().count();
+
+        let unique = format!("pixelpass_observer_test_{}", std::process::id());
+        let capture_name = format!("{unique}_capture");
+        let playback_name = format!("{unique}_playback");
+        let null_sink = PactlModule::load(
+            "module-null-sink",
+            &[format!("sink_name={unique}")],
+        );
+        let with_sink = wait_for(&observer, |projection| has_node(projection, &unique));
+        let sink_links = with_sink.snapshot.links().count();
+
+        let loopback = PactlModule::load(
+            "module-loopback",
+            &[
+                format!("source={unique}.monitor"),
+                format!("sink={unique}"),
+                format!("source_output_properties=node.name={capture_name}"),
+                format!("sink_input_properties=node.name={playback_name}"),
+            ],
+        );
+        wait_for(&observer, |projection| {
+            has_node(projection, &capture_name)
+                && has_node(projection, &playback_name)
+                && projection.snapshot.links().count() > sink_links
+        });
+
+        loopback.unload();
+        null_sink.unload();
+        wait_for(&observer, |projection| {
+            !has_node(projection, &unique)
+                && !has_node(projection, &capture_name)
+                && !has_node(projection, &playback_name)
+                && projection.snapshot.links().count() <= baseline_links
+        });
+    }
+}
diff --git a/src/host/observer/mod.rs b/src/host/observer/mod.rs
index cb0abbe..549accf 100644
--- a/src/host/observer/mod.rs
+++ b/src/host/observer/mod.rs
@@ -26,6 +26,7 @@
 
 #![allow(dead_code)] // Wired by the phase-3 adapter (Codex's half) and consumed by later phases.
 
+pub mod adapter;
 pub mod classify;
 pub mod pulse_pid;