Fix connected devices

CHANGELOG.md

@@ -0,0 +1,5 @@
+# Changelog
+
+## v0.10.6 — 2026-09-02
+
+- Recover federation automatically after sleep, prolonged idle, or a degraded rendezvous transport.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "furumusic"
-version = "0.10.5"
+version = "0.10.7"
 edition = "2024"
 description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL"
 
@@ -45,4 +45,4 @@ uuid = "1"
 librqbit = { version = "8.1.1", features = ["disable-upload"] }
 # P2P federation: publishes the library into a shared DHT and serves audio /
 # catalogs to furumi peers (TUI clients) over the frid stack.
-music-dht = "0.4.0"
+music-dht = "0.5.0"

src/federation/devices.rs

@@ -5,6 +5,9 @@
 //! protocol as the TUI clients on `furumi/sync/1` and maps operations into
 //! user-scoped Postgres state.
 
+use music_dht::playback::{
+    Announcement, Checkpoint, CommandStamp, Config as PlaybackConfig, Engine,
+};
 use std::collections::BTreeMap;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -24,7 +27,7 @@ use super::{TransportStats, record_stream_transport};
 pub const SYNC_ALPN: &[u8] = b"furumi/sync/2";
 
 const CLIENT_VERSION: &str = env!("CARGO_PKG_VERSION");
