Added 'connected devices' to external lib

ARCHITECTURE.md

@@ -166,6 +166,24 @@ node.
 
 ### Extension protocols
 
+Personal playback coordination is a pure domain state machine in
+`music_dht::playback`. It owns the common claim ordering, output-heartbeat
+rules, command fences, configurable startup/failover policy and generic
+snapshot/command envelopes. Applications supply authenticated events,
+monotonic time and actual output availability, persist scoped checkpoints,
+and apply the resolved owner to their UI/audio adapters. They must not elect
+an owner from an online-device list or reject a winning claim just because
+their local output is playing. The web gateway uses a passive server profile;
+browser startup and explicit selection are separate inputs.
+
+This register provides eventual exclusivity after a partition heals, not a
+quorum lock during a partition. Device-sync profiles now advertise version 3;
+the additive coordination envelope is independently versioned and transported
+over the existing JSON `furumi/sync/2` stream. Legacy library records remain
+readable, but legacy unfenced personal playback messages cannot control an
+upgraded output. See [PLAYBACK_PROTOCOL.md](PLAYBACK_PROTOCOL.md) for the
+contract, adapter obligations, configuration and rollout rules.
+
 Music discovery identifies an owner and content; it does not force all domain
 traffic through DHT messages. `music-dht` exposes the underlying authenticated
 stream capability and defines shared wire models for Furumi catalog and device

Cargo.lock

