Recover routed streams after PipeWire relink

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 0.8.2 - 2026-05-16
+
+- Fixed routed audio recovery after PipeWire or WirePlumber relinks playback
+  streams while system audio routing is enabled.
+- Fixed output recovery when speakers, headphones, or displays are recreated
+  under the same PipeWire name.
+
 ## 0.8.1 - 2026-05-14
 
 - Fix first-run Flatpak window sizing and state restoration.

data/io.github.bhack.mini-eq.metainfo.xml

@@ -33,18 +33,26 @@
   </description>
   <screenshots>
     <screenshot type="default">
-      <image>https://raw.githubusercontent.com/bhack/mini-eq/v0.8.1/docs/screenshots/mini-eq.png</image>
+      <image>https://raw.githubusercontent.com/bhack/mini-eq/v0.8.2/docs/screenshots/mini-eq.png</image>
       <caption>Adjust sound output with equalizer controls</caption>
     </screenshot>
     <screenshot>
-      <image>https://raw.githubusercontent.com/bhack/mini-eq/v0.8.1/docs/screenshots/mini-eq-dark.png</image>
+      <image>https://raw.githubusercontent.com/bhack/mini-eq/v0.8.2/docs/screenshots/mini-eq-dark.png</image>
       <caption>Use the equalizer with dark style</caption>
     </screenshot>
   </screenshots>
   <url type="homepage">https://github.com/bhack/mini-eq</url>
   <url type="bugtracker">https://github.com/bhack/mini-eq/issues</url>
   <url type="vcs-browser">https://github.com/bhack/mini-eq</url>
   <releases>
+    <release version="0.8.2" date="2026-05-16">
+      <description>
+        <ul>
+          <li>Fixed routed audio recovery after PipeWire or WirePlumber relinks playback streams while system audio routing is enabled.</li>
+          <li>Fixed output recovery when speakers, headphones, or displays are recreated under the same PipeWire name.</li>
+        </ul>
+      </description>
+    </release>
     <release version="0.8.1" date="2026-05-14">
       <description>
         <ul>

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "mini-eq"
-version = "0.8.1"
+version = "0.8.2"
 description = "Compact PipeWire system-wide parametric equalizer for Linux desktops."
 readme = "README.md"
 requires-python = ">=3.11"

src/mini_eq/pipewire_stream_router.py

@@ -143,6 +143,9 @@ def _processing_path_node_ids(self) -> set[int]:
     def _link_touches_processing_path(self, link: PipeWireLink) -> bool:
         return bool(self._processing_path_node_ids() & {link.output_node_id, link.input_node_id})
 
+    def _link_touches_routed_stream(self, link: PipeWireLink) -> bool:
+        return bool(self.routed_stream_ids & {link.output_node_id, link.input_node_id})
+
     def handle_link_state_changed(self, state: str | None) -> None:
         if state == LINK_STATE_ACTIVE:
             self.schedule_refresh(route_applied=True)
@@ -318,14 +321,19 @@ def handle_object_added(self, _manager, node) -> None:
         if self._link_touches_processing_path(link):
             self.track_processing_link_state(link)
             self.schedule_refresh(route_applied=True)
+        elif self._link_touches_routed_stream(link):
+            self.schedule_refresh()
 
     def handle_object_removed(self, _manager, node) -> None:
         try:
             link = self.backend.link_from_proxy(node)
         except Exception:
             return
 
+        should_refresh = self._link_touches_processing_path(link) or self._link_touches_routed_stream(link)
         self.untrack_processing_link_state(link)
+        if should_refresh:
+            self.schedule_refresh()
 
     def untrack_processing_link_states(self) -> None:
         for handler_id in list(self.link_state_handler_ids.values()):

src/mini_eq/routing.py

@@ -70,6 +70,7 @@ def __init__(self, output_sink: str | None) -> None:
         self.filter_node_id: int | None = None
         self.output_event_source_id = 0
         self.pending_followed_output_sink: str | None = None
