Added fix

ARCHITECTURE.md

@@ -190,6 +190,13 @@ must propagate and converge, not an ephemeral server-side session flag.
 
 ## Playback across devices
 
+A persisted device identity has exactly one live coordinator. The TUI holds
+an OS file lock beside its device-sync database for the entire runtime; two
+installations using the same user data directory cannot open it concurrently.
+Network polling is independent per trusted peer, with one in-flight exchange
+per peer and a bounded deadline. A stalled peer must not delay the next poll
+of a healthy output. Poll tasks are cancelled when federation shuts down.
+
 Playback has one logical state but remains physically local to the device
 producing audio.
 

Cargo.lock

@@ -1698,9 +1698,9 @@ dependencies = [
 
 [[package]]
 name = "furumi-library"
-version = "0.1.0"
+version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d6f15ab98c65d89ea18e55ed8852abd2f029f63876d357270c4647d73601d0a2"
+checksum = "f78466b6345cb90d7bb37db5d9dd67476ab6c2c3b2202876d24f448c314a9a08"
 dependencies = [
  "anyhow",
  "blake3",
@@ -1715,7 +1715,7 @@ dependencies = [
 
 [[package]]
 name = "furumi_tui"
-version = "0.3.0"
+version = "0.3.1"
 dependencies = [
  "anyhow",
  "base64",
@@ -3239,9 +3239,9 @@ dependencies = [
 
 [[package]]
 name = "music-dht"
-version = "0.4.1"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7bfe5cee89fa891b00e84738f947c924c6845fd84fedb4697cb073e3fe63bb81"
+checksum = "4c3f75d49d3a742a6777a1c4cc53f397399dfc9ddb70034c57aeb44d75d2ee4f"
 dependencies = [
  "async-trait",
  "blake3",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "furumi_tui"
-version = "0.3.0"
+version = "0.3.1"
 edition = "2024"
 rust-version = "1.97"
 description = "A federated P2P player for personal music libraries"
@@ -27,12 +27,12 @@ crokey = "1.4.0"
 crossterm = { version = "0.29.0", features = ["event-stream"] }
 directories = "6.0.0"
 futures-util = "0.3.32"
-furumi-library = "0.1.0"
+furumi-library = "0.2.0"
 image = { version = "0.25.10", default-features = false, features = ["jpeg", "png", "webp", "gif", "bmp"] }
 lofty = "0.22"
 # P2P federation: library index in a shared DHT + audio streaming between
 # peers (same protocol as furumi-fd).
-music-dht = "0.4.1"
+music-dht = "0.5.0"
 ratatui = "0.30.1"
 reqwest = { version = "0.12.28", default-features = false, features = ["rustls-tls", "stream", "blocking"] }
 rhai = { version = "1", features = ["sync"] }

README.md

@@ -195,6 +195,55 @@ cargo check --all-targets
 cargo test --all-targets
 ```
 
+## Playback coordination
+
+Connected Devices uses the shared `music_dht::playback` engine from frid.
+Newly started players take an idle/paused output after discovery and become
+controllers when another device is playing. Missing output reports trigger
+automatic failover; concurrent claims converge to one owner. The web gateway
+uses a passive server profile and reports actual browser activity.
+
+Only one TUI process may use a device identity. Installed and locally built
+binaries use the same user data directory: close the installed player before
+starting a development build. An OS file lock prevents duplicate coordinators
+and is released automatically on exit or a crash. Older builds do not take
+this lock, so close those explicitly when upgrading.
+
+Connected-device polling runs independently per peer, so an offline or stalled
+device does not postpone updates to live players.
+
+The existing device menu handles manual transfers. Advanced policy can be set
+in `settings.toml` without changing the UI:
+
+```toml
+[playback]
+claim_on_startup = true
+automatic_failover = true
+take_paused_on_startup = true
+discovery_ms = 3000
+owner_timeout_ms = 120000
+```
+
+All clients must support the new coordination envelope for eventual single
+output ownership. frid's `PLAYBACK_PROTOCOL.md` describes the protocol, adapter
+contract and rollout. Local Cargo patches are only for development; publish
+frid and bump the client dependency versions before releasing these changes.
+
+Run the local protocol checks from the TUI repository:
+
+```bash
+cargo test localhost_devices_exchange_state_and_handoff
+python scripts/test_device_interop.py ../furumusic
+```
+
+The first test uses real iroh streams, isolated SQLite databases, ownership
+handoff, and a stalled peer. The second builds both player test binaries and
+exchanges their actual JSON messages over `127.0.0.1`: queue metadata,
+bidirectional handoff, pause/seek and duplicate command fencing. It exercises
+the TUI command adapter and web player hub, but does not start a browser or
+PostgreSQL and does not cover web database migrations. Both tests use temporary
+identities and data; no running player or user library is used.
+
 ## License
 
 Furumi is released under the

scripts/test_device_interop.py

@@ -0,0 +1,65 @@
+"""Build and run the real TUI and web protocol adapters against each other.
+
+Usage: python scripts/test_device_interop.py [path/to/furumusic]
+No user databases, accounts, audio outputs, or external servers are used.
+"""
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+
+
+def test_binary(repo):
+    result = subprocess.run(
+        ["cargo", "test", "--locked", "--no-run", "--message-format=json"],
+        cwd=repo, text=True, encoding="utf-8", stdout=subprocess.PIPE,
+    )
+    result.check_returncode()
+    binaries = []
+    for line in result.stdout.splitlines():
+        event = json.loads(line)
+        if event.get("reason") == "compiler-artifact" and event.get("profile", {}).get("test") and event.get("executable"):
+            binaries.append(event["executable"])
+    if len(binaries) != 1:
+        raise RuntimeError(f"Expected one player test binary in {repo}, found {binaries}")
+    return binaries[0]
+
+
+def main():
+    tui = Path(__file__).resolve().parents[1]
+    web = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else tui.parent / "furumusic"
+    tui_bin, web_bin = test_binary(tui), test_binary(web)
+    with tempfile.TemporaryDirectory(prefix="furumi-interop-") as directory:
+        env = dict(os.environ, FURUMI_INTEROP_DIR=directory)
+        processes = []
+        try:
+            for binary, test in [(web_bin, "federation::devices::interop_tests::localhost_tui_peer"),
+                                 (tui_bin, "devices::interop_tests::localhost_web_peer")]:
+                listing = subprocess.check_output([binary, "--list"], text=True, encoding="utf-8")
+                if f"{test}: test" not in listing:
+                    raise RuntimeError(f"Required test {test} is missing from {binary}")
+                processes.append(subprocess.Popen(
+                    [binary, "--ignored", "--exact", test, "--nocapture"], env=env,
+                    stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                    text=True, encoding="utf-8",
+                    creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
+                ))
+            codes = []
+            for process in processes:
+                output, _ = process.communicate(timeout=45)
+                print(output, end="")
+                codes.append(process.returncode)
+            if any(codes):
+                raise RuntimeError(f"Cross-player test failed: exit codes {codes}")
+        finally:
+            for process in processes:
+                if process.poll() is None:
+                    process.kill()
+                process.wait()
+    print("PASS: TUI <-> WEB state, bidirectional handoff, pause/seek, duplicate fencing")
+
+
+if __name__ == "__main__":
+    main()

src/app/event.rs

@@ -185,7 +185,11 @@ pub enum AppEvent {
     /// Trusted device playback state, delivered by personal-device sync.
     DevicePlayback(crate::devices::PlaybackSnapshot),
     /// Playback command addressed to this device.
-    PlaybackCommand(crate::devices::PlaybackCommand),
+    PlaybackCommand {
+        command: crate::devices::PlaybackCommand,
+        authority: music_dht::playback::CommandStamp,
+        origin: String,
+    },
     /// Current lifecycle/status of the federation Jam.
     JamStatus(crate::jam::JamStatus),
     /// Host playback snapshot received by a Jam participant.

src/app/mod.rs

@@ -32,7 +32,6 @@ use update::{Effect, update};
 
 const TICK_INTERVAL: Duration = Duration::from_millis(250);
 const VISUALIZER_TICK_INTERVAL: Duration = Duration::from_millis(50);
-const ACTIVE_IDLE_LEASE_MS: i64 = 5 * 60 * 1000;
 
 /// Handles shared by background tasks; AppState stays pure UI data.
 pub struct Runtime {
@@ -297,6 +296,8 @@ pub async fn run(
     }
 
     let devices = crate::devices::DeviceSync::new(Arc::clone(&library))?;
+    devices.configure_playback(settings.playback.clone())?;
+    state.device_playback.config = settings.playback.clone();
     devices.set_event_tx(event_tx.clone());
     let jam = crate::jam::JamManager::new(event_tx.clone());
     let similarity = crate::similarity::Manager::new(
@@ -321,7 +322,6 @@ pub async fn run(
         state.device_playback.self_device_name = device_name.clone();
         state.device_playback.active_device_id = Some(device_id);
         state.device_playback.active_device_name = Some(device_name);
-        state.device_playback.startup_takeover_pending = true;
     }
     let player_events = event_tx.clone();
     let mut runtime = Runtime {
@@ -416,6 +416,7 @@ pub async fn run(
             },
             Some(app_event) = event_rx.recv() => handle_app_event(&mut state, &mut runtime, app_event),
             _ = tick.tick() => {
+                reconcile_personal_playback(&mut state, &mut runtime);
                 state.advance_spinner();
                 expire_quit_confirmation(&mut state);
                 sync_player_shared(&mut state, &runtime);
@@ -626,6 +627,7 @@ fn publish_playback_snapshot_with_active(state: &mut AppState, runtime: &Runtime
         state.device_playback.active_device_name = Some(device_name.clone());
     }
     let snapshot = crate::devices::PlaybackSnapshot {
+        coordination: None,
         device_id,
         device_name,
         active,
@@ -639,6 +641,7 @@ fn publish_playback_snapshot_with_active(state: &mut AppState, runtime: &Runtime
         runtime
             .jam
             .publish_host_playback(crate::devices::PlaybackSnapshot {
+                coordination: None,
                 device_id: state.device_playback.self_device_id.clone(),
                 device_name: state.device_playback.self_device_name.clone(),
                 active: true,
@@ -658,36 +661,6 @@ fn update_local_idle_since(state: &mut AppState) {
     }
 }
 
-fn active_snapshot_idle_since(snapshot: &crate::devices::PlaybackSnapshot) -> Option<i64> {
-    if snapshot.state.playing && !snapshot.state.paused {
-        None
-    } else {
-        snapshot
-            .state
-            .idle_since_ms
-            .or(Some(snapshot.updated_at_ms))
-    }
-}
-
-fn active_idle_lease_expired(snapshot: &crate::devices::PlaybackSnapshot, now: i64) -> bool {
-    active_snapshot_idle_since(snapshot)
-        .is_some_and(|idle_since| now.saturating_sub(idle_since) >= ACTIVE_IDLE_LEASE_MS)
-}
-
-fn local_active_lease_protected(state: &mut AppState, now: i64) -> bool {
-    if !state.device_playback.is_audio_owner() || !state.player.playing {
-        return false;
-    }
-    if !state.player.paused {
-        return true;
-    }
-    update_local_idle_since(state);
-    state
-        .device_playback
-        .local_idle_since_ms
-        .is_some_and(|idle_since| now.saturating_sub(idle_since) < ACTIVE_IDLE_LEASE_MS)
-}
-
 fn extrapolate_control_position(state: &mut AppState) {
     let Some(snapshot) = state.device_playback.last_remote_snapshot.as_ref() else {
         return;
@@ -719,6 +692,7 @@ pub(crate) fn become_control_device(
     runtime: &Runtime,
     snapshot: crate::devices::PlaybackSnapshot,
 ) {
+    runtime.player.set_playback_allowed(false);
     if state.device_playback.is_audio_owner() {
         runtime.player.stop();
         publish_inactive_playback_snapshot(state, runtime);
@@ -737,6 +711,7 @@ pub(crate) fn become_control_device(
 }
 
 pub(crate) fn become_active_device(state: &mut AppState, runtime: &mut Runtime, start_audio: bool) {
+    runtime.player.set_playback_allowed(true);
     let was_control = state.device_playback.role == state::DevicePlaybackRole::Control;
     state.device_playback.role = state::DevicePlaybackRole::Active;
     state.device_playback.jam_host = false;
@@ -761,6 +736,13 @@ pub(crate) fn become_active_device(state: &mut AppState, runtime: &mut Runtime,
 }
 
 pub(crate) fn transfer_active_to_this_device(state: &mut AppState, runtime: &mut Runtime) {
+    if let Err(error) = runtime
+        .devices
+        .claim_playback(&state.device_playback.self_device_id)
+    {
+        state.status_message = Some(format!("device handoff: {error}"));
+        return;
+    }
     if state.device_playback.is_audio_owner() {
         publish_playback_snapshot(state, runtime);
         request_urgent_device_sync(runtime);
@@ -800,6 +782,10 @@ pub(crate) fn transfer_active_to_remote_device(
         transfer_active_to_this_device(state, runtime);
         return;
     }
+    if let Err(error) = runtime.devices.claim_playback(&target_device_id) {
+        state.status_message = Some(format!("device handoff: {error}"));
+        return;
+    }
     extrapolate_control_position(state);
     let previous_active_id = state.device_playback.active_device_id.clone();
     if state.player.current.is_none() && !state.player.queue.is_empty() {
@@ -824,6 +810,7 @@ pub(crate) fn transfer_active_to_remote_device(
         record_playback_command_async(runtime, previous, command.clone(), "device handoff");
     }
     let snapshot = crate::devices::PlaybackSnapshot {
+        coordination: None,
         device_id: target_device_id.clone(),
         device_name: target_device_name.clone(),
         active: true,
@@ -2877,6 +2864,7 @@ fn refresh_artists(state: &mut AppState, runtime: &Runtime) {
 
 fn save_app_settings(state: &AppState) {
     let settings = crate::config::settings::AppSettings {
+        playback: state.device_playback.config.clone(),
         volume: state.player.volume,
         library: state.global.filters,
         music_dir: state.music_dir.clone(),
@@ -2983,81 +2971,67 @@ fn apply_queue_refresh(
     }
 }
 
+fn reconcile_personal_playback(state: &mut AppState, runtime: &mut Runtime) {
+    if state.device_playback.role == state::DevicePlaybackRole::Jam {
+        runtime
+            .player
+            .set_playback_allowed(state.device_playback.is_audio_owner());
+        let _ = runtime.devices.playback_tick(false, false);
+        return;
+    }
+    let playing =
+        state.device_playback.is_audio_owner() && state.player.playing && !state.player.paused;
+    let owner = match runtime.devices.playback_tick(true, playing) {
+        Ok(owner) => owner,
+        Err(error) => {
+            runtime.player.set_playback_allowed(false);
+            runtime.player.stop();
+            state.status_message = Some(format!("playback coordination: {error}"));
+            return;
+        }
+    };
+    let Some(owner) = owner else {
+        runtime.player.set_playback_allowed(false);
+        runtime.player.stop();
+        runtime.player_start_pending = false;
+        state.device_playback.role = state::DevicePlaybackRole::Control;
+        state.device_playback.active_device_id = None;
+        return;
+    };
+    if owner == state.device_playback.self_device_id {
+        runtime.player.set_playback_allowed(true);
+        if !state.device_playback.is_audio_owner() {
+            become_active_device(state, runtime, true);
+        }
+    } else if let Some(snapshot) = state.device_playback.remote.get(&owner).cloned() {
+        if state.device_playback.last_remote_snapshot.as_ref() != Some(&snapshot)
+            || state.device_playback.is_audio_owner()
+        {
+            become_control_device(state, runtime, snapshot);
+        }
+    } else {
+        runtime.player.set_playback_allowed(false);
+        runtime.player.stop();
+        runtime.player_start_pending = false;
+        state.device_playback.role = state::DevicePlaybackRole::Control;
+        state.device_playback.active_device_id = Some(owner);
+    }
+    publish_playback_snapshot_with_active(state, runtime, state.device_playback.is_audio_owner());
+}
+
 fn handle_device_playback_snapshot(
     state: &mut AppState,
     runtime: &mut Runtime,
     snapshot: crate::devices::PlaybackSnapshot,
 ) {
-    // Personal-device reconciliation must never change Jam ownership. Jam
-    // has its own authority and lifecycle even when the same TUI also belongs
-    // to a trusted-device group.
-    if state.device_playback.role == state::DevicePlaybackRole::Jam {
-        return;
-    }
     if snapshot.device_id == state.device_playback.self_device_id {
         return;
     }
     state
         .device_playback
         .remote
-        .insert(snapshot.device_id.clone(), snapshot.clone());
-    let now = unix_time_ms();
-    state.device_playback.online_devices = state
-        .device_playback
-        .remote
-        .values()
-        .filter(|snapshot| {
-            now.saturating_sub(snapshot.updated_at_ms) <= state::DEVICE_ONLINE_TTL_MS
-        })
-        .count()
-        + 1;
-
-    if !snapshot.active {
-        return;
-    }
-    // Starting a player is an explicit claim of the active role. Import the
-    // current queue/position from the previously active peer, then announce a
-    // normal handoff so that the old owner becomes a control device. This is
-    // intentionally one-shot: subsequent snapshots use the regular lease and
-    // explicit-transfer rules.
-    if state.device_playback.startup_takeover_pending {
-        state.device_playback.startup_takeover_pending = false;
-        become_control_device(state, runtime, snapshot);
-        transfer_active_to_this_device(state, runtime);
-        state.status_message = Some("playback moved to this newly started player".to_string());
-        return;
-    }
-    if local_active_lease_protected(state, now) {
-        tracing::debug!(
-            remote = %snapshot.device_id,
-            "ignored remote active snapshot while local active playback is protected"
-        );
-        return;
-    }
-    let lease_expired = active_idle_lease_expired(&snapshot, now);
-    let already_controls_this_device = state.device_playback.is_personal_control()
-        && state.device_playback.active_device_id.as_deref() == Some(snapshot.device_id.as_str());
-    if !lease_expired || already_controls_this_device {
-        let was_active = state.device_playback.is_audio_owner();
-        let was_paused = state.player.playing && state.player.paused;
-        become_control_device(state, runtime, snapshot.clone());
-        if was_active && was_paused {
-            state.popup = Some(state::Popup::FedText {
-                title: "Active device moved".to_string(),
-                text: format!("Playback is now controlled by {}.", snapshot.device_name),
-            });
-        }
-        return;
-    }
-
-    if state.device_playback.is_personal_control() {
-        return;
-    }
-    become_active_device(state, runtime, false);
-    state.status_message = Some(format!(
-        "active playback moved here; {} was idle for 5m",
-        snapshot.device_name
-    ));
+        .insert(snapshot.device_id.clone(), snapshot);
+    reconcile_personal_playback(state, runtime);
 }
 
 fn handle_playback_command(
@@ -3112,10 +3086,19 @@ fn handle_playback_command(
             state: wire,
         } => {
             if active_device_id == state.device_playback.self_device_id {
+                handle_playback_command(
+                    state,
+                    runtime,
+                    crate::devices::PlaybackCommand::SetState {
+                        state: wire,
+                        seek: true,
+                    },
+                );
                 return;
             }
             let was_active = state.device_playback.is_audio_owner();
             let snapshot = crate::devices::PlaybackSnapshot {
+                coordination: None,
                 device_id: active_device_id,
                 device_name: active_device_name,
                 active: true,
@@ -3237,47 +3220,6 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent
                 })
                 .unwrap_or(1)
                 .max(1);
-            let active_revoked = state.device_playback.is_personal_control()
-                && state
-                    .device_playback
-                    .active_device_id
-                    .as_ref()
-                    .is_some_and(|active| {
-                        state
-                            .federation
-                            .devices
-                            .as_ref()
-                            .and_then(|status| {
-                                status
-                                    .devices
-                                    .iter()
-                                    .find(|device| device.device_id == *active)
-                            })
-                            .is_some_and(|device| device.revoked)
-                    });
-            let active_missing = state.device_playback.is_personal_control()
-                && state
-                    .device_playback
-                    .active_device_id
-                    .as_ref()
-                    .is_some_and(|active| {
-                        active != &state.device_playback.self_device_id
-                            && !state.federation.devices.as_ref().is_some_and(|status| {
-                                status
-                                    .devices
-                                    .iter()
-                                    .any(|device| device.device_id == *active)
-                            })
-                    });
-            if active_revoked || active_missing {
-                runtime.player.stop();
-                state.player.playing = false;
-                state.player.current = None;
-                state.player.paused = false;
-                state.player.position_secs = 0.0;
-                become_active_device(state, runtime, false);
-                state.status_message = Some("active playback moved to this device".into());
-            }
             clamp_settings_cursor(state);
         }
         AppEvent::FedSyncFinished(message) => {
@@ -3318,18 +3260,28 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent
         AppEvent::DevicePlayback(snapshot) => {
             handle_device_playback_snapshot(state, runtime, snapshot);
         }
-        AppEvent::PlaybackCommand(_)
+        AppEvent::PlaybackCommand { .. }
             if state.device_playback.role == state::DevicePlaybackRole::Jam =>
         {
             tracing::debug!("ignored personal-device playback command while Jam is active");
         }
-        AppEvent::PlaybackCommand(command) => {
-            handle_playback_command(state, runtime, command);
+        AppEvent::PlaybackCommand {
+            command,
+            authority,
+            origin,
+        } => {
+            if runtime
+                .devices
+                .playback_command_is_current(&origin, &authority)
+            {
+                handle_playback_command(state, runtime, command);
+            }
         }
         AppEvent::JamStatus(status) => {
             state.jam = status.clone();
             match status.role {
                 crate::jam::JamRole::Host => {
+                    runtime.player.set_playback_allowed(true);
                     state.device_playback.role = state::DevicePlaybackRole::Jam;
                     state.device_playback.jam_host = true;
                     publish_playback_snapshot(state, runtime);
@@ -3848,6 +3800,9 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent
         }
         AppEvent::Player(event) if state.device_playback.is_control() => {
             runtime.player_start_pending = false;
+            // A background decoder may finish after ownership was handed off.
+            // Ignoring Started alone would leave its obsolete audio playing.
+            runtime.player.stop();
             tracing::debug!(
                 ?event,
                 "ignored local player event while controlling remote playback"

src/app/state.rs

@@ -1541,6 +1541,7 @@ impl DevicePlaybackRole {
 
 #[derive(Debug, Default)]
 pub struct DevicePlaybackState {
+    pub config: music_dht::playback::Config,
     pub role: DevicePlaybackRole,
     pub self_device_id: String,
     pub self_device_name: String,
@@ -1551,9 +1552,6 @@ pub struct DevicePlaybackState {
     pub remote: BTreeMap<String, crate::devices::PlaybackSnapshot>,
     pub last_remote_snapshot: Option<crate::devices::PlaybackSnapshot>,
     pub jam_host: bool,
-    /// A freshly started TUI owns playback by protocol. The first active
-    /// snapshot discovered during startup is imported and handed off here.
-    pub startup_takeover_pending: bool,
 }
 
 impl DevicePlaybackState {
@@ -1562,10 +1560,6 @@ impl DevicePlaybackState {
             || (self.role == DevicePlaybackRole::Jam && !self.jam_host)
     }
 
-    pub fn is_personal_control(&self) -> bool {
-        self.role == DevicePlaybackRole::Control
-    }
-
     pub fn is_audio_owner(&self) -> bool {
         self.role == DevicePlaybackRole::Active
             || (self.role == DevicePlaybackRole::Jam && self.jam_host)

src/config/settings.rs

@@ -50,6 +50,8 @@ impl Default for SimilaritySettings {
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct AppSettings {
+    #[serde(default)]
+    pub playback: music_dht::playback::Config,
     #[serde(default = "default_volume")]
     pub volume: u8,
     #[serde(default)]
@@ -64,6 +66,7 @@ pub struct AppSettings {
 impl Default for AppSettings {
     fn default() -> Self {
         Self {
+            playback: music_dht::playback::Config::default(),
             volume: default_volume(),
             library: LibraryFilters::default(),
             music_dir: default_music_dir(),

src/devices.rs

@@ -5,6 +5,7 @@
 //! materialized tables, so offline clients can merge likes, playlists and
 //! membership changes deterministically.
 
+use music_dht::playback::{Checkpoint, CommandStamp, Config as PlaybackConfig, Engine};
 use std::collections::{BTreeMap, BTreeSet};
 use std::path::PathBuf;
 use std::str::FromStr;
@@ -24,7 +25,7 @@ use crate::library::models::{ArtistRef, TrackItem};
 
 pub const SYNC_ALPN: &[u8] = b"furumi/sync/2";
 const CLIENT_VERSION: &str = env!("CARGO_PKG_VERSION");
-pub const PROTOCOL_VERSION: u16 = 2;
+pub const PROTOCOL_VERSION: u16 = 3;
 const INVITE_TTL_MS: i64 = 10 * 60 * 1000;
 const PAIRING_WAIT_MS: i64 = 5 * 60 * 1000;
 const PAIRING_RETRY_DELAY: Duration = Duration::from_secs(1);
@@ -81,6 +82,10 @@ pub struct DeviceSyncStatus {
 
 #[derive(Clone)]
 pub struct DeviceSync {
+    // A device identity has exactly one coordinator, including across binaries
+    // launched from different installation directories. The OS releases this
+    // lock on crashes; the file itself is deliberately never removed.
+    _identity_lock: Option<Arc<std::fs::File>>,
     conn: Arc<std::sync::Mutex<Connection>>,
     library: Arc<Library>,
     event_tx: Arc<std::sync::Mutex<Option<tokio::sync::mpsc::UnboundedSender<AppEvent>>>>,
@@ -245,32 +250,13 @@ pub struct PlaybackStateWire {
     pub repeat: PlaybackRepeat,
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-pub struct PlaybackSnapshot {
-    pub device_id: String,
-    pub device_name: String,
-    pub active: bool,
-    pub updated_at_ms: i64,
-    pub state: PlaybackStateWire,
-}
-
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-#[serde(tag = "kind", rename_all = "snake_case")]
-pub enum PlaybackCommand {
-    SetState {
-        state: PlaybackStateWire,
-        #[serde(default)]
-        seek: bool,
-    },
-    ActiveChanged {
-        active_device_id: String,
-        active_device_name: String,
-        state: PlaybackStateWire,
-    },
-}
+pub type PlaybackSnapshot = music_dht::playback::Snapshot<PlaybackStateWire>;
+pub type PlaybackCommand = music_dht::playback::Command<PlaybackStateWire>;
 
 #[derive(Debug, Clone, Default)]
 struct PlaybackShared {
+    engine: Option<Engine>,
+    engine_group: String,
     local: Option<PlaybackSnapshot>,
     remote: BTreeMap<String, PlaybackSnapshot>,
 }
@@ -369,6 +355,8 @@ pub enum SyncOpPayload {
     PlaybackCommand {
         target_device_id: String,
         command: PlaybackCommand,
+        #[serde(default)]
+        authority: Option<CommandStamp>,
     },
     ListenRecorded {
         event: ListenEvent,
@@ -615,16 +603,33 @@ fn default_db_path() -> PathBuf {
         .unwrap_or_else(|| PathBuf::from("devices").join("sync.sqlite3"))
 }
 
+fn acquire_identity_lock(database: &std::path::Path) -> Result<std::fs::File> {
+    let path = database.with_extension("lock");
+    let file = std::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(&path)
+        .with_context(|| format!("opening device identity lock {}", path.display()))?;
+    file.try_lock().with_context(|| format!(
+        "Cannot acquire device identity lock {}. Another Furumi instance may already be using this device. Close it before launching another binary; two players must not share one device identity.", path.display()
+    ))?;
+    Ok(file)
+}
+
 impl DeviceSync {
     pub fn new(library: Arc<Library>) -> Result<Arc<Self>> {
         let path = default_db_path();
         if let Some(parent) = path.parent() {
             std::fs::create_dir_all(parent)?;
         }
+        let identity_lock = acquire_identity_lock(&path)?;
         let conn =
             Connection::open(&path).with_context(|| format!("opening {}", path.display()))?;
         init_schema(&conn)?;
         let sync = Arc::new(Self {
+            _identity_lock: Some(Arc::new(identity_lock)),
             conn: Arc::new(std::sync::Mutex::new(conn)),
             library,
             event_tx: Arc::new(std::sync::Mutex::new(None)),
@@ -648,10 +653,92 @@ impl DeviceSync {
         format!("{}-{}", now_ms(), random_hex(12))
     }
 
+    pub fn configure_playback(&self, config: PlaybackConfig) -> Result<()> {
+        set_meta(
+            &lock(&self.conn),
+            "playback_config_v1",
+            &serde_json::to_string(&config)?,
+        )?;
+        lock(&self.playback).engine = None;
+        Ok(())
+    }
+
+    fn with_playback_engine<R>(&self, f: impl FnOnce(&mut Engine) -> R) -> Result<R> {
+        let identity = self.ensure_identity()?;
+        let mut shared = lock(&self.playback);
+        if shared.engine.is_none() || shared.engine_group != identity.group_id {
+            shared.engine_group = identity.group_id.clone();
+            let conn = lock(&self.conn);
+            let durable = get_meta(&conn, "playback_coordination_v1")?
+                .map(|json| serde_json::from_str::<Checkpoint>(&json))
+                .transpose()?
+                .filter(|checkpoint| checkpoint.scope == identity.group_id)
+                .map(|checkpoint| checkpoint.state)
+                .unwrap_or_default();
+            let config = get_meta(&conn, "playback_config_v1")?
+                .map(|json| serde_json::from_str::<PlaybackConfig>(&json))
+                .transpose()?
+                .unwrap_or_default();
+            shared.engine = Some(Engine::new(
+                identity.device_id,
+                config,
+                durable,
+                playback_clock(),
+            ));
+        }
+        let engine = shared.engine.as_mut().expect("initialized playback engine");
+        let previous = engine.clone();
+        let result = f(engine);
+        if previous.durable() != engine.durable() {
+            let saved = set_meta(
+                &lock(&self.conn),
+                "playback_coordination_v1",
+                &serde_json::to_string(&Checkpoint {
+                    scope: identity.group_id,
+                    state: engine.durable().clone(),
+                })?,
+            );
+            if let Err(error) = saved {
+                *engine = previous;
+                return Err(error);
+            }
+        }
+        Ok(result)
+    }
+
+    /// Evaluate shared ownership independently of the UI's online-device list.
+    pub fn playback_tick(&self, available: bool, playing: bool) -> Result<Option<String>> {
+        self.with_playback_engine(|engine| {
+            engine.set_output(available, playing);
+            engine.heartbeat(playback_clock());
+            engine.tick(playback_clock());
+            engine.owner().map(str::to_string)
+        })
+    }
+
+    pub fn playback_command_is_current(&self, origin: &str, stamp: &CommandStamp) -> bool {
+        self.with_playback_engine(|engine| engine.command_is_current(origin, stamp))
+            .unwrap_or(false)
+    }
+
+    pub fn claim_playback(&self, target: &str) -> Result<()> {
+        self.with_playback_engine(|engine| engine.transfer(target, playback_clock()))?
+            .then_some(())
+            .context("cannot allocate playback ownership term")
+    }
+
     pub fn publish_playback(&self, mut snapshot: PlaybackSnapshot) {
         if snapshot.updated_at_ms <= 0 {
             snapshot.updated_at_ms = now_ms();
         }
+        let Ok(coordination) = self.with_playback_engine(|engine| engine.announcement()) else {
+            return;
+        };
+        snapshot.active = coordination
+            .claim
+            .as_ref()
+            .is_some_and(|claim| claim.owner == snapshot.device_id);
+        snapshot.coordination = Some(coordination);
         lock(&self.playback).local = Some(snapshot);
     }
 
@@ -663,9 +750,22 @@ impl DeviceSync {
         if target_device_id.trim().is_empty() {
             return Ok(());
         }
+        let authority = self
+            .with_playback_engine(|engine| {
+                if let PlaybackCommand::ActiveChanged {
+                    active_device_id, ..
+                } = &command
+                    && engine.owner() != Some(active_device_id.as_str())
+                {
+                    engine.transfer(active_device_id, playback_clock());
+                }
+                engine.stamp()
+            })?
+            .context("no playback owner; wait for discovery or select an output")?;
         self.record_local_op(SyncOpPayload::PlaybackCommand {
             target_device_id: target_device_id.to_string(),
             command,
+            authority: Some(authority),
         })
     }
 
@@ -845,7 +945,7 @@ impl DeviceSync {
                 if let Some(profile) = profile {
                     self.apply_device_profile(&profile, true)?;
                     if let Some(playback) = playback {
-                        self.apply_playback_snapshot(playback)?;
+                        self.apply_playback_snapshot(&profile.device_id, playback)?;
                     }
                 }
                 self.apply_device_profiles(&devices)?;
@@ -1305,7 +1405,7 @@ impl DeviceSync {
             } => {
                 self.apply_device_profiles(&devices)?;
                 if let Some(playback) = playback {
-                    self.apply_playback_snapshot(playback)?;
+                    self.apply_playback_snapshot(&device.device_id, playback)?;
                 }
                 self.apply_snapshot(snapshot)?;
                 self.apply_ops(ops)?;
@@ -1636,8 +1736,15 @@ impl DeviceSync {
             SyncOpPayload::PlaybackCommand {
                 target_device_id,
                 command,
+                authority,
             } => {
-                self.apply_playback_command(target_device_id, command, &op.op_id)?;
+                self.apply_playback_command(
+                    target_device_id,
+                    command,
+                    authority.as_ref(),
+                    &op.origin_device_id,
+                    &op.op_id,
+                )?;
                 false
             }
             SyncOpPayload::ListenRecorded { event } => self
@@ -1651,10 +1758,32 @@ impl DeviceSync {
         &self,
         target_device_id: &str,
         command: &PlaybackCommand,
+        authority: Option<&CommandStamp>,
+        origin: &str,
         op_id: &str,
     ) -> Result<()> {
         let identity = self.ensure_identity()?;
-        if target_device_id != identity.device_id {
+        // Legacy commands still replicate with the library log, but cannot
+        // override the versioned personal-playback protocol.
+        let Some(authority) = authority else {
+            return Ok(());
+        };
+        let handoff = matches!(command, PlaybackCommand::ActiveChanged { .. });
+        if !handoff && target_device_id != identity.device_id {
+            return Ok(());
+        }
+        if let PlaybackCommand::ActiveChanged {
+            active_device_id, ..
+        } = command
+            && active_device_id != &authority.claim.owner
+        {
+            return Ok(());
+        }
+        let accepted = self.with_playback_engine(|engine| {
+            engine.accept_command(origin, authority, handoff, playback_clock())
+                && (handoff || engine.is_owner())
+        })?;
+        if !accepted || target_device_id != identity.device_id {
             return Ok(());
         }
         let inserted = {
@@ -1669,7 +1798,11 @@ impl DeviceSync {
             return Ok(());
         }
         if let Some(tx) = lock(&self.event_tx).as_ref() {
-            let _ = tx.send(AppEvent::PlaybackCommand(command.clone()));
+            let _ = tx.send(AppEvent::PlaybackCommand {
+                command: command.clone(),
+                authority: authority.clone(),
+                origin: origin.to_string(),
+            });
         }
         Ok(())
     }
@@ -2595,19 +2728,29 @@ impl DeviceSync {
         lock(&self.playback).local.clone()
     }
 
-    fn apply_playback_snapshot(&self, snapshot: PlaybackSnapshot) -> Result<()> {
+    fn apply_playback_snapshot(&self, sender: &str, snapshot: PlaybackSnapshot) -> Result<()> {
         let identity = self.ensure_identity()?;
-        if snapshot.device_id == identity.device_id {
+        if snapshot.device_id != sender || snapshot.device_id == identity.device_id {
+            return Ok(());
+        }
+        let Some(coordination) = &snapshot.coordination else {
+            return Ok(());
+        };
+        if !self.with_playback_engine(|engine| {
+            engine.observe(&snapshot.device_id, coordination, playback_clock())
+        })? {
             return Ok(());
         }
         let should_send = {
-            let mut playback = lock(&self.playback);
-            let changed = playback
-                .remote
-                .get(&snapshot.device_id)
-                .is_none_or(|current| snapshot.updated_at_ms > current.updated_at_ms);
+            let mut shared = lock(&self.playback);
+            let changed = shared.remote.get(&snapshot.device_id).is_none_or(|old| {
+                old.coordination.as_ref().is_none_or(|c| {
+                    coordination.claim > c.claim
+                        || (coordination.claim == c.claim && coordination.heartbeat > c.heartbeat)
+                })
+            });
             if changed {
-                playback
+                shared
                     .remote
                     .insert(snapshot.device_id.clone(), snapshot.clone());
             }
@@ -2707,13 +2850,45 @@ pub async fn sync_loop(
     transport_stats: Arc<crate::federation::TransportStats>,
 ) {
     let mut interval = tokio::time::interval(SYNC_INTERVAL);
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    // Independent polls: an offline device must not hold up live outputs.
+    // JoinSet aborts outstanding IO when federation stops or restarts.
+    let mut polls = tokio::task::JoinSet::new();
+    let mut active = std::collections::HashMap::new();
     loop {
-        interval.tick().await;
-        if let Err(err) = sync
-            .sync_once(Arc::clone(&service), Arc::clone(&transport_stats))
-            .await
-        {
-            tracing::debug!("personal sync tick failed: {err:#}");
+        tokio::select! {
+            _ = interval.tick() => {
+                match sync.active_remote_devices() {
+                    Ok(devices) => for device in devices {
+                        if device.endpoint_ticket.trim().is_empty()
+                            || active.values().any(|id| id == &device.device_id) { continue; }
+                        let device_id = device.device_id.clone();
+                        let sync = Arc::clone(&sync);
+                        let service = Arc::clone(&service);
+                        let stats = Arc::clone(&transport_stats);
+                        let handle = polls.spawn(async move {
+                            tokio::time::timeout(Duration::from_secs(30), sync.sync_device(service, &device, stats))
+                                .await.context("device sync exchange timed out")?
+                        });
+                        active.insert(handle.id(), device_id);
+                    },
+                    Err(err) => { let _ = sync.set_last_error(Some(format!("{err:#}"))); }
+                }
+                if let Err(err) = sync.gc_tombstones() {
+                    tracing::debug!("personal sync cleanup failed: {err:#}");
+                }
+            }
+            Some(completed) = polls.join_next_with_id(), if !polls.is_empty() => {
+                let (task, result) = match completed {
+                    Ok((task, result)) => (task, result),
+                    Err(error) => (error.id(), Err(anyhow::Error::from(error))),
+                };
+                if let Some(device_id) = active.remove(&task)
+                    && let Err(err) = result {
+                        tracing::debug!(device = %device_id, "device sync failed: {err:#}");
+                        let _ = sync.set_last_error(Some(format!("{}: {err:#}", short_id(&device_id))));
+                    }
+            }
         }
         if let Some(tx) = lock(&sync.event_tx).as_ref() {
             let _ = tx.send(AppEvent::DeviceSyncStatus(sync.status()));
@@ -2936,7 +3111,7 @@ async fn handle_pair_request(
     }
     sync.apply_device_profile(&profile, true)?;
     if let Some(playback) = playback {
-        sync.apply_playback_snapshot(playback)?;
+        sync.apply_playback_snapshot(&profile.device_id, playback)?;
     }
     sync.apply_snapshot(snapshot)?;
     sync.apply_ops(ops)?;
@@ -3026,7 +3201,7 @@ async fn handle_hello(
     sync.apply_device_profile(&profile, false)?;
     sync.apply_device_profiles(&devices)?;
     if let Some(playback) = playback {
-        sync.apply_playback_snapshot(playback)?;
+        sync.apply_playback_snapshot(&profile.device_id, playback)?;
     }
     sync.apply_snapshot(snapshot)?;
     sync.apply_ops(ops)?;
@@ -3635,3 +3810,14 @@ fn base64url_decode(value: &str) -> Result<Vec<u8>> {
 #[cfg(test)]
 #[path = "devices/tests.rs"]
 mod tests;
+
+#[cfg(test)]
+mod interop_tests;
+
+fn playback_clock() -> u64 {
+    static START: std::sync::OnceLock<std::time::Instant> = std::sync::OnceLock::new();
+    START
+        .get_or_init(std::time::Instant::now)
+        .elapsed()
+        .as_millis() as u64
+}

src/devices/interop_tests.rs

@@ -0,0 +1,133 @@
+//! Cross-binary wire test. Run with scripts/test_device_interop.py; the other
+//! endpoint is compiled from furumusic's real protocol types and player hub.
+use super::*;
+use tokio::io::AsyncWriteExt;
+
+#[tokio::test]
+#[ignore = "run scripts/test_device_interop.py to start both player test binaries"]
+async fn localhost_web_peer() {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        let dir = PathBuf::from(std::env::var("FURUMI_INTEROP_DIR").expect("interop runner"));
+        let address = loop {
+            if let Ok(address) = std::fs::read_to_string(dir.join("web-address")) {
+                break address;
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        };
+        let mut stream = tokio::net::TcpStream::connect(address.trim())
+            .await
+            .unwrap();
+        let conn = Connection::open_in_memory().unwrap();
+        init_schema(&conn).unwrap();
+        let sync = DeviceSync {
+            _identity_lock: None,
+            conn: Arc::new(std::sync::Mutex::new(conn)),
+            library: Arc::new(Library::open(&dir.join("interop-library.sqlite3")).unwrap()),
+            event_tx: Default::default(),
+            playback: Default::default(),
+        };
+        let id = sync.ensure_identity().unwrap();
+        let (tx, mut events) = tokio::sync::mpsc::unbounded_channel();
+        sync.set_event_tx(tx);
+        sync.claim_playback(&id.device_id).unwrap();
+        sync.playback_tick(true, true).unwrap();
+        let mut state: PlaybackStateWire = serde_json::from_value(serde_json::json!({
+            "queue": [{ "id": 7, "title": "Interop track", "duration_seconds": 123.5,
+                "release_id": 1, "release_title": "Interop album", "artist_names": ["Interop artist"],
+                "content_id": "b3:0000000000000000000000000000000000000000000000000000000000000000" }],
+            "queue_pos": 0, "playing": true, "paused": false,
+            "position_secs": 42.5, "volume": 73, "shuffle": true, "repeat": "all"
+        }))
+        .unwrap();
+        for phase in 0..3 {
+            sync.publish_playback(PlaybackSnapshot {
+                device_id: id.device_id.clone(),
+                device_name: "TUI interop".into(),
+                active: true,
+                updated_at_ms: now_ms(),
+                state: state.clone(),
+                coordination: None,
+            });
+            let hello = WireMessage::Hello {
+                group_id: id.group_id.clone(),
+                profile: sync.own_profile("").unwrap(),
+                devices: vec![],
+                vector: BTreeMap::new(),
+                ops: vec![],
+                snapshot: SyncSnapshot::default(),
+                playback: sync.local_playback_snapshot(),
+            };
+            let mut bytes = serde_json::to_vec(&hello).unwrap();
+            bytes.push(b'\n');
+            stream.write_all(&bytes).await.unwrap();
+            let response: WireMessage =
+                serde_json::from_slice(&read_line(&mut stream).await.unwrap()).unwrap();
+            let WireMessage::SyncResponse {
+                accepted: true,
+                playback: Some(snapshot),
+                ops,
+                ..
+            } = response
+            else {
+                panic!("expected web response")
+            };
+            let web_id = snapshot.device_id.clone();
+            sync.apply_playback_snapshot(&web_id, snapshot).unwrap();
+            assert_eq!(ops.len(), 1);
+            let op = &ops[0];
+            // Production command adapter including durable fencing/deduplication.
+            sync.apply_op(op).unwrap();
+            sync.apply_op(op).unwrap();
+            let mut commands = Vec::new();
+            while let Ok(event) = events.try_recv() {
+                if let AppEvent::PlaybackCommand {
+                    command,
+                    authority,
+                    origin,
+                } = event
+                {
+                    assert!(sync.playback_command_is_current(&origin, &authority));
+                    commands.push(command);
+                }
+            }
+            assert_eq!(commands.len(), 1, "one event even after duplicate delivery");
+            match commands.pop().unwrap() {
+                PlaybackCommand::ActiveChanged {
+                    active_device_id,
+                    state: next,
+                    ..
+                } => {
+                    assert_eq!(
+                        active_device_id,
+                        if phase == 0 {
+                            web_id.clone()
+                        } else {
+                            id.device_id.clone()
+                        }
+                    );
+                    assert_eq!(next.position_secs, 42.5);
+                    assert_eq!(next.queue, state.queue);
+                    state = next;
+                }
+                PlaybackCommand::SetState { state: next, seek } => {
+                    assert_eq!(phase, 2);
+                    assert!(seek);
+                    assert!(next.paused);
+                    assert_eq!(next.position_secs, 87.0);
+                    state = next;
+                }
+            }
+            assert_eq!(
+                sync.playback_tick(true, false).unwrap(),
+                Some(if phase == 0 {
+                    web_id
+                } else {
+                    id.device_id.clone()
+                })
+            );
+        }
+        stream.write_all(b"ok\n").await.unwrap();
+    })
+    .await
+    .expect("web/TUI exchange timed out");
+}

src/devices/tests.rs

@@ -1,5 +1,329 @@
 use super::*;
 
+/// Exercises the production stream handlers and SQLite adapter, not just the
+/// ownership reducer. Each peer has a fresh identity and an isolated library.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn localhost_devices_exchange_state_and_handoff() {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        let dirs = [tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap()];
+        let network = NetworkId::from_name(&format!("device-test-{}", random_hex(16)));
+        let mut peers = Vec::new();
+        let mut servers = Vec::new();
+        let mut receivers = Vec::new();
+        for dir in &dirs {
+            std::fs::create_dir_all(dir.path().join("network")).unwrap();
+            let config = music_dht::MusicDhtConfig::builder()
+                .data_dir(dir.path().join("network"))
+                .network_id(network)
+                .stream_protocol(SYNC_ALPN)
+                .build()
+                .unwrap();
+            let (service, events) = MusicDhtService::start(config).await.unwrap();
+            let service = Arc::new(service);
+            let conn = Connection::open_in_memory().unwrap();
+            init_schema(&conn).unwrap();
+            let sync = Arc::new(DeviceSync {
+                _identity_lock: Some(Arc::new(
+                    acquire_identity_lock(&dir.path().join("sync.sqlite3")).unwrap(),
+                )),
+                conn: Arc::new(std::sync::Mutex::new(conn)),
+                library: Arc::new(Library::open(&dir.path().join("library.sqlite3")).unwrap()),
+                event_tx: Default::default(),
+                playback: Default::default(),
+            });
+            let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+            sync.set_event_tx(tx);
+            let stats = Arc::new(crate::federation::TransportStats::default());
+            servers.push(tokio::spawn(serve_peers(
+                service.stream_acceptor(SYNC_ALPN).unwrap(),
+                sync.clone(),
+                service.clone(),
+                stats.clone(),
+            )));
+            receivers.push(rx);
+            peers.push((sync, service, stats, events));
+        }
+        let (a, service_a, stats_a, _) = &peers[0];
+        let (b, service_b, stats_b, _) = &peers[1];
+        b.ensure_identity().unwrap();
+        b.set_group_id(&a.ensure_identity().unwrap().group_id)
+            .unwrap();
+        let profile_a = a
+            .own_profile(&service_a.ticket().await.unwrap().to_string())
+            .unwrap();
+        let profile_b = b
+            .own_profile(&service_b.ticket().await.unwrap().to_string())
+            .unwrap();
+        a.apply_device_profile(&profile_b, true).unwrap();
+        b.apply_device_profile(&profile_a, true).unwrap();
+        a.claim_playback(&profile_a.device_id).unwrap();
+        a.playback_tick(true, true).unwrap();
+        let mut state = empty_playback_state();
+        state.playing = true;
+        state.position_secs = 42.5;
+        a.publish_playback(PlaybackSnapshot {
+            device_id: profile_a.device_id.clone(),
+            device_name: "TUI".into(),
+            active: true,
+            updated_at_ms: now_ms(),
+            state: state.clone(),
+            coordination: None,
+        });
+        a.sync_device(
+            service_a.clone(),
+            &a.active_remote_devices().unwrap()[0],
+            stats_a.clone(),
+        )
+        .await
+        .unwrap();
+        assert_eq!(
+            b.with_playback_engine(|e| e.owner().map(str::to_owned))
+                .unwrap(),
+            Some(profile_a.device_id.clone())
+        );
+        assert_eq!(lock(&b.playback).remote[&profile_a.device_id].state, state);
+        assert!(
+            b.status()
+                .devices
+                .iter()
+                .any(|d| d.device_id == profile_a.device_id && d.last_seen_ms.is_some())
+        );
+
+        a.record_playback_command(
+            &profile_b.device_id,
+            PlaybackCommand::ActiveChanged {
+                active_device_id: profile_b.device_id.clone(),
+                active_device_name: "Second player".into(),
+                state: state.clone(),
+            },
+        )
+        .unwrap();
+        a.sync_device(
+            service_a.clone(),
+            &a.active_remote_devices().unwrap()[0],
+            stats_a.clone(),
+        )
+        .await
+        .unwrap();
+        let mut transferred = false;
+        while let Ok(event) = receivers[1].try_recv() {
+            if let AppEvent::PlaybackCommand {
+                command:
+                    PlaybackCommand::ActiveChanged {
+                        state: received, ..
+                    },
+                authority,
+                origin,
+            } = event
+            {
+                assert_eq!(received, state);
+                assert!(b.playback_command_is_current(&origin, &authority));
+                transferred = true;
+            }
+        }
+        assert!(transferred, "handoff must reach the player's event loop");
+        assert_eq!(
+            b.playback_tick(true, true).unwrap(),
+            Some(profile_b.device_id.clone())
+        );
+        b.publish_playback(PlaybackSnapshot {
+            device_id: profile_b.device_id.clone(),
+            device_name: "Second player".into(),
+            active: true,
+            updated_at_ms: now_ms(),
+            state,
+            coordination: None,
+        });
+        b.sync_device(
+            service_b.clone(),
+            &b.active_remote_devices().unwrap()[0],
+            stats_b.clone(),
+        )
+        .await
+        .unwrap();
+        assert_eq!(
+            a.playback_tick(true, false).unwrap(),
+            Some(profile_b.device_id)
+        );
+        // A paired peer that accepts a connection but never answers must not
+        // serialize or stop subsequent polls to the responsive peer.
+        let silent_dir = tempfile::tempdir().unwrap();
+        let (silent, _events) = MusicDhtService::start(
+            music_dht::MusicDhtConfig::builder()
+                .data_dir(silent_dir.path())
+                .network_id(network)
+                .stream_protocol(SYNC_ALPN)
+                .build()
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+        let _silent_acceptor = silent.stream_acceptor(SYNC_ALPN).unwrap();
+        let mut silent_profile = profile_a.clone();
+        silent_profile.device_id = "silent-peer".into();
+        silent_profile.endpoint_id = silent.endpoint_id().to_string();
+        silent_profile.endpoint_ticket = silent.ticket().await.unwrap().to_string();
+        a.apply_device_profile(&silent_profile, true).unwrap();
+        lock(&a.conn)
+            .execute(
+                "UPDATE sync_devices SET last_seen_ms = ?1 WHERE device_id = 'silent-peer'",
+                [now_ms() + 1_000],
+            )
+            .unwrap();
+        assert_eq!(
+            a.active_remote_devices().unwrap()[0].device_id,
+            "silent-peer"
+        );
+        a.claim_playback(&profile_a.device_id).unwrap();
+        let poller = tokio::spawn(sync_loop(a.clone(), service_a.clone(), stats_a.clone()));
+        for position in [99.0, 100.0] {
+            a.playback_tick(true, true).unwrap();
+            let mut next = empty_playback_state();
+            next.playing = true;
+            next.position_secs = position;
+            a.publish_playback(PlaybackSnapshot {
+                device_id: profile_a.device_id.clone(),
+                device_name: "TUI".into(),
+                active: true,
+                updated_at_ms: now_ms(),
+                state: next,
+                coordination: None,
+            });
+            tokio::time::timeout(Duration::from_secs(6), async {
+                loop {
+                    if lock(&b.playback)
+                        .remote
+                        .get(&profile_a.device_id)
+                        .is_some_and(|snapshot| snapshot.state.position_secs == position)
+                    {
+                        break;
+                    }
+                    tokio::time::sleep(Duration::from_millis(20)).await;
+                }
+            })
+            .await
+            .expect("a stalled peer must not delay live playback polls");
+        }
+        poller.abort();
+        let _ = poller.await;
+        silent.shutdown().await.unwrap();
+        for server in servers {
+            server.abort();
+        }
+        for (_, service, _, _) in &peers {
+            service.shutdown().await.unwrap();
+        }
+    })
+    .await
+    .expect("localhost sync must complete within 30 seconds");
+}
+
+fn empty_playback_state() -> PlaybackStateWire {
+    PlaybackStateWire {
+        queue: vec![],
+        queue_pos: 0,
+        playing: false,
+        paused: false,
+        idle_since_ms: None,
+        position_secs: 0.0,
+        volume: 80,
+        shuffle: false,
+        repeat: PlaybackRepeat::Off,
+    }
+}
+
+#[test]
+fn coordination_ignores_legacy_commands_and_wrong_snapshot_sender() {
+    let sync = test_sync();
+    let identity = sync.ensure_identity().unwrap();
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+    sync.set_event_tx(tx);
+    let command = PlaybackCommand::SetState {
+        state: empty_playback_state(),
+        seek: false,
+    };
+    sync.apply_playback_command(&identity.device_id, &command, None, "remote", "legacy")
+        .unwrap();
+    assert!(rx.try_recv().is_err());
+    let mut remote = Engine::new(
+        "remote".into(),
+        PlaybackConfig::default(),
+        Default::default(),
+        0,
+    );
+    remote.transfer("remote", 0);
+    remote.set_output(true, true);
+    remote.heartbeat(1);
+    let snapshot = PlaybackSnapshot {
+        device_id: "remote".into(),
+        device_name: "Remote".into(),
+        active: true,
+        updated_at_ms: now_ms(),
+        state: empty_playback_state(),
+        coordination: Some(remote.announcement()),
+    };
+    sync.apply_playback_snapshot("different-sender", snapshot.clone())
+        .unwrap();
+    assert!(
+        sync.with_playback_engine(|engine| engine.owner().is_none())
+            .unwrap()
+    );
+    sync.apply_playback_snapshot("remote", snapshot).unwrap();
+    assert_eq!(
+        sync.playback_tick(true, false).unwrap().as_deref(),
+        Some("remote")
+    );
+    sync.publish_playback(PlaybackSnapshot {
+        device_id: identity.device_id,
+        device_name: identity.name,
+        active: false,
+        updated_at_ms: now_ms(),
+        state: empty_playback_state(),
+        coordination: None,
+    });
+    let gossip = sync
+        .local_playback_snapshot()
+        .unwrap()
+        .coordination
+        .unwrap();
+    assert_eq!(gossip.claim.unwrap().owner, "remote");
+}
+
+#[test]
+fn checkpoint_is_not_reused_after_changing_trusted_group() {
+    let sync = test_sync();
+    sync.claim_playback("old-group-owner").unwrap();
+    sync.set_group_id("new-test-group").unwrap();
+    assert!(
+        sync.with_playback_engine(|engine| engine.owner().is_none())
+            .unwrap()
+    );
+}
+
+#[test]
+fn queued_command_loses_its_fence_when_another_owner_wins() {
+    let sync = test_sync();
+    let identity = sync.ensure_identity().unwrap();
+    sync.claim_playback(&identity.device_id).unwrap();
+    let stamp = sync
+        .with_playback_engine(|engine| engine.stamp().unwrap())
+        .unwrap();
+    sync.apply_playback_command(
+        &identity.device_id,
+        &PlaybackCommand::SetState {
+            state: empty_playback_state(),
+            seek: false,
+        },
+        Some(&stamp),
+        &identity.device_id,
+        "queued",
+    )
+    .unwrap();
+    assert!(sync.playback_command_is_current(&identity.device_id, &stamp));
+    sync.claim_playback("new-owner").unwrap();
+    assert!(!sync.playback_command_is_current(&identity.device_id, &stamp));
+}
+
 static NEXT_TEST_DB: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
 
 fn test_sync() -> DeviceSync {
@@ -13,6 +337,7 @@ fn test_sync() -> DeviceSync {
         unique
     ));
     let sync = DeviceSync {
+        _identity_lock: None,
         conn: Arc::new(std::sync::Mutex::new(conn)),
         library: Arc::new(Library::open(&library_path).unwrap()),
         event_tx: Arc::new(std::sync::Mutex::new(None)),
@@ -22,6 +347,38 @@ fn test_sync() -> DeviceSync {
     sync
 }
 
+#[test]
+fn device_identity_has_one_coordinator_across_installation_paths() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("sync.sqlite3");
+    let first = acquire_identity_lock(&path).unwrap();
+    assert!(acquire_identity_lock(&path).is_err());
+    let child = std::process::Command::new(std::env::current_exe().unwrap())
+        .args([
+            "--exact",
+            "devices::tests::identity_lock_child_process",
+            "--nocapture",
+        ])
+        .env("FURUMI_IDENTITY_LOCK_TEST_PATH", &path)
+        .output()
+        .unwrap();
+    assert!(
+        child.status.success(),
+        "{}",
+        String::from_utf8_lossy(&child.stderr)
+    );
+    drop(first);
+    // Reopening a leftover lock file after shutdown must succeed.
+    assert!(acquire_identity_lock(&path).is_ok());
+}
+
+#[test]
+fn identity_lock_child_process() {
+    if let Some(path) = std::env::var_os("FURUMI_IDENTITY_LOCK_TEST_PATH") {
+        assert!(acquire_identity_lock(std::path::Path::new(&path)).is_err());
+    }
+}
+
 fn device_revoked(sync: &DeviceSync, device_id: &str) -> bool {
     let conn = lock(&sync.conn);
     conn.query_row(
@@ -212,19 +569,41 @@ fn playback_command_is_targeted_and_deduplicated() {
         seek: false,
     };
 
-    sync.apply_playback_command("dev_other", &command, "op_other")
+    sync.claim_playback(&identity.device_id).unwrap();
+    let stamp = sync
+        .with_playback_engine(|engine| engine.stamp().unwrap())
         .unwrap();
+    sync.apply_playback_command(
+        "dev_other",
+        &command,
+        Some(&stamp),
+        &identity.device_id,
+        "op_other",
+    )
+    .unwrap();
     assert!(rx.try_recv().is_err());
 
-    sync.apply_playback_command(&identity.device_id, &command, "op_1")
-        .unwrap();
+    sync.apply_playback_command(
+        &identity.device_id,
+        &command,
+        Some(&stamp),
+        &identity.device_id,
+        "op_1",
+    )
+    .unwrap();
     assert!(matches!(
         rx.try_recv().unwrap(),
-        crate::app::event::AppEvent::PlaybackCommand(_)
+        crate::app::event::AppEvent::PlaybackCommand { .. }
     ));
 
-    sync.apply_playback_command(&identity.device_id, &command, "op_1")
-        .unwrap();
+    sync.apply_playback_command(
+        &identity.device_id,
+        &command,
+        Some(&stamp),
+        &identity.device_id,
+        "op_1",
+    )
+    .unwrap();
     assert!(rx.try_recv().is_err());
 }
 
@@ -245,6 +624,7 @@ fn playback_commands_are_caught_up_only_by_their_target_while_fresh() {
         },
         seek: false,
     };
+    sync.claim_playback("dev_target").unwrap();
     sync.record_playback_command("dev_target", command.clone())
         .unwrap();
     sync.record_playback_command("dev_other", command).unwrap();

src/jam.rs

@@ -668,6 +668,7 @@ mod tests {
             repeat: crate::devices::PlaybackRepeat::Off,
         };
         host.publish_host_playback(PlaybackSnapshot {
+            coordination: None,
             device_id: "dev_host".into(),
             device_name: "Host".into(),
             active: true,

src/main.rs

@@ -61,7 +61,9 @@ fn main() -> Result<()> {
     if let Err(err) = config::logging::init() {
         startup_warning = Some(format!("logging disabled: {err:#}"));
     }
-    capture_stderr();
+    // Restore stderr before main returns: Rust prints a returned error only
+    // after local guards have been dropped and the terminal has been restored.
+    let _stderr_capture = capture_stderr();
     let (keymap, keymap_warning) = config::keymap::Keymap::load();
     let startup_warning = keymap_warning.or(startup_warning);
 
@@ -132,25 +134,46 @@ fn run_app(
 /// every line into tracing — it lands in the Logs tab and the log file
 /// instead of the screen.
 #[cfg(unix)]
-fn capture_stderr() {
+struct StderrCapture {
+    original: std::os::fd::RawFd,
+}
+
+#[cfg(unix)]
+impl Drop for StderrCapture {
+    fn drop(&mut self) {
+        // SAFETY: original is our owned duplicate, kept open for this scope.
+        unsafe {
+            libc::dup2(self.original, libc::STDERR_FILENO);
+            libc::close(self.original);
+        }
+    }
+}
+
+#[cfg(unix)]
+fn capture_stderr() -> Option<StderrCapture> {
     use std::io::BufRead as _;
     use std::os::fd::FromRawFd as _;
 
     let mut fds = [0i32; 2];
     // SAFETY: plain pipe/dup2 syscalls on freshly created fds.
     unsafe {
+        let original = libc::dup(libc::STDERR_FILENO);
+        if original == -1 {
+            return None;
+        }
+        let capture = StderrCapture { original };
         if libc::pipe(fds.as_mut_ptr()) != 0 {
-            return;
+            return None;
         }
         let [read_fd, write_fd] = fds;
         if libc::dup2(write_fd, libc::STDERR_FILENO) == -1 {
             libc::close(read_fd);
             libc::close(write_fd);
-            return;
+            return None;
         }
         libc::close(write_fd);
         let reader = std::fs::File::from_raw_fd(read_fd);
-        std::thread::Builder::new()
+        let reader_thread = std::thread::Builder::new()
             .name("stderr".to_string())
             .spawn(move || {
                 for line in std::io::BufReader::new(reader).lines() {
@@ -159,33 +182,65 @@ fn capture_stderr() {
                         tracing::warn!(target: "stderr", "{line}");
                     }
                 }
-            })
-            .ok();
+            });
+        if reader_thread.is_err() {
+            return None;
+        }
+        Some(capture)
     }
 }
 
 #[cfg(windows)]
-fn capture_stderr() {
+struct StderrCapture {
+    original: windows_sys::Win32::Foundation::HANDLE,
+    writer: windows_sys::Win32::Foundation::HANDLE,
+}
+
+#[cfg(windows)]
+impl Drop for StderrCapture {
+    fn drop(&mut self) {
+        // SAFETY: original is borrowed from the process; writer is owned by
+        // this guard. Restoring the original also preserves shell redirection.
+        unsafe {
+            windows_sys::Win32::System::Console::SetStdHandle(
+                windows_sys::Win32::System::Console::STD_ERROR_HANDLE,
+                self.original,
+            );
+            windows_sys::Win32::Foundation::CloseHandle(self.writer);
+        }
+    }
+}
+
+#[cfg(windows)]
+fn capture_stderr() -> Option<StderrCapture> {
     use std::io::BufRead as _;
     use std::os::windows::io::FromRawHandle as _;
 
-    use windows_sys::Win32::Foundation::CloseHandle;
-    use windows_sys::Win32::System::Console::{STD_ERROR_HANDLE, SetStdHandle};
+    use windows_sys::Win32::Foundation::{CloseHandle, INVALID_HANDLE_VALUE};
+    use windows_sys::Win32::System::Console::{GetStdHandle, STD_ERROR_HANDLE, SetStdHandle};
     use windows_sys::Win32::System::Pipes::CreatePipe;
 
     unsafe {
+        let original = GetStdHandle(STD_ERROR_HANDLE);
+        if original.is_null() || original == INVALID_HANDLE_VALUE {
+            return None;
+        }
         let mut read = core::ptr::null_mut();
         let mut write = core::ptr::null_mut();
         if CreatePipe(&mut read, &mut write, core::ptr::null(), 0) == 0 {
-            return;
+            return None;
         }
         if SetStdHandle(STD_ERROR_HANDLE, write) == 0 {
             CloseHandle(read);
             CloseHandle(write);
-            return;
+            return None;
         }
+        let capture = StderrCapture {
+            original,
+            writer: write,
+        };
         let reader = std::fs::File::from_raw_handle(read);
-        std::thread::Builder::new()
+        let reader_thread = std::thread::Builder::new()
             .name("stderr".to_string())
             .spawn(move || {
                 for line in std::io::BufReader::new(reader).lines() {
@@ -194,13 +249,49 @@ fn capture_stderr() {
                         tracing::warn!(target: "stderr", "{line}");
                     }
                 }
-            })
-            .ok();
+            });
+        if reader_thread.is_err() {
+            return None;
+        }
+        Some(capture)
     }
 }
 
 #[cfg(not(any(unix, windows)))]
-fn capture_stderr() {}
+fn capture_stderr() -> Option<()> {
+    None
+}
+
+#[cfg(test)]
+mod stderr_tests {
+    #[test]
+    fn error_output_is_restored_after_capture() {
+        const CHILD: &str = "FURUMI_STDERR_CAPTURE_TEST";
+        if std::env::var_os(CHILD).is_some() {
+            {
+                let _capture = super::capture_stderr().expect("capture stderr");
+            }
+            eprintln!("furumi-test: visible error after terminal shutdown");
+            return;
+        }
+        // Run in another process: redirecting global stderr inside a parallel
+        // test suite would interfere with unrelated tests and panic reporting.
+        let output = std::process::Command::new(std::env::current_exe().unwrap())
+            .args([
+                "--exact",
+                "stderr_tests::error_output_is_restored_after_capture",
+                "--nocapture",
+            ])
+            .env(CHILD, "1")
+            .output()
+            .unwrap();
+        assert!(output.status.success());
+        assert!(
+            String::from_utf8_lossy(&output.stderr)
+                .contains("furumi-test: visible error after terminal shutdown")
+        );
+    }
+}
 
 /// Kitty keyboard protocol, where supported, disambiguates Esc from alt-keys
 /// and modifier combos. The flags are popped on exit and on panic — leaving

src/player/mod.rs

@@ -90,9 +90,18 @@ impl Shared {
 pub struct Controller {
     tx: Sender<Command>,
     pub shared: Arc<Shared>,
+    playback_allowed: Arc<AtomicBool>,
 }
 
 impl Controller {
+    /// Revoke the output at the audio thread boundary, including work queued
+    /// by background decoders before an ownership transfer.
+    pub fn set_playback_allowed(&self, allowed: bool) {
+        if self.playback_allowed.swap(allowed, Ordering::AcqRel) && !allowed {
+            self.stop();
+        }
+    }
+
     pub fn play(&self, reader: TrackReader, byte_len: Option<u64>, volume: f32) {
         let _ = self.tx.send(Command::Play {
             reader,
@@ -142,11 +151,17 @@ pub fn spawn(on_event: impl Fn(PlayerEvent) + Send + 'static) -> Controller {
     let (tx, rx) = std::sync::mpsc::channel();
     let shared = Arc::new(Shared::default());
     let thread_shared = Arc::clone(&shared);
+    let playback_allowed = Arc::new(AtomicBool::new(true));
+    let thread_allowed = Arc::clone(&playback_allowed);
     std::thread::Builder::new()
         .name("audio".to_string())
-        .spawn(move || run(rx, thread_shared, on_event))
+        .spawn(move || run(rx, thread_shared, thread_allowed, on_event))
         .expect("spawning the audio thread cannot fail");
-    Controller { tx, shared }
+    Controller {
+        tx,
+        shared,
+        playback_allowed,
+    }
 }
 
 struct Output {
@@ -155,7 +170,12 @@ struct Output {
     player: Player,
 }
 
-fn run(rx: Receiver<Command>, shared: Arc<Shared>, on_event: impl Fn(PlayerEvent)) {
+fn run(
+    rx: Receiver<Command>,
+    shared: Arc<Shared>,
+    playback_allowed: Arc<AtomicBool>,
+    on_event: impl Fn(PlayerEvent),
+) {
     let mut output: Option<Output> = None;
     let mut track_loaded = false;
     let mut last_len = 0usize;
@@ -163,6 +183,14 @@ fn run(rx: Receiver<Command>, shared: Arc<Shared>, on_event: impl Fn(PlayerEvent
     loop {
         match rx.recv_timeout(Duration::from_millis(100)) {
             Ok(command) => {
+                if !playback_allowed.load(Ordering::Acquire)
+                    && matches!(
+                        command,
+                        Command::Play { .. } | Command::Enqueue { .. } | Command::Resume
+                    )
+                {
+                    continue;
+                }
                 // Commands change the source queue legitimately; resync the
                 // length so the next tick doesn't read it as a track ending.
                 handle(command, &shared, &mut output, &mut track_loaded, &on_event);