@@ -912,7 +912,7 @@ dependencies = [
 
 [[package]]
 name = "furumi-library"
-version = "0.1.0"
+version = "0.2.0"
 dependencies = [
  "anyhow",
  "blake3",
@@ -2073,7 +2073,7 @@ dependencies = [
 
 [[package]]
 name = "music-dht"
-version = "0.4.1"
+version = "0.5.0"
 dependencies = [
  "anyhow",
  "async-trait",

Cargo.toml

@@ -14,7 +14,7 @@ repository = "https://gt.hexor.cy/ab/frid"
 
 [workspace.dependencies]
 federation-net = { path = "crates/federation-net", version = "0.3.1" }
-music-dht = { path = "crates/music-dht", version = "0.4.1" }
+music-dht = { path = "crates/music-dht", version = "0.5.0" }
 iroh = "1"
 iroh-base = "1"
 iroh-tickets = "1"

PLAYBACK_PROTOCOL.md

@@ -0,0 +1,114 @@
+# Personal playback coordination
+
+`music_dht::playback` owns the personal-device state machine, policy parameters,
+snapshot/command envelopes and conflict rules. It has no networking, database,
+clock, UI or audio dependency. The existing JSON-lines device-sync transport
+continues to carry snapshots and logged commands. Jam is a separate authority
+scope; its capability-based protocol is not a personal ownership claim.
+
+## Contract
+
+Each logical device identity must have one coordinator and one selected output
+adapter. Independent live installations must not reuse a device identity or
+checkpoint. The web hub is a single authority for its user's browser outputs;
+running multiple independent hubs for that same identity requires an external
+actor/leader arrangement, not two copies of this in-memory adapter.
+
+An owner is selected by an immutable `Claim { counter, issuer, owner }`.
+Claims are ordered lexicographically in that order. A locally issued claim
+increments the greatest counter observed so far. Persist the winning claim;
+never reset it because a peer disappears from an online list. Controllers
+gossip the winning claim as well. Concurrent startup, transfer or failover
+claims therefore converge to the same winner regardless of delivery order.
+
+Only the winning output may render personal audio. A losing player stops its
+audio even if it was playing; it cannot reject a higher claim on that basis.
+This is eventual exclusivity, not a quorum lock: partitions can temporarily
+produce two outputs. Convergence assumes eventual delivery between trusted
+devices, functioning polling/event loops, and no continuing stream of explicit
+conflicting user transfers. A disconnected browser cannot be forcibly silenced.
+
+Availability, actual unpaused playback, ownership and network connectivity are
+different inputs. Only new reports from the owner's audio endpoint renew its
+liveness. Duplicate snapshots and relayed claims do not. Gateway timers must
+not turn cached browser state into new output heartbeats. The engine reserves
+heartbeat sequence ranges in its checkpoint so a restart cannot roll them back
+and every UI tick does not require a database write.
+
+Commands carry `CommandStamp { version, claim, sequence }`. A command can carry
+its existing claim ahead of the corresponding announcement. An explicit
+handoff must name that claim's owner and originate from its issuer. In a term,
+commands are applied in increasing sequence per sender; old, duplicate and
+incompatible envelopes are ignored. The command-sender map is bounded to 128
+entries per term. Concurrent controllers are serialized by the selected
+output, whose snapshots distribute the resulting queue and transport state.
+This does not claim exactly-once audio effects across process crashes.
+
+## Adapter obligations
+
+1. Authenticate the peer and trusted group using the existing transport and
+   operation-log membership checks. Bind snapshot `device_id` to the actual
+   sender. Names and snapshot wall-clock timestamps do not grant ownership.
+2. Restore `Checkpoint` only for its matching group. Construct `Engine` with a
+   stable device id, local `Config` and a monotonic clock. Do not restore wall
+   clock liveness deadlines.
+3. Feed real output availability and activity. Use `heartbeat` for a local
+   output or `output_report` with an increasing source-report id for a gateway.
+4. Feed announcements, tick the engine, and route explicit transfers through
+   `transfer`. Ordinary controls only stamp a command in the current term.
+5. Persist changed durable state before publishing decisions or executing
+   commands. Serialize transitions with checkpoint commits and restore the
+   previous engine if persistence fails. Recheck queued command fences before
+   their effects reach the output.
+6. Apply the resolved owner to the existing UI/audio adapter. Gossip its claim
+   even while controlling another output. Online lists are presentation only.
+
+Startup has a discovery grace period. If a restored/relayed owner has not yet
+reported, keep the startup decision pending until its heartbeat or timeout.
+Discovery is necessarily bounded: an entirely unknown peer across a partition
+cannot prevent a local startup claim. Once the peers communicate, normal claim
+ordering resolves the conflict.
+
+## Configuration
+
+The complete serde `Config` accepts these fields (unknown fields are errors):
+
+| Field | Default | Meaning |
+| --- | --- | --- |
+| `claim_on_startup` | `true` | One startup intent for this application instance |
+| `automatic_failover` | `true` | Claim when the owner stops reporting |
+| `take_paused_on_startup` | `true` | Startup may take a paused/idle output |
+| `discovery_ms` | `3000` | Initial discovery grace |
+| `owner_timeout_ms` | `120000` | Missing fresh output reports before failover |
+
+`Config::passive()` disables application-startup claims and automatic failover.
+Furumusic uses this for the always-on server. A real browser startup supplies a
+separate one-shot `request_startup`; explicit selection supplies a transfer.
+These events are not generated by server polling. Local browser selection uses
+the same `Config::should_claim` policy, with automatic cross-federation claims
+disabled. Browser presence expiry preserves the current owner and queue.
+
+TUI exposes the configuration as the `playback` section of its existing settings
+file and preserves it when saving other settings. The web adapter reads optional
+`playback_config_json` on the user's `furumusic__fed_device_identity` row at
+session initialization; null selects the passive profile. Audio availability
+is a runtime input, not a persisted setting.
+
+## Versioning and rollout
+
+The library-sync JSON ALPN remains `furumi/sync/2`; catalog, pairing and library
+data do not require a new transport. Device profiles advertise protocol version
+3. Coordination envelopes independently carry version 1. Snapshot coordination
+and operation authority are additive optional JSON fields so library-sync
+records from older clients still deserialize. Upgraded clients ignore legacy
+personal playback snapshots/commands without the coordination envelope.
+The exclusivity guarantee applies after all audio clients have upgraded.
+
+The shared generic `Snapshot<S>` and `Command<S>` keep state conversion in the
+applications while preventing divergent control message definitions. These
+envelopes use JSON, not a backwards-compatible promise for postcard layouts.
+Do not change Jam's independent authority by interpreting personal stamps there.
+
+For local integration checks, patch the three frid crates with Cargo's external
+`--config` mechanism. Keep released dependency declarations and lockfiles in
+the applications; publish frid and then bump client dependencies for rollout.

crates/furumi-library/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "furumi-library"
-version = "0.1.0"
+version = "0.2.0"
 description = "Client-independent SQLite music library for Furumi players"
 readme = "README.md"
 documentation = "https://docs.rs/furumi-library"

crates/music-dht/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "music-dht"
-version = "0.4.1"
+version = "0.5.0"
 description = "Distributed music library search: a Kademlia-style DHT on top of federation-net"
 readme = "README.md"
 documentation = "https://docs.rs/music-dht"

crates/music-dht/README.md

@@ -183,7 +183,15 @@ to know that the integration exists.
 Protocol v1 uses `furumi/sync/1`; v2 uses `furumi/sync/2`. During migration,
 new clients should accept both ALPNs and only send `ListenRecorded` when
 `supports_listen_history(peer.protocol_version)` returns true. Likes,
-playlists, membership and playback coordination remain compatible with v1.
+playlists and membership remain compatible with v1. Profile version 3 adds
+versioned personal playback coordination over the existing JSON sync stream;
+legacy unfenced playback messages are ignored by upgraded clients.
+
+`music_dht::playback` provides the shared pure ownership engine and generic
+snapshot/command envelopes. Configure startup and failover with `Config`, feed
+actual output reports, and persist `Checkpoint` before applying decisions.
+`Config::passive()` is intended for always-running gateways. See the workspace
+`PLAYBACK_PROTOCOL.md` for invariants, configuration and migration details.
 
 ## Verification
 

crates/music-dht/src/device_sync.rs

@@ -27,7 +27,7 @@ pub const SYNC_ALPN_V2: &[u8] = b"furumi/sync/2";
 pub const SYNC_ALPN: &[u8] = SYNC_ALPN_V2;
 
 /// Version of the personal-device sync wire protocol.
-pub const DEVICE_SYNC_PROTOCOL_VERSION: u16 = 2;
+pub const DEVICE_SYNC_PROTOCOL_VERSION: u16 = 3;
 
 /// Minimum actual listening time retained for an interrupted listen.
 pub const MIN_RECORDED_LISTEN_MS: i64 = 5_000;
@@ -327,43 +327,10 @@ pub struct PlaybackStateWire {
     pub repeat: PlaybackRepeat,
 }
 
-/// Playback status broadcast by a device during sync.
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-pub struct PlaybackSnapshot {
-    /// Device id.
-    pub device_id: String,
-    /// Device name.
-    pub device_name: String,
-    /// Whether this device considers itself active.
-    pub active: bool,
-    /// Update timestamp in unix milliseconds.
-    pub updated_at_ms: i64,
-    /// Playback state.
-    pub state: PlaybackStateWire,
-}
-
-/// Playback command delivered through the sync op log.
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-#[serde(tag = "kind", rename_all = "snake_case")]
-pub enum PlaybackCommand {
-    /// Replace target state. `seek` means the target should also seek audio.
-    SetState {
-        /// New state.
-        state: PlaybackStateWire,
-        /// Whether to seek the audio sink to `state.position_secs`.
-        #[serde(default)]
-        seek: bool,
-    },
-    /// Another device became active.
-    ActiveChanged {
-        /// New active device id.
-        active_device_id: String,
-        /// New active device name.
-        active_device_name: String,
-        /// Transferred playback state.
-        state: PlaybackStateWire,
-    },
-}
+/// Playback snapshot with the common coordination envelope.
+pub type PlaybackSnapshot = crate::playback::Snapshot<PlaybackStateWire>;
+/// Shared command vocabulary with the portable music queue DTO.
+pub type PlaybackCommand = crate::playback::Command<PlaybackStateWire>;
 
 /// One operation in the device-sync log.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -461,6 +428,9 @@ pub enum SyncOpPayload {
         target_device_id: String,
         /// Command.
         command: PlaybackCommand,
+        /// Ownership and ordering fence. Legacy commands cannot change v3 playback.
+        #[serde(default)]
+        authority: Option<crate::playback::CommandStamp>,
     },
     /// Append one immutable listening-history event.
     ///

crates/music-dht/src/lib.rs

@@ -137,3 +137,5 @@ pub use federation_net::{
     NetworkHealthSnapshot, NetworkHealthState, NetworkId, PeerTicket, RecvStream, RendezvousConfig,
     SecretKey, SendStream, StreamAcceptor,
 };
+
+pub mod playback;

crates/music-dht/src/playback.rs

@@ -0,0 +1,624 @@
+//! Transport-independent personal playback coordination.
+//!
+//! This is an eventually consistent register, not a distributed lock. During a
+//! partition multiple outputs may play. Once announcements are delivered, all
+//! participants adopt the greatest claim and only its owner may keep playing.
+//! Applications authenticate membership, persist [`DurableState`] before acting,
+//! feed monotonic time and actual output heartbeats, and execute ownership changes.
+//! Relaying an announcement must never manufacture an output heartbeat.
+
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
+
+/// Version of the additive playback coordination envelope.
+pub const VERSION: u16 = 1;
+/// Maximum command senders retained within one ownership term.
+pub const MAX_COMMAND_SENDERS: usize = 128;
+
+/// Local choices; these do not weaken conflict resolution or command fencing.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct Config {
+    /// Claim idle playback once after discovery at application startup.
+    pub claim_on_startup: bool,
+    /// Become the output when its owner stops producing heartbeats.
+    pub automatic_failover: bool,
+    /// A newly started output may take over a paused owner.
+    pub take_paused_on_startup: bool,
+    /// Wait for initial peer discovery before considering a startup claim.
+    pub discovery_ms: u64,
+    /// Missing output heartbeats, not server connectivity, trigger failover.
+    pub owner_timeout_ms: u64,
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            claim_on_startup: true,
+            automatic_failover: true,
+            take_paused_on_startup: true,
+            discovery_ms: 3_000,
+            owner_timeout_ms: 120_000,
+        }
+    }
+}
+
+impl Config {
+    /// Shared candidate policy, also used when a gateway selects a browser.
+    /// `owner` is `(actually_playing, milliseconds_since_fresh_output_report)`.
+    /// Startup intent is a one-shot event, never inferred from reconnection.
+    pub fn should_claim(&self, startup: bool, owner: Option<(bool, u64)>) -> bool {
+        let vacant = owner.is_none();
+        let expired = owner.is_some_and(|(_, age)| age >= self.owner_timeout_ms.max(1));
+        let idle = owner.is_some_and(|(playing, _)| !playing);
+        (startup && (vacant || expired || (self.take_paused_on_startup && idle)))
+            || (self.automatic_failover && (vacant || expired))
+    }
+
+    /// An always-running gateway only transfers ownership on user intent.
+    pub fn passive() -> Self {
+        Self {
+            claim_on_startup: false,
+            automatic_failover: false,
+            ..Self::default()
+        }
+    }
+}
+
+/// A Lamport term with a deterministic tie-breaker. No wall-clock ordering.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+pub struct Claim {
+    /// Greater than every term the issuer has observed.
+    pub counter: u64,
+    /// Authenticated device issuing the claim.
+    pub issuer: String,
+    /// Device allowed to render personal audio in this term.
+    pub owner: String,
+}
+
+impl Claim {
+    fn valid(&self) -> bool {
+        self.counter > 0
+            && [&self.issuer, &self.owner]
+                .into_iter()
+                .all(|id| !id.is_empty() && id.len() <= 256)
+    }
+}
+
+/// Persist atomically before publishing a new claim or applying a command.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct DurableState {
+    /// Winning claim, retained even after the owner becomes unavailable.
+    pub claim: Option<Claim>,
+    /// Last accepted command sequence per sender in the winning term.
+    pub commands: BTreeMap<String, u64>,
+    /// Locally authored command sequence; never reset on ownership changes.
+    pub command_sequence: u64,
+    /// Locally authored output heartbeat; retained across process restarts.
+    pub heartbeat_sequence: u64,
+}
+
+/// Gossip carried beside a playback snapshot, including by controllers.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Announcement {
+    /// Envelope version; unknown versions are ignored.
+    pub version: u16,
+    /// Greatest observed ownership claim.
+    pub claim: Option<Claim>,
+    /// Advances only when the sender's actual output reports activity.
+    pub heartbeat: u64,
+    /// Sender can currently render audio; a server alone cannot.
+    pub available: bool,
+    /// Sender is actually playing unpaused audio.
+    pub playing: bool,
+}
+
+/// Commands are scoped to an ownership term and ordered per sender.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CommandStamp {
+    /// Coordination envelope version.
+    pub version: u16,
+    /// Ownership term in which the command is valid.
+    pub claim: Claim,
+    /// Increasing sequence of the authenticated operation origin.
+    pub sequence: u64,
+}
+
+/// Application checkpoint scoped to a trusted-device group.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Checkpoint {
+    /// Application-defined membership scope. Do not reuse state across groups.
+    pub scope: String,
+    /// Durable coordinator state in that scope.
+    pub state: DurableState,
+}
+
+/// Shared snapshot envelope; applications supply their portable queue DTO.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Snapshot<S> {
+    /// Authenticated sender, checked against the transport by the adapter.
+    pub device_id: String,
+    /// Display name only; never an identity or ordering key.
+    pub device_name: String,
+    /// Compatibility/UI projection. Ownership comes only from coordination.
+    pub active: bool,
+    /// Position extrapolation timestamp, never used for claim ordering.
+    pub updated_at_ms: i64,
+    /// Output state or the controller's cached presentation state.
+    pub state: S,
+    /// Absent on older clients and on the independently scoped Jam protocol.
+    #[serde(default)]
+    pub coordination: Option<Announcement>,
+}
+
+/// Shared personal/Jam command vocabulary. Personal commands additionally
+/// require [`CommandStamp`]; Jam keeps its own capability and authority scope.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum Command<S> {
+    /// Update the current owner's output state.
+    SetState {
+        /// Queue and transport state.
+        state: S,
+        /// Also seek the actual audio output.
+        #[serde(default)]
+        seek: bool,
+    },
+    /// Explicit transfer with state for the newly selected output.
+    ActiveChanged {
+        /// Selected output identity.
+        active_device_id: String,
+        /// Selected output display name.
+        active_device_name: String,
+        /// State to render after transfer.
+        state: S,
+    },
+}
+
+/// Pure engine. All times are monotonic milliseconds supplied by the adapter.
+#[derive(Debug, Clone)]
+pub struct Engine {
+    id: String,
+    config: Config,
+    durable: DurableState,
+    started_at: u64,
+    startup_pending: bool,
+    owner_seen_at: u64,
+    owner_heartbeat: u64,
+    owner_playing: bool,
+    available: bool,
+    playing: bool,
+    output_report: u64,
+    heartbeat_sequence: u64,
+}
+
+impl Engine {
+    /// Restore durable state, but never restore persisted liveness timestamps.
+    pub fn new(id: String, config: Config, durable: DurableState, now: u64) -> Self {
+        let heartbeat_sequence = durable.heartbeat_sequence;
+        let owner_playing = durable.claim.is_some();
+        Self {
+            id,
+            startup_pending: config.claim_on_startup,
+            config,
+            durable,
+            started_at: now,
+            owner_seen_at: now,
+            owner_heartbeat: 0,
+            owner_playing,
+            available: false,
+            playing: false,
+            output_report: 0,
+            heartbeat_sequence,
+        }
+    }
+
+    /// Durable checkpoint. Save it before executing decisions.
+    pub fn durable(&self) -> &DurableState {
+        &self.durable
+    }
+
+    /// Winning output, whether currently reachable or not.
+    pub fn owner(&self) -> Option<&str> {
+        self.durable
+            .claim
+            .as_ref()
+            .map(|claim| claim.owner.as_str())
+    }
+
+    /// Whether this device currently owns personal playback.
+    pub fn is_owner(&self) -> bool {
+        self.owner() == Some(self.id.as_str())
+    }
+
+    /// Update local capabilities without interpreting connectivity as playback.
+    pub fn set_output(&mut self, available: bool, playing: bool) {
+        if self.available && !available {
+            self.advance_heartbeat();
+        }
+        self.available = available;
+        self.playing = available && playing;
+    }
+
+    /// Feed one fresh report from the actual local output, even if paused.
+    pub fn heartbeat(&mut self, now: u64) {
+        if self.available {
+            self.advance_heartbeat();
+            if self.is_owner() {
+                self.owner_seen_at = now;
+                self.owner_playing = self.playing;
+            }
+        }
+    }
+
+    fn advance_heartbeat(&mut self) {
+        self.heartbeat_sequence = self.heartbeat_sequence.saturating_add(1);
+        // Reserve a range before publishing. Restart skips unused values.
+        if self.heartbeat_sequence > self.durable.heartbeat_sequence {
+            self.durable.heartbeat_sequence = self.heartbeat_sequence.saturating_add(1024);
+        }
+    }
+
+    /// Merge gossip received from an authenticated trusted device.
+    /// Returns false for unsupported or malformed envelopes.
+    pub fn observe(&mut self, sender: &str, message: &Announcement, now: u64) -> bool {
+        if message.version != VERSION || message.claim.as_ref().is_some_and(|c| !c.valid()) {
+            return false;
+        }
+        if let Some(claim) = &message.claim {
+            self.merge(claim, now);
+        }
+        if self.owner() == Some(sender)
+            && self.durable.claim == message.claim
+            && message.heartbeat > self.owner_heartbeat
+        {
+            self.owner_heartbeat = message.heartbeat;
+            self.owner_playing = message.available && message.playing;
+            if message.available {
+                self.owner_seen_at = now;
+            }
+        }
+        true
+    }
+
+    fn merge(&mut self, claim: &Claim, now: u64) {
+        if self
+            .durable
+            .claim
+            .as_ref()
+            .is_none_or(|current| claim > current)
+        {
+            self.durable.claim = Some(claim.clone());
+            self.durable.commands.clear();
+            self.owner_seen_at = now;
+            self.owner_heartbeat = 0;
+            // A transferred owner gets a full grace period before failover.
+            self.owner_playing = true;
+        }
+    }
+
+    /// Apply explicit user intent. Target eligibility is checked by the adapter.
+    /// Returns false rather than wrapping an exhausted Lamport counter.
+    pub fn transfer(&mut self, target: &str, now: u64) -> bool {
+        if target.is_empty() || target.len() > 256 {
+            return false;
+        }
+        let Some(counter) = self
+            .durable
+            .claim
+            .as_ref()
+            .map_or(Some(1), |c| c.counter.checked_add(1))
+        else {
+            return false;
+        };
+        let claim = Claim {
+            counter,
+            issuer: self.id.clone(),
+            owner: target.into(),
+        };
+        self.merge(&claim, now);
+        self.startup_pending = false;
+        true
+    }
+
+    /// A real browser/application startup, never a server timer or reconnect.
+    pub fn request_startup(&mut self) {
+        self.startup_pending = true;
+    }
+
+    /// Feed a monotonically increasing output-report identifier. Polling the
+    /// same cached report cannot renew owner liveness.
+    pub fn output_report(&mut self, report: u64, now: u64) {
+        if report > self.output_report {
+            self.output_report = report;
+            self.heartbeat(now);
+        }
+    }
+
+    /// Fence an effect again immediately before executing it in an event loop.
+    pub fn command_is_current(&self, sender: &str, stamp: &CommandStamp) -> bool {
+        self.durable.claim.as_ref() == Some(&stamp.claim)
+            && self.durable.commands.get(sender) == Some(&stamp.sequence)
+    }
+
+    /// Whether an owner snapshot is at least as recent as accepted liveness.
+    pub fn announcement_is_current(&self, sender: &str, message: &Announcement) -> bool {
+        self.owner() == Some(sender)
+            && self.durable.claim == message.claim
+            && message.heartbeat >= self.owner_heartbeat
+    }
+
+    /// Evaluate startup/failover. A losing node cannot immediately reclaim:
+    /// the winning term always receives an entire heartbeat grace period.
+    pub fn tick(&mut self, now: u64) -> bool {
+        if now.saturating_sub(self.started_at) < self.config.discovery_ms {
+            return false;
+        }
+        // A restored or relayed owner has unknown activity until it reports.
+        // Keep startup intent pending through discovery, rather than treating
+        // an unknown playing state as an idle output.
+        if self.startup_pending
+            && self.owner().is_some()
+            && !self.is_owner()
+            && self.owner_heartbeat == 0
+            && now.saturating_sub(self.owner_seen_at) < self.config.owner_timeout_ms.max(1)
+        {
+            return false;
+        }
+        let startup = self.startup_pending;
+        self.startup_pending = false;
+        let eligible = self.available && !self.is_owner();
+        let owner = self
+            .owner()
+            .map(|_| (self.owner_playing, now.saturating_sub(self.owner_seen_at)));
+        if eligible && self.config.should_claim(startup, owner) {
+            return self.transfer(&self.id.clone(), now);
+        }
+        false
+    }
+
+    /// Current gossip. Controllers repeat the claim, never the owner's heartbeat.
+    pub fn announcement(&self) -> Announcement {
+        Announcement {
+            version: VERSION,
+            claim: self.durable.claim.clone(),
+            heartbeat: self.heartbeat_sequence,
+            available: self.available,
+            playing: self.playing,
+        }
+    }
+
+    /// Stamp a command with the current ownership term.
+    pub fn stamp(&mut self) -> Option<CommandStamp> {
+        let claim = self.durable.claim.clone()?;
+        self.durable.command_sequence = self.durable.command_sequence.checked_add(1)?;
+        Some(CommandStamp {
+            version: VERSION,
+            claim,
+            sequence: self.durable.command_sequence,
+        })
+    }
+
+    /// Merge the referenced term and fence a command. Commands may carry a
+    /// previously issued claim ahead of its gossip (transport can reorder).
+    /// Creating claims remains exclusive to transfer/startup/failover.
+    /// Sender identity comes from the authenticated operation log.
+    pub fn accept_command(
+        &mut self,
+        sender: &str,
+        stamp: &CommandStamp,
+        handoff: bool,
+        now: u64,
+    ) -> bool {
+        if stamp.version != VERSION
+            || !stamp.claim.valid()
+            || sender.is_empty()
+            || sender.len() > 256
+            || stamp.sequence == 0
+        {
+            return false;
+        }
+        if handoff && stamp.claim.issuer != sender {
+            return false;
+        }
+        self.merge(&stamp.claim, now);
+        if self.durable.claim.as_ref() != Some(&stamp.claim) {
+            return false;
+        }
+        if !self.durable.commands.contains_key(sender)
+            && self.durable.commands.len() >= MAX_COMMAND_SENDERS
+        {
+            return false;
+        }
+        let previous = self.durable.commands.entry(sender.to_string()).or_default();
+        if stamp.sequence <= *previous {
+            return false;
+        }
+        *previous = stamp.sequence;
+        true
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    fn node(id: &str, config: Config) -> Engine {
+        let mut e = Engine::new(id.into(), config, DurableState::default(), 0);
+        e.set_output(true, false);
+        e
+    }
+    #[test]
+    fn partition_converges_without_reclaim_loop() {
+        let mut a = node("a", Config::default());
+        let mut b = node("b", Config::default());
+        assert!(a.tick(3_000));
+        assert!(b.tick(3_000));
+        a.set_output(true, true);
+        b.set_output(true, true);
+        a.heartbeat(3_001);
+        b.heartbeat(3_001);
+        let aa = a.announcement();
+        let bb = b.announcement();
+        a.observe("b", &bb, 3_002);
+        b.observe("a", &aa, 3_002);
+        assert_eq!(a.owner(), b.owner());
+        assert!(!a.is_owner());
+        assert!(b.is_owner());
+        assert!(!a.tick(3_003));
+        a.observe("a", &aa, 3_004);
+        assert_eq!(a.owner(), Some("b"));
+    }
+    #[test]
+    fn startup_respects_playing_but_can_take_paused() {
+        for playing in [true, false] {
+            let mut owner = node("a", Config::default());
+            owner.tick(3_000);
+            owner.set_output(true, playing);
+            owner.heartbeat(3_001);
+            let mut newcomer = node("b", Config::default());
+            newcomer.observe("a", &owner.announcement(), 2_000);
+            assert_eq!(newcomer.tick(3_000), !playing);
+        }
+    }
+    #[test]
+    fn passive_gateway_never_claims_without_intent() {
+        let mut web = node("web", Config::passive());
+        assert!(!web.tick(3_000));
+        assert!(!web.tick(1_000_000));
+        assert!(web.transfer("web", 1_000_001));
+        assert!(web.is_owner());
+    }
+    #[test]
+    fn duplicates_and_relays_do_not_extend_liveness() {
+        let mut a = node("a", Config::default());
+        a.tick(3_000);
+        a.set_output(true, true);
+        a.heartbeat(3_001);
+        let mut b = node("b", Config::default());
+        b.observe("a", &a.announcement(), 3_002);
+        b.tick(3_003);
+        b.observe("a", &a.announcement(), 100_000);
+        b.observe("relay", &a.announcement(), 122_000);
+        assert!(b.tick(123_003));
+    }
+    #[test]
+    fn commands_are_fenced_ordered_and_survive_restore() {
+        let mut a = node("a", Config::default());
+        a.tick(3_000);
+        let old = a.stamp().unwrap();
+        assert!(a.accept_command("a", &old, false, 3_001));
+        let mut restored = Engine::new("a".into(), Config::default(), a.durable().clone(), 4_000);
+        assert!(!restored.accept_command("a", &old, false, 4_001));
+        restored.transfer("b", 4_002);
+        assert!(!restored.accept_command("a", &old, false, 4_003));
+        let forged = restored.stamp().unwrap();
+        assert!(!restored.accept_command("spoof", &forged, true, 4_004));
+    }
+
+    #[test]
+    fn three_outputs_converge_in_every_delivery_order() {
+        for order in [
+            [0, 1, 2],
+            [0, 2, 1],
+            [1, 0, 2],
+            [1, 2, 0],
+            [2, 0, 1],
+            [2, 1, 0],
+        ] {
+            let mut nodes = [
+                node("a", Config::default()),
+                node("b", Config::default()),
+                node("c", Config::default()),
+            ];
+            for node in &mut nodes {
+                node.tick(3_000);
+                node.set_output(true, true);
+                node.heartbeat(3_001);
+            }
+            let messages = nodes.each_ref().map(Engine::announcement);
+            for node in &mut nodes {
+                for index in order {
+                    node.observe(["a", "b", "c"][index], &messages[index], 4_000);
+                }
+                assert_eq!(node.owner(), Some("c"));
+                assert!(!node.tick(4_001));
+            }
+            assert_eq!(nodes.iter().filter(|node| node.is_owner()).count(), 1);
+        }
+    }
+
+    #[test]
+    fn cached_browser_reports_do_not_keep_a_suspended_output_alive() {
+        let mut web = node("web", Config::passive());
+        web.transfer("web", 0);
+        web.set_output(true, true);
+        web.output_report(1, 1);
+        let mut tui = node("tui", Config::default());
+        tui.observe("web", &web.announcement(), 2);
+        tui.tick(3_000);
+        for now in [30_000, 60_000, 90_000, 120_003] {
+            web.output_report(1, now);
+            tui.observe("web", &web.announcement(), now);
+        }
+        assert!(tui.tick(120_003));
+        web.observe("tui", &tui.announcement(), 120_004);
+        assert!(!web.is_owner());
+    }
+
+    #[test]
+    fn heartbeat_reservation_survives_restart() {
+        let mut a = node("a", Config::default());
+        a.tick(3_000);
+        a.heartbeat(3_001);
+        let before = a.announcement();
+        let mut restored = Engine::new("a".into(), Config::default(), a.durable().clone(), 0);
+        restored.set_output(true, false);
+        restored.heartbeat(1);
+        assert!(restored.announcement().heartbeat > before.heartbeat);
+    }
+
+    #[test]
+    fn command_can_arrive_before_handoff_and_older_state_cannot_overwrite_it() {
+        let mut a = node("a", Config::default());
+        a.transfer("b", 1);
+        let handoff = a.stamp().unwrap();
+        let play = a.stamp().unwrap();
+        let mut b = node("b", Config::default());
+        assert!(b.accept_command("a", &play, false, 2));
+        assert!(b.is_owner());
+        assert!(!b.accept_command("a", &handoff, true, 3));
+        assert!(b.command_is_current("a", &play));
+        let mut incompatible = a.stamp().unwrap();
+        incompatible.version += 1;
+        assert!(!b.accept_command("a", &incompatible, false, 4));
+    }
+
+    #[test]
+    fn disabled_automatic_policy_and_unavailable_outputs_never_claim() {
+        let mut disabled = node(
+            "a",
+            Config {
+                claim_on_startup: false,
+                automatic_failover: false,
+                ..Config::default()
+            },
+        );
+        assert!(!disabled.tick(1_000_000));
+        let mut unavailable = node("b", Config::default());
+        unavailable.set_output(false, false);
+        assert!(!unavailable.tick(1_000_000));
+    }
+
+    #[test]
+    fn command_sender_memory_is_bounded() {
+        let mut a = node("a", Config::default());
+        a.tick(3_000);
+        let stamp = a.stamp().unwrap();
+        for index in 0..MAX_COMMAND_SENDERS {
+            assert!(a.accept_command(&format!("sender{index}"), &stamp, false, 3_001));
+        }
+        assert!(!a.accept_command("overflow", &stamp, false, 3_002));
+        assert_eq!(a.durable().commands.len(), MAX_COMMAND_SENDERS);
+    }
+}