+        self.pending_current_output_sink_refresh = False
         self.output_object_added_handler_id = 0
         self.output_object_removed_handler_id = 0
         self.output_metadata_changed_handler_id = 0
@@ -148,6 +149,20 @@ def get_sink(self, sink_name: str) -> PipeWireNode | None:
 
         return self.output_backend.audio_sink_by_name(sink_name)
 
+    def filter_output_already_targets_sink(self, sink: PipeWireNode) -> bool:
+        if not self.running or self.filter_node_id is None or not sink.object_serial:
+            return False
+
+        try:
+            filter_output = self.output_backend.output_stream_by_name(self.filter_output_name)
+            if filter_output is None:
+                return False
+            target = self.output_backend.stream_target(filter_output.bound_id)
+        except Exception:
+            return False
+
+        return target.target_object == sink.object_serial
+
     def output_preset_keys(self) -> tuple[str, ...]:
         return self.output_preset_target().keys
 
@@ -271,9 +286,33 @@ def on_error(exc: Exception) -> None:
         return analyzer.set_enabled(enabled)
 
     def switch_output_sink(self, sink_name: str, explicit: bool) -> None:
-        if not sink_name or sink_name == self.output_sink:
+        if not sink_name:
+            if explicit:
+                self.follow_default_output = False
+            return
+
+        if sink_name == self.output_sink:
             if explicit:
                 self.follow_default_output = False
+
+            output_sink = self.get_sink(sink_name)
+            if output_sink is None:
+                return
+
+            self.refresh_output_route_param_monitor()
+            if self.stream_router is not None:
+                self.stream_router.set_output_sink_name(sink_name)
+            if self.output_analyzer is not None:
+                output_sink_description = output_sink.node_description if output_sink is not None else None
+                self.output_analyzer.set_output_sink_name(sink_name, output_sink_description)
+
+            if self.filter_output_already_targets_sink(output_sink):
+                return
+
+            if self.retarget_filter_output():
+                return
+
+            self.restart_engine()
             return
 
         if not self.is_valid_output_sink(sink_name):
@@ -356,6 +395,8 @@ def handle_output_object_added(self, _manager, proxy) -> None:
             return
 
         if node.is_audio_sink:
+            if node.node_name == getattr(self, "output_sink", ""):
+                self.pending_current_output_sink_refresh = True
             self.schedule_output_event_refresh()
 
     def handle_output_object_removed(self, _manager, _proxy) -> None:
@@ -423,11 +464,23 @@ def on_output_event_idle(self) -> bool:
 
         self.invalidate_output_preset_target()
         pending_followed_output_sink = getattr(self, "pending_followed_output_sink", None)
+        pending_current_output_sink_refresh = getattr(self, "pending_current_output_sink_refresh", False)
         self.pending_followed_output_sink = None
+        self.pending_current_output_sink_refresh = False
+        followed_output_refreshed = False
         if pending_followed_output_sink is not None:
-            self.refresh_followed_output_sink_from_event(pending_followed_output_sink)
+            followed_output_refreshed = self.refresh_followed_output_sink_from_event(pending_followed_output_sink)
         else:
-            self.refresh_followed_output_sink(snapshot=True)
+            followed_output_refreshed = self.refresh_followed_output_sink(snapshot=True)
+        if (
+            pending_current_output_sink_refresh
+            and not followed_output_refreshed
+            and self.get_sink(getattr(self, "output_sink", "")) is not None
+        ):
+            try:
+                self.switch_output_sink(self.output_sink, explicit=False)
+            except Exception as exc:
+                self.emit_status(f"output refresh warning: {exc}")
         self.refresh_output_route_param_monitor()
 
         if self.outputs_changed_callback is not None:

tests/test_check_flatpak_runtime.py

@@ -5,6 +5,28 @@
 from tools import check_flatpak_runtime
 
 
