Recover routed streams after PipeWire relink

Author: bhack

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()):

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_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")])

tests/test_release_preflight.py

@@ -88,6 +88,7 @@ def test_release_preflight_runs_headless_pipewire_runtime_smoke(monkeypatch) ->
     monkeypatch.setenv("MINI_EQ_HEADLESS_PIPEWIRE_TIMEOUT", "12")
     monkeypatch.setenv("MINI_EQ_HEADLESS_PIPEWIRE_CYCLES", "4")
     monkeypatch.setenv("MINI_EQ_HEADLESS_PIPEWIRE_AUDIO_DURATION", "34")
+    monkeypatch.setenv("MINI_EQ_HEADLESS_PIPEWIRE_IDLE_GAP", "5")
     monkeypatch.setattr(release_preflight, "run", lambda command, **_kwargs: commands.append(command))
 
     release_preflight.run_headless_pipewire_runtime_smoke(Path("/python"))
@@ -102,6 +103,8 @@ def test_release_preflight_runs_headless_pipewire_runtime_smoke(monkeypatch) ->
             "4",
             "--audio-duration",
             "34",
+            "--idle-gap",
+            "5",
         ]
     ]
 
@@ -132,6 +135,7 @@ def test_release_preflight_uses_hosted_headless_pipewire_defaults(monkeypatch) -
     monkeypatch.delenv("MINI_EQ_HEADLESS_PIPEWIRE_TIMEOUT", raising=False)
     monkeypatch.delenv("MINI_EQ_HEADLESS_PIPEWIRE_CYCLES", raising=False)
     monkeypatch.delenv("MINI_EQ_HEADLESS_PIPEWIRE_AUDIO_DURATION", raising=False)
+    monkeypatch.delenv("MINI_EQ_HEADLESS_PIPEWIRE_IDLE_GAP", raising=False)
     monkeypatch.setattr(release_preflight, "run", lambda command, **_kwargs: commands.append(command))
 
     release_preflight.run_headless_pipewire_runtime_smoke(Path("/python"))
@@ -146,6 +150,8 @@ def test_release_preflight_uses_hosted_headless_pipewire_defaults(monkeypatch) -
             "2",
             "--audio-duration",
             "180",
+            "--idle-gap",
+            "8",
         ]
     ]