merge: integrate qbzd event hooks from PR #700

cx
This commit is contained in:
vicrodh committed 2026-08-27 19:26:13 -06:00
commit e109e88ec2
13 files changed
+682 -9

No files matched your search

+4
View File
@@ -384,6 +384,10 @@ streamer.
- HiFi wizard with copyable audio-stack config blocks (clipboard works over SSH)
- MPRIS out of the box, live JSON events (`qbzd watch`), service files for
systemd/OpenRC/runit
- Event hooks: `qbzd settings set hooks.script /path/to/script` runs your script
on playback/session events with `QBZ_*` environment variables — push
integration for audio-box distros (moOde, Volumio, DIY setups), no polling
required
Full manual:
**[Headless Daemon (qbzd) — Wiki](https://github.com/vicrodh/qbz/wiki/Headless-Daemon)**
+15 -1
View File
@@ -16,6 +16,12 @@ pub struct DaemonPrefs {
/// (CONSOLE ext). Default ON; the `QBZD_MPRIS` env var overrides it when
/// set. Toggling this needs a daemon restart (the service spawns at boot).
pub mpris_enabled: bool,
/// Absolute path of an executable the daemon runs on playback/session
/// events (CONSOLE ext), with the event in `QBZ_*` environment variables.
/// Empty = hooks off (the default). The `QBZD_HOOK` env var overrides it
/// when set. Changing it needs a daemon restart (the dispatcher spawns at
/// boot). Machine-local by design — NEVER part of the settings bundle.
pub hook_script: String,
}
impl Default for DaemonPrefs {
fn default() -> Self {
@@ -23,6 +29,7 @@ impl Default for DaemonPrefs {
streaming_quality: "hires_plus".into(),
volume: 0.5,
mpris_enabled: true,
hook_script: String::new(),
}
}
}
@@ -35,7 +42,11 @@ pub fn load_at(data_root: &Path) -> DaemonPrefs {
// forward-compatible: unknown fields ignored WITH a warning (01 §10.3)
if let Some(obj) = v.as_object() {
for k in obj.keys() {
if k != "streaming_quality" && k != "volume" && k != "mpris_enabled" {
if k != "streaming_quality"
&& k != "volume"
&& k != "mpris_enabled"
&& k != "hook_script"
{
log::warn!("[daemon_prefs] unknown field ignored: {k}");
}
}
@@ -92,6 +103,8 @@ mod tests {
// Non-default (default is true) so the roundtrip actually exercises
// the field, not just its presence.
mpris_enabled: false,
// Non-default (default is empty) for the same reason.
hook_script: "/usr/local/bin/qbz-hook.sh".into(),
};
save_at(&prefs, &dir).expect("save daemon prefs");
@@ -100,6 +113,7 @@ mod tests {
assert_eq!(loaded.streaming_quality, "cd");
assert_eq!(loaded.volume, 0.72);
assert!(!loaded.mpris_enabled);
assert_eq!(loaded.hook_script, "/usr/local/bin/qbz-hook.sh");
let _ = std::fs::remove_dir_all(&dir);
}
+11
View File
@@ -47,6 +47,17 @@ pub enum CoreEvent {
/// Repeat mode changed
RepeatModeChanged { mode: crate::playback::RepeatMode },
// ============ Qobuz Connect Events ============
/// Qobuz Connect session lifecycle changed (headless renderer)
QconnectSessionChanged {
/// "off" | "connecting" | "connected" | "retrying" | "exhausted"
state: String,
/// The advertised renderer name, when one is set
device_name: Option<String>,
/// True only while a session is established ("connected")
session_active: bool,
},
// ============ Authentication Events ============
/// User logged in successfully
LoggedIn { session: UserSession },
+21
View File
@@ -438,6 +438,7 @@ fn create_output_stream_with_config(
sample_rate: u32,
channels: u16,
exclusive_mode: bool,
state: SharedState,
) -> Result<MixerDeviceSink, String> {
log::info!(
"Creating MixerDeviceSink: {}Hz, {} channels, exclusive: {}",
@@ -508,9 +509,25 @@ fn create_output_stream_with_config(
// Create MixerDeviceSink with custom config
match DeviceSinkBuilder::from_device(device) {
Ok(builder) => {
// rodio's default error callback only eprintln!s a live stream
// error (e.g. an ALSA buffer underrun mid-playback), so playback
// dies without the driver ever seeing a message to latch. Record
// it on the shared state instead so it drains to the event bus as
// a PlaybackError. Rate-limited: ALSA can repeat EPIPE every
// period on a wedged device, and each recorded message is one
// bus event (and one forked hook script) after the drain.
let mut last_reported: Option<std::time::Instant> = None;
match builder
.with_supported_config(&supported_config)
.with_buffer_size(cpal_buffer_size)
.with_error_callback(move |err| {
log::error!("Audio stream error: {err}");
let now = std::time::Instant::now();
if last_reported.map_or(true, |t| now.duration_since(t).as_secs() >= 5) {
last_reported = Some(now);
state.record_stream_error(format!("Audio stream error: {err}"));
}
})
.open_stream()
{
Ok(mixer_sink) => {
@@ -1994,6 +2011,7 @@ impl Player {
sample_rate,
channels,
dac_passthrough,
thread_state.clone(),
)
.map(StreamType::rodio)
};
@@ -2085,6 +2103,7 @@ impl Player {
sample_rate,
channels,
dac_passthrough,
thread_state.clone(),
)
.map(StreamType::rodio)
};
@@ -2513,6 +2532,7 @@ impl Player {
sample_rate,
channels,
dac_passthrough,
thread_state.clone(),
)
.map(StreamType::rodio)
};
@@ -2587,6 +2607,7 @@ impl Player {
sample_rate,
channels,
dac_passthrough,
thread_state.clone(),
)
.map(StreamType::rodio)
};
+53 -1
View File
@@ -56,7 +56,8 @@ pub enum ApplyClass {
/// lists exactly these, in this order; `settings set` accepts exactly these
/// keys and nothing else. Domains: `audio.*` (`AudioSettingsStore`),
/// `playback.*` (daemon_prefs.streaming_quality + `PlaybackPreferencesStore`),
/// `qconnect.*` (the daemon-root `qconnect_settings.db` KV, T9).
/// `qconnect.*` (the daemon-root `qconnect_settings.db` KV, T9),
/// `hooks.*` (daemon_prefs, CONSOLE ext).
const KEY_TABLE: &[(&str, ApplyClass)] = &[
// --- audio (Reinit — 03-setup-tui.md §4.3's 9-field list) -------------
("audio.backend", ApplyClass::Reinit),
@@ -92,6 +93,8 @@ const KEY_TABLE: &[(&str, ApplyClass)] = &[
("qconnect.device_name", ApplyClass::None),
("qconnect.startup_mode", ApplyClass::None),
("qconnect.volume_mode", ApplyClass::None),
// --- hooks (daemon_prefs, CONSOLE ext) ----------------------------------
("hooks.script", ApplyClass::None),
];
fn classify(key: &str) -> Option<ApplyClass> {
@@ -192,6 +195,24 @@ fn render_opt_string(v: &Option<String>) -> String {
v.clone().unwrap_or_else(|| "system".to_string())
}
/// `hooks.script`: empty clears (hooks off, the default); anything else must
/// be an absolute path — the daemon spawns it verbatim with no PATH lookup or
/// shell, so a relative value would silently depend on the daemon's cwd. Not
/// checked for existence here (a headless `settings set` must work before the
/// script is installed); the daemon warns at spawn time instead.
fn parse_hook_script(v: &str) -> Result<String, String> {
let trimmed = v.trim();
if trimmed.is_empty() {
return Ok(String::new());
}
if !std::path::Path::new(trimmed).is_absolute() {
return Err(format!(
"invalid hook script '{trimmed}' — expected an absolute path (or empty to disable)"
));
}
Ok(trimmed.to_string())
}
/// `audio.device_max_sample_rate`: "none"/"" clears the limit (Hz, matching
/// the stored unit directly — e.g. `192000`, not `192` kHz).
fn parse_opt_u32(v: &str) -> Result<Option<u32>, String> {
@@ -332,6 +353,7 @@ fn read_all(roots: &ProfileRoots) -> Result<Vec<(&'static str, String)>, String>
"qconnect.volume_mode" => {
qconnect_kv::load_volume_mode_at(&db).unwrap_or_else(|| "software".to_string())
}
"hooks.script" => prefs.hook_script.clone(),
other => unreachable!("KEY_TABLE/read_all drifted apart on key: {other}"),
};
out.push((*key, value));
@@ -557,6 +579,12 @@ pub(crate) fn write_one(roots: &ProfileRoots, key: &str, raw: &str) -> Result<Ap
let v = parse_volume_mode(raw).map_err(SetError::Usage)?;
qconnect_kv::save_volume_mode_at(&qconnect_db(roots), &v)
}
"hooks.script" => {
let v = parse_hook_script(raw).map_err(SetError::Usage)?;
let mut prefs = daemon_prefs::load_at(&roots.data);
prefs.hook_script = v;
daemon_prefs::save_at(&prefs, &roots.data).map_err(SetError::Io)?;
}
other => unreachable!("KEY_TABLE/write_one drifted apart on key: {other}"),
}
Ok(class)
@@ -1215,6 +1243,30 @@ mod tests {
}
}
#[test]
fn hook_script_accepts_absolute_or_empty_and_rejects_relative() {
// Empty clears (hooks off, the default — and what `show`'s value must
// round-trip through `set`); absolute paths pass verbatim; a relative
// path is a USAGE mistake (the daemon spawns with no PATH lookup).
assert_eq!(parse_hook_script(""), Ok(String::new()));
assert_eq!(parse_hook_script(" "), Ok(String::new()));
assert_eq!(
parse_hook_script("/usr/local/bin/qbz-hook.sh"),
Ok("/usr/local/bin/qbz-hook.sh".to_string())
);
assert!(parse_hook_script("qbz-hook.sh").is_err());
assert!(parse_hook_script("./hook.sh").is_err());
let roots = scratch_roots("hook-script");
let err = write_one(&roots, "hooks.script", "relative.sh").unwrap_err();
assert!(matches!(err, SetError::Usage(_)), "{err:?}");
write_one(&roots, "hooks.script", "/opt/hook.sh").expect("absolute path writes");
assert_eq!(daemon_prefs::load_at(&roots.data).hook_script, "/opt/hook.sh");
write_one(&roots, "hooks.script", "").expect("empty clears");
assert_eq!(daemon_prefs::load_at(&roots.data).hook_script, "");
cleanup(&roots);
}
#[test]
fn show_json_round_trips_into_set_for_every_canonical_key() {
// The brief's Step 1 property: `settings show --json` includes every
+68 -6
View File
@@ -96,6 +96,12 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec<String>) -> Re
// ordering — aborted alongside the scrobbler for a clean shutdown.
let memwatch = qbz_app::memory_watchdog::spawn(booted.runtime.core().player());
// Attach the CoreEvent bus to the shared state so the qconnect lifecycle
// latches can publish `QconnectSessionChanged` (SSE + the event hook).
if let Ok(mut s) = booted.shared.lock() {
s.bus = Some(booted.bus.clone());
}
// 10. playback driver (T4). Spawn the 450 ms headless orchestrator on the
// booted runtime + shared state. It runs safely regardless of auth: with
// no session the queue is empty and each tick is a near-no-op. The
@@ -117,7 +123,16 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec<String>) -> Re
// QConnect report scheduler (step 12) waits on it. Created BEFORE the driver
// so `on_edge` can capture it, and shared with `qconnect::start`.
let report_notify = Arc::new(tokio::sync::Notify::new());
let deps = build_driver_deps(quality_cell.clone(), booted.shared.clone(), report_notify.clone());
// The same edge, pulsed onto a SECOND Notify for the events bridge (10e):
// two subscribers on one Notify would race for the single stored permit.
let edge_notify = Arc::new(tokio::sync::Notify::new());
let deps = build_driver_deps(
quality_cell.clone(),
booted.shared.clone(),
report_notify.clone(),
edge_notify.clone(),
booted.bus.clone(),
);
let driver = tokio::spawn(playback_driver::run_driver(
booted.runtime.clone(),
deps,
@@ -139,6 +154,27 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec<String>) -> Re
// outside the #521/§8.2 ordering — aborted for a clean shutdown below.
let scrobbler = crate::scrobble_engine::spawn(roots.clone(), booted.bus.subscribe());
// 10c½. Events bridge: translate the driver's transition edges into the
// playback CoreEvents the bus consumers above (and SSE, and the event
// hook below) are written for — TrackStarted / PlaybackStateChanged /
// PositionUpdated / VolumeChanged. Holds only a Weak<AppRuntime>
// upgraded per wake, but is still aborted+joined ahead of
// `drop(booted)` so a mid-wake strong Arc can't outlive the ordering.
let events_bridge =
crate::events_bridge::spawn(&booted.runtime, booted.bus.clone(), edge_notify);
// 10c¾. Event hook (CONSOLE): fork `hooks.script` (or `QBZD_HOOK`) once per
// forwarded bus event with QBZ_* variables in its environment — push
// integration for headless boxes (moOde-style renderer coordination).
// Holds no Arc<AppRuntime>; aborted for a clean shutdown.
let hooks = match crate::hooks::script(&roots) {
Some(path) => {
log::info!("[hooks] running {} on daemon events (hooks.script)", path.display());
Some(crate::hooks::spawn(path, booted.bus.subscribe()))
}
None => None,
};
// 10d. MPRIS media controls (CONSOLE): publish org.mpris.MediaPlayer2 so a
// KDE/GNOME media widget, a plasmoid, or hardware media keys drive the
// daemon with no custom client. The inbound callback holds a
@@ -236,6 +272,15 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec<String>) -> Re
// device — the same ordering constraint as the driver (§8.2).
memwatch.abort();
let _ = memwatch.await;
// Stop the events bridge BEFORE `drop(booted)`: it upgrades its Weak to a
// strong Arc<AppRuntime> for the span of each wake (#521 ordering).
events_bridge.abort();
let _ = events_bridge.await;
// Stop the event-hook dispatcher (holds no Arc<AppRuntime>; order-free).
if let Some(hooks) = hooks {
hooks.abort();
let _ = hooks.await;
}
// Tear down MPRIS: abort its updater and drop the D-Bus handle. Its inbound
// callback held only a Weak<AppRuntime>, so this is order-free too.
if let Some(mpris) = mpris {
@@ -384,11 +429,14 @@ async fn boot(roots: &ProfileRoots, cfg: &QbzdConfig, warn_count: usize) -> Resu
/// Assemble the driver's host side channels: the streaming-quality resolver and
/// the daemon-shared latching / tick-timestamping hooks. T10: `on_edge` now
/// pulses the QConnect report `Notify` so the report scheduler reports on the
/// same transition/periodic edges the driver detects (§7.2).
/// same transition/periodic edges the driver detects (§7.2) — and, on a second
/// `Notify`, the events bridge (10c½) that publishes those edges to the bus.
fn build_driver_deps(
quality_cell: Arc<std::sync::Mutex<qbz_models::Quality>>,
shared: Arc<Mutex<DaemonShared>>,
report_notify: Arc<tokio::sync::Notify>,
edge_notify: Arc<tokio::sync::Notify>,
bus: tokio::sync::broadcast::Sender<qbz_models::CoreEvent>,
) -> DriverDeps {
let latch_shared = shared.clone();
let tick_shared = shared;
@@ -402,16 +450,28 @@ fn build_driver_deps(
// T10: signal the report scheduler on every ReportEdge. `notify_one`
// stores a single permit if the scheduler is mid-report, so no edge is
// lost and rapid edges coalesce into one report.
on_edge: Arc::new(move || report_notify.notify_one()),
on_edge: Arc::new(move || {
report_notify.notify_one();
edge_notify.notify_one();
}),
on_latch: Arc::new(move |category, message| {
if let Ok(mut s) = latch_shared.lock() {
match category {
"stream" => s.last_errors.stream = Some(message),
"transport" => s.last_errors.transport = Some(message),
"auth" => s.last_errors.auth = Some(message),
"stream" => s.last_errors.stream = Some(message.clone()),
"transport" => s.last_errors.transport = Some(message.clone()),
"auth" => s.last_errors.auth = Some(message.clone()),
_ => {}
}
}
// Surface stream failures on the bus too, so `/api/events` and the
// event hook see them live (e.g. a busy ALSA device at play time)
// instead of only the polled /api/status error latch.
if category == "stream" {
let _ = bus.send(qbz_models::CoreEvent::PlaybackError {
track_id: 0,
message,
});
}
}),
on_tick: Arc::new(move || {
if let Ok(mut s) = tick_shared.lock() {
@@ -517,6 +577,8 @@ fn new_shared(cfg: &QbzdConfig) -> Arc<Mutex<DaemonShared>> {
qconnect: QconnectStatus::default(),
credential_fingerprint: None,
network_online: std::sync::atomic::AtomicBool::new(true),
// Attached by daemon::run right after boot (the bus outlives boot).
bus: None,
}))
}
+107
View File
@@ -0,0 +1,107 @@
// crates/qbzd/src/events_bridge.rs — bridge playback-driver edges onto the
// CoreEvent bus.
//
// The 450 ms playback driver detects track/play-state transitions
// (`DriverAction::ReportEdge` -> `deps.on_edge`), but nothing translated those
// edges into the playback CoreEvent variants the bus consumers were written
// for: mpris.rs and scrobble_engine.rs both carry TrackStarted /
// PlaybackStateChanged / PositionUpdated / VolumeChanged arms that never
// fired, and `/api/events` (SSE) / `qbzd watch` carried no playback traffic at
// all. This task closes that gap: it wakes on the same edge Notify pulse the
// QConnect report scheduler uses (with a 2 s poll fallback, because
// `plan_tick` only reports edges while a track id is present — a stop that
// clears the track id would otherwise go unseen), reads the live player state,
// dedups against what it last published, and sends the delta to the bus.
//
// Holds only a Weak<AppRuntime> (upgraded per wake, dropped before the next
// wait), so it sits outside the #521 audio-release ordering — the caller
// aborts it for a clean shutdown, same as the queue-persist subscriber.
use std::sync::{Arc, Weak};
use std::time::Duration;
use qbz_app::shell::AppRuntime;
use qbz_models::{CoreEvent, PlaybackState};
use tokio::sync::{broadcast, Notify};
use tokio::task::JoinHandle;
use crate::adapter::DaemonAdapter;
type Runtime = Arc<AppRuntime<DaemonAdapter>>;
/// The poll fallback cadence: only exercised when no edge pulse arrives (e.g.
/// a stop cleared the track id, which suppresses `ReportEdge`).
const FALLBACK_POLL: Duration = Duration::from_secs(2);
/// Spawn the bridge task. Emits, deduped against the last published values:
/// * `TrackStarted` on a track-id change while playing (with the queue's
/// current-track metadata),
/// * `PlaybackStateChanged` on a playing/paused/stopped change,
/// * `PositionUpdated` on every wake while playing (~2 s cadence, the
/// scrobbler's timing source and the MPRIS progress feed),
/// * `VolumeChanged` on a volume change.
pub fn spawn(runtime: &Runtime, bus: broadcast::Sender<CoreEvent>, edge: Arc<Notify>) -> JoinHandle<()> {
let weak: Weak<AppRuntime<DaemonAdapter>> = Arc::downgrade(runtime);
tokio::spawn(async move {
let mut last_track_id: u64 = 0;
let mut last_state: Option<PlaybackState> = None;
let mut last_volume: Option<f32> = None;
loop {
// Wake on a driver edge, or fall back to a slow poll so a
// track-id-clearing stop still surfaces as a transition.
let _ = tokio::time::timeout(FALLBACK_POLL, edge.notified()).await;
let Some(rt) = weak.upgrade() else { return };
let core = rt.core();
let player = core.player();
let ev = player.get_playback_event();
let has_audio = player.has_loaded_audio();
if ev.track_id != 0 && ev.track_id != last_track_id && ev.is_playing {
// The queue's current track carries the metadata (title/artist/
// artwork). Skip when the cursor hasn't caught up yet — the next
// wake retries because last_track_id only advances on a send.
let queue = core.get_queue_state().await;
if let Some(track) = queue.current_track.as_ref().filter(|t| t.id == ev.track_id) {
let _ = bus.send(CoreEvent::TrackStarted {
track: track.clone(),
position_secs: ev.position,
});
last_track_id = ev.track_id;
}
}
// Same three-way mapping as the MPRIS seed: loaded-but-not-playing
// audio is Paused, nothing loaded is Stopped. Dedup on the MAPPED
// state, not `is_playing` — paused and stopped are both
// not-playing, and a pause -> stop transition must still emit.
let state = if ev.is_playing {
PlaybackState::Playing
} else if has_audio {
PlaybackState::Paused
} else {
PlaybackState::Stopped
};
if last_state != Some(state) {
let _ = bus.send(CoreEvent::PlaybackStateChanged { state });
last_state = Some(state);
// A stop ends the "current track": forget it so replaying the
// same track emits a fresh TrackStarted (scrobbling, hooks).
if state == PlaybackState::Stopped {
last_track_id = 0;
}
}
if ev.is_playing {
let _ = bus.send(CoreEvent::PositionUpdated {
position_secs: ev.position,
duration_secs: ev.duration,
});
}
if last_volume.is_none_or(|v| (v - ev.volume).abs() > 0.001) {
let _ = bus.send(CoreEvent::VolumeChanged { volume: ev.volume });
last_volume = Some(ev.volume);
}
}
})
}
+256
View File
@@ -0,0 +1,256 @@
// crates/qbzd/src/hooks.rs — run a user script on daemon events (CONSOLE ext).
//
// Headless integrators (moOde, Volumio, home-grown setups) need PUSH
// notifications to coordinate the daemon with the rest of an audio box — stop
// the local player when a Qobuz Connect session starts playing, restore the
// volume when it ends — without keeping a poller or an SSE client alive.
// `hooks.script` names an executable the daemon forks once per bus event with
// the event described in `QBZ_*` environment variables, the same integration
// contract pleezer (`--hook`) and shairport-sync (`run_this_...`) established.
//
// Enablement: the `QBZD_HOOK` env var wins when set (empty disables);
// otherwise the persisted `daemon_prefs.hook_script` path, written by `qbzd
// settings set hooks.script`. Changing it needs a daemon restart (the
// dispatcher spawns at boot, like MPRIS).
//
// The dispatcher never waits for the script inside the recv loop (a slow
// script would lag the broadcast bus) — each event forks detached and a small
// reaper task collects the exit status. Scripts therefore may overlap; an
// integrator needing strict ordering can flock inside the script.
use std::path::PathBuf;
use qbz_models::{CoreEvent, PlaybackState};
use tokio::sync::broadcast;
use tokio::task::JoinHandle;
use crate::paths::ProfileRoots;
/// Resolve the configured hook script: `QBZD_HOOK` when set (empty string
/// disables), else the persisted `hooks.script` setting; None = hooks off.
pub fn script(roots: &ProfileRoots) -> Option<PathBuf> {
let raw = match std::env::var("QBZD_HOOK") {
Ok(v) => v,
Err(_) => qbz_app::settings::daemon_prefs::load_at(&roots.data).hook_script,
};
let trimmed = raw.trim();
if trimmed.is_empty() {
None
} else {
Some(PathBuf::from(trimmed))
}
}
/// Spawn the hook dispatcher. Holds no `Arc<AppRuntime>` (only the script path
/// and the bus receiver), so it sits outside the §8.2 audio-release ordering —
/// the caller aborts it for a clean shutdown, like the scrobbler.
pub fn spawn(script: PathBuf, mut rx: broadcast::Receiver<CoreEvent>) -> JoinHandle<()> {
use broadcast::error::RecvError;
tokio::spawn(async move {
loop {
let ev = match rx.recv().await {
Ok(ev) => ev,
Err(RecvError::Lagged(_)) => continue,
Err(RecvError::Closed) => return,
};
let Some(vars) = hook_env(&ev) else { continue };
let mut cmd = tokio::process::Command::new(&script);
cmd.envs(vars).stdin(std::process::Stdio::null());
match cmd.spawn() {
Ok(mut child) => {
// Detached reaper: never block the recv loop on the script.
tokio::spawn(async move {
let _ = child.wait().await;
});
}
Err(e) => log::warn!("[hooks] could not run {}: {e}", script.display()),
}
}
})
}
fn state_label(state: PlaybackState) -> &'static str {
match state {
PlaybackState::Playing => "playing",
PlaybackState::Paused => "paused",
PlaybackState::Stopped => "stopped",
PlaybackState::Loading => "loading",
}
}
/// Map one bus event onto the script's `QBZ_*` environment, or None for
/// events the hook does not forward. Deliberately excluded: `PositionUpdated`
/// (a fork every ~2 s), `QueueUpdated` (bulky, and queue state is one
/// `GET /api/status` away), and everything on the SSE suppress list. The
/// `LoggedIn` session payload (auth token) is NEVER forwarded — the event
/// name alone is.
fn hook_env(ev: &CoreEvent) -> Option<Vec<(String, String)>> {
use CoreEvent::*;
let mut vars: Vec<(String, String)> = Vec::new();
let mut push = |k: &str, v: String| vars.push((format!("QBZ_{k}"), v));
match ev {
TrackStarted { track, position_secs } => {
push("EVENT", "TrackStarted".into());
push("TRACK_ID", track.id.to_string());
push("TITLE", track.title.clone());
push("ARTIST", track.artist.clone());
push("ALBUM", track.album.clone());
push("DURATION", track.duration_secs.to_string());
push("POSITION", position_secs.to_string());
if let Some(url) = &track.artwork_url {
push("COVER_URL", url.clone());
}
if let Some(bits) = track.bit_depth {
push("BIT_DEPTH", bits.to_string());
}
if let Some(rate) = track.sample_rate {
push("SAMPLE_RATE", rate.to_string());
}
}
PlaybackStateChanged { state } => {
push("EVENT", "PlaybackStateChanged".into());
push("STATE", state_label(*state).into());
}
TrackEnded { track_id } => {
push("EVENT", "TrackEnded".into());
push("TRACK_ID", track_id.to_string());
}
VolumeChanged { volume } => {
push("EVENT", "VolumeChanged".into());
// 0-100, the same scale the `qbzd volume` verb speaks.
push("VOLUME", (((*volume).clamp(0.0, 1.0) * 100.0).round() as u32).to_string());
}
QconnectSessionChanged { state, device_name, session_active } => {
push("EVENT", "QconnectSessionChanged".into());
push("STATE", state.clone());
push("SESSION_ACTIVE", session_active.to_string());
if let Some(name) = device_name {
push("DEVICE_NAME", name.clone());
}
}
LoggedIn { .. } => push("EVENT", "LoggedIn".into()),
LoggedOut => push("EVENT", "LoggedOut".into()),
SessionExpired => push("EVENT", "SessionExpired".into()),
Error { code, message, recoverable } => {
push("EVENT", "Error".into());
push("CODE", code.clone());
push("MESSAGE", message.clone());
push("RECOVERABLE", recoverable.to_string());
}
PlaybackError { track_id, message } => {
push("EVENT", "PlaybackError".into());
push("TRACK_ID", track_id.to_string());
push("MESSAGE", message.clone());
}
AudioDeviceChanged { device_name } => {
push("EVENT", "AudioDeviceChanged".into());
push("DEVICE_NAME", device_name.clone());
}
_ => return None,
}
// The full tagged JSON for scripts that prefer jq over positional vars —
// except LoggedIn, whose session payload stays out of child environments.
if !matches!(ev, LoggedIn { .. }) {
if let Ok(json) = serde_json::to_string(ev) {
vars.push(("QBZ_EVENT_JSON".to_string(), json));
}
}
Some(vars)
}
#[cfg(test)]
mod tests {
use super::*;
fn get<'a>(vars: &'a [(String, String)], key: &str) -> Option<&'a str> {
vars.iter().find(|(k, _)| k == key).map(|(_, v)| v.as_str())
}
/// A minimal QueueTrack via its serde defaults (the struct has no Default
/// impl, and adding one to the shared models is not this feature's call).
fn test_track() -> qbz_models::QueueTrack {
serde_json::from_value(serde_json::json!({
"id": 42,
"title": "Red Beans",
"artist": "Jon Batiste",
"album": "Monk Movements",
"duration_secs": 163,
"artwork_url": "https://example.com/cover.jpg",
"bit_depth": 24,
"sample_rate": 48.0,
"album_id": null,
"artist_id": null,
}))
.expect("QueueTrack from JSON")
}
#[test]
fn track_started_carries_metadata_and_json() {
let track = test_track();
let vars = hook_env(&CoreEvent::TrackStarted { track, position_secs: 7 })
.expect("TrackStarted is forwarded");
assert_eq!(get(&vars, "QBZ_EVENT"), Some("TrackStarted"));
assert_eq!(get(&vars, "QBZ_TITLE"), Some("Red Beans"));
assert_eq!(get(&vars, "QBZ_ARTIST"), Some("Jon Batiste"));
assert_eq!(get(&vars, "QBZ_DURATION"), Some("163"));
assert_eq!(get(&vars, "QBZ_POSITION"), Some("7"));
assert_eq!(get(&vars, "QBZ_BIT_DEPTH"), Some("24"));
let json = get(&vars, "QBZ_EVENT_JSON").expect("JSON payload present");
assert!(json.contains("\"type\":\"TrackStarted\""));
}
#[test]
fn playback_state_uses_lowercase_labels() {
for (state, label) in [
(PlaybackState::Playing, "playing"),
(PlaybackState::Paused, "paused"),
(PlaybackState::Stopped, "stopped"),
(PlaybackState::Loading, "loading"),
] {
let vars = hook_env(&CoreEvent::PlaybackStateChanged { state }).unwrap();
assert_eq!(get(&vars, "QBZ_STATE"), Some(label));
}
}
#[test]
fn qconnect_session_changed_is_forwarded() {
let vars = hook_env(&CoreEvent::QconnectSessionChanged {
state: "connected".into(),
device_name: Some("Moode Qobuz".into()),
session_active: true,
})
.unwrap();
assert_eq!(get(&vars, "QBZ_EVENT"), Some("QconnectSessionChanged"));
assert_eq!(get(&vars, "QBZ_STATE"), Some("connected"));
assert_eq!(get(&vars, "QBZ_SESSION_ACTIVE"), Some("true"));
assert_eq!(get(&vars, "QBZ_DEVICE_NAME"), Some("Moode Qobuz"));
}
#[test]
fn logged_in_forwards_the_event_name_but_never_the_session() {
let session: qbz_models::UserSession = serde_json::from_value(serde_json::json!({
"user_auth_token": "SECRET-TOKEN",
"user_id": 1,
"email": "user@example.com",
"display_name": "User",
"subscription_label": "Studio",
}))
.expect("UserSession from JSON");
let vars = hook_env(&CoreEvent::LoggedIn { session }).unwrap();
assert_eq!(get(&vars, "QBZ_EVENT"), Some("LoggedIn"));
// Redaction: no JSON payload, no token, exactly one variable.
assert_eq!(vars.len(), 1);
}
#[test]
fn chatty_and_bulky_events_are_not_forwarded() {
assert!(hook_env(&CoreEvent::PositionUpdated { position_secs: 1, duration_secs: 2 }).is_none());
assert!(hook_env(&CoreEvent::ShuffleChanged { enabled: true }).is_none());
assert!(hook_env(&CoreEvent::LoadingStarted { operation: "x".into() }).is_none());
}
#[test]
fn volume_is_forwarded_on_the_cli_percent_scale() {
let vars = hook_env(&CoreEvent::VolumeChanged { volume: 0.45 }).unwrap();
assert_eq!(get(&vars, "QBZ_VOLUME"), Some("45"));
}
}
+2
View File
@@ -5,6 +5,8 @@ mod api;
mod cli;
mod config;
mod daemon;
mod events_bridge;
mod hooks;
mod lock;
mod login;
mod mpris;
+4
View File
@@ -103,6 +103,9 @@ fn latch_lifecycle_into_shared(shared: &SharedState, state: QconnectLifecycleSta
if matches!(state, QconnectLifecycleState::Connected) {
s.set_network_online(true);
}
// Surface the transition on the CoreEvent bus (SSE, `qbzd watch`,
// the event hook) alongside the /api/status latch.
s.emit_qconnect_session_changed();
}
}
@@ -309,6 +312,7 @@ impl DaemonQconnectService {
if let Ok(mut s) = self.shared.lock() {
s.qconnect.state = "off".to_string();
s.qconnect.session_active = false;
s.emit_qconnect_session_changed();
}
Ok(())
}
+1
View File
@@ -129,6 +129,7 @@ impl SessionLoopHost for DaemonSessionLoopHost {
// latch `network.online` false (the success side latches true in
// `latch_lifecycle_into_shared` on the next `Connected` transition).
s.set_network_online(false);
s.emit_qconnect_session_changed();
}
log::warn!("[QConnect] Reconnect exhausted ({attempts}): {last_reason}");
// Daemon has no offline MODE, so the desktop's idle-retry offline park is
+21
View File
@@ -45,6 +45,10 @@ pub struct DaemonShared {
/// funnels through) or a successful QConnect (re)connect. Defaults true
/// (optimistic) until the first outcome latches it.
pub network_online: std::sync::atomic::AtomicBool,
/// CoreEvent bus handle so state mutations here can publish matching bus
/// events (`emit_qconnect_session_changed`). None until `daemon::run`
/// attaches it right after boot; tests leave it None.
pub bus: Option<tokio::sync::broadcast::Sender<qbz_models::CoreEvent>>,
}
impl DaemonShared {
@@ -62,6 +66,21 @@ impl DaemonShared {
self.network_online
.store(online, std::sync::atomic::Ordering::Relaxed);
}
/// Publish the CURRENT `qconnect` block as a `QconnectSessionChanged` bus
/// event (SSE `/api/events`, `qbzd watch`, the event hook). Call AFTER
/// mutating `self.qconnect`; a best-effort no-op before the bus attaches
/// or when no receiver is subscribed.
pub fn emit_qconnect_session_changed(&self) {
if let Some(bus) = &self.bus {
let _ = bus.send(qbz_models::CoreEvent::QconnectSessionChanged {
state: self.qconnect.state.clone(),
device_name: (!self.qconnect.device_name.is_empty())
.then(|| self.qconnect.device_name.clone()),
session_active: self.qconnect.session_active,
});
}
}
}
/// A non-reversible-in-practice fingerprint of a credential token (SipHash via
@@ -141,6 +160,7 @@ mod tests {
qconnect: QconnectStatus::default(),
credential_fingerprint: None,
network_online: std::sync::atomic::AtomicBool::new(true),
bus: None,
};
assert_eq!(shared.auth, AuthState::LoggedIn);
assert_eq!(shared.user_id, Some(1234567));
@@ -164,6 +184,7 @@ mod tests {
qconnect: QconnectStatus::default(),
credential_fingerprint: None,
network_online: std::sync::atomic::AtomicBool::new(true),
bus: None,
};
assert!(shared.network_online(), "defaults true (optimistic)");
+119 -1
View File
@@ -439,7 +439,43 @@ pub async fn apply_renderer_command(
if let Some(value) = resolved_playing_state {
match value {
PLAYING_STATE_PLAYING => {
engine.resume()?;
// A state-only resume (current_track = null, e.g. a
// mid-track handoff from a peer renderer after the
// engine restarted) can land on an engine whose queue
// cursor is set but which holds NO loaded audio — the
// session store restores the queue paused and
// unloaded. A bare resume() then dies in the audio
// thread ("cannot resume - no audio data available")
// while the cloud keeps reporting paused 0:00 to the
// controller forever. Cold-load the cloud's current
// track at its position first, exactly like the
// SetActive takeback path; the has_loaded_audio gate
// keeps echoes and live playback on the plain resume.
let cold_engine = !engine.has_loaded_audio();
let cold_track = projection_renderer_state.current_track.as_ref();
if cold_engine && cold_track.is_some() {
let track_id = cold_track.map(|t| t.track_id).unwrap_or(0);
let start_position_secs = renderer_state
.current_position_ms
.or(*current_position_ms)
.map(|ms| ms / 1000)
.unwrap_or(0);
if let Err(err) = force_remote_track_stream(
engine,
sync_state,
track_id,
projection_renderer_state.max_audio_quality,
start_position_secs,
)
.await
{
log::warn!(
"[QConnect] Cold-start load of remote track {track_id} failed: {err}"
);
}
} else {
engine.resume()?;
}
}
PLAYING_STATE_PAUSED => {
engine.pause()?;
@@ -1244,6 +1280,88 @@ mod tests {
assert_eq!(engine.calls().seeks, vec![40]);
}
/// A state-only resume (current_track = null) on a COLD engine — queue
/// cursor restored but no audio loaded, as after a daemon restart — must
/// cold-load the cloud's current track at the handed-off position instead
/// of issuing a bare resume that dies in the audio thread ("cannot resume
/// - no audio data available") and wedges the controller at paused 0:00.
#[tokio::test]
async fn apply_renderer_command_cold_resume_loads_current_track() {
let mut engine = MockEngine::new();
// Cursor reports the restored track, but nothing is loaded: the plain
// track-id guard would skip, which is why the force path is used.
engine.playback = PlaybackState {
track_id: 7,
..Default::default()
};
engine.queue_tracks = vec![mock_queue_track(7)];
engine.queue_index = Some(0);
engine.loaded_audio = false;
let sync = sync();
let cmd = RendererCommand::SetState {
playing_state: Some(PLAYING_STATE_PLAYING),
current_position_ms: None,
current_track: None,
next_track: None,
};
// The cloud's view carries the session's current track + position.
let renderer_state = QConnectRendererState {
current_track: Some(qi(7, 2)),
current_position_ms: Some(242_491),
..Default::default()
};
apply_renderer_command(&engine, &sync, &cmd, &renderer_state)
.await
.unwrap();
let calls = engine.calls();
assert_eq!(
calls.start_track_streams,
vec![7],
"cold resume must load the session's current track"
);
assert_eq!(
calls.start_positions,
vec![242],
"load must resume at the handed-off position"
);
assert_eq!(calls.resumes, 0, "no bare resume on a cold engine");
}
/// The cold-start load never fires while audio is loaded: a resume during
/// live playback (or a cloud echo) stays a plain resume, no re-stream.
#[tokio::test]
async fn apply_renderer_command_warm_resume_stays_plain() {
let mut engine = MockEngine::new();
engine.playback = PlaybackState {
track_id: 7,
position: 30,
..Default::default()
};
engine.queue_tracks = vec![mock_queue_track(7)];
engine.queue_index = Some(0);
engine.loaded_audio = true;
let sync = sync();
let cmd = RendererCommand::SetState {
playing_state: Some(PLAYING_STATE_PLAYING),
current_position_ms: None,
current_track: None,
next_track: None,
};
let renderer_state = QConnectRendererState {
current_track: Some(qi(7, 2)),
..Default::default()
};
apply_renderer_command(&engine, &sync, &cmd, &renderer_state)
.await
.unwrap();
let calls = engine.calls();
assert!(
calls.start_track_streams.is_empty(),
"no re-stream while audio is loaded"
);
assert_eq!(calls.resumes, 1, "plain resume on a warm engine");
}
/// #4 — WS-authoritative shuffle: a standalone SetShuffleMode must flip the
/// flag ONLY (set_shuffle_flag), NEVER generate a local order. It must not
/// call the order-generating set_shuffle, and must not apply any queue order