+def node_item(item_id: int, name: str) -> dict:
+    return {
+        "id": item_id,
+        "type": "PipeWire:Interface:Node",
+        "info": {"props": {"node.name": name, "object.serial": str(item_id + 1000)}},
+    }
+
+
+def link_item(item_id: int, output_node: int, input_node: int, state: str) -> dict:
+    return {
+        "id": item_id,
+        "type": "PipeWire:Interface:Link",
+        "info": {
+            "state": state,
+            "props": {
+                "link.output.node": str(output_node),
+                "link.input.node": str(input_node),
+            },
+        },
+    }
+
+
 @pytest.mark.parametrize(
     "app_ref",
     [
@@ -37,3 +59,37 @@ def test_flatpak_runtime_smoke_includes_extra_flatpak_run_args(monkeypatch: pyte
         "io.github.bhack.mini-eq//master",
         "--check-deps",
     ]
+
+
+def test_flatpak_runtime_recognizes_active_processing_path(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        check_flatpak_runtime,
+        "read_pw_dump",
+        lambda: [
+            node_item(10, "mini_eq_sink"),
+            node_item(20, "mini_eq_sink_output"),
+            node_item(30, "ci_null_sink"),
+            node_item(40, "browser"),
+            link_item(90, 40, 10, "active"),
+            link_item(91, 20, 30, "active"),
+        ],
+    )
+
+    assert check_flatpak_runtime.processing_path_has_active_links() is True
+
+
+def test_flatpak_runtime_rejects_inactive_processing_path(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        check_flatpak_runtime,
+        "read_pw_dump",
+        lambda: [
+            node_item(10, "mini_eq_sink"),
+            node_item(20, "mini_eq_sink_output"),
+            node_item(30, "ci_null_sink"),
+            node_item(40, "browser"),
+            link_item(90, 40, 10, "active"),
+            link_item(91, 20, 30, "paused"),
+        ],
+    )
+
+    assert check_flatpak_runtime.processing_path_has_active_links() is False

tests/test_check_headless_pipewire_runtime.py

@@ -25,6 +25,26 @@ def link_item(item_id: int, output_node: int, input_node: int, state: str) -> di
     }
 
 
+class AlwaysPendingContext:
+    def __init__(self) -> None:
+        self.iterations = 0
+
+    def pending(self) -> bool:
+        return True
+
+    def iteration(self, may_block: bool) -> None:
+        assert may_block is False
+        self.iterations += 1
+
+
+def test_drain_main_context_limits_continuous_pending_events() -> None:
+    context = AlwaysPendingContext()
+
+    headless.drain_main_context(context, max_iterations=3)
+
+    assert context.iterations == 3
+
+
 def test_headless_runtime_recognizes_active_processing_path(monkeypatch) -> None:
     monkeypatch.setattr(
         headless.live,
@@ -71,3 +91,25 @@ def test_headless_runtime_rejects_stale_virtual_route(monkeypatch) -> None:
     monkeypatch.setattr(headless.live, "metadata_targets", lambda: {42: ("old-serial", "Spa:Id")})
 
     assert headless.route_to_current_virtual(42, "mini_eq_sink") is None
+
+
+def test_dynamic_sink_properties_create_hotplug_audio_sink() -> None:
+    properties = headless.dynamic_sink_properties("ci_hotplug_sink")
+
+    assert 'node.name = "ci_hotplug_sink"' in properties
+    assert 'media.class = "Audio/Sink"' in properties
+    assert "object.linger = true" in properties
+    assert "factory.name = support.null-audio-sink" in properties
+    assert "session.suspend-timeout-seconds = 1" in properties
+
+
+def test_alsa_null_sink_properties_create_alsa_pcm_audio_sink() -> None:
+    properties = headless.alsa_null_sink_properties("ci_alsa_null_sink")
+
+    assert 'node.name = "ci_alsa_null_sink"' in properties
+    assert 'media.class = "Audio/Sink"' in properties
+    assert "object.linger = true" in properties
+    assert "factory.name = api.alsa.pcm.sink" in properties
+    assert 'api.alsa.path = "null"' in properties
+    assert 'audio.format = "S16LE"' in properties
+    assert "session.suspend-timeout-seconds = 1" in properties

tests/test_github_workflows.py

@@ -110,6 +110,7 @@ def test_headless_pipewire_runtime_smoke_is_optional_ci_gate_without_nested_gnom
     assert "python3-pyatspi" not in job
     assert 'timeout="${MINI_EQ_HEADLESS_PIPEWIRE_TIMEOUT:-90}"' in script
     assert 'audio_duration="${MINI_EQ_HEADLESS_PIPEWIRE_AUDIO_DURATION:-180}"' in script
+    assert 'idle_gap="${MINI_EQ_HEADLESS_PIPEWIRE_IDLE_GAP:-8}"' in script
 
 
 def test_live_ui_runtime_smoke_uses_host_gir_build_environment() -> None:
@@ -129,3 +130,4 @@ def test_flatpak_runtime_smoke_tolerates_pipewire_startup_race() -> None:
     assert "{ pw-dump 2>/dev/null || true; }" in script
     assert "first(.[] | select(" in script
     assert "| head -n 1" not in script
+    assert 'idle_gap="${MINI_EQ_FLATPAK_SMOKE_IDLE_GAP:-8}"' in script

tests/test_mini_eq_pipewire_stream_router.py

@@ -536,6 +536,61 @@ def test_pipewire_router_reapplies_controls_when_processing_link_becomes_active(
     assert applied == ["apply", "apply"]
 
 
+def test_pipewire_router_reroutes_tracked_stream_when_relinked_away(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    spotify = make_node(1, pw_backend.STREAM_OUTPUT_AUDIO, "spotify", "Spotify")
+    speakers = make_node(22, pw_backend.AUDIO_SINK, "speakers")
+    backend = FakePipeWireBackend([spotify], sinks=[speakers])
+    router = pw_router.PipeWireStreamRouter("mini_eq_sink", "mini_eq_sink_output", lambda _message: None, backend)
+    scheduled_callbacks: list[object] = []
+
+    monkeypatch.setattr(
+        pw_router.GLib,
+        "idle_add",
+        lambda callback: scheduled_callbacks.append(callback) or 321,
+    )
+
+    router.enabled = True
+    router.accept_stream_events = True
+    router.routed_stream_ids = {spotify.bound_id}
+    router.handle_object_added(None, make_link(92, output_node_id=spotify.bound_id, input_node_id=speakers.bound_id))
+
+    assert len(scheduled_callbacks) == 1
+    scheduled_callbacks[0]()
+
+    assert backend.moves == [(spotify.bound_id, "mini_eq_sink")]
+
+
+def test_pipewire_router_reroutes_tracked_stream_when_current_link_disappears(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    spotify = make_node(1, pw_backend.STREAM_OUTPUT_AUDIO, "spotify", "Spotify")
+    virtual_sink = make_node(11, pw_backend.AUDIO_SINK, "mini_eq_sink")
+    backend = FakePipeWireBackend([spotify], sinks=[virtual_sink])
+    router = pw_router.PipeWireStreamRouter("mini_eq_sink", "mini_eq_sink_output", lambda _message: None, backend)
+    scheduled_callbacks: list[object] = []
+
+    monkeypatch.setattr(
+        pw_router.GLib,
+        "idle_add",
+        lambda callback: scheduled_callbacks.append(callback) or 321,
+    )
+
+    router.enabled = True
+    router.accept_stream_events = True
+    router.routed_stream_ids = {spotify.bound_id}
+    router.handle_object_removed(
+        None,
+        make_link(92, output_node_id=spotify.bound_id, input_node_id=virtual_sink.bound_id),
+    )
+
+    assert len(scheduled_callbacks) == 1
+    scheduled_callbacks[0]()
+
+    assert backend.moves == [(spotify.bound_id, "mini_eq_sink")]
+
+
 def test_pipewire_router_tracks_internal_output_links(monkeypatch: pytest.MonkeyPatch) -> None:
     internal_output = make_node(90, pw_backend.STREAM_OUTPUT_AUDIO, "mini_eq_sink_output")
     backend = FakePipeWireBackend([internal_output], sinks=[make_node(22, pw_backend.AUDIO_SINK, "speakers")])