-const PROTOCOL_VERSION: u16 = 2;
+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);
@@ -193,29 +196,8 @@ struct PlaybackStateWire {
     repeat: PlaybackRepeat,
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-struct PlaybackSnapshot {
-    device_id: String,
-    device_name: String,
-    active: bool,
-    updated_at_ms: i64,
-    state: PlaybackStateWire,
-}
-
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-#[serde(tag = "kind", rename_all = "snake_case")]
-enum PlaybackCommand {
-    SetState {
-        state: PlaybackStateWire,
-        #[serde(default)]
-        seek: bool,
-    },
-    ActiveChanged {
-        active_device_id: String,
-        active_device_name: String,
-        state: PlaybackStateWire,
-    },
-}
+type PlaybackSnapshot = music_dht::playback::Snapshot<PlaybackStateWire>;
+type PlaybackCommand = music_dht::playback::Command<PlaybackStateWire>;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 struct SyncOpWire {
@@ -273,6 +255,8 @@ enum SyncOpPayload {
     PlaybackCommand {
         target_device_id: String,
         command: PlaybackCommand,
+        #[serde(default)]
+        authority: Option<CommandStamp>,
     },
     ListenRecorded {
         event: ListenEvent,
@@ -1097,50 +1081,53 @@ pub async fn sync_loop(
     transport_stats: Arc<TransportStats>,
 ) {
     let mut interval = tokio::time::interval(DEVICE_SYNC_INTERVAL);
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    let mut polls = tokio::task::JoinSet::new();
+    let mut active = std::collections::HashMap::new();
     loop {
-        interval.tick().await;
-        if let Err(err) = sync_once_all(
-            &pool,
-            Arc::clone(&service),
-            Arc::clone(&hub),
-            Arc::clone(&transport_stats),
-        )
-        .await
-        {
-            tracing::debug!("web fed device sync tick failed: {err:#}");
+        tokio::select! {
+            _ = interval.tick() => {
+                let users: Vec<i64> = match sqlx::query_scalar(
+                    "SELECT DISTINCT user_id FROM furumusic__fed_device WHERE trusted_at_ms IS NOT NULL AND revoked_at_ms IS NULL"
+                ).fetch_all(&pool).await {
+                    Ok(users) => users,
+                    Err(error) => { tracing::warn!("device poll listing failed: {error:#}"); continue; }
+                };
+                for user_id in users {
+                    let devices = match active_remote_devices(&pool, user_id).await {
+                        Ok(devices) => devices,
+                        Err(error) => { let _ = set_last_error(&pool, user_id, Some(&format!("{error:#}"))).await; continue; }
+                    };
+                    for device in devices {
+                        let key = (user_id, device.device_id.clone());
+                        if device.endpoint_ticket.trim().is_empty() || active.values().any(|id| id == &key) { continue; }
+                        let pool = pool.clone();
+                        let service = Arc::clone(&service);
+                        let hub = Arc::clone(&hub);
+                        let stats = Arc::clone(&transport_stats);
+                        let handle = polls.spawn(async move {
+                            tokio::time::timeout(Duration::from_secs(30), sync_device(&pool, service, hub, stats, user_id, &device))
+                                .await.context("device sync exchange timed out")?
+                        });
+                        active.insert(handle.id(), key);
+                    }
+                }
+            }
+            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((user_id, device_id)) = active.remove(&task)
+                    && let Err(error) = result {
+                        tracing::debug!(device = %device_id, "web fed device sync failed: {error:#}");
+                        let _ = set_last_error(&pool, user_id, Some(&format!("{}: {error:#}", short_id(&device_id)))).await;
+                    }
+            }
         }
     }
 }
 
-pub async fn sync_once_all(
-    pool: &sqlx::PgPool,
-    service: Arc<MusicDhtService>,
-    hub: Arc<PlayerDeviceHub>,
-    transport_stats: Arc<TransportStats>,
-) -> Result<()> {
-    let rows = sqlx::query(
-        "SELECT DISTINCT user_id FROM furumusic__fed_device
-         WHERE trusted_at_ms IS NOT NULL AND revoked_at_ms IS NULL",
-    )
-    .fetch_all(pool)
-    .await?;
-    for row in rows {
-        let user_id: i64 = row.get("user_id");
-        if let Err(err) = sync_once(
-            pool,
-            Arc::clone(&service),
-            Arc::clone(&hub),
-            Arc::clone(&transport_stats),
-            user_id,
-        )
-        .await
-        {
-            set_last_error(pool, user_id, Some(&format!("{err:#}"))).await?;
-        }
-    }
-    Ok(())
-}
-
 pub async fn sync_once(
     pool: &sqlx::PgPool,
     service: Arc<MusicDhtService>,
@@ -1235,7 +1222,14 @@ async fn try_connect_invite(
                 enforce_single_user_binding(pool, user_id, &profile.device_id).await?;
                 apply_device_profile(pool, user_id, &profile, true).await?;
                 if let Some(playback) = playback {
-                    apply_playback_snapshot(Arc::clone(&hub), pool, user_id, playback).await?;
+                    apply_playback_snapshot(
+                        Arc::clone(&hub),
+                        pool,
+                        user_id,
+                        &profile.device_id,
+                        playback,
+                    )
+                    .await?;
                 }
             }
             apply_device_profiles(pool, user_id, &devices).await?;
@@ -1454,7 +1448,14 @@ async fn handle_pair_request(
     let own_profile = own_profile(pool, user_id, "", &own_ticket).await?;
     apply_device_profile(pool, user_id, &profile, true).await?;
     if let Some(playback) = playback {
-        apply_playback_snapshot(Arc::clone(&hub), pool, user_id, playback).await?;
+        apply_playback_snapshot(
+            Arc::clone(&hub),
+            pool,
+            user_id,
+            &profile.device_id,
+            playback,
+        )
+        .await?;
     }
     apply_snapshot(pool, user_id, incoming_snapshot).await?;
     apply_ops(pool, Arc::clone(&hub), user_id, ops).await?;
@@ -1524,7 +1525,14 @@ async fn handle_hello(
     apply_device_profile(pool, user_id, &profile, false).await?;
     apply_device_profiles(pool, user_id, &devices).await?;
     if let Some(playback) = playback {
-        apply_playback_snapshot(Arc::clone(&hub), pool, user_id, playback).await?;
+        apply_playback_snapshot(
+            Arc::clone(&hub),
+            pool,
+            user_id,
+            &profile.device_id,
+            playback,
+        )
+        .await?;
     }
     apply_snapshot(pool, user_id, incoming_snapshot).await?;
     apply_ops(pool, Arc::clone(&hub), user_id, ops).await?;
@@ -1618,7 +1626,14 @@ async fn sync_device(
         } => {
             apply_device_profiles(pool, user_id, &devices).await?;
             if let Some(playback) = playback {
-                apply_playback_snapshot(Arc::clone(&hub), pool, user_id, playback).await?;
+                apply_playback_snapshot(
+                    Arc::clone(&hub),
+                    pool,
+                    user_id,
+                    &device.device_id,
+                    playback,
+                )
+                .await?;
             }
             apply_snapshot(pool, user_id, snapshot).await?;
             apply_ops(pool, hub, user_id, ops).await?;
@@ -1952,8 +1967,33 @@ async fn own_profile(
     })
 }
 
-async fn record_local_op(pool: &sqlx::PgPool, user_id: i64, payload: SyncOpPayload) -> Result<()> {
+async fn record_local_op(
+    pool: &sqlx::PgPool,
+    user_id: i64,
+    mut payload: SyncOpPayload,
+) -> Result<()> {
     let identity = ensure_identity(pool, user_id, "").await?;
+    if let SyncOpPayload::PlaybackCommand {
+        command, authority, ..
+    } = &mut payload
+    {
+        *authority = Some(
+            with_playback_engine(pool, user_id, &identity, |engine| {
+                if let PlaybackCommand::ActiveChanged {
+                    active_device_id, ..
+                } = command
+                {
+                    if engine.owner() != Some(active_device_id.as_str()) {
+                        engine.transfer(active_device_id, playback_clock());
+                    }
+                }
+                engine.stamp()
+            })
+            .await?
+            .context("no playback owner; select an output first")?,
+        );
+    }
+
     let now = now_ms();
     let row = sqlx::query(
         "UPDATE furumusic__fed_device_identity
@@ -2193,12 +2233,8 @@ async fn apply_op(
             )
             .await
         }
-        SyncOpPayload::PlaybackCommand {
-            target_device_id,
-            command,
-        } => {
-            apply_playback_command(pool, hub, user_id, target_device_id, command, &op.op_id)
-                .await?;
+        SyncOpPayload::PlaybackCommand { .. } => {
+            apply_playback_command(pool, hub, user_id, op).await?;
             Ok(false)
         }
         SyncOpPayload::ListenRecorded { event } => {
@@ -2798,12 +2834,41 @@ async fn apply_playback_command(
     pool: &sqlx::PgPool,
     hub: Arc<PlayerDeviceHub>,
     user_id: i64,
-    target_device_id: &str,
-    command: &PlaybackCommand,
-    op_id: &str,
+    op: &SyncOpWire,
 ) -> Result<()> {
+    let SyncOpPayload::PlaybackCommand {
+        target_device_id,
+        command,
+        authority,
+    } = &op.payload
+    else {
+        return Ok(());
+    };
+    let authority = authority.as_ref();
+    let origin = &op.origin_device_id;
+    let op_id = &op.op_id;
     let identity = ensure_identity(pool, user_id, "").await?;
-    if target_device_id != identity.device_id {
+    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
+    {
+        if active_device_id != &authority.claim.owner {
+            return Ok(());
+        }
+    }
+    let accepted = with_playback_engine(pool, user_id, &identity, |engine| {
+        engine.accept_command(origin, authority, handoff, playback_clock())
+            && (handoff || engine.is_owner())
+    })
+    .await?;
+    if !accepted || target_device_id != &identity.device_id {
         return Ok(());
     }
     let inserted = sqlx::query(
@@ -2822,17 +2887,36 @@ async fn apply_playback_command(
     }
     match command {
         PlaybackCommand::SetState { state, .. } => {
-            enqueue_web_transfer(pool, hub, user_id, state).await?;
+            enqueue_web_transfer(pool, hub, user_id, state, op).await?;
         }
         PlaybackCommand::ActiveChanged {
             active_device_id,
             state,
             ..
         } if active_device_id == &identity.device_id => {
-            enqueue_web_transfer(pool, hub, user_id, state).await?;
+            enqueue_web_transfer(pool, hub, user_id, state, op).await?;
         }
-        PlaybackCommand::ActiveChanged { .. } => {
-            let _ = hub.enqueue_fed_command(user_id, "pause", serde_json::json!({}));
+        PlaybackCommand::ActiveChanged {
+            active_device_id,
+            active_device_name,
+            state,
+        } => {
+            let payload = web_playback_payload(pool, state).await?;
+            if !with_playback_engine(pool, user_id, &identity, |engine| {
+                engine.command_is_current(origin, authority)
+            })
+            .await?
+            {
+                return Ok(());
+            }
+            hub.apply_fed_playback_state_json(
+                user_id,
+                active_device_id,
+                active_device_name,
+                true,
+                payload,
+            )
+            .map_err(|message| anyhow::anyhow!(message))?;
         }
     }
     Ok(())
@@ -2842,14 +2926,36 @@ async fn apply_playback_snapshot(
     hub: Arc<PlayerDeviceHub>,
     pool: &sqlx::PgPool,
     user_id: i64,
+    sender: &str,
     snapshot: PlaybackSnapshot,
 ) -> Result<()> {
+    if snapshot.device_id != sender {
+        return Ok(());
+    }
+    let identity = ensure_identity(pool, user_id, "").await?;
+    let Some(coordination) = &snapshot.coordination else {
+        return Ok(());
+    };
+    let owner = with_playback_engine(pool, user_id, &identity, |engine| {
+        if !engine.observe(&snapshot.device_id, coordination, playback_clock()) {
+            return None;
+        }
+        engine
+            .announcement_is_current(&snapshot.device_id, coordination)
+            .then(|| snapshot.device_id.clone())
+    })
+    .await?;
     let payload = web_playback_payload(pool, &snapshot.state).await?;
+    let active = owner.as_deref() == Some(snapshot.device_id.as_str())
+        && with_playback_engine(pool, user_id, &identity, |engine| {
+            engine.announcement_is_current(&snapshot.device_id, coordination)
+        })
+        .await?;
     hub.apply_fed_playback_state_json(
         user_id,
         &snapshot.device_id,
         &snapshot.device_name,
-        snapshot.active,
+        active,
         payload,
     )
     .map_err(|message| anyhow::anyhow!(message))?;
@@ -2861,8 +2967,24 @@ async fn enqueue_web_transfer(
     hub: Arc<PlayerDeviceHub>,
     user_id: i64,
     state: &PlaybackStateWire,
+    op: &SyncOpWire,
 ) -> Result<()> {
+    let identity = ensure_identity(pool, user_id, "").await?;
     let payload = web_playback_payload(pool, state).await?;
+    let SyncOpPayload::PlaybackCommand {
+        authority: Some(authority),
+        ..
+    } = &op.payload
+    else {
+        return Ok(());
+    };
+    if !with_playback_engine(pool, user_id, &identity, |engine| {
+        engine.command_is_current(&op.origin_device_id, authority)
+    })
+    .await?
+    {
+        return Ok(());
+    }
     if payload
         .get("tracks")
         .and_then(serde_json::Value::as_array)
@@ -2891,6 +3013,7 @@ pub async fn record_web_playback_command(
         pool,
         user_id,
         SyncOpPayload::PlaybackCommand {
+            authority: None,
             target_device_id: target_device_id.to_string(),
             command: PlaybackCommand::SetState {
                 state: wire,
@@ -2915,14 +3038,22 @@ pub async fn record_web_active_transfer(
     ensure_web_playback_target(pool, user_id, target_device_id).await?;
     let wire = playback_state_from_browser_json(pool, state).await?;
     let target_name = web_playback_target_name(pool, user_id, target_device_id).await?;
+    let identity = ensure_identity(pool, user_id, "").await?;
+    with_playback_engine(pool, user_id, &identity, |engine| {
+        engine.transfer(target_device_id, playback_clock())
+    })
+    .await?;
+
     record_local_op(
         pool,
         user_id,
         SyncOpPayload::PlaybackCommand {
+            authority: None,
             target_device_id: target_device_id.to_string(),
-            command: PlaybackCommand::SetState {
+            command: PlaybackCommand::ActiveChanged {
+                active_device_id: target_device_id.to_string(),
+                active_device_name: target_name.clone(),
                 state: wire.clone(),
-                seek: true,
             },
         },
     )
@@ -2940,6 +3071,7 @@ pub async fn record_web_active_transfer(
             pool,
             user_id,
             SyncOpPayload::PlaybackCommand {
+                authority: None,
                 target_device_id: previous_device_id.to_string(),
                 command,
             },
@@ -2958,10 +3090,15 @@ pub async fn record_web_active_takeover(
     ensure_web_playback_target(pool, user_id, previous_device_id).await?;
     let identity = ensure_identity(pool, user_id, "").await?;
     let wire = playback_state_from_browser_json(pool, state).await?;
+    with_playback_engine(pool, user_id, &identity, |engine| {
+        engine.transfer(&identity.device_id, playback_clock())
+    })
+    .await?;
     record_local_op(
         pool,
         user_id,
         SyncOpPayload::PlaybackCommand {
+            authority: None,
             target_device_id: previous_device_id.to_string(),
             command: PlaybackCommand::ActiveChanged {
                 active_device_id: identity.device_id,
@@ -3423,11 +3560,16 @@ async fn local_playback_snapshot(
     user_id: i64,
     identity: &Identity,
 ) -> Option<PlaybackSnapshot> {
-    // Keep publishing an inactive snapshot after a handoff. Omitting the
-    // snapshot left the last `active: true` value alive on trusted peers until
-    // its TTL elapsed, allowing the always-on web peer to reclaim playback.
-    let active = hub.federation_playback_is_local(user_id);
-    let state = hub.playback_state_json_for_commands(user_id)?;
+    let coordination = coordinate_web_output(pool, Arc::clone(&hub), user_id, identity)
+        .await
+        .ok()?;
+    let active = coordination
+        .claim
+        .as_ref()
+        .is_some_and(|claim| claim.owner == identity.device_id);
+    let state = hub
+        .playback_state_json_for_commands(user_id)
+        .unwrap_or_else(|| serde_json::json!({}));
     let wire = playback_state_from_browser_json(pool, state).await.ok()?;
     Some(PlaybackSnapshot {
         device_id: identity.device_id.clone(),
@@ -3435,9 +3577,123 @@ async fn local_playback_snapshot(
         active,
         updated_at_ms: now_ms(),
         state: wire,
+        coordination: Some(coordination),
     })
 }
 
+/// Drive coordination from browser HTTP traffic even before the first peer
+/// connects. This does not require a running federation transport.
+pub async fn refresh_web_output(
+    pool: &sqlx::PgPool,
+    hub: Arc<PlayerDeviceHub>,
+    user_id: i64,
+) -> Result<()> {
+    let identity = ensure_identity(pool, user_id, "").await?;
+    coordinate_web_output(pool, hub, user_id, &identity).await?;
+    Ok(())
+}
+
+async fn coordinate_web_output(
+    pool: &sqlx::PgPool,
+    hub: Arc<PlayerDeviceHub>,
+    user_id: i64,
+    identity: &Identity,
+) -> Result<Announcement> {
+    let (available, playing, report, startup) = hub.federation_output_report(user_id);
+    let coordination = with_playback_engine(pool, user_id, identity, |engine| {
+        engine.set_output(available, playing);
+        if startup {
+            engine.request_startup();
+        }
+        // A server poll cannot refresh the heartbeat of a suspended browser.
+        engine.output_report(report, playback_clock());
+        engine.tick(playback_clock());
+        engine.announcement()
+    })
+    .await?;
+    if let Some(owner) = &coordination.claim {
+        hub.enforce_federation_owner(user_id, &identity.device_id, &owner.owner);
+    }
+    Ok(coordination)
+}
+
+// One serialized coordinator per account. The lock covers checkpoint commit so
+// a later request cannot publish a term before its predecessor is durable.
+struct PlaybackSession {
+    engine: Option<Engine>,
+    group_id: String,
+}
+type PlaybackSessions = std::sync::Mutex<BTreeMap<i64, Arc<tokio::sync::Mutex<PlaybackSession>>>>;
+
+async fn with_playback_engine<R>(
+    pool: &sqlx::PgPool,
+    user_id: i64,
+    identity: &Identity,
+    f: impl FnOnce(&mut Engine) -> R,
+) -> Result<R> {
+    static SESSIONS: std::sync::OnceLock<PlaybackSessions> = std::sync::OnceLock::new();
+    let session = {
+        let mut sessions = SESSIONS
+            .get_or_init(Default::default)
+            .lock()
+            .expect("playback sessions");
+        sessions
+            .entry(user_id)
+            .or_insert_with(|| {
+                Arc::new(tokio::sync::Mutex::new(PlaybackSession {
+                    engine: None,
+                    group_id: String::new(),
+                }))
+            })
+            .clone()
+    };
+    let mut session = session.lock().await;
+    if session.engine.is_none() || session.group_id != identity.group_id {
+        session.group_id = identity.group_id.clone();
+        let row = sqlx::query("SELECT playback_coordination_json, playback_config_json FROM furumusic__fed_device_identity WHERE user_id = $1")
+            .bind(user_id).fetch_one(pool).await?;
+        let durable = row
+            .get::<Option<serde_json::Value>, _>("playback_coordination_json")
+            .map(serde_json::from_value::<Checkpoint>)
+            .transpose()?
+            .filter(|checkpoint| checkpoint.scope == identity.group_id)
+            .map(|checkpoint| checkpoint.state)
+            .unwrap_or_default();
+        let config = row
+            .get::<Option<serde_json::Value>, _>("playback_config_json")
+            .map(serde_json::from_value::<PlaybackConfig>)
+            .transpose()?
+            .unwrap_or_else(PlaybackConfig::passive);
+        session.engine = Some(Engine::new(
+            identity.device_id.clone(),
+            config,
+            durable,
+            playback_clock(),
+        ));
+    }
+    let engine = session
+        .engine
+        .as_mut()
+        .expect("initialized playback session");
+    let previous = engine.clone();
+    let result = f(engine);
+    if previous.durable() != engine.durable() {
+        if let Err(error) = sqlx::query("UPDATE furumusic__fed_device_identity SET playback_coordination_json = $2 WHERE user_id = $1")
+            .bind(user_id).bind(serde_json::to_value(&Checkpoint { scope: identity.group_id.clone(), state: engine.durable().clone() })?).execute(pool).await {
+            *engine = previous; return Err(error.into());
+        }
+    }
+    Ok(result)
+}
+
+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
+}
+
 async fn playback_state_from_browser_json(
     pool: &sqlx::PgPool,
     state: serde_json::Value,
@@ -4787,3 +5043,6 @@ fn base64url_decode(value: &str) -> Result<Vec<u8>> {
     }
     Ok(out)
 }
+
+#[cfg(test)]
+mod interop_tests;

src/federation/devices/interop_tests.rs

@@ -0,0 +1,137 @@
+//! Cross-binary protocol contract; the peer is the TUI's production adapter.
+use super::*;
+use tokio::io::AsyncWriteExt;
+
+#[tokio::test]
+#[ignore = "run furumi_tui/scripts/test_device_interop.py"]
+async fn localhost_tui_peer() {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        let dir =
+            std::path::PathBuf::from(std::env::var("FURUMI_INTEROP_DIR").expect("interop runner"));
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        std::fs::write(
+            dir.join("web-address"),
+            listener.local_addr().unwrap().to_string(),
+        )
+        .unwrap();
+        let (mut stream, _) = listener.accept().await.unwrap();
+        let hub = PlayerDeviceHub::default();
+        let web_id = "web-interop";
+        let mut engine = Engine::new(
+            web_id.into(),
+            PlaybackConfig::passive(),
+            Default::default(),
+            0,
+        );
+        for phase in 0..3 {
+            let hello: WireMessage =
+                serde_json::from_slice(&read_line(&mut stream).await.unwrap()).unwrap();
+            let WireMessage::Hello {
+                profile,
+                playback: Some(snapshot),
+                ..
+            } = hello
+            else {
+                panic!("expected TUI hello")
+            };
+            assert_eq!(profile.protocol_version, PROTOCOL_VERSION);
+            assert_eq!(snapshot.device_id, profile.device_id);
+            let announcement = snapshot
+                .coordination
+                .as_ref()
+                .expect("versioned playback envelope");
+            assert!(engine.observe(&profile.device_id, announcement, phase * 1000));
+            if phase == 0 {
+                assert_eq!(engine.owner(), Some(profile.device_id.as_str()));
+                assert!(
+                    !engine.tick(10_000),
+                    "passive web server must not seize playback"
+                );
+                assert_eq!(snapshot.state.position_secs, 42.5);
+                assert_eq!(snapshot.state.volume, 73);
+                assert!(snapshot.state.shuffle);
+                // Exercise the production browser hub projection without a
+                // PostgreSQL library or an audio device.
+                hub.apply_fed_playback_state_json(
+                    1,
+                    &profile.device_id,
+                    &profile.name,
+                    true,
+                    serde_json::json!({"tracks": [], "index": 0, "track": null,
+                        "position_seconds": 42.5, "duration_seconds": 100.0,
+                        "paused": false, "shuffle": true, "repeat_mode": "all",
+                        "volume": 0.73, "updated_at_ms": now_ms()}),
+                )
+                .unwrap();
+                assert_eq!(
+                    hub.active_device_id_for_commands(1),
+                    Some(format!("fed:{}", profile.device_id))
+                );
+            } else {
+                assert_eq!(
+                    engine.owner(),
+                    Some(if phase == 1 {
+                        web_id
+                    } else {
+                        profile.device_id.as_str()
+                    })
+                );
+            }
+            let mut state = snapshot.state;
+            let command = if phase < 2 {
+                let owner = if phase == 0 {
+                    web_id
+                } else {
+                    profile.device_id.as_str()
+                };
+                assert!(engine.transfer(owner, (phase + 1) * 1000));
+                PlaybackCommand::ActiveChanged {
+                    active_device_id: owner.into(),
+                    active_device_name: "interop".into(),
+                    state: state.clone(),
+                }
+            } else {
+                state.paused = true;
+                state.position_secs = 87.0;
+                PlaybackCommand::SetState {
+                    state: state.clone(),
+                    seek: true,
+                }
+            };
+            engine.set_output(true, engine.is_owner());
+            engine.heartbeat((phase + 1) * 1000);
+            let response = WireMessage::SyncResponse {
+                accepted: true,
+                error: None,
+                devices: vec![],
+                vector: BTreeMap::new(),
+                snapshot: SyncSnapshot::default(),
+                playback: Some(PlaybackSnapshot {
+                    device_id: web_id.into(),
+                    device_name: "WEB".into(),
+                    active: engine.is_owner(),
+                    updated_at_ms: now_ms(),
+                    state,
+                    coordination: Some(engine.announcement()),
+                }),
+                ops: vec![SyncOpWire {
+                    op_id: format!("{web_id}:{}", phase + 1),
+                    origin_device_id: web_id.into(),
+                    seq: (phase + 1) as i64,
+                    hlc_ms: now_ms(),
+                    payload: SyncOpPayload::PlaybackCommand {
+                        target_device_id: profile.device_id,
+                        command,
+                        authority: engine.stamp(),
+                    },
+                }],
+            };
+            let mut bytes = serde_json::to_vec(&response).unwrap();
+            bytes.push(b'\n');
+            stream.write_all(&bytes).await.unwrap();
+        }
+        assert_eq!(read_line(&mut stream).await.unwrap(), b"ok");
+    })
+    .await
+    .expect("TUI/web exchange timed out");
+}

src/federation/mod.rs

@@ -24,8 +24,9 @@ mod storage;
 
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::path::PathBuf;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, OnceLock};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use anyhow::{Context, Result};
 use music_dht::capabilities::CAPABILITIES_ALPN;
@@ -47,6 +48,8 @@ pub use similarity::SIMILARITY_ALPN;
 
 /// How often the published library is re-synchronized with the database.
 const SYNC_INTERVAL: Duration = Duration::from_secs(60);
+const SUPERVISOR_INTERVAL: Duration = Duration::from_secs(15);
+const RECOVERY_COOLDOWN: Duration = Duration::from_secs(5 * 60);
 const TRANSPORT_SAMPLE_LIMIT: usize = 16;
 
 struct Running {
@@ -273,6 +276,7 @@ pub struct Federation {
     data_dir: PathBuf,
     database_url: std::sync::Mutex<String>,
     storage_dir: std::sync::Mutex<String>,
+    desired_network: std::sync::Mutex<Option<String>>,
     save_on_listen: std::sync::atomic::AtomicBool,
     content_cache: std::sync::Mutex<HashMap<i64, (String, String)>>,
     content_pending: std::sync::Mutex<HashSet<i64>>,
@@ -281,6 +285,8 @@ pub struct Federation {
     download_locks: std::sync::Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
     pool: tokio::sync::OnceCell<PgPool>,
     running: tokio::sync::Mutex<Option<Running>>,
+    supervisor_started: AtomicBool,
+    recovery_count: AtomicU64,
     last_sync: std::sync::Mutex<Option<String>>,
     last_error: std::sync::Mutex<Option<String>>,
     transport_stats: Arc<TransportStats>,
@@ -304,6 +310,7 @@ pub fn handle() -> Arc<Federation> {
             data_dir: PathBuf::from(crate::media_paths::resolve_config_path("federation")),
             database_url: std::sync::Mutex::new(String::new()),
             storage_dir: std::sync::Mutex::new(String::new()),
+            desired_network: std::sync::Mutex::new(None),
             save_on_listen: std::sync::atomic::AtomicBool::new(false),
             content_cache: std::sync::Mutex::new(Default::default()),
             content_pending: std::sync::Mutex::new(Default::default()),
@@ -312,6 +319,8 @@ pub fn handle() -> Arc<Federation> {
             download_locks: std::sync::Mutex::new(Default::default()),
             pool: tokio::sync::OnceCell::new(),
             running: tokio::sync::Mutex::new(None),
+            supervisor_started: AtomicBool::new(false),
+            recovery_count: AtomicU64::new(0),
             last_sync: std::sync::Mutex::new(None),
             last_error: std::sync::Mutex::new(None),
             transport_stats: Arc::new(TransportStats::default()),
@@ -324,6 +333,16 @@ impl Federation {
         *lock(&self.last_error) = message;
     }
 
+    fn start_supervisor(self: &Arc<Self>) {
+        if self.supervisor_started.swap(true, Ordering::SeqCst) {
+            return;
+        }
+        let federation = Arc::clone(self);
+        tokio::spawn(async move {
+            federation.supervisor_loop().await;
+        });
+    }
+
     async fn pool(&self) -> Result<PgPool> {
         let url = lock(&self.database_url).clone();
         anyhow::ensure!(!url.is_empty(), "database is not configured");
@@ -343,6 +362,7 @@ impl Federation {
     /// settings live in the config KV table, so this waits for the database
     /// and resolves the same default → DB → env precedence the config uses.
     pub async fn boot(self: &Arc<Self>, config: &AppConfig) {
+        self.start_supervisor();
         *lock(&self.database_url) = config.database_url.clone();
         if config.database_url.is_empty() {
             return;
@@ -404,11 +424,13 @@ impl Federation {
         );
         let network = config.federation_network_id.trim().to_string();
         if config.federation_enabled && !network.is_empty() {
+            *lock(&self.desired_network) = Some(network.clone());
             if let Err(err) = self.start(network, config.agent_storage_dir.clone()).await {
                 tracing::error!("federation start failed: {err:#}");
                 self.set_error(Some(format!("start failed: {err}")));
             }
         } else {
+            *lock(&self.desired_network) = None;
             self.stop().await;
         }
     }
@@ -416,13 +438,38 @@ impl Federation {
     /// Starts the DHT node. Idempotent per network name; a node on another
     /// network is stopped and re-joined.
     async fn start(self: &Arc<Self>, network_name: String, storage_dir: String) -> Result<()> {
+        self.start_mode(network_name, storage_dir, false)
+            .await
+            .map(|_| ())
+    }
+
+    async fn start_mode(
+        self: &Arc<Self>,
+        network_name: String,
+        storage_dir: String,
+        recovery_only: bool,
+    ) -> Result<bool> {
         let pool = self.pool().await?;
         let mut guard = self.running.lock().await;
+        if recovery_only && lock(&self.desired_network).as_deref() != Some(network_name.as_str()) {
+            return Ok(false);
+        }
         if let Some(running) = guard.as_ref() {
             if running.network_name == network_name {
-                return Ok(());
+                if !recovery_only || !running.service.network_health().restart_recommended {
+                    return Ok(false);
+                }
+                tracing::warn!(
+                    network = %network_name,
+                    health = %running.service.network_health().state,
+                    "restarting degraded federation service"
+                );
+            } else if recovery_only {
+                return Ok(false);
             }
             stop_running(guard.take()).await;
+        } else if recovery_only {
+            tracing::warn!(network = %network_name, "retrying stopped federation service");
         }
 
         let dht_storage = Arc::new(PostgresFederationStorage::new(pool.clone()).await?);
@@ -565,7 +612,40 @@ impl Federation {
         drop(guard);
         // Publish right away instead of waiting for the first timer tick.
         self.spawn_sync_soon().await;
-        Ok(())
+        Ok(true)
+    }
+
+    async fn supervisor_loop(self: Arc<Self>) {
+        let mut interval = tokio::time::interval(SUPERVISOR_INTERVAL);
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+        interval.tick().await;
+        let mut last_attempt = None;
+        loop {
+            interval.tick().await;
+            if last_attempt.is_some_and(|attempt: Instant| attempt.elapsed() < RECOVERY_COOLDOWN) {
+                continue;
+            }
+            match self.recover_if_needed().await {
+                Ok(false) => {}
+                Ok(true) => {
+                    last_attempt = Some(Instant::now());
+                    self.recovery_count.fetch_add(1, Ordering::Relaxed);
+                }
+                Err(err) => {
+                    last_attempt = Some(Instant::now());
+                    tracing::error!(error = %err, "federation recovery failed");
+                    self.set_error(Some(format!("network recovery failed: {err}")));
+                }
+            }
+        }
+    }
+
+    async fn recover_if_needed(self: &Arc<Self>) -> Result<bool> {
+        let Some(network_name) = lock(&self.desired_network).clone() else {
+            return Ok(false);
+        };
+        let storage_dir = lock(&self.storage_dir).clone();
+        self.start_mode(network_name, storage_dir, true).await
     }
 
     async fn stop(&self) {
@@ -883,6 +963,7 @@ impl Federation {
         let node = match guard.as_ref() {
             Some(running) => {
                 let service = &running.service;
+                let health = service.network_health();
                 let published = service
                     .list_local_items()
                     .await
@@ -903,6 +984,13 @@ impl Federation {
                     "endpoint_id": service.endpoint_id().to_string(),
                     "connected_peers": peers,
                     "known_contacts": service.known_peers().len(),
+                    "network_health": health.state.to_string(),
+                    "rendezvous_failures": health.consecutive_rendezvous_failures,
+                    "peer_dial_failures": health.consecutive_peer_dial_failures,
+                    "rendezvous_restarts": health.rendezvous_restarts,
+                    "last_rendezvous_success_seconds": health.last_rendezvous_success_ago.map(|age| age.as_secs()),
+                    "last_rendezvous_error": health.last_rendezvous_error,
+                    "recovery_count": self.recovery_count.load(Ordering::Relaxed),
                     "similarity_routing_peers": running.similarity_dht.known_peers(),
                     "published_items": published,
                     "transport": transport,
@@ -1035,6 +1123,11 @@ impl Federation {
         .await
     }
 
+    pub async fn fed_device_web_refresh(&self, user_id: i64) -> Result<()> {
+        let pool = self.pool().await?;
+        devices::refresh_web_output(&pool, crate::player::PlayerDeviceHub::shared(), user_id).await
+    }
+
     pub async fn fed_device_web_command(
         &self,
         user_id: i64,

src/main.rs

@@ -26,7 +26,7 @@ use cot::auth::PasswordVerificationResult;
 use cot::cli::CliMetadata;
 use cot::common_types::Password;
 use cot::config::{
-    DatabaseConfig, MiddlewareConfig, ProjectConfig, SameSite, SessionMiddlewareConfig,
+    DatabaseConfig, Expiry, MiddlewareConfig, ProjectConfig, SameSite, SessionMiddlewareConfig,
     SessionStoreConfig, SessionStoreTypeConfig,
 };
 use cot::db::Database;
@@ -522,6 +522,9 @@ impl Project for FuruProject {
                 MiddlewareConfig::builder()
                     .session(
                         SessionMiddlewareConfig::builder()
+                            .expiry(Expiry::OnInactivity(std::time::Duration::from_secs(
+                                365 * 24 * 60 * 60,
+                            )))
                             .secure(false)
                             .same_site(SameSite::Lax)
                             .store(

src/music/mod.rs

@@ -1984,6 +1984,8 @@ pub mod db_migrations {
                 )",
             )
             .await?;
+        ctx.db.raw("ALTER TABLE furumusic__fed_device_identity ADD COLUMN IF NOT EXISTS playback_coordination_json JSONB").await?;
+        ctx.db.raw("ALTER TABLE furumusic__fed_device_identity ADD COLUMN IF NOT EXISTS playback_config_json JSONB").await?;
         ctx.db
             .raw(
                 "CREATE TABLE IF NOT EXISTS furumusic__fed_device (

src/player/mod.rs

@@ -108,7 +108,7 @@ struct LocalUploadResponse {
     upload: LocalUploadDto,
 }
 
-const PLAYER_DEVICE_TTL_MS: i64 = 30_000;
+const PLAYER_DEVICE_TTL_MS: i64 = 120_000;
 const PLAYER_DEVICE_RETURN_TAKEOVER_MS: i64 = 30 * 60 * 1_000;
 const PLAYER_DEVICE_COMMAND_TTL_MS: i64 = 20_000;
 const PLAYER_DEVICE_MAX_COMMANDS: usize = 32;
@@ -124,6 +124,7 @@ struct PlayerDevice {
     id: String,
     name: String,
     kind: String,
+    report_sequence: u64,
     last_seen_ms: i64,
 }
 
@@ -165,6 +166,8 @@ struct PlayerDeviceHubState {
     commands_by_device: HashMap<(i64, String), VecDeque<PendingPlayerDeviceCommand>>,
     playback_state_by_user: HashMap<i64, PlayerDevicePlaybackStateDto>,
     jams_by_id: HashMap<String, PlayerJamSession>,
+    playback_startup_by_user: HashMap<i64, std::time::Instant>,
+    output_report_sequence: u64,
 }
 
 #[derive(Debug, Default)]
@@ -187,6 +190,9 @@ impl PlayerDeviceHub {
         let now = current_millis();
         let mut state = self.state.lock().expect("player device hub lock");
         self.prune_locked(&mut state, now);
+        if self.user_has_joined_jam_locked(&state, user_id) {
+            return Ok(());
+        }
         let devices = state
             .devices_by_user
             .get(&user_id)
@@ -205,17 +211,105 @@ impl PlayerDeviceHub {
                     .map(|device| device.id.clone())
             })
             .ok_or("no browser playback device")?;
-        state.active_device_by_user.insert(user_id, target.clone());
+        if command == "transfer_state" {
+            state.active_device_by_user.insert(user_id, target.clone());
+        }
         self.enqueue_command_locked(&mut state, user_id, &target, command, payload, now);
         Ok(())
     }
 
-    pub(crate) fn federation_playback_is_local(&self, user_id: i64) -> bool {
-        let state = self.state.lock().expect("player device hub lock");
-        !state
+    pub(crate) fn federation_output_report(&self, user_id: i64) -> (bool, bool, u64, bool) {
+        let mut state = self.state.lock().expect("player device hub lock");
+        if self.user_has_joined_jam_locked(&state, user_id) {
+            return (false, false, 0, false);
+        }
+        let now = current_millis();
+        let active = state.active_device_by_user.get(&user_id);
+        let active_browser = active
+            .filter(|id| !is_fed_virtual_device_id(id))
+            .and_then(|id| state.devices_by_user.get(&user_id)?.get(id))
+            .filter(|device| now.saturating_sub(device.last_seen_ms) < PLAYER_DEVICE_TTL_MS);
+        let candidate = state.devices_by_user.get(&user_id).and_then(|devices| {
+            devices
+                .values()
+                .filter(|device| {
+                    !is_fed_virtual_device_id(&device.id)
+                        && now.saturating_sub(device.last_seen_ms) < PLAYER_DEVICE_TTL_MS
+                })
+                .max_by_key(|device| (device.report_sequence, &device.id))
+        });
+        let available = candidate.is_some();
+        // Only the selected browser can renew the gateway's owned output.
+        let report = active_browser.map_or(0, |device| device.report_sequence);
+        let playing = active_browser.is_some()
+            && state
+                .playback_state_by_user
+                .get(&user_id)
+                .is_some_and(|playback| playback.track.is_some() && !playback.paused);
+        let startup = state
+            .playback_startup_by_user
+            .remove(&user_id)
+            .is_some_and(|started| {
+                started.elapsed().as_millis() <= PLAYER_DEVICE_COMMAND_TTL_MS as u128
+            });
+        (available, playing, report, startup)
+    }
+
+    pub(crate) fn enforce_federation_owner(&self, user_id: i64, local: &str, owner: &str) {
+        let mut state = self.state.lock().expect("player device hub lock");
+        if self.user_has_joined_jam_locked(&state, user_id) {
+            return;
+        }
+        if owner == local {
+            let now = current_millis();
+            let active_local = state
+                .active_device_by_user
+                .get(&user_id)
+                .is_some_and(|id| !is_fed_virtual_device_id(id));
+            if !active_local {
+                let candidate = state.devices_by_user.get(&user_id).and_then(|devices| {
+                    devices
+                        .values()
+                        .filter(|device| {
+                            !is_fed_virtual_device_id(&device.id)
+                                && now.saturating_sub(device.last_seen_ms) < PLAYER_DEVICE_TTL_MS
+                        })
+                        .max_by_key(|device| (device.report_sequence, &device.id))
+                        .map(|device| device.id.clone())
+                });
+                if let Some(candidate) = candidate {
+                    state
+                        .active_device_by_user
+                        .insert(user_id, candidate.clone());
+                    if let Some(playback) =
+                        state
+                            .playback_state_by_user
+                            .get(&user_id)
+                            .and_then(|playback| {
+                                serde_json::to_value(playback_state_at(playback.clone(), now)).ok()
+                            })
+                    {
+                        self.enqueue_command_locked(
+                            &mut state,
+                            user_id,
+                            &candidate,
+                            "transfer_state",
+                            playback,
+                            now,
+                        );
+                    }
+                }
+            }
+            return;
+        }
+        state
             .active_device_by_user
-            .get(&user_id)
-            .is_some_and(|id| is_fed_virtual_device_id(id))
+            .insert(user_id, fed_virtual_device_id(owner));
+        // Poll responses also identify the winner. Purge delayed play/transfer
+        // commands so reconnecting browsers cannot resume an obsolete session.
+        state
+            .commands_by_device
+            .retain(|(user, _), _| *user != user_id);
     }
 
     pub(crate) fn playback_state_json_for_commands(
@@ -269,39 +363,27 @@ impl PlayerDeviceHub {
         state.devices_by_user.entry(user_id).or_default().insert(
             virtual_id.clone(),
             PlayerDevice {
+                report_sequence: 0,
                 id: virtual_id.clone(),
                 name: fed_device_name.to_string(),
                 kind: "fed".to_string(),
                 last_seen_ms: now,
             },
         );
-        // Match the trusted-device playback contract used by the TUI: a
-        // background/stale active snapshot must not steal playback from a
-        // browser that is actively playing. An explicit web handoff changes
-        // `active_device_by_user` to the federated virtual device before the
-        // snapshot arrives, so it still passes through here.
-        let local_playback_is_protected = state
-            .active_device_by_user
-            .get(&user_id)
-            .is_some_and(|active_id| !is_fed_virtual_device_id(active_id))
-            && state
-                .playback_state_by_user
-                .get(&user_id)
-                .is_some_and(|playback| playback.track.is_some() && !playback.paused);
-        if active && local_playback_is_protected {
+        // The shared coordinator has already resolved ownership. A local
+        // playing flag is not permission to reject its winning claim.
+        if self.user_has_joined_jam_locked(&state, user_id) {
             return Ok(());
         }
-        let should_update_playback = active
-            || state
-                .active_device_by_user
-                .get(&user_id)
-                .is_some_and(|active_id| active_id == &virtual_id);
         if active {
+            state
+                .commands_by_device
+                .retain(|(user, _), _| *user != user_id);
             state
                 .active_device_by_user
                 .insert(user_id, virtual_id.clone());
         }
-        if should_update_playback {
+        if active {
             state.playback_state_by_user.insert(user_id, playback_state);
         }
         Ok(())
@@ -330,9 +412,32 @@ impl PlayerDeviceHub {
             .playback_state_by_user
             .get(&user_id)
             .is_some_and(|playback| playback.track.is_some() && !playback.paused);
-        let should_claim_idle_playback = is_new_or_returning
-            && previous_active_id.as_deref() != Some(device_id)
-            && !active_is_playing;
+        let mut policy = music_dht::playback::Config::default();
+        // Local browser failover is permitted. The always-on gateway is not
+        // an automatic candidate against another federated output.
+        if previous_active_id
+            .as_deref()
+            .is_some_and(is_fed_virtual_device_id)
+        {
+            policy.automatic_failover = false;
+        }
+        let owner = previous_active_id.as_ref().map(|id| {
+            let age = state
+                .devices_by_user
+                .get(&user_id)
+                .and_then(|devices| devices.get(id))
+                .map_or(u64::MAX, |device| {
+                    now.saturating_sub(device.last_seen_ms).max(0) as u64
+                });
+            (active_is_playing, age)
+        });
+        let should_claim_idle_playback = previous_active_id.as_deref() != Some(device_id)
+            && policy.should_claim(is_new_or_returning, owner);
+        if is_new_or_returning {
+            state
+                .playback_startup_by_user
+                .insert(user_id, std::time::Instant::now());
+        }
         if should_claim_idle_playback {
             let transfer_state = state
                 .playback_state_by_user
@@ -379,9 +484,57 @@ impl PlayerDeviceHub {
         let now = current_millis();
         let mut state = self.state.lock().expect("player device hub lock");
         self.prune_locked(&mut state, now);
+        let previous = state.active_device_by_user.get(&user_id).cloned();
+        if let Some(previous) =
+            previous.filter(|id| id != device_id && !is_fed_virtual_device_id(id))
+        {
+            let age = state
+                .devices_by_user
+                .get(&user_id)
+                .and_then(|devices| devices.get(&previous))
+                .map_or(u64::MAX, |device| {
+                    now.saturating_sub(device.last_seen_ms).max(0) as u64
+                });
+            let playing = state
+                .playback_state_by_user
+                .get(&user_id)
+                .is_some_and(|playback| playback.track.is_some() && !playback.paused);
+            if music_dht::playback::Config::default().should_claim(false, Some((playing, age))) {
+                state
+                    .active_device_by_user
+                    .insert(user_id, device_id.to_string());
+                if let Some(payload) =
+                    state
+                        .playback_state_by_user
+                        .get(&user_id)
+                        .and_then(|playback| {
+                            serde_json::to_value(playback_state_at(playback.clone(), now)).ok()
+                        })
+                {
+                    self.enqueue_command_locked(
+                        &mut state,
+                        user_id,
+                        device_id,
+                        "transfer_state",
+                        payload,
+                        now,
+                    );
+                }
+            }
+        }
         self.touch_locked(&mut state, user_id, device_id, user_agent, now);
         self.update_playback_state_locked(&mut state, user_id, device_id, playback_state, now);
         self.touch_jam_locked(&mut state, user_id, device_id, current_jam_id, now);
+        if current_jam_id.is_none()
+            && state
+                .active_device_by_user
+                .get(&user_id)
+                .is_some_and(|active| active != device_id)
+        {
+            state
+                .commands_by_device
+                .remove(&(user_id, device_id.to_string()));
+        }
         let commands = state
             .commands_by_device
             .remove(&(user_id, device_id.to_string()))
@@ -534,8 +687,11 @@ impl PlayerDeviceHub {
         user_agent: Option<&str>,
         now: i64,
     ) {
+        state.output_report_sequence = state.output_report_sequence.saturating_add(1);
+        let report_sequence = state.output_report_sequence;
         let devices = state.devices_by_user.entry(user_id).or_default();
         let device = PlayerDevice {
+            report_sequence,
             id: device_id.to_string(),
             name: device_name_from_user_agent(user_agent),
             kind: device_kind_from_user_agent(user_agent).to_string(),
@@ -546,15 +702,7 @@ impl PlayerDeviceHub {
             .device_last_seen_ms
             .insert((user_id, device_id.to_string()), now);
 
-        let active_online = state
-            .active_device_by_user
-            .get(&user_id)
-            .is_some_and(|active_id| devices.contains_key(active_id));
-        if !active_online {
-            state
-                .active_device_by_user
-                .insert(user_id, device_id.to_string());
-        }
+        // Discovery only registers devices; startup/select decides ownership.
     }
 
     fn update_playback_state_locked(
@@ -976,25 +1124,11 @@ impl PlayerDeviceHub {
             devices.retain(|_, device| {
                 now.saturating_sub(device.last_seen_ms) <= PLAYER_DEVICE_TTL_MS
             });
-            let active_valid = state
-                .active_device_by_user
-                .get(user_id)
-                .is_some_and(|active_id| devices.contains_key(active_id));
-            if !active_valid {
-                if let Some(first_device_id) = devices.keys().next().cloned() {
-                    state
-                        .active_device_by_user
-                        .insert(*user_id, first_device_id);
-                } else {
-                    state.active_device_by_user.remove(user_id);
-                    state.playback_state_by_user.remove(user_id);
-                }
-            }
+            // Keep ownership and queue when presence expires. The shared
+            // protocol decides failover; HashMap order must never choose audio.
+            let _ = user_id;
             !devices.is_empty()
         });
-        state
-            .playback_state_by_user
-            .retain(|user_id, _| state.devices_by_user.contains_key(user_id));
 
         state
             .commands_by_device
@@ -1170,6 +1304,83 @@ fn device_kind_from_user_agent(user_agent: Option<&str>) -> &'static str {
 mod device_tests {
     use super::*;
 
+    #[test]
+    fn gateway_timer_does_not_manufacture_browser_reports() {
+        let hub = PlayerDeviceHub::default();
+        assert_eq!(hub.federation_output_report(1), (false, false, 0, false));
+        hub.heartbeat(1, "browser", None, None, None);
+        let first = hub.federation_output_report(1);
+        let repeated = hub.federation_output_report(1);
+        assert!(first.0);
+        assert!(first.3);
+        assert_eq!(first.2, repeated.2);
+        assert!(!repeated.3);
+    }
+
+    #[test]
+    fn an_old_browser_startup_does_not_claim_a_late_peer() {
+        let hub = PlayerDeviceHub::default();
+        hub.heartbeat(1, "browser", None, None, None);
+        hub.state.lock().unwrap().playback_startup_by_user.insert(
+            1,
+            std::time::Instant::now()
+                - std::time::Duration::from_millis(PLAYER_DEVICE_COMMAND_TTL_MS as u64 + 1),
+        );
+        assert!(!hub.federation_output_report(1).3);
+    }
+
+    #[test]
+    fn expired_presence_does_not_replace_federated_owner() {
+        let hub = PlayerDeviceHub::default();
+        hub.state
+            .lock()
+            .unwrap()
+            .active_device_by_user
+            .insert(1, "fed:remote".into());
+        let response = hub.poll(1, "browser", None, None, None);
+        assert_eq!(response.active_device_id.as_deref(), Some("fed:remote"));
+        assert!(response.commands.is_empty());
+    }
+
+    #[test]
+    fn browser_poll_fails_over_only_after_local_owner_timeout() {
+        let hub = PlayerDeviceHub::default();
+        hub.heartbeat(1, "first", None, None, None);
+        assert_eq!(
+            hub.poll(1, "second", None, None, None)
+                .active_device_id
+                .as_deref(),
+            Some("first")
+        );
+        hub.state
+            .lock()
+            .unwrap()
+            .devices_by_user
+            .get_mut(&1)
+            .unwrap()
+            .get_mut("first")
+            .unwrap()
+            .last_seen_ms = current_millis() - PLAYER_DEVICE_TTL_MS - 1;
+        assert_eq!(
+            hub.poll(1, "second", None, None, None)
+                .active_device_id
+                .as_deref(),
+            Some("second")
+        );
+    }
+
+    #[test]
+    fn pruning_keeps_the_owner_when_every_device_is_offline() {
+        let hub = PlayerDeviceHub::default();
+        hub.heartbeat(1, "browser", None, None, None);
+        let mut state = hub.state.lock().unwrap();
+        hub.prune_locked(&mut state, current_millis() + PLAYER_DEVICE_TTL_MS + 1);
+        assert_eq!(
+            state.active_device_by_user.get(&1).map(String::as_str),
+            Some("browser")
+        );
+    }
+
     #[test]
     fn detects_furumi_android_native_client() {
         let user_agent = Some("FurumiAndroid/1.0 Android Mobile");
@@ -1217,7 +1428,7 @@ mod device_tests {
     }
 
     #[test]
-    fn federated_snapshot_does_not_steal_active_browser_playback() {
+    fn resolved_federation_owner_overrides_a_playing_browser() {
         let hub = PlayerDeviceHub::default();
         let user_id = 7;
         {
@@ -1225,6 +1436,7 @@ mod device_tests {
             state.devices_by_user.entry(user_id).or_default().insert(
                 "browser".to_string(),
                 PlayerDevice {
+                    report_sequence: 0,
                     id: "browser".to_string(),
                     name: "Browser".to_string(),
                     kind: "computer".to_string(),
@@ -1276,7 +1488,7 @@ mod device_tests {
                 .active_device_by_user
                 .get(&user_id)
                 .map(String::as_str),
-            Some("browser")
+            Some("fed:remote")
         );
         assert!(
             state
@@ -1372,6 +1584,7 @@ mod device_tests {
             state.devices_by_user.entry(user_id).or_default().insert(
                 "browser".to_string(),
                 PlayerDevice {
+                    report_sequence: 0,
                     id: "browser".to_string(),
                     name: "Browser".to_string(),
                     kind: "computer".to_string(),
@@ -5347,6 +5560,12 @@ async fn devices_heartbeat_handler(
             return Ok(json_error(StatusCode::BAD_REQUEST, &format!("{err}")));
         }
     }
+    if let Err(error) = crate::federation::handle()
+        .fed_device_web_refresh(user.id)
+        .await
+    {
+        tracing::warn!(user_id = user.id, %error, "playback coordination startup failed");
+    }
     Json(response).into_response()
 }
 
@@ -5364,6 +5583,12 @@ async fn devices_poll_handler(
         return Ok(json_error(StatusCode::BAD_REQUEST, "invalid device id"));
     };
 
+    if let Err(error) = crate::federation::handle()
+        .fed_device_web_refresh(user.id)
+        .await
+    {
+        tracing::warn!(user_id = user.id, %error, "playback coordination refresh failed");
+    }
     let response = hub.poll(
         user.id,
         &device_id,

templates/player/scripts.html

@@ -1,5 +1,23 @@
 <script defer src="https://cdn.jsdelivr.net/npm/alpinejs@3/dist/cdn.min.js"></script>
 <script>
+// Handle expired sessions centrally, including background player requests.
+let loginRedirectStarted = false;
+function redirectOnUnauthorized(status) {
+    if (status !== 401 || loginRedirectStarted) return;
+    loginRedirectStarted = true;
+    window.location.replace('/login');
+}
+
+const playerFetch = window.fetch.bind(window);
+window.fetch = async (input, init) => {
+    const response = await playerFetch(input, init);
+    const url = new URL(input instanceof Request ? input.url : input, window.location.href);
+    if (url.origin === window.location.origin && url.pathname.startsWith('/api/')) {
+        redirectOnUnauthorized(response.status);
+    }
+    return response;
+};
+
 const T = {
     info: "{{ t.player_info }}",
     noDetails: "{{ t.player_no_details }}",
@@ -1688,7 +1706,7 @@ document.addEventListener('alpine:init', () => {
                 }
 
                 const player = Alpine.store('player');
-                if (player && Array.isArray(data.commands)) {
+                if (player && (this.isActive() || this.shouldPlayJamLocally()) && Array.isArray(data.commands)) {
                     data.commands.forEach(command => player._executeRemoteCommand(command));
                 }
                 if (player && !this.isActive()) {
@@ -1706,7 +1724,6 @@ document.addEventListener('alpine:init', () => {
         },
 
         _apply(data) {
-            const wasActive = this.isActive();
             const previousJamId = this.currentJamId;
             this.activeDeviceId = data.active_device_id || null;
             this.devices = Array.isArray(data.devices) ? data.devices : [];
@@ -1722,7 +1739,7 @@ document.addEventListener('alpine:init', () => {
             if (previousJamId !== this.currentJamId || !this.canPlayJamLocally()) {
                 this._setJamLocalPlayback(false, { pauseLocal: true });
             }
-            if (wasActive && !this.isActive()) {
+            if (!this.isActive() && !this.shouldPlayJamLocally()) {
                 Alpine.store('player')?._pauseLocal();
             }
             this._maybeShowRemoteHint();
@@ -6010,6 +6027,7 @@ document.addEventListener('alpine:init', () => {
                     this.uploadProgressText = this.uploadProgress.toFixed(1) + '%';
                 };
                 xhr.onload = () => {
+                    redirectOnUnauthorized(xhr.status);
                     let data = {};
                     try { data = JSON.parse(xhr.responseText || '{}'); } catch {}
                     if (xhr.status >= 200 && xhr.status < 300) resolve(data);